DNS Tunneling

#!/usr/bin/env python3
"""
DNS tunneling detection from PCAP or Zeek dns.log.
Multiple detection methods: volume, entropy, query length, TXT frequency.
"""
import subprocess, sys, math, re
from collections import defaultdict, Counter

PCAP = sys.argv[1]

def extract_dns(pcap: str) -> list:
    """Extract DNS queries via tshark."""
    result = subprocess.run([
        "tshark", "-r", pcap, "-n",
        "-Y", "dns.flags.response==0",
        "-T", "fields", "-E", "separator=\t",
        "-e", "frame.time_epoch",
        "-e", "ip.src",
        "-e", "dns.qry.name",
        "-e", "dns.qry.type",
    ], capture_output=True, text=True)

    queries = []
    for line in result.stdout.splitlines():
        parts = line.split("\t")
        if len(parts) >= 4:
            try:
                queries.append({
                    "ts": float(parts[0]),
                    "src": parts[1],
                    "name": parts[2].lower().rstrip("."),
                    "qtype": parts[3],
                })
            except ValueError:
                pass
    return queries

def get_sld(name: str) -> str:
    """Extract second-level domain (e.g., 'google' from 'mail.google.com')."""
    parts = name.split(".")
    if len(parts) >= 2:
        return ".".join(parts[-2:])
    return name

def char_entropy(s: str) -> float:
    if not s:
        return 0.0
    counts = Counter(s)
    total = len(s)
    return -sum((c/total)*math.log2(c/total) for c in counts.values())

def is_base32_like(s: str) -> bool:
    """Check if string looks like base32-encoded data."""
    b32_chars = set("abcdefghijklmnopqrstuvwxyz234567=")
    if not s:
        return False
    proportion = sum(1 for c in s.lower() if c in b32_chars) / len(s)
    return proportion > 0.90 and len(s) > 15

def is_base64_like(s: str) -> bool:
    """Check if string looks like base64-encoded data."""
    b64_chars = set("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789+/=-_")
    if not s:
        return False
    proportion = sum(1 for c in s if c in b64_chars) / len(s)
    return proportion > 0.90 and len(s) > 15

# Load queries
queries = extract_dns(PCAP)
print(f"Total DNS queries: {len(queries)}")

# Analysis 1: High query volume per source to single SLD
by_src_sld = defaultdict(list)
for q in queries:
    sld = get_sld(q["name"])
    by_src_sld[(q["src"], sld)].append(q)

print("\n=== High Query Volume per Source×Domain (>50 queries) ===")
for (src, sld), qs in sorted(by_src_sld.items(), key=lambda x: -len(x[1])):
    if len(qs) > 50:
        txt_count = sum(1 for q in qs if q["qtype"] == "16")
        print(f"  {src} → {sld}: {len(qs)} queries "
              f"({txt_count} TXT queries)")

# Analysis 2: Long subdomain labels
print("\n=== Long Subdomain Labels (>40 chars in leftmost label) ===")
seen = set()
for q in queries:
    labels = q["name"].split(".")
    if labels:
        label = labels[0]
        if len(label) > 40 and q["name"] not in seen:
            seen.add(q["name"])
            entropy = char_entropy(label)
            encoded = is_base32_like(label) or is_base64_like(label)
            print(f"  [{q['src']}] len={len(label)} ent={entropy:.2f} "
                  f"{'BASE-ENCODED' if encoded else ''} {q['name'][:70]}")

# Analysis 3: Reconstruct tunneled data
print("\n=== Attempting DNS Tunnel Stream Reconstruction ===")
# dnscat2 pattern: session_id (hex) + encoded data + fixed domain
for (src, sld), qs in by_src_sld.items():
    if len(qs) < 20:
        continue
    # Check if queries follow dnscat2-like pattern
    payloads = []
    for q in qs:
        labels = q["name"].split(".")
        if len(labels) >= 3:
            # First label might be session ID + data
            first = labels[0]
            # dnscat2: starts with hex session ID (4-8 chars) then data
            if re.match(r'^[0-9a-f]{4,8}', first) and len(first) > 10:
                payload_part = first[8:]  # Skip session ID
                payloads.append(payload_part)

    if len(payloads) > 10:
        # Try to decode accumulated base32
        accumulated = "".join(payloads[:20])
        try:
            import base64
            decoded = base64.b32decode(accumulated.upper() + "=" * ((8 - len(accumulated) % 8) % 8))
            print(f"  Possible dnscat2 stream: {src} → {sld}")
            print(f"  Decoded sample (first 50 bytes): {decoded[:50]}"  )
        except Exception:
            pass