DNS Security Analyzer

#!/usr/bin/env python3
"""
DNS Security Analyzer

READ THE DOCUMENTATION BEFORE YOU RUN THIS:
    https://permitany.com/scripts/dns-security-analyzer

It covers what the script does, whether it is safe to run, the
inventory file and credentials it needs, how to read the output,
and what to do when something fails.

    pip install "dnspython>=2.4" "requests>=2.31" "colorama>=0.4.6"
"""

import argparse
import json
import math
import re
import socket
import string
import sys
from collections import Counter
from datetime import datetime, timezone
from typing import List, Optional
# Windows consoles default to a legacy code page (cp1252), which cannot encode
# the arrows, box-drawing and check-mark characters used in the output below.
# Printing one raises UnicodeEncodeError and kills the script mid-scan, so make
# stdout UTF-8 and degrade unencodable characters instead of crashing.
for _stream in (sys.stdout, sys.stderr):
    try:
        _stream.reconfigure(encoding="utf-8", errors="replace")
    except (AttributeError, ValueError):
        pass        # already wrapped, or not a real stream (piped/redirected)

try:
    import dns.resolver
    import dns.query
    import dns.zone
    import dns.rdatatype
    import dns.exception
except ImportError:
    print("ERROR: dnspython not installed — pip install dnspython")
    sys.exit(1)

try:
    import requests
except ImportError:
    print("ERROR: requests not installed — pip install requests")
    sys.exit(1)

try:
    from colorama import Fore, Style, init as colorama_init
    colorama_init(autoreset=True)
except ImportError:
    class Fore:
        RED = GREEN = YELLOW = CYAN = WHITE = MAGENTA = ''
    class Style:
        RESET_ALL = BRIGHT = ''

FINDINGS = []

def finding(severity, check, resource, issue, detail='', fix=''):
    FINDINGS.append({'severity': severity, 'check': check, 'resource': resource,
                     'issue': issue, 'detail': detail, 'fix': fix})
    color = {'CRITICAL': Fore.RED, 'HIGH': Fore.YELLOW, 'MEDIUM': Fore.CYAN, 'LOW': Fore.WHITE}.get(severity, Fore.WHITE)
    mark = '\u2717' if severity in ('CRITICAL', 'HIGH') else '\u26a0'
    print(f"    {color}{mark} {severity:<10}{Style.RESET_ALL} {resource}")
    if detail:
        print(f"               {Fore.WHITE}{detail}{Style.RESET_ALL}")
    if fix:
        print(f"               {Fore.CYAN}Fix: {fix}{Style.RESET_ALL}")

def ok(resource, note=''):
    print(f"    {Fore.GREEN}\u2713{Style.RESET_ALL} {resource}  {note}")

def resolve_ns(domain: str) -> List[str]:
    try:
        answers = dns.resolver.resolve(domain, 'NS')
        ns_list = []
        for rr in answers:
            ns_name = str(rr.target).rstrip('.')
            try:
                ip = socket.gethostbyname(ns_name)
                ns_list.append((ns_name, ip))
            except socket.gaierror:
                ns_list.append((ns_name, None))
        return ns_list
    except Exception:
        return []

def test_zone_transfer(domain: str, nameservers: List):
    print(f"\n{Fore.CYAN}[*] ZONE TRANSFER TEST (AXFR){Style.RESET_ALL}")
    for ns_name, ns_ip in nameservers:
        target = ns_ip or ns_name
        try:
            zone = dns.zone.from_xfr(dns.query.xfr(target, domain, timeout=5))
            records = list(zone.iterate_rdatasets())
            finding('CRITICAL', 'AXFR', ns_name,
                    f'ZONE TRANSFER SUCCEEDED — {len(records)} records exposed',
                    f'Attacker can enumerate every hostname, IP, and service in {domain}',
                    f'Restrict AXFR in named.conf: allow-transfer {{ <secondary-ns-ip>; }};')
            print(f"               {Fore.RED}Exposed records (sample):{Style.RESET_ALL}")
            for i, (name, ttl, rdataset) in enumerate(records[:5]):
                print(f"               {Fore.RED}  {name}.{domain} → {rdataset}{Style.RESET_ALL}")
        except dns.exception.FormError:
            ok(ns_name, 'AXFR REFUSED ✓ (correct)')
        except dns.query.TransferError:
            ok(ns_name, 'AXFR REFUSED ✓ (correct)')
        except Exception as e:
            if 'REFUSED' in str(e) or 'denied' in str(e).lower():
                ok(ns_name, 'AXFR REFUSED ✓ (correct)')
            else:
                print(f"    {Fore.YELLOW}AXFR test error for {ns_name}: {e}{Style.RESET_ALL}")

def check_dnssec(domain: str):
    print(f"\n{Fore.CYAN}[*] DNSSEC{Style.RESET_ALL}")
    resolver = dns.resolver.Resolver()
    resolver.use_edns(0, dns.flags.DO, 4096)
    try:
        answers = resolver.resolve(domain, 'DNSKEY')
        ok('DNSKEY', f'{len(list(answers))} key(s) found')
        try:
            resolver.resolve(domain, 'RRSIG')
            ok('RRSIG', 'Signatures present — chain of trust active ✓')
        except Exception:
            finding('HIGH', 'DNSSEC', 'RRSIG', 'DNSKEY present but no RRSIG signatures',
                    'DNSSEC is partially configured — responses are not signed',
                    'Sign zone records with your DNS provider DNSSEC signing tool')
    except dns.resolver.NoAnswer:
        finding('MEDIUM', 'DNSSEC', domain, 'DNSSEC not enabled — no DNSKEY record',
                'DNS responses can be forged (cache poisoning / BGP hijack attacks)',
                'Enable DNSSEC at your DNS registrar — most support it at no extra cost')
    except Exception:
        finding('MEDIUM', 'DNSSEC', domain, 'DNSSEC status could not be determined', '', '')

def check_spf(domain: str):
    try:
        answers = dns.resolver.resolve(domain, 'TXT')
        for rr in answers:
            txt = b''.join(rr.strings).decode('utf-8', errors='replace')
            if txt.startswith('v=spf1'):
                print(f"    SPF: {Fore.WHITE}{txt[:80]}{Style.RESET_ALL}")
                if '+all' in txt:
                    finding('CRITICAL', 'SPF', domain, '+all — anyone can send as this domain',
                            'SPF record explicitly permits ALL senders — phishing trivially bypassed',
                            'Change +all to -all immediately and test with mail-tester.com')
                elif '~all' in txt:
                    finding('MEDIUM', 'SPF', domain, '~all SoftFail — spoofed emails may reach inbox',
                            'With ~all, spoofed emails are marked but not rejected',
                            'Change ~all to -all after testing that all legitimate mail sources are listed')
                elif '-all' in txt:
                    ok(domain, 'SPF -all (HardFail) ✓ — unauthorized senders rejected')
                else:
                    finding('MEDIUM', 'SPF', domain, 'No all mechanism found', '', 'Add -all at end of SPF record')
                return
        finding('HIGH', 'SPF', domain, 'No SPF record found',
                f'Anyone can send email claiming to be from @{domain}',
                f'Add TXT record: v=spf1 include:_spf.yourmailprovider.com -all')
    except Exception:
        finding('HIGH', 'SPF', domain, 'SPF lookup failed', '', '')

def check_dmarc(domain: str):
    dmarc_domain = f'_dmarc.{domain}'
    try:
        answers = dns.resolver.resolve(dmarc_domain, 'TXT')
        for rr in answers:
            txt = b''.join(rr.strings).decode('utf-8', errors='replace')
            if txt.startswith('v=DMARC1'):
                print(f"    DMARC: {Fore.WHITE}{txt[:100]}{Style.RESET_ALL}")
                if 'p=none' in txt:
                    finding('HIGH', 'DMARC', domain, 'p=none — monitoring only, no enforcement',
                            'Spoofed emails are logged but NOT blocked or quarantined',
                            'Start with p=quarantine to move spoofed mail to spam; upgrade to p=reject')
                elif 'p=quarantine' in txt:
                    finding('LOW', 'DMARC', domain, 'p=quarantine — not fully enforced',
                            'Spoofed emails go to spam but are not rejected',
                            'Upgrade to p=reject after confirming rua reports show no legitimate mail flagged')
                elif 'p=reject' in txt:
                    ok(domain, 'DMARC p=reject ✓ — spoofed emails fully blocked')
                return
        finding('HIGH', 'DMARC', domain, 'No DMARC record found',
                'No visibility into email spoofing attempts against this domain',
                f'Add TXT record for _dmarc.{domain}: v=DMARC1; p=none; rua=mailto:dmarc-reports@{domain}')
    except Exception:
        finding('HIGH', 'DMARC', domain, 'DMARC lookup failed', '', '')

def check_dkim_selectors(domain: str):
    SELECTORS = ['default', 'google', 'k1', 'mail', 'smtp', 'selector1', 'selector2', 'dkim', 'protonmail']
    found = []
    for sel in SELECTORS:
        try:
            dns.resolver.resolve(f'{sel}._domainkey.{domain}', 'TXT')
            found.append(sel)
        except Exception:
            pass
    if found:
        ok(domain, f'DKIM selectors found: {', '.join(found)}')
    else:
        finding('MEDIUM', 'DKIM', domain, 'No DKIM selectors found for common names',
                'Email may be signed under non-standard selector names (not necessarily an issue)',
                f'Check your mail provider documentation for the correct DKIM selector')

def check_caa(domain: str):
    print(f"\n{Fore.CYAN}[*] CAA RECORDS{Style.RESET_ALL}")
    try:
        answers = dns.resolver.resolve(domain, 'CAA')
        cas = [str(rr) for rr in answers]
        ok(domain, f'CAA configured: {'; '.join(cas[:3])}')
    except dns.resolver.NoAnswer:
        finding('MEDIUM', 'CAA', domain, 'No CAA record — any CA can issue TLS certificates',
                'Without CAA, a rogue CA could issue a cert for your domain enabling MITM attacks',
                f'Add: 0 issue "letsencrypt.org" or your CA of choice')
    except Exception:
        finding('MEDIUM', 'CAA', domain, 'CAA lookup failed', '', '')

def enumerate_subdomains_crt(domain: str) -> List[str]:
    print(f"\n{Fore.CYAN}[*] SUBDOMAINS (Certificate Transparency — passive){Style.RESET_ALL}")
    try:
        resp = requests.get(f'https://crt.sh/?q=%.{domain}&output=json', timeout=15)
        resp.raise_for_status()
        data = resp.json()
        subs = set()
        for entry in data:
            name = entry.get('name_value', '')
            for sub in name.split('\n'):
                sub = sub.strip().lstrip('*.')
                if sub.endswith(f'.{domain}') or sub == domain:
                    subs.add(sub)
        print(f"    Found {len(subs)} unique subdomains via crt.sh")
        return list(subs)
    except Exception as e:
        print(f"    {Fore.YELLOW}crt.sh lookup failed: {e}{Style.RESET_ALL}")
        return []

TAKEOVER_PATTERNS = [
    (r'github\.io', 'GitHub Pages — check if page returns 404'),
    (r'herokuapp\.com', 'Heroku — check if app name is available'),
    (r's3\.amazonaws\.com', 'S3 bucket — check if bucket exists'),
    (r'azurewebsites\.net', 'Azure Web App — check if app exists'),
    (r'netlify\.app', 'Netlify — check if site is active'),
    (r'wpengine\.com', 'WP Engine — check if account exists'),
    (r'ghost\.io', 'Ghost — check if blog exists'),
]

def check_subdomain_takeover(subs: List[str], domain: str):
    print(f"\n{Fore.CYAN}[*] SUBDOMAIN TAKEOVER CHECK{Style.RESET_ALL}")
    for sub in subs[:100]:  # cap to avoid rate limiting
        try:
            answers = dns.resolver.resolve(sub, 'CNAME')
            for rr in answers:
                cname = str(rr.target).rstrip('.')
                for pattern, description in TAKEOVER_PATTERNS:
                    if re.search(pattern, cname, re.IGNORECASE):
                        # Check if target resolves / returns 404
                        try:
                            resp = requests.get(f'https://{cname}', timeout=5, allow_redirects=True)
                            if resp.status_code == 404:
                                finding('HIGH', 'Subdomain Takeover', sub,
                                        f'CNAME → {cname} ({description})',
                                        f'Target returns 404 — attacker can claim this resource and serve content as {sub}',
                                        f'Remove DNS record for {sub} or reclaim the resource at {cname}')
                            else:
                                ok(sub, f'CNAME → {cname} (active)')
                        except Exception:
                            finding('MEDIUM', 'Subdomain Takeover', sub,
                                    f'CNAME → {cname} ({description}) — could not verify',
                                    'Manually verify the target is still claimed',
                                    f'Check {cname} is active and owned by your organization')
        except Exception:
            pass

def check_open_resolver(nameservers: List):
    print(f"\n{Fore.CYAN}[*] OPEN RESOLVER CHECK{Style.RESET_ALL}")
    for ns_name, ns_ip in nameservers:
        if not ns_ip:
            continue
        try:
            resolver = dns.resolver.Resolver(configure=False)
            resolver.nameservers = [ns_ip]
            resolver.timeout = 3
            resolver.lifetime = 3
            resolver.resolve('google.com', 'A')
            finding('HIGH', 'Open Resolver', ns_name,
                    'Accepts recursive queries from external IPs',
                    'Open resolvers can be abused for DNS amplification DDoS attacks (up to 70x amplification)',
                    f'Restrict recursion in named.conf: allow-recursion {{ <trusted-nets>; }};')
        except dns.resolver.NoNameservers:
            ok(ns_name, 'Not an open resolver ✓')
        except Exception:
            ok(ns_name, 'Not an open resolver ✓')

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

def analyze_dns_tunneling(log_path: str):
    print(f"\n{Fore.CYAN}[*] DNS TUNNELING DETECTION — {log_path}{Style.RESET_ALL}")
    HIGH_ENTROPY = 3.8
    try:
        with open(log_path, 'r', errors='replace') as fh:
            lines = fh.readlines()[:50000]
    except Exception as e:
        print(f"    ERROR reading log: {e}"); return
    subdomain_re = re.compile(r'query\[A\]\s+([\w\.-]+)\.', re.IGNORECASE)
    query_counts: Counter = Counter()
    high_entropy_queries = []
    for line in lines:
        m = subdomain_re.search(line)
        if m:
            fqdn = m.group(1)
            parts = fqdn.split('.')
            if len(parts) > 2:
                sub = parts[0]
                if len(sub) > 20:
                    ent = entropy(sub)
                    if ent > HIGH_ENTROPY:
                        high_entropy_queries.append((sub, ent, fqdn))
            query_counts[fqdn] += 1
    if high_entropy_queries:
        finding('HIGH', 'DNS Tunneling', log_path,
                f'{len(high_entropy_queries)} high-entropy subdomains detected',
                'High-entropy subdomains are a strong indicator of DNS tunneling (data exfiltration via DNS)',
                'Block with RPZ or DNS firewall; investigate source IPs making these queries')
        for sub, ent, fqdn in sorted(high_entropy_queries, key=lambda x: -x[1])[:5]:
            print(f"               {Fore.RED}entropy={ent:.2f}  {fqdn}{Style.RESET_ALL}")
    else:
        ok(log_path, f'No high-entropy patterns detected in {len(lines)} log lines')
    # Top queried domains
    top = query_counts.most_common(5)
    if top:
        print(f"    Top queried: {', '.join(f'{d} ({c}x)' for d, c in top)}")

# ─── MAIN ─────────────────────────────────────────────────────────────────────

def main():
    parser = argparse.ArgumentParser(description='DNS Security Analyzer')
    parser.add_argument('--domain', required=True)
    parser.add_argument('--nameservers', default='', help='Comma-separated nameserver hostnames')
    parser.add_argument('--wordlist', default='', help='Subdomain brute-force wordlist')
    parser.add_argument('--query-log', default='', help='DNS query log for tunneling detection')
    parser.add_argument('--output', default='', help='Output JSON report path')
    args = parser.parse_args()

    domain = args.domain.lower().strip()
    print(f"{Fore.CYAN}[*] DNS Security Analyzer | Domain: {domain}{Style.RESET_ALL}")

    if args.nameservers:
        nameservers = [(ns.strip(), None) for ns in args.nameservers.split(',')]
        for i, (ns, _) in enumerate(nameservers):
            try:
                ip = socket.gethostbyname(ns)
                nameservers[i] = (ns, ip)
            except Exception:
                pass
    else:
        nameservers = resolve_ns(domain)
    if nameservers:
        ns_str = ', '.join(f'{n} ({ip})' for n, ip in nameservers)
        print(f"{Fore.CYAN}[*] Nameservers: {ns_str}{Style.RESET_ALL}")
    else:
        print(f"{Fore.YELLOW}WARNING: Could not resolve nameservers for {domain}{Style.RESET_ALL}")

    test_zone_transfer(domain, nameservers)
    check_dnssec(domain)

    print(f"\n{Fore.CYAN}[*] EMAIL SECURITY (SPF / DKIM / DMARC){Style.RESET_ALL}")
    check_spf(domain)
    check_dkim_selectors(domain)
    check_dmarc(domain)

    check_caa(domain)

    subs = enumerate_subdomains_crt(domain)
    if subs:
        check_subdomain_takeover(subs, domain)

    check_open_resolver(nameservers)

    if args.query_log:
        analyze_dns_tunneling(args.query_log)

    # Summary
    counts = {s: sum(1 for f in FINDINGS if f['severity'] == s) for s in ['CRITICAL', 'HIGH', 'MEDIUM', 'LOW']}
    print(f"\n{'─'*70}")
    print(f"{Fore.RED}{counts['CRITICAL']} CRITICAL{Style.RESET_ALL}  "
          f"{Fore.YELLOW}{counts['HIGH']} HIGH{Style.RESET_ALL}  "
          f"{Fore.CYAN}{counts['MEDIUM']} MEDIUM{Style.RESET_ALL}  "
          f"{counts['LOW']} LOW")

    output = args.output or f"dns_audit_{datetime.now().strftime('%Y-%m-%d')}.json"
    report = {'domain': domain, 'generated': datetime.now(timezone.utc).isoformat(),
              'summary': counts, 'findings': FINDINGS}
    with open(output, 'w') as fh:
        json.dump(report, fh, indent=2)
    print(f"{Fore.GREEN}Report saved: {output}{Style.RESET_ALL}")
    return 2 if counts['CRITICAL'] else (1 if counts['HIGH'] or counts['MEDIUM'] else 0)

if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyboardInterrupt:
        print("\nInterrupted.", file=sys.stderr)
        sys.exit(130)