nordisk/holger
server/lib/src/promote/intake.rs
//! **The intake chain — every artifact has exactly one state.**
//!
//! Holger is a secure artifact gateway. The thing it sells is the walk an artifact
//! takes from the wire to a download, and that walk is this module:
//!
//! ```text
//! UNTRUSTED ─▶ INGESTED ─▶ SCANNED ─▶ POLICY_CHECKED ─▶ APPROVED ─▶ SEALED ─▶ PROMOTED ─▶ AVAILABLE
//! (the wire) (incoming) (quarantine) (quarantine) (approved) (production store ─────────────▶)
//!
//! any state before SEALED ─▶ QUARANTINED (held for a human; released back to the state it left)
//! any state ─▶ BLOCKED (terminal; never served)
//! ```
//!
//! ## What each state means, and what moves it
//!
//! * **UNTRUSTED** — bytes on the wire. Nothing is recorded for them.
//! * **INGESTED** — a door (HTTP PUT, a native upload path that names a
//! coordinate, or gRPC `PutArtifact`) wrote the bytes into the `incoming`
//! repository after the INGEST gates allowed them. The sha256 digest of those
//! bytes is recorded here and every later transition re-checks it.
//! * **SCANNED** — the hop `incoming → quarantine` on the promotion ladder. The
//! quarantine rung's gates run, and the transition carries the scanner evidence
//! slots (L-P2 scanner framework, L-P3 malware/file inspection fill them).
//! * **POLICY_CHECKED** — the policy for this state ran (its named gates, its
//! required evidence such as a licence verdict from L-P4). Still in quarantine.
//! * **APPROVED** — the hop `quarantine → approved`. The approved rung's gates run
//! and the transition carries the approval evidence slot (L-P5 separation of
//! duties fills it; this module only requires the slot when the policy asks).
//! * **SEALED** — the hop `approved → production` through
//! [`PromotionEngine::seal`]: blake3 content address, blob, create-once ref.
//! The production rung's gates run first, exactly as for any promotion.
//! * **PROMOTED** — the ladder's own lineage agrees that the artifact stands at
//! the production rung. If the seal's lineage write had failed, this refuses.
//! * **AVAILABLE** — the sealed bytes are read back from the sealed store and
//! their sha256 equals the ingested digest. Only now is the artifact served,
//! and it is served from the sealed store, never from a repository backend.
//!
//! ## The rules every transition obeys
//!
//! 1. **One state at a time.** [`check_transition`] allows exactly the next state
//! on the chain, a hold (QUARANTINED, BLOCKED), or the release of a hold back
//! to the state it left. A skip is refused BY NAME, listing the states it would
//! have skipped ([`IntakeError::SkipsState`], `holger-intake-1`).
//! 2. **A policy check on every transition.** The structural check, the digest
//! re-check, the transition's [`TransitionPolicy`] (named gates and required
//! evidence), and on a repository hop the ladder rung's own gates.
//! 3. **Audit first, fail closed.** The audit record is written BEFORE the effect
//! and before the state changes. If the audit write fails, nothing happens
//! ([`IntakeError::AuditFailed`], `holger-intake-8`).
//! 4. **Append-only.** The state is the reduction of an append-only JSON-lines
//! record under the promotion store's `lineage/` root, beside the ladder's own
//! stage lineage, written with the same discipline ([`super::stages::append_jsonl`]).
//!
//! ## What this module reuses, and does not replace
//!
//! The four repositories ARE the promotion ladder: `promotion.stages` must name
//! them in order (`incoming, quarantine, approved, production`), so a rung's
//! `gates: [...]` is the policy for the repository hop onto it, the lineage is
//! the ladder's, and the seal is the one seal holger has. The ingest gates are
//! the `promotion.ingest_gates` the gRPC door already ran. Quarantine is a STATE
//! here, not a hand-set property behind an environment variable.
//!
//! ## What it leaves for later lanes
//!
//! The [`Evidence`] slots are hooks. Nothing here scans, inspects or detects a
//! licence: a transition carries what those lanes will produce, a
//! [`TransitionPolicy`] can require a slot, and a slot whose verdict is `fail`
//! refuses the transition.
//!
//! The `approval` slot is filled HERE, by separation of duties
//! ([`super::duties`]): a move to APPROVED is an approval vote, the importer
//! never casts one, `approvals: N` needs N distinct approvers, and each vote is
//! an attested statement whose digest is the slot's reference.
use crate::lock::LockOrRecover as _;
use std::collections::{BTreeMap, BTreeSet};
use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use serde::{Deserialize, Serialize};
use traits::{ArtifactId, RepositoryBackendTrait};
use super::{
coord_key, GateSet, PromoteError, PromotedArtifact, PromotionContext, PromotionEngine,
};
use crate::audit::{AuditAction, AuditEvent, AuditLog, TrailKind};
// ── the states ──────────────────────────────────────────────────────────────
/// The one explicit state an artifact is in.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
pub enum ArtifactState {
/// Bytes on the wire; nothing is recorded for them.
Untrusted,
/// Written into the `incoming` repository after the ingest gates allowed it.
Ingested,
/// Moved onto the quarantine rung; the scanner evidence slots are filled.
Scanned,
/// The policy for the artifact ran and allowed it. Still in quarantine.
PolicyChecked,
/// Moved onto the approved rung; the approval evidence slot is filled.
Approved,
/// Sealed into the immutable, content-addressed production store.
Sealed,
/// The ladder's lineage agrees the artifact stands at the production rung.
Promoted,
/// Served from the sealed store. The only state that is downloadable.
Available,
/// Refused for good. Terminal; never served.
Blocked,
/// Held for a human. Released back to the state it was held at.
Quarantined,
}
impl ArtifactState {
/// The chain, in order. [`Blocked`](Self::Blocked) and
/// [`Quarantined`](Self::Quarantined) are not on it: they are holds.
pub const CHAIN: [ArtifactState; 8] = [
ArtifactState::Untrusted,
ArtifactState::Ingested,
ArtifactState::Scanned,
ArtifactState::PolicyChecked,
ArtifactState::Approved,
ArtifactState::Sealed,
ArtifactState::Promoted,
ArtifactState::Available,
];
/// Every state, chain first, then the holds.
pub const ALL: [ArtifactState; 10] = [
ArtifactState::Untrusted,
ArtifactState::Ingested,
ArtifactState::Scanned,
ArtifactState::PolicyChecked,
ArtifactState::Approved,
ArtifactState::Sealed,
ArtifactState::Promoted,
ArtifactState::Available,
ArtifactState::Blocked,
ArtifactState::Quarantined,
];
/// The name operators read and write: `POLICY_CHECKED`.
pub fn name(&self) -> &'static str {
match self {
ArtifactState::Untrusted => "UNTRUSTED",
ArtifactState::Ingested => "INGESTED",
ArtifactState::Scanned => "SCANNED",
ArtifactState::PolicyChecked => "POLICY_CHECKED",
ArtifactState::Approved => "APPROVED",
ArtifactState::Sealed => "SEALED",
ArtifactState::Promoted => "PROMOTED",
ArtifactState::Available => "AVAILABLE",
ArtifactState::Blocked => "BLOCKED",
ArtifactState::Quarantined => "QUARANTINED",
}
}
/// Parse a state name. Case, spaces, dashes and underscores are ignored, so
/// `POLICY CHECKED`, `policy-checked` and `PolicyChecked` are one state.
pub fn parse(raw: &str) -> Result<Self, IntakeError> {
let norm: String = raw
.chars()
.filter(|c| !matches!(c, ' ' | '-' | '_'))
.flat_map(|c| c.to_uppercase())
.collect();
Self::ALL
.into_iter()
.find(|s| s.name().replace('_', "") == norm)
.ok_or_else(|| IntakeError::UnknownState(raw.to_string()))
}
/// The position on the chain, or `None` for a hold.
pub fn rank(&self) -> Option<usize> {
Self::CHAIN.iter().position(|s| s == self)
}
/// The next state on the chain, or `None` at AVAILABLE and for a hold.
pub fn next(&self) -> Option<Self> {
let i = self.rank()?;
Self::CHAIN.get(i + 1).copied()
}
/// Is this a hold rather than a chain state?
pub fn is_hold(&self) -> bool {
matches!(self, ArtifactState::Blocked | ArtifactState::Quarantined)
}
/// Is the artifact sealed (its bytes are immutable in the production store)?
pub fn is_sealed(&self) -> bool {
matches!(
self,
ArtifactState::Sealed | ArtifactState::Promoted | ArtifactState::Available
)
}
/// Which intake repository an artifact in this chain state stands in.
/// `None` for UNTRUSTED (the wire) and for the holds.
pub fn repository(&self) -> Option<IntakeRepo> {
match self {
ArtifactState::Untrusted | ArtifactState::Blocked | ArtifactState::Quarantined => None,
ArtifactState::Ingested => Some(IntakeRepo::Incoming),
ArtifactState::Scanned | ArtifactState::PolicyChecked => Some(IntakeRepo::Quarantine),
ArtifactState::Approved => Some(IntakeRepo::Approved),
ArtifactState::Sealed | ArtifactState::Promoted | ArtifactState::Available => {
Some(IntakeRepo::Production)
}
}
}
}
impl std::fmt::Display for ArtifactState {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(self.name())
}
}
/// The four intake repositories.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum IntakeRepo {
Incoming,
Quarantine,
Approved,
Production,
}
// ── the evidence slots ──────────────────────────────────────────────────────
/// What kind of evidence a slot holds. The later lanes fill these: the scanner
/// framework (`scan`), malware and file inspection (`malware`), licence policy
/// (`license`), separation of duties (`approval`).
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum EvidenceKind {
/// A vulnerability / CVE scan result.
Scan,
/// Malware and file inspection.
Malware,
/// Licence detection and licence policy.
License,
/// Signature and provenance verification.
Provenance,
/// A human approval (separation of duties).
Approval,
/// Anything else a policy produces.
Other,
}
impl EvidenceKind {
/// The wire spelling.
pub fn name(&self) -> &'static str {
match self {
EvidenceKind::Scan => "scan",
EvidenceKind::Malware => "malware",
EvidenceKind::License => "license",
EvidenceKind::Provenance => "provenance",
EvidenceKind::Approval => "approval",
EvidenceKind::Other => "other",
}
}
}
/// What a piece of evidence concluded.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum EvidenceVerdict {
Pass,
Warn,
Fail,
}
/// One filled evidence slot, carried by a transition and kept in its record.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Evidence {
pub kind: EvidenceKind,
/// Who produced it: a scanner name, an approver's subject.
pub source: String,
pub verdict: EvidenceVerdict,
/// A digest of the full report, or a short reference to it.
#[serde(default)]
pub reference: String,
}
impl Evidence {
pub fn new(kind: EvidenceKind, source: impl Into<String>, verdict: EvidenceVerdict) -> Self {
Self {
kind,
source: source.into(),
verdict,
reference: String::new(),
}
}
pub fn with_reference(mut self, reference: impl Into<String>) -> Self {
self.reference = reference.into();
self
}
/// Parse the wire spelling a door receives (`scan`, `pass`, …).
pub fn from_wire(
kind: &str,
source: &str,
verdict: &str,
reference: &str,
) -> Result<Self, String> {
let k = match kind.trim().to_ascii_lowercase().as_str() {
"scan" => EvidenceKind::Scan,
"malware" => EvidenceKind::Malware,
"license" => EvidenceKind::License,
"provenance" => EvidenceKind::Provenance,
"approval" => EvidenceKind::Approval,
"other" => EvidenceKind::Other,
other => return Err(format!("'{other}' is not an evidence kind (scan, malware, license, provenance, approval, other)")),
};
let v = match verdict.trim().to_ascii_lowercase().as_str() {
"pass" => EvidenceVerdict::Pass,
"warn" => EvidenceVerdict::Warn,
"fail" => EvidenceVerdict::Fail,
other => {
return Err(format!(
"'{other}' is not an evidence verdict (pass, warn, fail)"
))
}
};
Ok(Evidence::new(k, source, v).with_reference(reference))
}
}
// ── the refusals ────────────────────────────────────────────────────────────
/// Why an intake transition did not happen. Every variant leaves the state as it
/// was.
#[derive(Debug)]
pub enum IntakeError {
/// The transition jumps past states on the chain.
SkipsState {
from: ArtifactState,
to: ArtifactState,
skipped: Vec<ArtifactState>,
},
/// The transition goes backwards, or to the state the artifact is in.
NotForward {
from: ArtifactState,
to: ArtifactState,
},
/// The artifact's state cannot be left this way.
Terminal {
state: ArtifactState,
to: ArtifactState,
},
/// Not an intake state name.
UnknownState(String),
/// The coordinate has no intake record.
NotIngested(String),
/// The policy refused: a gate, a failing evidence verdict, or the ladder.
PolicyRefused {
to: ArtifactState,
reasons: Vec<String>,
},
/// The policy requires evidence the transition did not carry.
EvidenceMissing {
to: ArtifactState,
missing: Vec<EvidenceKind>,
},
/// The audit record could not be written; nothing happened.
AuditFailed(String),
/// The bytes do not match the digest recorded at ingest.
Integrity { want: String, got: String },
/// The coordinate is sealed; it cannot be ingested again.
Immutable(String),
/// The repository is fed only by the intake chain.
DoorRefused { repo: String, why: String },
/// A write into a repository outside the chain where the policy requires a scan (P19).
SideDoor { repo: String, why: String },
/// The upload path names no artifact coordinate.
NoCoordinate(String),
/// Storage failed (the record, the incoming backend, or the sealed store).
Store(String),
/// Separation of duties refused ([`super::duties`], `holger-duty-<n>`).
Duty(super::duties::DutyError),
/// The npm door refused ([`super::npm`], `holger-npm-<n>`).
Npm(super::npm::NpmError),
/// The cargo door refused (lane HOL-CARGO-PUBLISH-DOOR).
Cargo(super::cargo::CargoError),
}
impl From<super::npm::NpmError> for IntakeError {
fn from(e: super::npm::NpmError) -> Self {
IntakeError::Npm(e)
}
}
impl From<super::duties::DutyError> for IntakeError {
fn from(e: super::duties::DutyError) -> Self {
IntakeError::Duty(e)
}
}
impl IntakeError {
/// The frozen `holger-intake-<n>` code. Exhaustive on purpose.
pub fn code(&self) -> &'static holger_errcode::ErrCode {
match self {
IntakeError::SkipsState { .. } => &holger_errcode::INTAKE_SKIPS_STATE,
IntakeError::NotForward { .. } => &holger_errcode::INTAKE_NOT_FORWARD,
IntakeError::Terminal { .. } => &holger_errcode::INTAKE_TERMINAL,
IntakeError::UnknownState(_) => &holger_errcode::INTAKE_UNKNOWN_STATE,
IntakeError::NotIngested(_) => &holger_errcode::INTAKE_NOT_INGESTED,
IntakeError::PolicyRefused { .. } => &holger_errcode::INTAKE_POLICY_REFUSED,
IntakeError::EvidenceMissing { .. } => &holger_errcode::INTAKE_EVIDENCE_MISSING,
IntakeError::AuditFailed(_) => &holger_errcode::INTAKE_AUDIT_FAILED,
IntakeError::Integrity { .. } => &holger_errcode::INTAKE_INTEGRITY,
IntakeError::Immutable(_) => &holger_errcode::INTAKE_IMMUTABLE,
IntakeError::DoorRefused { .. } => &holger_errcode::INTAKE_DOOR_REFUSED,
IntakeError::SideDoor { .. } => &holger_errcode::INTAKE_SIDE_DOOR,
IntakeError::NoCoordinate(_) => &holger_errcode::INTAKE_NO_COORDINATE,
IntakeError::Store(_) => &holger_errcode::INTAKE_STORE,
IntakeError::Duty(d) => d.code(),
IntakeError::Npm(n) => n.code(),
IntakeError::Cargo(c) => c.code(),
}
}
/// **Is this a storage failure, no verdict?** The intake storage or the audit trail could
/// not be written (a full disk, `EIO`, a store that cannot be written): nothing about the
/// artifact was judged (lane HOL-ENOSPC-SETS, field row HOL-PULL-ENOSPC-BLOCK).
pub fn is_storage_failure(&self) -> bool {
matches!(self, IntakeError::Store(_) | IntakeError::AuditFailed(_))
}
/// Both channels in one string: `[holger-intake-1] 'INGESTED' → 'APPROVED' skips …`.
pub fn tagged(&self) -> String {
self.code().tag(self)
}
/// The HTTP-like status a door answers this refusal with.
pub fn status(&self) -> u16 {
match self {
IntakeError::SkipsState { .. }
| IntakeError::NotForward { .. }
| IntakeError::Terminal { .. }
| IntakeError::Immutable(_) => 409,
IntakeError::UnknownState(_) | IntakeError::NoCoordinate(_) => 400,
IntakeError::NotIngested(_) => 404,
IntakeError::PolicyRefused { .. } | IntakeError::EvidenceMissing { .. } => 412,
IntakeError::DoorRefused { .. } | IntakeError::SideDoor { .. } => 403,
IntakeError::AuditFailed(_) | IntakeError::Integrity { .. } | IntakeError::Store(_) => {
500
}
IntakeError::Duty(d) => d.status(),
IntakeError::Npm(n) => n.status(),
IntakeError::Cargo(c) => c.status(),
}
}
}
fn names(states: &[ArtifactState]) -> String {
states
.iter()
.map(|s| s.name())
.collect::<Vec<_>>()
.join(", ")
}
impl std::fmt::Display for IntakeError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
IntakeError::SkipsState { from, to, skipped } => write!(
f,
"'{from}' → '{to}' skips {}; every artifact passes each intake state in order \
(the next state is '{}')",
names(skipped),
from.next().map(|s| s.name()).unwrap_or("none"),
),
IntakeError::NotForward { from, to } => write!(
f,
"'{from}' → '{to}' does not move forward on the intake chain"
),
IntakeError::Terminal { state, to } => {
write!(f, "an artifact in '{state}' cannot move to '{to}'")
}
IntakeError::UnknownState(s) => write!(
f,
"'{s}' is not an intake state (states: {})",
names(&ArtifactState::ALL)
),
IntakeError::NotIngested(a) => write!(
f,
"'{a}' has no intake record; it was never ingested through a door"
),
IntakeError::PolicyRefused { to, reasons } => {
write!(f, "the policy for '{to}' refused: {}", reasons.join("; "))
}
IntakeError::EvidenceMissing { to, missing } => write!(
f,
"the policy for '{to}' requires evidence that was not supplied: {}",
missing
.iter()
.map(|k| k.name())
.collect::<Vec<_>>()
.join(", ")
),
IntakeError::AuditFailed(e) => write!(
f,
"the audit record could not be written ({e}); the transition did not happen"
),
IntakeError::Integrity { want, got } => write!(
f,
"the bytes no longer match the ingested digest (want sha256 {want}, got {got})"
),
IntakeError::Immutable(a) => {
write!(f, "'{a}' is sealed; its bytes cannot be ingested again")
}
IntakeError::DoorRefused { repo, why } => write!(
f,
"repository '{repo}' is fed only by the intake chain: {why}"
),
IntakeError::SideDoor { repo, why } => write!(
f,
"repository '{repo}' is outside the intake chain, and {why}"
),
IntakeError::NoCoordinate(p) => write!(
f,
"the upload '{p}' names no artifact coordinate, so the intake chain cannot \
track it; upload with a path the repository maps (PUT /<repo>/<path>) or \
over gRPC PutArtifact"
),
IntakeError::Store(e) => write!(f, "intake storage failed: {e}"),
IntakeError::Duty(d) => d.fmt(f),
IntakeError::Npm(n) => n.fmt(f),
IntakeError::Cargo(c) => c.fmt(f),
}
}
}
impl std::error::Error for IntakeError {}
impl From<PromoteError> for IntakeError {
fn from(e: PromoteError) -> Self {
match e {
PromoteError::PolicyBlocked(reasons) => IntakeError::PolicyRefused {
to: ArtifactState::Sealed,
reasons,
},
PromoteError::Ladder(l) => IntakeError::PolicyRefused {
to: ArtifactState::Sealed,
reasons: vec![l.tagged()],
},
other => IntakeError::Store(other.tagged()),
}
}
}
// ── the pure machine ────────────────────────────────────────────────────────
/// **Judge one transition.** `from` is the current state, `held_from` the state a
/// QUARANTINED artifact was held at (ignored otherwise), `to` the state asked for.
///
/// Allowed, and nothing else:
/// * the next state on the chain;
/// * BLOCKED from any state but BLOCKED;
/// * QUARANTINED from INGESTED, SCANNED, POLICY_CHECKED or APPROVED;
/// * from QUARANTINED: BLOCKED, or back to the state it was held at.
pub fn check_transition(
from: ArtifactState,
held_from: Option<ArtifactState>,
to: ArtifactState,
) -> Result<(), IntakeError> {
use ArtifactState::*;
if from == Blocked {
return Err(IntakeError::Terminal { state: from, to });
}
if to == Blocked {
return Ok(());
}
if from == Quarantined {
return match held_from {
Some(h) if h == to => Ok(()),
_ => Err(IntakeError::Terminal { state: from, to }),
};
}
if to == Quarantined {
return if matches!(from, Ingested | Scanned | PolicyChecked | Approved) {
Ok(())
} else {
Err(IntakeError::Terminal { state: from, to })
};
}
if from == Available {
return Err(IntakeError::Terminal { state: from, to });
}
let (Some(fi), Some(ti)) = (from.rank(), to.rank()) else {
return Err(IntakeError::NotForward { from, to });
};
if ti <= fi {
return Err(IntakeError::NotForward { from, to });
}
if ti > fi + 1 {
return Err(IntakeError::SkipsState {
from,
to,
skipped: ArtifactState::CHAIN[fi + 1..ti].to_vec(),
});
}
Ok(())
}
// ── configuration ───────────────────────────────────────────────────────────
/// The policy for arriving at one state, from the RON `promotion.intake` block.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct TransitionPolicy {
/// The state this policy guards the arrival at.
pub to: ArtifactState,
/// Gates from the same catalog the rungs use, run over the artifact.
#[serde(default)]
pub gates: Vec<String>,
/// Evidence slots the transition must carry.
#[serde(default)]
pub require_evidence: Vec<EvidenceKind>,
}
/// The `intake:` key inside the RON `promotion:` block.
///
/// ```ron
/// promotion: Some((
/// ladder_name: "intake",
/// stages: [
/// (name: "incoming", repo: "incoming"),
/// (name: "quarantine", repo: "quarantine", gates: ["osv"]),
/// (name: "approved", repo: "approved"),
/// (name: "production", repo: "production", gates: ["provenance"]),
/// ],
/// intake: Some((
/// incoming: "incoming", quarantine: "quarantine",
/// approved: "approved", production: "production",
/// transitions: [ (to: APPROVED, require_evidence: [approval]) ],
/// )),
/// )),
/// ```
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct IntakeSettings {
pub incoming: String,
pub quarantine: String,
pub approved: String,
pub production: String,
#[serde(default)]
pub transitions: Vec<TransitionPolicy>,
/// The file and archive inspector run on every ingest (lane S3): its
/// expansion limits, per-format policies and the optional ClamAV / YARA
/// scanners. Absent = on, with the safe defaults.
#[cfg(feature = "scanners")]
#[serde(default)]
pub inspect: holger_inspect::InspectConfig,
/// The secrets scanner's rule file (`(rules: "…")`, a `holger.secret-rules/v1`
/// RON carried over the airgap like the YARA rules). Absent: the built-in set.
#[cfg(feature = "scanners")]
#[serde(default)]
pub secrets: Option<crate::scanner::secrets::SecretsSettings>,
/// **The intake's rules** (lane SANDBOX-SCAN): `rules: (sandbox: required)`,
/// the default. Every scan that unpacks, parses or runs an engine over an
/// artifact runs in the scan sandbox; `off` is allowed in a test build only
/// (`holger-sandbox-6`).
/// The same rule decides where every job runs (lane HOL-SANDBOX-TRAIT): the
/// verification of every bundle that comes in, on a downstream too.
#[serde(default)]
pub rules: crate::job_runner::IntakeRules,
/// The sandbox: the engine, the worker image and its limits. Absent: the
/// defaults (`localhost/holger-scan-worker`, rootless podman). Every job runs
/// in it (`crate::job_runner`), on a downstream too.
#[serde(default)]
pub sandbox: Option<holger_sandbox::SandboxConfig>,
/// The scanners the SCANNED transition runs (`crate::scanner`). Absent ⇒ the
/// transition carries only the evidence its caller hands it.
#[serde(default)]
pub scan: Option<IntakeScanSettings>,
/// **How many distinct approvers APPROVED needs** (two-person approval is
/// `approvals: 2`). None of them may be the importer. Default 1.
#[serde(default = "one_approval")]
pub approvals: u32,
/// **The trust policy an imported bundle verifies against** at the box's
/// import doors (`AdminService.ImportBundle`, `POST /-/bundle/import`, the
/// `bundle.inbox` job; [`crate::import_door`]): a `holger.bundle-trust-policy/v1`
/// RON file. `holger-server bundle import --policy` overrides it. Absent ⇒
/// the doors refuse by name (`holger-import-4`); the CLI then needs `--policy`.
#[serde(default)]
pub trust_policy: Option<String>,
/// **The chain's npm door** (lane P5, [`super::npm`]): the repository (`ron_repo_type:
/// "npm"`, no archive) whose `npm publish` ingests into `incoming` and whose packument and
/// tarballs are the AVAILABLE set. Absent ⇒ no npm door.
#[serde(default)]
pub npm: Option<String>,
/// **The chain's cargo door** (lane HOL-CARGO-PUBLISH-DOOR, [`super::cargo`]): the
/// repository (`ron_repo_type: "rust"`, no storage) whose `cargo publish` ingests into
/// `incoming`, whose sparse index and downloads are the AVAILABLE set, and whose yanks are
/// recorded beside the chain. Absent ⇒ no cargo door.
#[serde(default)]
pub cargo: Option<String>,
/// **The chain's bundles door** (lane HOL-BUNDLES-DOOR): the raw repository downstreams
/// mirror (the estate's `bundles`). An HTTP PUT there is a signed Holger bundle, verified in
/// the `verify-bundle` job against [`Self::bundles_trust_policy`] (the release signature, the
/// approvals, every member) before a byte is stored; anything else is refused by name
/// (`holger-bundles-1`). Its members were scanned and approved where the bundle was made,
/// so it is a door of the chain, not a side door. Absent ⇒ no bundles door.
#[serde(default)]
pub bundles: Option<String>,
/// **The trust policy of the bundles door** (`holger.bundle-trust-policy/v1` RON). Absent ⇒
/// [`Self::trust_policy`]; neither ⇒ every PUT into the door is refused (fail closed).
#[serde(default)]
pub bundles_trust_policy: Option<String>,
/// **Scanned pushes** (lane HOL-OCI-EVIDENCE, [`super::oci_push`]): every writable `oci`
/// repository takes its pushes INTO this chain (scan, policy, approvals, then served); a
/// row `(repository: "<name>", scan: off)` opts one out, refused by name where the
/// policy requires a scan (`holger-oci-1`).
#[serde(default)]
pub oci_push: Vec<super::oci_push::OciPushRule>,
/// **Repositories outside the chain that take writes unscanned** (lane HOL-FIELD-2 P19).
/// Where the policy requires a scan, every other repository outside the chain refuses
/// writes (`holger-intake-15`), and naming one here is refused at boot
/// (`holger-intake-16`).
#[serde(default)]
pub unscanned: Vec<String>,
}
/// The `scan:` key inside `promotion.intake`.
///
/// ```ron
/// scan: Some((required: [vulnerability, sbom, malware])),
/// ```
///
/// With it, the SCANNED transition runs the server's scanners over the
/// artifact's bytes, keeps the evidence (append-only, audited, signed when an
/// attestation signer is configured) and fills the transition's evidence slots
/// from it. A required kind with no scanner installed fills its slot `fail`, so
/// the transition is refused; it never passes because a check was missing.
#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct IntakeScanSettings {
/// Scanner kinds that must report; an error from one blocks.
#[serde(default)]
pub required: Vec<crate::scanner::ScannerKind>,
/// The provenance scanner's policy (lane G3a): the `rules` floor and each
/// source's `gates.provenance`. Absent ⇒ the provenance slot stands empty.
#[cfg(feature = "scanners")]
#[serde(default)]
pub provenance: Option<crate::scanner::provenance::ProvenanceScanSettings>,
}
/// The most rendered pages (or packuments) one chain keeps; past it the cache starts again.
const PAGE_CACHE_CAP: usize = 65_536;
/// The page a simple-index path names, as the cache keys it: the normalized project (`""`
/// for the root page) and the form. `None` for a path that is no index page.
fn simple_page_key(sub: &str, json: bool) -> Option<(String, bool)> {
let sub = sub.split('?').next().unwrap_or(sub);
let (sub, json) = match sub.trim_end_matches('/').strip_suffix("/json") {
Some(s) if s.trim_start_matches('/').starts_with("simple/") => (s, true),
_ => (sub, json),
};
let rest = sub.trim_start_matches('/').strip_prefix("simple")?;
if !(rest.is_empty() || rest.starts_with('/')) {
return None;
}
let project = rest.trim_matches('/');
if project.contains('/') {
return None;
}
Some((
if project.is_empty() {
String::new()
} else {
normalize_project(project)
},
json,
))
}
/// Why a build without scanners refuses an upload (`holger-downstream-3`).
#[cfg(not(feature = "scanners"))]
pub fn no_scanners_no_ingest() -> String {
holger_errcode::DOWNSTREAM_NO_UPLOAD.tag(
"this Holger is a downstream and runs no scanner: a package arrives only from its upstream, \
as a signed bundle (carried in or pulled), never as an upload",
)
}
fn one_approval() -> u32 {
1
}
impl IntakeSettings {
/// The four repositories in ladder order.
pub fn repos(&self) -> [&str; 4] {
[
&self.incoming,
&self.quarantine,
&self.approved,
&self.production,
]
}
/// Refuse a declaration that cannot mean what it says. `ladder_repos` is the
/// promotion ladder's repositories in rung order.
pub fn validate(&self, ladder_repos: &[&str], allow_skipping: bool) -> anyhow::Result<()> {
let repos = self.repos();
let mut seen = std::collections::BTreeSet::new();
for r in repos {
if !seen.insert(r) {
anyhow::bail!(
"config `promotion.intake`: repository '{r}' is named for two intake roles; \
incoming, quarantine, approved and production must be four repositories"
);
}
}
if ladder_repos != repos {
anyhow::bail!(
"config `promotion.intake`: the promotion ladder must be exactly the intake \
repositories in order ({}), so each rung's gates guard one intake hop; the \
ladder declares ({})",
repos.join(" → "),
ladder_repos.join(" → "),
);
}
if allow_skipping {
anyhow::bail!(
"config `promotion.intake`: `allow_skipping: true` would let an artifact jump a \
rung of the intake chain; the intake chain passes every state in order"
);
}
if let Some(npm) = &self.npm {
if repos.contains(&npm.as_str()) {
anyhow::bail!(
"config `promotion.intake`: `npm: \"{npm}\"` names an intake repository; the npm \
door is a repository of its own (`ron_repo_type: \"npm\"`) that feeds '{}'",
self.incoming
);
}
}
if let Some(cargo) = &self.cargo {
if repos.contains(&cargo.as_str()) || self.npm.as_deref() == Some(cargo.as_str()) {
anyhow::bail!(
"config `promotion.intake`: `cargo: \"{cargo}\"` names an intake repository or \
the npm door; the cargo door is a repository of its own (`ron_repo_type: \
\"rust\"`) that feeds '{}'",
self.incoming
);
}
}
if self.approvals == 0 {
anyhow::bail!(
"config `promotion.intake`: `approvals: 0` would let an artifact reach APPROVED \
with nobody approving it; the least is 1, two-person approval is 2"
);
}
let mut seen_to = std::collections::BTreeSet::new();
for p in &self.transitions {
if p.to == ArtifactState::Untrusted {
anyhow::bail!(
"config `promotion.intake`: no transition arrives at UNTRUSTED; a policy for \
it would never run"
);
}
if !seen_to.insert(p.to) {
anyhow::bail!(
"config `promotion.intake`: two policies for '{}'; one policy per state",
p.to
);
}
}
Ok(())
}
}
/// A [`TransitionPolicy`] with its gates resolved against the catalog at boot.
#[derive(Debug, Default)]
pub struct ResolvedPolicy {
pub gates: GateSet,
pub require_evidence: Vec<EvidenceKind>,
}
// ── the append-only record ──────────────────────────────────────────────────
/// One line of an artifact's intake record.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct IntakeEntry {
pub from: ArtifactState,
pub to: ArtifactState,
/// The coordinate, as `namespace/name@version`.
pub artifact: String,
/// sha256 of the ingested bytes; every later transition re-checks it.
pub sha256: String,
/// The intake repository the artifact stands in after this transition.
#[serde(default)]
pub repo: String,
pub actor: String,
/// Unix seconds.
pub at: i64,
/// The evidence slots this transition carried.
#[serde(default)]
pub evidence: Vec<Evidence>,
/// What the policy check decided, in words.
#[serde(default)]
pub policy: String,
#[serde(default)]
pub reason: String,
/// **The source the artifact came from** (tier 1 `sources:`, [`crate::sources`]).
/// Empty on a chain declared with the old `promotion:` spelling.
#[serde(default, skip_serializing_if = "String::is_empty")]
pub origin: String,
/// **The artifact's effective policy**: the floor plus its source's gates, fixed at
/// ingest and carried on every later line, so an auditor reads which rule refused
/// it and who set that rule.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub effective_policy: Option<crate::sources::EffectivePolicy>,
}
impl IntakeEntry {
/// The label a record names a coordinate by: `ns/name@version`.
pub fn label_of(id: &ArtifactId) -> String {
super::artifact_label(id)
}
}
/// Where an artifact stands now: the reduction of its record.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct IntakeStatus {
pub state: ArtifactState,
/// The state a QUARANTINED artifact was held at.
pub held_from: Option<ArtifactState>,
pub sha256: String,
pub last: IntakeEntry,
}
/// The append-only intake record: one JSON-lines file per coordinate under
/// `<prod_root>/lineage/intake/`, and the approval votes beside it under
/// `<prod_root>/lineage/intake-approvals/`.
#[derive(Debug, Clone)]
pub struct IntakeLog {
root: PathBuf,
votes: PathBuf,
/// The record's reduction in memory (HOL-PERF-1), shared by every clone.
catalog: Arc<super::catalog::Catalog>,
}
impl IntakeLog {
pub fn new(root: impl Into<PathBuf>) -> Self {
let root = root.into();
let votes = match root.parent() {
Some(p) => p.join("intake-approvals"),
None => root.join("approvals"),
};
let catalog = Arc::new(super::catalog::Catalog::new(root.clone()));
Self {
root,
votes,
catalog,
}
}
/// The catalog over this record.
pub(crate) fn catalog(&self) -> &super::catalog::Catalog {
&self.catalog
}
fn votes_path(&self, id: &ArtifactId) -> PathBuf {
self.votes.join(format!("{}.jsonl", coord_key(id)))
}
pub fn append_vote(&self, id: &ArtifactId, vote: &ApprovalVote) -> std::io::Result<()> {
super::stages::append_jsonl(&self.votes_path(id), vote)
}
/// Every approval vote ever cast for `id`, oldest first.
pub fn votes(&self, id: &ArtifactId) -> Vec<ApprovalVote> {
let Ok(text) = std::fs::read_to_string(self.votes_path(id)) else {
return Vec::new();
};
text.lines()
.filter(|l| !l.trim().is_empty())
.filter_map(|l| serde_json::from_str::<ApprovalVote>(l).ok())
.collect()
}
fn path(&self, id: &ArtifactId) -> PathBuf {
self.root.join(format!("{}.jsonl", coord_key(id)))
}
pub fn append(&self, id: &ArtifactId, entry: &IntakeEntry) -> std::io::Result<()> {
super::stages::append_jsonl(&self.path(id), entry)?;
// The record took: the catalog follows (here, and through the journal elsewhere).
self.catalog.applied(&coord_key(id), entry);
Ok(())
}
/// Every entry, oldest first. A torn line is skipped, as the stage lineage does.
pub fn history(&self, id: &ArtifactId) -> Vec<IntakeEntry> {
let Ok(text) = std::fs::read_to_string(self.path(id)) else {
return Vec::new();
};
text.lines()
.filter(|l| !l.trim().is_empty())
.filter_map(|l| serde_json::from_str::<IntakeEntry>(l).ok())
.collect()
}
/// Every coordinate with a record, as `(file key, status)`, sorted by key: the
/// catalog's (the record is scanned once, then kept).
pub fn all(&self) -> Vec<(String, IntakeStatus)> {
self.catalog.all()
}
/// The current state: the last entry's `to`. From the catalog once it is built (one
/// `stat` of its journal), otherwise from the coordinate's file.
pub fn status(&self, id: &ArtifactId) -> Option<IntakeStatus> {
if let Some(st) = self.catalog.status_if_built(&coord_key(id)) {
return st;
}
std::fs::read_to_string(self.path(id))
.ok()
.as_deref()
.and_then(super::catalog::reduce)
}
}
// ── the security dashboard's queues ─────────────────────────────────────────
/// One queue of the security dashboard: a stable name, the words an operator
/// reads, and the states it holds. The SERVER owns this grouping; a console
/// only renders it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct IntakeQueue {
pub name: &'static str,
pub title: &'static str,
pub states: &'static [ArtifactState],
}
/// The five queues, in the order a dashboard shows them. Every state is in
/// exactly one.
pub const QUEUES: [IntakeQueue; 5] = [
IntakeQueue {
name: "incoming",
title: "Incoming",
states: &[
ArtifactState::Untrusted,
ArtifactState::Ingested,
ArtifactState::Scanned,
],
},
IntakeQueue {
name: "quarantined",
title: "Quarantined",
states: &[ArtifactState::Quarantined],
},
IntakeQueue {
name: "awaiting-approval",
title: "Awaiting approval",
states: &[ArtifactState::PolicyChecked],
},
IntakeQueue {
name: "blocked",
title: "Blocked",
states: &[ArtifactState::Blocked],
},
IntakeQueue {
name: "available",
title: "Approved / available",
states: &[
ArtifactState::Approved,
ArtifactState::Sealed,
ArtifactState::Promoted,
ArtifactState::Available,
],
},
];
/// The queue a state stands in.
pub fn queue_of(state: ArtifactState) -> &'static IntakeQueue {
QUEUES
.iter()
.find(|q| q.states.contains(&state))
.expect("every state is in a queue")
}
/// The queue named `name`, if there is one.
pub fn queue_named(name: &str) -> Option<&'static IntakeQueue> {
QUEUES.iter().find(|q| q.name == name)
}
// ── approvals (separation of duties, `super::duties`) ───────────────────────
/// Who approves: the authenticated principal and how it authenticated.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Approver {
/// The stable subject (OIDC `sub`, static-token subject, mTLS CN).
pub principal: String,
/// `oidc`, `mtls`, `bearer`, `passkey` … or `in-process` for a local caller.
pub method: String,
/// The human name for the audit line; authorizes nothing.
pub display: Option<String>,
}
impl Approver {
/// The approver a door authenticated.
pub fn from_identity(id: Option<&crate::auth::AuthIdentity>) -> Self {
match id {
Some(i) => Self {
principal: i.subject.clone(),
method: i.method.clone(),
display: i.display.clone(),
},
None => Self {
principal: super::duties::ANONYMOUS.into(),
method: "none".into(),
display: None,
},
}
}
/// An in-process caller (tests, the local CLI) that names itself.
pub fn in_process(actor: impl Into<String>) -> Self {
Self {
principal: actor.into(),
method: "in-process".into(),
display: None,
}
}
}
/// **One approval vote**: an in-toto Statement v1 about the artifact's sha256,
/// attested by the approver's authenticated principal. Kept append-only; the
/// APPROVED entry carries one `approval` evidence slot per vote whose
/// `reference` is `sha256:<statement_sha256>`.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ApprovalVote {
pub artifact: String,
pub sha256: String,
pub approver: String,
pub method: String,
#[serde(default)]
pub display: Option<String>,
pub at: i64,
/// The intake record's length when the vote was cast: a vote counts only
/// while nothing else happened to the artifact since.
pub seq: usize,
#[serde(default)]
pub reason: String,
pub statement: serde_json::Value,
pub statement_sha256: String,
}
impl ApprovalVote {
#[allow(clippy::too_many_arguments)]
fn new(
artifact: &str,
sha256: &str,
approver: &Approver,
importer: &str,
seq: usize,
quorum: u32,
reason: &str,
at: i64,
) -> Self {
let statement = serde_json::json!({
"_type": "https://in-toto.io/Statement/v1",
"subject": [{ "name": artifact, "digest": { "sha256": sha256 } }],
"predicateType": super::duties::APPROVAL_PREDICATE,
"predicate": {
"approver": approver.principal,
"method": approver.method,
"display": approver.display,
"importer": importer,
"quorum": quorum,
"reason": reason,
"at": at,
},
});
let bytes = serde_json::to_vec(&statement).unwrap_or_default();
Self {
artifact: artifact.to_string(),
sha256: sha256.to_string(),
approver: approver.principal.clone(),
method: approver.method.clone(),
display: approver.display.clone(),
at,
seq,
reason: reason.to_string(),
statement,
statement_sha256: sha256_hex(&bytes),
}
}
/// The `approval` evidence slot this vote fills.
pub fn evidence(&self) -> Evidence {
Evidence::new(
EvidenceKind::Approval,
&self.approver,
EvidenceVerdict::Pass,
)
.with_reference(format!("sha256:{}", self.statement_sha256))
}
}
/// What an approval vote did.
#[derive(Debug, Clone)]
// One outcome per call, returned and matched at once, never stored in bulk: the entry stays inline.
#[allow(clippy::large_enum_variant)]
pub enum ApproveOutcome {
/// Recorded; the quorum is not reached and the state did not move.
Pending {
have: u32,
need: u32,
vote: ApprovalVote,
},
/// The quorum is reached and the artifact is APPROVED.
Approved {
have: u32,
need: u32,
entry: IntakeEntry,
},
}
/// The words an invalidation's QUARANTINED entry starts with
/// ([`IntakeChain::invalidate`], run by the sync protocol, `crate::sync`).
pub const INVALIDATED_PREFIX: &str = "invalidated by upstream";
/// What [`IntakeChain::invalidate`] did to one artifact.
#[derive(Debug, Clone)]
// One outcome per call, returned and matched at once, never stored in bulk: the entry stays inline.
#[allow(clippy::large_enum_variant)]
pub enum InvalidateOutcome {
/// Held QUARANTINED now; the entry says from which state.
Held(IntakeEntry),
/// Already QUARANTINED or BLOCKED; nothing changed.
AlreadyHeld(ArtifactState),
}
// ── the live chain ──────────────────────────────────────────────────────────
/// A request to move one artifact to one state.
#[derive(Debug, Clone)]
pub struct TransitionRequest {
pub id: ArtifactId,
pub to: ArtifactState,
pub actor: String,
/// The client address, for the audit record.
pub source: String,
pub evidence: Vec<Evidence>,
pub reason: String,
}
impl TransitionRequest {
pub fn new(id: ArtifactId, to: ArtifactState, actor: impl Into<String>) -> Self {
Self {
id,
to,
actor: actor.into(),
source: String::new(),
evidence: Vec::new(),
reason: String::new(),
}
}
pub fn with_evidence(mut self, e: Evidence) -> Self {
self.evidence.push(e);
self
}
pub fn with_reason(mut self, r: impl Into<String>) -> Self {
self.reason = r.into();
self
}
pub fn from_source(mut self, s: impl Into<String>) -> Self {
self.source = s.into();
self
}
}
/// A sealed wheel's core metadata (PEP 658): the `METADATA` bytes and their sha256, or `None` when the wheel has none.
type CoreMeta = Option<(Arc<Vec<u8>>, String)>;
/// **The intake chain of one server.** Built once at boot and shared by the
/// HTTP door, the gRPC door and the CLI.
pub struct IntakeChain {
settings: IntakeSettings,
engine: PromotionEngine,
log: IntakeLog,
audit: Arc<dyn AuditLog>,
incoming: Arc<dyn RepositoryBackendTrait>,
ecosystem: String,
ingest_gates: GateSet,
policies: BTreeMap<ArtifactState, ResolvedPolicy>,
/// The in-process inspector: a TEST build's chain without a scan worker
/// (`with_inspector`). A release build never walks an archive here.
#[cfg(feature = "scanners")]
inspector: holger_inspect::Inspector,
/// **The scan worker** (`crate::sandbox`): the INGEST inspection runs in it.
#[cfg(feature = "scanners")]
worker: Option<Arc<crate::sandbox::ScanWorker>>,
/// The server's ONE job runner (lane HOL-SANDBOX-TRAIT), for the Security page.
jobs: Option<Arc<crate::job_runner::Jobs>>,
/// The incoming repository's format name (`pip`, `npm`, `raw`, …), which
/// with the file name gives the format an artifact declares.
#[cfg(feature = "scanners")]
format_name: String,
/// The scanners the SCANNED transition runs, when the config names them.
scanner: Option<Arc<crate::scanner::ScanPipeline>>,
/// Where SCANNED writes a model file's `model.*` properties, when set.
properties: Option<Arc<crate::properties::PropertyStore>>,
/// **One transition at a time PER ARTIFACT**, so two concurrent requests cannot
/// both pass the check against the same state. Keyed by the coordinate: two
/// different artifacts ingest, scan and move independently (the bench found one
/// process-wide lock held across the SCANNED step, scans included, which ran 64
/// parallel packages on one core).
locks: Mutex<std::collections::HashMap<String, Arc<Mutex<()>>>>,
/// The npm door's dist-tag record (lane P5), beside the intake record.
npm_tags: super::npm::TagLog,
/// The cargo door's yank record (lane HOL-CARGO-PUBLISH-DOOR), beside the intake record.
cargo_yanks: super::cargo::YankLog,
/// The public base the cargo door's `config.json` names (the door's `ron_base_url`);
/// `None` ⇒ the door derives it from the request.
cargo_base: Option<String>,
/// The packument's view of a sealed tarball, by its sha256 (the bytes never change).
npm_meta: Mutex<BTreeMap<String, super::npm::VersionMeta>>,
/// A sealed wheel's core metadata (PEP 658), by the wheel's sha256: the
/// `METADATA` bytes and their sha256, or `None` when the wheel has none.
core_meta: Mutex<BTreeMap<String, CoreMeta>>,
/// **The index rows** (lane HOL-PERF-2): what the scan worker read out of each package
/// for its format's index, stored at ingest by sha256.
rows: super::rows::RowStore,
/// **The simple index pages, rendered once per change** (lane HOL-PERF-2, X5): by the
/// normalized project (`""` for the root page) and the form, with the catalog
/// generation each was rendered at.
pages: Mutex<std::collections::HashMap<(String, bool), RenderedPage>>,
/// **The packuments, rendered once per change**: by package and tarball base, with the
/// catalog generation and the dist-tag record's length each was rendered at.
packuments: Mutex<std::collections::HashMap<(String, String), RenderedPackument>>,
/// **The sealed blobs verified so far** (HOL-PERF-3), by content address: the file's
/// identity, size and times when its blake3 matched the seal and its sha256 the intake
/// record. A download whose blob file still has that stamp is served without hashing;
/// a blob changed on disk (written, replaced, truncated) has another stamp and is
/// verified again, so a tampered seal is still refused.
verified: Mutex<std::collections::HashMap<String, (BlobStamp, String)>>,
/// How many sealed blobs were hashed in full to serve them.
verifications: std::sync::atomic::AtomicU64,
/// The public base the packument's tarball links start with (the npm repository's
/// `ron_base_url`); `None` ⇒ the door derives it from the request.
npm_base: Option<String>,
/// The effective policy of each upload source (tier 1 `sources:` + `rules:`); `None`
/// on a chain declared with the old `promotion:` spelling.
sources: Option<crate::sources::SourcePolicies>,
/// **The pull-through into this chain** ([`crate::pull_intake`]): remote
/// sources whose fetched packages are INGESTED here like an upload.
#[cfg(feature = "jobs")]
pull: std::sync::OnceLock<Arc<crate::pull_intake::PullIntake>>,
/// **The chain views** (`name` → `ron_repo_type`: deb, rpm, helm, maven3, go): an upload at
/// `<view>/<path>` is a package of that format and is scanned as one (lane HOL-FIELD-TEST P12).
views: std::sync::OnceLock<std::collections::BTreeMap<String, String>>,
/// **The scanned registries** ([`super::oci_push`]): pushes to them come in here.
oci_push: Option<Arc<super::oci_push::OciPush>>,
/// **The side doors** (lane HOL-FIELD-2 P19): repositories outside the chain whose writes
/// are refused because the policy requires a scan, each with the words why.
side_doors: std::sync::OnceLock<std::collections::BTreeMap<String, String>>,
/// **Re-read the Sigstore trust root** (lane HOL-OCI-EVIDENCE): set at boot, called by the
/// import door once a signed refresh of the root is installed.
trust_reloader: std::sync::OnceLock<TrustReloader>,
/// **This Holger's part in the sync protocol** ([`crate::sync`], H4).
#[cfg(feature = "sync")]
sync: std::sync::OnceLock<Arc<crate::sync::SyncHub>>,
}
/// A simple index page as rendered (lane HOL-PERF-2): the catalog generation it was
/// rendered at, its Content-Type and its bytes.
type RenderedPage = (u64, &'static str, traits::Bytes);
/// A packument as rendered: the catalog generation, the dist-tag record's length, the bytes.
type RenderedPackument = (u64, usize, traits::Bytes);
/// What re-reads the trust root (the provenance scanner's and the pull-through's) and says how.
pub type TrustReloader = Box<dyn Fn() -> Result<String, String> + Send + Sync>;
/// A blob file's identity, size and times: what makes a verification reusable.
type BlobStamp = (u64, u64, u64, i64, i64, i64, i64);
fn stamp_of(m: &std::fs::Metadata) -> BlobStamp {
use std::os::unix::fs::MetadataExt as _;
(
m.dev(),
m.ino(),
m.len(),
m.mtime(),
m.mtime_nsec(),
m.ctime(),
m.ctime_nsec(),
)
}
fn now_secs() -> i64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs() as i64)
.unwrap_or(0)
}
fn sha256_hex(bytes: &[u8]) -> String {
crate::attest::ArtifactDigest::of_sha256(bytes)
.hex()
.to_string()
}
impl IntakeChain {
/// Assemble the chain. `engine` must carry the intake ladder (the server
/// derives both from one config, see `Holger::intake_chain`).
pub fn new(
settings: IntakeSettings,
engine: PromotionEngine,
prod_root: impl Into<PathBuf>,
audit: Arc<dyn AuditLog>,
incoming: Arc<dyn RepositoryBackendTrait>,
) -> Self {
let prod_root: PathBuf = prod_root.into();
let format_name = format!("{:?}", incoming.format()).to_ascii_lowercase();
#[cfg(feature = "scanners")]
let ecosystem = crate::security::ecosystem_for_format(incoming.format());
// A build without scanners scans nothing; the name only labels.
#[cfg(not(feature = "scanners"))]
let ecosystem = format_name.clone();
#[cfg(feature = "scanners")]
let inspector = holger_inspect::Inspector::from_config(&settings.inspect);
Self {
#[cfg(feature = "scanners")]
inspector,
#[cfg(feature = "scanners")]
worker: None,
jobs: None,
#[cfg(feature = "scanners")]
format_name,
settings,
engine,
npm_tags: super::npm::TagLog::new(prod_root.join("lineage").join("npm-dist-tags")),
cargo_yanks: super::cargo::YankLog::new(prod_root.join("lineage").join("cargo-yanks")),
cargo_base: None,
npm_meta: Mutex::new(BTreeMap::new()),
core_meta: Mutex::new(BTreeMap::new()),
rows: super::rows::RowStore::new(prod_root.join("lineage").join("index-rows")),
pages: Mutex::new(std::collections::HashMap::new()),
packuments: Mutex::new(std::collections::HashMap::new()),
verified: Mutex::new(std::collections::HashMap::new()),
verifications: std::sync::atomic::AtomicU64::new(0),
npm_base: None,
log: IntakeLog::new(prod_root.join("lineage").join("intake")),
audit,
incoming,
ecosystem,
ingest_gates: GateSet::empty(),
policies: BTreeMap::new(),
scanner: None,
properties: None,
locks: Mutex::new(std::collections::HashMap::new()),
sources: None,
#[cfg(feature = "jobs")]
pull: std::sync::OnceLock::new(),
views: std::sync::OnceLock::new(),
side_doors: std::sync::OnceLock::new(),
oci_push: None,
trust_reloader: std::sync::OnceLock::new(),
#[cfg(feature = "sync")]
sync: std::sync::OnceLock::new(),
}
}
/// Attach the scanned registries (lane HOL-OCI-EVIDENCE).
pub fn with_oci_push(mut self, push: Arc<super::oci_push::OciPush>) -> Self {
self.oci_push = Some(push);
self
}
/// Set what re-reads the trust root (once, at boot).
pub fn set_trust_reloader(&self, r: TrustReloader) {
let _ = self.trust_reloader.set(r);
}
/// **Re-read the trust root** the config names (after an import installed a refresh).
/// `Ok(None)`: nothing here reads one.
pub fn reload_trust(&self) -> Result<Option<String>, String> {
match self.trust_reloader.get() {
Some(r) => r().map(Some),
None => Ok(None),
}
}
/// The scanned registries, when this chain has any.
pub fn oci_push(&self) -> Option<&Arc<super::oci_push::OciPush>> {
self.oci_push.as_ref()
}
/// Does the registry `repo` push into this chain?
pub fn scans_pushes_to(&self, repo: &str) -> bool {
self.oci_push.as_ref().is_some_and(|p| p.scans(repo))
}
/// Attach the sync hub (`sync:` block). Once; a second is ignored.
#[cfg(feature = "sync")]
pub fn attach_sync(&self, hub: Arc<crate::sync::SyncHub>) {
let _ = self.sync.set(hub);
}
/// The sync hub, when the config has a `sync:` block.
#[cfg(feature = "sync")]
pub fn sync(&self) -> Option<Arc<crate::sync::SyncHub>> {
self.sync.get().cloned()
}
/// The effective policy of each source, when the config declares `sources:` (the Security page).
pub fn sources(&self) -> Option<&crate::sources::SourcePolicies> {
self.sources.as_ref()
}
/// **The per-source policies** (tier 1): every upload resolves its source, and its
/// effective policy rides on its record through the chain.
pub fn with_sources(mut self, sources: crate::sources::SourcePolicies) -> Self {
self.sources = Some(sources);
self
}
/// Attach the pull-through (once, at boot).
/// The chain views this config declares (`name` → `ron_repo_type`), set once at the build.
pub fn attach_views(&self, views: std::collections::BTreeMap<String, String>) {
let _ = self.views.set(views);
}
/// Is `repo` a side door: outside the chain, its writes refused (lane HOL-FIELD-2 P19)?
pub fn is_side_door(&self, repo: &str) -> bool {
self.side_doors.get().is_some_and(|d| d.contains_key(repo))
}
/// The repositories outside the chain whose writes are refused (`name` → why), set once
/// at the build (lane HOL-FIELD-2 P19).
pub fn attach_side_doors(&self, doors: std::collections::BTreeMap<String, String>) {
let _ = self.side_doors.set(doors);
}
#[cfg(feature = "jobs")]
pub fn attach_pull(&self, pull: Arc<crate::pull_intake::PullIntake>) {
let _ = self.pull.set(pull);
}
/// The pull-through into this chain, when the config declares remote sources.
#[cfg(feature = "jobs")]
pub fn pull(&self) -> Option<&Arc<crate::pull_intake::PullIntake>> {
self.pull.get()
}
/// **UNTRUSTED → INGESTED for a package a pull-through FETCHED**: the same
/// door as an upload ([`ingest`](Self::ingest)): the INGEST gates, the
/// inspector and the malware engines, the audit record first, then the
/// bytes into the incoming repository. `actor` is `pull-through:<source>`,
/// `source` the upstream it came from.
pub fn ingest_pulled(
&self,
actor: &str,
source: &str,
id: &ArtifactId,
bytes: &[u8],
) -> Result<IntakeEntry, IntakeError> {
let incoming = self.settings.incoming.clone();
let backend = Arc::clone(&self.incoming);
self.ingest(actor, source, &incoming, id, bytes, || {
backend.put(id, bytes)
})
}
/// **An image pushed to a scanned registry** (lane HOL-OCI-EVIDENCE): its layout INGESTED
/// into `incoming` as the pusher (`actor`, from `peer`), its origin the registry, so the
/// provenance gate named after the registry applies at SCANNED. The pusher, as any
/// importer, never approves it.
pub fn ingest_pushed(
&self,
actor: &str,
peer: &str,
registry: &str,
id: &ArtifactId,
bytes: &[u8],
) -> Result<IntakeEntry, IntakeError> {
let incoming = self.settings.incoming.clone();
let backend = Arc::clone(&self.incoming);
self.ingest_from(actor, peer, &incoming, id, bytes, Some(registry), || {
backend.put(id, bytes)
})
}
/// The bytes an INGESTED (not yet sealed) artifact holds in the incoming
/// repository, re-checked against its digest.
pub fn incoming_artifact(&self, id: &ArtifactId) -> Result<Vec<u8>, IntakeError> {
let st = self
.log
.status(id)
.ok_or_else(|| IntakeError::NotIngested(super::artifact_label(id)))?;
self.incoming_bytes(id, &st.sha256)
}
/// The gates an upload passes before it is INGESTED.
pub fn with_ingest_gates(mut self, gates: GateSet) -> Self {
self.ingest_gates = gates;
self
}
/// Replace the inspector the settings declared (tests, embedders).
#[cfg(feature = "scanners")]
pub fn with_inspector(mut self, inspector: holger_inspect::Inspector) -> Self {
self.inspector = inspector;
self
}
/// Run the INGEST inspection in this scan worker (the sandbox).
#[cfg(feature = "scanners")]
pub fn with_worker(mut self, worker: Arc<crate::sandbox::ScanWorker>) -> Self {
self.worker = Some(worker);
self
}
/// The server's job runner, as the Security page states it.
pub fn with_jobs(mut self, jobs: Option<Arc<crate::job_runner::Jobs>>) -> Self {
self.jobs = jobs;
self
}
/// The server's job runner, when the chain was given it.
pub fn jobs(&self) -> Option<&Arc<crate::job_runner::Jobs>> {
self.jobs.as_ref()
}
/// The scan worker, when the chain has one.
#[cfg(feature = "scanners")]
pub fn worker(&self) -> Option<&Arc<crate::sandbox::ScanWorker>> {
self.worker.as_ref()
}
/// The subject the SCANNED transition scans `id` as: its file name, its
/// ecosystem, and its coordinates when the id carries them.
#[cfg(feature = "scanners")]
fn scan_subject<'a>(
&'a self,
id: &'a ArtifactId,
bytes: &'a [u8],
) -> crate::scanner::ScanSubject<'a> {
let file_name = id.name.rsplit('/').next().unwrap_or(&id.name);
let npm = super::npm::coords_of(id);
let ecosystem = if npm.is_some() {
"npm"
} else if go_coords_of(id).is_some() {
"go"
} else if self.cargo_coords_of(id).is_some() {
"cargo"
} else {
crate::scanner::ecosystem_hint(file_name, &self.ecosystem)
};
let subject = crate::scanner::ScanSubject::new(file_name, ecosystem, bytes);
if !id.version.is_empty() {
subject.with_coords(&id.name, &id.version)
} else {
subject
}
}
/// The `(name, version)` of a crate the cargo door published (`<door>/<name>/<name>-
/// <version>.crate`), when `id` is one (lane HOL-CARGO-PUBLISH-DOOR).
pub fn cargo_coords_of(&self, id: &ArtifactId) -> Option<(String, String)> {
super::cargo::coords_of(self.settings.cargo.as_deref()?, id)
}
/// The inspector every ingest runs.
#[cfg(feature = "scanners")]
pub fn inspector(&self) -> &holger_inspect::Inspector {
&self.inspector
}
/// Inspect one artifact's bytes as the ingest does: the format comes from
/// its file name and the incoming repository's format.
#[cfg(feature = "scanners")]
pub fn inspect(&self, id: &ArtifactId, bytes: &[u8]) -> holger_inspect::Findings {
let file = id.name.rsplit('/').next().unwrap_or(&id.name);
let format = holger_inspect::Format::declared(file, &self.format_name);
if let Some(w) = &self.worker {
// The same subject SCANNED will scan, so both share one worker run.
let subject = self
.scan_subject(id, bytes)
.with_repo(&self.settings.incoming);
return w.inspect(&subject);
}
if crate::sandbox::IN_PROCESS_ALLOWED {
return self.inspector.inspect(bytes, format, file);
}
// A release build never walks an archive in this process.
crate::sandbox::refused_findings(
format,
file,
&holger_sandbox::Refusal::NoEngine(
"this intake chain has no scan worker attached".into(),
),
)
}
/// **The index row of an artifact being ingested** (lane HOL-PERF-2): the one the scan
/// worker answered for the inspection that just ran (the same subject, so the worker's
/// memo, no second run); a test build without a worker reads it in-process, as it
/// inspects. A release build never parses a package here.
#[cfg(feature = "scanners")]
fn row_at_ingest(&self, id: &ArtifactId, bytes: &[u8]) -> Option<holger_index_rows::IndexRow> {
let subject = self
.scan_subject(id, bytes)
.with_repo(&self.settings.incoming);
if let Some(w) = &self.worker {
return w.report(&subject).ok().and_then(|r| r.index_row.clone());
}
if crate::sandbox::IN_PROCESS_ALLOWED {
return holger_index_rows::row_of(subject.ecosystem, subject.file_name, bytes);
}
None
}
/// The resolved policy for arriving at `to`.
/// Run these scanners on every transition into SCANNED, and fill the
/// transition's evidence slots from what they report.
pub fn with_scanner(mut self, pipeline: Arc<crate::scanner::ScanPipeline>) -> Self {
self.scanner = Some(pipeline);
self
}
/// Write a model file's metadata (`model.format`, `model.architecture`,
/// `model.parameters`, `model.tensors`, `model.dtype`, `model.quantization`)
/// into this property store when SCANNED reads it, on the artifact's
/// coordinate in each of the four intake repositories, so the property
/// search axis finds the model wherever it stands on the ladder.
pub fn with_properties(mut self, store: Arc<crate::properties::PropertyStore>) -> Self {
self.properties = Some(store);
self
}
/// The scanners the SCANNED transition runs, if any.
pub fn scanner(&self) -> Option<&Arc<crate::scanner::ScanPipeline>> {
self.scanner.as_ref()
}
pub fn with_policy(mut self, to: ArtifactState, policy: ResolvedPolicy) -> Self {
self.policies.insert(to, policy);
self
}
pub fn settings(&self) -> &IntakeSettings {
&self.settings
}
/// Which intake role `repo` plays, if any.
pub fn role_of(&self, repo: &str) -> Option<IntakeRepo> {
let s = &self.settings;
if repo == s.incoming {
Some(IntakeRepo::Incoming)
} else if repo == s.quarantine {
Some(IntakeRepo::Quarantine)
} else if repo == s.approved {
Some(IntakeRepo::Approved)
} else if repo == s.production {
Some(IntakeRepo::Production)
} else {
None
}
}
fn repo_name(&self, role: IntakeRepo) -> &str {
match role {
IntakeRepo::Incoming => &self.settings.incoming,
IntakeRepo::Quarantine => &self.settings.quarantine,
IntakeRepo::Approved => &self.settings.approved,
IntakeRepo::Production => &self.settings.production,
}
}
/// `Some(refusal)` when a door must not write into `repo`: quarantine,
/// approved and production are fed only by the chain.
/// A repository outside the chain refuses writes where the policy requires a scan
/// (`holger-intake-15`, lane HOL-FIELD-2 P19).
pub fn write_refusal(&self, repo: &str) -> Option<IntakeError> {
let Some(role) = self.role_of(repo) else {
let why = self.side_doors.get()?.get(repo)?;
return Some(IntakeError::SideDoor {
repo: repo.to_string(),
why: why.clone(),
});
};
match role {
IntakeRepo::Incoming => None,
_ => Some(IntakeError::DoorRefused {
repo: repo.to_string(),
why: format!(
"upload into '{}' and move it with an intake transition",
self.settings.incoming
),
}),
}
}
/// `Some(refusal)` when a write of RAW BYTES into `repo`'s backend must not happen: a door
/// that does not speak the repository's own protocol (gRPC `PutArtifact` /
/// `PutArtifactStream`, a queued upload, a replication, a promotion) writing there
/// (SEC-REVIEW 2026-11-01 P3). Everything [`Self::write_refusal`] refuses, and also the
/// doors whose writes walk the chain only through their own protocol over HTTP: a scanned
/// `oci` registry (`docker push`), the npm door (`npm publish`) and the cargo door
/// (`cargo publish`). A raw write there would land in the store the door's view reads,
/// with no intake record.
pub fn raw_write_refusal(&self, repo: &str) -> Option<IntakeError> {
if let Some(refusal) = self.write_refusal(repo) {
return Some(refusal);
}
let how = if self.scans_pushes_to(repo) {
format!("a scanned registry takes pushes only over its OCI door (`docker push` to /v2/{repo}/…), which walks the chain")
} else if self.npm_door() == Some(repo) {
"the npm door takes writes only as `npm publish`, which walks the chain".to_string()
} else if self.cargo_door() == Some(repo) {
"the cargo door takes writes only as `cargo publish`, which walks the chain".to_string()
} else if self.bundles_door() == Some(repo) {
"the bundles door takes writes only as an HTTP PUT of a signed Holger bundle, which it verifies".to_string()
} else {
return None;
};
Some(IntakeError::DoorRefused {
repo: repo.to_string(),
why: how,
})
}
/// `Some(refusal)` when a replication `from` → `to` would carry bytes past the chain
/// (SEC-REVIEW 2026-11-01 P4). Out of the chain only production replicates, read through
/// its view (the AVAILABLE set); incoming, quarantine and approved hold bytes nobody
/// approved. Into the chain nothing replicates (the chain is fed by its own transitions),
/// and the destination must take raw writes ([`Self::raw_write_refusal`]): where a scan is
/// required that refuses every served repository outside the chain, so an Admin-only
/// repository's bytes never land where a reader is served.
pub fn replication_refusal(&self, from: &str, to: &str) -> Option<IntakeError> {
if matches!(
self.role_of(from),
Some(IntakeRepo::Incoming | IntakeRepo::Quarantine | IntakeRepo::Approved)
) {
return Some(IntakeError::DoorRefused {
repo: from.to_string(),
why: format!(
"its artifacts are not AVAILABLE; only '{}' replicates out of the chain",
self.settings.production
),
});
}
if self.role_of(to).is_some() {
return Some(IntakeError::DoorRefused {
repo: to.to_string(),
why: format!(
"a replication never writes into the chain; upload into '{}' and walk it through",
self.settings.incoming
),
});
}
self.raw_write_refusal(to)
}
/// `Some(refusal)` when a promotion `from` → `to` must not happen (SEC-REVIEW 2026-11-01
/// P9): [`Self::promote_refusal`] (the chain's repositories move only by its transitions),
/// and a destination that takes no raw write ([`Self::raw_write_refusal`]): a side door
/// where a scan is required, a scanned registry, the npm or the cargo door.
pub fn promotion_refusal(&self, from: &str, to: &str) -> Option<IntakeError> {
self.promote_refusal(from, to)
.or_else(|| self.raw_write_refusal(to))
}
/// `Some(refusal)` when a promotion names an intake repository: the seal into
/// production happens only as the SEALED transition.
pub fn promote_refusal(&self, from_repo: &str, to_repo: &str) -> Option<IntakeError> {
let repo = [from_repo, to_repo]
.into_iter()
.find(|r| self.role_of(r).is_some())?;
Some(IntakeError::DoorRefused {
repo: repo.to_string(),
why: "a promotion into or out of the intake chain is the SEALED transition, not a \
promote"
.to_string(),
})
}
/// The lock of one coordinate: held for that artifact's transition (its state
/// check, its scans, its record), and nothing else's.
fn lock_for(&self, id: &ArtifactId) -> Arc<Mutex<()>> {
let key = super::coord_key(id);
let mut m = self.locks.lock_or_recover();
if m.len() > 4096 {
m.retain(|_, l| Arc::strong_count(l) > 1);
}
Arc::clone(m.entry(key).or_default())
}
pub fn status(&self, id: &ArtifactId) -> Option<IntakeStatus> {
self.log.status(id)
}
pub fn history(&self, id: &ArtifactId) -> Vec<IntakeEntry> {
self.log.history(id)
}
/// Every artifact with an intake record, where it stands now, newest
/// transition first (the security dashboard's order).
pub fn list(&self) -> Vec<(ArtifactId, IntakeStatus)> {
let mut out = self.all();
out.sort_by(|a, b| {
b.1.last
.at
.cmp(&a.1.last.at)
.then_with(|| a.1.last.artifact.cmp(&b.1.last.artifact))
});
out
}
/// The state of `id`; UNTRUSTED when it has no record.
pub fn state(&self, id: &ArtifactId) -> ArtifactState {
self.status(id)
.map(|s| s.state)
.unwrap_or(ArtifactState::Untrusted)
}
fn audit_first(
&self,
entry: &IntakeEntry,
action: AuditAction,
bytes: u64,
source: &str,
) -> Result<(), IntakeError> {
self.audit_entry(entry, action, bytes, source, true)
}
/// [`audit_first`](Self::audit_first), with or without the trail step. An
/// airgap landing's mirror entries carry no trail step of their own: the
/// secure side's trail is the import's `hash_verified` and `airgap_imported`.
fn audit_entry(
&self,
entry: &IntakeEntry,
action: AuditAction,
bytes: u64,
source: &str,
trail: bool,
) -> Result<(), IntakeError> {
let mut detail = format!("intake {} → {}", entry.from, entry.to);
if !entry.evidence.is_empty() {
let ev: Vec<String> = entry
.evidence
.iter()
.map(|e| format!("{}:{}={:?}", e.kind.name(), e.source, e.verdict))
.collect();
detail.push_str(&format!("; evidence: {}", ev.join(", ")));
}
if !entry.policy.is_empty() {
detail.push_str(&format!("; policy: {}", entry.policy));
}
if !entry.reason.is_empty() {
detail.push_str(&format!("; reason: {}", entry.reason));
}
detail.push_str(&format!("; sha256 {}", entry.sha256));
let mut ev = AuditEvent::new(
&entry.actor,
action,
&entry.repo,
&entry.artifact,
source,
200,
bytes,
)
.with_detail(detail);
ev.sha256 = entry.sha256.clone();
ev.trail = if !trail {
None
} else {
match entry.to {
ArtifactState::Ingested => Some(TrailKind::Imported),
ArtifactState::PolicyChecked => Some(TrailKind::PolicyPassed),
ArtifactState::Approved => Some(TrailKind::Approved),
ArtifactState::Sealed => Some(TrailKind::Sealed),
_ => None,
}
};
self.audit
.record(ev)
.map_err(|e| IntakeError::AuditFailed(e.to_string()))
}
/// The trail's `hash_verified` step: the bytes in `store` were read back and
/// hash to the ingested digest. Written before anything relies on them, and
/// fail-closed like every intake record.
#[allow(clippy::too_many_arguments)]
fn audit_hash_verified(
&self,
actor: &str,
repo: &str,
artifact: &str,
sha256: &str,
nbytes: u64,
source: &str,
store: &str,
) -> Result<(), IntakeError> {
let ev = AuditEvent::new(
actor,
AuditAction::Other,
repo,
artifact,
source,
200,
nbytes,
)
.with_detail(format!(
"hash verified: the bytes in the {store} hash to the ingested sha256 {sha256}"
))
.on_trail(TrailKind::HashVerified, sha256);
self.audit
.record(ev)
.map_err(|e| IntakeError::AuditFailed(e.to_string()))
}
/// **The trail's `exported` step**: an AVAILABLE artifact left this holger
/// for `destination` (a bundle, a transfer medium). Only a sealed, available
/// artifact is exported; the record is written before the caller's effect
/// and a failed write refuses the export (`holger-intake-8`).
pub fn record_export(
&self,
actor: &str,
id: &ArtifactId,
destination: &str,
source: &str,
) -> Result<IntakeEntry, IntakeError> {
let artifact = super::artifact_label(id);
let status = self
.log
.status(id)
.ok_or_else(|| IntakeError::NotIngested(artifact.clone()))?;
if status.state != ArtifactState::Available {
return Err(IntakeError::DoorRefused {
repo: status.last.repo.clone(),
why: format!(
"'{artifact}' is {}, and only an AVAILABLE artifact is exported",
status.state
),
});
}
let ev = AuditEvent::new(
actor,
AuditAction::Other,
&status.last.repo,
&artifact,
source,
200,
0,
)
.with_detail(format!(
"exported to {destination}; sha256 {}",
status.sha256
))
.on_trail(TrailKind::Exported, &status.sha256);
self.audit
.record(ev)
.map_err(|e| IntakeError::AuditFailed(e.to_string()))?;
Ok(status.last)
}
/// Audit a refusal a door decided (separation of duties) beside the chain's own.
pub fn record_refusal(
&self,
actor: &str,
repo: &str,
artifact: &str,
source: &str,
err: &IntakeError,
) {
self.audit_refusal(actor, repo, artifact, source, AuditAction::Promote, err, "")
}
#[allow(clippy::too_many_arguments)]
fn audit_refusal(
&self,
actor: &str,
repo: &str,
artifact: &str,
source: &str,
action: AuditAction,
err: &IntakeError,
sha256: &str,
) {
let mut ev = AuditEvent::new(actor, action, repo, artifact, source, err.status(), 0)
.with_detail(format!("intake refused: {}", err.tagged()));
ev.sha256 = sha256.to_string();
// A policy, a gate or missing evidence said no: the trail's `policy_denied`.
if !sha256.is_empty()
&& matches!(
err,
IntakeError::PolicyRefused { .. } | IntakeError::EvidenceMissing { .. }
)
{
ev.trail = Some(TrailKind::PolicyDenied);
}
if let Err(e) = self.audit.record(ev) {
log::warn!("intake: the refusal's audit record failed too: {e}");
}
}
fn evaluate(
&self,
gates: &GateSet,
actor: &str,
from_repo: &str,
to_repo: &str,
id: &ArtifactId,
bytes: &[u8],
) -> nornir_gate::Decision {
let set = [PromotedArtifact {
id,
ecosystem: self.ecosystem_of(id),
digest: crate::attest::ArtifactDigest::of_sha256(bytes),
bytes,
attestations: &[],
}];
let ctx = if from_repo.is_empty() {
PromotionContext::ingest(actor, to_repo, &set)
} else {
PromotionContext {
principal: actor,
from_repo,
to_repo,
hop: None,
set: &set,
}
};
gates.evaluate(&ctx)
}
/// The ecosystem the gates judge `id` as: a wheel in a `raw` intake
/// repository is a `pip` package (its licence is read from its METADATA),
/// exactly as the scanners see it.
fn ecosystem_of(&self, id: &ArtifactId) -> &str {
if let Some((ecosystem, _)) = self.format_target(id) {
return ecosystem;
}
if super::npm::coords_of(id).is_some() {
return "npm";
}
if go_coords_of(id).is_some() {
return "go";
}
if self.cargo_coords_of(id).is_some() {
return "cargo";
}
let file_name = id.name.rsplit('/').next().unwrap_or(&id.name);
crate::scanner::ecosystem_hint(file_name, &self.ecosystem)
}
/// **What a pulled coordinate is scanned as** (lane HOL-FIELD-TEST P11): a pull-through
/// coordinate is `<source>/<rest>`, and the source's kind says which ecosystem's advisories
/// it is matched against (`pull_intake::scan_target`). Without it a pulled sdist, npm
/// tarball, Maven jar or crate was matched in the `raw` intake repository's ecosystem,
/// which has no advisories, and every known CVE passed.
#[cfg(feature = "jobs")]
fn pulled_target(&self, id: &ArtifactId) -> Option<(&'static str, Option<(String, String)>)> {
let (source, rest) = id.name.split_once('/')?;
let src = self.pull()?.source(source)?;
if src.cfg.kind == crate::pull_intake::RemoteKind::Apk {
// An APK is matched in the advisories of the distribution its repository serves
// (Wolfi, Chainguard, Alpine); never as raw (P12 holds a raw `.apk`).
return Some((crate::pull_intake::apk_distro(&src.cfg.url), None));
}
crate::pull_intake::scan_target(src.cfg.kind, rest)
}
#[cfg(not(feature = "jobs"))]
fn pulled_target(&self, _id: &ArtifactId) -> Option<(&'static str, Option<(String, String)>)> {
None
}
/// **What a coordinate is scanned as, by where it came in**: a pull-through source's
/// (P11), else the chain view it was uploaded under (P12).
fn format_target(&self, id: &ArtifactId) -> Option<(&'static str, Option<(String, String)>)> {
self.pulled_target(id).or_else(|| self.view_target(id))
}
/// **An upload under a chain view is a package of the view's format** (lane HOL-FIELD-TEST
/// P12): `<view>/<path>` in `incoming` is matched in that format's advisories with the
/// coordinates its path names. Before, it was matched as `raw` and every known CVE passed
/// (a Log4Shell POM under `maven/` walked through SCANNED). A Go view's zip is the chain's
/// own (`go_coords_of`).
fn view_target(&self, id: &ArtifactId) -> Option<(&'static str, Option<(String, String)>)> {
let (view, rest) = id.name.split_once('/')?;
let ty = self.views.get()?.get(view)?;
let file = rest.rsplit('/').next().unwrap_or(rest);
match ty.as_str() {
// A .deb is named by its control file, under its source package (the scanner
// reads it, P14): never by the file name, which apt does not install by.
"deb" => Some(("deb", None)),
"rpm" => Some(("rpm", rpm_coords(file))),
"maven3" => Some(("maven", maven_coords(rest))),
"helm" => Some(("helm", None)),
_ => None,
}
}
/// **UNTRUSTED → INGESTED.** Run the ingest gates and the INGESTED policy over
/// the bytes, write the audit record, run `store` (the door's own write into
/// the incoming backend), then record the state.
pub fn ingest(
&self,
actor: &str,
source: &str,
repo: &str,
id: &ArtifactId,
bytes: &[u8],
store: impl FnOnce() -> anyhow::Result<()>,
) -> Result<IntakeEntry, IntakeError> {
self.ingest_from(actor, source, repo, id, bytes, None, store)
}
/// [`ingest`](Self::ingest), the record's origin `origin` when no tier-1 source names one.
#[allow(clippy::too_many_arguments)]
fn ingest_from(
&self,
actor: &str,
source: &str,
repo: &str,
id: &ArtifactId,
bytes: &[u8],
origin: Option<&str>,
store: impl FnOnce() -> anyhow::Result<()>,
) -> Result<IntakeEntry, IntakeError> {
let artifact = super::artifact_label(id);
// A build without scanners (the downstream, H3) never ingests: what it
// serves arrives from its upstream, verified (a signed bundle or a pull).
#[cfg(not(feature = "scanners"))]
let out = {
let _ = (store, origin);
Err(IntakeError::PolicyRefused {
to: ArtifactState::Ingested,
reasons: vec![no_scanners_no_ingest()],
})
};
#[cfg(feature = "scanners")]
let out = self.ingest_inner(actor, source, repo, id, &artifact, bytes, origin, store);
if let Err(e) = &out {
self.audit_refusal(
actor,
repo,
&artifact,
source,
AuditAction::Upload,
e,
&sha256_hex(bytes),
);
}
out
}
#[allow(clippy::too_many_arguments)]
#[cfg(feature = "scanners")]
fn ingest_inner(
&self,
actor: &str,
source: &str,
repo: &str,
id: &ArtifactId,
artifact: &str,
bytes: &[u8],
origin: Option<&str>,
store: impl FnOnce() -> anyhow::Result<()>,
) -> Result<IntakeEntry, IntakeError> {
if self.role_of(repo) != Some(IntakeRepo::Incoming) {
return Err(self
.write_refusal(repo)
.unwrap_or_else(|| IntakeError::DoorRefused {
repo: repo.to_string(),
why: "not an intake repository".into(),
}));
}
let lock = self.lock_for(id);
let _held = lock.lock_or_recover();
let sha256 = sha256_hex(bytes);
let from = match self.log.status(id) {
None => ArtifactState::Untrusted,
Some(s) if s.state.is_sealed() => {
return Err(IntakeError::Immutable(artifact.to_string()))
}
Some(s) if s.state == ArtifactState::Blocked && s.sha256 == sha256 => {
return Err(IntakeError::Terminal {
state: s.state,
to: ArtifactState::Ingested,
});
}
// A new upload of an unsealed coordinate is a new artifact: it starts
// again from the wire.
Some(_) => ArtifactState::Untrusted,
};
check_transition(from, None, ArtifactState::Ingested)?;
// Tier 1: which source this upload is from, and the policy it carries.
let effective = match &self.sources {
Some(sp) => {
Some(
sp.resolve(actor, &id.name)
.map_err(|why| IntakeError::PolicyRefused {
to: ArtifactState::Ingested,
reasons: vec![why],
})?,
)
}
None => None,
};
if let Some(p) = &effective {
let file_name = id.name.rsplit('/').next().unwrap_or(&id.name);
if let Some(why) = p.ingest_refusal(file_name) {
return Err(IntakeError::PolicyRefused {
to: ArtifactState::Ingested,
reasons: vec![why],
});
}
}
let mut reasons = Vec::new();
let mut summaries = Vec::new();
for gates in [
Some(&self.ingest_gates),
self.policies
.get(&ArtifactState::Ingested)
.map(|p| &p.gates),
]
.into_iter()
.flatten()
{
if gates.is_empty() {
continue;
}
let d = self.evaluate(gates, actor, "", repo, id, bytes);
summaries.push(d.summary());
if !d.is_allowed() && gates.enforcement() == super::Enforcement::Block {
reasons.extend(d.rejections.iter().map(|r| r.to_string()));
}
}
if !reasons.is_empty() {
return Err(IntakeError::PolicyRefused {
to: ArtifactState::Ingested,
reasons,
});
}
// File and archive inspection + the malware hook, on every ingest. A
// blocking finding refuses the artifact before a byte is stored.
let findings = self.inspect(id, bytes);
let verdict = match findings.verdict() {
holger_inspect::Verdict::Pass => EvidenceVerdict::Pass,
holger_inspect::Verdict::Warn => EvidenceVerdict::Warn,
holger_inspect::Verdict::Fail => EvidenceVerdict::Fail,
};
if verdict == EvidenceVerdict::Fail {
let reasons = findings
.blocking()
.map(|f| format!("inspect: {f}"))
.collect();
return Err(IntakeError::PolicyRefused {
to: ArtifactState::Ingested,
reasons,
});
}
let mut evidence = vec![
Evidence::new(EvidenceKind::Malware, "holger-inspect", verdict)
.with_reference(findings.summary()),
];
for f in findings
.findings
.iter()
.filter(|f| f.severity == holger_inspect::Severity::Warn)
{
evidence.push(
Evidence::new(
EvidenceKind::Malware,
"holger-inspect",
EvidenceVerdict::Warn,
)
.with_reference(f.to_string()),
);
}
summaries.push(findings.summary());
if let Some(p) = &effective {
summaries.push(format!("effective policy: {}", p.summary()));
}
let entry = IntakeEntry {
from,
to: ArtifactState::Ingested,
artifact: artifact.to_string(),
sha256,
repo: repo.to_string(),
actor: actor.to_string(),
at: now_secs(),
evidence,
policy: summaries.join("; "),
reason: String::new(),
origin: effective
.as_ref()
.map(|p| p.source.clone())
.or_else(|| origin.map(str::to_string))
.unwrap_or_default(),
effective_policy: effective,
};
self.audit_first(&entry, AuditAction::Upload, bytes.len() as u64, source)?;
store().map_err(|e| IntakeError::Store(format!("{e:#}")))?;
// The package's index row (lane HOL-PERF-2), beside the evidence: what every index
// renders from once the artifact is AVAILABLE.
if let Some(row) = self.row_at_ingest(id, bytes) {
self.rows
.put(&entry.sha256, bytes.len() as u64, &row)
.map_err(|e| {
IntakeError::Store(format!(
"the index row of '{artifact}' could not be stored: {e}"
))
})?;
}
self.log
.append(id, &entry)
.map_err(|e| IntakeError::Store(e.to_string()))?;
Ok(entry)
}
/// The bytes in the incoming backend, re-checked against the ingested digest.
fn incoming_bytes(&self, id: &ArtifactId, want: &str) -> Result<Vec<u8>, IntakeError> {
let bytes = self
.incoming
.fetch(id)
.map_err(|e| IntakeError::Store(format!("{e:#}")))?
.ok_or_else(|| {
IntakeError::Store(format!(
"'{}' is no longer in '{}'",
super::artifact_label(id),
self.settings.incoming
))
})?;
let got = sha256_hex(&bytes);
if got != want {
return Err(IntakeError::Integrity {
want: want.to_string(),
got,
});
}
Ok(bytes.to_vec())
}
/// The bytes in the sealed store, re-checked against the ingested digest (once per
/// blob file, see [`open_sealed`](Self::open_sealed)).
fn sealed_bytes(&self, id: &ArtifactId, want: &str) -> Result<Vec<u8>, IntakeError> {
use std::io::Read as _;
let (mut file, len) = self.open_sealed(id, want)?;
let mut bytes = Vec::with_capacity(len as usize);
file.read_to_end(&mut bytes).map_err(|e| {
IntakeError::Store(format!(
"the sealed blob of '{}': {e}",
super::artifact_label(id)
))
})?;
Ok(bytes)
}
/// **The sealed blob of `id`, opened and verified once** (HOL-PERF-3). Its blake3 must
/// be the content address the seal's ref names and its sha256 `want`, the intake
/// record's digest. A blob file verified before, with the same identity, size and
/// times, is not hashed again; any change to it is. The file comes back at offset 0.
fn open_sealed(
&self,
id: &ArtifactId,
want: &str,
) -> Result<(std::fs::File, u64), IntakeError> {
use std::io::{Read as _, Seek as _};
let label = || super::artifact_label(id);
let (address, path) = self
.engine
.sealed_blob(&self.settings.production, id)
.map_err(|e| IntakeError::Store(e.tagged()))?
.ok_or_else(|| {
IntakeError::Store(format!("'{}' is not in the sealed store", label()))
})?;
let store = |e: std::io::Error| IntakeError::Store(PromoteError::Io(e).tagged());
let mut file = std::fs::File::open(&path).map_err(store)?;
let stamp = stamp_of(&file.metadata().map_err(store)?);
if self
.verified
.lock_or_recover()
.get(&address)
.is_some_and(|(s, w)| *s == stamp && w == want)
{
return Ok((file, stamp.2));
}
let mut b3 = blake3::Hasher::new();
let mut s2 = <sha2::Sha256 as sha2::Digest>::new();
let mut buf = vec![0u8; 1 << 20];
loop {
let n = file.read(&mut buf).map_err(store)?;
if n == 0 {
break;
}
b3.update(&buf[..n]);
sha2::Digest::update(&mut s2, &buf[..n]);
}
if b3.finalize().to_hex().as_str() != address {
return Err(IntakeError::Store(
PromoteError::Io(std::io::Error::other(
"integrity: promoted blob blake3 does not match the sealed ref",
))
.tagged(),
));
}
let got: String = sha2::Digest::finalize(s2)
.iter()
.map(|b| format!("{b:02x}"))
.collect();
if got != want {
return Err(IntakeError::Integrity {
want: want.to_string(),
got,
});
}
// Changed while it was read (written, truncated): what was hashed is not one state
// of the file. Its ctime alone may move (a link count, a chmod) without its bytes.
let after = stamp_of(&file.metadata().map_err(store)?);
if (after.0, after.1, after.2, after.3, after.4)
!= (stamp.0, stamp.1, stamp.2, stamp.3, stamp.4)
{
return Err(IntakeError::Store(format!(
"the sealed blob of '{}' changed while it was verified",
label()
)));
}
self.verifications
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
self.verified
.lock_or_recover()
.insert(address, (stamp, want.to_string()));
file.rewind().map_err(store)?;
Ok((file, stamp.2))
}
/// **A download, streamed from the sealed store** (HOL-PERF-3): the AVAILABLE
/// artifact's blob file in pieces of at most `chunk_bytes`, verified once (see
/// [`open_sealed`](Self::open_sealed)). `Ok(None)` for every other state, as
/// [`serve`](Self::serve).
pub fn serve_stream(
&self,
id: &ArtifactId,
chunk_bytes: usize,
) -> Result<Option<traits::ChunkIter>, IntakeError> {
match self.log.status(id) {
Some(s) if s.state == ArtifactState::Available => {
let (file, _) = self.open_sealed(id, &s.sha256)?;
Ok(Some(traits::backend::file_chunks(file, chunk_bytes)))
}
_ => Ok(None),
}
}
/// How many sealed blobs this chain hashed in full to serve them (each blob file
/// once, and again after it changed on disk).
pub fn sealed_verifications(&self) -> u64 {
self.verifications
.load(std::sync::atomic::Ordering::Relaxed)
}
/// **Move one artifact to one state.** Every check, the audit record, the
/// effect, then the record; any refusal leaves the state as it was.
///
/// A move to APPROVED is an approval vote by `req.actor` (see
/// [`approve`](Self::approve)); short of the quorum it is recorded and
/// answered with [`DutyError::QuorumPending`](super::duties::DutyError::QuorumPending).
pub fn advance(&self, req: TransitionRequest) -> Result<IntakeEntry, IntakeError> {
let approver = Approver::in_process(req.actor.clone());
match self.approve_or_advance(req, &approver)? {
ApproveOutcome::Approved { entry, .. } => Ok(entry),
ApproveOutcome::Pending { have, need, vote } => {
Err(IntakeError::Duty(super::duties::DutyError::QuorumPending {
artifact: vote.artifact,
have,
need,
}))
}
}
}
/// **Cast an approval vote, or make any other move.** The door's entry: a
/// move to APPROVED from POLICY_CHECKED is a vote by `approver` (whose
/// principal replaces `req.actor`); every other move is
/// [`advance`](Self::advance)'s, answered as `Approved` with the entry.
pub fn approve_or_advance(
&self,
mut req: TransitionRequest,
approver: &Approver,
) -> Result<ApproveOutcome, IntakeError> {
req.actor = approver.principal.clone();
let artifact = super::artifact_label(&req.id);
let out = {
let lock = self.lock_for(&req.id);
let _held = lock.lock_or_recover();
let status = self.log.status(&req.id);
let is_vote = req.to == ArtifactState::Approved
&& status
.as_ref()
.is_some_and(|s| s.state != ArtifactState::Quarantined);
if is_vote {
self.vote_locked(&req, approver, &artifact)
} else {
self.advance_inner(&req, &artifact)
.map(|entry| ApproveOutcome::Approved {
have: 0,
need: 0,
entry,
})
}
};
if let Err(e) = &out {
let now = self.status(&req.id);
let repo = now
.as_ref()
.and_then(|s| s.state.repository())
.map(|r| self.repo_name(r).to_string())
.unwrap_or_default();
let sha = now.map(|s| s.sha256).unwrap_or_default();
self.audit_refusal(
&req.actor,
&repo,
&artifact,
&req.source,
AuditAction::Promote,
e,
&sha,
);
}
out
}
/// The principal recorded on the artifact's latest INGESTED entry.
pub fn importer(&self, id: &ArtifactId) -> Option<String> {
Self::importer_in(&self.log.history(id))
}
fn importer_in(history: &[IntakeEntry]) -> Option<String> {
history
.iter()
.rev()
.find(|e| e.to == ArtifactState::Ingested)
.map(|e| e.actor.clone())
}
/// How many distinct approvers APPROVED needs.
pub fn approvals_required(&self) -> u32 {
self.settings.approvals.max(1)
}
/// The votes that count now: cast since the artifact's last transition,
/// over its current bytes.
pub fn pending_approvals(&self, id: &ArtifactId) -> Vec<ApprovalVote> {
let history = self.log.history(id);
let Some(last) = history.last() else {
return Vec::new();
};
self.log
.votes(id)
.into_iter()
.filter(|v| v.seq == history.len() && v.sha256 == last.sha256)
.collect()
}
/// One approval vote, under the lock. The rules of separation of duties
/// that need no identity provider are enforced here, for every caller.
fn vote_locked(
&self,
req: &TransitionRequest,
approver: &Approver,
artifact: &str,
) -> Result<ApproveOutcome, IntakeError> {
use super::duties::{DutyError, ANONYMOUS};
if let Some(e) = req
.evidence
.iter()
.find(|e| e.kind == EvidenceKind::Approval)
{
return Err(DutyError::ForgedApproval {
source: e.source.clone(),
}
.into());
}
let history = self.log.history(&req.id);
let status = self
.log
.status(&req.id)
.ok_or_else(|| IntakeError::NotIngested(artifact.to_string()))?;
check_transition(status.state, status.held_from, ArtifactState::Approved)?;
let principal = approver.principal.trim();
if principal.is_empty() || principal == ANONYMOUS {
return Err(DutyError::AnonymousApprover {
artifact: artifact.to_string(),
}
.into());
}
let importer = Self::importer_in(&history).unwrap_or_default();
if importer == principal {
return Err(DutyError::ImporterCannotApprove {
artifact: artifact.to_string(),
principal: principal.to_string(),
}
.into());
}
let need = self.approvals_required();
let votes: Vec<ApprovalVote> = self
.log
.votes(&req.id)
.into_iter()
.filter(|v| v.seq == history.len() && v.sha256 == status.sha256)
.collect();
if votes.iter().any(|v| v.approver == principal) {
return Err(DutyError::DuplicateApprover {
artifact: artifact.to_string(),
principal: principal.to_string(),
have: votes.len() as u32,
need,
}
.into());
}
let vote = ApprovalVote::new(
artifact,
&status.sha256,
approver,
&importer,
history.len(),
need,
&req.reason,
now_secs(),
);
let have = votes.len() as u32 + 1;
if have < need {
// Audit first, then the vote: a vote nobody can see is not cast.
let ev = AuditEvent::new(principal, AuditAction::Promote, &status.last.repo, artifact, &req.source, 202, 0)
.with_detail(format!(
"approval {have} of {need} for {artifact} by {principal} ({}); statement sha256 {}; sha256 {}",
approver.method, vote.statement_sha256, status.sha256
))
.on_trail(TrailKind::Approved, &status.sha256);
self.audit
.record(ev)
.map_err(|e| IntakeError::AuditFailed(e.to_string()))?;
self.log
.append_vote(&req.id, &vote)
.map_err(|e| IntakeError::Store(e.to_string()))?;
return Ok(ApproveOutcome::Pending { have, need, vote });
}
// The quorum: the APPROVED transition carries every vote's evidence slot.
let mut moved = req.clone();
moved.evidence.extend(
votes
.iter()
.chain(std::iter::once(&vote))
.map(|v| v.evidence()),
);
let entry = self.advance_inner(&moved, artifact)?;
if let Err(e) = self.log.append_vote(&req.id, &vote) {
log::warn!("intake: the deciding vote's record failed after APPROVED (its slot is in the entry): {e}");
}
Ok(ApproveOutcome::Approved { have, need, entry })
}
/// The move itself; the caller holds the lock.
fn advance_inner(
&self,
req: &TransitionRequest,
artifact: &str,
) -> Result<IntakeEntry, IntakeError> {
use ArtifactState::*;
let status = self
.log
.status(&req.id)
.ok_or_else(|| IntakeError::NotIngested(artifact.to_string()))?;
let from = status.state;
let to = req.to;
// An invalidation from the upstream is never released here: only a Block moves it.
if from == Quarantined
&& to != Blocked
&& status.last.reason.starts_with(INVALIDATED_PREFIX)
{
return Err(IntakeError::DoorRefused {
repo: status.last.repo.clone(),
why: format!("'{artifact}' is held by an invalidation from its upstream ({}); only a Block moves it on", status.last.reason),
});
}
check_transition(from, status.held_from, to)?;
// A release is a judgement like an approval: never the importer's own.
if from == Quarantined && to != Blocked {
let importer = self.importer(&req.id).unwrap_or_default();
if importer == req.actor {
return Err(super::duties::DutyError::ImporterCannotApprove {
artifact: artifact.to_string(),
principal: req.actor.clone(),
}
.into());
}
}
// SCANNED runs the configured scanners and fills the evidence slots from
// their evidence, beside whatever the caller handed in.
let mut evidence = req.evidence.clone();
// SCANNED reads the bytes first: the trail's `hash_verified` is that read
// matching the ingested digest, before any scanner sees them.
let mut prefetched: Option<Vec<u8>> = None;
let mut scan_blocks: Vec<String> = Vec::new();
// Tier 1: the scanners' records and the subject's ecosystem, for the
// artifact's effective policy after the scan.
let mut scan_records: Vec<crate::scanner::ScanEvidence> = Vec::new();
// HOL-DEPSET-APPROVE: a Maven POM's advisories warn, never refuse (no code in it).
let mut advisories_warn_only = false;
#[cfg_attr(
not(feature = "scanners"),
allow(unused_variables, unused_assignments, unused_mut)
)]
let mut scan_eco = String::new();
if to == Scanned {
let b = self.incoming_bytes(&req.id, &status.sha256)?;
self.audit_hash_verified(
&req.actor,
&status.last.repo,
artifact,
&status.sha256,
b.len() as u64,
&req.source,
"incoming repository",
)?;
prefetched = Some(b);
}
if to == Scanned {
if self.scanner.is_none() {
// No pipeline here: the caller's scanner slots are the scans, one
// trail entry each, with its verdict.
for e in evidence.iter().filter(|e| e.kind != EvidenceKind::Approval) {
let ev = AuditEvent::new(
&req.actor,
AuditAction::Scan,
&status.last.repo,
artifact,
&req.source,
200,
0,
)
.with_detail(format!(
"scanner={} kind={} verdict={:?} (evidence handed in)",
e.source,
e.kind.name(),
e.verdict
))
.on_trail(TrailKind::Scanned, &status.sha256);
self.audit
.record(ev)
.map_err(|e| IntakeError::AuditFailed(e.to_string()))?;
}
}
if let Some(pipeline) = &self.scanner {
let b = prefetched.clone().unwrap_or_default();
let file_name = req.id.name.rsplit('/').next().unwrap_or(&req.id.name);
let npm = super::npm::coords_of(&req.id);
let go = go_coords_of(&req.id);
let cargo = self.cargo_coords_of(&req.id);
let pulled = self.format_target(&req.id);
let ecosystem = if let Some((e, _)) = &pulled {
e
} else if npm.is_some() {
"npm"
} else if go.is_some() {
"go"
} else if cargo.is_some() {
"cargo"
} else {
crate::scanner::ecosystem_hint(file_name, &self.ecosystem)
};
// HOL-CARGO-PUBLISH-DOOR: a crate's pinned dependency versions, from the row the
// scan worker read at intake, are matched beside the crate itself.
let row = (ecosystem == "cargo")
.then(|| self.rows.get(&status.sha256))
.flatten();
let pinned: &[holger_index_rows::Pinned] = match row.as_deref().map(|r| &r.row) {
Some(holger_index_rows::IndexRow::Crate { pinned, .. }) => pinned,
_ => &[],
};
let mut subject =
crate::scanner::ScanSubject::new(file_name, ecosystem, &b).with_pinned(pinned);
// Tier 1: the source the artifact came from (its intake record), so the
// provenance verifier applies that source's gate.
if !status.last.origin.is_empty() {
subject = subject.with_source(&status.last.origin);
}
if !req.id.version.is_empty() {
subject = subject.with_coords(&req.id.name, &req.id.version);
} else if let Some((_, Some((package, version)))) = &pulled {
// A pulled npm tarball or Maven file: the source's coordinates (P11).
subject = subject.with_coords(package, version);
} else if let Some((package, version)) = &npm {
// A scoped package's tarball file name drops its scope; the scan names it whole.
subject = subject.with_coords(package, version);
} else if let Some((module, version)) = &go {
// A Go zip's file name is its version only (`v1.0.0.zip`): the module names it.
subject = subject.with_coords(module, version);
} else if let Some((name, version)) = &cargo {
// A published crate: its name and version, as the advisories name them.
subject = subject.with_coords(name, version);
}
// W5: the stored, signed evidence about these bytes is read back
// first; the scanners run only when it is missing or stale.
let scanned = pipeline
.scan_or_reuse(&req.actor, &self.settings.incoming, &subject)
.map_err(|e| IntakeError::Store(e.tagged()))?;
evidence.extend(crate::scanner::intake_slots(&scanned.evidence, &scanned.id));
scan_records = scanned.evidence.records.clone();
advisories_warn_only = crate::scanner::advisories_warn_only(ecosystem, file_name);
#[cfg(feature = "scanners")]
{
scan_eco = ecosystem.to_string();
}
// A block names its reason (and its code) in the refusal itself.
scan_blocks.extend(
scanned
.evidence
.records
.iter()
.filter(|r| r.verdict == crate::scanner::ScanVerdict::Block)
.map(|r| format!("{} ({}): {}", r.kind, r.scanner_id, r.detail)),
);
#[cfg(feature = "scanners")]
if let (Some(props), Some(model)) = (
&self.properties,
crate::scanner::model::metadata_of(&scanned.evidence),
) {
for repo in self.settings.repos() {
for (key, values) in model.properties() {
props.set(repo, &req.id, &key, values).map_err(|e| {
IntakeError::Store(format!("the model properties of {artifact} in '{repo}' could not be written: {e}"))
})?;
}
}
}
}
}
// Tier 1: the artifact's effective policy (the floor plus its source's gates),
// against what the scanners found. Each refusal names the rule and who set it.
if to == Scanned {
if let (Some(p), Some(b)) = (&status.last.effective_policy, prefetched.as_deref()) {
#[cfg(not(feature = "scanners"))]
let license = {
let _ = b;
None
};
#[cfg(feature = "scanners")]
let license = (!p.deny_licenses.is_empty()).then(|| {
let deny = p.deny_licenses.iter().map(|d| d.value.clone()).collect();
crate::scanner::license::LicenseScanner::new(
crate::scanner::license::LicensePolicy::new(
Vec::new(),
deny,
crate::scanner::license::UnknownLicense::Pass,
),
)
.judge(&scan_eco, b)
});
// A Maven POM's advisories are no refusal (`advisories_warn_only`): its
// vulnerability records stay on the evidence, and the rule `cve_max` skips them.
let judged: Vec<crate::scanner::ScanEvidence> = scan_records
.iter()
.filter(|r| {
!(advisories_warn_only
&& r.kind == crate::scanner::ScannerKind::Vulnerability)
})
.cloned()
.collect();
scan_blocks.extend(p.refusals(&judged, license.as_ref()));
}
}
// Evidence: required slots present, no slot failing.
let policy = self.policies.get(&to);
if let Some(p) = policy {
let missing: Vec<EvidenceKind> = p
.require_evidence
.iter()
.filter(|k| !evidence.iter().any(|e| e.kind == **k))
.copied()
.collect();
if !missing.is_empty() {
return Err(IntakeError::EvidenceMissing { to, missing });
}
}
let failing: Vec<String> = evidence
.iter()
.filter(|e| e.verdict == EvidenceVerdict::Fail)
.map(|e| {
// A required scanner that could not run says so (HOL-DOGFOOD-5): no verdict.
match e.reference.starts_with(crate::scanner::COULD_NOT_RUN) {
true => format!(
"{} evidence from '{}' failed ({})",
e.kind.name(),
e.source,
e.reference.split("; ").next().unwrap_or_default()
),
false => format!("{} evidence from '{}' failed", e.kind.name(), e.source),
}
})
.chain(scan_blocks)
.collect();
if !failing.is_empty() && to != Blocked {
return Err(IntakeError::PolicyRefused {
to,
reasons: failing,
});
}
let s = &self.settings;
let mut summaries: Vec<String> = Vec::new();
let mut reasons: Vec<String> = Vec::new();
let mut take = |d: nornir_gate::Decision, block: bool, reasons: &mut Vec<String>| {
summaries.push(d.summary());
if !d.is_allowed() && block {
reasons.extend(d.rejections.iter().map(|r| r.to_string()));
}
};
// The bytes this transition judges, re-checked against the digest.
// A hold needs none: holding an artifact must never depend on reading it.
let bytes: Option<Vec<u8>> = match to {
Blocked | Quarantined => None,
Scanned if prefetched.is_some() => prefetched.take(),
_ if from == Quarantined => Some(self.incoming_bytes(&req.id, &status.sha256)?),
Scanned | PolicyChecked | Approved | Sealed => {
Some(self.incoming_bytes(&req.id, &status.sha256)?)
}
Promoted | Available => Some(self.sealed_bytes(&req.id, &status.sha256)?),
Untrusted | Ingested => None,
};
if let (Available, Some(b)) = (to, bytes.as_deref()) {
self.audit_hash_verified(
&req.actor,
&status.last.repo,
artifact,
&status.sha256,
b.len() as u64,
&req.source,
"sealed store",
)?;
}
// The transition's own policy gates.
if let (Some(p), Some(b)) = (policy, bytes.as_deref()) {
if !p.gates.is_empty() {
let (fr, tr) = match to.repository() {
Some(r) => (status.last.repo.clone(), self.repo_name(r).to_string()),
None => (status.last.repo.clone(), status.last.repo.clone()),
};
let d = self.evaluate(&p.gates, &req.actor, &fr, &tr, &req.id, b);
take(
d,
p.gates.enforcement() == super::Enforcement::Block,
&mut reasons,
);
}
}
// The ladder rung's gates on a repository hop. A release from QUARANTINED
// re-enters a state it already passed and does not hop.
let hop = match to {
Scanned if from != Quarantined => Some((s.incoming.as_str(), s.quarantine.as_str())),
Approved if from != Quarantined => Some((s.quarantine.as_str(), s.approved.as_str())),
Sealed => Some((s.approved.as_str(), s.production.as_str())),
_ => None,
};
if let (Some((fr, tr)), Some(b)) = (hop, bytes.as_deref()) {
let set = [PromotedArtifact {
id: &req.id,
ecosystem: self.ecosystem_of(&req.id),
digest: crate::attest::ArtifactDigest::of_sha256(b),
bytes: b,
attestations: &[],
}];
match self.engine.gate_check_hop(fr, tr, &set, &req.actor) {
Ok(d) => take(d, false, &mut reasons),
Err(PromoteError::PolicyBlocked(r)) => reasons.extend(r),
Err(PromoteError::Ladder(l)) => reasons.push(l.tagged()),
Err(e) => return Err(IntakeError::Store(e.tagged())),
}
}
// The state's built-in check.
match to {
Promoted => {
let at = self.engine.current_stage(&req.id);
let want = self
.engine
.ladder()
.stage_of_repo(&s.production)
.map(|st| st.name.clone());
if at.is_none() || at != want {
reasons.push(format!(
"the ladder's lineage places the artifact at {:?}, not at the production \
rung {:?}",
at, want
));
} else {
summaries.push("the ladder's lineage stands at the production rung".into());
}
}
Available => {
summaries.push("the sealed bytes read back and match the ingested sha256".into())
}
_ => {}
}
if !reasons.is_empty() {
return Err(IntakeError::PolicyRefused { to, reasons });
}
let repo = match to.repository() {
Some(r) => self.repo_name(r).to_string(),
None => status.last.repo.clone(),
};
let entry = IntakeEntry {
from,
to,
artifact: artifact.to_string(),
sha256: status.sha256.clone(),
repo,
actor: req.actor.clone(),
at: now_secs(),
evidence,
policy: if summaries.is_empty() {
"structural check only".into()
} else {
summaries.join("; ")
},
reason: req.reason.clone(),
origin: status.last.origin.clone(),
effective_policy: status.last.effective_policy.clone(),
};
let nbytes = bytes.as_ref().map(|b| b.len() as u64).unwrap_or(0);
self.audit_first(&entry, AuditAction::Promote, nbytes, &req.source)?;
// The effect.
match (to, bytes.as_deref()) {
(Scanned, Some(_)) | (Approved, Some(_)) if from != Quarantined => {
let (fr, tr) = hop.expect("a hop for SCANNED and APPROVED");
self.engine
.record_hop(
fr,
tr,
&req.id,
&crate::digest::blake3_hex(bytes.as_deref().unwrap_or_default()),
&req.actor,
entry.at,
)
.map_err(|e| IntakeError::Store(e.tagged()))?;
}
(Sealed, Some(b)) => {
match self
.engine
.seal(&s.approved, &s.production, &req.id, b, &req.actor, entry.at)
{
Ok(_) => {}
// A retry after an audit or record failure: the seal took, and
// its bytes are these bytes, so the transition may complete.
Err(PromoteError::AlreadyPromoted) => {
self.sealed_bytes(&req.id, &status.sha256)?;
}
Err(e) => return Err(e.into()),
}
}
_ => {}
}
self.log
.append(&req.id, &entry)
.map_err(|e| IntakeError::Store(e.to_string()))?;
Ok(entry)
}
/// **Hold `id` at once because its upstream invalidated it** (Holger v2.0 H4).
///
/// From ANY state but QUARANTINED and BLOCKED, AVAILABLE included, the artifact
/// goes to QUARANTINED: quarantine, not delete. The sealed bytes stay where they
/// are, the record keeps the state it was held at, and from this entry on the
/// download gate ([`serve`](Self::serve)) and every index (they are views of the
/// AVAILABLE set) leave it out. No release brings it back
/// ([`INVALIDATED_PREFIX`]); a Security Approver may only Block it. `reason`
/// must start with [`INVALIDATED_PREFIX`]. The audit record is written first
/// (the transition, and the artifact's `invalidated` trail step); no record, no
/// move.
pub fn invalidate(
&self,
id: &ArtifactId,
actor: &str,
source: &str,
reason: &str,
) -> Result<InvalidateOutcome, IntakeError> {
let artifact = super::artifact_label(id);
if !reason.starts_with(INVALIDATED_PREFIX) {
return Err(IntakeError::Store(format!("an invalidation's reason starts with '{INVALIDATED_PREFIX}', and '{reason}' does not")));
}
let lock = self.lock_for(id);
let _held = lock.lock_or_recover();
let status = self
.log
.status(id)
.ok_or_else(|| IntakeError::NotIngested(artifact.clone()))?;
if matches!(
status.state,
ArtifactState::Quarantined | ArtifactState::Blocked
) {
return Ok(InvalidateOutcome::AlreadyHeld(status.state));
}
let entry = IntakeEntry {
from: status.state,
to: ArtifactState::Quarantined,
artifact: artifact.clone(),
sha256: status.sha256.clone(),
repo: status.last.repo.clone(),
actor: actor.to_string(),
at: now_secs(),
evidence: Vec::new(),
policy: "an invalidation in the upstream's signed sync: held at once, the bytes kept"
.into(),
reason: reason.to_string(),
origin: status.last.origin.clone(),
effective_policy: status.last.effective_policy.clone(),
};
self.audit_entry(&entry, AuditAction::Promote, 0, source, false)?;
let ev = AuditEvent::new(
actor,
AuditAction::Other,
&entry.repo,
&artifact,
source,
200,
0,
)
.with_detail(format!(
"{reason}; held QUARANTINED from {}; sha256 {}",
entry.from, entry.sha256
))
.on_trail(TrailKind::Invalidated, &entry.sha256);
self.audit
.record(ev)
.map_err(|e| IntakeError::AuditFailed(e.to_string()))?;
self.log
.append(id, &entry)
.map_err(|e| IntakeError::Store(e.to_string()))?;
Ok(InvalidateOutcome::Held(entry))
}
/// The invalidation holding `id`, when one does: its reason.
pub fn invalidation_of(&self, id: &ArtifactId) -> Option<String> {
let s = self.log.status(id)?;
(s.state == ArtifactState::Quarantined && s.last.reason.starts_with(INVALIDATED_PREFIX))
.then_some(s.last.reason)
}
/// **The download gate.** `Ok(Some(bytes))` only for an AVAILABLE artifact,
/// and the bytes come from the sealed store, re-checked against the ingested
/// digest. Every other state is `Ok(None)`: not downloadable.
pub fn serve(&self, id: &ArtifactId) -> Result<Option<Vec<u8>>, IntakeError> {
match self.log.status(id) {
Some(s) if s.state == ArtifactState::Available => {
self.sealed_bytes(id, &s.sha256).map(Some)
}
_ => Ok(None),
}
}
// ── the npm door (lane P5, [`super::npm`]) ──────────────────────────────
/// The repository that is this chain's npm door, when `promotion.intake.npm` names one.
pub fn npm_door(&self) -> Option<&str> {
self.settings.npm.as_deref()
}
/// The public base of the packument's tarball links (the npm repository's `ron_base_url`).
pub fn with_npm_base(mut self, base: Option<String>) -> Self {
self.npm_base = base
.map(|b| b.trim().trim_end_matches('/').to_string())
.filter(|b| !b.is_empty());
self
}
pub fn npm_base(&self) -> Option<&str> {
self.npm_base.as_deref()
}
/// **`npm publish`** (`PUT /<door>/<package>`): the document is read and its one tarball
/// checked ([`super::npm::parse_publish`]), then INGESTED into `incoming` like any upload
/// (the ingest gates, the inspector, audit first) at the tarball's registry path; the
/// publish's dist-tags are recorded, to take effect when the version is AVAILABLE.
pub fn npm_publish(
&self,
actor: &str,
source: &str,
url_package: &str,
body: &[u8],
) -> Result<(IntakeEntry, super::npm::Publish), IntakeError> {
let incoming = self.settings.incoming.clone();
let p = super::npm::parse_publish(url_package, body).map_err(|e| {
let e = IntakeError::Npm(e);
self.audit_refusal(
actor,
&incoming,
url_package,
source,
AuditAction::Upload,
&e,
&sha256_hex(body),
);
e
})?;
let id = super::npm::coordinate(&p.name, &p.version);
let entry = self.ingest(actor, source, &incoming, &id, &p.tarball, || {
self.incoming.put(&id, &p.tarball)
})?;
let at = now_secs();
for tag in &p.tags {
let rec = super::npm::TagRecord {
tag: tag.clone(),
version: Some(p.version.clone()),
actor: actor.to_string(),
at,
};
self.npm_tags.append(&p.name, &rec).map_err(|e| {
IntakeError::Store(format!(
"{}@{} is INGESTED; its dist-tag '{tag}' could not be recorded: {e}",
p.name, p.version
))
})?;
}
Ok((entry, p))
}
// ── the cargo door (lane HOL-CARGO-PUBLISH-DOOR, [`super::cargo`]) ──────
/// The repository that is this chain's cargo door, when `promotion.intake.cargo` names one.
pub fn cargo_door(&self) -> Option<&str> {
self.settings.cargo.as_deref()
}
/// The chain's bundles door, when it has one (lane HOL-BUNDLES-DOOR).
pub fn bundles_door(&self) -> Option<&str> {
self.settings.bundles.as_deref()
}
/// The bundles door's trust policy: its own, else the chain's import policy.
pub fn bundles_trust_policy(&self) -> Option<&str> {
self.settings
.bundles_trust_policy
.as_deref()
.or(self.settings.trust_policy.as_deref())
}
/// The public base the door's `config.json` names (the door's `ron_base_url`).
pub fn with_cargo_base(mut self, base: Option<String>) -> Self {
self.cargo_base = base
.map(|b| b.trim().trim_end_matches('/').to_string())
.filter(|b| !b.is_empty());
self
}
pub fn cargo_base(&self) -> Option<&str> {
self.cargo_base.as_deref()
}
/// **`cargo publish`** (`PUT /<door>/api/v1/crates/new`): the body is read
/// ([`super::cargo::parse_publish`]), the crate's own `Cargo.toml` checked against it (the
/// row the scan worker reads, [`super::cargo::check_row`]), then it is INGESTED into
/// `incoming` like any upload (the ingest gates, the inspector, audit first) at
/// `<door>/<name>/<name>-<version>.crate`.
pub fn cargo_publish(
&self,
actor: &str,
source: &str,
body: &[u8],
) -> Result<(IntakeEntry, super::cargo::Publish), IntakeError> {
let incoming = self.settings.incoming.clone();
let door = self.settings.cargo.clone().unwrap_or_default();
let refused = |what: &str, e: IntakeError, sha: &str| {
self.audit_refusal(actor, &incoming, what, source, AuditAction::Upload, &e, sha);
e
};
let p = super::cargo::parse_publish(body).map_err(|e| {
refused(
"api/v1/crates/new",
IntakeError::Cargo(e),
&sha256_hex(body),
)
})?;
let id = super::cargo::id(&door, &p.name, &p.version);
#[cfg(feature = "scanners")]
{
let row = self.row_at_ingest(&id, &p.crate_bytes);
super::cargo::check_row(&p, row.as_ref()).map_err(|e| {
refused(&id.name, IntakeError::Cargo(e), &sha256_hex(&p.crate_bytes))
})?;
}
let entry = self.ingest(actor, source, &incoming, &id, &p.crate_bytes, || {
self.incoming.put(&id, &p.crate_bytes)
})?;
Ok((entry, p))
}
/// **One crate's sparse index file** over its AVAILABLE versions: each line the sealed
/// crate's row (read by the scan worker at intake; the crate is never opened here),
/// `cksum` the sealed sha256, `yanked` the yank record's. `None` when nothing of it is
/// AVAILABLE.
pub fn cargo_index(&self, name: &str) -> Option<String> {
let door = self.settings.cargo.as_deref()?;
let name = name.to_ascii_lowercase();
let (_, files) = self.available_under(&format!("{door}/"));
let mut versions = Vec::new();
for (rel, id) in files {
if !rel.starts_with(&format!("{name}/")) {
continue;
}
let Some((n, version)) = super::cargo::coords_of(door, &id) else {
continue;
};
if n != name {
continue;
}
let Some(st) = self.status(&id) else {
continue;
};
match self.rows.get(&st.sha256).as_deref().map(|r| &r.row) {
Some(holger_index_rows::IndexRow::Crate { line, .. }) => {
versions.push((version, line.clone(), st.sha256.clone()))
}
_ => log::warn!(
"cargo door: {name} {version} has no index row; it is left out of its index"
),
}
}
if versions.is_empty() {
return None;
}
Some(super::cargo::index_file(
&mut versions,
&self.cargo_yanks.yanked(&name),
))
}
/// The coordinate a download below the cargo door names.
pub fn cargo_download_id(&self, name: &str, version: &str) -> Option<ArtifactId> {
Some(super::cargo::id(
self.settings.cargo.as_deref()?,
name,
version,
))
}
/// **`cargo yank`** (`yanked: true`) **and `cargo yank --undo`**: of an AVAILABLE version
/// only, audited first, then recorded.
pub fn cargo_yank(
&self,
actor: &str,
source: &str,
name: &str,
version: &str,
yanked: bool,
admin: bool,
) -> Result<(), IntakeError> {
use super::cargo::CargoError;
let door = self.settings.cargo.clone().unwrap_or_default();
let id = super::cargo::id(&door, name, version);
if self.state(&id) != ArtifactState::Available {
return Err(IntakeError::Cargo(CargoError::Yank {
status: 404,
why: format!(
"{name}@{version} is not AVAILABLE; only a version that passed the intake chain is yanked or unyanked"
),
}));
}
// SEC-REVIEW 2026-11-01 P12: the publisher is who INGESTED the version (its first
// record); a yank is the publisher's or an Admin's, and an Admin's yank is undone
// only by an Admin.
if !admin {
let publisher = self.history(&id).into_iter().next().map(|e| e.actor);
if publisher.as_deref() != Some(actor) {
return Err(IntakeError::Cargo(CargoError::Yank {
status: 403,
why: format!(
"{name}@{version} was published by {}; only its publisher or a Repository Admin yanks or unyanks it",
publisher.unwrap_or_else(|| "nobody on record".into())
),
}));
}
let last = self
.cargo_yanks
.records(name)
.into_iter()
.rev()
.find(|r| r.version == version);
if !yanked && last.is_some_and(|r| r.yanked && r.by_admin) {
return Err(IntakeError::Cargo(CargoError::Yank {
status: 403,
why: format!(
"{name}@{version} was yanked by a Repository Admin; only an Admin undoes it"
),
}));
}
}
let verb = if yanked { "yank" } else { "unyank" };
let ev = AuditEvent::new(actor, AuditAction::Other, &door, &id.name, source, 200, 0)
.with_detail(format!("cargo {verb} {name}@{version}"));
self.audit
.record(ev)
.map_err(|e| IntakeError::AuditFailed(e.to_string()))?;
let rec = super::cargo::YankRecord {
version: version.to_string(),
yanked,
actor: actor.to_string(),
at: now_secs(),
by_admin: admin,
};
self.cargo_yanks
.append(name, &rec)
.map_err(|e| IntakeError::Store(e.to_string()))
}
/// The AVAILABLE tarballs of `package`: (version, coordinate, status).
fn npm_available(&self, package: &str) -> Vec<(String, ArtifactId, IntakeStatus)> {
let mut out: Vec<(String, ArtifactId, IntakeStatus)> = self
.log
.catalog()
.npm_versions(package)
.into_iter()
.filter_map(|(id, st)| match super::npm::coords_of(&id) {
Some((p, v)) if p == package => Some((v, id, st)),
_ => None,
})
.collect();
out.sort_by(|a, b| super::npm::compare_versions(&a.0, &b.0));
out
}
/// The dist-tags of `package` in force over its AVAILABLE versions.
pub fn npm_dist_tags(&self, package: &str) -> BTreeMap<String, String> {
let available = self
.npm_available(package)
.into_iter()
.map(|(v, _, _)| v)
.collect();
super::npm::effective_tags(&self.npm_tags.records(package), &available)
}
/// **The packument** of `package` over its AVAILABLE versions, each manifest read from the
/// sealed tarball (once per sha256). `None` when nothing of it is AVAILABLE.
pub fn npm_packument(
&self,
package: &str,
tarball_base: &str,
) -> Result<Option<serde_json::Value>, IntakeError> {
let available = self.npm_available(package);
if available.is_empty() {
return Ok(None);
}
let mut versions = Vec::with_capacity(available.len());
for (version, _, st) in &available {
let cached = self.npm_meta.lock_or_recover().get(&st.sha256).cloned();
let meta = match cached {
Some(m) => m,
// The version's row (lane HOL-PERF-2): the manifest and integrity the scan
// worker read at intake; the sealed tarball is never opened for it.
None => match self
.rows
.get(&st.sha256)
.as_deref()
.and_then(super::npm::meta_of_row)
{
Some(m) => {
self.npm_meta
.lock_or_recover()
.insert(st.sha256.clone(), m.clone());
m
}
None => {
log::warn!("npm: {package}@{version} has no index row; it is left out of its packument");
continue;
}
},
};
versions.push((version.clone(), meta));
}
let set = available.iter().map(|(v, _, _)| v.clone()).collect();
let tags = super::npm::effective_tags(&self.npm_tags.records(package), &set);
Ok(Some(super::npm::packument(
package,
&versions,
&tags,
tarball_base,
)))
}
/// The coordinate an npm tarball route names, when it is a tarball of its package.
pub fn npm_tarball_id(route: &super::npm::Route) -> Option<ArtifactId> {
let super::npm::Route::Tarball { package, file } = route else {
return None;
};
let id = ArtifactId {
namespace: None,
name: format!("{package}/-/{file}"),
version: String::new(),
};
matches!(super::npm::coords_of(&id), Some((p, _)) if p == *package).then_some(id)
}
/// **A read below the npm door**: the packument, a tarball (the sealed bytes of an
/// AVAILABLE version) or the dist-tags; `Ok(None)` for anything not AVAILABLE or not npm's.
pub fn npm_read(
&self,
route: &super::npm::Route,
tarball_base: &str,
) -> Result<Option<(&'static str, traits::Bytes)>, IntakeError> {
use super::npm::Route;
match route {
Route::Packument(package) => {
// Rendered once per change (lane HOL-PERF-2, X5): a new AVAILABLE version
// moves the generation, a dist-tag the record's length.
let generation = self.generation();
let tags = self.npm_tags.records(package).len();
let key = (package.clone(), tarball_base.to_string());
if let Some((g, t, b)) = self.packuments.lock_or_recover().get(&key) {
if *g == generation && *t == tags {
return Ok(Some(("application/json", b.clone())));
}
}
let Some(doc) = self.npm_packument(package, tarball_base)? else {
return Ok(None);
};
let body = traits::Bytes::from(doc.to_string());
let mut cache = self.packuments.lock_or_recover();
// A bound on what request-derived bases can make it hold.
if cache.len() >= PAGE_CACHE_CAP {
cache.clear();
}
cache.insert(key, (generation, tags, body.clone()));
Ok(Some(("application/json", body)))
}
Route::Tarball { .. } => match Self::npm_tarball_id(route) {
Some(id) => Ok(self
.serve(&id)?
.map(|b| ("application/octet-stream", traits::Bytes::from(b)))),
None => Ok(None),
},
Route::DistTags(package) => {
let tags = self.npm_dist_tags(package);
Ok((!tags.is_empty()).then(|| {
(
"application/json",
traits::Bytes::from(serde_json::json!(tags).to_string()),
)
}))
}
_ => Ok(None),
}
}
/// **`npm dist-tag add` / `rm`**: `version: Some` points `tag` at an AVAILABLE version,
/// `None` removes it (`latest` never). Audited first; the tags in force after it.
pub fn npm_tag(
&self,
actor: &str,
source: &str,
package: &str,
tag: &str,
version: Option<&str>,
) -> Result<BTreeMap<String, String>, IntakeError> {
use super::npm::NpmError;
let refuse = |status: u16, why: String| IntakeError::Npm(NpmError::Tag { status, why });
if !super::npm::valid_tag(tag) {
return Err(refuse(400, format!("'{tag}' is not a dist-tag name")));
}
let available: BTreeSet<String> = self
.npm_available(package)
.into_iter()
.map(|(v, _, _)| v)
.collect();
if available.is_empty() {
return Err(refuse(404, format!("'{package}' has no AVAILABLE version")));
}
let current = super::npm::effective_tags(&self.npm_tags.records(package), &available);
let detail = match version {
Some(v) if !available.contains(v) => {
return Err(refuse(404, format!("{package}@{v} is not AVAILABLE; a tag names only a version that passed the intake chain")));
}
Some(v) => format!("npm dist-tag {tag} → {v}"),
None if tag == super::npm::LATEST => {
return Err(refuse(
400,
"`latest` is never removed; point it at another version".into(),
))
}
None if !current.contains_key(tag) => {
return Err(refuse(404, format!("'{package}' has no dist-tag '{tag}'")))
}
None => format!("npm dist-tag {tag} removed"),
};
let door = self.settings.npm.clone().unwrap_or_default();
let ev = AuditEvent::new(actor, AuditAction::Other, &door, package, source, 200, 0)
.with_detail(detail);
self.audit
.record(ev)
.map_err(|e| IntakeError::AuditFailed(e.to_string()))?;
let rec = super::npm::TagRecord {
tag: tag.to_string(),
version: version.map(str::to_string),
actor: actor.to_string(),
at: now_secs(),
};
self.npm_tags
.append(package, &rec)
.map_err(|e| IntakeError::Store(e.to_string()))?;
Ok(super::npm::effective_tags(
&self.npm_tags.records(package),
&available,
))
}
/// Every coordinate with an intake record and its status, in key order.
pub fn all(&self) -> Vec<(ArtifactId, IntakeStatus)> {
self.log
.all()
.into_iter()
.filter_map(|(key, st)| id_from_label(&st.last.artifact, &key).map(|id| (id, st)))
.collect()
}
/// Every AVAILABLE artifact: what production serves.
/// **A wheel's core metadata** (PEP 658/714): the `<name>-<version>.dist-info/METADATA`
/// of the AVAILABLE wheel at `id`, with its sha256, from the wheel's index row (lane
/// HOL-PERF-2: the scan worker read it at intake; the sealed wheel is never opened for
/// it). `None` when `id` is not an AVAILABLE wheel, or its row has no metadata.
pub fn core_metadata(&self, id: &ArtifactId) -> Option<(Arc<Vec<u8>>, String)> {
let base = id.name.rsplit('/').next().unwrap_or(&id.name);
if !base.ends_with(".whl") || self.state(id) != ArtifactState::Available {
return None;
}
let st = self.status(id)?;
if let Some(hit) = self.core_meta.lock_or_recover().get(&st.sha256) {
return hit.clone();
}
let row = self.rows.get(&st.sha256)?;
let found = match &row.row {
holger_index_rows::IndexRow::Wheel {
metadata: Some(m),
sha256: Some(h),
} => Some((Arc::new(m.clone()), h.clone())),
_ => None,
};
self.core_meta
.lock_or_recover()
.insert(st.sha256, found.clone());
found
}
/// **The index row of the AVAILABLE artifact `id`** (lane HOL-PERF-2), when one is
/// stored: what its format's index renders from.
pub fn index_row(&self, id: &ArtifactId) -> Option<Arc<holger_index_rows::StoredRow>> {
let st = self
.status(id)
.filter(|s| s.state == ArtifactState::Available)?;
self.rows.get(&st.sha256)
}
/// Every AVAILABLE artifact: what production serves.
pub fn available(&self) -> Vec<(ArtifactId, IntakeStatus)> {
self.log.catalog().available()
}
/// **The generation of the AVAILABLE set** (HOL-PERF-1): it changes whenever the set does.
pub fn generation(&self) -> u64 {
self.log.catalog().generation()
}
/// The generation the AVAILABLE files under `prefix` (`<name>/`) last changed at.
pub fn prefix_generation(&self, prefix: &str) -> u64 {
self.log.catalog().prefix_generation(prefix)
}
/// The AVAILABLE files under `prefix` (`<name>/`, raw coordinates only), the prefix
/// stripped, with the generation the set is at: what a chain view serves.
pub fn available_under(&self, prefix: &str) -> (u64, Vec<(String, ArtifactId)>) {
self.log.catalog().available_under(prefix)
}
/// **The PEP 503 simple index over the AVAILABLE set** of the production
/// repository. `sub` is the path below the repository (`simple/`,
/// `simple/<project>/`). `None` for any other path.
///
/// Every link carries `#sha256=` from the intake record — the digest the
/// artifact was ingested with and re-checked against at every state — so
/// `pip` verifies the wheel it downloads. Links point at the repository's
/// own coordinate path, which the read door serves from the sealed store.
pub fn simple_index(&self, sub: &str) -> Option<(&'static str, traits::Bytes)> {
self.simple_index_as(sub, false)
}
/// [`simple_index`](Self::simple_index) in either form: `json` gives the
/// PEP 691 JSON document (`application/vnd.pypi.simple.v1+json`, W1) over the
/// same AVAILABLE set, the same links and the same sha256 — what pip asks for
/// first ([`wants_pep691`]). `simple/<project>/json` asks for it explicitly.
///
/// Each page is rendered once per change of the AVAILABLE set and handed out as the
/// same bytes (lane HOL-PERF-2, X5).
pub fn simple_index_as(&self, sub: &str, json: bool) -> Option<(&'static str, traits::Bytes)> {
let key = simple_page_key(sub, json)?;
let generation = self.generation();
if let Some((g, c, b)) = self.pages.lock_or_recover().get(&key) {
if *g == generation {
return Some((c, b.clone()));
}
}
let (ctype, body) = self.render_simple_index(sub, json)?;
let body = traits::Bytes::from(body);
let mut pages = self.pages.lock_or_recover();
if pages.len() >= PAGE_CACHE_CAP {
pages.clear();
}
pages.insert(key, (generation, ctype, body.clone()));
Some((ctype, body))
}
/// The simple index page `sub`, rendered over the catalog.
fn render_simple_index(&self, sub: &str, json: bool) -> Option<(&'static str, Vec<u8>)> {
let sub = sub.split('?').next().unwrap_or(sub);
let (sub, json) = match sub.trim_end_matches('/').strip_suffix("/json") {
Some(s) if s.trim_start_matches('/').starts_with("simple/") => (s, true),
_ => (sub, json),
};
let sub = sub.split('?').next().unwrap_or(sub).trim_start_matches('/');
let rest = sub.strip_prefix("simple")?;
if !(rest.is_empty() || rest.starts_with('/')) {
return None;
}
let project = rest.trim_matches('/');
if project.contains('/') {
return None;
}
// The catalog's pip index (HOL-PERF-1): the root page needs the project names, a
// project page that project's files; nothing scans the AVAILABLE set.
let catalog = self.log.catalog();
let files: Vec<(String, String, String)> = if project.is_empty() {
catalog
.pip_projects()
.into_iter()
.map(|p| (p, String::new(), String::new()))
.collect()
} else {
let want = normalize_project(project);
catalog
.pip_files(&want)
.into_iter()
.map(|(path, sha)| (want.clone(), path, sha))
.collect()
};
if json {
let meta = serde_json::json!({ "api-version": "1.0" });
let doc = if project.is_empty() {
let projects: std::collections::BTreeSet<&str> =
files.iter().map(|(p, _, _)| p.as_str()).collect();
let list: Vec<serde_json::Value> = projects
.into_iter()
.map(|p| serde_json::json!({ "name": p }))
.collect();
serde_json::json!({ "meta": meta, "projects": list })
} else {
let want = normalize_project(project);
let mine: Vec<serde_json::Value> = files
.iter()
.filter(|(p, _, _)| *p == want)
.map(|(_, path, sha)| {
let base = path.rsplit('/').next().unwrap_or(path);
let mut f = serde_json::json!({ "filename": base, "url": format!("../../{path}"), "hashes": { "sha256": sha } });
// PEP 658/714: the wheel's core metadata, served at `<url>.metadata`.
if let Some((_, msha)) = self.core_metadata(&ArtifactId { namespace: None, name: path.clone(), version: String::new() }) {
f["core-metadata"] = serde_json::json!({ "sha256": msha });
f["dist-info-metadata"] = serde_json::json!({ "sha256": msha });
}
f
})
.collect();
if mine.is_empty() {
return None;
}
serde_json::json!({ "meta": meta, "name": want, "files": mine })
};
return Some((PEP691_JSON, doc.to_string().into_bytes()));
}
let mut body = String::from(
"<!DOCTYPE html>\n<html><head><meta name=\"pypi:repository-version\" content=\"1.0\"><title>",
);
if project.is_empty() {
let projects: std::collections::BTreeSet<&str> =
files.iter().map(|(p, _, _)| p.as_str()).collect();
body.push_str("Simple index</title></head><body>\n");
for p in projects {
let p = html_escape(p);
body.push_str(&format!("<a href=\"{p}/\">{p}</a><br/>\n"));
}
} else {
let want = normalize_project(project);
let mine: Vec<&(String, String, String)> =
files.iter().filter(|(p, _, _)| *p == want).collect();
if mine.is_empty() {
return None;
}
body.push_str(&format!(
"Links for {}</title></head><body>\n<h1>Links for {}</h1>\n",
html_escape(&want),
html_escape(&want)
));
for (_, path, sha) in mine {
let base = path.rsplit('/').next().unwrap_or(path);
// PEP 658/714: the wheel's core metadata, served at `<href>.metadata`.
let meta = self
.core_metadata(&ArtifactId {
namespace: None,
name: path.clone(),
version: String::new(),
})
.map(|(_, m)| {
let m = html_escape(&format!("sha256={m}"));
format!(" data-core-metadata=\"{m}\" data-dist-info-metadata=\"{m}\"")
})
.unwrap_or_default();
body.push_str(&format!(
"<a href=\"../../{}#sha256={}\"{meta}>{}</a><br/>\n",
html_escape(path),
html_escape(sha),
html_escape(base)
));
}
}
body.push_str("</body></html>\n");
Some(("text/html", body.into_bytes()))
}
/// **Land an artifact a verified airgap bundle carried as AVAILABLE** in this
/// side's production repository (`super::airgap`).
///
/// The caller verified the bundle's release signature, the artifact's hashes
/// and its evidence attestations. This writes the whole chain for it in one
/// step: an entry per state from INGESTED to AVAILABLE, each carrying the
/// online side's evidence for that state and naming the bundle, so the record
/// still reads one state at a time. Every entry's audit record is written
/// FIRST; if one fails, nothing lands (fail closed). Then the ladder's hops,
/// the seal into the sealed store, the read-back, and the record.
///
/// A coordinate already AVAILABLE with the same bytes is an idempotent
/// re-import. Any other existing record refuses.
pub fn land_imported(&self, imp: &ImportedArtifact<'_>) -> Result<LandOutcome, IntakeError> {
use ArtifactState::*;
let artifact = super::artifact_label(&imp.id);
let lock = self.lock_for(&imp.id);
let _held = lock.lock_or_recover();
let sha256 = sha256_hex(imp.bytes);
if let Some(last) = imp.online_history.last() {
if last.sha256 != sha256 {
return Err(IntakeError::Integrity {
want: last.sha256.clone(),
got: sha256,
});
}
}
match self.log.status(&imp.id) {
None => {}
Some(s) if s.state == Available && s.sha256 == sha256 => {
return Ok(LandOutcome::AlreadyAvailable)
}
Some(s) if s.state.is_sealed() => return Err(IntakeError::Immutable(artifact)),
Some(s) => {
return Err(IntakeError::DoorRefused {
repo: self.settings.production.clone(),
why: format!(
"'{artifact}' already has an intake record here (state {}); an airgap import \
lands only a coordinate this side has never seen",
s.state
),
})
}
}
// G1: THIS side's own malware engines (its YARA rules, its ClamAV with the
// signature database carried over the airgap) re-scan the bytes before a
// byte or an audit record is written. Only the malware hook weighs here:
// the online side already ran the file inspection, and its evidence is
// verified. A side that declares no engine lands exactly as before.
#[cfg(feature = "scanners")]
let findings = self.inspect(&imp.id, imp.bytes);
#[cfg(feature = "scanners")]
let refused: Vec<String> = findings
.blocking()
.filter(|f| {
matches!(
f.kind,
holger_inspect::FindingKind::Malware
| holger_inspect::FindingKind::ScannerNotConfigured
The file goes on; the console shows all of it.