企业官网建设流程全解析

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的代码结构是典型的四层管道模型，每一层只管自己的事，互不干扰：

层与层之间用queue.Queue传递数据，每层是独立的Python线程。这样设计的好处是：采集层被网卡中断打得再狠，也不会阻塞规则匹配——前提是队列长度设置合理，我默认用的是maxsize=10000，满了就丢弃并记录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 domain"

filter字段是预过滤条件，在采集层就执行，比如只保留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_packets=0): 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( iface=self.interface, filter=self.bpf_filter, prn=self._packet_handler, store=False ) self.sniffer.start() def stop(self): if self.sniffer: self.sniffer.stop()

注意几个关键点：

store=False是必须的。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", errors="ignore") 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_mode=WAL") conn.execute("PRAGMA synchronous=NORMAL")

WAL模式最大的好处是读写不互斥，读流量的同时还能执行查询，不会被写入阻塞。synchronous=NORMAL则是在WAL模式下相对安全地降低fsync频率，换取写入吞吐。这两个设置搭配实测，SQLite的写入能力从每秒几百条提升到三千条以上，能扛住中等规模办公网的Alert生成频率。

告警推送我用的是通用Webhook方式：

def send_alert(alert): if not alert.webhook_url: logging.warning(f"Alert {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, json=payload, timeout=5) resp.raise_for_status() except Exception as e: logging.error(f"Failed 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(src="02:00:00:00:00:01", dst="ff:ff:ff:ff:ff:ff") / \ IP(src="192.168.50.10", dst="8.8.8.8") / \ UDP(sport=54321, dport=53) / \ DNS(id=1000, qr=0, qd=DNSQR(qname="example.xyz")) sendp(packet, iface="eth1", count=1)

我特意把域名设为example.xyz，因为规则文件里有一条专门检测.xyz域名的DNS查询。构造流量发出去之后，等几秒钟查MiroFish的SQLite库：

sqlite3 alerts.db "select * from alerts;"

看到一条severity为medium的告警记录被写进去了，规则引擎的整条链路就算验证通过。这个“构造流量-发送-查结果”的测试模型，在整个开发过程中帮我省了无数时间。每写完一条新规则，都先构造正例、反例各一组数据，确保命中与否都符合预期，再扩散到其他规则。

4.2 性能拐点：每个包三微秒是分水岭

性能测试用的是虚拟机里的一个脚本，它不断发送混合流量包，同时用MiroFish的--stats参数每秒输出一次处理包数和丢弃数。我记录了几组关键数据：

拐点非常清晰。包速率高于每秒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两个capability：

sudo setcap cap_net_raw,cap_net_admin+ep $(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’（PSH+ACK）这样的表示。但有一次我写了个判断条件tcp_flags: contains "S"，怎么都不命中。手动查看解析出来的tcp_flags字段，发现TCP握手包显示的是’S’，明明包含S，contains判断却返回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从想法到落地，前前后后两周多的时间。它不是什么高深的东西，但解决了我实际工作里的一个具体痛点。如果你也受够了“抓到包但看不出问题”的尴尬处境，完全可以自己动手搓一个类似的小工具——流量分析的门槛远没有想象中那么高，关键是找到自己真正需要它做的事。