Security Skills Atlas
threat hunting

Hunting for Cobalt Strike Beacons

Detect Cobalt Strike beacon network activity using default TLS certificate signatures (serial 8BB00EE), JA3/JA3S/JARM fingerprints, HTTP C2 profile pattern matching, beacon jitter analysis, and named pipe detection via Zeek, Suricata, and Python PCAP analysis.

beaconc2cobalt-strikeja3jarmnetwork-forensicssuricatathreat-hunting
Install this skill
npx skills add mukul975/Anthropic-Cybersecurity-Skills
Framework mappings

MITRE ATT&CK

T1046 on the official MITRE ATT&CK siteT1057 on the official MITRE ATT&CK siteT1071 on the official MITRE ATT&CK siteT1071.001 on the official MITRE ATT&CK siteT1082 on the official MITRE ATT&CK siteT1083 on the official MITRE ATT&CK siteT1090 on the official MITRE ATT&CK siteT1105 on the official MITRE ATT&CK siteT1571 on the official MITRE ATT&CK siteT1573.002 on the official MITRE ATT&CK site

NIST CSF 2.0

DE.AE-02 on the official NIST CSF 2.0 siteDE.AE-07 on the official NIST CSF 2.0 siteDE.CM-01 on the official NIST CSF 2.0 siteID.RA-05 on the official NIST CSF 2.0 site

Overview

Cobalt Strike is the most prevalent command-and-control framework used by both red teams and threat actors. Beacon, its primary payload, communicates with team servers using configurable HTTP/HTTPS/DNS profiles that can mimic legitimate traffic. However, default configurations and behavioral patterns remain detectable through TLS certificate analysis (default serial 8BB00EE), JA3/JA3S fingerprinting, beacon interval jitter analysis, and HTTP malleable profile pattern matching. This skill covers building detection capabilities using Zeek network logs, Suricata IDS rules, and Python-based PCAP analysis to identify beacon callbacks in network traffic.

When to Use

  • When investigating security incidents that require hunting for cobalt strike beacons
  • When building detection rules or threat hunting queries for this domain
  • When SOC analysts need structured procedures for this analysis type
  • When validating security monitoring coverage for related attack techniques

Prerequisites

  • Zeek 6.0+ with JA3 and HASSH packages installed
  • Suricata 7.0+ with Emerging Threats ruleset
  • Python 3.9+ with scapy and dpkt libraries
  • Network traffic captures (PCAP) or live Zeek logs
  • RITA (Real Intelligence Threat Analytics) for beacon scoring
  • Threat intelligence feeds with known Cobalt Strike IOCs

Steps

Step 1: TLS Certificate Analysis

Detect default Cobalt Strike certificates using JA3S fingerprints, certificate serial numbers, and JARM fingerprints in Zeek ssl.log.

Step 2: Beacon Interval Analysis

Analyze connection timing patterns to identify regular callback intervals with configurable jitter, characteristic of beacon behavior.

Step 3: HTTP Profile Detection

Match HTTP request patterns (URI paths, headers, user-agents) against known malleable C2 profiles.

Step 4: Correlate and Score

Combine multiple indicators (TLS + timing + HTTP profile) into a composite beacon confidence score.

Expected Output

JSON report containing detected beacon candidates with confidence scores, TLS fingerprints, timing analysis, HTTP profile matches, and recommended response actions.

Source materials

References and resources

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

References 1

api-reference.md4.6 KB

API Reference: Hunting for Cobalt Strike Beacons

Cobalt Strike Default TLS Indicators

Indicator Value Detection Confidence
Default cert serial 8BB00EE 95% (unmodified teamserver)
Default cert issuer Major Cobalt Strike 95%
JA3S hash (Java TLS) ae4edc6faf64d08308082ad26be60767 80%
JA3S hash (alt) a0e9f5d64349fb13191bc781f81f42e1 80%
JARM fingerprint 07d14d16d21d21d07c42d41d00041d24a458a375eef0c576d23a7bab9a9fb1 90%

Zeek Log Fields for Detection

ssl.log Key Fields

Field Index Name Use
0 ts Connection timestamp
2 id.orig_h Source IP
4 id.resp_h Destination IP (C2 server)
5 id.resp_p Destination port
20 cert_chain_fps Certificate serial number
21 ja3s JA3S server fingerprint hash

conn.log Beacon Timing Fields

Field Index Name Use
0 ts Connection epoch timestamp
2 id.orig_h Beaconing host
4 id.resp_h C2 destination
5 id.resp_p C2 port
8 duration Session length
9 orig_bytes Bytes sent (check size)
10 resp_bytes Bytes received (check size)

RITA Beacon Analysis

# Import Zeek logs into RITA
rita import /opt/zeek/logs/current rita_dataset
 
# Show beaconing connections ranked by score
rita show-beacons rita_dataset --human-readable
 
# Show long connections (persistent C2)
rita show-long-connections rita_dataset
 
# Export beacon results as CSV
rita show-beacons rita_dataset -H > beacons.csv
 
# Show DNS tunneling (alternate C2 channel)
rita show-exploded-dns rita_dataset

Suricata Detection Rules

# Detect default Cobalt Strike TLS certificate
alert tls any any -> any any (msg:"ET MALWARE Cobalt Strike Default Certificate"; \
  tls.cert_serial; content:"8BB00EE"; sid:2029560; rev:3;)
 
# Detect known Cobalt Strike JA3S
alert tls any any -> any any (msg:"ET MALWARE Cobalt Strike JA3S"; \
  ja3s.hash; content:"ae4edc6faf64d08308082ad26be60767"; sid:2029561; rev:2;)
 
# Detect Cobalt Strike default HTTP beacon URI
alert http any any -> any any (msg:"ET MALWARE CobaltStrike Beacon URI"; \
  content:"GET"; http_method; pcre:"/^\/[a-zA-Z]{4}$/U"; sid:2029562; rev:1;)
 
# Detect Cobalt Strike named pipe (SMB beacon)
alert smb any any -> any any (msg:"ET MALWARE CobaltStrike Named Pipe"; \
  content:"|MSRPC|"; content:"\\\\pipe\\\\"; content:"MSSE-"; sid:2029563; rev:1;)

Malleable C2 Profile HTTP Indicators

Pattern URI Regex Context
Default GET ^/[a-zA-Z]{4}$ 4-char alpha URI (e.g., /aGth)
submit.php ^/submit\.php\?id=\d+$ POST callback with numeric ID
Pixel tracking ^/pixel\.(gif|png)$ Fake tracking pixel
UTM beacon ^/__utm\.gif$ Mimics Google Analytics
RSS feed ^/updates\.(rss|json)$ Fake feed endpoint
JS beacon ^/visit\.js$ Fake JavaScript resource

Default User-Agent Strings

Mozilla/5.0 (compatible; MSIE 9.0; Windows NT 6.1; Trident/5.0)
Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 5.1)
Mozilla/5.0 (compatible; MSIE 10.0; Windows NT 6.2; WOW64; Trident/6.0)

Beacon Timing Analysis Formula

# Jitter percentage calculation
intervals = [t[i+1] - t[i] for i in range(len(t) - 1)]
avg = sum(intervals) / len(intervals)
std = sqrt(sum((x - avg)**2 for x in intervals) / len(intervals))
jitter_pct = (std / avg) * 100
 
# Beacon score (0-100, higher = more likely beacon)
beacon_score = max(0, 1 - (jitter_pct / 100)) * 100
# Score >= 85 = critical, >= 60 = high suspicion

JARM Scanner CLI

# Scan single host for JARM fingerprint
python3 jarm.py -p 443 suspicious-host.example.com
 
# Known Cobalt Strike JARM
# 07d14d16d21d21d07c42d41d00041d24a458a375eef0c576d23a7bab9a9fb1
 
# Compare against threat intel JARM database
python3 jarm.py -p 8443 10.0.0.50 | grep -f cs_jarm_list.txt

MITRE ATT&CK Mapping

Technique ID Beacon Indicator
Application Layer Protocol T1071.001 HTTP/HTTPS beaconing pattern
Encrypted Channel T1573.002 Default TLS cert / JA3S match
Non-Standard Port T1571 HTTPS on 8080, 8443, 444
Ingress Tool Transfer T1105 Large resp_bytes in beacon
Proxy T1090 Redirector infrastructure

References

Scripts 1

agent.py8.1 KB
Display-only source. This catalog never executes bundled scripts.
#!/usr/bin/env python3
"""Cobalt Strike Beacon Hunter - detects beacon signatures in network traffic and Zeek logs."""

import json
import argparse
import logging
import os
import re
import math
from collections import defaultdict
from datetime import datetime

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)

CS_DEFAULT_CERT_SERIAL = "8bb00ee"
CS_KNOWN_JA3S = [
    "ae4edc6faf64d08308082ad26be60767",
    "a0e9f5d64349fb13191bc781f81f42e1",
]
CS_KNOWN_JARM = "07d14d16d21d21d07c42d41d00041d24a458a375eef0c576d23a7bab9a9fb1"


def parse_zeek_ssl_log(ssl_log_path):
    """Parse Zeek ssl.log for default Cobalt Strike certificates."""
    findings = []
    try:
        with open(ssl_log_path) as f:
            for line in f:
                if line.startswith("#"):
                    continue
                fields = line.strip().split("\t")
                if len(fields) < 22:
                    continue
                ts, uid, src_ip, src_port, dst_ip, dst_port = fields[0], fields[1], fields[2], fields[3], fields[4], fields[5]
                serial = fields[20] if len(fields) > 20 else ""
                ja3s = fields[21] if len(fields) > 21 else ""
                if serial.lower().replace(":", "") == CS_DEFAULT_CERT_SERIAL:
                    findings.append({
                        "indicator": "default_cs_certificate",
                        "src_ip": src_ip, "dst_ip": dst_ip, "dst_port": dst_port,
                        "cert_serial": serial, "timestamp": ts,
                        "severity": "critical", "confidence": 95,
                    })
                if ja3s in CS_KNOWN_JA3S:
                    findings.append({
                        "indicator": "known_cs_ja3s",
                        "src_ip": src_ip, "dst_ip": dst_ip, "dst_port": dst_port,
                        "ja3s_hash": ja3s, "timestamp": ts,
                        "severity": "high", "confidence": 80,
                    })
    except FileNotFoundError:
        logger.warning("Zeek ssl.log not found: %s", ssl_log_path)
    return findings


def analyze_beacon_timing(conn_log_path, min_connections=20, max_jitter_pct=25):
    """Analyze connection timing for beacon-like regular intervals."""
    connections = defaultdict(list)
    try:
        with open(conn_log_path) as f:
            for line in f:
                if line.startswith("#"):
                    continue
                fields = line.strip().split("\t")
                if len(fields) < 7:
                    continue
                ts = float(fields[0])
                src_ip, dst_ip, dst_port = fields[2], fields[4], fields[5]
                key = (src_ip, dst_ip, dst_port)
                connections[key].append(ts)
    except (FileNotFoundError, ValueError):
        logger.warning("Zeek conn.log parse failed: %s", conn_log_path)
        return []
    beacons = []
    for (src, dst, port), timestamps in connections.items():
        if len(timestamps) < min_connections:
            continue
        timestamps.sort()
        intervals = [timestamps[i + 1] - timestamps[i] for i in range(len(timestamps) - 1)]
        if not intervals:
            continue
        avg_interval = sum(intervals) / len(intervals)
        if avg_interval < 1:
            continue
        std_interval = math.sqrt(sum((x - avg_interval) ** 2 for x in intervals) / len(intervals))
        jitter_pct = (std_interval / avg_interval) * 100 if avg_interval > 0 else 100
        if jitter_pct <= max_jitter_pct:
            beacon_score = round(max(0, 1 - (jitter_pct / 100)) * 100, 1)
            if beacon_score >= 60:
                beacons.append({
                    "indicator": "beacon_timing",
                    "src_ip": src, "dst_ip": dst, "dst_port": port,
                    "connections": len(timestamps),
                    "avg_interval_sec": round(avg_interval, 1),
                    "jitter_pct": round(jitter_pct, 1),
                    "beacon_score": beacon_score,
                    "severity": "critical" if beacon_score >= 85 else "high",
                    "confidence": int(beacon_score),
                })
    return sorted(beacons, key=lambda x: x["beacon_score"], reverse=True)


def check_http_profiles(http_log_path):
    """Detect known Cobalt Strike HTTP malleable C2 profile patterns."""
    cs_uri_patterns = [
        r"^/[a-zA-Z]{4}$", r"^/submit\.php\?id=\d+$", r"^/pixel\.(gif|png)$",
        r"^/__utm\.gif$", r"^/updates\.(rss|json)$", r"^/visit\.js$",
    ]
    cs_ua_patterns = [
        "Mozilla/5.0 (compatible; MSIE 9.0; Windows NT 6.1; Trident/5.0)",
        "Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 5.1)",
    ]
    findings = []
    try:
        with open(http_log_path) as f:
            for line in f:
                if line.startswith("#"):
                    continue
                fields = line.strip().split("\t")
                if len(fields) < 12:
                    continue
                src_ip, dst_ip = fields[2], fields[4]
                uri = fields[9] if len(fields) > 9 else ""
                user_agent = fields[12] if len(fields) > 12 else ""
                for pattern in cs_uri_patterns:
                    if re.match(pattern, uri):
                        findings.append({
                            "indicator": "cs_http_profile",
                            "src_ip": src_ip, "dst_ip": dst_ip,
                            "uri": uri, "user_agent": user_agent[:100],
                            "matched_pattern": pattern,
                            "severity": "high", "confidence": 60,
                        })
                        break
                if user_agent in cs_ua_patterns:
                    findings.append({
                        "indicator": "cs_default_user_agent",
                        "src_ip": src_ip, "dst_ip": dst_ip,
                        "user_agent": user_agent,
                        "severity": "high", "confidence": 70,
                    })
    except FileNotFoundError:
        logger.warning("Zeek http.log not found: %s", http_log_path)
    return findings


def generate_report(tls_findings, beacon_findings, http_findings):
    all_findings = tls_findings + beacon_findings + http_findings
    critical = sum(1 for f in all_findings if f.get("severity") == "critical")
    by_dst = defaultdict(int)
    for f in all_findings:
        by_dst[f.get("dst_ip", "")] += 1
    return {
        "timestamp": datetime.utcnow().isoformat(),
        "tls_certificate_hits": len(tls_findings),
        "beacon_timing_detections": len(beacon_findings),
        "http_profile_matches": len(http_findings),
        "total_indicators": len(all_findings),
        "critical_indicators": critical,
        "top_suspect_destinations": dict(sorted(by_dst.items(), key=lambda x: x[1], reverse=True)[:10]),
        "findings": all_findings[:30],
        "cobalt_strike_likely": critical > 0,
    }


def main():
    parser = argparse.ArgumentParser(description="Cobalt Strike Beacon Hunting Agent")
    parser.add_argument("--zeek-dir", required=True, help="Directory containing Zeek log files")
    parser.add_argument("--min-connections", type=int, default=20, help="Minimum connections for beacon analysis")
    parser.add_argument("--max-jitter", type=int, default=25, help="Maximum jitter percentage for beacon scoring")
    parser.add_argument("--output", default="cobalt_strike_hunt_report.json")
    args = parser.parse_args()

    ssl_log = os.path.join(args.zeek_dir, "ssl.log")
    conn_log = os.path.join(args.zeek_dir, "conn.log")
    http_log = os.path.join(args.zeek_dir, "http.log")
    tls_findings = parse_zeek_ssl_log(ssl_log)
    beacon_findings = analyze_beacon_timing(conn_log, args.min_connections, args.max_jitter)
    http_findings = check_http_profiles(http_log)
    report = generate_report(tls_findings, beacon_findings, http_findings)
    with open(args.output, "w") as f:
        json.dump(report, f, indent=2, default=str)
    logger.info("CS Hunt: %d TLS hits, %d beacons, %d HTTP matches, CS likely: %s",
                len(tls_findings), len(beacon_findings), len(http_findings), report["cobalt_strike_likely"])
    print(json.dumps(report, indent=2, default=str))


if __name__ == "__main__":
    main()
Keep exploring
Analyzing PowerShell Empire ArtifactsDetect PowerShell Empire framework artifacts in Windows event logs by identifying Base64 encoded launcher patterns, default user agents, staging URL structures, stager IOCs, and known Empire module signatures in Script Block Logging events.Analyzing Ransomware Network IndicatorsIdentify ransomware network indicators including C2 beaconing patterns, TOR exit node connections, data exfiltration flows, and encryption key exchange via Zeek conn.log and NetFlow analysisHunting for Command and Control BeaconingDetect C2 beaconing patterns in network traffic using frequency analysis, jitter detection, and domain reputation to identify compromised endpoints communicating with adversary infrastructure.Hunting for DNS Tunneling with ZeekDetect DNS tunneling and data exfiltration by analyzing Zeek dns.log for high-entropy subdomain queries, excessive query volume, long query lengths, and unusual DNS record types indicating covert channel communication.Hunting For Living Off The Cloud TechniquesHunt for adversary abuse of legitimate cloud services for C2, data staging, and exfiltration including abuse of Azure, AWS, GCP services, and SaaS platforms.Hunting For Unusual Network ConnectionsHunt for unusual network connections by analyzing outbound traffic patterns, rare destinations, non-standard ports, and anomalous connection frequencies from endpoints.