server/lib/src/promote/mod.rs
//! 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 crate::lock::RwLockOrRecover as _;
use std::fs;
use std::path::PathBuf;
use std::sync::Arc;
/// The Artifactory-shaped **stage ladder** (`sparring → staging → release`) layered
/// over the pointer-move seal below: which hops are legal, which run the gate, and
/// the append-only lineage that makes a promotion queryable and reversible without
/// ever rewriting a sealed generation.
pub mod stages;
/// **The airgap**: the signed export of the approved set (verdict computed from
/// the stored evidence) and its verified import on the secure side.
pub mod airgap;
/// **The cargo door of the intake chain** (lane HOL-CARGO-PUBLISH-DOOR): `cargo publish` ingests into
/// `incoming`, the sparse index and the downloads are a view of the AVAILABLE set, yanks are
/// recorded for AVAILABLE versions only. See its docs.
pub mod cargo;
/// **The AVAILABLE catalog** (HOL-PERF-1): the intake record's reduction in memory, kept
/// current by every append and, across processes, by the record's journal.
pub(crate) mod catalog;
/// **Separation of duties** on the intake chain: Developer, Importer,
/// SecurityApprover, RepositoryAdmin, Auditor; the importer never approves;
/// N distinct, attested approvers. See its docs.
pub mod duties;
/// The **gate seam** — a [`gate::Gate`] trait over a [`gate::PromotionContext`]
/// that carries the principal, the hop and the whole promoted set, with each
/// member's DIGEST and BYTES as required fields. It replaces the environment-variable
/// `PolicyGate` that used to live in this file. See the module docs for what
/// that struct cost.
pub mod gate;
/// **The package identity, as a purl** (Holger 3.2, H32-PURL): one required `identify()` per
/// format; the intake chain refuses a package without one before any other step.
pub mod identity;
/// **The intake chain** — UNTRUSTED → INGESTED → SCANNED → POLICY_CHECKED →
/// APPROVED → SEALED → PROMOTED → AVAILABLE, with QUARANTINED and BLOCKED as
/// holds. Layered over the ladder, the gates and the seal below; see its docs.
pub mod intake;
/// **The whole join between a loaded meter guest and the gate seam**: one
/// `WasmGate` holding an `Arc<dyn MeterRuntime>`, with one `impl Gate` for it.
/// Behind `meter-guests`, because without that feature there is no
/// `MeterRuntime` to hold.
#[cfg(feature = "meter-guests")]
pub mod meter_gate;
/// **The npm door of the intake chain**: `npm publish` ingests into `incoming`, the packument
/// and the tarballs are a view of the AVAILABLE set, dist-tags name AVAILABLE versions only,
/// unpublish is refused by policy. See its docs.
pub mod npm;
/// Scanned pushes into a writable `oci` registry (lane HOL-OCI-EVIDENCE).
#[cfg(feature = "intake")]
pub mod oci_push;
/// **The index rows** (lane HOL-PERF-2): what the scan worker read out of each package for
/// its format's index, stored at intake by sha256; every index renders from them.
pub mod rows;
/// **Waivers** (HOL-VEX-WAIVER): one advisory on one package coordinate, requested on a
/// BLOCKED package, approved by the chain's N approvers, until a date; exported as OpenVEX.
pub mod waiver;
pub use gate::{
Gate, GateCatalog, GateConfigError, GateSet, PromotedArtifact, PromotionContext,
ProvenanceGate, ProvenancePolicy,
};
pub use stages::{Hop, LadderError, Stage, StageEntry, StageLadder, StageLog, StageOp};
use traits::ArtifactId;
// The scan seam itself now lives behind `gate::OsvGate`; this module keeps only
// the enforcement level and the engine that carries the resolved set.
/// 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 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, thiserror::Error)]
pub enum PromoteError {
/// The coordinate is already sealed in the target PROD repo — immutable,
/// overwrite refused (the re-promote guard).
#[error("artifact already promoted (immutable — overwrite refused)")]
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.
#[error("promotion blocked by policy gate: {}", .0.join("; "))]
PolicyBlocked(Vec<String>),
/// Filesystem failure writing the blob or ref.
#[error("promotion store io error: {0}")]
Io(#[source] std::io::Error),
/// The hop is not legal on the configured **stage ladder** — it skips a rung,
/// goes backwards, or names a repository that is not a stage. Refused BEFORE
/// anything is sealed, so an illegal promotion writes nothing at all.
#[error(transparent)]
Ladder(stages::LadderError),
/// The promotion TARGET names a repository this server does not serve.
///
/// `to_repo` arrives verbatim from an untrusted request and used to be
/// SANITIZED ([`sanitize_repo`]) into a directory segment and never RESOLVED —
/// two different jobs that were doing one. Containment is not existence: a
/// promotion to `made-up-repo` passed containment, sealed the bytes into
/// `refs/made-up-repo/`, returned `success: true` and a content address, and
/// nothing served them. The caller was told a promotion happened and no
/// repository had it.
///
/// Refused BY NAME now, before the ladder and before anything touches disk.
/// Carries the served set so the message can say what WOULD have been legal.
#[error(
"promotion target '{to_repo}' is not a repository this server serves ({})",
served_words(served)
)]
UnknownRepository {
to_repo: String,
served: Vec<String>,
},
}
/// `served: a, b` or `it serves none`, for [`PromoteError::UnknownRepository`].
fn served_words(served: &[String]) -> String {
if served.is_empty() {
"it serves none".to_string()
} else {
format!("served: {}", served.join(", "))
}
}
impl PromoteError {
/// The FROZEN `holger-promote-<n>` code for this failure — the machine half of
/// the dual channel (`Display` below is the human half). A `Ladder` failure
/// DELEGATES to the inner [`stages::LadderError::code`], so the specific rung
/// verdict survives the wrapping instead of collapsing into one generic code.
///
/// Exhaustive on purpose: a new `PromoteError` variant without a code is a
/// COMPILE error.
pub fn code(&self) -> &'static holger_errcode::ErrCode {
match self {
PromoteError::AlreadyPromoted => &holger_errcode::PROMOTE_ALREADY_PROMOTED,
PromoteError::PolicyBlocked(_) => &holger_errcode::PROMOTE_POLICY_BLOCKED,
PromoteError::Io(_) => &holger_errcode::PROMOTE_IO,
PromoteError::Ladder(e) => e.code(),
PromoteError::UnknownRepository { .. } => &holger_errcode::PROMOTE_UNKNOWN_REPOSITORY,
}
}
/// BOTH channels in one string.
pub fn tagged(&self) -> String {
self.code().tag(self)
}
}
impl From<std::io::Error> for PromoteError {
fn from(e: std::io::Error) -> Self {
PromoteError::Io(e)
}
}
impl From<stages::LadderError> for PromoteError {
fn from(e: stages::LadderError) -> Self {
PromoteError::Ladder(e)
}
}
/// Human-readable artifact label used in the audit log and the record.
pub(crate) 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),
)
}
/// Is `name` ALREADY the exact single path segment it would be stored under?
///
/// [`sanitize_repo`] is deliberately total — it maps every string to something
/// writable — and therefore many-to-one: `rust/prod`, `rust:prod` and
/// `rust_prod` all collapse to `rust_prod`. That is the right behaviour for the
/// PROMOTE path, where the destination repository has already been resolved and
/// authorized as an object.
///
/// It is the WRONG behaviour anywhere a caller's raw string is both the
/// authorization subject and the storage key, because those two then disagree:
/// the ACL and the admin floor are consulted for `rust/prod` (which matches no
/// per-repo rule, so it takes the permissive default) while the bytes land under
/// `rust_prod` (a real repository the caller may be denied). This predicate is
/// how a caller refuses that divergence instead of papering over it: if a name
/// is not already its own storage key, it is not a name this server issued, and
/// the request is refused rather than redirected.
///
/// Empty is not a valid segment (`sanitize_repo` would turn it into `_`).
///
/// This is deliberately asked about the REPOSITORY only, not about the artifact
/// coordinate. The coordinate is also stored through a many-to-one sanitizer
/// (`coord_key`), so `serde/x` and `serde_x` do share one metadata document —
/// but both sit inside the same repository, and holger's ACLs are per-repository,
/// so that collision crosses no authorization boundary and refusing it would
/// reject artifact names no ecosystem forbids. The day ACLs become per-artifact,
/// the coordinate needs the same treatment (or an injective key) and this comment
/// is the record that it was considered and scoped out, not missed.
pub(crate) fn is_own_storage_key(name: &str) -> bool {
!name.is_empty() && sanitize_repo(name) == name
}
/// 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 promotion job ledger, as a slot a data-set growth can empty.**
///
/// `<data_root>/holger-jobs/jobs.redb` is a redb file the engine holds OPEN for
/// the process's life — one of the four descriptors that made `/data`
/// unmountable (MEASURED 2026-09-27 over holger-registers-own-plugin: the audit
/// segment, this ledger, the job queue's, every znippy archive). The growth
/// (`crate::grow`) [`release`](Self::release)s it before the unmount and
/// [`reacquire`](Self::reacquire)s it after; a seal in between runs un-recorded,
/// exactly as a server whose ledger could not be opened does — and no seal
/// happens in between, because the growth holds every admin mutation first.
pub struct JobLedger {
/// Where it reopens. `None` for a ledger handed in already open (tests): it
/// cannot be released, and says so.
root: Option<std::path::PathBuf>,
store: std::sync::RwLock<Option<Arc<jera::jobs::JobStore>>>,
}
impl JobLedger {
/// The open store, `None` while released.
pub fn get(&self) -> Option<Arc<jera::jobs::JobStore>> {
self.store.read_or_recover().clone()
}
/// Close `jobs.redb`: the slot is emptied and the store dropped once no seal
/// holds it (the growth has already held every seal off).
pub fn release(&self) -> Result<(), String> {
let Some(root) = &self.root else {
return Err("the promotion job ledger was handed in open and has no path to reopen it from, so it is not released".into());
};
let taken = self.store.write_or_recover().take();
if let Some(store) = taken {
if Arc::strong_count(&store) > 1 {
// Put it back: somebody is mid-seal, and a redb file closed under
// them is a torn ledger row.
*self.store.write_or_recover() = Some(store);
return Err(format!(
"the promotion job ledger {}/jobs.redb is in use by a seal right now",
root.display()
));
}
}
Ok(())
}
/// Open `<root>/jobs.redb` again.
pub fn reacquire(&self) -> Result<(), String> {
let Some(root) = &self.root else {
return Ok(());
};
let mut slot = self.store.write_or_recover();
if slot.is_none() {
let store = jera::jobs::JobStore::open(root).map_err(|e| {
format!(
"reopen the promotion job ledger {}/jobs.redb: {e}",
root.display()
)
})?;
*slot = Some(Arc::new(store));
}
Ok(())
}
}
/// 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 resolved [`GateSet`] for this engine (design §4). Empty ⇒ no gate is
/// configured, which is NOT the same thing as "every gate allowed": the
/// [`nornir_gate::Decision`] an empty set produces says "none configured" in
/// its summary, so the audit record distinguishes the two.
///
/// The set is resolved at boot from the RON `promotion` block against
/// [`GateCatalog`]. A rung naming a gate this build does not carry is a boot
/// error (HOL-GUN-007) and never a silent skip.
gates: GateSet,
/// 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<JobLedger>>,
/// 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)>,
/// The **stage ladder** this engine promotes along (`sparring → staging →
/// release`). Empty by default ⇒ no stage model: `seal` judges nothing extra and
/// behaves exactly as it did before stages existed. Configured ⇒ every hop is
/// checked against it before anything is written, and every successful hop is
/// recorded in the lineage below.
ladder: stages::StageLadder,
/// The ladder's name — the lineage namespace (one ecosystem's ladder per name).
ladder_name: String,
/// The append-only stage lineage, rooted at `<prod_root>/lineage`. Present iff a
/// ladder is configured.
lineage: Option<stages::StageLog>,
/// The set of repositories this server actually SERVES — the resolver for
/// `to_repo`.
///
/// `None` (the default, and what [`new`](Self::new) builds) ⇒ no registry is
/// attached, so the engine cannot resolve and does not judge: the behaviour is
/// byte-for-byte what it was, which is what the persistence and store-level unit
/// tests exercise. `Some` ⇒ a `to_repo` outside the set is
/// [`UnknownRepository`](PromoteError::UnknownRepository), refused before the
/// ladder and before any disk write.
///
/// The SERVER always attaches one (`HolgerConfig::promotion_engine`), so both
/// surfaces that promote — the gRPC `PromoteArtifact` RPC and the
/// `holger-server promote` CLI verb — resolve against the same served set.
served: Option<std::collections::BTreeSet<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(),
gates: GateSet::empty(),
jobs: None,
require_property: None,
ladder: stages::StageLadder::default(),
ladder_name: String::new(),
lineage: None,
served: None,
}
}
/// Attach the set of repositories this server SERVES, so `to_repo` is RESOLVED
/// and not merely sanitized into a directory name.
///
/// Without it a promotion to a repository that does not exist SUCCEEDED: the
/// name was collapsed to a filesystem-safe segment, the bytes were sealed under
/// `refs/<that segment>/`, and the caller got `success: true` and a content
/// address for a promotion no repository could serve. Sanitizing answers "can
/// this name escape `refs/`"; it never answered "does this repository exist".
///
/// With it, an unserved target is [`PromoteError::UnknownRepository`] — refused
/// BY NAME, before the ladder and before anything is written.
pub fn with_served_repositories<I, S>(mut self, names: I) -> Self
where
I: IntoIterator<Item = S>,
S: Into<String>,
{
self.served = Some(names.into_iter().map(Into::into).collect());
self
}
/// The served repository set, if a registry is attached.
pub fn served_repositories(&self) -> Option<&std::collections::BTreeSet<String>> {
self.served.as_ref()
}
/// 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. `Holger::promotion_engine` sets it from
/// `promotion.require_property` (`key=value`, or bare `key`).
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,
}
}
/// Promote along a **stage ladder** (`sparring → staging → release`), recording
/// every hop in an append-only lineage under `<prod_root>/lineage/<name>`.
///
/// Additive in the strict sense: with no ladder configured (the default) `seal`
/// judges nothing extra, writes no lineage, and is byte-for-byte the promotion
/// holger has always done. With one configured, an illegal hop is refused BEFORE
/// any blob or ref is written — a rejected promotion leaves the store untouched.
pub fn with_ladder(mut self, name: impl Into<String>, ladder: stages::StageLadder) -> Self {
self.lineage = Some(stages::StageLog::new(self.prod_root.join("lineage")));
self.ladder = ladder;
self.ladder_name = name.into();
self
}
/// The configured ladder (empty when there is no stage model).
pub fn ladder(&self) -> &stages::StageLadder {
&self.ladder
}
/// Where `id` currently sits on the ladder — the highest rung it has reached and
/// not been withdrawn from. `None` with no ladder configured, or when the
/// artifact has never been promoted.
pub fn current_stage(&self, id: &ArtifactId) -> Option<String> {
self.lineage
.as_ref()?
.current_stage(&self.ladder, &self.ladder_name, id)
}
/// How `id` got where it is: every promotion and every withdrawal, oldest first.
pub fn stage_history(&self, id: &ArtifactId) -> Vec<stages::StageEntry> {
self.lineage
.as_ref()
.map(|l| l.history(&self.ladder_name, id))
.unwrap_or_default()
}
/// **Withdraw** `id` from `stage` — the reversal.
///
/// Appends a `Withdrawn` entry to the lineage. The sealed blob and its ref are
/// NOT touched: an immutable generation is never rewritten, so an artifact that
/// already crossed an airgap does not become invalid because it was withdrawn on
/// the other side. What changes is [`current_stage`](Self::current_stage)'s
/// answer — it falls back to the rung below.
pub fn withdraw(
&self,
id: &ArtifactId,
stage: &str,
actor: &str,
at: i64,
reason: &str,
) -> Result<stages::StageEntry, PromoteError> {
let log = self
.lineage
.as_ref()
.ok_or_else(|| PromoteError::Ladder(stages::LadderError::UnknownStage(stage.into())))?;
if self.ladder.stage(stage).is_none() {
return Err(PromoteError::Ladder(stages::LadderError::UnknownStage(
stage.into(),
)));
}
Ok(log.withdraw(&self.ladder_name, id, stage, actor, at, reason)?)
}
/// Record that `id` stands on the rung of `to_repo`, arriving from the rung of
/// `from_repo`, WITHOUT a seal: the intake chain's hops onto the quarantine and
/// approved rungs, which hold no sealed bytes. The hop is judged by the ladder
/// first, exactly as [`seal`](Self::seal) judges its hop, and refused before
/// anything is written.
pub fn record_hop(
&self,
from_repo: &str,
to_repo: &str,
id: &ArtifactId,
content_address: &str,
actor: &str,
at: i64,
) -> Result<stages::StageEntry, PromoteError> {
let log = self.lineage.as_ref().ok_or_else(|| {
PromoteError::Ladder(stages::LadderError::UnknownStage(to_repo.into()))
})?;
let hop = self.ladder.check_repo_hop(from_repo, to_repo)?;
Ok(log.record_promotion(&self.ladder_name, id, &hop, content_address, actor, at)?)
}
/// Attach the resolved [`GateSet`] so promotions are gated (design §4).
///
/// Replaces `with_gate(PolicyGate)`. The set is built at boot from the RON
/// `promotion` block against a [`GateCatalog`]; there is no environment
/// variable and no `Option`, because "no gate configured" and "gate
/// configured and it allowed" must not be the same value.
pub fn with_gates(mut self, gates: GateSet) -> Self {
self.gates = gates;
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(Arc::new(JobLedger {
root: None,
store: std::sync::RwLock::new(Some(store)),
}));
self
}
/// [`with_job_ledger`](Self::with_job_ledger), opened at `root` — and so
/// RELEASABLE: `serve` attaches it this way, and a data-set growth
/// (`crate::grow`) closes `<root>/jobs.redb` before the unmount and opens it
/// again after ([`JobLedger`]).
pub fn attach_job_ledger_at(&mut self, root: &std::path::Path) -> anyhow::Result<()> {
let store = jera::jobs::JobStore::open(root)?;
self.jobs = Some(Arc::new(JobLedger {
root: Some(root.to_path_buf()),
store: std::sync::RwLock::new(Some(Arc::new(store))),
}));
Ok(())
}
/// The attached ledger, for the growth's release registry.
pub fn job_ledger(&self) -> Option<Arc<JobLedger>> {
self.jobs.clone()
}
/// Whether any gate is configured.
pub fn has_gate(&self) -> bool {
!self.gates.is_empty()
}
/// The configured gate names, in order.
///
/// A surface should expose this. gunnar's reason, which holds here
/// unchanged: a server whose gate configuration failed to load looks
/// exactly like one that has no gates, unless somebody can ask.
pub fn gate_names(&self) -> Vec<&str> {
self.gates.names()
}
/// Whether a durable job ledger is attached (seals are recorded as jobs).
pub fn has_job_ledger(&self) -> bool {
self.jobs.is_some()
}
/// Run every configured gate over the WHOLE promoted set, before any seal.
///
/// Returns the [`nornir_gate::Decision`] in both outcomes, because the
/// caller must audit what was decided whether or not it refused. At
/// [`Enforcement::Block`] a refusing decision is a
/// [`PromoteError::PolicyBlocked`]; at [`Enforcement::Warn`] the decision is
/// returned and the promotion continues.
///
/// An `Err` from any gate is inside the decision as `undecided: true` and
/// refuses at `Block`. A gate that cannot decide is not a gate that allowed.
pub fn gate_check_set(
&self,
ctx: &PromotionContext<'_>,
) -> Result<nornir_gate::Decision, PromoteError> {
let decision = self.gates.evaluate(ctx);
if decision.is_allowed() || self.gates.enforcement() == Enforcement::Warn {
return Ok(decision);
}
Err(PromoteError::PolicyBlocked(
decision.rejections.iter().map(|r| r.to_string()).collect(),
))
}
/// The gates, **scoped to the hop**.
///
/// With a ladder configured the gates run only when the TARGET rung declares
/// gates — rehearsal is usually ungated, release never is. With no ladder
/// this is exactly [`gate_check_set`](Self::gate_check_set).
///
/// The hop is resolved first and put into the context, so a gate can see
/// which rung it is guarding.
pub fn gate_check_hop(
&self,
from_repo: &str,
to_repo: &str,
set: &[PromotedArtifact<'_>],
principal: &str,
) -> Result<nornir_gate::Decision, PromoteError> {
let hop = if self.ladder.is_empty() {
None
} else {
let hop = self.ladder.check_repo_hop(from_repo, to_repo)?;
if !hop.is_gated() {
return Ok(nornir_gate::Decision::default());
}
Some(hop)
};
let ctx = PromotionContext {
principal,
from_repo,
to_repo,
hop: hop.as_ref(),
set,
};
self.gate_check_set(&ctx)
}
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 store = self.jobs.as_ref().and_then(|l| l.get());
let job = store.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> {
// RESOLVE THE TARGET FIRST. Before the ladder, before the blob, before the
// ref — a promotion to a repository that does not exist must write NOTHING
// and must say so by name. `sanitize_repo` below guarantees CONTAINMENT (the
// ref cannot escape `refs/`), which is a different question from EXISTENCE,
// and for a long time only the first one was asked: `to_repo` was collapsed
// to a path segment and never resolved, so `promote … --to made-up-repo`
// returned success and a content address for bytes no repository served.
//
// No registry attached (`None`) ⇒ unchanged: the store-level engine judges
// nothing it was not given the facts to judge.
if let Some(served) = &self.served {
if !served.contains(to_repo) {
return Err(PromoteError::UnknownRepository {
to_repo: to_repo.to_string(),
served: served.iter().cloned().collect(),
});
}
}
// STAGE LADDER next: an illegal hop must write NOTHING. Checked before the
// blob, before the ref, before anything touches disk. No ladder configured ⇒
// no judgement, exactly as before.
let hop = if self.ladder.is_empty() {
None
} else {
Some(self.ladder.check_repo_hop(from_repo, to_repo)?)
};
let content_address = crate::digest::blake3_hex(data);
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)?;
}
//
// The staging name is unique per write (HOL-RACE-1). Two coordinates with
// the same bytes (a Go module's `.mod` across its versions, the same `.info`
// across modules) share one blob; with one fixed `<hash>.tmp` two concurrent
// seals wrote the same staging file and the second rename found it already
// moved ("No such file or directory"). Each writer now renames its own file;
// the rename replaces identical bytes atomically, so the last one wins
// harmlessly.
//
// A blob that exists is never replaced (HOL-PERF-3): the staged file is LINKED to
// its address, which fails when another seal got there first, and the bytes at an
// address are the same bytes by construction. A replace would unlink the inode a
// download is reading and verifying. A filesystem without hard links renames.
if !blob_path.exists() {
let tmp = staging_path(&blob_path);
let staged =
fs::write(&tmp, data).and_then(|()| match fs::hard_link(&tmp, &blob_path) {
Ok(()) => Ok(()),
Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => Ok(()),
Err(_) => fs::rename(&tmp, &blob_path),
});
let _ = fs::remove_file(&tmp);
staged?;
}
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)),
}
// The record FOLLOWS the fact: the lineage entry is appended only once the
// seal actually took, so the history never claims a promotion that did not
// happen. A lineage write that fails is non-fatal — the artifact IS sealed and
// saying otherwise would be the bigger lie; it is logged loudly instead.
if let (Some(hop), Some(log)) = (hop.as_ref(), self.lineage.as_ref()) {
if let Err(e) = log.record_promotion(
&self.ladder_name,
id,
hop,
&record.content_address,
promoted_by,
promoted_at,
) {
log::error!(
"promotion of {} to stage '{}' SEALED but its lineage entry could not be \
written ({e}) — `current_stage` will under-report until it is replayed",
record.artifact,
hop.to.name,
);
}
}
Ok(record)
}
/// **Where a sealed coordinate's bytes are**: the content address its ref names and
/// the blob file at that address, unread (HOL-PERF-3). `None` when not promoted. The
/// caller verifies what it reads (`IntakeChain` does it once per blob file).
pub fn sealed_blob(
&self,
to_repo: &str,
id: &ArtifactId,
) -> Result<Option<(String, PathBuf)>, PromoteError> {
let ref_path = self.ref_path(to_repo, id);
let ref_json = match fs::read_to_string(&ref_path) {
Ok(j) => j,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(e) => return Err(PromoteError::Io(e)),
};
let want = PromotionRecord::content_address_from_json(&ref_json).ok_or_else(|| {
PromoteError::Io(std::io::Error::other("corrupt ref: no content_address"))
})?;
// The address names a file under `blobs/`, nothing else: a ref is ours, but a
// path out of it is still checked.
if want.is_empty() || !want.bytes().all(|b| b.is_ascii_hexdigit()) {
return Err(PromoteError::Io(std::io::Error::other(
"corrupt ref: the content address is not a digest",
)));
}
let blob = self.blob_path(&want);
Ok(Some((want, blob)))
}
/// 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 = crate::digest::blake3_hex(&data);
if got != want {
return Err(PromoteError::Io(std::io::Error::other(
"integrity: promoted blob blake3 does not match the sealed ref",
)));
}
Ok(Some(data))
}
}
/// A staging name beside `path` that no other write in this process (or another
/// process: the PID is in it) uses at the same time: `.<name>.<pid>.<seq>.tmp`. The
/// leading dot keeps it out of a listing of content addresses (HOL-RACE-1).
pub(crate) fn staging_path(path: &std::path::Path) -> PathBuf {
static SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
let seq = SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let name = path
.file_name()
.map(|n| n.to_string_lossy().into_owned())
.unwrap_or_default();
path.with_file_name(format!(".{name}.{}.{seq}.tmp", std::process::id()))
}
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(all(test, feature = "scanners"))]
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 = crate::digest::blake3_hex(data);
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"
);
}
/// A promotion to a repository this server does not serve is REFUSED BY NAME,
/// and writes nothing.
///
/// ⚠ WHY THE OLD ASSERTION WAS WRONG. The test above
/// (`crafted_to_repo_cannot_escape_prod_root`) asserted
/// `.expect("seal returns (path is contained, not that it fails)")` — i.e. it
/// asserted, as correct behaviour, that promoting to `"../../../../outside/pwn"`
/// SUCCEEDS. It is still right about the thing it was written to prove (the
/// crafted name cannot escape `refs/`), but it was standing in for a question it
/// never asked: `to_repo` was SANITIZED into a directory segment and never
/// RESOLVED. Containment is not existence. So a promotion to `made-up-repo` —
/// no crafting, no attack, just a typo — sealed the bytes under
/// `refs/made-up-repo/`, returned a `PromotionRecord` with a content address,
/// and reported `success: true` over gRPC for a promotion no repository could
/// ever serve. A verb that confirms work it did not do is worse than one that
/// refuses.
///
/// The engine now RESOLVES the target against the served set
/// (`with_served_repositories`), and an unserved target is
/// `PromoteError::UnknownRepository` — checked before the ladder and before any
/// disk write. `Holger::promotion_engine` attaches the set ALWAYS, from the same
/// repository list the stage ladder is validated against, with or without a
/// `promotion:` block (until 2026-10-22 it did not, and only tests attached one;
/// the production-path guard is
/// `the_production_engine_refuses_a_promotion_to_an_unserved_repository_by_name`
/// in `lib.rs`).
///
/// RED-when-broken: drop the `served` check in `seal_inner` and every case here
/// goes back to `Ok`.
#[test]
fn a_promotion_to_an_unserved_repository_is_refused_by_name() {
let root = tempfile::tempdir().unwrap();
let prod = root.path().join("prod");
let engine =
PromotionEngine::new(&prod).with_served_repositories(["rust-dev", "rust-prod"]);
// The plain typo is the case that used to LIE, not just the crafted ones.
for (i, absent) in [
"made-up-repo",
"rust-prod ",
"RUST-PROD",
"../../../../outside/pwn",
"..",
]
.into_iter()
.enumerate()
{
let coord = id("serde", &format!("1.0.{i}"));
let err = engine
.seal("rust-dev", absent, &coord, b"bytes", "someone", 1)
.expect_err("a promotion to a repository that is not served must be REFUSED");
match &err {
PromoteError::UnknownRepository { to_repo, served } => {
// Refused BY NAME: the message names what was asked for…
assert_eq!(to_repo, absent);
// …and what would have been legal.
assert_eq!(served, &["rust-dev".to_string(), "rust-prod".to_string()]);
}
other => panic!("expected UnknownRepository for {absent:?}, got {other:?}"),
}
// Refused BEFORE the store is touched: no ref, no blob, no prod_root.
assert!(
!engine.ref_path(absent, &coord).exists(),
"a refused promotion to {absent:?} still wrote a ref"
);
assert!(!prod.exists(), "a refused promotion created the prod store");
}
// A SERVED target still seals, so the check refuses the absent and only the
// absent.
let coord = id("serde", "2.0.0");
engine
.seal("rust-dev", "rust-prod", &coord, b"bytes", "someone", 1)
.expect("a served target still promotes");
assert!(engine.is_promoted("rust-prod", &coord));
}
/// With NO registry attached the engine judges nothing — the store-level default
/// is byte-for-byte what it was, which is what every other test here (and
/// `tests/store_persistence.rs`) relies on.
#[test]
fn without_a_served_set_the_target_is_not_resolved() {
let dir = tempfile::tempdir().unwrap();
let engine = PromotionEngine::new(dir.path());
assert!(engine.served_repositories().is_none());
engine
.seal(
"dev",
"a-repo-nobody-declared",
&id("serde", "1.0.0"),
b"x",
"t",
1,
)
.expect("no registry ⇒ no judgement, unchanged");
}
// ── 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 = crate::digest::blake3_hex(data);
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::attest::ArtifactDigest;
use crate::security::{
Advisory, AdvisoryDb, Decision as ScanDecision, LicensePolicy, OsvScanner, Policy,
ProvenancePolicy, Severity,
};
use gate::{GateCatalog, OsvGate, PromotedArtifact};
/// Build a one-artifact promoted set for the gate tests. The DIGEST and the
/// BYTES are required now, which is the point of the change.
fn set_of<'a>(id: &'a ArtifactId, bytes: &'a [u8]) -> [PromotedArtifact<'a>; 1] {
[PromotedArtifact {
id,
ecosystem: "cargo",
digest: ArtifactDigest::of_sha256(bytes),
bytes,
attestations: &[],
}]
}
fn gates_of(scanner: OsvScanner, e: Enforcement) -> GateSet {
GateCatalog::new()
.with(std::sync::Arc::new(OsvGate::new(scanner)))
.resolve(&[OsvGate::NAME.to_string()])
.expect("the built-in osv gate resolves")
.with_enforcement(e)
}
/// 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: ScanDecision::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_gates(gates_of(real_blocking_scanner(), Enforcement::Block));
assert!(engine.has_gate());
assert_eq!(engine.gate_names(), vec!["osv"]);
let bad_id = id("badcrate", "1.0.0");
let bad = engine.gate_check_hop("dev", "prod", &set_of(&bad_id, b"bytes"), "tester");
assert!(
matches!(bad, Err(PromoteError::PolicyBlocked(_))),
"a High-severity advisory must block the promotion, got {bad:?}"
);
let ok_id = id("serde", "1.0.0");
let clean = engine.gate_check_hop("dev", "prod", &set_of(&ok_id, b"bytes"), "tester");
assert!(clean.is_ok(), "a clean coordinate must pass the gate");
assert!(clean.unwrap().summary().contains("osv"));
}
/// 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 =
crate::security::advisory_db_from_osv_file(osv.to_str().unwrap()).expect("import OSV");
// This test is about advisories; its fixture bytes declare no licence,
// so the licence stance is stated (an unknown licence is held by default).
let scanner = OsvScanner::new(
crate::security::promotion_policy_for(
Vec::new(),
crate::security::UnknownLicense::Pass,
crate::security::ProvenancePolicy::Ignore,
),
db,
);
let engine =
PromotionEngine::new(dir.path()).with_gates(gates_of(scanner, Enforcement::Block));
let bad_id = id("badcrate", "1.0.0");
assert!(
matches!(
engine.gate_check_hop("dev", "prod", &set_of(&bad_id, b"bytes"), "tester"),
Err(PromoteError::PolicyBlocked(_))
),
"the imported advisory must block the affected coordinate"
);
let ok_id = id("goodcrate", "9.9.9");
assert!(
engine
.gate_check_hop("dev", "prod", &set_of(&ok_id, b"bytes"), "tester")
.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_gates(gates_of(real_blocking_scanner(), Enforcement::Warn));
let bad_id = id("badcrate", "1.0.0");
let d = engine
.gate_check_hop("dev", "prod", &set_of(&bad_id, b"bytes"), "tester")
.expect("Warn enforcement must permit the promotion");
assert!(
!d.is_allowed(),
"Warn permits, but the refusal is still IN the decision so the audit records it"
);
}
/// HOL-GUN-007 at the engine: a gate name this build does not carry is an
/// error naming what it could have been, never a silent skip.
#[test]
fn an_unknown_gate_name_is_refused_with_the_available_list() {
let cat =
GateCatalog::new().with(std::sync::Arc::new(OsvGate::new(real_blocking_scanner())));
let err = cat.resolve(&["attested".to_string()]).unwrap_err();
assert_eq!(err.named, "attested");
assert_eq!(err.available, vec!["osv".to_string()]);
}
/// The digest reaches the scan. Two different byte strings under the same
/// coordinate are two subjects — the thing the coordinate-only seam could
/// not express, which collapsed modgunn's verdict cache onto one key.
#[test]
fn the_gate_sees_the_bytes_not_only_the_coordinate() {
let a = id("serde", "1.0.0");
let one = set_of(&a, b"one");
let two = set_of(&a, b"two");
assert_ne!(one[0].digest.hex(), two[0].digest.hex());
assert!(!one[0].digest.hex().is_empty());
}
/// The DEFAULT, unconfigured path is UNCHANGED: no gate ⇒ the hop proceeds,
/// and a full seal still succeeds without any scan. What IS new: the
/// decision says "none configured" rather than looking like a pass.
#[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.
let bad_id = id("badcrate", "1.0.0");
let d = engine
.gate_check_hop("dev", "prod", &set_of(&bad_id, b"bytes"), "tester")
.expect("no gate configured must not refuse");
assert!(d.is_allowed());
assert!(
d.summary().contains("none configured"),
"an allow with no gates must be distinguishable from an allow by gates: {}",
d.summary()
);
// 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);
}
// ── The stage ladder, driving REAL seals (not the ladder in isolation). ───────
fn rungs() -> StageLadder {
StageLadder::new(vec![
Stage::new("sparring", "rust-sparring"),
Stage::new("staging", "rust-staging"),
Stage::new("release", "rust-release").gated(),
])
}
/// An artifact walks the ladder rung by rung through the EXISTING pointer-move
/// seal, the lineage records each hop, and `current_stage` answers where it is.
/// Nothing about the seal changed — the bytes are still content-addressed once
/// and still verify at every rung.
#[test]
fn an_artifact_walks_the_ladder_and_the_lineage_says_where_it_is() {
let tmp = tempfile::tempdir().unwrap();
let engine = PromotionEngine::new(tmp.path()).with_ladder("rust", rungs());
let id = ArtifactId {
namespace: None,
name: "serde".into(),
version: "1.0.200".into(),
};
let bytes = b"the-sealed-bytes".to_vec();
assert_eq!(engine.current_stage(&id), None, "not on the ladder yet");
let r1 = engine
.seal(
"rust-sparring",
"rust-staging",
&id,
&bytes,
"release-bot",
1000,
)
.expect("one rung up is legal");
assert_eq!(engine.current_stage(&id).as_deref(), Some("staging"));
let r2 = engine
.seal(
"rust-staging",
"rust-release",
&id,
&bytes,
"release-bot",
2000,
)
.expect("the next rung is legal");
assert_eq!(engine.current_stage(&id).as_deref(), Some("release"));
// The POINTER moved, not the bytes: identical content de-dups to one blob and
// both rungs name the same content address.
assert_eq!(
r1.content_address, r2.content_address,
"promotion is a pointer move"
);
assert_eq!(
engine.fetch("rust-release", &id).unwrap(),
Some(bytes.clone()),
"and it still verifies"
);
let hist = engine.stage_history(&id);
assert_eq!(hist.len(), 2);
assert_eq!(
(hist[0].from_stage.as_str(), hist[0].stage.as_str()),
("sparring", "staging")
);
assert_eq!(
(hist[1].from_stage.as_str(), hist[1].stage.as_str()),
("staging", "release")
);
}
/// **An illegal hop writes NOTHING.** The ladder is judged before the blob, before
/// the ref, before anything touches disk — so a refused promotion leaves the store
/// exactly as it was. RED-when-broken if the check ever slides below the write.
#[test]
fn an_illegal_hop_is_refused_before_the_store_is_touched() {
let tmp = tempfile::tempdir().unwrap();
let engine = PromotionEngine::new(tmp.path()).with_ladder("rust", rungs());
let id = ArtifactId {
namespace: None,
name: "serde".into(),
version: "1.0.200".into(),
};
let err = engine
.seal("rust-sparring", "rust-release", &id, b"bytes", "bot", 1000)
.expect_err("skipping staging must be refused");
assert!(
matches!(err, PromoteError::Ladder(LadderError::SkipsStages { .. })),
"{err}"
);
assert!(err.to_string().contains("skips 1 stage"), "{err}");
assert!(
!engine.is_promoted("rust-release", &id),
"no ref was written"
);
assert!(
!tmp.path().join("blobs").exists(),
"no blob was written either"
);
assert!(
engine.stage_history(&id).is_empty(),
"and nothing was recorded"
);
// Backwards is refused too, and a repo that is not a stage at all.
assert!(matches!(
engine
.seal("rust-release", "rust-staging", &id, b"b", "bot", 1)
.unwrap_err(),
PromoteError::Ladder(LadderError::NotForward { .. })
));
assert!(matches!(
engine
.seal("rust-sparring", "some-other-repo", &id, b"b", "bot", 1)
.unwrap_err(),
PromoteError::Ladder(LadderError::UnknownStage(_))
));
}
/// **No ladder configured ⇒ nothing changed.** The default engine promotes
/// between any two repositories exactly as it always did, and writes no lineage.
/// This is the additive guarantee.
#[test]
fn without_a_ladder_promotion_is_exactly_what_it_was() {
let tmp = tempfile::tempdir().unwrap();
let engine = PromotionEngine::new(tmp.path());
let id = ArtifactId {
namespace: None,
name: "serde".into(),
version: "1.0.200".into(),
};
// Two arbitrary repos, no ordering, no stage model.
engine
.seal("anything", "anywhere", &id, b"bytes", "bot", 1000)
.unwrap();
assert!(engine.is_promoted("anywhere", &id));
assert_eq!(
engine.current_stage(&id),
None,
"no ladder ⇒ no stage answer"
);
assert!(engine.stage_history(&id).is_empty(), "and no lineage file");
assert!(engine.ladder().is_empty());
}
/// Withdrawing a release does not rewrite the sealed generation: the bytes still
/// fetch and still verify, the ref still exists, and re-promoting the coordinate
/// is still refused as immutable. Only the ANSWER moves.
#[test]
fn withdrawal_moves_the_answer_not_the_sealed_bytes() {
let tmp = tempfile::tempdir().unwrap();
let engine = PromotionEngine::new(tmp.path()).with_ladder("rust", rungs());
let id = ArtifactId {
namespace: None,
name: "serde".into(),
version: "1.0.200".into(),
};
let bytes = b"the-sealed-bytes".to_vec();
engine
.seal("rust-sparring", "rust-staging", &id, &bytes, "bot", 1000)
.unwrap();
engine
.seal("rust-staging", "rust-release", &id, &bytes, "bot", 2000)
.unwrap();
let entry = engine
.withdraw(&id, "release", "oncall", 3000, "CVE-2026-0001")
.expect("a release can be withdrawn");
assert_eq!(entry.op, StageOp::Withdrawn);
assert_eq!(
engine.current_stage(&id).as_deref(),
Some("staging"),
"it fell one rung"
);
// …but the sealed generation is untouched in every observable way.
assert!(
engine.is_promoted("rust-release", &id),
"the ref still stands"
);
assert_eq!(
engine.fetch("rust-release", &id).unwrap(),
Some(bytes.clone()),
"the bytes still verify"
);
assert!(
matches!(
engine
.seal("rust-staging", "rust-release", &id, &bytes, "bot", 4000)
.unwrap_err(),
PromoteError::AlreadyPromoted
),
"immutability survives a withdrawal — the coordinate is still sealed"
);
assert_eq!(
engine.stage_history(&id).len(),
3,
"the history keeps all three records"
);
// Withdrawing from a rung that is not on the ladder is refused.
assert!(engine
.withdraw(&id, "prod", "oncall", 5000, "typo")
.is_err());
}
/// Per-rung gating: the gate runs on the hop into a `gated` rung and is skipped on
/// the ungated ones — one scanner, enforcement where it matters.
#[test]
fn the_policy_gate_runs_per_rung_not_per_engine() {
let tmp = tempfile::tempdir().unwrap();
// A gate that BLOCKS the coordinate under test, so "did it run" is observable
// from the outside: a hop that consults it is refused, one that skips it is not.
let engine = PromotionEngine::new(tmp.path())
.with_ladder("rust", rungs())
.with_gates(gates_of(real_blocking_scanner(), Enforcement::Block));
let bad = id("badcrate", "1.0.0");
let bad_set = set_of(&bad, b"bytes");
let good = id("goodcrate", "1.0.0");
let good_set = set_of(&good, b"bytes");
// Ungated rung: the gate is NOT consulted, so the blocking verdict never fires.
assert!(
engine
.gate_check_hop("rust-sparring", "rust-staging", &bad_set, "tester")
.is_ok(),
"an ungated rung must not run the gate"
);
// Gated rung: the gate runs, and its Block verdict refuses the hop.
assert!(
matches!(
engine
.gate_check_hop("rust-staging", "rust-release", &bad_set, "tester")
.unwrap_err(),
PromoteError::PolicyBlocked(_)
),
"a gated rung must run the gate"
);
// A clean coordinate passes the same gated rung.
assert!(engine
.gate_check_hop("rust-staging", "rust-release", &good_set, "tester")
.is_ok());
// An illegal hop is refused by the ladder before the gate is even reached.
assert!(matches!(
engine
.gate_check_hop("rust-sparring", "rust-release", &bad_set, "tester")
.unwrap_err(),
PromoteError::Ladder(LadderError::SkipsStages { .. })
));
}
// ── The seal as a STORE: races, torn state, and what a read may never say ──
/// **Only one of many racing seals may win.** `create_new` on the ref is the
/// immutability commit, so N threads promoting the same coordinate with
/// DIFFERENT bytes must produce exactly one `Ok` and N-1 `AlreadyPromoted` —
/// and the bytes that read back must be the winner's, verifying against the
/// winner's content address. RED-when-broken if the ref write ever becomes a
/// check-then-write (two winners, one silently overwritten).
#[test]
fn only_one_of_many_racing_seals_wins_and_its_bytes_are_what_reads_back() {
const RACERS: usize = 8;
let dir = tempfile::tempdir().unwrap();
let engine = PromotionEngine::new(dir.path());
let coord = id("serde", "1.0.0");
let winners = std::sync::Mutex::new(Vec::<PromotionRecord>::new());
let refused = std::sync::atomic::AtomicUsize::new(0);
std::thread::scope(|scope| {
for n in 0..RACERS {
let engine = &engine;
let coord = &coord;
let winners = &winners;
let refused = &refused;
// A32-14, stays a task: scoped threads, joined when the scope ends.
scope.spawn(move || {
let bytes = format!("payload-from-racer-{n}").into_bytes();
match engine.seal("dev", "prod", coord, &bytes, "bot", 1) {
Ok(rec) => winners.lock().unwrap().push(rec),
Err(PromoteError::AlreadyPromoted) => {
refused.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
}
Err(e) => panic!("a racing seal must win or be refused, got {e}"),
}
});
}
});
let winners = winners.into_inner().unwrap();
assert_eq!(winners.len(), 1, "exactly one seal may take the coordinate");
assert_eq!(
refused.load(std::sync::atomic::Ordering::Relaxed),
RACERS - 1,
"every loser is refused as already promoted"
);
let sealed = engine
.fetch("prod", &coord)
.unwrap()
.expect("the coordinate is sealed");
assert_eq!(
blake3::hash(&sealed).to_hex().to_string(),
winners[0].content_address,
"the bytes that read back are the WINNER's, and they verify"
);
}
/// A promoted coordinate whose blob has gone missing is an ERROR, never
/// `Ok(None)`: "the ref says this is sealed" and "there are no bytes" is
/// corruption, and a caller must not mistake it for "was never promoted".
#[test]
fn a_sealed_coordinate_whose_blob_vanished_is_an_error_not_absent() {
let dir = tempfile::tempdir().unwrap();
let engine = PromotionEngine::new(dir.path());
let coord = id("tokio", "1.38.0");
let rec = engine
.seal("dev", "prod", &coord, b"the-bytes", "admin", 1)
.unwrap();
fs::remove_file(dir.path().join("blobs").join(&rec.content_address)).unwrap();
let err = engine
.fetch("prod", &coord)
.expect_err("a missing blob behind a live ref must be an error");
assert!(matches!(err, PromoteError::Io(_)), "{err}");
assert!(
engine.is_promoted("prod", &coord),
"the seal itself still stands"
);
}
/// A ref that does not carry a readable `content_address` (a torn or hand-
/// mangled seal) is reported as an error, never silently read as "no such
/// promotion" — the same rule the missing-blob case obeys.
#[test]
fn a_corrupt_or_empty_ref_is_an_error_not_absent() {
let dir = tempfile::tempdir().unwrap();
let engine = PromotionEngine::new(dir.path());
let coord = id("anyhow", "1.0.0");
engine
.seal("dev", "prod", &coord, b"bytes", "admin", 1)
.unwrap();
let ref_path = engine.ref_path("prod", &coord);
for corrupt in [&b""[..], &b"{"[..], &b"{\"artifact\":\"anyhow@1.0.0\"}"[..]] {
fs::write(&ref_path, corrupt).unwrap();
assert!(
engine.fetch("prod", &coord).is_err(),
"a ref without a readable content address must error, not read as absent"
);
}
}
/// **The ref JSON cannot be spoofed from an untrusted field.** `to_json` is
/// hand-rolled, so an identity or repo name carrying `","content_address":"…`
/// must be ESCAPED, not obeyed: the integrity check on read has to find the
/// real digest. RED-when-broken the moment the escaping is dropped — `fetch`
/// would then look for an attacker-named blob.
#[test]
fn a_crafted_identity_cannot_spoof_the_sealed_content_address() {
let dir = tempfile::tempdir().unwrap();
let engine = PromotionEngine::new(dir.path());
let coord = ArtifactId {
namespace: Some(r#"ns","content_address":"deadbeef"#.into()),
name: r#"pkg\","content_address":"deadbeef"#.into(),
version: "1.0.0".into(),
};
let data = b"honest-bytes";
let rec = engine
.seal(
r#"from","content_address":"deadbeef"#,
"prod",
&coord,
data,
r#"mallory","content_address":"deadbeef"#,
1,
)
.expect("a hostile field is data, not syntax");
assert_eq!(rec.content_address, blake3::hash(data).to_hex().to_string());
assert_eq!(
engine.fetch("prod", &coord).unwrap().as_deref(),
Some(data.as_slice()),
"the read path must resolve the REAL content address, not the injected one"
);
assert!(
!dir.path().join("blobs").join("deadbeef").exists(),
"nothing was written under the injected address"
);
}
/// A completed seal leaves no staging file behind — the blob write is
/// temp-then-rename, and a reader listing `blobs/` must see content addresses
/// only. RED if the rename is ever dropped for a plain write, or the temp file
/// is left in place.
#[test]
fn a_seal_leaves_no_staging_file_behind() {
let dir = tempfile::tempdir().unwrap();
let engine = PromotionEngine::new(dir.path());
let rec = engine
.seal("dev", "prod", &id("serde", "1.0.0"), b"bytes", "admin", 1)
.unwrap();
let names: Vec<String> = fs::read_dir(dir.path().join("blobs"))
.unwrap()
.flatten()
.map(|e| e.file_name().to_string_lossy().to_string())
.collect();
assert_eq!(
names,
vec![rec.content_address],
"the blob dir holds exactly the content address"
);
}
/// The boundaries a content-addressed store has to get right: an EMPTY
/// artifact is a real artifact (sealed, addressed by the blake3 of nothing,
/// and read back as empty bytes — never as "absent"), and a unicode coordinate
/// round-trips through the sanitized on-disk key.
#[test]
fn an_empty_payload_and_a_unicode_coordinate_both_round_trip() {
let dir = tempfile::tempdir().unwrap();
let engine = PromotionEngine::new(dir.path());
let empty = id("empty-crate", "0.0.0");
let rec = engine.seal("dev", "prod", &empty, b"", "admin", 1).unwrap();
assert_eq!(rec.size, 0);
assert_eq!(rec.content_address, blake3::hash(b"").to_hex().to_string());
assert_eq!(
engine.fetch("prod", &empty).unwrap(),
Some(Vec::new()),
"an empty artifact reads back as empty bytes, NOT as not-promoted"
);
let uni = ArtifactId {
namespace: Some("se.vetra.förråd".into()),
name: "räksmörgås".into(),
version: "1.0.0-β".into(),
};
engine
.seal("dev", "prod", &uni, "smörgås".as_bytes(), "admin", 1)
.unwrap();
assert_eq!(
engine.fetch("prod", &uni).unwrap().as_deref(),
Some("smörgås".as_bytes()),
"a unicode coordinate round-trips through the sanitized key"
);
assert!(
engine
.seal("dev", "prod", &uni, b"other", "admin", 2)
.is_err(),
"and it is sealed exactly once, like any other coordinate"
);
}
/// Both path keys reduce ANY input to exactly one filesystem segment: no
/// separator survives, and the dot-only escapes (`.`, `..`, empty) are
/// neutralised. This is the guard that keeps a wire-supplied repo name or
/// coordinate inside `refs/`.
#[test]
fn repo_and_coordinate_keys_are_always_one_safe_segment() {
let hostile = [
"../../../../etc/passwd",
"/absolute/path",
"..",
".",
"",
"a/b/c",
"back\\slash",
"nul\0byte",
"spaces and \"quotes\"",
"räksmörgås",
"%2e%2e%2f",
];
for raw in hostile {
let repo = sanitize_repo(raw);
assert!(!repo.is_empty(), "a repo segment is never empty ({raw:?})");
assert!(
!repo.contains('/') && !repo.contains('\\') && !repo.contains('\0'),
"sanitize_repo({raw:?}) leaked a separator: {repo:?}"
);
assert!(
repo != "." && repo != "..",
"sanitize_repo({raw:?}) stayed a dot escape"
);
assert_eq!(
std::path::Path::new(&repo).components().count(),
1,
"sanitize_repo({raw:?}) is not a single component: {repo:?}"
);
let coord = coord_key(&ArtifactId {
namespace: Some(raw.into()),
name: raw.into(),
version: raw.into(),
});
assert!(
!coord.contains('/') && !coord.contains('\\') && !coord.contains('\0'),
"coord_key({raw:?}) leaked a separator: {coord:?}"
);
assert_eq!(
std::path::Path::new(&coord).components().count(),
1,
"coord_key({raw:?}) is not a single component: {coord:?}"
);
}
// A long name stays one segment (it is not chopped into directories).
let long = "x".repeat(600);
assert_eq!(sanitize_repo(&long).len(), 600);
assert_eq!(
std::path::Path::new(&sanitize_repo(&long))
.components()
.count(),
1
);
}
/// HOL-RACE-1: two coordinates with the SAME bytes (a Go module's `.mod` is
/// identical across its versions) share one content-addressed blob. Sealed at
/// the same time from many threads, every seal must take. RED while the blob's
/// staging file had one fixed name (`<hash>.tmp`): the loser's rename found the
/// file already moved and failed with "No such file or directory".
#[test]
fn concurrent_seals_of_identical_bytes_under_different_coordinates_all_take() {
const THREADS: usize = 16;
const ROUNDS: usize = 64;
let dir = tempfile::tempdir().unwrap();
let engine = std::sync::Arc::new(PromotionEngine::new(dir.path()));
let barrier = std::sync::Arc::new(std::sync::Barrier::new(THREADS));
let handles: Vec<_> = (0..THREADS)
.map(|t| {
let (engine, barrier) = (engine.clone(), barrier.clone());
std::thread::spawn(move || {
let mut failed = Vec::new();
for round in 0..ROUNDS {
let data = format!("module example.com/m/pkg{round:05}\n");
barrier.wait();
if let Err(e) = engine.seal(
"dev",
"prod",
&id(&format!("pkg{round}"), &format!("v{t}")),
data.as_bytes(),
"admin",
1,
) {
failed.push(format!("round {round} thread {t}: {e}"));
}
}
failed
})
})
.collect();
let failed: Vec<String> = handles
.into_iter()
.flat_map(|h| h.join().unwrap())
.collect();
assert!(
failed.is_empty(),
"{} of {} seals failed: {:?}",
failed.len(),
THREADS * ROUNDS,
&failed[..failed.len().min(5)]
);
let names: Vec<String> = fs::read_dir(dir.path().join("blobs"))
.unwrap()
.flatten()
.map(|e| e.file_name().to_string_lossy().to_string())
.collect();
assert_eq!(
names.len(),
ROUNDS,
"one blob per distinct content, no staging file left: {names:?}"
);
for round in 0..ROUNDS {
for t in 0..THREADS {
let got = engine
.fetch("prod", &id(&format!("pkg{round}"), &format!("v{t}")))
.unwrap()
.unwrap();
assert_eq!(
got,
format!("module example.com/m/pkg{round:05}\n").as_bytes()
);
}
}
}
}