简介:这是一份面向计算机网络课程设计的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], raw=raw_packet )) return parsed_packets这个调度流程只处理了最常见的以太网+IPv4+TCP组合,实际项目里会继续判断协议号,如果遇到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), ethertype=ethertype, payload=payload )这里__slots__能减少实例内存占用的不确定因素,尤其当你一次解析数万包时能明显感觉到差别。struct.unpack('!H')中的!表示网络字节序(大端),因为IEEE 802.3规定的以太网头部字段都是大端传输。如果你用原生字节序解析,在x86机器上得到的值会被翻转,常见现象是0x0800(IPv4)变成了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[i+1] 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 b'PORT ' 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:offset+remaining_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...') # len=40 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十六进制,否则容易多空格或大小写错误。从那以后我每次增加协议支持,都强制走一遍先写测试、再改解析器的流程,这个方法让我少走了很多弯路。希望帮到你。
本文还有配套的精品资源,点击获取