正反向隔离装置下TCP/UDP穿透Demo：原理、实现与避坑指南

简介：这份资源面向从事网络隔离与安全传输的开发者，聚焦正反向隔离装置场景下的TCP/UDP穿透实现，在原有TCP代理基础上补充了UDP穿透能力，适合需要打通隔离环境双向通信、验证代理转发逻辑的中高级网络编程人员参考。压缩包共11个文件，约1.68MB，以cpp源码与ini配置为主，辅以可执行测试程序、说明文档及一份整体方案演示文稿，覆盖代理服务、TCP与UDP收发测试等模块，便于对照源码理解穿透流程与配置方式。目前已有1253人学习下载。资源提供了可运行的测试程序与配置样例，读者可据此搭建本地验证环境，观察TCP与UDP数据在隔离装置两侧的转发路径，结合说明文档与方案演示梳理整体架构，为自研隔离穿透模块或排查代理转发异常提供可借鉴的实现思路与调试参考。

1. 正反向隔离装置下的 TCP/UDP 穿透：一个能跑起来的 Demo 到底解决了什么

做电力、工控或者某些内网项目的人，大概率都遇到过同一个坎：业务系统之间明明网络是通的，但中间横着一台正反向隔离装置，TCP 连接建不起来，UDP 报文也过不去。你 ping 得通、telnet 不通，抓包看到 SYN 发出去石沉大海，最后只能对着拓扑图干瞪眼。这个 Demo 就是冲着这个场景来的——它用一套最小可运行的代码，演示了在正反向隔离装置约束下，如何把 TCP 和 UDP 流量“穿透”过去，让两侧的业务进程能正常通信。适合谁？适合手里有类似隔离环境、需要做协议适配或代理转发的开发者，也适合想搞懂隔离装置前后到底该怎么设计通信链路的同学。它不是生产级方案，但把核心链路拆得足够清楚，拿来改比从零搭要省太多事。

2. 架构拆解：穿透链路是怎么在隔离装置两侧搭起来的

2.1 整体拓扑与角色划分

先把这张图在脑子里立起来。整个 Demo 的通信链路可以抽象成四个角色：客户端、近端代理、隔离装置、远端代理、服务端。客户端发起 TCP 连接或 UDP 报文，近端代理负责在隔离装置这一侧“接住”流量，然后通过隔离装置允许的通道把数据摆渡到另一侧，远端代理再把数据还原成原始协议发给真正的服务端。隔离装置本身对 TCP/UDP 是不透明的，它只认自己那套摆渡规则，所以穿透的本质不是“骗过”隔离装置，而是在它两侧各放一个代理，把原始协议拆成它能接受的形式，过去之后再拼回来。

这个设计里最关键的一点是：近端代理和远端代理之间不能假设有长连接。正反向隔离装置的一个典型特征是单向性或准单向性，正向和反向的数据通道可能是分开的，甚至反向通道的带宽和时延都跟正向不一样。所以 Demo 里代理之间的数据摆渡没有用“一条 TCP 连接扛到底”的思路，而是做成了带会话标识的报文转发，每个会话用唯一 ID 关联，正向走正向通道，反向走反向通道，两侧各自维护会话表。这样即使某一向的通道短暂中断，会话状态也不会立刻崩掉。

从代码组织上看，Demo 大致分成这么几块：协议解析层（区分 TCP 流和 UDP 报文）、会话管理层（建会话、查会话、销毁会话）、摆渡适配层（对接隔离装置允许的传输方式）、以及两侧代理的入口。这个分层不是为了好看，是为了让你改的时候知道该动哪——比如你换一种隔离装置的摆渡方式，只需要动摆渡适配层，会话管理和协议解析不用碰。

2.2 目录结构与模块职责

Demo 的目录结构不复杂，但每个文件都有明确归属，我按实际拆解的习惯给你列一下：

tcp-udp-penetration-demo/ ├── common/ │ ├── protocol.py # 报文头定义、序列化/反序列化 │ ├── session.py # 会话表、会话生命周期管理 │ └── config.py # 端口、超时、缓冲区大小等配置 ├── near_agent/ │ ├── tcp_listener.py # 近端 TCP 监听与连接接入 │ ├── udp_listener.py # 近端 UDP 报文接收 │ └── forwarder.py # 向隔离装置方向摆渡 ├── far_agent/ │ ├── tcp_connector.py # 远端向真实服务端建连 │ ├── udp_sender.py # 远端向真实服务端发 UDP │ └── receiver.py # 从隔离装置方向收数据 └── run_demo.py # 一键启动两侧代理的入口

common/protocol.py是整个链路的地基。因为数据要在隔离装置两侧来回摆渡，必须有一个自描述的报文格式，让接收方知道这段数据属于哪个会话、是 TCP 还是 UDP、是正向还是反向。common/session.py管的是“谁跟谁是一伙的”，每个客户端连接或 UDP 五元组对应一个会话，会话里存着缓冲区、最后活跃时间、状态机。near_agent和far_agent是对称的两套入口，区别在于近端是“接流量”，远端是“发流量”。run_demo.py把两侧拉起来，方便你本地先跑通再拆开部署。

这里有个设计取舍值得说：Demo 没有引入任何消息队列或中间件，摆渡通道就是最朴素的 socket 转发。原因是隔离装置环境里往往不允许你随便装东西，越少依赖越容易落地。你二次开发时如果环境允许，可以把摆渡层换成文件摆渡或者共享内存，接口留好了就行。

2.3 报文格式设计：让隔离装置两侧能对上话

穿透能不能成，一半看报文格式设计得够不够自描述。Demo 里定义了一个固定头加可变体的结构，头里带会话 ID、协议类型、方向标志、载荷长度，体里就是原始 TCP 数据或 UDP 载荷。下面这段是protocol.py里的核心打包逻辑：

import struct # 报文头格式：会话ID(4B) + 协议类型(1B) + 方向(1B) + 载荷长度(2B) HEADER_FMT = "!IBBH" HEADER_LEN = struct.calcsize(HEADER_FMT) PROTO_TCP = 1 PROTO_UDP = 2 DIR_FORWARD = 0 # 近端 -> 远端 DIR_BACKWARD = 1 # 远端 -> 近端 def pack_message(session_id, proto, direction, payload): """把原始数据封装成可摆渡的报文""" if len(payload) > 65535: raise ValueError("payload too large for demo header") header = struct.pack(HEADER_FMT, session_id, proto, direction, len(payload)) return header + payload def unpack_message(raw): """从摆渡通道收到的字节流里还原出会话信息和载荷""" if len(raw) < HEADER_LEN: return None session_id, proto, direction, length = struct.unpack(HEADER_FMT, raw[:HEADER_LEN]) payload = raw[HEADER_LEN:HEADER_LEN + length] if len(payload) != length: return None # 半包，交给上层缓冲处理 return session_id, proto, direction, payload

这段代码的逻辑很直白：pack_message把会话 ID、协议类型、方向和载荷长度打进 8 字节的头，后面跟原始数据。unpack_message反过来解析，长度对不上就返回None，表示这是个半包，需要上层继续缓冲。参数上，session_id用 4 字节无符号整数，Demo 里用自增计数器分配，实际部署时建议换成哈希或者带时间戳的生成方式，避免重启后 ID 冲突。proto和direction各占 1 字节，够用且省空间。载荷长度用 2 字节，意味着单条报文最大 64KB，对 TCP 流来说需要上层做分片，对 UDP 来说刚好覆盖常见 MTU 范围。

提示：!IBBH里的!表示网络字节序，跨机器摆渡时千万别省，否则大小端不一致会让会话 ID 解析成完全不同的值，这种问题抓包都不一定看得出来。

2.4 会话管理：TCP 流和 UDP 报文怎么统一管

TCP 是有连接的流，UDP 是无连接的报文，两者在穿透场景下的管理方式不一样，但 Demo 用一张会话表把它们统一了。TCP 会话以“客户端地址 + 端口”为键，UDP 会话以五元组为键，会话对象里存协议类型、远端连接句柄、接收缓冲区、最后活跃时间。下面这段是session.py的核心：

import time import threading class Session: def __init__(self, session_id, proto, peer_addr): self.session_id = session_id self.proto = proto self.peer_addr = peer_addr self.buffer = b"" self.last_active = time.time() self.remote_sock = None # TCP 时是 socket 对象，UDP 时为 None self.lock = threading.Lock() def touch(self): self.last_active = time.time() class SessionTable: def __init__(self, idle_timeout=120): self._sessions = {} self._lock = threading.Lock() self.idle_timeout = idle_timeout def create(self, session_id, proto, peer_addr): with self._lock: sess = Session(session_id, proto, peer_addr) self._sessions[session_id] = sess return sess def get(self, session_id): with self._lock: sess = self._sessions.get(session_id) if sess: sess.touch() return sess def sweep(self): """清理超时会话，避免表无限增长""" now = time.time() with self._lock: expired = [sid for sid, s in self._sessions.items() if now - s.last_active > self.idle_timeout] for sid in expired: sess = self._sessions.pop(sid) if sess.remote_sock: try: sess.remote_sock.close() except OSError: pass

Session里那个lock不是摆设。TCP 场景下，近端可能在收数据的同时远端在回数据，两个线程操作同一个会话的缓冲区，不加锁就会出现数据交错。SessionTable.sweep是必须定时调的，Demo 在run_demo.py里起了个后台线程每 30 秒扫一次，idle_timeout默认 120 秒。这个值怎么定？看你的业务心跳间隔，比心跳大两三倍比较稳妥。太短会把正常但安静的长连接误杀，太长会让会话表在异常场景下膨胀。

UDP 会话有个特殊处理：因为 UDP 没有连接状态，Demo 在收到第一个报文时建会话，后续同五元组的报文复用会话，超时后销毁。TCP 则是连接建立时建会话，连接关闭或超时销毁。这个差异在near_agent的入口代码里体现得很清楚，下一章会展开。

3. 跑起来：从本地回环到模拟隔离环境的两级验证

3.1 环境准备与依赖

Demo 的依赖压到了最低，Python 3.8 以上，标准库就够，不需要额外装包。这一点是刻意的——隔离装置环境里装依赖往往要走审批，能少一个是一个。你拿到代码后先确认版本：

python3 --version # 期望输出 Python 3.8.x 及以上

如果版本低于 3.8，struct和线程相关的用法基本兼容，但建议还是升一下，避免一些边界行为差异。代码里没有用asyncio，全是阻塞 socket 加线程，原因是隔离环境里事件循环的调试成本高，出问题时线程栈比协程栈好排查。这个选择不优雅，但实用。

配置集中在common/config.py，几个关键参数先认识一下：

本地验证时这些默认值不用改，直接能跑。模拟隔离环境时，你需要把BRIDGE_PORT对应的摆渡通道换成隔离装置允许的方式，这个后面讲。

3.2 本地回环跑通 TCP 穿透

第一步先在单机上把链路跑通，确认逻辑没问题再谈隔离。开三个终端。第一个终端起一个最简单的 TCP 回声服务端，模拟真实业务：

# echo_server.py import socket srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) srv.bind(("127.0.0.1", 8001)) srv.listen(5) print("echo server on 8001") while True: conn, addr = srv.accept() while True: data = conn.recv(4096) if not data: break conn.sendall(b"echo:" + data) conn.close()

第二个终端起 Demo 的两侧代理：

python3 run_demo.py --mode all # 输出类似： # near tcp listener on 9001 # near udp listener on 9002 # bridge on 9100 # far agent connected to 127.0.0.1:8001

第三个终端用nc或者一段 Python 客户端打近端端口：

printf 'hello-penetration' | nc 127.0.0.1 9001 # 期望收到 echo:hello-penetration

如果收到回声，说明“客户端 → 近端 → 摆渡 → 远端 → 回声服务端 → 原路返回”整条链路是通的。这一步看着简单，但它验证了三件事：报文封装解封装没丢数据、会话 ID 在两侧对得上、TCP 半关闭处理没把连接提前掐断。很多穿透 Demo 卡在最后一点上，客户端发完数据关写端，代理误以为整个会话结束，回声还没回来就把会话销毁了。Demo 里对recv返回空做了区分处理，只有读写都结束才销毁。

3.3 UDP 穿透验证与抓包确认

UDP 的验证方式类似，但因为没有连接概念，要用能收发的工具。先起一个 UDP 回声服务端：

# udp_echo.py import socket srv = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) srv.bind(("127.0.0.1", 8002)) print("udp echo on 8002") while True: data, addr = srv.recvfrom(4096) srv.sendto(b"udp-echo:" + data, addr)

然后客户端往近端 UDP 端口发：

# udp_client.py import socket cli = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) cli.settimeout(3) cli.sendto(b"ping-through-isolation", ("127.0.0.1", 9002)) try: data, _ = cli.recvfrom(4096) print("got:", data) except socket.timeout: print("timeout - check session table and bridge")

收到udp-echo:ping-through-isolation就算通。UDP 这里最容易翻车的是会话匹配：客户端每次sendto的源端口如果变了，近端会当成新会话，远端就不知道该把回包发给谁。Demo 里近端以“客户端 IP + 端口 + 目标端口”做键，客户端用同一个 socket 发就没问题。如果你用nc -u测试，注意它可能每次换端口，表现就是收不到回包，别急着怀疑代码。

抓包确认是必要的一步。在摆渡端口上抓一下，看报文头是不是符合预期：

sudo tcpdump -i lo -X port 9100 -c 10

你能看到 8 字节头后面跟载荷，会话 ID 在正向和反向报文里是一致的。如果反向报文的会话 ID 对不上，问题一定出在远端的会话查找逻辑，而不是网络。

3.4 模拟隔离环境：把摆渡通道换掉

本地回环跑通只证明逻辑对，真正要验证的是隔离装置场景。Demo 没法真的接一台隔离装置，但可以模拟它的约束：单向通道、带延迟、可能丢包。做法是把forwarder.py里的摆渡 socket 换成一个带方向限制的转发器，正向只允许近端发远端收，反向反过来，中间加 50ms 延迟模拟摆渡时延。

# bridge_sim.py - 模拟单向摆渡通道 import socket, threading, time def one_way(src_port, dst_port, delay=0.05): src = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) src.bind(("127.0.0.1", src_port)) dst = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) while True: data, _ = src.recvfrom(65535) time.sleep(delay) # 模拟摆渡延迟 dst.sendto(data, ("127.0.0.1", dst_port)) # 正向通道 9100 -> 9101，反向通道 9102 -> 9103 threading.Thread(target=one_way, args=(9100, 9101), daemon=True).start() threading.Thread(target=one_way, args=(9102, 9103), daemon=True).start() print("bridge simulator running")

然后把近端的发送目标改成 9100、接收监听 9103，远端的接收监听 9101、发送目标 9102。这样正向和反向走的是两条独立通道，跟正反向隔离装置的行为接近。跑一遍 TCP 和 UDP 用例，如果还能通，说明你的会话管理和报文格式扛得住单向链路。这个模拟器我建议你保留，后面改代码时拿它做回归，比每次上真实环境省事得多。

注意：模拟器里time.sleep是阻塞的，单线程处理单方向没问题，但如果你把延迟调大又并发多会话，会看到吞吐明显下降。这不是穿透逻辑的问题，是模拟器本身的瓶颈，别误判。

4. 核心代码拆解：近端接入与远端还原的关键几段

4.1 近端 TCP 接入：accept 之后做了什么

近端 TCP 监听是整个链路的入口，它的职责是接住客户端连接、分配会话 ID、把数据封包往摆渡通道送。下面这段是tcp_listener.py的主循环：

import socket, threading from common.protocol import pack_message, PROTO_TCP, DIR_FORWARD from common.session import SessionTable from common import config table = SessionTable(idle_timeout=config.IDLE_TIMEOUT) _next_id = 0 def handle_client(conn, addr): global _next_id _next_id += 1 sid = _next_id sess = table.create(sid, PROTO_TCP, addr) bridge = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) try: while True: data = conn.recv(config.BUFFER_SIZE) if not data: break sess.touch() pkt = pack_message(sid, PROTO_TCP, DIR_FORWARD, data) bridge.sendto(pkt, ("127.0.0.1", config.BRIDGE_PORT)) finally: conn.close() # 发送一个空载荷的结束报文，通知远端关闭 pkt = pack_message(sid, PROTO_TCP, DIR_FORWARD, b"") bridge.sendto(pkt, ("127.0.0.1", config.BRIDGE_PORT)) bridge.close() def start(): srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) srv.bind(("0.0.0.0", config.NEAR_TCP_PORT)) srv.listen(64) print(f"near tcp listener on {config.NEAR_TCP_PORT}") while True: conn, addr = srv.accept() threading.Thread(target=handle_client, args=(conn, addr), daemon=True).start()

逐段看。handle_client里每个客户端连接起一个线程，分配自增sid，建会话，然后循环recv。收到数据就pack_message封包，通过 UDP 摆渡通道发出去。这里用 UDP 做摆渡是 Demo 的简化，真实隔离装置可能是文件、消息队列或者专有协议，但接口形态一样：给一段字节，它负责送到对面。finally里发一个空载荷报文作为结束信号，这个约定很重要——远端收到空载荷就知道该关对应的服务端连接了，否则服务端会一直挂着等数据。

_next_id用全局变量加自增，单进程没问题，多进程部署时会冲突。二次开发时我一般会换成uuid4().int & 0xFFFFFFFF或者带机器标识的拼接，保证全局唯一。BUFFER_SIZE设 4096 是个折中，太小系统调用多，太大单报文可能超过摆渡通道的 MTU 限制，按你的通道特性调。

4.2 远端还原：收到报文后怎么找到正确的连接

远端代理干的是反向的事：从摆渡通道收报文，按会话 ID 找到对应的服务端连接，把载荷写进去。receiver.py的核心逻辑：

import socket, threading from common.protocol import unpack_message, PROTO_TCP, PROTO_UDP, DIR_FORWARD from common.session import SessionTable from common import config table = SessionTable(idle_timeout=config.IDLE_TIMEOUT) def ensure_remote(sess): """TCP 会话首次收到数据时，向真实服务端建连""" if sess.remote_sock is None: s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.connect((config.FAR_TCP_TARGET_HOST, config.FAR_TCP_TARGET_PORT)) sess.remote_sock = s # 起一个线程把服务端回包送回摆渡通道 threading.Thread(target=pump_back, args=(sess,), daemon=True).start() return sess.remote_sock def pump_back(sess): bridge = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) try: while True: data = sess.remote_sock.recv(config.BUFFER_SIZE) if not data: break pkt = pack_message(sess.session_id, PROTO_TCP, 1, data) # 1 = 反向 bridge.sendto(pkt, ("127.0.0.1", config.BRIDGE_BACK_PORT)) finally: bridge.close() def on_bridge_packet(raw): parsed = unpack_message(raw) if parsed is None: return sid, proto, direction, payload = parsed if direction != DIR_FORWARD: return sess = table.get(sid) if sess is None: sess = table.create(sid, proto, None) if proto == PROTO_TCP: if payload == b"": if sess.remote_sock: sess.remote_sock.close() return sock = ensure_remote(sess) sock.sendall(payload) elif proto == PROTO_UDP: udp = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) udp.sendto(payload, (config.FAR_UDP_TARGET_HOST, config.FAR_UDP_TARGET_PORT)) udp.close()

关键点在ensure_remote：远端不是一上来就连服务端，而是等第一个带载荷的报文到了才建连。这叫懒连接，好处是避免近端接了连接但客户端一直不发数据时，远端白白占一个服务端连接。pump_back是每个 TCP 会话一个线程，专门把服务端回包封成反向报文送回摆渡通道。UDP 那边更简单，来一个报文发一次，不维护连接，但这也意味着 UDP 的回包路径需要服务端主动往远端的监听地址回，Demo 里靠udp_sender配合，实际部署时要注意 NAT 和端口映射。

on_bridge_packet里对空载荷的处理和近端呼应，收到就关服务端连接。这里有个隐患：如果空载荷报文丢了（UDP 摆渡可能丢），服务端连接会一直挂着直到超时。生产环境要么给摆渡通道加确认，要么靠sweep兜底。Demo 选了后者，简单但不够及时，你改的时候心里有数。

4.3 UDP 会话的特殊处理：无连接怎么维持状态

UDP 穿透比 TCP 多一层麻烦：没有连接建立和关闭的信号，会话全靠报文驱动。近端udp_listener.py的做法是收到报文先查表，查不到就建会话，然后把报文封包送走：

import socket from common.protocol import pack_message, PROTO_UDP, DIR_FORWARD from common.session import SessionTable from common import config table = SessionTable(idle_timeout=config.IDLE_TIMEOUT) _key_map = {} # (client_ip, client_port) -> session_id _next_id = 100000 def start(): global _next_id sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.bind(("0.0.0.0", config.NEAR_UDP_PORT)) bridge = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) print(f"near udp listener on {config.NEAR_UDP_PORT}") while True: data, addr = sock.recvfrom(65535) key = (addr[0], addr[1]) sid = _key_map.get(key) if sid is None: _next_id += 1 sid = _next_id _key_map[key] = sid table.create(sid, PROTO_UDP, addr) else: table.get(sid).touch() pkt = pack_message(sid, PROTO_UDP, DIR_FORWARD, data) bridge.sendto(pkt, ("127.0.0.1", config.BRIDGE_PORT))

_key_map是近端独有的，因为只有近端知道客户端地址，远端不需要。这个映射表要跟SessionTable同步清理，否则客户端换了端口重连，旧键会一直留着。Demo 里在sweep之后顺手清_key_map里对应已销毁会话的项，代码在run_demo.py的维护线程里。这个细节容易被忽略，跑久了内存慢慢涨，排查起来很烦。

UDP 回包路径是这样的：服务端往远端代理的 UDP 监听地址回，远端封成反向报文送摆渡通道，近端收到后按会话 ID 查_key_map反查客户端地址，再sendto回去。反查这一步 Demo 里用会话对象里存的peer_addr，所以Session的peer_addr对 UDP 会话必须填，不能像 TCP 那样懒填。

5. 避坑与排查：穿透链路上最容易翻车的几个点

5.1 现象：TCP 能建连但数据发不出去

原因通常是摆渡通道的 MTU 或报文长度限制。Demo 头里载荷长度是 2 字节，单报文最大 64KB，但如果你把BUFFER_SIZE调大又碰上摆渡通道实际只支持 1500 字节左右，sendto会报Message too long或者静默截断。解决方法是把BUFFER_SIZE压到通道 MTU 以下，或者在pack_message之前做分片，每片单独封包，接收端按会话 ID 拼回去。分片逻辑 Demo 没内置，因为不同通道差异太大，但接口留好了，你在forwarder.py里加一层切分就行。

5.2 现象：UDP 第一个包通，后续包收不到回包

八成是会话键对不上。客户端如果每次sendto换源端口，近端会建多个会话，远端回包时按会话 ID 找客户端地址，找到的是旧端口，自然收不到。排查方法是抓近端入口的包，看源端口是否稳定；不稳定就要求客户端复用 socket，或者在近端做“同 IP 短时间内的新端口复用最近会话”的宽松匹配。后者有串话风险，只适合单客户端场景，别在多人环境用。

5.3 现象：会话表越跑越大，最后内存吃满

SessionTable.sweep没被调用，或者idle_timeout设得太大。Demo 默认 120 秒，后台线程 30 秒扫一次，正常不会涨。但如果你把run_demo.py里的维护线程注释掉做调试，忘了恢复，跑一晚上就能看出来。另一个隐蔽原因是 TCP 会话的remote_sock关了但会话没删，sweep里只按时间清，不检查 socket 状态。改的时候在sweep里加一句if sess.remote_sock and sess.remote_sock.fileno() == -1提前清理，能少踩一次。

5.4 现象：模拟单向通道后，反向数据全丢

检查反向通道的端口配置是否和pump_back里的BRIDGE_BACK_PORT一致。Demo 里正向和反向用了不同端口，改配置时容易只改一边。另一个常见原因是反向报文的direction字段写错，on_bridge_packet里if direction != DIR_FORWARD: return会把反向报文直接丢掉，表现就是正向通、反向死。抓包看方向字节，正向是 0，反向是 1，对不上就是封包时传错了常量。

5.5 现象：本地全通，上真实隔离装置后连不上

先确认隔离装置允许的摆渡方式到底是什么。Demo 用 UDP socket 只是模拟，真实环境可能是文件摆渡、单向光闸或者专有 API。这时候要改的是forwarder.py和receiver.py里的发送接收实现，pack_message/unpack_message和会话管理不用动。改完先用bridge_sim.py回归，再上真实环境。还有一个容易忽略的点：隔离装置两侧的时钟可能不同步，如果你的会话超时依赖绝对时间戳，会误判，Demo 用的是time.time()相对差值，问题不大，但换成datetime做序列化时要注意。

6. 二次开发：把 Demo 改成能落地的穿透组件

拿到这个 Demo 之后，大多数人不会直接拿去用，而是把它当成骨架，换成自己环境里的摆渡方式。我一般按这个顺序改：先定摆渡接口，再换传输实现，最后补可靠性。摆渡接口其实就两个函数——发和收，Demo 里散在forwarder.py和receiver.py，二次开发时第一步是把它们抽成一个Bridge抽象类，正向和反向各一个实例，这样换通道只改一个文件。

class Bridge: def send(self, raw: bytes): ... def recv(self) -> bytes: ... class UdpBridge(Bridge): def __init__(self, local_port, remote_port): self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) self.sock.bind(("0.0.0.0", local_port)) self.remote = ("127.0.0.1", remote_port) def send(self, raw): self.sock.sendto(raw, self.remote) def recv(self): data, _ = self.sock.recvfrom(65535) return data

抽完之后，文件摆渡就写个FileBridge，send往目录里写带序号的文件，recv轮询目录读文件；专有 API 就包一层。会话管理和协议解析完全不用动，这是分层带来的好处。验证方法也简单：把bridge_sim.py换成你的Bridge实现，跑一遍 TCP 和 UDP 用例，通了再谈性能。

可靠性这块，Demo 是“尽力而为”，真实环境要补三样东西。一是摆渡报文的确认与重传，UDP 摆渡会丢包，丢了就是数据缺口，TCP 流上表现为卡住。简单做法是给每个报文加序号，接收端发现跳号就请求重发，但单向通道没法请求，只能靠发送端定时重发最近 N 个报文，接收端去重。二是会话的持久化，代理重启后会话表清空，正在传输的连接全断，如果业务不能接受，得把会话状态落盘，重启后按会话 ID 恢复。三是背压，近端收得快、摆渡通道慢的时候，缓冲区会涨，Demo 没做流控，长时间大流量会 OOM，加一个固定大小的队列，满了就暂停recv，让 TCP 自身的窗口机制把压力传回客户端。

有个技巧值得单独说：调试穿透问题时，别一上来就抓全链路，先在摆渡通道两侧各打一条日志，记录会话 ID、方向、载荷长度，跑一个用例对比两侧日志。正向报文在近端发出的 ID 和远端收到的 ID 必须一一对应，对不上就定位到是封包还是解包的问题；对得上但数据不对，就是载荷处理的问题。这个习惯帮我省过很多时间，比盲抓包高效得多。

最后说个我自己的教训。早期改这类穿透代码，我总想着先把功能跑通再补超时和清理，结果有次测试环境跑了两天，会话表涨到几十万条，内存直接打满，服务假死。从那以后我每次动会话相关的代码，都强制先把sweep和超时路径跑一遍，哪怕功能还没完全通。这个 Demo 里sweep是现成的，你改的时候别把它弄丢了。希望帮到你。

本文还有配套的精品资源，点击获取