简介:网络流量分析器源码面向网络运维、安全监控人员及学习 VC++ 网络编程的开发者,基于抓包数据实现多维度流量统计:可查看单点流量、点到点流量与协议流量排名,并支持按流量数值和占比两种方式呈现,同时给出基于协议及目标端口的累计流量,提示 ARP 流量占比并辅助 ARP 流量分析。它可用于业务流量监控、病毒木马后门流量排查、ARP 病毒监控和网络异常发现等场景,关键技术涉及内存表扫描与 BarChart 图表控件使用。压缩包共 36 个文件,约 168KB,包含 7 个 .h 头文件、5 个 .cpp 源文件,以及 bmp/ico/rc 等界面资源、sln/dsp/dsw/vcxproj 等工程文件和 exe、mdb、xlsx、ini 等辅助与数据文件,便于直接打开工程研究或运行示例。已有 909 人学习,适合作为流量分析工具原型、课程设计或二次开发参考。
1. 抓包脚本写了几十个,最后还是回头啃这份网络流量分析器源码
排障时最常见的场面:出口带宽跑满,监控面板一片正常,只能临时开个抓包窗口手工翻包。写个几十行的抓包脚本谁都会,难的是把抓到的原始帧变成能看的指标——谁在占带宽、哪些会话异常、DNS 解析慢在哪一环。这份网络流量分析器源码要解决的就是这段脏活:从网卡采集开始,做协议逐层解析、会话重组、指标聚合,最后落成可读的报表。它适合两类人:手上只有一堆 pcap 却不知道怎么下手的运维,以及想把抓包能力嵌进自己平台的开发者。代码量不大,但链路上的每个环节都得自己扛。
2. 架构拆解:一个数据包从网卡到报表的四层链路
整个项目按数据流分成四层,层与层之间只靠结构体或者字典传递,耦合度压得很低:采集层负责从网卡拿原始帧,解析层负责把字节拆成协议字段,会话层负责把零散的包拼回一条连接,统计层负责把连接汇总成指标并输出。这样分层的好处是每一层都能单独拿出去测——解析层可以喂离线的 pcap,统计层可以直接喂构造好的字典,不用每次都动真网卡。
很多人拿到这类源码,第一反应是从入口文件从头往下读,结果读两百行就陷进struct.unpack的偏移量里出不来了。我的建议是反着读:先看统计层输出了哪些字段,再看会话层的键是什么,最后才回去看解析层怎么把字节对齐到这些字段上。下面按这个顺序拆。
2.1 采集层:原始套接字、收包缓冲与内核态过滤
采集层的核心只有一件事:尽可能少丢包地把帧取出来。在 Linux 上常见的有两条路,一条是直接开AF_PACKET原始套接字,一条是套一层抓包库。源码里两种实现往往都留着,原始套接字版本没有额外依赖,适合塞进容器镜像;抓包库版本跨平台,Windows 上也能跑。下面这段是原始套接字的骨架:
import socket import time ETH_P_ALL = 0x0003 def open_capture(ifname="eth0", rcvbuf=8 * 1024 * 1024): # AF_PACKET + ETH_P_ALL:直接拿二层全部帧,包含非本机 MAC 的帧 sock = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(ETH_P_ALL)) sock.bind((ifname, 0)) # 内核收包缓冲,默认值通常只有 208KB,高 PPS 场景必调 sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, rcvbuf) sock.settimeout(1.0) # 让 recvfrom 可以被中断,便于优雅退出 return sock def capture_loop(sock, on_frame, stop_flag): dropped = 0 while not stop_flag.is_set(): try: frame, addr = sock.recvfrom(65535) except socket.timeout: continue # addr = (ifname, proto, pkttype, hatype, hwaddr) on_frame(time.time(), frame) return dropped参数上有几个地方决定了这份源码能不能上生产。rcvbuf是最容易被忽略的一项,8MB 是常见起点,流量越大越要往上加,加不上去说明系统的net.core.rmem_max卡住了,得先改 sysctl 再重启进程,光改代码没用。settimeout(1.0)看着不起眼,但没有它,退出信号进不来,Ctrl+C 之后进程会挂住——这是我在自己项目里踩过好几次的坑,后来养成习惯,凡是阻塞读都带超时。
recvfrom(65535)里的 65535 是单帧上限。真要抠性能可以降到 128 甚至 96,只抓包头不抓载荷,吞吐能翻上去,代价是应用层解析全废,做纯流量统计时这个取舍很划算。另外要注意,AF_PACKET拿到的帧可能比实际链路层帧长(网卡开了某些卸载特性时尤其明显),解析层必须做长度校验,不能无脑unpack。
再说过滤。用户态过滤意味着所有包都要从内核拷到用户态再被丢弃,高 PPS 下这是纯浪费。合格的做法是把过滤表达式下沉到内核:抓包库版本直接用pcap_compile生成 cBPF 程序,再通过SO_ATTACH_FILTER挂到套接字上。源码里常见的过滤字符串就三类:
# 只看 HTTP 和 DNS,适合做应用层分析 tcp port 80 or tcp port 8080 or udp port 53 # 排除本机 SSH,避免自己抓自己造成回声 not port 22 # 只看某个网段之间的流量 net 10.20.0.0/16 and not broadcast and not multicast规则写宽了用户态压力大,写窄了容易漏。有个反直觉的结论:过滤表达式越复杂,内核里的 cBPF 执行开销也越高,所以别写几十个or串起来的巨型规则,按分析目标拆成几个采集进程反而更好维护。
2.2 解析层:以太网帧头到传输层的逐层剥离
解析层是整个源码里最容易写错的地方,错一个字节偏移,后面全歪。它本质上是三段定长解析加一段变长载荷:14 字节以太网头、20 到 60 字节 IP 头、20 到 60 字节 TCP 头。下面这段是项目里的主解析流程,我按源码结构做了精简:
import socket import struct def parse_ethernet(frame): if len(frame) < 14: return None _dst, _src, ethertype = struct.unpack("!6s6sH", frame[:14]) offset = 14 # 802.1Q VLAN 标签:4 字节,会把 ethertype 往后推 while ethertype in (0x8100, 0x88A8) and len(frame) >= offset + 4: ethertype = struct.unpack("!H", frame[offset + 2:offset + 4])[0] offset += 4 return ethertype, offset def parse_ipv4(data, offset): if len(data) < offset + 20: return None ver_ihl = data[offset] ihl = (ver_ihl & 0x0F) * 4 # 首部长度以 4 字节为单位 total_len = struct.unpack("!H", data[offset + 2:offset + 4])[0] flags_frag = struct.unpack("!H", data[offset + 6:offset + 8])[0] proto = data[offset + 9] src = socket.inet_ntoa(data[offset + 12:offset + 16]) dst = socket.inet_ntoa(data[offset + 16:offset + 20]) end = min(offset + total_len, len(data)) # 抓包截断时 total_len 会超出实际长度 return { "src": src, "dst": dst, "proto": proto, "frag_off": flags_frag & 0x1FFF, "more_frag": bool(flags_frag & 0x2000), "payload": data[offset + ihl:end], } def parse_tcp(payload): if len(payload) < 20: return None sport, dport, seq, ack = struct.unpack("!HHII", payload[:12]) data_off = (payload[12] >> 4) * 4 # 高 4 位才是首部长度 flags = payload[13] return { "sport": sport, "dport": dport, "seq": seq, "ack": ack, "flags": flags, "payload": payload[data_off:], }这段代码里有三个参数值得单独说。ihl来自首字节低 4 位乘 4,正常是 20,带选项时能到 60,硬编码 20 是新手最常见的翻车点。total_len是 IP 层声明的总长,但抓包时如果 snaplen 设小了,实际拿到的len(data)会更小,所以必须取min,否则后面切片会静默丢数据,你只会看到统计数字对不上,查一天也查不出原因。frag_off和more_frag用来判断分片,非首片没有 TCP 头,直接拿去解析传输层会得到垃圾字段,源码里对分片一般只做计数不做重组,因为 IP 分片在现代网络里占比极低,为它写完整重组逻辑性价比不高。
TCP 头里的data_off同理,从第 12 字节的高 4 位取,低 4 位是保留位,直接用整个字节会算出 240 这种荒谬值。flags是标志位集合,SYN 是 0x02、ACK 是 0x10、FIN 是 0x01、RST 是 0x04,判断连接建立和拆除全靠它。
再往上就是应用层。HTTP 是明文,直接在载荷里找\r\n\r\n分隔头体,第一行按空格切三段拿到方法、路径、版本;DNS 走 UDP 53,头部 12 字节定长,问题区从第 12 字节开始,按域名标签逐段读长度前缀。这里有个约束要写进注释:遇到非标准端口(比如 8080 上的 HTTP、5353 上的 DNS)需要靠端口映射表或者载荷特征来识别,纯端口判断会漏一大片,这也是为什么很多分析结果看起来「少了一截」。
2.3 统计层:指标怎么聚合、怎么输出
统计层负责把成千上万条会话压成几十个数字。源码里的指标大致分四类:按主机的收发字节和包数、按协议的流量占比、按会话的时延和重传次数、按时间片的 PPS 曲线。数据结构上最省事的是字典加计数器:
import heapq import json from collections import defaultdict class TrafficStats: def __init__(self): self.by_host = defaultdict(lambda: [0, 0]) # ip -> [bytes, packets] self.by_proto = defaultdict(int) # proto -> bytes self.total_bytes = 0 def add(self, src, dst, proto, nbytes): h = self.by_host[src] h[0] += nbytes h[1] += 1 self.by_proto[proto] += nbytes self.total_bytes += nbytes def top_talkers(self, n=10): # 按字节排序,包数只作参考:小包洪水刷包数很快,但占不满带宽 return heapq.nlargest(n, self.by_host.items(), key=lambda kv: kv[1][0]) def dump(self, path): with open(path, "w", encoding="utf-8") as f: json.dump({ "total_bytes": self.total_bytes, "top": [{"ip": ip, "bytes": v[0], "packets": v[1]} for ip, v in self.top_talkers(20)], "proto": dict(self.by_proto), }, f, ensure_ascii=False, indent=2)top_talkers用heapq.nlargest而不是sorted,是因为主机数可能到十万级,全排序是 O(n log n) 且要额外内存,堆只要 O(n log k)。by_host用defaultdict省掉存在性判断,但代价是键无界增长——一个跑了三天的采集进程,字典里可能攒了几十万个 IP,内存肉眼可见地涨。源码里的处理方式通常是按时间窗口定期整体重置,或者对长时间不活跃的键做衰减,两种都行,关键是要有,不能放着不管。
输出格式上,JSON 适合给前端或者下游接口用,CSV 适合丢进表格里画图。如果源码带了个简易的 Web 面板,统计层一般会额外维护一份最近 N 个时间片的环形缓冲,前端轮询拿数组直接渲染折线,不用每次重算全量。
提示:统计层的所有计数器都要考虑回绕和清零时机,尤其是做「最近一分钟 PPS」这类滑动窗口指标时,用总数相减而不是每秒清零,否则采样抖动会让曲线出现毛刺。
3. 目录结构与核心模块:源码包里每个文件该看哪几行
拿到源码包先别急着跑,花十分钟把目录看一遍,后面定位问题能省几个小时。这类项目的目录结构大同小异,按职责切分,命名也比较直白。
3.1 目录分层:六个目录各自的职责与阅读优先级
traffic-analyzer/ ├── capture/ # 采集层:原始套接字、抓包库封装、BPF 过滤 │ ├── sniffer.py # 主循环,入口在这里 │ └── filter.py # 过滤表达式编译与挂载 ├── parser/ # 解析层:逐协议剥离 │ ├── ethernet.py │ ├── ipv4.py │ ├── tcp.py │ ├── udp.py │ └── app/ # 应用层:http / dns / tls 握手字段 ├── session/ # 会话层:流表、TCP 重组、超时老化 │ ├── flowtable.py │ └── reassembly.py ├── stats/ # 统计层:指标聚合、TopN、阈值判定 │ └── metrics.py ├── output/ # 输出层:JSON / CSV / 简易 Web 面板 └── conf/ # 配置文件、网卡与过滤规则| 目录 | 主要职责 | 建议阅读顺序 | 改动频率 |
|---|---|---|---|
| capture | 拿帧、内核过滤、丢包统计 | 第 3 | 低 |
| parser | 字节到字段的映射 | 第 1 | 高(加协议都在这) |
| session | 五元组归并、字节流拼接 | 第 2 | 中 |
| stats | 聚合与阈值 | 第 4 | 高(指标口径常变) |
| output | 序列化与展示 | 第 5 | 中 |
| conf | 网卡名、过滤串、超时参数 | 先看 | 高 |
读的顺序建议是conf→parser→session→capture→stats→output。先看配置知道默认抓哪块网卡、过滤什么,再看解析知道能识别哪些协议,然后看会话层怎么归并,最后才回头看采集。这么读的好处是每一步都有明确的输入输出,不会一上来就被套接字和信号处理绕晕。
3.2 流表设计:五元组哈希、双向归并与超时老化
流表是整份源码里最值得抄的一段。它决定了一条连接在内存里长什么样、什么时候被回收。核心是一个以五元组为键的字典,加上一个老化机制:
import time from dataclasses import dataclass @dataclass class Flow: key: tuple first_seen: float last_seen: float packets: int = 0 nbytes: int = 0 syn_seen: bool = False fin_seen: bool = False class FlowTable: def __init__(self, idle_timeout=120, max_flows=200000): self.flows = {} self.idle_timeout = idle_timeout self.max_flows = max_flows def touch(self, ts, src, dst, sport, dport, proto, nbytes, flags=0): key = (src, dst, sport, dport, proto) flow = self.flows.get(key) if flow is None: # 双向归并:先试反方向,避免同一连接被拆成两条 rev = (dst, src, dport, sport, proto) flow = self.flows.get(rev) if flow is None: flow = Flow(key=key, first_seen=ts, last_seen=ts) self.flows[key] = flow flow.last_seen = ts flow.packets += 1 flow.nbytes += nbytes if flags & 0x02: flow.syn_seen = True if flags & 0x01: flow.fin_seen = True return flow def expire(self, now): dead = [k for k, f in self.flows.items() if now - f.last_seen > self.idle_timeout] for k in dead: del self.flows[k] return len(dead)idle_timeout默认 120 秒是经验值:太长,端口扫描和短连接会把表撑爆;太短,长连接的统计会被切成好几段,TopN 排名失真。max_flows是硬上限,超了就得触发强制淘汰,通常淘汰最久未活跃的那批,源码里如果没写这个兜底,跑在高并发环境里迟早 OOM。
双向归并那段是整个流表的关键设计。一条 TCP 连接的两个方向五元组是镜像的,如果不做归并,一次请求会被算成两条流,字节数各记一半,时延也算不出来。归并的代价是每次新流都要多查一次字典,命中率还行——绝大多数包都属于已存在的流。
expire这里是全表扫描,流量大的时候这一下能卡住好几毫秒。生产做法是用最小堆按last_seen排序,或者按时间分桶,只扫当前桶和上一个桶。这是我在自己项目里优化过的点,改完之后 CPU 曲线明显平了一段。
3.3 TCP 会话重组:按 seq 拼字节流,认出请求行
只做统计不做内容分析的话,会话层到流表就够了。但只要你想知道「这个请求返回了什么状态码」,就必须把乱序到达的 TCP 段拼回字节流:
def reassemble(segs, isn=None): """segs: [(seq, payload)],按到达顺序排列;isn: 首个序列号的基准值""" segs = [(s, p) for s, p in segs if p] if not segs: return b"" base = isn if isn is not None else min(s for s, _ in segs) buf = bytearray() for seq, payload in segs: off = (seq - base) & 0xFFFFFFFF # 32 位序列号回绕 if off >= 0x80000000: # 负偏移,说明是重传的旧段 continue if off > len(buf): buf.extend(b"\x00" * (off - len(buf))) # 空洞占位 buf[off:off + len(payload)] = payload return bytes(buf)& 0xFFFFFFFF那一行是处理序列号回绕的,32 位计数器跑满一圈会从 0 重新开始,直接相减会得到负数或者天文数字。用off >= 0x80000000判断「这其实是负偏移」,是把有符号比较转成无符号比较的常见写法,比先转 int32 再比较省事。
中间的空洞补零是权宜之计:真丢包了,补零之后你在字节流里会看到一段\x00,HTTP 解析会在这里断掉,正好可以当作丢包的信号。比起花力气做完整的重传等待和选择性确认,这个做法在分析场景里够用了。
拿到字节流之后,取前 4KB 找第一行\r\n,按空格切三段就是方法、路径、版本;从头部里找Host和User-Agent,从响应里找状态码。这套逻辑简单粗暴,但对明文协议命中率很高。
注意:重组只对明文有意义。加密流量在会话层只能看到握手阶段的字段(SNI、证书长度、协商的套件),内容一律是随机字节,别指望从中提取路径。
4. 跑起来:环境、权限与第一次出图
源码能跑通和能分析是两回事。这一章按「先离线、再在线」的顺序走,能少踩不少坑。
4.1 依赖与编译:三条命令把环境搭起来
Python 版本的项目一般带requirements.txt,依赖不会太多,抓包封装、序列化、可选的图表或 Web 框架,加起来通常不超过十个包。C 版本则是configure && make或者 CMake 一套。下面按 Python 版本走:
# 1. 建独立环境,别往系统 Python 里装,抓包相关的依赖容易和系统包打架 python3 -m venv .venv && source .venv/bin/activate # 2. 装依赖 pip install -r requirements.txt # 3. 试跑,先不指定网卡,看能不能打印帮助 python -m analyzer --help--help是最快的健康检查。如果这一步就报ModuleNotFoundError,说明依赖没装全或者虚拟环境没激活;如果报语法错误,多半是 Python 版本太低,源码里用了 3.8 以上的语法(比如海象运算符或者dataclass的某些参数),升到 3.9 以上基本能解决。C 版本如果make失败,先看是不是缺libpcap的开发头文件,这类错误信息很直白,fatal error: pcap.h: No such file or directory就是缺开发包。
4.2 权限与网卡:为什么一运行就是 Permission denied
这是新手第一个必然踩的坑。裸套接字需要CAP_NET_RAW能力,普通用户没有,直接跑就是PermissionError: [Errno 1] Operation not permitted。三种解法,按推荐程度排:
# 方案一:临时提权,最省事,但进程整体都在 root 下,风险最大 sudo python -m analyzer --iface eth0 # 方案二:只给解释器加抓包能力,比全量 root 好,但等于给整个解释器开权限 sudo setcap cap_net_raw,cap_net_admin=eip $(readlink -f $(which python3)) # 方案三:把采集进程拆出来单独跑,分析进程用普通用户,通过管道或本地队列传数据 sudo python -m analyzer.capture --iface eth0 | python -m analyzer.stats方案三最干净,也是我一般推荐的结构:采集进程权限高但逻辑极简,分析进程逻辑复杂但零权限,中间用标准输入输出或者本地队列隔开。缺点是多了进程间通信的开销,做高吞吐时要留意缓冲区大小。
网卡名也别写死。eth0在有的机器上是ens33,在容器里可能是eth0@if12这种带后缀的形式。先ip link看一眼实际名字,或者干脆用--iface any让套接字抓全部接口——代价是混杂模式下的流量会翻倍,分析结果的去重逻辑要跟上。
提示:在容器里跑抓包,要确认容器是否有独立的网络命名空间。没有的话,抓到的可能是宿主机全部流量,统计口径完全不对。
4.3 离线回放:先用 pcap 把链路跑通再上真网卡
强烈建议第一次运行喂离线文件。线上抓包出问题,你分不清是解析错了还是采集丢了;离线文件是固定的,同样的输入必然得到同样的输出,排查范围小一个数量级。
# 用现成的 pcap 跑一遍,输出到 result.json python -m analyzer --read sample.pcap --out result.json --topn 20 # 检查是否解析出会话 python -c "import json;d=json.load(open('result.json'));print(len(d['top']), d['total_bytes'])"--read分支和实时采集分支在源码里通常共用一个解析入口,区别只在于帧从哪来:一个是recvfrom,一个是文件读取循环。如果离线能跑出结果、实时跑不出,问题一定在采集层;如果两边都跑不出,问题在解析层。这个二分法能省掉大量猜测。
4.4 常用参数表与调参建议
| 参数 | 含义 | 默认值 | 调参建议 |
|---|---|---|---|
--iface | 抓包网卡 | 无 | 先用ip link确认真实名字 |
--filter | BPF 过滤表达式 | 空(全抓) | 生产必填,写窄不写宽 |
--snaplen | 单帧截取长度 | 65535 | 纯统计降到 128 |
--rcvbuf | 内核收包缓冲 | 8MB | 高 PPS 调到 32MB 以上 |
--idle-timeout | 流老化秒数 | 120 | 短连接多降到 30 |
--max-flows | 流表上限 | 200000 | 按内存预算调,单流约 200 字节 |
--topn | TopN 输出条数 | 10 | 报表用 20,排查用 100 |
--out | 输出路径 | stdout | JSON 给程序,CSV 给表格 |
调参顺序建议是:先定--filter缩小范围,再调--rcvbuf解决丢包,最后才动--idle-timeout和--max-flows。反过来调容易陷入「改了参数指标变了但不知道是哪个参数起作用」的玄学状态。
5. 避坑与排查:六个把新手劝退的现场
这一节全是血泪经验,按「现象 → 原因 → 解决」写,遇到对得上的直接照做。
5.1 采集阶段:抓不到、丢得多、混杂模式失效
现象一:程序正常退出,输出全零。原因通常是网卡名写错或者流量根本没过这块网卡。先ip -s link show eth0看 RX 计数是否在涨,不涨说明选错网卡;涨但程序没输出,就是过滤表达式把包全挡了,把--filter置空再跑一次。还有个隐蔽情况:程序跑在容器里,容器网络命名空间和宿主机隔离,抓的是容器自己的虚拟网卡,自然什么也抓不到。
现象二:抓到了,但和实际流量对不上,总字节数偏小。八成是内核缓冲不够导致丢包。原始套接字不会主动告诉用户态丢了多少,得主动查:用抓包库版本的话,结束时会打印ps_drop;用裸套接字的话,读PACKET_STATISTICS才知道。解决办法就是把rcvbuf往上加,同时确认net.core.rmem_max也放开了,两个值必须同时改,只改一个等于没改。
现象三:只能看到本机的流量,看不到同网段其他机器的。这是没开混杂模式。AF_PACKET默认只收目标 MAC 是本机的帧,要抓全量得设置PACKET_MR_PROMISC;抓包库版本是pcap_set_promisc传 1;命令行工具一般是-p的反面。注意在虚拟化和云环境里,混杂模式经常在底层就被挡住了,你设了也没用,这种情况只能靠镜像端口或者在本机做采集代理。
5.2 解析与统计阶段:字段错位、内存上涨、口径不一致
现象四:解析出来的端口号是几万这种鬼数字,或者 IP 显示成乱码。一定是字节序搞反了。网络字节序是大端,Python 里struct的格式串必须带!前缀;抓包库返回的字段有时已经是主机序,混用就会错。排查方法很土但有效:抓一个已知的包(比如自己curl一下),把解析结果和 Wireshark 里看到的逐字段比对,第一个对不上的字段就是出错的位置。
现象五:跑几个小时后内存一路涨,最后被 OOM kill。流表和主机字典都是无界键空间,端口扫描或者短连接洪水能把表撑到几十万条。要检查三件事:expire有没有被定期调用(很多人写了但忘了在主循环里调)、有没有max_flows兜底、主机统计字典有没有按窗口重置。三者缺一,长时间运行必然涨。
现象六:两份报表的「总流量」对不上。这是口径问题,不是代码问题。常见分歧点有三个:是否计入二层头部(14 字节)、是否计入以太网填充、双向流量是合并算还是分开算。源码里一般在常量定义处会写清楚,改口径的时候把这几个开关集中到一个地方,别散落在各个模块里,否则下次对账还得再翻一遍。
注意:改解析逻辑之前,先拿一份固定的 pcap 存个基线输出。改完 diff 一下,能立刻看出哪些指标变了、变的方向对不对,比凭感觉判断可靠得多。
6. 二次开发:加一个自定义协议解析器并验证它真的生效
这类项目最典型的二次开发需求就是加协议。假设你有个内部服务跑在自定义端口上,包头是 4 字节魔数加 2 字节长度。改造的关键是别去动主解析流程,而是走注册表:
# parser/app/__init__.py 里维护一张注册表 PARSERS = [] def register(proto, ports): def wrapper(fn): PARSERS.append({"proto": proto, "ports": set(ports), "fn": fn}) return fn return wrapper # parser/app/myproto.py import struct from . import register @register("myproto", ports=[9527]) def parse_myproto(payload): if len(payload) < 6: return None magic, body_len = struct.unpack("!IH", payload[:6]) if magic != 0x4D595052: return None body = payload[6:6 + body_len] return {"cmd": body[:2].hex(), "size": body_len}主流程只要在 TCP 载荷进应用层解析前遍历一次PARSERS,端口匹配或者魔数匹配就交给对应函数,返回None视为不识别,继续往下走。这样加协议不用改核心代码,也不会影响已有协议。
验证分三步走,别跳。第一步造一个最小 pcap:用脚本拼一个带目标端口的 TCP 包写进文件,跑离线模式,看解析结果里有没有目标字段;第二步用真实流量的副本跑,对比改动前后的会话数和字节数,差值应该只落在你新增的协议上;第三步挂到实时网卡上跑半小时,看 CPU 和内存曲线有没有异常抬升——注册表是线性遍历的,协议一多就是 O(n),到时候该换成按端口建索引。
改造里最容易翻车的地方是异常处理。解析函数抛异常会直接打断抓包主循环,一条畸形包就能让整个采集停掉。我一般会在调用处套一层try/except,把异常连同原始载荷前 64 字节记进错误日志,跑一天之后翻日志,哪些解析器在裸奔一目了然。
从那以后我每次加新解析器,都强制先跑一遍离线基线、再跑一遍对比、最后才上真网卡,三步少一步就得回滚重来。希望帮到你。
本文还有配套的精品资源,点击获取