From d659a4e7f1df21bf0163731fb992e4f9a8d3dc9c Mon Sep 17 00:00:00 2001 From: yu <191972707@qq.com> Date: Fri, 23 Oct 2026 17:30:27 +0800 Subject: [PATCH] =?UTF-8?q?feat(tools):=20=E6=89=93=E6=B4=9E=E6=8E=A2?= =?UTF-8?q?=E9=92=88=E4=B8=8E=E6=9E=81=E7=AE=80=20rendezvous=20=E6=9C=8D?= =?UTF-8?q?=E5=8A=A1=E5=99=A8=EF=BC=88Python=20POC=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - punch_probe.py: UDP 打洞探针，支持 STUN 查映射、--peer 手工配对、--server/--room 信令配对 - rendezvous.py: 极简 UDP 信令服务器，只做房间配对与地址回推，不承载媒体 - 已在 47.98.245.66 实测跨公网注册/配对可用 --- tools/punch_probe.py | 343 +++++++++++++++++++++++++++++++++++++++++++ tools/rendezvous.py | 90 ++++++++++++ 2 files changed, 433 insertions(+) create mode 100644 tools/punch_probe.py create mode 100644 tools/rendezvous.py diff --git a/tools/punch_probe.py b/tools/punch_probe.py new file mode 100644 index 0000000..6c8abdf --- /dev/null +++ b/tools/punch_probe.py @@ -0,0 +1,343 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""UDP 打洞探针 —— 两台不同网络的机器各跑一个，用来摸清你们线路的底牌。 + +回答四个问题： + 1. 这条线路的 NAT 能不能打洞？ + 2. 要几轮才能打通？（轮数越少说明 NAT 越宽松） + 3. 对方的源端口有没有被 NAT 改写？（判断是不是对称 NAT） + 4. 有没有公网 IPv6？IPv6 直连通不通？ + +用法 +---- +A 机（假设公网 1.2.3.4）: + py punch_probe.py --peer 2.3.4.5:47800 + +B 机（假设公网 2.3.4.5）: + py punch_probe.py --peer 1.2.3.4:47800 + +IPv6（双方都有公网 v6 时）: + py punch_probe.py --peer [2408:8207:1234::1]:47800 + +先查自己的公网映射（走 STUN，不需要对方）: + py punch_probe.py --stun + +被动监听（不发包，只看能不能收到对方打进来的包）: + py punch_probe.py + +说明 +---- +- 两台机器必须**同时**启动，脚本靠「双方同时向对方发包」来穿透 NAT。 +- 先各自查公网 IP（浏览器搜 "my ip" 或 --stun），再互相填。 +- 只依赖 Python 标准库。 +""" + +import argparse +import socket +import struct +import sys +import threading +import time + +MAGIC = b"PUNCHv1|" + + +# ---------------------------------------------------------------- 地址解析 + +def parse_addr(s): + """解析 'ip:port' 或 '[v6]:port' -> (host, port, family)""" + s = s.strip() + if s.startswith("["): + end = s.index("]") + host = s[1:end] + port = int(s[end + 2:]) + return host, port, socket.AF_INET6 + host, _, p = s.rpartition(":") + return host, int(p), socket.AF_INET + + +def fmt_addr(sockaddr): + host, port = sockaddr[0], sockaddr[1] + if ":" in host: + return "[%s]:%d" % (host, port) + return "%s:%d" % (host, port) + + +# ---------------------------------------------------------------- STUN + +def stun_mapped(server="stun.l.google.com", port=19302, timeout=3.0, sock=None): + """最小 STUN Binding 请求，返回 (ip, port)；失败返回 None。 + + sock 不为 None 时复用它（用来测「同一个本地端口对另一个目标」的映射端口， + 这是判断 NAT 是不是对称型的关键）。 + """ + txid = b"\x5a" * 12 + req = struct.pack(">HHI12s", 0x0001, 0, 0x2112A442, txid) + try: + infos = socket.getaddrinfo(server, port, 0, socket.SOCK_DGRAM) + except OSError: + return None + fam, sockaddr = infos[0][0], infos[0][4] + own = sock is None + s = socket.socket(fam, socket.SOCK_DGRAM) if own else sock + s.settimeout(timeout) + try: + s.sendto(req, sockaddr) + data, _ = s.recvfrom(2048) + except OSError: + return None + finally: + if own: + s.close() + + if len(data) < 20: + return None + off = 20 + while off + 4 <= len(data): + atype, alen = struct.unpack(">HH", data[off:off + 4]) + val = data[off + 4:off + 4 + alen] + if atype == 0x0020 and len(val) >= 8: # XOR-MAPPED-ADDRESS + famv = val[1] + xport = struct.unpack(">H", val[2:4])[0] ^ 0x2112 + if famv == 0x01 and len(val) >= 8: + ip = bytes(a ^ b for a, b in zip(val[4:8], b"\x21\x12\xA4\x42")) + return socket.inet_ntoa(ip), xport + if famv == 0x02 and len(val) >= 20: + mask = b"\x21\x12\xA4\x42" + txid + ip = bytes(a ^ b for a, b in zip(val[4:20], mask)) + return socket.inet_ntop(socket.AF_INET6, ip), xport + off += 4 + alen + ((4 - alen % 4) % 4) + return None + + +# ---------------------------------------------------------------- 信令 + +def discover_peer(sock, server, room, timeout=60.0): + """注册到房间，等服务器回推对方的公网地址。 + + 注意：这里**复用同一个 socket**——服务器看到的源地址，就是 NAT 为 + 「本机 → 服务器」这条流分配的映射。对锥形 NAT 而言，它与「本机 → 对端」 + 的映射相同，所以拿它去打洞是有效的；对对称 NAT 则不同，打不通属预期。 + """ + host, port = server[0], server[1] + deadline = time.time() + timeout + last_send = 0.0 + while time.time() < deadline: + now = time.time() + if now - last_send > 1.0: + try: + sock.sendto(b"REGISTER|" + room.encode("utf-8"), (host, port)) + except OSError as e: + print(" 注册发送失败：%s" % e) + return None + last_send = now + try: + data, _src = sock.recvfrom(2048) + except socket.timeout: + continue + except ConnectionResetError: + continue + except OSError as e: + if getattr(e, "winerror", None) == 10054: + continue + return None + txt = data.decode("utf-8", "replace").strip() + if txt.startswith("PEER|"): + h, p, _f = parse_addr(txt[5:]) + return (h, p) + return None + + +# ---------------------------------------------------------------- 主逻辑 + +class Probe: + def __init__(self, bind_host, port, family): + self.sock = socket.socket(family, socket.SOCK_DGRAM) + self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + if family == socket.AF_INET6: + try: + self.sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 1) + except OSError: + pass + self.sock.bind((bind_host, port)) + self.sock.settimeout(0.2) + # Windows 特有坑：曾经发过包给对方，但对方端口没开 → 会收到 ICMP 端口不可达， + # 之后下一次 recv 直接抛 WSAECONNRESET(10054)。不打这个 ioctl， + # 接收循环会被"自己之前的失败发送"打断，之后再也收不到包。 + # （Rust 实现同样要处理：tokio recv 返回 ConnectionReset 时不能退出循环） + if sys.platform == "win32": + try: + self.sock.ioctl(socket.SIO_UDP_CONNRESET, False) + except (AttributeError, OSError): + pass + self.port = self.sock.getsockname()[1] + self.stop = threading.Event() + self.first_src = None + self.packets = 0 + + def receiver(self): + while not self.stop.is_set(): + try: + data, src = self.sock.recvfrom(2048) + except socket.timeout: + continue + except ConnectionResetError: + # ICMP 端口不可达留下的错误，忽略并继续收（绝不能 break） + continue + except OSError as e: + if getattr(e, "winerror", None) == 10054: + continue + break + if self.first_src is None: + self.first_src = src + print("\n ✅ 收到对方的包！") + print(" 对方源地址（NAT 之后）= %s" % fmt_addr(src)) + if src[1] != self.port: + print(" 对方源端口 = %d（本机端口 %d）；若双方约定端口相同，" + "说明 NAT 改写了端口" % (src[1], self.port)) + else: + print(" 对方源端口与约定端口一致 → NAT 保留了端口") + self.packets += 1 + if self.packets <= 3: + print(" [%s] 收到 %d 字节" % (fmt_addr(src), len(data))) + + def send_rounds(self, peer_sockaddr, rounds, interval): + for i in range(1, rounds + 1): + if self.stop.is_set(): + return + payload = MAGIC + struct.pack(">d", time.time()) + try: + self.sock.sendto(payload, peer_sockaddr) + except OSError as e: + print(" 发送失败：%s" % e) + return + if i == 1 or i % 5 == 0 or self.first_src is not None: + mark = "✅" if self.first_src is not None else " " + print(" %s 第 %2d/%d 轮已发出" % (mark, i, rounds)) + if self.first_src is not None: + return + time.sleep(interval) + + +def main(): + ap = argparse.ArgumentParser(add_help=True, description="UDP 打洞探针") + ap.add_argument("--peer", help="对方地址 ip:port 或 [v6]:port；不给则只被动监听") + ap.add_argument("--port", type=int, default=47800, help="本地端口（默认 47800）") + ap.add_argument("--rounds", type=int, default=25, help="发包轮数（默认 25）") + ap.add_argument("--interval", type=float, default=0.2, help="轮间隔秒（默认 0.2）") + ap.add_argument("--grace", type=float, default=2.5, + help="发完后继续监听多少秒等回包（默认 2.5）") + ap.add_argument("--server", help="信令服务器 ip:port（与 --room 配合，自动配对）") + ap.add_argument("--room", help="房间名；同一房间的两个客户端会被配对") + ap.add_argument("--wait", type=float, default=60.0, + help="等对方加入房间的秒数（默认 60）") + ap.add_argument("--stun", action="store_true", help="只查询自己的公网映射后退出") + ap.add_argument("--stun-port", type=int, default=0, + help="配合 --stun：用指定本地端口发 STUN（与打洞端口一致时才能对比映射）") + args = ap.parse_args() + + # ---- 只查 STUN + if args.stun: + s = None + if args.stun_port: + s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + s.bind(("0.0.0.0", args.stun_port)) + print("[STUN] 用本机端口 %d 查询（与打洞同端口）..." % args.stun_port) + else: + print("[STUN] 查询公网映射（临时端口）...") + r = stun_mapped(sock=s) + if s: + s.close() + if r: + print(" 对 Google STUN 的映射 = %s:%d" % r) + if args.stun_port: + print(" 本机端口 = %d" % args.stun_port) + print(" 下一步：对比信令服务器日志里记录的源端口") + print(" 两者相同 → 端点无关映射（锥形 NAT）→ 打洞大概率成功") + print(" 两者不同 → 对称 NAT → 打洞困难，需要端口预测或中继") + else: + print(" 查询失败（STUN 19302 可能被封；可稍后重试）") + return 0 + + # ---- 决定地址族与模式 + peer = None + if args.peer: + host, pport, family = parse_addr(args.peer) + peer = (host, pport) + bind_host = "::" if family == socket.AF_INET6 else "0.0.0.0" + mode = "双向打洞（手工填地址）" + elif args.server and args.room: + _h, _p, family = parse_addr(args.server) + bind_host = "::" if family == socket.AF_INET6 else "0.0.0.0" + mode = "信令自动配对 + 打洞" + else: + family = socket.AF_INET + bind_host = "0.0.0.0" + mode = "被动监听" + + probe = Probe(bind_host, args.port, family) + print("=" * 60) + print(" UDP 打洞探针 模式=%s" % mode) + print(" 本机监听 %s:%d" % (bind_host, probe.port)) + if peer: + print(" 发往 %s （%s）" % (fmt_addr(peer + (0, 0)), + "IPv6" if family == socket.AF_INET6 else "IPv4")) + print("=" * 60) + + # ---- 信令配对：不用自己查公网 IP，服务器看到的就是你的公网映射 + if args.server and args.room and peer is None: + srv = parse_addr(args.server)[:2] + print("\n向信令服务器 %s 注册 room=%s ..." % (fmt_addr(srv), args.room)) + found = discover_peer(probe.sock, srv, args.room, timeout=args.wait) + if found is None: + print(" ❌ 超时：没等到对方加入同一个房间") + probe.sock.close() + return 1 + peer = found + print(" ✅ 配对成功，对方公网地址 = %s" % fmt_addr(found)) + + t = threading.Thread(target=probe.receiver, daemon=True) + t.start() + + try: + if peer: + print("\n开始同时发包（若对方也在发包，第 1~3 轮通常就通）...") + probe.send_rounds(peer, args.rounds, args.interval) + # 对方可能比我晚启动：发完后继续监听，否则会把"晚到的成功"误判成失败 + if probe.first_src is None: + print("\n发包结束，继续监听 %.1f 秒等对方回包 ..." % args.grace) + waited = 0.0 + while waited < args.grace and probe.first_src is None: + time.sleep(0.1) + waited += 0.1 + else: + print("\n被动监听中，Ctrl+C 退出 ...") + while not probe.stop.is_set(): + time.sleep(0.3) + except KeyboardInterrupt: + pass + finally: + probe.stop.set() + time.sleep(0.3) + probe.sock.close() + + print("\n" + "-" * 60) + if probe.first_src is not None: + print("结论：✅ 打洞成功") + print(" 对方实际源地址 = %s" % fmt_addr(probe.first_src)) + print(" 共收到 %d 个包" % probe.packets) + else: + print("结论：❌ 本轮没有收到对方的包") + print(" 可能原因（按概率排序）：") + print(" 1. 一端或两端是对称型 NAT / 运营商级 NAT(CGNAT) → 需要端口预测或中继") + print(" 2. 一方根本没发包（对方没启动、或地址填错/端口填错）") + print(" 3. UDP 被防火墙拦截（企业网络常见）") + print(" 4. 双方没有真正同时发包（时间差超过 NAT 表项寿命）") + print(" 下一步：两端各跑一次 --stun，看看是不是落在 100.64~100.127 网段（CGNAT）") + print("-" * 60) + return 0 if probe.first_src is not None else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tools/rendezvous.py b/tools/rendezvous.py new file mode 100644 index 0000000..39b3e8b --- /dev/null +++ b/tools/rendezvous.py @@ -0,0 +1,90 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""极简 rendezvous（信令）服务器 —— POC 用 + +作用只有一个：**告诉双方「对方看到的你」是什么地址**。 +双方注册到同一个房间后，服务器把各自观察到的公网地址回推给对方， +之后双方直接打洞，服务器不再参与（也不承载任何媒体流量）。 + +用法（跑在有公网 IP 的机器上，比如 43.140.215.244）: + py rendezvous.py --port 47999 + +客户端： + py punch_probe.py --server <服务器IP>:47999 --room test01 --port 47800 +""" + +import argparse +import socket +import sys +import time + +# 房间成员多久没动静就清掉 +MEMBER_TTL = 30.0 + + +def fmt(sockaddr): + host, port = sockaddr[0], sockaddr[1] + return "[%s]:%d" % (host, port) if ":" in host else "%s:%d" % (host, port) + + +def main(): + ap = argparse.ArgumentParser(description="极简 rendezvous 服务器") + ap.add_argument("--bind", default="0.0.0.0") + ap.add_argument("--port", type=int, default=47999) + args = ap.parse_args() + + # 重定向到文件时 stdout 会全缓冲，日志要等进程退出才出现；强制行缓冲 + try: + sys.stdout.reconfigure(line_buffering=True) + except (AttributeError, OSError): + pass + + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + sock.bind((args.bind, args.port)) + + print("[rendezvous] 监听 udp://%s:%d" % (args.bind, args.port)) + print("[rendezvous] 房间成员 TTL = %.0fs，Ctrl+C 退出\n" % MEMBER_TTL) + + rooms = {} # room -> [(addr, last_seen)] + + try: + while True: + data, src = sock.recvfrom(2048) + txt = data.decode("utf-8", "replace").strip() + now = time.time() + + if not txt.startswith("REGISTER|"): + continue + room = txt.split("|", 1)[1].strip() + if not room: + continue + + # 清掉过期成员 + 去掉本机的旧记录（支持重复注册） + members = [(a, t) for (a, t) in rooms.get(room, []) + if now - t < MEMBER_TTL and a[1] != src[1]] + members.append((src, now)) + rooms[room] = members + + print("[rendezvous] %s 注册 room=%s (现有 %d 人) 源地址=%s" + % (fmt(src), room, len(members), fmt(src))) + + if len(members) >= 2: + a1, _ = members[0] + a2, _ = members[1] + sock.sendto(b"PEER|" + fmt(a2).encode(), a1) + sock.sendto(b"PEER|" + fmt(a1).encode(), a2) + print("[rendezvous] -> 配对成功：%s <==> %s" + % (fmt(a1), fmt(a2))) + else: + sock.sendto(b"WAIT", src) + print("[rendezvous] -> 等待第二个人加入 ...") + except KeyboardInterrupt: + print("\n[rendezvous] 退出") + finally: + sock.close() + return 0 + + +if __name__ == "__main__": + sys.exit(main())