cellos-supervisor 0.6.0-pre

//! SEC-25 Phase 2 — supervisor-side wiring of `CELLOS_TRUST_VERIFY_KEYS_PATH`
//! and `CELLOS_TRUST_KEYSET_PATH`.
//!
//! W2 SEC-25 Phase 1 shipped the dataplane verifier
//! `cellos_core::verify_signed_trust_keyset_envelope`. This module turns the
//! three operator-facing env vars into a startup-time trust-keyset
//! establishment step:
//!
//! | Env var                                | Required | Effect                                                                 |
//! |----------------------------------------|----------|------------------------------------------------------------------------|
//! | `CELLOS_TRUST_VERIFY_KEYS_PATH`        | optional | Operator's `kid → base64url-pubkey` JSON file (see `cellos_core::trust_keys`). |
//! | `CELLOS_TRUST_KEYSET_PATH`             | optional | Path to a `signed-trust-keyset-envelope-v1` document to verify.        |
//! | `CELLOS_REQUIRE_TRUST_VERIFY_KEYS=1`   | optional | Fail-closed: build_supervisor errors if the verifying-keys file is missing OR if the envelope does not verify. |
//!
//! When the verifying-keys path is unset and `REQUIRE` is unset, the
//! supervisor runs with an empty trust keyring (the legacy / zero-config
//! posture); any envelope reference will surface a `keyset_verification_failed`
//! event instead of failing startup.

use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;

use cellos_core::types::{OrgAuthorityCeilingManifestV1, AUTHORITY_CEILING_V1_PAYLOAD_TYPE};
use cellos_core::{
    cloud_event_v1_keyset_verification_failed, cloud_event_v1_keyset_verified,
    load_trust_verify_keys_file, ports::EventSink, verify_signed_trust_keyset_chain,
    verify_signed_trust_keyset_envelope, SignedTrustKeysetEnvelope, TrustAnchorPublicKey,
};
use serde_json::Value;

/// Outcome of the optional `CELLOS_TRUST_KEYSET_PATH` envelope verification.
///
/// Returned by [`load_and_verify_trust_keyset_from_env`]. The caller (typically
/// `build_supervisor`) is responsible for emitting the resulting CloudEvent
/// through whichever sink is wired up.
#[derive(Debug, Clone)]
pub enum KeysetLoadOutcome {
    /// `CELLOS_TRUST_KEYSET_PATH` was unset — no verification attempted.
    NotConfigured,
    /// Envelope verified successfully against the supplied keyring.
    Verified(KeysetVerifiedDetails),
    /// Envelope load or verification failed in fail-open mode (the supervisor
    /// continues to run; an event surfaces the degraded posture). Fail-closed
    /// errors are returned from the call as `Err` and never surface here.
    Failed(KeysetVerificationFailedDetails),
}

#[derive(Debug, Clone)]
pub struct KeysetVerifiedDetails {
    pub keyset_id: String,
    pub payload_digest: String,
    pub verified_signer_kid: String,
}

#[derive(Debug, Clone)]
pub struct KeysetVerificationFailedDetails {
    pub attempted_keyset_basename: String,
    pub reason: String,
}

/// Read `CELLOS_TRUST_VERIFY_KEYS_PATH` and parse the operator's keyring.
///
/// - Unset and `require_trust_verify_keys` is `false` → returns an empty map
///   (legacy / zero-config posture). The supervisor still runs; envelope
///   references surface `keyset_verification_failed` events.
/// - Unset and `require_trust_verify_keys` is `true` → returns an error.
/// - Set and load fails:
///   - `require_trust_verify_keys: true` → returns the load error.
///   - `require_trust_verify_keys: false` → returns the load error (fail-open
///     does not apply to the keyring file itself — only to the envelope
///     verification step). A misconfigured keys file is unambiguously an
///     operator-side bug; silently empty-keyring-falling-through would mask
///     it.
pub fn load_trust_verify_keys_from_env(
    require_trust_verify_keys: bool,
) -> Result<Arc<HashMap<String, TrustAnchorPublicKey>>, anyhow::Error> {
    match std::env::var_os("CELLOS_TRUST_VERIFY_KEYS_PATH") {
        None => {
            if require_trust_verify_keys {
                return Err(anyhow::anyhow!(
                    "CELLOS_REQUIRE_TRUST_VERIFY_KEYS is set but CELLOS_TRUST_VERIFY_KEYS_PATH is unset"
                ));
            }
            tracing::debug!(
                target: "cellos.supervisor.trust",
                "CELLOS_TRUST_VERIFY_KEYS_PATH unset; supervisor runs with empty trust keyring"
            );
            Ok(Arc::new(HashMap::new()))
        }
        Some(path_os) => {
            let path = PathBuf::from(&path_os);
            let keys = load_trust_verify_keys_file(&path).map_err(|e| {
                anyhow::anyhow!(
                    "CELLOS_TRUST_VERIFY_KEYS_PATH: cannot load '{}': {e}",
                    path.display()
                )
            })?;
            tracing::info!(
                target: "cellos.supervisor.trust",
                path = %path.display(),
                kid_count = keys.len(),
                "trust verifying keys loaded"
            );
            Ok(Arc::new(keys))
        }
    }
}

/// Read `CELLOS_TRUST_KEYSET_PATH` and verify the envelope against `keys`.
///
/// Behavior matrix:
///
/// | Env var unset | Outcome                                |
/// |---------------|----------------------------------------|
/// | yes           | [`KeysetLoadOutcome::NotConfigured`]   |
///
/// | Env var set, fail-closed (`require_trust_verify_keys=true`) | Outcome on failure |
/// |-------------------------------------------------------------|--------------------|
/// | file open / read error                                      | `Err`              |
/// | JSON parse error                                            | `Err`              |
/// | `verify_signed_trust_keyset_envelope` error                 | `Err`              |
///
/// | Env var set, fail-open (`require_trust_verify_keys=false`) | Outcome on failure |
/// |------------------------------------------------------------|--------------------|
/// | file open / read error                                     | [`KeysetLoadOutcome::Failed`] |
/// | JSON parse error                                           | [`KeysetLoadOutcome::Failed`] |
/// | `verify_signed_trust_keyset_envelope` error                | [`KeysetLoadOutcome::Failed`] |
///
/// On success, the supervisor decodes the inner `keysetId` from the raw
/// payload bytes (minimal `{"keysetId":"..."}` extraction — full keyset shape
/// validation is not the supervisor's job; the cellos-trustd sibling repo
/// owns that). If decoding the inner `keysetId` fails (the payload is not a
/// JSON object or has no string-typed `keysetId`), the envelope is still
/// considered verified at the signature level, but the event reports
/// `keysetId = "(unknown)"`.
pub fn load_and_verify_trust_keyset_from_env(
    keys: &HashMap<String, TrustAnchorPublicKey>,
    require_trust_verify_keys: bool,
    now: std::time::SystemTime,
) -> Result<KeysetLoadOutcome, anyhow::Error> {
    // scope: chain-mode wins when both env vars are set. The HEAD envelope
    // of the chain becomes the "verified envelope" for event-emit purposes;
    // the chain integrity check is logged additionally.
    if let Some(chain_path_os) = std::env::var_os("CELLOS_TRUST_KEYSET_CHAIN_PATH") {
        return load_and_verify_chain_from_env(chain_path_os, keys, require_trust_verify_keys, now);
    }

    let Some(path_os) = std::env::var_os("CELLOS_TRUST_KEYSET_PATH") else {
        return Ok(KeysetLoadOutcome::NotConfigured);
    };
    let path = PathBuf::from(&path_os);
    let basename = path
        .file_name()
        .map(|s| s.to_string_lossy().into_owned())
        .unwrap_or_else(|| "(unknown)".to_string());

    match attempt_load_and_verify(&path, keys, now) {
        Ok(details) => {
            tracing::info!(
                target: "cellos.supervisor.trust",
                keyset_id = %details.keyset_id,
                payload_digest = %details.payload_digest,
                verified_signer_kid = %details.verified_signer_kid,
                "signed trust keyset envelope verified at supervisor startup"
            );
            Ok(KeysetLoadOutcome::Verified(details))
        }
        Err(reason_string) => {
            if require_trust_verify_keys {
                return Err(anyhow::anyhow!(
                    "CELLOS_TRUST_KEYSET_PATH: verification failed under CELLOS_REQUIRE_TRUST_VERIFY_KEYS=1: {reason_string}"
                ));
            }
            tracing::warn!(
                target: "cellos.supervisor.trust",
                attempted_keyset_basename = %basename,
                reason = %reason_string,
                "signed trust keyset envelope verification failed; continuing in degraded mode"
            );
            Ok(KeysetLoadOutcome::Failed(KeysetVerificationFailedDetails {
                attempted_keyset_basename: basename,
                reason: reason_string,
            }))
        }
    }
}

/// scope: chain-mode loader — reads the comma-or-newline-separated list of
/// envelope paths from `CELLOS_TRUST_KEYSET_CHAIN_PATH` (oldest-first) and
/// verifies the chain via `verify_signed_trust_keyset_chain`.
///
/// On success, the HEAD envelope is what surfaces in the
/// `keyset_verified` CloudEvent (its `keysetId`, `payloadDigest`, and
/// verifying signer kid). The fact that an N-envelope chain was walked is
/// logged at the `cellos.supervisor.trust` target for operator audit; it
/// does not currently appear in the event payload.
///
/// Path-list parsing: splits on commas AND newlines; empty entries are
/// skipped (so `path1,,path2` parses as `[path1, path2]`).
fn load_and_verify_chain_from_env(
    chain_path_os: std::ffi::OsString,
    keys: &HashMap<String, TrustAnchorPublicKey>,
    require_trust_verify_keys: bool,
    now: std::time::SystemTime,
) -> Result<KeysetLoadOutcome, anyhow::Error> {
    let raw_str = chain_path_os.to_string_lossy().into_owned();
    let paths: Vec<PathBuf> = raw_str
        .split([',', '\n'])
        .map(str::trim)
        .filter(|s| !s.is_empty())
        .map(PathBuf::from)
        .collect();

    if paths.is_empty() {
        let reason = "CELLOS_TRUST_KEYSET_CHAIN_PATH set but parsed no envelope paths".to_string();
        if require_trust_verify_keys {
            return Err(anyhow::anyhow!(
                "CELLOS_TRUST_KEYSET_CHAIN_PATH: verification failed under CELLOS_REQUIRE_TRUST_VERIFY_KEYS=1: {reason}"
            ));
        }
        tracing::warn!(
            target: "cellos.supervisor.trust",
            reason = %reason,
            "signed trust keyset chain verification failed; continuing in degraded mode"
        );
        return Ok(KeysetLoadOutcome::Failed(KeysetVerificationFailedDetails {
            attempted_keyset_basename: "(empty chain)".into(),
            reason,
        }));
    }

    // Use the HEAD envelope's basename for event emission — that's the one
    // an operator will compare against their rotation publish step.
    let head_basename = paths
        .last()
        .and_then(|p| p.file_name())
        .map(|s| s.to_string_lossy().into_owned())
        .unwrap_or_else(|| "(unknown)".to_string());

    match attempt_load_and_verify_chain(&paths, keys, now) {
        Ok(details) => {
            tracing::info!(
                target: "cellos.supervisor.trust",
                envelope_count = paths.len(),
                keyset_id = %details.keyset_id,
                payload_digest = %details.payload_digest,
                verified_signer_kid = %details.verified_signer_kid,
                "verified {}-envelope chain (head digest: {})",
                paths.len(),
                details.payload_digest
            );
            Ok(KeysetLoadOutcome::Verified(details))
        }
        Err(reason_string) => {
            if require_trust_verify_keys {
                return Err(anyhow::anyhow!(
                    "CELLOS_TRUST_KEYSET_CHAIN_PATH: verification failed under CELLOS_REQUIRE_TRUST_VERIFY_KEYS=1: {reason_string}"
                ));
            }
            tracing::warn!(
                target: "cellos.supervisor.trust",
                attempted_keyset_basename = %head_basename,
                envelope_count = paths.len(),
                reason = %reason_string,
                "signed trust keyset chain verification failed; continuing in degraded mode"
            );
            Ok(KeysetLoadOutcome::Failed(KeysetVerificationFailedDetails {
                attempted_keyset_basename: head_basename,
                reason: reason_string,
            }))
        }
    }
}

/// Internal: parse every envelope file in `paths`, run the chain verifier,
/// and surface the HEAD envelope's verification details on success.
fn attempt_load_and_verify_chain(
    paths: &[PathBuf],
    keys: &HashMap<String, TrustAnchorPublicKey>,
    now: std::time::SystemTime,
) -> Result<KeysetVerifiedDetails, String> {
    let mut chain: Vec<SignedTrustKeysetEnvelope> = Vec::with_capacity(paths.len());
    for path in paths {
        let raw =
            read_envelope_file(path).map_err(|e| format!("cannot read {}: {e}", path.display()))?;
        let envelope: SignedTrustKeysetEnvelope = serde_json::from_str(&raw)
            .map_err(|e| format!("JSON parse error in {}: {e}", path.display()))?;
        chain.push(envelope);
    }

    let head_payload_bytes =
        verify_signed_trust_keyset_chain(&chain, keys, now).map_err(|e| format!("{e}"))?;

    // The chain verifier already proved each envelope (including HEAD)
    // verifies; we re-walk HEAD's signatures only to surface the winning kid
    // for the event, mirroring the single-envelope path.
    let head_envelope = chain.last().expect("chain non-empty");
    let verified_signer_kid =
        pick_verified_signer_kid(head_envelope, &head_payload_bytes, keys, now)
            .unwrap_or_else(|| "(unknown)".to_string());
    let keyset_id =
        decode_inner_keyset_id(&head_payload_bytes).unwrap_or_else(|| "(unknown)".into());

    Ok(KeysetVerifiedDetails {
        keyset_id,
        payload_digest: head_envelope.payload_digest.clone(),
        verified_signer_kid,
    })
}

/// Internal: open + parse + verify an envelope at `path`. Returns the
/// `KeysetVerifiedDetails` on success or a stringified reason on any failure.
fn attempt_load_and_verify(
    path: &Path,
    keys: &HashMap<String, TrustAnchorPublicKey>,
    now: std::time::SystemTime,
) -> Result<KeysetVerifiedDetails, String> {
    let raw =
        read_envelope_file(path).map_err(|e| format!("cannot read {}: {e}", path.display()))?;

    let envelope: SignedTrustKeysetEnvelope = serde_json::from_str(&raw)
        .map_err(|e| format!("JSON parse error in {}: {e}", path.display()))?;

    let payload_bytes =
        verify_signed_trust_keyset_envelope(&envelope, keys, now).map_err(|e| format!("{e}"))?;

    // Find which signature actually verified — we need to surface it in the
    // event. The verifier returns the raw payload bytes on success but does
    // not tell us the kid that won; we re-walk the signatures here using the
    // same predicate the verifier did.
    let verified_signer_kid = pick_verified_signer_kid(&envelope, &payload_bytes, keys, now)
        .unwrap_or_else(|| "(unknown)".to_string());

    let keyset_id = decode_inner_keyset_id(&payload_bytes).unwrap_or_else(|| "(unknown)".into());

    Ok(KeysetVerifiedDetails {
        keyset_id,
        payload_digest: envelope.payload_digest.clone(),
        verified_signer_kid,
    })
}

/// Re-walk the envelope signatures to find the kid whose signature verified.
///
/// Mirrors the verifier's per-signature loop in
/// `cellos_core::verify_signed_trust_keyset_envelope`: ed25519 algorithm,
/// known kid, optional `notBefore`/`notAfter` window check (delegated to the
/// dalek primitive for now — the dalek `Signature::from_bytes` accepts any
/// 64-byte input), `verify_strict` over the raw payload bytes.
fn pick_verified_signer_kid(
    envelope: &SignedTrustKeysetEnvelope,
    payload_bytes: &[u8],
    keys: &HashMap<String, TrustAnchorPublicKey>,
    now: std::time::SystemTime,
) -> Option<String> {
    use base64::engine::general_purpose::URL_SAFE_NO_PAD;
    use base64::Engine as _;
    use chrono::DateTime;

    for sig in &envelope.signatures {
        if sig.algorithm != "ed25519" {
            continue;
        }
        let Some(verifying_key) = keys.get(&sig.signer_kid) else {
            continue;
        };
        if !window_contains(now, sig.not_before.as_deref(), sig.not_after.as_deref()) {
            continue;
        }
        let sig_b64 = sig.signature.trim_end_matches('=');
        let Ok(sig_bytes) = URL_SAFE_NO_PAD.decode(sig_b64) else {
            continue;
        };
        if sig_bytes.len() != 64 {
            continue;
        }
        // Strict Ed25519 verify through the crypto provider port (ADR-0027).
        if cellos_core::provider()
            .verify_ed25519(verifying_key.as_bytes(), payload_bytes, &sig_bytes)
            .is_ok()
        {
            return Some(sig.signer_kid.clone());
        }
    }
    // Suppress unused-import warning when neither path is exercised.
    let _ = DateTime::<chrono::Utc>::from_timestamp(0, 0);
    None
}

/// Mirrors `cellos_core::spec_validation::signature_window_contains` (private).
fn window_contains(
    now: std::time::SystemTime,
    not_before: Option<&str>,
    not_after: Option<&str>,
) -> bool {
    use chrono::DateTime;
    let now_chrono: DateTime<chrono::Utc> = now.into();
    if let Some(nb) = not_before {
        match DateTime::parse_from_rfc3339(nb) {
            Ok(t) => {
                if now_chrono < t.with_timezone(&chrono::Utc) {
                    return false;
                }
            }
            Err(_) => return false,
        }
    }
    if let Some(na) = not_after {
        match DateTime::parse_from_rfc3339(na) {
            Ok(t) => {
                if now_chrono > t.with_timezone(&chrono::Utc) {
                    return false;
                }
            }
            Err(_) => return false,
        }
    }
    true
}

/// Pull the inner keyset's `keysetId` out of the raw payload bytes via a
/// minimal `{ "keysetId": "..." }` parse — full schema validation is the
/// trustd sibling repo's job.
fn decode_inner_keyset_id(payload_bytes: &[u8]) -> Option<String> {
    let v: Value = serde_json::from_slice(payload_bytes).ok()?;
    v.as_object()?.get("keysetId")?.as_str().map(String::from)
}

/// Reads `path` with `O_NOFOLLOW` on Unix (matching `composition.rs`'s
/// `CELLOS_POLICY_PACK_PATH` / `CELLOS_AUTHORITY_KEYS_PATH` policy — SEC-15b).
fn read_envelope_file(path: &Path) -> Result<String, std::io::Error> {
    #[cfg(unix)]
    {
        use std::io::Read;
        use std::os::unix::fs::OpenOptionsExt;
        let mut opts = std::fs::OpenOptions::new();
        opts.read(true);
        opts.custom_flags(libc::O_RDONLY | libc::O_NOFOLLOW);
        let mut file = opts.open(path)?;
        let mut buf = String::new();
        file.read_to_string(&mut buf)?;
        Ok(buf)
    }
    #[cfg(not(unix))]
    {
        // S37: best-effort Windows symlink-swap defense (TOCTOU residual) — the
        // same reparse-point refusal cellos-core applies to trust-verify keys.
        #[cfg(windows)]
        cellos_core::trust_keys::reject_reparse_point(path)
            .map_err(|e| std::io::Error::other(e.to_string()))?;
        std::fs::read_to_string(path)
    }
}

/// Convenience: emit the [`KeysetLoadOutcome`] as a single CloudEvent through
/// `event_sink`. No-op for `KeysetLoadOutcome::NotConfigured`.
///
/// `now` is the timestamp embedded in the `verifiedAt` / `failedAt` field —
/// passed in for testability (unit tests pin it to a known value).
///
/// Errors from sink emit are surfaced as `anyhow::Error`. Phase 2 callers
/// today pass `?` — sink failures during startup-time keyset establishment
/// are operator-visible (a misbehaving JetStream connection still surfaces).
pub async fn emit_keyset_outcome(
    outcome: &KeysetLoadOutcome,
    event_sink: &Arc<dyn EventSink>,
    jsonl_sink: Option<&Arc<dyn EventSink>>,
    now: chrono::DateTime<chrono::Utc>,
) -> Result<(), anyhow::Error> {
    let timestamp = now.to_rfc3339_opts(chrono::SecondsFormat::Secs, true);
    match outcome {
        KeysetLoadOutcome::NotConfigured => Ok(()),
        KeysetLoadOutcome::Verified(details) => {
            let envelope = cloud_event_v1_keyset_verified(
                "cellos-supervisor",
                &timestamp,
                &details.keyset_id,
                &details.payload_digest,
                &details.verified_signer_kid,
                &timestamp,
                None,
            )?;
            event_sink
                .emit(&envelope)
                .await
                .map_err(|e| anyhow::anyhow!("emit keyset_verified: {e}"))?;
            if let Some(secondary) = jsonl_sink {
                secondary
                    .emit(&envelope)
                    .await
                    .map_err(|e| anyhow::anyhow!("emit keyset_verified to jsonl sink: {e}"))?;
            }
            Ok(())
        }
        KeysetLoadOutcome::Failed(details) => {
            let envelope = cloud_event_v1_keyset_verification_failed(
                "cellos-supervisor",
                &timestamp,
                &details.attempted_keyset_basename,
                &details.reason,
                &timestamp,
                None,
            )?;
            event_sink
                .emit(&envelope)
                .await
                .map_err(|e| anyhow::anyhow!("emit keyset_verification_failed: {e}"))?;
            if let Some(secondary) = jsonl_sink {
                secondary.emit(&envelope).await.map_err(|e| {
                    anyhow::anyhow!("emit keyset_verification_failed to jsonl sink: {e}")
                })?;
            }
            Ok(())
        }
    }
}

// ── ADR-0024: org-root authority-ceiling manifest loader ────────────────────
//
// Mirrors `load_and_verify_trust_keyset_from_env` semantics for the ceiling
// manifest: read `CELLOS_AUTHORITY_CEILING_PATH`, verify the envelope with the
// SAME payload-agnostic verifier and the SAME org-root verifying keys, then
// — because the verifier accepts the closed two-member sibling media-type set
// (T02) — **re-assert** the payload_type equals the ceiling sibling constant so
// a validly-signed trust-keyset envelope presented where a ceiling is expected
// is rejected. The inner `OrgAuthorityCeilingManifestV1` shape is parsed here,
// per the "verifier proves origin, caller parses shape" SEC-25 seam.
//
// Fail-closed gate (`require_authority_ceiling`, auto-set by the hardened
// profile in `main.rs`): unset + not-required => `NotConfigured` (the labeled
// fail-OPEN path; the caller turns it into the signed `ceiling.degraded`
// receipt and proceeds); set / hardened + missing-or-invalid => `Err`.
//
// `ceilingEpoch` monotonicity is enforced against an on-disk freshness anchor
// file (`CELLOS_AUTHORITY_CEILING_ANCHOR_PATH`) so a stale-but-validly-signed
// manifest is rejected across separate supervisor processes (ADR-0024
// §Manifest replay). The anchor records the highest accepted epoch; a manifest
// whose `ceilingEpoch` is below it is `ceiling_epoch_stale`.

/// Outcome of the optional `CELLOS_AUTHORITY_CEILING_PATH` envelope
/// verification, returned by [`load_and_verify_ceiling_from_env`].
///
/// Mirrors [`KeysetLoadOutcome`]: fail-closed failures are returned as `Err`
/// from the call and never surface here. `NotConfigured` is the labeled
/// fail-OPEN path (ADR-0024 §Fail-closed posture) — the caller emits a single
/// signed degraded receipt and proceeds; it is reachable only when the gate is
/// **not** required (unset env, not hardened).
#[derive(Debug, Clone)]
pub enum CeilingLoadOutcome {
    /// `CELLOS_AUTHORITY_CEILING_PATH` was unset and the gate is not required —
    /// no verification attempted. The labeled fail-open path.
    NotConfigured,
    /// Envelope verified, payload_type re-asserted, inner manifest parsed, and
    /// the `ceilingEpoch` freshness check passed against the on-disk anchor.
    Verified(CeilingVerifiedDetails),
}

/// The verified ceiling manifest plus the envelope-level provenance the
/// admission gate (T07) needs to populate the [`crate`]-side admission verdict.
#[derive(Debug, Clone)]
pub struct CeilingVerifiedDetails {
    /// The parsed inner manifest carrying the named `AuthorityCapability`
    /// ceilings the gate compares `spec.authority` against.
    pub manifest: OrgAuthorityCeilingManifestV1,
    /// `sha256:<hex>` over the raw decoded payload bytes (the envelope's
    /// `payloadDigest`, surfaced into the verdict's `manifestDigest`).
    pub manifest_digest: String,
    /// The `kid` of the org-root signature that verified.
    pub verified_signer_kid: String,
}

/// Read `CELLOS_AUTHORITY_CEILING_PATH` and verify the ceiling-manifest
/// envelope against `keys`.
///
/// Behavior matrix (mirrors [`load_and_verify_trust_keyset_from_env`]):
///
/// | Env var unset, `require_authority_ceiling=false` | Outcome                    |
/// |--------------------------------------------------|----------------------------|
/// | yes                                              | [`CeilingLoadOutcome::NotConfigured`] (labeled fail-open) |
///
/// | Env var unset, `require_authority_ceiling=true` (hardened) | Outcome      |
/// |------------------------------------------------------------|--------------|
/// | yes                                                        | `Err`        |
///
/// | Env var set (any require value) — failure cause            | Outcome      |
/// |------------------------------------------------------------|--------------|
/// | file open / read error                                     | `Err`        |
/// | JSON parse error (envelope)                                | `Err`        |
/// | `verify_signed_trust_keyset_envelope` error (sig/digest/threshold/window) | `Err` |
/// | verified payload_type != ceiling sibling constant          | `Err`        |
/// | inner manifest unparseable                                 | `Err`        |
/// | `ceilingEpoch` below the on-disk freshness anchor          | `Err`        |
///
/// Once the env var is set, **every** failure is `Err` regardless of the
/// require flag: a ceiling manifest an operator has explicitly pointed the
/// supervisor at but that does not verify is unambiguously misconfigured, and
/// fail-open does not apply to it (the only fail-open path is the unset +
/// not-required `NotConfigured` arm above). The require flag governs the
/// **unset** case only: unset + not-required proceeds degraded; unset +
/// required aborts.
///
/// On success, the anchor file is advanced to the manifest's `ceilingEpoch`
/// (write-through) so a later process cannot accept an older epoch.
pub fn load_and_verify_ceiling_from_env(
    keys: &HashMap<String, TrustAnchorPublicKey>,
    require_authority_ceiling: bool,
    now: std::time::SystemTime,
) -> Result<CeilingLoadOutcome, anyhow::Error> {
    let Some(path_os) = std::env::var_os("CELLOS_AUTHORITY_CEILING_PATH") else {
        if require_authority_ceiling {
            return Err(anyhow::anyhow!(
                "CELLOS_REQUIRE_AUTHORITY_CEILING is set (or hardened profile) but \
                 CELLOS_AUTHORITY_CEILING_PATH is unset"
            ));
        }
        tracing::warn!(
            target: "cellos.supervisor.ceiling",
            "CELLOS_AUTHORITY_CEILING_PATH unset and ceiling not required; admission proceeds \
             WITHOUT an org-root authority ceiling (labeled fail-open path)"
        );
        return Ok(CeilingLoadOutcome::NotConfigured);
    };

    let path = PathBuf::from(&path_os);
    let details = attempt_load_and_verify_ceiling(&path, keys, now)
        .map_err(|reason| anyhow::anyhow!("CELLOS_AUTHORITY_CEILING_PATH: {reason}"))?;

    tracing::info!(
        target: "cellos.supervisor.ceiling",
        ceiling_id = %details.manifest.ceiling_id,
        ceiling_epoch = details.manifest.ceiling_epoch,
        manifest_digest = %details.manifest_digest,
        verified_signer_kid = %details.verified_signer_kid,
        ceilings = details.manifest.ceilings.len(),
        "org-root authority-ceiling manifest verified at supervisor startup"
    );
    Ok(CeilingLoadOutcome::Verified(details))
}

/// Internal: open + parse + verify the ceiling envelope at `path`, re-assert
/// the ceiling media type on the verified payload, parse the inner manifest,
/// and enforce `ceilingEpoch` monotonicity against the on-disk anchor. Returns
/// the verified details or a stringified reason on any failure.
fn attempt_load_and_verify_ceiling(
    path: &Path,
    keys: &HashMap<String, TrustAnchorPublicKey>,
    now: std::time::SystemTime,
) -> Result<CeilingVerifiedDetails, String> {
    let raw =
        read_envelope_file(path).map_err(|e| format!("cannot read {}: {e}", path.display()))?;

    let envelope: SignedTrustKeysetEnvelope = serde_json::from_str(&raw)
        .map_err(|e| format!("JSON parse error in {}: {e}", path.display()))?;

    // SEC-25 verifier proves origin over the raw bytes (digest, Ed25519, time
    // window, N-of-M threshold). It accepts the closed two-member sibling
    // media-type set, so we MUST re-assert the exact ceiling media type below.
    let payload_bytes =
        verify_signed_trust_keyset_envelope(&envelope, keys, now).map_err(|e| format!("{e}"))?;

    // Re-assert the verified payload is a ceiling, not a validly-signed
    // trust-keyset envelope (ADR-0024 Decision 1 type-confusion mitigation).
    if envelope.payload_type != AUTHORITY_CEILING_V1_PAYLOAD_TYPE {
        return Err(format!(
            "verified envelope payloadType is '{}', expected ceiling sibling '{}' \
             (a validly-signed trust-keyset is not a ceiling)",
            envelope.payload_type, AUTHORITY_CEILING_V1_PAYLOAD_TYPE
        ));
    }

    let manifest: OrgAuthorityCeilingManifestV1 = serde_json::from_slice(&payload_bytes)
        .map_err(|e| format!("inner ceiling manifest is unparseable: {e}"))?;

    // Replay bound across fresh supervisor processes: the anchor records the
    // highest epoch ever accepted; a manifest below it is stale (ADR-0024
    // §Manifest replay). When no anchor path is configured the freshness check
    // is skipped — the residual replay risk is named in the ADR.
    enforce_ceiling_epoch_freshness(manifest.ceiling_epoch)?;

    let verified_signer_kid = pick_verified_signer_kid(&envelope, &payload_bytes, keys, now)
        .unwrap_or_else(|| "(unknown)".to_string());

    Ok(CeilingVerifiedDetails {
        manifest,
        manifest_digest: envelope.payload_digest.clone(),
        verified_signer_kid,
    })
}

/// Enforce `ceilingEpoch` monotonicity against the on-disk freshness anchor at
/// `CELLOS_AUTHORITY_CEILING_ANCHOR_PATH`.
///
/// The anchor is a tiny operator-managed file (alongside the keys) holding the
/// single highest epoch the supervisor has ever accepted, as a decimal `u64`.
/// Semantics:
///
/// - Anchor path **unset** → the freshness check is skipped (the manifest's own
///   `notBefore`/`notAfter` window, verified by the envelope verifier, is the
///   only replay bound). The residual replay risk is named in ADR-0024.
/// - Anchor file **absent or empty** → first observation; any epoch is
///   accepted and the anchor is initialised to it.
/// - Anchor file present with a recorded epoch `a`:
///   - `ceiling_epoch < a` → `Err` (`ceiling_epoch_stale`).
///   - `ceiling_epoch >= a` → accepted; the anchor is advanced to
///     `ceiling_epoch` (write-through, monotonic).
///
/// An anchor file that exists but cannot be read or holds non-`u64` content is
/// an operator-side bug and fails closed (`Err`) rather than silently resetting
/// the replay floor.
fn enforce_ceiling_epoch_freshness(ceiling_epoch: u64) -> Result<(), String> {
    let Some(anchor_os) = std::env::var_os("CELLOS_AUTHORITY_CEILING_ANCHOR_PATH") else {
        tracing::warn!(
            target: "cellos.supervisor.ceiling",
            ceiling_epoch,
            "CELLOS_AUTHORITY_CEILING_ANCHOR_PATH unset; ceilingEpoch replay floor NOT enforced \
             across processes (residual replay risk — see ADR-0024 §Manifest replay)"
        );
        return Ok(());
    };
    let anchor_path = PathBuf::from(&anchor_os);

    let recorded = read_ceiling_epoch_anchor(&anchor_path)?;
    if let Some(anchor_epoch) = recorded {
        if ceiling_epoch < anchor_epoch {
            return Err(format!(
                "ceiling_epoch_stale: ceilingEpoch {ceiling_epoch} is below the freshness anchor \
                 floor {anchor_epoch} at {}",
                anchor_path.display()
            ));
        }
    }

    // Monotonic write-through. Only ever raises the floor: when the new epoch
    // equals the anchor we still rewrite (idempotent) so the anchor is created
    // on first observation.
    std::fs::write(&anchor_path, ceiling_epoch.to_string()).map_err(|e| {
        format!(
            "cannot advance ceiling freshness anchor at {}: {e}",
            anchor_path.display()
        )
    })?;
    Ok(())
}

/// Read the recorded epoch from the freshness anchor file. Returns `Ok(None)`
/// when the file is absent or empty (first observation); `Err` when it exists
/// but cannot be read or parsed (operator-side bug → fail closed).
fn read_ceiling_epoch_anchor(anchor_path: &Path) -> Result<Option<u64>, String> {
    let raw = match std::fs::read_to_string(anchor_path) {
        Ok(s) => s,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
        Err(e) => {
            return Err(format!(
                "cannot read ceiling freshness anchor at {}: {e}",
                anchor_path.display()
            ));
        }
    };
    let trimmed = raw.trim();
    if trimmed.is_empty() {
        return Ok(None);
    }
    let epoch: u64 = trimmed.parse().map_err(|e| {
        format!(
            "ceiling freshness anchor at {} holds non-u64 content '{trimmed}': {e}",
            anchor_path.display()
        )
    })?;
    Ok(Some(epoch))
}

#[cfg(test)]
mod tests {
    //! Unit-level smoke tests for the env-driven loaders. Full integration
    //! coverage (env-flag gating, event emission through a real sink) lives in
    //! `tests/supervisor_trust_keyset_verify.rs`.
    use super::{
        load_and_verify_trust_keyset_from_env, load_trust_verify_keys_from_env, KeysetLoadOutcome,
    };
    use std::sync::{Mutex, MutexGuard};

    static ENV_MUTEX: Mutex<()> = Mutex::new(());

    fn lock_env() -> MutexGuard<'static, ()> {
        ENV_MUTEX.lock().unwrap_or_else(|e| e.into_inner())
    }

    #[test]
    fn unset_path_with_require_unset_yields_empty_map() {
        let _guard = lock_env();
        std::env::remove_var("CELLOS_TRUST_VERIFY_KEYS_PATH");
        let keys = load_trust_verify_keys_from_env(false).expect("legacy posture");
        assert!(keys.is_empty());
    }

    #[test]
    fn unset_path_with_require_set_errors() {
        let _guard = lock_env();
        std::env::remove_var("CELLOS_TRUST_VERIFY_KEYS_PATH");
        let err = load_trust_verify_keys_from_env(true).expect_err("require set + unset path");
        assert!(format!("{err}").contains("CELLOS_TRUST_VERIFY_KEYS_PATH"));
    }

    #[test]
    fn keyset_path_unset_returns_not_configured() {
        let _guard = lock_env();
        std::env::remove_var("CELLOS_TRUST_KEYSET_PATH");
        let outcome = load_and_verify_trust_keyset_from_env(
            &Default::default(),
            false,
            std::time::SystemTime::now(),
        )
        .expect("unset path + fail-open");
        assert!(matches!(outcome, KeysetLoadOutcome::NotConfigured));
    }

    // ── ADR-0024 ceiling-manifest loader (T03) ──────────────────────────────
    use super::{
        load_and_verify_ceiling_from_env, CeilingLoadOutcome, AUTHORITY_CEILING_V1_PAYLOAD_TYPE,
    };
    use base64::engine::general_purpose::URL_SAFE_NO_PAD;
    use base64::Engine as _;
    use cellos_core::types::TRUST_KEYSET_V1_PAYLOAD_TYPE;
    use cellos_core::{SignedTrustKeysetEnvelope, TrustAnchorPublicKey, TrustKeysetSignature};
    use ed25519_dalek::{Signer as _, SigningKey};
    use sha2::{Digest, Sha256};
    use std::collections::HashMap;

    const CEILING_KID: &str = "org-root";

    fn org_root_signing_key() -> SigningKey {
        SigningKey::from_bytes(&[0x24; 32])
    }

    fn org_root_keyring() -> HashMap<String, TrustAnchorPublicKey> {
        let mut keys = HashMap::new();
        keys.insert(
            CEILING_KID.to_string(),
            TrustAnchorPublicKey::from_bytes_unchecked(
                org_root_signing_key().verifying_key().to_bytes(),
            ),
        );
        keys
    }

    fn sha256_prefixed(bytes: &[u8]) -> String {
        let digest = Sha256::digest(bytes);
        let mut s = String::from("sha256:");
        for byte in digest {
            use std::fmt::Write as _;
            write!(s, "{byte:02x}").expect("hex write");
        }
        s
    }

    /// Minimal valid ceiling manifest JSON (empty ceilings list = deny-all,
    /// which is irrelevant to the loader — the loader only proves origin,
    /// re-asserts the media type, parses the shape, and checks epoch freshness).
    fn ceiling_manifest_json(epoch: u64) -> Vec<u8> {
        serde_json::to_vec(&serde_json::json!({
            "schemaVersion": "1.0.0",
            "ceilingId": "ceiling-test",
            "ceilingEpoch": epoch,
            "ceilings": [],
        }))
        .expect("serialize ceiling manifest")
    }

    /// Build a signed envelope wrapping `payload_bytes` under `payload_type`.
    fn build_signed_envelope(
        payload_bytes: &[u8],
        payload_type: &str,
    ) -> SignedTrustKeysetEnvelope {
        let key = org_root_signing_key();
        let signature = key.sign(payload_bytes);
        SignedTrustKeysetEnvelope {
            schema_version: "1.0.0".into(),
            payload_type: payload_type.into(),
            payload: URL_SAFE_NO_PAD.encode(payload_bytes),
            signatures: vec![TrustKeysetSignature {
                signer_kid: CEILING_KID.into(),
                algorithm: "ed25519".into(),
                signature: URL_SAFE_NO_PAD.encode(signature.to_bytes()),
                not_before: None,
                not_after: None,
            }],
            payload_digest: sha256_prefixed(payload_bytes),
            produced_at: "2026-06-01T00:00:00Z".into(),
            replaces_envelope_digest: None,
            required_signer_count: None,
        }
    }

    fn write_envelope(
        dir: &std::path::Path,
        envelope: &SignedTrustKeysetEnvelope,
    ) -> std::path::PathBuf {
        let path = dir.join("ceiling.json");
        std::fs::write(
            &path,
            serde_json::to_vec(envelope).expect("serialize envelope"),
        )
        .expect("write envelope");
        path
    }

    /// Snapshot + restore the ceiling env vars these tests mutate.
    struct CeilingEnvGuard {
        path: Option<std::ffi::OsString>,
        anchor: Option<std::ffi::OsString>,
    }

    impl CeilingEnvGuard {
        fn new() -> Self {
            let g = Self {
                path: std::env::var_os("CELLOS_AUTHORITY_CEILING_PATH"),
                anchor: std::env::var_os("CELLOS_AUTHORITY_CEILING_ANCHOR_PATH"),
            };
            std::env::remove_var("CELLOS_AUTHORITY_CEILING_PATH");
            std::env::remove_var("CELLOS_AUTHORITY_CEILING_ANCHOR_PATH");
            g
        }
    }

    impl Drop for CeilingEnvGuard {
        fn drop(&mut self) {
            match self.path.take() {
                Some(v) => std::env::set_var("CELLOS_AUTHORITY_CEILING_PATH", v),
                None => std::env::remove_var("CELLOS_AUTHORITY_CEILING_PATH"),
            }
            match self.anchor.take() {
                Some(v) => std::env::set_var("CELLOS_AUTHORITY_CEILING_ANCHOR_PATH", v),
                None => std::env::remove_var("CELLOS_AUTHORITY_CEILING_ANCHOR_PATH"),
            }
        }
    }

    #[test]
    fn ceiling_unset_not_required_returns_not_configured() {
        let _guard = lock_env();
        let _env = CeilingEnvGuard::new();
        let outcome =
            load_and_verify_ceiling_from_env(&HashMap::new(), false, std::time::SystemTime::now())
                .expect("unset + not required is the labeled fail-open path");
        assert!(matches!(outcome, CeilingLoadOutcome::NotConfigured));
    }

    #[test]
    fn ceiling_required_but_unset_errors() {
        let _guard = lock_env();
        let _env = CeilingEnvGuard::new();
        let err =
            load_and_verify_ceiling_from_env(&HashMap::new(), true, std::time::SystemTime::now())
                .expect_err("required + unset must abort");
        assert!(format!("{err}").contains("CELLOS_AUTHORITY_CEILING_PATH"));
    }

    #[test]
    fn ceiling_required_missing_file_errors() {
        let _guard = lock_env();
        let _env = CeilingEnvGuard::new();
        let dir = tempfile::tempdir().expect("tempdir");
        std::env::set_var(
            "CELLOS_AUTHORITY_CEILING_PATH",
            dir.path().join("does-not-exist.json"),
        );
        let err = load_and_verify_ceiling_from_env(
            &org_root_keyring(),
            true,
            std::time::SystemTime::now(),
        )
        .expect_err("required + missing file must abort");
        assert!(format!("{err}").contains("cannot read"));
    }

    #[test]
    fn ceiling_tampered_signature_errors() {
        let _guard = lock_env();
        let _env = CeilingEnvGuard::new();
        let dir = tempfile::tempdir().expect("tempdir");
        let payload = ceiling_manifest_json(1);
        let mut envelope = build_signed_envelope(&payload, AUTHORITY_CEILING_V1_PAYLOAD_TYPE);
        // Flip a byte in the signature so the Ed25519 verify fails closed.
        let mut sig_bytes = URL_SAFE_NO_PAD
            .decode(envelope.signatures[0].signature.trim_end_matches('='))
            .expect("decode sig");
        sig_bytes[0] ^= 0xff;
        envelope.signatures[0].signature = URL_SAFE_NO_PAD.encode(&sig_bytes);
        let path = write_envelope(dir.path(), &envelope);
        std::env::set_var("CELLOS_AUTHORITY_CEILING_PATH", &path);

        let err = load_and_verify_ceiling_from_env(
            &org_root_keyring(),
            true,
            std::time::SystemTime::now(),
        )
        .expect_err("tampered signature must abort");
        assert!(format!("{err}").contains("CELLOS_AUTHORITY_CEILING_PATH"));
    }

    #[test]
    fn ceiling_wrong_payload_type_rejected() {
        let _guard = lock_env();
        let _env = CeilingEnvGuard::new();
        let dir = tempfile::tempdir().expect("tempdir");
        // A validly-signed envelope, but carrying the trust-keyset sibling type
        // where a ceiling is expected: the verifier's media-type guard accepts
        // it, but the loader's re-assertion must reject it.
        let payload = ceiling_manifest_json(1);
        let envelope = build_signed_envelope(&payload, TRUST_KEYSET_V1_PAYLOAD_TYPE);
        let path = write_envelope(dir.path(), &envelope);
        std::env::set_var("CELLOS_AUTHORITY_CEILING_PATH", &path);

        let err = load_and_verify_ceiling_from_env(
            &org_root_keyring(),
            true,
            std::time::SystemTime::now(),
        )
        .expect_err("trust-keyset envelope presented as ceiling must be rejected");
        assert!(format!("{err}").contains("expected ceiling sibling"));
    }

    #[test]
    fn ceiling_valid_envelope_verifies_and_parses() {
        let _guard = lock_env();
        let _env = CeilingEnvGuard::new();
        let dir = tempfile::tempdir().expect("tempdir");
        let payload = ceiling_manifest_json(7);
        let envelope = build_signed_envelope(&payload, AUTHORITY_CEILING_V1_PAYLOAD_TYPE);
        let path = write_envelope(dir.path(), &envelope);
        std::env::set_var("CELLOS_AUTHORITY_CEILING_PATH", &path);

        let outcome = load_and_verify_ceiling_from_env(
            &org_root_keyring(),
            true,
            std::time::SystemTime::now(),
        )
        .expect("valid ceiling envelope must verify");
        match outcome {
            CeilingLoadOutcome::Verified(details) => {
                assert_eq!(details.manifest.ceiling_id, "ceiling-test");
                assert_eq!(details.manifest.ceiling_epoch, 7);
                assert_eq!(details.verified_signer_kid, CEILING_KID);
            }
            other => panic!("expected Verified, got {other:?}"),
        }
    }

    #[test]
    fn ceiling_epoch_below_anchor_rejected() {
        let _guard = lock_env();
        let _env = CeilingEnvGuard::new();
        let dir = tempfile::tempdir().expect("tempdir");
        // Anchor records a high epoch; a lower-epoch manifest is stale.
        let anchor = dir.path().join("ceiling.anchor");
        std::fs::write(&anchor, "10").expect("write anchor");
        std::env::set_var("CELLOS_AUTHORITY_CEILING_ANCHOR_PATH", &anchor);

        let payload = ceiling_manifest_json(3);
        let envelope = build_signed_envelope(&payload, AUTHORITY_CEILING_V1_PAYLOAD_TYPE);
        let path = write_envelope(dir.path(), &envelope);
        std::env::set_var("CELLOS_AUTHORITY_CEILING_PATH", &path);

        let err = load_and_verify_ceiling_from_env(
            &org_root_keyring(),
            true,
            std::time::SystemTime::now(),
        )
        .expect_err("ceilingEpoch below anchor must abort");
        let msg = format!("{err}");
        assert!(msg.contains("ceiling_epoch_stale"), "got: {msg}");
        // Anchor floor must NOT be lowered by a rejected stale manifest.
        assert_eq!(
            std::fs::read_to_string(&anchor)
                .expect("read anchor")
                .trim(),
            "10"
        );
    }

    #[test]
    fn ceiling_epoch_advances_anchor_monotonically() {
        let _guard = lock_env();
        let _env = CeilingEnvGuard::new();
        let dir = tempfile::tempdir().expect("tempdir");
        let anchor = dir.path().join("ceiling.anchor");
        std::env::set_var("CELLOS_AUTHORITY_CEILING_ANCHOR_PATH", &anchor);

        // First observation (no anchor file yet): epoch 5 accepted, anchor set.
        let payload = ceiling_manifest_json(5);
        let envelope = build_signed_envelope(&payload, AUTHORITY_CEILING_V1_PAYLOAD_TYPE);
        let path = write_envelope(dir.path(), &envelope);
        std::env::set_var("CELLOS_AUTHORITY_CEILING_PATH", &path);
        load_and_verify_ceiling_from_env(&org_root_keyring(), true, std::time::SystemTime::now())
            .expect("first observation accepted");
        assert_eq!(
            std::fs::read_to_string(&anchor)
                .expect("read anchor")
                .trim(),
            "5"
        );
    }
}