网络流量分析器源码解析：从抓包到会话统计的完整链路

简介这份网络流量分析器源码面向从事网络监控、安全运维与VC开发的技术人员提供一套可编译运行的抓包流量分析程序可用于业务流量监控、病毒木马后门流量监控、ARP病毒监控及网络异常监控等实际场景。压缩包共36个文件约168KB以h头文件与cpp源文件为主体配合rc资源脚本、ico与bmp图标位图、mdb与db数据文件、sln与vcxproj工程文件及exe可执行程序构成完整的VC工程结构。程序实现单点流量、点到点流量与协议流量排名分析支持按流量统计和按流量占比统计可显示基于协议和目标端口的累计流量并提示ARP流量占比、开展ARP流量分析。源码涉及内存表扫描、BarChart图表控件使用等关键技术点读者可据此理解抓包记录的组织方式、统计逻辑与图表呈现方法并在此基础上扩展监控指标或移植到自身业务环境。目前已有911人学习下载。1. 抓包抓不到重点这套网络流量分析器源码把链路拆开给你看线上接口突然变慢日志里全是正常返回服务端 CPU 也不高最后发现是某个下游连接在反复重传。这种问题靠打日志基本无解得看链路上真正跑了什么。这套网络流量分析器源码解决的就是这件事从网卡抓包、按协议解析、统计会话、输出可读的分析结果一条链路走完。它适合两类人——想搞懂抓包解析全流程的后端和运维以及要拿现成骨架做流量监控、协议识别、异常检测的开发者。源码结构不复杂但把抓包、解码、统计、展示这几层切得比较清楚拿来当网络方向的学习底子或者二次开发起点都合适。下面按“能干什么、怎么跑、怎么改”的顺序拆。2. 架构与目录抓包、解码、统计、输出四层怎么串起来2.1 整体分层与数据流这套源码的核心思路是流水线抓包层负责从网卡或离线文件拿到原始字节解码层把字节按以太网帧、IP 包、TCP/UDP 段逐层剥开统计层按五元组聚合会话并累计字节数、包数、时长输出层把结果打成表格或写文件。四层之间用统一的数据结构传递解码层不关心数据从哪来统计层不关心字节怎么解这样离线分析和实时抓包能复用同一套逻辑。常见做法是定义一个Packet结构体字段包括时间戳、源/目的 IP、源/目的端口、协议号、载荷长度、TCP 标志位。抓包层填时间戳和原始字节解码层填其余字段统计层只读这个结构。这样分层的好处是加新协议解析时只动解码层换输出格式时只动输出层互不影响。# 统一的数据载体四层之间只传这个结构 class Packet: def __init__(self): self.ts 0.0 # 抓包时间戳秒 self.src_ip # 源 IP self.dst_ip # 目的 IP self.src_port 0 # 源端口 self.dst_port 0 # 目的端口 self.proto 0 # 协议号6TCP, 17UDP self.payload_len 0 # 载荷长度 self.tcp_flags 0 # TCP 标志位用于识别 SYN/FIN/RST这段结构体是整个项目的“通用货币”。ts用浮点秒而不是整数毫秒是为了在统计会话时长时保留精度tcp_flags单独留字段是因为后面判断连接建立和断开要靠 SYN、FIN、RST如果只存协议号就没法区分一次正常关闭和一次异常重置。实际写的时候我一般会再加一个raw字段存原始字节方便调试时回看但统计层不读它避免内存占用过大。2.2 目录结构与模块职责源码目录大致是这样组织的不同版本可能略有出入但分层逻辑一致traffic-analyzer/ ├── capture/ # 抓包层网卡抓包、pcap 文件读取 │ ├── live.py # 实时抓包 │ └── offline.py # 离线 pcap 读取 ├── decode/ # 解码层以太网/IP/TCP/UDP 解析 │ ├── ethernet.py │ ├── ip.py │ └── transport.py ├── stats/ # 统计层会话聚合、流量统计 │ └── session.py ├── output/ # 输出层表格、文件、JSON │ └── report.py ├── config.py # 过滤规则、抓包参数 └── main.py # 入口串起四层capture/下实时和离线分开是因为两者拿数据的方式不同实时抓包要绑定网卡、设混杂模式离线读取只是按顺序读文件。但两者对外都暴露同一个迭代器接口main.py里换一行配置就能切换。decode/按协议层拆文件是为了每层只处理自己那一段以太网层剥掉 14 字节头交给 IP 层IP 层剥掉 20 字节头交给传输层职责清晰。stats/只有一个会话聚合模块因为统计逻辑相对集中。output/独立出来方便后面加新的展示形式。2.3 入口怎么把四层串起来main.py的逻辑很短就是取配置、建抓包源、循环解码、喂给统计、最后输出。关键在于循环里对异常包的处理解码失败不能整个程序崩掉要跳过继续。from capture.live import LiveCapture from capture.offline import OfflineCapture from decode.ethernet import decode_ethernet from stats.session import SessionTable from output.report import print_report import config def build_source(cfg): # 根据配置决定实时抓包还是读离线文件 if cfg.mode live: return LiveCapture(cfg.iface, cfg.filter) return OfflineCapture(cfg.pcap_path) def run(): cfg config.load() source build_source(cfg) table SessionTable() for raw in source: try: pkt decode_ethernet(raw) # 逐层解码返回 Packet except ValueError: continue # 坏包跳过不中断 if pkt is None: continue # 非 IP 包直接忽略 table.add(pkt) # 按五元组聚合 print_report(table, top_ncfg.top_n)build_source把抓包源的创建收在一处cfg.mode是live还是offline决定走哪个分支这样切换分析方式不用改主循环。decode_ethernet返回None表示这个包不是 IP 流量比如 ARP直接忽略抛ValueError表示字节不够或格式不对跳过。table.add是统计层的入口内部按五元组找已有会话或新建。print_report的top_n控制输出前几名避免终端刷屏。这个入口没有花哨设计但把“坏包不崩、非目标流量不统计”这两个实际场景处理了跑真实流量时很关键。3. 跑起来环境、抓包权限和第一份分析结果3.1 依赖安装与配置这套源码用 Python 写抓包底层依赖 libpcapPython 侧常见做法是用 scapy 或 pcapy 这类库。按项目里requirements.txt装依赖即可核心就两三个包。装之前确认系统有 libpcap 开发库否则底层库编译会失败。# 系统层依赖Debian/Ubuntu 系 sudo apt-get install -y libpcap-dev # Python 依赖 pip install -r requirements.txtlibpcap-dev提供抓包的系统调用接口Python 的抓包库只是它的封装缺了它pip install阶段就会报找不到头文件。requirements.txt里一般会有抓包库、解析库和表格输出库版本不用刻意锁死但抓包库和系统 libpcap 版本差太多时会有兼容问题遇到报错先看这一层。配置文件config.py里几个参数要按场景改参数含义常用值mode抓包模式live / offlineiface网卡名eth0 / ens33pcap_path离线文件路径./test.pcapfilterBPF 过滤表达式tcp port 80top_n输出前 N 名10iface用ip addr或ifconfig查实际网卡名不同系统命名不一样填错会直接报“找不到设备”。filter是 BPF 语法写tcp port 80就只抓 80 端口的 TCP能大幅减少无关流量跑大流量时必设。3.2 抓包权限这个坎实时抓包要读网卡原始数据普通用户没权限直接跑会报PermissionError或底层返回空。常见做法是用sudo跑或者给 Python 解释器加抓包能力。# 方式一直接提权运行 sudo python main.py # 方式二给解释器加能力之后免 sudo sudo setcap cap_net_raw,cap_net_admineip $(readlink -f $(which python3))方式一简单但每次都要 sudo调试时环境变量和路径容易和普通用户不一致。方式二给解释器加了原始套接字和管理网卡的能力之后普通用户也能抓适合反复调试。注意方式二改的是解释器权限换虚拟环境后要重新设。如果只是分析已有的 pcap 文件用offline模式完全不需要这些权限新手建议先从这个模式入手。3.3 用离线文件跑通第一遍手上没有现成流量时先拿一个 pcap 文件跑离线模式确认整条链路通。把config.py里mode改成offlinepcap_path指向文件然后运行。python main.py正常输出是按会话聚合的表格每行一个五元组列出包数、字节数、持续时间。第一次跑建议先看三件事总包数对不对、有没有明显解析失败的计数、前几名会话是不是符合预期。如果总包数是 0先查pcap_path和文件格式如果解析失败计数很高多半是链路层类型不匹配比如文件是 Linux cooked 格式而不是以太网帧需要在解码层加分支。跑通离线再切实时能少踩很多权限和网卡的坑。4. 核心代码拆解解码层和会话统计怎么写4.1 以太网到 IP 的逐层剥离解码层是整个项目最需要小心的地方因为网络字节序、可变头长度、分片这些细节都在这里。以太网帧头固定 14 字节其中前 12 字节是源/目的 MAC后 2 字节是上层协议类型0x0800表示 IPv4。剥掉这 14 字节剩下的交给 IP 层。import struct ETH_HEADER_LEN 14 ETH_TYPE_IPV4 0x0800 def decode_ethernet(raw): if len(raw) ETH_HEADER_LEN: raise ValueError(frame too short) # 前 12 字节 MAC 这里不解析直接跳过 eth_type struct.unpack(!H, raw[12:14])[0] if eth_type ! ETH_TYPE_IPV4: return None # 非 IPv4忽略 return decode_ip(raw[ETH_HEADER_LEN:])struct.unpack(!H, ...)里的!表示网络字节序大端这是网络协议解析最容易翻车的地方——x86 是小端不显式指定就会把0x0800读成0x0008然后所有包都被当成未知协议。raw[12:14]取的是以太网类型字段判断不等于 IPv4 就返回None让上层跳过。这里没解析 MAC 地址因为统计层用不到需要时再加。实际抓包里还有 VLAN 标签0x8100带标签的帧头会多 4 字节遇到时要在这一步先判断再偏移否则后面全错位。4.2 IP 头解析与可变长度处理IP 头不是固定 20 字节IHL字段表示头长度以 4 字节为单位有选项时会更长。解析时必须按IHL算偏移写死 20 字节在带选项的包上会解析出错。def decode_ip(data): if len(data) 20: raise ValueError(ip header too short) ver_ihl data[0] version ver_ihl 4 ihl (ver_ihl 0x0F) * 4 # 头长度单位字节 if version ! 4: return None proto data[9] # 协议号 src_ip ..join(str(b) for b in data[12:16]) dst_ip ..join(str(b) for b in data[16:20]) total_len struct.unpack(!H, data[2:4])[0] payload data[ihl:total_len] # 按总长度截去掉以太网填充 return decode_transport(proto, src_ip, dst_ip, payload)ver_ihl 4取高 4 位是版本号 0x0F取低 4 位是头长度再乘 4 得到字节数这是 IP 头解析的标准操作。data[9]是协议号6 是 TCP、17 是 UDP。源/目的 IP 各取 4 字节转成点分十进制。total_len从偏移 2 取 2 字节用它截payload而不是直接取到末尾是因为以太网帧有最小长度限制小包后面会填充无用字节不按总长度截会把填充当载荷统计字节数就偏大。这个细节不注意小包多的场景统计结果会明显虚高。4.3 会话聚合五元组和超时淘汰统计层的核心是用五元组源 IP、源端口、目的 IP、目的端口、协议做键把同一个连接的所有包累加到一起。难点在于连接什么时候算结束TCP 有 FIN/RST 可以判断UDP 没有连接概念只能靠超时。import time class SessionTable: def __init__(self, timeout60): self.sessions {} # key: 五元组, value: 统计信息 self.timeout timeout # 空闲超时秒数 def _key(self, pkt): # 双向流量归一化小端口在前保证请求和响应落到同一会话 a (pkt.src_ip, pkt.src_port) b (pkt.dst_ip, pkt.dst_port) pair tuple(sorted([a, b])) return (pair[0], pair[1], pkt.proto) def add(self, pkt): key self._key(pkt) now pkt.ts s self.sessions.get(key) if s is None: s {packets: 0, bytes: 0, start: now, last: now} self.sessions[key] s s[packets] 1 s[bytes] pkt.payload_len s[last] now # TCP 收到 FIN 或 RST标记会话可关闭 if pkt.proto 6 and (pkt.tcp_flags 0x01 or pkt.tcp_flags 0x04): s[closed] True_key里对两端排序是关键设计同一个连接的两个方向源和目的会互换不归一化就会变成两个会话统计翻倍。sorted按 IP 和端口排序保证两个方向算出同一个键。add里累加包数和字节数更新最后活跃时间。TCP 的 FIN 标志是0x01、RST 是0x04收到就标记closed输出时可以区分正常关闭和异常重置。timeout字段留给清理逻辑用长时间没流量的会话要淘汰否则跑久了内存一直涨。清理一般在主循环里定期扫一遍把now - last timeout的删掉。UDP 会话只能靠这个超时判断结束超时设太短会把长连接拆成多个设太长内存回收慢60 秒是个折中值按业务调。4.4 输出层把统计结果打成能看的表输出层不复杂但决定了结果好不好读。常见做法是按字节数降序排取前 N 名每行列出五元组、包数、字节数、持续时长。def print_report(table, top_n10): rows [] for key, s in table.sessions.items(): (sip, sport), (dip, dport), proto key duration s[last] - s[start] rows.append((s[bytes], sip, sport, dip, dport, proto, s[packets], duration)) rows.sort(reverseTrue) # 按字节数降序 print(f{源:22}{目的:22}{协议:6}{包数:8}{字节:10}{时长}) for r in rows[:top_n]: src f{r[1]}:{r[2]} dst f{r[3]}:{r[4]} print(f{src:22}{dst:22}{r[5]:6}{r[6]:8}{r[0]:10}{r[7]:.1f}s)rows.sort(reverseTrue)直接按元组第一个元素字节数降序因为元组比较从第一项开始字节数相同才比后面的实际场景够用。格式化里:22是左对齐占 22 字符保证列对齐IP 加端口最长差不多这个宽度。duration用最后减开始反映会话活跃跨度。这个输出适合终端快速看要长期留存或做进一步分析把print换成写 CSV 或 JSON 就行统计层的数据结构不用动这也是分层的好处。5. 避坑与排查跑真实流量时最容易翻车的几处5.1 抓到的包全是 0 或者只有自己的流量现象是实时模式跑起来统计结果里只有本机发起的连接看不到其他机器的流量。原因是网卡没进混杂模式默认只收发给自己的帧。解决是在打开抓包设备时显式设置混杂模式抓包库一般有对应参数比如promiscTrue。另外虚拟机环境里网卡可能是 NAT 模式宿主机和其他虚拟机的流量根本到不了这块虚拟网卡这种情况要么换桥接模式要么在网关侧抓。先确认抓包位置对不对再怀疑代码。5.2 字节数统计明显偏大现象是统计出来的总字节数比实际流量大不少小包多的场景尤其明显。原因是 IP 头解析时没按total_len截断把以太网帧的填充字节也算进了载荷。以太网最小帧 64 字节小包后面会补零到最小长度不截断就会多算。解决是按 IP 头里的总长度字段截payload代码里data[ihl:total_len]就是干这个的。改完再对比一次抓包工具的总字节数能对上就说明修对了。5.3 同一连接被统计成两条现象是请求和响应分别出现在两行包数和字节数各算一半。原因是会话键没做双向归一化源和目的互换后算出了不同的键。解决是在生成键时对两端排序保证A→B和B→A落到同一个键。注意排序要按 IP 和端口一起排只排 IP 不排端口同一对 IP 上的多个连接会混在一起。改完后看输出一个连接的上下行应该合并成一行。5.4 跑一段时间内存持续上涨现象是长时间抓包后进程内存越来越大最后被系统杀掉。原因是会话表只增不删短连接和扫描流量会不断产生新键。解决是加超时淘汰定期扫描把空闲超过阈值的会话删掉TCP 已关闭的会话可以更早清理。清理频率不用太高几十秒一次就够太频繁反而增加开销。另外raw字段如果存了原始字节统计完要及时释放别一直挂在会话上。5.5 离线文件解析全失败现象是离线模式跑起来解析失败计数等于总包数一条有效会话都没有。原因是链路层类型不匹配pcap 文件可能不是标准以太网帧而是 Linux cooked 格式SLL或原始 IP。解决是先读 pcap 全局头里的链路类型字段按类型选择对应的解码入口SLL格式没有 MAC 头偏移量不一样。抓包工具导出时也可以指定格式统一成以太网帧最省事。遇到这种问题先打印前几个包的十六进制看头部结构就能判断。6. 二次开发加协议识别和异常检测的落点拿到这套源码后最常见的改造方向是加协议识别和异常检测。协议识别不用改抓包和统计层在解码层拿到端口和载荷后加一个判断函数就行按知名端口先做粗分类80/443 是 Web、22 是 SSH端口不明确的再看载荷特征。异常检测则挂在统计层之后对会话表做一轮扫描比如单会话字节数超过阈值、短时间内大量 SYN 没有对应 ACK扫描特征、某个目的端口连接数异常集中都可以作为规则。这些规则读的都是现成的会话结构不用动底层。改造时我建议先加一个可配置的规则表把阈值和开关放配置里别写死在代码里。验证方法很直接拿一份已知包含异常的 pcap 跑一遍看规则能不能命中再拿正常流量跑一遍看误报率。下面是一个规则扫描的骨架挂在print_report之前def detect_anomalies(table, rules): alerts [] for key, s in table.sessions.items(): (sip, sport), (dip, dport), proto key # 规则一单会话流量过大 if s[bytes] rules[max_session_bytes]: alerts.append((大流量会话, sip, dip, s[bytes])) # 规则二短时间大量包疑似扫描或攻击 duration max(s[last] - s[start], 0.001) if s[packets] / duration rules[max_pps]: alerts.append((高速率会话, sip, dip, s[packets])) return alertsrules从配置读max_session_bytes和max_pps按实际网络规模调内网和公网阈值差很多。duration取max(..., 0.001)是防止除零单个包的会话时长为 0 会直接崩。这个骨架只做了两条规则加规则就是往循环里加判断结构不用变。验证时把告警和人工分析结果对一遍误报多的规则先调阈值再考虑删。血泪经验是改解码层一定要拿真实 pcap 回归别只用手造的包测。我见过按标准文档写的解析在真实流量上因为 VLAN 标签和 IP 选项直接错位手造包永远测不出来。从那以后我每次动解码逻辑都强制拿几份不同来源的 pcap 各跑一遍对比总包数和字节数再提交。希望帮到你。本文还有配套的精品资源点击获取