fix(uot-relay): 数据报级错误不再杀隧道 + 地址族健壮性
实测复现：目标为关闭的 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 问题。
This commit is contained in:
RRRosmontis
@@ -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():
|
||||
|
||||
Reference in New Issue
Block a user