car-inference 0.47.0

//! Refreshable signed model catalog (Phase E1).
//!
//! The built-in catalog (`builtin_catalog.json`) is fixed at build time,
//! so a genuinely new model — say a new small-active-param MoE worth
//! trying on a constrained machine — can't surface until the next
//! release. This lets the catalog be refreshed from a **signed** remote
//! source: the daemon fetches the catalog bytes + a detached ed25519
//! signature, verifies it against a pinned public key, and caches the
//! authenticated body + signature envelope. Startup re-verifies that exact
//! body before models load into the registry (which is immutable at runtime),
//! becoming `recommend()` candidates and therefore concierge suggestions —
//! always grounded, never hallucinated.
//!
//! Security: a catalog is only ever trusted if its signature verifies
//! against the configured public key. No key configured → no remote
//! catalog (safe default). Verification is over the exact fetched bytes
//! (detached signature), so there's no canonicalization to get wrong.

use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::OnceLock;

use serde::{Deserialize, Serialize};

use crate::schema::ModelSchema;

/// Cache file name (sibling of `models.json` under `~/.car/`).
pub const CATALOG_CACHE_FILE: &str = "catalog-cache.json";

/// Max catalog body we'll download before verifying — a backstop so a
/// malicious/oversized URL can't OOM the daemon ahead of verification.
const MAX_CATALOG_BYTES: u64 = 8 * 1024 * 1024;

/// The signed catalog document. `version` is a monotonic counter the
/// publisher increments each release; the daemon refuses any catalog
/// whose version is not strictly greater than the last accepted one.
/// This is the anti-rollback guard: a signature proves authenticity but
/// not freshness, so an attacker (or stale cache) replaying an *old but
/// validly signed* catalog would otherwise pass — the version check
/// stops it.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CatalogDoc {
    pub version: u64,
    #[serde(default)]
    pub models: Vec<ModelSchema>,
}

/// An authenticated catalog retained in memory after detached-signature
/// verification. The parsed document is derived from `signed_body`; it is never
/// persisted as a separate trusted field.
#[derive(Debug, Clone)]
pub struct VerifiedCatalog {
    doc: CatalogDoc,
    signed_body: String,
    signature: String,
}

impl VerifiedCatalog {
    pub fn version(&self) -> u64 {
        self.doc.version
    }

    pub fn model_count(&self) -> usize {
        self.doc.models.len()
    }

    pub fn into_models(self) -> Vec<ModelSchema> {
        self.doc.models
    }
}

/// Single-file cache envelope. Only the exact signed body and detached
/// signature are persisted: all catalog fields, including `version` and every
/// schema's trust/capabilities/source, are reparsed only after verification.
#[derive(Debug, Serialize, Deserialize)]
struct CatalogCacheEnvelope {
    signed_body: String,
    signature: String,
}

/// Where the verified catalog is cached, given the models dir.
pub fn cache_path(models_dir: &Path) -> PathBuf {
    models_dir
        .parent()
        .unwrap_or(models_dir)
        .join(CATALOG_CACHE_FILE)
}

/// Load cached catalog models only after re-verifying the exact signed body.
/// Missing key, legacy plain JSON, malformed envelope, or invalid signature all
/// fail closed to an empty list without breaking startup.
pub fn load_cache(path: &Path, public_key_b64: Option<&str>) -> Vec<ModelSchema> {
    load_verified(path, public_key_b64)
        .map(VerifiedCatalog::into_models)
        .unwrap_or_default()
}

/// Load the authenticated cached catalog (including its anti-rollback version).
pub fn load_verified(path: &Path, public_key_b64: Option<&str>) -> Option<VerifiedCatalog> {
    let public_key_b64 = public_key_b64
        .map(str::trim)
        .filter(|key| !key.is_empty())?;
    let json = std::fs::read_to_string(path).ok()?;
    let envelope: CatalogCacheEnvelope = serde_json::from_str(&json).ok()?;
    verify_signed_catalog(envelope.signed_body, envelope.signature, public_key_b64).ok()
}

/// Load only the authenticated document for anti-rollback comparison.
pub fn load_doc(path: &Path, public_key_b64: Option<&str>) -> Option<CatalogDoc> {
    load_verified(path, public_key_b64).map(|verified| verified.doc)
}

fn verify_signed_catalog(
    signed_body: String,
    signature: String,
    public_key_b64: &str,
) -> Result<VerifiedCatalog, String> {
    car_bundle::verify_detached(
        signed_body.as_bytes(),
        signature.trim(),
        public_key_b64.trim(),
    )
    .map_err(|e| format!("catalog signature verification failed: {e}"))?;
    let doc = serde_json::from_str(&signed_body).map_err(|e| format!("parse catalog: {e}"))?;
    Ok(VerifiedCatalog {
        doc,
        signed_body,
        signature: signature.trim().to_string(),
    })
}

/// Fetch a catalog document + its detached signature, verify against
/// `public_key_b64`, and retain the exact signed bytes + signature for the
/// authenticated cache envelope. The signature is `{url}.sig` (base64 ed25519
/// over the exact catalog bytes). The caller enforces version monotonicity.
pub async fn fetch_and_verify(
    http: &reqwest::Client,
    url: &str,
    public_key_b64: &str,
) -> Result<VerifiedCatalog, String> {
    let resp = http
        .get(url)
        .send()
        .await
        .map_err(|e| format!("fetch catalog: {e}"))?
        .error_for_status()
        .map_err(|e| format!("fetch catalog: {e}"))?;
    if let Some(len) = resp.content_length() {
        if len > MAX_CATALOG_BYTES {
            return Err(format!("catalog too large ({len} bytes)"));
        }
    }
    let bytes = resp
        .bytes()
        .await
        .map_err(|e| format!("read catalog: {e}"))?;
    if bytes.len() as u64 > MAX_CATALOG_BYTES {
        return Err(format!("catalog too large ({} bytes)", bytes.len()));
    }
    let sig = http
        .get(format!("{url}.sig"))
        .send()
        .await
        .map_err(|e| format!("fetch signature: {e}"))?
        .error_for_status()
        .map_err(|e| format!("fetch signature: {e}"))?
        .text()
        .await
        .map_err(|e| format!("read signature: {e}"))?;

    let signed_body =
        String::from_utf8(bytes.to_vec()).map_err(|e| format!("parse catalog: {e}"))?;
    verify_signed_catalog(signed_body, sig, public_key_b64)
}

/// Persist the authenticated envelope atomically. No parsed catalog field is
/// serialized outside the signed body.
pub fn save_verified(path: &Path, verified: &VerifiedCatalog) -> std::io::Result<()> {
    if let Some(parent) = path.parent() {
        std::fs::create_dir_all(parent)?;
    }
    let envelope = CatalogCacheEnvelope {
        signed_body: verified.signed_body.clone(),
        signature: verified.signature.clone(),
    };
    let json = serde_json::to_string_pretty(&envelope).map_err(std::io::Error::other)?;
    let tmp = unique_temp_path(path);
    std::fs::write(&tmp, json)?;
    if let Err(error) = std::fs::rename(&tmp, path) {
        let _ = std::fs::remove_file(&tmp);
        return Err(error);
    }
    Ok(())
}

fn unique_temp_path(path: &Path) -> PathBuf {
    static TEMP_SEQUENCE: AtomicU64 = AtomicU64::new(0);
    let sequence = TEMP_SEQUENCE.fetch_add(1, Ordering::Relaxed);
    let file_name = path
        .file_name()
        .and_then(|name| name.to_str())
        .unwrap_or(CATALOG_CACHE_FILE);
    path.with_file_name(format!(
        ".{file_name}.{}.{}.tmp",
        std::process::id(),
        sequence
    ))
}

fn cache_update_lock() -> &'static tokio::sync::Mutex<()> {
    static LOCK: OnceLock<tokio::sync::Mutex<()>> = OnceLock::new();
    LOCK.get_or_init(|| tokio::sync::Mutex::new(()))
}

fn catalog_lock_path(path: &Path) -> PathBuf {
    let mut lock_path = path.as_os_str().to_owned();
    lock_path.push(".lock");
    PathBuf::from(lock_path)
}

fn acquire_catalog_lock(path: &Path) -> std::io::Result<std::fs::File> {
    let lock_path = catalog_lock_path(path);
    if let Some(parent) = lock_path.parent() {
        std::fs::create_dir_all(parent)?;
    }
    let lock = std::fs::OpenOptions::new()
        .read(true)
        .write(true)
        .create(true)
        .truncate(false)
        .open(&lock_path)?;

    #[cfg(test)]
    {
        use std::fs::TryLockError;

        match lock.try_lock() {
            Ok(()) => Ok(lock),
            Err(TryLockError::WouldBlock) => {
                if let Some(contended_path) = std::env::var_os("CAR_CATALOG_TEST_LOCK_CONTENDED") {
                    std::fs::write(contended_path, b"contended")?;
                }
                lock.lock()?;
                Ok(lock)
            }
            Err(TryLockError::Error(error)) => Err(error),
        }
    }

    #[cfg(not(test))]
    {
        lock.lock()?;
        Ok(lock)
    }
}

#[cfg(test)]
fn pause_install_after_authenticated_read() -> Result<(), String> {
    let Some(ready_path) = std::env::var_os("CAR_CATALOG_TEST_READ_READY") else {
        return Ok(());
    };
    let release_path = std::env::var_os("CAR_CATALOG_TEST_READ_RELEASE")
        .ok_or_else(|| "CAR_CATALOG_TEST_READ_RELEASE is required with READ_READY".to_string())?;
    std::fs::write(&ready_path, b"ready")
        .map_err(|error| format!("signal authenticated catalog read: {error}"))?;
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
    while !Path::new(&release_path).exists() {
        if std::time::Instant::now() >= deadline {
            return Err("timed out waiting to release authenticated catalog read".to_string());
        }
        std::thread::sleep(std::time::Duration::from_millis(10));
    }
    Ok(())
}

fn install_if_newer_locked(
    path: &Path,
    verified: &VerifiedCatalog,
    public_key_b64: &str,
) -> Result<usize, String> {
    // Hold the sibling advisory lock across the authenticated read, monotonic
    // comparison, and atomic replacement. The File owns the OS lock and releases
    // it on every return path via RAII; the lock file itself deliberately stays.
    let _lock = acquire_catalog_lock(path).map_err(|error| {
        format!(
            "acquire catalog cache lock {}: {error}",
            catalog_lock_path(path).display()
        )
    })?;
    let cached_version = load_doc(path, Some(public_key_b64))
        .map(|doc| doc.version)
        .unwrap_or(0);
    #[cfg(test)]
    pause_install_after_authenticated_read()?;
    if verified.version() <= cached_version {
        return Err(format!(
            "catalog version {} is not newer than the cached version {} (rejected)",
            verified.version(),
            cached_version
        ));
    }
    let count = verified.model_count();
    save_verified(path, verified).map_err(|e| e.to_string())?;
    Ok(count)
}

/// Serialize authenticated version comparison and atomic replacement across
/// tasks and processes so a lower version can never win last.
pub async fn install_if_newer(
    path: &Path,
    verified: &VerifiedCatalog,
    public_key_b64: &str,
) -> Result<usize, String> {
    let _guard = cache_update_lock().lock().await;
    let path = path.to_path_buf();
    let verified = verified.clone();
    let public_key_b64 = public_key_b64.to_string();
    tokio::task::spawn_blocking(move || install_if_newer_locked(&path, &verified, &public_key_b64))
        .await
        .map_err(|error| format!("catalog cache install task failed: {error}"))?
}

#[cfg(test)]
pub(crate) fn signed_test_catalog(doc: CatalogDoc, seed: u8) -> (VerifiedCatalog, String) {
    use base64::Engine;
    use ed25519_dalek::{Signer, SigningKey};

    let signed_body = serde_json::to_string(&doc).unwrap();
    let signing_key = SigningKey::from_bytes(&[seed; 32]);
    let signature = base64::engine::general_purpose::STANDARD
        .encode(signing_key.sign(signed_body.as_bytes()).to_bytes());
    let public_key =
        base64::engine::general_purpose::STANDARD.encode(signing_key.verifying_key().as_bytes());
    let verified = verify_signed_catalog(signed_body, signature, &public_key).unwrap();
    (verified, public_key)
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::process::{Child, Command, ExitStatus};
    use std::time::{Duration, Instant};

    const CROSS_PROCESS_TEST: &str =
        "catalog::tests::cross_process_installs_preserve_highest_authenticated_version";

    fn wait_for_path(path: &Path, timeout: Duration) -> Result<(), String> {
        let deadline = Instant::now() + timeout;
        while !path.exists() {
            if Instant::now() >= deadline {
                return Err(format!("timed out waiting for {}", path.display()));
            }
            std::thread::sleep(Duration::from_millis(10));
        }
        Ok(())
    }

    fn wait_for_child(child: &mut Child, timeout: Duration) -> Result<ExitStatus, String> {
        let deadline = Instant::now() + timeout;
        loop {
            match child.try_wait() {
                Ok(Some(status)) => return Ok(status),
                Ok(None) if Instant::now() < deadline => {
                    std::thread::sleep(Duration::from_millis(10));
                }
                Ok(None) => {
                    let _ = child.kill();
                    let _ = child.wait();
                    return Err("timed out waiting for catalog child process".to_string());
                }
                Err(error) => {
                    let _ = child.kill();
                    let _ = child.wait();
                    return Err(format!("wait for catalog child process: {error}"));
                }
            }
        }
    }

    fn spawn_catalog_child(
        cache_path: &Path,
        version: u64,
        ready_path: Option<&Path>,
        release_path: Option<&Path>,
        contended_path: Option<&Path>,
    ) -> Child {
        let mut command = Command::new(std::env::current_exe().unwrap());
        command
            .arg(CROSS_PROCESS_TEST)
            .arg("--exact")
            .arg("--nocapture")
            .arg("--test-threads=1")
            .env("CAR_CATALOG_TEST_CHILD_VERSION", version.to_string())
            .env("CAR_CATALOG_TEST_CHILD_CACHE", cache_path);
        if let Some(path) = ready_path {
            command.env("CAR_CATALOG_TEST_READ_READY", path);
        }
        if let Some(path) = release_path {
            command.env("CAR_CATALOG_TEST_READ_RELEASE", path);
        }
        if let Some(path) = contended_path {
            command.env("CAR_CATALOG_TEST_LOCK_CONTENDED", path);
        }
        command.spawn().expect("spawn catalog test child")
    }

    #[test]
    fn missing_cache_is_empty() {
        let p = std::env::temp_dir().join("car-catalog-none-xyz.json");
        let _ = std::fs::remove_file(&p);
        assert!(load_cache(&p, None).is_empty());
    }

    #[test]
    fn cache_path_is_sibling_of_models_dir() {
        let p = cache_path(Path::new("/home/u/.car/models"));
        assert_eq!(p, Path::new("/home/u/.car/catalog-cache.json"));
    }

    #[test]
    fn atomic_temp_paths_are_unique_siblings() {
        let path = Path::new("/home/u/.car/catalog-cache.json");
        let first = unique_temp_path(path);
        let second = unique_temp_path(path);

        assert_eq!(first.parent(), path.parent());
        assert_eq!(second.parent(), path.parent());
        assert_ne!(first, second);
        assert!(first
            .file_name()
            .unwrap()
            .to_string_lossy()
            .ends_with(".tmp"));
        assert!(second
            .file_name()
            .unwrap()
            .to_string_lossy()
            .ends_with(".tmp"));
    }

    #[test]
    fn valid_signed_envelope_round_trips_curated_models() {
        let tmp = tempfile::tempdir().unwrap();
        let path = cache_path(&tmp.path().join("models"));
        let schema = crate::openrouter::builtin_schemas()
            .into_iter()
            .next()
            .expect("managed OpenRouter schema");
        let (verified, public_key) = signed_test_catalog(
            CatalogDoc {
                version: 7,
                models: vec![schema.clone()],
            },
            7,
        );

        save_verified(&path, &verified).unwrap();
        let loaded = load_cache(&path, Some(&public_key));
        assert_eq!(loaded.len(), 1);
        assert_eq!(loaded[0].id, schema.id);
        assert_eq!(loaded[0].trust_tier, crate::schema::TrustTier::Curated);
        assert_eq!(
            load_doc(&path, Some(&public_key))
                .map(|doc| doc.version)
                .unwrap(),
            7
        );
    }

    #[tokio::test]
    async fn tampered_body_or_version_and_wrong_or_missing_key_fail_closed() {
        let tmp = tempfile::tempdir().unwrap();
        let path = cache_path(&tmp.path().join("models"));
        let schema = crate::openrouter::builtin_schemas()
            .into_iter()
            .next()
            .expect("managed OpenRouter schema");
        let (verified, public_key) = signed_test_catalog(
            CatalogDoc {
                version: 7,
                models: vec![schema],
            },
            11,
        );
        let (_, wrong_key) = signed_test_catalog(
            CatalogDoc {
                version: 1,
                models: vec![],
            },
            12,
        );

        save_verified(&path, &verified).unwrap();
        assert!(load_cache(&path, None).is_empty());
        assert!(load_cache(&path, Some("")).is_empty());
        assert!(load_cache(&path, Some(&wrong_key)).is_empty());

        let mut envelope: CatalogCacheEnvelope =
            serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap();
        envelope.signed_body = envelope
            .signed_body
            .replace("\"version\":7", "\"version\":18446744073709551615");
        std::fs::write(&path, serde_json::to_vec_pretty(&envelope).unwrap()).unwrap();
        assert!(load_cache(&path, Some(&public_key)).is_empty());
        assert!(
            load_doc(&path, Some(&public_key)).is_none(),
            "a tampered cached version must not participate in anti-rollback comparison"
        );

        let (newer, same_public_key) = signed_test_catalog(
            CatalogDoc {
                version: 8,
                models: vec![],
            },
            11,
        );
        assert_eq!(public_key, same_public_key);
        assert_eq!(
            install_if_newer(&path, &newer, &public_key).await,
            Ok(0),
            "a forged high cached version must not block a newer authenticated refresh"
        );
        assert_eq!(
            load_doc(&path, Some(&public_key)).map(|doc| doc.version),
            Some(8)
        );
    }

    #[test]
    fn invalid_or_missing_signature_and_legacy_plain_json_fail_closed() {
        let tmp = tempfile::tempdir().unwrap();
        let path = cache_path(&tmp.path().join("models"));
        let doc = CatalogDoc {
            version: u64::MAX,
            models: crate::openrouter::builtin_schemas()
                .into_iter()
                .take(1)
                .collect(),
        };
        let (verified, public_key) = signed_test_catalog(doc.clone(), 21);

        save_verified(&path, &verified).unwrap();
        let mut value: serde_json::Value =
            serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap();
        value["signature"] = serde_json::Value::String("AAAA".to_string());
        std::fs::write(&path, serde_json::to_vec_pretty(&value).unwrap()).unwrap();
        assert!(load_cache(&path, Some(&public_key)).is_empty());

        save_verified(&path, &verified).unwrap();
        let mut value: serde_json::Value =
            serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap();
        value.as_object_mut().unwrap().remove("signature");
        std::fs::write(&path, serde_json::to_vec_pretty(&value).unwrap()).unwrap();
        assert!(load_cache(&path, Some(&public_key)).is_empty());

        std::fs::write(&path, serde_json::to_vec_pretty(&doc).unwrap()).unwrap();
        assert!(load_cache(&path, Some(&public_key)).is_empty());
        assert!(load_doc(&path, Some(&public_key)).is_none());
    }

    #[tokio::test]
    async fn concurrent_installs_leave_the_highest_authenticated_version() {
        let tmp = tempfile::tempdir().unwrap();
        let path = cache_path(&tmp.path().join("models"));
        let (version_n, public_key) = signed_test_catalog(
            CatalogDoc {
                version: 41,
                models: vec![],
            },
            31,
        );
        let (version_n_plus_one, same_public_key) = signed_test_catalog(
            CatalogDoc {
                version: 42,
                models: vec![],
            },
            31,
        );
        assert_eq!(public_key, same_public_key);

        let (_lower, higher) = tokio::join!(
            install_if_newer(&path, &version_n, &public_key),
            install_if_newer(&path, &version_n_plus_one, &public_key)
        );
        assert!(higher.is_ok());
        assert_eq!(
            load_doc(&path, Some(&public_key)).map(|doc| doc.version),
            Some(42)
        );

        let replay = install_if_newer(&path, &version_n, &public_key).await;
        assert!(replay.is_err());
        assert_eq!(
            load_doc(&path, Some(&public_key)).map(|doc| doc.version),
            Some(42)
        );
    }

    #[test]
    fn cross_process_installs_preserve_highest_authenticated_version() {
        if let Some(version) = std::env::var_os("CAR_CATALOG_TEST_CHILD_VERSION") {
            let version = version
                .to_string_lossy()
                .parse::<u64>()
                .expect("child catalog version");
            let cache_path = PathBuf::from(
                std::env::var_os("CAR_CATALOG_TEST_CHILD_CACHE").expect("child catalog cache path"),
            );
            let (verified, public_key) = signed_test_catalog(
                CatalogDoc {
                    version,
                    models: vec![],
                },
                31,
            );
            let runtime = tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
                .unwrap();
            runtime
                .block_on(install_if_newer(&cache_path, &verified, &public_key))
                .expect("child catalog install");
            return;
        }

        let tmp = tempfile::tempdir().unwrap();
        let path = cache_path(&tmp.path().join("models"));
        let ready = tmp.path().join("version-41-read");
        let release = tmp.path().join("release-version-41");
        let (initial, public_key) = signed_test_catalog(
            CatalogDoc {
                version: 40,
                models: vec![],
            },
            31,
        );
        save_verified(&path, &initial).unwrap();

        let contended = tmp.path().join("version-42-contended");
        let mut version_41 = spawn_catalog_child(&path, 41, Some(&ready), Some(&release), None);
        if let Err(error) = wait_for_path(&ready, Duration::from_secs(10)) {
            let _ = version_41.kill();
            let _ = version_41.wait();
            panic!("{error}");
        }

        let mut version_42 = spawn_catalog_child(&path, 42, None, None, Some(&contended));
        let contention_status = wait_for_path(&contended, Duration::from_secs(10));
        std::fs::write(&release, b"release").unwrap();
        let version_42_status = wait_for_child(&mut version_42, Duration::from_secs(10));
        let version_41_status = wait_for_child(&mut version_41, Duration::from_secs(10));

        assert!(
            contention_status.is_ok(),
            "version 42 never contended on the cross-process cache lock: {contention_status:?}"
        );
        assert!(
            version_42_status.unwrap().success(),
            "version 42 child failed"
        );
        assert!(
            version_41_status.unwrap().success(),
            "version 41 child failed"
        );
        assert_eq!(
            load_doc(&path, Some(&public_key)).map(|doc| doc.version),
            Some(42),
            "a stale cross-process writer must not replace a newer authenticated catalog"
        );

        let (stale, same_public_key) = signed_test_catalog(
            CatalogDoc {
                version: 41,
                models: vec![],
            },
            31,
        );
        assert_eq!(public_key, same_public_key);
        let runtime = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let stale_result = runtime.block_on(install_if_newer(&path, &stale, &public_key));
        assert!(
            stale_result.is_err(),
            "an authenticated but stale replay must be rejected after the process race"
        );
        assert_eq!(
            load_doc(&path, Some(&public_key)).map(|doc| doc.version),
            Some(42)
        );
    }
}