MiroFish：基于Python的轻量级局域网流量镜像分析工具

1. 为什么我要写MiroFish现有工具的“最后一公里”问题先说个真实场景。前阵子公司办公网间歇性卡顿领导让我排查是不是有设备在偷偷跑大流量。我按老套路在核心交换机上做了个端口镜像把Uplink口的流量复制到分析口然后tcpdump一堆pcap文件拖回电脑用Wireshark慢慢翻。翻了一个下午眼睛都快瞎了——两千多万个包我要找的特征是“凌晨三点到五点之间某个MAC地址对外发起的HTTP POST请求”Wireshark的过滤器能查但查出来的结果依然要一条条看而且多文件之间来回切效率极其低下。这就是我写MiroFish的起点。MiroFish这个名字取自Mirror镜像和Fish鱼的组合它像鱼群一样安静地在网络里游动把交换机上镜像过来的数据包“捞”出来分析。项目定位很清楚——一个跑在Linux小主机上的轻量级局域网流量镜像分析工具专门解决从“抓到包”到“发现异常”这段最后一公里的事。它不追求替代Wireshark做深度协议解码也不打算做成Zeek那样庞大的入侵检测框架核心功能就四个持续抓包、按规则过滤、命中告警、落库查证。什么人适合用这个东西两类。一类是像我这样需要经常处理局域网流量排查的网络运维另一类是刚开始学协议分析的开发者——用MiroFish的规则文件练手能比直接在Wireshark里点点点更快理解“流量特征”是怎么回事。整个项目大概两千行Python依赖只有scapy和一个SQLite驱动树莓派4B就能跑。下面我把设计思路、核心代码、实际验证结果、踩过的坑全部摊开讲想自己搭一套的可以直接照着做。2. 总体设计轻量、低依赖、能跑在树莓派上的协议分析工具2.1 技术选型为什么是Python加scapy而不是C或者go做流量分析工具技术选型上天然有三条路C语言直接调libpcap、Go语言用gopacket、Python用scapy。我最终选了Python加scapy原因很实际。第一开发效率。MiroFish最核心的价值在规则逻辑和分析视角不在抓包引擎本身。用scapy的sniff函数几行代码就能拿到带完整协议栈解析的包对象用C写同样的功能光处理数据链路层到传输层的指针偏移就得折腾半天。第二生态匹配。scapy解DNS、HTTP、TLS等应用层协议时能直接以字段方式访问写规则的时候人脑负担小。比如说我要判断一个DNS响应里是否包含特定域名用C得自己算DNS报文头的标志位、问题计数、回答计数然后逐个解析压缩域名指针而scapy一行pkt[DNS].qd.qname就搞定了。第三部署成本。scapy对Linux发行版的支持极好树莓派上apt install python3-scapy直接装不需要像libpcap那样考虑交叉编译。当然代价也有抓包性能上限不如纯C方案。实测下来在树莓派4B单核上MiroFish稳定处理每秒3到4万个包超过这个量就开始丢包。对办公网这种规模完全够用但如果是核心交换机上全量镜像几千兆的流量那还是老老实实用Zeek或者ntopng。2.2 数据链路的整体设计采集、解析、规则、存储四层分离MiroFish的代码结构是典型的四层管道模型每一层只管自己的事互不干扰层级模块职责采集层collector.py从网卡或pcap文件读取数据包做基础过滤交给上层解析层parser.py把scapy包对象拆解为统一格式的Flow记录和Event记录规则层engine.py加载YAML规则对Event记录做匹配命中后生成Alert存储与告警层storage.py,alerter.pyAlert和Flow写入SQLite同时通过Webhook推送通知层与层之间用queue.Queue传递数据每层是独立的Python线程。这样设计的好处是采集层被网卡中断打得再狠也不会阻塞规则匹配——前提是队列长度设置合理我默认用的是maxsize10000满了就丢弃并记录drop计数避免内存暴涨。有个细节很重要规则引擎不应该直接拿scapy的对象做匹配。scapy对象解析是惰性的访问字段时才真正做内存读取如果规则里每个条件都去触发一次解析性能损耗非常惊人。我把解析层做成了“一次性投影”——每个包进来后立即把它所有可能用到的字段抽取成普通Python字典后续规则匹配只查字典不再碰原始包。这个决策让整体性能提升了大约三倍。2.3 规则引擎的设计思路把匹配逻辑交给YAML而不是代码写这个项目之前我想明白一个问题用户排查流量时问的最多的根本不是“这个包的完整协议树长什么样”而是“有没有出现满足某几个条件的包”。所以MiroFish把核心交互收敛到了规则文件上。每个规则是一个YAML片段大致长这样- name: detect_dns_query_for_suspicious_domain enabled: true filter: udp_port_53 conditions: - field: dns_qname op: endswith value: .xyz - field: dns_qr op: eq value: 0 action: type: alert severity: medium message: Detected DNS query for .xyz domainfilter字段是预过滤条件在采集层就执行比如只保留UDP端口53的包可以大幅减少进入规则引擎的数据量。conditions是精确匹配条件多个条件之间是AND关系。op支持eq、contains、startswith、endswith、gt、lt、regex七种操作符。我把规则引擎做成YAML驱动之后最大的感受是排查新问题时不用再改代码、重启服务了。现场发现可疑流量特征写一条规则扔进目录热加载后十秒钟内就能看到是否命中。对于“搞不清到底该查什么”的初期阶段这种试错效率提升特别明显。3. 核心模块实现从抓包到告警的全链路代码拆解3.1 采集模块不能只用scapy的sniff裸奔scapy的sniff函数用起来很简单但直接在生产环境跑会有两个隐患一是它默认会做完整的协议解析导致CPU开销偏高二是它的回调函数在抓包线程里执行如果回调处理太慢内核buffer会被塞满整个抓包会话直接崩。MiroFish的采集器把sniff的参数精细调过一遍核心代码是这样的from scapy.all import AsyncSniffer, Ether, IP, IPv6, TCP, UDP, DNS import queue class PacketCollector: def __init__(self, interface, bpf_filter, output_queue, max_packets0): self.interface interface self.bpf_filter bpf_filter self.queue output_queue self.sniffer None self.stopped False self.count 0 self.dropped 0 self.max_packets max_packets def _packet_handler(self, pkt): if self.max_packets 0 and self.count self.max_packets: return try: self.queue.put_nowait(pkt) self.count 1 except queue.Full: self.dropped 1 def start(self): self.sniffer AsyncSniffer( ifaceself.interface, filterself.bpf_filter, prnself._packet_handler, storeFalse ) self.sniffer.start() def stop(self): if self.sniffer: self.sniffer.stop()注意几个关键点storeFalse是必须的。sniff默认会把所有抓到的包存在内存里几百万个包下去内存直接爆掉MiroFish只需要实时回调不需要保留原始包。prn回调里用put_nowait而不是put。规则引擎处理速度跟不上时丢弃新包比无限阻塞要好——阻塞会直接传导到抓包线程导致内核socket buffer溢出丢包那就不是丢几个包的问题了是整段流量都断片。BPF过滤器在采集层就用上了。默认过滤器是tcp or udp or icmp把ARP、LLDP这些管理帧全部挡在外面。这些帧对协议分析用处不大放进来只会白白消耗CPU。3.2 解析层的字段投影把scapy对象变成一张扁平的字典规则引擎不应该关心pkt[IP].src这种层级访问方式更不应该处理pkt[Ether].src的Ethernet头部跟IP层MAC地址之间谁是谁的问题。解析层做的事情是把这些投影成一张扁平的字段表def project_packet(pkt): fields { time: pkt.time, len: len(pkt), } if pkt.haslayer(Ether): fields[eth_src] pkt[Ether].src fields[eth_dst] pkt[Ether].dst if pkt.haslayer(IP): fields[ip_src] pkt[IP].src fields[ip_dst] pkt[IP].dst fields[ip_proto] pkt[IP].proto fields[ip_ttl] pkt[IP].ttl elif pkt.haslayer(IPv6): fields[ip_src] pkt[IPv6].src fields[ip_dst] pkt[IPv6].dst fields[ip_proto] pkt[IPv6].nh fields[ip_ttl] pkt[IPv6].hlim if pkt.haslayer(TCP): fields[tcp_sport] pkt[TCP].sport fields[tcp_dport] pkt[TCP].dport fields[tcp_flags] pkt[TCP].flags fields[tcp_seq] pkt[TCP].seq fields[tcp_ack] pkt[TCP].ack elif pkt.haslayer(UDP): fields[udp_sport] pkt[UDP].sport fields[udp_dport] pkt[UDP].dport if pkt.haslayer(DNS): dns pkt[DNS] fields[dns_id] dns.id fields[dns_qr] dns.qr fields[dns_opcode] dns.opcode fields[dns_rcode] dns.rcode if dns.qd: fields[dns_qname] dns.qd.qname.decode(utf-8, errorsignore) if dns.an: ans dns.an fields[dns_an_count] dns.ancount if ans.type 1 and hasattr(ans, rdata): fields[dns_an_rdata] ans.rdata return fields为什么要把TCP flags单独投影成tcp_flags字段因为在规则里eq、regex这些操作符只能处理字符串和数字scapy的FlagValue对象无法直接比较。投影的时候把它转成字符串规则引擎就能写tcp_flags: contains R这种自然条件了。投影后解析层把包对象转成两条记录一条是Flow记录以五元组为key的会话聚合一条是Event记录单包原始投影。规则引擎只消费Event记录而要查看某个IP的整体行为时直接查Flow表。这个双轨设计是后来加上的——一开始只有Event记录排查时发现单个包看不出问题模式必须聚合到会话维度。3.3 告警与存储SQLite的WAL模式和Webhook推送存储层选了SQLite而不是MySQL或者PostgreSQL理由就一个MiroFish是单机部署的工具SQLite的零配置特性让它在树莓派上格外省心。但要踩过坑才知道——SQLite的默认回滚日志模式在频繁写入时会产生剧烈的磁盘I/O抖动而且多个线程同时写会报database is locked。解决办法是两件事配合使用。一是让所有写入操作收敛到单一写入线程通过队列接收来自规则引擎的Alert和Flow记录绝对不允许两个线程同时执行INSERT。二是打开WAL模式conn.execute(PRAGMA journal_modeWAL) conn.execute(PRAGMA synchronousNORMAL)WAL模式最大的好处是读写不互斥读流量的同时还能执行查询不会被写入阻塞。synchronousNORMAL则是在WAL模式下相对安全地降低fsync频率换取写入吞吐。这两个设置搭配实测SQLite的写入能力从每秒几百条提升到三千条以上能扛住中等规模办公网的Alert生成频率。告警推送我用的是通用Webhook方式def send_alert(alert): if not alert.webhook_url: logging.warning(fAlert {alert.id} has no webhook_url configured) return payload { msg_type: text, content: f[MiroFish] {alert.severity}: {alert.message} } try: resp requests.post(alert.webhook_url, jsonpayload, timeout5) resp.raise_for_status() except Exception as e: logging.error(fFailed to send alert {alert.id}: {e})具体接什么Webhook你随意钉钉、企业微信、Slack都行只要提供一个可用的URL。推送失败不影响主流程最多记一条日志。告警这东西宁可漏推一条也不能让推送阻塞了采集分析主链路。4. 真刀真枪的验证构造流量、对比tcpdump、看性能拐点4.1 用scapy构造测试流量不依赖真实环境也能验证规则写代码是一回事验证代码是另一回事。真实网络流量不可控没法精准验证某条规则到底有没有漏报。我把测试环境搭在一台Ubuntu 22.04虚拟机上使用scapy自己构造测试流量再送到MiroFish的分析网卡上。构造一个DNS查询包的代码很简单from scapy.all import Ether, IP, UDP, DNS, DNSQR, sendp packet Ether(src02:00:00:00:00:01, dstff:ff:ff:ff:ff:ff) / \ IP(src192.168.50.10, dst8.8.8.8) / \ UDP(sport54321, dport53) / \ DNS(id1000, qr0, qdDNSQR(qnameexample.xyz)) sendp(packet, ifaceeth1, count1)我特意把域名设为example.xyz因为规则文件里有一条专门检测.xyz域名的DNS查询。构造流量发出去之后等几秒钟查MiroFish的SQLite库sqlite3 alerts.db select * from alerts;看到一条severity为medium的告警记录被写进去了规则引擎的整条链路就算验证通过。这个“构造流量-发送-查结果”的测试模型在整个开发过程中帮我省了无数时间。每写完一条新规则都先构造正例、反例各一组数据确保命中与否都符合预期再扩散到其他规则。4.2 性能拐点每个包三微秒是分水岭性能测试用的是虚拟机里的一个脚本它不断发送混合流量包同时用MiroFish的--stats参数每秒输出一次处理包数和丢弃数。我记录了几组关键数据发送速率包/秒处理速率包/秒丢弃率CPU单核占用5,0005,0000%22%10,00010,0000%41%20,00020,0000%78%40,00032,00020%100%拐点非常清晰。包速率高于每秒3万个之后处理链路的耗时开始呈现指数增长——因为Python的GIL在大量对象创建和字段访问时竞争加剧队列也开始堆积。这个测试结果直接影响了我的部署建议MiroFish适合处理中小规模的汇聚流量比如办公室出口、仓库分支、校园网的某个网段不适合放在核心机房镜像全量流量。如果确实需要处理更高吞吐我的做法是把采集层拆出去利用交换机的流量过滤功能比如只镜像特定VLAN先降量再分析。分析工具最忌讳贪心抓到十万个包但每个都来不及细看不如精准抓一千个全部看清。4.3 树莓派上72小时连续运行的真实表现性能测试都是在虚拟机上做的真正的考验是搬到树莓派4B上连续跑72小时。我把MiroFish部署在一个4GB内存版本的树莓派上网卡接的是交换机镜像端口镜像了办公室一半员工的流量。运行结果非常有意思内存占用非常稳定始终保持在180MB到220MB之间。队列在流量高峰期偶尔堆积但因为是预定义的10000上限满了就丢弃并计数没有出现内存泄漏。SQLite数据库三天长到了1.2GB——这个增速说明Flow记录的量比想象中大很多因为每个TCP会话即使没有任何规则命中也会在结束时写一条Flow记录落库。磁盘写入量是唯一需要认真对待的问题。我把数据库放到了树莓派的外接USB固态盘上如果用机器本身的TF卡这个写入频率大概率会把卡写穿。这个经验后来我写到了项目README里存储务必放在SSD或者机械硬盘上不要放SD卡和U盘。5. 踩坑记录流量分析工具开发中最容易翻车的六个细节5.1 混杂模式、root权限、BPF过滤器“过于严格”第一个坑是权限。在Linux上直接用scapy抓包必须以root身份运行。用普通用户跑scapy启动时一脸茫然地告诉你”Permission denied“。解决方法是安装时给Python解释器加cap_net_raw、cap_net_admin两个capabilitysudo setcap cap_net_raw,cap_net_adminep $(which python3)第二个坑是网卡不是混杂模式。交换机的镜像端口把流量吐给树莓派的某个网卡但Linux网卡默认只接收目的MAC是自身的帧。如果忘了开启混杂模式MiroFish抓到的只有广播帧和发给自己IP的流量分析结果会完全失真。scapy的sniff函数不会自动帮你开启混杂模式需要在启动采集器之前手动设置sudo ip link set eth0 promisc on第三个坑是BPF过滤器过于严格。一开始我用tcp or udp过滤一切后来发现某些交换机发出的VLAN Tagged帧802.1Q根本无法通过这个过滤器。排查半天才发现——BPF过滤器匹配的是裸以太网帧VLAN Tag会改变包头的偏移导致tcp关键字匹配不到。解决方法是把过滤条件改成vlan and tcp or vlan and udp or tcp or udp确保带VLAN标签的流量也能放进来。这个坑极其经典新学者十有八九会踩。5.2 时间戳精度造成的排序混乱scapy的pkt.time是浮点数记录的是抓包时刻的系统时间。单个包看没问题但把它存SQLite的REAL字段时就有意思了——SQLite的浮点数精度在纳秒级而Python的浮点精度在微秒级存储时会出现微小的舍入差异。当时我遇到的现象是查出来的记录按时间排序后部分相邻包的时间戳竟然是“倒挂”的——先抓的包显示的时间比后抓的包晚了几微秒。原因在于包装进队列时pkt.time已经在抓包线程里被取出来了但入队后等规则引擎处理时真实时间已经前进了一截。两个包几乎同时到达时内核的时间戳精度不够导致先入队的包时间戳反而更小。解决方案是统一用time.time()在解析层重新打一个高精度时间戳丢弃scapy自带的pkt.time。虽然少了内核态纳秒级精度但至少保证“先入队的时间戳一定更早”这个排列逻辑是对的。流量分析不是硬件抓包器微秒级精度完全够用一致性比精度更重要。5.3 SQLite写入抖动导致丢包最开始我让规则引擎命中后立即写SQLite每命中一条就INSERT一条。结果流量一上来SQLite的写入速度跟不上WAL文件疯狂膨胀磁盘I/O被打满然后内核socket buffer溢出采集线程大面积丢包。后来我改成了批量插入策略把待写入的Alert和Flow记录先攒在内存列表里每500条或者5秒钟先到先写批量执行一次事务。实测同样流量下SQLite的写入次数从每秒几百次降到了每5秒一次磁盘I/O完全平稳。这是整个项目里收益最大的性能优化没有之一。5.4 大小端字节序TCP flags解析的玄学TCP flags在scapy里是个FlagValue对象转成字符串时会得到类似’A’ACK、’PA’PSHACK这样的表示。但有一次我写了个判断条件tcp_flags: contains S怎么都不命中。手动查看解析出来的tcp_flags字段发现TCP握手包显示的是’S’明明包含Scontains判断却返回false。查了半天才发现contains对字符串的判断是区分大小写的而scapy的FlagValue转字符串时不管字母本身大小写一律以大写输出。我的规则文件里写的是contains s自然永远匹配不上。修正方式是在解析层统一把flags转成大写fields[tcp_flags] str(pkt[TCP].flags).upper()这个坑提醒我规则匹配的输入输出格式必须在一开始就明确约定好否则排查问题的过程会变成“为什么我的规则不生效”的玄学问题。我把所有枚举类型字段的格式约定写进了项目的README比如tcp_flags一律大写、dns_qr用0和1、ip_proto用数字编号避免使用者因为格式差异产生困惑。6. 项目扩展方向与个人使用体会6.1 从检测工具走向轻量级网络可视化MiroFish现在的版本做的是“检测-告警-存储”这条流水线。但我实际用了两个月之后最大的不满足来自“看到异常但看不到全景”。告警告诉我某个IP在过去一小时发起了大量DNS查询但我还想知道这个IP在所有流量中的占比是多少它主要和哪些目标通信一天之内的流量趋势是什么样这就引出了下一步最值得做的扩展在Flow记录的基础上加一个时间序列聚合模块。把五元组聚合结果按分钟维度写入另一个表然后暴露一个HTTP接口前端用简单的折线图把数据可视化出来相当于给MiroFish装上一个“流量监控面板”。这件事的工程量不算大底层的Flow数据已经齐全缺的是聚合逻辑和展示层。另一个值得尝试的方向是PCAP导出。现在规则命中时会记录原始包的基本字段但如果有取证需求必须拿到原始包的完整内容。我计划在命中告警时把原始包同时写入一个按日期滚动的pcap文件这样既能保证轻量常态运行又能在需要取证的时候拿出完整证据链。6.2 我的几条实际使用心得我自己把MiroFish跑在办公室快三个月聊几个真实感受。第一规则要写“巧”而不是写“多”。我一开始照着网上各种入侵检测规则库抄了几百条结果告警轰炸到根本看不过来。后来精简到二十多条每条都是针对我们办公网实际观察到的异常行为特征——比如非工作时间的大流量外发、DNS请求中出现未知顶级域、某个IP在短时间内发起了大量TCP SYN。量少了每条规则的准确率反而高了很多。第二SQLite数据要定期清理。Flow记录增长太快我写了个每天凌晨自动执行的任务删除30天前的数据。数据库文件不要一味撑大查询速度会劣化。第三一定要加drop计数监控。MiroFish的--stats参数每秒打一行日志我会让巡检脚本盯着dropped字段一旦丢包率持续高于1%立即去查镜像口流量是不是超出了设备处理能力。丢包是流量分析的隐形杀手——它不会报错但会让你错过真正要发现的东西。如果有人想基于这个项目做二次开发我的建议是从规则引擎下手。它是最核心也最灵活的部分把YAML规则文件改成支持OR逻辑、加上正则分组、增加时间窗口状态检测比如“5分钟内超过10次才触发”这些扩展会让MiroFish从单包匹配升级到会话行为分析能力会上一个台阶。代码结构上四层分离的设计已经给你留好了扩展位改起来不会伤筋动骨。整个MiroFish从想法到落地前前后后两周多的时间。它不是什么高深的东西但解决了我实际工作里的一个具体痛点。如果你也受够了“抓到包但看不出问题”的尴尬处境完全可以自己动手搓一个类似的小工具——流量分析的门槛远没有想象中那么高关键是找到自己真正需要它做的事。