DNS Forensics

#!/bin/bash
PCAP="$1"

echo "=== DNS Query Volume by Destination ==="
tshark -r "$PCAP" -n -Y "dns.flags.response==0" -T fields -e ip.dst \
  | sort | uniq -c | sort -rn | head -20

echo ""
echo "=== Top queried domains ==="
tshark -r "$PCAP" -n -Y "dns.flags.response==0" -T fields -e dns.qry.name \
  | sort | uniq -c | sort -rn | head -50

echo ""
echo "=== NXDOMAIN responses (failed lookups) ==="
tshark -r "$PCAP" -n -Y "dns.flags.rcode==3" -T fields \
  -E separator="\t" \
  -e frame.time_epoch -e ip.src -e dns.qry.name \
  | sort | uniq -c | sort -rn | head -50

echo ""
echo "=== Long subdomain queries (tunneling indicator: >50 chars) ==="
tshark -r "$PCAP" -n -Y "dns.flags.response==0" -T fields -e dns.qry.name \
  | awk 'length($0) > 50' | sort | uniq -c | sort -rn | head -30

echo ""
echo "=== DNS TXT queries (data exfil via TXT records) ==="
tshark -r "$PCAP" -n -Y "dns.qry.type==16" -T fields \
  -E separator="\t" \
  -e frame.time_epoch -e ip.src -e dns.qry.name

echo ""
echo "=== High-entropy domain labels (DGA detection) ==="
tshark -r "$PCAP" -n -Y "dns.flags.response==0" -T fields -e dns.qry.name \
  | python3 -c "
import sys, math, re
from collections import Counter

def entropy(s):
    if not s: return 0
    counts = Counter(s)
    total = len(s)
    return -sum((c/total)*math.log2(c/total) for c in counts.values())

for line in sys.stdin:
    domain = line.strip()
    # Extract leftmost label (subdomain)
    labels = domain.split('.')
    if labels:
        label = labels[0]
        if len(label) >= 8:  # Skip short labels
            e = entropy(label)
            if e > 3.5:  # High entropy threshold
                print(f'{e:.2f}  {domain}')
" | sort -rn | head -30

#!/usr/bin/env python3
"""
Detect Domain Generation Algorithm (DGA) domains in DNS traffic.
Uses entropy + n-gram analysis + NXDOMAIN rate.
"""
import subprocess, math, re, sys
from collections import Counter, defaultdict

# English bigram frequencies (top pairs in English text)
ENGLISH_BIGRAMS = {
    'th':0.0271,'he':0.0233,'in':0.0203,'er':0.0178,'an':0.0161,
    'on':0.0157,'en':0.0151,'at':0.0145,'es':0.0145,'st':0.0139,
    'or':0.0128,'nt':0.0122,'nd':0.0118,'to':0.0117,'re':0.0116,
    'it':0.0110,'ha':0.0105,'is':0.0100,'ed':0.0099,'ar':0.0097,
}

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

def bigram_score(s: str) -> float:
    """Score how 'English-like' a string is based on bigram frequency."""
    s = s.lower()
    if len(s) < 2:
        return 0.0
    total = 0.0
    for i in range(len(s)-1):
        bg = s[i:i+2]
        total += ENGLISH_BIGRAMS.get(bg, 0)
    return total / (len(s) - 1)

def is_dga_candidate(domain: str) -> tuple[bool, dict]:
    labels = domain.rstrip('.').split('.')
    if len(labels) < 2:
        return False, {}

    # Analyze the SLD (second-level domain, e.g., 'google' from 'google.com')
    sld = labels[-2] if len(labels) >= 2 else labels[0]

    if len(sld) < 6:  # Too short to be DGA
        return False, {}

    entropy = char_entropy(sld)
    bigram = bigram_score(sld)
    digit_ratio = sum(1 for c in sld if c.isdigit()) / len(sld)
    consonant_ratio = sum(1 for c in sld.lower() if c in 'bcdfghjklmnpqrstvwxyz') / len(sld)

    is_dga = (entropy > 3.5 or bigram < 0.015) and len(sld) >= 8
    metrics = {
        "sld": sld,
        "entropy": round(entropy, 3),
        "bigram_score": round(bigram, 4),
        "digit_ratio": round(digit_ratio, 3),
        "length": len(sld),
    }
    return is_dga, metrics

# Extract domains from PCAP
pcap = sys.argv[1]
result = subprocess.run(
    ["tshark", "-r", pcap, "-n", "-Y", "dns.flags.response==0",
     "-T", "fields", "-e", "dns.qry.name"],
    capture_output=True, text=True
)

domain_nxdomain = defaultdict(lambda: {"queries": 0, "nxdomain": 0})

# Also track NXDOMAIN responses
nxdomain_result = subprocess.run(
    ["tshark", "-r", pcap, "-n", "-Y", "dns.flags.rcode==3",
     "-T", "fields", "-e", "dns.qry.name"],
    capture_output=True, text=True
)

for line in nxdomain_result.stdout.splitlines():
    d = line.strip()
    if d:
        domain_nxdomain[d]["nxdomain"] += 1

dga_candidates = []
seen = set()
for line in result.stdout.splitlines():
    domain = line.strip().lower()
    if not domain or domain in seen:
        continue
    seen.add(domain)
    domain_nxdomain[domain]["queries"] += 1

    is_dga, metrics = is_dga_candidate(domain)
    if is_dga:
        metrics["domain"] = domain
        metrics["nxdomain"] = domain_nxdomain[domain]["nxdomain"]
        dga_candidates.append(metrics)

# Sort by entropy descending
dga_candidates.sort(key=lambda x: -x["entropy"])
print(f"{'Entropy':8}  {'Bigram':7}  {'Len':4}  {'NXDOM':6}  Domain")
print("-" * 70)
for c in dga_candidates[:30]:
    print(f"{c['entropy']:8.3f}  {c['bigram_score']:7.4f}  {c['length']:4d}  "
          f"{c['nxdomain']:6d}  {c['domain']}")
print(f"\nTotal DGA candidates: {len(dga_candidates)}")