PCAP 网络分析指南

本技能提供对网络数据包捕获文件（PCAP 文件）进行分析及使用 Python 计算网络统计指标的指导。

快速入门：使用辅助模块

本目录下提供一个工具模块（pcap_utils.py），其中包含经过**充分测试、逻辑正确**的常用分析函数实现。该模块提供若干实用函数，用于统计中间结果，帮助你更快得出关键结论。请直接调用这些函数，而非自行重新实现相同逻辑——因其已正确处理各类边界情况。

# 推荐做法：导入并直接使用辅助函数
import sys
sys.path.insert(0, '/root/skills/pcap-analysis')  # 将技能目录加入 Python 路径
from pcap_utils import (
    load_packets, split_by_protocol, graph_metrics,
    detect_port_scan, detect_dos_pattern, detect_beaconing,
    port_counters, ip_counters, iat_stats, flow_metrics,
    packets_per_minute_stats, producer_consumer_counts, shannon_entropy
)

packets = load_packets('/root/packets.pcap')
parts = split_by_protocol(packets)

# 图论/拓扑指标（入度/出度统计的是 UNIQUE IP 数量，而非数据包数量！）
g = graph_metrics(parts['ip'])
print(g['max_indegree'], g['max_outdegree'])

# 检测函数采用严格阈值，所有条件必须同时满足才返回 True
print(detect_port_scan(parts['tcp']))      # 返回 True/False
print(detect_dos_pattern(ppm_avg, ppm_max)) # 返回 True/False
print(detect_beaconing(iat_cv))             # 返回 True/False

辅助函数使用特定的检测阈值（详见下文说明），这些阈值经过校准以确保结果准确。若自行实现逻辑并采用不同阈值，极可能导致错误结论。

概述

网络流量分析涉及读取数据包捕获文件，并计算各类统计指标：

• 基础计数（数据包数量、字节数、协议分布）
• 分布分析（熵值）
• 图论/拓扑结构指标
• 时间序列模式
• 流级别分析

使用 Scapy 读取 PCAP 文件

Scapy 是 Python 中标准的数据包操作库：

from scapy.all import rdpcap, IP, TCP, UDP, ICMP, ARP

# 加载全部数据包
packets = rdpcap('packets.pcap')

# 按协议过滤
ip_packets = [p for p in packets if IP in p]
tcp_packets = [p for p in packets if TCP in p]
udp_packets = [p for p in packets if UDP in p]

基础统计指标

数据包与字节数统计

total_packets = len(packets)
total_bytes = sum(len(p) for p in packets)
avg_packet_size = total_bytes / total_packets

协议分布统计

tcp_count = len([p for p in packets if TCP in p])
udp_count = len([p for p in packets if UDP in p])
icmp_count = len([p for p in packets if ICMP in p])
arp_count = len([p for p in packets if ARP in p])

熵值计算

香农熵（Shannon entropy）用于衡量分布的“随机性”：

import math
from collections import Counter

def shannon_entropy(counter):
    """
    计算香农熵：H(X) = -Σ p(x) log₂(p(x))

    低熵值：流量集中在少数目标上（正常行为）
    高熵值：流量分散在大量目标上（扫描行为）
    """
    total = sum(counter.values())
    if total == 0:
        return 0.0

    entropy = 0.0
    for count in counter.values():
        if count > 0:
            p = count / total
            entropy -= p * math.log2(p)
    return entropy

# 示例：目的端口熵值
dst_ports = Counter()
for pkt in tcp_packets:
    dst_ports[pkt[TCP].dport] += 1
for pkt in udp_packets:
    if IP in pkt:
        dst_ports[pkt[UDP].dport] += 1

port_entropy = shannon_entropy(dst_ports)

图论/拓扑结构指标

重要提示：务必使用 pcap_utils.py 中的 graph_metrics() 函数获取正确结果！

将网络建模为有向图，节点为 IP 地址，边为通信关系对（src → dst）。

关键要点：度（degree）指与之通信的 UNIQUE IP 数量，而非数据包总数！

• max_indegree = 任一目的地址所接收数据包的唯一源 IP 数量的最大值
• max_outdegree = 任一源地址所发送数据包的唯一目的 IP 数量的最大值

常见错误：统计数据包总数而非唯一 IP 数量。对于含 38 个节点的网络，max_indegree 最大值应为 37，而非数千！

# 推荐做法：使用辅助函数
from pcap_utils import graph_metrics
g = graph_metrics(ip_packets)
print(g['max_indegree'])   # 唯一 IP 数量，通常 < 50
print(g['max_outdegree'])  # 唯一 IP 数量，通常 < 50

# 或手动实现时：
from collections import defaultdict

# 构建图：节点 = IP，边 = (src, dst) 对
edges = set()
indegree = defaultdict(set)   # dst -> 向该 dst 发送数据包的源 IP 集合
outdegree = defaultdict(set)  # src -> src 所发送数据包的目的 IP 集合

for pkt in ip_packets:
    src, dst = pkt[IP].src, pkt[IP].dst
    edges.add((src, dst))
    indegree[dst].add(src)      # dst 收到来自 src 的数据包
    outdegree[src].add(dst)     # src 向 dst 发送数据包

all_nodes = set(indegree.keys()) | set(outdegree.keys())
num_nodes = len(all_nodes)
num_edges = len(edges)

# 网络密度 = 实际边数 / 可能边数
# 有向图中可能边数 = n × (n−1)
network_density = num_edges / (num_nodes * (num_nodes - 1))

# 度中心性 —— 统计 UNIQUE IP 数量，非数据包数量！
# 使用 len(set) 获取唯一 IP 数量
max_indegree = max(len(v) for v in indegree.values())    # len(set) = 唯一 IP 数量
max_outdegree = max(len(v) for v in outdegree.values())  # len(set) = 唯一 IP 数量

时间维度指标

数据包到达间隔时间（IAT）

# 获取按时间戳排序的列表
timestamps = sorted(float(p.time) for p in packets)

# 计算到达间隔时间
iats = [timestamps[i+1] - timestamps[i] for i in range(len(timestamps)-1)]

iat_mean = sum(iats) / len(iats)
iat_variance = sum((x - iat_mean)**2 for x in iats) / len(iats)
iat_std = math.sqrt(iat_variance)

# 变异系数 CV = std / mean
# CV 较低（<0.5）：规律性强/机器人式流量（可疑）
# CV 较高（>1.0）：突发性/人类操作流量（正常）
iat_cv = iat_std / iat_mean if iat_mean > 0 else 0

生产者/消费者比（PCR）

# PCR = (bytes_sent - bytes_recv) / (bytes_sent + bytes_recv)
# 正值：生产者/服务器；负值：消费者/客户端
bytes_sent = defaultdict(int)
bytes_recv = defaultdict(int)

for pkt in ip_packets:
    size = len(pkt)
    bytes_sent[pkt[IP].src] += size
    bytes_recv[pkt[IP].dst] += size

num_producers = 0
num_consumers = 0
for ip in all_nodes:
    sent = bytes_sent.get(ip, 0)
    recv = bytes_recv.get(ip, 0)
    total = sent + recv
    if total > 0:
        pcr = (sent - recv) / total
        if pcr > 0.2:
            num_producers += 1
        elif pcr < -0.2:
            num_consumers += 1

流分析

一条流定义为五元组：(src_ip, dst_ip, src_port, dst_port, protocol)

重要提示：仅统计同时包含 IP 层和传输层的数据包所构成的流！

# 收集唯一流
flows = set()
for pkt in tcp_packets:
    if IP in pkt:  # 必须检查 IP 层存在
        flow = (pkt[IP].src, pkt[IP].dst, pkt[TCP].sport, pkt[TCP].dport, "TCP")
        flows.add(flow)

for pkt in udp_packets:
    if IP in pkt:  # 必须检查 IP 层存在
        flow = (pkt[IP].src, pkt[IP].dst, pkt[UDP].sport, pkt[UDP].dport, "UDP")
        flows.add(flow)

unique_flows = len(flows)
tcp_flows = len([f for f in flows if f[4] == "TCP"])
udp_flows = len([f for f in flows if f[4] == "UDP"])

# 双向流：统计正向与反向均存在的流对数量
# 即存在 A→B 且 B→A（端口相同）的流对
bidirectional_count = 0
for flow in flows:
    src_ip, dst_ip, src_port, dst_port, proto = flow
    reverse = (dst_ip, src_ip, dst_port, src_port, proto)
    if reverse in flows:
        bidirectional_count += 1

# 每个双向流对被统计两次（A→B 和 B→A），故需除以 2
bidirectional_flows = bidirectional_count // 2

时间序列分析

按时间区间对数据包分桶：

from collections import defaultdict

timestamps = [float(p.time) for p in packets]
start_time = min(timestamps)

# 按分钟分桶
minute_buckets = defaultdict(int)
for ts in timestamps:
    minute = int((ts - start_time) / 60)
    minute_buckets[minute] += 1

duration_seconds = max(timestamps) - start_time
packets_per_min = list(minute_buckets.values())
ppm_avg = sum(packets_per_min) / len(packets_per_min)
ppm_max = max(packets_per_min)
ppm_min = min(packets_per_min)

写入 CSV 结果文件

import csv

results = {
    "total_packets": total_packets,
    "protocol_tcp": tcp_count,
    # ... 其他指标
}

# 读取模板并填充数值
with open('network_stats.csv', 'r') as f:
    reader = csv.DictReader(f)
    rows = list(reader)

with open('network_stats.csv', 'w', newline='') as f:
    writer = csv.DictWriter(f, fieldnames=['metric', 'value'])
    writer.writeheader()
    for row in rows:
        metric = row['metric']
        if metric.startswith('#'):
            writer.writerow(row)  # 保留注释行
        elif metric in results:
            writer.writerow({'metric': metric, 'value': results[metric]})

流量分析问题

完成各项指标计算后，你可能需要回答有关该流量的具体分析问题。

主导协议识别

找出数据包数量最多的协议：

protocol_counts = {
    "tcp": tcp_count,
    "udp": udp_count,
    "icmp": icmp_count,
    "arp": arp_count,
}
dominant_protocol = max(protocol_counts, key=protocol_counts.get)

端口扫描检测（鲁棒方法）

重要提示：请使用 pcap_utils.py 中的 detect_port_scan() 函数以获得准确结果！

基于简单阈值的检测方法不具备鲁棒性。原因如下：

• 合法用户可能在较长时间内访问大量端口（时间不敏感）
• 分布式扫描由多个源发起，每个源仅探测少量端口
• 半开放扫描（仅发送 SYN）可能未完成三次握手

鲁棒检测要求同时满足以下三项信号：

1. 端口熵值 > 6.0：扫描器均匀访问大量端口（高熵值）。正常流量反复访问相同端口（低熵值，约 4–5 bits）。
2. 仅 SYN 包比例 > 0.7：半开放扫描仅发送 SYN 而不完成握手（占比 >70%）
3. 唯一端口数 > 100：必须具备足够多的唯一端口才构成有意义的扫描行为

若任意一项条件未满足，则判定为无端口扫描行为！

# 推荐做法：使用辅助函数
from pcap_utils import detect_port_scan
has_port_scan = detect_port_scan(tcp_packets)  # 返回 True/False

# 或手动实现时，必须同时满足全部三项条件：
from collections import Counter, defaultdict

src_port_counts = defaultdict(Counter)  # src -> 目的端口计数器
src_syn_only = defaultdict(int)         # src -> 仅 SYN 数据包计数
src_total_tcp = defaultdict(int)        # src -> TCP 数据包总数

for pkt in tcp_packets:
    if IP in pkt and TCP in pkt:
        src = pkt[IP].src
        dst_port = pkt[TCP].dport
        flags = pkt[TCP].flags

        src_port_counts[src][dst_port] += 1
        src_total_tcp[src] += 1

        # 仅 SYN：SYN 标志位（0x02）置位，且 ACK 标志位（0x10）未置位
        if flags & 0x02 and not (flags & 0x10):
            src_syn_only[src] += 1

def calc_port_entropy(port_counter):
    """计算端口分布的香农熵值。"""
    total = sum(port_counter.values())
    if total == 0:
        return 0.0
    entropy = 0.0
    for count in port_counter.values():
        if count > 0:
            p = count / total
            entropy -= p * math.log2(p)
    return entropy

has_port_scan = False
for src in src_port_counts:
    total_pkts = src_total_tcp[src]
    if total_pkts < 50:  # 忽略低流量源
        continue

    port_entropy = calc_port_entropy(src_port_counts[src])
    syn_only_ratio = src_syn_only[src] / total_pkts
    unique_ports = len(src_port_counts[src])

    # 扫描特征：必须同时满足以下三项条件！
    # - 高熵值（端口分布均匀，非重复访问固定端口）
    # - 高仅 SYN 比例（半开放连接，未完成握手）
    # - 大量唯一端口（实际探测多种服务）
    if port_entropy > 6.0 and syn_only_ratio > 0.7 and unique_ports > 100:
        has_port_scan = True
        break

DoS 模式检测

重要提示：请使用 pcap_utils.py 中的 detect_dos_pattern() 函数以获得准确结果！

DoS 攻击会导致极端的流量激增。判定阈值为>20 倍于平均速率。

最大/平均比值必须严格大于 20 才可判定为 DoS。5 倍、10 倍甚至 15 倍均为正常流量波动，不可视为 DoS！

# 推荐做法：使用辅助函数
from pcap_utils import detect_dos_pattern
has_dos = detect_dos_pattern(packets_per_minute_avg, packets_per_minute_max)

# 或手动实现时：
ppm_ratio = packets_per_minute_max / packets_per_minute_avg
# 仅当比值 > 20 时才表示 DoS！较低比值属于正常流量波动。
has_dos_pattern = ppm_ratio > 20

C2 信标行为检测

命令与控制（C2）信标行为表现为规律、周期性的通信模式。 可通过低到达间隔时间变异系数（CV < 0.5）进行检测：

# CV = std / mean
# CV 较低（<0.5）：规律性强/机器人式流量（可疑信标行为）
# CV 较高（>1.0）：突发性/人类操作流量（正常）
has_beaconing = iat_cv < 0.5

良性流量评估

当且仅当以下全部条件为假时，流量才可判定为良性：

• 未检测到端口扫描（使用 detect_port_scan()）
• 未检测到 DoS 模式（使用 detect_dos_pattern()）
• 未检测到信标行为（使用 detect_beaconing()）

重要提示：请使用具备正确阈值的检测辅助函数。若自行实现检测逻辑并采用错误阈值，将导致错误结论！

# 推荐做法：使用辅助函数进行检测
from pcap_utils import detect_port_scan, detect_dos_pattern, detect_beaconing

has_port_scan = detect_port_scan(tcp_packets)
has_dos_pattern = detect_dos_pattern(ppm_avg, ppm_max)
has_beaconing = detect_beaconing(iat_cv)

# 仅当全部检测结果均为 False 时，流量才为良性
is_benign = (
    not has_port_scan and
    not has_dos_pattern and
    not has_beaconing
)

常见问题

大型 PCAP 文件的内存占用

对于大型捕获文件，建议分块处理或应用具体过滤条件：

from scapy.all import PcapReader

# 流式读取大型文件
with PcapReader('large.pcap') as pcap:
    for pkt in pcap:
        # 逐个处理数据包
        pass

非 IP 数据包

部分数据包（如 ARP 等）不含 IP 层。务必始终检查：

for pkt in packets:
    if IP in pkt:
        src = pkt[IP].src
        # ... 处理逻辑

无 IP 层的 UDP 数据包

链路层协议可能存在不含 IP 层的 UDP 数据包：

for pkt in udp_packets:
    if IP in pkt:  # 安全检查
        # ... 处理逻辑