📚 课程目录第4章 Python流量分析入门

4. Python流量分析入门

好，咱们进入实战环节了。前面聊了那么多理论，现在终于要亲手碰一碰真正的网络流量了。这一章，我带你搞定Python里最强大的流量分析工具——Scapy。说实话，我当年刚接触网络分析时，被各种二进制格式搞得头大，直到遇见了Scapy，才觉得「原来流量分析可以这么爽」。

4.1 Scapy库的安装与环境配置

先说说安装。Scapy的安装其实很简单，但有几个坑我得提前告诉你。

⚠️ 避坑指南： 我曾经在Windows上直接pip install scapy，结果运行时报错说缺少WinPcap。折腾了半天才发现，Windows下需要额外安装Npcap驱动。Linux和macOS就省心多了。

推荐用pip安装，干净利落：

# 基础安装
pip install scapy

# 如果你需要图形界面（我个人很少用）
pip install scapy[complete]

# 验证安装
python -c "from scapy.all import *; print('Scapy安装成功')"

嗯，这里要注意：Scapy需要管理员/root权限才能发送原始数据包。如果只是读取PCAP文件做分析，普通用户权限就够了。我平时做分析时，90%的时间都在读PCAP，只有做主动探测时才需要提权。

4.2 Scapy的核心数据结构

Scapy最牛的地方，是它把网络协议栈变成了Python对象。你想想看，一个数据包就是一个多层嵌套的字典，每一层协议都可以直接访问。

我个人习惯把Scapy的数据包想象成一个「洋葱」：

最外层：Ethernet帧（MAC地址）
中间层：IP层（源/目标IP）
内层：TCP/UDP等传输层（端口号）
核心：Payload（实际数据）

看个例子就明白了：

from scapy.all import *

# 手动构造一个数据包
pkt = IP(src="192.168.1.1", dst="10.0.0.1")/TCP(sport=12345, dport=80)/"GET / HTTP/1.1"

# 查看数据包结构
pkt.show()

# 输出类似这样：
# ###[ IP ]### 
#   src       = 192.168.1.1
#   dst       = 10.0.0.1
# ###[ TCP ]### 
#   sport     = 12345
#   dport     = 80
# ###[ Raw ]### 
#   load      = 'GET / HTTP/1.1'

看到了吗？用/符号就能把不同协议层「叠」在一起。这个设计太优雅了，我第一次用的时候就觉得：这才叫符合直觉的API设计。

4.3 读取PCAP文件

实际工作中，我们很少手动构造数据包。更多时候是分析现成的PCAP文件。比如从Wireshark导出的抓包文件，或者从IDS/IPS系统里拿到的流量样本。

读取PCAP文件就一行代码：

from scapy.all import *

# 读取PCAP文件
packets = rdpcap("sample.pcap")

# 看看有多少个包
print(f"总共 {len(packets)} 个数据包")

# 查看第一个包
packets[0].show()

# 遍历所有包
for i, pkt in enumerate(packets[:5]):  # 只看前5个
    print(f"包 {i+1}: {pkt.summary()}")

💡 小技巧： 如果PCAP文件很大（比如几个GB），用rdpcap一次性读入内存可能会爆掉。我建议用PcapReader迭代读取：

from scapy.all import PcapReader
for pkt in PcapReader("large_file.pcap"):
    # 逐包处理，内存友好
    process_packet(pkt)

4.4 解析数据包关键字段

读进来只是第一步，关键是怎么提取我们想要的字段。我在做异常流量分析时，最常关注这几个维度：

协议层	关键字段	常见用途
Ethernet	src, dst, type	识别MAC地址、判断协议类型
IP	src, dst, ttl, len	追踪源/目标、检测TTL异常
TCP	sport, dport, flags, seq	端口扫描检测、连接状态分析
UDP	sport, dport, len	DNS查询、NTP放大攻击
ICMP	type, code	ping扫描、Traceroute

来看看实际怎么提取：

from scapy.all import *

packets = rdpcap("sample.pcap")

for pkt in packets:
    # 检查是否有IP层
    if pkt.haslayer(IP):
        ip_layer = pkt[IP]
        src_ip = ip_layer.src
        dst_ip = ip_layer.dst
        ttl = ip_layer.ttl
        
        # 检查是否有TCP层
        if pkt.haslayer(TCP):
            tcp_layer = pkt[TCP]
            sport = tcp_layer.sport
            dport = tcp_layer.dport
            flags = tcp_layer.flags
            
            # 输出关键信息
            print(f"{src_ip}:{sport} -> {dst_ip}:{dport} [Flags: {flags}]")
            
            # 检测SYN扫描（只有SYN标志位）
            if flags == 0x02:  # SYN标志
                print(f"⚠️ 发现SYN扫描行为: {src_ip} -> {dst_ip}:{dport}")

🔍 实战经验： 我曾经处理过一个DDoS攻击事件，攻击者用了大量伪造源IP的SYN包。当时就是靠Scapy快速解析了PCAP文件，统计了每个源IP的SYN包数量，发现某个IP段异常集中。嗯，这个经验告诉我：流量分析的第一步，永远是「统计」。

4.5 常用过滤与统计技巧

实际分析时，我们往往不需要所有包，而是关注特定类型的流量。Scapy提供了很灵活的过滤方式：

from scapy.all import *
from collections import Counter

packets = rdpcap("capture.pcap")

# 1. 过滤特定协议
tcp_packets = [p for p in packets if p.haslayer(TCP)]
udp_packets = [p for p in packets if p.haslayer(UDP)]

# 2. 统计目标端口分布
dst_ports = [p[TCP].dport for p in tcp_packets if p.haslayer(TCP)]
port_counts = Counter(dst_ports)
print("Top 10 目标端口:")
for port, count in port_counts.most_common(10):
    print(f"  端口 {port}: {count} 次")

# 3. 提取HTTP请求（80端口）
http_packets = [p for p in tcp_packets 
                if p.haslayer(TCP) and 
                (p[TCP].dport == 80 or p[TCP].sport == 80)]

# 4. 查看数据包大小分布
sizes = [len(p) for p in packets]
print(f"最小包: {min(sizes)} bytes")
print(f"最大包: {max(sizes)} bytes")
print(f"平均包: {sum(sizes)/len(sizes):.0f} bytes")

💡 我的习惯： 拿到一个PCAP文件后，我通常会先做三件事：

统计协议分布（TCP/UDP/ICMP各占多少）
找出Top 10通信IP
检查是否有异常端口（比如非标准端口上的服务）

这三步走完，基本就能判断流量是否正常了。

4.6 实战：快速分析一个PCAP样本

来，咱们把上面学的串起来，写一个完整的分析脚本：

#!/usr/bin/env python3
"""
快速PCAP分析工具
用法: python analyze_pcap.py sample.pcap
"""

from scapy.all import *
import sys
from collections import Counter

def analyze_pcap(filename):
    print(f"[*] 正在分析: {filename}")
    
    # 读取PCAP
    packets = rdpcap(filename)
    print(f"[*] 总数据包数: {len(packets)}")
    
    # 统计协议
    proto_count = Counter()
    ip_src = Counter()
    ip_dst = Counter()
    tcp_ports = Counter()
    
    for pkt in packets:
        if pkt.haslayer(IP):
            ip = pkt[IP]
            ip_src[ip.src] += 1
            ip_dst[ip.dst] += 1
            
            if pkt.haslayer(TCP):
                proto_count['TCP'] += 1
                tcp = pkt[TCP]
                tcp_ports[tcp.dport] += 1
            elif pkt.haslayer(UDP):
                proto_count['UDP'] += 1
            elif pkt.haslayer(ICMP):
                proto_count['ICMP'] += 1
        else:
            proto_count['Other'] += 1
    
    # 输出结果
    print("\n=== 协议分布 ===")
    for proto, count in proto_count.most_common():
        print(f"  {proto}: {count} ({count/len(packets)*100:.1f}%)")
    
    print("\n=== Top 5 源IP ===")
    for ip, count in ip_src.most_common(5):
        print(f"  {ip}: {count}")
    
    print("\n=== Top 5 目标IP ===")
    for ip, count in ip_dst.most_common(5):
        print(f"  {ip}: {count}")
    
    print("\n=== Top 10 目标端口 ===")
    for port, count in tcp_ports.most_common(10):
        print(f"  端口 {port}: {count}")

if __name__ == "__main__":
    if len(sys.argv) != 2:
        print("用法: python analyze_pcap.py <pcap_file>")
        sys.exit(1)
    analyze_pcap(sys.argv[1])

⚠️ 注意： 这个脚本只是个起点。真实场景中，你可能会遇到加密流量、分片包、IPv6等情况。我建议你先用这个脚本跑几个干净的PCAP文件练手，等熟悉了再处理复杂场景。

好了，这一章的内容就到这里。说白了，Scapy就是一把瑞士军刀，能切能砍能撬。你只要掌握了读取PCAP和提取字段这两个基本功，后面那些高级玩法——流量重放、协议伪造、自动化检测——就都水到渠成了。