Security Skills Atlas
threat hunting

Hunting for Domain Fronting C2 Traffic

Detect domain fronting C2 traffic by analyzing SNI vs HTTP Host header mismatches in proxy logs and TLS certificate discrepancies using pyOpenSSL for certificate inspection

c2-detectiondomain-frontingnetwork-securityproxy-logspyopensslthreat-huntingtls-inspection
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 siteT1082 on the official MITRE ATT&CK siteT1083 on the official MITRE ATT&CK siteT1090.004 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

MITRE D3FEND

Application Protocol Command Analysis on the official MITRE D3FEND siteClient-server Payload Profiling on the official MITRE D3FEND siteNetwork Isolation on the official MITRE D3FEND siteNetwork Traffic Analysis on the official MITRE D3FEND siteNetwork Traffic Community Deviation on the official MITRE D3FEND site

Overview

Domain fronting (MITRE ATT&CK T1090.004) is a technique where attackers use different domain names in the TLS SNI field and the HTTP Host header to disguise C2 traffic behind legitimate CDN-hosted domains. This skill detects domain fronting by parsing proxy/web gateway logs for SNI-Host header mismatches, analyzing TLS certificates for CDN provider identification, flagging connections where the SNI points to a high-reputation domain but the Host header targets an attacker-controlled domain, and correlating with known CDN provider IP ranges.

When to Use

  • When investigating security incidents that require hunting for domain fronting c2 traffic
  • 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

  • Web proxy or secure web gateway logs with SNI and Host header fields
  • Python 3.8+ with pyOpenSSL and cryptography libraries
  • TLS inspection enabled on proxy for Host header visibility
  • CDN provider IP range lists (CloudFront, Azure CDN, Cloudflare)

Steps

  1. Parse proxy logs for connections with both SNI and Host header fields
  2. Compare SNI domain against HTTP Host header for mismatches
  3. Extract TLS certificate Subject and SAN fields using pyOpenSSL
  4. Identify CDN-hosted connections via certificate issuer and IP ranges
  5. Flag high-confidence domain fronting where SNI and Host differ on CDN IPs
  6. Score alerts based on domain reputation differential
  7. Generate detection report with network flow context

Expected Output

JSON report containing detected domain fronting indicators with SNI-Host pairs, certificate details, CDN provider identification, confidence scores, and MITRE ATT&CK technique mapping.

Source materials

References and resources

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

References 1

api-reference.md2.1 KB

Domain Fronting C2 Traffic Detection API Reference

Domain Fronting Mechanism

TLS ClientHello:  SNI = legitimate-cdn-domain.cloudfront.net
HTTP Request:     Host: attacker-c2-server.evil.com

The CDN accepts the TLS connection based on SNI, then routes the HTTP request to the backend specified in the Host header. Network monitoring sees only the legitimate SNI domain.

MITRE ATT&CK

Technique ID Description
Proxy: Domain Fronting T1090.004 Route C2 through CDN using SNI/Host mismatch

CDN Provider Identification

Certificate Issuers

CDN Certificate CN Pattern
CloudFront *.cloudfront.net
Azure CDN *.azureedge.net
Cloudflare sni.cloudflaressl.com
Akamai *.akamaiedge.net
Fastly *.fastly.net

Proxy Log Detection

Squid Proxy Log Fields

timestamp src_ip CONNECT sni:443 -> status Host: host_header

Palo Alto Threat ID

Threat ID 86467: Domain fronting detected (SNI/Host mismatch)

Splunk Detection Query

index=proxy sourcetype=squid OR sourcetype=bluecoat
| eval sni_root=mvindex(split(sni, "."), -2) + "." + mvindex(split(sni, "."), -1)
| eval host_root=mvindex(split(host_header, "."), -2) + "." + mvindex(split(host_header, "."), -1)
| where sni_root != host_root
| stats count by sni, host_header, src_ip
| sort -count

pyOpenSSL Certificate Inspection

from OpenSSL import crypto
import ssl, socket
 
ctx = ssl.create_default_context()
with ctx.wrap_socket(socket.socket(), server_hostname=hostname) as s:
    s.connect((hostname, 443))
    der_cert = s.getpeercert(True)
 
x509 = crypto.load_certificate(crypto.FILETYPE_ASN1, der_cert)
subject_cn = x509.get_subject().CN
issuer_cn = x509.get_issuer().CN
 
for i in range(x509.get_extension_count()):
    ext = x509.get_extension(i)
    if ext.get_short_name() == b"subjectAltName":
        print(str(ext))  # DNS:*.cloudfront.net, DNS:cloudfront.net

CLI Usage

python agent.py --proxy-log squid_access.csv --output fronting_report.json
python agent.py --proxy-log logs.csv --check-certs

Scripts 1

agent.py6.5 KB
Display-only source. This catalog never executes bundled scripts.
#!/usr/bin/env python3
"""Detect domain fronting C2 traffic via SNI/Host header mismatch and TLS certificate analysis."""

import json
import csv
import ssl
import socket
import argparse
from collections import defaultdict
from datetime import datetime

try:
    from OpenSSL import crypto
    HAS_PYOPENSSL = True
except ImportError:
    HAS_PYOPENSSL = False


CDN_PROVIDERS = {
    "cloudfront.net": "Amazon CloudFront",
    "azureedge.net": "Azure CDN",
    "cloudflare.com": "Cloudflare",
    "akamaiedge.net": "Akamai",
    "fastly.net": "Fastly",
    "googleapis.com": "Google Cloud CDN",
    "azurefd.net": "Azure Front Door",
}


def load_proxy_logs(filepath):
    """Load proxy logs CSV with columns: timestamp, src_ip, sni, host_header, dst_ip, dst_port, method, url, status."""
    records = []
    with open(filepath, newline="", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        for row in reader:
            records.append({
                "timestamp": row.get("timestamp", ""),
                "src_ip": row.get("src_ip", row.get("c-ip", "")),
                "sni": row.get("sni", row.get("cs-ssl-sni", "")).lower().strip(),
                "host_header": row.get("host_header", row.get("cs-host", "")).lower().strip(),
                "dst_ip": row.get("dst_ip", row.get("s-ip", "")),
                "dst_port": int(row.get("dst_port", row.get("s-port", "443"))),
                "method": row.get("method", row.get("cs-method", "")),
                "url": row.get("url", row.get("cs-uri-stem", "")),
                "status": row.get("status", row.get("sc-status", "")),
                "bytes": int(row.get("bytes", row.get("sc-bytes", "0"))),
            })
    return records


def extract_domain_root(domain):
    """Extract root domain from FQDN (e.g., sub.example.com -> example.com)."""
    parts = domain.rstrip(".").split(".")
    return ".".join(parts[-2:]) if len(parts) >= 2 else domain


def identify_cdn_provider(domain):
    """Check if a domain belongs to a known CDN provider."""
    for cdn_suffix, provider in CDN_PROVIDERS.items():
        if domain.endswith(cdn_suffix):
            return provider
    return None


def detect_sni_host_mismatch(records):
    """Detect connections where SNI and Host header point to different domains."""
    alerts = []
    for rec in records:
        sni = rec["sni"]
        host = rec["host_header"]
        if not sni or not host:
            continue
        sni_root = extract_domain_root(sni)
        host_root = extract_domain_root(host)
        if sni_root != host_root:
            cdn = identify_cdn_provider(sni) or identify_cdn_provider(host)
            confidence = "high" if cdn else "medium"
            alerts.append({
                "detection": "SNI/Host Header Mismatch",
                "mitre_technique": "T1090.004",
                "timestamp": rec["timestamp"],
                "src_ip": rec["src_ip"],
                "sni": sni,
                "host_header": host,
                "sni_root": sni_root,
                "host_root": host_root,
                "cdn_provider": cdn,
                "dst_ip": rec["dst_ip"],
                "confidence": confidence,
                "severity": "critical" if cdn else "high",
                "description": f"Domain fronting: SNI={sni} but Host={host}",
            })
    return alerts


def get_tls_certificate_info(hostname, port=443, timeout=5):
    """Retrieve TLS certificate details for a given hostname using pyOpenSSL."""
    if not HAS_PYOPENSSL:
        return {"error": "pyOpenSSL not installed"}
    try:
        ctx = ssl.create_default_context()
        with ctx.wrap_socket(socket.socket(), server_hostname=hostname) as s:
            s.settimeout(timeout)
            s.connect((hostname, port))
            der_cert = s.getpeercert(True)
        x509 = crypto.load_certificate(crypto.FILETYPE_ASN1, der_cert)
        subject = dict(x509.get_subject().get_components())
        issuer = dict(x509.get_issuer().get_components())
        san_list = []
        for i in range(x509.get_extension_count()):
            ext = x509.get_extension(i)
            if ext.get_short_name() == b"subjectAltName":
                san_list = [s.strip().replace("DNS:", "") for s in str(ext).split(",")]
        return {
            "subject_cn": subject.get(b"CN", b"").decode(),
            "issuer_cn": issuer.get(b"CN", b"").decode(),
            "issuer_o": issuer.get(b"O", b"").decode(),
            "san": san_list[:20],
            "not_before": str(x509.get_notBefore()),
            "not_after": str(x509.get_notAfter()),
            "serial": str(x509.get_serial_number()),
        }
    except Exception as e:
        return {"error": str(e)}


def analyze_fronting_pairs(alerts):
    """Aggregate and rank domain fronting pairs by frequency."""
    pair_counts = defaultdict(int)
    for a in alerts:
        pair_counts[(a["sni"], a["host_header"])] += 1
    ranked = sorted(pair_counts.items(), key=lambda x: -x[1])
    return [{"sni": p[0], "host": p[1], "count": c} for (p, c) in ranked[:20]]


def main():
    parser = argparse.ArgumentParser(description="Domain Fronting C2 Traffic Hunter")
    parser.add_argument("--proxy-log", required=True, help="CSV proxy log with SNI and Host header fields")
    parser.add_argument("--check-certs", action="store_true", help="Fetch TLS certs for top fronting domains")
    parser.add_argument("--output", default="domain_fronting_report.json", help="Output report path")
    args = parser.parse_args()

    records = load_proxy_logs(args.proxy_log)
    print(f"[+] Loaded {len(records)} proxy log entries")

    alerts = detect_sni_host_mismatch(records)
    print(f"[+] Detected {len(alerts)} SNI/Host mismatches")

    fronting_pairs = analyze_fronting_pairs(alerts)
    cert_info = {}
    if args.check_certs and fronting_pairs:
        for pair in fronting_pairs[:5]:
            cert_info[pair["sni"]] = get_tls_certificate_info(pair["sni"])

    report = {
        "analysis_time": datetime.utcnow().isoformat() + "Z",
        "total_proxy_entries": len(records),
        "detections": alerts[:50],
        "total_mismatches": len(alerts),
        "fronting_pairs_ranked": fronting_pairs,
        "certificate_analysis": cert_info,
        "mitre_technique": "T1090.004",
        "cdn_involved": list({a["cdn_provider"] for a in alerts if a.get("cdn_provider")}),
    }

    with open(args.output, "w") as f:
        json.dump(report, f, indent=2)
    print(f"[+] Top fronting pairs: {len(fronting_pairs)}")
    print(f"[+] Report saved to {args.output}")


if __name__ == "__main__":
    main()
Keep exploring
Hunting for Beaconing with Frequency AnalysisIdentify command-and-control beaconing patterns in network traffic by applying statistical frequency analysis, jitter calculation, and coefficient of variation scoring to detect periodic callbacks from compromised endpoints.Analyzing Persistence Mechanisms in LinuxDetect and analyze Linux persistence mechanisms including crontab entries, systemd service units, LD_PRELOAD hijacking, bashrc modifications, and authorized_keys backdoors using auditd and file integrity monitoringAnalyzing 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.Building Threat Hunt Hypothesis FrameworkBuild a systematic threat hunt hypothesis framework that transforms threat intelligence, attack patterns, and environmental data into testable hunting hypotheses.Detecting DCSync Attack in Active DirectoryDetect DCSync attacks where adversaries abuse Active Directory replication privileges to extract password hashes by monitoring for non-domain-controller accounts requesting directory replication via DsGetNCChanges.Detecting DLL Sideloading AttacksDetect DLL side-loading attacks where adversaries place malicious DLLs alongside legitimate applications to hijack execution flow for defense evasion.