3. 流量特征提取：统计特征、协议特征、行为特征

好，咱们进入第三章。上一章聊了数据源怎么抓、怎么存，这一章要动真格的了——特征提取。

说白了，流量本身只是一堆字节流，你得把它变成能喂给分析模型的特征向量。我刚开始做威胁狩猎那会儿，总想着把所有字段都塞进去，结果模型跑出来一团浆糊。后来才明白，特征不在多，在精。

僵尸网络的流量，其实有很明显的“指纹”。我把它分成三大类：统计特征、协议特征、行为特征。咱们一个一个拆。

3.1 统计特征：包长、间隔、速率

统计特征是最直观的。你想想看，人上网和机器上网，节奏完全不一样。

包长分布——这是我最爱看的特征之一。正常用户浏览网页，包长分布很散，从几十字节到上千字节都有。但僵尸网络呢？很多控制指令就几十个字节，心跳包更是固定长度。我遇到过一个大灰狼远控样本，它的心跳包永远是 0xFE 0xED 0xBE 0xEF 开头，后面跟着固定长度的加密载荷。这种特征，一眼就能看出来。

包间隔（IAT）——这个更有意思。人操作电脑，点击鼠标、切换页面，间隔是随机的、不规则的。但僵尸程序是定时器驱动的。我曾经处理过一个案例，受害主机每 60 秒整点向 C2 发送一次心跳，误差不超过 0.5 秒。这种“秒级精度”的周期性，人做不到。

速率特征——包括每秒包数（PPS）、每秒字节数（BPS）。DDoS 型僵尸在攻击阶段速率会飙升，但潜伏期速率很低。我习惯把速率做成时间序列图，看有没有“脉冲式”的突发现象。

统计特征	正常用户	僵尸网络
包长分布	离散、范围广（40~1500 字节）	集中、固定长度（如 64 字节心跳）
包间隔	随机、不规则	高度规律、定时触发
速率（PPS/BPS）	波动平缓	潜伏期低、攻击期脉冲式飙升

3.2 协议特征：HTTP、DNS、IRC

协议特征，说白了就是看流量“长什么样”。不同的僵尸网络，用的协议天差地别。

HTTP 特征——现在最常见的 C2 协议。我见过太多 HTTP 僵尸了，它们的共同点是：User-Agent 很老（比如 IE6）、请求 URL 路径很奇怪（如 /gate.php?id=123&cmd=heartbeat）、Host 头指向 IP 而非域名。还有一个细节——Content-Type。很多僵尸用 application/x-www-form-urlencoded 传加密数据，但正常浏览器很少这么干。

DNS 特征——这个我要重点说。DNS 是僵尸网络的“生命线”。它们需要解析 C2 域名。我总结了几条铁律：

• 高频查询：正常主机几分钟查一次 DNS，僵尸可能每秒查一次。
• NXDOMAIN 响应：很多僵尸用 DGA（域名生成算法），生成的域名大部分解析失败。如果一台主机 DNS 查询失败率超过 50%，基本可以断定有问题。
• TXT 记录：有些高级僵尸把指令藏在 DNS TXT 记录里。我见过一个样本，它的 C2 指令就是通过 TXT 查询返回的 Base64 编码字符串。

IRC 特征——老牌僵尸协议，现在少了，但还在。IRC 僵尸的特征很明显：连接 IRC 服务器、加入特定频道、接收 PRIVMSG 指令。我记得 2016 年处理过一个 Mirai 变种，它用的就是 IRC 协议，频道名是 #mirai_ctl，指令格式是 !scan 192.168.0.0/16。这种特征，写个正则就能匹配。

3.3 行为特征：周期性、心跳包

行为特征，是统计特征和协议特征的“升华”。它关注的是模式，而不是单个数据包。

周期性检测——这是我最依赖的手段。僵尸网络为了维持“在线状态”，必须定期和 C2 通信。这个周期可能是 30 秒、5 分钟、1 小时。怎么检测？我常用自相关分析。把时间序列数据做自相关，如果某个滞后值出现明显峰值，就说明存在周期性。

举个例子：假设你抓了 1 小时的流量，每 5 分钟就有一个 DNS 查询到 evil.com。自相关函数在滞后 300 秒处会出现一个尖峰。这个尖峰，就是僵尸的“心跳”。

心跳包特征——心跳包本身也有特征。我总结了几种常见模式：

• 固定载荷：每次发送的内容一模一样（如 0x01 0x02 0x03）。
• 递增序列号：载荷里包含一个递增的计数器，用于同步状态。
• 时间戳加密：载荷里包含加密后的当前时间戳，用于防重放。

我个人习惯，提取心跳包特征时，会做载荷相似度计算。把同一源 IP 的所有出站包做两两比较，如果相似度超过 95%，基本就是心跳包。

3.4 知识体系总览

下面这张图，是我自己梳理的流量特征提取框架。你可以把它当成一个“检查清单”，做分析时对着看，不容易漏项。

3.5 实战：提取一个心跳包特征

光说不练假把式。我写个简单的 Python 代码片段，演示怎么从 PCAP 里提取心跳包特征。

import scapy.all as scapy
import numpy as np

def extract_heartbeat_features(pcap_file, src_ip, dst_ip, time_window=300):
    """
    提取指定源IP到目的IP的心跳包特征
    time_window: 时间窗口（秒），默认5分钟
    """
    packets = scapy.rdpcap(pcap_file)
    
    # 过滤出源IP到目的IP的TCP/UDP包
    target_pkts = []
    for pkt in packets:
        if pkt.haslayer(scapy.IP):
            if pkt[scapy.IP].src == src_ip and pkt[scapy.IP].dst == dst_ip:
                target_pkts.append(pkt)
    
    if len(target_pkts) < 2:
        print("数据包太少，无法分析")
        return None
    
    # 提取时间戳和载荷长度
    timestamps = []
    payload_lens = []
    for pkt in target_pkts:
        timestamps.append(float(pkt.time))
        if pkt.haslayer(scapy.Raw):
            payload_lens.append(len(pkt[scapy.Raw].load))
        else:
            payload_lens.append(0)
    
    # 计算包间隔
    intervals = np.diff(timestamps)
    
    # 计算载荷相似度（简单版本：比较相邻包的前10字节）
    payload_similarities = []
    for i in range(len(target_pkts) - 1):
        pkt1 = target_pkts[i]
        pkt2 = target_pkts[i+1]
        if pkt1.haslayer(scapy.Raw) and pkt2.haslayer(scapy.Raw):
            load1 = pkt1[scapy.Raw].load[:10]
            load2 = pkt2[scapy.Raw].load[:10]
            if len(load1) > 0 and len(load2) > 0:
                similarity = sum(1 for a, b in zip(load1, load2) if a == b) / max(len(load1), len(load2))
                payload_similarities.append(similarity)
    
    # 输出特征
    features = {
        'packet_count': len(target_pkts),
        'mean_interval': np.mean(intervals),
        'std_interval': np.std(intervals),
        'mean_payload_len': np.mean(payload_lens),
        'std_payload_len': np.std(payload_lens),
        'mean_similarity': np.mean(payload_similarities) if payload_similarities else 0,
        'is_periodic': np.std(intervals) < 0.5  # 间隔标准差小于0.5秒，认为有周期性
    }
    
    return features

# 使用示例
features = extract_heartbeat_features('capture.pcap', '192.168.1.100', '203.0.113.5')
if features:
    print(f"心跳包数量: {features['packet_count']}")
    print(f"平均间隔: {features['mean_interval']:.2f}秒")
    print(f"间隔标准差: {features['std_interval']:.2f}秒")
    print(f"载荷相似度: {features['mean_similarity']:.2%}")
    print(f"是否周期性: {features['is_periodic']}")