#!/usr/bin/env python3 """Focused DNS traffic analyzer for malware detection. Analyzes DNS queries in PCAP files to detect DGA domains, DNS tunneling, fast-flux networks, and other DNS-based malicious activity. Usage: python dns_analyzer.py --pcap capture.pcap --output dns_report.json python dns_analyzer.py --pcap capture.pcap --entropy-threshold 3.5 """ from __future__ import annotations import argparse import collections import json import logging import math import os import platform import re import sys from datetime import datetime, timezone from pathlib import Path from typing import Any, Optional logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", ) logger = logging.getLogger(__name__) try: from scapy.all import PcapReader, DNS, DNSQR, DNSRR, IP, UDP, conf conf.verb = 0 HAS_SCAPY = True except ImportError: HAS_SCAPY = False try: import dpkt HAS_DPKT = True except ImportError: HAS_DPKT = False # Common English bigrams for DGA detection COMMON_BIGRAMS = { "th", "he", "in", "er", "an", "re", "on", "at", "en", "nd", "ti", "es", "or", "te", "of", "ed", "is", "it", "al", "ar", "st", "to", "nt", "ng", "se", "ha", "as", "ou", "io", "le", "ve", "co", "me", "de", "hi", "ri", "ro", "ic", "ne", "ea", "ra", "ce", "li", "ch", "ll", "be", "ma", "si", "om", "ur", } # Suspicious TLDs SUSPICIOUS_TLDS = { "tk", "ml", "ga", "cf", "gq", # Free TLDs often abused "top", "xyz", "club", "online", "site", "icu", "buzz", "work", "click", "loan", "racing", "win", "bid", "stream", "download", "review", "date", "faith", "party", "science", } # Known DNS tunneling tool domains KNOWN_TUNNEL_PATTERNS = [ r"\.dns2tcp\.", r"\.iodine\.", r"\.dnscat\.", r"\.dnscapy\.", ] def shannon_entropy(text: str) -> float: """Calculate Shannon entropy of a string.""" if not text: return 0.0 freq = collections.Counter(text.lower()) length = len(text) return -sum( (count / length) * math.log2(count / length) for count in freq.values() ) def bigram_score(domain: str) -> float: """Score a domain based on English bigram frequency. Lower = more random.""" label = domain.split(".")[0].lower() if len(label) < 4: return 1.0 # Short labels are not useful for bigram analysis bigrams = [label[i:i + 2] for i in range(len(label) - 1)] if not bigrams: return 0.0 matches = sum(1 for b in bigrams if b in COMMON_BIGRAMS) return matches / len(bigrams) def consonant_ratio(text: str) -> float: """Calculate the ratio of consonants to total alphabetic characters.""" alpha = [c for c in text.lower() if c.isalpha()] if not alpha: return 0.0 vowels = set("aeiou") consonants = sum(1 for c in alpha if c not in vowels) return consonants / len(alpha) def digit_ratio(text: str) -> float: """Calculate the ratio of digits in the string.""" if not text: return 0.0 return sum(1 for c in text if c.isdigit()) / len(text) def is_hex_string(text: str) -> bool: """Check if a string appears to be hex-encoded.""" clean = re.sub(r"[^a-fA-F0-9]", "", text) return len(clean) > 10 and len(clean) / max(len(text), 1) > 0.8 class DnsAnalyzer: """Analyzes DNS traffic for malicious patterns.""" def __init__(self, entropy_threshold: float = 3.5, tunnel_length: int = 30): self.entropy_threshold = entropy_threshold self.tunnel_subdomain_length = tunnel_length self.queries: list[dict] = [] self.responses: list[dict] = [] self.domain_query_count: dict[str, int] = collections.defaultdict(int) self.domain_ips: dict[str, set] = collections.defaultdict(set) self.src_query_count: dict[str, int] = collections.defaultdict(int) def parse_pcap_dpkt(self, pcap_path: str) -> None: """Parse DNS from PCAP using dpkt.""" import socket as _socket with open(pcap_path, "rb") as f: try: pcap = dpkt.pcap.Reader(f) except ValueError: f.seek(0) pcap = dpkt.pcapng.Reader(f) for timestamp, buf in pcap: try: eth = dpkt.ethernet.Ethernet(buf) if not isinstance(eth.data, dpkt.ip.IP): continue ip = eth.data if not isinstance(ip.data, dpkt.udp.UDP): continue udp = ip.data if udp.sport != 53 and udp.dport != 53: continue if not udp.data: continue dns = dpkt.dns.DNS(udp.data) src_ip = _socket.inet_ntoa(ip.src) dst_ip = _socket.inet_ntoa(ip.dst) if dns.qr == 0: # Query for q in dns.qd: self.queries.append({ "timestamp": timestamp, "src_ip": src_ip, "dst_ip": dst_ip, "query": q.name, "type": {1: "A", 2: "NS", 5: "CNAME", 15: "MX", 16: "TXT", 28: "AAAA"}.get(q.type, str(q.type)), }) self.domain_query_count[q.name] += 1 self.src_query_count[src_ip] += 1 else: # Response qname = dns.qd[0].name if dns.qd else "" answers = [] for rr in dns.an: if rr.type == 1: try: addr = _socket.inet_ntoa(rr.rdata) answers.append(addr) self.domain_ips[qname].add(addr) except Exception: pass rcode_map = {0: "NOERROR", 3: "NXDOMAIN", 2: "SERVFAIL", 5: "REFUSED"} self.responses.append({ "timestamp": timestamp, "query": qname, "answers": answers, "rcode": rcode_map.get(dns.rcode, str(dns.rcode)), "answer_count": len(answers), }) except Exception: continue def parse_pcap_scapy(self, pcap_path: str) -> None: """Parse DNS from PCAP using Scapy.""" with PcapReader(pcap_path) as reader: for pkt in reader: if not pkt.haslayer(DNS): continue ts = float(pkt.time) if hasattr(pkt, 'time') else 0 src_ip = pkt[IP].src if pkt.haslayer(IP) else "unknown" dst_ip = pkt[IP].dst if pkt.haslayer(IP) else "unknown" dns = pkt[DNS] if dns.qr == 0 and dns.haslayer(DNSQR): qname = dns[DNSQR].qname.decode("utf-8", errors="replace").rstrip(".") qtype_map = {1: "A", 2: "NS", 5: "CNAME", 15: "MX", 16: "TXT", 28: "AAAA"} self.queries.append({ "timestamp": ts, "src_ip": src_ip, "dst_ip": dst_ip, "query": qname, "type": qtype_map.get(dns[DNSQR].qtype, str(dns[DNSQR].qtype)), }) self.domain_query_count[qname] += 1 self.src_query_count[src_ip] += 1 elif dns.qr == 1: qname = dns[DNSQR].qname.decode("utf-8", errors="replace").rstrip(".") if dns.haslayer(DNSQR) else "" answers = [] for i in range(dns.ancount): try: rr = dns.an[i] if hasattr(rr, "rdata"): rdata = str(rr.rdata) answers.append(rdata) self.domain_ips[qname].add(rdata) except Exception: pass rcode_map = {0: "NOERROR", 3: "NXDOMAIN", 2: "SERVFAIL", 5: "REFUSED"} self.responses.append({ "timestamp": ts, "query": qname, "answers": answers, "rcode": rcode_map.get(dns.rcode, str(dns.rcode)), "answer_count": len(answers), }) def detect_dga(self) -> list[dict]: """Detect likely DGA (Domain Generation Algorithm) domains.""" dga_candidates = [] seen = set() for query in self.queries: domain = query["query"] if domain in seen: continue seen.add(domain) # Get the registrable domain parts parts = domain.split(".") if len(parts) < 2: continue # Analyze the label (part before TLD) label = parts[0] if len(parts) == 2 else ".".join(parts[:-2]) if len(label) < 5: continue ent = shannon_entropy(label) bg_score = bigram_score(label) cons_ratio = consonant_ratio(label) dig_ratio = digit_ratio(label) is_hex = is_hex_string(label) # DGA scoring score = 0 reasons = [] if ent > self.entropy_threshold: score += 2 reasons.append(f"high entropy ({ent:.2f})") if bg_score < 0.2: score += 2 reasons.append(f"low bigram score ({bg_score:.2f})") if cons_ratio > 0.75: score += 1 reasons.append(f"high consonant ratio ({cons_ratio:.2f})") if dig_ratio > 0.3: score += 1 reasons.append(f"high digit ratio ({dig_ratio:.2f})") if is_hex: score += 2 reasons.append("appears hex-encoded") tld = parts[-1].lower() if tld in SUSPICIOUS_TLDS: score += 1 reasons.append(f"suspicious TLD (.{tld})") if score >= 3: dga_candidates.append({ "domain": domain, "label": label, "score": score, "entropy": round(ent, 2), "bigram_score": round(bg_score, 2), "reasons": reasons, "query_count": self.domain_query_count.get(domain, 0), }) return sorted(dga_candidates, key=lambda x: x["score"], reverse=True) def detect_tunneling(self) -> list[dict]: """Detect DNS tunneling indicators.""" tunnel_candidates = [] domain_subdomains: dict[str, list] = collections.defaultdict(list) for query in self.queries: domain = query["query"] parts = domain.split(".") if len(parts) < 3: continue # Base domain (last 2 parts for common TLDs) base = ".".join(parts[-2:]) subdomain = ".".join(parts[:-2]) domain_subdomains[base].append({ "subdomain": subdomain, "full_query": domain, "timestamp": query["timestamp"], "type": query["type"], }) for base_domain, subs in domain_subdomains.items(): indicators = [] score = 0 # Check for long subdomains long_subs = [s for s in subs if len(s["subdomain"]) > self.tunnel_subdomain_length] if long_subs: score += 2 indicators.append( f"{len(long_subs)} queries with long subdomains (>{self.tunnel_subdomain_length} chars)" ) # High query volume to single domain if len(subs) > 50: score += 2 indicators.append(f"high query volume ({len(subs)} queries)") # Unique subdomain count unique_subs = set(s["subdomain"] for s in subs) if len(unique_subs) > 20 and len(unique_subs) / max(len(subs), 1) > 0.8: score += 2 indicators.append( f"high unique subdomain ratio ({len(unique_subs)}/{len(subs)})" ) # TXT record queries (common for DNS tunneling) txt_queries = [s for s in subs if s["type"] == "TXT"] if len(txt_queries) > 5: score += 2 indicators.append(f"{len(txt_queries)} TXT record queries") # Check for hex/base64 encoded subdomains encoded_count = sum( 1 for s in subs if is_hex_string(s["subdomain"]) or re.match(r"^[a-zA-Z0-9+/=]{20,}$", s["subdomain"]) ) if encoded_count > 5: score += 2 indicators.append(f"{encoded_count} encoded-looking subdomains") # Known tunneling tool patterns for pattern in KNOWN_TUNNEL_PATTERNS: if re.search(pattern, base_domain, re.IGNORECASE): score += 5 indicators.append(f"matches known tunnel tool pattern: {pattern}") if score >= 3: avg_sub_len = sum(len(s["subdomain"]) for s in subs) / max(len(subs), 1) tunnel_candidates.append({ "base_domain": base_domain, "score": score, "total_queries": len(subs), "unique_subdomains": len(unique_subs), "avg_subdomain_length": round(avg_sub_len, 1), "indicators": indicators, "sample_queries": [s["full_query"] for s in subs[:5]], }) return sorted(tunnel_candidates, key=lambda x: x["score"], reverse=True) def detect_fast_flux(self) -> list[dict]: """Detect fast-flux DNS behavior (single domain resolving to many IPs).""" fast_flux = [] for domain, ips in self.domain_ips.items(): if len(ips) >= 5: fast_flux.append({ "domain": domain, "unique_ips": len(ips), "ips": sorted(ips), "query_count": self.domain_query_count.get(domain, 0), }) return sorted(fast_flux, key=lambda x: x["unique_ips"], reverse=True) def detect_suspicious_tlds(self) -> list[dict]: """Find queries to suspicious TLDs.""" suspicious = [] seen = set() for query in self.queries: domain = query["query"] if domain in seen: continue seen.add(domain) tld = domain.split(".")[-1].lower() if tld in SUSPICIOUS_TLDS: suspicious.append({ "domain": domain, "tld": tld, "query_count": self.domain_query_count.get(domain, 0), }) return sorted(suspicious, key=lambda x: x["query_count"], reverse=True) def get_nxdomain_analysis(self) -> dict: """Analyze NXDOMAIN responses for DGA indicators.""" nxdomains = [r["query"] for r in self.responses if r.get("rcode") == "NXDOMAIN"] total_responses = len(self.responses) nxdomain_count = len(nxdomains) return { "count": nxdomain_count, "total_responses": total_responses, "ratio": round(nxdomain_count / max(total_responses, 1), 4), "domains": sorted(set(nxdomains))[:100], "high_ratio_warning": nxdomain_count / max(total_responses, 1) > 0.3, } def generate_report(self) -> dict[str, Any]: """Generate comprehensive DNS analysis report.""" dga = self.detect_dga() tunneling = self.detect_tunneling() fast_flux = self.detect_fast_flux() suspicious_tlds = self.detect_suspicious_tlds() nxdomain = self.get_nxdomain_analysis() # Top queried domains top_domains = sorted( self.domain_query_count.items(), key=lambda x: x[1], reverse=True, )[:50] # Top querying hosts top_sources = sorted( self.src_query_count.items(), key=lambda x: x[1], reverse=True, )[:20] return { "metadata": { "tool": "dns_analyzer.py", "timestamp": datetime.now(timezone.utc).isoformat(), "platform": platform.system(), "entropy_threshold": self.entropy_threshold, "tunnel_subdomain_length": self.tunnel_subdomain_length, }, "overview": { "total_queries": len(self.queries), "total_responses": len(self.responses), "unique_domains": len(set(q["query"] for q in self.queries)), "unique_sources": len(self.src_query_count), "dga_candidates": len(dga), "tunnel_candidates": len(tunneling), "fast_flux_domains": len(fast_flux), "suspicious_tld_domains": len(suspicious_tlds), }, "dga_detection": dga[:50], "tunneling_detection": tunneling[:20], "fast_flux_detection": fast_flux[:20], "suspicious_tlds": suspicious_tlds[:50], "nxdomain_analysis": nxdomain, "top_queried_domains": [ {"domain": d, "count": c} for d, c in top_domains ], "top_querying_hosts": [ {"ip": ip, "count": c} for ip, c in top_sources ], } def main() -> None: parser = argparse.ArgumentParser( description="DNS traffic analyzer for malware detection", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: %(prog)s --pcap capture.pcap --output dns_report.json %(prog)s --pcap capture.pcap --entropy-threshold 3.5 %(prog)s --pcap capture.pcap --tunnel-length 25 """, ) parser.add_argument("--input", "--pcap", "-f", dest="pcap", required=True, help="Path to PCAP file") parser.add_argument("--output", "-o", help="Output JSON report path (default: stdout)") parser.add_argument( "--entropy-threshold", type=float, default=3.5, help="Entropy threshold for DGA detection (default: 3.5)", ) parser.add_argument( "--tunnel-length", type=int, default=30, help="Minimum subdomain length for tunneling detection (default: 30)", ) parser.add_argument("--verbose", "-v", action="store_true", help="Verbose logging") parser.add_argument("--format", default="json", choices=["json", "text", "csv"], help="Output format (default: json)") args = parser.parse_args() if args.verbose: logging.getLogger().setLevel(logging.DEBUG) if not os.path.isfile(args.pcap): logger.error(f"PCAP file not found: {args.pcap}") sys.exit(1) analyzer = DnsAnalyzer( entropy_threshold=args.entropy_threshold, tunnel_length=args.tunnel_length, ) # Parse with best available library if HAS_DPKT: try: analyzer.parse_pcap_dpkt(args.pcap) except Exception as e: logger.warning(f"dpkt failed: {e}") if HAS_SCAPY: analyzer.parse_pcap_scapy(args.pcap) else: logger.error("No working parser available") sys.exit(1) elif HAS_SCAPY: analyzer.parse_pcap_scapy(args.pcap) else: logger.error("Install scapy or dpkt: pip install scapy dpkt") sys.exit(1) report = analyzer.generate_report() report_json = json.dumps(report, indent=2, default=str) if args.output: with open(args.output, "w") as f: f.write(report_json) logger.info(f"Report written to: {args.output}") else: print(report_json) # Summary ov = report["overview"] print(f"\n=== DNS Analysis Summary ===", file=sys.stderr) print(f"Total queries: {ov['total_queries']} ({ov['unique_domains']} unique domains)", file=sys.stderr) print(f"DGA candidates: {ov['dga_candidates']}", file=sys.stderr) print(f"Tunnel candidates: {ov['tunnel_candidates']}", file=sys.stderr) print(f"Fast-flux domains: {ov['fast_flux_domains']}", file=sys.stderr) print(f"Suspicious TLD domains: {ov['suspicious_tld_domains']}", file=sys.stderr) if __name__ == "__main__": main()