cellos-supervisor 0.5.1

//! SEC-21 Phase 3e — per-cell DNS rebinding state tracker.
//!
//! Pure data, no I/O. The [`RebindingState`] holds a per-hostname history of
//! every distinct IP ever observed across all refresh ticks for the cell's
//! lifetime. On each tick the ticker calls [`RebindingState::evaluate`] BEFORE
//! emitting the standard `dns_authority_drift` event, gets back a
//! [`RebindingDecision`] describing:
//!
//! 1. Which IPs in the new response are net-new (`novel_ips`).
//! 2. Whether adding them would push the cumulative distinct-IP count above
//!    `policy.max_novel_ips_per_hostname` (`threshold_exceeded`).
//! 3. Which IPs violate the operator-declared allowlist (`allowlist_violations`).
//! 4. The `effective_targets` the workload should see — equal to
//!    `new_targets` in audit-only mode (`reject_on_rebind=false`); filtered to
//!    drop allowlist violations and over-cap novel IPs in enforcement mode
//!    (`reject_on_rebind=true`).
//!
//! The ticker emits `dns_authority_rebind_threshold` and
//! `dns_authority_rebind_rejected` events from the decision, then calls
//! [`RebindingState::commit`] to persist the new observation.
//!
//! Combined with the P3a TTL floor (`refresh_policy.min_ttl_seconds`), the
//! P3e tracker structurally closes the v0.4.0 honest residual "DNS rebinding
//! (TTL=0 / fast-flux) — resolver-side mitigation territory". The TTL floor
//! limits how often responses can change; the per-hostname IP tracker limits
//! what they can change to.
//!
//! ## Allowlist format
//!
//! [`DnsRebindingPolicy::response_ip_allowlist`] entries are
//! `hostname:ip-or-cidr` strings — split on the first `:`, the prefix
//! identifies the hostname, the suffix is either an IPv4/IPv6 literal or a
//! CIDR prefix (e.g. `203.0.113.0/24`). Schema-side validation should
//! pre-filter malformed entries; this module silently skips entries it can't
//! parse so resolver-refresh never crashes on user input.

use std::collections::{HashMap, HashSet};
use std::net::IpAddr;
use std::str::FromStr;

use ipnet::IpNet;

use cellos_core::DnsRebindingPolicy;

/// Per-hostname history of every distinct IP ever observed across all
/// refresh ticks for the cell's lifetime.
///
/// Owned by the [`super::ticker::TickerHandle`] (via the spawned task's
/// internal state) so observations persist across ticks without leaking out
/// of the ticker's lifetime. Reset to empty when a new cell starts.
#[derive(Debug, Default, Clone)]
pub struct RebindingState {
    /// hostname → distinct IPs observed so far, in insertion order.
    histories: HashMap<String, Vec<String>>,
}

/// Result of a single per-hostname [`RebindingState::evaluate`] call.
///
/// Borrows from the input `new_targets` slice for the IP-list fields to
/// avoid an allocation when the caller only needs to iterate; the
/// `effective_targets` field is owned because filtering produces a new
/// `Vec<String>` regardless.
pub struct RebindingDecision<'a> {
    /// IPs from the new response that are net-new (not in prior history),
    /// in the order they appeared in `new_targets`.
    pub novel_ips: Vec<&'a str>,
    /// True when at least one novel IP exists AND
    /// `policy.max_novel_ips_per_hostname` would be exceeded after adding
    /// them. Caller emits a `dns_authority_rebind_threshold` event per
    /// hostname (one per tick, regardless of how many novel IPs).
    pub threshold_exceeded: bool,
    /// IPs that fail `policy.response_ip_allowlist` (when set), in the
    /// order they appeared in `new_targets`. Caller emits one
    /// `dns_authority_rebind_rejected` event per IP. Empty when the
    /// allowlist is empty (allowlist enforcement is opt-in).
    pub allowlist_violations: Vec<&'a str>,
    /// The resolved-target set the workload should see.
    ///
    /// - When `policy.reject_on_rebind == false` (audit-only, the Phase 3e
    ///   default): equal to `new_targets`. Events fire but the IPs remain
    ///   in the workload's resolution.
    /// - When `policy.reject_on_rebind == true`: filtered to drop
    ///   (a) every IP in `allowlist_violations` AND
    ///   (b) novel IPs that push the cumulative count past
    ///   `policy.max_novel_ips_per_hostname`. The "last `len() - max`
    ///   novel IPs" are the ones dropped — we keep the IPs that were
    ///   observed first, on the assumption that legitimate CDN rotation
    ///   looks like a small steady set.
    pub effective_targets: Vec<String>,
}

impl RebindingState {
    /// Create an empty state. Callers reuse a single instance for the
    /// lifetime of the cell so prior observations persist.
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    /// Number of hostnames currently tracked. Test affordance / metrics.
    #[must_use]
    pub fn hostname_count(&self) -> usize {
        self.histories.len()
    }

    /// Distinct-IP history for `hostname`, or empty when no observation has
    /// been committed yet. Test affordance / introspection.
    #[must_use]
    pub fn history(&self, hostname: &str) -> &[String] {
        self.histories
            .get(hostname)
            .map(|v| v.as_slice())
            .unwrap_or(&[])
    }

    /// Evaluate `new_targets` for `hostname` against the per-hostname
    /// history and the operator's [`DnsRebindingPolicy`].
    ///
    /// Pure function — DOES NOT mutate state. Caller emits events from the
    /// returned [`RebindingDecision`], then calls [`Self::commit`] to
    /// persist the observation (the commit takes the EFFECTIVE targets so
    /// the history reflects what the workload actually saw).
    ///
    /// Semantics:
    ///
    /// - `novel_ips = new_targets - history[hostname]`. Order preserved
    ///   from `new_targets`.
    /// - `threshold_exceeded = history[hostname].len() + novel_ips.len()`
    ///   strictly greater than `policy.max_novel_ips_per_hostname`. False
    ///   when no novel IPs exist (a steady CDN with churn within the prior
    ///   history is fine).
    /// - `allowlist_violations` is empty when `policy.response_ip_allowlist`
    ///   is empty (allowlist enforcement is strictly opt-in). Otherwise,
    ///   each IP in `new_targets` is checked against the parsed allowlist
    ///   filtered to entries with this `hostname` prefix; IPs failing all
    ///   entries are violations.
    /// - `effective_targets` is `new_targets` verbatim when
    ///   `policy.reject_on_rebind == false`. Otherwise, allowlist
    ///   violations AND over-cap novel IPs are filtered out.
    pub fn evaluate<'a>(
        &self,
        hostname: &str,
        new_targets: &'a [String],
        policy: &DnsRebindingPolicy,
    ) -> RebindingDecision<'a> {
        let prior: HashSet<&String> = self
            .histories
            .get(hostname)
            .map(|v| v.iter().collect())
            .unwrap_or_default();

        // Build novel_ips deterministically (order from new_targets), but
        // de-dupe within the response itself so a response that contains
        // [1.1.1.1, 1.1.1.1, 1.0.0.1] doesn't double-count the dupe.
        let mut novel_ips: Vec<&str> = Vec::new();
        let mut novel_seen: HashSet<&str> = HashSet::new();
        for ip in new_targets {
            let s: &str = ip.as_str();
            if !prior.iter().any(|p| p.as_str() == s) && novel_seen.insert(s) {
                novel_ips.push(s);
            }
        }

        let prior_len = prior.len() as u64;
        let novel_len = novel_ips.len() as u64;
        let cap = u64::from(policy.max_novel_ips_per_hostname);
        // Threshold fires only when there is something novel AND adding it
        // would push the cumulative distinct-IP count strictly past the
        // cap. A steady CDN with churn within prior history (no novel IPs)
        // is silent.
        let threshold_exceeded = novel_len > 0 && (prior_len.saturating_add(novel_len)) > cap;

        // Allowlist evaluation — only when allowlist is non-empty. We parse
        // every entry once per call (allowlist is small in practice; the
        // cell's spec usually declares a few dozen at most) and filter by
        // hostname prefix. An IP is a violation if it fails ALL applicable
        // entries; when no entries apply to this hostname, EVERY IP in the
        // response is a violation (operator declared an allowlist that
        // doesn't cover this hostname → fail-closed for the hostname).
        let mut allowlist_violations: Vec<&str> = Vec::new();
        if !policy.response_ip_allowlist.is_empty() {
            let entries = parse_allowlist_for_hostname(&policy.response_ip_allowlist, hostname);
            for ip_str in new_targets {
                let s: &str = ip_str.as_str();
                if !ip_in_allowlist(s, &entries) {
                    allowlist_violations.push(s);
                }
            }
        }

        // Effective targets — verbatim in audit mode; filtered in enforce
        // mode. Filtering: drop (a) allowlist violations AND (b) novel IPs
        // beyond the cap (i.e. the LAST `novel_len - keep_novel` novel IPs
        // in `novel_ips`).
        let effective_targets: Vec<String> = if policy.reject_on_rebind {
            // Compute the set of dropped novel IPs (over the cap).
            let mut dropped_novel: HashSet<&str> = HashSet::new();
            if threshold_exceeded {
                // How many novel IPs may we keep before exceeding cap?
                // keep_novel = max(0, cap - prior_len). Saturating because
                // prior_len could already be > cap (operator lowered the cap
                // mid-cell run, or never set it from default 4 and the
                // cell already churned past).
                let keep_novel = cap.saturating_sub(prior_len) as usize;
                for &novel in novel_ips.iter().skip(keep_novel) {
                    dropped_novel.insert(novel);
                }
            }
            let dropped_allowlist: HashSet<&str> = allowlist_violations.iter().copied().collect();

            new_targets
                .iter()
                .filter(|t| {
                    !dropped_novel.contains(t.as_str()) && !dropped_allowlist.contains(t.as_str())
                })
                .cloned()
                .collect()
        } else {
            new_targets.to_vec()
        };

        RebindingDecision {
            novel_ips,
            threshold_exceeded,
            allowlist_violations,
            effective_targets,
        }
    }

    /// Persist the current observation. Must be called AFTER the caller has
    /// emitted any threshold/rejected events so the state reflects the
    /// post-tick view.
    ///
    /// Takes the EFFECTIVE targets (post-rejection) so the history reflects
    /// what the workload actually saw. In audit-only mode the effective
    /// targets equal the raw response, so commit is functionally equivalent
    /// to "remember everything we observed."
    pub fn commit(&mut self, hostname: &str, effective_targets: &[String]) {
        let history = self.histories.entry(hostname.to_string()).or_default();
        for t in effective_targets {
            if !history.iter().any(|h| h == t) {
                history.push(t.clone());
            }
        }
    }
}

/// Parsed allowlist entry: a hostname-prefix and an IP-or-CIDR matcher.
#[derive(Debug)]
enum AllowlistMatcher {
    /// Exact IPv4/IPv6 literal (no prefix length).
    Ip(IpAddr),
    /// CIDR network — checked via `IpNet::contains`.
    Net(IpNet),
}

/// Parse the operator's `response_ip_allowlist` into matchers applicable to
/// `hostname`. Skips malformed entries silently. Pure — no I/O, no panic.
fn parse_allowlist_for_hostname(entries: &[String], hostname: &str) -> Vec<AllowlistMatcher> {
    let mut out: Vec<AllowlistMatcher> = Vec::new();
    for raw in entries {
        // Format is `hostname:ip-or-cidr`. Split on FIRST `:` so an IPv6
        // literal in the suffix (which contains its own `:`s) survives.
        let Some((prefix, suffix)) = raw.split_once(':') else {
            continue; // malformed, no separator
        };
        if prefix != hostname {
            continue; // entry applies to a different hostname
        }
        let suffix = suffix.trim();
        if suffix.is_empty() {
            continue; // malformed, empty matcher
        }
        if suffix.contains('/') {
            if let Ok(net) = IpNet::from_str(suffix) {
                out.push(AllowlistMatcher::Net(net));
            }
            // else: silently drop — schema validation should pre-filter,
            // resolver-refresh never crashes on operator input.
        } else if let Ok(ip) = IpAddr::from_str(suffix) {
            out.push(AllowlistMatcher::Ip(ip));
        }
        // else: silently drop malformed literal.
    }
    out
}

/// Test whether `ip_str` matches any entry in `entries`. Returns false when
/// `entries` is empty (caller is responsible for the "allowlist not
/// applicable to this hostname" semantic — empty `entries` here ALWAYS
/// means the IP fails). Also returns false when `ip_str` is not a valid IP
/// literal — defensive against resolvers returning hostname-typed targets.
fn ip_in_allowlist(ip_str: &str, entries: &[AllowlistMatcher]) -> bool {
    let Ok(ip) = IpAddr::from_str(ip_str) else {
        return false;
    };
    entries.iter().any(|e| match e {
        AllowlistMatcher::Ip(matcher) => *matcher == ip,
        AllowlistMatcher::Net(net) => net.contains(&ip),
    })
}

#[cfg(test)]
mod tests {
    use super::*;

    fn policy_default() -> DnsRebindingPolicy {
        DnsRebindingPolicy::default()
    }

    fn policy_with(max: u32, reject: bool, allowlist: Vec<&str>) -> DnsRebindingPolicy {
        DnsRebindingPolicy {
            response_ip_allowlist: allowlist.into_iter().map(String::from).collect(),
            max_novel_ips_per_hostname: max,
            reject_on_rebind: reject,
        }
    }

    fn s(items: &[&str]) -> Vec<String> {
        items.iter().map(|s| (*s).to_string()).collect()
    }

    // ------------------------------------------------------------
    // Novel-IP detection.
    // ------------------------------------------------------------

    #[test]
    fn evaluate_returns_all_novel_when_first_observation() {
        let state = RebindingState::new();
        let new_targets = s(&["1.1.1.1", "1.0.0.1"]);
        let policy = policy_default();
        let decision = state.evaluate("api.example.com", &new_targets, &policy);
        assert_eq!(decision.novel_ips, vec!["1.1.1.1", "1.0.0.1"]);
    }

    #[test]
    fn evaluate_returns_no_novel_when_repeat_observation() {
        let mut state = RebindingState::new();
        let first = s(&["1.1.1.1", "1.0.0.1"]);
        state.commit("api.example.com", &first);
        let new_targets = s(&["1.1.1.1", "1.0.0.1"]);
        let policy = policy_default();
        let decision = state.evaluate("api.example.com", &new_targets, &policy);
        assert!(decision.novel_ips.is_empty());
        assert!(!decision.threshold_exceeded);
    }

    // ------------------------------------------------------------
    // Threshold semantics — boundary checks.
    // ------------------------------------------------------------

    #[test]
    fn evaluate_threshold_exceeded_above_max_novel_ips() {
        let mut state = RebindingState::new();
        // Prior history has 3 IPs; cap is 4; new response adds 2 more →
        // cumulative 5 > 4 → threshold exceeded.
        state.commit("h", &s(&["1.0.0.1", "1.0.0.2", "1.0.0.3"]));
        let policy = policy_with(4, false, vec![]);
        let new_targets = s(&["1.0.0.4", "1.0.0.5"]);
        let decision = state.evaluate("h", &new_targets, &policy);
        assert!(decision.threshold_exceeded);
        assert_eq!(decision.novel_ips, vec!["1.0.0.4", "1.0.0.5"]);
    }

    #[test]
    fn evaluate_threshold_not_exceeded_at_exact_max() {
        let mut state = RebindingState::new();
        // Prior history has 3 IPs; cap is 4; new response adds 1 more →
        // cumulative exactly 4 → NOT exceeded (boundary is strict >).
        state.commit("h", &s(&["1.0.0.1", "1.0.0.2", "1.0.0.3"]));
        let policy = policy_with(4, false, vec![]);
        let new_targets = s(&["1.0.0.4"]);
        let decision = state.evaluate("h", &new_targets, &policy);
        assert!(!decision.threshold_exceeded);
        assert_eq!(decision.novel_ips, vec!["1.0.0.4"]);
    }

    // ------------------------------------------------------------
    // Allowlist semantics.
    // ------------------------------------------------------------

    #[test]
    fn evaluate_allowlist_violations_when_set() {
        let state = RebindingState::new();
        let policy = policy_with(
            10,
            false,
            vec!["api.example.com:1.1.1.1", "api.example.com:1.0.0.1"],
        );
        let new_targets = s(&["1.1.1.1", "198.51.100.7"]);
        let decision = state.evaluate("api.example.com", &new_targets, &policy);
        assert_eq!(decision.allowlist_violations, vec!["198.51.100.7"]);
    }

    #[test]
    fn evaluate_no_allowlist_violations_when_unset() {
        let state = RebindingState::new();
        let policy = policy_with(10, false, vec![]);
        let new_targets = s(&["198.51.100.7"]);
        let decision = state.evaluate("api.example.com", &new_targets, &policy);
        assert!(decision.allowlist_violations.is_empty());
    }

    // ------------------------------------------------------------
    // Reject-on-rebind filtering.
    // ------------------------------------------------------------

    #[test]
    fn evaluate_reject_on_rebind_filters_novel_above_threshold() {
        let mut state = RebindingState::new();
        state.commit("h", &s(&["1.0.0.1", "1.0.0.2", "1.0.0.3", "1.0.0.4"])); // already at cap=4
        let policy = policy_with(4, true, vec![]);
        // Two novel IPs come in; both should be filtered out under reject.
        let new_targets = s(&["1.0.0.4", "1.0.0.5", "1.0.0.6"]);
        let decision = state.evaluate("h", &new_targets, &policy);
        assert!(decision.threshold_exceeded);
        // Only the prior-known 1.0.0.4 survives; 1.0.0.5 and 1.0.0.6 dropped.
        assert_eq!(decision.effective_targets, vec!["1.0.0.4".to_string()]);
    }

    #[test]
    fn evaluate_reject_on_rebind_filters_allowlist_violations() {
        let state = RebindingState::new();
        let policy = policy_with(10, true, vec!["api.example.com:1.1.1.1"]);
        let new_targets = s(&["1.1.1.1", "198.51.100.7"]);
        let decision = state.evaluate("api.example.com", &new_targets, &policy);
        assert_eq!(decision.allowlist_violations, vec!["198.51.100.7"]);
        assert_eq!(decision.effective_targets, vec!["1.1.1.1".to_string()]);
    }

    #[test]
    fn evaluate_audit_only_keeps_violations_in_effective_targets() {
        let state = RebindingState::new();
        // reject=false → audit-only: violations recorded but NOT filtered.
        let policy = policy_with(10, false, vec!["api.example.com:1.1.1.1"]);
        let new_targets = s(&["1.1.1.1", "198.51.100.7"]);
        let decision = state.evaluate("api.example.com", &new_targets, &policy);
        assert_eq!(decision.allowlist_violations, vec!["198.51.100.7"]);
        assert_eq!(decision.effective_targets, new_targets);
    }

    // ------------------------------------------------------------
    // Commit semantics.
    // ------------------------------------------------------------

    #[test]
    fn commit_persists_observation() {
        let mut state = RebindingState::new();
        assert_eq!(state.hostname_count(), 0);
        state.commit("h", &s(&["1.1.1.1"]));
        assert_eq!(state.hostname_count(), 1);
        assert_eq!(state.history("h"), &["1.1.1.1".to_string()]);
        // Idempotent on duplicate IPs.
        state.commit("h", &s(&["1.1.1.1", "1.0.0.1"]));
        assert_eq!(
            state.history("h"),
            &["1.1.1.1".to_string(), "1.0.0.1".to_string()]
        );
    }

    // ------------------------------------------------------------
    // Allowlist parsing — malformed inputs MUST NOT panic.
    // ------------------------------------------------------------

    #[test]
    fn parse_allowlist_skips_malformed_entries() {
        let entries: Vec<String> = vec![
            "no-colon-here".into(),        // malformed: no separator
            "h:".into(),                   // malformed: empty suffix
            "h:not-an-ip".into(),          // malformed: junk literal
            "h:999.999.999.999".into(),    // malformed: out-of-range octets
            "h:1.1.1.1/notanumber".into(), // malformed: bad CIDR prefix
            "h:1.1.1.1".into(),            // valid
        ];
        let parsed = parse_allowlist_for_hostname(&entries, "h");
        assert_eq!(parsed.len(), 1, "only the well-formed entry survives");
    }

    #[test]
    fn parse_allowlist_supports_cidr() {
        let entries: Vec<String> = vec!["h:203.0.113.0/24".into()];
        let parsed = parse_allowlist_for_hostname(&entries, "h");
        assert_eq!(parsed.len(), 1);
        // 203.0.113.42 should match the /24.
        assert!(ip_in_allowlist("203.0.113.42", &parsed));
        // 203.0.114.42 (different /24) must not match.
        assert!(!ip_in_allowlist("203.0.114.42", &parsed));
    }

    // ------------------------------------------------------------
    // Bonus: hostname filtering of allowlist entries.
    // ------------------------------------------------------------

    #[test]
    fn allowlist_entries_for_other_hostname_are_ignored() {
        let state = RebindingState::new();
        // Allowlist applies to a DIFFERENT hostname → for "h", every IP
        // fails (no applicable entries → fail-closed).
        let policy = policy_with(10, false, vec!["other.example.com:1.1.1.1"]);
        let new_targets = s(&["1.1.1.1"]);
        let decision = state.evaluate("h", &new_targets, &policy);
        assert_eq!(decision.allowlist_violations, vec!["1.1.1.1"]);
    }
}