Analyzing Command And Control Communication

Analyzing Command-and-Control Communication

When to Use

• Reverse engineering a malware sample has revealed network communication that needs protocol analysis
• Building network-level detection signatures for a specific C2 framework (Cobalt Strike, Metasploit, Sliver)
• Mapping C2 infrastructure including primary servers, fallback domains, and dead drops
• Analyzing encrypted or encoded C2 traffic to understand the command set and data format
• Attributing malware to a threat actor based on C2 infrastructure patterns and tooling

Do not use for general network anomaly detection; this is specifically for understanding known or suspected C2 protocols from malware analysis.

Prerequisites

• PCAP capture of malware network traffic (from sandbox, network tap, or full packet capture)
• Wireshark/tshark for packet-level analysis
• Reverse engineering tools (Ghidra, dnSpy) for understanding C2 code in the malware binary
• Python 3.8+ with scapy, dpkt, and requests for protocol analysis and replay
• Threat intelligence databases for C2 infrastructure correlation (VirusTotal, Shodan, Censys)
• JA3/JA3S fingerprint databases for TLS-based C2 identification

Workflow

Step 1: Identify the C2 Channel

Determine the protocol and transport used for C2 communication:

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

Step 2: Analyze Beacon Pattern

Characterize the periodic communication pattern:

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")

Step 3: Decode C2 Protocol Structure

Reverse engineer the message format from captured traffic:

# 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

Step 4: Identify C2 Framework

Match observed patterns to known C2 frameworks:

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

Step 5: Map C2 Infrastructure

Document the full C2 infrastructure and failover mechanisms:

# 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"],
            })