简介在电力、工控等安全隔离场景中正反向隔离装置常用于单向数据传递但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 bIOTP FRAME_HEADER struct.Struct(4sBBH) # 表示大端4s 是4字节magicB 是帧类型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(fmagic 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), timeout5) 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(targetforward_loop, args(client_sock,), daemonTrue).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 21 python3 src/reverse_tcp.py logs/reverse_tcp.log 21 python3 src/udp_through.py logs/udp_through.log 21 第一条命令里的--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 seq1 ftype1 payload17 bytes [tunnel_endpoint] recv frame from forward tunnel, seq1, ftype1 [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(fbad 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_mapkey是目标服务地址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(bGET /)把这个函数插进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(bGET /)在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都会先把通道模型在纸上画一遍确认“这帧数据到底走正向还是反向”画清楚再动手改协议半包问题也就没再犯过。希望帮到你。本文还有配套的精品资源点击获取