malware analysis

Analyzing Command-and-Control Communication

Analyzes malware command-and-control (C2) communication protocols to understand beacon patterns, command structures, data encoding, and infrastructure. Covers HTTP, HTTPS, DNS, and custom protocol C2 analysis for detection development and threat intelligence. Activates for requests involving C2 analysis, beacon detection, C2 protocol reverse engineering, or command-and-control infrastructure mapping.

beaconc2command-and-controlmalwareprotocol-analysis
Install this skill
npx skills add mukul975/Anthropic-Cybersecurity-Skills
Framework mappings

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"],
            })
        return resolutions
    return []

Step 6: Create Network Detection Signatures

Build detection rules based on analyzed C2 characteristics:

# 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

Key Concepts

Term Definition
Beaconing Periodic check-in communication from malware to C2 server at regular intervals, often with jitter to avoid pattern detection
Jitter Randomization applied to beacon interval (e.g., 60s +/- 15%) to make the timing pattern less predictable and harder to detect
Malleable C2 Cobalt Strike feature allowing operators to customize all aspects of C2 traffic (URIs, headers, encoding) to mimic legitimate services
Dead Drop Intermediate location (paste site, cloud storage, social media) where C2 commands are posted for the malware to retrieve
Domain Fronting Using a trusted CDN domain in the TLS SNI while routing to a different backend, making C2 traffic appear to go to a legitimate service
Fast Flux Rapidly changing DNS records for C2 domains to distribute across many IPs and resist takedown efforts
C2 Framework Software toolkit providing C2 server, implant generator, and operator interface (Cobalt Strike, Metasploit, Sliver, Covenant)

Tools & Systems

  • Wireshark: Packet analyzer for detailed C2 protocol analysis at the packet level
  • RITA (Real Intelligence Threat Analytics): Open-source tool analyzing Zeek logs for beacon detection and DNS tunneling
  • CobaltStrikeParser: Tool extracting Cobalt Strike beacon configuration from samples and memory dumps
  • JA3/JA3S: TLS fingerprinting method for identifying C2 frameworks by their TLS implementation characteristics
  • Shodan/Censys: Internet scanning platforms for mapping C2 infrastructure and identifying related servers

Common Scenarios

Scenario: Reverse Engineering a Custom C2 Protocol

Context: A malware sample communicates with its C2 server using an unknown binary protocol over TCP port 8443. The protocol needs to be decoded to understand the command set and build detection signatures.

Approach:

  1. Filter PCAP for TCP port 8443 conversations and extract the TCP streams
  2. Analyze the first few exchanges to identify the handshake/authentication mechanism
  3. Map the message structure (length prefix, type field, payload encoding)
  4. Cross-reference with Ghidra disassembly of the send/receive functions in the malware
  5. Identify the command dispatcher and document each command code's function
  6. Build a protocol decoder in Python for ongoing traffic analysis
  7. Create Suricata rules matching the protocol handshake or static header bytes

Pitfalls:

  • Assuming the protocol is static; some C2 frameworks negotiate encryption during the handshake
  • Not capturing enough traffic to see all command types (some commands are rare)
  • Missing fallback C2 channels (DNS, ICMP) that activate when the primary channel fails
  • Confusing encrypted payload data with the protocol framing structure

Output Format

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
Source materials

References and resources

Everything below is rendered for inspection. Script files are read-only and never run.

References 1

api-reference.md2.7 KB

API Reference: C2 Communication Analysis Tools

Scapy - Packet Analysis Library (Python)

Reading PCAPs

from scapy.all import rdpcap, IP, TCP, UDP, DNS, DNSQR
packets = rdpcap("capture.pcap")

Filtering Packets

# TCP SYN packets (connection initiation)
syn_pkts = [p for p in packets if TCP in p and (p[TCP].flags & 0x02)]
 
# DNS queries
dns_pkts = [p for p in packets if DNS in p and p[DNS].qr == 0]
 
# Access fields
pkt[IP].src        # Source IP
pkt[IP].dst        # Destination IP
pkt[TCP].sport     # Source port
pkt[TCP].dport     # Destination port
pkt[TCP].flags     # TCP flags (0x02 = SYN)
float(pkt.time)    # Packet timestamp

dpkt - Packet Parsing Library (Python)

Reading PCAPs

import dpkt
with open("capture.pcap", "rb") as f:
    pcap = dpkt.pcap.Reader(f)
    for timestamp, buf in pcap:
        eth = dpkt.ethernet.Ethernet(buf)
        ip = eth.data
        tcp = ip.data

HTTP Request Parsing

http = dpkt.http.Request(tcp.data)
http.method     # GET, POST
http.uri        # /path
http.headers    # dict of headers
http.body       # POST body

tshark - CLI Wireshark

Beacon Analysis

tshark -r capture.pcap -T fields -e ip.dst -e tcp.dstport -e frame.time_epoch \
  -Y "tcp.flags.syn==1" > syn_times.csv

HTTP Extraction

tshark -r capture.pcap -Y "http.request" -T fields \
  -e http.request.method -e http.host -e http.request.uri -e http.user_agent

DNS Extraction

tshark -r capture.pcap -Y "dns.qr==0" -T fields \
  -e dns.qry.name -e dns.qry.type -e ip.src

JA3 TLS Fingerprinting

tshark -r capture.pcap -Y "tls.handshake.type==1" -T fields \
  -e ip.src -e tls.handshake.ja3

CobaltStrikeParser - Beacon Config Extraction

Usage

from cobalt_strike_parser import BeaconConfig
config = BeaconConfig.from_file("beacon.bin")
for key, value in config.items():
    print(f"{key}: {value}")

Key Config Fields

Field Description
BeaconType HTTP, HTTPS, DNS, SMB
C2Server Primary C2 URL
SleepTime Beacon interval (ms)
Jitter Jitter percentage
UserAgent HTTP User-Agent string
Watermark License watermark ID

Suricata - Network IDS Rules

Rule Syntax

alert <proto> <src> <port> -> <dst> <port> (msg:""; <options>; sid:N; rev:N;)

Key Keywords

Keyword Purpose
http.method Match HTTP method
http.uri Match request URI
http.header Match header content
ja3.hash Match JA3 TLS fingerprint
dns.query Match DNS query name
tls.cert_subject Match TLS certificate CN
threshold Rate-based detection

Scripts 1

agent.py7.7 KB
Display-only source. This catalog never executes bundled scripts.
#!/usr/bin/env python3
"""C2 communication analysis agent for beacon detection and protocol decoding."""

import statistics
import base64
import os
import sys
from collections import defaultdict

try:
    from scapy.all import rdpcap, IP, TCP, DNS, DNSQR
    HAS_SCAPY = True
except ImportError:
    HAS_SCAPY = False

try:
    import dpkt
    HAS_DPKT = True
except ImportError:
    HAS_DPKT = False


def detect_beacons(pcap_path, min_connections=5, max_jitter_pct=25.0):
    """Analyze PCAP for periodic beacon patterns using TCP SYN timing."""
    if not HAS_SCAPY:
        print("[ERROR] scapy not installed: pip install scapy")
        return []
    packets = rdpcap(pcap_path)
    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))
    beacons = []
    for dst, times in sorted(connections.items()):
        if len(times) < min_connections:
            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
        is_beacon = 5 < avg_interval < 7200 and jitter_pct < max_jitter_pct
        record = {
            "destination": dst,
            "connections": len(times),
            "duration_seconds": round(times[-1] - times[0], 1),
            "avg_interval_seconds": round(avg_interval, 1),
            "stdev_seconds": round(stdev, 1),
            "jitter_percent": round(jitter_pct, 1),
            "is_beacon": is_beacon,
        }
        if is_beacon:
            beacons.append(record)
    return beacons


def extract_http_requests(pcap_path):
    """Extract HTTP requests from a PCAP file using dpkt."""
    if not HAS_DPKT:
        print("[ERROR] dpkt not installed: pip install dpkt")
        return []
    requests = []
    with open(pcap_path, "rb") as f:
        pcap = dpkt.pcap.Reader(f)
        for ts, buf in pcap:
            try:
                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 len(tcp.data) == 0:
                    continue
                try:
                    http = dpkt.http.Request(tcp.data)
                    decoded_body = None
                    if http.body:
                        try:
                            decoded_body = base64.b64decode(http.body).decode("utf-8", errors="replace")
                        except Exception:
                            decoded_body = http.body[:200]
                    requests.append({
                        "timestamp": ts,
                        "src_ip": ".".join(str(b) for b in ip.src),
                        "dst_ip": ".".join(str(b) for b in ip.dst),
                        "dst_port": tcp.dport,
                        "method": http.method,
                        "uri": http.uri,
                        "host": http.headers.get("host", ""),
                        "user_agent": http.headers.get("user-agent", ""),
                        "body_size": len(http.body) if http.body else 0,
                        "decoded_body_preview": decoded_body,
                    })
                except (dpkt.dpkt.NeedData, dpkt.dpkt.UnpackError):
                    pass
            except Exception:
                continue
    return requests


def extract_dns_queries(pcap_path):
    """Extract DNS queries from a PCAP for C2 domain identification."""
    if not HAS_SCAPY:
        return []
    packets = rdpcap(pcap_path)
    queries = []
    for pkt in packets:
        if DNS in pkt and pkt[DNS].qr == 0 and DNSQR in pkt:
            qname = pkt[DNSQR].qname.decode("utf-8", errors="replace").rstrip(".")
            queries.append({
                "src_ip": pkt[IP].src if IP in pkt else "?",
                "query": qname,
                "type": pkt[DNSQR].qtype,
            })
    return queries


def identify_c2_framework(http_requests):
    """Match HTTP request patterns against known C2 framework signatures."""
    cs_uris = ["/pixel", "/submit.php", "/__utm.gif", "/ca", "/dpixel",
               "/push", "/visit.js", "/tab_icon"]
    framework_hits = []
    for req in http_requests:
        uri = req.get("uri", "")
        ua = req.get("user_agent", "")
        for cs_uri in cs_uris:
            if cs_uri in uri:
                framework_hits.append({
                    "framework": "Cobalt Strike",
                    "indicator": f"URI pattern: {cs_uri}",
                    "request": req,
                })
                break
        if "MeterSSL" in ua or len(uri) == 5 and uri.startswith("/"):
            framework_hits.append({
                "framework": "Metasploit/Meterpreter",
                "indicator": f"URI/UA pattern: {uri} / {ua[:50]}",
                "request": req,
            })
    return framework_hits


def generate_suricata_rules(beacons, http_requests):
    """Generate Suricata IDS rules from observed C2 patterns."""
    rules = []
    sid = 9000100
    for beacon in beacons:
        dst_ip, dst_port = beacon["destination"].rsplit(":", 1)
        rules.append(
            f'alert tcp $HOME_NET any -> {dst_ip} {dst_port} ('
            f'msg:"MALWARE Detected C2 Beacon to {dst_ip}:{dst_port}"; '
            f'flow:established,to_server; '
            f'threshold:type threshold, track by_src, count 5, seconds 600; '
            f'sid:{sid}; rev:1;)'
        )
        sid += 1
    for req in http_requests[:5]:
        if req.get("uri"):
            uri = req["uri"]
            rules.append(
                f'alert http $HOME_NET any -> $EXTERNAL_NET any ('
                f'msg:"MALWARE Suspected C2 HTTP Request {uri}"; '
                f'flow:established,to_server; '
                f'http.method; content:"{req["method"]}"; '
                f'http.uri; content:"{uri}"; '
                f'sid:{sid}; rev:1;)'
            )
            sid += 1
    return rules


if __name__ == "__main__":
    print("=" * 60)
    print("C2 Communication Analysis Agent")
    print("Beacon detection, protocol decoding, signature generation")
    print("=" * 60)

    pcap_file = sys.argv[1] if len(sys.argv) > 1 else None

    if pcap_file and os.path.exists(pcap_file):
        print(f"\n[*] Analyzing PCAP: {pcap_file}")

        print("\n--- Beacon Detection ---")
        beacons = detect_beacons(pcap_file)
        for b in beacons:
            print(f"[!] BEACON: {b['destination']} "
                  f"interval={b['avg_interval_seconds']}s "
                  f"jitter={b['jitter_percent']}% "
                  f"sessions={b['connections']}")

        print("\n--- HTTP Requests ---")
        http_reqs = extract_http_requests(pcap_file)
        for r in http_reqs[:10]:
            print(f"  {r['method']} {r['host']}{r['uri']}")

        print("\n--- DNS Queries ---")
        dns_qs = extract_dns_queries(pcap_file)
        for q in dns_qs[:10]:
            print(f"  {q['src_ip']} -> {q['query']}")

        print("\n--- C2 Framework Identification ---")
        hits = identify_c2_framework(http_reqs)
        for h in hits:
            print(f"[!] {h['framework']}: {h['indicator']}")

        print("\n--- Suricata Rules ---")
        rules = generate_suricata_rules(beacons, http_reqs)
        for r in rules:
            print(r)
    else:
        print("\n[DEMO] Usage: python agent.py <capture.pcap>")
        print("[*] Provide a PCAP file to analyze for C2 communication patterns.")
Keep exploring