"""本地 SOCKS5 → HTTP CONNECT/UoT v2 桥接。 tun2socks 只需要一个标准 SOCKS5 出口；TCP 连接转成当前节点的 HTTP CONNECT，UDP 连接转成 magic address 上的 UoT v2 流。系统代理设置完全不参与。 """ from __future__ import annotations import base64 import ipaddress import select import socket import ssl import struct import threading from typing import Any, Callable from .dns_observer import DnsObserver from .uot_protocol import ( Address, MAGIC_ADDRESS, encode_datagram_frame, encode_request, read_datagram_frame, ) # relay 的内网监听端口与外层 CONNECT 端口是两个概念：nginx 接收 # CONNECT magic-host:443 后，将字节流转发到 127.0.0.1:15201。 DEFAULT_UOT_RELAY_PORT = 15201 DEFAULT_UOT_CONNECT_PORT = 443 def _recv_exact(stream: socket.socket, size: int) -> bytes: """按 SOCKS5/TCP 协议长度读取，不能假设一次 recv 就拿全。""" data = bytearray() while len(data) < size: chunk = stream.recv(size - len(data)) if not chunk: raise ConnectionError("SOCKS5 连接在握手中断开") data.extend(chunk) return bytes(data) def _node_http(node: dict[str, Any]) -> tuple[str, int, str, str]: methods = (node.get("methods") or {}).get("http") or {} host = str(node.get("domain") or "").strip() port = methods.get("port") or node.get("port") or 443 username = methods.get("username") or node.get("username") or "" password = methods.get("password") or node.get("password") or "" if not host: raise ValueError("节点缺少 HTTP CONNECT 地址") return host, int(port), str(username), str(password) def _node_uot_connect_port(node: dict[str, Any]) -> int: """返回 magic CONNECT 的对外端口，不能使用 0。""" chain = node.get("_jump_chain") if isinstance(node, dict) else None if isinstance(chain, list) and chain: return _node_uot_connect_port(chain[-1]) methods = (node.get("methods") or {}).get("http") or {} raw = methods.get("uot_connect_port") if raw is None: raw = node.get("uot_connect_port") # 兼容早期把 uot_port 当作 CONNECT 端口的本地配置；服务端 nginx # 会按 magic host 映射到隐藏的 15201 relay 端口。 if raw is None: raw = methods.get("uot_port") if raw is None: raw = node.get("uot_port") if raw is None: raw = DEFAULT_UOT_CONNECT_PORT port = int(raw) if not 1 <= port <= 65535: raise ValueError("UoT CONNECT 端口超出范围") return port def _socks_address(stream: socket.socket, kind: int | None = None) -> Address: # SOCKS5 请求头已经读取过 ATYP 时直接传入；独立解析时才从流中读取。 if kind is None: kind = _recv_exact(stream, 1)[0] if kind == 1: host = socket.inet_ntoa(_recv_exact(stream, 4)) elif kind == 4: host = socket.inet_ntop(socket.AF_INET6, _recv_exact(stream, 16)) elif kind == 3: size = _recv_exact(stream, 1)[0] host = _recv_exact(stream, size).decode("idna") else: raise ValueError("SOCKS5 地址类型不支持") port = struct.unpack("!H", _recv_exact(stream, 2))[0] return Address(host, port) def _pack_socks_udp(destination: Address, payload: bytes) -> bytes: try: ip = ipaddress.ip_address(destination.host) except ValueError: raw = destination.host.encode("idna") return b"\x00\x00\x00\x03" + bytes([len(raw)]) + raw + struct.pack("!H", destination.port) + payload atyp = 1 if ip.version == 4 else 4 return b"\x00\x00\x00" + bytes([atyp]) + ip.packed + struct.pack("!H", destination.port) + payload def _unpack_socks_udp(packet: bytes) -> tuple[Address, bytes]: if len(packet) < 4 or packet[:2] != b"\x00\x00": raise ValueError("SOCKS5 UDP 包头无效") if packet[2] != 0: # RFC 1928 允许分片，但 tun2socks 当前不会发送分片；静默拼接会把 # 不完整的报文当成 DNS/QUIC 数据，造成更难排查的协议损坏。 raise ValueError("暂不支持 SOCKS5 UDP 分片") atyp = packet[3] offset = 4 if atyp == 1: if len(packet) < offset + 4: raise ValueError("SOCKS5 UDP IPv4 地址不完整") host = socket.inet_ntoa(packet[offset:offset + 4]); offset += 4 elif atyp == 4: if len(packet) < offset + 16: raise ValueError("SOCKS5 UDP IPv6 地址不完整") host = socket.inet_ntop(socket.AF_INET6, packet[offset:offset + 16]); offset += 16 elif atyp == 3: if len(packet) < offset + 1: raise ValueError("SOCKS5 UDP 域名长度不完整") size = packet[offset]; offset += 1 if len(packet) < offset + size: raise ValueError("SOCKS5 UDP 域名地址不完整") host = packet[offset:offset + size].decode("idna"); offset += size else: raise ValueError("SOCKS5 UDP 地址类型不支持") if len(packet) < offset + 2: raise ValueError("SOCKS5 UDP 端口不完整") port = struct.unpack("!H", packet[offset:offset + 2])[0] return Address(host, port), packet[offset + 2:] class HttpConnectDialer: def __init__(self, node: dict[str, Any], timeout: float = 15.0, source_address: str | None = None, proxy_addresses: list[tuple] | None = None, proxy_addresses_by_node: dict[str, list[tuple]] | None = None): self.node = node self.timeout = timeout # TUN 启动后默认路由会指向虚拟网卡。节点本身必须从物理网卡发出， # 否则上游 CONNECT 会再次进入本地 SOCKS，形成递归回环。 self.source_address = source_address self.proxy_addresses = proxy_addresses self.proxy_addresses_by_node = proxy_addresses_by_node or {} @staticmethod def _node_key(node: dict[str, Any]) -> str: methods = (node.get("methods") or {}).get("http") or {} return "|".join((str(node.get("domain") or ""), str(methods.get("port") or node.get("port") or 443), str(methods.get("username") or node.get("username") or ""))) @staticmethod def _node_credentials(node: dict[str, Any]) -> tuple[str, str]: methods = (node.get("methods") or {}).get("http") or {} username = methods.get("username") or node.get("username") or "" password = methods.get("password") or node.get("password") or "" return str(username), str(password) def _open_proxy_socket(self, host: str, port: int, addresses: list[tuple] | None = None) -> socket.socket: addresses = addresses or self.proxy_addresses if not addresses: addresses = [item[4] for item in socket.getaddrinfo( host, port, socket.AF_UNSPEC, socket.SOCK_STREAM)] last_error: OSError | None = None for sockaddr in addresses: family = socket.AF_INET6 if len(sockaddr) == 4 else socket.AF_INET raw = socket.socket(family, socket.SOCK_STREAM) try: raw.settimeout(self.timeout) if self.source_address: if family == socket.AF_INET and ":" not in self.source_address: raw.bind((self.source_address, 0)) elif family == socket.AF_INET6 and ":" in self.source_address: raw.bind((self.source_address, 0, 0, 0)) raw.connect(sockaddr) return raw except OSError as exc: last_error = exc raw.close() if last_error: raise last_error raise OSError(f"无法解析 HTTP CONNECT 节点: {host}") def _jump_addresses(self, node: dict[str, Any]) -> list[tuple] | None: return self.proxy_addresses_by_node.get(self._node_key(node)) def _open_jump(self, target: Address) -> socket.socket: """通过 2-3 个 HTTP CONNECT 节点逐层建立 TLS，再连接最终目标。""" chain = list(self.node.get("_jump_chain") or []) if not 2 <= len(chain) <= 3: raise RuntimeError("跳板链路必须包含 2-3 个节点") ctx = ssl.create_default_context() first = chain[0] first_host, first_port, _username, _password = _node_http(first) stream = self._open_proxy_socket( first_host, first_port, self._jump_addresses(first)) try: stream = ctx.wrap_socket(stream, server_hostname=first_host) for index, next_node in enumerate(chain[1:], 1): next_host, next_port, _username, _password = _node_http(next_node) leftover = _http_connect_over( stream, chain[index - 1], next_host, next_port, resolve_target=True) stream = _LayeredTlsTransport(stream, ctx, next_host, leftover) _http_connect_over( stream, chain[-1], target.host, target.port, resolve_target=False) return stream except Exception: try: stream.close() except OSError: pass raise def open(self, target: Address) -> socket.socket: if self.node.get("_jump_chain"): return self._open_jump(target) host, port, username, password = _node_http(self.node) raw = self._open_proxy_socket(host, port) ctx = ssl.create_default_context() stream = ctx.wrap_socket(raw, server_hostname=host) authority = f"[{target.host}]:{target.port}" if ":" in target.host else f"{target.host}:{target.port}" auth = "" if username or password: token = base64.b64encode(f"{username}:{password}".encode()).decode("ascii") auth = f"Authorization: Basic {token}\r\n" request = ( f"CONNECT {authority} HTTP/1.1\r\n" f"Host: {authority}\r\n" f"{auth}Connection: Keep-Alive\r\n\r\n" ).encode("ascii") stream.sendall(request) response = b"" while b"\r\n\r\n" not in response and len(response) < 65536: chunk = stream.recv(4096) if not chunk: break response += chunk first = response.split(b"\r\n", 1)[0].decode("latin1", "replace") if not first.startswith(("HTTP/1.1 200", "HTTP/1.0 200")): stream.close() raise ConnectionError(f"HTTP CONNECT 失败: {first[:160]}") return stream def _resolve_jump_address(host: str, port: int) -> tuple: infos = socket.getaddrinfo(host, port, socket.AF_UNSPEC, socket.SOCK_STREAM) infos.sort(key=lambda item: 0 if item[0] == socket.AF_INET else 1) if not infos: raise OSError(f"无法解析跳板地址: {host}") return infos[0][4] def _http_connect_over(transport, proxy_node: dict[str, Any], target_host: str, target_port: int, *, resolve_target: bool) -> bytes: """在当前 TLS 层发送 CONNECT；中间跳板使用解析后的 IP，末跳保留目标名。""" username, password = HttpConnectDialer._node_credentials(proxy_node) auth_header = "" if username or password: token = base64.b64encode(f"{username}:{password}".encode()).decode("ascii") auth_header = f"Authorization: Basic {token}\r\n" if resolve_target: address = _resolve_jump_address(target_host, target_port) authority = f"[{address[0]}]:{address[1]}" if ":" in address[0] else f"{address[0]}:{address[1]}" else: authority = f"[{target_host}]:{target_port}" if ":" in target_host else f"{target_host}:{target_port}" request = ( f"CONNECT {authority} HTTP/1.1\r\n" f"Host: {authority}\r\n" f"{auth_header}Connection: Keep-Alive\r\n\r\n" ).encode("ascii") transport.sendall(request) response = b"" while b"\r\n\r\n" not in response and len(response) < 65536: chunk = transport.recv(4096) if not chunk: break response += chunk marker = b"\r\n\r\n" end = response.find(marker) first = response.split(b"\r\n", 1)[0].decode("latin-1", "replace") if not (first.startswith("HTTP/1.1 200") or first.startswith("HTTP/1.0 200")): raise ConnectionError(f"跳板 CONNECT 失败: {first[:160]}") return response[end + len(marker):] if end >= 0 else b"" class _LayeredTlsTransport: """在已有 TLS 字节流上承载下一层 TLS。""" def __init__(self, transport, ctx, server_hostname: str, initial: bytes = b""): self._transport = transport self._incoming = ssl.MemoryBIO() self._outgoing = ssl.MemoryBIO() self._pending = initial self._ssl = ctx.wrap_bio( self._incoming, self._outgoing, server_side=False, server_hostname=server_hostname) self._handshake() def _flush(self): data = self._outgoing.read() if data: self._transport.sendall(data) def _feed(self): if self._pending: data, self._pending = self._pending, b"" else: data = self._transport.recv(16384) if not data: raise ConnectionError("分层 TLS 下层连接已关闭") self._incoming.write(data) def _handshake(self): while True: try: self._ssl.do_handshake() self._flush() return except ssl.SSLWantReadError: self._flush() self._feed() except ssl.SSLWantWriteError: self._flush() def sendall(self, data): view = memoryview(data) while view: try: written = self._ssl.write(view) view = view[written:] self._flush() except ssl.SSLWantReadError: self._flush() self._feed() except ssl.SSLWantWriteError: self._flush() def recv(self, size): while True: try: return self._ssl.read(size) except ssl.SSLWantReadError: self._flush() self._feed() except ssl.SSLWantWriteError: self._flush() def settimeout(self, value): self._transport.settimeout(value) def gettimeout(self): return self._transport.gettimeout() def close(self): self._transport.close() class Socks5Bridge: def __init__(self, node: dict[str, Any], bypass_lan: bool = True, route_rules: list[dict[str, Any]] | None = None, log: Callable[[str], None] | None = None, source_address: str | None = None, proxy_addresses: list[tuple] | None = None, direct_target_resolver: Callable[[str, str, int], tuple[str, int] | None] | None = None, route_rule_resolver: Callable[[str], dict[str, Any] | None] | None = None, route_node_resolver: Callable[[dict[str, Any]], dict[str, Any] | None] | None = None, proxy_addresses_by_node: dict[str, list[tuple]] | None = None, direct_route_ensurer: Callable[[str], None] | None = None): self.node = node self.bypass_lan = bypass_lan self.route_rules = route_rules or [] self.log = log or (lambda _message: None) self.source_address = source_address self.proxy_addresses = proxy_addresses # 由 main.py 注入 Lite 的同一套 PAC/路由判定，TUN 不复制另一份规则语义。 self.direct_target_resolver = direct_target_resolver self.route_rule_resolver = route_rule_resolver self.route_node_resolver = route_node_resolver self.proxy_addresses_by_node = proxy_addresses_by_node or {} self.direct_route_ensurer = direct_route_ensurer self.listener: socket.socket | None = None self._running = threading.Event() self._threads: list[threading.Thread] = [] self._up = 0 self._down = 0 self._counter_lock = threading.Lock() self._dns = DnsObserver(log=self.log) self._udp_resources: list[socket.socket] = [] self._udp_resources_lock = threading.RLock() def start(self) -> int: self.listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self.listener.bind(("127.0.0.1", 0)) self.listener.listen(32) self.listener.settimeout(1) self._running.set() thread = threading.Thread(target=self._accept_loop, daemon=True) thread.start(); self._threads.append(thread) return self.listener.getsockname()[1] def _accept_loop(self): while self._running.is_set() and self.listener: try: client, _ = self.listener.accept() except socket.timeout: continue except OSError: break thread = threading.Thread(target=self._client, args=(client,), daemon=True) thread.start(); self._threads.append(thread) def _client(self, client: socket.socket): try: client.settimeout(15) greeting = _recv_exact(client, 2) if greeting[0] != 5: return methods = _recv_exact(client, greeting[1]) if 0 not in methods: client.sendall(b"\x05\xff") return client.sendall(b"\x05\x00") header = _recv_exact(client, 4) if header[0] != 5: return command = header[1] destination = _socks_address(client, header[3]) if command == 1: self.log(f"[桥接] TCP 目标: {self._target_label(destination)}") self._tcp(client, destination) elif command == 3: self._udp(client) else: client.sendall(b"\x05\x07\x00\x01\x00\x00\x00\x00\x00\x00") except Exception as exc: # 停止 TUN 时主动关闭仍在转发的连接，Windows 可能返回 # WSAECONNRESET (10054)；这是收尾动作，不应作为用户错误记录。 if self._running.is_set(): self.log(f"[桥接] SOCKS5 请求失败: {exc}") finally: try: client.close() except OSError: pass def _target_label(self, target: Address) -> str: host = target.host try: ipaddress.ip_address(host) except ValueError: ips = self._dns.lookup_domain(host) resolved = ",".join(ips) if ips else "未知" return f"domain={host} ip={resolved} port={target.port}" domain = self._dns.lookup_ip(host) or "未知" return f"domain={domain} ip={host} port={target.port}" def _policy_host(self, host: str) -> str: """对已观察到的唯一 DNS 映射应用域名规则；实际连接仍使用原目标 IP。""" try: ipaddress.ip_address(host) except ValueError: return host return self._dns.lookup_ip(host) or host def _direct_target(self, host: str, port: int) -> tuple[str, int] | None: explicit = False normalized = str(host).rstrip(".").lower() for rule in self.route_rules: if not isinstance(rule, dict) or not rule.get("enabled", True): continue if str(rule.get("action") or "direct").lower() != "direct": continue value = str(rule.get("domain") or rule.get("value") or "").strip().lower().rstrip(".") if value.startswith("*."): value = value[2:] try: if ipaddress.ip_address(normalized) in ipaddress.ip_network(value, strict=False): explicit = True except ValueError: if normalized == value or normalized.endswith("." + value): explicit = True if explicit: break if not self.bypass_lan and not explicit: return None try: address = ipaddress.ip_address(host) return (host, port) if explicit or address.is_private else None except ValueError: try: for item in socket.getaddrinfo(host, port, socket.AF_UNSPEC, socket.SOCK_STREAM): address = item[4][0].split("%", 1)[0] if explicit or ipaddress.ip_address(address).is_private: return address, port except OSError: pass return None def _route_rule(self, host: str, original_host: str | None = None) -> dict[str, Any] | None: if self.route_rule_resolver is not None: try: rule = self.route_rule_resolver(host) if isinstance(rule, dict): return rule if original_host and original_host != host: rule = self.route_rule_resolver(original_host) if isinstance(rule, dict): return rule except Exception as exc: self.log(f"[路由] 规则判断失败: {exc}") for rule in self.route_rules: if not isinstance(rule, dict) or not rule.get("enabled", True): continue value = str(rule.get("domain") or rule.get("value") or "").strip().lower().rstrip(".") normalized = str(host).rstrip(".").lower() base = value[2:] if value.startswith("*.") else value if base and (normalized == base or normalized.endswith("." + base)): return rule return None def _direct_target_for(self, host: str, port: int, policy_host: str | None = None) -> tuple[str, int] | None: policy_host = policy_host or self._policy_host(host) rule = self._route_rule(policy_host, host) action = str((rule or {}).get("action") or "").lower() if action == "node": return None if self.direct_target_resolver is not None: try: return self.direct_target_resolver(policy_host, host, port) except Exception as exc: self.log(f"[路由] PAC 直连判断失败: {exc}") return None if action == "direct": return (host, port) return self._direct_target(host, port) def _route_node(self, rule: dict[str, Any] | None) -> dict[str, Any] | None: if not rule or str(rule.get("action") or "").lower() != "node": return None if self.route_node_resolver is None: return None try: node = self.route_node_resolver(rule) return node if isinstance(node, dict) else None except Exception as exc: self.log(f"[路由] 固定节点解析失败: {exc}") return None @staticmethod def _node_key(node: dict[str, Any]) -> str: chain = node.get("_jump_chain") if isinstance(chain, list) and chain: return "jump|" + ">".join( Socks5Bridge._node_key(item) for item in chain if isinstance(item, dict) ) methods = (node.get("methods") or {}).get("http") or {} return "|".join((str(node.get("domain") or ""), str(methods.get("port") or node.get("port") or 443), str(methods.get("username") or node.get("username") or ""))) def _dialer(self, node: dict[str, Any]) -> HttpConnectDialer: proxy_addresses = self.proxy_addresses chain = node.get("_jump_chain") if isinstance(chain, list) and chain: proxy_addresses = self.proxy_addresses_by_node.get( self._node_key(chain[0])) return HttpConnectDialer( node, source_address=self.source_address, proxy_addresses=(proxy_addresses if node is self.node else self.proxy_addresses_by_node.get(self._node_key(node), proxy_addresses)), proxy_addresses_by_node=self.proxy_addresses_by_node, ) def _ensure_direct_routes(self, host: str, port: int) -> None: if self.direct_route_ensurer is None: return try: self.direct_route_ensurer(str(ipaddress.ip_address(host))) return except ValueError: pass try: for item in socket.getaddrinfo(host, port, socket.AF_UNSPEC, socket.SOCK_STREAM): self.direct_route_ensurer(str(item[4][0]).split("%", 1)[0]) except OSError: pass def _open_target(self, target: Address) -> socket.socket: policy_host = self._policy_host(target.host) rule = self._route_rule(policy_host, target.host) route_node = self._route_node(rule) if rule and str(rule.get("action") or "").lower() == "node": if route_node is None: raise ConnectionError("域名路由指定的节点不可用") return self._dialer(route_node).open(target) direct = self._direct_target_for(target.host, target.port, policy_host) if direct is not None: self._ensure_direct_routes(direct[0], direct[1]) return socket.create_connection(direct, 15) return self._dialer(self.node).open(target) def _tcp(self, client: socket.socket, target: Address): upstream = self._open_target(target) upstream.settimeout(None); client.settimeout(None) client.sendall(b"\x05\x00\x00\x01\x00\x00\x00\x00\x00\x00") self._relay(client, upstream) def _relay(self, left: socket.socket, right: socket.socket): sockets = [left, right] try: while self._running.is_set(): readable, _, _ = select.select(sockets, [], [], 1) for source in readable: data = source.recv(65536) if not data: return target = right if source is left else left target.sendall(data) with self._counter_lock: if source is left: self._up += len(data) else: self._down += len(data) finally: for item in sockets: try: item.close() except OSError: pass def _udp(self, control: socket.socket): # tun2socks v2.7 在 Windows 上通过 ListenPacket("udp", "") # 创建 IPv6 wildcard socket（日志形如 [::]:port）。SOCKS5 返回的 # UDP relay 地址必须与这个地址族一致，否则 Windows 会在 # wsasendto 阶段以 WSAEINVAL 拒绝向 127.0.0.1:port 发包。 # 优先使用 IPv6 loopback；老系统没有 IPv6 时再回退 IPv4。 try: udp = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM) udp.bind(("::1", 0)) relay_address = b"\x04" + socket.inet_pton(socket.AF_INET6, "::1") relay_bind = f"[::1]:{udp.getsockname()[1]}" except OSError: udp = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) udp.bind(("127.0.0.1", 0)) relay_address = b"\x01" + socket.inet_aton("127.0.0.1") relay_bind = f"127.0.0.1:{udp.getsockname()[1]}" port = udp.getsockname()[1] udp.settimeout(1) with self._udp_resources_lock: self._udp_resources.append(udp) control.sendall(b"\x05\x00\x00" + relay_address + struct.pack("!H", port)) self.log(f"[桥接] SOCKS5 UDP 关联已建立: {relay_bind}") control.settimeout(1) streams: dict[str, socket.socket] = {} direct_flows: dict[tuple[str, int], tuple[socket.socket, tuple, Address]] = {} malformed = 0 try: while self._running.is_set(): try: packet, peer = udp.recvfrom(65535) except socket.timeout: try: if not control.recv(1): break except socket.timeout: continue continue try: destination, payload = _unpack_socks_udp(packet) except Exception as exc: malformed += 1 if malformed <= 5: self.log( f"[桥接] SOCKS5 UDP 数据包解析失败: {exc} len={len(packet)}" ) continue if malformed: malformed = 0 if not streams and not direct_flows: self.log( f"[桥接] SOCKS5 UDP 数据报: target={destination.host}:{destination.port} " f"bytes={len(payload)}" ) if destination.port == 53: domain = self._dns.observe_query(payload) if domain: self.log( f"[DNS] 查询: domain={domain} server={destination.host}:{destination.port}" ) policy_host = self._policy_host(destination.host) rule = self._route_rule(policy_host, destination.host) route_node = self._route_node(rule) if rule and str(rule.get("action") or "").lower() == "node" and route_node is None: raise ConnectionError("UDP 域名路由指定的节点不可用") direct = None if route_node is not None else self._direct_target_for( destination.host, destination.port, policy_host) if direct is not None: key = (str(direct[0]), int(direct[1])) flow = direct_flows.get(key) if flow is None: flow_socket, sockaddr = self._open_direct_udp(direct) flow = (flow_socket, sockaddr, destination) direct_flows[key] = flow threading.Thread( target=self._direct_udp_reader, args=(flow_socket, udp, destination, peer), daemon=True, ).start() self.log(f"[UDP] 直连通道: target={destination.host}:{destination.port}") flow_socket, direct_addr, _flow_destination = flow flow_socket.sendto(payload, direct_addr) with self._counter_lock: self._up += len(payload) continue selected_node = route_node or self.node stream_key = self._node_key(selected_node) stream = streams.get(stream_key) if stream is None: self.log( f"[UoT] 建立 UDP 通道: target={destination.host}:{destination.port}" + ("（固定节点）" if route_node is not None else "") ) stream = self._dialer(selected_node).open( Address(MAGIC_ADDRESS, _node_uot_connect_port(selected_node)) ) stream.settimeout(None) stream.sendall(encode_request(False, destination)) streams[stream_key] = stream with self._udp_resources_lock: self._udp_resources.append(stream) threading.Thread( target=self._uot_reader, args=(stream, udp, peer), daemon=True ).start() # UoT v2 non-connect 模式每个数据报都带目标地址。 stream.sendall(encode_datagram_frame(destination, payload)) with self._counter_lock: self._up += len(payload) finally: udp.close() for stream in streams.values(): try: stream.close() except OSError: pass for flow_socket, _sockaddr, _destination in direct_flows.values(): try: flow_socket.close() except OSError: pass with self._udp_resources_lock: self._udp_resources = [ item for item in self._udp_resources if item is not udp and item not in streams.values() ] def _open_direct_udp_target(self, target: tuple[str, int]) -> tuple[socket.socket, tuple]: host, port = target last_error = None for item in socket.getaddrinfo(host, port, socket.AF_UNSPEC, socket.SOCK_DGRAM): family, socktype, proto, _canonname, sockaddr = item raw = socket.socket(family, socktype, proto) try: destination_ip = str(sockaddr[0]).split("%", 1)[0] if self.direct_route_ensurer is not None: self.direct_route_ensurer(destination_ip) is_loopback = False try: is_loopback = ipaddress.ip_address(destination_ip).is_loopback except ValueError: pass if (family == socket.AF_INET and self.source_address and ":" not in self.source_address and not is_loopback): raw.bind((self.source_address, 0)) elif (family == socket.AF_INET6 and self.source_address and ":" in self.source_address and not is_loopback): raw.bind((self.source_address, 0, 0, 0)) if family == socket.AF_INET: try: raw.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) except OSError: pass raw.settimeout(1) return raw, sockaddr except OSError as exc: last_error = exc raw.close() raise last_error or OSError(f"无法解析 UDP 目标: {host}") def _open_direct_udp(self, target: tuple[str, int]) -> tuple[socket.socket, tuple]: raw, sockaddr = self._open_direct_udp_target(target) # 解析后的 sockaddr 直接用于 sendto，保留 IPv6 scope id。 return raw, sockaddr def _direct_udp_reader(self, raw: socket.socket, udp: socket.socket, destination: Address, peer): try: while self._running.is_set(): try: payload, _source = raw.recvfrom(65535) except socket.timeout: continue if not payload: continue if destination.port == 53: observed = self._dns.observe_response(payload) if observed: domain, ips = observed self.log(f"[DNS] 直连回包: domain={domain} ip={','.join(ips)}") udp.sendto(_pack_socks_udp(destination, payload), peer) with self._counter_lock: self._down += len(payload) except OSError: pass def _uot_reader(self, stream, udp, peer): try: while self._running.is_set(): destination, payload = read_datagram_frame(stream) observed = self._dns.observe_response(payload) if observed: domain, ips = observed self.log(f"[DNS] 回包: domain={domain} ip={','.join(ips)}") udp.sendto(_pack_socks_udp(destination, payload), peer) with self._counter_lock: self._down += len(payload) except Exception as exc: if self._running.is_set(): self.log(f"[UoT] UDP 回包读取失败: {exc}") def stop(self): self._running.clear() if self.listener: try: self.listener.close() except OSError: pass self.listener = None with self._udp_resources_lock: resources = list(self._udp_resources) self._udp_resources.clear() for resource in resources: try: resource.close() except OSError: pass def traffic(self) -> tuple[int, int, int, int]: with self._counter_lock: return self._up, self._down, self._up, self._down