858 lines
37 KiB
Python
858 lines
37 KiB
Python
"""本地 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 _proxy_target(self, target: Address) -> Address:
|
||
"""为上游 CONNECT 恢复唯一 DNS 域名，给 IPv6 目标提供 IPv4 回退。"""
|
||
try:
|
||
ipaddress.ip_address(target.host)
|
||
except ValueError:
|
||
return target
|
||
domain = self._dns.lookup_ip(target.host)
|
||
if not domain:
|
||
return target
|
||
self.log(
|
||
f"[桥接] IPv6 目标回退域名解析: {target.host} -> {domain}:{target.port}"
|
||
)
|
||
return Address(domain, target.port)
|
||
|
||
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(self._proxy_target(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(self._proxy_target(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
|