openlatch-client 0.1.18

//! On-disk bundle cache: load, store, and digest verification (D37).
//!
//! Two files live under `<base>/policy/`:
//!
//! | File | Role |
//! |---|---|
//! | `bundle.json` | The active bundle, **exactly the bytes received**. |
//! | `bundle.meta.json` | The commit marker plus poll/activation state. |
//!
//! The base directory is a [`Path`] **parameter**, never
//! `crate::config::openlatch_dir()` — that keeps this module a leaf and makes
//! the whole store testable against a `tempfile::TempDir`.
//!
//! > **The client never re-serializes or re-canonicalizes the bundle.** It
//! > hashes the raw bytes it received and stores those exact bytes. Do not
//! > reach for `serde_json_canonicalizer` on this path (it exists in the tree
//! > for the tamper-evidence and config-plane hashes): hashing the received
//! > bytes is the invariant that makes cross-language JCS byte-parity a
//! > non-issue.

use std::path::{Path, PathBuf};

use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};

use crate::core::error::{ERR_BUNDLE_FETCH_FAILED, ERR_BUNDLE_INVALID, ERR_BUNDLE_REJECTED};
use crate::generated::types::PolicyBundle;

/// Directory name under the caller-supplied base directory.
pub const POLICY_DIR: &str = "policy";
/// The bundle body — the exact bytes received from the platform.
pub const BUNDLE_FILE: &str = "bundle.json";
/// The commit marker. Its existence is what makes a body "activated".
pub const META_FILE: &str = "bundle.meta.json";

/// `<base>/policy/`.
pub fn policy_dir(base: &Path) -> PathBuf {
    base.join(POLICY_DIR)
}

/// `<base>/policy/bundle.json`.
pub fn bundle_path(base: &Path) -> PathBuf {
    policy_dir(base).join(BUNDLE_FILE)
}

/// `<base>/policy/bundle.meta.json`.
pub fn meta_path(base: &Path) -> PathBuf {
    policy_dir(base).join(META_FILE)
}

/// Sidecar state for the cached bundle.
///
/// # Deviation from the PRD, deliberate (D28)
///
/// The PRD's on-disk shape carries two timestamps, `fetched_at` and
/// `last_fetch_ok`. This struct carries **three tiers** instead, because a
/// `304` advances the poll tier without advancing the download tier and a
/// single `fetched_at` cannot express that:
///
/// | Field | Advances on |
/// |---|---|
/// | `last_poll_ok_at` | any `2xx` **or** `304` — drives the `OL-1213` staleness clock |
/// | `last_download_verified_at` | a `200` whose digest verified |
/// | `last_activated_at` | the hot-swap completed |
///
/// Plus `last_error`, so a host that is quietly failing to refresh says why
/// without anyone grepping the daemon log.
///
/// Unknown fields are tolerated (no `deny_unknown_fields`) and every non-core
/// field defaults, so a meta file written by a newer or older client still
/// parses rather than being discarded along with a perfectly good bundle.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct BundleMeta {
    /// `sha256:<hex>` over `bundle.json`'s bytes. Re-checked on every load.
    pub digest: String,
    /// The **last successfully activated** entity-tag, sent as
    /// `If-None-Match` (D25). Never the merely-received one: recording a tag
    /// for a bundle that failed activation wedges the client on the old bundle
    /// forever (the bug OPA shipped in opa#2220).
    #[serde(default)]
    pub etag: Option<String>,
    /// The activated bundle's revision.
    pub revision: i64,
    /// Written on the FIRST successful activation and never rewritten — this is
    /// the trust-on-first-use anchor every later bundle is checked against.
    pub organization_id: String,
    /// Copied from the bundle body; drives `olpolicybundleage`.
    pub built_at: String,
    /// Any `2xx` **or** `304`. RFC 3339 UTC.
    #[serde(default)]
    pub last_poll_ok_at: Option<String>,
    /// A `200` whose digest verified. RFC 3339 UTC.
    #[serde(default)]
    pub last_download_verified_at: Option<String>,
    /// The hot-swap completed. RFC 3339 UTC.
    #[serde(default)]
    pub last_activated_at: Option<String>,
    /// → `olpolicyoffline` (inverted). `false` after a failed poll attempt.
    #[serde(default = "default_true")]
    pub last_fetch_ok: bool,
    /// e.g. `"OL-1211 digest mismatch"`. `None` once a poll succeeds.
    #[serde(default)]
    pub last_error: Option<String>,
    /// The `agent_id` the [`Self::etag`] above was fetched under.
    ///
    /// The bundle body is composed **per agent** — the platform reads
    /// `X-OpenLatch-Agent-Id` and stamps `client_config.agent_context` from it
    /// — so the validator is only meaningful paired with the identity that
    /// obtained it. Sending it under a different (or absent) identity invites a
    /// `304` for a body this install has never seen: the client would keep a
    /// context-less bundle until the org's rules next changed. The poller
    /// therefore omits `If-None-Match` whenever this does not equal its own
    /// `agent_id`, and a full download re-establishes the pair.
    ///
    /// Absent in every meta written before the header existed, and that reads
    /// as `None` — which compares unequal to any configured `agent_id` and
    /// costs exactly one extra download on upgrade. (`Option` is already
    /// optional to serde; the attribute matches its siblings above.)
    #[serde(default)]
    pub agent_id: Option<String>,
}

fn default_true() -> bool {
    true
}

impl BundleMeta {
    /// A fresh meta for a bundle that has just verified and activated.
    pub fn activated(bundle: &PolicyBundle, digest: String, etag: Option<String>) -> Self {
        let now = now_rfc3339();
        Self {
            digest,
            etag,
            revision: bundle.revision,
            organization_id: bundle.organization_id.clone(),
            built_at: bundle.built_at.clone(),
            last_poll_ok_at: Some(now.clone()),
            last_download_verified_at: Some(now.clone()),
            last_activated_at: Some(now),
            last_fetch_ok: true,
            last_error: None,
            // Stamped by the poller, which is the only place that knows which
            // identity the response was served for.
            agent_id: None,
        }
    }
}

/// A verified body + meta pair.
#[derive(Debug, Clone, PartialEq)]
pub struct CachedBundle {
    pub bundle: PolicyBundle,
    pub meta: BundleMeta,
}

/// Everything that can go wrong reading or writing the cache.
///
/// [`StoreError::code`] maps each variant onto the `OL-####` registry so the
/// caller logs the right code without re-deriving the mapping.
#[derive(Debug, thiserror::Error)]
pub enum StoreError {
    /// The policy directory is unreadable/unwritable, or the disk is full. The
    /// caller keeps the in-memory bundle and keeps enforcing — a disk error
    /// must never fail the verdict path.
    #[error("policy store I/O error at {path}: {source}")]
    Io {
        path: PathBuf,
        #[source]
        source: std::io::Error,
    },
    /// `bundle.json`'s bytes no longer hash to what `bundle.meta.json` claims.
    #[error("cached bundle digest mismatch: meta claims {expected}, body hashes to {actual}")]
    DigestMismatch { expected: String, actual: String },
    /// `bundle.meta.json` is not parseable, so the body cannot be verified at
    /// all.
    #[error("cached bundle metadata is unreadable: {0}")]
    MetaMalformed(String),
    /// The body verified against the digest but is not a valid policy bundle —
    /// which means a valid digest was recorded for an invalid document.
    #[error("cached bundle body is not a valid policy bundle: {0}")]
    BodyMalformed(String),
}

impl StoreError {
    /// The `OL-####` code to log this failure under.
    pub fn code(&self) -> &'static str {
        match self {
            // "policy dir unwritable or disk full -> keep enforcing, log OL-1210".
            StoreError::Io { .. } => ERR_BUNDLE_FETCH_FAILED,
            StoreError::DigestMismatch { .. } | StoreError::MetaMalformed(_) => ERR_BUNDLE_REJECTED,
            StoreError::BodyMalformed(_) => ERR_BUNDLE_INVALID,
        }
    }
}

fn io(path: &Path, source: std::io::Error) -> StoreError {
    StoreError::Io {
        path: path.to_path_buf(),
        source,
    }
}

/// `sha256:<hex>` over the given bytes — the exact form the platform puts in
/// the `ETag`.
pub fn digest_of(bytes: &[u8]) -> String {
    let mut hasher = Sha256::new();
    hasher.update(bytes);
    format!("sha256:{}", hex::encode(hasher.finalize()))
}

/// Verify `body` against an expected `sha256:<hex>` digest.
///
/// The comparison is a plain equality check on purpose: the digest is not a
/// secret and there is no timing channel worth defending here — and neither
/// side is proof of authenticity anyway (that is D32's Ed25519 signature, v1.1).
pub fn verify_digest(body: &[u8], expected: &str) -> Result<(), StoreError> {
    let actual = digest_of(body);
    if actual == expected {
        Ok(())
    } else {
        Err(StoreError::DigestMismatch {
            expected: expected.to_string(),
            actual,
        })
    }
}

/// Current time as RFC 3339 UTC with a `Z` suffix, matching the format the
/// platform emits for `built_at`.
pub fn now_rfc3339() -> String {
    chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true)
}

/// Read `bundle.meta.json` alone, without touching the body.
///
/// Returns `Ok(None)` when the file simply is not there. Useful to the poller,
/// which wants the stored `If-None-Match` validator and the poll clock.
pub fn read_meta(base: &Path) -> Result<Option<BundleMeta>, StoreError> {
    let path = meta_path(base);
    let raw = match std::fs::read(&path) {
        Ok(raw) => raw,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
        Err(e) => return Err(io(&path, e)),
    };
    serde_json::from_slice(&raw)
        .map(Some)
        .map_err(|e| StoreError::MetaMalformed(e.to_string()))
}

/// Load and **re-verify** the cached bundle.
///
/// The digest is re-checked against the body on EVERY load from disk, not only
/// after a fetch. Without this a local user edits `bundle.json` to delete the
/// rule that blocks them and the daemon happily runs the edited version.
///
/// - `Ok(None)` — nothing cached. Either the pair was absent, or exactly one of
///   the two files existed (an orphan, see below) and both were removed.
/// - `Ok(Some(_))` — the pair verified.
/// - `Err(_)` — the pair was present and failed verification. **Both files have
///   already been deleted** and the caller must run with no policy until the
///   next successful fetch, logging [`StoreError::code`].
///
/// # Why a failed verification deletes the files
///
/// The stored `etag` lives in the same meta file. Keeping a rejected pair on
/// disk would leave that validator in place, so the next poll sends
/// `If-None-Match`, receives `304`, never re-downloads — and the host runs with
/// no policy **permanently**. Discarding both files forces a full download on
/// the next poll, which is the only path back to enforcing.
///
/// # Orphans
///
/// `bundle.json` is written FIRST and `bundle.meta.json` SECOND, so the meta
/// file is the commit marker. A crash between the two leaves an orphan body
/// with no meta; the inverse is unreachable but handled identically. Either way
/// it is "no bundle": delete both, refetch.
pub fn load(base: &Path) -> Result<Option<CachedBundle>, StoreError> {
    let body_path = bundle_path(base);
    let meta_file = meta_path(base);

    let body = match std::fs::read(&body_path) {
        Ok(b) => Some(b),
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => None,
        Err(e) => return Err(io(&body_path, e)),
    };
    let meta_raw = match std::fs::read(&meta_file) {
        Ok(b) => Some(b),
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => None,
        Err(e) => return Err(io(&meta_file, e)),
    };

    let (body, meta_raw) = match (body, meta_raw) {
        (None, None) => return Ok(None),
        (Some(_), None) | (None, Some(_)) => {
            tracing::warn!(
                target: "policy",
                "orphaned policy cache (body and meta must both be present); deleting both and refetching"
            );
            discard(base);
            return Ok(None);
        }
        (Some(body), Some(meta_raw)) => (body, meta_raw),
    };

    let meta: BundleMeta = match serde_json::from_slice(&meta_raw) {
        Ok(m) => m,
        Err(e) => {
            discard(base);
            return Err(StoreError::MetaMalformed(e.to_string()));
        }
    };

    if let Err(e) = verify_digest(&body, &meta.digest) {
        discard(base);
        return Err(e);
    }

    // Tolerant on the way in: a cached bundle written by a NEWER client, then
    // read back by this one after a downgrade, must not be thrown away whole for
    // one rule this build cannot read (see `parse_bundle_tolerant`).
    let bundle: PolicyBundle =
        match serde_json::from_slice(&body).and_then(super::parse_bundle_tolerant) {
            Ok(b) => b,
            Err(e) => {
                discard(base);
                return Err(StoreError::BodyMalformed(e.to_string()));
            }
        };

    Ok(Some(CachedBundle { bundle, meta }))
}

/// Persist a verified bundle: body FIRST, meta SECOND.
///
/// Order matters: `bundle.json` lands first, then `bundle.meta.json`. The meta
/// file's existence is the **commit marker**. A crash between the two leaves an
/// orphan body with no meta → treated as "no bundle", both deleted, refetched.
/// The inverse (meta claiming a digest for a body that is not there) is
/// therefore unreachable.
///
/// Both writes go through write-temp-then-rename, mirroring
/// `src/core/update.rs::write_sentinel`, so a reader never observes a partial
/// file.
pub fn store(base: &Path, body: &[u8], meta: &BundleMeta) -> Result<(), StoreError> {
    write_body(base, body)?;
    write_meta(base, meta)
}

/// Write `bundle.json` only. Split out from [`store`] so the commit ordering is
/// expressible — and testable — rather than implied.
pub fn write_body(base: &Path, body: &[u8]) -> Result<(), StoreError> {
    let dir = policy_dir(base);
    std::fs::create_dir_all(&dir).map_err(|e| io(&dir, e))?;
    atomic_write(&bundle_path(base), body)
}

/// Write `bundle.meta.json` only — the commit marker.
///
/// Also the 304 path: a revalidation updates the poll clock in place without
/// touching the body.
pub fn write_meta(base: &Path, meta: &BundleMeta) -> Result<(), StoreError> {
    let dir = policy_dir(base);
    std::fs::create_dir_all(&dir).map_err(|e| io(&dir, e))?;
    let body = serde_json::to_vec_pretty(meta).map_err(|e| {
        io(
            &meta_path(base),
            std::io::Error::new(std::io::ErrorKind::InvalidData, e),
        )
    })?;
    atomic_write(&meta_path(base), &body)
}

/// Remove both cache files, best effort.
///
/// Deliberately infallible: this runs on the rejection path, and failing to
/// delete a file the daemon has already decided not to trust must not turn into
/// a second error the caller has to handle. A `NotFound` is the normal case for
/// at least one of the two.
pub fn discard(base: &Path) {
    for path in [bundle_path(base), meta_path(base)] {
        match std::fs::remove_file(&path) {
            Ok(()) => {}
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
            Err(e) => tracing::warn!(
                target: "policy",
                code = ERR_BUNDLE_FETCH_FAILED,
                path = %path.display(),
                error = %e,
                "could not remove cached policy file"
            ),
        }
    }
}

/// Write-temp-then-rename, matching `src/core/update.rs`.
fn atomic_write(path: &Path, bytes: &[u8]) -> Result<(), StoreError> {
    let tmp = path.with_extension("json.tmp");
    std::fs::write(&tmp, bytes).map_err(|e| io(&tmp, e))?;
    std::fs::rename(&tmp, path).map_err(|e| io(path, e))
}

#[cfg(test)]
mod tests {
    use super::super::test_support::*;
    use super::*;
    use crate::generated::types::{PolicyRuleMode, PolicyRuleSeverity};

    fn fixture() -> (PolicyBundle, Vec<u8>) {
        let bundle = wire_bundle(
            vec![wire_rule(
                "OL-CMD-001",
                "*rm -rf*",
                PolicyRuleMode::Enforce,
                PolicyRuleSeverity::Critical,
            )],
            true,
        );
        let body = serde_json::to_vec(&bundle).expect("serialise fixture");
        (bundle, body)
    }

    fn seed(base: &Path) -> (PolicyBundle, Vec<u8>, BundleMeta) {
        let (bundle, body) = fixture();
        let meta = BundleMeta::activated(
            &bundle,
            digest_of(&body),
            Some("\"sha256:deadbeef\"".to_string()),
        );
        store(base, &body, &meta).expect("store");
        (bundle, body, meta)
    }

    #[test]
    fn digest_is_sha256_of_the_exact_bytes() {
        // Known vector: sha256("") .
        assert_eq!(
            digest_of(b""),
            "sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
        );
        assert_ne!(digest_of(b"a"), digest_of(b"b"));
        assert!(verify_digest(b"a", &digest_of(b"a")).is_ok());
        assert!(verify_digest(b"a", &digest_of(b"b")).is_err());
    }

    #[test]
    fn round_trips_body_and_meta() {
        let dir = tempfile::tempdir().expect("tempdir");
        let (bundle, _body, meta) = seed(dir.path());

        let loaded = load(dir.path()).expect("load ok").expect("bundle present");
        assert_eq!(loaded.bundle, bundle);
        assert_eq!(loaded.meta, meta);
        assert_eq!(loaded.meta.etag.as_deref(), Some("\"sha256:deadbeef\""));
        assert_eq!(loaded.meta.revision, 42);
        assert!(loaded.meta.last_activated_at.is_some());
        assert!(loaded.meta.last_download_verified_at.is_some());
        assert!(loaded.meta.last_poll_ok_at.is_some());
    }

    #[test]
    fn empty_cache_is_not_an_error() {
        let dir = tempfile::tempdir().expect("tempdir");
        assert!(load(dir.path()).expect("load ok").is_none());
        assert!(read_meta(dir.path()).expect("read ok").is_none());
    }

    /// THE tamper test: a local user edits `bundle.json` to delete the rule
    /// that blocks them.
    #[test]
    fn tampered_body_is_rejected_on_load() {
        let dir = tempfile::tempdir().expect("tempdir");
        seed(dir.path());

        let tampered = serde_json::to_vec(&wire_bundle(vec![], true)).expect("serialise");
        std::fs::write(bundle_path(dir.path()), &tampered).expect("tamper");

        let err = load(dir.path()).expect_err("tampered body must not load");
        assert!(matches!(err, StoreError::DigestMismatch { .. }));
        assert_eq!(err.code(), ERR_BUNDLE_REJECTED);

        // Both files are gone, so the next poll cannot 304 its way back into
        // running with no policy.
        assert!(!bundle_path(dir.path()).exists());
        assert!(!meta_path(dir.path()).exists());
        assert!(load(dir.path()).expect("load ok").is_none());
    }

    #[test]
    fn malformed_meta_is_rejected_and_discarded() {
        let dir = tempfile::tempdir().expect("tempdir");
        seed(dir.path());
        std::fs::write(meta_path(dir.path()), b"{ not json").expect("corrupt meta");

        let err = load(dir.path()).expect_err("unverifiable pair must not load");
        assert!(matches!(err, StoreError::MetaMalformed(_)));
        assert_eq!(err.code(), ERR_BUNDLE_REJECTED);
        assert!(!bundle_path(dir.path()).exists());
        assert!(!meta_path(dir.path()).exists());
    }

    #[test]
    fn body_matching_the_digest_but_not_the_schema_is_rejected() {
        let dir = tempfile::tempdir().expect("tempdir");
        let (bundle, _) = fixture();
        let body = br#"{"schema_version":1}"#.to_vec();
        let meta = BundleMeta::activated(&bundle, digest_of(&body), None);
        store(dir.path(), &body, &meta).expect("store");

        let err = load(dir.path()).expect_err("incomplete document must not load");
        assert!(matches!(err, StoreError::BodyMalformed(_)));
        assert_eq!(err.code(), ERR_BUNDLE_INVALID);
        assert!(!bundle_path(dir.path()).exists());
    }

    /// A crash between the two writes leaves an orphan body. Both files go, and
    /// the caller sees "no bundle" rather than an error.
    #[test]
    fn orphan_body_without_meta_is_treated_as_no_bundle() {
        let dir = tempfile::tempdir().expect("tempdir");
        let (_, body) = fixture();
        write_body(dir.path(), &body).expect("write body");

        // The commit ordering, asserted: the body exists and the marker does
        // not, which is the only torn state `store` can produce.
        assert!(bundle_path(dir.path()).exists());
        assert!(!meta_path(dir.path()).exists());

        assert!(load(dir.path()).expect("load ok").is_none());
        assert!(!bundle_path(dir.path()).exists());
    }

    #[test]
    fn orphan_meta_without_body_is_treated_as_no_bundle() {
        let dir = tempfile::tempdir().expect("tempdir");
        let (bundle, body) = fixture();
        write_meta(
            dir.path(),
            &BundleMeta::activated(&bundle, digest_of(&body), None),
        )
        .expect("write meta");

        assert!(load(dir.path()).expect("load ok").is_none());
        assert!(!meta_path(dir.path()).exists());
    }

    /// `store` must leave nothing behind but the two committed files — a
    /// leftover `.tmp` would be picked up by nothing, but it is a smell.
    #[test]
    fn store_leaves_no_temp_files() {
        let dir = tempfile::tempdir().expect("tempdir");
        seed(dir.path());
        let names: Vec<String> = std::fs::read_dir(policy_dir(dir.path()))
            .expect("read dir")
            .map(|e| e.expect("entry").file_name().to_string_lossy().into_owned())
            .collect();
        assert_eq!(names.len(), 2, "unexpected files: {names:?}");
        assert!(names.contains(&BUNDLE_FILE.to_string()));
        assert!(names.contains(&META_FILE.to_string()));
    }

    /// The 304 path: the poll clock advances, `bundle.json` is untouched.
    #[test]
    fn write_meta_updates_the_poll_clock_without_touching_the_body() {
        let dir = tempfile::tempdir().expect("tempdir");
        let (_, body, mut meta) = seed(dir.path());

        meta.last_poll_ok_at = Some("2026-07-22T10:00:00.000Z".to_string());
        write_meta(dir.path(), &meta).expect("write meta");

        assert_eq!(
            std::fs::read(bundle_path(dir.path())).expect("read body"),
            body
        );
        let reloaded = read_meta(dir.path()).expect("read ok").expect("meta");
        assert_eq!(
            reloaded.last_poll_ok_at.as_deref(),
            Some("2026-07-22T10:00:00.000Z")
        );
        // The download/activation tiers did NOT move — that is the whole point
        // of splitting them (D28).
        assert_eq!(
            reloaded.last_download_verified_at,
            meta.last_download_verified_at
        );
        assert_eq!(reloaded.last_activated_at, meta.last_activated_at);
    }

    /// A meta file written by a client that predates the three-tier fields must
    /// still parse — discarding it would throw away a valid bundle.
    #[test]
    fn meta_tolerates_missing_and_unknown_fields() {
        let dir = tempfile::tempdir().expect("tempdir");
        let (_, body) = fixture();
        write_body(dir.path(), &body).expect("write body");
        let legacy = serde_json::json!({
            "digest": digest_of(&body),
            "revision": 7,
            "organization_id": "0192f8a1-4c3b-7e2a-9f10-5d8c3b1a7e42",
            "built_at": "2026-07-21T09:00:00Z",
            "fetched_at": "2026-07-21T09:00:01Z"
        });
        std::fs::write(
            meta_path(dir.path()),
            serde_json::to_vec(&legacy).expect("serialise"),
        )
        .expect("write legacy meta");

        let loaded = load(dir.path()).expect("load ok").expect("bundle present");
        assert_eq!(loaded.meta.revision, 7);
        assert!(loaded.meta.etag.is_none());
        assert!(loaded.meta.last_fetch_ok, "defaults to true");
        assert!(loaded.meta.last_poll_ok_at.is_none());
        // A meta written before the agent-id header existed carries no
        // identity, so the poller cannot reuse its validator.
        assert!(loaded.meta.agent_id.is_none());
    }

    #[test]
    fn discard_is_idempotent() {
        let dir = tempfile::tempdir().expect("tempdir");
        discard(dir.path());
        seed(dir.path());
        discard(dir.path());
        discard(dir.path());
        assert!(load(dir.path()).expect("load ok").is_none());
    }

    #[test]
    fn now_rfc3339_is_utc_with_a_z_suffix() {
        let now = now_rfc3339();
        assert!(now.ends_with('Z'), "{now}");
        assert!(chrono::DateTime::parse_from_rfc3339(&now).is_ok(), "{now}");
    }
}