detecting-command-and-control-over-dns

What this skill does

# Detecting Command and Control Over DNS

## When to Use

- Investigating suspected DNS tunneling used for C2 communication or data exfiltration
- Analyzing DNS query logs for signs of encoded payloads in subdomain strings
- Classifying domains as DGA-generated vs. legitimate using statistical or ML methods
- Detecting DNS beaconing patterns (regular intervals, consistent query sizes)
- Hunting for Iodine, dnscat2, dns2tcp, Cobalt Strike DNS, or Sliver DNS traffic
- Monitoring TXT record abuse for command delivery or staged payload download
- Building DNS anomaly detection rules for SOC/SIEM deployment

**Do not use** for general DNS performance monitoring or DNS configuration auditing; use DNS health monitoring tools for those. For HTTP/HTTPS-based C2 detection, use network traffic analysis skills focused on web protocols.

**DISCLAIMER**: DNS tunneling tools referenced in this skill (Iodine, dnscat2, dns2tcp) are dual-use. They have legitimate uses (bypassing captive portals, security research) and malicious uses (C2 channels, exfiltration). Only deploy detection in networks you are authorized to monitor. Testing tunneling tools requires explicit authorization.

## Prerequisites

- DNS query logs from recursive resolver, Zeek/Bro, Suricata, or passive DNS tap
- Python 3.9+ with `numpy`, `scikit-learn`, `pandas`, `tldextract`, and `dnspython`
- Zeek (formerly Bro) with dns.log output or Suricata with DNS EVE JSON logging
- SIEM access (Splunk, Elastic, Microsoft Sentinel) for log correlation
- Passive DNS database access (CIRCL pDNS, Farsight DNSDB, or internal) for enrichment
- Wireshark/tshark for packet-level DNS inspection
- Known-good domain whitelist (Alexa/Tranco top 1M or Majestic Million)

## Workflow

### Step 1: Collect and Parse DNS Query Logs

Ingest DNS traffic from network sensors and parse into analyzable format:

```bash
# Zeek - extract dns.log fields
# Default Zeek dns.log columns:
# ts uid id.orig_h id.orig_p id.resp_h id.resp_p proto trans_id rtt query
# qclass qclass_name qtype qtype_name rcode rcode_name AA TC RD RA Z
# answers TTLs rejected

# Filter for potentially suspicious record types
cat dns.log | zeek-cut ts id.orig_h query qtype_name answers rcode_name | \
    grep -E "TXT|NULL|CNAME|MX" > suspicious_qtypes.log

# Extract unique queried domains
cat dns.log | zeek-cut query | sort -u > unique_domains.txt

# Suricata EVE JSON - extract DNS events
cat eve.json | jq -r 'select(.event_type=="dns") |
    [.timestamp, .src_ip, .dns.rrname, .dns.rrtype, .dns.rcode] |
    @tsv' > dns_events.tsv

# tshark - extract DNS queries from pcap
tshark -r capture.pcap -T fields \
    -e frame.time -e ip.src -e ip.dst \
    -e dns.qry.name -e dns.qry.type \
    -e dns.resp.type -e dns.txt \
    -Y "dns" > dns_queries.tsv

# Count queries per domain (find high-volume destinations)
cat dns.log | zeek-cut query | \
    awk -F. '{print $(NF-1)"."$NF}' | \
    sort | uniq -c | sort -rn | head -50
```

### Step 2: Shannon Entropy Analysis of DNS Queries

Calculate entropy of subdomain strings to identify encoded/encrypted data:

```python
#!/usr/bin/env python3
"""Shannon entropy analysis for DNS query subdomains."""

import math
import csv
import sys
from collections import Counter

try:
    import tldextract
    HAS_TLDEXTRACT = True
except ImportError:
    HAS_TLDEXTRACT = False


def shannon_entropy(data):
    """Calculate Shannon entropy of a string (bits per character)."""
    if not data:
        return 0.0
    counter = Counter(data)
    length = len(data)
    entropy = -sum(
        (count / length) * math.log2(count / length)
        for count in counter.values()
    )
    return entropy


def extract_subdomain(fqdn):
    """Extract the subdomain portion from a fully qualified domain name."""
    if HAS_TLDEXTRACT:
        ext = tldextract.extract(fqdn)
        if ext.subdomain:
            return ext.subdomain, f"{ext.domain}.{ext.suffix}"
        return "", f"{ext.domain}.{ext.suffix}"
    else:
        # Fallback: assume last two labels are domain + TLD
        parts = fqdn.rstrip(".").split(".")
        if len(parts) > 2:
            return ".".join(parts[:-2]), ".".join(parts[-2:])
        return "", fqdn


def analyze_dns_entropy(queries, entropy_threshold=3.5, length_threshold=30):
    """
    Analyze DNS queries for tunneling indicators using entropy.

    Thresholds (tunable per environment):
      - entropy_threshold: Shannon entropy above this flags as suspicious (3.5-4.0 typical)
      - length_threshold: Subdomain length above this flags as suspicious (30-50 chars)

    Returns list of flagged queries with scores.
    """
    results = []

    for query_record in queries:
        fqdn = query_record.get("query", "").lower().rstrip(".")
        if not fqdn:
            continue

        subdomain, base_domain = extract_subdomain(fqdn)
        if not subdomain:
            continue

        # Remove dots from subdomain for entropy calculation
        subdomain_flat = subdomain.replace(".", "")
        if not subdomain_flat:
            continue

        entropy = shannon_entropy(subdomain_flat)
        length = len(subdomain_flat)
        label_count = subdomain.count(".") + 1

        # Scoring: higher = more suspicious
        score = 0.0
        flags = []

        if entropy > entropy_threshold:
            score += (entropy - entropy_threshold) * 25
            flags.append(f"high_entropy:{entropy:.2f}")

        if length > length_threshold:
            score += (length - length_threshold) * 0.5
            flags.append(f"long_subdomain:{length}")

        if label_count > 4:
            score += label_count * 2
            flags.append(f"many_labels:{label_count}")

        # Check for hex/base32/base64 encoding patterns
        hex_ratio = sum(1 for c in subdomain_flat if c in "0123456789abcdef") / max(len(subdomain_flat), 1)
        if hex_ratio > 0.85 and length > 20:
            score += 20
            flags.append("hex_encoded")

        b32_chars = set("abcdefghijklmnopqrstuvwxyz234567")
        b32_ratio = sum(1 for c in subdomain_flat if c in b32_chars) / max(len(subdomain_flat), 1)
        if b32_ratio > 0.95 and length > 20:
            score += 15
            flags.append("base32_encoded")

        # Only report if at least one flag triggered
        if flags:
            results.append({
                "fqdn": fqdn,
                "subdomain": subdomain,
                "base_domain": base_domain,
                "entropy": round(entropy, 4),
                "subdomain_length": length,
                "label_count": label_count,
                "score": round(score, 2),
                "flags": flags,
                "src_ip": query_record.get("src_ip", ""),
                "timestamp": query_record.get("timestamp", ""),
                "qtype": query_record.get("qtype", ""),
            })

    # Sort by score descending
    results.sort(key=lambda x: x["score"], reverse=True)
    return results


# Thresholds for known tunneling tools
TOOL_SIGNATURES = {
    "iodine": {
        "subdomain_pattern": r"^[a-z0-9]{50,}$",  # Long hex-like subdomains
        "common_qtypes": ["NULL", "TXT", "CNAME", "MX", "A"],
        "typical_entropy": (3.8, 4.2),
        "description": "Iodine DNS tunnel - IPv4 over DNS, uses NULL/TXT records",
    },
    "dnscat2": {
        "subdomain_pattern": r"^dnscat\.|^[a-f0-9]{16,}",
        "common_qtypes": ["TXT", "CNAME", "MX", "A"],
        "typical_entropy": (3.5, 4.5),
        "description": "dnscat2 encrypted C2 channel over DNS",
    },
    "dns2tcp": {
        "subdomain_pattern": r"^[a-z2-7]{20,}",  # Base32 encoding
        "common_qtypes": ["TXT", "KEY"],
        "typical_entropy": (3.6, 4.0),
        "description": "dns2tcp tunnel - TCP over DNS using TXT/KEY records",
    },
    "cobalt_strike_dns": {
        "subdomain_pattern": r"^[a-f0-9]{12,}\.",
        "common_qtypes": ["A", "AAAA", "TXT"],
        "typical_entropy": (3.2, 4.0),
        "description": "Co