cellos_supervisor/resolver_refresh/

mod.rs

//! SEC-21 host-controlled resolver refresh + DNS authority drift detection.
//!
//! This module is the dataplane companion to the W1 T13 contract layer
//! (`spec.authority.dnsAuthority.refreshPolicy`). On each tick it:
//!
//! 1. For every declared hostname, calls an injectable resolver function and
//!    receives a (possibly empty) set of target strings.
//! 2. Compares the answer against the prior observation in [`ResolverState`].
//! 3. When the target set has changed, emits a
//!    `dev.cellos.events.cell.observability.v1.dns_authority_drift` CloudEvent.
//! 4. Honors `policy.minTtlSeconds` (floor — skip refreshes within this window)
//!    and `policy.maxStaleSeconds` (ceiling — force refresh once exceeded).
//! 5. With `strategy: manual`, automatic ticks are skipped — caller must opt
//!    in explicitly.
//!
//! The resolver function is injected via the [`ResolverFn`] type alias so unit
//! tests can run without contacting the network. The supervisor wires the
//! production path to a `to_socket_addrs`-style lookup; see
//! [`crate::supervisor`] for the gating behind `CELLOS_DNS_AUTHORITY_REFRESH=1`.
//!
//! ## Scope
//!
//! - **Phase 1 (W2 SEC-21, shipped):** the supervisor calls
//!   [`ResolverRefresh::tick`] exactly once at startup, before the workload
//!   runs. `tick()` itself stays sync — callers run it from a context where
//!   blocking is acceptable, e.g. inside `tokio::task::spawn_blocking`.
//! - **Phase 2 (this slice, shipped):** the [`ticker`] submodule wraps
//!   `tick()` in a long-running tokio task that fires every
//!   `min(refreshPolicy.minTtlSeconds, 60s).max(5s)` for the lifetime of
//!   the cell. The ticker owns its own [`ResolverState`] (independent of
//!   the startup tick) and dispatches drift events through a sync
//!   [`ticker::DriftEmitter`] adapter — see [`sink_emitter`] for the
//!   sync→async bridge to [`cellos_core::ports::EventSink`].
//! - **Phase 3 (shipped):** real upstream TTL via [`hickory_resolve`]. The
//!   resolver function now returns a [`ResolvedAnswer`] carrying both the
//!   target set AND the minimum TTL the upstream nameserver advertised. Two
//!   downstream effects:
//!
//!     1. The `dns_authority_drift` event's `ttlSeconds` field is now real
//!        (was always `0`); `staleSeconds` becomes meaningful when the
//!        ticker observes drift past the prior TTL.
//!     2. `refreshPolicy.minTtlSeconds` clamps short upstream TTLs as a
//!        DNS-rebinding-mitigation floor (operators can refuse to honour a
//!        TTL=0 fast-flux record by setting a positive floor) — see
//!        [`ticker::TickerConfig`] / [`ResolverRefresh::tick`] for the
//!        clamp logic.
//!
//!   Per-hostname response-IP allowlisting (the *full* rebinding closure)
//!   remains a separate slice; the floor is the partial-mitigation hook.

pub mod dnssec;
pub mod hickory_resolve;
pub mod rebinding;
pub mod sink_emitter;
pub mod ticker;

#[allow(unused_imports)]
pub use dnssec::{TrustAnchors, ENV_TRUST_ANCHORS_PATH, TRUST_ANCHOR_SOURCE_IANA_DEFAULT};
#[allow(unused_imports)]
pub use hickory_resolve::{
    extract_rrsig_metadata, proof_to_validation_result_with_rrsig, resolve_with_ttl,
    resolve_with_ttl_validated, DnssecValidationResult, ResolvedAnswer, ValidatedResolvedAnswer,
};
#[allow(unused_imports)]
pub use rebinding::{RebindingDecision, RebindingState};

use std::collections::{BTreeSet, HashMap};
use std::io;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use sha2::{Digest, Sha256};

use cellos_core::{
    cloud_event_v1_dns_authority_dnssec_failed, cloud_event_v1_dns_authority_drift,
    cloud_event_v1_dns_authority_rebind_rejected, cloud_event_v1_dns_authority_rebind_threshold,
    CloudEventV1, DnsAuthorityDnssecFailed, DnsAuthorityDnssecFailureReason, DnsAuthorityDrift,
    DnsAuthorityRebindRejected, DnsAuthorityRebindThreshold, DnsRebindingPolicy, DnsRefreshPolicy,
    DnsRefreshStrategy, DnsResolver, DnsResolverDnssecPolicy,
};

/// Injectable resolver function.
///
/// scope: returns a [`ResolvedAnswer`] carrying both the target set AND
/// the upstream TTL. Tests pass a closure returning canned answers;
/// production wires this to a `hickory-resolver`-backed call (see
/// [`hickory_resolve::resolve_with_ttl`]).
///
/// An `Err` is treated the same as "no targets observed this tick" — drift
/// events are not emitted for a transient resolver failure, and prior state
/// is preserved.
///
/// **Migration note** (legacy signature): an older signature was
/// `Fn(&str) -> io::Result<Vec<String>>`. Existing call sites that have
/// already canonicalized to `Vec<String>` can adapt with a one-line shim:
/// `Ok(ResolvedAnswer { targets: vec, ttl_seconds: 0, resolver_addr: ... })`.
pub type ResolverFn<'a> = dyn Fn(&str) -> io::Result<ResolvedAnswer> + 'a;

/// SEC-21 Phase 3h — DNSSEC-validating resolver function.
///
/// Returns a [`ValidatedResolvedAnswer`] carrying both the standard
/// [`ResolvedAnswer`] AND the [`DnssecValidationResult`] discriminator
/// (Validated / Failed / Unsigned). When the [`ResolverRefresh`] has a
/// `dnssec_policy` configured, this callback is preferred over the
/// plain [`ResolverFn`]; otherwise it is unused.
///
/// An `Err` is treated identically to the [`ResolverFn`] err path —
/// transient resolver failure, no drift event, prior state preserved.
pub type ValidatedResolverFn<'a> = dyn Fn(&str) -> io::Result<ValidatedResolvedAnswer> + 'a;

/// Per-hostname state tracked across ticks.
#[derive(Debug, Clone, Default)]
struct HostState {
    /// Previous canonicalized target list (sorted, deduped).
    previous_targets: Vec<String>,
    /// Previous canonical digest, or `"empty"` on first observation.
    previous_digest: String,
    /// Wall-clock time of the last *successful* refresh — used to enforce
    /// `min_ttl_seconds` (floor) and `max_stale_seconds` (ceiling).
    last_refresh_at: Option<SystemTime>,
    /// TTL the upstream returned for the last successful answer; `0` when
    /// unknown (`to_socket_addrs` does not surface DNS TTL today).
    last_ttl_seconds: u32,
}

/// Persistent state for [`ResolverRefresh::tick`] — one entry per hostname.
#[derive(Debug, Default)]
pub struct ResolverState {
    hosts: HashMap<String, HostState>,
}

impl ResolverState {
    /// Create an empty resolver state. Callers reuse a single instance across
    /// ticks for a given cell so prior observations persist.
    pub fn new() -> Self {
        Self::default()
    }

    /// Number of hostnames currently tracked. Test affordance.
    pub fn len(&self) -> usize {
        self.hosts.len()
    }

    /// Convenience: is there any tracked state.
    pub fn is_empty(&self) -> bool {
        self.hosts.is_empty()
    }
}

/// One refresh-tick execution.
///
/// Borrows the spec-side configuration (no allocation), runs each declared
/// hostname through the injected resolver, and produces drift CloudEvents the
/// caller publishes via the configured event sink.
pub struct ResolverRefresh<'a> {
    /// Refresh policy from `spec.authority.dnsAuthority.refreshPolicy`.
    /// `None` → defaults: no floor, no ceiling, `ttl-honor` strategy.
    pub policy: Option<&'a DnsRefreshPolicy>,
    /// SEC-21 Phase 3e — DNS rebinding mitigation policy from
    /// `spec.authority.dnsAuthority.rebindingPolicy`. `None` → no
    /// per-hostname response-IP tracking. Combined with the P3a TTL floor
    /// (`policy.min_ttl_seconds`), this structurally closes the v0.4.0
    /// rebinding residual.
    pub rebinding_policy: Option<&'a DnsRebindingPolicy>,
    /// Declared resolvers from `spec.authority.dnsAuthority.resolvers[]`.
    /// The first entry's `resolverId` is used as the event's `resolverId`
    /// when present; the empty string is used when none are declared.
    pub resolvers: &'a [DnsResolver],
    /// Hostnames the supervisor may refresh — typically derived from
    /// `spec.authority.egressRules[].host` ∪
    /// `spec.authority.dnsAuthority.hostnameAllowlist`.
    pub hostnames: &'a [String],
    /// Optional `keysetId` to bind into emitted events.
    pub keyset_id: Option<&'a str>,
    /// Optional `issuerKid` to bind into emitted events.
    pub issuer_kid: Option<&'a str>,
    /// Optional policy-bundle digest (`sha256:<hex>`) to bind into emitted events.
    pub policy_digest: Option<&'a str>,
    /// Optional pass-through correlation id.
    pub correlation_id: Option<&'a str>,
    /// CloudEvent `source` field — defaults to `cellos-supervisor` when the
    /// caller passes `None`.
    pub source: Option<&'a str>,
    /// SEC-21 Phase 3h — opt-in DNSSEC validation policy. `None` (default)
    /// preserves P3a/P3e behaviour exactly: no validation, no
    /// `dns_authority_dnssec_failed` events, no `dnssec_status` tagging on
    /// drift events. When `Some`, the [`Self::tick_with_dnssec`] method
    /// honours `validate` / `failClosed` per resolver — see method docs.
    pub dnssec_policy: Option<&'a DnsResolverDnssecPolicy>,
    /// Trust anchors loaded from the env / spec / IANA-default precedence
    /// chain. Used purely for stamping the source descriptor into emitted
    /// events; hickory 0.24's resolver does not accept custom anchors via
    /// public API. See [`super::dnssec`] module docs for the residual.
    pub trust_anchors: Option<&'a TrustAnchors>,
}

impl<'a> ResolverRefresh<'a> {
    /// Run one refresh tick.
    ///
    /// For each declared hostname:
    ///
    /// - If the strategy is `manual`, skip (operator must opt in).
    /// - If `min_ttl_seconds` says we refreshed recently, skip — *unless*
    ///   `max_stale_seconds` has been exceeded (ceiling overrides floor).
    /// - Call the injected resolver. On error: leave prior state untouched and
    ///   emit no event (transient failures are not drift).
    /// - Canonicalize, hash, compare against prior digest.
    /// - On change, build a [`DnsAuthorityDrift`] payload, wrap it in a
    ///   CloudEvent envelope, and append to the returned vector.
    /// - Update `state` with the new observation.
    ///
    /// Returns the events the caller should publish — possibly empty.
    pub fn tick(
        &self,
        state: &mut ResolverState,
        resolver: &ResolverFn<'_>,
        now: SystemTime,
        cell_id: &str,
        run_id: &str,
    ) -> Vec<CloudEventV1> {
        // Backward-compat: callers that don't track rebinding state pass
        // through to the rebinding-aware variant with a throwaway state.
        // The throwaway is fine because, when `rebinding_policy is None`,
        // the rebinding-aware path emits no extra events and the
        // throwaway state never has anything committed to it.
        let mut throwaway = RebindingState::new();
        self.tick_with_rebinding(state, &mut throwaway, resolver, now, cell_id, run_id)
    }

    /// SEC-21 Phase 3e variant of [`Self::tick`] that also threads a
    /// per-cell [`RebindingState`] for DNS rebinding mitigation.
    ///
    /// Behaves identically to `tick` when `self.rebinding_policy` is
    /// `None`: emits only `dns_authority_drift` events, never touches
    /// `rebinding_state`. When `rebinding_policy` is `Some`:
    ///
    /// 1. Resolves every hostname (same as `tick`).
    /// 2. For each successful resolution, calls
    ///    [`RebindingState::evaluate`] to decide which IPs are novel,
    ///    whether the per-hostname cap is exceeded, and which IPs violate
    ///    the operator allowlist.
    /// 3. Emits one `dns_authority_rebind_threshold` per hostname when
    ///    the cap is exceeded, and one `dns_authority_rebind_rejected`
    ///    per allowlist violation.
    /// 4. Stamps the EFFECTIVE targets (post-rejection when
    ///    `reject_on_rebind=true`) into the subsequent
    ///    `dns_authority_drift` event so the drift digest reflects what
    ///    the workload actually resolves to.
    /// 5. Calls [`RebindingState::commit`] with the effective targets so
    ///    the per-cell history persists across ticks.
    pub fn tick_with_rebinding(
        &self,
        state: &mut ResolverState,
        rebinding_state: &mut RebindingState,
        resolver: &ResolverFn<'_>,
        now: SystemTime,
        cell_id: &str,
        run_id: &str,
    ) -> Vec<CloudEventV1> {
        let mut out: Vec<CloudEventV1> = Vec::new();

        // Manual strategy: caller must opt in explicitly via a different code
        // path (not yet implemented). For Phase 1 we treat manual as "no-op".
        let strategy = self
            .policy
            .and_then(|p| p.strategy)
            .unwrap_or(DnsRefreshStrategy::TtlHonor);
        if matches!(strategy, DnsRefreshStrategy::Manual) {
            return out;
        }

        let min_ttl = self.policy.and_then(|p| p.min_ttl_seconds).unwrap_or(0);
        let max_stale = self.policy.and_then(|p| p.max_stale_seconds);

        let resolver_id_owned: String = self
            .resolvers
            .first()
            .map(|r| r.resolver_id.clone())
            .unwrap_or_default();

        let observed_at_rfc3339 = system_time_to_rfc3339(now);

        for hostname in self.hostnames {
            let prior = state.hosts.get(hostname).cloned().unwrap_or_default();

            // Floor (min_ttl_seconds) — skip refresh when we last looked up
            // this hostname less than `min_ttl_seconds` ago.
            //
            // Ceiling (max_stale_seconds) — *override* the floor: if the prior
            // answer is older than max_stale, refresh regardless.
            if let Some(last) = prior.last_refresh_at {
                let age = now.duration_since(last).unwrap_or_default();
                let age_secs = age.as_secs();
                let floor_active = age_secs < u64::from(min_ttl);
                let ceiling_breached = max_stale.is_some_and(|ms| age_secs >= u64::from(ms));
                if floor_active && !ceiling_breached {
                    // Within the floor and not past the ceiling — skip.
                    continue;
                }
            }

            let answer = match resolver(hostname) {
                Ok(answer) => answer,
                Err(_) => {
                    // Transient resolver failure — do not emit drift, do not
                    // update prior state, do not bump `last_refresh_at`.
                    continue;
                }
            };

            let canonical_targets = canonicalize_targets(&answer.targets);
            let previous_digest = if prior.previous_digest.is_empty() {
                "empty".to_string()
            } else {
                prior.previous_digest.clone()
            };

            // scope: DNS rebinding mitigation. When the operator has
            // declared a `rebindingPolicy`, evaluate the canonical targets
            // against the per-cell rebinding state BEFORE digest / drift
            // emission. The decision's `effective_targets` replaces
            // `canonical_targets` for everything downstream — the digest,
            // the drift event payload, and the persisted state — so the
            // operator-visible `dns_authority_drift` reflects what the
            // workload actually resolves to (and so a `reject_on_rebind`
            // policy actually withholds attacker IPs from the workload's
            // view).
            //
            // When `rebinding_policy is None`, the decision is a no-op:
            // effective_targets = canonical_targets, no events fire.
            let (current_targets, rebind_events) = match self.rebinding_policy {
                Some(rb_policy) => {
                    let decision =
                        rebinding_state.evaluate(hostname, &canonical_targets, rb_policy);
                    let mut events: Vec<CloudEventV1> = Vec::new();

                    if decision.threshold_exceeded {
                        // One threshold event per hostname per tick. We
                        // pick the FIRST novel IP as `novelIp` for SIEM
                        // ergonomics (the cumulative count + max give the
                        // operator everything they need to triage).
                        if let Some(first_novel) = decision.novel_ips.first() {
                            let prior_count = rebinding_state.history(hostname).len() as u32;
                            let cumulative =
                                prior_count.saturating_add(decision.novel_ips.len() as u32);
                            let payload = DnsAuthorityRebindThreshold {
                                schema_version: "1.0.0".into(),
                                cell_id: cell_id.to_string(),
                                run_id: run_id.to_string(),
                                hostname: hostname.clone(),
                                novel_ip: (*first_novel).to_string(),
                                previous_ip_count: prior_count,
                                cumulative_ip_count: cumulative,
                                max_novel_ips_per_hostname: rb_policy.max_novel_ips_per_hostname,
                                policy_digest: self
                                    .policy_digest
                                    .map(str::to_string)
                                    .unwrap_or_else(empty_policy_digest),
                                keyset_id: self.keyset_id.map(str::to_string),
                                issuer_kid: self.issuer_kid.map(str::to_string),
                                correlation_id: self.correlation_id.map(str::to_string),
                                resolver_id: if resolver_id_owned.is_empty() {
                                    None
                                } else {
                                    Some(resolver_id_owned.clone())
                                },
                                observed_at: observed_at_rfc3339.clone(),
                            };
                            let source = self.source.unwrap_or("cellos-supervisor");
                            if let Ok(ev) = cloud_event_v1_dns_authority_rebind_threshold(
                                source,
                                &observed_at_rfc3339,
                                &payload,
                            ) {
                                events.push(ev);
                            }
                        }
                    }

                    if !decision.allowlist_violations.is_empty() {
                        // One rejected event per offending IP. We pre-build
                        // the per-hostname allowlist echo once.
                        let echo: Vec<String> = rb_policy
                            .response_ip_allowlist
                            .iter()
                            .filter(|raw| {
                                raw.split_once(':')
                                    .is_some_and(|(prefix, _)| prefix == hostname)
                            })
                            .cloned()
                            .collect();
                        let prior_count = rebinding_state.history(hostname).len() as u32;
                        let cumulative =
                            prior_count.saturating_add(decision.novel_ips.len() as u32);
                        for &offending in &decision.allowlist_violations {
                            let payload = DnsAuthorityRebindRejected {
                                schema_version: "1.0.0".into(),
                                cell_id: cell_id.to_string(),
                                run_id: run_id.to_string(),
                                hostname: hostname.clone(),
                                novel_ip: offending.to_string(),
                                previous_ip_count: prior_count,
                                cumulative_ip_count: cumulative,
                                response_ip_allowlist: echo.clone(),
                                policy_digest: self
                                    .policy_digest
                                    .map(str::to_string)
                                    .unwrap_or_else(empty_policy_digest),
                                keyset_id: self.keyset_id.map(str::to_string),
                                issuer_kid: self.issuer_kid.map(str::to_string),
                                correlation_id: self.correlation_id.map(str::to_string),
                                resolver_id: if resolver_id_owned.is_empty() {
                                    None
                                } else {
                                    Some(resolver_id_owned.clone())
                                },
                                observed_at: observed_at_rfc3339.clone(),
                            };
                            let source = self.source.unwrap_or("cellos-supervisor");
                            if let Ok(ev) = cloud_event_v1_dns_authority_rebind_rejected(
                                source,
                                &observed_at_rfc3339,
                                &payload,
                            ) {
                                events.push(ev);
                            }
                        }
                    }

                    // Use the effective (post-rejection) targets for the
                    // drift event + state commit.
                    let effective = canonicalize_targets(&decision.effective_targets);
                    (effective, events)
                }
                None => (canonical_targets, Vec::new()),
            };

            let current_digest = digest_target_set(&current_targets);

            // Emit the rebinding events BEFORE the drift event so a SIEM
            // sees them in causal order ("threshold/rejected → drift").
            for ev in rebind_events {
                out.push(ev);
            }

            // scope: DNS-rebinding-mitigation floor. `refreshPolicy.minTtlSeconds`
            // doubles as a *clamp* on the upstream TTL we record: an operator
            // who has explicitly set a positive floor refuses to honour
            // sub-floor TTLs (typical fast-flux indicator). RFC 1035 §3.2.1
            // permits resolvers to clamp; the cellos floor is conservative
            // (it only raises sub-floor TTLs, never lowers above-floor ones).
            // `min_ttl == 0` is the "no floor" sentinel — pass through verbatim.
            let clamped_ttl: u32 = if min_ttl > 0 && answer.ttl_seconds < min_ttl {
                min_ttl
            } else {
                answer.ttl_seconds
            };

            // Stale window observed: how long was the prior answer served past
            // its TTL before this refresh fired? Phase 3: now uses the *real*
            // prior TTL (was always 0 in Phase 1, making this trivially 0).
            let stale_seconds: u32 = match (prior.last_refresh_at, prior.last_ttl_seconds) {
                (Some(last), ttl) if ttl > 0 => {
                    let age = now.duration_since(last).unwrap_or_default().as_secs();
                    age.saturating_sub(u64::from(ttl)).min(u32::MAX as u64) as u32
                }
                _ => 0,
            };

            // Only emit a drift event when the digest actually changed.
            if current_digest != previous_digest {
                let prev_set: BTreeSet<&String> = prior.previous_targets.iter().collect();
                let curr_set: BTreeSet<&String> = current_targets.iter().collect();
                let added: Vec<String> = curr_set
                    .difference(&prev_set)
                    .map(|s| (*s).clone())
                    .collect();
                let removed: Vec<String> = prev_set
                    .difference(&curr_set)
                    .map(|s| (*s).clone())
                    .collect();

                let drift = DnsAuthorityDrift {
                    schema_version: "1.0.0".into(),
                    cell_id: cell_id.to_string(),
                    run_id: run_id.to_string(),
                    resolver_id: resolver_id_owned.clone(),
                    hostname: hostname.clone(),
                    previous_targets: prior.previous_targets.clone(),
                    current_targets: current_targets.clone(),
                    added_targets: added,
                    removed_targets: removed,
                    previous_digest,
                    current_digest: current_digest.clone(),
                    // scope: real upstream TTL via hickory-resolver, clamped
                    // to `refreshPolicy.minTtlSeconds` when the operator has
                    // declared a floor (DNS-rebinding fast-flux mitigation).
                    ttl_seconds: clamped_ttl,
                    stale_seconds,
                    keyset_id: self.keyset_id.map(str::to_string),
                    issuer_kid: self.issuer_kid.map(str::to_string),
                    policy_digest: self.policy_digest.map(str::to_string),
                    correlation_id: self.correlation_id.map(str::to_string),
                    // SEC-21 Phase 3h — `tick_with_rebinding` (the
                    // pre-DNSSEC path) leaves dnssec_status as `None`
                    // so the schema treats the field as omitted. The
                    // dnssec-aware sibling `tick_with_dnssec` populates
                    // it with one of the four enum literals.
                    dnssec_status: None,
                    observed_at: observed_at_rfc3339.clone(),
                };

                let source = self.source.unwrap_or("cellos-supervisor");
                match cloud_event_v1_dns_authority_drift(source, &observed_at_rfc3339, &drift) {
                    Ok(ev) => out.push(ev),
                    Err(e) => {
                        // Emit-side serialization failure is reported via tracing
                        // by the caller; the resolver-refresh module is pure
                        // data + has no logger. Drop the event silently here.
                        let _ = e;
                    }
                }
            }

            // Always update state on a successful resolution — even when the
            // digest matched — so `last_refresh_at` reflects the latest probe.
            // We persist the *clamped* TTL so the next stale-window calc uses
            // the floor-honouring value, matching the event we just emitted.
            state.hosts.insert(
                hostname.clone(),
                HostState {
                    previous_targets: current_targets.clone(),
                    previous_digest: current_digest,
                    last_refresh_at: Some(now),
                    last_ttl_seconds: clamped_ttl,
                },
            );

            // scope: commit the per-hostname rebinding observation.
            // Takes the EFFECTIVE targets (post-rejection in enforce mode)
            // so the cumulative history reflects what the workload saw.
            // No-op when `rebinding_policy is None`.
            if self.rebinding_policy.is_some() {
                rebinding_state.commit(hostname, &current_targets);
            }
        }

        out
    }
}

/// SEC-21 Phase 3h — `dnssec_status` enum literals for the
/// `dns_authority_drift` event payload. Centralized so the ticker tests
/// can match on the exact string the schema expects without re-typing it.
pub const DNSSEC_STATUS_VALIDATED: &str = "validated";
pub const DNSSEC_STATUS_VALIDATION_FAILED: &str = "validation_failed";
pub const DNSSEC_STATUS_UNSIGNED: &str = "unsigned";
pub const DNSSEC_STATUS_NOT_ATTEMPTED: &str = "not_attempted";

impl<'a> ResolverRefresh<'a> {
    /// SEC-21 Phase 3h variant of [`Self::tick_with_rebinding`] that
    /// also processes DNSSEC validation outcomes per hostname.
    ///
    /// Behaviour:
    ///
    /// - When `self.dnssec_policy` is `None`, this method behaves exactly
    ///   like [`Self::tick_with_rebinding`] except it pulls per-hostname
    ///   answers from `validated_resolved` and stamps
    ///   `dnssec_status: not_attempted` on every emitted drift event.
    /// - When `self.dnssec_policy` is `Some(policy)`:
    ///     - `Validated` → `dnssec_status: validated`, normal flow.
    ///     - `Failed{reason}` → emit `dns_authority_dnssec_failed`
    ///       (reason: `validation_failed`); when `policy.fail_closed`,
    ///       drop `answer.targets` to empty BEFORE the rebinding
    ///       evaluator runs and tag drift `dnssec_status: validation_failed`.
    ///     - `Unsigned` → emit `dns_authority_dnssec_failed`
    ///       (reason: `unsigned_zone`); same `fail_closed` semantics.
    ///
    /// Order: dnssec_failed events are emitted BEFORE the drift /
    /// rebinding events for that hostname so a SIEM sees them in
    /// causal order ("dnssec_failed → drift").
    pub fn tick_with_dnssec(
        &self,
        state: &mut ResolverState,
        rebinding_state: &mut RebindingState,
        validated_resolved: &std::collections::HashMap<String, io::Result<ValidatedResolvedAnswer>>,
        now: SystemTime,
        cell_id: &str,
        run_id: &str,
    ) -> Vec<CloudEventV1> {
        let mut out: Vec<CloudEventV1> = Vec::new();

        let strategy = self
            .policy
            .and_then(|p| p.strategy)
            .unwrap_or(DnsRefreshStrategy::TtlHonor);
        if matches!(strategy, DnsRefreshStrategy::Manual) {
            return out;
        }

        let min_ttl = self.policy.and_then(|p| p.min_ttl_seconds).unwrap_or(0);
        let max_stale = self.policy.and_then(|p| p.max_stale_seconds);

        let resolver_id_owned: String = self
            .resolvers
            .first()
            .map(|r| r.resolver_id.clone())
            .unwrap_or_default();
        let observed_at_rfc3339 = system_time_to_rfc3339(now);
        let trust_anchor_source: String = self
            .trust_anchors
            .map(|ta| ta.source.clone())
            .unwrap_or_else(|| dnssec::TRUST_ANCHOR_SOURCE_IANA_DEFAULT.to_string());
        let policy_active = self.dnssec_policy.is_some();
        let fail_closed = self.dnssec_policy.map(|p| p.fail_closed).unwrap_or(false);

        for hostname in self.hostnames {
            let prior = state.hosts.get(hostname).cloned().unwrap_or_default();

            // Floor / ceiling — same predicate as the non-DNSSEC path.
            if let Some(last) = prior.last_refresh_at {
                let age = now.duration_since(last).unwrap_or_default();
                let age_secs = age.as_secs();
                let floor_active = age_secs < u64::from(min_ttl);
                let ceiling_breached = max_stale.is_some_and(|ms| age_secs >= u64::from(ms));
                if floor_active && !ceiling_breached {
                    continue;
                }
            }

            let validated = match validated_resolved.get(hostname) {
                Some(Ok(v)) => v.clone(),
                Some(Err(_)) | None => {
                    // Transient resolver failure or missing entry — same
                    // semantics as the unvalidated path: no drift, no
                    // state mutation, no DNSSEC event (can't tell what
                    // went wrong).
                    continue;
                }
            };

            // Decide the dnssec_status string + per-hostname effective
            // answer (drop on fail_closed) + whether to emit a
            // dns_authority_dnssec_failed event.
            let (dnssec_status, dnssec_event, effective_answer): (
                &'static str,
                Option<CloudEventV1>,
                ResolvedAnswer,
            ) = match (&validated.validation, policy_active) {
                (DnssecValidationResult::Validated { .. }, true) => {
                    (DNSSEC_STATUS_VALIDATED, None, validated.answer.clone())
                }
                (DnssecValidationResult::Failed { reason }, true) => {
                    let payload = DnsAuthorityDnssecFailed {
                        schema_version: "1.0.0".into(),
                        cell_id: cell_id.to_string(),
                        run_id: run_id.to_string(),
                        resolver_id: resolver_id_owned.clone(),
                        hostname: hostname.clone(),
                        reason: DnsAuthorityDnssecFailureReason::ValidationFailed
                            .as_str()
                            .to_string(),
                        fail_closed,
                        trust_anchor_source: trust_anchor_source.clone(),
                        policy_digest: self.policy_digest.map(str::to_string),
                        keyset_id: self.keyset_id.map(str::to_string),
                        issuer_kid: self.issuer_kid.map(str::to_string),
                        correlation_id: self.correlation_id.map(str::to_string),
                        // SEC-21 Phase 3h.1 — additive `source` field. The
                        // resolver-refresh path always stamps this literal.
                        // The dataplane (in-netns DNS proxy) emits the same
                        // event with `source = "dataplane"` from
                        // `dns_proxy::dnssec`.
                        source: Some("resolver_refresh".into()),
                        observed_at: observed_at_rfc3339.clone(),
                    };
                    let _ = reason; // reason is logged by the caller via tracing if needed
                    let source = self.source.unwrap_or("cellos-supervisor");
                    let event = cloud_event_v1_dns_authority_dnssec_failed(
                        source,
                        &observed_at_rfc3339,
                        &payload,
                    )
                    .ok();
                    let answer = if fail_closed {
                        ResolvedAnswer {
                            targets: Vec::new(),
                            ttl_seconds: validated.answer.ttl_seconds,
                            resolver_addr: validated.answer.resolver_addr,
                        }
                    } else {
                        validated.answer.clone()
                    };
                    (DNSSEC_STATUS_VALIDATION_FAILED, event, answer)
                }
                (DnssecValidationResult::Unsigned, true) => {
                    let payload = DnsAuthorityDnssecFailed {
                        schema_version: "1.0.0".into(),
                        cell_id: cell_id.to_string(),
                        run_id: run_id.to_string(),
                        resolver_id: resolver_id_owned.clone(),
                        hostname: hostname.clone(),
                        reason: DnsAuthorityDnssecFailureReason::UnsignedZone
                            .as_str()
                            .to_string(),
                        fail_closed,
                        trust_anchor_source: trust_anchor_source.clone(),
                        policy_digest: self.policy_digest.map(str::to_string),
                        keyset_id: self.keyset_id.map(str::to_string),
                        issuer_kid: self.issuer_kid.map(str::to_string),
                        correlation_id: self.correlation_id.map(str::to_string),
                        // SEC-21 Phase 3h.1 — additive `source` field; see
                        // matching comment on the Failed-arm above.
                        source: Some("resolver_refresh".into()),
                        observed_at: observed_at_rfc3339.clone(),
                    };
                    let source = self.source.unwrap_or("cellos-supervisor");
                    let event = cloud_event_v1_dns_authority_dnssec_failed(
                        source,
                        &observed_at_rfc3339,
                        &payload,
                    )
                    .ok();
                    let answer = if fail_closed {
                        ResolvedAnswer {
                            targets: Vec::new(),
                            ttl_seconds: validated.answer.ttl_seconds,
                            resolver_addr: validated.answer.resolver_addr,
                        }
                    } else {
                        validated.answer.clone()
                    };
                    (DNSSEC_STATUS_UNSIGNED, event, answer)
                }
                // policy off → status:not_attempted, no event, pass-through answer.
                (_, false) => (DNSSEC_STATUS_NOT_ATTEMPTED, None, validated.answer.clone()),
            };

            // Emit dnssec_failed BEFORE the drift event so SIEM sees
            // causal order. When `dnssec_event` is None this is a no-op.
            if let Some(ev) = dnssec_event {
                out.push(ev);
            }

            // From here, the path mirrors `tick_with_rebinding` exactly,
            // operating on `effective_answer` (the post-DNSSEC-policy
            // ResolvedAnswer).
            let canonical_targets = canonicalize_targets(&effective_answer.targets);
            let previous_digest = if prior.previous_digest.is_empty() {
                "empty".to_string()
            } else {
                prior.previous_digest.clone()
            };

            let (current_targets, rebind_events) = match self.rebinding_policy {
                Some(rb_policy) => {
                    let decision =
                        rebinding_state.evaluate(hostname, &canonical_targets, rb_policy);
                    let mut events: Vec<CloudEventV1> = Vec::new();
                    if decision.threshold_exceeded {
                        if let Some(first_novel) = decision.novel_ips.first() {
                            let prior_count = rebinding_state.history(hostname).len() as u32;
                            let cumulative =
                                prior_count.saturating_add(decision.novel_ips.len() as u32);
                            let payload = DnsAuthorityRebindThreshold {
                                schema_version: "1.0.0".into(),
                                cell_id: cell_id.to_string(),
                                run_id: run_id.to_string(),
                                hostname: hostname.clone(),
                                novel_ip: (*first_novel).to_string(),
                                previous_ip_count: prior_count,
                                cumulative_ip_count: cumulative,
                                max_novel_ips_per_hostname: rb_policy.max_novel_ips_per_hostname,
                                policy_digest: self
                                    .policy_digest
                                    .map(str::to_string)
                                    .unwrap_or_else(empty_policy_digest),
                                keyset_id: self.keyset_id.map(str::to_string),
                                issuer_kid: self.issuer_kid.map(str::to_string),
                                correlation_id: self.correlation_id.map(str::to_string),
                                resolver_id: if resolver_id_owned.is_empty() {
                                    None
                                } else {
                                    Some(resolver_id_owned.clone())
                                },
                                observed_at: observed_at_rfc3339.clone(),
                            };
                            let source = self.source.unwrap_or("cellos-supervisor");
                            if let Ok(ev) = cloud_event_v1_dns_authority_rebind_threshold(
                                source,
                                &observed_at_rfc3339,
                                &payload,
                            ) {
                                events.push(ev);
                            }
                        }
                    }
                    if !decision.allowlist_violations.is_empty() {
                        let echo: Vec<String> = rb_policy
                            .response_ip_allowlist
                            .iter()
                            .filter(|raw| {
                                raw.split_once(':')
                                    .is_some_and(|(prefix, _)| prefix == hostname)
                            })
                            .cloned()
                            .collect();
                        let prior_count = rebinding_state.history(hostname).len() as u32;
                        let cumulative =
                            prior_count.saturating_add(decision.novel_ips.len() as u32);
                        for &offending in &decision.allowlist_violations {
                            let payload = DnsAuthorityRebindRejected {
                                schema_version: "1.0.0".into(),
                                cell_id: cell_id.to_string(),
                                run_id: run_id.to_string(),
                                hostname: hostname.clone(),
                                novel_ip: offending.to_string(),
                                previous_ip_count: prior_count,
                                cumulative_ip_count: cumulative,
                                response_ip_allowlist: echo.clone(),
                                policy_digest: self
                                    .policy_digest
                                    .map(str::to_string)
                                    .unwrap_or_else(empty_policy_digest),
                                keyset_id: self.keyset_id.map(str::to_string),
                                issuer_kid: self.issuer_kid.map(str::to_string),
                                correlation_id: self.correlation_id.map(str::to_string),
                                resolver_id: if resolver_id_owned.is_empty() {
                                    None
                                } else {
                                    Some(resolver_id_owned.clone())
                                },
                                observed_at: observed_at_rfc3339.clone(),
                            };
                            let source = self.source.unwrap_or("cellos-supervisor");
                            if let Ok(ev) = cloud_event_v1_dns_authority_rebind_rejected(
                                source,
                                &observed_at_rfc3339,
                                &payload,
                            ) {
                                events.push(ev);
                            }
                        }
                    }
                    let effective = canonicalize_targets(&decision.effective_targets);
                    (effective, events)
                }
                None => (canonical_targets, Vec::new()),
            };

            let current_digest = digest_target_set(&current_targets);

            for ev in rebind_events {
                out.push(ev);
            }

            let clamped_ttl: u32 = if min_ttl > 0 && effective_answer.ttl_seconds < min_ttl {
                min_ttl
            } else {
                effective_answer.ttl_seconds
            };

            let stale_seconds: u32 = match (prior.last_refresh_at, prior.last_ttl_seconds) {
                (Some(last), ttl) if ttl > 0 => {
                    let age = now.duration_since(last).unwrap_or_default().as_secs();
                    age.saturating_sub(u64::from(ttl)).min(u32::MAX as u64) as u32
                }
                _ => 0,
            };

            if current_digest != previous_digest {
                let prev_set: BTreeSet<&String> = prior.previous_targets.iter().collect();
                let curr_set: BTreeSet<&String> = current_targets.iter().collect();
                let added: Vec<String> = curr_set
                    .difference(&prev_set)
                    .map(|s| (*s).clone())
                    .collect();
                let removed: Vec<String> = prev_set
                    .difference(&curr_set)
                    .map(|s| (*s).clone())
                    .collect();

                let drift = DnsAuthorityDrift {
                    schema_version: "1.0.0".into(),
                    cell_id: cell_id.to_string(),
                    run_id: run_id.to_string(),
                    resolver_id: resolver_id_owned.clone(),
                    hostname: hostname.clone(),
                    previous_targets: prior.previous_targets.clone(),
                    current_targets: current_targets.clone(),
                    added_targets: added,
                    removed_targets: removed,
                    previous_digest,
                    current_digest: current_digest.clone(),
                    ttl_seconds: clamped_ttl,
                    stale_seconds,
                    keyset_id: self.keyset_id.map(str::to_string),
                    issuer_kid: self.issuer_kid.map(str::to_string),
                    policy_digest: self.policy_digest.map(str::to_string),
                    correlation_id: self.correlation_id.map(str::to_string),
                    dnssec_status: Some(dnssec_status.to_string()),
                    observed_at: observed_at_rfc3339.clone(),
                };

                let source = self.source.unwrap_or("cellos-supervisor");
                if let Ok(ev) =
                    cloud_event_v1_dns_authority_drift(source, &observed_at_rfc3339, &drift)
                {
                    out.push(ev);
                }
            }

            state.hosts.insert(
                hostname.clone(),
                HostState {
                    previous_targets: current_targets.clone(),
                    previous_digest: current_digest,
                    last_refresh_at: Some(now),
                    last_ttl_seconds: clamped_ttl,
                },
            );

            if self.rebinding_policy.is_some() {
                rebinding_state.commit(hostname, &current_targets);
            }
        }

        out
    }
}

/// Sentinel `policyDigest` used in the SEC-21 Phase 3e rebinding events
/// when the caller didn't supply one. The schema requires a digest, so we
/// stamp the deterministic empty-string digest rather than failing the
/// emission. Operators who want the real digest must wire
/// `policy_digest` into [`ResolverRefresh`] (every production call site
/// already does — this fallback only fires in unit tests that omit it).
fn empty_policy_digest() -> String {
    // sha256("") — `e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855`
    "sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855".to_string()
}

/// Canonicalize a target set: trim whitespace, drop empties, dedupe, sort.
fn canonicalize_targets(raw: &[String]) -> Vec<String> {
    let mut set: BTreeSet<String> = BTreeSet::new();
    for t in raw {
        let trimmed = t.trim();
        if !trimmed.is_empty() {
            set.insert(trimmed.to_string());
        }
    }
    set.into_iter().collect()
}

/// Compute `sha256:<hex>` over the canonicalized target list joined with `\n`.
/// `[]` (empty set) hashes deterministically — every empty set produces the
/// same digest.
fn digest_target_set(canonical: &[String]) -> String {
    let mut hasher = Sha256::new();
    for (i, t) in canonical.iter().enumerate() {
        if i > 0 {
            hasher.update(b"\n");
        }
        hasher.update(t.as_bytes());
    }
    let bytes = hasher.finalize();
    let mut out = String::with_capacity(7 + 64);
    out.push_str("sha256:");
    for b in bytes {
        use std::fmt::Write;
        let _ = write!(out, "{b:02x}");
    }
    out
}

/// Convert a `SystemTime` to an RFC3339 string. Avoids pulling chrono into the
/// pure-data section of the module — the supervisor passes its own
/// chrono-based timestamp through `now` for production paths.
fn system_time_to_rfc3339(t: SystemTime) -> String {
    let secs = t
        .duration_since(UNIX_EPOCH)
        .unwrap_or(Duration::ZERO)
        .as_secs() as i64;
    let dt =
        chrono::DateTime::<chrono::Utc>::from_timestamp(secs, 0).unwrap_or_else(chrono::Utc::now);
    dt.to_rfc3339()
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::cell::RefCell;
    use std::net::{IpAddr, Ipv4Addr, SocketAddr};
    use std::time::Duration;

    /// Test helper — wrap a `Vec<String>` answer in the Phase 3 [`ResolvedAnswer`]
    /// shape. The pre-Phase-3 tests didn't care about TTL, so we default to 0
    /// here and let the Phase 3 ticker tests build their own answers with
    /// real TTLs.
    fn answer_zero_ttl(targets: Vec<String>) -> ResolvedAnswer {
        ResolvedAnswer {
            targets,
            ttl_seconds: 0,
            resolver_addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 53),
        }
    }

    fn fixed_resolvers() -> Vec<DnsResolver> {
        vec![DnsResolver {
            resolver_id: "resolver-test-001".into(),
            endpoint: "1.1.1.1:53".into(),
            protocol: cellos_core::DnsResolverProtocol::Do53Udp,
            trust_kid: None,
            dnssec: None,
        }]
    }

    fn make<'a>(
        policy: Option<&'a DnsRefreshPolicy>,
        resolvers: &'a [DnsResolver],
        hostnames: &'a [String],
    ) -> ResolverRefresh<'a> {
        ResolverRefresh {
            policy,
            rebinding_policy: None,
            resolvers,
            hostnames,
            keyset_id: Some("keyset-test-001"),
            issuer_kid: Some("kid-test-001"),
            policy_digest: None,
            correlation_id: None,
            source: Some("cellos-supervisor-test"),
            // SEC-21 Phase 3h test default — DNSSEC off so the existing
            // P3a tests (~17 of them, untouched) keep their original
            // semantics. The DNSSEC-specific tests below build their
            // own ResolverRefresh with the policy populated.
            dnssec_policy: None,
            trust_anchors: None,
        }
    }

    #[test]
    fn canonicalize_dedupes_and_sorts() {
        let raw: Vec<String> = vec!["1.0.0.1".into(), "1.1.1.1".into(), "1.0.0.1".into()];
        let canon = canonicalize_targets(&raw);
        assert_eq!(canon, vec!["1.0.0.1".to_string(), "1.1.1.1".to_string()]);
    }

    #[test]
    fn digest_changes_when_targets_change() {
        let a = canonicalize_targets(&["1.1.1.1".to_string()]);
        let b = canonicalize_targets(&["1.0.0.1".to_string()]);
        assert_ne!(digest_target_set(&a), digest_target_set(&b));
    }

    #[test]
    fn first_observation_emits_drift_with_empty_previous() {
        let hostnames = vec!["api.example.com".to_string()];
        let resolvers = fixed_resolvers();
        let refresher = make(None, &resolvers, &hostnames);
        let mut state = ResolverState::new();
        let resolver = |_h: &str| Ok(answer_zero_ttl(vec!["1.1.1.1".to_string()]));
        let events = refresher.tick(
            &mut state,
            &resolver,
            SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000),
            "cell-001",
            "run-001",
        );
        assert_eq!(events.len(), 1, "first observation must emit drift");
        let data = events[0].data.as_ref().expect("data present");
        assert_eq!(data["previousDigest"], "empty");
        assert_eq!(data["addedTargets"], serde_json::json!(["1.1.1.1"]));
        assert!(data["removedTargets"].as_array().unwrap().is_empty());
    }

    #[test]
    fn stable_targets_emit_no_drift_after_first_observation() {
        let hostnames = vec!["api.example.com".to_string()];
        let resolvers = fixed_resolvers();
        let policy = DnsRefreshPolicy {
            min_ttl_seconds: Some(0),
            max_stale_seconds: None,
            strategy: None,
        };
        let refresher = make(Some(&policy), &resolvers, &hostnames);
        let mut state = ResolverState::new();
        let resolver = |_h: &str| Ok(answer_zero_ttl(vec!["1.1.1.1".to_string()]));

        let now = SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000);
        let _ = refresher.tick(&mut state, &resolver, now, "c", "r");
        let events = refresher.tick(
            &mut state,
            &resolver,
            now + Duration::from_secs(60),
            "c",
            "r",
        );
        assert!(events.is_empty(), "stable targets must not emit drift");
    }

    #[test]
    fn manual_strategy_emits_nothing() {
        let hostnames = vec!["api.example.com".to_string()];
        let resolvers = fixed_resolvers();
        let policy = DnsRefreshPolicy {
            min_ttl_seconds: None,
            max_stale_seconds: None,
            strategy: Some(DnsRefreshStrategy::Manual),
        };
        let refresher = make(Some(&policy), &resolvers, &hostnames);
        let mut state = ResolverState::new();
        let resolver = |_h: &str| Ok(answer_zero_ttl(vec!["1.1.1.1".to_string()]));
        let events = refresher.tick(
            &mut state,
            &resolver,
            SystemTime::UNIX_EPOCH + Duration::from_secs(1),
            "c",
            "r",
        );
        assert!(events.is_empty(), "manual strategy must skip ticks");
        assert!(state.is_empty(), "state must remain untouched");
    }

    #[test]
    fn floor_skips_refresh_within_min_ttl() {
        let hostnames = vec!["api.example.com".to_string()];
        let resolvers = fixed_resolvers();
        let policy = DnsRefreshPolicy {
            min_ttl_seconds: Some(300),
            max_stale_seconds: None,
            strategy: None,
        };
        let refresher = make(Some(&policy), &resolvers, &hostnames);
        let mut state = ResolverState::new();
        let calls: RefCell<u32> = RefCell::new(0);
        let resolver = |_h: &str| -> io::Result<ResolvedAnswer> {
            *calls.borrow_mut() += 1;
            Ok(answer_zero_ttl(vec!["1.1.1.1".to_string()]))
        };

        let t0 = SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000);
        let _ = refresher.tick(&mut state, &resolver, t0, "c", "r");
        // 60s later — well within min_ttl_seconds (300) — must skip.
        let _ = refresher.tick(
            &mut state,
            &resolver,
            t0 + Duration::from_secs(60),
            "c",
            "r",
        );
        assert_eq!(
            *calls.borrow(),
            1,
            "floor must skip the second resolver call"
        );
    }

    #[test]
    fn ceiling_forces_refresh_after_max_stale() {
        let hostnames = vec!["api.example.com".to_string()];
        let resolvers = fixed_resolvers();
        let policy = DnsRefreshPolicy {
            min_ttl_seconds: Some(3_000),
            max_stale_seconds: Some(60),
            strategy: None,
        };
        let refresher = make(Some(&policy), &resolvers, &hostnames);
        let mut state = ResolverState::new();
        let calls: RefCell<u32> = RefCell::new(0);
        let resolver = |_h: &str| -> io::Result<ResolvedAnswer> {
            *calls.borrow_mut() += 1;
            Ok(answer_zero_ttl(vec!["1.1.1.1".to_string()]))
        };

        let t0 = SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000);
        let _ = refresher.tick(&mut state, &resolver, t0, "c", "r");
        // 120s later — past max_stale (60s) — ceiling must override the floor.
        let _ = refresher.tick(
            &mut state,
            &resolver,
            t0 + Duration::from_secs(120),
            "c",
            "r",
        );
        assert_eq!(
            *calls.borrow(),
            2,
            "ceiling must override the floor and force a refresh"
        );
    }

    #[test]
    fn resolver_failure_does_not_emit_drift() {
        let hostnames = vec!["api.example.com".to_string()];
        let resolvers = fixed_resolvers();
        let refresher = make(None, &resolvers, &hostnames);
        let mut state = ResolverState::new();
        let resolver =
            |_h: &str| -> io::Result<ResolvedAnswer> { Err(io::Error::other("transient")) };
        let events = refresher.tick(
            &mut state,
            &resolver,
            SystemTime::UNIX_EPOCH + Duration::from_secs(1),
            "c",
            "r",
        );
        assert!(
            events.is_empty(),
            "transient resolver error must not be reported as drift"
        );
        assert!(state.is_empty(), "failed lookup must not commit state");
    }

    // ============================================================
    // scope: TTL-clamp tests for `refreshPolicy.minTtlSeconds`.
    // The clamp is the partial DNS-rebinding mitigation: an operator
    // who sets `minTtlSeconds: 60` refuses to honour TTL=0 fast-flux
    // records, which would otherwise force per-millisecond cache
    // invalidation. RFC 1035 §3.2.1 permits the clamp; only the
    // recorded `ttlSeconds` field is affected — the resolver still
    // probes upstream on every tick.
    // ============================================================

    fn answer_with_ttl(targets: Vec<String>, ttl: u32) -> ResolvedAnswer {
        ResolvedAnswer {
            targets,
            ttl_seconds: ttl,
            resolver_addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 53),
        }
    }

    #[test]
    fn ticker_floors_ttl_to_min_ttl_seconds() {
        let hostnames = vec!["api.example.com".to_string()];
        let resolvers = fixed_resolvers();
        // Operator floor at 60s; upstream returns TTL=5 (suspect).
        let policy = DnsRefreshPolicy {
            min_ttl_seconds: Some(60),
            max_stale_seconds: None,
            strategy: None,
        };
        let refresher = make(Some(&policy), &resolvers, &hostnames);
        let mut state = ResolverState::new();
        let resolver = |_h: &str| -> io::Result<ResolvedAnswer> {
            Ok(answer_with_ttl(vec!["1.1.1.1".to_string()], 5))
        };
        let events = refresher.tick(
            &mut state,
            &resolver,
            SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000),
            "c",
            "r",
        );
        assert_eq!(events.len(), 1, "first observation emits drift");
        let data = events[0].data.as_ref().expect("data");
        assert_eq!(
            data["ttlSeconds"], 60,
            "sub-floor TTL must be clamped up to min_ttl_seconds"
        );
    }

    #[test]
    fn ticker_passes_through_ttl_above_floor() {
        let hostnames = vec!["api.example.com".to_string()];
        let resolvers = fixed_resolvers();
        // Operator floor at 60s; upstream returns TTL=300 (above the floor).
        let policy = DnsRefreshPolicy {
            min_ttl_seconds: Some(60),
            max_stale_seconds: None,
            strategy: None,
        };
        let refresher = make(Some(&policy), &resolvers, &hostnames);
        let mut state = ResolverState::new();
        let resolver = |_h: &str| -> io::Result<ResolvedAnswer> {
            Ok(answer_with_ttl(vec!["1.1.1.1".to_string()], 300))
        };
        let events = refresher.tick(
            &mut state,
            &resolver,
            SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000),
            "c",
            "r",
        );
        let data = events[0].data.as_ref().expect("data");
        assert_eq!(
            data["ttlSeconds"], 300,
            "above-floor TTL must pass through verbatim"
        );
    }

    #[test]
    fn ticker_zero_min_ttl_means_no_floor() {
        let hostnames = vec!["api.example.com".to_string()];
        let resolvers = fixed_resolvers();
        // Operator declared `minTtlSeconds: 0` — the "no floor" sentinel.
        // Sub-second TTLs from upstream pass through unmolested so the
        // operator can observe them.
        let policy = DnsRefreshPolicy {
            min_ttl_seconds: Some(0),
            max_stale_seconds: None,
            strategy: None,
        };
        let refresher = make(Some(&policy), &resolvers, &hostnames);
        let mut state = ResolverState::new();
        let resolver = |_h: &str| -> io::Result<ResolvedAnswer> {
            Ok(answer_with_ttl(vec!["1.1.1.1".to_string()], 1))
        };
        let events = refresher.tick(
            &mut state,
            &resolver,
            SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000),
            "c",
            "r",
        );
        let data = events[0].data.as_ref().expect("data");
        assert_eq!(
            data["ttlSeconds"], 1,
            "min_ttl_seconds=0 means no clamp — pass through TTL=1"
        );
    }
}