简介：在电力、工控等安全隔离场景中，正反向隔离装置常用于单向数据传递，但TCP/UDP业务常因隔离策略受阻。这套示例针对该痛点，给出了基于TCP与UDP的穿透实现，重点演示了增加UDP穿透后的代理配置与测试方法，适合从事网络安全、工业控制系统集成或隔离装置二次开发的工程师参考，也适合作为协议穿透原理学习的入门范例。资源共11个文件，压缩包约1.68MB，包括两份C++源码、2份INI配置、可运行的代理程序与TCP/UDP测试程序、一份文本说明以及隔离服务整体方案PPT，覆盖从配置、编译到联调的关键要素，目录结构清晰便于按模块查看。已有1250人学习下载，可用于快速搭建隔离装置穿透验证环境，并结合PPT中整体方案理解正反向隔离机制，为实际项目中的协议适配提供可复用的代码参考。
1. 正反向隔离装置TCP/UDP穿透Demo：一个能跑的隔离网模拟器
开发电力监控这类内外网数据交换功能时，最耗时间的往往不是业务逻辑，而是你写的代码要穿过正反向隔离装置，手边却只有设备的协议文档。这个Demo用Python在本地模拟了一套隔离装置双通道，把TCP和UDP业务数据封装成自定义帧，通过正向、反向两条单向隧道穿透到对端，让你在没有真实装置的情况下先把链路调通、把协议验证掉。它针对的是网络安全分区里的常规数据交换场景，不是绕过边界，而是把合法业务按隔离装置允许的方式送过去。适合正在做隔离装置对接的开发者，也适合拿隔离网通信做毕业设计的人：整个工程没有第三方依赖，改一个ini文件就能跑起来。
2. 架构设计：先理解正反向双通道模型，再动手搭目录
正反向隔离装置在TCP/IP网络里是个特殊存在：它在安全区之间做物理隔离，让业务数据只能单向流动。如果你按常规Socket的思路，建一条TCP连接把请求和响应都塞进去，到装置这层会被直接丢掉。所以动手写代码之前，先把通道模型画清楚。
2.1 装置的双通道模型
隔离装置对外呈现的并不是一个透明网桥，而是两个方向相反的通道。正向通道从内网侧发起，把数据摆渡到外网侧；反向通道反过来，只允许外网侧把数据摆渡回内网。真实装置通常要求应用自己封帧，把数据放到约定位置，装置侧定时扫描搬运。业务要做到双向通信，就必须把原来的双向交互拆成两条独立的单工链路：请求走正向，响应走反向；具体方向取决于谁在内网、谁在外网。
这个Demo把真实装置的“摆渡定时器”简化成了两条常驻TCP隧道。每条隧道只允许数据从一端流向另一端，接收端只负责从隧道里读帧，绝不反向写数据。这样一简化，你就能在本地用普通Socket精确复现隔离装置的半双工行为。隧道之间不共享状态，唯一的关联是穿透协议帧里的序列号。需要双向应答时，业务层必须自己维护请求ID，并在另一条通道里找到对应的响应——这也是真实装置对接项目里最常见的工作习惯。
2.2 Demo的组件划分与目录结构
项目按“配置—协议—通道—入口”四个层次拆分，启动入口和业务入口分开，方便不同角色的人快速定位文件。目录结构如下。
isolation-through-demo/ ├── config/ │ └── demo.ini # 隧道地址、监听端口、目标服务配置 ├── src/ │ ├── protocol.py # 穿透协议帧编码与解码 │ ├── common.py # 日志、配置读取、Socket接收缓冲区 │ ├── forward_tcp.py # 正向通道TCP转发 │ ├── reverse_tcp.py # 反向通道TCP转发 │ ├── udp_through.py # UDP会话转发 │ └── tunnel_endpoint.py # 建立隧道Socket，接收并处理对端帧 ├── scripts/ │ ├── start_demo.sh # 一键启动正向、反向、UDP三组进程 │ └── test_traffic.py # 产生测试流量并打印帧序号 └── README.md每个文件只做一件事。protocol.py不依赖网络，可以单独做单元测试；common.py里的ReceiveBuffer专门处理TCP半包问题，算是这个Demo里最值得复用的工具类；forward_tcp.py和reverse_tcp.py分别实现两个方向的TCP业务转发，代码结构几乎对称，便于对照阅读；udp_through.py因为UDP无连接的特性，额外维护一张会话表；tunnel_endpoint.py是隧道对端，负责从隧道里读帧、按帧类型分发给目标连接。
业务数据流按这样走：业务端连接到本机监听端口，forward_tcp.py接收数据后用protocol.py组成帧，把帧写入正向隧道Socket；对端的tunnel_endpoint.py从隧道读出帧，解出payload后连接到配置的target_host并写进去。UDP链路略有不同：udp_through.py先记录客户端源地址与session_id的映射，再把UDP报文封装成帧穿过隧道；对端收到后查reply_map，决定回包时把数据写回哪条隧道。从业务端和目标端看，两边都以为自己在和一个普通Socket通信，隔离装置被模拟成了半透明的数据管道。
这样的架构还有一个额外好处：只要不修改业务协议，隧道可以从TCP换成Unix Socket，或者把单条隧道改成多条并联。因为业务和通道之间隔着一层帧协议，通道的物理实现被隔离在业务代码之外。
真实装置与Demo模块的对应关系，可以看下面这张表，方便你拿真实项目来对齐。
| 真实装置环节 | Demo对应实现 |
|---|---|
| 正向通道摆渡 | forward_tcp.py + 正向隧道Socket |
| 反向通道摆渡 | reverse_tcp.py + 反向隧道Socket |
| 报文拆分/重组 | protocol.py 帧编解码 |
| 应用层协议白名单 | config中的allowed_request预留字段 |
为什么不直接用UDP over UDP来做UDP穿透？隔离装置对UDP的处理通常比较保守，UDP本身丢包后没有重传和拥塞控制，直接穿过去会出现大量乱序。常见做法是把UDP载荷放进TCP隧道里，牺牲一点时延换可靠性。这个Demo也沿用了这个思路，UDP报文只是作为Frame的payload，依靠TCP隧道保证顺序。
3. 核心模块拆解：从Socket到穿透帧的完整链路
看代码之前先明确一点：穿透不是把Socket原样搬过去，而是把数据流切成一帧一帧，再通过隧道传输。帧协议是整个Demo的基石，后面所有转发逻辑都在围绕它转。
3.1 协议帧：一个可扩展的报文头
协议帧用了固定8字节头部加变长载荷的结构，帧类型和序列号都放在头部，方便接收方在不解析完整数据的情况下快速判断。代码如下。
import struct PROTO_MAGIC = b'IOTP' FRAME_HEADER = struct.Struct('>4sBBH') # > 表示大端；4s 是4字节magic；B 是帧类型；B 是序列号；H 是payload长度 class Frame: def __init__(self, ftype: int, seq: int, payload: bytes = b''): self.ftype = ftype self.seq = seq self.payload = payload def encode(self) -> bytes: return FRAME_HEADER.pack(PROTO_MAGIC, self.ftype, self.seq, len(self.payload)) + self.payload @staticmethod def decode(data: bytes): if len(data) < FRAME_HEADER.size: return None, data magic, ftype, seq, length = FRAME_HEADER.unpack_from(data, 0) if magic != PROTO_MAGIC: raise ValueError(f'magic mismatch: {magic}') total = FRAME_HEADER.size + length if len(data) < total: return None, data payload = data[FRAME_HEADER.size:total] return Frame(ftype, seq, payload), data[total:]这里的ftype定义了四类帧：1表示业务数据，2表示隧道心跳，3表示连接结束，4表示UDP会话首包。seq是8位序列号，用来在接收端做乱序统计，也可以用于日志里追踪一帧数据的完整路径。decode返回的第二个值是多出来的字节，用于解决TCP粘包问题——一次recv可能包含多个帧，也可能只包含一个帧的前半部分。
3.2 TCP穿透：把字节流切片成帧再塞进单向隧道
TCP穿透的核心逻辑在forward_tcp.py里：业务端发来的是连续字节流，转发进程把它切成多个Frame，逐帧写入隧道。反向过程由对端tunnel_endpoint.py完成，把帧还原成字节流写进目标服务。
import socket import threading from protocol import Frame from common import ReceiveBuffer def tcp_forward(local_listen, target_host, target_port, tunnel_sock): server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server.bind(local_listen) server.listen(5) def forward_loop(client_sock): remote_sock = socket.create_connection((target_host, target_port), timeout=5) buf = ReceiveBuffer() seq = 0 while True: chunk = client_sock.recv(4096) if not chunk: tunnel_sock.sendall(Frame(3, seq, b'').encode()) break buf.push(chunk) for payload in buf.pop_frames(): frame = Frame(1, seq, payload) tunnel_sock.sendall(frame.encode()) seq = (seq + 1) & 0xFF remote_sock.close() client_sock.close() while True: client_sock, addr = server.accept() threading.Thread(target=forward_loop, args=(client_sock,), daemon=True).start()这里的local_listen是一个元组，比如("127.0.0.1", 9001)，表示本机监听端口。target_host和target_port是隔离装置对外的业务服务器地址。remote_sock在第一次有客户端连接时就建立，这里故意简化成固定连接；如果目标服务要求每请求一个连接，可以改成在收到第一帧后再创建远端连接。
参数上需要注意seq是8位，到255后会回绕，这对测试流量足够了，但生产环境建议换成16位。ReceiveBuffer.pop_frames()返回的是已拼完整的payload列表，不包含头部，这样业务数据里哪怕有IOTP这四个字节也不会误判。
3.3 UDP穿透：用会话表解决无连接的回包问题
UDP没有连接概念，收到一个包之后，对端回包时不知道应该发给谁。Demo的UDP穿透用session_id解决：第一次收到某个客户端的包时分配一个ID，之后所有来自该客户端的包都带着这个ID穿过隧道；对端回包时也带上同一个ID，这样客户端侧就能把回包映射回正确的本地Socket。
import socket from protocol import Frame session_map = {} def udp_forward(local_listen, target_host, target_port, tunnel_sock): sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.bind(local_listen) next_sid = 1 while True: data, addr = sock.recvfrom(65507) sid = session_map.get(addr) if sid is None: sid = next_sid session_map[addr] = sid next_sid += 1 tunnel_sock.sendall(Frame(4, sid, data).encode())对端解出UDP帧后，需要把回包按会话ID写回隧道，代码如下。
reply_map = {} def udp_receiver(tunnel_sock, target_host, target_port): sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) while True: data = tunnel_sock.recv(4096) frame, _ = Frame.decode(data) if frame is None or frame.ftype != 4: continue sid = frame.seq sock.sendto(frame.payload, (target_host, target_port)) reply_data, _ = sock.recvfrom(65507) tunnel_sock.sendall(Frame(4, sid, reply_data).encode())这段代码把回包处理简化成了阻塞等待，同一个会话只能有一个包在途。实际项目中，这里应该用异步循环，或者干脆在udp_receiver里每收到一个UDP包就新起一个线程去等回包。不过作为Demo，当前实现足以把UDP穿透的思路讲清楚：会话ID是整个映射的关键，它不关心IP地址和端口，只关心逻辑上的“这个客户端的流”。
4. 运行与验证：用一条nc命令确认数据穿过了模拟装置
Demo的定位是“能跑、能验证、能改”，所以运行部分也被压缩到只有三步：改配置、起进程、发流量。不要跳过配置直接起进程，因为转发目标指向哪里都写在demo.ini里。
4.1 环境与配置说明
运行环境要求Python 3.8及以上，不需要pip install任何第三方库。配置文件放在config/demo.ini，示例内容如下。
[channel] # 正向隧道和内网转发进程之间的通信地址 forward_tunnel_host = 127.0.0.1 forward_tunnel_port = 19001 # 反向隧道地址 reverse_tunnel_host = 127.0.0.1 reverse_tunnel_port = 19002 [forward] # 业务端连接到这里，模拟内网侧入口 local_listen = 127.0.0.1:9001 # 数据穿过正向隧道后，会被转发到这个目标服务 target_host = 127.0.0.1 target_port = 8080 [reverse] # 反向通道入口 local_listen = 127.0.0.1:9002 target_host = 127.0.0.1 target_port = 3306 [udp] # UDP业务入口 local_listen = 127.0.0.1:9003 target_host = 127.0.0.1 target_port = 514注意[forward]和[reverse]里的local_listen是业务端访问的端口，而[channel]里的forward_tunnel_port是转发出进程和隧道对端进程之间的内部通信端口，两者不要混填。target_host和target_port写的是你本机已经存在的服务，比如8080上的HTTP服务、514上的Syslog服务，如果本地没有对应服务，验证时会看到连接失败日志。
4.2 启动Demo的四种进程
打开两个终端，第一个终端先启动两条隧道对端，第二个终端启动三个转发进程。隧道对端是常驻进程，先起它可以避免后面转发进程一启动就报连接拒绝。
python3 src/tunnel_endpoint.py --channel forward --listen 127.0.0.1:19001 & python3 src/tunnel_endpoint.py --channel reverse --listen 127.0.0.1:19002 & python3 src/forward_tcp.py > logs/forward_tcp.log 2>&1 & python3 src/reverse_tcp.py > logs/reverse_tcp.log 2>&1 & python3 src/udp_through.py > logs/udp_through.log 2>&1 &第一条命令里的--channel forward告诉进程这是正向隧道的对端，它只负责从19001端口读帧并转发到[forward]节里配的target。--listen参数指定了隧道对端监听在哪个地址，这个地址和demo.ini里的forward_tunnel_port必须一致，否则forward_tcp.py创建的Socket找不到人接收。
三条转发进程都依赖config/demo.ini里的配置，所以启动前改动ini文件后，需要kill掉进程再重启才生效。
4.3 用nc命令验证TCP穿透成功
最简单的方法是把业务数据发到正向监听端口9001，再去看目标服务是否收到。假设8080端口有一个HTTP服务，执行下面的命令。
echo -e "GET / HTTP/1.1\r\nHost: test.local\r\n\r\n" | nc 127.0.0.1 9001正常情况下，forward_tcp日志里会看到：
[forward_tcp] accept from ('127.0.0.1', 51234) [forward_tcp] send frame seq=1 ftype=1 payload=17 bytes [tunnel_endpoint] recv frame from forward tunnel, seq=1, ftype=1 [tunnel_endpoint] connect target 127.0.0.1:8080 success如果最后一行没出现，优先检查target_host和target_port是否指向了一个真实监听的服务。可以用nc -l 8080临时起一个裸TCP服务，然后在9001端口发数据，看看8080这边能不能收到同样的payload。这个验证方式能帮你把问题定位在“Demo转发链路”还是“目标服务”。
UDP验证可以用自带脚本，它会发一条UDP报文并等待回包。
python3 scripts/test_traffic.py --type udp --data "hello"预期输出是received hello from target。如果只看到发送成功没有回包，基本就是5.2节说的回包映射问题。
5. 避坑指南：本机模拟隔离装置最容易踩的四个坑
本地模拟和真实装置相比，最大的区别是没有那些物理层面的限制，但这反而会掩盖协议层的问题。下面四条是我在这个Demo上反复踩过的坑，每条都按现象、原因、解决的顺序写清楚。
5.1 现象：转发过去的数据少了一段，而且毫无规律
本地起一个HTTP服务做测试，发现有些请求体到目标端只有一半，有时候整包丢了。一开始以为是网络玄学，后来才发现是TCP半包问题。
原因在于TCP是流式协议，业务端一次send(4096)到转发进程后，recv可能只返回了200字节，而转发进程直接把200字节封装成一帧，导致对端拿到的payload不完整。解决方法是加一个ReceiveBuffer，把所有recv到的字节先缓存起来，等攒够一个完整帧再交给业务逻辑。核心实现如下。
class ReceiveBuffer: def __init__(self): self._buf = b'' def push(self, data: bytes): self._buf += data def read_frame(self): if len(self._buf) < FRAME_HEADER.size: return None magic, ftype, seq, length = FRAME_HEADER.unpack_from(self._buf, 0) if magic != PROTO_MAGIC: raise ValueError(f'bad magic: {magic}') total = FRAME_HEADER.size + length if len(self._buf) < total: return None payload = self._buf[FRAME_HEADER.size:total] self._buf = self._buf[total:] return Frame(ftype, seq, payload)注意read_frame是返回一帧，而前面tcp_forward循环里的pop_frames实际上会循环调用read_frame直到返回None。这样无论隧道对端怎么分包、粘包，都能从字节流里切出完整帧。
5.2 现象：UDP包能发过去，但回包永远收不到
用test_traffic.py发UDP数据，目标端确实收到了，但客户端一直等不到回复。原因在udp_receiver里，回包时只执行了tunnel_sock.sendall(Frame(4, sid, reply_data).encode())，却没有把这个回包关联到最初的客户端地址。
解决思路是建立一张reply_map，key是目标服务地址，value是session_id，这样从目标服务收到回包时，就知道该把回包写回哪个会话。实际代码里，session_id在udp_forward里已经维护了客户端地址到ID的映射，udp_receiver应该把目标地址到ID的反向映射也补上。
5.3 现象：正向、反向进程都活着，数据却一动不动
这种问题最容易出现在模拟双向请求时：内网业务端通过正向通道发请求，期望响应从同一条连接回来。但正向通道本质上只允许内网到外网单向流动，外网侧发回来的响应并不在这条隧道里，于是客户端傻等，服务端也在等。
原因是没有把业务协议拆成两条单工链路。解决方法是应用层自己加请求ID，正向通道传输请求，反向通道传输响应；或者让外网侧主动把结果放进反向通道，内网侧用轮询去拉。这个约束不是Demo才有，真实隔离装置比这更严格，所以写业务代码时一开始就要按这个模型设计。
5.4 现象：本地Demo完全正常，接上真实隔离装置后偶发丢包
把同一套转发代码换到真实装置后，偶尔出现某条数据没过去，重发就正常。出现这种“时好时坏”的情况，第一反应不是查网络，而是查真实装置的报文白名单。
原因在于真实装置通常有最大报文长度限制、源IP/MAC绑定、协议类型白名单，而Demo默认放行一切数据。解决方法是给Demo也加一层白名单过滤，在转发前检查payload长度和起始特征。下面是一段示例过滤函数，可以按需扩展成从config里读规则。
def policy_filter(data: bytes) -> bool: if len(data) > 1024: return False return data.startswith(b'GET /')把这个函数插进forward_loop里，不满足条件就直接丢弃并打印warning。这样在本地就能把真实装置的限制模拟出一部分，不至于到现场才翻车。
6. 二次开发：把Demo协议改成自己的业务转发规则
这个Demo本身不绑定具体业务协议，所以二次开发时最值得改的是两个地方：帧类型和过滤策略。如果你想限制只允许某类请求穿过隔离装置，可以保留现有Frame格式，但在转发前加一个规则层。
下面这段代码演示如何新增一个“只放行HTTP GET”的规则，并把规则函数接入forward_loop。
def policy_filter(data: bytes) -> bool: if not data: return False return data.startswith(b'GET /')在forward_loop里封装成帧之前调用：
if not policy_filter(chunk): print(f'[filter] drop {len(chunk)} bytes') break验证方法很简单：启动Demo后分别用nc发送GET和POST请求，GET应该正常穿透，POST会在日志里看到drop记录。这个改造思路可以扩展到消息队列、Modbus、Syslog等任何协议，只需要把policy_filter换成对应的特征判断，比如Syslog可以判断是否以<加数字开头，Modbus可以判断事务标识符后跟协议标识符。
如果你需要支持多条连接复用，而不是一个转发进程只服务一个目标端口，可以考虑把target_host和target_port也做成Frame的一部分，在帧头后面追加一个4字节的目标端口字段。这样对端不需要在配置里写死目标，而是从每一帧里动态取目标地址。改动点集中在protocol.py的编码和解码，forward_tcp.py只需要在组帧时把目标信息填进去。
从那以后我每次改动这个Demo，都会先把通道模型在纸上画一遍，确认“这帧数据到底走正向还是反向”，画清楚再动手改协议，半包问题也就没再犯过。希望帮到你。
本文还有配套的精品资源，点击获取
