analyzing-command-and-control-communication

分析命令与控制通信

何时使用

• 对恶意软件样本进行逆向工程时发现需要协议分析的网络通信
• 为特定 C2 框架（Cobalt Strike、Metasploit、Sliver）构建网络层检测特征
• 梳理 C2 基础设施，包括主服务器、备用域名和死投点
• 分析加密或编码的 C2 流量，以了解命令集和数据格式
• 基于 C2 基础设施模式和工具将恶意软件归因于某个威胁行为者

请勿使用于一般网络异常检测；本内容专门用于从恶意软件分析中理解已知或疑似 C2 协议。

前提条件

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

工作流

步骤 1：识别 C2 通道

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

C2 Communication Channels:
━━━━━━━━━━━━━━━━━━━━━━━━━
HTTP/HTTPS:     Most common; uses standard web traffic to blend in
                Indicators: Regular POST/GET requests, specific URI patterns, custom headers

DNS:            Tunneling data through DNS queries and responses
                Indicators: High-volume TXT queries, long subdomain names, high entropy

Custom TCP/UDP: Proprietary binary protocol on non-standard port
                Indicators: Non-HTTP traffic on high ports, unknown protocol

ICMP:           Data encoded in ICMP echo/reply payloads
                Indicators: ICMP packets with large or non-standard payloads

WebSocket:      Persistent bidirectional connection for real-time C2
                Indicators: WebSocket upgrade followed by binary frames

Cloud Services: Using legitimate APIs (Telegram, Discord, Slack, GitHub)
                Indicators: API calls to cloud services from unexpected processes

Email:          SMTP/IMAP for C2 commands and data exfiltration
                Indicators: Automated email operations from non-email processes

步骤 2：分析 Beacon 模式

刻画周期性通信模式的特征：

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

packets = rdpcap("c2_traffic.pcap")

# Group TCP SYN packets by destination
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))

# Analyze each destination for beaconing
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 DETECTED: {dst}")
        print(f"    Interval: {avg_interval:.0f}s +/- {stdev:.0f}s ({jitter_pct:.0f}% jitter)")
        print(f"    Sessions: {len(times)} over {duration:.0f}s")

步骤 3：解码 C2 协议结构

从捕获的流量中逆向工程消息格式：

# HTTP-based C2 protocol analysis
import dpkt
import base64

with open("c2_traffic.pcap", "rb") as f:
    pcap = dpkt.pcap.Reader(f)

for ts, buf in pcap:
    eth = dpkt.ethernet.Ethernet(buf)
    if not isinstance(eth.data, dpkt.ip.IP):
        continue
    ip = eth.data
    if not isinstance(ip.data, dpkt.tcp.TCP):
        continue
    tcp = ip.data

    if tcp.dport == 80 or tcp.dport == 443:
        if len(tcp.data) > 0:
            try:
                http = dpkt.http.Request(tcp.data)
                print(f"\n--- C2 REQUEST ---")
                print(f"Method: {http.method}")
                print(f"URI: {http.uri}")
                print(f"Headers: {dict(http.headers)}")
                if http.body:
                    print(f"Body ({len(http.body)} bytes):")
                    # Try Base64 decode
                    try:
                        decoded = base64.b64decode(http.body)
                        print(f"  Decoded: {decoded[:200]}")
                    except:
                        print(f"  Raw: {http.body[:200]}")
            except:
                pass

步骤 4：识别 C2 框架

将观察到的模式与已知 C2 框架进行匹配：

Known C2 Framework Signatures:
━━━━━━━━━━━━━━━━━━━━━━━━━━━━
Cobalt Strike:
  - Default URIs: /pixel, /submit.php, /___utm.gif, /ca, /dpixel
  - Malleable C2 profiles customize all traffic characteristics
  - JA3: varies by profile, catalog at ja3er.com
  - Watermark in beacon config (unique per license)
  - Config extraction: use CobaltStrikeParser or 1768.py

Metasploit/Meterpreter:
  - Default staging URI patterns: random 4-char checksum
  - Reverse HTTP(S) handler patterns
  - Meterpreter TLV (Type-Length-Value) protocol structure

Sliver:
  - mTLS, HTTP, DNS, WireGuard transport options
  - Protobuf-encoded messages
  - Unique implant ID in communication

Covenant:
  - .NET-based C2 framework
  - HTTP with customizable profiles
  - Task-based command execution

PoshC2:
  - PowerShell/C# based
  - HTTP with encrypted payloads
  - Cookie-based session management

# Extract Cobalt Strike beacon configuration from PCAP or sample
python3 << 'PYEOF'
# Using CobaltStrikeParser (pip install cobalt-strike-parser)
from cobalt_strike_parser import BeaconConfig

try:
    config = BeaconConfig.from_file("suspect.exe")
    print("Cobalt Strike Beacon Configuration:")
    for key, value in config.items():
        print(f"  {key}: {value}")
except Exception as e:
    print(f"Not a Cobalt Strike beacon or parse error: {e}")
PYEOF

步骤 5：梳理 C2 基础设施

记录完整的 C2 基础设施和故障转移机制：

# Infrastructure mapping
import requests
import json

c2_indicators = {
    "primary_c2": "185.220.101.42",
    "domains": ["update.malicious.com", "backup.evil.net"],
    "ports": [443, 8443],
    "failover_dns": ["ns1.malicious-dns.com"],
}

# Enrich with Shodan
def shodan_lookup(ip, api_key):
    resp = requests.get(f"https://api.shodan.io/shodan/host/{ip}?key={api_key}")
    if resp.status_code == 200:
        data = resp.json()
        return {
            "ip": ip,
            "ports": data.get("ports", []),
            "os": data.get("os"),
            "org": data.get("org"),
            "asn": data.get("asn"),
            "country": data.get("country_code"),
            "hostnames": data.get("hostnames", []),
            "last_update": data.get("last_update"),
        }
    return None

# Enrich with passive DNS
def pdns_lookup(domain):
    # Using VirusTotal passive DNS
    resp = requests.get(
        f"https://www.virustotal.com/api/v3/domains/{domain}/resolutions",
        headers={"x-apikey": VT_API_KEY}
    )
    if resp.status_code == 200:
        data = resp.json()
        resolutions = []
        for r in data.get("data", []):
            resolutions.append({
                "ip": r["attributes"]["ip_address"],
                "date": r["attributes"]["date"],
            })
        return resolutions
    return []

步骤 6：创建网络检测特征

基于分析出的 C2 特征构建检测规则：

# Suricata rules for the analyzed C2
cat << 'EOF' > c2_detection.rules
# HTTP beacon pattern
alert http $HOME_NET any -> $EXTERNAL_NET any (
    msg:"MALWARE MalwareX C2 HTTP Beacon";
    flow:established,to_server;
    http.method; content:"POST";
    http.uri; content:"/gate.php"; startswith;
    http.header; content:"User-Agent: Mozilla/5.0 (compatible; MSIE 10.0)";
    threshold:type threshold, track by_src, count 5, seconds 600;
    sid:9000010; rev:1;
)

# JA3 fingerprint match
alert tls $HOME_NET any -> $EXTERNAL_NET any (
    msg:"MALWARE MalwareX TLS JA3 Fingerprint";
    ja3.hash; content:"a0e9f5d64349fb13191bc781f81f42e1";
    sid:9000011; rev:1;
)

# DNS beacon detection (high-entropy subdomain)
alert dns $HOME_NET any -> any any (
    msg:"MALWARE Suspected DNS C2 Tunneling";
    dns.query; pcre:"/^[a-z0-9]{20,}\./";
    threshold:type threshold, track by_src, count 10, seconds 60;
    sid:9000012; rev:1;
)

# Certificate-based detection
alert tls $HOME_NET any -> $EXTERNAL_NET any (
    msg:"MALWARE MalwareX Self-Signed C2 Certificate";
    tls.cert_subject; content:"CN=update.malicious.com";
    sid:9000013; rev:1;
)
EOF

关键概念

术语	定义
Beaconing	恶意软件以固定间隔向 C2 服务器进行的周期性签到通信，通常带有抖动以避免模式检测
Jitter	施加到 beacon 间隔上的随机化（例如 60s +/- 15%），使时间模式更难以预测和检测
Malleable C2	Cobalt Strike 的一项功能，允许操作者自定义 C2 流量的所有方面（URI、标头、编码），以模仿合法服务
Dead Drop	中间位置（粘贴站点、云存储、社交媒体），C2 命令在此发布以供恶意软件获取
Domain Fronting	在 TLS SNI 中使用受信任的 CDN 域名，同时路由到不同的后端，使 C2 流量看起来像是发往合法服务
Fast Flux	快速更改 C2 域名的 DNS 记录，以分散到众多 IP 并抵抗下架行动
C2 Framework	提供 C2 服务器、植入物生成器和操作者界面的软件工具包（Cobalt Strike、Metasploit、Sliver、Covenant）

工具与系统

• Wireshark：数据包分析器，用于在数据包层面进行详细的 C2 协议分析
• RITA (Real Intelligence Threat Analytics)：开源工具，分析 Zeek 日志以进行 beacon 检测和 DNS 隧道检测
• CobaltStrikeParser：从样本和内存转储中提取 Cobalt Strike beacon 配置的工具
• JA3/JA3S：TLS 指纹识别方法，用于根据 C2 框架的 TLS 实现特征来识别它们
• Shodan/Censys：互联网扫描平台，用于梳理 C2 基础设施并识别相关服务器

常见场景

场景：逆向工程自定义 C2 协议

背景：一个恶意软件样本通过 TCP 端口 8443 使用未知的二进制协议与其 C2 服务通信。需要解码该协议以理解命令集并构建检测规则。

方法：

1. 在 PCAP 中过滤 TCP 端口 8443 的会话并提取 TCP 流
2. 分析最初几次交互以识别握手/认证机制
3. 梳理消息结构（长度前缀、类型字段、载荷编码）
4. 与 Ghidra 中恶意软件发送/接收函数的反汇编交叉比对
5. 识别命令分发器并记录每个命令码的功能
6. 用 Python 构建协议解码器，用于持续的流量分析
7. 创建匹配协议握手或静态头部字节的 Suricata 规则

陷阱：

• 假设协议是静态的；一些 C2 框架在握手期间协商加密
• 未捕获足够的流量以看到所有命令类型（有些命令很罕见）
• 遗漏在主通道失败时激活的备用 C2 通道（DNS、ICMP）
• 将加密载荷数据与协议分帧结构混淆

输出格式

C2 COMMUNICATION ANALYSIS REPORT
===================================
Sample:           malware.exe (SHA-256: e3b0c44...)
C2 Framework:     Cobalt Strike 4.9

BEACON CONFIGURATION
C2 Server:        hxxps://185.220.101[.]42/updates
Beacon Type:      HTTPS (reverse)
Sleep:            60 seconds
Jitter:           15%
User-Agent:       Mozilla/5.0 (Windows NT 10.0; Win64; x64)
URI (GET):        /dpixel
URI (POST):       /submit.php
Watermark:        1234567890

PROTOCOL ANALYSIS
Transport:        HTTPS (TLS 1.2)
JA3 Hash:         a0e9f5d64349fb13191bc781f81f42e1
Certificate:      CN=Microsoft Update (self-signed)
Encoding:         Base64 with XOR key 0x69
Command Format:   [4B length][4B command_id][payload]

COMMAND SET
0x01 - Sleep          Change beacon interval
0x02 - Shell          Execute cmd.exe command
0x03 - Download       Transfer file from C2
0x04 - Upload         Exfiltrate file to C2
0x05 - Inject         Process injection
0x06 - Keylog         Start keylogger
0x07 - Screenshot     Capture screen

INFRASTRUCTURE
Primary:          185.220.101[.]42 (AS12345, Hosting Co, NL)
Failover:         91.215.85[.]17 (AS67890, VPS Provider, RU)
DNS:              update.malicious[.]com -> 185.220.101[.]42
Registrar:        NameCheap
Registration:     2025-09-01

DETECTION SIGNATURES
SID 9000010:      HTTP beacon pattern
SID 9000011:      JA3 TLS fingerprint
SID 9000013:      C2 certificate match