资讯动态

Python实战:从零构建网络协议编辑器与封包修改工具

发布时间:2026/10/23 9:34:59 来源:尧图企业网站定制
简介这是一份面向计算机网络课程设计的Python实战资源实现了网络协议编辑器与网络协议分析器两个工具适合需要完成类似选题、或想深入理解网络协议报文封装/解析过程的学生。资源共4个文件核心包括packet_sender.py负责报文构造与发送和packet_analysisor.py负责流量捕获、协议解码与统计两个Python脚本另附README说明文档与.gitignore压缩包仅13KB结构精简便于直接阅读和复用。目前已有433人学习下载。项目通过socket库创建连接并收发数据利用struct库对二进制报文进行打包与解包编辑器中可自定义源/目的IP、端口号等字段并预览数据包分析器则真实抓取网络流量按TCP、UDP、IP等协议层次解析字段并展示基本的协议分布统计。代码同时体现了模块化设计、异常处理与注释规范读者既能掌握Python网络编程的核心方法也能基于此扩展自定义协议、图形界面或更丰富的过滤规则从而高效完成课程设计、实验报告或进一步的项目开发。1. 项目定位为什么课程设计里要自己写一个协议编辑器说到网络协议编辑器很多人第一时间想到的是Wireshark自带的修改功能但在课程设计里很多时候要求你用Python独立实现一套能解析、能编辑、能重发的协议工具箱。这个项目就是这么个东西它不是一个只能读pcap文件的静态分析器而是把协议解析、字段编辑、封包重发合在一起的命令行工具。你可以在终端里打开一个抓包文件像展开JSON一样逐层看以太网帧、IP头、TCP头的每个字段然后直接改目标端口、改序列号、改负载内容写完重新算checksum再发给真实服务器。适合的人群很明确网络工程或计算机专业的学生做课程设计以及需要在本地快速验证协议逻辑的测试工程师。它把Wireshark里“看得见摸不着”的二进制流转成了Python对象你既能看清每个字节的含义也能用几行代码把修改后的包重新发出去。比起直接用scapy这个项目胜在“编辑器”的体验——它有一套独立的字段模型修改后按模型重新构建二进制更适合理解协议字段之间的依赖关系。整个项目是纯Python写的不依赖Windows特有API跨平台都能跑。2. 架构设计从pcap文件到控制台的管道式数据流2.1 目录结构与模块划分这个项目没有搞成单文件脚本而是按职责拆出了几个独立的模块。整体结构大致是这样的network_protocol_tools/ ├── cli.py # 入口解析命令行参数调度各模块 ├── analyzer/ │ ├── __init__.py │ ├── pcap_reader.py # 读取pcap文件处理字节流和分片 │ ├── ethernet_frame.py # 以太网帧解析 │ ├── ip_packet.py # IPv4头解析与组装 │ ├── tcp_segment.py # TCP段解析 │ ├── udp_datagram.py # UDP解析 │ └── payload_handler.py # 应用层负载识别HTTP/DNS等 ├── editor/ │ ├── __init__.py │ ├── field_model.py # 协议字段的树状数据模型 │ ├── packet_builder.py # 从模型重新序列化为二进制 │ └── checksum.py # 校验和计算 ├── sender/ │ ├── __init__.py │ ├── raw_sender.py # 基于原始套接字发送 │ └── pcap_writer.py # 导出修改后的包为pcap文件 ├── tests/ │ ├── test_checksum.py │ └── test_fields.py └── README.md这个目录设计遵循了一个简单的原则解析和编辑是两条独立的链路。analyzer负责把二进制变成字段editor负责把字段变成二进制sender负责把二进制发出去。中间衔接的字段模型是整个项目的核心它既承担解析结果的存储也是编辑器的修改目标。这样拆分的好处是你可以单独测试协议解析逻辑也可以单独测试字段修改后的重新组装不会互相污染。2.2 数据流动pcap文件到网络发送的完整路径一条数据从pcap文件到最终发送会经过四个阶段。第一阶段是pcap_reader读取文件按pcap全局头里的链路类型字段判断是以太网还是其他数据链路然后逐包读取成原始字节。第二阶段是解析器按协议栈逐层剥壳从以太网帧到IP包再到TCP/UDP段最后把负载交给payload_handler尝试识别HTTP、DNS等常见协议。第三阶段是field_model把解析出来的字段组织成嵌套字典结构用户可以通过指定路径访问某个字段比如修改ip_packet.ttl或tcp_segment.dport。第四阶段是packet_builder做反向操作把修改过的字段重新压成二进制并对涉及校验和的字段重新计算。# cli.py 中的核心调度逻辑简化 def process_pcap(input_file): packets pcap_reader.read_all(input_file) parsed_packets [] for raw_packet in packets: frame ethernet_frame.parse(raw_packet) if frame.ethertype 0x0800: ip_pkt ip_packet.parse(frame.payload) if ip_pkt.protocol 6: tcp tcp_segment.parse(ip_pkt.payload) parsed_packets.append(ProtocolTree( layer_data[frame, ip_pkt, tcp], rawraw_packet )) return parsed_packets这个调度流程只处理了最常见的以太网IPv4TCP组合实际项目里会继续判断协议号如果遇到ICMP协议号1就走另一个分支。参数说明方面pcap_reader.read_all返回的是原始包字节列表每个元素是(timestamp, rawnetworkheader_bytes, packet_bytes)的命名元组。ethernet_frame.parse要求输入去掉pcap文件头的完整帧数据如果传入了带文件头的字节解析会在前六字节处出错这是新手容易踩的坑。3. 核心模块拆解从原始字节到树状字段模型3.1 以太网帧与IPv4头解析实现协议解析最基础的工作是把连续的二进制按字段宽度切开。以太网帧头固定14字节6字节目的MAC、6字节源MAC、2字节类型。这里有一个容易出错的地方MAC地址要用十六进制字符串显示而类型字段要用struct.unpack(!H)按大端序读取否则得到的数字会颠倒。# ethernet_frame.py import struct class EthernetFrame: __slots__ (dst_mac, src_mac, ethertype, payload) classmethod def parse(cls, data: bytes): dst_mac data[:6] src_mac data[6:12] ethertype struct.unpack(!H, data[12:14])[0] payload data[14:] return cls( dst_mac:.join(%02x % b for b in dst_mac), src_mac:.join(%02x % b for b in src_mac), ethertypeethertype, payloadpayload )这里__slots__能减少实例内存占用的不确定因素尤其当你一次解析数万包时能明显感觉到差别。struct.unpack(!H)中的!表示网络字节序大端因为IEEE 802.3规定的以太网头部字段都是大端传输。如果你用原生字节序解析在x86机器上得到的值会被翻转常见现象是0x0800IPv4变成了0x0008。IPv4头的解析比以太网帧复杂因为头部长度可变必须从IHL字段读取头长再根据total_length确定包体边界。# ip_packet.py import struct class IPv4Packet: def parse(self, data: bytes): # 固定20字节基础头 可选字段 version_ihl data[0] version version_ihl 4 ihl (version_ihl 0x0F) * 4 total_length struct.unpack(!H, data[2:4])[0] protocol data[9] src_ip socket.inet_ntoa(data[12:16]) dst_ip socket.inet_ntoa(data[16:20]) return { version: version, header_length: ihl, total_length: total_length, protocol: protocol, src_ip: src_ip, dst_ip: dst_ip, payload: data[ihl:total_length] }这个解析器的关键点是payload的切分边界。很多人会直接取data[ihl:]但这会带上IP层尾部可能的填充字节正确做法是用total_length来截断。如果抓包文件里有分片total_length还要和MF标志位配合判断。socket.inet_ntoa这个函数在Python 3里输入必须是4字节对象如果你传进来的是字符串它会直接抛TypeError而不是自动转换。3.2 TCP和UDP有状态解析与四层映射TCP段解析里最容易翻车的是头部长度字段。TCP头最短20字节但因为有可选字段头部长度由数据偏移字段4位表示这个值是4字节的倍数。正确解析后数据负载从偏移量开始。判断TCP是否携带有用负载不能只看序号还要结合PSH标志位。# tcp_segment.py def parse_tcp(data: bytes): if len(data) 20: raise ValueError(TCP segment too short) sport, dport struct.unpack(!HH, data[0:4]) seq struct.unpack(!I, data[4:8])[0] ack struct.unpack(!I, data[8:12])[0] offset_reserved data[12] data_offset (offset_reserved 4) * 4 flags data[12:14] psh_flag flags[1] 0x08 payload data[data_offset:] # 判断是否有负载 has_payload psh_flag or len(payload) 0 return { sport: sport, dport: dport, seq: seq, ack: ack, data_offset: data_offset, payload: payload, has_payload: has_payload }flags的解析需要按位与操作比如psh_flag flags[1] 0x08因为TCP标志位跨越两个字节URG/ACK/PSH在第二个字节的低4位。新手如果直接读整个两字节再做位运算容易算错位。UDP解析相对简单固定8字节头长度字段包含头数据算负载长度时要减8。但在实际项目里UDP更容易踩坑的是分片如果IP层说这是分片UDP负载可能不完整直接解析会得到乱码这时候需要分片重组模块来处理。协议编辑器最大的价值在于它不只是“读字段”而是让你能改字段后重新组装。所以每一项解析都要保留原始字节和偏移方便编辑器写回。我在设计时给每个字段都存了一个可选的byte_offset元信息这样packet_builder能精确覆盖那些未修改的字节段而不是整个重建——重建往往会让未设置的填充字段变成0比如TCP选项区可能改变对端解析行为。3.3 编辑器的数据模型与packet_builder回写编辑器端的核心是FieldModel类。这个类不关心包的具体语义只维护一棵字段树。每个字段节点包含字段名、值、起始位、位宽、读取器和写入器。修改值后通过写入器把值转成二进制。这比直接对字典赋值更加工程化因为协议字段经常以位为单位偏移而不是字节。# field_model.py class FieldNode: def __init__(self, name, bit_offset, bit_width, reader, writer): self.name name self.bit_offset bit_offset self.bit_width bit_width self.reader reader # 函数从bytearray读取值 self.writer writer # 函数把值写入bytearray self.children [] def read(self, data): return self.reader(data, self.bit_offset, self.bit_width) def write(self, data, value): self.writer(data, self.bit_offset, self.bit_width, value)一个典型的写入器如下先把所属字节段读出来用位掩码清空目标位域再把新值左移放到对应位置。这里有点容易被位运算绕晕但是一旦封装成FieldNode后面新增协议就只需要配置节点不用再关心具体操作。def write_bits(data: bytearray, bit_offset: int, bit_width: int, value: int): start_byte bit_offset // 8 end_byte (bit_offset bit_width 7) // 8 # 提取涉及的所有字节并拼接成整数 byte_slice data[start_byte:end_byte] seg_val int.from_bytes(byte_slice, big) shift 8 * (end_byte - start_byte) - (bit_offset % 8) - bit_width mask ((1 bit_width) - 1) shift seg_val (seg_val ~mask) | ((value shift) mask) data[start_byte:end_byte] seg_val.to_bytes(end_byte - start_byte, big)参数定义里shift是位域右边界到切片末端的距离这样无论位域跨越几个字节都能正确对齐。如果你修改的是跨字节字段比如IPv4中的16位总长度它通常正好在字节边界上但TCP的标志位却跨在字节13的末位和字节14的首位单纯用字节操作是完不成的这种位级操作就派上了用场。packet_builder的策略是先克隆原始字节然后把所有被用户修改过的FieldNode依次write回去。这就意味着未修改的选项字段、padding字段都会原样保留不会因为重建而丢失。最后一步是重新计算必要校验和。比如修改IPv4头的TTL后IP头校验和必须重算修改TCP负载后TCP校验和必须重算同时IP头总长度也要更新。这个依赖关系在项目里用了一个简单的dependency_graph字典来维护。4. 踩坑避坑笔记协议工具开发的血泪经验4.1 pcap文件读取时的字节序陷阱现象用open(pcap_path, rb).read()后按Pcap文件头解析版本号、时间戳得到的数据跟Wireshark显示完全对不上比如链路类型明明写着以太网程序却解析成别的值。原因pcap文件头里所有数值字段都是按特定字节序存储的但pcap的“魔法字节”决定了我们该用哪种字节序解析。很多抓包工具默认保存为小端序但Linux上某些工具会保存为大端序。直接按固定字节序解析必然在另一半文件上翻车。解决解析pcap头时先读前4字节magic data[0:4]。如果magic为d4 c3 b2 a1则小端序如果为a1 b2 c3 d4则大端序。之后的字段都用对应字节序去struct.unpack。我一般会把读到的字节序标记存到一个上下文对象里供后续所有包解析使用避免每个函数重复判断。def pcap_header(data: bytes): magic data[:4] endian if magic b\xd4\xc3\xb2\xa1 else if endian : magic magic[::-1] # 转换为网络序 version_major struct.unpack(endian H, data[4:6])[0] link_type struct.unpack(endian I, data[20:24])[0] return version_major, link_type这段代码用endian字符串直接传给struct.unpack让读取规则跟着文件走而不是跟着代码走。后来我把link_type也用查表法映射成“以太网/令牌环/虚拟以太网”总之别硬编码链路类型。4.2 分片重组解析HTTP负载时总差一段现象明明抓到了包含HTTP GET的TCP包但解析器拿到的负载只有前半截后半段在下一个包里。单独看每个包的payload字段都正常一旦拼接HTTP请求体却发现不完整。原因IP分片导致TCP头被拆到多个包中。比如一条1500字节的HTTP请求以太网最大传输单元是1500字节IP头占20字节TCP头通常占40字节那负载最多只有1440字节。剩余的负载必须放在第二个分片里。而解析器直接按total_length截断每个包忽略了分片标志和偏移。解决在IP层加一个分片重组模块。维护一个字典键是(src_ip, dst_ip, identification)值是收集到的分片列表。只有当所有分片的MF标志为0且偏移连续时才按偏移拼回完整的IP包再交给TCP解析。这个逻辑放在pcap_reader和ethernet_frame之间因为重组的是IP负载。def reassemble_ip(ip_parse_result, data): key (ip_parse_result[src_ip], ip_parse_result[dst_ip], ip_parse_result[identification]) if ip_parse_result[flags_mf] or ip_parse_result[fragment_offset]: fragment_buffer[key].append(ip_parse_result[payload]) return None else: # 当前分片是最后一个尝试拼接 if key in fragment_buffer: fragment_buffer[key].append(ip_parse_result[payload]) full_payload b.join(fragment_buffer.pop(key)) return full_payload return ip_parse_result[payload]这段代码只演示了基本思路实际还要考虑分片超时、重叠分片等攻击场景。对于课程设计而言能把正常抓到的分片拼起来就够用了。很多时候报“内容长度不足”其实是这里没做重组。4.3 TCP校验和计算的offset与伪头部长度现象修改了TCP负载后重新计算校验和发出去的包在Wireshark里显示校验和错误。但检查代码逻辑看起来是按RFC 793算的补码和为什么不对原因TCP校验和覆盖的范围不仅是TCP头负载还要包含12字节的伪头部pseudo header其中长度字段是TCP段的长度。这个长度不能直接用IP层的total_length减去IP头长因为IP层可能还有填充字节。常见错误是把伪头里的长度算成len(tcp_segment_payload) 20但payload里有填充时就会出错。解决计算校验和前先把TCP段切到精确长度。IP头里total_length已经标明TCP段的结束位置用tcp_end ip_total_length - ip_header_length再用tcp_end作为TCP数据的边界。伪头部按这个边界算长度。def tcp_checksum(src_ip, dst_ip, tcp_segment): # tcp_segment 是包含TCP头负载的精确字节 pseudo_header struct.pack(!4s4sBBH, socket.inet_aton(src_ip), socket.inet_aton(dst_ip), 0, 6, len(tcp_segment)) total pseudo_header tcp_segment if len(total) % 2 1: total b\x00 # 奇数长度时末尾补零 res 0 for i in range(0, len(total), 2): res (total[i] 8) total[i1] res (res 16) (res 0xFFFF) return (~res) 0xFFFF校验和的玄学之处在于如果你把伪头部长度算错本地计算显示是对的但接收方的校验和验证总是失败。后来我在校验和函数入口处加了一个断言len(tcp_segment) ip_total_length - ip_header_length一旦不满足就报错省去很多排查时间。4.4 动态端口与TCP状态跟踪现象解析FTP或HTTP的会话明明前后两个包目的端口分别是21和50000但想在同一窗口里把两个包关联成一次对话却找不到相同的五元组查询会话记录时总是中断。原因像FTP活动模式这样的协议会通过控制连接动态协商数据连接端口。要正确关联必须解析应用层数据从FTP响应里提取新的端口号更新会话状态表。许多协议解析器初期不做状态机只按TCP四元组匹配无法处理动态协商。解决在我的工具里加了一个SessionTracker它不按固定的元组索引而是按tracking_key函数动态生成。默认情况下tracking_key是(src_ip, sport, dst_ip, dport)但对FTP会额外解析PORT 192,168,1,1,50,10指令把新连接的五元组映射到现有会话上。class SessionTracker: def __init__(self): self.sessions {} def get(self, packet_tuple): key self.generate_key(packet_tuple) return self.sessions.get(key) def generate_key(self, packet_tuple): if packet_tuple[sport] 21 or packet_tuple[dport] 21: # 解析FTP命令获得数据端口 if bPORT in packet_tuple[payload]: port parse_ftp_port(packet_tuple[payload]) return (ftp-data, packet_tuple[src_ip], port) return tuple(packet_tuple[k] for k in (src_ip,sport,dst_ip,dport))这个模块虽然代码量不大但让“会话分析”功能提升了一个档次。很多专用流量分析工具的做法也是类似只是端口协商规则表更复杂。课程设计里能处理FTP主动模式这一项已经能超过班里大部分人了。4.5 编辑后的包在Wireshark里提示“previous segment lost”现象用编辑器把TCP包的SEQ号改小后发出去让对端接收对端总是回复一个ACK但抓包软件报“TCP Previous Segment Lost”的警告。原因TCP的SEQ号不仅代表当前包的起始序号还隐含了与前后包的连续性。如果只修改包内某个字段而没调整相邻包的SEQ或没更新ACK号发送方和接收方的窗口状态就无法对齐表现为丢包警告。这其实不是网络丢包而是TCP状态机认为你跳序了。解决编辑器在修改SEQ后一定要同时更新后续所有包的SEQ号偏移量。我在编辑器接口里加了adapter层当修改tcp.seq时会扫描当前会话中所有同方向的包把它们的seq和ack都加上相同偏移。这个适配逻辑不是自动的得用户确认因为有些场景就是希望故意制造乱序来测试接收方行为。默认情况下不自动调整但会弹一个提示确认是否同步更新相关会话包这能让用户避免因漏改导致的“看似改了一个字段实际整条链路都受影响”的情况。5. 二次开发扩展新协议与快速验证的工程习惯5.1 新增一个自定义协议解析器如果计划支持一个新的应用层协议比如MQTT你不需要改动底层模块。只需在analyzer/payload_handler.py里根据TCP或UDP的端口号分发到对应的解析函数。我的一般做法是先写一个纯函数输入是应用层负载字节输出是字段字典。比如MQTT固定头第一字节是报文类型后四位是标志剩余长度用变长编码。然后在payload_handler里生成一个PayloadInfo对象挂到tcp_segment.payload_info上。# payload_handler.py def parse_mqtt(payload: bytes): if len(payload) 2: return {protocol: mqtt, error: incomplete} header payload[0] msg_type header 4 flags header 0x0F # 剩余长度最多4字节变长编码 remaining_length 0 multiplier 1 offset 1 while True: digit payload[offset] remaining_length (digit 0x7F) * multiplier multiplier * 128 offset 1 if digit 0x80 0: break return { protocol: mqtt, msg_type: msg_type, flags: flags, remaining_length: remaining_length, body: payload[offset:offsetremaining_length] }把新解析器接进主流程时只需要在网络栈解析之后加一行判断。我倾向于把端口判断放在payload_handler而不是TCP解析器里因为端口是应用层约定不属于TCP层职责。这样如果以后想解析TLS握手都不用动TCP代码。5.2 用单元测试锁定解析和编辑行为改协议字段最容易引发关联bug比如改了IP长度但没改总长度或者改了TTL但没重算校验和。为避免这种问题我习惯给每个解析器都写一个回归测试。测试数据不需要手动构造可以从pcap文件里截取一段真实流量的十六进制字节存成测试集。每次改完代码跑一遍确保解析结果不变编辑后再序列化的字节与原包在未修改字段上完全一致。# tests/test_checksum.py import unittest from editor.checksum import tcp_checksum class TestChecksum(unittest.TestCase): def test_tcp_checksum_valid(self): # 用一段真实的TCP包字节作为测试向量 tcp_segment bytes.fromhex(0035...) # len40 src_ip 192.168.1.10 dst_ip 93.184.216.34 result tcp_checksum(src_ip, dst_ip, tcp_segment) self.assertEqual(result, 0xABCD) # 从Wireshark抄来的正确校验和测试向量的来源很关键。我都是先在Wireshark里找一个典型的TCP三次握手包复制原始十六进制再把它们转成Python字节字面量放进测试文件。注意拷贝的是原始字节而不是显示出来的ASCII十六进制否则容易多空格或大小写错误。从那以后我每次增加协议支持都强制走一遍先写测试、再改解析器的流程这个方法让我少走了很多弯路。希望帮到你。本文还有配套的精品资源点击获取

读完文章,也想定制专属网站?

尧图设计师 24 小时内与您沟通定制方案

免费获取报价 →
↑