holger-server-lib 0.6.9

//! Promotion — the DEV → PROD transition (`.nornir/promotion-design.md`).
//!
//! Promotion moves an artifact from a **mutable DEV** repository into an
//! **immutable, content-addressed PROD store**. Per the design (§1, §10.1), the
//! thing that moves is the artifact's **content address** (blake3), and the seal
//! is the commit: once a coordinate is sealed into a PROD repo it is
//! **immutable** — a re-promote or overwrite is refused. Bytes are stored once,
//! content-addressed, so identical content de-dups.
//!
//! Scope of THIS slice (faithful minimal reading of the design):
//!   * [`PromotionEngine::seal`] performs the storage transition: content-address
//!     the bytes (blake3), write the blob under its hash, and create an exclusive
//!     **ref** (the seal) for the `(to_repo, artifact)` coordinate. The ref's
//!     existence *is* the immutability guard — a second seal of the same
//!     coordinate returns [`PromoteError::AlreadyPromoted`].
//!   * Integrity on read ([`PromotionEngine::fetch`]) re-verifies the blob's
//!     blake3 against the ref, so a tampered blob is detected offline.
//!   * The audited unit is [`PromotionRecord`] (the design's §1 record, minus the
//!     deferred fields).
//!
//! **Deferred, per design §6** (noted, not faked): CMS/ed25519 **signing** of the
//! sealed manifest (the ref records the blake3 content address — the manifest
//! anchor a signature would later cover); generation chaining / deletion-vector
//! overlays (§10.4); the policy gate (§4, currently pass-through — promotion is
//! unconditional here); znippy `seal()` of a whole generation (§2). The store is
//! a content-addressed blob+ref tree, which is the immutable, blake3-addressed
//! substrate those build on.

use std::fs;
use std::path::PathBuf;
use std::sync::Arc;

use traits::ArtifactId;

use crate::security::{self, Decision, OsvScanner, ScanEngine};

/// How hard the promotion policy gate (design §4) bites at a boundary.
///
/// The gate is a **seam**: when no gate is configured on the engine the promote
/// is unconditional (the default, unconfigured path is unchanged). When one IS
/// configured the enforcement level decides what a failing verdict does.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Enforcement {
    /// Scan + record the verdict but never refuse (observability only).
    Warn,
    /// Fail closed: a `Block` verdict REFUSES the promotion (design's prod
    /// boundary). This is the real gate.
    Block,
}

/// The promotion policy gate (design §4) — the CVE / license / provenance check
/// run at the boundary an artifact crosses. It wraps the **real** modgunn scan
/// seam ([`OsvScanner`] via [`security::scan_listing`]'s underlying
/// `scanner.scan`), NOT the `demo_*` stub: the caller supplies the scanner
/// (real policy + an imported advisory DB), so nothing demo leaks into the
/// enforcement path. A promote is refused when the verdict is `Block` and
/// enforcement is [`Enforcement::Block`].
pub struct PolicyGate {
    scanner: OsvScanner,
    enforcement: Enforcement,
}

impl PolicyGate {
    /// A gate over `scanner` at the given `enforcement` level.
    pub fn new(scanner: OsvScanner, enforcement: Enforcement) -> Self {
        Self { scanner, enforcement }
    }

    /// Build a gate from the environment — the operator **config flag** — or
    /// `Ok(None)` when promotion enforcement is not configured (the default path,
    /// unchanged). Fail-closed on misconfiguration: a configured advisory feed
    /// that cannot be read/parsed is an `Err` (serve refuses to start) rather
    /// than a silently-empty gate that would wave promotions through.
    ///
    /// - `HOLGER_PROMOTE_ENFORCE` = `block` | `warn`  (unset / `off` ⇒ `Ok(None)`)
    /// - `HOLGER_PROMOTE_ADVISORY_DB` = path to an OSV/GHSA/RUSTSEC JSON snapshot
    ///   (the imported CVE feed); absent ⇒ empty CVE set, license/provenance
    ///   policy still applies.
    /// - `HOLGER_PROMOTE_DENY_LICENSES` = comma-separated SPDX ids to deny.
    pub fn from_env() -> anyhow::Result<Option<PolicyGate>> {
        let enforcement = match std::env::var("HOLGER_PROMOTE_ENFORCE") {
            Ok(v) => match v.trim().to_ascii_lowercase().as_str() {
                "block" => Enforcement::Block,
                "warn" => Enforcement::Warn,
                "" | "off" => return Ok(None),
                other => {
                    anyhow::bail!("HOLGER_PROMOTE_ENFORCE must be block|warn|off, got '{other}'")
                }
            },
            Err(_) => return Ok(None),
        };
        let db = match std::env::var("HOLGER_PROMOTE_ADVISORY_DB") {
            Ok(path) if !path.trim().is_empty() => {
                security::advisory_db_from_osv_file(path.trim())? // fail-closed on config error
            }
            _ => security::AdvisoryDb::default(), // no feed → license/provenance only
        };
        let deny_licenses = std::env::var("HOLGER_PROMOTE_DENY_LICENSES")
            .ok()
            .map(|s| {
                s.split(',')
                    .map(|x| x.trim().to_string())
                    .filter(|x| !x.is_empty())
                    .collect::<Vec<_>>()
            })
            .unwrap_or_default();
        let scanner = OsvScanner::new(security::promotion_policy(deny_licenses), db);
        Ok(Some(PolicyGate::new(scanner, enforcement)))
    }

    /// Evaluate one artifact (`ecosystem` + coords) against the real scanner.
    /// Returns the decision plus human-readable finding reasons (for the audit
    /// detail and the refusal message).
    pub fn evaluate(&self, ecosystem: &str, id: &ArtifactId) -> (Decision, Vec<String>) {
        let art = security::artifact_from_coords(ecosystem, &id.name, &id.version, false);
        let verdict = self.scanner.scan(&art);
        let reasons = verdict
            .findings
            .iter()
            .map(|f| format!("{} [{:?}] {}", f.id, f.severity, f.summary))
            .collect();
        (verdict.decision, reasons)
    }
}

/// The audited outcome of one promotion — the design's §1 `PromotionRecord`
/// restricted to the fields this slice fills (generation/signature fields are
/// deferred). `promoted_at` is a unix timestamp supplied by the caller.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PromotionRecord {
    /// `namespace/name@version` (namespace segment dropped when empty).
    pub artifact: String,
    /// Source (mutable DEV) repository.
    pub from_repo: String,
    /// Target (immutable PROD) repository.
    pub to_repo: String,
    /// blake3 content address of the promoted bytes — the thing that moves.
    pub content_address: String,
    /// Promoted byte count.
    pub size: u64,
    /// Identity that performed the promotion (mTLS CN or OIDC `sub`).
    pub promoted_by: String,
    /// Unix timestamp (seconds) the promotion was recorded.
    pub promoted_at: i64,
}

/// Why a promotion could not be sealed.
#[derive(Debug)]
pub enum PromoteError {
    /// The coordinate is already sealed in the target PROD repo — immutable,
    /// overwrite refused (the re-promote guard).
    AlreadyPromoted,
    /// The policy gate (design §4) REFUSED the promotion: the artifact's scan
    /// verdict was `Block` under `Enforcement::Block`. Carries the finding
    /// reasons for the audit log and the caller message. This is the fail-closed
    /// boundary — it only fires when a gate is configured.
    PolicyBlocked(Vec<String>),
    /// Filesystem failure writing the blob or ref.
    Io(std::io::Error),
}

impl std::fmt::Display for PromoteError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            PromoteError::AlreadyPromoted => {
                write!(f, "artifact already promoted (immutable — overwrite refused)")
            }
            PromoteError::PolicyBlocked(reasons) => {
                write!(f, "promotion blocked by policy gate: {}", reasons.join("; "))
            }
            PromoteError::Io(e) => write!(f, "promotion store io error: {e}"),
        }
    }
}

impl std::error::Error for PromoteError {}

impl From<std::io::Error> for PromoteError {
    fn from(e: std::io::Error) -> Self {
        PromoteError::Io(e)
    }
}

/// Human-readable artifact label used in the audit log and the record.
fn artifact_label(id: &ArtifactId) -> String {
    match id.namespace.as_deref() {
        Some(ns) if !ns.is_empty() => format!("{}/{}@{}", ns, id.name, id.version),
        _ => format!("{}@{}", id.name, id.version),
    }
}

/// A filesystem-safe key for a `(namespace, name, version)` coordinate. Path
/// separators and other awkward characters are replaced so the ref lives in a
/// single directory level per repo and can't escape it.
pub(crate) fn coord_key(id: &ArtifactId) -> String {
    let sanitize = |s: &str| {
        s.chars()
            .map(|c| if c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-') { c } else { '_' })
            .collect::<String>()
    };
    format!(
        "{}__{}__{}",
        sanitize(id.namespace.as_deref().unwrap_or("")),
        sanitize(&id.name),
        sanitize(&id.version),
    )
}

/// A filesystem-safe **single path segment** for a `to_repo` name.
///
/// `to_repo` arrives verbatim from an untrusted `PromoteArtifactRequest`, so it
/// MUST NOT be joined into the PROD store path as-is: a value like
/// `"../../../../tmp/pwn"` would let a seal climb out of `refs/` and write (or a
/// probe read) an arbitrary-location `.ref` file. This collapses path
/// separators and every other awkward character to `_` (path separators can
/// therefore never survive), and neutralises the dot-only escapes (`.`/`..`, and
/// the empty string), so the result is always exactly one directory level that
/// stays inside `refs/`. A legitimate repo name (a RON node key like
/// `rust-prod`) passes through unchanged; the same philosophy already guards the
/// coordinate via [`coord_key`].
pub(crate) fn sanitize_repo(to_repo: &str) -> String {
    let s: String = to_repo
        .chars()
        .map(|c| if c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-') { c } else { '_' })
        .collect();
    match s.as_str() {
        "" | "." | ".." => "_".to_string(),
        _ => s,
    }
}

/// The immutable, content-addressed PROD store.
///
/// Layout under `prod_root`:
/// ```text
/// blobs/<blake3-hex>              — the artifact bytes, content-addressed (dedup)
/// refs/<to_repo>/<coord>.ref      — the seal: JSON PromotionRecord (exclusive)
/// ```
/// A blob is written idempotently (same content ⇒ same file). A ref is created
/// with `create_new`, so a coordinate can be sealed **exactly once** — the seal
/// is the immutability commit and the re-promote guard, race-safe.
pub struct PromotionEngine {
    prod_root: PathBuf,
    /// The optional policy gate (design §4). `None` ⇒ promotion is unconditional
    /// (the default, unconfigured path — unchanged). `Some` ⇒ the boundary runs
    /// the real scan and, at `Enforcement::Block`, fails closed.
    gate: Option<PolicyGate>,
    /// The optional durable **job ledger** (jera). `None` ⇒ no job row is written
    /// (the default, unconfigured path — behaviour identical to before). `Some` ⇒
    /// every [`seal`](Self::seal) is recorded as one tracked job (`running →
    /// done|failed` + timing) in the shared `jera::jobs::JobStore`, so a reader
    /// can list the server's recent promotions. jera keeps the ledger bounded
    /// (the most recent `jera::jobs::CAP` terminal jobs) — this is the operational
    /// recent-jobs view, not the permanent record; that is holger's append-only
    /// audit log. Additive: the seal's storage transition is unchanged whether or
    /// not a ledger is attached.
    jobs: Option<Arc<jera::jobs::JobStore>>,
    /// Optional **custom-property requirement** for promotion (property-policy §).
    /// `Some((key, value))` ⇒ an artifact may only be promoted when its SOURCE
    /// coordinate carries this property (e.g. `("signed","true")`, or `("signed","")`
    /// for "any value under `signed`") — the classic "only promote signed builds"
    /// gate. `None` (default) ⇒ no property requirement. The engine only *holds*
    /// the requirement (it does no I/O); the promote boundary resolves the source
    /// coordinate's properties against the `PropertyStore` and calls
    /// [`property_requirement_met`](Self::property_requirement_met).
    require_property: Option<(String, String)>,
}

impl PromotionEngine {
    /// A store rooted at `prod_root` (created on first seal). No policy gate —
    /// promotion is unconditional (the default path; configure one with
    /// [`with_gate`](Self::with_gate) to enforce). No job ledger — seals are not
    /// recorded (attach one with [`with_job_ledger`](Self::with_job_ledger)).
    pub fn new(prod_root: impl Into<PathBuf>) -> Self {
        Self { prod_root: prod_root.into(), gate: None, jobs: None, require_property: None }
    }

    /// Require a custom **property** on the source coordinate before a promotion is
    /// allowed (property-policy §). Additive: with none set (the default) promotion
    /// is unconditional on properties. Reads `HOLGER_PROMOTE_REQUIRE_PROPERTY`
    /// (`key=value`, or bare `key`) when built via [`require_property_from_env`].
    pub fn with_required_property(mut self, key: String, value: String) -> Self {
        self.require_property = Some((key, value));
        self
    }

    /// The configured property requirement, if any (`(key, value)`).
    pub fn required_property(&self) -> Option<&(String, String)> {
        self.require_property.as_ref()
    }

    /// Parse `HOLGER_PROMOTE_REQUIRE_PROPERTY` into a `(key, value)` requirement —
    /// the env config surface, mirroring [`PolicyGate::from_env`]. `key=value`, or a
    /// bare `key` (any value). Absent/empty ⇒ `None` (no requirement). Pure over the
    /// supplied string so it is testable without touching the environment.
    pub fn parse_property_requirement(raw: &str) -> Option<(String, String)> {
        crate::properties::parse_predicate(raw)
    }

    /// Whether a `props` map satisfies this engine's property requirement. `true`
    /// when no requirement is configured (unconditional). The promote boundary
    /// passes the SOURCE coordinate's property map; the match rule is the shared
    /// [`properties::map_matches`](crate::properties::map_matches) — so a
    /// `signed=true` requirement means exactly what the search axis / retention pin
    /// mean (no twin).
    pub fn property_requirement_met(&self, props: &crate::properties::PropertyMap) -> bool {
        match &self.require_property {
            Some((key, value)) => crate::properties::map_matches(props, key, value),
            None => true,
        }
    }

    /// Attach a policy gate so promotions are gated (design §4). Additive: the
    /// gate only bites when configured, so existing callers that don't set one
    /// keep the unconditional behaviour.
    pub fn with_gate(mut self, gate: PolicyGate) -> Self {
        self.gate = Some(gate);
        self
    }

    /// Route every seal through the shared **jera job ledger** (`jera::jobs`).
    /// Additive: with no ledger attached (the default) a promotion behaves
    /// exactly as before and writes no job row; with one attached each seal opens
    /// a `promote` job that transitions `running → done` (with the sealed content
    /// address as its `result_ref`) or `running → failed` (carrying the reason),
    /// so the ledger holds a record of the server's recent promotions (jera bounds
    /// it to the most recent `jera::jobs::CAP`; the permanent record is the audit log).
    pub fn with_job_ledger(mut self, store: Arc<jera::jobs::JobStore>) -> Self {
        self.jobs = Some(store);
        self
    }

    /// Whether a policy gate is configured (the boundary will scan before it
    /// seals).
    pub fn has_gate(&self) -> bool {
        self.gate.is_some()
    }

    /// Whether a durable job ledger is attached (seals are recorded as jobs).
    pub fn has_job_ledger(&self) -> bool {
        self.jobs.is_some()
    }

    /// Run the policy gate for `id` in `ecosystem` BEFORE a seal. Fail-closed:
    /// with a gate at [`Enforcement::Block`] a `Block` verdict returns
    /// [`PromoteError::PolicyBlocked`]; `Warn` records but permits; no gate is a
    /// pass-through (`Ok`). The caller (the gRPC promote handler) invokes this
    /// at the boundary and audits the outcome.
    pub fn gate_check(&self, ecosystem: &str, id: &ArtifactId) -> Result<(), PromoteError> {
        let Some(gate) = self.gate.as_ref() else {
            return Ok(()); // unconfigured ⇒ unconditional (default path)
        };
        let (decision, reasons) = gate.evaluate(ecosystem, id);
        match (gate.enforcement, decision) {
            (Enforcement::Block, Decision::Block) => Err(PromoteError::PolicyBlocked(reasons)),
            _ => Ok(()),
        }
    }

    fn blob_path(&self, hash: &str) -> PathBuf {
        self.prod_root.join("blobs").join(hash)
    }

    fn ref_path(&self, to_repo: &str, id: &ArtifactId) -> PathBuf {
        // `to_repo` is untrusted (straight off the wire); sanitize it to a single
        // safe segment so a crafted name can never escape `refs/` (path
        // traversal). The stored `PromotionRecord.to_repo` keeps the original
        // requested string — only the on-disk path is sanitized.
        self.prod_root
            .join("refs")
            .join(sanitize_repo(to_repo))
            .join(format!("{}.ref", coord_key(id)))
    }

    /// Whether `id` is already sealed into `to_repo` (immutable — a further seal
    /// would be refused).
    pub fn is_promoted(&self, to_repo: &str, id: &ArtifactId) -> bool {
        self.ref_path(to_repo, id).exists()
    }

    /// Seal `data` for `id` into the immutable PROD store under `to_repo`.
    ///
    /// Content-addresses the bytes (blake3), writes the blob under its hash, then
    /// creates the ref **exclusively** — refusing to overwrite an already-sealed
    /// coordinate ([`PromoteError::AlreadyPromoted`]). Returns the audited record.
    ///
    /// When a **job ledger** is attached ([`with_job_ledger`](Self::with_job_ledger))
    /// the whole seal runs as one tracked `jera` job: a `running` row is written
    /// before the storage transition and flipped to `done` (with the content
    /// address as `result_ref`) or `failed` (with the reason) when it settles. The
    /// storage transition itself is byte-for-byte the same either way, so this is
    /// purely additive observability over the existing seal.
    pub fn seal(
        &self,
        from_repo: &str,
        to_repo: &str,
        id: &ArtifactId,
        data: &[u8],
        promoted_by: &str,
        promoted_at: i64,
    ) -> Result<PromotionRecord, PromoteError> {
        // Open a tracked job iff a ledger is attached. `None` ⇒ the fast, original
        // path with no ledger write.
        let job = self.jobs.as_ref().map(|store| {
            store.start(
                "promote",
                &artifact_label(id),
                to_repo,
                serde_json::json!({
                    "from_repo": from_repo,
                    "to_repo": to_repo,
                    "size": data.len(),
                    "promoted_by": promoted_by,
                }),
            )
        });

        let out = self.seal_inner(from_repo, to_repo, id, data, promoted_by, promoted_at);

        if let Some(job) = job {
            match &out {
                Ok(rec) => job.finish(
                    serde_json::json!({
                        "content_address": rec.content_address,
                        "size": rec.size,
                    }),
                    &rec.content_address,
                ),
                Err(e) => job.fail_with_detail(serde_json::json!({ "error": e.to_string() })),
            }
        }
        out
    }

    /// The storage transition itself — content-address, blob write, exclusive ref
    /// create. [`seal`](Self::seal) wraps this in the optional job ledger; every
    /// existing behaviour lives here unchanged.
    fn seal_inner(
        &self,
        from_repo: &str,
        to_repo: &str,
        id: &ArtifactId,
        data: &[u8],
        promoted_by: &str,
        promoted_at: i64,
    ) -> Result<PromotionRecord, PromoteError> {
        let content_address = blake3::hash(data).to_hex().to_string();
        let ref_path = self.ref_path(to_repo, id);

        // Immutability: the coordinate must not already be sealed.
        if ref_path.exists() {
            return Err(PromoteError::AlreadyPromoted);
        }

        // Content-addressed blob — write atomically (tmp + rename); idempotent, so
        // identical content shared across coordinates is stored once.
        let blob_path = self.blob_path(&content_address);
        if let Some(parent) = blob_path.parent() {
            fs::create_dir_all(parent)?;
        }
        if !blob_path.exists() {
            let tmp = blob_path.with_extension("tmp");
            fs::write(&tmp, data)?;
            fs::rename(&tmp, &blob_path)?;
        }

        let record = PromotionRecord {
            artifact: artifact_label(id),
            from_repo: from_repo.to_string(),
            to_repo: to_repo.to_string(),
            content_address,
            size: data.len() as u64,
            promoted_by: promoted_by.to_string(),
            promoted_at,
        };

        // The ref is the seal. `create_new` fails if it already exists, so two
        // concurrent seals of the same coordinate can't both win — the loser gets
        // AlreadyPromoted, preserving immutability under a race.
        if let Some(parent) = ref_path.parent() {
            fs::create_dir_all(parent)?;
        }
        let json = record.to_json();
        match fs::OpenOptions::new().write(true).create_new(true).open(&ref_path) {
            Ok(mut f) => {
                use std::io::Write;
                f.write_all(json.as_bytes())?;
                f.flush()?;
            }
            Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
                return Err(PromoteError::AlreadyPromoted);
            }
            Err(e) => return Err(PromoteError::Io(e)),
        }
        Ok(record)
    }

    /// Read a promoted artifact's bytes back, re-verifying the blob's blake3
    /// against the sealed ref (integrity on read). `None` when not promoted;
    /// `Err` on IO or an integrity mismatch (a tampered blob).
    pub fn fetch(&self, to_repo: &str, id: &ArtifactId) -> Result<Option<Vec<u8>>, PromoteError> {
        let ref_path = self.ref_path(to_repo, id);
        if !ref_path.exists() {
            return Ok(None);
        }
        let ref_json = fs::read_to_string(&ref_path)?;
        let want = PromotionRecord::content_address_from_json(&ref_json)
            .ok_or_else(|| PromoteError::Io(std::io::Error::other("corrupt ref: no content_address")))?;
        let data = fs::read(self.blob_path(&want))?;
        let got = blake3::hash(&data).to_hex().to_string();
        if got != want {
            return Err(PromoteError::Io(std::io::Error::other(
                "integrity: promoted blob blake3 does not match the sealed ref",
            )));
        }
        Ok(Some(data))
    }
}

impl PromotionRecord {
    /// Serialize to a compact JSON line (the ref body / audit detail). Hand-rolled
    /// to avoid a serde derive on a type that never needs to round-trip fully —
    /// only `content_address` is read back (for the integrity check).
    fn to_json(&self) -> String {
        let esc = |s: &str| s.replace('\\', "\\\\").replace('"', "\\\"");
        format!(
            r#"{{"artifact":"{}","from_repo":"{}","to_repo":"{}","content_address":"{}","size":{},"promoted_by":"{}","promoted_at":{}}}"#,
            esc(&self.artifact),
            esc(&self.from_repo),
            esc(&self.to_repo),
            esc(&self.content_address),
            self.size,
            esc(&self.promoted_by),
            self.promoted_at,
        )
    }

    /// Pull the `content_address` value out of a ref JSON line (the only field the
    /// read path needs). Returns `None` if absent/malformed.
    fn content_address_from_json(json: &str) -> Option<String> {
        let needle = "\"content_address\":\"";
        let start = json.find(needle)? + needle.len();
        let rest = &json[start..];
        let end = rest.find('"')?;
        Some(rest[..end].to_string())
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn id(name: &str, version: &str) -> ArtifactId {
        ArtifactId { namespace: None, name: name.into(), version: version.into() }
    }

    /// A seal transitions the bytes into the content-addressed store: the record
    /// carries the blake3 address, the blob is stored under that hash, and the
    /// bytes read back intact.
    #[test]
    fn seal_transitions_and_content_addresses() {
        let dir = tempfile::tempdir().unwrap();
        let engine = PromotionEngine::new(dir.path());
        let data = b"the-promoted-crate-bytes";
        let expected = blake3::hash(data).to_hex().to_string();

        let rec = engine
            .seal("rust-dev", "rust-prod", &id("serde", "1.0.0"), data, "alice@corp", 1_700_000_000)
            .expect("seal ok");

        assert_eq!(rec.content_address, expected, "record carries the blake3 content address");
        assert_eq!(rec.size, data.len() as u64);
        assert_eq!(rec.from_repo, "rust-dev");
        assert_eq!(rec.to_repo, "rust-prod");
        assert_eq!(rec.promoted_by, "alice@corp");
        // The blob lives under its content address, and reads back intact.
        assert!(dir.path().join("blobs").join(&expected).is_file(), "blob stored under blake3 hash");
        let back = engine.fetch("rust-prod", &id("serde", "1.0.0")).unwrap();
        assert_eq!(back.as_deref(), Some(data.as_slice()));
    }

    /// A promoted artifact is IMMUTABLE: a second seal of the same coordinate is
    /// refused, whether the bytes are identical or different.
    #[test]
    fn promoted_artifact_is_immutable() {
        let dir = tempfile::tempdir().unwrap();
        let engine = PromotionEngine::new(dir.path());
        engine
            .seal("dev", "prod", &id("tokio", "1.38.0"), b"v1", "admin", 1)
            .expect("first seal ok");
        assert!(engine.is_promoted("prod", &id("tokio", "1.38.0")));

        // Re-promote with different bytes — refused (immutable).
        let err = engine
            .seal("dev", "prod", &id("tokio", "1.38.0"), b"tampered", "admin", 2)
            .expect_err("re-promote must be refused");
        assert!(matches!(err, PromoteError::AlreadyPromoted));

        // Re-promote with identical bytes — still refused (the coordinate is sealed).
        let err = engine
            .seal("dev", "prod", &id("tokio", "1.38.0"), b"v1", "admin", 3)
            .expect_err("re-promote must be refused even if identical");
        assert!(matches!(err, PromoteError::AlreadyPromoted));

        // The original bytes are untouched.
        assert_eq!(engine.fetch("prod", &id("tokio", "1.38.0")).unwrap().as_deref(), Some(b"v1".as_slice()));
    }

    /// A crafted `to_repo` full of path-traversal cannot escape `refs/`: the seal
    /// stays contained under `prod_root`, and nothing is written outside it (the
    /// path-traversal / arbitrary-file-write finding). The record still reports
    /// the requested name verbatim; only the on-disk location is sanitized.
    #[test]
    fn crafted_to_repo_cannot_escape_prod_root() {
        let root = tempfile::tempdir().unwrap();
        let prod = root.path().join("prod");
        let engine = PromotionEngine::new(&prod);

        // A sibling directory an escape would land in — must stay empty.
        let outside = root.path().join("outside");
        fs::create_dir_all(&outside).unwrap();

        for (i, evil) in ["../../../../outside/pwn", "..", ".", "a/b/c", "../outside"]
            .into_iter()
            .enumerate()
        {
            // Distinct coordinate per case so two evils that sanitize to the same
            // segment (".", "..") don't collide on the immutability guard.
            let coord = id("serde", &format!("1.0.{i}"));
            let rec = engine
                .seal(evil, evil, &coord, b"bytes", "attacker", 1)
                .expect("seal returns (path is contained, not that it fails)");
            // The record keeps the requested (untrusted) name verbatim…
            assert_eq!(rec.to_repo, evil);
            // …but the ref file is physically inside prod_root/refs, never outside.
            let ref_path = engine.ref_path(evil, &coord);
            let canon = ref_path.canonicalize().expect("ref exists");
            assert!(
                canon.starts_with(prod.canonicalize().unwrap()),
                "ref for {evil:?} escaped prod_root: {canon:?}"
            );
        }
        // Nothing was ever written into the sibling directory.
        assert_eq!(fs::read_dir(&outside).unwrap().count(), 0, "no file escaped prod_root");
    }

    // ── jera job ledger — every promotion is a tracked, durable job ───────────

    /// Open a fresh jera job ledger under `root`.
    fn ledger(root: &std::path::Path) -> Arc<jera::jobs::JobStore> {
        Arc::new(jera::jobs::JobStore::open(root).expect("open jera job ledger"))
    }

    /// A successful seal with a ledger attached writes ONE `promote` job that
    /// settles `done`, carrying the sealed content address as its `result_ref`.
    #[test]
    fn seal_records_a_done_job_in_the_ledger() {
        let store_dir = tempfile::tempdir().unwrap();
        let led_dir = tempfile::tempdir().unwrap();
        let store = ledger(led_dir.path());
        let engine = PromotionEngine::new(store_dir.path()).with_job_ledger(store.clone());
        assert!(engine.has_job_ledger());

        let data = b"promoted-bytes";
        let expected = blake3::hash(data).to_hex().to_string();
        let rec = engine
            .seal("rust-dev", "rust-prod", &id("serde", "1.0.0"), data, "alice@corp", 1)
            .expect("seal ok");
        assert_eq!(rec.content_address, expected);

        let jobs = store.list(&jera::jobs::JobSelector::All).expect("list jobs");
        assert_eq!(jobs.len(), 1, "exactly one promotion job recorded");
        let job = &jobs[0];
        assert_eq!(job.kind, "promote");
        assert_eq!(job.status, "done", "successful seal settles done");
        assert_eq!(job.target, artifact_label(&id("serde", "1.0.0")));
        assert_eq!(job.workspace, "rust-prod");
        assert_eq!(job.result_ref, expected, "done job carries the content address");
        assert!(job.elapsed_ms.is_some(), "a terminal job has timing");
    }

    /// A refused re-promote (immutable coordinate) with a ledger attached records
    /// the second attempt as a `failed` job — the durable trail shows both the
    /// success and the refused overwrite.
    #[test]
    fn refused_reseal_records_a_failed_job() {
        let store_dir = tempfile::tempdir().unwrap();
        let led_dir = tempfile::tempdir().unwrap();
        let store = ledger(led_dir.path());
        let engine = PromotionEngine::new(store_dir.path()).with_job_ledger(store.clone());

        engine.seal("dev", "prod", &id("tokio", "1.38.0"), b"v1", "admin", 1).expect("first seal");
        engine
            .seal("dev", "prod", &id("tokio", "1.38.0"), b"v2", "admin", 2)
            .expect_err("re-promote refused");

        let jobs = store.list(&jera::jobs::JobSelector::All).expect("list jobs");
        assert_eq!(jobs.len(), 2, "both attempts recorded");
        let statuses: std::collections::BTreeSet<_> = jobs.iter().map(|j| j.status.as_str()).collect();
        assert!(statuses.contains("done"), "the first seal is done");
        assert!(statuses.contains("failed"), "the refused re-seal is failed");
    }

    /// Without a ledger (the default) no job is written and the seal behaves
    /// exactly as before — the ledger is purely additive.
    #[test]
    fn seal_without_ledger_writes_no_job() {
        let store_dir = tempfile::tempdir().unwrap();
        let engine = PromotionEngine::new(store_dir.path());
        assert!(!engine.has_job_ledger());
        // Seal succeeds identically to the plain path.
        engine.seal("dev", "prod", &id("anyhow", "1.0.0"), b"bytes", "admin", 1).expect("seal ok");
        // (No ledger to inspect — the point is that nothing is required.)
    }

    /// Identical content promoted under two coordinates is stored once (content
    /// addressing dedups), but each coordinate has its own ref.
    #[test]
    fn identical_content_dedups_but_refs_are_distinct() {
        let dir = tempfile::tempdir().unwrap();
        let engine = PromotionEngine::new(dir.path());
        let data = b"shared-bytes";
        engine.seal("dev", "prod", &id("a", "1"), data, "admin", 1).unwrap();
        engine.seal("dev", "prod", &id("b", "1"), data, "admin", 1).unwrap();
        let blobs: Vec<_> = fs::read_dir(dir.path().join("blobs"))
            .unwrap()
            .filter_map(|e| e.ok())
            .filter(|e| e.path().extension().is_none())
            .collect();
        assert_eq!(blobs.len(), 1, "identical content stored once");
        assert!(engine.is_promoted("prod", &id("a", "1")));
        assert!(engine.is_promoted("prod", &id("b", "1")));
    }

    // ── Policy gate (design §4) — REAL scan seam, not `demo_*` ────────────────

    use crate::security::{Advisory, AdvisoryDb, LicensePolicy, Policy, ProvenancePolicy, Severity};

    /// A REAL scanner (NOT `demo_scanner`) built for the gate tests: one High
    /// `cargo` advisory on `badcrate@1.0.0`, a permissive license policy, and
    /// provenance ignored. A High vuln scores `Block` in modgunn's mapping.
    fn real_blocking_scanner() -> OsvScanner {
        let db = AdvisoryDb::new(
            "holger-promote-gate-test-db",
            vec![Advisory::basic(
                "RUSTSEC-TEST-0001",
                "cargo",
                "badcrate",
                vec!["1.0.0".into()],
                Severity::High,
                "test high-severity advisory",
            )],
        );
        let policy = Policy {
            version: "gate-test".into(),
            license: LicensePolicy { allow: Vec::new(), deny: Vec::new(), on_unknown: Decision::Pass },
            provenance: ProvenancePolicy::Ignore,
            ..Default::default()
        };
        OsvScanner::new(policy, db)
    }

    /// With a gate at `Block`, promoting a vulnerable coordinate is REFUSED at the
    /// boundary (`PolicyBlocked`) — the gate is real, not pass-through. A clean
    /// coordinate through the same gate passes.
    #[test]
    fn gate_blocks_vulnerable_and_passes_clean() {
        let dir = tempfile::tempdir().unwrap();
        let engine = PromotionEngine::new(dir.path())
            .with_gate(PolicyGate::new(real_blocking_scanner(), Enforcement::Block));
        assert!(engine.has_gate());

        let bad = engine.gate_check("cargo", &id("badcrate", "1.0.0"));
        assert!(
            matches!(bad, Err(PromoteError::PolicyBlocked(_))),
            "a High-severity advisory must block the promotion, got {bad:?}"
        );

        let clean = engine.gate_check("cargo", &id("serde", "1.0.0"));
        assert!(clean.is_ok(), "a clean coordinate must pass the gate");
    }

    /// The REAL imported-feed path end to end: an OSV/GHSA JSON snapshot is
    /// imported into the advisory DB, wired into a `Block` gate, and it refuses
    /// promotion of the affected coordinate while passing an unaffected one. No
    /// `demo_*` anywhere in this path.
    #[test]
    fn gate_from_imported_osv_snapshot_blocks_affected() {
        let dir = tempfile::tempdir().unwrap();
        let osv = dir.path().join("advisories.json");
        std::fs::write(
            &osv,
            r#"{ "id": "CVE-2099-0001", "summary": "test advisory",
                 "affected": [ { "package": { "ecosystem": "crates.io", "name": "badcrate" },
                                 "versions": ["1.0.0"] } ],
                 "database_specific": { "severity": "HIGH" } }"#,
        )
        .unwrap();
        let db = security::advisory_db_from_osv_file(osv.to_str().unwrap()).expect("import OSV");
        let scanner = OsvScanner::new(security::promotion_policy(Vec::new()), db);
        let engine = PromotionEngine::new(dir.path())
            .with_gate(PolicyGate::new(scanner, Enforcement::Block));

        assert!(
            matches!(
                engine.gate_check("cargo", &id("badcrate", "1.0.0")),
                Err(PromoteError::PolicyBlocked(_))
            ),
            "the imported advisory must block the affected coordinate"
        );
        assert!(
            engine.gate_check("cargo", &id("goodcrate", "9.9.9")).is_ok(),
            "an unaffected coordinate passes the imported-feed gate"
        );
    }

    /// A gate at `Warn` records the verdict but never refuses — even a blocking
    /// verdict is permitted through.
    #[test]
    fn warn_gate_permits_even_a_blocking_verdict() {
        let dir = tempfile::tempdir().unwrap();
        let engine = PromotionEngine::new(dir.path())
            .with_gate(PolicyGate::new(real_blocking_scanner(), Enforcement::Warn));
        assert!(
            engine.gate_check("cargo", &id("badcrate", "1.0.0")).is_ok(),
            "Warn enforcement must permit the promotion"
        );
    }

    /// The DEFAULT, unconfigured path is UNCHANGED: no gate ⇒ `gate_check` is a
    /// pass-through, and a full seal still succeeds without any scan.
    #[test]
    fn no_gate_is_pass_through_default_unchanged() {
        let dir = tempfile::tempdir().unwrap();
        let engine = PromotionEngine::new(dir.path());
        assert!(!engine.has_gate());
        // Even a coordinate that WOULD block under a gate passes when none is set.
        assert!(engine.gate_check("cargo", &id("badcrate", "1.0.0")).is_ok());
        // And the seal path is untouched.
        assert!(engine
            .seal("dev", "prod", &id("badcrate", "1.0.0"), b"bytes", "admin", 1)
            .is_ok());
    }

    /// A tampered blob is caught on read by the blake3 integrity re-check.
    #[test]
    fn tampered_blob_fails_integrity_on_read() {
        let dir = tempfile::tempdir().unwrap();
        let engine = PromotionEngine::new(dir.path());
        let data = b"honest-bytes";
        let rec = engine.seal("dev", "prod", &id("x", "1"), data, "admin", 1).unwrap();
        // Overwrite the content-addressed blob with different bytes.
        fs::write(dir.path().join("blobs").join(&rec.content_address), b"evil").unwrap();
        assert!(engine.fetch("prod", &id("x", "1")).is_err(), "tamper must fail integrity");
    }

    // ─── Property-requirement gate (property-policy §) ───────────────────────

    /// `HOLGER_PROMOTE_REQUIRE_PROPERTY` parsing: `key=value`, bare `key`, and the
    /// degenerate forms.
    #[test]
    fn parse_property_requirement_forms() {
        assert_eq!(
            PromotionEngine::parse_property_requirement("signed=true"),
            Some(("signed".into(), "true".into()))
        );
        assert_eq!(
            PromotionEngine::parse_property_requirement("  signed "),
            Some(("signed".into(), String::new())),
            "bare key ⇒ any value"
        );
        assert_eq!(PromotionEngine::parse_property_requirement(""), None, "empty ⇒ no requirement");
        assert_eq!(PromotionEngine::parse_property_requirement("=true"), None, "empty key ⇒ none");
    }

    /// RED-when-broken: with a `signed=true` requirement, a props map WITHOUT it is
    /// refused (`property_requirement_met` false); with it, allowed. No requirement
    /// ⇒ always allowed.
    #[test]
    fn property_requirement_met_gates_on_the_map() {
        use crate::properties::PropertyMap;
        let engine = PromotionEngine::new("/tmp/unused-prod")
            .with_required_property("signed".into(), "true".into());

        // Missing the property → NOT met.
        let mut empty = PropertyMap::new();
        assert!(!engine.property_requirement_met(&empty), "no property ⇒ refused");
        // Wrong value → NOT met.
        empty.insert("signed".into(), vec!["false".into()]);
        assert!(!engine.property_requirement_met(&empty), "wrong value ⇒ refused");
        // Correct value → met.
        let mut ok = PropertyMap::new();
        ok.insert("signed".into(), vec!["true".into()]);
        assert!(engine.property_requirement_met(&ok), "signed=true ⇒ allowed");

        // No requirement configured ⇒ always allowed.
        let open = PromotionEngine::new("/tmp/unused-prod2");
        assert!(open.property_requirement_met(&PropertyMap::new()), "no requirement ⇒ unconditional");
        assert_eq!(open.required_property(), None);
    }
}