基础功能，主机和子节点可用

__init__.py

@@ -0,0 +1,4 @@
+"""mynetspeeder multi-relay accelerator."""
+
+__all__ = ["__version__"]
+__version__ = "2.0.0"

__main__.py

@@ -0,0 +1,4 @@
+from .cli import main
+
+if __name__ == "__main__":
+    raise SystemExit(main())

cli.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+
+from . import __version__
+from .config import Config
+from .relay_server import RelayServer
+from .relay_client import RelayManager
+from .transparent_edge import TransparentEdge
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(prog="mynetspeeder")
+    parser.add_argument("--version", action="version", version=f"%(prog)s {__version__}")
+    sub = parser.add_subparsers(dest="command", required=True)
+
+    relay = sub.add_parser("relay", help="在子节点 VPS 上启动 relay")
+    relay.add_argument("--listen-host", default="0.0.0.0")
+    relay.add_argument("--listen-port", type=int, default=9009)
+    relay.add_argument("--token", required=True)
+    relay.set_defaults(handler=handle_relay)
+
+    edge = sub.add_parser("edge", help="在当前主 VPS 上启动透明 direct 出站加速")
+    edge.add_argument("--listen-host", default="127.0.0.1")
+    edge.add_argument("--listen-port", type=int, default=19080)
+    edge.add_argument("--config", required=True)
+    edge.set_defaults(handler=handle_edge)
+
+    probe = sub.add_parser("probe", help="查看子节点探测与在线状态")
+    probe.add_argument("--config", required=True)
+    probe.add_argument("--once", action="store_true")
+    probe.set_defaults(handler=handle_probe)
+    return parser
+
+
+def handle_relay(args: argparse.Namespace) -> int:
+    asyncio.run(RelayServer(args.token).start(args.listen_host, args.listen_port))
+    return 0
+
+
+def handle_edge(args: argparse.Namespace) -> int:
+    asyncio.run(TransparentEdge(args.listen_host, args.listen_port, Config.load(args.config)).start())
+    return 0
+
+
+def handle_probe(args: argparse.Namespace) -> int:
+    async def run_probe() -> None:
+        manager = RelayManager(Config.load(args.config))
+        await manager.start()
+        await asyncio.sleep(2)
+        print(json.dumps(manager.snapshot(), ensure_ascii=False, indent=2))
+        if not args.once:
+            while True:
+                await asyncio.sleep(5)
+                print(json.dumps(manager.snapshot(), ensure_ascii=False, indent=2))
+    asyncio.run(run_probe())
+    return 0
+
+
+def main() -> int:
+    parser = build_parser()
+    args = parser.parse_args()
+    return args.handler(args)

config.json

@@ -0,0 +1,7 @@
+{
+  "strategy": "top2",
+  "redundancy": 2,
+  "tcp_warmup_bytes": 65536,
+  "probe_interval": 3,
+  "relays": [{"name": "hk1", "host": "141.140.15.30", "port": 9009, "token": "130", "weight": 100}]
+}

config.py

@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Literal
+
+Strategy = Literal["broadcast", "top2", "backup"]
+
+
+@dataclass
+class RelayNode:
+    name: str
+    host: str
+    port: int
+    token: str
+    weight: int = 100
+
+
+@dataclass
+class Config:
+    relays: list[RelayNode]
+    strategy: Strategy = "top2"
+    redundancy: int = 2
+    tcp_warmup_bytes: int = 65536
+    probe_interval: float = 15.0
+
+    @classmethod
+    def load(cls, path: str) -> "Config":
+        raw = json.loads(Path(path).read_text())
+        relays = [RelayNode(**item) for item in raw["relays"]]
+        return cls(
+            relays=relays,
+            strategy=raw.get("strategy", "top2"),
+            redundancy=raw.get("redundancy", 2),
+            tcp_warmup_bytes=raw.get("tcp_warmup_bytes", 65536),
+            probe_interval=raw.get("probe_interval", 15.0),
+        )

demo-config copy.json

@@ -0,0 +1,7 @@
+{
+  "strategy": "top2",
+  "redundancy": 2,
+  "tcp_warmup_bytes": 65536,
+  "probe_interval": 3,
+  "relays": []
+}

protocol.py

@@ -0,0 +1,52 @@
+from __future__ import annotations
+
+import json
+import struct
+from dataclasses import dataclass
+from typing import Any
+
+HEADER = struct.Struct("!BIIQII")
+AUTH = 1
+TCP_OPEN = 2
+TCP_STATUS = 3
+TCP_DATA = 4
+TCP_CLOSE = 5
+UDP_SEND = 6
+UDP_RECV = 7
+PING = 8
+PONG = 9
+
+STATUS_OK = 0
+STATUS_ERR = 1
+
+
+@dataclass
+class Frame:
+    kind: int
+    session_id: int
+    stream_id: int
+    seq: int
+    packet_id: int
+    payload: bytes = b""
+
+
+def encode_json(data: Any) -> bytes:
+    return json.dumps(data, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
+
+
+def decode_json(data: bytes) -> Any:
+    return json.loads(data.decode("utf-8"))
+
+
+async def write_frame(writer, frame: Frame) -> None:
+    writer.write(HEADER.pack(frame.kind, frame.session_id, frame.stream_id, frame.seq, frame.packet_id, len(frame.payload)))
+    if frame.payload:
+        writer.write(frame.payload)
+    await writer.drain()
+
+
+async def read_frame(reader) -> Frame:
+    header = await reader.readexactly(HEADER.size)
+    kind, session_id, stream_id, seq, packet_id, length = HEADER.unpack(header)
+    payload = await reader.readexactly(length) if length else b""
+    return Frame(kind=kind, session_id=session_id, stream_id=stream_id, seq=seq, packet_id=packet_id, payload=payload)

relay_client.py

@@ -0,0 +1,117 @@
+from __future__ import annotations
+
+import asyncio
+import contextlib
+from dataclasses import dataclass
+from typing import Awaitable, Callable, Dict
+
+from .config import Config, RelayNode
+from .protocol import AUTH, STATUS_OK, Frame, encode_json, read_frame, write_frame
+from .scheduler import Scheduler
+
+FrameHandler = Callable[["RelayConnection", Frame], Awaitable[None]]
+
+
+@dataclass
+class RelayConnection:
+    node: RelayNode
+    manager: "RelayManager"
+    reader: asyncio.StreamReader
+    writer: asyncio.StreamWriter
+    closed: bool = False
+    handlers: Dict[tuple[int, int], FrameHandler] = None
+    pump_task: asyncio.Task | None = None
+
+    def __post_init__(self) -> None:
+        if self.handlers is None:
+            self.handlers = {}
+
+    async def start(self) -> None:
+        print(f"[edge] connecting relay name={self.node.name} addr={self.node.host}:{self.node.port}")
+        await write_frame(self.writer, Frame(AUTH, 0, 0, 0, 0, encode_json({"token": self.node.token})))
+        frame = await read_frame(self.reader)
+        if frame.kind != AUTH or frame.packet_id != STATUS_OK:
+            raise ConnectionError(f"relay auth failed: {self.node.name}")
+        print(f"[edge] relay connected name={self.node.name} addr={self.node.host}:{self.node.port}")
+        self.pump_task = asyncio.create_task(self._pump())
+
+    async def _pump(self) -> None:
+        try:
+            while True:
+                frame = await read_frame(self.reader)
+                handler = self.handlers.get((frame.session_id, frame.stream_id))
+                if handler:
+                    await handler(self, frame)
+        except asyncio.IncompleteReadError:
+            print(f"[edge] relay disconnected name={self.node.name} eof=true")
+        except Exception as exc:
+            print(f"[edge] relay pump error name={self.node.name} error={exc!r}")
+        finally:
+            await self.close()
+
+    async def send(self, frame: Frame) -> None:
+        if self.closed:
+            raise ConnectionError(f"relay closed: {self.node.name}")
+        await write_frame(self.writer, frame)
+
+    def bind(self, session_id: int, stream_id: int, handler: FrameHandler) -> None:
+        self.handlers[(session_id, stream_id)] = handler
+
+    def unbind(self, session_id: int, stream_id: int) -> None:
+        self.handlers.pop((session_id, stream_id), None)
+
+    async def close(self) -> None:
+        if self.closed:
+            return
+        self.closed = True
+        self.manager.on_closed(self)
+        self.writer.close()
+        with contextlib.suppress(Exception):
+            await self.writer.wait_closed()
+
+
+class RelayManager:
+    def __init__(self, config: Config) -> None:
+        self.config = config
+        self.scheduler = Scheduler(config)
+        self.connections: Dict[str, RelayConnection] = {}
+        self.tasks: list[asyncio.Task] = []
+
+    async def start(self) -> None:
+        await self.scheduler.start()
+        for node in self.config.relays:
+            self.tasks.append(asyncio.create_task(self._maintain(node)))
+
+    async def _maintain(self, node: RelayNode) -> None:
+        while True:
+            if node.name in self.connections and not self.connections[node.name].closed:
+                await asyncio.sleep(2)
+                continue
+            try:
+                reader, writer = await asyncio.open_connection(node.host, node.port)
+                connection = RelayConnection(node=node, manager=self, reader=reader, writer=writer)
+                await connection.start()
+                self.connections[node.name] = connection
+                await connection.pump_task
+            except Exception as exc:
+                print(f"[edge] relay connect failed name={node.name} addr={node.host}:{node.port} error={exc!r}")
+                await asyncio.sleep(3)
+
+    def on_closed(self, connection: RelayConnection) -> None:
+        current = self.connections.get(connection.node.name)
+        if current is connection:
+            self.connections.pop(connection.node.name, None)
+
+    def available(self) -> list[RelayConnection]:
+        chosen = {node.name for node in self.scheduler.choose()}
+        preferred = [self.connections[name] for name in chosen if name in self.connections and not self.connections[name].closed]
+        if preferred:
+            return preferred
+        return [conn for conn in self.connections.values() if not conn.closed]
+
+    def snapshot(self) -> list[dict[str, object]]:
+        data = self.scheduler.snapshot()
+        online = {name for name, conn in self.connections.items() if not conn.closed}
+        for item in data:
+            item["online"] = item["name"] in online
+        return data

relay_server.py

@@ -0,0 +1,183 @@
+from __future__ import annotations
+
+import asyncio
+import contextlib
+from dataclasses import dataclass, field
+from typing import Dict
+
+from .protocol import AUTH, PING, PONG, STATUS_ERR, STATUS_OK, TCP_CLOSE, TCP_DATA, TCP_OPEN, TCP_STATUS, UDP_RECV, UDP_SEND, Frame, decode_json, read_frame, write_frame
+
+
+@dataclass
+class TcpSession:
+    session_id: int
+    stream_id: int
+    writer: asyncio.StreamWriter
+    task: asyncio.Task
+
+
+@dataclass
+class UdpSession:
+    session_id: int
+    stream_id: int
+    transport: asyncio.DatagramTransport | None = None
+    protocol: "RelayUdpProtocol | None" = None
+
+
+class RelayUdpProtocol(asyncio.DatagramProtocol):
+    def __init__(self, channel: "RelayChannel", session_id: int, stream_id: int) -> None:
+        self.channel = channel
+        self.session_id = session_id
+        self.stream_id = stream_id
+
+    def datagram_received(self, data: bytes, _addr) -> None:
+        if self.channel.closed:
+            return
+        asyncio.create_task(self.channel.safe_send(Frame(UDP_RECV, self.session_id, self.stream_id, 0, 0, data)))
+
+
+@dataclass
+class RelayChannel:
+    reader: asyncio.StreamReader
+    writer: asyncio.StreamWriter
+    token: str
+    tcp_sessions: Dict[tuple[int, int], TcpSession] = field(default_factory=dict)
+    udp_sessions: Dict[tuple[int, int], UdpSession] = field(default_factory=dict)
+    closed: bool = False
+
+    async def run(self) -> None:
+        peer = self.writer.get_extra_info("peername")
+        authed = False
+        try:
+            auth = await read_frame(self.reader)
+            if auth.kind != AUTH or decode_json(auth.payload).get("token") != self.token:
+                raise PermissionError("invalid token")
+            authed = True
+            print(f"[relay] auth ok peer={peer}")
+            await self.safe_send(Frame(AUTH, 0, 0, 0, STATUS_OK, b"ok"))
+            while True:
+                frame = await read_frame(self.reader)
+                await self.handle(frame)
+        except asyncio.IncompleteReadError:
+            if authed:
+                print(f"[relay] disconnected peer={peer}")
+        except asyncio.CancelledError:
+            pass
+        except Exception as exc:
+            if authed:
+                print(f"[relay] channel error peer={peer} error={exc!r}")
+        finally:
+            await self.close()
+
+    async def safe_send(self, frame: Frame) -> bool:
+        if self.closed:
+            return False
+        try:
+            await write_frame(self.writer, frame)
+            return True
+        except (BrokenPipeError, ConnectionResetError, RuntimeError, OSError, asyncio.CancelledError):
+            return False
+
+    async def handle(self, frame: Frame) -> None:
+        key = (frame.session_id, frame.stream_id)
+        if frame.kind == PING:
+            await self.safe_send(Frame(PONG, 0, 0, frame.seq, 0, b"pong"))
+            return
+        if frame.kind == TCP_OPEN:
+            meta = decode_json(frame.payload)
+            family = int(meta.get("family", 0)) or 0
+            try:
+                reader, writer = await asyncio.open_connection(meta["host"], int(meta["port"]), family=family or 0)
+                task = asyncio.create_task(self._tcp_pump(frame.session_id, frame.stream_id, reader))
+                self.tcp_sessions[key] = TcpSession(frame.session_id, frame.stream_id, writer, task)
+                await self.safe_send(Frame(TCP_STATUS, frame.session_id, frame.stream_id, 0, STATUS_OK, b"ok"))
+            except Exception as exc:
+                await self.safe_send(Frame(TCP_STATUS, frame.session_id, frame.stream_id, 0, STATUS_ERR, str(exc).encode()))
+            return
+        if frame.kind == TCP_DATA:
+            session = self.tcp_sessions.get(key)
+            if session:
+                try:
+                    session.writer.write(frame.payload)
+                    await session.writer.drain()
+                except Exception:
+                    await self._close_tcp(key)
+            return
+        if frame.kind == TCP_CLOSE:
+            await self._close_tcp(key)
+            return
+        if frame.kind == UDP_SEND:
+            meta = decode_json(frame.payload[: frame.packet_id])
+            payload = frame.payload[frame.packet_id :]
+            session = self.udp_sessions.get(key)
+            if session is None:
+                family = int(meta.get("family", 0)) or 0
+                transport, protocol = await asyncio.get_running_loop().create_datagram_endpoint(
+                    lambda: RelayUdpProtocol(self, frame.session_id, frame.stream_id),
+                    remote_addr=(meta["host"], int(meta["port"])),
+                    family=family or 0,
+                )
+                session = UdpSession(frame.session_id, frame.stream_id, transport, protocol)
+                self.udp_sessions[key] = session
+            with contextlib.suppress(Exception):
+                session.transport.sendto(payload)
+            return
+
+    async def _tcp_pump(self, session_id: int, stream_id: int, reader: asyncio.StreamReader) -> None:
+        try:
+            while True:
+                chunk = await reader.read(65536)
+                if not chunk:
+                    break
+                sent = await self.safe_send(Frame(TCP_DATA, session_id, stream_id, 0, 0, chunk))
+                if not sent:
+                    break
+        except asyncio.CancelledError:
+            pass
+        except Exception:
+            pass
+        finally:
+            if not self.closed:
+                await self.safe_send(Frame(TCP_CLOSE, session_id, stream_id, 0, 0, b""))
+            await self._close_tcp((session_id, stream_id), from_task=True)
+
+    async def _close_tcp(self, key: tuple[int, int], from_task: bool = False) -> None:
+        session = self.tcp_sessions.pop(key, None)
+        if session is None:
+            return
+        if not from_task and session.task is not asyncio.current_task():
+            session.task.cancel()
+            with contextlib.suppress(Exception):
+                await session.task
+        session.writer.close()
+        with contextlib.suppress(Exception):
+            await session.writer.wait_closed()
+
+    async def close(self) -> None:
+        if self.closed:
+            return
+        self.closed = True
+        for key in list(self.tcp_sessions):
+            await self._close_tcp(key)
+        for session in self.udp_sessions.values():
+            if session.transport:
+                session.transport.close()
+        self.udp_sessions.clear()
+        self.writer.close()
+        with contextlib.suppress(Exception):
+            await self.writer.wait_closed()
+
+
+class RelayServer:
+    def __init__(self, token: str) -> None:
+        self.token = token
+
+    async def start(self, host: str, port: int) -> None:
+        server = await asyncio.start_server(self._accept, host, port)
+        sockets = ", ".join(str(sock.getsockname()) for sock in server.sockets or [])
+        print(f"[relay] listening on {sockets}")
+        async with server:
+            await server.serve_forever()
+
+    async def _accept(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
+        await RelayChannel(reader, writer, self.token).run()

scheduler.py

@@ -0,0 +1,74 @@
+from __future__ import annotations
+
+import asyncio
+import time
+from dataclasses import dataclass
+
+from .config import Config, RelayNode
+
+
+@dataclass
+class RelayScore:
+    node: RelayNode
+    latency_ms: float = 9999.0
+    failures: int = 0
+    last_ok: float = 0.0
+
+    @property
+    def score(self) -> float:
+        penalty = self.failures * 250.0
+        return self.latency_ms + penalty - (self.node.weight * 0.1)
+
+
+class Scheduler:
+    def __init__(self, config: Config) -> None:
+        self.config = config
+        self.scores = {node.name: RelayScore(node=node) for node in config.relays}
+        self._task: asyncio.Task | None = None
+
+    async def start(self) -> None:
+        if self._task is None:
+            self._task = asyncio.create_task(self._probe_loop())
+
+    async def _probe_loop(self) -> None:
+        while True:
+            await asyncio.gather(*(self._probe(node) for node in self.config.relays), return_exceptions=True)
+            await asyncio.sleep(self.config.probe_interval)
+
+    async def _probe(self, node: RelayNode) -> None:
+        started = time.perf_counter()
+        try:
+            reader, writer = await asyncio.wait_for(asyncio.open_connection(node.host, node.port), timeout=3)
+            writer.close()
+            await writer.wait_closed()
+            elapsed = (time.perf_counter() - started) * 1000
+            score = self.scores[node.name]
+            score.latency_ms = elapsed
+            score.last_ok = time.time()
+            score.failures = max(0, score.failures - 1)
+        except Exception:
+            self.scores[node.name].failures += 1
+
+    def choose(self) -> list[RelayNode]:
+        ordered = sorted(self.scores.values(), key=lambda item: item.score)
+        if self.config.strategy == "broadcast":
+            limit = min(len(ordered), max(1, self.config.redundancy))
+            return [item.node for item in ordered[:limit]]
+        if self.config.strategy == "backup":
+            return [item.node for item in ordered[:1]]
+        limit = min(len(ordered), max(1, self.config.redundancy, 2))
+        return [item.node for item in ordered[:limit]]
+
+    def snapshot(self) -> list[dict[str, object]]:
+        ordered = sorted(self.scores.values(), key=lambda item: item.score)
+        return [
+            {
+                "name": item.node.name,
+                "host": item.node.host,
+                "port": item.node.port,
+                "latency_ms": round(item.latency_ms, 2),
+                "failures": item.failures,
+                "score": round(item.score, 2),
+            }
+            for item in ordered
+        ]

scripts/install.sh

@@ -0,0 +1,45 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+PREFIX="${1:-/opt/mynetspeeder}"
+BIN_PATH="/usr/local/bin/mynetspeeder"
+PYTHON_BIN="${PYTHON_BIN:-$(command -v python3 || true)}"
+PACKAGE_PARENT="$(dirname "$PREFIX")"
+PACKAGE_NAME="$(basename "$PREFIX")"
+
+if [[ $EUID -ne 0 ]]; then
+  echo "need root"
+  exit 1
+fi
+
+if [[ -z "$PYTHON_BIN" ]]; then
+  echo "python3 not found"
+  exit 1
+fi
+
+if ! command -v rsync >/dev/null 2>&1; then
+  echo "rsync not found"
+  exit 1
+fi
+
+mkdir -p "$PREFIX"
+rsync -a --delete /home/mynetspeeder/ "$PREFIX/"
+find "$PREFIX" -name '__pycache__' -type d -prune -exec rm -rf {} +
+
+cat > "$BIN_PATH" <<EOF
+#!/usr/bin/env bash
+set -euo pipefail
+export PYTHONPATH="$PACKAGE_PARENT\${PYTHONPATH:+:\$PYTHONPATH}"
+cd "$PACKAGE_PARENT"
+exec "$PYTHON_BIN" -m "$PACKAGE_NAME" "\$@"
+EOF
+chmod +x "$BIN_PATH"
+
+chmod +x "$PREFIX/scripts/start-transparent.sh" "$PREFIX/scripts/stop-transparent.sh" "$PREFIX/scripts/install.sh"
+[[ -f "$PREFIX/scripts/start-relay.sh" ]] && chmod +x "$PREFIX/scripts/start-relay.sh"
+[[ -f "$PREFIX/scripts/stop-relay.sh" ]] && chmod +x "$PREFIX/scripts/stop-relay.sh"
+
+echo "installed to $PREFIX"
+echo "package parent: $PACKAGE_PARENT"
+echo "command: $BIN_PATH"
+echo "test: mynetspeeder --help"

scripts/start-relay.sh

@@ -0,0 +1,61 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+usage() {
+  cat <<'EOF'
+Usage: start-relay.sh <token>
+
+Options via env:
+  MYNETSPEEDER_RELAY_HOST   relay 监听地址，默认 0.0.0.0
+  MYNETSPEEDER_RELAY_PORT   relay 监听端口，默认 9009
+  MYNETSPEEDER_INSTALL_DIR  安装目录，默认 /opt/mynetspeeder
+  MYNETSPEEDER_USER         运行用户，默认 mynetspeeder
+EOF
+}
+
+LISTEN_HOST="${MYNETSPEEDER_RELAY_HOST:-0.0.0.0}"
+LISTEN_PORT="${MYNETSPEEDER_RELAY_PORT:-9009}"
+TOKEN="${MYNETSPEEDER_RELAY_TOKEN:-${1:-}}"
+INSTALL_DIR="${MYNETSPEEDER_INSTALL_DIR:-/opt/mynetspeeder}"
+INSTALL_PARENT="$(dirname "$INSTALL_DIR")"
+PACKAGE_NAME="$(basename "$INSTALL_DIR")"
+RUNTIME_USER="${MYNETSPEEDER_USER:-mynetspeeder}"
+PID_FILE="/var/run/mynetspeeder-relay.pid"
+LOG_FILE="/var/log/mynetspeeder-relay.log"
+
+if [[ $EUID -ne 0 ]]; then
+  echo "need root"
+  exit 1
+fi
+
+if [[ -z "$TOKEN" ]]; then
+  usage
+  exit 1
+fi
+
+if [[ ! -d "$INSTALL_DIR" ]]; then
+  echo "install dir not found: $INSTALL_DIR"
+  exit 1
+fi
+
+id -u "$RUNTIME_USER" >/dev/null 2>&1 || useradd --system --no-create-home --shell /usr/sbin/nologin "$RUNTIME_USER"
+mkdir -p /var/log
+: > "$LOG_FILE"
+chown "$RUNTIME_USER":"$RUNTIME_USER" "$LOG_FILE"
+
+pkill -f 'python3 -m mynetspeeder relay' || true
+
+runuser -u "$RUNTIME_USER" -- bash -lc "export PYTHONUNBUFFERED=1; export PYTHONPATH=${INSTALL_PARENT}; cd ${INSTALL_PARENT} && exec nohup python3 -m ${PACKAGE_NAME} relay --listen-host ${LISTEN_HOST} --listen-port ${LISTEN_PORT} --token ${TOKEN}" >>"$LOG_FILE" 2>&1 &
+RELAY_PID=$!
+echo "$RELAY_PID" > "$PID_FILE"
+
+sleep 1
+if ! ss -ln | grep -qE "[:.]${LISTEN_PORT}( |$)"; then
+  echo "relay failed to listen on ${LISTEN_HOST}:${LISTEN_PORT}"
+  tail -n 50 "$LOG_FILE" || true
+  exit 1
+fi
+
+echo "relay started on ${LISTEN_HOST}:${LISTEN_PORT}"
+echo "pid file: $PID_FILE"
+echo "log file: $LOG_FILE"

scripts/start-transparent.sh

@@ -0,0 +1,113 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+usage() {
+  cat <<'EOF'
+Usage: start-transparent.sh [-v|--verbose] [--enable-udp] [--capture-uid UID] [config_path]
+
+Options:
+  -v, --verbose        启动后实时输出 mynetspeeder 日志
+  --capture-uid UID    只接管该 UID 发起的 TCP 出站
+  --enable-udp         额外启用 UDP 透明接管（实验性，默认关闭）
+  -h, --help           显示帮助
+EOF
+}
+
+VERBOSE=0
+ENABLE_UDP=0
+CONFIG_PATH="/home/mynetspeeder/config.json"
+CAPTURE_UID="${MYNETSPEEDER_CAPTURE_UID:-}"
+
+while [[ $# -gt 0 ]]; do
+  case "$1" in
+    -v|--verbose)
+      VERBOSE=1; shift ;;
+    --enable-udp)
+      ENABLE_UDP=1; shift ;;
+    --capture-uid)
+      CAPTURE_UID="${2:-}"
+      [[ -n "$CAPTURE_UID" ]] || { echo "missing value for --capture-uid"; exit 1; }
+      shift 2 ;;
+    -h|--help)
+      usage; exit 0 ;;
+    *)
+      CONFIG_PATH="$1"; shift ;;
+  esac
+done
+
+LISTEN_HOST="${MYNETSPEEDER_LISTEN_HOST:-127.0.0.1}"
+LISTEN_PORT="${MYNETSPEEDER_LISTEN_PORT:-19080}"
+RUNTIME_USER="${MYNETSPEEDER_USER:-mynetspeeder}"
+PID_FILE="/var/run/mynetspeeder-edge.pid"
+LOG_FILE="/var/log/mynetspeeder-edge.log"
+CHAIN4="MYNETSPEEDER"
+CHAIN6="MYNETSPEEDER6"
+
+if [[ $EUID -ne 0 ]]; then echo "need root"; exit 1; fi
+if [[ ! -f "$CONFIG_PATH" ]]; then echo "config not found: $CONFIG_PATH"; exit 1; fi
+if [[ -z "$CAPTURE_UID" ]]; then echo "refusing unsafe global capture"; exit 1; fi
+if ! [[ "$CAPTURE_UID" =~ ^[0-9]+$ ]]; then echo "capture uid must be numeric"; exit 1; fi
+
+id -u "$RUNTIME_USER" >/dev/null 2>&1 || useradd --system --no-create-home --shell /usr/sbin/nologin "$RUNTIME_USER"
+mkdir -p /var/log
+: > "$LOG_FILE"
+chown "$RUNTIME_USER":"$RUNTIME_USER" "$LOG_FILE"
+
+pkill -f 'python3 -m mynetspeeder edge' || true
+runuser -u "$RUNTIME_USER" -- bash -lc "export PYTHONUNBUFFERED=1; export PYTHONPATH=/home; cd /home && exec nohup python3 -m mynetspeeder edge --listen-host ${LISTEN_HOST} --listen-port ${LISTEN_PORT} --config ${CONFIG_PATH}" >>"$LOG_FILE" 2>&1 &
+EDGE_PID=$!
+echo "$EDGE_PID" > "$PID_FILE"
+sleep 1
+ss -ln | grep -qE "[:.]${LISTEN_PORT}( |$)" || { echo "edge failed to listen"; tail -n 50 "$LOG_FILE" || true; exit 1; }
+
+iptables -t nat -N "$CHAIN4" 2>/dev/null || true
+iptables -t nat -F "$CHAIN4"
+iptables -t nat -A "$CHAIN4" -d 127.0.0.0/8 -j RETURN
+iptables -t nat -A "$CHAIN4" -m owner --uid-owner "$RUNTIME_USER" -j RETURN
+while read -r host; do
+  [[ -n "$host" && "$host" != *:* ]] && iptables -t nat -A "$CHAIN4" -d "$host" -j RETURN
+done < <(python3 - <<'PY' "$CONFIG_PATH"
+import json, sys
+cfg = json.load(open(sys.argv[1]))
+for relay in cfg.get('relays', []):
+    print(relay['host'])
+PY
+)
+iptables -t nat -A "$CHAIN4" -p tcp -m owner --uid-owner "$CAPTURE_UID" -j REDIRECT --to-ports "$LISTEN_PORT"
+iptables -t nat -C OUTPUT -p tcp -j "$CHAIN4" 2>/dev/null || iptables -t nat -A OUTPUT -p tcp -j "$CHAIN4"
+if [[ "$ENABLE_UDP" == "1" ]]; then
+  iptables -t nat -A "$CHAIN4" -p udp -m owner --uid-owner "$CAPTURE_UID" -j REDIRECT --to-ports "$LISTEN_PORT"
+  iptables -t nat -C OUTPUT -p udp -j "$CHAIN4" 2>/dev/null || iptables -t nat -A OUTPUT -p udp -j "$CHAIN4"
+fi
+
+if command -v ip6tables >/dev/null 2>&1; then
+  ip6tables -t nat -N "$CHAIN6" 2>/dev/null || true
+  ip6tables -t nat -F "$CHAIN6"
+  ip6tables -t nat -A "$CHAIN6" -d ::1/128 -j RETURN
+  ip6tables -t nat -A "$CHAIN6" -m owner --uid-owner "$RUNTIME_USER" -j RETURN
+  while read -r host; do
+    [[ -n "$host" && "$host" == *:* ]] && ip6tables -t nat -A "$CHAIN6" -d "$host" -j RETURN
+  done < <(python3 - <<'PY' "$CONFIG_PATH"
+import json, sys
+cfg = json.load(open(sys.argv[1]))
+for relay in cfg.get('relays', []):
+    print(relay['host'])
+PY
+)
+  ip6tables -t nat -A "$CHAIN6" -p tcp -m owner --uid-owner "$CAPTURE_UID" -j REDIRECT --to-ports "$LISTEN_PORT"
+  ip6tables -t nat -C OUTPUT -p tcp -j "$CHAIN6" 2>/dev/null || ip6tables -t nat -A OUTPUT -p tcp -j "$CHAIN6"
+  if [[ "$ENABLE_UDP" == "1" ]]; then
+    ip6tables -t nat -A "$CHAIN6" -p udp -m owner --uid-owner "$CAPTURE_UID" -j REDIRECT --to-ports "$LISTEN_PORT"
+    ip6tables -t nat -C OUTPUT -p udp -j "$CHAIN6" 2>/dev/null || ip6tables -t nat -A OUTPUT -p udp -j "$CHAIN6"
+  fi
+fi
+
+echo "mynetspeeder transparent mode started on ${LISTEN_HOST}:${LISTEN_PORT}"
+echo "capture uid: $CAPTURE_UID"
+echo "udp capture: $ENABLE_UDP"
+echo "log file: $LOG_FILE"
+
+if [[ "$VERBOSE" == "1" ]]; then
+  echo "verbose mode: press Ctrl+C to stop viewing logs, service keeps running"
+  exec tail -n 80 -f "$LOG_FILE"
+fi

scripts/start.sh

@@ -0,0 +1 @@
+./start-transparent.sh -v --capture-uid 99

scripts/stop-relay.sh

@@ -0,0 +1,17 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+PID_FILE="/var/run/mynetspeeder-relay.pid"
+
+if [[ $EUID -ne 0 ]]; then
+  echo "need root"
+  exit 1
+fi
+
+if [[ -f "$PID_FILE" ]]; then
+  kill "$(cat "$PID_FILE")" 2>/dev/null || true
+  rm -f "$PID_FILE"
+fi
+pkill -f 'python3 -m mynetspeeder relay' || true
+
+echo "relay stopped"

scripts/stop-transparent.sh

@@ -0,0 +1,31 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+CHAIN4="MYNETSPEEDER"
+CHAIN6="MYNETSPEEDER6"
+PID_FILE="/var/run/mynetspeeder-edge.pid"
+
+if [[ $EUID -ne 0 ]]; then
+  echo "need root"
+  exit 1
+fi
+
+if [[ -f "$PID_FILE" ]]; then
+  kill "$(cat "$PID_FILE")" 2>/dev/null || true
+  rm -f "$PID_FILE"
+fi
+pkill -f 'python3 -m mynetspeeder edge' || true
+
+iptables -t nat -D OUTPUT -p tcp -j "$CHAIN4" 2>/dev/null || true
+iptables -t nat -D OUTPUT -p udp -j "$CHAIN4" 2>/dev/null || true
+iptables -t nat -F "$CHAIN4" 2>/dev/null || true
+iptables -t nat -X "$CHAIN4" 2>/dev/null || true
+
+if command -v ip6tables >/dev/null 2>&1; then
+  ip6tables -t nat -D OUTPUT -p tcp -j "$CHAIN6" 2>/dev/null || true
+  ip6tables -t nat -D OUTPUT -p udp -j "$CHAIN6" 2>/dev/null || true
+  ip6tables -t nat -F "$CHAIN6" 2>/dev/null || true
+  ip6tables -t nat -X "$CHAIN6" 2>/dev/null || true
+fi
+
+echo "mynetspeeder transparent mode stopped"

socks_edge.py

@@ -0,0 +1,317 @@
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import itertools
+import socket
+import struct
+from dataclasses import dataclass, field
+from typing import Dict
+
+from .config import Config, RelayNode
+from .protocol import AUTH, STATUS_OK, TCP_CLOSE, TCP_DATA, TCP_OPEN, TCP_STATUS, UDP_RECV, UDP_SEND, Frame, decode_json, encode_json, read_frame, write_frame
+from .scheduler import Scheduler
+
+SOCKS_VERSION = 5
+
+
+async def read_exact(reader: asyncio.StreamReader, size: int) -> bytes:
+    return await reader.readexactly(size)
+
+
+@dataclass(eq=False)
+class RelayLink:
+    node: RelayNode
+    reader: asyncio.StreamReader
+    writer: asyncio.StreamWriter
+    pump: asyncio.Task | None = None
+    tcp_sessions: Dict[tuple[int, int], "TcpRaceSession"] = field(default_factory=dict)
+    udp_server: "UdpAssociateServer | None" = None
+    closed: bool = False
+
+    async def start(self) -> None:
+        await write_frame(self.writer, Frame(AUTH, 0, 0, 0, 0, encode_json({"token": self.node.token})))
+        frame = await read_frame(self.reader)
+        if frame.kind != AUTH or frame.packet_id != STATUS_OK:
+            raise ConnectionError(f"relay auth failed: {self.node.name}")
+        self.pump = asyncio.create_task(self._pump())
+
+    async def _pump(self) -> None:
+        try:
+            while True:
+                frame = await read_frame(self.reader)
+                key = (frame.session_id, frame.stream_id)
+                if frame.kind in (TCP_STATUS, TCP_DATA, TCP_CLOSE):
+                    session = self.tcp_sessions.get(key)
+                    if session:
+                        await session.handle_frame(self, frame)
+                elif frame.kind == UDP_RECV and self.udp_server:
+                    await self.udp_server.handle_from_relay(frame, self)
+        except asyncio.IncompleteReadError:
+            pass
+        finally:
+            await self.close()
+
+    async def send(self, frame: Frame) -> None:
+        if self.closed:
+            raise ConnectionError(f"relay closed: {self.node.name}")
+        await write_frame(self.writer, frame)
+
+    async def close(self) -> None:
+        if self.closed:
+            return
+        self.closed = True
+        self.writer.close()
+        with contextlib.suppress(Exception):
+            await self.writer.wait_closed()
+
+
+@dataclass
+class TcpRaceSession:
+    session_id: int
+    stream_id: int
+    target_host: str
+    target_port: int
+    local_reader: asyncio.StreamReader
+    local_writer: asyncio.StreamWriter
+    links: list[RelayLink]
+    warmup_bytes: int
+    winning_link: RelayLink | None = None
+    opened: int = 0
+    open_errors: list[str] = field(default_factory=list)
+    uplink_bytes: int = 0
+    closed: bool = False
+    open_event: asyncio.Event = field(default_factory=asyncio.Event)
+    winner_event: asyncio.Event = field(default_factory=asyncio.Event)
+    pump_task: asyncio.Task | None = None
+
+    async def start(self) -> None:
+        meta = encode_json({"host": self.target_host, "port": self.target_port})
+        for link in self.links:
+            link.tcp_sessions[(self.session_id, self.stream_id)] = self
+            await link.send(Frame(TCP_OPEN, self.session_id, self.stream_id, 0, 0, meta))
+        await asyncio.wait_for(self.open_event.wait(), timeout=10)
+        if self.opened == 0:
+            raise ConnectionError(self.open_errors[0] if self.open_errors else "all relays failed")
+        self.pump_task = asyncio.create_task(self._pump_local())
+
+    async def _pump_local(self) -> None:
+        try:
+            while True:
+                chunk = await self.local_reader.read(65536)
+                if not chunk:
+                    break
+                self.uplink_bytes += len(chunk)
+                if self.winning_link is None and self.uplink_bytes <= self.warmup_bytes:
+                    await asyncio.gather(*(link.send(Frame(TCP_DATA, self.session_id, self.stream_id, 0, 0, chunk)) for link in self.links if not link.closed), return_exceptions=True)
+                else:
+                    if self.winning_link is None:
+                        await self.winner_event.wait()
+                    if self.winning_link:
+                        await self.winning_link.send(Frame(TCP_DATA, self.session_id, self.stream_id, 0, 0, chunk))
+        except Exception:
+            pass
+        finally:
+            await self.close()
+
+    async def handle_frame(self, link: RelayLink, frame: Frame) -> None:
+        if self.closed:
+            return
+        if frame.kind == TCP_STATUS:
+            if frame.packet_id == STATUS_OK:
+                self.opened += 1
+            else:
+                self.open_errors.append(frame.payload.decode("utf-8", errors="replace"))
+            if self.opened > 0 or len(self.open_errors) == len(self.links):
+                self.open_event.set()
+            return
+        if frame.kind == TCP_DATA:
+            if self.winning_link is None:
+                self.winning_link = link
+                self.winner_event.set()
+                await self._close_losers(except_link=link)
+            if link is self.winning_link:
+                self.local_writer.write(frame.payload)
+                await self.local_writer.drain()
+            return
+        if frame.kind == TCP_CLOSE:
+            if self.winning_link is None:
+                self.winning_link = link
+                self.winner_event.set()
+            if link is self.winning_link:
+                await self.close()
+
+    async def _close_losers(self, except_link: RelayLink) -> None:
+        await asyncio.gather(*(link.send(Frame(TCP_CLOSE, self.session_id, self.stream_id, 0, 0, b"")) for link in self.links if link is not except_link and not link.closed), return_exceptions=True)
+
+    async def close(self) -> None:
+        if self.closed:
+            return
+        self.closed = True
+        if self.pump_task and self.pump_task is not asyncio.current_task():
+            self.pump_task.cancel()
+            with contextlib.suppress(Exception):
+                await self.pump_task
+        await asyncio.gather(*(link.send(Frame(TCP_CLOSE, self.session_id, self.stream_id, 0, 0, b"")) for link in self.links if not link.closed), return_exceptions=True)
+        for link in self.links:
+            link.tcp_sessions.pop((self.session_id, self.stream_id), None)
+        self.local_writer.close()
+        with contextlib.suppress(Exception):
+            await self.local_writer.wait_closed()
+
+
+class UdpAssociateServer(asyncio.DatagramProtocol):
+    def __init__(self, edge: "SocksEdge") -> None:
+        self.edge = edge
+        self.transport: asyncio.DatagramTransport | None = None
+        self.client_addr = None
+        self.packet_counter = itertools.count(1)
+        self.pending: set[int] = set()
+
+    def connection_made(self, transport) -> None:
+        self.transport = transport
+
+    def datagram_received(self, data: bytes, addr) -> None:
+        if len(data) < 10:
+            return
+        if self.client_addr is None:
+            self.client_addr = addr
+        if addr != self.client_addr:
+            return
+        host, port, payload = self._parse_socks_udp(data)
+        packet_id = next(self.packet_counter)
+        self.pending.add(packet_id)
+        asyncio.create_task(self.edge.forward_udp(host, port, payload, packet_id, self))
+
+    async def handle_from_relay(self, frame: Frame, _link: RelayLink) -> None:
+        if frame.packet_id not in self.pending or self.transport is None or self.client_addr is None:
+            return
+        self.pending.discard(frame.packet_id)
+        host = self.edge.udp_targets.get(frame.packet_id, ("0.0.0.0", 0))[0]
+        port = self.edge.udp_targets.get(frame.packet_id, ("0.0.0.0", 0))[1]
+        packet = self._build_socks_udp(host, port, frame.payload)
+        self.transport.sendto(packet, self.client_addr)
+
+    def _parse_socks_udp(self, packet: bytes) -> tuple[str, int, bytes]:
+        atyp = packet[3]
+        offset = 4
+        if atyp == 1:
+            host = socket.inet_ntoa(packet[offset:offset + 4])
+            offset += 4
+        elif atyp == 3:
+            size = packet[offset]
+            offset += 1
+            host = packet[offset:offset + size].decode()
+            offset += size
+        else:
+            raise ValueError("unsupported udp atyp")
+        port = struct.unpack("!H", packet[offset:offset + 2])[0]
+        offset += 2
+        return host, port, packet[offset:]
+
+    def _build_socks_udp(self, host: str, port: int, payload: bytes) -> bytes:
+        try:
+            addr = socket.inet_aton(host)
+            header = b"\x00\x00\x00\x01" + addr + struct.pack("!H", port)
+        except OSError:
+            raw = host.encode()
+            header = b"\x00\x00\x00\x03" + bytes([len(raw)]) + raw + struct.pack("!H", port)
+        return header + payload
+
+
+class SocksEdge:
+    def __init__(self, listen_host: str, listen_port: int, config: Config) -> None:
+        self.listen_host = listen_host
+        self.listen_port = listen_port
+        self.config = config
+        self.scheduler = Scheduler(config)
+        self.links: list[RelayLink] = []
+        self.session_ids = itertools.count(1)
+        self.udp_targets: dict[int, tuple[str, int]] = {}
+        self.udp_server: UdpAssociateServer | None = None
+
+    async def start(self) -> None:
+        await self.scheduler.start()
+        await self._connect_relays()
+        server = await asyncio.start_server(self._accept, self.listen_host, self.listen_port)
+        sockets = ", ".join(str(sock.getsockname()) for sock in server.sockets or [])
+        print(f"[edge] socks5 listening on {sockets}")
+        async with server:
+            await server.serve_forever()
+
+    async def _connect_relays(self) -> None:
+        for node in self.config.relays:
+            reader, writer = await asyncio.open_connection(node.host, node.port)
+            link = RelayLink(node, reader, writer)
+            await link.start()
+            self.links.append(link)
+        loop = asyncio.get_running_loop()
+        transport, protocol = await loop.create_datagram_endpoint(lambda: UdpAssociateServer(self), local_addr=(self.listen_host, 0))
+        self.udp_server = protocol
+        for link in self.links:
+            link.udp_server = protocol
+        self.udp_transport = transport
+
+    async def _accept(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
+        try:
+            host, port, udp_mode = await self._handshake(reader, writer)
+            if udp_mode:
+                return
+            links = self._selected_links()
+            session = TcpRaceSession(
+                session_id=next(self.session_ids),
+                stream_id=0,
+                target_host=host,
+                target_port=port,
+                local_reader=reader,
+                local_writer=writer,
+                links=links,
+                warmup_bytes=self.config.tcp_warmup_bytes,
+            )
+            await session.start()
+        except Exception:
+            writer.close()
+            with contextlib.suppress(Exception):
+                await writer.wait_closed()
+
+    def _selected_links(self) -> list[RelayLink]:
+        chosen = {node.name for node in self.scheduler.choose()}
+        links = [link for link in self.links if link.node.name in chosen and not link.closed]
+        return links or [link for link in self.links if not link.closed][:1]
+
+    async def forward_udp(self, host: str, port: int, payload: bytes, packet_id: int, udp_server: UdpAssociateServer) -> None:
+        self.udp_targets[packet_id] = (host, port)
+        meta = encode_json({"host": host, "port": port})
+        links = self._selected_links()
+        for index, link in enumerate(links):
+            body = meta + payload if index == 0 else payload
+            await link.send(Frame(UDP_SEND, 1, index, 0, packet_id if index == 0 else 0, body))
+
+    async def _handshake(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> tuple[str, int, bool]:
+        version, methods_len = (await read_exact(reader, 2))
+        if version != SOCKS_VERSION:
+            raise ValueError("unsupported socks version")
+        await read_exact(reader, methods_len)
+        writer.write(b"\x05\x00")
+        await writer.drain()
+        version, command, _, atyp = await read_exact(reader, 4)
+        if version != SOCKS_VERSION:
+            raise ValueError("unsupported socks version")
+        if atyp == 1:
+            host = socket.inet_ntoa(await read_exact(reader, 4))
+        elif atyp == 3:
+            size = (await read_exact(reader, 1))[0]
+            host = (await read_exact(reader, size)).decode()
+        else:
+            raise ValueError("unsupported atyp")
+        port = struct.unpack("!H", await read_exact(reader, 2))[0]
+        if command == 1:
+            writer.write(b"\x05\x00\x00\x01\x00\x00\x00\x00\x00\x00")
+            await writer.drain()
+            return host, port, False
+        if command == 3 and self.udp_server and self.udp_server.transport:
+            bind_host, bind_port = self.udp_server.transport.get_extra_info("sockname")[:2]
+            writer.write(b"\x05\x00\x00\x01" + socket.inet_aton(bind_host) + struct.pack("!H", bind_port))
+            await writer.drain()
+            return host, port, True
+        raise ValueError("unsupported socks command")

transparent_edge.py

@@ -0,0 +1,565 @@
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import itertools
+import socket
+import struct
+from dataclasses import dataclass, field
+from typing import Awaitable, Callable
+
+from .config import Config
+from .protocol import STATUS_OK, TCP_CLOSE, TCP_DATA, TCP_OPEN, TCP_STATUS, UDP_RECV, UDP_SEND, Frame, encode_json
+from .relay_client import RelayConnection, RelayManager
+
+SO_ORIGINAL_DST = 80
+IP6T_SO_ORIGINAL_DST = 80
+IP_RECVORIGDSTADDR = 20
+IPV6_RECVORIGDSTADDR = 74
+
+
+@dataclass(frozen=True)
+class TargetAddress:
+    host: str
+    port: int
+    family: int
+
+
+@dataclass(frozen=True)
+class PeerAddress:
+    host: str
+    port: int
+    family: int
+
+
+def parse_sockaddr(raw: bytes) -> TargetAddress:
+    if len(raw) < 8:
+        raise ValueError("invalid transparent destination payload")
+    family = struct.unpack_from("=H", raw, 0)[0]
+    port = struct.unpack_from("!H", raw, 2)[0]
+    if family == socket.AF_INET:
+        host = socket.inet_ntoa(raw[4:8])
+        return TargetAddress(host=host, port=port, family=family)
+    if family == socket.AF_INET6:
+        if len(raw) < 28:
+            raise ValueError("invalid IPv6 transparent destination payload")
+        host = socket.inet_ntop(socket.AF_INET6, raw[8:24])
+        return TargetAddress(host=host, port=port, family=family)
+    raise ValueError(f"unsupported family={family}")
+
+
+class BasePath:
+    def __init__(self, name: str, on_frame: Callable[["BasePath", str, bytes | None], Awaitable[None]]) -> None:
+        self.name = name
+        self.on_frame = on_frame
+        self.opened = False
+        self.closed = False
+
+    async def open(self, target: TargetAddress) -> None:
+        raise NotImplementedError
+
+    async def send(self, data: bytes) -> None:
+        raise NotImplementedError
+
+    async def close(self) -> None:
+        raise NotImplementedError
+
+
+class DirectTcpPath(BasePath):
+    def __init__(self, name: str, on_frame: Callable[[BasePath, str, bytes | None], Awaitable[None]]) -> None:
+        super().__init__(name, on_frame)
+        self.reader: asyncio.StreamReader | None = None
+        self.writer: asyncio.StreamWriter | None = None
+        self.pump_task: asyncio.Task | None = None
+
+    async def open(self, target: TargetAddress) -> None:
+        try:
+            family = socket.AF_INET6 if target.family == socket.AF_INET6 else socket.AF_INET
+            self.reader, self.writer = await asyncio.open_connection(host=target.host, port=target.port, family=family)
+            self.opened = True
+            self.pump_task = asyncio.create_task(self._pump())
+            await self.on_frame(self, "status", b"ok")
+        except Exception as exc:
+            await self.on_frame(self, "status", str(exc).encode())
+
+    async def _pump(self) -> None:
+        assert self.reader is not None
+        try:
+            while True:
+                chunk = await self.reader.read(65536)
+                if not chunk:
+                    break
+                await self.on_frame(self, "data", chunk)
+        except Exception:
+            pass
+        finally:
+            await self.on_frame(self, "close", None)
+
+    async def send(self, data: bytes) -> None:
+        if self.closed or self.writer is None:
+            return
+        self.writer.write(data)
+        await self.writer.drain()
+
+    async def close(self) -> None:
+        if self.closed:
+            return
+        self.closed = True
+        if self.pump_task and self.pump_task is not asyncio.current_task():
+            self.pump_task.cancel()
+            with contextlib.suppress(Exception):
+                await self.pump_task
+        if self.writer:
+            self.writer.close()
+            with contextlib.suppress(Exception):
+                await self.writer.wait_closed()
+
+
+class RelayTcpPath(BasePath):
+    def __init__(self, name: str, on_frame: Callable[[BasePath, str, bytes | None], Awaitable[None]], connection: RelayConnection, session_id: int, stream_id: int) -> None:
+        super().__init__(name, on_frame)
+        self.connection = connection
+        self.session_id = session_id
+        self.stream_id = stream_id
+
+    async def open(self, target: TargetAddress) -> None:
+        if self.connection.closed:
+            await self.on_frame(self, "status", b"relay unavailable")
+            return
+        self.connection.bind(self.session_id, self.stream_id, self._handle_frame)
+        try:
+            await self.connection.send(Frame(TCP_OPEN, self.session_id, self.stream_id, 0, 0, encode_json({"host": target.host, "port": target.port, "family": target.family})))
+        except Exception as exc:
+            await self.on_frame(self, "status", str(exc).encode())
+
+    async def _handle_frame(self, _conn: RelayConnection, frame: Frame) -> None:
+        if frame.kind == TCP_STATUS:
+            if frame.packet_id == STATUS_OK:
+                self.opened = True
+                await self.on_frame(self, "status", b"ok")
+            else:
+                await self.on_frame(self, "status", frame.payload)
+            return
+        if frame.kind == TCP_DATA:
+            await self.on_frame(self, "data", frame.payload)
+            return
+        if frame.kind == TCP_CLOSE:
+            await self.on_frame(self, "close", None)
+
+    async def send(self, data: bytes) -> None:
+        if self.closed or self.connection.closed:
+            return
+        await self.connection.send(Frame(TCP_DATA, self.session_id, self.stream_id, 0, 0, data))
+
+    async def close(self) -> None:
+        if self.closed:
+            return
+        self.closed = True
+        self.connection.unbind(self.session_id, self.stream_id)
+        if not self.connection.closed:
+            with contextlib.suppress(Exception):
+                await self.connection.send(Frame(TCP_CLOSE, self.session_id, self.stream_id, 0, 0, b""))
+
+
+@dataclass
+class TransparentSession:
+    session_id: int
+    target: TargetAddress
+    reader: asyncio.StreamReader
+    writer: asyncio.StreamWriter
+    paths: list[BasePath]
+    warmup_bytes: int
+    opened_count: int = 0
+    status_count: int = 0
+    errors: list[str] = field(default_factory=list)
+    winner: BasePath | None = None
+    uplink_bytes: int = 0
+    open_event: asyncio.Event = field(default_factory=asyncio.Event)
+    winner_event: asyncio.Event = field(default_factory=asyncio.Event)
+    closed: bool = False
+    pump_task: asyncio.Task | None = None
+
+    async def start(self) -> None:
+        await asyncio.gather(*(path.open(self.target) for path in self.paths), return_exceptions=True)
+        await asyncio.wait_for(self.open_event.wait(), timeout=10)
+        if self.opened_count == 0:
+            raise ConnectionError(self.errors[0] if self.errors else "all paths failed")
+        self.pump_task = asyncio.create_task(self._pump_local())
+
+    async def _pump_local(self) -> None:
+        try:
+            while True:
+                chunk = await self.reader.read(65536)
+                if not chunk:
+                    break
+                self.uplink_bytes += len(chunk)
+                active = [path for path in self.paths if path.opened and not path.closed]
+                if not active:
+                    break
+                if self.winner is None and self.uplink_bytes <= self.warmup_bytes:
+                    await asyncio.gather(*(path.send(chunk) for path in active), return_exceptions=True)
+                else:
+                    if self.winner is None:
+                        await self.winner_event.wait()
+                    if self.winner:
+                        await self.winner.send(chunk)
+        except Exception:
+            pass
+        finally:
+            await self.close()
+
+    async def handle_path(self, path: BasePath, event: str, payload: bytes | None) -> None:
+        if self.closed:
+            return
+        if event == "status":
+            self.status_count += 1
+            if payload == b"ok":
+                self.opened_count += 1
+            elif payload is not None:
+                self.errors.append(payload.decode("utf-8", errors="replace"))
+            if self.opened_count > 0 or self.status_count == len(self.paths):
+                self.open_event.set()
+            return
+        if event == "data":
+            if self.winner is None:
+                self.winner = path
+                print(f"[edge] session={self.session_id} winner={path.name} target={self.target.host}:{self.target.port}")
+                self.winner_event.set()
+                await self._close_losers(path)
+            if path is self.winner and payload is not None:
+                self.writer.write(payload)
+                await self.writer.drain()
+            return
+        if event == "close":
+            path.closed = True
+            if self.winner is None:
+                remaining = [candidate for candidate in self.paths if candidate.opened and not candidate.closed]
+                if not remaining:
+                    await self.close()
+            elif path is self.winner:
+                await self.close()
+
+    async def _close_losers(self, winner: BasePath) -> None:
+        await asyncio.gather(*(path.close() for path in self.paths if path is not winner), return_exceptions=True)
+
+    async def close(self) -> None:
+        if self.closed:
+            return
+        self.closed = True
+        print(f"[edge] session={self.session_id} closed target={self.target.host}:{self.target.port}")
+        if self.pump_task and self.pump_task is not asyncio.current_task():
+            self.pump_task.cancel()
+            with contextlib.suppress(Exception):
+                await self.pump_task
+        await asyncio.gather(*(path.close() for path in self.paths), return_exceptions=True)
+        self.writer.close()
+        with contextlib.suppress(Exception):
+            await self.writer.wait_closed()
+
+
+class DirectUdpPath(BasePath):
+    def __init__(self, name: str, on_frame: Callable[[BasePath, str, bytes | None], Awaitable[None]], target: TargetAddress) -> None:
+        super().__init__(name, on_frame)
+        self.target = target
+        self.socket: socket.socket | None = None
+        self.read_task: asyncio.Task | None = None
+
+    async def open(self, _target: TargetAddress) -> None:
+        try:
+            family = socket.AF_INET6 if self.target.family == socket.AF_INET6 else socket.AF_INET
+            self.socket = socket.socket(family, socket.SOCK_DGRAM)
+            self.socket.setblocking(False)
+            await asyncio.get_running_loop().sock_connect(self.socket, (self.target.host, self.target.port))
+            self.opened = True
+            self.read_task = asyncio.create_task(self._pump())
+            await self.on_frame(self, "status", b"ok")
+        except Exception as exc:
+            await self.on_frame(self, "status", str(exc).encode())
+
+    async def _pump(self) -> None:
+        assert self.socket is not None
+        loop = asyncio.get_running_loop()
+        try:
+            while True:
+                data = await loop.sock_recv(self.socket, 65535)
+                if not data:
+                    break
+                await self.on_frame(self, "data", data)
+        except Exception:
+            pass
+        finally:
+            await self.on_frame(self, "close", None)
+
+    async def send(self, data: bytes) -> None:
+        if self.closed or self.socket is None:
+            return
+        await asyncio.get_running_loop().sock_sendall(self.socket, data)
+
+    async def close(self) -> None:
+        if self.closed:
+            return
+        self.closed = True
+        if self.read_task and self.read_task is not asyncio.current_task():
+            self.read_task.cancel()
+            with contextlib.suppress(Exception):
+                await self.read_task
+        if self.socket:
+            self.socket.close()
+
+
+class RelayUdpPath(BasePath):
+    def __init__(self, name: str, on_frame: Callable[[BasePath, str, bytes | None], Awaitable[None]], connection: RelayConnection, session_id: int, stream_id: int, target: TargetAddress) -> None:
+        super().__init__(name, on_frame)
+        self.connection = connection
+        self.session_id = session_id
+        self.stream_id = stream_id
+        self.target = target
+
+    async def open(self, _target: TargetAddress) -> None:
+        if self.connection.closed:
+            await self.on_frame(self, "status", b"relay unavailable")
+            return
+        self.connection.bind(self.session_id, self.stream_id, self._handle_frame)
+        self.opened = True
+        await self.on_frame(self, "status", b"ok")
+
+    async def _handle_frame(self, _conn: RelayConnection, frame: Frame) -> None:
+        if frame.kind == UDP_RECV:
+            await self.on_frame(self, "data", frame.payload)
+
+    async def send(self, data: bytes) -> None:
+        if self.closed or self.connection.closed:
+            return
+        meta = encode_json({"host": self.target.host, "port": self.target.port, "family": self.target.family})
+        payload = meta + data
+        await self.connection.send(Frame(UDP_SEND, self.session_id, self.stream_id, 0, len(meta), payload))
+
+    async def close(self) -> None:
+        if self.closed:
+            return
+        self.closed = True
+        self.connection.unbind(self.session_id, self.stream_id)
+
+
+@dataclass
+class UdpFlow:
+    flow_id: int
+    source: PeerAddress
+    target: TargetAddress
+    send_response: Callable[[PeerAddress, bytes], Awaitable[None]]
+    paths: list[BasePath]
+    winner: BasePath | None = None
+    closed: bool = False
+    last_activity: float = 0.0
+
+    async def start(self) -> None:
+        await asyncio.gather(*(path.open(self.target) for path in self.paths), return_exceptions=True)
+
+    async def send(self, payload: bytes) -> None:
+        self.last_activity = asyncio.get_running_loop().time()
+        active = [path for path in self.paths if path.opened and not path.closed]
+        if self.winner is None:
+            await asyncio.gather(*(path.send(payload) for path in active), return_exceptions=True)
+        elif not self.winner.closed:
+            await self.winner.send(payload)
+
+    async def handle_path(self, path: BasePath, event: str, payload: bytes | None) -> None:
+        self.last_activity = asyncio.get_running_loop().time()
+        if event == "data" and payload is not None:
+            if self.winner is None:
+                self.winner = path
+                print(f"[edge] udp flow={self.flow_id} winner={path.name} target={self.target.host}:{self.target.port}")
+            if path is self.winner:
+                await self.send_response(self.source, payload)
+        if event == "close":
+            path.closed = True
+
+    async def close(self) -> None:
+        if self.closed:
+            return
+        self.closed = True
+        await asyncio.gather(*(path.close() for path in self.paths), return_exceptions=True)
+
+
+class TransparentUdpListener:
+    def __init__(self, edge: "TransparentEdge", family: int, bind_host: str, port: int) -> None:
+        self.edge = edge
+        self.family = family
+        self.bind_host = bind_host
+        self.port = port
+        self.socket: socket.socket | None = None
+
+    def start(self) -> None:
+        sock = socket.socket(self.family, socket.SOCK_DGRAM)
+        sock.setblocking(False)
+        if self.family == socket.AF_INET:
+            sock.setsockopt(socket.SOL_IP, IP_RECVORIGDSTADDR, 1)
+            sock.bind((self.bind_host, self.port))
+        else:
+            sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 1)
+            sock.setsockopt(socket.IPPROTO_IPV6, IPV6_RECVORIGDSTADDR, 1)
+            sock.bind((self.bind_host, self.port, 0, 0))
+        self.socket = sock
+        asyncio.get_running_loop().add_reader(sock.fileno(), self._on_readable)
+        print(f"[edge] transparent udp listening on {sock.getsockname()}")
+
+    def _on_readable(self) -> None:
+        assert self.socket is not None
+        try:
+            data, ancdata, _flags, src = self.socket.recvmsg(65535, 512)
+        except BlockingIOError:
+            return
+        except Exception as exc:
+            print(f"[edge] udp recv failed family={self.family} error={exc!r}")
+            return
+        original = None
+        for level, ctype, cdata in ancdata:
+            if self.family == socket.AF_INET and level == socket.SOL_IP and ctype == IP_RECVORIGDSTADDR:
+                original = parse_sockaddr(cdata)
+                break
+            if self.family == socket.AF_INET6 and level == socket.IPPROTO_IPV6 and ctype == IPV6_RECVORIGDSTADDR:
+                original = parse_sockaddr(cdata)
+                break
+        if original is None:
+            print(f"[edge] udp missing original dst family={self.family} src={src}")
+            return
+        if self.family == socket.AF_INET:
+            source = PeerAddress(host=src[0], port=src[1], family=socket.AF_INET)
+        else:
+            source = PeerAddress(host=src[0], port=src[1], family=socket.AF_INET6)
+        asyncio.create_task(self.edge.handle_udp_datagram(source, original, data, self))
+
+    async def send_response(self, source: PeerAddress, payload: bytes) -> None:
+        assert self.socket is not None
+        if source.family == socket.AF_INET:
+            self.socket.sendto(payload, (source.host, source.port))
+        else:
+            self.socket.sendto(payload, (source.host, source.port, 0, 0))
+
+    async def close(self) -> None:
+        if self.socket is None:
+            return
+        asyncio.get_running_loop().remove_reader(self.socket.fileno())
+        self.socket.close()
+        self.socket = None
+
+
+class TransparentEdge:
+    def __init__(self, listen_host: str, listen_port: int, config: Config) -> None:
+        self.listen_host = listen_host
+        self.listen_port = listen_port
+        self.config = config
+        self.manager = RelayManager(config)
+        self.session_ids = itertools.count(1)
+        self.stream_ids = itertools.count(1)
+        self.udp_listeners: list[TransparentUdpListener] = []
+        self.udp_flows: dict[tuple[PeerAddress, TargetAddress], UdpFlow] = {}
+        self.udp_flow_ids = itertools.count(1)
+        self.udp_gc_task: asyncio.Task | None = None
+
+    async def start(self) -> None:
+        await self.manager.start()
+        print(f"[edge] relay snapshot: {self.manager.snapshot()}")
+        server4 = await asyncio.start_server(self._accept, self.listen_host, self.listen_port, family=socket.AF_INET)
+        sockets = [str(sock.getsockname()) for sock in server4.sockets or []]
+        server6 = None
+        if self.listen_host in ("::", "::1", "0.0.0.0", "127.0.0.1"):
+            host6 = "::1" if self.listen_host == "127.0.0.1" else "::"
+            try:
+                server6 = await asyncio.start_server(self._accept, host6, self.listen_port, family=socket.AF_INET6)
+                sockets.extend(str(sock.getsockname()) for sock in server6.sockets or [])
+            except Exception as exc:
+                print(f"[edge] ipv6 tcp listener skipped: {exc!r}")
+        self._start_udp_listeners()
+        self.udp_gc_task = asyncio.create_task(self._gc_udp_flows())
+        print(f"[edge] transparent tcp listening on {', '.join(sockets)}")
+        if server6 is None:
+            async with server4:
+                await server4.serve_forever()
+        else:
+            async with server4, server6:
+                await asyncio.gather(server4.serve_forever(), server6.serve_forever())
+
+    def _start_udp_listeners(self) -> None:
+        binds = []
+        if self.listen_host == "127.0.0.1":
+            binds = [(socket.AF_INET, "127.0.0.1"), (socket.AF_INET6, "::1")]
+        elif self.listen_host == "0.0.0.0":
+            binds = [(socket.AF_INET, "0.0.0.0"), (socket.AF_INET6, "::")]
+        else:
+            family = socket.AF_INET6 if ":" in self.listen_host else socket.AF_INET
+            binds = [(family, self.listen_host)]
+        for family, host in binds:
+            try:
+                listener = TransparentUdpListener(self, family, host, self.listen_port)
+                listener.start()
+                self.udp_listeners.append(listener)
+            except Exception as exc:
+                print(f"[edge] udp listener skipped family={family} host={host} error={exc!r}")
+
+    async def _accept(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
+        peer = writer.get_extra_info("peername")
+        try:
+            target = self._get_original_dst(writer)
+            session_id = next(self.session_ids)
+            session = TransparentSession(session_id=session_id, target=target, reader=reader, writer=writer, paths=[], warmup_bytes=self.config.tcp_warmup_bytes)
+            paths: list[BasePath] = [DirectTcpPath(name="direct", on_frame=lambda path, event, payload, s=session: self._handle_tcp_session(s, path, event, payload))]
+            for connection in self.manager.available():
+                stream_id = next(self.stream_ids)
+                paths.append(RelayTcpPath(name=connection.node.name, on_frame=lambda path, event, payload, s=session: self._handle_tcp_session(s, path, event, payload), connection=connection, session_id=session_id, stream_id=stream_id))
+            session.paths = paths
+            print(f"[edge] accept peer={peer} session={session_id} target={target.host}:{target.port} candidates={[path.name for path in paths]}")
+            await session.start()
+        except Exception as exc:
+            print(f"[edge] accept failed peer={peer} error={exc!r}")
+            writer.close()
+            with contextlib.suppress(Exception):
+                await writer.wait_closed()
+
+    async def _handle_tcp_session(self, session: TransparentSession, path: BasePath, event: str, payload: bytes | None) -> None:
+        await session.handle_path(path, event, payload)
+
+    def _get_original_dst(self, writer: asyncio.StreamWriter) -> TargetAddress:
+        sock = writer.get_extra_info("socket")
+        if sock is None:
+            raise RuntimeError("socket unavailable")
+        family = sock.family
+        if family == socket.AF_INET:
+            raw = sock.getsockopt(socket.SOL_IP, SO_ORIGINAL_DST, 16)
+            return parse_sockaddr(raw)
+        if family == socket.AF_INET6:
+            raw = sock.getsockopt(socket.IPPROTO_IPV6, IP6T_SO_ORIGINAL_DST, 128)
+            return parse_sockaddr(raw)
+        raise RuntimeError(f"unsupported socket family={family}")
+
+    async def handle_udp_datagram(self, source: PeerAddress, target: TargetAddress, payload: bytes, listener: TransparentUdpListener) -> None:
+        key = (source, target)
+        flow = self.udp_flows.get(key)
+        if flow is None:
+            flow_id = next(self.udp_flow_ids)
+            paths: list[BasePath] = [DirectUdpPath(name="direct", on_frame=lambda path, event, data, fid=flow_id: self._handle_udp_path(fid, path, event, data), target=target)]
+            for connection in self.manager.available():
+                stream_id = next(self.stream_ids)
+                paths.append(RelayUdpPath(name=connection.node.name, on_frame=lambda path, event, data, fid=flow_id: self._handle_udp_path(fid, path, event, data), connection=connection, session_id=flow_id, stream_id=stream_id, target=target))
+            flow = UdpFlow(flow_id=flow_id, source=source, target=target, send_response=listener.send_response, paths=paths)
+            self.udp_flows[key] = flow
+            print(f"[edge] udp flow={flow_id} target={target.host}:{target.port} candidates={[path.name for path in paths]}")
+            await flow.start()
+        await flow.send(payload)
+
+    async def _handle_udp_path(self, flow_id: int, path: BasePath, event: str, payload: bytes | None) -> None:
+        for flow in list(self.udp_flows.values()):
+            if flow.flow_id == flow_id:
+                await flow.handle_path(path, event, payload)
+                break
+
+    async def _gc_udp_flows(self) -> None:
+        loop = asyncio.get_running_loop()
+        while True:
+            await asyncio.sleep(30)
+            now = loop.time()
+            stale = [key for key, flow in self.udp_flows.items() if flow.last_activity and now - flow.last_activity > 120]
+            for key in stale:
+                flow = self.udp_flows.pop(key, None)
+                if flow:
+                    await flow.close()