analyzing-command-and-control-communication

分析命令与控制通信

适用场景

• 对恶意软件样本的逆向工程揭示了需要协议分析的网络通信
• 为特定 C2 框架（Cobalt Strike、Metasploit、Sliver）构建网络级检测签名
• 绘制 C2 基础设施图，包括主服务器、备用域名和投弃站
• 分析加密或编码的 C2 流量以了解命令集和数据格式
• 基于 C2 基础设施模式和工具对恶意软件进行威胁行为者溯源归因

不适用于一般网络异常检测；本技能专门用于从恶意软件分析中了解已知或疑似 C2 协议。

前置条件

• 恶意软件网络流量的 PCAP 捕获（来自沙箱、网络分流或全包捕获）
• Wireshark/tshark 用于数据包级分析
• 逆向工程工具（Ghidra、dnSpy）用于了解恶意软件二进制文件中的 C2 代码
• Python 3.8+，安装 scapy、dpkt 和 requests 用于协议分析和重放
• 威胁情报数据库用于 C2 基础设施关联（VirusTotal、Shodan、Censys）
• JA3/JA3S 指纹数据库用于基于 TLS 的 C2 识别

工作流程

步骤 1：识别 C2 通道

确定用于 C2 通信的协议和传输方式：

C2 通信通道：
━━━━━━━━━━━━━━━━━━━━━━━━━
HTTP/HTTPS：    最常见；使用标准 Web 流量进行混入
                指标：定期 POST/GET 请求、特定 URI 模式、自定义请求头

DNS：           通过 DNS 查询和响应进行数据隧道传输
                指标：大量 TXT 查询、长子域名、高熵值

自定义 TCP/UDP：使用非标准端口的私有二进制协议
                指标：高端口上的非 HTTP 流量、未知协议

ICMP：          编码在 ICMP 回显/回复载荷中的数据
                指标：具有大型或非标准载荷的 ICMP 数据包

WebSocket：     用于实时 C2 的持久双向连接
                指标：WebSocket 升级后跟随二进制帧

云服务：        使用合法 API（Telegram、Discord、Slack、GitHub）
                指标：非预期进程向云服务进行 API 调用

电子邮件：      使用 SMTP/IMAP 进行 C2 命令和数据外泄
                指标：非电子邮件进程的自动化邮件操作

步骤 2：分析 Beacon 模式

对周期性通信模式进行特征分析：

from scapy.all import rdpcap, IP, TCP
from collections import defaultdict
import statistics
import json

packets = rdpcap("c2_traffic.pcap")

# 按目标分组 TCP SYN 数据包
connections = defaultdict(list)
for pkt in packets:
    if IP in pkt and TCP in pkt and (pkt[TCP].flags & 0x02):
        key = f"{pkt[IP].dst}:{pkt[TCP].dport}"
        connections[key].append(float(pkt.time))

# 分析每个目标的 beacon 行为
for dst, times in sorted(connections.items()):
    if len(times) < 3:
        continue

    intervals = [times[i+1] - times[i] for i in range(len(times)-1)]
    avg_interval = statistics.mean(intervals)
    stdev = statistics.stdev(intervals) if len(intervals) > 1 else 0
    jitter_pct = (stdev / avg_interval * 100) if avg_interval > 0 else 0
    duration = times[-1] - times[0]

    beacon_data = {
        "destination": dst,
        "connections": len(times),
        "duration_seconds": round(duration, 1),
        "avg_interval_seconds": round(avg_interval, 1),
        "stdev_seconds": round(stdev, 1),
        "jitter_percent": round(jitter_pct, 1),
        "is_beacon": 5 < avg_interval < 7200 and jitter_pct < 25,
    }

    if beacon_data["is_beacon"]:
        print(f"[!] 检测到 BEACON：{dst}")
        print(f"    间隔：{avg_interval:.0f}s +/- {stdev:.0f}s（{jitter_pct:.0f}% 抖动）")
        print(f"    会话：{len(times)} 次，持续 {duration:.0f}s")