Python Scapy 2.5.0 网络抓包实战：构建企业级流量监控工具

当你的服务器突然出现异常流量，或是需要排查某个微服务接口的通信问题时，传统抓包工具如Wireshark往往显得笨重且难以自动化。这就是为什么越来越多的运维工程师开始转向Python的Scapy库——它不仅能以编程方式精准捕获目标流量，还能直接对接数据分析管道。本文将带你从零构建一个生产环境可用的网络监控工具，涵盖协议过滤、实时统计和可视化三大核心功能。

1. 环境配置与基础抓包

在开始前，确保使用Python 3.8+环境，并通过以下命令安装最新版Scapy：

pip install scapy==2.5.0
# 可选：安装数据分析依赖
pip install pandas matplotlib

1.1 基础抓包脚本

下面这个脚本演示了如何捕获100个TCP数据包并打印简要信息：

from scapy.all import sniff, TCP

def packet_callback(packet):
    if packet.haslayer(TCP):
        print(f"TCP Packet: {packet[TCP].sport} -> {packet[TCP].dport}")

sniff(filter="tcp", prn=packet_callback, count=100)

关键参数说明：

• filter ：BPF语法过滤规则，支持 host 、 port 、 net 等条件
• prn ：每个数据包触发的回调函数
• count ：捕获数据包总数

提示：在生产环境中运行前，请确保具有足够的网络权限（Linux/Mac需要sudo）

1.2 高级过滤技巧

Scapy支持组合过滤条件，以下是几种实用场景：

过滤需求	BPF语法示例
捕获特定端口的HTTP流量	tcp port 80 or tcp port 443
监控两个主机间的通信	host 192.168.1.1 and host 8.8.8.8
排除ARP广播包	not arp

2. 协议分析与流量统计

2.1 实时流量统计器

改进版的回调函数可以维护协议统计：

from collections import defaultdict

stats = defaultdict(int)

def analyze_packet(packet):
    if packet.haslayer(TCP):
        stats['tcp'] += 1
    elif packet.haslayer(UDP):
        stats['udp'] += 1
        # 记录DNS查询
        if packet[UDP].dport == 53:
            stats['dns'] += 1
    elif packet.haslayer(ICMP):
        stats['icmp'] += 1

sniff(prn=analyze_packet, timeout=60)  # 运行1分钟
print(dict(stats))

2.2 流量可视化

将统计结果通过Matplotlib生成饼图：

import matplotlib.pyplot as plt

data = {
    'TCP': stats['tcp'],
    'UDP': stats['udp'],
    'ICMP': stats['icmp'],
    'DNS': stats.get('dns', 0)
}

plt.pie(data.values(), labels=data.keys(), autopct='%1.1f%%')
plt.title('Network Protocol Distribution')
plt.savefig('protocol_distribution.png')

3. 生产环境部署方案

3.1 长期运行的守护进程

使用Python的 logging 模块和定时任务构建稳定服务：

import logging
from threading import Thread

logging.basicConfig(
    filename='packet_monitor.log',
    level=logging.INFO,
    format='%(asctime)s - %(message)s'
)

class PacketMonitor(Thread):
    def run(self):
        while True:
            sniff(prn=self.log_packet, timeout=300)  # 每5分钟轮询
            
    def log_packet(self, packet):
        if packet.haslayer(TCP):
            logging.info(f"TCP {packet[TCP].sport}->{packet[TCP].dport}")

monitor = PacketMonitor(daemon=True)
monitor.start()

3.2 异常流量检测

通过阈值触发告警的示例实现：

alert_threshold = 1000  # 每分钟TCP包数量阈值

def detect_anomaly(packet):
    if packet.haslayer(TCP):
        stats['tcp_last_min'] += 1

def check_threshold():
    if stats.get('tcp_last_min', 0) > alert_threshold:
        print(f"ALERT: TCP flood detected ({stats['tcp_last_min']} packets)")
    stats['tcp_last_min'] = 0  # 重置计数器

# 每分钟检查一次
from threading import Timer
Timer(60, check_threshold).start()

4. 高级应用场景

4.1 HTTP请求解析

提取HTTP头部信息需要处理Raw层数据：

def parse_http(packet):
    if packet.haslayer(TCP) and packet.haslayer(Raw):
        payload = str(packet[Raw].load)
        if 'HTTP' in payload:
            headers = payload.split('\r\n')
            print("HTTP Headers:")
            for h in headers[:10]:  # 打印前10行头部
                print(f"  {h}")

sniff(filter="tcp port 80", prn=parse_http)

4.2 自定义协议分析

以解析DNS查询为例：

from scapy.layers.dns import DNSQR

def dns_handler(packet):
    if packet.haslayer(DNSQR):
        query = packet[DNSQR].qname.decode()
        print(f"DNS Query: {query}")

sniff(filter="udp port 53", prn=dns_handler)

实际项目中，我们会将这些功能模块化，构建成可配置的监控系统。比如通过YAML文件定义监控规则：

monitor_rules:
  - name: web_traffic
    filter: tcp port 80 or tcp port 443
    alert:
      threshold: 500/min
      action: email_admin
  - name: dns_monitor
    filter: udp port 53
    log_format: "DNS query: {qname}"