backend/scanners/dns_security_scanner.py

"""
dns_security_scanner.py — DNS security analysis using DNSPython.
Performs comprehensive DNS security checks including zone transfer, DNSSEC, and record analysis.
"""
import dns.resolver, dns.query, dns.message, dns.rdatatype
from scanners.base_scanner import BaseScanner

class DNSSecurityScanner(BaseScanner):
    SCANNER_NAME = "DNS Security Analyzer"

    def run(self):
        self.log("INFO", f"[DNS Security] Starting comprehensive DNS analysis for {self.domain}...")
        
        # Perform various DNS security checks
        self._check_zone_transfer()
        self._check_dnssec()
        self._check_mx_records()
        self._check_spf()
        self._check_dmarc()
        self._check_wildcard_dns()
        self._check_dns_cache_poisoning()
        self._check_caa_records()
        
        self.log("SUCCESS", "[DNS Security] DNS analysis complete.")
        return self.vulns

    def _check_zone_transfer(self):
        """Check for DNS zone transfer vulnerability."""
        self.log("INFO", "[DNS Security] Checking for zone transfer vulnerability...")
        
        try:
            # Get authoritative nameservers
            ns_answers = dns.resolver.resolve(self.domain, 'NS')
            nameservers = [rdata.target.to_text().rstrip('.') for rdata in ns_answers]
            
            for ns in nameservers[:2]:  # Check first 2 nameservers only (faster)
                try:
                    # Attempt AXFR (zone transfer) — short timeout for speed
                    zone_query = dns.query.xfr(ns, self.domain, timeout=5)
                    zone_response = next(zone_query)
                    
                    if zone_response:
                        self.log("CRITICAL", f"[DNS Security] ZONE TRANSFER VULNERABLE on {ns}!")
                        self.add_vuln(
                            title="DNS Zone Transfer Vulnerability",
                            severity="Critical", category="DNS Security", cvss_score=9.8,
                            description=f"DNS zone transfer (AXFR) is allowed from {ns}. This exposes all DNS records including internal hostnames, mail servers, and other sensitive infrastructure details.",
                            remediation="Disable zone transfers to unauthorized servers:\n  BIND: allow-transfer { trusted_ips; };\n  Ensure only authorized secondary nameservers can perform AXFR."
                        )
                        return  # Found vulnerability, no need to check others
                except dns.exception.DNSException:
                    # Expected if zone transfer is disabled
                    pass
                except OSError as e:
                    # Network-level errors (timeout, connection refused) are expected
                    err_msg = str(e).strip()
                    if err_msg:
                        self.log("DEBUG", f"[DNS Security] Zone transfer network error on {ns}: {err_msg}")
                except Exception as e:
                    err_msg = str(e).strip()
                    if err_msg:
                        self.log("DEBUG", f"[DNS Security] Nameserver query error on {ns}: {err_msg}")
            
            self.log("SUCCESS", "[DNS Security] Zone transfer properly secured.")
            
        except dns.resolver.NoNameservers:
            self.log("WARNING", "[DNS Security] Could not retrieve nameservers.")
        except Exception as e:
            self.log("WARNING", f"[DNS Security] Zone transfer check failed: {e}")

    def _check_dnssec(self):
        """Check for DNSSEC implementation."""
        self.log("INFO", "[DNS Security] Checking DNSSEC implementation...")
        
        try:
            # Check for DNSKEY records
            dnskey_answers = dns.resolver.resolve(self.domain, 'DNSKEY')
            if dnskey_answers:
                self.log("SUCCESS", "[DNS Security] DNSSEC is implemented.")
                return
        except dns.resolver.NoAnswer:
            pass
        except Exception as e:
            self.log("ERROR", f"[DNS Security] DNSKEY query error: {e}")
        
        self.log("WARNING", "[DNS Security] DNSSEC not implemented.")
        self.add_vuln(
            title="DNSSEC Not Implemented",
            severity="Medium", category="DNS Security", cvss_score=5.3,
            description=f"The domain {self.domain} does not have DNSSEC implemented. This makes it vulnerable to DNS cache poisoning and spoofing attacks.",
            remediation="Implement DNSSEC:\n  1. Generate DNSSEC keys\n  2. Sign your zone files\n  3. Upload DS records to your registrar\n  4. Verify with: dig +dnssec {self.domain}"
        )

    def _check_mx_records(self):
        """Check MX records for security issues."""
        self.log("INFO", "[DNS Security] Analyzing MX records...")
        
        try:
            mx_answers = dns.resolver.resolve(self.domain, 'MX')
            mx_servers = [rdata.exchange.to_text().rstrip('.') for rdata in mx_answers]
            
            if mx_servers:
                self.log("SUCCESS", f"[DNS Security] MX Records: {', '.join(mx_servers[:3])}")
                
                # Check for common mail server security issues
                for mx in mx_servers:
                    # Check if MX points to IP directly (bad practice)
                    if mx.replace('.', '').isdigit():
                        self.log("WARNING", f"[DNS Security] MX record points to IP: {mx}")
                        self.add_vuln(
                            title="MX Record Points to IP Address",
                            severity="Low", category="DNS Security", cvss_score=3.1,
                            description=f"MX record for {self.domain} points to IP address {mx} instead of a hostname. This is not a best practice.",
                            remediation="Update MX records to point to hostnames instead of IP addresses."
                        )
            else:
                self.log("INFO", "[DNS Security] No MX records found.")
                
        except dns.resolver.NoAnswer:
            self.log("INFO", "[DNS Security] No MX records found.")
        except Exception as e:
            self.log("WARNING", f"[DNS Security] MX check failed: {e}")

    def _check_spf(self):
        """Check for SPF record."""
        self.log("INFO", "[DNS Security] Checking SPF record...")
        
        try:
            txt_answers = dns.resolver.resolve(self.domain, 'TXT')
            spf_found = False
            
            for rdata in txt_answers:
                txt = rdata.to_text()
                if 'v=spf1' in txt:
                    spf_found = True
                    self.log("SUCCESS", f"[DNS Security] SPF Record: {txt[:100]}")
                    
                    # Check for ~all vs -all
                    if '~all' in txt:
                        self.log("WARNING", "[DNS Security] SPF uses ~all (soft fail)")
                        self.add_vuln(
                            title="SPF Uses Soft Fail (~all)",
                            severity="Low", category="DNS Security", cvss_score=3.1,
                            description=f"SPF record for {self.domain} uses ~all (soft fail) instead of -all (hard fail). This may allow some spoofed emails to pass.",
                            remediation="Consider changing ~all to -all in your SPF record for stricter email validation."
                        )
                    break
            
            if not spf_found:
                self.log("WARNING", "[DNS Security] No SPF record found.")
                self.add_vuln(
                    title="SPF Record Missing",
                    severity="Medium", category="DNS Security", cvss_score=5.3,
                    description=f"No SPF (Sender Policy Framework) record found for {self.domain}. This allows email spoofing and increases spam risk.",
                    remediation="Add an SPF record to your DNS:\n  Example: v=spf1 ip4:192.0.2.0/24 -all\n  Use SPF record generator tools for proper configuration."
                )
                
        except dns.resolver.NoAnswer:
            self.log("WARNING", "[DNS Security] No TXT records found (SPF missing).")
            self.add_vuln(
                title="SPF Record Missing",
                severity="Medium", category="DNS Security", cvss_score=5.3,
                description=f"No SPF record found for {self.domain}.",
                remediation="Add an SPF record to your DNS configuration."
            )
        except Exception as e:
            self.log("WARNING", f"[DNS Security] SPF check failed: {e}")

    def _check_dmarc(self):
        """Check for DMARC record."""
        self.log("INFO", "[DNS Security] Checking DMARC record...")
        
        dmarc_domain = f"_dmarc.{self.domain}"
        
        try:
            txt_answers = dns.resolver.resolve(dmarc_domain, 'TXT')
            dmarc_found = False
            
            for rdata in txt_answers:
                txt = rdata.to_text()
                if 'v=DMARC1' in txt:
                    dmarc_found = True
                    self.log("SUCCESS", f"[DNS Security] DMARC Record: {txt[:100]}")
                    
                    # Check for p=none (monitoring only)
                    if 'p=none' in txt:
                        self.log("WARNING", "[DNS Security] DMARC policy is p=none (monitoring only)")
                        self.add_vuln(
                            title="DMARC Policy Set to None",
                            severity="Low", category="DNS Security", cvss_score=3.1,
                            description=f"DMARC record for {self.domain} has policy p=none, which means no action is taken on failed SPF/DKIM. This is only suitable for monitoring.",
                            remediation="Update DMARC policy to p=quarantine or p=reject for better email security:\n  v=DMARC1; p=quarantine; rua=mailto:dmarc@example.com"
                        )
                    break
            
            if not dmarc_found:
                self.log("WARNING", "[DNS Security] No DMARC record found.")
                self.add_vuln(
                    title="DMARC Record Missing",
                    severity="Medium", category="DNS Security", cvss_score=5.3,
                    description=f"No DMARC (Domain-based Message Authentication, Reporting & Conformance) record found for {self.domain}. DMARC helps prevent email spoofing.",
                    remediation="Add a DMARC record to your DNS:\n  Example: v=DMARC1; p=quarantine; rua=mailto:dmarc@example.com\n  Start with p=none for monitoring, then move to p=quarantine or p=reject."
                )
                
        except dns.resolver.NoAnswer:
            self.log("WARNING", "[DNS Security] No DMARC record found.")
            self.add_vuln(
                title="DMARC Record Missing",
                severity="Medium", category="DNS Security", cvss_score=5.3,
                description=f"No DMARC record found for {self.domain}.",
                remediation="Add a DMARC record to your DNS configuration."
            )
        except dns.resolver.NXDOMAIN:
            self.log("WARNING", "[DNS Security] DMARC subdomain does not exist.")
        except Exception as e:
            self.log("WARNING", f"[DNS Security] DMARC check failed: {e}")

    def _check_wildcard_dns(self):
        """Check for wildcard DNS records."""
        self.log("INFO", "[DNS Security] Checking for wildcard DNS...")
        
        random_subdomain = f"random-nonexistent-{hash(self.domain) % 10000}.{self.domain}"
        
        try:
            dns.resolver.resolve(random_subdomain, 'A')
            self.log("WARNING", "[DNS Security] Wildcard DNS record detected.")
            self.add_vuln(
                title="Wildcard DNS Record Detected",
                severity="Low", category="DNS Security", cvss_score=3.1,
                description=f"Wildcard DNS record detected for {self.domain}. This can lead to subdomain takeover vulnerabilities if not properly managed.",
                remediation="Review wildcard DNS usage. Ensure all subdomains are properly claimed and monitored. Consider removing wildcard if not necessary."
            )
        except dns.resolver.NXDOMAIN:
            self.log("SUCCESS", "[DNS Security] No wildcard DNS detected.")
        except Exception as e:
            self.log("WARNING", f"[DNS Security] Wildcard check failed: {e}")

    def _check_dns_cache_poisoning(self):
        """Check for DNS cache poisoning vulnerabilities."""
        self.log("INFO", "[DNS Security] Checking DNS cache poisoning risks...")
        
        try:
            # Check for random source ports (good practice)
            # This is a heuristic check - we can't directly test this without more complex probing
            self.log("INFO", "[DNS Security] Ensure DNS resolver uses random source ports and transaction IDs.")
            
            # Check for DNSSEC (already done in _check_dnssec)
            # This is informational
            self.log("INFO", "[DNS Security] DNS cache poisoning protection relies on DNSSEC implementation.")
            
        except Exception as e:
            self.log("WARNING", f"[DNS Security] Cache poisoning check failed: {e}")

    def _check_caa_records(self):
        """Check for CAA (Certification Authority Authorization) records."""
        self.log("INFO", "[DNS Security] Checking CAA records...")
        
        try:
            caa_answers = dns.resolver.resolve(self.domain, 'CAA')
            caa_records = []
            
            for rdata in caa_answers:
                caa_records.append(f"{rdata.flags} {rdata.tag} {rdata.value}")
            
            if caa_records:
                self.log("SUCCESS", f"[DNS Security] CAA Records: {', '.join(caa_records)}")
            else:
                self.log("INFO", "[DNS Security] No CAA records found (informational).")
                self.add_vuln(
                    title="CAA Record Not Configured",
                    severity="Low", category="DNS Security", cvss_score=2.1,
                    description=f"No CAA (Certification Authority Authorization) record found for {self.domain}. CAA records specify which CAs are allowed to issue certificates for the domain.",
                    remediation="Consider adding CAA records to restrict which certificate authorities can issue certificates:\n  Example: issue ca.example.com; issuewild ca.example.com"
                )
                
        except dns.resolver.NoAnswer:
            self.log("INFO", "[DNS Security] No CAA records found.")
        except Exception as e:
            self.log("WARNING", f"[DNS Security] CAA check failed: {e}")