#!/usr/bin/env python3 """ ClawTunnel: lightweight single-port HTTP/WebSocket/encrypted TCP/UDP tunnel. Server: python3 clawtunnel.py server --token devtoken --bind 127.0.0.1 --http 8080 --ws 8765 --raw-port 3680 HTTP agent: clawtunnel agent --mode http --token devtoken --name demo --server ws://127.0.0.1:8765 --target http://127.0.0.1:9000 Public: http://127.0.0.1:8080/t/demo/ WebSocket agent: clawtunnel agent --mode ws --token devtoken --name echo --server ws://127.0.0.1:8765 --target ws://127.0.0.1:8766/ws Public: ws://echo.example.com/ws TCP bind/access: clawtunnel bind --mode tcp --token devtoken --name sshdemo --server ws://127.0.0.1:8765 --target 127.0.0.1:22 Bind prints a generated access code. Clients use: clawtunnel access --server example.com:3680 --code --listen 127.0.0.1:2222 ssh -p 2222 user@127.0.0.1 UDP bind/access: clawtunnel bind --mode udp --token devtoken --name dnsdemo --server ws://127.0.0.1:8765 --target 127.0.0.1:53 clawtunnel access --udp --server example.com:3680 --code --listen 127.0.0.1:5353 Security model: - Agents connect outbound over ws:// or wss://. - Server requires a shared admin token before registering tunnels. - Raw TCP and UDP ingress use encrypted frames/datagrams derived from the per-tunnel generated access code. - Use wss:// behind Caddy for encrypted agent transport on real deployments. """ from __future__ import annotations import argparse import asyncio import base64 import hashlib import hmac import http.client import json import os import secrets import signal import socket import struct import sys import threading import time import urllib.parse import urllib.request import urllib.error import uuid from dataclasses import dataclass, field from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Any import websockets from websockets.server import WebSocketServerProtocol try: from cryptography.hazmat.primitives.ciphers.aead import AESGCM from cryptography.hazmat.primitives.kdf.hkdf import HKDF from cryptography.hazmat.primitives import hashes except Exception: # pragma: no cover - exercised only on minimal installs AESGCM = None # type: ignore[assignment] HKDF = None # type: ignore[assignment] hashes = None # type: ignore[assignment] HOP_BY_HOP_HEADERS = {"connection", "keep-alive", "proxy-authenticate", "proxy-authorization", "te", "trailer", "transfer-encoding", "upgrade"} REGISTRY: TunnelRegistry SERVER_LOOP: asyncio.AbstractEventLoop SERVER_TOKEN = "" USER_STORE_PATH = "" USER_STORE_LOCK = threading.RLock() VPN_CACHE_PATH = "" BANNED_IPS_PATH = "" VPN_CACHE_TTL = 7 * 24 * 3600 PUBLIC_BIND = "0.0.0.0" RAW_SERVER: asyncio.AbstractServer | None = None UDP_TRANSPORT: asyncio.DatagramTransport | None = None TCP_MAGIC = b"CLAW2TCP\0" UDP_MAGIC = b"CLAW2UDP\0" KEY_ID_LEN = 16 SALT_LEN = 16 NONCE_LEN = 12 MAX_FRAME = 1024 * 1024 MAX_UDP_PAYLOAD = 60 * 1024 ACCESS_CODE_ALPHABET = "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789" ACCESS_CODE_LEN = 6 def b64(data: bytes) -> str: return base64.b64encode(data).decode("ascii") def unb64(data: str) -> bytes: return base64.b64decode(data.encode("ascii")) def filtered_headers(headers: dict[str, str]) -> dict[str, str]: return {k: v for k, v in headers.items() if k.lower() not in HOP_BY_HOP_HEADERS} def parse_host_port(value: str) -> tuple[str, int]: if ":" not in value: raise ValueError("target must be host:port") host, port_s = value.rsplit(":", 1) return host or "127.0.0.1", int(port_s) def websocket_request_path(websocket: Any) -> str: path = getattr(websocket, "path", None) if path: return path request = getattr(websocket, "request", None) if request is not None: path = getattr(request, "path", None) if path: return path return "/" def make_connect_key() -> str: return "".join(secrets.choice(ACCESS_CODE_ALPHABET) for _ in range(ACCESS_CODE_LEN)) def make_user_token() -> str: return "ctu_" + secrets.token_urlsafe(32) def hash_password(password: str, salt: str | None = None) -> tuple[str, str]: salt = salt or secrets.token_hex(16) digest = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), bytes.fromhex(salt), 200_000).hex() return salt, digest def load_user_store() -> dict[str, Any]: with USER_STORE_LOCK: if not USER_STORE_PATH or not os.path.exists(USER_STORE_PATH): return {"users": {}, "tokens": {}} try: with open(USER_STORE_PATH, "r", encoding="utf-8") as f: data = json.load(f) if not isinstance(data, dict): raise ValueError("bad store") data.setdefault("users", {}); data.setdefault("tokens", {}) return data except Exception: return {"users": {}, "tokens": {}} def save_user_store(data: dict[str, Any]) -> None: with USER_STORE_LOCK: if not USER_STORE_PATH: return os.makedirs(os.path.dirname(USER_STORE_PATH), exist_ok=True) tmp = USER_STORE_PATH + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump(data, f, indent=2, sort_keys=True) os.replace(tmp, USER_STORE_PATH) try: os.chmod(USER_STORE_PATH, 0o600) except Exception: pass def token_owner(token: str) -> str | None: if token and SERVER_TOKEN and hmac.compare_digest(token, SERVER_TOKEN): return "admin" if not token: return None data = load_user_store() info = data.get("tokens", {}).get(token) if not isinstance(info, dict) or info.get("disabled"): return None email = normalize_email(str(info.get("email", ""))) user = data.get("users", {}).get(email) if not isinstance(user, dict) or user.get("disabled") or user.get("banned"): return None if ip_is_banned(str(user.get("signup_ip", "") or "")): return None return email def token_is_allowed(token: str) -> bool: return token_owner(token) is not None def admin_authorized(headers: Any, path: str = "") -> bool: supplied = str(headers.get("X-Admin-Key", "") or "").strip() auth = str(headers.get("Authorization", "") or "").strip() if not supplied and auth.lower().startswith("bearer "): supplied = auth.split(None, 1)[1].strip() if not supplied and path: try: qs = urllib.parse.parse_qs(urllib.parse.urlparse(path).query) supplied = str((qs.get("admin_key") or qs.get("key") or [""])[0] or "").strip() except Exception: supplied = "" return bool(supplied and SERVER_TOKEN and hmac.compare_digest(supplied, SERVER_TOKEN)) def normalize_email(email: str) -> str: return email.strip().lower() def client_ip_from_headers(headers: Any) -> str: xff_values = [p.strip() for p in str(headers.get("X-Forwarded-For", "") or "").split(",") if p.strip()] # Caddy appends the real peer to X-Forwarded-For; use the last hop to avoid client-supplied spoof bypasses. if xff_values: return xff_values[-1] xr = str(headers.get("X-Real-IP", "") or "").strip() if xr: return xr return "" def is_private_or_local_ip(ip: str) -> bool: try: import ipaddress obj = ipaddress.ip_address(ip) return obj.is_private or obj.is_loopback or obj.is_link_local or obj.is_reserved except Exception: return False def load_vpn_cache() -> dict[str, Any]: if not VPN_CACHE_PATH or not os.path.exists(VPN_CACHE_PATH): return {} try: with open(VPN_CACHE_PATH, "r", encoding="utf-8") as f: return json.load(f) except Exception: return {} def save_vpn_cache(cache: dict[str, Any]) -> None: if not VPN_CACHE_PATH: return os.makedirs(os.path.dirname(VPN_CACHE_PATH), exist_ok=True) tmp = VPN_CACHE_PATH + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump(cache, f, indent=2, sort_keys=True) os.replace(tmp, VPN_CACHE_PATH) try: os.chmod(VPN_CACHE_PATH, 0o600) except Exception: pass def load_banned_ips() -> dict[str, Any]: if not BANNED_IPS_PATH or not os.path.exists(BANNED_IPS_PATH): return {"ips": {}} try: with open(BANNED_IPS_PATH, "r", encoding="utf-8") as f: data = json.load(f) if not isinstance(data, dict): return {"ips": {}} data.setdefault("ips", {}) return data except Exception: return {"ips": {}} def save_banned_ips(data: dict[str, Any]) -> None: if not BANNED_IPS_PATH: return data.setdefault("ips", {}) os.makedirs(os.path.dirname(BANNED_IPS_PATH), exist_ok=True) tmp = BANNED_IPS_PATH + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump(data, f, indent=2, sort_keys=True) os.replace(tmp, BANNED_IPS_PATH) try: os.chmod(BANNED_IPS_PATH, 0o600) except Exception: pass def ip_is_banned(ip: str) -> bool: if not ip: return False return ip in load_banned_ips().get("ips", {}) def ban_ip(ip: str, email: str, message: str) -> bool: if not ip or is_private_or_local_ip(ip): return False data = load_banned_ips(); ips = data.setdefault("ips", {}) ips[ip] = {"email": email, "message": message, "banned_at": int(time.time())} save_banned_ips(data) return True def unban_ip(ip: str) -> bool: if not ip: return False data = load_banned_ips(); ips = data.setdefault("ips", {}) existed = ip in ips if existed: ips.pop(ip, None) save_banned_ips(data) return existed def vpn_reputation(ip: str) -> dict[str, Any]: if not ip or is_private_or_local_ip(ip): return {"blocked": False, "reason": "local/private bypass"} now = int(time.time()); cache = load_vpn_cache(); cached = cache.get(ip) if isinstance(cached, dict) and now - int(cached.get("ts", 0)) < VPN_CACHE_TTL: blocked = bool(cached.get("proxy") or cached.get("hosting")) return {"blocked": blocked, "reason": "cached", "data": cached} try: url = "http://ip-api.com/json/" + urllib.parse.quote(ip) + "?fields=status,message,proxy,hosting,query" with urllib.request.urlopen(url, timeout=4) as resp: data = json.loads(resp.read().decode("utf-8")) data["ts"] = now; cache[ip] = data; save_vpn_cache(cache) return {"blocked": bool(data.get("proxy") or data.get("hosting")), "reason": "live", "data": data} except Exception as exc: # Fail open on reputation provider outage; don't lock out real users because ip-api hiccuped. return {"blocked": False, "reason": f"lookup failed: {exc}"} def signup_origin_allowed(headers: Any) -> bool: origin = str(headers.get("Origin", "") or "").rstrip("/") referer = str(headers.get("Referer", "") or "") app_client = str(headers.get("X-ClawTunnel-Client", "") or "").strip().lower() allowed = {"https://www.animefu.space", "https://animefu.space"} if app_client == "android": return True if origin in allowed: return True return any(referer == a or referer.startswith(a + "/") for a in allowed) def require_crypto() -> None: if AESGCM is None or HKDF is None or hashes is None: raise RuntimeError("encrypted raw TCP/UDP requires the 'cryptography' package") def key_bytes(connect_key: str) -> bytes: return connect_key.encode("utf-8") def key_id_for(connect_key: str) -> bytes: return hmac.new(key_bytes(connect_key), b"clawtunnel-v2-key-id", hashlib.sha256).digest()[:KEY_ID_LEN] def hkdf_key(connect_key: str, salt: bytes | None, info: bytes) -> bytes: require_crypto() return HKDF(algorithm=hashes.SHA256(), length=32, salt=salt, info=info).derive(key_bytes(connect_key)) # type: ignore[union-attr] def tcp_keys(connect_key: str, salt: bytes) -> tuple[bytes, bytes]: material = hkdf_key(connect_key, salt, b"clawtunnel-v2-tcp") c2s = hmac.new(material, b"client-to-server", hashlib.sha256).digest() s2c = hmac.new(material, b"server-to-client", hashlib.sha256).digest() return c2s, s2c def udp_key(connect_key: str) -> bytes: return hkdf_key(connect_key, None, b"clawtunnel-v2-udp") def nonce_from_counter(counter: int) -> bytes: return b"\0\0\0\0" + struct.pack("!Q", counter) async def encrypted_write(writer: asyncio.StreamWriter, aes: Any, counter: int, plaintext: bytes, aad: bytes = b"") -> int: if len(plaintext) > MAX_FRAME: raise ValueError("encrypted frame too large") ct = aes.encrypt(nonce_from_counter(counter), plaintext, aad) writer.write(struct.pack("!I", len(ct)) + ct); await writer.drain() return counter + 1 async def encrypted_read(reader: asyncio.StreamReader, aes: Any, counter: int, aad: bytes = b"") -> tuple[bytes, int]: hdr = await reader.readexactly(4) n = struct.unpack("!I", hdr)[0] if n <= 0 or n > MAX_FRAME + 32: raise ValueError("bad encrypted frame length") ct = await reader.readexactly(n) return aes.decrypt(nonce_from_counter(counter), ct, aad), counter + 1 def encrypt_udp(connect_key: str, plaintext: bytes) -> bytes: if len(plaintext) > MAX_UDP_PAYLOAD: raise ValueError("UDP payload too large") kid = key_id_for(connect_key); nonce = secrets.token_bytes(NONCE_LEN); key = udp_key(connect_key) aad = UDP_MAGIC + kid + nonce return aad + AESGCM(key).encrypt(nonce, plaintext, aad) # type: ignore[operator] def decrypt_udp(connect_key: str, packet: bytes) -> bytes: if len(packet) < len(UDP_MAGIC) + KEY_ID_LEN + NONCE_LEN + 16 or not packet.startswith(UDP_MAGIC): raise ValueError("bad UDP packet") kid_off = len(UDP_MAGIC); nonce_off = kid_off + KEY_ID_LEN; ct_off = nonce_off + NONCE_LEN nonce = packet[nonce_off:ct_off]; aad = packet[:ct_off]; ct = packet[ct_off:] return AESGCM(udp_key(connect_key)).decrypt(nonce, ct, aad) # type: ignore[operator] @dataclass class AgentConnection: name: str mode: str websocket: WebSocketServerProtocol | None connect_key: str = "" owner_email: str = "" bind_token: str = "" connected_at: float = field(default_factory=time.time) pending_http: dict[str, asyncio.Future] = field(default_factory=dict) public_tcp_writers: dict[str, asyncio.StreamWriter] = field(default_factory=dict) public_tcp_crypto: dict[str, tuple[Any, int, bytes]] = field(default_factory=dict) local_tcp_writers: dict[str, asyncio.StreamWriter] = field(default_factory=dict) class TunnelRegistry: def __init__(self) -> None: self._agents: dict[str, AgentConnection] = {} self._connect_keys: dict[str, AgentConnection] = {} self._key_ids: dict[bytes, AgentConnection] = {} self._lock = threading.RLock() def get(self, name: str) -> AgentConnection | None: with self._lock: return self._agents.get(name) def get_by_key(self, key: str) -> AgentConnection | None: with self._lock: return self._connect_keys.get(key) def get_by_key_id(self, key_id: bytes) -> AgentConnection | None: with self._lock: return self._key_ids.get(key_id) def names(self) -> list[str]: with self._lock: return sorted(f"{a.name}:{a.mode}" for a in self._agents.values()) def snapshot(self) -> list[dict[str, Any]]: with self._lock: return [ { "name": a.name, "mode": a.mode, "owner_email": a.owner_email, "connected_at": int(a.connected_at), "has_access_code": bool(a.connect_key), "public_tcp_connections": len(a.public_tcp_writers), "pending_http": len(a.pending_http), } for a in sorted(self._agents.values(), key=lambda x: x.name) ] def by_name(self, name: str) -> list[AgentConnection]: with self._lock: return [a for a in self._agents.values() if a.name == name] def by_owner(self, email: str) -> list[AgentConnection]: email = normalize_email(email) with self._lock: return [a for a in self._agents.values() if normalize_email(a.owner_email) == email] def set(self, agent: AgentConnection) -> None: with self._lock: old = self._agents.get(agent.name) if old and old.websocket is not agent.websocket: asyncio.run_coroutine_threadsafe(close_agent(old, "replaced"), SERVER_LOOP) self._agents[agent.name] = agent if agent.mode in {"tcp", "udp"} and agent.connect_key: old_key_agent = self._connect_keys.get(agent.connect_key) if old_key_agent and old_key_agent.websocket is not agent.websocket: asyncio.run_coroutine_threadsafe(close_agent(old_key_agent, "connect key replaced"), SERVER_LOOP) self._connect_keys[agent.connect_key] = agent self._key_ids[key_id_for(agent.connect_key)] = agent def remove(self, agent: AgentConnection) -> None: with self._lock: if self._agents.get(agent.name) is agent: self._agents.pop(agent.name, None) if agent.connect_key and self._connect_keys.get(agent.connect_key) is agent: self._connect_keys.pop(agent.connect_key, None) if agent.connect_key and self._key_ids.get(key_id_for(agent.connect_key)) is agent: self._key_ids.pop(key_id_for(agent.connect_key), None) for fut in agent.pending_http.values(): if not fut.done(): fut.set_exception(ConnectionError("agent disconnected")) REGISTRY = TunnelRegistry() async def ws_send(agent_or_ws: Any, payload: dict[str, Any]) -> None: ws = getattr(agent_or_ws, "websocket", agent_or_ws) if ws is None: raise ConnectionError("websocket is closed") await ws.send(json.dumps(payload)) async def close_agent(agent: AgentConnection, reason: str) -> None: try: if agent.websocket: try: await ws_send(agent, {"type": "admin_disconnect", "reason": reason, "message": reason}) await asyncio.sleep(0.05) except Exception: pass await agent.websocket.close(code=4000, reason=reason[:120]) except Exception: pass # ---------------- HTTP + public WebSocket proxy ---------------- def parse_public_route(handler: BaseHTTPRequestHandler) -> tuple[str | None, str]: parsed = urllib.parse.urlsplit(handler.path) parts = parsed.path.split("/") if len(parts) >= 3 and parts[1] == "t" and parts[2]: name = parts[2] rest = "/" + "/".join(parts[3:]) if parsed.query: rest += "?" + parsed.query return name, rest host = handler.headers.get("Host", "").split(":", 1)[0] if host and "." in host: return host.split(".", 1)[0], handler.path return None, handler.path class PublicHTTPHandler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" def log_message(self, fmt: str, *args: Any) -> None: sys.stderr.write("public-http: " + fmt % args + "\n") def do_GET(self) -> None: self._handle() def do_POST(self) -> None: self._handle() def do_PUT(self) -> None: self._handle() def do_PATCH(self) -> None: self._handle() def do_DELETE(self) -> None: self._handle() def do_HEAD(self) -> None: self._handle() def do_OPTIONS(self) -> None: self._handle() def _send_text(self, status: int, text: str) -> None: body = text.encode("utf-8") self.send_response(status) self.send_header("Content-Type", "text/plain; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() if self.command != "HEAD": self.wfile.write(body) def _send_json(self, status: int, obj: dict[str, Any]) -> None: body = json.dumps(obj, separators=(",", ":")).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Headers", "Content-Type, Authorization, X-Admin-Key, X-ClawTunnel-Client") self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.send_header("Content-Length", str(len(body))) self.end_headers() if self.command != "HEAD": self.wfile.write(body) def _read_json(self) -> dict[str, Any]: length = int(self.headers.get("Content-Length", "0") or "0") if length <= 0 or length > 64 * 1024: return {} return json.loads(self.rfile.read(length).decode("utf-8")) def _handle_signup_or_login(self, signup: bool) -> None: if self.command == "OPTIONS": return self._send_json(200, {"ok": True}) if self.command != "POST": return self._send_json(405, {"ok": False, "error": "POST required"}) try: payload = self._read_json() except Exception: return self._send_json(400, {"ok": False, "error": "invalid json"}) email = normalize_email(str(payload.get("email", ""))) password = str(payload.get("password", "")) if "@" not in email or len(email) > 254: return self._send_json(400, {"ok": False, "error": "valid email required"}) if len(password) < 8: return self._send_json(400, {"ok": False, "error": "password must be at least 8 characters"}) data = load_user_store(); users = data.setdefault("users", {}); tokens = data.setdefault("tokens", {}) now = int(time.time()) user = users.get(email) if signup: if not signup_origin_allowed(self.headers): return self._send_json(403, {"ok": False, "error": "website signup required", "message": "Signup is only available from https://www.animefu.space/clawtunnel/. VPN/proxy/datacenter signups are blocked."}) ip = client_ip_from_headers(self.headers) if ip_is_banned(ip): return self._send_json(403, {"ok": False, "error": "signup source banned", "message": "Signup cannot be completed from this IP address."}) rep = vpn_reputation(ip) if rep.get("blocked"): return self._send_json(403, {"ok": False, "error": "vpn/proxy signup source blocked", "message": "Signup cannot be completed from a VPN, proxy, or datacenter network. Turn off VPN/proxy and try again."}) if user: return self._send_json(409, {"ok": False, "error": "account already exists; use login"}) salt, pw_hash = hash_password(password); token = make_user_token() users[email] = {"email": email, "salt": salt, "password_hash": pw_hash, "created_at": now, "signup_ip": ip, "disabled": False, "tokens": [token]} tokens[token] = {"email": email, "created_at": now, "disabled": False} save_user_store(data) return self._send_json(200, {"ok": True, "email": email, "token": token, "message": "Use this token as CLAWTUNNEL_TOKEN for bind."}) if not user or user.get("disabled"): return self._send_json(401, {"ok": False, "error": "bad email or password"}) login_ip = client_ip_from_headers(self.headers) if ip_is_banned(login_ip): return self._send_json(403, {"ok": False, "error": "login source banned", "message": "Login cannot be completed from this IP address."}) salt = str(user.get("salt", "")); expected = str(user.get("password_hash", "")); _, actual = hash_password(password, salt) if not hmac.compare_digest(actual, expected): return self._send_json(401, {"ok": False, "error": "bad email or password"}) token = make_user_token(); user.setdefault("tokens", []).append(token); user["last_login_at"] = now; tokens[token] = {"email": email, "created_at": now, "disabled": False} save_user_store(data) return self._send_json(200, {"ok": True, "email": email, "token": token, "message": "Use this token as CLAWTUNNEL_TOKEN for bind."}) def _admin_forbidden(self) -> bool: if not admin_authorized(self.headers, self.path): self._send_json(401, {"ok": False, "error": "admin key required"}) return True return False def _admin_user_rows(self) -> list[dict[str, Any]]: data = load_user_store(); users = data.get("users", {}); tokens = data.get("tokens", {}) rows = [] for email, user in sorted(users.items()): user_tokens = list(user.get("tokens", []) or []) disabled_tokens = sum(1 for t in user_tokens if isinstance(tokens.get(t), dict) and tokens.get(t, {}).get("disabled")) rows.append({ "email": email, "disabled": bool(user.get("disabled")), "banned": bool(user.get("banned")), "created_at": user.get("created_at"), "last_login_at": user.get("last_login_at"), "ban_message": user.get("ban_message", ""), "signup_ip": user.get("signup_ip", ""), "token_count": len(user_tokens), "disabled_token_count": disabled_tokens, "active_tunnel_count": len(REGISTRY.by_owner(email)), }) return rows def _admin_kick_agents(self, agents: list[AgentConnection], message: str) -> int: count = 0 for agent in agents: if SERVER_LOOP and agent.websocket: asyncio.run_coroutine_threadsafe(close_agent(agent, message), SERVER_LOOP) count += 1 return count def _handle_admin(self, path: str) -> None: if self.command == "OPTIONS": return self._send_json(200, {"ok": True}) if self._admin_forbidden(): return if path in {"/api/admin/state", "/api/admin/users"} and self.command == "GET": return self._send_json(200, {"ok": True, "users": self._admin_user_rows(), "tunnels": REGISTRY.snapshot()}) if path == "/api/admin/tunnels" and self.command == "GET": return self._send_json(200, {"ok": True, "tunnels": REGISTRY.snapshot()}) if self.command != "POST": return self._send_json(405, {"ok": False, "error": "POST required"}) try: payload = self._read_json() except Exception: return self._send_json(400, {"ok": False, "error": "invalid json"}) email = normalize_email(str(payload.get("email", ""))) name = str(payload.get("name", "")).strip().lower() message = str(payload.get("message", "") or "").strip() data = load_user_store(); users = data.setdefault("users", {}); tokens = data.setdefault("tokens", {}) if path == "/api/admin/reset_password": if not email or email not in users: return self._send_json(404, {"ok": False, "error": "user not found"}) new_password = str(payload.get("new_password", "") or "") generated = False if not new_password: new_password = "Ct-" + secrets.token_urlsafe(12) generated = True if len(new_password) < 8: return self._send_json(400, {"ok": False, "error": "password must be at least 8 characters"}) salt, pw_hash = hash_password(new_password) users[email]["salt"] = salt; users[email]["password_hash"] = pw_hash; users[email]["password_reset_at"] = int(time.time()) save_user_store(data) out = {"ok": True, "email": email, "message": "password reset"} if generated: out["new_password"] = new_password return self._send_json(200, out) if path == "/api/admin/ban": if not email or email not in users: return self._send_json(404, {"ok": False, "error": "user not found"}) if not message: message = "Banned by ClawTunnel admin" users[email]["disabled"] = True; users[email]["banned"] = True; users[email]["ban_message"] = message; users[email]["banned_at"] = int(time.time()) for t in users[email].get("tokens", []) or []: if isinstance(tokens.get(t), dict): tokens[t]["disabled"] = True signup_ip = str(users[email].get("signup_ip", "") or "").strip() ip_banned = ban_ip(signup_ip, email, message) save_user_store(data) kicked = self._admin_kick_agents(REGISTRY.by_owner(email), message) return self._send_json(200, {"ok": True, "email": email, "banned": True, "kicked": kicked, "message": message, "signup_ip": signup_ip, "ip_banned": ip_banned}) if path == "/api/admin/unban": if not email or email not in users: return self._send_json(404, {"ok": False, "error": "user not found"}) users[email]["disabled"] = False; users[email]["banned"] = False; users[email]["unbanned_at"] = int(time.time()) for t in users[email].get("tokens", []) or []: if isinstance(tokens.get(t), dict): tokens[t]["disabled"] = False signup_ip = str(users[email].get("signup_ip", "") or "").strip() ip_unbanned = unban_ip(signup_ip) save_user_store(data) return self._send_json(200, {"ok": True, "email": email, "banned": False, "signup_ip": signup_ip, "ip_unbanned": ip_unbanned}) if path == "/api/admin/kick": if not message: message = "Kicked by ClawTunnel admin" agents: list[AgentConnection] = [] if email: agents.extend(REGISTRY.by_owner(email)) if name: agents.extend(REGISTRY.by_name(name)) # preserve order but drop duplicates seen: set[int] = set(); unique: list[AgentConnection] = [] for a in agents: if id(a) not in seen: seen.add(id(a)); unique.append(a) kicked = self._admin_kick_agents(unique, message) return self._send_json(200, {"ok": True, "kicked": kicked, "message": message}) return self._send_json(404, {"ok": False, "error": "unknown admin endpoint"}) def _handle(self) -> None: path = urllib.parse.urlsplit(self.path).path if path == "/api/signup": return self._handle_signup_or_login(True) if path == "/api/login": return self._handle_signup_or_login(False) if path.startswith("/api/admin/"): return self._handle_admin(path) if path == "/__claw/health": return self._send_text(200, "ok\n") if path == "/__claw/tunnels": return self._send_text(200, "\n".join(REGISTRY.names()) + "\n") name, upstream_path = parse_public_route(self) if not name: return self._send_text(404, "No tunnel selected. Use /t//... or a wildcard host.\n") agent = REGISTRY.get(name) if not agent or agent.mode not in {"http", "ws"}: return self._send_text(502, f"Tunnel '{name}' is not connected.\n") if name.lower().startswith("jelly") and (upstream_path or "/") in {"/", "/web"}: host = self.headers.get("Host", name + ".animefu.space") self.send_response(302) self.send_header("Location", f"https://{host}/web/") self.send_header("Content-Length", "0") self.end_headers() return if self.headers.get("Upgrade", "").lower() == "websocket": if agent.mode not in {"http", "ws"}: return self._send_text(426, "Tunnel is not registered for websocket upgrade forwarding.\n") return self._handle_websocket(agent, upstream_path) if agent.mode != "http": return self._send_text(426, "This tunnel only accepts WebSocket Upgrade requests.\n") length = int(self.headers.get("Content-Length", "0") or "0") body = self.rfile.read(length) if length else b"" low_path = (upstream_path or "/").lower() if name.lower() == "book" and ("/public/session/" in low_path or "/hls/" in low_path or "/stream" in low_path or "/download" in low_path): return self._handle_raw_http_stream(agent, upstream_path, body) request_id = uuid.uuid4().hex payload = {"type":"http_request","id":request_id,"method":self.command,"path":upstream_path or "/","headers":filtered_headers({k:v for k,v in self.headers.items()}),"body_b64":b64(body)} fut = asyncio.run_coroutine_threadsafe(send_http_request(agent, payload), SERVER_LOOP) try: response = fut.result(timeout=45) except Exception as exc: return self._send_text(504, f"Tunnel request failed: {exc}\n") status = int(response.get("status", 502)); resp_headers = filtered_headers(response.get("headers", {})); resp_body = unb64(response.get("body_b64", "")) self.send_response(status) for key, value in resp_headers.items(): if key.lower() in {"content-length"}: continue self.send_header(key, value) self.send_header("X-ClawTunnel", "1"); self.send_header("Content-Length", str(len(resp_body))); self.end_headers() if self.command != "HEAD": self.wfile.write(resp_body) def _raw_initial_request(self, upstream_path: str, body: bytes = b"", force_close: bool = False) -> bytes: host = self.headers.get("Host", "") lines = [f"{self.command} {upstream_path or '/'} HTTP/1.1"] saw_connection = False for k, v in self.headers.items(): lk = k.lower() if lk == "host": lines.append(f"Host: {host}") elif lk == "connection": saw_connection = True lines.append("Connection: close" if force_close else f"{k}: {v}") else: lines.append(f"{k}: {v}") if force_close and not saw_connection: lines.append("Connection: close") return ("\r\n".join(lines) + "\r\n\r\n").encode("iso-8859-1") + (body or b"") def _handle_raw_http_stream(self, agent: AgentConnection, upstream_path: str, body: bytes = b"") -> None: stream_id = uuid.uuid4().hex initial = self._raw_initial_request(upstream_path, body, force_close=True) self.connection.settimeout(3600) fut = asyncio.run_coroutine_threadsafe(open_public_tcp_stream(agent, stream_id, self.connection, initial), SERVER_LOOP) try: fut.result(timeout=3600) except Exception: pass self.close_connection = True def _handle_websocket(self, agent: AgentConnection, upstream_path: str) -> None: stream_id = uuid.uuid4().hex initial = self._raw_initial_request(upstream_path, b"", force_close=False) self.connection.settimeout(3600) fut = asyncio.run_coroutine_threadsafe(open_public_tcp_stream(agent, stream_id, self.connection, initial), SERVER_LOOP) try: fut.result(timeout=3600) except Exception: pass self.close_connection = True async def send_http_request(agent: AgentConnection, payload: dict[str, Any]) -> dict[str, Any]: request_id = payload["id"] fut = asyncio.get_running_loop().create_future(); agent.pending_http[request_id] = fut try: await ws_send(agent, payload) return await asyncio.wait_for(fut, timeout=40) finally: agent.pending_http.pop(request_id, None) def run_public_http(bind: str, port: int) -> ThreadingHTTPServer: httpd = ThreadingHTTPServer((bind, port), PublicHTTPHandler) threading.Thread(target=httpd.serve_forever, daemon=True).start(); return httpd # ---------------- Single-port raw TCP ingress ---------------- async def open_public_tcp_stream(agent: AgentConnection, stream_id: str, sock: socket.socket, initial_data: bytes = b"") -> None: if agent.websocket is None: raise ConnectionError("agent disconnected") sock.setblocking(False) loop = asyncio.get_running_loop() reader, writer = await asyncio.open_connection(sock=sock) agent.public_tcp_writers[stream_id] = writer peer = writer.get_extra_info("peername") try: await ws_send(agent, {"type":"tcp_open","id":stream_id,"peer":str(peer)}) if initial_data: await ws_send(agent, {"type":"tcp_data","id":stream_id,"body_b64":b64(initial_data)}) while True: data = await reader.read(32768) if not data: break await ws_send(agent, {"type":"tcp_data","id":stream_id,"body_b64":b64(data)}) except Exception: pass finally: agent.public_tcp_writers.pop(stream_id, None) try: await ws_send(agent, {"type":"tcp_close","id":stream_id}) except Exception: pass writer.close(); await writer.wait_closed() async def handle_raw_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: stream_id = uuid.uuid4().hex agent: AgentConnection | None = None try: require_crypto() header = await asyncio.wait_for(reader.readexactly(len(TCP_MAGIC) + KEY_ID_LEN + SALT_LEN), timeout=10) if not header.startswith(TCP_MAGIC): writer.close(); await writer.wait_closed(); return kid_off = len(TCP_MAGIC); salt_off = kid_off + KEY_ID_LEN key_id = header[kid_off:salt_off]; salt = header[salt_off:salt_off + SALT_LEN] agent = REGISTRY.get_by_key_id(key_id) if not agent or agent.mode != "tcp" or agent.websocket is None or not agent.connect_key: writer.close(); await writer.wait_closed(); return c2s_key, s2c_key = tcp_keys(agent.connect_key, salt) c2s = AESGCM(c2s_key); s2c = AESGCM(s2c_key) # type: ignore[operator] recv_ctr = send_ctr = 0 opened, recv_ctr = await encrypted_read(reader, c2s, recv_ctr, header) if opened != b"OPEN": writer.close(); await writer.wait_closed(); return send_ctr = await encrypted_write(writer, s2c, send_ctr, b"OK", header) agent.public_tcp_writers[stream_id] = writer agent.public_tcp_crypto[stream_id] = (s2c, send_ctr, header) peer = writer.get_extra_info("peername") await ws_send(agent, {"type":"tcp_open","id":stream_id,"peer":str(peer),"encrypted":True}) while True: data, recv_ctr = await encrypted_read(reader, c2s, recv_ctr, header) if not data: break await ws_send(agent, {"type":"tcp_data","id":stream_id,"body_b64":b64(data)}) except Exception: pass finally: for a in list(REGISTRY._agents.values()): a.public_tcp_writers.pop(stream_id, None); a.public_tcp_crypto.pop(stream_id, None) try: if agent: await ws_send(agent, {"type":"tcp_close","id":stream_id}) except Exception: pass writer.close(); await writer.wait_closed() async def start_raw_listener(bind: str, port: int) -> asyncio.AbstractServer: server = await asyncio.start_server(handle_raw_client, bind, port) sockets = ", ".join(str(s.getsockname()) for s in (server.sockets or [])) print(f"single-port raw TCP ingress listening on {sockets}", flush=True) return server # ---------------- Single-port encrypted UDP ingress ---------------- UDP_SESSIONS: dict[str, tuple[tuple[str, int], AgentConnection, float]] = {} def udp_plain(session_id: str, body: bytes) -> bytes: return json.dumps({"id":session_id,"body_b64":b64(body)}, separators=(",", ":")).encode("utf-8") def parse_udp_plain(data: bytes) -> tuple[str, bytes]: obj = json.loads(data.decode("utf-8")) return str(obj.get("id", "")), unb64(obj.get("body_b64", "")) class RawUDPProtocol(asyncio.DatagramProtocol): def connection_made(self, transport: asyncio.BaseTransport) -> None: global UDP_TRANSPORT UDP_TRANSPORT = transport # type: ignore[assignment] def datagram_received(self, data: bytes, addr: tuple[str, int]) -> None: asyncio.create_task(handle_udp_datagram(data, addr)) async def handle_udp_datagram(data: bytes, addr: tuple[str, int]) -> None: try: if not data.startswith(UDP_MAGIC): return kid_off = len(UDP_MAGIC); key_id = data[kid_off:kid_off + KEY_ID_LEN] agent = REGISTRY.get_by_key_id(key_id) if not agent or agent.mode != "udp" or agent.websocket is None or not agent.connect_key: return session_id, body = parse_udp_plain(decrypt_udp(agent.connect_key, data)) if not session_id or len(body) > MAX_UDP_PAYLOAD: return UDP_SESSIONS[session_id] = (addr, agent, time.time()) await ws_send(agent, {"type":"udp_datagram","id":session_id,"peer":str(addr),"body_b64":b64(body)}) except Exception: pass def send_udp_to_client(agent: AgentConnection, session_id: str, body: bytes) -> None: if not UDP_TRANSPORT or not session_id or not agent.connect_key: return sess = UDP_SESSIONS.get(session_id) if not sess or sess[1] is not agent: return addr = sess[0] try: UDP_TRANSPORT.sendto(encrypt_udp(agent.connect_key, udp_plain(session_id, body)), addr) except Exception: pass async def start_udp_listener(bind: str, port: int) -> asyncio.DatagramTransport: loop = asyncio.get_running_loop() transport, _ = await loop.create_datagram_endpoint(lambda: RawUDPProtocol(), local_addr=(bind, port)) print(f"single-port encrypted UDP ingress listening on {(bind, port)}", flush=True) return transport # type: ignore[return-value] # ---------------- Server WebSocket handler ---------------- async def agent_ws_handler(websocket: WebSocketServerProtocol) -> None: parsed = urllib.parse.urlsplit(websocket_request_path(websocket)); qs = urllib.parse.parse_qs(parsed.query) token = qs.get("token", [""])[0]; name = qs.get("name", [""])[0].strip().lower(); mode = qs.get("mode", ["http"])[0].strip().lower(); connect_key = qs.get("connect_key", [""])[0].strip() owner = token_owner(token) if owner is None: await websocket.close(code=4001, reason="bad token"); return if not name or not all(c.isalnum() or c in "-_" for c in name): await websocket.close(code=4002, reason="bad tunnel name"); return if mode not in {"http", "ws", "tcp", "udp"}: await websocket.close(code=4003, reason="bad mode"); return if mode in {"tcp", "udp"} and not connect_key: for _ in range(64): candidate = make_connect_key() if REGISTRY.get_by_key(candidate) is None: connect_key = candidate break else: await websocket.close(code=4004, reason="access code space exhausted"); return agent = AgentConnection(name=name, mode=mode, websocket=websocket, connect_key=connect_key, owner_email=owner, bind_token=token) try: REGISTRY.set(agent) print(f"agent connected: {name} mode={mode}" + (" access_code=yes" if connect_key else ""), flush=True) if mode in {"tcp", "udp"}: await websocket.send(json.dumps({"type":"registered","name":name,"mode":mode,"connect_key":connect_key})) async for message in websocket: data = json.loads(message); typ = data.get("type") if typ == "http_response": fut = agent.pending_http.get(data.get("id")) if fut and not fut.done(): fut.set_result(data) elif typ == "tcp_data": writer = agent.public_tcp_writers.get(data.get("id")) if writer: body = unb64(data.get("body_b64", "")); enc = agent.public_tcp_crypto.get(data.get("id")) if enc: aes, counter, aad = enc; counter = await encrypted_write(writer, aes, counter, body, aad); agent.public_tcp_crypto[data.get("id")] = (aes, counter, aad) else: try: writer.write(body) await writer.drain() except Exception as exc: print(f"public tcp stream write error {agent.name}:{data.get('id')}: {type(exc).__name__}: {exc!r}", flush=True) agent.public_tcp_writers.pop(data.get("id"), None) try: writer.close(); await writer.wait_closed() except Exception: pass elif typ == "tcp_close": writer = agent.public_tcp_writers.pop(data.get("id"), None) agent.public_tcp_crypto.pop(data.get("id"), None) if writer: writer.close(); await writer.wait_closed() elif typ == "udp_datagram": send_udp_to_client(agent, data.get("id", ""), unb64(data.get("body_b64", ""))) except Exception as exc: print(f"agent error {name}: {exc}", flush=True) finally: REGISTRY.remove(agent) for writer in list(agent.public_tcp_writers.values()) + list(agent.local_tcp_writers.values()): writer.close() agent.public_tcp_writers.clear(); agent.public_tcp_crypto.clear(); agent.local_tcp_writers.clear() print(f"agent disconnected: {name}", flush=True) async def run_server(args: argparse.Namespace) -> None: global SERVER_LOOP, SERVER_TOKEN, USER_STORE_PATH, VPN_CACHE_PATH, BANNED_IPS_PATH, PUBLIC_BIND, RAW_SERVER SERVER_LOOP = asyncio.get_running_loop(); SERVER_TOKEN = args.token or os.environ.get("CLAWTUNNEL_TOKEN", ""); USER_STORE_PATH = args.user_store or os.environ.get("CLAWTUNNEL_USER_STORE", ""); VPN_CACHE_PATH = args.vpn_cache or os.environ.get("CLAWTUNNEL_VPN_CACHE", ""); BANNED_IPS_PATH = os.environ.get("CLAWTUNNEL_BANNED_IPS", ""); PUBLIC_BIND = args.public_bind if not BANNED_IPS_PATH and USER_STORE_PATH: BANNED_IPS_PATH = os.path.join(os.path.dirname(USER_STORE_PATH), "banned_ips.json") if not SERVER_TOKEN: raise SystemExit("server requires --token or CLAWTUNNEL_TOKEN") if USER_STORE_PATH: save_user_store(load_user_store()) httpd = run_public_http(args.bind, args.http) RAW_SERVER = await start_raw_listener(PUBLIC_BIND, args.raw_port) udp_transport = await start_udp_listener(PUBLIC_BIND, args.raw_port) print(f"public HTTP listening on http://{args.bind}:{args.http}", flush=True); print(f"agent WebSocket listening on ws://{args.bind}:{args.ws}", flush=True) stop = asyncio.Event() for sig in (signal.SIGINT, signal.SIGTERM): try: SERVER_LOOP.add_signal_handler(sig, stop.set) except NotImplementedError: pass async with websockets.serve(agent_ws_handler, args.bind, args.ws, ping_interval=30, ping_timeout=120, max_size=16*1024*1024, max_queue=1024): await stop.wait() RAW_SERVER.close(); await RAW_SERVER.wait_closed(); udp_transport.close(); httpd.shutdown() # ---------------- Agent-side forwarding ---------------- def forward_http_to_local(target: str, req: dict[str, Any]) -> dict[str, Any]: parsed_target = urllib.parse.urlsplit(target); scheme = parsed_target.scheme or "http"; host = parsed_target.hostname or "127.0.0.1"; port = parsed_target.port or (443 if scheme == "https" else 80); base_path = parsed_target.path.rstrip("/"); path = base_path + req.get("path", "/") headers = filtered_headers(req.get("headers", {})); headers["Host"] = f"{host}:{port}"; headers["Accept-Encoding"] = "identity"; body = unb64(req.get("body_b64", "")) conn_cls = http.client.HTTPSConnection if scheme == "https" else http.client.HTTPConnection; conn = conn_cls(host, port, timeout=30) try: conn.request(req.get("method", "GET"), path, body=body, headers=headers); resp = conn.getresponse(); resp_body = resp.read() return {"type":"http_response","id":req["id"],"status":resp.status,"headers":{k:v for k,v in resp.getheaders()},"body_b64":b64(resp_body)} except Exception as exc: return {"type":"http_response","id":req["id"],"status":502,"headers":{"Content-Type":"text/plain; charset=utf-8"},"body_b64":b64(f"local target error: {exc}\n".encode())} finally: conn.close() async def agent_open_local_tcp(agent: AgentConnection, ws: Any, target: str, stream_id: str) -> None: parsed = urllib.parse.urlsplit(target) if parsed.scheme in {"ws", "wss", "http", "https"}: host = parsed.hostname or "127.0.0.1"; port = parsed.port or (443 if parsed.scheme in {"wss", "https"} else 80) else: host, port = parse_host_port(target) try: reader, writer = await asyncio.open_connection(host, port, ssl=(parsed.scheme in {"wss", "https"} if parsed.scheme else False)) except Exception as exc: await ws.send(json.dumps({"type":"tcp_data","id":stream_id,"body_b64":b64(f"local tcp connect error: {exc}\n".encode())})); await ws.send(json.dumps({"type":"tcp_close","id":stream_id})); return agent.local_tcp_writers[stream_id] = writer try: while True: data = await reader.read(32768) if not data: break await ws.send(json.dumps({"type":"tcp_data","id":stream_id,"body_b64":b64(data)})) except Exception: pass finally: agent.local_tcp_writers.pop(stream_id, None) try: await ws.send(json.dumps({"type":"tcp_close","id":stream_id})) except Exception: pass writer.close(); await writer.wait_closed() class AgentUDPProtocol(asyncio.DatagramProtocol): def __init__(self, ws: Any, session_id: str) -> None: self.ws = ws; self.session_id = session_id def datagram_received(self, data: bytes, addr: tuple[str, int]) -> None: asyncio.create_task(self.ws.send(json.dumps({"type":"udp_datagram","id":self.session_id,"body_b64":b64(data)}))) async def agent_send_local_udp(agent: AgentConnection, ws: Any, target: str, session_id: str, body: bytes) -> None: transports = getattr(agent, "local_udp_transports", None) if transports is None: transports = {}; setattr(agent, "local_udp_transports", transports) transport = transports.get(session_id) if transport is None or transport.is_closing(): host, port = parse_host_port(target) loop = asyncio.get_running_loop() transport, _ = await loop.create_datagram_endpoint(lambda: AgentUDPProtocol(ws, session_id), remote_addr=(host, port)) transports[session_id] = transport transport.sendto(body) async def run_agent(args: argparse.Namespace) -> None: token = args.token or os.environ.get("CLAWTUNNEL_TOKEN", "") if not token: raise SystemExit("agent requires --token or CLAWTUNNEL_TOKEN") qs = urllib.parse.urlencode({"token":token,"name":args.name,"mode":args.mode,"connect_key":args.connect_key or ""}) url = args.server.rstrip("/") + "/agent?" + qs local_agent = AgentConnection(name=args.name, mode=args.mode, websocket=None, connect_key=args.connect_key or "") while True: try: print(f"connecting to {url.split('token=')[0]}token=REDACTED...; forwarding to {args.target}", flush=True) async with websockets.connect(url, ping_interval=30, ping_timeout=120, max_size=16*1024*1024, max_queue=1024) as ws: print(f"connected as tunnel '{args.name}' mode={args.mode}", flush=True) if args.mode == "http": print(f"public URL: https://{args.name}.animefu.space/", flush=True) print("no access code is generated for http mode; use the public URL above", flush=True) elif args.mode == "ws": print(f"public WebSocket URL: wss://{args.name}.animefu.space/", flush=True) print("no access code is generated for ws mode; use the public WSS URL above", flush=True) elif args.mode in {"tcp", "udp"}: print(f"waiting for generated {args.mode.upper()} access code...", flush=True) async for message in ws: req = json.loads(message); typ = req.get("type") if typ == "admin_disconnect": print(f"admin disconnect: {req.get('message') or req.get('reason') or 'disconnected by admin'}", flush=True) elif typ == "registered" and req.get("connect_key"): print(f"access code for {args.name}: {req.get('connect_key')}", flush=True) print(f"compat connect key for {args.name}: {req.get('connect_key')}", flush=True) elif typ == "http_request": await ws.send(json.dumps(await asyncio.to_thread(forward_http_to_local, args.target, req))) elif typ == "tcp_open": asyncio.create_task(agent_open_local_tcp(local_agent, ws, args.target, req["id"])) elif typ == "tcp_data": writer = local_agent.local_tcp_writers.get(req.get("id")) if not writer: for _ in range(100): await asyncio.sleep(0.01); writer = local_agent.local_tcp_writers.get(req.get("id")) if writer: break if writer: try: writer.write(unb64(req.get("body_b64", ""))) await writer.drain() except Exception as exc: print(f"local tcp stream write error {req.get('id')}: {type(exc).__name__}: {exc!r}", flush=True) local_agent.local_tcp_writers.pop(req.get("id"), None) try: writer.close(); await writer.wait_closed() except Exception: pass try: await ws.send(json.dumps({"type":"tcp_close","id":req.get("id")})) except Exception: pass elif typ == "tcp_close": writer = local_agent.local_tcp_writers.pop(req.get("id"), None) if writer: writer.close(); await writer.wait_closed() elif typ == "udp_datagram": asyncio.create_task(agent_send_local_udp(local_agent, ws, args.target, req.get("id", ""), unb64(req.get("body_b64", "")))) except KeyboardInterrupt: return except Exception as exc: print(f"agent disconnected/error: {exc}; reconnecting in 2s", flush=True); await asyncio.sleep(2) async def pipe_streams(src: asyncio.StreamReader, dst: asyncio.StreamWriter) -> None: try: while True: data = await src.read(32768) if not data: break dst.write(data); await dst.drain() except Exception: pass finally: dst.close() async def pipe_local_to_encrypted(src: asyncio.StreamReader, dst: asyncio.StreamWriter, aes: Any, aad: bytes) -> None: counter = 1 try: while True: data = await src.read(32768) if not data: break counter = await encrypted_write(dst, aes, counter, data, aad) except Exception: pass finally: dst.close() async def pipe_encrypted_to_local(src: asyncio.StreamReader, dst: asyncio.StreamWriter, aes: Any, aad: bytes) -> None: counter = 1 try: while True: data, counter = await encrypted_read(src, aes, counter, aad) if not data: break dst.write(data); await dst.drain() except Exception: pass finally: dst.close() class ConnectUDPProtocol(asyncio.DatagramProtocol): def __init__(self, server: tuple[str, int], key: str) -> None: self.server = server; self.key = key; self.transport: asyncio.DatagramTransport | None = None self.peer_to_session: dict[tuple[str, int], str] = {}; self.session_to_peer: dict[str, tuple[str, int]] = {} def connection_made(self, transport: asyncio.BaseTransport) -> None: self.transport = transport # type: ignore[assignment] def datagram_received(self, data: bytes, addr: tuple[str, int]) -> None: if addr == self.server: try: session_id, body = parse_udp_plain(decrypt_udp(self.key, data)) peer = self.session_to_peer.get(session_id) if peer and self.transport: self.transport.sendto(body, peer) except Exception: pass return session_id = self.peer_to_session.get(addr) if not session_id: session_id = uuid.uuid4().hex; self.peer_to_session[addr] = session_id; self.session_to_peer[session_id] = addr try: if self.transport: self.transport.sendto(encrypt_udp(self.key, udp_plain(session_id, data)), self.server) except Exception: pass async def run_udp_connect(args: argparse.Namespace) -> None: require_crypto() server_host, server_port = parse_host_port(args.server); listen_host, listen_port = parse_host_port(args.listen) loop = asyncio.get_running_loop(); server_addr = (socket.gethostbyname(server_host), server_port) transport, _ = await loop.create_datagram_endpoint(lambda: ConnectUDPProtocol(server_addr, args.key), local_addr=(listen_host, listen_port)) print(f"local UDP connector listening on {listen_host}:{listen_port} -> {args.server} key={args.key[:6]}...", flush=True) try: await asyncio.Future() finally: transport.close() async def run_connect(args: argparse.Namespace) -> None: if not getattr(args, "key", "") and getattr(args, "code", ""): args.key = args.code if not getattr(args, "key", ""): raise SystemExit("access/connect requires --code or --key") if getattr(args, "udp", False): await run_udp_connect(args); return require_crypto() server_host, server_port = parse_host_port(args.server); listen_host, listen_port = parse_host_port(args.listen) async def handle_local(local_reader: asyncio.StreamReader, local_writer: asyncio.StreamWriter) -> None: try: remote_reader, remote_writer = await asyncio.open_connection(server_host, server_port) salt = secrets.token_bytes(SALT_LEN); header = TCP_MAGIC + key_id_for(args.key) + salt c2s_key, s2c_key = tcp_keys(args.key, salt) c2s = AESGCM(c2s_key); s2c = AESGCM(s2c_key) # type: ignore[operator] remote_writer.write(header); await remote_writer.drain() await encrypted_write(remote_writer, c2s, 0, b"OPEN", header) status, _ = await encrypted_read(remote_reader, s2c, 0, header) if status != b"OK": local_writer.write(b"ERR connect failed\n"); await local_writer.drain(); local_writer.close(); return await asyncio.gather(pipe_local_to_encrypted(local_reader, remote_writer, c2s, header), pipe_encrypted_to_local(remote_reader, local_writer, s2c, header)) except Exception as exc: try: local_writer.write(f"connect error: {exc}\n".encode()); await local_writer.drain() except Exception: pass finally: local_writer.close() server = await asyncio.start_server(handle_local, listen_host, listen_port) print(f"local encrypted TCP connector listening on {listen_host}:{listen_port} -> {args.server} key={args.key[:6]}...", flush=True) async with server: await server.serve_forever() def run_account(args: argparse.Namespace, signup: bool) -> None: if signup: raise SystemExit("Signup is website-only. Open https://www.animefu.space/clawtunnel/ and use the signup form. VPN/proxy/datacenter signups are blocked; turn off VPN before signing up.") email = args.email.strip().lower() password = args.password if not email: email = input("Email: ").strip().lower() if not password: import getpass password = getpass.getpass("Password: ") base = args.server.rstrip("/") url = base + ("/api/signup" if signup else "/api/login") body = json.dumps({"email": email, "password": password}).encode("utf-8") req = urllib.request.Request(url, data=body, headers={"Content-Type":"application/json"}, method="POST") try: with urllib.request.urlopen(req, timeout=20) as resp: data = json.loads(resp.read().decode("utf-8")) except urllib.error.HTTPError as exc: try: data = json.loads(exc.read().decode("utf-8")) except Exception: data = {"ok": False, "error": str(exc)} raise SystemExit(f"{('signup' if signup else 'login')} failed: {data.get('error', data)}") if not data.get("ok") or not data.get("token"): raise SystemExit(f"{('signup' if signup else 'login')} failed: {data.get('error', data)}") token = str(data["token"]) print("Signup OK." if signup else "Login OK.") print("Use this for bind:") print("export CLAWTUNNEL_TOKEN=" + token) print("Then run, for example:") print("./clawtunnel bind --mode tcp --name phone-ssh --server wss://claw.animefu.space --target 127.0.0.1:8022") def main() -> None: parser = argparse.ArgumentParser(prog="clawtunnel"); sub = parser.add_subparsers(dest="mode_cmd", required=True) s = sub.add_parser("server", help="run public controller"); s.add_argument("--bind", default="127.0.0.1"); s.add_argument("--public-bind", default="0.0.0.0"); s.add_argument("--http", type=int, default=8080); s.add_argument("--ws", type=int, default=8765); s.add_argument("--raw-port", type=int, default=3680); s.add_argument("--token", default=""); s.add_argument("--user-store", default="", help="JSON user/token store for signup/login tokens"); s.add_argument("--vpn-cache", default="", help="JSON cache for signup VPN/proxy reputation checks") a = sub.add_parser("agent", help="run local tunnel agent"); a.add_argument("--server", default="ws://127.0.0.1:8765"); a.add_argument("--token", default=""); a.add_argument("--name", required=True); a.add_argument("--target", required=True); a.add_argument("--mode", choices=["http","ws","tcp","udp"], default="http"); a.add_argument("--connect-key", default="", help="optional fixed raw TCP/UDP connect key; generated by server if omitted") b = sub.add_parser("bind", help="bind a local service and print an access code (alias for agent)"); b.add_argument("--server", default="ws://127.0.0.1:8765"); b.add_argument("--token", default=""); b.add_argument("--name", required=True); b.add_argument("--target", required=True); b.add_argument("--mode", choices=["http","ws","tcp","udp"], default="http"); b.add_argument("--code", dest="connect_key", default="", help="optional fixed access code; generated by server if omitted") c = sub.add_parser("connect", help="local connector for encrypted raw TCP/UDP tunnels"); c.add_argument("--server", required=True, help="host:raw_port, e.g. example.com:3680"); c.add_argument("--key", required=True, help="generated connect key from tcp/udp agent"); c.add_argument("--listen", required=True, help="local host:port, e.g. 127.0.0.1:2222"); c.add_argument("--udp", action="store_true", help="listen and forward UDP instead of TCP") x = sub.add_parser("access", help="create a local TCP/UDP listener from a bind access code"); x.add_argument("--server", required=True, help="host:raw_port, e.g. example.com:3680"); x.add_argument("--code", default="", help="access code printed by bind"); x.add_argument("--key", default="", help="compat alias for --code"); x.add_argument("--listen", required=True, help="local host:port, e.g. 127.0.0.1:2222"); x.add_argument("--udp", action="store_true", help="listen and forward UDP instead of TCP") su = sub.add_parser("signup", help="create a ClawTunnel account and print your bind token"); su.add_argument("--server", default="https://claw.animefu.space"); su.add_argument("--email", default=""); su.add_argument("--password", default="") li = sub.add_parser("login", help="log in and print a fresh bind token"); li.add_argument("--server", default="https://claw.animefu.space"); li.add_argument("--email", default=""); li.add_argument("--password", default="") args = parser.parse_args() if args.mode_cmd == "server": asyncio.run(run_server(args)) elif args.mode_cmd in {"agent", "bind"}: asyncio.run(run_agent(args)) elif args.mode_cmd in {"connect", "access"}: asyncio.run(run_connect(args)) elif args.mode_cmd == "signup": run_account(args, True) elif args.mode_cmd == "login": run_account(args, False) if __name__ == "__main__": main()