cellos-supervisor 0.5.1

//! 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::{
    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,
};
use ed25519_dalek::VerifyingKey;
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, VerifyingKey>>, 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, VerifyingKey>,
    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, VerifyingKey>,
    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, VerifyingKey>,
    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, VerifyingKey>,
    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, VerifyingKey>,
    now: std::time::SystemTime,
) -> Option<String> {
    use base64::engine::general_purpose::URL_SAFE_NO_PAD;
    use base64::Engine as _;
    use chrono::DateTime;
    use ed25519_dalek::Signature;

    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;
        };
        let Ok(sig_array) = <[u8; 64]>::try_from(sig_bytes.as_slice()) else {
            continue;
        };
        let signature = Signature::from_bytes(&sig_array);
        if verifying_key
            .verify_strict(payload_bytes, &signature)
            .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))]
    {
        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(())
        }
    }
}

#[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));
    }
}