cellos-core 0.7.3

//! Policy pack — operator-defined execution constraints applied at admission.
//!
//! A `PolicyPack` document declares a named set of rules that constrain what
//! an [`ExecutionCellSpec`](crate::ExecutionCellSpec) is allowed to express.
//! The spec carries a [`PolicyRef`](crate::PolicyRef) (`spec.policy.packId` /
//! `spec.policy.packVersion`) so emitted CloudEvents attribute every run to the
//! policy pack in effect.
//!
//! # Validation surface
//!
//! Two entry points:
//!
//! - [`validate_policy_pack_document`] — structural validity of the pack itself.
//! - [`validate_spec_against_policy`] — admission gate: returns all violations
//!   the given spec has against the pack's rules.
//!
//! Both functions return structured errors rather than panicking, so callers can
//! surface them as operator-readable messages.

use std::collections::HashMap;

use serde::{Deserialize, Serialize};

use crate::{CellosError, ExecutionCellSpec, PlacementSpec, SecretDeliveryMode};

// ── Document types ────────────────────────────────────────────────────────────

/// Top-level policy pack document (`apiVersion: cellos.io/v1`, `kind: PolicyPack`).
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PolicyPackDocument {
    pub api_version: String,
    pub kind: String,
    pub spec: PolicyPackSpec,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PolicyPackSpec {
    /// Portable identifier for this pack — used in `spec.policy.packId`.
    pub id: String,
    #[serde(default)]
    pub description: Option<String>,
    /// Pack schema version — semver `MAJOR.MINOR.PATCH` (optional pre-release).
    ///
    /// When present, the runtime compares this against
    /// [`MIN_SUPPORTED_POLICY_PACK_VERSION`] at admission and rejects packs
    /// whose version is **lower than** the runtime's compiled-in floor unless
    /// the operator sets `CELLOS_POLICY_ALLOW_DOWNGRADE=1`. When absent, the
    /// pack is treated as version [`MIN_SUPPORTED_POLICY_PACK_VERSION`] for
    /// backwards compatibility with packs authored before the field existed.
    /// See P4-04.
    #[serde(default)]
    pub version: Option<String>,
    /// T11 — optional placement scope. When set, the pack's rules apply only
    /// to specs whose `spec.placement` matches every populated field of this
    /// scope (a `None` field on the policy means "any" for that axis).
    ///
    /// A pack without a `placement` scope is global — applied to every spec
    /// regardless of where it runs. The matching rule lives in
    /// [`spec_matches_placement_scope`] so the contract is testable
    /// independently of the rest of admission.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub placement: Option<PlacementSpec>,
    pub rules: PolicyRules,
}

/// Compiled-in floor for the policy pack schema version. Packs declaring a
/// `spec.version` strictly lower than this triple are rejected at admission
/// unless `CELLOS_POLICY_ALLOW_DOWNGRADE=1` is set. P4-04.
pub const MIN_SUPPORTED_POLICY_PACK_VERSION: &str = "1.0.0";

/// Environment variable name that, when truthy
/// (`1` / `true` / `yes` / `on`, case-insensitive), suppresses the policy pack
/// downgrade check at admission. P4-04.
pub const POLICY_ALLOW_DOWNGRADE_ENV: &str = "CELLOS_POLICY_ALLOW_DOWNGRADE";

/// Constraint rules within a policy pack.
///
/// All fields are optional — an absent field means "no restriction on this axis."
/// A `PolicyPack` with an empty `PolicyRules` is valid but does not constrain
/// any spec (useful as an explicit "audit-only" marker).
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PolicyRules {
    /// `spec.lifetime.ttlSeconds` must not exceed this value.
    #[serde(default)]
    pub max_lifetime_ttl_seconds: Option<u64>,

    /// `spec.run.limits.memoryMaxBytes` must not exceed this value.
    #[serde(default)]
    pub max_memory_max_bytes: Option<u64>,

    /// `spec.run.timeoutMs` must not exceed this value.
    #[serde(default)]
    pub max_run_timeout_ms: Option<u64>,

    /// When `true`, `spec.authority.egressRules` must be non-empty.
    ///
    /// Use this to force explicit declaration of all outbound network intent —
    /// specs that declare no egress rules are rejected.
    #[serde(default)]
    pub require_egress_declared: bool,

    /// When `true`, specs must declare zero outbound egress rules.
    ///
    /// Use this to enforce fully air-gapped workloads. A spec with any
    /// `spec.authority.egressRules` is rejected.
    #[serde(default)]
    pub forbid_outbound_egress_rules: bool,

    /// When non-empty, every `spec.authority.egressRules[].host` must match
    /// at least one pattern in this list.
    ///
    /// Patterns support a single leading `*.` wildcard (e.g. `*.internal`).
    /// Exact hostnames are also accepted. An empty list means no restriction.
    #[serde(default)]
    pub allowed_egress_hosts: Vec<String>,

    /// When `true`, `spec.run.secretDelivery` must not be `env`.
    ///
    /// Use this to prohibit the default env-variable secret delivery mode
    /// and require `runtimeBroker` or `runtimeLeasedBroker` instead.
    #[serde(default)]
    pub require_runtime_secret_delivery: bool,

    /// When `true`, `spec.run.limits` must be present.
    ///
    /// Use this to ensure every spec explicitly declares resource bounds.
    #[serde(default)]
    pub require_resource_limits: bool,

    /// When `true`, any spec that declares an egress rule on port 53 (DNS) is
    /// flagged with a [`PolicyViolation`] unless the spec also sets
    /// `authority.egressRules[].protocol = "dns-acknowledged"` on every port-53
    /// rule.
    ///
    /// CellOS's egress policy operates at L3/L4 (IP:port via nftables) and has
    /// no visibility into DNS query content. A cell with declared port 53
    /// egress can therefore exfiltrate arbitrary data via DNS TXT query labels
    /// to an attacker-controlled authoritative nameserver, completely outside
    /// CellOS observability. See `docs/sec_roadmap.md` R10 / SEC-15.
    ///
    /// The `dns-acknowledged` protocol value is a forcing function, not a
    /// security control: operators who understand the covert-channel risk can
    /// set it to suppress the violation. Future work (L2-04 extension) is a
    /// DNS proxy inside the cell netns with query-level CloudEvents.
    #[serde(default)]
    pub flag_dns_egress_without_acknowledgment: Option<bool>,

    /// When `true`, any port-53 egress rule with `protocol = "dns-acknowledged"`
    /// must also supply a non-empty `dnsEgressJustification` string.
    ///
    /// This closes SEAM-2 (illusory consent): without it, an operator can
    /// suppress [`flag_dns_egress_without_acknowledgment`](Self::flag_dns_egress_without_acknowledgment)
    /// by typing four words and zero further effort, defeating the forcing
    /// function. With the rule enabled, the operator must record an auditable
    /// justification alongside the acknowledgment, raising the cost of
    /// reflexive suppression and creating evidence of stated reasoning.
    /// See `docs/sec_roadmap.md` R10 / SEC-15c.
    #[serde(default, rename = "requireDnsEgressJustification")]
    pub require_dns_egress_justification: Option<bool>,

    /// A2-02: per-caller-identity allowlist for `spec.authority.secretRefs`.
    ///
    /// Map shape: `caller-identity → [allowed secretRef key]`. The caller
    /// identity is host-stamped via `CELLOS_CALLER_IDENTITY` at the supervisor
    /// boundary (per ADR-0007 / D11: `cellos-core` itself does NOT read process
    /// env vars; the resolved identity is threaded through to
    /// [`validate_secret_refs_against_allowlist`] as a parameter).
    ///
    /// When this field is `None`, the allowlist gate is **skipped entirely** —
    /// admission proceeds against the rest of the rules unchanged. This is the
    /// "audit-only" / opt-in posture documented in ADR-0007 §"What 1.0 ships".
    ///
    /// When this field is `Some(map)`:
    /// - The caller's identity MUST appear as a key in `map` (otherwise the
    ///   caller is "unmapped" and rejected with [`CellosError::InvalidSpec`]
    ///   carrying a `caller_unmapped` reason string).
    /// - Every entry in `spec.authority.secretRefs` MUST appear in
    ///   `map[caller_identity]` (otherwise admission is rejected with a
    ///   `secret_ref_denied` reason naming the offending key).
    ///
    /// The keys (caller identities) and values (secretRef names) are operator-
    /// curated portable strings; this field carries no implicit role hierarchy
    /// (see ADR-0007 non-goal "Hierarchical roles / role inheritance"). Two
    /// callers needing the same set of refs are two map entries, not one.
    ///
    /// See ADR-0007 (`docs/adr/0007-rbac-secret-ref-admission.md`) for the
    /// contract; A2-03 (multi-tenant event isolation) is a separate surface.
    #[serde(default, rename = "secretRefAllowlist")]
    pub secret_ref_allowlist: Option<HashMap<String, Vec<String>>>,
}

// ── Violation ─────────────────────────────────────────────────────────────────

/// A single rule violation produced by [`validate_spec_against_policy`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PolicyViolation {
    /// The `camelCase` rule name that was violated (e.g. `"maxLifetimeTtlSeconds"`).
    pub rule: String,
    /// Human-readable description of the violation.
    pub message: String,
}

impl std::fmt::Display for PolicyViolation {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "[{}] {}", self.rule, self.message)
    }
}

// ── Version compatibility (P4-04) ─────────────────────────────────────────────

/// Parse a semver `MAJOR.MINOR.PATCH` string into a comparable triple.
///
/// Accepts an optional `-pre` suffix per SemVer 2.0 (e.g. `"1.2.3-rc.1"`); the
/// pre-release portion is ignored for ordering — a downgrade-vs-floor check
/// only needs the numeric core. Leading zeros on numeric components (e.g.
/// `"01.02.03"`) are rejected to keep the surface canonical.
///
/// Returns `Err(CellosError::InvalidSpec)` with a descriptive message on
/// malformed input.
fn parse_semver_triple(value: &str) -> Result<(u64, u64, u64), CellosError> {
    let core = match value.split_once('-') {
        Some((core, pre)) => {
            if pre.is_empty()
                || !pre
                    .chars()
                    .all(|c| c.is_ascii_alphanumeric() || matches!(c, '.' | '-'))
            {
                return Err(CellosError::InvalidSpec(format!(
                    "policy pack spec.version {value:?} has malformed pre-release suffix"
                )));
            }
            core
        }
        None => value,
    };
    let parts: Vec<&str> = core.split('.').collect();
    if parts.len() != 3 {
        return Err(CellosError::InvalidSpec(format!(
            "policy pack spec.version {value:?} must be a MAJOR.MINOR.PATCH semver string"
        )));
    }
    let mut triple = [0u64; 3];
    for (i, p) in parts.iter().enumerate() {
        if p.is_empty() || !p.chars().all(|c| c.is_ascii_digit()) {
            return Err(CellosError::InvalidSpec(format!(
                "policy pack spec.version {value:?} component {p:?} is not a non-negative integer"
            )));
        }
        if p.len() > 1 && p.starts_with('0') {
            return Err(CellosError::InvalidSpec(format!(
                "policy pack spec.version {value:?} component {p:?} has a leading zero"
            )));
        }
        triple[i] = p.parse::<u64>().map_err(|_| {
            CellosError::InvalidSpec(format!(
                "policy pack spec.version {value:?} component {p:?} overflows u64"
            ))
        })?;
    }
    Ok((triple[0], triple[1], triple[2]))
}

/// Validate a declared policy pack `spec.version` against the runtime's
/// compiled-in floor [`MIN_SUPPORTED_POLICY_PACK_VERSION`]. P4-04.
///
/// - When `declared` is `None`, the pack is treated as the floor version
///   (backwards compatibility for packs authored before the field existed).
/// - When `declared` parses to a triple **strictly lower** than the floor,
///   the pack is rejected unless `allow_downgrade` is `true`.
/// - When `declared` is malformed, the pack is rejected unconditionally — the
///   downgrade-allow override does **not** suppress structural errors.
///
/// Pure free function — `cellos-core` does not read process env vars (D11).
/// The supervisor admission path reads `CELLOS_POLICY_ALLOW_DOWNGRADE` and
/// passes the resolved bool through [`spec_validation::check_policy_pack_version`]
/// or directly here. [`validate_policy_pack_document`] always calls this with
/// `allow_downgrade=false` so structural validation stays strict; the env
/// override only kicks in when the supervisor explicitly opts in.
pub fn check_policy_pack_version_compatibility(
    declared: Option<&str>,
    allow_downgrade: bool,
) -> Result<(), CellosError> {
    let declared_triple = match declared {
        Some(v) => parse_semver_triple(v)?,
        None => return Ok(()),
    };
    let floor_triple = parse_semver_triple(MIN_SUPPORTED_POLICY_PACK_VERSION)
        .expect("MIN_SUPPORTED_POLICY_PACK_VERSION must parse");

    if declared_triple < floor_triple {
        if allow_downgrade {
            return Ok(());
        }
        return Err(CellosError::InvalidSpec(format!(
            "policy pack spec.version {} is older than runtime-supported floor {} \
             (set {}=1 to override)",
            declared.unwrap_or(""),
            MIN_SUPPORTED_POLICY_PACK_VERSION,
            POLICY_ALLOW_DOWNGRADE_ENV
        )));
    }
    Ok(())
}

// ── Document validation ───────────────────────────────────────────────────────

/// Validate the structural integrity of a [`PolicyPackDocument`].
///
/// Checks:
/// - `apiVersion == "cellos.io/v1"`
/// - `kind == "PolicyPack"`
/// - `spec.id` is a valid portable identifier
/// - All numeric bounds are > 0 when set
/// - All `allowedEgressHosts` patterns are non-empty
pub fn validate_policy_pack_document(doc: &PolicyPackDocument) -> Result<(), CellosError> {
    if doc.api_version != "cellos.io/v1" {
        return Err(CellosError::InvalidSpec(format!(
            "policy pack apiVersion must be \"cellos.io/v1\", got {:?}",
            doc.api_version
        )));
    }
    if doc.kind != "PolicyPack" {
        return Err(CellosError::InvalidSpec(format!(
            "policy pack kind must be \"PolicyPack\", got {:?}",
            doc.kind
        )));
    }

    if !crate::spec_validation::is_portable_identifier(&doc.spec.id) {
        return Err(CellosError::InvalidSpec(format!(
            "policy pack spec.id {:?} is not a valid portable identifier",
            doc.spec.id
        )));
    }

    // P4-04: reject packs whose declared schema version is older than the
    // runtime's compiled-in floor. Strict by default — `validate_policy_pack_document`
    // is the structural-correctness gate and must not depend on process env;
    // the supervisor reads `CELLOS_POLICY_ALLOW_DOWNGRADE` and calls
    // `check_policy_pack_version_compatibility(declared, allow_downgrade=true)`
    // separately when the operator opted in.
    check_policy_pack_version_compatibility(doc.spec.version.as_deref(), false)?;

    let rules = &doc.spec.rules;

    if let Some(v) = rules.max_lifetime_ttl_seconds {
        if v == 0 {
            return Err(CellosError::InvalidSpec(
                "policy pack rules.maxLifetimeTtlSeconds must be > 0".into(),
            ));
        }
    }
    if let Some(v) = rules.max_memory_max_bytes {
        if v == 0 {
            return Err(CellosError::InvalidSpec(
                "policy pack rules.maxMemoryMaxBytes must be > 0".into(),
            ));
        }
    }
    if let Some(v) = rules.max_run_timeout_ms {
        if v == 0 {
            return Err(CellosError::InvalidSpec(
                "policy pack rules.maxRunTimeoutMs must be > 0".into(),
            ));
        }
    }
    if rules.require_egress_declared && rules.forbid_outbound_egress_rules {
        return Err(CellosError::InvalidSpec(
            "policy pack rules.requireEgressDeclared and rules.forbidOutboundEgressRules \
             are mutually exclusive"
                .into(),
        ));
    }
    for pattern in &rules.allowed_egress_hosts {
        if pattern.is_empty() {
            return Err(CellosError::InvalidSpec(
                "policy pack rules.allowedEgressHosts contains an empty pattern".into(),
            ));
        }
    }

    Ok(())
}

// ── Spec admission gate ───────────────────────────────────────────────────────

/// T11 — match a spec's placement against a policy pack's placement scope.
///
/// Returns `true` when **every populated field of the scope** equals the
/// matching field in `spec_placement`. Empty fields on the scope are
/// wildcards. When the scope has any populated field and the spec has no
/// `placement` block at all, the match fails (the spec is unscoped).
///
/// This intentionally requires exact, case-sensitive equality on each
/// populated axis — placement identifiers are operator-curated portable
/// strings already validated at admission and don't carry hierarchy.
pub fn spec_matches_placement_scope(
    spec_placement: Option<&PlacementSpec>,
    scope: &PlacementSpec,
) -> bool {
    let scope_has_any = scope.pool_id.is_some()
        || scope.kubernetes_namespace.is_some()
        || scope.queue_name.is_some();
    if !scope_has_any {
        return true;
    }
    let Some(spec_placement) = spec_placement else {
        return false;
    };
    if let Some(pool) = scope.pool_id.as_deref() {
        if spec_placement.pool_id.as_deref() != Some(pool) {
            return false;
        }
    }
    if let Some(ns) = scope.kubernetes_namespace.as_deref() {
        if spec_placement.kubernetes_namespace.as_deref() != Some(ns) {
            return false;
        }
    }
    if let Some(queue) = scope.queue_name.as_deref() {
        if spec_placement.queue_name.as_deref() != Some(queue) {
            return false;
        }
    }
    true
}

/// Check `spec` against all rules in `pack`.
///
/// Returns the full list of violations. An empty `Vec` means admission is
/// granted. Callers MAY treat any violation as a hard reject, or log them for
/// audit purposes without blocking execution.
pub fn validate_spec_against_policy(
    spec: &ExecutionCellSpec,
    pack: &PolicyPackSpec,
) -> Vec<PolicyViolation> {
    let mut violations = Vec::new();

    // T11 — placement-scoped policy packs.
    //
    // A pack with `placement` constraints applies only to specs whose
    // placement matches every populated field. A pack without `placement`
    // is global. When the pack does NOT apply to this spec, we return an
    // empty violation list — the spec is out of scope, not "policy clean".
    // Callers wanting "did any pack apply" should track scope separately.
    if let Some(scope) = &pack.placement {
        if !spec_matches_placement_scope(spec.placement.as_ref(), scope) {
            return violations;
        }
    }

    let rules = &pack.rules;

    // ── Lifetime cap ─────────────────────────────────────────────────────────
    if let Some(max) = rules.max_lifetime_ttl_seconds {
        if spec.lifetime.ttl_seconds > max {
            violations.push(PolicyViolation {
                rule: "maxLifetimeTtlSeconds".into(),
                message: format!(
                    "spec.lifetime.ttlSeconds {} exceeds policy maximum {}",
                    spec.lifetime.ttl_seconds, max
                ),
            });
        }
    }

    // ── Memory cap ───────────────────────────────────────────────────────────
    if let Some(max) = rules.max_memory_max_bytes {
        let actual = spec
            .run
            .as_ref()
            .and_then(|r| r.limits.as_ref())
            .and_then(|l| l.memory_max_bytes);
        if let Some(actual) = actual {
            if actual > max {
                violations.push(PolicyViolation {
                    rule: "maxMemoryMaxBytes".into(),
                    message: format!(
                        "spec.run.limits.memoryMaxBytes {actual} exceeds policy maximum {max}"
                    ),
                });
            }
        }
    }

    // ── Run timeout cap ───────────────────────────────────────────────────────
    if let Some(max) = rules.max_run_timeout_ms {
        let actual = spec.run.as_ref().and_then(|r| r.timeout_ms);
        if let Some(actual) = actual {
            if actual > max {
                violations.push(PolicyViolation {
                    rule: "maxRunTimeoutMs".into(),
                    message: format!("spec.run.timeoutMs {actual} exceeds policy maximum {max}"),
                });
            }
        }
    }

    // ── Egress rules ─────────────────────────────────────────────────────────
    let egress_rules = spec.authority.egress_rules.as_deref().unwrap_or_default();

    if rules.require_egress_declared && egress_rules.is_empty() {
        violations.push(PolicyViolation {
            rule: "requireEgressDeclared".into(),
            message: "policy requires spec.authority.egressRules to be non-empty".into(),
        });
    }

    if rules.forbid_outbound_egress_rules && !egress_rules.is_empty() {
        violations.push(PolicyViolation {
            rule: "forbidOutboundEgressRules".into(),
            message: format!(
                "policy forbids outbound egress rules but spec declares {} rule(s)",
                egress_rules.len()
            ),
        });
    }

    if !rules.allowed_egress_hosts.is_empty() {
        for rule in egress_rules {
            if !rules
                .allowed_egress_hosts
                .iter()
                .any(|pat| host_matches_pattern(&rule.host, pat))
            {
                violations.push(PolicyViolation {
                    rule: "allowedEgressHosts".into(),
                    message: format!(
                        "egress host {:?} does not match any allowed pattern in {:?}",
                        rule.host, rules.allowed_egress_hosts
                    ),
                });
            }
        }
    }

    // ── Secret delivery ───────────────────────────────────────────────────────
    if rules.require_runtime_secret_delivery {
        let delivery = spec
            .run
            .as_ref()
            .map(|r| &r.secret_delivery)
            .unwrap_or(&SecretDeliveryMode::Env);
        if *delivery == SecretDeliveryMode::Env {
            violations.push(PolicyViolation {
                rule: "requireRuntimeSecretDelivery".into(),
                message: "policy requires spec.run.secretDelivery to be runtimeBroker or \
                           runtimeLeasedBroker, not env"
                    .into(),
            });
        }
    }

    // ── Resource limits required ──────────────────────────────────────────────
    if rules.require_resource_limits {
        let has_limits = spec.run.as_ref().and_then(|r| r.limits.as_ref()).is_some();
        if !has_limits {
            violations.push(PolicyViolation {
                rule: "requireResourceLimits".into(),
                message: "policy requires spec.run.limits to be declared".into(),
            });
        }
    }

    // ── DNS egress acknowledgment (SEC-15) ────────────────────────────────────
    //
    // Port-53 egress is a covert exfiltration channel: DNS TXT queries can
    // carry arbitrary data via labels to an attacker-controlled authoritative
    // nameserver, outside CellOS visibility. Surface this as a policy
    // violation unless the operator explicitly acknowledged the risk by
    // setting `protocol: dns-acknowledged` on every port-53 rule.
    if rules.flag_dns_egress_without_acknowledgment == Some(true) {
        let mut has_dns_egress = false;
        let mut all_acknowledged = true;
        for rule in egress_rules {
            if rule.port == 53 {
                has_dns_egress = true;
                let acknowledged = rule
                    .protocol
                    .as_deref()
                    .is_some_and(|p| p.eq_ignore_ascii_case("dns-acknowledged"));
                if !acknowledged {
                    all_acknowledged = false;
                }
            }
        }
        if has_dns_egress && !all_acknowledged {
            violations.push(PolicyViolation {
                rule: "flagDnsEgressWithoutAcknowledgment".into(),
                message: "spec declares port 53 (DNS) egress without acknowledgment — \
                          DNS can be used as a covert exfiltration channel; set \
                          protocol: dns-acknowledged to acknowledge this risk"
                    .into(),
            });
        }
    }

    // ── DNS egress justification (SEC-15c / SEAM-2) ──────────────────────────
    //
    // The `dns-acknowledged` protocol value alone is too cheap a suppression —
    // an operator can silence the acknowledgment check by typing four words
    // with zero risk evaluation. When this rule is active, every port-53
    // dns-acknowledged rule must also carry a non-empty
    // `dnsEgressJustification` string. The justification is operator metadata
    // (free-text), not a capability constraint, so it does not participate in
    // egress allowlist or capability subset checks.
    if rules.require_dns_egress_justification == Some(true) {
        for rule in egress_rules {
            if rule.port != 53 {
                continue;
            }
            let acknowledged = rule
                .protocol
                .as_deref()
                .is_some_and(|p| p.eq_ignore_ascii_case("dns-acknowledged"));
            if !acknowledged {
                continue;
            }
            let justified = rule
                .dns_egress_justification
                .as_deref()
                .is_some_and(|s| !s.trim().is_empty());
            if !justified {
                violations.push(PolicyViolation {
                    rule: "requireDnsEgressJustification".into(),
                    message: "port-53 egress rule with protocol dns-acknowledged \
                              requires a non-empty dnsEgressJustification field"
                        .into(),
                });
            }
        }
    }

    violations
}

/// A2-02 / ADR-0007: per-caller-identity secretRef allowlist gate.
///
/// Returns `Ok(())` when admission may proceed. Returns
/// [`CellosError::InvalidSpec`] with a structured, operator-readable reason
/// string when the caller is not mapped, or when the spec requests a secretRef
/// outside the caller's allowlist.
///
/// Behaviour:
/// - When `rules.secret_ref_allowlist` is `None`, the gate is **skipped** and
///   `Ok(())` is returned regardless of the spec. This is the opt-in posture
///   from ADR-0007: an operator who has not authored an allowlist is not
///   suddenly subject to one. The supervisor side is responsible for emitting
///   the `cell.admission.v1.caller_identity_check_skipped` info event when it
///   chooses; this function is silent in that case.
/// - When `rules.secret_ref_allowlist` is `Some(map)`:
///   - If `caller_identity` is not a key of `map`, the caller is rejected as
///     `caller_unmapped`.
///   - Otherwise, every entry in `spec.authority.secret_refs` (when present)
///     MUST be present in `map[caller_identity]`. The first denied ref is
///     named in the error message; the function returns on first violation.
///
/// **D11 boundary**: this function does NOT read process env vars. The caller
/// identity is supplied as a parameter; the supervisor's startup path is
/// responsible for reading `CELLOS_CALLER_IDENTITY` (defaulting to `"default"`
/// when unset) and threading it in.
pub fn validate_secret_refs_against_allowlist(
    spec: &ExecutionCellSpec,
    rules: &PolicyRules,
    caller_identity: &str,
) -> Result<(), CellosError> {
    let Some(allowlist) = rules.secret_ref_allowlist.as_ref() else {
        // Audit-only / opt-in: pack lacks the allowlist. Admission proceeds.
        return Ok(());
    };

    let Some(allowed) = allowlist.get(caller_identity) else {
        return Err(CellosError::InvalidSpec(format!(
            "caller_unmapped: caller identity {caller_identity:?} is not present in \
             policy pack rules.secretRefAllowlist; admission rejected per ADR-0007"
        )));
    };

    let Some(requested) = spec.authority.secret_refs.as_ref() else {
        // No secretRefs requested — nothing to check.
        return Ok(());
    };

    for ref_name in requested {
        if !allowed.iter().any(|granted| granted == ref_name) {
            return Err(CellosError::InvalidSpec(format!(
                "secret_ref_denied: caller {caller_identity:?} is not granted secretRef \
                 {ref_name:?} by policy pack rules.secretRefAllowlist; admission rejected \
                 per ADR-0007"
            )));
        }
    }

    Ok(())
}

/// Returns `true` when `host` matches `pattern`.
///
/// Patterns:
/// - `"*.example.com"` — matches any subdomain of `example.com` (at least one label prefix)
/// - `"api.example.com"` — exact match
/// - `"*"` — matches anything
fn host_matches_pattern(host: &str, pattern: &str) -> bool {
    if pattern == "*" {
        return true;
    }
    if let Some(suffix) = pattern.strip_prefix("*.") {
        // "*.example.com" must match "foo.example.com" but NOT "example.com" itself.
        host.ends_with(&format!(".{suffix}"))
    } else {
        host == pattern
    }
}

// ── T12: AuthorizationPolicy ──────────────────────────────────────────────────
//
// T12 RBAC ships an operator-defined authorization policy that gates cell
// admission on the subject (tenant identity), the target pool, and the
// referenced policy pack — independent of the existing `PolicyPack` admission
// gate. The supervisor loads one `AuthorizationPolicyDocument` from
// `CELLOS_AUTHZ_POLICY_PATH` at startup and evaluates every spec against it
// before `host.create()`. See `contracts/schemas/authorization-policy-v1.schema.json`.
//
// Why separate from `PolicyPack`? PolicyPack constrains *what a spec is
// allowed to express* (egress hosts, TTL caps, etc). AuthorizationPolicy
// constrains *who is allowed to submit a spec at all* — a different rule
// namespace with a different rejection event type. Both gates fire
// independently.

/// Top-level authorization policy document (`apiVersion: cellos.io/v1`,
/// `kind: AuthorizationPolicy`). Loaded by the supervisor from
/// `CELLOS_AUTHZ_POLICY_PATH` at startup.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct AuthorizationPolicyDocument {
    pub api_version: String,
    pub kind: String,
    pub spec: AuthorizationPolicy,
}

/// T12: authorization policy gate.
///
/// `subjects` is the allowlist of operator identities (currently a tenant id
/// in 1.0). `allowed_pools` and `allowed_policy_packs` narrow the surface
/// further — empty means "no restriction on this axis". `max_cells_per_hour`
/// is an optional rolling-hour rate cap.
///
/// All sets are matched by exact equality; no glob/regex semantics. Identity
/// strings are opaque tokens — `oidc:github:org/team`,
/// `k8s:serviceaccount:ns/name`, or `tenant:<id>` are all valid.
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct AuthorizationPolicy {
    /// Operator identities authorized by this policy. The supervisor compares
    /// `spec.correlation.tenantId` against this list at admission. An empty
    /// list rejects *every* spec — there is no implicit allow-all.
    pub subjects: Vec<String>,

    /// Pool IDs the subject may target via `spec.placement.poolId`. Empty
    /// means no pool restriction (all pools allowed).
    #[serde(default)]
    pub allowed_pools: Vec<String>,

    /// Policy pack IDs the subject may reference via `spec.policy.packId`.
    /// Empty means no pack restriction (all packs allowed).
    #[serde(default)]
    pub allowed_policy_packs: Vec<String>,

    /// Optional rolling-hour rate cap. When unset (`None`) there is no rate
    /// limit; when set (`Some(n)`), at most `n` admitted cells per hour per
    /// subject. The supervisor maintains the per-subject counter in-memory.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub max_cells_per_hour: Option<u32>,
}

/// Validate the structural integrity of an [`AuthorizationPolicyDocument`].
///
/// Checks:
/// - `apiVersion == "cellos.io/v1"`
/// - `kind == "AuthorizationPolicy"`
/// - `spec.subjects` is non-empty (an empty subjects list would reject every
///   spec; the operator almost certainly meant to delete the file instead).
/// - No subject / pool / pack id is empty or whitespace-only.
/// - `maxCellsPerHour`, when set, is > 0.
pub fn validate_authorization_policy(doc: &AuthorizationPolicyDocument) -> Result<(), CellosError> {
    if doc.api_version != "cellos.io/v1" {
        return Err(CellosError::InvalidSpec(format!(
            "authorization policy apiVersion must be \"cellos.io/v1\", got {:?}",
            doc.api_version
        )));
    }
    if doc.kind != "AuthorizationPolicy" {
        return Err(CellosError::InvalidSpec(format!(
            "authorization policy kind must be \"AuthorizationPolicy\", got {:?}",
            doc.kind
        )));
    }
    let policy = &doc.spec;
    if policy.subjects.is_empty() {
        return Err(CellosError::InvalidSpec(
            "authorization policy spec.subjects must be non-empty — \
             an empty subjects list would reject every spec; \
             remove CELLOS_AUTHZ_POLICY_PATH to disable the gate instead"
                .into(),
        ));
    }
    for s in &policy.subjects {
        if s.trim().is_empty() {
            return Err(CellosError::InvalidSpec(
                "authorization policy spec.subjects contains an empty / whitespace-only entry"
                    .into(),
            ));
        }
    }
    for p in &policy.allowed_pools {
        if p.trim().is_empty() {
            return Err(CellosError::InvalidSpec(
                "authorization policy spec.allowedPools contains an empty entry".into(),
            ));
        }
    }
    for p in &policy.allowed_policy_packs {
        if p.trim().is_empty() {
            return Err(CellosError::InvalidSpec(
                "authorization policy spec.allowedPolicyPacks contains an empty entry".into(),
            ));
        }
    }
    if let Some(0) = policy.max_cells_per_hour {
        return Err(CellosError::InvalidSpec(
            "authorization policy spec.maxCellsPerHour must be > 0 when set".into(),
        ));
    }
    Ok(())
}

// ── Tests ─────────────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;
    use crate::types::{AuthorityBundle, EgressRule, Lifetime, RunLimits, RunSpec};

    fn minimal_spec() -> ExecutionCellSpec {
        ExecutionCellSpec {
            id: "test-cell".into(),
            correlation: None,
            ingress: None,
            environment: None,
            placement: None,
            policy: None,
            identity: None,
            run: Some(RunSpec {
                argv: vec!["/usr/bin/true".into()],
                working_directory: None,
                timeout_ms: None,
                limits: None,
                secret_delivery: SecretDeliveryMode::Env,
            }),
            authority: AuthorityBundle {
                filesystem: None,
                network: None,
                egress_rules: None,
                secret_refs: None,
                authority_derivation: None,
                dns_authority: None,
                cdn_authority: None,
            },
            lifetime: Lifetime { ttl_seconds: 300 },
            export: None,
            telemetry: None,
        }
    }

    fn minimal_pack(rules: PolicyRules) -> PolicyPackSpec {
        PolicyPackSpec {
            id: "test-policy".into(),
            description: None,
            version: None,
            placement: None,
            rules,
        }
    }

    fn minimal_doc(rules: PolicyRules) -> PolicyPackDocument {
        PolicyPackDocument {
            api_version: "cellos.io/v1".into(),
            kind: "PolicyPack".into(),
            spec: minimal_pack(rules),
        }
    }

    // ── validate_policy_pack_document ────────────────────────────────────────

    #[test]
    fn valid_doc_passes_structural_check() {
        let doc = minimal_doc(PolicyRules::default());
        assert!(validate_policy_pack_document(&doc).is_ok());
    }

    #[test]
    fn wrong_api_version_is_rejected() {
        let mut doc = minimal_doc(PolicyRules::default());
        doc.api_version = "v1".into();
        assert!(validate_policy_pack_document(&doc).is_err());
    }

    #[test]
    fn wrong_kind_is_rejected() {
        let mut doc = minimal_doc(PolicyRules::default());
        doc.kind = "ExecutionCell".into();
        assert!(validate_policy_pack_document(&doc).is_err());
    }

    #[test]
    fn invalid_spec_id_is_rejected() {
        let mut doc = minimal_doc(PolicyRules::default());
        doc.spec.id = "-bad".into();
        assert!(validate_policy_pack_document(&doc).is_err());
    }

    #[test]
    fn zero_max_ttl_is_rejected() {
        let doc = minimal_doc(PolicyRules {
            max_lifetime_ttl_seconds: Some(0),
            ..Default::default()
        });
        assert!(validate_policy_pack_document(&doc).is_err());
    }

    #[test]
    fn require_and_forbid_egress_together_is_rejected() {
        let doc = minimal_doc(PolicyRules {
            require_egress_declared: true,
            forbid_outbound_egress_rules: true,
            ..Default::default()
        });
        assert!(validate_policy_pack_document(&doc).is_err());
    }

    #[test]
    fn empty_egress_host_pattern_is_rejected() {
        let doc = minimal_doc(PolicyRules {
            allowed_egress_hosts: vec!["".into()],
            ..Default::default()
        });
        assert!(validate_policy_pack_document(&doc).is_err());
    }

    // ── validate_spec_against_policy ─────────────────────────────────────────

    #[test]
    fn spec_passes_empty_policy() {
        let spec = minimal_spec();
        let pack = minimal_pack(PolicyRules::default());
        assert!(validate_spec_against_policy(&spec, &pack).is_empty());
    }

    #[test]
    fn ttl_exceeds_max_is_violation() {
        let spec = minimal_spec(); // ttl_seconds = 300
        let pack = minimal_pack(PolicyRules {
            max_lifetime_ttl_seconds: Some(60),
            ..Default::default()
        });
        let violations = validate_spec_against_policy(&spec, &pack);
        assert_eq!(violations.len(), 1);
        assert_eq!(violations[0].rule, "maxLifetimeTtlSeconds");
    }

    #[test]
    fn ttl_at_exact_max_passes() {
        let spec = minimal_spec(); // ttl_seconds = 300
        let pack = minimal_pack(PolicyRules {
            max_lifetime_ttl_seconds: Some(300),
            ..Default::default()
        });
        assert!(validate_spec_against_policy(&spec, &pack).is_empty());
    }

    #[test]
    fn memory_exceeds_max_is_violation() {
        let mut spec = minimal_spec();
        spec.run = Some(RunSpec {
            argv: vec!["/usr/bin/true".into()],
            working_directory: None,
            timeout_ms: None,
            limits: Some(RunLimits {
                memory_max_bytes: Some(8 * 1024 * 1024 * 1024), // 8 GiB
                cpu_max: None,
                graceful_shutdown_seconds: None,
            }),
            secret_delivery: SecretDeliveryMode::Env,
        });
        let pack = minimal_pack(PolicyRules {
            max_memory_max_bytes: Some(4 * 1024 * 1024 * 1024), // 4 GiB cap
            ..Default::default()
        });
        let violations = validate_spec_against_policy(&spec, &pack);
        assert_eq!(violations.len(), 1);
        assert_eq!(violations[0].rule, "maxMemoryMaxBytes");
    }

    #[test]
    fn run_timeout_exceeds_max_is_violation() {
        let mut spec = minimal_spec();
        spec.run = Some(RunSpec {
            argv: vec!["/usr/bin/true".into()],
            working_directory: None,
            timeout_ms: Some(7_200_000), // 2 hours
            limits: None,
            secret_delivery: SecretDeliveryMode::Env,
        });
        let pack = minimal_pack(PolicyRules {
            max_run_timeout_ms: Some(3_600_000), // 1 hour cap
            ..Default::default()
        });
        let violations = validate_spec_against_policy(&spec, &pack);
        assert_eq!(violations.len(), 1);
        assert_eq!(violations[0].rule, "maxRunTimeoutMs");
    }

    #[test]
    fn require_egress_declared_fails_when_no_egress_rules() {
        let spec = minimal_spec(); // no egress rules
        let pack = minimal_pack(PolicyRules {
            require_egress_declared: true,
            ..Default::default()
        });
        let violations = validate_spec_against_policy(&spec, &pack);
        assert_eq!(violations.len(), 1);
        assert_eq!(violations[0].rule, "requireEgressDeclared");
    }

    #[test]
    fn require_egress_declared_passes_when_egress_present() {
        let mut spec = minimal_spec();
        spec.authority.egress_rules = Some(vec![EgressRule {
            host: "api.github.com".into(),
            port: 443,
            protocol: None,
            dns_egress_justification: None,
        }]);
        let pack = minimal_pack(PolicyRules {
            require_egress_declared: true,
            ..Default::default()
        });
        assert!(validate_spec_against_policy(&spec, &pack).is_empty());
    }

    #[test]
    fn forbid_outbound_egress_fails_when_rules_declared() {
        let mut spec = minimal_spec();
        spec.authority.egress_rules = Some(vec![EgressRule {
            host: "external.example.com".into(),
            port: 443,
            protocol: None,
            dns_egress_justification: None,
        }]);
        let pack = minimal_pack(PolicyRules {
            forbid_outbound_egress_rules: true,
            ..Default::default()
        });
        let violations = validate_spec_against_policy(&spec, &pack);
        assert_eq!(violations.len(), 1);
        assert_eq!(violations[0].rule, "forbidOutboundEgressRules");
    }

    #[test]
    fn forbid_outbound_egress_passes_when_no_rules() {
        let spec = minimal_spec(); // no egress rules
        let pack = minimal_pack(PolicyRules {
            forbid_outbound_egress_rules: true,
            ..Default::default()
        });
        assert!(validate_spec_against_policy(&spec, &pack).is_empty());
    }

    #[test]
    fn allowed_egress_hosts_rejects_unlisted_host() {
        let mut spec = minimal_spec();
        spec.authority.egress_rules = Some(vec![EgressRule {
            host: "evil.example.com".into(),
            port: 443,
            protocol: None,
            dns_egress_justification: None,
        }]);
        let pack = minimal_pack(PolicyRules {
            allowed_egress_hosts: vec!["*.internal".into(), "api.github.com".into()],
            ..Default::default()
        });
        let violations = validate_spec_against_policy(&spec, &pack);
        assert_eq!(violations.len(), 1);
        assert_eq!(violations[0].rule, "allowedEgressHosts");
    }

    #[test]
    fn allowed_egress_hosts_accepts_wildcard_subdomain() {
        let mut spec = minimal_spec();
        spec.authority.egress_rules = Some(vec![EgressRule {
            host: "cache.internal".into(),
            port: 443,
            protocol: None,
            dns_egress_justification: None,
        }]);
        let pack = minimal_pack(PolicyRules {
            allowed_egress_hosts: vec!["*.internal".into()],
            ..Default::default()
        });
        assert!(validate_spec_against_policy(&spec, &pack).is_empty());
    }

    #[test]
    fn wildcard_subdomain_does_not_match_bare_domain() {
        // "*.internal" should NOT match "internal" itself.
        assert!(!host_matches_pattern("internal", "*.internal"));
        assert!(host_matches_pattern("foo.internal", "*.internal"));
    }

    #[test]
    fn require_runtime_secret_delivery_rejects_env_mode() {
        let spec = minimal_spec(); // default delivery = Env
        let pack = minimal_pack(PolicyRules {
            require_runtime_secret_delivery: true,
            ..Default::default()
        });
        let violations = validate_spec_against_policy(&spec, &pack);
        assert_eq!(violations.len(), 1);
        assert_eq!(violations[0].rule, "requireRuntimeSecretDelivery");
    }

    #[test]
    fn require_runtime_secret_delivery_accepts_broker_mode() {
        let mut spec = minimal_spec();
        spec.run = Some(RunSpec {
            argv: vec!["/usr/bin/true".into()],
            working_directory: None,
            timeout_ms: None,
            limits: None,
            secret_delivery: SecretDeliveryMode::RuntimeBroker,
        });
        let pack = minimal_pack(PolicyRules {
            require_runtime_secret_delivery: true,
            ..Default::default()
        });
        assert!(validate_spec_against_policy(&spec, &pack).is_empty());
    }

    #[test]
    fn require_resource_limits_rejects_spec_without_limits() {
        let spec = minimal_spec(); // no limits
        let pack = minimal_pack(PolicyRules {
            require_resource_limits: true,
            ..Default::default()
        });
        let violations = validate_spec_against_policy(&spec, &pack);
        assert_eq!(violations.len(), 1);
        assert_eq!(violations[0].rule, "requireResourceLimits");
    }

    #[test]
    fn require_resource_limits_passes_with_limits_set() {
        let mut spec = minimal_spec();
        spec.run = Some(RunSpec {
            argv: vec!["/usr/bin/true".into()],
            working_directory: None,
            timeout_ms: None,
            limits: Some(RunLimits {
                memory_max_bytes: Some(512 * 1024 * 1024),
                cpu_max: None,
                graceful_shutdown_seconds: None,
            }),
            secret_delivery: SecretDeliveryMode::Env,
        });
        let pack = minimal_pack(PolicyRules {
            require_resource_limits: true,
            ..Default::default()
        });
        assert!(validate_spec_against_policy(&spec, &pack).is_empty());
    }

    #[test]
    fn multiple_violations_are_all_reported() {
        // Spec violates: TTL too long + runtime secret delivery required.
        let spec = minimal_spec(); // ttl=300, delivery=Env
        let pack = minimal_pack(PolicyRules {
            max_lifetime_ttl_seconds: Some(60),
            require_runtime_secret_delivery: true,
            ..Default::default()
        });
        let violations = validate_spec_against_policy(&spec, &pack);
        assert_eq!(violations.len(), 2);
        let rules: Vec<&str> = violations.iter().map(|v| v.rule.as_str()).collect();
        assert!(rules.contains(&"maxLifetimeTtlSeconds"));
        assert!(rules.contains(&"requireRuntimeSecretDelivery"));
    }

    // ── flagDnsEgressWithoutAcknowledgment (SEC-15) ──────────────────────────

    #[test]
    fn dns_egress_flagged_when_rule_enabled() {
        let mut spec = minimal_spec();
        spec.authority.egress_rules = Some(vec![EgressRule {
            host: "ns.example.com".into(),
            port: 53,
            protocol: None,
            dns_egress_justification: None,
        }]);
        let pack = minimal_pack(PolicyRules {
            flag_dns_egress_without_acknowledgment: Some(true),
            ..Default::default()
        });
        let violations = validate_spec_against_policy(&spec, &pack);
        assert_eq!(violations.len(), 1);
        assert_eq!(violations[0].rule, "flagDnsEgressWithoutAcknowledgment");
        assert!(violations[0].message.contains("dns-acknowledged"));
    }

    #[test]
    fn dns_egress_not_flagged_when_protocol_acknowledged() {
        let mut spec = minimal_spec();
        spec.authority.egress_rules = Some(vec![EgressRule {
            host: "ns.example.com".into(),
            port: 53,
            protocol: Some("dns-acknowledged".into()),
            dns_egress_justification: None,
        }]);
        let pack = minimal_pack(PolicyRules {
            flag_dns_egress_without_acknowledgment: Some(true),
            ..Default::default()
        });
        assert!(validate_spec_against_policy(&spec, &pack).is_empty());
    }

    #[test]
    fn dns_egress_acknowledgment_is_case_insensitive() {
        let mut spec = minimal_spec();
        spec.authority.egress_rules = Some(vec![EgressRule {
            host: "ns.example.com".into(),
            port: 53,
            protocol: Some("DNS-Acknowledged".into()),
            dns_egress_justification: None,
        }]);
        let pack = minimal_pack(PolicyRules {
            flag_dns_egress_without_acknowledgment: Some(true),
            ..Default::default()
        });
        assert!(validate_spec_against_policy(&spec, &pack).is_empty());
    }

    #[test]
    fn dns_egress_not_checked_when_rule_disabled() {
        let mut spec = minimal_spec();
        spec.authority.egress_rules = Some(vec![EgressRule {
            host: "ns.example.com".into(),
            port: 53,
            protocol: None,
            dns_egress_justification: None,
        }]);
        // Rule unset (None) — the default — should not flag.
        let pack = minimal_pack(PolicyRules::default());
        assert!(validate_spec_against_policy(&spec, &pack).is_empty());
    }

    #[test]
    fn dns_egress_not_checked_when_rule_explicitly_false() {
        let mut spec = minimal_spec();
        spec.authority.egress_rules = Some(vec![EgressRule {
            host: "ns.example.com".into(),
            port: 53,
            protocol: None,
            dns_egress_justification: None,
        }]);
        let pack = minimal_pack(PolicyRules {
            flag_dns_egress_without_acknowledgment: Some(false),
            ..Default::default()
        });
        assert!(validate_spec_against_policy(&spec, &pack).is_empty());
    }

    #[test]
    fn dns_egress_rule_does_not_affect_non_dns_ports() {
        let mut spec = minimal_spec();
        spec.authority.egress_rules = Some(vec![EgressRule {
            host: "api.github.com".into(),
            port: 443,
            protocol: None,
            dns_egress_justification: None,
        }]);
        let pack = minimal_pack(PolicyRules {
            flag_dns_egress_without_acknowledgment: Some(true),
            ..Default::default()
        });
        assert!(validate_spec_against_policy(&spec, &pack).is_empty());
    }

    #[test]
    fn dns_egress_flagged_when_some_rules_acknowledged_but_not_all() {
        let mut spec = minimal_spec();
        spec.authority.egress_rules = Some(vec![
            EgressRule {
                host: "ns1.example.com".into(),
                port: 53,
                protocol: Some("dns-acknowledged".into()),
                dns_egress_justification: None,
            },
            EgressRule {
                host: "ns2.example.com".into(),
                port: 53,
                protocol: None,
                dns_egress_justification: None,
            },
        ]);
        let pack = minimal_pack(PolicyRules {
            flag_dns_egress_without_acknowledgment: Some(true),
            ..Default::default()
        });
        let violations = validate_spec_against_policy(&spec, &pack);
        assert_eq!(violations.len(), 1);
        assert_eq!(violations[0].rule, "flagDnsEgressWithoutAcknowledgment");
    }

    // ── FC-34b / FC-34e: SEC-15 ack gate is protocol-agnostic on port 53 ─
    //
    // The W1 audit (`docs/firecracker-dns-audit.md`) drift item 3 / FC-34e
    // posited a possible bypass: a spec with `port: 53, protocol: "tcp"`
    // might pass the SEC-15 acknowledgment gate, because the gate's wording
    // is DNS-flavoured ("dns-acknowledged") and could be read as UDP-only.
    //
    // Reading the actual evaluation at `policy.rs:361-371`, the gate
    // condition is `if rule.port == 53` — it keys off the port number and
    // is **completely indifferent** to the L4 protocol value. So any
    // port-53 rule (UDP, TCP, or anything else) is gated identically:
    // either every port-53 rule is `dns-acknowledged`, or the spec is
    // rejected. FC-34e is **resolved as finding A** (gate is protocol-
    // agnostic; no bypass exists). These tests pin that property so a
    // future refactor that splits the gate by protocol cannot silently
    // re-introduce a TCP/53 hole.

    #[test]
    fn dns_egress_ack_gate_covers_tcp_protocol() {
        // FC-34e finding A: a TCP/53 rule WITHOUT acknowledgment must be
        // rejected by the SEC-15 gate, exactly like UDP/53.
        let mut spec = minimal_spec();
        spec.authority.egress_rules = Some(vec![EgressRule {
            host: "1.1.1.1".into(),
            port: 53,
            protocol: Some("tcp".into()),
            dns_egress_justification: None,
        }]);
        let pack = minimal_pack(PolicyRules {
            flag_dns_egress_without_acknowledgment: Some(true),
            ..Default::default()
        });
        let violations = validate_spec_against_policy(&spec, &pack);
        assert_eq!(
            violations.len(),
            1,
            "TCP/53 without dns-acknowledged must violate the SEC-15 gate; \
             got: {violations:?}"
        );
        assert_eq!(violations[0].rule, "flagDnsEgressWithoutAcknowledgment");
    }

    #[test]
    fn dns_egress_ack_gate_admits_acknowledged_tcp_53() {
        // FC-34b: an explicit `protocol: "tcp"` rule on port 53 with
        // `dns-acknowledged` MUST be admitted — but per the SEC-15 gate's
        // current contract, "dns-acknowledged" is the literal protocol
        // marker. This test pins the operator path: declare a single rule
        // with `protocol: "dns-acknowledged"` and rely on the FC backend's
        // `udp dport 53 accept` mapping (the host nft layer treats
        // dns-acknowledged as UDP/53 — see
        // `cellos-host-firecracker/src/lib.rs:1715-1719`). DNS-over-TCP
        // (large responses, AXFR) is intentionally not granted by a single
        // port-53 rule; an operator who needs both UDP/53 and TCP/53 must
        // declare two separate rules and accept the SEC-15 gate on each.
        let mut spec = minimal_spec();
        spec.authority.egress_rules = Some(vec![EgressRule {
            host: "1.1.1.1".into(),
            port: 53,
            protocol: Some("dns-acknowledged".into()),
            dns_egress_justification: None,
        }]);
        let pack = minimal_pack(PolicyRules {
            flag_dns_egress_without_acknowledgment: Some(true),
            ..Default::default()
        });
        assert!(
            validate_spec_against_policy(&spec, &pack).is_empty(),
            "acknowledged port-53 rule must pass the SEC-15 gate"
        );
    }

    #[test]
    fn dns_egress_ack_gate_rejects_mixed_acknowledged_and_tcp_53() {
        // FC-34e finding A, mixed-rule edge case: a spec that declares
        // both an acknowledged UDP/53 rule AND a bare TCP/53 rule must
        // still be rejected, because the gate requires every port-53 rule
        // to carry `dns-acknowledged`. This pins protocol-agnosticism
        // even when the operator partially complies.
        let mut spec = minimal_spec();
        spec.authority.egress_rules = Some(vec![
            EgressRule {
                host: "1.1.1.1".into(),
                port: 53,
                protocol: Some("dns-acknowledged".into()),
                dns_egress_justification: None,
            },
            EgressRule {
                host: "8.8.8.8".into(),
                port: 53,
                protocol: Some("tcp".into()),
                dns_egress_justification: None,
            },
        ]);
        let pack = minimal_pack(PolicyRules {
            flag_dns_egress_without_acknowledgment: Some(true),
            ..Default::default()
        });
        let violations = validate_spec_against_policy(&spec, &pack);
        assert_eq!(
            violations.len(),
            1,
            "mixed ack+TCP/53 must violate the SEC-15 gate; got: {violations:?}"
        );
        assert_eq!(violations[0].rule, "flagDnsEgressWithoutAcknowledgment");
    }

    #[test]
    fn policy_violation_display_includes_rule_and_message() {
        let v = PolicyViolation {
            rule: "maxLifetimeTtlSeconds".into(),
            message: "300 exceeds 60".into(),
        };
        let s = v.to_string();
        assert!(s.contains("maxLifetimeTtlSeconds"));
        assert!(s.contains("300 exceeds 60"));
    }

    // ── requireDnsEgressJustification (SEC-15c / SEAM-2) ─────────────────────

    #[test]
    fn dns_justification_required_when_rule_enabled_and_acknowledged() {
        let mut spec = minimal_spec();
        spec.authority.egress_rules = Some(vec![EgressRule {
            host: "ns.example.com".into(),
            port: 53,
            protocol: Some("dns-acknowledged".into()),
            dns_egress_justification: None,
        }]);
        let pack = minimal_pack(PolicyRules {
            require_dns_egress_justification: Some(true),
            ..Default::default()
        });
        let violations = validate_spec_against_policy(&spec, &pack);
        assert_eq!(violations.len(), 1);
        assert_eq!(violations[0].rule, "requireDnsEgressJustification");
        assert!(violations[0].message.contains("dnsEgressJustification"));
    }

    #[test]
    fn dns_justification_satisfied_with_nonempty_string() {
        let mut spec = minimal_spec();
        spec.authority.egress_rules = Some(vec![EgressRule {
            host: "ns.example.com".into(),
            port: 53,
            protocol: Some("dns-acknowledged".into()),
            dns_egress_justification: Some("internal resolver at 10.0.0.1".into()),
        }]);
        let pack = minimal_pack(PolicyRules {
            require_dns_egress_justification: Some(true),
            ..Default::default()
        });
        assert!(validate_spec_against_policy(&spec, &pack).is_empty());
    }

    #[test]
    fn dns_justification_empty_string_rejected() {
        let mut spec = minimal_spec();
        spec.authority.egress_rules = Some(vec![EgressRule {
            host: "ns.example.com".into(),
            port: 53,
            protocol: Some("dns-acknowledged".into()),
            dns_egress_justification: Some("  ".into()),
        }]);
        let pack = minimal_pack(PolicyRules {
            require_dns_egress_justification: Some(true),
            ..Default::default()
        });
        let violations = validate_spec_against_policy(&spec, &pack);
        assert_eq!(violations.len(), 1);
        assert_eq!(violations[0].rule, "requireDnsEgressJustification");
    }

    #[test]
    fn dns_justification_not_required_when_rule_disabled() {
        let mut spec = minimal_spec();
        spec.authority.egress_rules = Some(vec![EgressRule {
            host: "ns.example.com".into(),
            port: 53,
            protocol: Some("dns-acknowledged".into()),
            dns_egress_justification: None,
        }]);
        // Rule unset (None) — the default — should not flag.
        let pack = minimal_pack(PolicyRules::default());
        let violations = validate_spec_against_policy(&spec, &pack);
        // No requireDnsEgressJustification violation; flagDnsEgress also unset.
        assert!(
            !violations
                .iter()
                .any(|v| v.rule == "requireDnsEgressJustification"),
            "unexpected requireDnsEgressJustification violation: {violations:?}"
        );
    }

    // ── P4-04: policy pack version compatibility ─────────────────────────────
    //
    // These tests exercise `check_policy_pack_version_compatibility` directly.
    // The integration end (downgrade rejection at admission via
    // `validate_policy_pack_document`) is covered by
    // `crates/cellos-core/tests/policy_pack_version_admission.rs`. The env-var
    // override path is also tested in the integration suite (Rust unit tests
    // share a process and would race on `CELLOS_POLICY_ALLOW_DOWNGRADE`).

    #[test]
    fn version_absent_is_accepted() {
        assert!(check_policy_pack_version_compatibility(None, false).is_ok());
    }

    #[test]
    fn version_at_floor_is_accepted() {
        assert!(check_policy_pack_version_compatibility(
            Some(MIN_SUPPORTED_POLICY_PACK_VERSION),
            false
        )
        .is_ok());
    }

    #[test]
    fn version_above_floor_is_accepted() {
        assert!(check_policy_pack_version_compatibility(Some("1.4.2"), false).is_ok());
        assert!(check_policy_pack_version_compatibility(Some("2.0.0"), false).is_ok());
    }

    #[test]
    fn version_with_prerelease_is_accepted() {
        // Pre-release is ignored for ordering; "1.0.0-rc.1" >= floor "1.0.0".
        assert!(check_policy_pack_version_compatibility(Some("1.0.0-rc.1"), false).is_ok());
    }

    #[test]
    fn malformed_version_is_rejected() {
        assert!(check_policy_pack_version_compatibility(Some("v1.0"), false).is_err());
        assert!(check_policy_pack_version_compatibility(Some("1.0"), false).is_err());
        assert!(check_policy_pack_version_compatibility(Some("01.00.00"), false).is_err());
        assert!(check_policy_pack_version_compatibility(Some(""), false).is_err());
    }

    #[test]
    fn document_validates_with_explicit_floor_version() {
        let mut doc = minimal_doc(PolicyRules::default());
        doc.spec.version = Some(MIN_SUPPORTED_POLICY_PACK_VERSION.into());
        assert!(validate_policy_pack_document(&doc).is_ok());
    }

    #[test]
    fn document_rejects_malformed_version() {
        let mut doc = minimal_doc(PolicyRules::default());
        doc.spec.version = Some("not-a-semver".into());
        assert!(validate_policy_pack_document(&doc).is_err());
    }

    // ── T11-3 — placement-scoped policy packs ──────────────────────────────
    //
    // A pack with `placement` constraints applies only to specs whose
    // placement matches every populated field of the scope. A pack without
    // `placement` is global and applies everywhere.

    fn pack_with_placement(rules: PolicyRules, placement: PlacementSpec) -> PolicyPackSpec {
        PolicyPackSpec {
            id: "scoped-policy".into(),
            description: None,
            version: None,
            placement: Some(placement),
            rules,
        }
    }

    fn spec_with_ttl_and_placement(
        ttl_seconds: u64,
        placement: Option<PlacementSpec>,
    ) -> ExecutionCellSpec {
        let mut s = minimal_spec();
        s.lifetime.ttl_seconds = ttl_seconds;
        s.placement = placement;
        s
    }

    #[test]
    fn placement_scoped_pack_applies_when_pool_matches() {
        // Pack scoped to pool "amd64" with a 60s TTL ceiling.
        let pack = pack_with_placement(
            PolicyRules {
                max_lifetime_ttl_seconds: Some(60),
                ..Default::default()
            },
            PlacementSpec {
                pool_id: Some("runner-pool-amd64".into()),
                kubernetes_namespace: None,
                queue_name: None,
            },
        );
        // Spec with 300s TTL on matching pool — pack applies, violation expected.
        let spec = spec_with_ttl_and_placement(
            300,
            Some(PlacementSpec {
                pool_id: Some("runner-pool-amd64".into()),
                kubernetes_namespace: None,
                queue_name: None,
            }),
        );
        let violations = validate_spec_against_policy(&spec, &pack);
        assert_eq!(violations.len(), 1, "scoped pack should apply on match");
        assert_eq!(violations[0].rule, "maxLifetimeTtlSeconds");
    }

    #[test]
    fn placement_scoped_pack_is_skipped_when_pool_differs() {
        let pack = pack_with_placement(
            PolicyRules {
                max_lifetime_ttl_seconds: Some(60),
                ..Default::default()
            },
            PlacementSpec {
                pool_id: Some("runner-pool-amd64".into()),
                kubernetes_namespace: None,
                queue_name: None,
            },
        );
        // Same offending spec but on a DIFFERENT pool — pack must not apply.
        let spec = spec_with_ttl_and_placement(
            300,
            Some(PlacementSpec {
                pool_id: Some("runner-pool-arm64".into()),
                kubernetes_namespace: None,
                queue_name: None,
            }),
        );
        let violations = validate_spec_against_policy(&spec, &pack);
        assert!(
            violations.is_empty(),
            "scoped pack must not apply to mismatched placement, got {violations:?}"
        );
    }

    #[test]
    fn unscoped_pack_applies_everywhere() {
        // No placement on the pack — global pack — must apply regardless.
        let pack = minimal_pack(PolicyRules {
            max_lifetime_ttl_seconds: Some(60),
            ..Default::default()
        });
        let spec_no_placement = spec_with_ttl_and_placement(300, None);
        let spec_with_pool = spec_with_ttl_and_placement(
            300,
            Some(PlacementSpec {
                pool_id: Some("runner-pool-amd64".into()),
                kubernetes_namespace: None,
                queue_name: None,
            }),
        );
        assert_eq!(
            validate_spec_against_policy(&spec_no_placement, &pack).len(),
            1,
            "unscoped pack must apply to specs without placement"
        );
        assert_eq!(
            validate_spec_against_policy(&spec_with_pool, &pack).len(),
            1,
            "unscoped pack must apply to specs with any placement"
        );
    }

    #[test]
    fn placement_scope_with_no_populated_fields_is_universal() {
        // An empty `PlacementSpec` (all fields None) on a pack means "apply
        // everywhere" — equivalent to `placement: None`.
        let pack = pack_with_placement(
            PolicyRules {
                max_lifetime_ttl_seconds: Some(60),
                ..Default::default()
            },
            PlacementSpec::default(),
        );
        let spec = spec_with_ttl_and_placement(300, None);
        let violations = validate_spec_against_policy(&spec, &pack);
        assert_eq!(violations.len(), 1, "empty scope must behave as universal");
    }

    #[test]
    fn scope_with_multiple_fields_requires_all_to_match() {
        // Scope demands both pool AND namespace — partial matches are misses.
        let pack = pack_with_placement(
            PolicyRules {
                max_lifetime_ttl_seconds: Some(60),
                ..Default::default()
            },
            PlacementSpec {
                pool_id: Some("runner-pool-amd64".into()),
                kubernetes_namespace: Some("cellos-prod".into()),
                queue_name: None,
            },
        );
        // Right pool, WRONG namespace → no application.
        let half_match = spec_with_ttl_and_placement(
            300,
            Some(PlacementSpec {
                pool_id: Some("runner-pool-amd64".into()),
                kubernetes_namespace: Some("cellos-staging".into()),
                queue_name: None,
            }),
        );
        assert!(validate_spec_against_policy(&half_match, &pack).is_empty());

        // Both match → pack applies.
        let full_match = spec_with_ttl_and_placement(
            300,
            Some(PlacementSpec {
                pool_id: Some("runner-pool-amd64".into()),
                kubernetes_namespace: Some("cellos-prod".into()),
                queue_name: None,
            }),
        );
        assert_eq!(validate_spec_against_policy(&full_match, &pack).len(), 1);
    }

    // ── T12: AuthorizationPolicy ────────────────────────────────────────────

    fn minimal_authz_doc(policy: AuthorizationPolicy) -> AuthorizationPolicyDocument {
        AuthorizationPolicyDocument {
            api_version: "cellos.io/v1".into(),
            kind: "AuthorizationPolicy".into(),
            spec: policy,
        }
    }

    #[test]
    fn authz_policy_valid_doc_passes() {
        let doc = minimal_authz_doc(AuthorizationPolicy {
            subjects: vec!["tenant:acme".into(), "oidc:github:foo/bar".into()],
            allowed_pools: vec!["pool-a".into()],
            allowed_policy_packs: vec!["strict-1".into()],
            max_cells_per_hour: Some(100),
        });
        assert!(validate_authorization_policy(&doc).is_ok());
    }

    #[test]
    fn authz_policy_empty_subjects_rejected() {
        let doc = minimal_authz_doc(AuthorizationPolicy {
            subjects: vec![],
            ..AuthorizationPolicy::default()
        });
        let err = validate_authorization_policy(&doc).expect_err("empty subjects must reject");
        assert!(
            err.to_string().contains("subjects must be non-empty"),
            "got: {err}"
        );
    }

    #[test]
    fn authz_policy_wrong_kind_rejected() {
        let mut doc = minimal_authz_doc(AuthorizationPolicy {
            subjects: vec!["tenant:acme".into()],
            ..AuthorizationPolicy::default()
        });
        doc.kind = "PolicyPack".into();
        assert!(validate_authorization_policy(&doc).is_err());
    }

    #[test]
    fn authz_policy_zero_rate_limit_rejected() {
        let doc = minimal_authz_doc(AuthorizationPolicy {
            subjects: vec!["tenant:acme".into()],
            max_cells_per_hour: Some(0),
            ..AuthorizationPolicy::default()
        });
        let err = validate_authorization_policy(&doc).expect_err("zero rate limit must reject");
        assert!(err.to_string().contains("maxCellsPerHour"), "got: {err}");
    }

    #[test]
    fn authz_policy_empty_pool_entry_rejected() {
        let doc = minimal_authz_doc(AuthorizationPolicy {
            subjects: vec!["tenant:acme".into()],
            allowed_pools: vec!["valid".into(), "  ".into()],
            ..AuthorizationPolicy::default()
        });
        assert!(validate_authorization_policy(&doc).is_err());
    }
}