server/lib/src/pull_intake/mod.rs
//! # Pull-through through the intake chain
//!
//! **Owner order 2026-10-23: a package Holger FETCHES from an upstream goes
//! through the same intake chain as an upload.** Before this module only an
//! upload into `incoming` was ingested; the proxy, the queued pull-through,
//! the crates.io upstream and the mirror cycle served or copied bytes that no
//! scanner had seen.
//!
//! A *remote source* (`pull_through.sources`, one per upstream; the kinds G0's
//! `sources:` names `python`, `oci`, `npm`, `maven`, `crates` and `apk` with a `url`) is a
//! door into the chain ([`ecosystems`] for npm, Maven, cargo and APK):
//!
//! ```text
//! client asks ──▶ AVAILABLE? ── yes ─▶ served from the SEALED store
//! │ no
//! ▼
//! one jera job per want (Job::PullIntake, deduplicated)
//! fetch from the source (every byte held against the digest the
//! source publishes) + the signatures and attestations that came with it
//! ▼
//! IntakeChain::ingest into `incoming` as `pull-through:<source>`
//! (inspector, malware engines, the INGEST gates)
//! ▼
//! SCANNED (the scanners + the provenance verifier over the material)
//! POLICY_CHECKED
//! ▼
//! auto-approval? rules.allow_auto_approve AND the source's
//! auto_approve: when(...) holds ─▶ one vote by `auto-approve:<source>`
//! (never for an OCI source: a container image always waits for people)
//! otherwise it waits for people (two where `approvals: 2`)
//! ▼
//! APPROVED ─▶ SEALED ─▶ PROMOTED ─▶ AVAILABLE (walked by the pull-through)
//! ```
//!
//! Until then the client gets its ecosystem's "not yet": pip a `503` with
//! `Retry-After` while the job runs (pip retries 503) and a `404` naming the
//! state once it waits for a person; a container client the registry error
//! `UNAVAILABLE` (503, `Retry-After`; podman's `--retry` retries it) and then
//! `MANIFEST_UNKNOWN` (404); npm, Maven and cargo as [`door`]'s table says. The
//! index (the PEP 503 page, the tag, the packument, `maven-metadata.xml`, the
//! sparse index file) names only AVAILABLE items.
//!
//! The RON shape is G0+G2's vocabulary (`sources`, `rules`, `gates`,
//! `auto_approve: off | when(provenance: require, cve_max: high)`, the floor's
//! `allow_auto_approve: false`), kept here under `pull_through:` until
//! `crate::sources` carries the remote kinds; the swap is a rename.
pub mod apk;
pub mod crates;
pub mod door;
mod ecosystems;
pub mod go;
pub use ecosystems::maven_ctype;
mod closures;
pub mod depset;
pub mod http;
pub mod maven;
pub mod npm;
pub mod oci;
pub mod pip;
pub mod pom;
pub mod manifest_job;
pub mod provenance;
use crate::lock::{LockOrRecover as _, RwLockOrRecover as _};
use std::collections::BTreeMap;
use std::path::PathBuf;
use std::sync::{Arc, Mutex, OnceLock};
use serde::{Deserialize, Serialize};
use traits::ArtifactId;
use crate::promote::intake::{
ApproveOutcome, ArtifactState, Evidence, EvidenceKind, EvidenceVerdict, IntakeChain,
IntakeError, TransitionRequest,
};
use crate::scanner::FindingSeverity;
use provenance::{Demand, Material, Mode, Verdict, Verifier};
// ── configuration (G0+G2's vocabulary) ──────────────────────────────────────
/// The `pull_through:` block.
///
/// Written here with G0's implicit `Some` (`crate::sources::ron_options`);
/// the strict top-level reader wants `Some(...)` around each optional value.
///
/// ```ron
/// pull_through: Some((
/// sources: [
/// (name: "cgr", kind: oci, url: "https://cgr.dev", platforms: ["linux/amd64"],
/// gates: (provenance: require(identity: (issuer: "https://token.actions.githubusercontent.com",
/// subject: "https://github.com/chainguard-images/images/.github/workflows/release.yaml@refs/heads/main")))),
/// (name: "pypi", kind: python, url: "https://pypi.org/simple/",
/// auto_approve: when(provenance: require, cve_max: high)),
/// (name: "cg-python", kind: python, url: "https://libraries.cgr.dev/python/simple/",
/// credential: (username: "<identity id>", password_env: "CHAINGUARD_PYTHON_TOKEN")),
/// ],
/// rules: (require_signed: true, cve_max: high, allow_auto_approve: true),
/// )),
/// ```
#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct PullThroughSettings {
#[serde(default)]
pub sources: Vec<RemoteSource>,
/// The floor every remote source stands on.
#[serde(default)]
pub rules: PullRules,
/// `Retry-After` while a pull is in flight. Default 2 s.
#[serde(default)]
pub retry_after_secs: Option<u64>,
}
/// What a remote source speaks.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum RemoteKind {
/// A PEP 503 / PEP 691 simple index (PyPI, Chainguard Libraries for Python).
Python,
/// An OCI distribution registry (`cgr.dev`).
Oci,
/// An npm registry (registry.npmjs.org, Chainguard Libraries for JavaScript).
Npm,
/// A Maven 2 repository (Maven Central, Chainguard Libraries for Java).
Maven,
/// A cargo sparse index (index.crates.io).
Crates,
/// An APK repository (Wolfi, Chainguard, Alpine): a signed `APKINDEX.tar.gz` per arch.
Apk,
/// Go modules (the GOPROXY protocol: `@v/list`, `.info`, `.mod`, `.zip`, `@latest`).
/// With a `url` (a GOPROXY, proxy.golang.org) an asked-for version is pulled (H6,
/// [`go`]); without one the source serves what arrived.
Go,
}
impl RemoteKind {
pub fn name(self) -> &'static str {
match self {
RemoteKind::Python => "python",
RemoteKind::Oci => "oci",
RemoteKind::Npm => "npm",
RemoteKind::Maven => "maven",
RemoteKind::Crates => "crates",
RemoteKind::Apk => "apk",
RemoteKind::Go => "go",
}
}
}
/// A basic credential. The password is read from the environment or a file at
/// boot and never written anywhere.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct RemoteCredential {
pub username: String,
#[serde(default)]
pub password_env: Option<String>,
#[serde(default)]
pub password_file: Option<String>,
}
/// A source's `gates.provenance` (G3a / G0 `ProvenanceGate`).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case", deny_unknown_fields)]
pub enum ProvenanceGate {
Require {
#[serde(default)]
identity: Option<provenance::Identity>,
#[serde(default)]
identities: Vec<provenance::Identity>,
/// The OpenPGP keys (names the trust bundle pins) that may sign this
/// source's `.asc` files (Maven Central). Empty = any pinned pgp key.
#[serde(default)]
pgp_keys: Vec<String>,
#[serde(default)]
slsa_min: Option<u8>,
/// `kind: apk`: the pinned apk keys (by name) that may sign this source's
/// index. Empty = any pinned apk key.
#[serde(default)]
apk_keys: Vec<String>,
},
Prefer,
Off,
}
/// `cve: (max: high)`.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct CveGate {
pub max: FindingSeverity,
}
/// A source's `gates:`: they tighten the floor, never loosen it.
#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct RemoteGates {
#[serde(default)]
pub provenance: Option<ProvenanceGate>,
#[serde(default)]
pub cve: Option<CveGate>,
}
/// **`auto_approve: off | when(provenance: require, cve_max: high)`.** Off by
/// default. It takes effect only when the floor's `allow_auto_approve` is true
/// too, and it is ONE approver: under `approvals: 2` a person still casts the
/// second vote.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
#[serde(rename_all = "snake_case", deny_unknown_fields)]
pub enum AutoApprove {
#[default]
Off,
When {
/// `require`: a provenance verdict of `pass` from `holger-provenance`
/// is on the SCANNED record. `off`: provenance is not a condition.
provenance: AutoNeed,
/// The highest advisory severity the vulnerability scan may report.
/// No vulnerability scan on record ⇒ the condition does not hold.
cve_max: FindingSeverity,
},
}
/// `require` | `off`.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum AutoNeed {
Require,
Off,
}
fn default_cve_max() -> FindingSeverity {
FindingSeverity::High
}
/// **The floor** (G0's `rules:` fields that a pull-through reads).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct PullRules {
/// Nothing unsigned comes in.
#[serde(default)]
pub require_signed: bool,
/// Nothing without verified provenance comes in.
#[serde(default)]
pub require_provenance: bool,
#[serde(default)]
pub slsa_min: u8,
/// The signer identities every source may be signed by.
#[serde(default)]
pub identities: Vec<provenance::Identity>,
/// The highest advisory severity allowed. Default `high`.
#[serde(default = "default_cve_max")]
pub cve_max: FindingSeverity,
/// **Whether any source may auto-approve.** Default false: every pulled
/// package waits for people.
#[serde(default)]
pub allow_auto_approve: bool,
/// The signing keys every source's material may be signed with (`rules.keys`):
/// npm's registry key verifies `dist.signatures` and npm's publish attestation.
#[serde(default)]
pub keys: Vec<holger_provenance::TrustedKey>,
/// The SLSA builders trusted at all (`rules.builders`). Empty = any builder.
#[serde(default)]
pub builders: Vec<holger_provenance::BuilderRule>,
/// The signed Sigstore trust bundle (`holger.sigstore-trust/v1`) keyless
/// signatures verify against (lane G3a), checked with
/// `promotion.intake.trust_policy`. Absent ⇒ a keyless signature is refused.
#[serde(default)]
pub trust_root: Option<String>,
/// **The apk keys an APK source's index may be signed with** (`rules.apk_keys`),
/// beside the ones the trust bundle pins: the online side's config names them as
/// it names npm's registry key in `keys`. Each is an RSA `PUBLIC KEY` PEM, inline
/// (`pem`) or in a file (`pem_file`), under the name its `.SIGN.*.<name>` writes.
#[serde(default)]
pub apk_keys: Vec<PinnedApkKey>,
}
/// One `rules.apk_keys` entry.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct PinnedApkKey {
/// `wolfi-signing.rsa.pub`.
pub name: String,
#[serde(default)]
pub pem: Option<String>,
#[serde(default)]
pub pem_file: Option<String>,
}
impl PullRules {
/// The `rules.apk_keys`, read: `(name, PEM)`, each checked to be an RSA public key.
pub fn read_apk_keys(&self) -> anyhow::Result<Vec<(String, Vec<u8>)>> {
let mut out = Vec::new();
for k in &self.apk_keys {
let pem = match (&k.pem, &k.pem_file) {
(Some(p), None) => p.clone().into_bytes(),
(None, Some(f)) => std::fs::read(f).map_err(|e| anyhow::anyhow!("config `pull_through.rules.apk_keys` {:?}: `pem_file` {f}: {e}", k.name))?,
_ => anyhow::bail!("config `pull_through.rules.apk_keys` {:?}: give exactly one of `pem` or `pem_file`", k.name),
};
if !holger_provenance::trust::apk_key_name_ok(&k.name) {
anyhow::bail!(
"config `pull_through.rules.apk_keys`: {:?} is not a key file name",
k.name
);
}
holger_provenance::apk::ApkKey::from_pem(&k.name, &pem).map_err(|e| {
anyhow::anyhow!("config `pull_through.rules.apk_keys`: {}", e.tagged())
})?;
out.push((k.name.clone(), pem));
}
Ok(out)
}
}
impl Default for PullRules {
fn default() -> Self {
Self {
require_signed: false,
require_provenance: false,
slsa_min: 0,
identities: Vec::new(),
cve_max: default_cve_max(),
allow_auto_approve: false,
keys: Vec::new(),
builders: Vec::new(),
trust_root: None,
apk_keys: Vec::new(),
}
}
}
/// One remote source.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct RemoteSource {
/// The name: the first path segment the source's packages are served under
/// (`/production/<name>/<file>`, `<holger>/production/<name>/<image>`), and
/// the actor `pull-through:<name>`.
pub name: String,
pub kind: RemoteKind,
/// `https://pypi.org/simple/`, `https://cgr.dev`. **Empty or absent: the source
/// SERVES ONLY** (a downstream's view, H3): the AVAILABLE set that arrived under its
/// name (a signed bundle, a pull from the upstream Holger) is served with this kind's
/// native index, and a miss is never fetched (`holger-downstream-4`).
#[serde(default)]
pub url: String,
#[serde(default)]
pub credential: Option<RemoteCredential>,
#[serde(default)]
pub gates: RemoteGates,
#[serde(default)]
pub auto_approve: AutoApprove,
/// `python`: which files of the newest release are pulled for a project
/// page (`sdist`, or a substring of the file name). Default: the sdist,
/// pure wheels, manylinux and musllinux wheels.
#[serde(default)]
pub select: Vec<String>,
/// `python`: at most this many files per release. Default 8.
#[serde(default)]
pub max_files: Option<usize>,
/// `oci`: the platforms kept from an image index (`linux/amd64`). Empty = all.
#[serde(default)]
pub platforms: Vec<String>,
/// The cap on one package (an image whole). Default 4 GiB.
#[serde(default)]
pub max_bytes: Option<u64>,
/// How old a project page's or tag's last pull may be before a request
/// asks the source again. Default 6 h.
#[serde(default)]
pub refresh_secs: Option<u64>,
/// How long a request waits for its pull before it is answered "not yet".
/// Default 0, and 30 s for `maven`: Maven before 3.9.5 does not retry a `503` (3.9.5 and
/// later retry it three times, honouring `Retry-After`; measured, HOL-DOCFIND i), so its
/// request is held (bounded) until the pull has walked as far as policy lets it.
#[serde(default)]
pub wait_secs: Option<u64>,
}
impl RemoteSource {
/// No `url`: the source serves what arrived under its name and fetches nothing.
pub fn serves_only(&self) -> bool {
self.url.trim().is_empty()
}
}
// ── what a request wants, and what it is told ──────────────────────────────
/// What one pull job fetches. Serialized into the jera row.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
#[serde(tag = "want", rename_all = "snake_case")]
pub enum Want {
/// The newest release of a project (a project page was asked for).
PipProject { project: String },
/// One distribution file (a file link was asked for).
PipFile { file: String },
/// An image by tag or digest.
Oci { name: String, reference: String },
/// The version an npm package's `latest` dist-tag names (its packument was asked for).
NpmPackage { package: String },
/// One npm version (its tarball was asked for).
NpmVersion { package: String, version: String },
/// A Maven version's POM and JAR (`group` is the path, `org/hamcrest`).
MavenVersion {
group: String,
artifact: String,
version: String,
},
/// One other file of a Maven version (a classifier, another extension).
MavenFile { path: String },
/// The newest release `maven-metadata.xml` names.
MavenLatest { group: String, artifact: String },
/// The newest release of a crate (its index file was asked for).
CrateName { name: String },
/// One crate version (its download was asked for).
CrateVersion { name: String, version: String },
/// The newest APK package of a name and its closure (`<arch>/-/pull/<name>`).
ApkName { arch: String, name: String },
/// One APK file (`<arch>/<name>-<version>.apk` was asked for).
ApkFile { arch: String, file: String },
/// One Go module version (its `.info`, `.mod` or `.zip` was asked for; `module` escaped).
GoVersion { module: String, version: String },
}
impl Want {
/// The key a request record and the job's content id use.
pub fn key(&self) -> String {
match self {
Want::PipProject { project } => format!("project:{project}"),
Want::PipFile { file } => format!("file:{file}"),
Want::Oci { name, reference } => format!("oci:{name}:{reference}"),
Want::NpmPackage { package } => format!("npm:{package}"),
Want::NpmVersion { package, version } => format!("npm:{package}@{version}"),
Want::MavenVersion {
group,
artifact,
version,
} => format!("maven:{group}:{artifact}:{version}"),
Want::MavenFile { path } => format!("maven-file:{path}"),
Want::MavenLatest { group, artifact } => format!("maven-latest:{group}:{artifact}"),
Want::CrateName { name } => format!("crate:{name}"),
Want::CrateVersion { name, version } => format!("crate:{name}@{version}"),
Want::ApkName { arch, name } => format!("apk:{arch}:{name}"),
Want::ApkFile { arch, file } => format!("apk-file:{arch}/{file}"),
Want::GoVersion { module, version } => format!("go:{module}@{version}"),
}
}
}
/// **What a request is told** while a package is not AVAILABLE.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Ask {
/// It is AVAILABLE now: read the chain again.
Ready,
/// A pull is in flight.
Fetching { retry_after_secs: u64, job: String },
/// It is in the chain and waits for a person (or a hold).
Held {
state: ArtifactState,
detail: String,
},
/// The chain refused it; it is never served.
Refused { why: String },
/// The source does not publish it, or the pull failed for good.
Unknown { why: String },
}
impl Ask {
fn rank(&self) -> u8 {
match self {
Ask::Ready => 0,
Ask::Fetching { .. } => 1,
Ask::Held { .. } => 2,
Ask::Refused { .. } => 3,
Ask::Unknown { .. } => 4,
}
}
/// The better of two answers across sources.
pub fn best(self, other: Ask) -> Ask {
if other.rank() < self.rank() {
other
} else {
self
}
}
}
/// One line of the request record.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
struct RequestRecord {
source: String,
want: String,
at: i64,
/// `fetched` | `unknown` | `failed` | `refused` | `scan-failed` | `store-failed`.
outcome: String,
/// The coordinates the pull brought into the chain (their labels' names).
#[serde(default)]
names: Vec<String>,
#[serde(default)]
why: String,
/// A `store-failed` pull: what to pull again after the retry interval.
#[serde(default, skip_serializing_if = "Option::is_none")]
retry: Option<Want>,
}
/// **What an upstream index listed for one project** at its last pull (lane HOL-FIELD-TEST
/// P2): every final, unyanked file with the sha256 the index published. The project page
/// lists those that are not AVAILABLE here as yanked, so a pinned install can ask for one.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
struct PipListing {
source: String,
project: String,
at: i64,
files: Vec<(String, String)>,
}
/// One file an upstream lists that is not AVAILABLE here: the project page names it yanked
/// (PEP 592) with `reason`, its link the source's file door (`/production/<source>/<file>`).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct HeldFile {
pub source: String,
pub filename: String,
pub sha256: String,
pub reason: String,
}
/// One OCI tag as the source answered it.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
struct TagRecord {
source: String,
name: String,
tag: String,
digest: String,
at: i64,
}
/// The answer to a miss on a serve-only source (`holger-downstream-4`): its upstream has
/// not sent it, and a downstream never fetches it itself.
fn not_sent(source: &str, want: &Want) -> String {
holger_errcode::DOWNSTREAM_NOT_SENT.tag(format_args!(
"{} is not on this Holger: source '{source}' serves what its upstream Holger sent, and the upstream has not sent it",
want.key()
))
}
/// **A pull the scanners could not judge** (lane HOL-FIELD-TEST P8): the sandbox refused the
/// scan, the worker is not the server's build, clamd did not answer. No verdict: nothing is
/// stored, nothing is served, and the pull is asked again ([`SCAN_FAILED`]), never recorded as
/// refused for good.
#[derive(Debug)]
pub struct ScanFailed(pub String);
impl std::fmt::Display for ScanFailed {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(&self.0)
}
}
impl std::error::Error for ScanFailed {}
/// The request outcome of a [`ScanFailed`] pull.
const SCAN_FAILED: &str = "scan-failed";
/// **A pull the storage could not take** (lane HOL-ENOSPC-SETS, field row
/// HOL-PULL-ENOSPC-BLOCK): the incoming store, the scan-evidence store, the provenance
/// material or the audit trail could not be written (a full disk, `EIO`, a store that cannot
/// be written). No verdict: nothing is served, the artifact stays where it stood, and the pull
/// is asked again after the retry interval ([`STORE_FAILED`]), never refused for good.
#[derive(Debug)]
pub struct StoreFailed {
/// The coordinate the storage failed for (its label's name).
pub name: String,
pub why: String,
}
impl std::fmt::Display for StoreFailed {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(&self.why)
}
}
impl std::error::Error for StoreFailed {}
/// The request outcome of a [`StoreFailed`] pull.
const STORE_FAILED: &str = "store-failed";
/// **Is this refusal no verdict?** Every reason is a scanner that could not run
/// (`holger_inspect::FindingKind::{ScannerError, ScannerNotConfigured}`, named
/// `scanner-error` and `scanner-not-configured`); a reason that judged the bytes
/// (malware, a bomb, a gate) makes it a refusal.
fn undecided(e: &IntakeError) -> bool {
match e {
IntakeError::PolicyRefused { reasons, .. } => {
!reasons.is_empty()
&& reasons
.iter()
.all(|r| r.contains("scanner-error ") || r.contains("scanner-not-configured "))
}
_ => false,
}
}
/// The record of the upstream pages the Python sources listed ([`PipListing`]).
const PIP_LISTINGS: &str = "pip-listings.jsonl";
fn now_secs() -> i64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs() as i64)
.unwrap_or(0)
}
/// The actor a pull ingests and walks as.
pub fn actor_of(source: &str) -> String {
format!("pull-through:{source}")
}
/// The principal an automatic approval is cast by.
pub fn auto_approver_of(source: &str) -> String {
format!("auto-approve:{source}")
}
/// The coordinate a pulled Python file stands at.
#[cfg(test)]
mod scan_target_tests {
use super::{scan_target, RemoteKind};
/// **What a pulled file is scanned as** (lane HOL-FIELD-TEST P11): the source's kind
/// names the ecosystem the advisories are matched in, and the coordinates where the file
/// name alone does not carry them (a scoped npm package, a Maven group).
#[test]
fn a_pulled_file_is_scanned_as_its_sources_ecosystem() {
assert_eq!(
scan_target(RemoteKind::Python, "PyYAML-5.3.tar.gz"),
Some(("pip", None))
);
assert_eq!(
scan_target(RemoteKind::Npm, "@scope/pkg/-/pkg-1.2.3.tgz"),
Some(("npm", Some(("@scope/pkg".into(), "1.2.3".into()))))
);
assert_eq!(
scan_target(RemoteKind::Npm, "lodash/-/lodash-4.17.15.tgz"),
Some(("npm", Some(("lodash".into(), "4.17.15".into()))))
);
assert_eq!(
scan_target(
RemoteKind::Maven,
"org/apache/logging/log4j/log4j-core/2.14.1/log4j-core-2.14.1.jar"
),
Some((
"maven",
Some((
"org.apache.logging.log4j:log4j-core".into(),
"2.14.1".into()
))
))
);
assert_eq!(
scan_target(RemoteKind::Crates, "smallvec/smallvec-1.6.0.crate"),
Some(("cargo", None))
);
assert_eq!(
scan_target(RemoteKind::Oci, "chainguard/static/sha256-x.oci.tar"),
None
);
}
}
/// **What a file a source pulled is scanned as** (lane HOL-FIELD-TEST P11): the ecosystem its
/// advisories are matched in and, where the file name does not carry them, its coordinates.
/// `rest` is the coordinate below the source (`lodash/-/lodash-4.17.15.tgz`). `None`: the
/// chain's own rules stand (an image, an APK, a Go zip the chain reads itself).
pub fn scan_target(
kind: RemoteKind,
rest: &str,
) -> Option<(&'static str, Option<(String, String)>)> {
match kind {
RemoteKind::Python => Some(("pip", None)),
RemoteKind::Npm => Some((
"npm",
crate::promote::npm::coords_of(&ArtifactId {
namespace: None,
name: rest.to_string(),
version: String::new(),
}),
)),
RemoteKind::Maven => match maven::parse(rest)? {
maven::MavenPath::File {
group,
artifact,
version,
..
} => Some((
"maven",
Some((format!("{}:{artifact}", group.replace('/', ".")), version)),
)),
_ => None,
},
RemoteKind::Crates => Some(("cargo", None)),
RemoteKind::Go | RemoteKind::Oci | RemoteKind::Apk => None,
}
}
/// **The distribution an APK repository serves** (lane HOL-FIELD-TEST P12): the advisory
/// ecosystem its packages are matched in, by the repository URL.
pub fn apk_distro(url: &str) -> &'static str {
let u = url.to_ascii_lowercase();
if u.contains("chainguard") {
"chainguard"
} else if u.contains("wolfi") {
"wolfi"
} else {
"alpine"
}
}
#[cfg(test)]
mod apk_distro_tests {
#[test]
fn an_apk_repository_names_its_distribution() {
assert_eq!(super::apk_distro("https://packages.wolfi.dev/os"), "wolfi");
assert_eq!(
super::apk_distro("https://apk.cgr.dev/chainguard"),
"chainguard"
);
assert_eq!(
super::apk_distro("https://dl-cdn.alpinelinux.org/alpine/v3.20/main"),
"alpine"
);
}
}
pub fn pip_id(source: &str, file: &str) -> ArtifactId {
ArtifactId {
namespace: None,
name: format!("{source}/{file}"),
version: String::new(),
}
}
/// The coordinate a pulled image (by its digest) stands at.
pub fn oci_id(source: &str, name: &str, digest: &str) -> ArtifactId {
ArtifactId {
namespace: None,
name: format!("{source}/{name}/{}.oci.tar", digest.replace(':', "-")),
version: String::new(),
}
}
/// A source at run time: its config, its credential and its effective policy.
pub struct SourceRt {
pub cfg: RemoteSource,
basic: Option<String>,
/// What the provenance verifier holds this source's packages to.
pub demand: Demand,
/// The highest advisory severity (the floor or the source's, the lower).
pub cve_max: FindingSeverity,
}
impl SourceRt {
fn auth(&self) -> Option<(&str, &str)> {
self.basic.as_deref().map(|b| (self.cfg.url.as_str(), b))
}
fn cap(&self) -> u64 {
self.cfg.max_bytes.unwrap_or(4 * 1024 * 1024 * 1024)
}
fn fetch_wait(&self) -> std::time::Duration {
let default = if self.cfg.kind == RemoteKind::Maven {
30
} else {
0
};
std::time::Duration::from_secs(self.cfg.wait_secs.unwrap_or(default).min(300))
}
fn refresh_secs(&self) -> i64 {
self.cfg.refresh_secs.unwrap_or(6 * 3600) as i64
}
/// The effective policy, in one line for the audit.
pub fn effective_summary(&self, floor: &PullRules) -> String {
format!(
"source '{}': {}; cve_max {:?}; auto_approve {}",
self.cfg.name,
self.demand.summary(),
self.cve_max,
match (floor.allow_auto_approve, self.cfg.auto_approve) {
(_, AutoApprove::Off) => "off".to_string(),
(false, AutoApprove::When { .. }) => "asked, floor refuses".to_string(),
(
true,
AutoApprove::When {
provenance,
cve_max,
},
) =>
format!("when(provenance: {provenance:?}, cve_max: {cve_max:?})").to_lowercase(),
}
)
}
}
fn basic_of(c: &RemoteCredential, source: &str) -> anyhow::Result<String> {
use base64::Engine as _;
let password = match (&c.password_env, &c.password_file) {
(Some(env), None) => std::env::var(env).map_err(|_| {
anyhow::anyhow!("config `pull_through`: source '{source}': the environment variable {env} named by `password_env` is not set")
})?,
(None, Some(file)) => std::fs::read_to_string(file)
.map_err(|e| anyhow::anyhow!("config `pull_through`: source '{source}': `password_file` {file}: {e}"))?
.trim()
.to_string(),
_ => anyhow::bail!("config `pull_through`: source '{source}': a credential needs exactly one of `password_env` or `password_file`"),
};
Ok(format!(
"Basic {}",
base64::engine::general_purpose::STANDARD.encode(format!("{}:{password}", c.username))
))
}
/// **Validate the block and compute each source's effective policy.** Every
/// refusal names the source and the rule.
pub fn build_sources(s: &PullThroughSettings) -> anyhow::Result<Vec<SourceRt>> {
let floor = &s.rules;
let mut seen = std::collections::BTreeSet::new();
let mut out = Vec::new();
for src in &s.sources {
let n = &src.name;
if n.is_empty()
|| !n
.bytes()
.all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-' || b == b'_')
{
anyhow::bail!("config `pull_through`: a source needs a name of lower-case letters, digits, '-' and '_' (got {n:?})");
}
if matches!(n.as_str(), "simple" | "legacy" | "-") {
anyhow::bail!("config `pull_through`: source name '{n}' is a path the index door uses");
}
if !seen.insert(n.as_str()) {
anyhow::bail!("config `pull_through`: two sources are named '{n}'");
}
// A cargo index may be written as cargo writes it (`sparse+https://…`).
let mut src = src.clone();
if src.kind == RemoteKind::Crates {
if let Some(u) = src.url.strip_prefix("sparse+") {
src.url = u.to_string();
}
}
let src = &src;
if !src.serves_only() {
if http::origin_of(&src.url).is_none() {
anyhow::bail!("config `pull_through`: source '{n}': `url` must be an absolute http(s) URL (got {:?})", src.url);
}
// T9b: a pin this build cannot check is refused by name, never run unpinned. (An `online`
// build always carries the verifier, so this names the pin before the client.)
if let Some(ProvenanceGate::Require { pgp_keys, .. }) = &src.gates.provenance {
if !holger_provenance::pgp::BUILT && !pgp_keys.is_empty() {
anyhow::bail!(
"config `pull_through`: source '{n}' has {}",
holger_provenance::pgp::not_built(&format!(
"`require(pgp_keys: {pgp_keys:?})`"
))
);
}
}
if !cfg!(feature = "online") {
anyhow::bail!(
"config `pull_through`: source '{n}' (kind `{}`) fetches from {}, and this build has no outbound \
client; build with `--features online`",
src.kind.name(),
src.url
);
}
if crate::egress::is_offline() {
return Err(crate::egress::config_refusal(format!(
"`pull_through` source '{n}' fetches from {}",
src.url
)));
}
}
if src.kind != RemoteKind::Python && (!src.select.is_empty() || src.max_files.is_some()) {
anyhow::bail!(
"config `pull_through`: source '{n}' (kind `{}`) takes no `select` or `max_files`",
src.kind.name()
);
}
if src.kind != RemoteKind::Oci && !src.platforms.is_empty() {
anyhow::bail!(
"config `pull_through`: source '{n}' (kind `{}`) takes no `platforms`",
src.kind.name()
);
}
// The provenance demand: the floor, tightened by the source's gate.
let floor_prov = if floor.require_provenance || floor.slsa_min > 0 {
Mode::Require
} else {
Mode::Off
};
let signed = if floor.require_signed {
Mode::Require
} else {
Mode::Off
};
let mut identities = floor.identities.clone();
let mut slsa_min = floor.slsa_min;
let mut pgp_keys = Vec::new();
let mut apk_keys = Vec::new();
let provenance = match &src.gates.provenance {
None => floor_prov,
Some(ProvenanceGate::Off) | Some(ProvenanceGate::Prefer) if floor_prov == Mode::Require => anyhow::bail!(
"config `pull_through`: source '{n}' has `gates: (provenance: {})`, which loosens the floor \
(`rules: (require_provenance: true)`); a source's gates may only tighten the floor",
if src.gates.provenance == Some(ProvenanceGate::Off) { "off" } else { "prefer" }
),
Some(ProvenanceGate::Off) if signed == Mode::Require => anyhow::bail!(
"config `pull_through`: source '{n}' has `gates: (provenance: off)`, which loosens the floor \
(`rules: (require_signed: true)`); a source's gates may only tighten the floor"
),
Some(ProvenanceGate::Off) => Mode::Off,
Some(ProvenanceGate::Prefer) => Mode::Prefer,
Some(ProvenanceGate::Require { identity, identities: ids, pgp_keys: pgp, slsa_min: s, apk_keys: k }) => {
pgp_keys = pgp.clone();
apk_keys = k.clone();
let mut own: Vec<provenance::Identity> = ids.clone();
own.extend(identity.clone());
if !own.is_empty() {
// A source may only narrow the floor's identities.
if !floor.identities.is_empty() && own.iter().any(|i| !floor.identities.contains(i)) {
anyhow::bail!(
"config `pull_through`: source '{n}' trusts a signer identity the floor's `rules.identities` \
does not; a source's gates may only narrow them"
);
}
identities = own;
}
if let Some(s) = s {
if *s < floor.slsa_min {
anyhow::bail!(
"config `pull_through`: source '{n}' has `slsa_min: {s}`, below the floor's {}",
floor.slsa_min
);
}
slsa_min = *s;
}
Mode::Require
}
};
let cve_max = match src.gates.cve {
Some(CveGate { max }) if max > floor.cve_max => anyhow::bail!(
"config `pull_through`: source '{n}' has `gates: (cve: (max: {max:?}))`, above the floor's `cve_max: {:?}`; \
a source's gates may only tighten the floor",
floor.cve_max
),
Some(CveGate { max }) => max,
None => floor.cve_max,
};
if let AutoApprove::When {
cve_max: auto_max, ..
} = src.auto_approve
{
if let Some(why) = never_auto_approved(src.kind, n) {
anyhow::bail!("config `pull_through`: {why}; set the source's `auto_approve: off`");
}
if !floor.allow_auto_approve {
anyhow::bail!(
"config `pull_through`: source '{n}' asks `auto_approve: when(...)`, and the floor says \
`rules: (allow_auto_approve: false)`; auto-approval needs both (set the floor's \
`allow_auto_approve: true`, or the source's `auto_approve: off`)"
);
}
if auto_max > cve_max {
anyhow::bail!(
"config `pull_through`: source '{n}' would auto-approve up to `cve_max: {auto_max:?}`, above \
its effective `cve_max: {cve_max:?}`"
);
}
}
let basic = match &src.credential {
Some(c) => Some(basic_of(c, n)?),
None => None,
};
out.push(SourceRt {
cfg: src.clone(),
basic,
demand: Demand {
provenance,
signed,
identities,
slsa_min,
pgp_keys,
apk_keys,
},
cve_max,
});
}
Ok(out)
}
// ── the driver ──────────────────────────────────────────────────────────────
/// **The pull-through of one server**, beside its intake chain.
pub struct PullIntake {
sources: Vec<SourceRt>,
rules: PullRules,
retry_after_secs: u64,
http: Arc<dyn http::Fetch>,
verifier: std::sync::RwLock<Arc<dyn Verifier>>,
queue: OnceLock<Arc<crate::jobs::JobQueue>>,
/// `<prod_root>/lineage/pull-intake`.
root: PathBuf,
write: Mutex<()>,
/// The sealed layouts read back most recently, by the artifact's sha256.
layouts: Mutex<Vec<(String, Arc<oci::Layout>)>>,
/// The packument's view of a sealed npm tarball, by its sha256.
npm_meta: Mutex<BTreeMap<String, crate::promote::npm::VersionMeta>>,
/// A serve-only crates source's index line, derived from the sealed `.crate`, by its sha256.
derived_lines: Mutex<BTreeMap<String, String>>,
/// The pinned apk keys an upstream APK index must be signed with (the trust
/// bundle's `apk-keys/`, [`PullIntake::set_apk_keys`]).
apk_keys: std::sync::RwLock<Vec<holger_provenance::apk::ApkKey>>,
/// Holger's own apk signing key, made at its first use.
apk_signer: Mutex<Option<Arc<apk::Signer>>>,
/// The chain's provenance inbox, when its own provenance scanner runs.
#[cfg(feature = "scanners")]
inbox: Option<crate::scanner::provenance::ProvenanceInbox>,
}
impl std::fmt::Debug for PullIntake {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("PullIntake")
.field(
"sources",
&self.sources.iter().map(|s| &s.cfg.name).collect::<Vec<_>>(),
)
.finish()
}
}
const LAYOUT_CACHE: usize = 8;
impl PullIntake {
/// Build it from the settings. `prod_root` is the chain's
/// `<data_root>/holger-prod-store`.
pub fn new(
settings: &PullThroughSettings,
prod_root: impl Into<PathBuf>,
) -> anyhow::Result<Self> {
let sources = build_sources(settings)?;
#[cfg(feature = "online")]
let http = http::UreqFetch::shared();
#[cfg(not(feature = "online"))]
let http: Arc<dyn http::Fetch> = Arc::new(NoFetch);
Ok(Self {
sources,
rules: settings.rules.clone(),
retry_after_secs: settings.retry_after_secs.unwrap_or(2).max(1),
http,
verifier: std::sync::RwLock::new(provenance::default_verifier()),
queue: OnceLock::new(),
root: prod_root.into().join("lineage").join("pull-intake"),
write: Mutex::new(()),
layouts: Mutex::new(Vec::new()),
npm_meta: Mutex::new(BTreeMap::new()),
derived_lines: Mutex::new(BTreeMap::new()),
apk_keys: std::sync::RwLock::new(Vec::new()),
apk_signer: Mutex::new(None),
#[cfg(feature = "scanners")]
inbox: None,
})
}
/// Replace the outbound client (tests use the real client against fakes).
pub fn with_fetch(mut self, http: Arc<dyn http::Fetch>) -> Self {
self.http = http;
self
}
/// File what comes with each pulled package in the chain's provenance inbox too,
/// so the chain's own provenance scanner sees the same material at SCANNED.
#[cfg(feature = "scanners")]
pub fn with_provenance_inbox(
mut self,
inbox: crate::scanner::provenance::ProvenanceInbox,
) -> Self {
self.inbox = Some(inbox);
self
}
/// Replace the provenance verifier (G3a's `holger-provenance`, or a test's).
pub fn with_verifier(self, v: Arc<dyn Verifier>) -> Self {
self.replace_verifier(v);
self
}
/// Swap the provenance verifier on a running server (an embedder, a test).
pub fn replace_verifier(&self, v: Arc<dyn Verifier>) {
*self.verifier.write_or_recover() = v;
}
fn verifier(&self) -> Arc<dyn Verifier> {
Arc::clone(&self.verifier.read_or_recover())
}
/// **Pin the apk keys** (`(name, PEM)`, as the trust bundle carries them) an
/// upstream APK index must be signed with. Every one must be an RSA public key.
pub fn set_apk_keys(&self, keys: Vec<(String, Vec<u8>)>) -> anyhow::Result<()> {
let mut out = Vec::new();
for (name, pem) in &keys {
out.push(
holger_provenance::apk::ApkKey::from_pem(name, pem)
.map_err(|e| anyhow::anyhow!("pull-through: apk key {name:?}: {e}"))?,
);
}
*self.apk_keys.write_or_recover() = out;
Ok(())
}
fn apk_keys(&self) -> Vec<holger_provenance::apk::ApkKey> {
self.apk_keys.read_or_recover().clone()
}
/// Holger's apk signing key (`<root>/apk-signing/`), made at its first use.
pub fn apk_signer(&self) -> anyhow::Result<Arc<apk::Signer>> {
let mut held = self.apk_signer.lock_or_recover();
if let Some(s) = held.as_ref() {
return Ok(Arc::clone(s));
}
let s = Arc::new(apk::Signer::load_or_create(&self.root.join("apk-signing"))?);
*held = Some(Arc::clone(&s));
Ok(s)
}
/// The job queue pulls run on. Set once at start.
pub fn attach_queue(&self, q: Arc<crate::jobs::JobQueue>) {
let _ = self.queue.set(q);
}
pub fn sources(&self) -> &[SourceRt] {
&self.sources
}
pub fn source(&self, name: &str) -> Option<&SourceRt> {
self.sources.iter().find(|s| s.cfg.name == name)
}
/// The Python sources, in declaration order.
pub fn python_sources(&self) -> impl Iterator<Item = &SourceRt> {
self.sources
.iter()
.filter(|s| s.cfg.kind == RemoteKind::Python)
}
pub fn retry_after_secs(&self) -> u64 {
self.retry_after_secs
}
/// **The files the Python sources list for `project` that are not AVAILABLE here**, as
/// the project page names them (lane HOL-FIELD-TEST P2): yanked, so pip and uv skip them
/// unless a requirement pins that version with `==`, and asking for one's link pulls it
/// ([`Self::ask_file`]). An AVAILABLE file is the page's own; nothing here is served.
///
/// **A refused file is listed too** (lane HOL-310-CHECK), yanked with "refused by Holger:
/// <the reason>": a `==` pin selects it, so pip prints the reason ("Reason for being yanked")
/// before its link answers 403 with the same reason. Before, pip said only "from versions:
/// none". Asking for it pulls nothing: the chain (or the door's record) answers.
pub fn pip_held(&self, chain: &IntakeChain, project: &str) -> Vec<HeldFile> {
let project = crate::promote::intake::normalize_project(project);
let mut last: BTreeMap<String, PipListing> = BTreeMap::new();
for l in self.read::<PipListing>(PIP_LISTINGS) {
if l.project == project {
last.insert(l.source.clone(), l);
}
}
let mut out = Vec::new();
for src in self.python_sources() {
let Some(listing) = last.remove(&src.cfg.name) else {
continue;
};
for (filename, sha256) in listing.files {
let id = pip_id(&src.cfg.name, &filename);
let reason = match chain.status(&id).map(|_| chain.state(&id)) {
// Refused at the door (never stored): yanked with the refusal, never pulled again.
None => match self.refused_at_door(&src.cfg.name, &id.name) {
Some(why) => format!("refused by Holger: {why}"),
None => "not in Holger yet: asking for this file pulls it into Holger's intake chain, where it waits for approval".to_string(),
},
// Refused for good: yanked with the reason, never served.
Some(ArtifactState::Blocked) => {
format!("refused by Holger: {}", last_reason(chain, &id))
}
// AVAILABLE: the page's own. Held for a reason (an invalidation): never
// offered, as the AVAILABLE page never lists it.
Some(ArtifactState::Available | ArtifactState::Quarantined) => continue,
Some(state) => format!(
"held by Holger's intake chain ({}), waiting for approval",
serde_json::to_value(state)
.ok()
.and_then(|v| v.as_str().map(str::to_string))
.unwrap_or_default()
),
};
out.push(HeldFile {
source: src.cfg.name.clone(),
filename,
sha256,
reason,
});
}
}
out
}
// ── the record ──
fn append<T: Serialize>(&self, file: &str, v: &T) -> std::io::Result<()> {
use std::io::Write as _;
let _held = self.write.lock_or_recover();
std::fs::create_dir_all(&self.root)?;
let mut f = std::fs::OpenOptions::new()
.create(true)
.append(true)
.open(self.root.join(file))?;
let mut line = serde_json::to_vec(v).map_err(std::io::Error::other)?;
line.push(b'\n');
f.write_all(&line)?;
f.sync_data()
}
fn read<T: for<'de> Deserialize<'de>>(&self, file: &str) -> Vec<T> {
let Ok(text) = std::fs::read_to_string(self.root.join(file)) else {
return Vec::new();
};
text.lines()
.filter_map(|l| serde_json::from_str(l).ok())
.collect()
}
/// The request record, as text (diagnostics).
pub fn debug_records(&self) -> String {
std::fs::read_to_string(self.root.join("requests.jsonl")).unwrap_or_default()
}
/// The refusal a pull of `name` met at the door, under any want.
fn refused_at_door(&self, source: &str, name: &str) -> Option<String> {
self.read::<RequestRecord>("requests.jsonl")
.into_iter()
.rev()
.find(|r| {
r.source == source && r.outcome == "refused" && r.names.iter().any(|n| n == name)
})
.map(|r| r.why)
}
fn last_request(&self, source: &str, want: &Want) -> Option<RequestRecord> {
let key = want.key();
self.read::<RequestRecord>("requests.jsonl")
.into_iter()
.rev()
.find(|r| r.source == source && r.want == key)
// P8: a pull the scanners could not judge is asked again once the retry interval
// passed, so a healthy worker scans it; until then its answer stands (not served).
// HOL-ENOSPC-SETS: so is a pull the storage could not take.
.filter(|r| {
(r.outcome != SCAN_FAILED && r.outcome != STORE_FAILED)
|| now_secs() - r.at < self.retry_after_secs as i64
})
}
/// The latest request record of `source` that names `name`, when the storage could not
/// take that pull ([`StoreFailed`]).
fn store_failed_of(&self, source: &str, name: &str) -> Option<RequestRecord> {
self.read::<RequestRecord>("requests.jsonl")
.into_iter()
.rev()
.find(|r| r.source == source && r.names.iter().any(|n| n == name))
.filter(|r| r.outcome == STORE_FAILED)
}
/// **An artifact the storage failed for at SCANNED or later** stands where it stood (not
/// served): pulled again after the retry interval, held meanwhile with the cause by name.
fn redrive(&self, src: &SourceRt, id: &ArtifactId, state: ArtifactState) -> Option<Ask> {
let source = src.cfg.name.as_str();
let r = self.store_failed_of(source, &id.name)?;
let want = r.retry?;
if let Some(job) = self.open_job(source, &want) {
return Some(Ask::Fetching {
retry_after_secs: self.retry_after_secs,
job,
});
}
if now_secs() - r.at >= self.retry_after_secs as i64 {
return Some(self.enqueue(source, &want));
}
Some(Ask::Held {
state,
detail: format!(
"{} stands at {state} in the intake chain: {}; retried in {} s",
id.name, r.why, self.retry_after_secs
),
})
}
/// The error of a pull the storage could not take, told to the operator by name (the log
/// here; the job's failure reason on the console's jobs page).
fn store_failed(&self, src: &SourceRt, id: &ArtifactId, cause: &str) -> anyhow::Error {
let why = format!(
"{}: the storage could not take it, so nothing is served and it is pulled again \
after {} s: {cause}",
id.name, self.retry_after_secs
);
log::error!("pull-through {}: {why}", src.cfg.name);
anyhow::Error::new(StoreFailed {
name: id.name.clone(),
why,
})
}
fn record(&self, source: &str, want: &Want, outcome: &str, names: Vec<String>, why: &str) {
let r = RequestRecord {
source: source.into(),
want: want.key(),
at: now_secs(),
outcome: outcome.into(),
names,
why: why.into(),
retry: (outcome == STORE_FAILED).then(|| want.clone()),
};
// The estate board's count of this source (HBOARD-1).
crate::upstreams::record_pull(source, outcome, crate::clock::now_unix_ms());
if let Err(e) = self.append("requests.jsonl", &r) {
log::warn!("pull-through: the request record could not be written: {e}");
}
}
fn tags(&self, source: &str, name: &str) -> BTreeMap<String, TagRecord> {
let mut out = BTreeMap::new();
for t in self.read::<TagRecord>("oci-tags.jsonl") {
if t.source == source && t.name == name {
out.insert(t.tag.clone(), t);
}
}
out
}
/// The tags of `name` this source knows: what it pulled (its tag record) and, for a
/// serve-only source (a downstream's view), the tag each AVAILABLE image's sealed
/// layout names, the newest landing winning a tag that moved.
fn tags_seen(
&self,
chain: &IntakeChain,
src: &SourceRt,
name: &str,
) -> BTreeMap<String, TagRecord> {
let source = src.cfg.name.as_str();
let mut out = self.tags(source, name);
if src.cfg.serves_only() {
let mut seen: Vec<(i64, String, String)> = Vec::new();
for id in self.image_ids(chain, source, name) {
if chain.state(&id) != ArtifactState::Available {
continue;
}
let at = chain.history(&id).last().map(|e| e.at).unwrap_or(0);
if let Some(l) = self.layout(chain, &id) {
if let Some(t) = &l.tag {
seen.push((at, t.clone(), l.digest.clone()));
}
}
}
seen.sort();
for (at, tag, digest) in seen {
out.insert(
tag.clone(),
TagRecord {
source: source.into(),
name: name.into(),
tag,
digest,
at,
},
);
}
}
out
}
fn materials_dir(&self, sha256: &str) -> PathBuf {
self.root.join("materials").join(sha256)
}
fn keep_materials(&self, sha256: &str, materials: &[Material]) -> std::io::Result<()> {
let dir = self.materials_dir(sha256);
std::fs::create_dir_all(&dir)?;
for m in materials {
let safe: String = m
.name
.chars()
.map(|c| {
if c.is_ascii_alphanumeric() || matches!(c, '.' | '-' | '_') {
c
} else {
'_'
}
})
.collect();
let dst = dir.join(safe);
// A large material many packages share (an APK index) is kept once, by its
// digest, and linked into each package's directory.
if m.bytes.len() >= 1 << 20 {
let blobs = self.root.join("material-blobs");
std::fs::create_dir_all(&blobs)?;
let blob = blobs.join(crate::attest::ArtifactDigest::of_sha256(&m.bytes).hex());
if !blob.exists() {
// A unique staging name per write: two packages sharing one
// material race no fixed `.tmp` (HOL-RACE-1).
static SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
let seq = SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let tmp = blobs.join(format!(".material.{}.{seq}.tmp", std::process::id()));
let staged =
std::fs::write(&tmp, &m.bytes).and_then(|()| std::fs::rename(&tmp, &blob));
if let Err(e) = staged {
let _ = std::fs::remove_file(&tmp);
return Err(e);
}
}
let _ = std::fs::remove_file(&dst);
if std::fs::hard_link(&blob, &dst).is_ok() {
continue;
}
}
std::fs::write(dst, &m.bytes)?;
}
Ok(())
}
fn kept_materials(&self, sha256: &str) -> Vec<Material> {
let Ok(rd) = std::fs::read_dir(self.materials_dir(sha256)) else {
return Vec::new();
};
let mut out: Vec<Material> = rd
.flatten()
.filter_map(|e| {
Some(Material {
name: e.file_name().to_string_lossy().to_string(),
bytes: std::fs::read(e.path()).ok()?,
})
})
.collect();
out.sort_by(|a, b| a.name.cmp(&b.name));
out
}
// ── asking ──
fn content_id(source: &str, want: &Want) -> String {
format!("pull-intake:{source}:{}", want.key())
}
fn enqueue(&self, source: &str, want: &Want) -> Ask {
// A source that serves only (a downstream's view) never fetches a miss.
if self.source(source).is_some_and(|s| s.cfg.serves_only()) {
return Ask::Unknown {
why: not_sent(source, want),
};
}
let Some(q) = self.queue.get() else {
return Ask::Unknown {
why: "this server runs no job queue, and a pull-through runs on it".into(),
};
};
let job = crate::jobs::Job::PullIntake {
source: source.into(),
want: want.clone(),
};
match q.enqueue(job) {
Ok(j) => Ask::Fetching {
retry_after_secs: self.retry_after_secs,
job: j.0,
},
Err(e) => Ask::Unknown {
why: format!("the pull could not be queued: {e:#}"),
},
}
}
fn open_job(&self, source: &str, want: &Want) -> Option<String> {
self.queue
.get()?
.open_job_for(&Self::content_id(source, want))
.map(|j| j.0)
}
/// **Walk every APPROVED, SEALED or PROMOTED pull of this source on to
/// AVAILABLE** (the approval was the judgement; the rest is the seal and its
/// read-back), and say where the coordinates stand.
pub fn settle(&self, chain: &IntakeChain, src: &SourceRt, ids: &[ArtifactId]) -> Ask {
let mut best = Ask::Unknown {
why: "nothing was pulled".into(),
};
for id in ids {
let actor = actor_of(&src.cfg.name);
// A serve-only source's coordinates arrived by import or pull (their importer is
// the bundle's); a fetching source settles only what it pulled itself.
if !src.cfg.serves_only() && chain.importer(id).as_deref() != Some(actor.as_str()) {
continue;
}
let state = match chain.state(id) {
ArtifactState::Approved | ArtifactState::Sealed | ArtifactState::Promoted => {
match self.finish(chain, &actor, id) {
Ok(s) => s,
Err(e) => {
log::warn!(
"pull-through {}: finishing {} failed: {e}",
src.cfg.name,
id.name
);
chain.state(id)
}
}
}
s => s,
};
// HOL-ENOSPC-SETS: a pull the storage could not take is walked again, never held.
if matches!(state, ArtifactState::Ingested | ArtifactState::Scanned)
&& !src.cfg.serves_only()
{
if let Some(ask) = self.redrive(src, id, state) {
best = best.best(ask);
continue;
}
}
let ask = match state {
ArtifactState::Available => Ask::Ready,
ArtifactState::Blocked => Ask::Refused {
why: last_reason(chain, id),
},
// Invalidated by its upstream (H4): refused with the reason, never "not yet".
ArtifactState::Quarantined if chain.invalidation_of(id).is_some() => Ask::Refused {
why: chain.invalidation_of(id).unwrap_or_default(),
},
ArtifactState::Untrusted => continue,
s => Ask::Held {
state: s,
detail: format!(
"{} stands at {s} in the intake chain{}",
id.name,
if s == ArtifactState::PolicyChecked {
", waiting for approval"
} else {
""
}
),
},
};
best = best.best(ask);
}
best
}
fn finish(
&self,
chain: &IntakeChain,
actor: &str,
id: &ArtifactId,
) -> Result<ArtifactState, IntakeError> {
loop {
let s = chain.state(id);
let next = match s {
ArtifactState::Approved => ArtifactState::Sealed,
ArtifactState::Sealed => ArtifactState::Promoted,
ArtifactState::Promoted => ArtifactState::Available,
s => return Ok(s),
};
chain.advance(
TransitionRequest::new(id.clone(), next, actor).from_source("pull-through"),
)?;
}
}
/// The coordinates a source has brought in for one Python project.
fn project_ids(&self, chain: &IntakeChain, source: &str, project: &str) -> Vec<ArtifactId> {
let prefix = format!("{source}/");
chain
.all()
.into_iter()
.map(|(id, _)| id)
.filter(|id| id.namespace.is_none() && id.version.is_empty())
.filter(|id| {
id.name.strip_prefix(&prefix).is_some_and(|f| {
!f.contains('/')
&& crate::promote::intake::python_project_of(f).as_deref() == Some(project)
})
})
.collect()
}
/// **A project page was asked for** and lists nothing AVAILABLE (or is due
/// a refresh). Settles what is approved, and pulls when nothing is on the way.
pub fn ask_project(&self, chain: &IntakeChain, project: &str) -> Ask {
let project = crate::promote::intake::normalize_project(project);
let mut best = Ask::Unknown {
why: format!("no Python source publishes '{project}'"),
};
for src in self.python_sources() {
let name = &src.cfg.name;
let want = Want::PipProject {
project: project.clone(),
};
let ids = self.project_ids(chain, name, &project);
let settled = self.settle(chain, src, &ids);
let last = self.last_request(name, &want);
let stale = last
.as_ref()
.is_none_or(|r| now_secs() - r.at >= src.refresh_secs());
let ask = if let Some(job) = self.open_job(name, &want) {
settled.best(Ask::Fetching {
retry_after_secs: self.retry_after_secs,
job,
})
} else if last.is_none()
|| (stale && settled == Ask::Ready)
|| last
.as_ref()
.is_some_and(|r| r.outcome == "failed" && stale)
{
settled.best(self.enqueue(name, &want))
} else {
match (&settled, &last) {
(Ask::Unknown { .. }, Some(r)) if r.outcome == "refused" => {
Ask::Refused { why: r.why.clone() }
}
(Ask::Unknown { .. }, Some(r)) if r.outcome != "fetched" => {
Ask::Unknown { why: r.why.clone() }
}
(s, _) => s.clone(),
}
};
best = best.best(ask);
}
best
}
/// **A file link was asked for** (`/production/<source>/<file>`) and the
/// file is not AVAILABLE.
pub fn ask_file(&self, chain: &IntakeChain, source: &str, file: &str) -> Ask {
let Some(src) = self
.source(source)
.filter(|s| s.cfg.kind == RemoteKind::Python)
else {
return Ask::Unknown {
why: format!("'{source}' is no Python source"),
};
};
if crate::promote::intake::python_project_of(file).is_none() || file.contains('/') {
return Ask::Unknown {
why: format!("'{file}' is not a Python distribution file"),
};
}
let id = pip_id(source, file);
let settled = self.settle(chain, src, std::slice::from_ref(&id));
if settled
!= (Ask::Unknown {
why: "nothing was pulled".into(),
})
{
return settled;
}
if let Some(why) = self.refused_at_door(source, &id.name) {
return Ask::Refused { why };
}
let want = Want::PipFile { file: file.into() };
if let Some(job) = self.open_job(source, &want) {
return Ask::Fetching {
retry_after_secs: self.retry_after_secs,
job,
};
}
match self.last_request(source, &want) {
Some(r) if r.outcome == "refused" => Ask::Refused { why: r.why },
Some(r) if r.outcome != "fetched" && now_secs() - r.at < src.refresh_secs() => {
Ask::Unknown { why: r.why }
}
_ => self.enqueue(source, &want),
}
}
/// **An image manifest was asked for.** `Ok(Some((media_type, bytes, digest)))`
/// when it is AVAILABLE; otherwise the answer to give.
pub fn oci_manifest(
&self,
chain: &IntakeChain,
source: &str,
name: &str,
reference: &str,
) -> Result<(String, Vec<u8>, String), Ask> {
let Some(src) = self
.source(source)
.filter(|s| s.cfg.kind == RemoteKind::Oci)
else {
return Err(Ask::Unknown {
why: format!("'{source}' is no OCI source"),
});
};
let want = Want::Oci {
name: name.into(),
reference: reference.into(),
};
let tag = (!oci::is_digest(reference)).then_some(reference);
// The digest the reference stands for, as far as Holger has pulled it.
let digest: Option<String> = match tag {
Some(t) => self
.tags_seen(chain, src, name)
.get(t)
.map(|r| r.digest.clone()),
None => Some(reference.to_string()),
};
if let Some(d) = &digest {
// The top manifest, or a platform manifest inside an AVAILABLE layout.
for id in self.image_ids(chain, source, name) {
let ask = self.settle(chain, src, std::slice::from_ref(&id));
if ask != Ask::Ready {
if id == oci_id(source, name, d) {
if let Some(job) = self.open_job(source, &want) {
return Err(Ask::Fetching {
retry_after_secs: self.retry_after_secs,
job,
});
}
return Err(ask);
}
continue;
}
if let Some(l) = self.layout(chain, &id) {
if let Some((mt, b)) = l.manifest(d) {
let out = (mt.to_string(), b.to_vec(), d.clone());
// A tag past its refresh age asks the source again; the
// AVAILABLE image is served meanwhile.
if let (Some(_), Some(r)) = (
tag,
tag.and_then(|t| self.tags(source, name).get(t).cloned()),
) {
if now_secs() - r.at >= src.refresh_secs()
&& self.open_job(source, &want).is_none()
{
let _ = self.enqueue(source, &want);
}
}
return Ok(out);
}
}
}
}
if let Some(job) = self.open_job(source, &want) {
return Err(Ask::Fetching {
retry_after_secs: self.retry_after_secs,
job,
});
}
match self.last_request(source, &want) {
Some(r) if r.outcome == "refused" => Err(Ask::Refused { why: r.why }),
Some(r) if r.outcome != "fetched" && now_secs() - r.at < src.refresh_secs() => {
Err(Ask::Unknown { why: r.why })
}
Some(r) if r.outcome == "fetched" && digest.is_some() => Err(Ask::Unknown {
why: format!(
"{name}:{reference} was pulled as {}, which is not AVAILABLE",
digest.unwrap_or_default()
),
}),
_ => Err(self.enqueue(source, &want)),
}
}
/// A blob of an AVAILABLE image.
pub fn oci_blob(
&self,
chain: &IntakeChain,
source: &str,
name: &str,
digest: &str,
) -> Option<Vec<u8>> {
for id in self.image_ids(chain, source, name) {
if chain.state(&id) != ArtifactState::Available {
continue;
}
if let Some(b) = self
.layout(chain, &id)
.and_then(|l| l.blob(digest).map(|b| b.to_vec()))
{
return Some(b);
}
}
None
}
/// The tags whose pulled digest is AVAILABLE.
pub fn oci_tags(&self, chain: &IntakeChain, source: &str, name: &str) -> Vec<String> {
let Some(src) = self.source(source) else {
return Vec::new();
};
self.tags_seen(chain, src, name)
.into_values()
.filter(|t| chain.state(&oci_id(source, name, &t.digest)) == ArtifactState::Available)
.map(|t| t.tag)
.collect()
}
fn image_ids(&self, chain: &IntakeChain, source: &str, name: &str) -> Vec<ArtifactId> {
let prefix = format!("{source}/{name}/");
chain
.all()
.into_iter()
.map(|(id, _)| id)
.filter(|id| {
id.name
.strip_prefix(&prefix)
.is_some_and(|r| !r.contains('/') && r.ends_with(".oci.tar"))
})
.collect()
}
/// The sealed layouts of `name` that are AVAILABLE (HOL-REFERRERS: the attached
/// artifacts and Holger's evidence are served from these, never from a pull).
pub fn oci_layouts(
&self,
chain: &IntakeChain,
source: &str,
name: &str,
) -> Vec<Arc<oci::Layout>> {
self.image_ids(chain, source, name)
.into_iter()
.filter(|id| chain.state(id) == ArtifactState::Available)
.filter_map(|id| self.layout(chain, &id))
.collect()
}
fn layout(&self, chain: &IntakeChain, id: &ArtifactId) -> Option<Arc<oci::Layout>> {
let sha = chain.status(id)?.sha256;
if let Some((_, l)) = self
.layouts
.lock_or_recover()
.iter()
.find(|(s, _)| *s == sha)
{
return Some(Arc::clone(l));
}
let bytes = chain.serve(id).ok().flatten()?;
let l = Arc::new(
oci::Layout::read(&bytes)
.map_err(|e| log::warn!("pull-through: the sealed layout of {}: {e}", id.name))
.ok()?,
);
let mut c = self.layouts.lock_or_recover();
c.push((sha, Arc::clone(&l)));
if c.len() > LAYOUT_CACHE {
c.remove(0);
}
Some(l)
}
// ── the job ──
/// **The pull job's body** (`Job::PullIntake`): fetch, ingest, walk.
/// A network failure is an `Err` the queue retries; a refusal by the chain
/// is recorded and is the job's result.
pub fn run(&self, chain: &IntakeChain, source: &str, want: &Want) -> anyhow::Result<String> {
let src = self.source(source).ok_or_else(|| {
anyhow::anyhow!(crate::jobs::JobPermanent(format!(
"no pull-through source '{source}' is configured"
)))
})?;
let out = match want {
Want::PipProject { project } => self.run_project(chain, src, project, want),
Want::PipFile { file } => self.run_file(chain, src, file, want),
Want::Oci { name, reference } => self.run_oci(chain, src, name, reference, want),
w => self.run_ecosystem(chain, src, w),
};
if let Err(e) = &out {
if let Some(sf) = e.downcast_ref::<StoreFailed>() {
let names = vec![sf.name.clone()];
self.record(source, want, STORE_FAILED, names, &format!("{e:#}"));
} else {
let outcome = if e.downcast_ref::<ScanFailed>().is_some() {
SCAN_FAILED
} else {
"failed"
};
self.record(source, want, outcome, Vec::new(), &format!("{e:#}"));
}
}
out
}
fn run_project(
&self,
chain: &IntakeChain,
src: &SourceRt,
project: &str,
want: &Want,
) -> anyhow::Result<String> {
let Some(files) =
pip::project_files(self.http.as_ref(), &src.cfg.url, project, src.auth())?
else {
let why = format!("source '{}' does not publish '{project}'", src.cfg.name);
self.record(&src.cfg.name, want, "unknown", Vec::new(), &why);
return Ok(why);
};
// What the upstream lists, for the project page's yanked links (P2): written before
// the newest release is driven, so a client's retry already sees the page.
self.remember_listing_pip(&src.cfg.name, project, &files);
let select = if src.cfg.select.is_empty() {
pip::default_select()
} else {
src.cfg.select.clone()
};
let chosen = pip::newest_release(&files, &select, src.cfg.max_files.unwrap_or(8));
if chosen.is_empty() {
let why = format!(
"source '{}' publishes no final release of '{project}' this source selects",
src.cfg.name
);
self.record(&src.cfg.name, want, "unknown", Vec::new(), &why);
return Ok(why);
}
let root = format!(
"{project}=={}",
chosen
.first()
.and_then(|f| pip::version_of(&f.filename))
.unwrap_or_default()
);
let w = self.pip_walk(chain, src, project, chosen)?;
Ok(self.record_walk(src, want, root, w))
}
/// Keep what the upstream listed for `project` ([`PipListing`]).
fn remember_listing_pip(&self, source: &str, project: &str, files: &[pip::RemoteFile]) {
let listing = PipListing {
source: source.to_string(),
project: project.to_string(),
at: now_secs(),
files: files
.iter()
.filter(|f| !f.yanked && pip::version_of(&f.filename).is_some())
.filter_map(|f| f.sha256.clone().map(|h| (f.filename.clone(), h)))
.collect(),
};
if let Err(e) = self.append(PIP_LISTINGS, &listing) {
log::warn!("pull-through: the upstream listing could not be written: {e}");
}
}
fn run_file(
&self,
chain: &IntakeChain,
src: &SourceRt,
file: &str,
want: &Want,
) -> anyhow::Result<String> {
let project = crate::promote::intake::python_project_of(file).ok_or_else(|| {
anyhow::anyhow!(crate::jobs::JobPermanent(format!(
"'{file}' is not a Python distribution file"
)))
})?;
let files = pip::project_files(self.http.as_ref(), &src.cfg.url, &project, src.auth())?
.unwrap_or_default();
let Some(f) = files.into_iter().find(|f| f.filename == file) else {
let why = format!("source '{}' does not publish {file}", src.cfg.name);
self.record(&src.cfg.name, want, "unknown", Vec::new(), &why);
return Ok(why);
};
if f.yanked {
let id = pip_id(&src.cfg.name, file);
let (line, refused) = self.drive_pip(chain, src, &id, &f)?;
return Ok(self.record_drives(&src.cfg.name, want, vec![(id.name, line, refused)]));
}
let root = format!("{project}=={}", pip::version_of(file).unwrap_or_default());
let w = self.pip_walk(chain, src, &project, vec![f])?;
Ok(self.record_walk(src, want, root, w))
}
/// `(line, refused at the door)`.
fn drive_pip(
&self,
chain: &IntakeChain,
src: &SourceRt,
id: &ArtifactId,
f: &pip::RemoteFile,
) -> anyhow::Result<(String, bool)> {
if chain.status(id).is_none() {
if f.yanked {
return Ok((format!("{}: yanked upstream, not pulled", f.filename), true));
}
let (bytes, materials) = pip::fetch_file(self.http.as_ref(), f, src.auth(), src.cap())?;
if let Some(line) = self.ingest(chain, src, id, &bytes, &materials)? {
return Ok((line, true));
}
}
Ok((self.walk(chain, src, id, None)?, false))
}
/// Record a pull's outcome: `refused` when nothing it fetched got through the door.
fn record_drives(
&self,
source: &str,
want: &Want,
drives: Vec<(String, String, bool)>,
) -> String {
let lines: Vec<String> = drives.iter().map(|(_, l, _)| l.clone()).collect();
let names: Vec<String> = drives
.iter()
.filter(|(_, _, refused)| !refused)
.map(|(n, _, _)| n.clone())
.collect();
if names.is_empty() {
let refused = drives.iter().map(|(n, _, _)| n.clone()).collect();
self.record(source, want, "refused", refused, &lines.join("; "));
} else {
self.record(source, want, "fetched", names, "");
}
lines.join("; ")
}
fn run_oci(
&self,
chain: &IntakeChain,
src: &SourceRt,
name: &str,
reference: &str,
want: &Want,
) -> anyhow::Result<String> {
if !oci::is_name(name) || !(oci::is_tag(reference) || oci::is_digest(reference)) {
anyhow::bail!(crate::jobs::JobPermanent(format!(
"'{name}:{reference}' is not an image reference"
)));
}
let reg = oci::Registry::new(self.http.as_ref(), &src.cfg.url, src.basic.clone());
// A tag is resolved first: an image already pulled is not pulled again.
let digest = if oci::is_digest(reference) {
reference.to_string()
} else {
match reg.manifest(name, reference)? {
Some((_, _, d)) => d,
None => {
let why = format!("source '{}' has no {name}:{reference}", src.cfg.name);
self.record(&src.cfg.name, want, "unknown", Vec::new(), &why);
return Ok(why);
}
}
};
let id = oci_id(&src.cfg.name, name, &digest);
let mut refused = false;
let line = if chain.status(&id).is_none() {
let Some(p) = oci::pull(®, name, &digest, &src.cfg.platforms, src.cap())? else {
let why = format!("source '{}' has no {name}@{digest}", src.cfg.name);
self.record(&src.cfg.name, want, "unknown", Vec::new(), &why);
return Ok(why);
};
let mut p = p;
p.tag = (!oci::is_digest(reference)).then(|| reference.to_string());
let tar = oci::layout_tar(&p)?;
match self.ingest(chain, src, &id, &tar, &p.materials)? {
Some(line) => {
refused = true;
line
}
None => self.walk(chain, src, &id, Some(p.top().to_vec()))?,
}
} else {
self.walk(chain, src, &id, None)?
};
if !oci::is_digest(reference) {
let t = TagRecord {
source: src.cfg.name.clone(),
name: name.into(),
tag: reference.into(),
digest: digest.clone(),
at: now_secs(),
};
if let Err(e) = self.append("oci-tags.jsonl", &t) {
log::warn!("pull-through: the tag record could not be written: {e}");
}
}
Ok(self.record_drives(&src.cfg.name, want, vec![(id.name.clone(), line, refused)]))
}
/// UNTRUSTED → INGESTED as `pull-through:<source>`. `Some(line)` when the
/// chain refused it (recorded, final); `None` when it is INGESTED.
fn ingest(
&self,
chain: &IntakeChain,
src: &SourceRt,
id: &ArtifactId,
bytes: &[u8],
materials: &[Material],
) -> anyhow::Result<Option<String>> {
let actor = actor_of(&src.cfg.name);
let sha = crate::attest::ArtifactDigest::of_sha256(bytes)
.hex()
.to_string();
if let Err(e) = self.keep_materials(&sha, materials) {
return Err(self.store_failed(
src,
id,
&format!("the provenance material could not be kept ({e})"),
));
}
#[cfg(feature = "scanners")]
if let Some(inbox) = &self.inbox {
for m in materials {
let safe: String = m
.name
.chars()
.map(|c| {
if c.is_ascii_alphanumeric() || matches!(c, '.' | '-' | '_') {
c
} else {
'_'
}
})
.collect();
if let Err(e) = inbox.put(&sha, safe.trim_start_matches('.'), &m.bytes) {
return Err(self.store_failed(
src,
id,
&format!("the provenance material could not be filed in the inbox ({e:#})"),
));
}
}
}
match chain.ingest_pulled(&actor, &src.cfg.url, id, bytes) {
Ok(_) => Ok(None),
Err(e) if e.is_storage_failure() => Err(self.store_failed(src, id, &e.tagged())),
Err(e) if undecided(&e) => Err(anyhow::Error::new(ScanFailed(format!(
"{}: the scanners could not judge it, so nothing was stored and it is pulled and \
scanned again at the next request: {}",
id.name,
e.tagged()
)))),
Err(e) => Ok(Some(format!(
"{}: refused at the door: {}",
id.name,
e.tagged()
))),
}
}
/// **Walk a pulled coordinate as far as policy lets it**: SCANNED (with the
/// provenance verdict), POLICY_CHECKED, the automatic approval when the
/// source and the floor allow it, then on to AVAILABLE. A refusal blocks it.
fn walk(
&self,
chain: &IntakeChain,
src: &SourceRt,
id: &ArtifactId,
subject: Option<Vec<u8>>,
) -> anyhow::Result<String> {
let actor = actor_of(&src.cfg.name);
let mut provenance_said: Option<String> = None;
loop {
let Some(status) = chain.status(id) else {
return Ok(format!("{}: not in the chain", id.name));
};
let read = status.state;
let step = match status.state {
ArtifactState::Ingested => {
let mut req =
TransitionRequest::new(id.clone(), ArtifactState::Scanned, &actor)
.from_source("pull-through");
if !src.demand.is_off() {
let ev = self.provenance_evidence(
chain,
src,
id,
&status.sha256,
subject.as_deref(),
);
if ev.verdict == EvidenceVerdict::Fail {
provenance_said = Some(ev.reference.clone());
}
req = req.with_evidence(ev);
}
chain.advance(req.with_reason(src.effective_summary(&self.rules)))
}
ArtifactState::Scanned => chain.advance(
TransitionRequest::new(id.clone(), ArtifactState::PolicyChecked, &actor)
.from_source("pull-through"),
),
ArtifactState::PolicyChecked => {
return Ok(match self.auto_approval(chain, src, id, &status.sha256) {
Err(why) => {
format!("{}: POLICY_CHECKED, waiting for approval ({why})", id.name)
}
Ok(reason) => {
let who = auto_approver(&src.cfg.name);
let req = TransitionRequest::new(
id.clone(),
ArtifactState::Approved,
&who.principal,
)
.from_source("pull-through")
.with_reason(reason);
match chain.approve_or_advance(req, &who) {
Ok(ApproveOutcome::Pending { have, need, .. }) => {
format!("{}: auto-approved as {have} of {need}; waiting for a person", id.name)
}
Ok(ApproveOutcome::Approved { .. }) => {
match self.finish(chain, &actor, id) {
Ok(s) => format!("{}: auto-approved, {s}", id.name),
Err(e) => format!(
"{}: auto-approved; the seal failed: {}",
id.name,
e.tagged()
),
}
}
Err(e) => format!(
"{}: the automatic approval was refused: {}",
id.name,
e.tagged()
),
}
}
});
}
ArtifactState::Approved | ArtifactState::Sealed | ArtifactState::Promoted => {
return Ok(match self.finish(chain, &actor, id) {
Ok(s) => format!("{}: {s}", id.name),
Err(e) => format!("{}: {}", id.name, e.tagged()),
});
}
s => return Ok(format!("{}: {s}", id.name)),
};
if let Err(e) = step {
// Another pull of the same coordinate moved it meanwhile (cargo asks for a
// crate's index and its download at once: two jobs, one artifact). That is no
// refusal: walk on from where it stands now (lane HOL-FIELD-2).
if chain.status(id).is_some_and(|s| s.state != read) {
provenance_said = None;
continue;
}
// HOL-ENOSPC-SETS: a store that could not be written judged nothing. The
// artifact stays where it stands (never served) and is walked again.
if e.is_storage_failure() {
return Err(self.store_failed(src, id, &e.tagged()));
}
let mut why = e.tagged();
if let Some(p) = provenance_said.take() {
why.push_str(&format!("; provenance: {p}"));
}
let block = TransitionRequest::new(id.clone(), ArtifactState::Blocked, &actor)
.from_source("pull-through")
.with_reason(format!(
"refused on the pull-through from '{}': {why}",
src.cfg.name
));
return Ok(match chain.advance(block) {
Ok(_) => format!("{}: BLOCKED ({why})", id.name),
Err(e2) => format!(
"{}: refused ({why}); the block failed: {}",
id.name,
e2.tagged()
),
});
}
}
}
fn provenance_evidence(
&self,
chain: &IntakeChain,
src: &SourceRt,
id: &ArtifactId,
sha256: &str,
subject: Option<&[u8]>,
) -> Evidence {
let materials = self.kept_materials(sha256);
// What the signatures are about: an image's manifest (its digest), a file's bytes.
let subject: Option<Vec<u8>> = match subject {
Some(s) => Some(s.to_vec()),
None => chain.incoming_artifact(id).ok().map(|b| {
if id.name.ends_with(".oci.tar") {
oci::Layout::read(&b)
.ok()
.and_then(|l| l.manifest(&l.digest).map(|(_, m)| m.to_vec()))
.unwrap_or_default()
} else {
b
}
}),
};
let Some(subject) = subject.filter(|s| !s.is_empty()) else {
return Evidence::new(
EvidenceKind::Provenance,
provenance::SCANNER_ID,
EvidenceVerdict::Fail,
)
.with_reference("the bytes the signatures are about could not be read");
};
let required = src.demand.signed == Mode::Require || src.demand.provenance == Mode::Require;
if materials.is_empty() && required {
return Evidence::new(EvidenceKind::Provenance, provenance::SCANNER_ID, EvidenceVerdict::Fail).with_reference(format!(
"no signature or attestation came with it from '{}', and its effective policy requires one ({})",
src.cfg.name,
src.demand.summary()
));
}
let verifier = self.verifier();
let o = verifier.verify(&subject, &materials, &src.demand);
let verdict = match o.verdict {
Verdict::Pass => EvidenceVerdict::Pass,
Verdict::Warn => EvidenceVerdict::Warn,
Verdict::Block => EvidenceVerdict::Fail,
};
Evidence::new(EvidenceKind::Provenance, verifier.id(), verdict).with_reference(format!(
"{}{} ({}; {} material(s), {} verified)",
o.code.map(|c| format!("[{c}] ")).unwrap_or_default(),
o.detail,
src.demand.summary(),
materials.len(),
o.verified
))
}
/// **May this pull be approved automatically?** `Ok(reason)` for the vote;
/// `Err(why not)`.
fn auto_approval(
&self,
chain: &IntakeChain,
src: &SourceRt,
id: &ArtifactId,
sha256: &str,
) -> Result<String, String> {
let AutoApprove::When {
provenance,
cve_max,
} = src.cfg.auto_approve
else {
return Err(format!("source '{}' has auto_approve: off", src.cfg.name));
};
if let Some(why) = never_auto_approved(src.cfg.kind, &src.cfg.name) {
return Err(why);
}
if !self.rules.allow_auto_approve {
return Err("the floor says allow_auto_approve: false".into());
}
// W1: a WARN that excuses what the floor blocks is a person's call.
for e in chain.history(id) {
if let Some(why) = excused_block(&e.evidence) {
return Err(why);
}
}
let mut said = vec![format!(
"auto-approved: source '{}' auto_approve: when(provenance: {}, cve_max: {}), floor allow_auto_approve: true",
src.cfg.name,
if provenance == AutoNeed::Require { "require" } else { "off" },
format!("{cve_max:?}").to_lowercase()
)];
if provenance == AutoNeed::Require {
let passed = chain.history(id).iter().any(|e| {
e.to == ArtifactState::Scanned
&& e.evidence.iter().any(|v| {
v.kind == EvidenceKind::Provenance
&& v.source == provenance::SCANNER_ID
&& v.verdict == EvidenceVerdict::Pass
})
});
if !passed {
return Err(
"no provenance pass from holger-provenance is on the SCANNED record".into(),
);
}
said.push("provenance verified by holger-provenance".into());
}
let pipeline = chain.scanner().ok_or_else(|| {
"this chain runs no scanners, so no vulnerability scan is on record".to_string()
})?;
let ev = match pipeline.stored(sha256) {
Ok(crate::scanner::Stored::Fresh { evidence, .. })
| Ok(crate::scanner::Stored::Stale { evidence, .. }) => evidence,
_ => return Err("no vulnerability scan is on record".into()),
};
let vulns: Vec<&crate::scanner::ScanEvidence> = ev
.records_of(crate::scanner::ScannerKind::Vulnerability)
.filter(|r| !r.is_not_configured() && r.verdict != crate::scanner::ScanVerdict::Error)
.collect();
if vulns.is_empty() {
return Err("no vulnerability scan reported on these bytes".into());
}
let worst = vulns
.iter()
.flat_map(|r| r.findings.iter().map(|f| f.severity))
.max();
if let Some(w) = worst {
if w > cve_max {
return Err(format!(
"the vulnerability scan found a {w:?} advisory, above cve_max {cve_max:?}"
));
}
}
said.push(format!(
"highest advisory severity: {}",
worst
.map(|w| format!("{w:?}").to_lowercase())
.unwrap_or_else(|| "none".into())
));
Ok(said.join("; "))
}
}
/// The inspection's Windows-launcher finding (holger-inspect `WINDOWS_LAUNCHER_FINDING`), held
/// here so a build without the scanners (a downstream reads carried evidence) needs no
/// holger-inspect; `the_launcher_finding_is_holger_inspects` keeps the two equal.
pub(crate) const WINDOWS_LAUNCHER_FINDING: &str = "Windows script launcher";
/// **Does a WARN on this record excuse what the floor would block?** (SEC-REVIEW 2026-11-01
/// W1): a Windows-launcher WARN is a native executable the policy lets through for a person to
/// look at; the robot never approves past it.
pub(crate) fn excused_block(evidence: &[crate::promote::intake::Evidence]) -> Option<String> {
use crate::promote::intake::{EvidenceKind, EvidenceVerdict};
evidence
.iter()
.find(|e| {
e.kind == EvidenceKind::Malware
&& e.verdict == EvidenceVerdict::Warn
&& e.reference.contains(WINDOWS_LAUNCHER_FINDING)
})
.map(|e| {
format!(
"the inspection excused a native executable as a Windows launcher ({}); a person approves it",
e.reference
)
})
}
#[cfg(test)]
mod excused_block_tests {
use super::excused_block;
use crate::promote::intake::{Evidence, EvidenceKind, EvidenceVerdict};
/// W1: the launcher WARN stops auto-approve; any other WARN does not.
#[cfg(feature = "scanners")]
#[test]
fn the_launcher_finding_is_holger_inspects() {
assert_eq!(
super::WINDOWS_LAUNCHER_FINDING,
holger_inspect::WINDOWS_LAUNCHER_FINDING
);
}
#[test]
fn a_launcher_warn_stops_the_robot() {
let launcher = Evidence::new(
EvidenceKind::Malware,
"holger-inspect",
EvidenceVerdict::Warn,
)
.with_reference(format!(
"warn executable demo/cli-64.exe: {} (PE .exe, 14336 bytes)",
super::WINDOWS_LAUNCHER_FINDING
));
let script = Evidence::new(
EvidenceKind::Malware,
"holger-inspect",
EvidenceVerdict::Warn,
)
.with_reference("warn script demo/run: #! script outside its script directories");
assert!(excused_block(&[script.clone(), launcher]).is_some());
assert!(excused_block(&[script]).is_none());
}
}
/// The automatic approver: its own principal, never the importer.
/// **A container image is NEVER auto-approved** (A29 row 11, owner): whatever the floor's
/// `allow_auto_approve` says, an OCI pull waits for the persons the approval rule names
/// (two in a gateway). `Some(why)` for a kind that may not take an automatic vote.
pub fn never_auto_approved(kind: RemoteKind, source: &str) -> Option<String> {
(kind == RemoteKind::Oci).then(|| {
format!(
"source '{source}' is an OCI registry, and a container image is never auto-approved (owner, A29 row 11): \
every image waits for its approvers, whatever `allow_auto_approve` says"
)
})
}
fn auto_approver(source: &str) -> crate::promote::intake::Approver {
crate::promote::intake::Approver {
principal: auto_approver_of(source),
method: "auto-approve".into(),
display: None,
}
}
fn last_reason(chain: &IntakeChain, id: &ArtifactId) -> String {
chain
.history(id)
.last()
.map(|e| {
if e.reason.is_empty() {
e.policy.clone()
} else {
e.reason.clone()
}
})
.unwrap_or_default()
}
/// The fetcher of a build with no outbound client: refuses by name.
#[cfg(not(feature = "online"))]
struct NoFetch;
#[cfg(not(feature = "online"))]
impl http::Fetch for NoFetch {
fn get(
&self,
url: &str,
_: &[(String, String)],
_: Option<(&str, &str)>,
_: u64,
) -> anyhow::Result<http::Reply> {
anyhow::bail!("GET {url}: this build has no outbound client (`online` is off)")
}
}
#[cfg(test)]
mod tests {
use super::*;
fn settings(r: &str) -> PullThroughSettings {
ron::from_str(r).unwrap_or_else(|e| panic!("{r}: {e}"))
}
#[test]
fn auto_approval_is_off_by_default_and_needs_the_floor() {
let s = settings(
r#"(sources: [(name: "pypi", kind: python, url: "https://pypi.org/simple/")])"#,
);
assert_eq!(s.sources[0].auto_approve, AutoApprove::Off);
assert!(!s.rules.allow_auto_approve);
let asks = settings(
r#"(sources: [(name: "pypi", kind: python, url: "https://pypi.org/simple/", auto_approve: when(provenance: require, cve_max: high))])"#,
);
let e = build_sources(&asks).err().expect("refused").to_string();
assert!(e.contains("allow_auto_approve: false"), "{e}");
let both = settings(
r#"(sources: [(name: "cgr", kind: oci, url: "https://cgr.dev", auto_approve: when(provenance: require, cve_max: high))],
rules: (allow_auto_approve: true))"#,
);
// A29 row 11 (owner): a container image is NEVER auto-approved, whatever the floor
// says. The config that asks for it is refused by name.
let e = build_sources(&both)
.err()
.expect("an OCI source never auto-approves")
.to_string();
assert!(
e.contains("never auto-approved") && e.contains("cgr"),
"{e}"
);
let pypi = settings(
r#"(sources: [(name: "pypi", kind: python, url: "https://pypi.org/simple/", auto_approve: when(provenance: off, cve_max: high))],
rules: (allow_auto_approve: true))"#,
);
if cfg!(feature = "online") {
assert!(
build_sources(&pypi).is_ok(),
"the other kinds keep the floor's rule"
);
}
}
#[test]
fn a_container_image_never_takes_an_automatic_vote() {
let why = never_auto_approved(RemoteKind::Oci, "cgr").expect("OCI is never auto-approved");
assert!(
why.contains("never auto-approved") && why.contains("cgr"),
"{why}"
);
for k in [
RemoteKind::Python,
RemoteKind::Npm,
RemoteKind::Maven,
RemoteKind::Crates,
] {
assert!(
never_auto_approved(k, "x").is_none(),
"{k:?} keeps the floor's rule"
);
}
}
#[test]
fn a_gate_may_only_tighten_the_floor() {
if !cfg!(feature = "online") {
return;
}
let loose = settings(
r#"(sources: [(name: "cgr", kind: oci, url: "https://cgr.dev", gates: (provenance: Some(off)))], rules: (require_signed: true))"#,
);
assert!(build_sources(&loose)
.err()
.unwrap()
.to_string()
.contains("loosens the floor"));
let cve = settings(
r#"(sources: [(name: "pypi", kind: python, url: "https://pypi.org/simple/", gates: (cve: Some((max: critical))))])"#,
);
assert!(build_sources(&cve)
.err()
.unwrap()
.to_string()
.contains("above the floor"));
let tight = settings(
r#"(sources: [(name: "cgr", kind: oci, url: "https://cgr.dev", gates: (provenance: Some(require(identity: Some((issuer: "https://token.actions.githubusercontent.com", subject: Some("s"))))), cve: Some((max: medium))))],
rules: (require_signed: true))"#,
);
let s = build_sources(&tight).unwrap();
assert_eq!(s[0].demand.provenance, Mode::Require);
assert_eq!(s[0].demand.signed, Mode::Require);
assert_eq!(s[0].cve_max, FindingSeverity::Medium);
}
/// T9b: OpenPGP is opt-in (`maven-pgp`; `online` carries it). A source that pins pgp
/// keys in a build without it is refused at load by name, never run unpinned.
#[test]
fn a_pgp_pin_without_the_feature_is_refused_by_name() {
let s = settings(
r#"(sources: [(name: "central", kind: maven, url: "https://repo1.maven.org/maven2/", gates: (provenance: Some(require(pgp_keys: ["povirk"]))))])"#,
);
let r = build_sources(&s);
if holger_provenance::pgp::BUILT {
if cfg!(feature = "online") {
assert_eq!(r.unwrap()[0].demand.pgp_keys, vec!["povirk".to_string()]);
}
} else {
let e = r
.err()
.expect("a pgp pin without the feature is refused")
.to_string();
assert!(e.contains("[provenance-pgp-not-built]"), "{e}");
assert!(
e.contains("'central'") && e.contains("`maven-pgp`"),
"it names the source and the feature: {e}"
);
}
// A source with no pgp pin is not touched by it.
let plain = settings(
r#"(sources: [(name: "central", kind: maven, url: "https://repo1.maven.org/maven2/")])"#,
);
if let Err(e) = build_sources(&plain) {
assert!(!e.to_string().contains("provenance-pgp-not-built"), "{e}");
}
}
#[test]
fn coordinates_name_the_source() {
assert_eq!(
pip_id("pypi", "a-1.0-py3-none-any.whl").name,
"pypi/a-1.0-py3-none-any.whl"
);
let d = format!("sha256:{}", "a".repeat(64));
assert_eq!(
oci_id("cgr", "chainguard/static", &d).name,
format!("cgr/chainguard/static/sha256-{}.oci.tar", "a".repeat(64))
);
assert_eq!(actor_of("cgr"), "pull-through:cgr");
assert_eq!(auto_approver_of("cgr"), "auto-approve:cgr");
}
}