feat(scanners): dns hijacking scanner

scanners/dns-scanner/dns_scanner/dns_scanner.py

@@ -12,12 +12,13 @@
 from dns.exception import Timeout
 
 from dns_scanner.email_scanners import DKIMScanner, DMARCScanner
+from dns_scanner.ns_registrar import check_ns_delegations, get_registrar_context
 
 logger = logging.getLogger(__name__)
 
 TIMEOUT = int(os.getenv("SCAN_TIMEOUT", "20"))
 DNSSEC_NAMESERVER_IP = os.getenv("DNSSEC_NAMESERVER_IP")
-DNSSEC_NAMESERVER_HOSTNAME= os.getenv("DNSSEC_NAMESERVER_HOSTNAME")
+DNSSEC_NAMESERVER_HOSTNAME = os.getenv("DNSSEC_NAMESERVER_HOSTNAME")
 
 
 @dataclass
@@ -27,6 +28,8 @@ class DNSScanResult:
     zone_apex: str = None
     record_exists: bool = None
     rcode: str = None
+    ns_delegations: dict = None
+    registrar_context: dict = None
     resolve_chain: list[list[str]] = None
     resolve_ips: [str] = None
     cname_record: str = None
@@ -50,9 +53,9 @@ def find_zone_apex(domain, resolver=None):
         while name != dns.name.root:
             logger.debug(f"Checking for SOA at {name}")
             try:
-                answers = resolver.resolve(name, 'SOA')
+                answers = resolver.resolve(name, "SOA")
                 logger.debug(f"Found SOA for {domain} at {name}: {answers[0]}")
-                zone_apex = str(name).rstrip('.')
+                zone_apex = str(name).rstrip(".")
                 return zone_apex
             except NoAnswer:
                 # Go up one level
@@ -105,7 +108,13 @@ def minimal_dnssec_check(domain, nameserver_ip, nameserver_hostname, _ttl_marker
     """
     try:
         q = dns.message.make_query(domain, dns.rdatatype.DNSKEY, want_dnssec=True)
-        resp = dns.query.tls(q, where=nameserver_ip, timeout=TIMEOUT, server_hostname=nameserver_hostname, verify=True)
+        resp = dns.query.tls(
+            q,
+            where=nameserver_ip,
+            timeout=TIMEOUT,
+            server_hostname=nameserver_hostname,
+            verify=True,
+        )
         if resp.rcode() != dns.rcode.NOERROR:
             return None
     except Timeout:
@@ -114,7 +123,9 @@ def minimal_dnssec_check(domain, nameserver_ip, nameserver_hostname, _ttl_marker
         )
         return None
     except Exception as e:
-        logger.error(f"Error while running minimal DNSSEC check for domain '{domain}': {e}")
+        logger.error(
+            f"Error while running minimal DNSSEC check for domain '{domain}': {e}"
+        )
         return None
     # Check if AD (Authenticated Data) flag is set, showing that the data is DNSSEC validated
     return bool(resp.flags & dns.flags.AD)
@@ -136,7 +147,7 @@ def get_wildcard_status(domain: str, resolver: Resolver, a_records: Answer):
     result = {"wildcard_entry": False, "wildcard_sibling": False}
     try:
         wildcard_sibling_domain = re.sub(r"^[^.]+", "*", domain)
-        wildcard_record = dns.resolver.resolve(
+        wildcard_record = resolver.resolve(
             wildcard_sibling_domain,
             rdtype=dns.rdatatype.A,
             raise_on_no_answer=False,
@@ -148,7 +159,7 @@ def get_wildcard_status(domain: str, resolver: Resolver, a_records: Answer):
                 try:
                     # check for mail-only subdomain (e.g. mail.example.com)
                     mx_records = resolver.resolve(qname=domain, rdtype=dns.rdatatype.MX)
-                    wildcard_mx = dns.resolver.resolve(
+                    wildcard_mx = resolver.resolve(
                         wildcard_sibling_domain,
                         rdtype=dns.rdatatype.MX,
                         raise_on_no_answer=False,
@@ -201,7 +212,12 @@ def scan_domain(domain, dkim_selectors=None):
 
     # Check if domain exists
     dns_answer_return_types = []
-    for query_type in [dns.rdatatype.A, dns.rdatatype.SOA, dns.rdatatype.NS]:
+    for query_type in [
+        dns.rdatatype.A,
+        dns.rdatatype.CNAME,
+        dns.rdatatype.SOA,
+        dns.rdatatype.NS,
+    ]:
         rtype = get_dns_return_type(domain, query_type)
         if rtype == "NOERROR":
             dns_answer_return_types.append(rtype)
@@ -284,10 +300,16 @@ def scan_domain(domain, dkim_selectors=None):
         logger.debug(f"Skipping DNSSEC check for {domain} - No zone apex found")
         zone_dnssec_enabled = None
     elif not DNSSEC_NAMESERVER_IP or not DNSSEC_NAMESERVER_HOSTNAME:
-        logger.debug(f"Skipping DNSSEC check for {domain} - DNSSEC nameserver environment variables not set")
+        logger.debug(
+            f"Skipping DNSSEC check for {domain} - DNSSEC nameserver environment variables not set"
+        )
         zone_dnssec_enabled = None
     else:
-        zone_dnssec_enabled = dnssec_check_with_ttl(domain=zone_apex, nameserver_ip=DNSSEC_NAMESERVER_IP, nameserver_hostname=DNSSEC_NAMESERVER_HOSTNAME)
+        zone_dnssec_enabled = dnssec_check_with_ttl(
+            domain=zone_apex,
+            nameserver_ip=DNSSEC_NAMESERVER_IP,
+            nameserver_hostname=DNSSEC_NAMESERVER_HOSTNAME,
+        )
 
     scan_result.zone_dnssec_enabled = zone_dnssec_enabled
 
@@ -297,12 +319,24 @@ def scan_domain(domain, dkim_selectors=None):
     dmarc_scanner = DMARCScanner(domain)
     dmarc_scan_result = dmarc_scanner.run()
     scan_result.base_domain = dmarc_scan_result.get("base_domain", "")
-    scan_result.ns_records = dmarc_scan_result.get("ns", {})
     scan_result.mx_records = dmarc_scan_result.get("mx", {})
     scan_result.spf = dmarc_scan_result.get("spf", {})
     scan_result.dmarc = dmarc_scan_result.get("dmarc", {})
     logger.debug(f"DMARC scan elapsed time: {time.monotonic() - dmarc_start_time}")
 
+    ns_records = dmarc_scan_result.get("ns", {"hostnames": [], "errors": []})
+    scan_result.ns_records = ns_records
+    # check nameserver delegations
+    scan_result.ns_delegations = check_ns_delegations(
+        domain=domain, zone_apex=zone_apex, ns_records=ns_records
+    )
+
+    registrar_domain = scan_result.base_domain or zone_apex or domain
+    scan_result.registrar_context = get_registrar_context(
+        base_domain=registrar_domain,
+        ns_hosts=scan_result.ns_delegations.get("ns_hosts", []),
+    )
+
     # If no MX records are found (with warnings), but there are CNAME records, check the CNAME target for MX records
     if (
         len(scan_result.mx_records.get("hosts", [])) == 0

scanners/dns-scanner/dns_scanner/ns_registrar.py

@@ -0,0 +1,196 @@
+import os
+
+import dns
+import dns.resolver
+import requests
+from dns.exception import Timeout
+from dns.resolver import NXDOMAIN, NoAnswer, NoNameservers
+
+TIMEOUT = int(os.getenv("SCAN_TIMEOUT", "20"))
+
+
+def probe_nameserver(
+    where: str, qname: str, qtype: str, recursion_desired: bool, timeout: int
+):
+    query = dns.message.make_query(
+        qname,
+        dns.rdatatype.from_text(qtype),
+        use_edns=True,
+    )
+    if not recursion_desired:
+        query.flags &= ~dns.flags.RD
+    return dns.query.udp(query, where=where, timeout=timeout)
+
+
+def get_ns_ip(host: str, resolver):
+    ns_ip = None
+    try:
+        ns_a = resolver.resolve(host, "A")
+        if ns_a:
+            ns_ip = ns_a[0].to_text()
+    except (NoAnswer, NXDOMAIN, NoNameservers, Timeout):
+        ns_ip = None
+
+    if ns_ip is None:
+        try:
+            ns_aaaa = resolver.resolve(host, "AAAA")
+            if ns_aaaa:
+                ns_ip = ns_aaaa[0].to_text()
+        except (NoAnswer, NXDOMAIN, NoNameservers, Timeout):
+            ns_ip = None
+
+    return ns_ip
+
+
+def check_ns_delegations(domain, zone_apex, ns_records, resolver=None, timeout_sec=10):
+    if resolver is None:
+        resolver = dns.resolver.get_default_resolver()
+
+    qname = zone_apex
+    if not zone_apex:
+        qname = domain
+
+    ns_hosts = ns_records.get("hostnames", [])
+    if len(ns_hosts) == 0:
+        try:
+            ns_res = resolver.resolve(domain, dns.rdatatype.NS)
+            ns_hosts = [host.to_text() for host in ns_res]
+        except (NoAnswer, NXDOMAIN, NoNameservers, Timeout):
+            ns_hosts = []
+
+    output = {
+        "ns_hosts": ns_hosts,
+        "ns_checks": [],
+        "ns_delegation": {
+            "total_ns": len(ns_hosts),
+            "authoritative_ok": 0,
+            "lame_count": 0,
+            "lame_type": "none",
+        },
+    }
+    if len(ns_hosts) == 0:
+        output["ns_delegation"]["lame_type"] = "unknown"
+        return output
+
+    for host in ns_hosts:
+        row = {
+            "ns_host": host,
+            "qname": qname,
+            "qtype": "SOA",
+            "rcode": None,
+            "answered_authoritatively": False,
+            "error": None,
+            "timeout": False,
+        }
+
+        try:
+            ns_ip = get_ns_ip(host, resolver)
+            if ns_ip is None:
+                row["error"] = "ns_ip_resolution_failed"
+                output["ns_delegation"]["lame_count"] += 1
+                output["ns_checks"].append(row)
+                continue
+
+            res = probe_nameserver(ns_ip, qname, "SOA", False, timeout_sec)
+            row["rcode"] = dns.rcode.to_text(res.rcode())
+            row["answered_authoritatively"] = bool(res.flags & dns.flags.AA)
+
+            if row["answered_authoritatively"] and row["rcode"] in [
+                "NOERROR",
+                "NXDOMAIN",
+            ]:
+                output["ns_delegation"]["authoritative_ok"] += 1
+            else:
+                output["ns_delegation"]["lame_count"] += 1
+        except Timeout:
+            row["timeout"] = True
+            row["error"] = "timeout"
+            output["ns_delegation"]["lame_count"] += 1
+        except Exception as e:
+            row["error"] = str(e)
+            output["ns_delegation"]["lame_count"] += 1
+
+        output["ns_checks"].append(row)
+
+    ok = output["ns_delegation"]["authoritative_ok"]
+    total = output["ns_delegation"]["total_ns"]
+
+    if ok == total:
+        output["ns_delegation"]["lame_type"] = "none"
+    elif ok == 0:
+        output["ns_delegation"]["lame_type"] = "full"
+    else:
+        output["ns_delegation"]["lame_type"] = "partial"
+
+    return output
+
+
+def get_registrar_context(base_domain, ns_hosts=None):
+    context = {
+        "base_domain": base_domain,
+        "lookup_success": False,
+        "rdap_url": None,
+        "registrar_name": None,
+        "registrar_id": None,
+        "rdap_nameservers": [],
+        "delegation_matches_rdap": None,
+        "error": None,
+    }
+
+    if not base_domain:
+        context["error"] = "missing_base_domain"
+        return context
+
+    rdap_url = f"https://rdap.org/domain/{base_domain}"
+    context["rdap_url"] = rdap_url
+
+    try:
+        response = requests.get(rdap_url, timeout=TIMEOUT)
+        response.raise_for_status()
+        payload = response.json()
+    except Exception as e:
+        context["error"] = str(e)
+        return context
+
+    context["lookup_success"] = True
+
+    nameservers = payload.get("nameservers", [])
+    context["rdap_nameservers"] = [
+        (ns.get("ldhName") or "").rstrip(".").lower()
+        for ns in nameservers
+        if ns.get("ldhName")
+    ]
+
+    entities = payload.get("entities", [])
+    for entity in entities:
+        roles = [r.lower() for r in entity.get("roles", [])]
+        if "registrar" not in roles:
+            continue
+
+        context["registrar_id"] = entity.get("handle")
+
+        vcard = entity.get("vcardArray", [])
+        if isinstance(vcard, list) and len(vcard) == 2 and isinstance(vcard[1], list):
+            for entry in vcard[1]:
+                if not isinstance(entry, list) or len(entry) < 4:
+                    continue
+                key = entry[0]
+                value = entry[3]
+                if key in {"fn", "org"} and value:
+                    context["registrar_name"] = value
+                    break
+
+        if context["registrar_name"] is None:
+            public_ids = entity.get("publicIds", [])
+            if public_ids:
+                context["registrar_name"] = public_ids[0].get("identifier")
+
+        break
+
+    if ns_hosts is not None:
+        normalized_hosts = {h.rstrip(".").lower() for h in ns_hosts if h}
+        rdap_hosts = set(context["rdap_nameservers"])
+        if rdap_hosts:
+            context["delegation_matches_rdap"] = normalized_hosts == rdap_hosts
+
+    return context

scanners/subdomain-takeover/.env.example

@@ -0,0 +1,7 @@
+NATS_URL=
+NATS_STREAM=
+SUBJECT_IN=
+SUBJECT_OUT=
+DURABLE_NAME=
+WORKER_COUNT=
+LOG_LEVEL=

scanners/subdomain-takeover/Dockerfile

@@ -0,0 +1,22 @@
+FROM golang:1.25-alpine AS builder
+
+WORKDIR /src
+RUN apk add --no-cache git ca-certificates
+
+COPY go.mod go.sum ./
+RUN go mod download
+
+COPY . .
+RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o /out/subdomain-takeover ./cmd/service
+
+# Runtime stage
+FROM alpine:3.22
+
+RUN addgroup -S scanner && adduser -S scanner -G scanner
+WORKDIR /app
+RUN apk add --no-cache ca-certificates tzdata
+
+COPY --from=builder /out/subdomain-takeover /app/subdomain-takeover
+
+USER scanner
+ENTRYPOINT ["/app/subdomain-takeover"]