简介这是一份面向网络管理与安全监控开发者的VC网络流量分析器源码围绕抓包记录实现多种条件下的流量统计与展示。程序可分析单点流量、点到点流量与协议流量排名支持按流量数值或占比统计并给出基于协议和目标端口的累计流量、ARP流量占比及ARP分析可用于业务流量、病毒木马后门、ARP病毒和网络异常监控等场景。压缩包共36个文件、约168KB包含h与cpp源码、rc及ico/bmp界面资源、sln/vcxproj/dsp等工程文件以及mdb、db、ini、xlsx等配置数据另附可执行文件便于对照运行。关键技术涉及内存表扫描和BarChart图表控件使用适合学习MFC界面、统计逻辑与抓包分析联动读者可据此拆解抓包数据结构、流量聚合统计、图表刷新与配置读取等模块并改造成自己的监控工具。已有911人学习下载。1. 一条上行被打满的排查把我推向了这份网络流量分析器源码前阵子内网某个业务模块的接口响应从 80ms 涨到 1.2s后端日志干干净净数据库也正常最后查出问题在带宽上——一台后台机在往外同步数据把上行占满了。当时用 tcpdump 抓了 200 多 MB 的包抓包工具打开直接卡死靠手工翻根本翻不完。后来换了个思路写脚本按五元组把包聚合成流按字节数排个序十分钟就锁定了那台机器。这份「网络流量分析器源码」做的就是这件事采集、解析、按流聚合、出统计报表一条链路走完输出直接是能拿去对账的 CSV。它适合需要在内网做流量画像、排查带宽占用、统计协议分布的开发和运维也适合拿它当学习抓包解析的练手基座——代码不长但每一层都能单独拎出来改。2. 技术栈与目录结构一张网卡到一张报表的模块划分拿到源码先别急着跑先看清楚它是怎么分层的。网络流量分析这类工具代码量不大但链路长分层不清楚的话改一个统计口径要动五个文件。这份源码的组织方式是把「包从哪来」「包是什么」「怎么统计」「结果给谁」拆成四段中间用生成器和普通字典传递数据没有引入消息队列也没有搞插件框架读起来比较直白。2.1 技术选型Python 抓包到底能不能扛住很多人一听 Python 抓包就摇头觉得性能不行。这个判断对一半。抓包解析的性能瓶颈不在语言在「每个包做多少次对象分配」。用纯 Python 逐层构造协议对象单核大概能到每秒一两万个包换成零拷贝风格的解析库能上到每秒几万到十几万真要跑到线速就得下沉到 C 或者用内核侧的过滤与采样能力。所以选型的正确问法是你的场景需要多少 pps以及你更缺性能还是更缺开发时间。这份源码默认走的是「Python 解析 内核 BPF 过滤」的组合量级上覆盖千兆内网的抽样分析、单机流量排查、离线 pcap 复盘完全够用。如果拿到的版本里还有一条dpkt的快路径那是给高吞吐场景准备的备选。三者的差别大致是这样方案典型吞吐量级开发成本适合场景scapy 逐层解析万级 pps最低字段名可读性最好离线复盘、协议调试、二次开发dpkt 手工拆包十万级 pps中要自己处理边界在线长跑、采样统计C/libpcap 或内核旁路线速高内存和生命周期都要自己管骨干链路、全量留存提示上表的数字只是量级参考实际受包长、CPU、是否开启混杂模式影响很大。评估自己的场景时用真实 pcap 回放跑一遍比看任何表格都准。我一般的做法是先用 scapy 版本把功能和统计口径跑通、结果和抓包工具对得上再决定要不要为性能换快路径。反过来先优化性能往往是白忙——因为大部分时候瓶颈根本不在解析而在你把包全存进内存了。2.2 目录结构与每个模块的职责源码的目录大体按四层切分模块之间是单向依赖上层可以调下层下层不认识上层。这个约束很关键它保证了你可以只替换采集层比如从网卡抓包换成读 pcap 文件而不动统计逻辑。traffic-analyzer/ ├── collector/ # 采集层在线抓包、离线 pcap 回放统一产出包迭代器 │ ├── live.py # 网卡实时采集 │ └── offline.py # pcap 文件流式读取 ├── parser/ # 解析层以太网帧 → IP → TCP/UDP产出五元组 │ ├── scapy_parser.py # 可读性优先的实现 │ └── fast_parser.py # 高性能快路径 ├── aggregator/ # 聚合层流表、时间窗口、空闲老化 │ └── flow_table.py ├── output/ # 输出层CLI 表格、CSV、JSON │ └── writer.py ├── rules/ # 过滤规则与应用识别注册表 │ └── apps.py └── cli.py # 入口参数解析把四层装配起来对应的职责边界用一张表说清楚更省事模块输入输出不该做的事collector网卡名或 pcap 路径原始包对象流不做协议判断不做统计parser原始包对象FlowKey、字节数、TCP 标志位不累计不落盘aggregator解析结果流记录字典不关心包从哪来output流记录字典表格、CSV、JSON不做计算只做格式化这个划分带来的直接好处是想加一种新的输出格式只动output/想支持 IPv6只动parser/想改成按源 IP 聚合而不是按五元组只动aggregator/的 key 生成。二次开发时按这条线找文件比全局搜索快得多。2.3 环境准备与第一次跑通依赖很轻一个虚拟环境加两个包就够。真正容易卡住的不是安装是权限——抓包要读原始套接字普通用户默认没有这个能力。# 建虚拟环境避免污染系统 Python python3 -m venv .venv source .venv/bin/activate # 核心依赖scapy 负责解析dpkt 在快路径里用 pip install scapy dpkt # 给解释器授抓包能力省掉每次 sudo # 注意 readlink -fvenv 里的 python 是软链直接对软链 setcap 无效 sudo setcap cap_net_raw,cap_net_admineip $(readlink -f $(which python3)) # 先跑离线模式不依赖网卡和权限验证代码链路通不通 python cli.py --pcap ./samples/demo.pcap --out ./result.csv --topn 20参数分三类改动时对应不同的模块--pcap传文件走离线回放不传就走在线采集这个分支在collector/里--out决定输出格式后缀是.csv走 CSV 写入器.json走 JSON 写入器--topn只影响输出阶段的截断不影响统计本身。第一次跑建议就用离线模式。原因很实际在线抓包时你没法控制网卡上流过的内容输出为空你分不清是代码有问题还是这段时间真没流量而离线 pcap 是确定的输入跑两遍结果必须一模一样这就是最好的自检。3. 核心链路拆解一个数据包从网卡到 CSV 的完整路径链路本身不复杂复杂的是每一层都会遇到「看起来能跑但其实不对」的写法。这一章按数据流顺序走一遍每段代码后面说清楚为什么这么写、参数怎么调。3.1 采集层把在线 sniff 和离线 pcap 收进同一个迭代器采集层最重要的设计决定是「用生成器而不是列表」。抓包函数如果返回列表意味着它要把所有包攒在内存里再交给你跑十分钟就能把机器吃干净。统一成生成器之后上层是for循环消费内存占用和数据量脱钩。# collector/__init__.py import scapy.all as scapy from scapy.utils import PcapReader def iter_packets(sourceNone, ifaceNone, bpfNone, count0): 统一的包来源。 source: pcap 文件路径传 None 表示在线采集 iface: 网卡名在线模式必填 bpf: BPF 过滤表达式尽量前置到内核减少用户态拷贝 count: 0 表示不限制 if source: # PcapReader 是流式读取不会把整个文件读进内存 with PcapReader(source) as reader: for i, pkt in enumerate(reader): if count and i count: break yield pkt return # 在线模式AsyncSniffer 边抓边消费storeFalse 是关键 sniffer scapy.AsyncSniffer(ifaceiface, filterbpf, storeFalse) sniffer.start() try: emitted 0 while not count or emitted count: pkt sniffer.results.get() # 队列抓一个取一个 emitted 1 yield pkt finally: sniffer.stop()storeFalse是这段代码里唯一不能省的参数。默认sniff()会把抓到的包存进一个列表并一直持有短时间跑没事长跑就是内存泄漏。改成storeFalse后包会被丢进一个队列消费端取走即释放。results.get()是阻塞的所以在线模式需要一个停止条件。实际使用里一般靠信号量中断finally里调stop()保证抓包线程能退出——少了这个CtrlC 之后进程会挂住不退。还有一个容易忽略的点bpf传的是tcp port 443这类表达式它会被下推到内核的过滤器里执行。也就是说这些包在进入 Python 之前就被丢掉了性能差别是数量级的。所以能用 BPF 表达的过滤条件千万别写成 Python 里的if。3.2 解析层scapy 版和 dpkt 快路径怎么选解析层只做一件事把一个原始包变成(五元组, 字节数, TCP 标志位)。听起来简单但这里有个隐性的性能陷阱——每次访问pkt[scapy.IP]都会触发一次层叠扫描。# parser/scapy_parser.py from collections import namedtuple import scapy.all as scapy FlowKey namedtuple(FlowKey, src dst sport dport proto) def parse_packet(pkt): 返回 (FlowKey, nbytes, tcp_flags)非 IPv4 返回 None。 # haslayer 比 scapy.IP in pkt 快后者会构造比较对象 if not pkt.haslayer(scapy.IP): return None ip pkt[scapy.IP] sport dport 0 flags 0 if ip.proto 6 and pkt.haslayer(scapy.TCP): tcp pkt[scapy.TCP] sport, dport, flags tcp.sport, tcp.dport, int(tcp.flags) elif ip.proto 17 and pkt.haslayer(scapy.UDP): udp pkt[scapy.UDP] sport, dport udp.sport, udp.dport else: # ICMP、GRE 等协议没有端口概念端口位补 0 便于统一聚合 pass return FlowKey(ip.src, ip.dst, sport, dport, ip.proto), len(pkt), flagslen(pkt)给的是整个以太帧长度包含链路层头。如果你的统计口径是「IP 层字节数」这里应该改成ip.len两者会差 14 字节每包在包长分布偏小的场景下误差能到 3% 以上。这个口径必须在开始统计前定死中途改会导致前后数据没法对比。当 pps 上到几万之后scapy 的对象构造开销就压不住了。这时候切到 dpkt 的快路径思路是「按固定偏移拆包不做完整的对象树」# parser/fast_parser.py import struct import socket import dpkt def parse_fast(buf): buf 是以太网帧的 bytes只解析到传输层失败返回 None。 if len(buf) 34: return None ethertype struct.unpack(!H, buf[12:14])[0] if ethertype ! 0x0800: # 只处理 IPv4IPv6 走另一条分支 return None ihl (buf[14] 0x0F) * 4 # IP 头长度单位是 4 字节 proto buf[23] src socket.inet_ntoa(buf[26:30]) dst socket.inet_ntoa(buf[30:34]) sport dport 0 if proto in (6, 17) and len(buf) 14 ihl 4: sport, dport struct.unpack(!HH, buf[14 ihl: 14 ihl 4]) return (src, dst, sport, dport, proto), len(buf)这里用struct.unpack而不是 dpkt 的完整解析是因为在只关心五元组的场景下构造dpkt.ethernet.Ethernet对象树的成本远高于直接切片。代价是你要自己处理 IP 选项ihl大于 5 的情况和分片包——分片包的传输层头只在第一片里有这里没做判断会拿到脏数据。注意快路径是拿可维护性换性能。如果你后面要加应用层识别老老实实回到 scapy 版本或者用 dpkt 的完整解析别在手工切片上加逻辑那种代码三个月后自己都看不懂。3.3 聚合层五元组流表与空闲老化流表就是一个字典key 是五元组value 是累计量。真正的技术在过期淘汰——如果不淘汰一个跑一天的进程流表条数只增不减最后还是 OOM。# aggregator/flow_table.py import time class FlowTable: def __init__(self, idle_timeout60): self.flows {} self.idle_timeout idle_timeout # 秒超过这个时间没新包就认为流结束 def add(self, key, ts, nbytes, flags): rec self.flows.get(key) if rec is None: # 首包时间用包自带的时间戳不用 time.time()否则离线回放时速率全错 rec self.flows[key] { first: ts, last: ts, bytes: 0, packets: 0, syn: 0, fin: 0, rst: 0, } rec[last] ts rec[bytes] nbytes rec[packets] 1 if flags 0x02: rec[syn] 1 if flags 0x01: rec[fin] 1 if flags 0x04: rec[rst] 1 def expire(self, now): 返回已结束的流并从活动表里摘掉。 dead [k for k, v in self.flows.items() if now - v[last] self.idle_timeout] return [self.flows.pop(k) for k in dead]idle_timeout是最需要按场景调的参数。设成 5 秒长连接会被切成很多段单条流的平均速率被低估设成 600 秒短连接会一直挂在内存里流表膨胀。内网 HTTP 为主的环境60 秒是个不错的起点。判断依据是观察expire每次返回的条数如果长期是 0说明超时太长如果远大于新增条数说明太短。TCP 标志位的计数是为了后面判断连接质量。一个只有 SYN 没有后续的流通常是扫描或者目标不可达SYN 数远大于 FIN 数的流说明有大量连接没有正常关闭。这类信号在排查「连接数莫名其妙很高」的问题时特别有用比看总字节数直观。3.4 指标与输出bps、pps、Top N 和落盘格式统计口径定了输出就是格式化的事。这里唯一需要小心的是单位换算——字节和比特差 8 倍写错了数据看起来「差不多」但结论会完全反过来。# output/writer.py import csv def to_rows(flows, topn20): rows [] for key, rec in flows.items(): dur max(rec[last] - rec[first], 1e-6) # 防止除零 rows.append({ src: key[0], dst: key[1], sport: key[2], dport: key[3], proto: key[4], bytes: rec[bytes], packets: rec[packets], bps: rec[bytes] * 8 / dur, # 比特每秒注意乘 8 pps: rec[packets] / dur, duration: round(dur, 3), }) rows.sort(keylambda r: r[bytes], reverseTrue) return rows[:topn] def dump_csv(rows, path): if not rows: return with open(path, w, newline, encodingutf-8) as f: writer csv.DictWriter(f, fieldnameslist(rows[0].keys())) writer.writeheader() writer.writerows(rows)max(dur, 1e-6)这个保护不是洁癖。单个包的流比如一次 UDP 查询首包和末包时间戳相同dur为 0直接除会抛ZeroDivisionError而这类流在 DNS 密集的环境里非常常见。用 1e-6 兜底后这类流的 bps 会是一个非常离谱的大数——这反而是个信号提醒你「单包流的速率没有意义看字节数就行」。排序按字节数降序、截断到 topn是为了输出可读。真正做容量规划时Top N 往往不够用因为流量分布是长尾的前 20 条可能只占 30%。这时候把topn设成一个很大的数比如 100000拿到全量再自己用表格工具做透视比在代码里加各种分组逻辑划算。4. 过滤与性能把无用的包挡在解析器之前跑通了之后下一个问题必然是「怎么跑得更快」。优化的顺序很重要先做不写代码就能拿到的收益再考虑改架构。按投入产出排第一是过滤第二是采样第三才是换解析器。4.1 BPF 表达式内核态丢包比用户态快一个数量级BPF 过滤器是在内核里执行的不匹配的包根本不会拷贝到用户态。这意味着它可以让你在一张跑满的网卡上只处理 5% 的流量而 CPU 占用降到接近 5%。相比之下在 Python 里写if pkt[IP].proto ! 6: continue是完全相反的做法——包已经拷贝过来了对象已经构造了你才决定丢掉。# 只看某个网段进出的 TCP排除掉本机回环和某个大流量端口 python cli.py --iface eth0 \ --bpf tcp and net 10.20.0.0/16 and not port 9000 \ --out live.csv --topn 50 # 先验证表达式对不对用 -c 限制包数避免脚本写错时刷屏 tcpdump -i eth0 -c 20 -nn tcp and net 10.20.0.0/16 and not port 9000养成先用tcpdump -c验证表达式的习惯能省掉大量「脚本跑了但输出是空的」的困惑。表达式语法错误时抓包库有的版本会直接抛异常有的版本会静默忽略过滤器然后抓全量——后者更坑你以为在过滤其实在裸奔。几个高频组合值得记一下not port 22排除管理流量避免干扰tcp[tcpflags] tcp-syn ! 0只看建连包用来快速判断谁在扫greater 500只看大包用于定位大文件传输。这些条件组合起来能覆盖八成排查场景。4.2 采样、环形缓冲与多进程分流的取舍当单核真的压不住时第一个念头通常是「多进程」。但在抓包场景里多进程的收益没有想象中大多个进程同时从同一张网卡抓包每个进程都会拿到一份完整拷贝总 CPU 消耗反而上升。正确的做法是「一次抓包分发处理」或者干脆上采样。采样要解决的核心问题是「可外推」。随机丢包会导致每次跑出来的结果都不一样没法做趋势对比而按固定步长采样每 N 个包取 1 个在流量平稳时还行遇到突发就会系统性偏差。比较稳的做法是用五元组做哈希只保留哈希值落在某个区间的包# 一致性采样同一个流要么全采要么全不采便于按流聚合 import hashlib def sampled(key, mod10): 保留 1/mod 的流同一条流每次判断结果一致。 h hashlib.md5((%s|%s|%s|%s|%s % key).encode()).digest() return int.from_bytes(h[:4], big) % mod 0按流采样而不是按包采样是因为统计目标是「每条流多少字节」。按包采会把一条流的字节数打成 1/N虽然乘回去也能估总量但流级别的 Top N 排序会失真——大流和小流被采到的概率一样短流反而更容易被完整采到。环形缓冲是另一条路只保留最近 N 秒的包更早的丢掉。它适合「出问题之后回溯最近发生了什么」的场景内存占用可控。实现上用一个collections.deque(maxlenN)就够代价是你能回答的问题变窄了——只有时间窗口内的问题能答。4.3 结果怎么验和 tcpdump 对账任何性能优化之后都必须对账否则你不知道自己优化的是速度还是正确性。最省事的对账方式是拿同一个 pcap 文件分别用这套代码和抓包工具自带的统计功能算总包数、总字节数能对上再谈别的。# 抓包工具侧的基准总包数 tcpdump -r demo.pcap -nn 2/dev/null | wc -l # 分析器侧不加过滤全量统计后求和 python cli.py --pcap demo.pcap --out all.csv --topn 1000000两边数字对不上先查三件事分析器是不是只统计了 IPv4tcpdump默认全协议帧长口径是不是含链路层头有没有开启采样。这三个原因能解释九成的差异。对上了再回头看看优化前后的 bps 曲线形状有没有变——形状变了说明采样或过滤引入了偏差数字对上只是巧合。5. 避坑与排查五个让我返工的细节这一章是血泪经验。前面几章的代码看着都能跑但真正部署到一台有流量的机器上翻车的地方全在这些边角。5.1 采集阶段权限和内存的两个坑现象一本地跑得好好的换台机器就报Operation not permitted加了sudo能跑但换个用户又不行。原因是抓包需要CAP_NET_RAW能力普通用户默认没有。用sudo是绕过而不是解决——它会让输出文件的属主变成 root后续再想用普通用户处理这些文件又是一堆权限问题。正确的做法是把能力授给解释器本身setcap cap_net_raw,cap_net_admineip python 路径。这里有个隐蔽的坑虚拟环境里的python是个软链接setcap挂在软链上不生效必须用readlink -f解析到真实路径。另外重新安装或升级 Python 之后这个能力会丢失需要重新授予别以为是代码退化了。现象二在线模式跑几个小时后内存一路涨到 OOM离线模式完全正常。八成是sniff()用了默认参数。默认storeTrue会把所有包攒在一个列表里直到抓包结束才返回——在线模式永远不会结束于是列表无限增长。另一个可能是AsyncSniffer的storeFalse设了但消费端太慢队列是无界的生产速度超过消费速度队列一样会涨。判断方法是看内存增长曲线是线性还是阶梯状线性的基本都是队列积压。解决办法是给队列设上限、满了就丢包并计数宁可丢包也不能把机器拖死。现象三统计出来的总字节数和网卡计数器差了将近一倍而且抓到的包里有很多 64KB 的超大帧。不是代码算错了是网卡在做分段卸载。开启 GRO/GSO 之后内核把多个小包聚合成一个超大帧交给上层抓包抓到的就是这个聚合后的帧。它会让包数统计严重偏低、平均包长严重偏高。解决方式是在采集前关掉对应网卡的卸载特性或者接受这个偏差但在报告里注明「包数为聚合后口径」。这个坑最坑的地方在于它不影响字节总数只影响包数和平均包长所以只看总量的对账发现不了。5.2 解析与统计阶段三个坑现象四离线回放一个 10 分钟的 pcap算出来的 bps 高得离谱和实际带宽对不上。去查时间戳的来源。人在写统计代码时很自然会写time.time()但那拿到的是「处理这个包的时刻」不是「这个包到达网卡的时刻」。离线回放时所有包在几秒内被处理完算出来的速率自然是实际值的上百倍。正确做法是统一用包自带的时间戳scapy 里是pkt.timedpkt 里是读取时返回的ts。这条规则对所有「事后分析」类工具都成立——时间基准必须来自数据本身不能来自运行环境。现象五同一个 TCP 连接在结果里出现两条记录方向相反的字节数分列两行。这不算 bug是五元组本身有方向。一条连接 A:1234 → B:80 和 B:80 → A:1234 是两个不同的 key分别累计是对的。但如果你的目标是「按会话统计总流量」就得做双向归一把源和目的按某种规则比如 IP 字符串比较、端口大小排序后重新组 key。def normalize(key): 双向归一把 (A,B) 和 (B,A) 映射到同一个 key。 src, dst, sport, dport, proto key if (src, sport) (dst, dport): return key return (dst, src, dport, sport, proto)代价是丢失方向信息。所以别在解析层做归一而是在输出层按需做——原始数据保留方向展示时再决定怎么合。反过来把方向丢了再想还原是没有后悔药吃的。6. 二次开发加一个自定义协议识别器真正用起来之后迟早会碰到「我想知道某个私有端口上跑的是什么」。别去改解析主循环那是最容易被后续更新冲突的地方。源码里rules/目录的设计意图就是给你挂扩展的维护一张「匹配函数 → 应用名」的注册表解析完之后按优先级逐条试命中就写进流的标签里。# rules/apps.py import struct REGISTRY [] # [(优先级, 匹配函数, 应用名)] def register(priority): def deco(fn): REGISTRY.append((priority, fn, fn.__name__.replace(match_, ))) REGISTRY.sort(keylambda x: -x[0]) # 优先级高的先试 return fn return deco register(100) def match_private_rpc(payload): 私有 RPC前 4 字节是魔数 0xRPC1后跟 2 字节版本号。 if len(payload) 6: return None magic, ver struct.unpack(!IH, payload[:6]) if magic 0x52504331: # RPC1 return {app: private-rpc, version: ver} return None关键参数是priority数值越大越先匹配。给自己写的识别器留 100 以上的空间内置的通用识别HTTP 请求行、TLS 握手头放在 50 以下这样自定义规则永远有优先权。返回None表示不匹配继续试下一条别抛异常——一条脏包不该让整个统计流程中断。验证方法很朴素但有效构造一个最小的 pcap里面只放三条流一条是你的私有协议、一条是普通 HTTP、一条是随机字节回放之后断言输出里恰好只有一条被打上private-rpc标签。把这个用例固化成脚本以后每次改识别逻辑都跑一遍。我吃过亏的地方是只测了命中路径忘了测「不命中时会不会误判」——结果一个长度判断写反把所有大包都标成了私有协议报表看起来还挺正常。从那以后我每次加识别规则都强制走一遍「命中 / 不命中 / 畸形包」三个用例缺一个都不合并。抓包分析这类工具的可怕之处在于它不会报错只会安静地给你一份看起来合理的错误数据。希望帮到你。本文还有配套的精品资源点击获取
