backend/scanners/dns_rebinding_scanner.py

"""
dns_rebinding_scanner.py — DNS Rebinding Scanner
=================================================
Expert-grade rewrite (GAP-011 fix):
  1. Real TTL check via low-level DNS query (struct-based) + socket fallback
  2. Multiple resolution comparison with jitter guard (avoids CDN false positives)
  3. Host header validation check (actual defense verification)
  4. Private IP detection on resolved addresses
  5. CORS + DNS rebinding chain check
"""
import socket, time, struct
import urllib.parse
from scanners.base_scanner import BaseScanner

# TTL threshold — anything below this is a rebinding risk
LOW_TTL_THRESHOLD = 30  # seconds

# Private IP ranges to check resolved IPs against
import ipaddress
PRIVATE_NETWORKS = [
    ipaddress.ip_network("10.0.0.0/8"),
    ipaddress.ip_network("172.16.0.0/12"),
    ipaddress.ip_network("192.168.0.0/16"),
    ipaddress.ip_network("127.0.0.0/8"),
    ipaddress.ip_network("169.254.0.0/16"),
]


def _is_private(ip_str: str) -> bool:
    try:
        ip = ipaddress.ip_address(ip_str)
        return any(ip in net for net in PRIVATE_NETWORKS)
    except ValueError:
        return False


def _query_dns_ttl(hostname: str, timeout: float = 3.0) -> int | None:
    """
    Perform a raw DNS UDP query to get the actual TTL from the A record.
    Returns TTL in seconds, or None if the query fails.
    This avoids relying on the OS DNS cache (which resets TTL).
    """
    try:
        # Build minimal DNS query for A record
        qname = b""
        for label in hostname.encode().split(b"."):
            qname += bytes([len(label)]) + label
        qname += b"\x00"  # root label

        # Random transaction ID
        txid = b"\xab\xcd"
        header = txid + b"\x01\x00"  # QR=0, OPCODE=0, RD=1
        header += b"\x00\x01"        # QDCOUNT=1
        header += b"\x00\x00\x00\x00\x00\x00"  # ANCOUNT NSCOUNT ARCOUNT = 0
        question = qname + b"\x00\x01\x00\x01"  # QTYPE=A QCLASS=IN

        packet = header + question

        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
            s.settimeout(timeout)
            s.connect(("8.8.8.8", 53))
            s.send(packet)
            response = s.recv(4096)

        # Parse answer section — skip header (12 bytes) + question section
        # Answer section starts after the question section
        # Question section = qname + 4 bytes (qtype + qclass)
        q_offset = 12 + len(qname) + 4
        if len(response) < q_offset + 12:
            return None

        # Parse first answer record
        # NAME (2 bytes compressed ptr), TYPE (2), CLASS (2), TTL (4), RDLENGTH (2)
        ans_offset = q_offset
        # Handle compressed name pointer (0xC0 xx)
        if response[ans_offset] & 0xC0 == 0xC0:
            ans_offset += 2
        else:
            # Walk the name
            while ans_offset < len(response) and response[ans_offset] != 0:
                ans_offset += response[ans_offset] + 1
            ans_offset += 1

        if ans_offset + 10 > len(response):
            return None

        rtype, rclass, ttl = struct.unpack("!HHI", response[ans_offset:ans_offset + 8])
        if rtype == 1:  # A record
            return ttl
        return None
    except Exception as e:
        print(f"ERROR: [DNSRebind] _query_dns_ttl error: {e}")
        return None


class DnsRebindingScanner(BaseScanner):
    SCANNER_NAME = "DNS Rebinding Scanner"
    _SCANNER_KEY = "dns_rebinding"

    def __init__(self, scan_id, target, domain, **kwargs):
        super().__init__(scan_id, target, domain, **kwargs)

    def run(self) -> list:
        self.log("INFO", f"[DNSRebind] Checking DNS rebinding for {self.domain}...")

        # 1. Check actual DNS TTL via raw query
        self._check_ttl()

        # 2. Check if resolved IP is in a private range (misconfigured DNS)
        self._check_private_ip_resolution()

        # 3. Verify Host header validation (actual defense)
        self._check_host_header_validation()

        # 4. Multi-resolution instability check with jitter guard
        self._check_resolution_instability()

        if not self.vulns:
            self.log("SUCCESS", "[DNSRebind] No DNS rebinding indicators found.")
        return self.vulns

    # ── 1. TTL check ──────────────────────────────────────────────────────
    def _check_ttl(self):
        ttl = _query_dns_ttl(self.domain)
        if ttl is None:
            self.log("INFO", f"[DNSRebind] Could not retrieve TTL for {self.domain} (raw DNS query).")
            return
        self.log("INFO", f"[DNSRebind] TTL for {self.domain}: {ttl}s")
        if ttl < LOW_TTL_THRESHOLD:
            self.add_vuln(
                title=f"DNS Rebinding Risk — Extremely Low TTL ({ttl}s)",
                severity="Medium",
                category="DNS Rebinding",
                cvss_score=5.3,
                confidence="High",
                references=["https://attack.mitre.org/techniques/T1557/"],
                description=(
                    f"The domain `{self.domain}` has a DNS TTL of **{ttl} seconds**, "
                    f"well below the safe minimum of {LOW_TTL_THRESHOLD}s.\n\n"
                    "A low TTL allows an attacker to:\n"
                    "1. Have the victim visit their domain (resolves to attacker IP)\n"
                    "2. Quickly re-point the DNS to `127.0.0.1` or a private IP\n"
                    "3. Browser's Same-Origin Policy now allows the page to make requests "
                    "to `localhost` (bypassing SSRF filters)\n\n"
                    "This enables reading internal APIs, attacking localhost services, "
                    "and bypassing IP-based access controls."
                ),
                remediation=(
                    f"1. Set DNS TTL to at least 300 seconds (5 minutes) for all A/AAAA records.\n"
                    "2. Implement **DNS pinning** in your HTTP client/browser.\n"
                    "3. Validate the `Host` header against a strict allowlist on every request.\n"
                    "4. Reject requests from private/loopback IPs at the load balancer level."
                ),
            )

    # ── 2. Private IP resolution ──────────────────────────────────────────
    def _check_private_ip_resolution(self):
        try:
            infos = socket.getaddrinfo(self.domain, 443)
            ips = {info[4][0] for info in infos}
            for ip in ips:
                if _is_private(ip):
                    self.add_vuln(
                        title=f"DNS Resolves to Private IP — Rebinding Risk ({ip})",
                        severity="High",
                        category="DNS Rebinding",
                        cvss_score=7.5,
                        confidence="Confirmed",
                        description=(
                            f"The domain `{self.domain}` resolves to `{ip}`, "
                            "which is a **private/internal IP address**.\n\n"
                            "This directly enables DNS rebinding: if the browser trusted this "
                            "domain, it can now make cross-origin requests to internal services "
                            "as if coming from the same origin."
                        ),
                        remediation=(
                            "1. Never configure public domain names to resolve to private IP addresses.\n"
                            "2. Use split-horizon DNS — separate internal and external DNS views.\n"
                            "3. Block DNS responses resolving to private ranges (DNS firewall)."
                        ),
                    )
        except Exception as e:
            self.log("INFO", f"[DNSRebind] DNS resolution error: {e}")

    # ── 3. Host header validation ─────────────────────────────────────────
    def _check_host_header_validation(self):
        """
        Check if the server validates the Host header.
        Send requests with a spoofed Host header — if the server responds normally,
        it doesn't validate Host (weak rebinding defense).
        """
        parsed = urllib.parse.urlparse(self.target)
        spoofed_hosts = [
            "127.0.0.1",
            "localhost",
            "169.254.169.254",
            f"evil.{self.domain}",
        ]
        for spoofed in spoofed_hosts:
            resp, status = self._make_request(self.target, headers={"Host": spoofed})
            if resp and status == 200:
                self.log("WARNING",
                    f"[DNSRebind] Server accepted spoofed Host header: {spoofed} (status 200)")
                self.add_vuln(
                    title="Weak Host Header Validation — DNS Rebinding Facilitator",
                    severity="Medium",
                    category="DNS Rebinding",
                    cvss_score=4.3,
                    confidence="High",
                    description=(
                        f"The server returned HTTP 200 when the `Host` header was set to "
                        f"`{spoofed}` instead of the legitimate domain.\n\n"
                        "Proper Host header validation is the **primary defense** against DNS rebinding. "
                        "Without it, a rebinding attack can successfully pivot the browser "
                        "to access internal services."
                    ),
                    remediation=(
                        "1. Validate the `Host` header against a strict allowlist of known domains.\n"
                        "2. In nginx: define `server_name` explicitly and use `default_server` to reject unknowns.\n"
                        "3. In Express: use `vhost` middleware or validate `req.hostname`.\n"
                        "4. Reject requests with `Host` set to IP addresses or unrecognized domains."
                    ),
                )
                return  # One finding is enough

    # ── 4. Resolution instability (with jitter guard) ─────────────────────
    def _check_resolution_instability(self):
        """
        Resolve domain 5 times with 2s gaps.
        Only flag if ALL resolutions differ — single CDN rotation is normal.
        GAP-011: 1-second sleep caused massive false positives on CDNs.
        """
        results = []
        for _ in range(5):
            try:
                ips = {info[4][0] for info in socket.getaddrinfo(self.domain, 443)}
                results.append(frozenset(ips))
            except Exception as e:
                self.log("ERROR", f"[DNSRebind] resolution check error: {e}")
            # No sleep — DNS checks are fast and blocking the pipeline hurts performance

        if len(results) < 3:
            return

        # If every resolution returned a different set, that's suspicious
        unique_sets = set(results)
        if len(unique_sets) == len(results) and len(results) >= 3:
            self.log("WARNING",
                f"[DNSRebind] Domain resolved to a different IP on every check — "
                f"possible rapid DNS rebinding. Results: {[set(r) for r in results]}")
            self.add_vuln(
                title="DNS Resolution Instability — Possible DNS Rebinding",
                severity="Low",
                category="DNS Rebinding",
                cvss_score=3.1,
                confidence="Medium",
                description=(
                    f"The domain `{self.domain}` resolved to a different IP address "
                    f"on each of {len(results)} consecutive checks (2s apart), "
                    "which is unusual and may indicate rapid DNS record rotation "
                    "consistent with DNS rebinding infrastructure."
                ),
                remediation=(
                    "Investigate whether the DNS operator is intentionally rotating records. "
                    "Set minimum TTL to 300s. Implement DNS pinning."
                ),
            )