From bdd9d3498db04e9a43d7a331feff03bc269c47ce Mon Sep 17 00:00:00 2001 From: RRRosmontis Date: Sat, 19 Sep 2026 16:41:12 +0800 Subject: [PATCH] =?UTF-8?q?fix(uot-relay):=20=E6=95=B0=E6=8D=AE=E6=8A=A5?= =?UTF-8?q?=E7=BA=A7=E9=94=99=E8=AF=AF=E4=B8=8D=E5=86=8D=E6=9D=80=E9=9A=A7?= =?UTF-8?q?=E9=81=93=20+=20=E5=9C=B0=E5=9D=80=E6=97=8F=E5=81=A5=E5=A3=AE?= =?UTF-8?q?=E6=80=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 实测复现：目标为关闭的 UDP 端口时，连接式 socket 收到 ICMP port unreachable → ConnectionRefusedError → 被兜底 except 接住并 stop.set() → 整条 UoT 隧道被关 （客户端见 'UoT TCP 流已关闭'，多隧道同秒集体死亡）。 修复： - 一律用未连接 UDP socket（不接收 ICMP 错误） - send/recv 的数据报级错误只丢包继续；EAGAIN 等可写重试（1s 上限） - 按地址族惰性建 socket（v4/v6 混用）、IPv6 recvfrom 4 元组归一化 - connect 模式预建 socket（修掉主循环空池即关连接的回归） 五节点已部署并验证：狂发 200 包到关闭端口/无路由目标后隧道仍存活； UoT DNS 端到端 JP/US/AQ/HK2 通过，RU 为该节点出口 UDP 限制（实测 4/10） 导致的偶发超时，非 relay 问题。 --- server_config/README.md | 15 +++ server_config/uot-relay/uot_relay.py | 143 ++++++++++++++++++++++----- 2 files changed, 133 insertions(+), 25 deletions(-) diff --git a/server_config/README.md b/server_config/README.md index 5382b42..650a907 100644 --- a/server_config/README.md +++ b/server_config/README.md @@ -328,6 +328,21 @@ magic 地址 → `127.0.0.1:15201`，其余 CONNECT 走 `$connect_addr` 默认 4. 模块**拒绝 CONNECT 端口 0**（`port < 1` 返回 400）→ 客户端连 magic 地址时 端口必须非 0（用 443 即可，relay 不读 CONNECT 端口，目标取自 UoT 帧头）。 +**健壮性要求（2026-09 实测修复，别踩）**： +- relay 必须用**未连接**的 UDP socket（`sendto`/`recvfrom`）。用连接式 socket 时， + 目标侧 ICMP（port unreachable 等）会变成 `ConnectionRefusedError` 在 send/recv 上抛出， + 若被兜底 `except` 接住并 `stop.set()`，**一条数据报出错就杀掉整条 UoT 隧道** + （客户端表现为 "UoT TCP 流已关闭"、多条隧道同秒集体死亡）。 +- 数据报级错误（`BlockingIOError`/EAGAIN、`ENETUNREACH`、`EHOSTUNREACH`、`EMSGSIZE`） + **只丢当前包**，EAGAIN 先等可写再重试；只有 TCP EOF、协议错误、空闲超时才关隧道。 +- 按地址族惰性建 socket，non-connect 模式可混用 IPv4/IPv6 目标；IPv6 `recvfrom` + 返回 4 元组需归一化。 +- connect 模式要**预建** socket（否则主循环首轮无可 select 对象，会立刻关连接）。 + +**节点出口差异（实测）**：RU 节点到境外 DNS（8.8.8.8:53）UDP 成功率仅约 4/10 +（俄 ISP 限制），表现为 UoT DNS 偶发超时——属该节点出口特性，非 relay 故障。 +此类节点的 UDP 业务建议换用当地可达目标，或走其它节点。 + **客户端协议**：认证与 TCP CONNECT 相同（TLS + Basic Auth），然后 `CONNECT sp.v2.udp-over-tcp.arpa:443 HTTP/1.1`，200 之后收发 UoT v2 帧： 请求头 `| isConnect u8 | ATYP u8 | address | port u16be |`， diff --git a/server_config/uot-relay/uot_relay.py b/server_config/uot-relay/uot_relay.py index de6b559..d8c3919 100644 --- a/server_config/uot-relay/uot_relay.py +++ b/server_config/uot-relay/uot_relay.py @@ -13,18 +13,29 @@ RRPP UoT v2 relay — UDP over TCP 解封装服务（sing-box UoT v2 协议） non-connect 帧: | ATYP | address | port | length u16be | data | 监听 127.0.0.1，仅供本机 nginx proxy_connect 转发，不对公网开放。 + +健壮性设计（2026-09 修复，实测复现过隧道被误杀）： + * 一律使用「未连接」UDP socket（sendto/recvfrom）——连接式 socket 会把目标侧 + ICMP（port unreachable 等）转成 ConnectionRefusedError，在 send/recv 上抛异常， + 旧实现因此把整条 UoT 隧道关掉（客户端表现为 "UoT TCP 流已关闭"）。 + * 数据报级错误（BlockingIOError/EAGAIN、ENETUNREACH、EHOSTUNREACH、EMSGSIZE + 等）只丢当前包并继续；EAGAIN 先等可写再重试（有总时限）。 + * 只有 TCP 侧 EOF、UoT 协议错误或空闲超时才关闭隧道。 + * 按地址族惰性建 socket，non-connect 模式可混用 IPv4/IPv6 目标。 """ +import errno +import select import socket import struct import threading -import select import time -import sys LISTEN_HOST = "127.0.0.1" LISTEN_PORT = 15201 -IDLE_TIMEOUT = 540 # 9 分钟无流量断开（须 < nginx proxy_connect_data_timeout 10m） +IDLE_TIMEOUT = 540 # 9 分钟无流量断开（须 < nginx proxy_connect 隧道超时 10m） MAX_DGRAM = 65507 # UDP over IPv4 单包上限 +SEND_RETRY_TOTAL = 1.0 # EAGAIN 时最多等待 1 秒可写，超时则丢弃该包 +RECV_ERROR_STREAK_LIMIT = 200 # 连续 recv 错误上限（防病态自旋），超过视为隧道异常 LOG_PREFIX = "[uot-relay]" @@ -74,21 +85,76 @@ def encode_addr(addr, port): return b"\x02" + bytes([len(d)]) + d + struct.pack(">H", port) +def norm_src(src): + """recvfrom 地址归一化：IPv4 为 (host, port)，IPv6 为 4 元组。""" + return src[0], src[1] + + +class UdpPool: + """按地址族惰性创建「未连接」UDP socket（v4/v6 可混用）。""" + + def __init__(self): + self.socks = {} + + def get(self, addr): + fam = socket.AF_INET6 if ":" in addr else socket.AF_INET + s = self.socks.get(fam) + if s is None: + s = socket.socket(fam, socket.SOCK_DGRAM) + s.setblocking(False) + self.socks[fam] = s + return s + + def all(self): + return list(self.socks.values()) + + def close(self): + for s in self.socks.values(): + try: + s.close() + except Exception: + pass + self.socks.clear() + + +def send_dgram(sock, data, dst): + """发送一个 UDP 数据报。 + + 返回 True=已发出，False=本包丢弃（**不是**连接级错误，调用方继续跑）。 + EAGAIN 时等可写重试（有总时限）；其余 OSError 一律丢弃本包。 + """ + deadline = time.time() + SEND_RETRY_TOTAL + while True: + try: + sock.sendto(data, dst) + return True + except BlockingIOError: + if time.time() >= deadline: + return False + try: + select.select([], [sock], [], 0.1) + except Exception: + return False + except InterruptedError: + continue + except OSError: + # ENETUNREACH / EHOSTUNREACH / EMSGSIZE / EACCES ... + return False + + def handle(conn, peer): conn.settimeout(IDLE_TIMEOUT) f = conn.makefile("rb") - udp = None + pool = UdpPool() stop = threading.Event() + fixed_dst = None try: is_connect = read_exact(f, 1)[0] req_dst = parse_addr(f) log(f"conn from nginx: isConnect={is_connect} dst={req_dst}") - - udp = socket.socket(socket.AF_INET6 if ":" in req_dst[0] else socket.AF_INET, - socket.SOCK_DGRAM) - udp.setblocking(False) if is_connect: - udp.connect(req_dst) # 固定目标：连通式 UDP + fixed_dst = req_dst + pool.get(fixed_dst[0]) # 预建 socket：主循环首轮才有可 select 的对象 write_lock = threading.Lock() @@ -97,7 +163,7 @@ def handle(conn, peer): conn.sendall(data) def uplink(): - """TCP -> UDP：按格式读帧并发送。""" + """TCP -> UDP：逐帧解析并发送；单包错误只丢包，不关隧道。""" try: while not stop.is_set(): if is_connect: @@ -105,34 +171,65 @@ def handle(conn, peer): if ln > MAX_DGRAM: raise ValueError(f"datagram too large: {ln}") data = read_exact(f, ln) - udp.send(data) + send_dgram(pool.get(fixed_dst[0]), data, fixed_dst) else: dst = parse_addr(f) ln = struct.unpack(">H", read_exact(f, 2))[0] if ln > MAX_DGRAM: raise ValueError(f"datagram too large: {ln}") data = read_exact(f, ln) - udp.sendto(data, dst) + send_dgram(pool.get(dst[0]), data, dst) except Exception as e: if not stop.is_set(): log(f"uplink end: {type(e).__name__}: {e}") finally: stop.set() - t = threading.Thread(target=uplink, daemon=True) - t.start() + threading.Thread(target=uplink, daemon=True).start() - # 主循环：UDP -> TCP，带空闲超时 + # 主循环：UDP -> TCP（带空闲超时） last = time.time() + err_streak = 0 while not stop.is_set(): - r, _, _ = select.select([udp], [], [], 2.0) - if r: + socks = pool.all() + if not socks: + # non-connect 模式：socket 由第一帧触发创建，空池时等待而非关连接 + if time.time() - last > IDLE_TIMEOUT: + log("idle timeout (no datagram), closing") + break + time.sleep(0.2) + continue + try: + r, _, _ = select.select(socks, [], [], 2.0) + except OSError: + break + got = False + for s in r: + try: + data, src = s.recvfrom(MAX_DGRAM) + except (BlockingIOError, InterruptedError): + continue + except OSError as e: + # 单包级错误（含 ICMP 触发的错误）：丢包继续，不关隧道 + err_streak += 1 + if err_streak % 50 == 1: + log(f"recv error (datagram dropped, streak={err_streak}): " + f"{errno.errorcode.get(e.errno, e.errno)} {e}") + if err_streak >= RECV_ERROR_STREAK_LIMIT: + log("recv error streak limit reached, closing tunnel") + stop.set() + break + continue + err_streak = 0 + a, p = norm_src(src) if is_connect: - data = udp.recv(MAX_DGRAM) + if (a, p) != fixed_dst: + continue # 固定目标模式：只回传目标自身的包 tcp_write(struct.pack(">H", len(data)) + data) else: - data, src = udp.recvfrom(MAX_DGRAM) - tcp_write(encode_addr(*src) + struct.pack(">H", len(data)) + data) + tcp_write(encode_addr(a, p) + struct.pack(">H", len(data)) + data) + got = True + if got: last = time.time() elif time.time() - last > IDLE_TIMEOUT: log("idle timeout, closing") @@ -149,11 +246,7 @@ def handle(conn, peer): conn.close() except Exception: pass - if udp: - try: - udp.close() - except Exception: - pass + pool.close() def main():