seer-core 0.54.0

//! NS delegation health: compares the parent zone's delegation NS set
//! against the zone's own authoritative NS RRset and probes each delegated
//! server for lameness.
//!
//! Flow (see [`DelegationChecker::check`]):
//! 1. Resolve the parent zone's NS set through the recursive resolver.
//! 2. Query up to two responsive parent servers directly with recursion
//!    disabled for the domain's NS — the delegation lives in the referral's
//!    AUTHORITY section (or the ANSWER section when the parent server also
//!    happens to be authoritative for the child). The union is `delegated_ns`.
//! 3. Query each delegated NS directly (RD=0) for the domain's own NS RRset.
//!    A server that fails to answer authoritatively for its own zone is lame
//!    (refused / timeout / non-authoritative / empty). The union of
//!    authoritative answers is `zone_ns`.
//! 4. Compare the two sets case-insensitively with trailing dots normalized.
//!
//! [`DelegationChecker::delegated_ns`] runs steps 1–2 only, on a shorter
//! deadline: [`crate::SmartLookup`] uses it to cross-check a WHOIS-only
//! result's nameservers ([`DnsNameservers`]), since some registries' WHOIS
//! lags their own zone.
//!
//! **Parent-zone simplification:** the parent zone is derived by stripping the
//! domain's leftmost label (`example.com` → `com`, `example.co.uk` → `co.uk`).
//! For multi-label public suffixes this is exactly the registrable parent, and
//! for registrable domains it is the TLD — the common cases. A PSL-aware
//! parent derivation (needed only for exotic zone cuts below private suffixes)
//! is future work; for a subdomain whose immediate parent is not a zone cut,
//! the NS lookup on the synthesized parent simply fails with a typed error.
//!
//! **Retry boundary (deliberate):** like [`crate::dns::resolver`], this module
//! does not wrap queries in [`crate::retry::RetryPolicy`]. hickory performs
//! its own retransmission (`opts.attempts = 2`) within the per-query timeout;
//! a lameness probe wrapped in an outer retry loop would only multiply
//! worst-case latency without changing the verdict.
//!
//! **SSRF:** delegated NS hostnames come from untrusted DNS data, so every
//! address they resolve to is checked against
//! [`crate::validation::describe_reserved_ip`] before a direct query is sent.
//! Servers that resolve only to reserved/private space are skipped with a
//! warning. Tests reach loopback fixtures through `#[cfg(test)]`-only seams
//! (`allowing_private_hosts` / `with_port_map` / `with_recursive_upstream`);
//! the production validation path is never weakened.
//!
//! **Bounded work:** at most [`MAX_PARENT_SERVERS_TRIED`] parent servers are
//! contacted (stopping after [`PARENT_SERVERS_WANTED`] respond), at most
//! [`MAX_NS_PROBED`] delegated servers are probed (concurrently), each
//! address lookup and query runs under the per-query timeout, and each server
//! is asked on one vetted address at a time, IPv4 first: an address this
//! host has no route to (IPv6 on an IPv4-only host, or the reverse) fails at
//! once and the server's next address is tried. The whole check runs under
//! one deadline of [`CHECK_BUDGET_TIMEOUTS`] per-query timeouts; a probe
//! still running when it passes is reported as not probed.

use std::collections::BTreeSet;
use std::net::IpAddr;
use std::time::Duration;

use hickory_resolver::config::ResolverOpts;
use hickory_resolver::net::runtime::TokioRuntimeProvider;
use hickory_resolver::net::{DnsError, NetError};
use hickory_resolver::proto::op::ResponseCode;
use hickory_resolver::proto::rr::{RData as HickoryRData, Record, RecordType as HickoryRecordType};
use hickory_resolver::TokioResolver;
use serde::{Deserialize, Serialize};
use tracing::{debug, instrument};

use super::resolver::{apply_standard_opts, fqdn, google_or_pinned, single_server_config};
use super::DEFAULT_DNS_TIMEOUT;
use crate::error::{Result, SeerError};
use crate::validation::normalize_domain;

/// The whole check's deadline, in per-query timeouts. An address lookup is
/// held to one timeout, but a lookup or query that hickory retransmits
/// (`attempts = 2`) may take two: the parent NS lookup (2), then
/// [`PARENT_SERVERS_WANTED`] parent servers that each answer only on the
/// retransmission (3 each: lookup + query), then the concurrent probes (3).
/// Without the bound, silent servers held the caller for up to 17 timeouts
/// (85s at the default 5s): the parent phase is sequential over
/// [`MAX_PARENT_SERVERS_TRIED`] servers.
const CHECK_BUDGET_TIMEOUTS: u32 = 2 + 3 * PARENT_SERVERS_WANTED as u32 + 3;

/// Overall deadline of [`DelegationChecker::delegated_ns`], in per-query
/// timeouts. It is supplementary detail on a lookup, so it gets far less
/// than a full [`DelegationChecker::check`]: a parent zone that answers at
/// all does so well within it, and one that doesn't is not worth waiting on.
const DELEGATED_NS_BUDGET_TIMEOUTS: u32 = 2;

/// Maximum delegated NS servers probed for lameness. Delegations larger than
/// this are unusual; the cap bounds total probe work and a warning notes the
/// truncation.
const MAX_NS_PROBED: usize = 6;

/// Maximum parent-zone servers contacted while hunting for responsive ones.
const MAX_PARENT_SERVERS_TRIED: usize = 4;

/// Stop querying parent servers once this many have answered.
const PARENT_SERVERS_WANTED: usize = 2;

/// Why a delegated nameserver is lame, for consumers that act on the cause
/// without parsing [`LameNs::reason`] (the security audit grades a server
/// that answered wrongly apart from one that never answered).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum LameKind {
    /// Answered NXDOMAIN for its own zone.
    Nxdomain,
    /// Answered authoritatively, but with no NS records.
    NoNsRecords,
    /// Answered the NS set without the AA bit (a cache, not the authority).
    NotAuthoritative,
    /// Referred the query elsewhere instead of answering.
    Referral,
    /// Answered with an empty, non-authoritative response.
    EmptyAnswer,
    /// Answered with an error RCODE (REFUSED, SERVFAIL, …).
    ErrorRcode,
    /// Gave no usable response (timeout, transport failure) — the server may
    /// be dead, or the probe may have been lost.
    Unresponsive,
}

/// A delegated nameserver that failed to answer authoritatively for its own
/// zone, with a structured [`LameKind`] and a human-readable reason.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct LameNs {
    /// Nameserver host name (normalized: lowercase, no trailing dot).
    pub host: String,
    /// What made the server lame.
    pub kind: LameKind,
    /// Why the server is considered lame, for display.
    pub reason: String,
}

/// Result of an NS delegation health check.
///
/// All NS host names are normalized (lowercase, trailing dot stripped) and
/// sorted, so the report is deterministic and set comparisons are exact.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DelegationReport {
    /// The checked domain (normalized).
    pub domain: String,
    /// The parent zone the delegation was read from (domain minus its
    /// leftmost label — see the module docs for this simplification).
    pub parent_zone: String,
    /// Parent-zone servers that answered the delegation query.
    pub parent_server_queried: Vec<String>,
    /// The parent's view: NS names in the delegation (referral) for the domain.
    pub delegated_ns: Vec<String>,
    /// The authoritative view: union of NS RRsets returned (with the AA bit
    /// set) by the delegated servers themselves.
    pub zone_ns: Vec<String>,
    /// True when the parent and zone NS sets match exactly and no probed
    /// server is lame.
    pub in_sync: bool,
    /// NS names the parent lists but the zone's own RRset omits.
    pub missing_from_zone: Vec<String>,
    /// NS names the zone's RRset lists but the parent delegation omits.
    pub missing_from_parent: Vec<String>,
    /// Delegated servers that failed to answer authoritatively.
    pub lame: Vec<LameNs>,
    /// Non-fatal anomalies (unresolvable glue, skipped servers, truncated
    /// probe list, ...).
    pub warnings: Vec<String>,
}

/// A registry's WHOIS nameserver list checked against the live parent-zone
/// delegation ([`DelegationChecker::delegated_ns`]).
///
/// Some registries' WHOIS lags their own zone (NIC Togo listed 2 of the 4
/// nameservers `.tg` delegates for `nic.tg`), and a WHOIS-only lookup has no
/// RDAP record to cross-check against, so [`crate::SmartLookup`] attaches
/// this to WHOIS-sourced results. Names are normalized (lowercase, no
/// trailing dot) and sorted.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct DnsNameservers {
    /// NS names the parent zone delegates for the domain.
    pub delegated: Vec<String>,
    /// True when WHOIS lists exactly the delegated set.
    pub in_sync: bool,
    /// Delegated NS names WHOIS does not list.
    pub missing_from_whois: Vec<String>,
    /// NS names WHOIS lists that the parent does not delegate.
    pub missing_from_dns: Vec<String>,
}

impl DnsNameservers {
    /// Compares `whois` (any case, trailing dots allowed) with the
    /// `delegated` set.
    pub fn compare(whois: &[String], delegated: Vec<String>) -> Self {
        let norm = |n: &String| n.trim_end_matches('.').to_ascii_lowercase();
        let whois: BTreeSet<String> = whois.iter().map(norm).collect();
        let delegated: BTreeSet<String> = delegated.iter().map(norm).collect();
        let missing_from_whois: Vec<String> = delegated.difference(&whois).cloned().collect();
        let missing_from_dns: Vec<String> = whois.difference(&delegated).cloned().collect();
        Self {
            in_sync: missing_from_whois.is_empty() && missing_from_dns.is_empty(),
            delegated: delegated.into_iter().collect(),
            missing_from_whois,
            missing_from_dns,
        }
    }

    /// True when WHOIS lists no nameservers but the parent delegates some.
    /// Many registries' WHOIS omits nameservers entirely, so unlike a partial
    /// list this is no sign of a stale record.
    pub fn whois_lists_none(&self) -> bool {
        // Any WHOIS name either shrinks `missing_from_whois` below the
        // delegated set or lands in `missing_from_dns`.
        !self.delegated.is_empty()
            && self.missing_from_dns.is_empty()
            && self.missing_from_whois.len() == self.delegated.len()
    }
}

/// The parent zone's answer for a delegation (steps 1–2 of
/// [`DelegationChecker::check`]).
struct ParentDelegation {
    /// Parent-zone servers that answered.
    servers_queried: Vec<String>,
    /// Union of the NS names they delegate (normalized, sorted).
    delegated: BTreeSet<String>,
    warnings: Vec<String>,
}

/// Classified result of a single non-recursive NS query against one server.
///
/// hickory folds a referral for an NS query (empty ANSWER, matching NS
/// records in AUTHORITY) into a successful `Lookup` — `contains_answer`
/// scans all sections for the queried type — so both the `Ok` and the
/// `NoRecordsFound` paths are normalized into the same `Response` shape
/// here, keeping ANSWER and AUTHORITY NS sets separate.
#[derive(Debug)]
enum DirectNs {
    /// The server produced a DNS response (any RCODE except hard errors).
    Response {
        /// The response AA bit. `false` for the `NoRecordsFound` path, which
        /// does not carry the header — for this module's queries that path
        /// only occurs for NODATA/NXDOMAIN responses, which are never
        /// treated as authoritative NS answers anyway.
        authoritative: bool,
        response_code: ResponseCode,
        /// NS targets for the queried domain from the ANSWER section.
        answer_ns: Vec<String>,
        /// NS targets for the queried domain from the AUTHORITY section
        /// (where a parent publishes the delegation / referral).
        authority_ns: Vec<String>,
    },
    /// The server responded with an error RCODE (REFUSED, SERVFAIL, ...).
    Rcode(ResponseCode),
    /// No usable response (timeout / transport failure), with the reason.
    Unreachable(String),
    /// The probe packet never left this host (no route for the address
    /// family, or no socket of that family at all — e.g. an IPv6-only
    /// server probed from an IPv4-only host).
    /// The same epistemic state as unresolvable glue: nothing is known
    /// about the server's authority, so this maps to a skip, not lameness.
    LocalNoRoute(String),
}

/// Per-server outcome of a lameness probe.
#[derive(Debug)]
enum ProbeOutcome {
    /// Authoritative NS answer — contributes to `zone_ns`.
    Healthy(Vec<String>),
    /// The server is lame: its kind and the display reason.
    Lame(LameKind, String),
    /// The server could not be probed at all (e.g. unresolvable glue);
    /// reported as a warning, not as lameness.
    Skipped(String),
}

/// Checks NS delegation health for a domain.
///
/// Construct with [`DelegationChecker::new`] (defaults) or
/// [`DelegationChecker::from_config`] (honors `~/.seer/config.toml`
/// `timeouts.dns_secs`), then call [`check`](DelegationChecker::check).
#[derive(Clone)]
pub struct DelegationChecker {
    /// Per-query timeout for both recursive and direct queries.
    timeout: Duration,
    /// Recursive resolver used to discover the parent zone's NS set and to
    /// resolve nameserver host addresses ("glue", loosely — resolution goes
    /// through recursion rather than the parent's additional section).
    recursive: TokioResolver,
    /// Test-only: pin the recursive resolver to a loopback mock.
    #[cfg(test)]
    recursive_upstream: Option<(IpAddr, u16)>,
    /// Test-only: per-host port override for direct queries, so each mock
    /// "server" can live on its own ephemeral loopback port.
    #[cfg(test)]
    port_map: Option<std::collections::HashMap<String, u16>>,
    /// Test-only: skip the SSRF/reserved-IP validation on direct targets.
    #[cfg(test)]
    allow_private_hosts: bool,
    /// Test-only: addresses whose direct queries fail as if this host had
    /// no route to them (see `with_unroutable`).
    #[cfg(test)]
    unroutable: Vec<IpAddr>,
    /// Test-only: the whole check's deadline, instead of
    /// [`CHECK_BUDGET_TIMEOUTS`] per-query timeouts.
    #[cfg(test)]
    budget: Option<Duration>,
}

impl std::fmt::Debug for DelegationChecker {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("DelegationChecker")
            .field("timeout", &self.timeout)
            .finish()
    }
}

impl Default for DelegationChecker {
    fn default() -> Self {
        Self::new()
    }
}

impl DelegationChecker {
    /// Creates a delegation checker with default settings (5s per-query
    /// timeout, Google DNS as the recursive resolver).
    pub fn new() -> Self {
        Self {
            timeout: DEFAULT_DNS_TIMEOUT,
            recursive: build_recursive_resolver(DEFAULT_DNS_TIMEOUT, None),
            #[cfg(test)]
            recursive_upstream: None,
            #[cfg(test)]
            port_map: None,
            #[cfg(test)]
            allow_private_hosts: false,
            #[cfg(test)]
            unroutable: Vec::new(),
            #[cfg(test)]
            budget: None,
        }
    }

    /// Builds a checker honoring `~/.seer/config.toml` settings.
    ///
    /// Reads `timeouts.dns_secs` (already clamped to 1–60s by
    /// [`crate::config::SeerConfig::load`]). Sugar over
    /// [`DelegationChecker::with_timeout`].
    pub fn from_config(config: &crate::config::SeerConfig) -> Self {
        Self::new().with_timeout(config.dns_timeout())
    }

    /// Sets the per-query timeout (applies to recursive lookups, parent
    /// delegation queries, and lameness probes alike).
    pub fn with_timeout(mut self, timeout: Duration) -> Self {
        self.timeout = timeout;
        self.recursive = build_recursive_resolver(timeout, self.recursive_upstream());
        self
    }

    /// Test-only: point the recursive resolver at a loopback mock server.
    #[cfg(test)]
    fn with_recursive_upstream(mut self, ip: IpAddr, port: u16) -> Self {
        self.recursive_upstream = Some((ip, port));
        self.recursive = build_recursive_resolver(self.timeout, self.recursive_upstream);
        self
    }

    /// Test-only: route direct queries for the given (normalized) host names
    /// to per-host mock ports instead of port 53.
    #[cfg(test)]
    fn with_port_map(mut self, map: std::collections::HashMap<String, u16>) -> Self {
        self.port_map = Some(map);
        self
    }

    /// Test-only: allow direct queries to loopback/private hosts (mock servers).
    #[cfg(test)]
    fn allowing_private_hosts(mut self) -> Self {
        self.allow_private_hosts = true;
        self
    }

    /// Test-only: direct queries to `addrs` fail at once with a local
    /// no-route error, as on a host without a route to that family.
    #[cfg(test)]
    fn with_unroutable(mut self, addrs: &[IpAddr]) -> Self {
        self.unroutable.extend_from_slice(addrs);
        self
    }

    /// Test-only: hold the whole check to `budget`.
    #[cfg(test)]
    fn with_budget(mut self, budget: Duration) -> Self {
        self.budget = Some(budget);
        self
    }

    /// The whole check's deadline: [`CHECK_BUDGET_TIMEOUTS`] per-query
    /// timeouts.
    /// Steps 1–2 of [`check`](Self::check): the parent zone's NS set, then the
    /// delegation for `domain` from up to [`PARENT_SERVERS_WANTED`] of its
    /// servers. Errors when the parent has no NS set or no parent server
    /// answers; an empty delegation is a warning.
    async fn parent_delegation(
        &self,
        domain: &str,
        parent_zone: &str,
        deadline: tokio::time::Instant,
        budget: Duration,
    ) -> Result<ParentDelegation> {
        let mut warnings = Vec::new();
        let out_of_time = || {
            SeerError::DnsError(format!(
                "delegation check for {domain} gave up after {budget:?} without a parent-zone answer"
            ))
        };

        // 1. Parent zone NS set via the recursive resolver.
        let parent_hosts = tokio::time::timeout_at(deadline, self.recursive_ns_set(parent_zone))
            .await
            .map_err(|_| out_of_time())??;
        if parent_hosts.is_empty() {
            return Err(SeerError::DnsError(format!(
                "parent zone {} has no NS records — cannot locate the delegation for {}",
                parent_zone, domain
            )));
        }

        // 2. Ask up to PARENT_SERVERS_WANTED responsive parent servers for the
        //    delegation (RD=0). Referrals carry the NS set in AUTHORITY; a
        //    parent that is also authoritative for the child answers in ANSWER.
        let mut parent_server_queried = Vec::new();
        let mut delegated: BTreeSet<String> = BTreeSet::new();
        for host in parent_hosts.iter().take(MAX_PARENT_SERVERS_TRIED) {
            if parent_server_queried.len() >= PARENT_SERVERS_WANTED {
                break;
            }
            let Ok(asked) = tokio::time::timeout_at(deadline, self.ask_host(host, domain)).await
            else {
                warnings.push(format!(
                    "stopped asking parent servers: the check's {budget:?} budget ran out"
                ));
                break;
            };
            let direct = match asked {
                Ok(direct) => direct,
                Err(reason) => {
                    warnings.push(format!("skipped parent server {}: {}", host, reason));
                    continue;
                }
            };
            match direct {
                DirectNs::Response {
                    response_code,
                    answer_ns,
                    authority_ns,
                    ..
                } => {
                    if response_code == ResponseCode::NXDomain {
                        warnings.push(format!(
                            "parent server {} returned NXDOMAIN for {}",
                            host, domain
                        ));
                    }
                    // Referrals carry the delegation in AUTHORITY; a parent
                    // that is also authoritative for the child answers in
                    // ANSWER. Union both.
                    delegated.extend(answer_ns);
                    delegated.extend(authority_ns);
                    parent_server_queried.push(host.clone());
                }
                DirectNs::Rcode(code) => {
                    warnings.push(format!("parent server {} answered with {}", host, code));
                }
                DirectNs::Unreachable(reason) => {
                    warnings.push(format!(
                        "parent server {} did not respond: {}",
                        host, reason
                    ));
                }
                DirectNs::LocalNoRoute(reason) => {
                    warnings.push(format!(
                        "parent server {} not probed — no route from this host ({})",
                        host, reason
                    ));
                }
            }
        }

        if parent_server_queried.is_empty() {
            return Err(SeerError::DnsError(format!(
                "no parent-zone ({}) nameserver answered the delegation query for {}",
                parent_zone, domain
            )));
        }
        if delegated.is_empty() {
            warnings.push(format!(
                "parent zone {} returned no NS delegation for {} — the domain may not be delegated",
                parent_zone, domain
            ));
        }

        Ok(ParentDelegation {
            servers_queried: parent_server_queried,
            delegated,
            warnings,
        })
    }

    /// The NS set the parent zone delegates for `domain` (normalized:
    /// lowercase, no trailing dot; sorted) — the registry's live view, without
    /// [`check`](Self::check)'s per-server lameness probes. Bounded by two
    /// per-query timeouts overall (`DELEGATED_NS_BUDGET_TIMEOUTS`), far less
    /// than a full check: it is supplementary detail on a lookup.
    #[instrument(skip(self), fields(domain = %domain))]
    pub async fn delegated_ns(&self, domain: &str) -> Result<Vec<String>> {
        let domain = normalize_domain(domain)?;
        let parent_zone = parent_zone_of(&domain);
        let budget = self.timeout.saturating_mul(DELEGATED_NS_BUDGET_TIMEOUTS);
        let deadline = tokio::time::Instant::now() + budget;
        let delegation = self
            .parent_delegation(&domain, &parent_zone, deadline, budget)
            .await?;
        Ok(delegation.delegated.into_iter().collect())
    }

    #[cfg(not(test))]
    fn budget(&self) -> Duration {
        self.timeout.saturating_mul(CHECK_BUDGET_TIMEOUTS)
    }

    #[cfg(test)]
    fn budget(&self) -> Duration {
        self.budget
            .unwrap_or_else(|| self.timeout.saturating_mul(CHECK_BUDGET_TIMEOUTS))
    }

    #[cfg(test)]
    fn simulated_no_route(&self, ip: IpAddr) -> bool {
        self.unroutable.contains(&ip)
    }

    #[cfg(not(test))]
    fn simulated_no_route(&self, _ip: IpAddr) -> bool {
        false
    }

    #[cfg(test)]
    fn recursive_upstream(&self) -> Option<(IpAddr, u16)> {
        self.recursive_upstream
    }

    #[cfg(not(test))]
    fn recursive_upstream(&self) -> Option<(IpAddr, u16)> {
        None
    }

    #[cfg(test)]
    fn allow_private(&self) -> bool {
        self.allow_private_hosts
    }

    #[cfg(not(test))]
    fn allow_private(&self) -> bool {
        false
    }

    /// Port used for a direct query to `host`. Always 53 in production; the
    /// `#[cfg(test)]` port map lets each mock server bind its own ephemeral
    /// loopback port.
    #[cfg(test)]
    fn direct_port(&self, host: &str) -> u16 {
        self.port_map
            .as_ref()
            .and_then(|map| map.get(&normalize_ns_name(host)).copied())
            .unwrap_or(53)
    }

    #[cfg(not(test))]
    fn direct_port(&self, _host: &str) -> u16 {
        53
    }

    /// Runs the delegation health check for `domain`.
    ///
    /// # Arguments
    /// * `domain` - The domain to check (e.g. "example.com"); normalized first
    ///
    /// # Returns
    /// * `Ok(DelegationReport)` - the comparison + lameness report
    /// * `Err(SeerError)` - when the input is invalid, the parent zone's NS
    ///   set cannot be resolved, or no parent server answers the delegation
    ///   query (without parent data there is nothing to compare)
    #[instrument(skip(self), fields(domain = %domain))]
    pub async fn check(&self, domain: &str) -> Result<DelegationReport> {
        let domain = normalize_domain(domain)?;
        let parent_zone = parent_zone_of(&domain);
        let budget = self.budget();
        let deadline = tokio::time::Instant::now() + budget;

        // 1–2. The parent zone's delegation.
        let ParentDelegation {
            servers_queried: parent_server_queried,
            delegated,
            mut warnings,
        } = self
            .parent_delegation(&domain, &parent_zone, deadline, budget)
            .await?;

        // 3. Probe the delegated servers (bounded, concurrent) for lameness.
        let probe_hosts: Vec<String> = delegated.iter().take(MAX_NS_PROBED).cloned().collect();
        if delegated.len() > MAX_NS_PROBED {
            warnings.push(format!(
                "probing only {} of {} delegated nameservers",
                MAX_NS_PROBED,
                delegated.len()
            ));
        }
        // Each probe under the check's deadline, so one that runs out is
        // reported on its own and the probes that finished are kept.
        let outcomes = futures::future::join_all(probe_hosts.iter().map(|host| async {
            tokio::time::timeout_at(deadline, self.probe_delegated_ns(host, &domain))
                .await
                .unwrap_or_else(|_| {
                    ProbeOutcome::Skipped(format!(
                        "not probed within the check's {budget:?} budget"
                    ))
                })
        }))
        .await;

        let mut zone_ns: BTreeSet<String> = BTreeSet::new();
        let mut lame = Vec::new();
        for (host, outcome) in probe_hosts.iter().zip(outcomes) {
            match outcome {
                ProbeOutcome::Healthy(ns) => zone_ns.extend(ns),
                ProbeOutcome::Lame(kind, reason) => lame.push(LameNs {
                    host: host.clone(),
                    kind,
                    reason,
                }),
                ProbeOutcome::Skipped(reason) => {
                    warnings.push(format!("could not probe {}: {}", host, reason));
                }
            }
        }

        // 4. Set comparison. When no server produced an authoritative answer
        //    there is no zone view to diff against — flag it instead of
        //    reporting every delegated NS as "missing from zone".
        let (missing_from_zone, missing_from_parent) = if zone_ns.is_empty() {
            if !delegated.is_empty() {
                warnings.push(
                    "no delegated nameserver returned an authoritative NS answer — \
                     parent/zone set comparison skipped"
                        .to_string(),
                );
            }
            (Vec::new(), Vec::new())
        } else {
            (
                delegated.difference(&zone_ns).cloned().collect(),
                zone_ns.difference(&delegated).cloned().collect(),
            )
        };

        let in_sync = !delegated.is_empty()
            && !zone_ns.is_empty()
            && missing_from_zone.is_empty()
            && missing_from_parent.is_empty()
            && lame.is_empty();

        lame.sort_by(|a, b| a.host.cmp(&b.host));

        Ok(DelegationReport {
            domain,
            parent_zone,
            parent_server_queried,
            delegated_ns: delegated.into_iter().collect(),
            zone_ns: zone_ns.into_iter().collect(),
            in_sync,
            missing_from_zone,
            missing_from_parent,
            lame,
            warnings,
        })
    }

    /// Resolves a zone's NS set through the recursive resolver, returning
    /// normalized host names (sorted, deduplicated). NXDOMAIN/NODATA fold to
    /// an empty set; transport failures surface as errors.
    async fn recursive_ns_set(&self, zone: &str) -> Result<Vec<String>> {
        match self
            .recursive
            .lookup(fqdn(zone), HickoryRecordType::NS)
            .await
        {
            Ok(lookup) => {
                let set: BTreeSet<String> =
                    ns_targets_for(lookup.answers(), zone).into_iter().collect();
                Ok(set.into_iter().collect())
            }
            Err(e) if e.is_no_records_found() => Ok(Vec::new()),
            Err(e) => Err(SeerError::DnsError(format!(
                "NS lookup for parent zone {} failed: {}",
                zone, e
            ))),
        }
    }

    /// Resolves the usable (public, unless the test seam allows private)
    /// addresses of a nameserver host, IPv4 first ([`crate::net::ipv4_first`]). Errors
    /// carry a human-readable reason for the report's warnings.
    async fn resolve_host_ips(&self, host: &str) -> std::result::Result<Vec<IpAddr>, String> {
        // The resolver re-sends a query that timed out (`attempts`), so only
        // this deadline holds the lookup to one per-query timeout.
        let response = tokio::time::timeout(self.timeout, self.recursive.lookup_ip(fqdn(host)))
            .await
            .unwrap_or(Err(NetError::Timeout))
            .map_err(|e| format!("could not resolve {}: {}", host, e))?;
        let ips: Vec<IpAddr> = response.iter().collect();
        if ips.is_empty() {
            return Err(format!("{} did not resolve to any address", host));
        }
        // SSRF protection: NS host names come from untrusted DNS data, so
        // reserved/private targets are refused. The test seam is
        // `#[cfg(test)]`-only; production always validates.
        let (vetted, refused) = partition_reserved(&ips, self.allow_private());
        if vetted.is_empty() {
            return Err(match refused.first() {
                Some((_, reason)) => format!("{} resolves to a blocked address ({})", host, reason),
                None => format!("{} did not resolve to a usable address", host),
            });
        }
        let mut vetted = vetted;
        crate::net::ipv4_first(&mut vetted, |ip| *ip);
        Ok(vetted)
    }

    /// Asks `host` for `domain`'s NS set (RD=0) on its addresses in turn,
    /// until a query leaves this host: an address with no local route
    /// ([`DirectNs::LocalNoRoute`]) fails at once and the next is tried —
    /// an IPv6-only server from an IPv4-only host, or the reverse. `Err` is
    /// an address-lookup failure; when no address is routable, the last
    /// `LocalNoRoute`.
    async fn ask_host(&self, host: &str, domain: &str) -> std::result::Result<DirectNs, String> {
        let mut last = None;
        for ip in self.resolve_host_ips(host).await? {
            debug!(server = %host, %ip, "direct NS query");
            match self.direct_ns_query(host, ip, domain).await {
                DirectNs::LocalNoRoute(reason) => {
                    last = Some(DirectNs::LocalNoRoute(format!("{ip}: {reason}")));
                }
                direct => return Ok(direct),
            }
        }
        // `resolve_host_ips` returns at least one address.
        Ok(last.unwrap_or_else(|| DirectNs::Unreachable("no address to query".to_string())))
    }

    /// Sends a single non-recursive NS query for `domain` to the server at
    /// `ip` and classifies the outcome. Never fails hard: construction and
    /// transport errors fold into [`DirectNs::Unreachable`] so callers can
    /// record them per-server.
    async fn direct_ns_query(&self, host: &str, ip: IpAddr, domain: &str) -> DirectNs {
        if self.simulated_no_route(ip) {
            return DirectNs::LocalNoRoute("network is unreachable (simulated)".to_string());
        }
        let resolver = match self.build_direct_resolver(host, ip) {
            Ok(resolver) => resolver,
            Err(e) => return DirectNs::Unreachable(format!("resolver construction failed: {}", e)),
        };
        match resolver.lookup(fqdn(domain), HickoryRecordType::NS).await {
            // The Lookup preserves the raw response message, so the AA bit
            // is the server's own claim of authority, and a referral's NS
            // records are still sitting in the AUTHORITY section.
            Ok(lookup) => {
                let message = lookup.message();
                DirectNs::Response {
                    authoritative: message.metadata.authoritative,
                    response_code: message.metadata.response_code,
                    answer_ns: ns_targets_for(&message.answers, domain),
                    authority_ns: ns_targets_for(&message.authorities, domain),
                }
            }
            Err(e) => classify_net_error(e, domain),
        }
    }

    /// Builds a one-shot stub resolver pinned to `ip` with recursion
    /// disabled, mirroring the construction pattern in
    /// [`crate::dns::resolver`]. UDP with TCP fallback, so truncated NS sets
    /// retry over TCP.
    fn build_direct_resolver(&self, host: &str, ip: IpAddr) -> Result<TokioResolver> {
        let config = single_server_config(ip, self.direct_port(host), true);
        let mut builder =
            TokioResolver::builder_with_config(config, TokioRuntimeProvider::default());
        apply_direct_opts(builder.options_mut(), self.timeout);
        builder
            .build()
            .map_err(|e| SeerError::DnsError(format!("failed to construct DNS resolver: {}", e)))
    }

    /// Probes one delegated nameserver: sends the RD=0 NS query
    /// ([`ask_host`](Self::ask_host)) and classifies the result.
    async fn probe_delegated_ns(&self, host: &str, domain: &str) -> ProbeOutcome {
        match self.ask_host(host, domain).await {
            Ok(direct) => outcome_for_direct(direct),
            // Unresolvable "glue" is a warning, not lameness: the server was
            // never reached, so nothing is known about its authority.
            Err(reason) => ProbeOutcome::Skipped(reason),
        }
    }
}

/// Maps a direct-query result onto the per-server probe verdict. Pure so the
/// lameness-vs-skip policy is unit-testable without a live probe.
fn outcome_for_direct(direct: DirectNs) -> ProbeOutcome {
    match direct {
        DirectNs::Response {
            response_code: ResponseCode::NXDomain,
            ..
        } => ProbeOutcome::Lame(
            LameKind::Nxdomain,
            "returned NXDOMAIN for its own zone".to_string(),
        ),
        DirectNs::Response {
            authoritative: true,
            answer_ns,
            ..
        } if !answer_ns.is_empty() => ProbeOutcome::Healthy(answer_ns),
        DirectNs::Response {
            authoritative: true,
            ..
        } => ProbeOutcome::Lame(
            LameKind::NoNsRecords,
            "authoritative answer contained no NS records".to_string(),
        ),
        DirectNs::Response { answer_ns, .. } if !answer_ns.is_empty() => ProbeOutcome::Lame(
            LameKind::NotAuthoritative,
            "answered non-authoritatively (AA bit not set)".to_string(),
        ),
        DirectNs::Response { authority_ns, .. } if !authority_ns.is_empty() => ProbeOutcome::Lame(
            LameKind::Referral,
            "not authoritative — referred the query elsewhere".to_string(),
        ),
        DirectNs::Response { .. } => ProbeOutcome::Lame(
            LameKind::EmptyAnswer,
            "returned an empty answer (no NS records)".to_string(),
        ),
        DirectNs::Rcode(code) => ProbeOutcome::Lame(
            LameKind::ErrorRcode,
            format!("refused the query ({})", code),
        ),
        DirectNs::Unreachable(reason) => ProbeOutcome::Lame(
            LameKind::Unresponsive,
            format!("no usable response: {}", reason),
        ),
        DirectNs::LocalNoRoute(reason) => ProbeOutcome::Skipped(format!(
            "no route from this host ({}) — server not probed (e.g. an IPv6-only nameserver on an IPv4-only host)",
            reason
        )),
    }
}

/// Options for a direct (RD=0) query to one server: the shared option set
/// (timeout, attempts, no hosts file, pinned server order — hand-copied here
/// it had drifted and lacked the ordering) with recursion disabled, since
/// delegation data must come from the server's own authority, not from
/// recursion or a forwarder's cache.
fn apply_direct_opts(opts: &mut ResolverOpts, timeout: Duration) {
    apply_standard_opts(opts, timeout);
    opts.recursion_desired = false;
}

/// Builds the recursive resolver: Google DNS (UDP+TCP) in production, or a
/// pinned loopback upstream for the `#[cfg(test)]` seam. Shared with the
/// trace walker (`dns::trace`), which resolves glueless nameservers through
/// it.
///
/// The `expect` expresses the same invariant as
/// `dns::resolver::build_default_resolver`: with plain UDP/TCP upstreams the
/// only fallible step (rustls TLS-context construction for DoT/DoH) never
/// runs, so construction cannot fail.
pub(crate) fn build_recursive_resolver(
    timeout: Duration,
    upstream: Option<(IpAddr, u16)>,
) -> TokioResolver {
    let mut builder = TokioResolver::builder_with_config(
        google_or_pinned(upstream),
        TokioRuntimeProvider::default(),
    );
    // Shared with the main resolver so the option sets cannot drift: the
    // hand-copied version here lacked the pinned `UserProvidedOrder` server
    // ordering, so parent-NS discovery could draw black-holed IPv6 upstreams.
    apply_standard_opts(builder.options_mut(), timeout);
    builder
        .build()
        .expect("plain UDP/TCP resolver build cannot fail with the bundled webpki root store")
}

/// Classifies a hickory lookup error into a [`DirectNs`] outcome, extracting
/// referral NS names for `domain` from the AUTHORITY section when present.
/// The `NoRecordsFound` path only occurs when no NS record for the domain
/// exists in any section (NODATA / NXDOMAIN) — see [`DirectNs`].
fn classify_net_error(err: NetError, domain: &str) -> DirectNs {
    match err {
        NetError::Dns(DnsError::NoRecordsFound(no_records)) => DirectNs::Response {
            authoritative: false,
            response_code: no_records.response_code,
            answer_ns: Vec::new(),
            authority_ns: no_records
                .authorities
                .as_deref()
                .map(|records| ns_targets_for(records, domain))
                .unwrap_or_default(),
        },
        NetError::Dns(DnsError::ResponseCode(code)) => DirectNs::Rcode(code),
        NetError::Timeout => DirectNs::Unreachable("timed out".to_string()),
        // What `resolver.lookup()` actually returns when every transport
        // attempt died at the local socket layer (verified against
        // hickory-resolver 0.26): the pool's error fold (`most_specific`)
        // always prefers the non-`Io` side and its accumulator starts as
        // `NoConnections`, so all-`Io` failures surface as `NoConnections`.
        // For these fresh single-server pools that can only mean local
        // socket errors (no route / address family unavailable): hickory's
        // UDP is unconnected (`send_to`/`recv_from`), so a *remote* refusal
        // never yields `Io` on the UDP leg — a dark or refusing server
        // times out instead, and `Timeout` outranks `NoConnections` in the
        // fold. The server was therefore never reached: local, not lame.
        NetError::NoConnections => DirectNs::LocalNoRoute(
            "every connection attempt failed at the local socket layer".to_string(),
        ),
        // Defense in depth should hickory ever surface the socket error
        // directly instead of folding it into `NoConnections`.
        NetError::Io(io_err) if is_local_no_route(&io_err) => {
            DirectNs::LocalNoRoute(io_err.to_string())
        }
        other => DirectNs::Unreachable(other.to_string()),
    }
}

/// True when an io transport error means the packet never left this host:
/// the kernel had no route to the destination (ENETUNREACH/EHOSTUNREACH), or
/// cannot open a socket of its address family at all (EAFNOSUPPORT — e.g. an
/// IPv6 server on a host whose kernel has IPv6 disabled). `TimedOut` never
/// reaches here — `From<io::Error> for NetError` folds it into
/// `NetError::Timeout` (which stays a lameness signal: packets were sent).
pub(crate) fn is_local_no_route(err: &std::io::Error) -> bool {
    use std::io::ErrorKind::{HostUnreachable, NetworkUnreachable};
    matches!(err.kind(), NetworkUnreachable | HostUnreachable)
        || err
            .raw_os_error()
            .is_some_and(|code| Some(code) == ADDRESS_FAMILY_UNSUPPORTED)
}

/// The OS error code for "address family not supported". std maps it to no
/// `io::ErrorKind` of its own on any platform (it reads as uncategorized), so
/// [`is_local_no_route`] matches the raw code: `EAFNOSUPPORT` on Unix, whose
/// value differs across targets (97 on x86 and Arm Linux, 47 on macOS) —
/// hence `libc` — and on Windows Winsock's `WSAEAFNOSUPPORT`, which is what
/// std and tokio report for a failed socket call there (not the C runtime's
/// `EAFNOSUPPORT`).
#[cfg(unix)]
pub(crate) const ADDRESS_FAMILY_UNSUPPORTED: Option<i32> = Some(libc::EAFNOSUPPORT);
#[cfg(windows)]
pub(crate) const ADDRESS_FAMILY_UNSUPPORTED: Option<i32> =
    Some(windows_sys::Win32::Networking::WinSock::WSAEAFNOSUPPORT);
#[cfg(not(any(unix, windows)))]
pub(crate) const ADDRESS_FAMILY_UNSUPPORTED: Option<i32> = None;

/// The parent zone of a normalized domain: everything after the leftmost
/// label. `normalize_domain` guarantees at least one interior dot, so the
/// fallback branch is unreachable in practice (kept panic-free regardless).
fn parent_zone_of(domain: &str) -> String {
    domain
        .split_once('.')
        .map(|(_, rest)| rest.to_string())
        .unwrap_or_else(|| domain.to_string())
}

/// SSRF vetting for nameserver addresses, which come from untrusted DNS data
/// (resolved NS names here, glue in the trace walker): splits `ips` into the
/// addresses a direct query may be sent to and the refused ones, each with
/// its [`crate::validation::describe_reserved_ip`] reason. `allow_private`
/// is the `#[cfg(test)]` seam's flag and is always `false` in production.
pub(crate) fn partition_reserved(
    ips: &[IpAddr],
    allow_private: bool,
) -> (Vec<IpAddr>, Vec<(IpAddr, &'static str)>) {
    let mut usable = Vec::new();
    let mut refused = Vec::new();
    for &ip in ips {
        match crate::validation::describe_reserved_ip(&ip) {
            Some(reason) if !allow_private => refused.push((ip, reason)),
            _ => usable.push(ip),
        }
    }
    (usable, refused)
}

/// Normalizes an NS host name for set comparison: lowercase, trailing dot
/// stripped.
fn normalize_ns_name(name: &str) -> String {
    name.trim_end_matches('.').to_ascii_lowercase()
}

/// Extracts the normalized NS target names owned by `owner` from a record
/// slice (answer or authority section). Records for other owners — e.g. the
/// parent zone's own SOA in a NODATA response — are ignored.
///
/// Names are rendered with `to_ascii()`, never `to_string()`: hickory's
/// `Display` decodes `xn--` labels to Unicode, so for every IDN the owner
/// (`münchen.de.`) never equalled the punycode `owner` from
/// `normalize_domain` (`xn--mnchen-3ya.de`). Every IDN domain came back with
/// an empty delegation ("not delegated"), and IDN TLDs hard-errored on an
/// empty parent NS set.
fn ns_targets_for(records: &[Record], owner: &str) -> Vec<String> {
    let owner = normalize_ns_name(owner);
    records
        .iter()
        .filter_map(|record| {
            if normalize_ns_name(&record.name.to_ascii()) != owner {
                return None;
            }
            match &record.data {
                HickoryRData::NS(ns) => Some(normalize_ns_name(&ns.0.to_ascii())),
                _ => None,
            }
        })
        .collect()
}

#[cfg(test)]
mod tests {
    //! Hermetic tests: every DNS "server" in a scenario (the recursive
    //! resolver, the parent server, and each delegated NS) is a loopback UDP
    //! mock on its own ephemeral port, reached through the `#[cfg(test)]`
    //! seams (`with_recursive_upstream` / `with_port_map` /
    //! `allowing_private_hosts`). Scenarios script the shared
    //! [`crate::dns::test_support`] fixture, whose replies cover the AA bit,
    //! AUTHORITY-section referrals and error RCODEs; its `MockMode::Ignore`
    //! serves the timeout case.

    use std::collections::HashMap;
    use std::net::Ipv4Addr;

    use hickory_resolver::proto::rr::rdata as wire;
    use hickory_resolver::proto::rr::Name;

    use super::*;
    use crate::dns::test_support::{spawn_mock_dns, spawn_mock_dns_fn, MockMode, MockReply};

    // --- pure helpers --------------------------------------------------

    #[test]
    fn parent_zone_strips_leftmost_label() {
        assert_eq!(parent_zone_of("example.com"), "com");
        assert_eq!(parent_zone_of("sub.example.co.uk"), "example.co.uk");
        assert_eq!(parent_zone_of("example.co.uk"), "co.uk");
    }

    #[test]
    fn direct_opts_share_the_standard_set_and_disable_recursion() {
        // Regression: the hand-copied option set lacked the pinned
        // `UserProvidedOrder` server ordering.
        let mut opts = ResolverOpts::default();
        apply_direct_opts(&mut opts, Duration::from_secs(7));
        assert_eq!(
            opts.server_ordering_strategy,
            hickory_resolver::config::ServerOrderingStrategy::UserProvidedOrder
        );
        assert_eq!(opts.timeout, Duration::from_secs(7));
        assert!(!opts.recursion_desired);
    }

    #[test]
    fn ns_targets_match_idn_owners_in_punycode() {
        // Regression: owners were compared via `Name`'s Display, which
        // renders `xn--` labels as Unicode, so this punycode owner (what
        // `normalize_domain("münchen.de")` produces) matched nothing.
        let owner = Name::from_ascii("xn--mnchen-3ya.de.").unwrap();
        let records = vec![
            Record::from_rdata(
                owner.clone(),
                300,
                HickoryRData::NS(wire::NS(
                    Name::from_ascii("ns1.xn--mnchen-3ya.de.").unwrap(),
                )),
            ),
            Record::from_rdata(
                owner,
                300,
                HickoryRData::NS(wire::NS(Name::from_ascii("NS2.Example.NET.").unwrap())),
            ),
        ];
        let mut targets = ns_targets_for(&records, "xn--mnchen-3ya.de");
        targets.sort();
        assert_eq!(
            targets,
            vec![
                "ns1.xn--mnchen-3ya.de".to_string(),
                "ns2.example.net".to_string()
            ]
        );
        // An IDN TLD (`рф` = `xn--p1ai`) as the owner, for the parent-zone
        // NS lookup path.
        let tld = vec![Record::from_rdata(
            Name::from_ascii("xn--p1ai.").unwrap(),
            300,
            HickoryRData::NS(wire::NS(Name::from_ascii("a.dns.ripn.net.").unwrap())),
        )];
        assert_eq!(
            ns_targets_for(&tld, "xn--p1ai"),
            vec!["a.dns.ripn.net".to_string()]
        );
    }

    #[test]
    fn ns_name_normalization_is_case_and_dot_insensitive() {
        assert_eq!(normalize_ns_name("NS1.Example.COM."), "ns1.example.com");
        assert_eq!(normalize_ns_name("ns1.example.com"), "ns1.example.com");
    }

    #[test]
    fn partition_reserved_refuses_reserved_addresses_with_their_reason() {
        let public: IpAddr = "192.5.6.30".parse().unwrap();
        let private: IpAddr = "10.0.0.53".parse().unwrap();
        let loopback: IpAddr = "::1".parse().unwrap();
        let (usable, refused) = partition_reserved(&[private, public, loopback], false);
        assert_eq!(usable, vec![public]);
        assert_eq!(
            refused,
            vec![
                (private, "private network (RFC 1918)"),
                (loopback, "IPv6 loopback (::1)")
            ]
        );
        // The test seam's flag lets everything through, in order.
        let (usable, refused) = partition_reserved(&[private, public], true);
        assert_eq!(usable, vec![private, public]);
        assert!(refused.is_empty());
    }

    #[test]
    fn local_no_route_transport_errors_classify_as_local() {
        use std::io;
        // ENETUNREACH/EHOSTUNREACH mean the probe packet never left this
        // host (e.g. an IPv6-only NS probed from an IPv4-only host); hickory
        // surfaces them as `NetError::Io`. They say nothing about the
        // server's authority, so they must not classify as a remote failure.
        for kind in [
            io::ErrorKind::NetworkUnreachable,
            io::ErrorKind::HostUnreachable,
        ] {
            let direct = classify_net_error(io::Error::new(kind, "no route").into(), "example.com");
            assert!(
                matches!(direct, DirectNs::LocalNoRoute(_)),
                "{kind:?} classified as {direct:?}"
            );
        }
        // What `resolver.lookup()` actually surfaces for the all-local-
        // failure case: the pool folds per-attempt `Io` errors away
        // (`most_specific` always prefers the non-`Io` side, and the
        // accumulator starts as `NoConnections`), so a no-route probe comes
        // back as `NoConnections`, never `Io`. It must classify as local.
        let pooled = classify_net_error(NetError::NoConnections, "example.com");
        assert!(
            matches!(pooled, DirectNs::LocalNoRoute(_)),
            "NoConnections classified as {pooled:?}"
        );
        // A refused connection is the remote host answering: a genuine
        // remote signal that must keep reading as Unreachable (→ lame).
        let refused = classify_net_error(
            io::Error::new(io::ErrorKind::ConnectionRefused, "refused").into(),
            "example.com",
        );
        assert!(
            matches!(refused, DirectNs::Unreachable(_)),
            "refused classified as {refused:?}"
        );
    }

    #[test]
    fn address_family_unsupported_classifies_as_local() {
        use std::io;
        // EAFNOSUPPORT (WSAEAFNOSUPPORT on Windows): no socket of the
        // server's address family can be opened on this host, e.g. an IPv6
        // server when the kernel has IPv6 disabled. Nothing was sent. std
        // gives the code no ErrorKind, so it is matched by the raw OS code,
        // which is how a failed socket call reports it.
        let code = ADDRESS_FAMILY_UNSUPPORTED.expect("a Unix or Windows target");
        let err = io::Error::from_raw_os_error(code);
        assert!(is_local_no_route(&err), "{err:?} ({:?})", err.kind());
        let direct = classify_net_error(err.into(), "example.com");
        assert!(
            matches!(direct, DirectNs::LocalNoRoute(_)),
            "EAFNOSUPPORT classified as {direct:?}"
        );
        // The code decides, not the wording or a nearby kind: an error with
        // no OS code, or an unsupported operation, stays a remote failure.
        for other in [
            io::Error::other("Address family not supported by protocol"),
            io::Error::from(io::ErrorKind::Unsupported),
        ] {
            assert!(!is_local_no_route(&other), "{other:?}");
        }
    }

    /// Every lameness branch carries its own structured kind, so consumers
    /// (the security audit) never parse `reason`.
    #[test]
    fn each_lame_outcome_has_its_kind() {
        let response =
            |authoritative, code, answer: &[&str], authority: &[&str]| DirectNs::Response {
                authoritative,
                response_code: code,
                answer_ns: answer.iter().map(|s| s.to_string()).collect(),
                authority_ns: authority.iter().map(|s| s.to_string()).collect(),
            };
        let kind = |direct| match outcome_for_direct(direct) {
            ProbeOutcome::Lame(kind, _) => kind,
            other => panic!("expected lame, got {other:?}"),
        };
        assert_eq!(
            kind(response(true, ResponseCode::NXDomain, &[], &[])),
            LameKind::Nxdomain
        );
        assert_eq!(
            kind(response(true, ResponseCode::NoError, &[], &[])),
            LameKind::NoNsRecords
        );
        assert_eq!(
            kind(response(false, ResponseCode::NoError, &["ns1.test"], &[])),
            LameKind::NotAuthoritative
        );
        assert_eq!(
            kind(response(
                false,
                ResponseCode::NoError,
                &[],
                &["ns.parent.test"]
            )),
            LameKind::Referral
        );
        assert_eq!(
            kind(response(false, ResponseCode::NoError, &[], &[])),
            LameKind::EmptyAnswer
        );
        assert_eq!(
            kind(DirectNs::Rcode(ResponseCode::Refused)),
            LameKind::ErrorRcode
        );
        assert_eq!(
            kind(DirectNs::Unreachable("timed out".to_string())),
            LameKind::Unresponsive
        );
    }

    #[test]
    fn local_no_route_probe_outcome_is_skipped_not_lame() {
        let outcome =
            outcome_for_direct(DirectNs::LocalNoRoute("Network is unreachable".to_string()));
        assert!(
            matches!(outcome, ProbeOutcome::Skipped(_)),
            "got {outcome:?}"
        );
    }

    #[test]
    fn from_config_applies_dns_timeout() {
        let mut config = crate::config::SeerConfig::default();
        config.timeouts.dns_secs = 9;
        let checker = DelegationChecker::from_config(&config);
        assert_eq!(checker.timeout, Duration::from_secs(9));
    }

    fn ns_rdata(target: &str) -> HickoryRData {
        HickoryRData::NS(wire::NS(
            Name::from_ascii(target).expect("valid test NS name"),
        ))
    }

    fn loopback_a() -> HickoryRData {
        HickoryRData::A(wire::A(Ipv4Addr::LOCALHOST))
    }

    /// Recursive-resolver mock for the standard scenario: parent zone `test`
    /// served by `a.parent.test`, and A records (→127.0.0.1) for every host
    /// in `resolvable`.
    async fn spawn_recursive_mock(resolvable: &[&str]) -> u16 {
        let resolvable: Vec<String> = resolvable.iter().map(|s| s.to_string()).collect();
        spawn_mock_dns_fn(move |qname, qtype| {
            match (qname, qtype) {
                ("test", HickoryRecordType::NS) => {
                    MockReply::Answer(vec![ns_rdata("a.parent.test.")])
                }
                (host, HickoryRecordType::A)
                    if host == "a.parent.test" || resolvable.iter().any(|r| r == host) =>
                {
                    MockReply::Answer(vec![loopback_a()])
                }
                // AAAA and unknown hosts: NODATA (lookup_ip falls back / fails).
                _ => MockReply::NoData,
            }
        })
        .await
    }

    /// An authoritative zone server answering `domain NS` with `rrset`.
    async fn spawn_zone_server(rrset: &[&str]) -> u16 {
        let records: Vec<HickoryRData> = rrset.iter().map(|ns| ns_rdata(ns)).collect();
        spawn_mock_dns_fn(move |_qname, qtype| {
            if qtype == HickoryRecordType::NS {
                MockReply::AuthoritativeAnswer(records.clone())
            } else {
                MockReply::NoData
            }
        })
        .await
    }

    /// A parent server that refers `seer.test` to `referral` (glueless).
    async fn spawn_parent_server(referral: &[&str]) -> u16 {
        let servers: Vec<(String, Vec<IpAddr>)> = referral
            .iter()
            .map(|ns| (ns.to_string(), Vec::new()))
            .collect();
        spawn_mock_dns_fn(move |qname, qtype| {
            if qname == "seer.test" && qtype == HickoryRecordType::NS {
                MockReply::Delegation {
                    zone: "seer.test".to_string(),
                    servers: servers.clone(),
                }
            } else {
                MockReply::NoData
            }
        })
        .await
    }

    fn checker(recursive_port: u16, port_map: HashMap<String, u16>) -> DelegationChecker {
        DelegationChecker::new()
            .with_timeout(Duration::from_millis(500))
            .allowing_private_hosts()
            .with_recursive_upstream(Ipv4Addr::LOCALHOST.into(), recursive_port)
            .with_port_map(port_map)
    }

    // --- end-to-end scenarios -------------------------------------------

    #[tokio::test]
    async fn healthy_in_sync_delegation() {
        let recursive = spawn_recursive_mock(&["ns1.seer.test", "ns2.seer.test"]).await;
        let parent = spawn_parent_server(&["ns1.seer.test.", "ns2.seer.test."]).await;
        let ns1 = spawn_zone_server(&["ns1.seer.test.", "ns2.seer.test."]).await;
        let ns2 = spawn_zone_server(&["ns1.seer.test.", "ns2.seer.test."]).await;

        let report = checker(
            recursive,
            HashMap::from([
                ("a.parent.test".to_string(), parent),
                ("ns1.seer.test".to_string(), ns1),
                ("ns2.seer.test".to_string(), ns2),
            ]),
        )
        .check("seer.test")
        .await
        .expect("healthy delegation check must succeed");

        assert_eq!(report.domain, "seer.test");
        assert_eq!(report.parent_zone, "test");
        assert_eq!(report.parent_server_queried, vec!["a.parent.test"]);
        assert_eq!(report.delegated_ns, vec!["ns1.seer.test", "ns2.seer.test"]);
        assert_eq!(report.zone_ns, vec!["ns1.seer.test", "ns2.seer.test"]);
        assert!(report.in_sync, "report should be in sync: {report:?}");
        assert!(report.missing_from_zone.is_empty());
        assert!(report.missing_from_parent.is_empty());
        assert!(report.lame.is_empty());
    }

    #[tokio::test]
    async fn parent_lists_ns_the_zone_omits() {
        // Parent delegates to ns1 + ns3; both answer authoritatively but the
        // zone RRset only contains ns1 → ns3 is missing_from_zone.
        let recursive = spawn_recursive_mock(&["ns1.seer.test", "ns3.seer.test"]).await;
        let parent = spawn_parent_server(&["ns1.seer.test.", "ns3.seer.test."]).await;
        let zone = spawn_zone_server(&["ns1.seer.test."]).await;
        let zone2 = spawn_zone_server(&["ns1.seer.test."]).await;

        let report = checker(
            recursive,
            HashMap::from([
                ("a.parent.test".to_string(), parent),
                ("ns1.seer.test".to_string(), zone),
                ("ns3.seer.test".to_string(), zone2),
            ]),
        )
        .check("seer.test")
        .await
        .expect("check must succeed");

        assert_eq!(report.delegated_ns, vec!["ns1.seer.test", "ns3.seer.test"]);
        assert_eq!(report.zone_ns, vec!["ns1.seer.test"]);
        assert_eq!(report.missing_from_zone, vec!["ns3.seer.test"]);
        assert!(report.missing_from_parent.is_empty());
        assert!(!report.in_sync);
        assert!(report.lame.is_empty(), "set mismatch is not lameness");
    }

    #[tokio::test]
    async fn zone_lists_ns_the_parent_omits() {
        // Parent only delegates ns1, but the zone's RRset advertises ns1 +
        // ns2 → ns2 is missing_from_parent.
        let recursive = spawn_recursive_mock(&["ns1.seer.test"]).await;
        let parent = spawn_parent_server(&["ns1.seer.test."]).await;
        let zone = spawn_zone_server(&["ns1.seer.test.", "ns2.seer.test."]).await;

        let report = checker(
            recursive,
            HashMap::from([
                ("a.parent.test".to_string(), parent),
                ("ns1.seer.test".to_string(), zone),
            ]),
        )
        .check("seer.test")
        .await
        .expect("check must succeed");

        assert_eq!(report.delegated_ns, vec!["ns1.seer.test"]);
        assert_eq!(report.zone_ns, vec!["ns1.seer.test", "ns2.seer.test"]);
        assert_eq!(report.missing_from_parent, vec!["ns2.seer.test"]);
        assert!(report.missing_from_zone.is_empty());
        assert!(!report.in_sync);
    }

    #[tokio::test]
    async fn refusing_server_is_lame() {
        let recursive = spawn_recursive_mock(&["ns1.seer.test", "ns2.seer.test"]).await;
        let parent = spawn_parent_server(&["ns1.seer.test.", "ns2.seer.test."]).await;
        let ns1 = spawn_zone_server(&["ns1.seer.test.", "ns2.seer.test."]).await;
        let ns2 = spawn_mock_dns_fn(|_, _| MockReply::Rcode(ResponseCode::Refused)).await;

        let report = checker(
            recursive,
            HashMap::from([
                ("a.parent.test".to_string(), parent),
                ("ns1.seer.test".to_string(), ns1),
                ("ns2.seer.test".to_string(), ns2),
            ]),
        )
        .check("seer.test")
        .await
        .expect("check must succeed");

        // The sets still match (ns1 serves the full RRset), but the lame
        // server alone must break in_sync.
        assert!(report.missing_from_zone.is_empty());
        assert!(report.missing_from_parent.is_empty());
        assert_eq!(report.lame.len(), 1, "lame: {:?}", report.lame);
        assert_eq!(report.lame[0].host, "ns2.seer.test");
        assert!(
            report.lame[0].reason.to_lowercase().contains("refused"),
            "reason should mention the refusal: {}",
            report.lame[0].reason
        );
        assert_eq!(report.lame[0].kind, LameKind::ErrorRcode);
        assert!(!report.in_sync);
    }

    #[tokio::test]
    async fn non_authoritative_answer_is_lame() {
        let recursive = spawn_recursive_mock(&["ns1.seer.test", "ns2.seer.test"]).await;
        let parent = spawn_parent_server(&["ns1.seer.test.", "ns2.seer.test."]).await;
        let ns1 = spawn_zone_server(&["ns1.seer.test.", "ns2.seer.test."]).await;
        // ns2 answers the right RRset but without the AA bit — a cache, not
        // an authority.
        let ns2 = spawn_mock_dns_fn(|_, qtype| {
            if qtype == HickoryRecordType::NS {
                MockReply::Answer(vec![ns_rdata("ns1.seer.test."), ns_rdata("ns2.seer.test.")])
            } else {
                MockReply::NoData
            }
        })
        .await;

        let report = checker(
            recursive,
            HashMap::from([
                ("a.parent.test".to_string(), parent),
                ("ns1.seer.test".to_string(), ns1),
                ("ns2.seer.test".to_string(), ns2),
            ]),
        )
        .check("seer.test")
        .await
        .expect("check must succeed");

        assert_eq!(report.lame.len(), 1, "lame: {:?}", report.lame);
        assert_eq!(report.lame[0].host, "ns2.seer.test");
        assert!(
            report.lame[0].reason.contains("non-authoritatively"),
            "reason: {}",
            report.lame[0].reason
        );
        assert_eq!(report.lame[0].kind, LameKind::NotAuthoritative);
        assert!(!report.in_sync);
        // The healthy server still provides the zone view.
        assert_eq!(report.zone_ns, vec!["ns1.seer.test", "ns2.seer.test"]);
    }

    #[tokio::test]
    async fn unresponsive_server_is_lame_with_timeout_reason() {
        let recursive = spawn_recursive_mock(&["ns1.seer.test", "ns2.seer.test"]).await;
        let parent = spawn_parent_server(&["ns1.seer.test.", "ns2.seer.test."]).await;
        let ns1 = spawn_zone_server(&["ns1.seer.test.", "ns2.seer.test."]).await;
        // Reuse the shared fixture's black-hole mode for the no-response case.
        let ns2 = spawn_mock_dns(MockMode::Ignore).await;

        let report = checker(
            recursive,
            HashMap::from([
                ("a.parent.test".to_string(), parent),
                ("ns1.seer.test".to_string(), ns1),
                ("ns2.seer.test".to_string(), ns2),
            ]),
        )
        .check("seer.test")
        .await
        .expect("check must succeed");

        assert_eq!(report.lame.len(), 1, "lame: {:?}", report.lame);
        assert_eq!(report.lame[0].host, "ns2.seer.test");
        assert!(
            report.lame[0].reason.contains("no usable response"),
            "reason: {}",
            report.lame[0].reason
        );
        assert_eq!(report.lame[0].kind, LameKind::Unresponsive);
        assert!(!report.in_sync);
    }

    #[tokio::test]
    async fn unresolvable_glue_is_warned_and_skipped_not_lame() {
        // ns2 has no A record at the recursive resolver → skipped with a
        // warning; the remaining server still proves the sets match.
        let recursive = spawn_recursive_mock(&["ns1.seer.test"]).await;
        let parent = spawn_parent_server(&["ns1.seer.test.", "ns2.seer.test."]).await;
        let ns1 = spawn_zone_server(&["ns1.seer.test.", "ns2.seer.test."]).await;

        let report = checker(
            recursive,
            HashMap::from([
                ("a.parent.test".to_string(), parent),
                ("ns1.seer.test".to_string(), ns1),
            ]),
        )
        .check("seer.test")
        .await
        .expect("check must succeed");

        assert!(report.lame.is_empty(), "skip must not be lame: {report:?}");
        assert!(
            report.warnings.iter().any(|w| w.contains("ns2.seer.test")),
            "warnings should mention the skipped host: {:?}",
            report.warnings
        );
        assert!(report.in_sync, "sets match and nothing is lame: {report:?}");
    }

    #[tokio::test]
    async fn no_responsive_parent_server_is_an_error() {
        // The parent host resolves but its server never answers.
        let recursive = spawn_recursive_mock(&[]).await;
        let parent = spawn_mock_dns(MockMode::Ignore).await;

        let err = checker(
            recursive,
            HashMap::from([("a.parent.test".to_string(), parent)]),
        )
        .check("seer.test")
        .await
        .expect_err("no parent answer must be an error");
        assert!(
            matches!(err, SeerError::DnsError(_)),
            "expected DnsError, got: {err:?}"
        );
        assert!(err.to_string().contains("parent"), "got: {err}");
    }

    /// Regression: each server was probed on one address only, so a server
    /// whose first (IPv4) address this host cannot route to — an IPv6-only
    /// host — was never reached. The next address is tried instead.
    #[tokio::test]
    async fn unroutable_address_falls_back_to_the_next() {
        let unroutable: IpAddr = Ipv4Addr::new(127, 0, 0, 2).into();
        let recursive = spawn_mock_dns_fn(move |qname, qtype| match (qname, qtype) {
            ("test", HickoryRecordType::NS) => MockReply::Answer(vec![ns_rdata("a.parent.test.")]),
            (_, HickoryRecordType::A) => MockReply::Answer(vec![
                HickoryRData::A(wire::A(Ipv4Addr::new(127, 0, 0, 2))),
                loopback_a(),
            ]),
            _ => MockReply::NoData,
        })
        .await;
        let parent = spawn_parent_server(&["ns1.seer.test."]).await;
        let ns1 = spawn_zone_server(&["ns1.seer.test."]).await;
        let ports = HashMap::from([
            ("a.parent.test".to_string(), parent),
            ("ns1.seer.test".to_string(), ns1),
        ]);

        let report = checker(recursive, ports.clone())
            .with_unroutable(&[unroutable])
            .check("seer.test")
            .await
            .expect("check must succeed");
        assert_eq!(report.parent_server_queried, vec!["a.parent.test"]);
        assert_eq!(report.zone_ns, vec!["ns1.seer.test"]);
        assert!(report.in_sync, "{report:?}");

        // With no routable address at all, the server is not reached.
        let err = checker(recursive, ports)
            .with_unroutable(&[unroutable, Ipv4Addr::LOCALHOST.into()])
            .check("seer.test")
            .await
            .expect_err("no parent server reachable");
        assert!(err.to_string().contains("parent"), "{err}");
    }

    /// Regression: the check had no overall deadline, so silent servers
    /// held the caller for up to 17 timeouts. A probe still running at the
    /// deadline is reported as not probed, and the answers that arrived are
    /// kept.
    #[tokio::test]
    async fn the_check_is_held_to_one_deadline() {
        assert_eq!(CHECK_BUDGET_TIMEOUTS, 11);
        let recursive = spawn_recursive_mock(&["ns1.seer.test", "ns2.seer.test"]).await;
        let parent = spawn_parent_server(&["ns1.seer.test.", "ns2.seer.test."]).await;
        let ns1 = spawn_zone_server(&["ns1.seer.test.", "ns2.seer.test."]).await;
        let ns2 = spawn_mock_dns(MockMode::Ignore).await;

        let started = std::time::Instant::now();
        let report = checker(
            recursive,
            HashMap::from([
                ("a.parent.test".to_string(), parent),
                ("ns1.seer.test".to_string(), ns1),
                ("ns2.seer.test".to_string(), ns2),
            ]),
        )
        .with_budget(Duration::from_millis(300))
        .check("seer.test")
        .await
        .expect("the parent answered in time");

        // ns2 would time out after 500ms (twice); the deadline cut it short.
        assert!(started.elapsed() < Duration::from_millis(500));
        assert!(report.lame.is_empty(), "{report:?}");
        assert!(
            report
                .warnings
                .iter()
                .any(|w| w.contains("ns2.seer.test") && w.contains("budget")),
            "{:?}",
            report.warnings
        );
        assert_eq!(report.zone_ns, vec!["ns1.seer.test", "ns2.seer.test"]);
    }

    // --- delegated_ns (lookup cross-check) ---------------------------------

    #[tokio::test]
    async fn delegated_ns_reads_the_parent_referral_without_probing() {
        // No zone servers exist: delegated_ns must not need them.
        let recursive = spawn_recursive_mock(&[]).await;
        let parent = spawn_parent_server(&["NS2.seer.test.", "ns1.seer.test."]).await;
        let delegated = checker(
            recursive,
            HashMap::from([("a.parent.test".to_string(), parent)]),
        )
        .delegated_ns("seer.test")
        .await
        .expect("parent answers the delegation");
        assert_eq!(delegated, vec!["ns1.seer.test", "ns2.seer.test"]);
    }

    #[tokio::test]
    async fn delegated_ns_errors_when_no_parent_server_answers() {
        let recursive = spawn_recursive_mock(&[]).await;
        // a.parent.test resolves, but nothing listens on its mapped port.
        let dead = std::net::UdpSocket::bind("127.0.0.1:0").unwrap();
        let port = dead.local_addr().unwrap().port();
        drop(dead);
        let result = checker(
            recursive,
            HashMap::from([("a.parent.test".to_string(), port)]),
        )
        .delegated_ns("seer.test")
        .await;
        assert!(result.is_err(), "{result:?}");
    }

    #[test]
    fn dns_nameservers_compare_normalizes_and_diffs() {
        // NIC Togo (.tg): WHOIS lists 2 of the 4 delegated hosts.
        let whois = vec!["NS1.nic.tg.".to_string(), "ns2.nic.tg".to_string()];
        let delegated = ["ns1.nic.tg", "ns2.nic.tg", "ns4.nic.tg", "ns5.netmaster.tg"]
            .map(String::from)
            .to_vec();
        let ns = DnsNameservers::compare(&whois, delegated);
        assert!(!ns.in_sync);
        assert_eq!(
            ns.missing_from_whois,
            vec!["ns4.nic.tg", "ns5.netmaster.tg"]
        );
        assert!(ns.missing_from_dns.is_empty());

        let same = DnsNameservers::compare(&whois, vec!["ns2.nic.tg".into(), "ns1.nic.tg".into()]);
        assert!(same.in_sync);
        assert_eq!(same.delegated, vec!["ns1.nic.tg", "ns2.nic.tg"]);
    }
}