server/lib/tests/pull_through_runs_the_intake_chain.rs
//! **A package Holger FETCHES goes through the same intake chain as an upload**
//! (owner order 2026-10-23, `server/lib/src/pull_intake`).
//!
//! Each test runs one real Holger (four `raw` intake repositories, static-token
//! principals with duties, the job queue) in front of a local FAKE upstream
//! that serves REAL bytes fetched once from the internet and kept under
//! `tests/fixtures/pull_through/` (fetched 2026-10-23):
//!
//! * `pypi/`: PyPI's PEP 691 page for `sampleproject` and the 4.0.0 wheel and
//! sdist with their PEP 740 attestations (`pypi.org/integrity/...`);
//! * `oci/chainguard/static/`: `cgr.dev/chainguard/static:latest`, the index
//! `sha256:17367e4d…` with both platform manifests, every config and layer
//! blob, and its cosign `.sig` and `.att` manifests with their layers.
//!
//! The fake upstream speaks what the real ones speak: PEP 691 JSON with
//! `hashes` and `provenance`, the registry's bearer-token challenge, the
//! `sha256-<digest>.sig|.att` tags. What the tests measure:
//!
//! * the first request is answered "not yet" (pip: 503 + Retry-After; OCI: the
//! registry error `UNAVAILABLE`), the package lands INGESTED as
//! `pull-through:<source>` and walks SCANNED and POLICY_CHECKED;
//! * without auto-approval it waits for a person (404 naming the state), and
//! after the approval a request is served from the SEALED store;
//! * a planted EICAR wheel is refused at the door, never stored, never served;
//! * auto-approval on (one vote by `auto-approve:<source>`; under
//! `approvals: 2` a person casts the second) and off (the default);
//! * an OCI image is pulled whole with its cosign signature and attestations,
//! which reach the provenance verifier; the default verifier (G3a's
//! `holger-provenance` is not in this build yet) fails CLOSED; an unsigned
//! image is refused by the floor `require_signed: true`.
// A server's test, never a gateway build's (lane HOL-CLEANUP): a gateway build is a
// downstream, and a pull-through fetches from its sources, which a downstream refuses (holger-downstream-1).
#![cfg(not(feature = "gateway"))]
/// The scan sandbox this run uses (lane SANDBOX-SCAN). `HOLGER_TEST_SANDBOX=required`
/// runs every scan in the real container: the worker image minted
/// (`HOLGER_SCAN_WORKER_IMAGE`, default `localhost/holger-scan-worker:<version>`),
/// podman's API socket at `HOLGER_TEST_SANDBOX_URL` and a job directory both sides
/// see at `HOLGER_TEST_SANDBOX_WORKDIR`. Otherwise the worker's own code runs in
/// this process (`sandbox: off`, a test build).
fn sandbox_ron() -> String {
match std::env::var("HOLGER_TEST_SANDBOX").as_deref() {
Ok("required") => {
let image = std::env::var("HOLGER_SCAN_WORKER_IMAGE").unwrap_or_else(|_| {
concat!("localhost/holger-scan-worker:", env!("CARGO_PKG_VERSION")).into()
});
let url = std::env::var("HOLGER_TEST_SANDBOX_URL")
.map(|u| format!("url: Some(\"{u}\"), "))
.unwrap_or_default();
let work = std::env::var("HOLGER_TEST_SANDBOX_WORKDIR")
.map(|w| format!("work_dir: Some(\"{w}\"), "))
.unwrap_or_default();
format!("rules: (sandbox: required), sandbox: Some(({url}{work}image: \"{image}\")),")
}
_ => "rules: (sandbox: off),".into(),
}
}
use std::collections::BTreeMap;
use std::io::{Read, Write};
use std::net::{TcpListener, TcpStream};
use std::path::Path;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use holger_server_lib::promote::intake::{
ArtifactState, EvidenceKind, EvidenceVerdict, IntakeChain, TransitionRequest,
};
use holger_server_lib::pull_intake::provenance::{Demand, Material, Outcome, Verdict, Verifier};
use holger_server_lib::pull_intake::{oci_id, pip_id};
use holger_server_lib::{read_ron_config, wire_holger, Holger};
use sha2::Digest as _;
use traits::test_support::hold_port;
const FX: &str = concat!(env!("CARGO_MANIFEST_DIR"), "/tests/fixtures/pull_through");
const WHEEL: &str = "sampleproject-4.0.0-py3-none-any.whl";
const SDIST: &str = "sampleproject-4.0.0.tar.gz";
const INDEX: &str = "sha256:17367e4d3ae2b46c3290858e483fac09be085c92b372360fe3bf6a6150848fde";
const AMD64: &str = "sha256:483300cd4680af8b3388dd6883a9fb09e9d9c1f700f59e8eac8d0870eb3b04fd";
const CHAINGUARD_ISSUER: &str = "https://token.actions.githubusercontent.com";
const CHAINGUARD_SUBJECT: &str =
"https://github.com/chainguard-images/images/.github/workflows/release.yaml@refs/heads/main";
fn sha256_hex(b: &[u8]) -> String {
hex::encode(sha2::Sha256::digest(b))
}
fn token(who: &str) -> String {
format!("{who}-token-of-sufficient-length-for-holger")
}
// ── the fake upstream ───────────────────────────────────────────────────────
struct Upstream {
base: String,
hits: Arc<Mutex<BTreeMap<String, usize>>>,
}
impl Upstream {
fn hits(&self, path: &str) -> usize {
self.hits.lock().unwrap().get(path).copied().unwrap_or(0)
}
}
struct Reply {
status: u16,
headers: Vec<(String, String)>,
body: Vec<u8>,
}
fn reply(status: u16, ctype: &str, body: Vec<u8>) -> Reply {
Reply {
status,
headers: vec![("Content-Type".into(), ctype.into())],
body,
}
}
fn fixture(p: &str) -> Option<Vec<u8>> {
std::fs::read(Path::new(FX).join(p)).ok()
}
/// A stored-only zip wheel with `members`, METADATA and WHEEL.
fn wheel_bytes(name: &str, version: &str, members: &[(&str, &[u8])]) -> Vec<u8> {
fn crc32(data: &[u8]) -> u32 {
let mut crc = 0xFFFF_FFFFu32;
for b in data {
crc ^= *b as u32;
for _ in 0..8 {
crc = if crc & 1 != 0 {
(crc >> 1) ^ 0xEDB8_8320
} else {
crc >> 1
};
}
}
!crc
}
let info = format!("{name}-{version}.dist-info");
let mut all: Vec<(String, Vec<u8>)> = members
.iter()
.map(|(n, b)| (n.to_string(), b.to_vec()))
.collect();
all.push((
format!("{info}/METADATA"),
format!(
"Metadata-Version: 2.1\nName: {name}\nVersion: {version}\nLicense-Expression: MIT\n"
)
.into_bytes(),
));
all.push((
format!("{info}/WHEEL"),
b"Wheel-Version: 1.0\nGenerator: holger-test\nRoot-Is-Purelib: true\nTag: py3-none-any\n"
.to_vec(),
));
let (mut out, mut central) = (Vec::new(), Vec::new());
for (n, body) in &all {
let crc = crc32(body);
let offset = out.len() as u32;
out.extend_from_slice(&0x0403_4b50u32.to_le_bytes());
out.extend_from_slice(&[20, 0, 0, 0, 0, 0, 0, 0, 0, 0]);
out.extend_from_slice(&crc.to_le_bytes());
out.extend_from_slice(&(body.len() as u32).to_le_bytes());
out.extend_from_slice(&(body.len() as u32).to_le_bytes());
out.extend_from_slice(&(n.len() as u16).to_le_bytes());
out.extend_from_slice(&0u16.to_le_bytes());
out.extend_from_slice(n.as_bytes());
out.extend_from_slice(body);
central.extend_from_slice(&0x0201_4b50u32.to_le_bytes());
central.extend_from_slice(&[20, 0, 20, 0, 0, 0, 0, 0, 0, 0, 0, 0]);
central.extend_from_slice(&crc.to_le_bytes());
central.extend_from_slice(&(body.len() as u32).to_le_bytes());
central.extend_from_slice(&(body.len() as u32).to_le_bytes());
central.extend_from_slice(&(n.len() as u16).to_le_bytes());
central.extend_from_slice(&[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]);
central.extend_from_slice(&offset.to_le_bytes());
central.extend_from_slice(n.as_bytes());
}
let start = out.len() as u32;
out.extend_from_slice(¢ral);
out.extend_from_slice(&0x0605_4b50u32.to_le_bytes());
out.extend_from_slice(&[0, 0, 0, 0]);
out.extend_from_slice(&(all.len() as u16).to_le_bytes());
out.extend_from_slice(&(all.len() as u16).to_le_bytes());
out.extend_from_slice(&(central.len() as u32).to_le_bytes());
out.extend_from_slice(&start.to_le_bytes());
out.extend_from_slice(&[0, 0]);
out
}
fn eicar() -> Vec<u8> {
// Two halves, so this source file is not itself the test string.
format!(
"{}{}",
"X5O!P%@AP[4\\PZX54(P^)7CC)7}$", "EICAR-STANDARD-ANTIVIRUS-TEST-FILE!$H+H*"
)
.into_bytes()
}
fn evil_wheel() -> Vec<u8> {
wheel_bytes(
"evil_pkg",
"1.0.0",
&[
("evil_pkg/__init__.py", b""),
("evil_pkg/eicar.com", &eicar()),
],
)
}
fn pip_page(base: &str, project: &str) -> Option<Vec<u8>> {
match project {
"sampleproject" => {
let mut v: serde_json::Value =
serde_json::from_slice(&fixture("pypi/sampleproject.simple.json")?).ok()?;
for f in v.get_mut("files")?.as_array_mut()? {
let name = f["filename"].as_str()?.to_string();
f["url"] = format!("{base}/files/{name}").into();
if f.get("provenance").is_some_and(|p| !p.is_null()) {
f["provenance"] = format!("{base}/integrity/{name}").into();
}
}
Some(serde_json::to_vec(&v).ok()?)
}
"evil-pkg" => {
let w = evil_wheel();
Some(
serde_json::json!({
"meta": {"api-version": "1.1"}, "name": "evil-pkg",
"files": [{"filename": "evil_pkg-1.0.0-py3-none-any.whl", "url": format!("{base}/files/evil_pkg-1.0.0-py3-none-any.whl"),
"hashes": {"sha256": sha256_hex(&w)}}]
})
.to_string()
.into_bytes(),
)
}
_ => None,
}
}
fn oci_reply(path: &str, auth: Option<&str>, base: &str) -> Reply {
if auth != Some("Bearer t0k-for-the-fake") {
return Reply {
status: 401,
headers: vec![
("Content-Type".into(), "application/json".into()),
(
"WWW-Authenticate".into(),
format!("Bearer realm=\"{base}/token\",service=\"fake-registry\""),
),
],
body: br#"{"errors":[{"code":"UNAUTHORIZED","message":"authentication required"}]}"#
.to_vec(),
};
}
let rest = path.trim_start_matches("/v2/");
let (image, signed) = if let Some(r) = rest.strip_prefix("chainguard/static/") {
(r, true)
} else if let Some(r) = rest.strip_prefix("unsigned/static/") {
(r, false)
} else {
return reply(
404,
"application/json",
br#"{"errors":[{"code":"NAME_UNKNOWN"}]}"#.to_vec(),
);
};
let dir = "oci/chainguard/static";
// HOL-REFERRERS: the signed image also has an OCI 1.1 referrer, an SPDX SBOM
// attached the way `oras attach` does it; the unsigned one has none.
let (sbom_manifest, sbom_layer) = sbom_referrer();
if image == format!("referrers/{INDEX}") && signed {
let desc = serde_json::json!({
"mediaType": "application/vnd.oci.image.manifest.v1+json", "digest": format!("sha256:{}", sha256_hex(&sbom_manifest)),
"size": sbom_manifest.len(), "artifactType": SPDX,
});
let body = serde_json::json!({ "schemaVersion": 2, "mediaType": "application/vnd.oci.image.index.v1+json", "manifests": [desc] });
return reply(
200,
"application/vnd.oci.image.index.v1+json",
body.to_string().into_bytes(),
);
}
if image == format!("manifests/sha256:{}", sha256_hex(&sbom_manifest)) && signed {
return reply(
200,
"application/vnd.oci.image.manifest.v1+json",
sbom_manifest,
);
}
if image == format!("blobs/sha256:{}", sha256_hex(&sbom_layer)) {
return reply(200, "application/octet-stream", sbom_layer);
}
if image == format!("blobs/sha256:{}", sha256_hex(b"{}")) {
return reply(200, "application/octet-stream", b"{}".to_vec());
}
if let Some(r) = image.strip_prefix("manifests/") {
if !signed && (r.ends_with(".sig") || r.ends_with(".att")) {
return reply(
404,
"application/json",
br#"{"errors":[{"code":"MANIFEST_UNKNOWN"}]}"#.to_vec(),
);
}
return match fixture(&format!("{dir}/manifests/{}", r.replace(':', "_"))) {
Some(b) => {
let mt = serde_json::from_slice::<serde_json::Value>(&b)
.ok()
.and_then(|v| v["mediaType"].as_str().map(str::to_string))
.unwrap_or_default();
let mut rp = reply(200, &mt, b.clone());
rp.headers.push((
"Docker-Content-Digest".into(),
format!("sha256:{}", sha256_hex(&b)),
));
rp
}
None => reply(
404,
"application/json",
br#"{"errors":[{"code":"MANIFEST_UNKNOWN"}]}"#.to_vec(),
),
};
}
if let Some(d) = image.strip_prefix("blobs/") {
return match fixture(&format!("{dir}/blobs/{}", d.replace(':', "_"))) {
Some(b) => reply(200, "application/octet-stream", b),
None => reply(
404,
"application/json",
br#"{"errors":[{"code":"BLOB_UNKNOWN"}]}"#.to_vec(),
),
};
}
reply(
404,
"application/json",
br#"{"errors":[{"code":"NOT_FOUND"}]}"#.to_vec(),
)
}
const SPDX: &str = "application/spdx+json";
/// An SPDX SBOM as an OCI 1.1 artifact whose subject is the image index: `(manifest, layer)`.
fn sbom_referrer() -> (Vec<u8>, Vec<u8>) {
let layer = br#"{"spdxVersion":"SPDX-2.3","name":"chainguard/static","packages":[]}"#.to_vec();
let index_len = fixture(&format!(
"oci/chainguard/static/manifests/{}",
INDEX.replace(':', "_")
))
.map(|b| b.len())
.unwrap_or(0);
let manifest = serde_json::json!({
"schemaVersion": 2,
"mediaType": "application/vnd.oci.image.manifest.v1+json",
"artifactType": SPDX,
"config": { "mediaType": "application/vnd.oci.empty.v1+json", "digest": format!("sha256:{}", sha256_hex(b"{}")), "size": 2 },
"layers": [ { "mediaType": SPDX, "digest": format!("sha256:{}", sha256_hex(&layer)), "size": layer.len() } ],
"subject": { "mediaType": "application/vnd.oci.image.index.v1+json", "digest": INDEX, "size": index_len },
"annotations": { "org.opencontainers.image.created": "2026-10-30T00:00:00Z" },
});
(serde_json::to_vec(&manifest).unwrap(), layer)
}
fn serve_one(mut s: TcpStream, base: &str, hits: &Mutex<BTreeMap<String, usize>>) {
let _ = s.set_read_timeout(Some(Duration::from_secs(10)));
let mut buf = Vec::new();
let mut chunk = [0u8; 4096];
while !buf.windows(4).any(|w| w == b"\r\n\r\n") {
match s.read(&mut chunk) {
Ok(0) | Err(_) => return,
Ok(n) => buf.extend_from_slice(&chunk[..n]),
}
}
let head = String::from_utf8_lossy(&buf).to_string();
let mut lines = head.lines();
let first = lines.next().unwrap_or("");
let path = first.split_whitespace().nth(1).unwrap_or("/").to_string();
let auth = lines
.find_map(|l| {
l.strip_prefix("authorization: ")
.or_else(|| l.strip_prefix("Authorization: "))
})
.map(str::to_string);
let bare = path.split('?').next().unwrap_or("").to_string();
*hits.lock().unwrap().entry(bare.clone()).or_default() += 1;
let r = if let Some(p) = bare.strip_prefix("/simple/") {
match pip_page(base, p.trim_matches('/')) {
Some(b) => reply(200, "application/vnd.pypi.simple.v1+json", b),
None => reply(404, "text/plain", b"no such project".to_vec()),
}
} else if let Some(f) = bare.strip_prefix("/files/") {
if f == "evil_pkg-1.0.0-py3-none-any.whl" {
reply(200, "application/octet-stream", evil_wheel())
} else {
match fixture(&format!("pypi/files/{f}")) {
Some(b) => reply(200, "application/octet-stream", b),
None => reply(404, "text/plain", Vec::new()),
}
}
} else if let Some(f) = bare.strip_prefix("/integrity/") {
match fixture(&format!("pypi/files/{f}.provenance.json")) {
Some(b) => reply(200, "application/vnd.pypi.integrity.v1+json", b),
None => reply(404, "text/plain", Vec::new()),
}
} else if bare == "/token" {
reply(
200,
"application/json",
br#"{"token":"t0k-for-the-fake"}"#.to_vec(),
)
} else if bare.starts_with("/v2/") {
oci_reply(&bare, auth.as_deref(), base)
} else {
reply(404, "text/plain", Vec::new())
};
let mut out = format!(
"HTTP/1.1 {} X\r\nContent-Length: {}\r\nConnection: close\r\n",
r.status,
r.body.len()
);
for (k, v) in &r.headers {
out.push_str(&format!("{k}: {v}\r\n"));
}
out.push_str("\r\n");
let _ = s.write_all(out.as_bytes());
let _ = s.write_all(&r.body);
}
fn upstream() -> Upstream {
let l = TcpListener::bind("127.0.0.1:0").unwrap();
let base = format!("http://{}", l.local_addr().unwrap());
let hits: Arc<Mutex<BTreeMap<String, usize>>> = Arc::default();
let (b2, h2) = (base.clone(), Arc::clone(&hits));
std::thread::spawn(move || {
for s in l.incoming().flatten() {
let (b3, h3) = (b2.clone(), Arc::clone(&h2));
std::thread::spawn(move || serve_one(s, &b3, &h3));
}
});
Upstream { base, hits }
}
// ── one Holger ──────────────────────────────────────────────────────────────
struct Running {
holger: Holger,
_rt: tokio::runtime::Runtime,
_data: tempfile::TempDir,
_cfg: tempfile::NamedTempFile,
http: String,
}
impl Running {
fn chain(&self) -> Arc<IntakeChain> {
self.holger.intake().expect("the intake chain")
}
}
/// The `promotion.intake.scan` every pull-through here names. A pull-through never feeds a
/// chain that runs no vulnerability scan (HOL-DOGFOOD-3, `holger-intake-21`); only the test
/// build's `sandbox: off` is exempt, so `scan: None` would be refused under
/// `HOLGER_TEST_SANDBOX=required`.
const VULN: &str = "Some((required: [vulnerability]))";
/// `pull_through` is the RON block's body; `approvals` the quorum; `scan` the
/// `promotion.intake.scan` value.
fn start(pull_through: &str, approvals: u32, scan: &str) -> Running {
start_with(pull_through, approvals, scan, "")
}
/// [`start`] with `extra` top-level blocks in the config (an `intel:`).
fn start_with(pull_through: &str, approvals: u32, scan: &str, extra: &str) -> Running {
start_inspecting(pull_through, approvals, scan, extra, "")
}
/// [`start_with`] with more `inspect:` engines beside the YARA rules (a `clamav:`).
fn start_inspecting(
pull_through: &str,
approvals: u32,
scan: &str,
extra: &str,
inspect: &str,
) -> Running {
let mut last = String::new();
for _ in 0..8 {
let (grpc, http) = (hold_port(), hold_port());
let extra = format!("{extra}\u{0}{inspect}");
match try_start(grpc, http, pull_through, approvals, scan, &extra) {
Ok(r) => return r,
Err(e) => last = e,
}
}
panic!("server never started; last error: {last}");
}
fn try_start(
grpc: u16,
http: u16,
pull_through: &str,
approvals: u32,
scan: &str,
extra: &str,
) -> Result<Running, String> {
// `extra` is `<top-level blocks>[\0<inspect engines>]` (see `start_inspecting`).
let (extra, inspect_more) = extra.split_once('\u{0}').unwrap_or((extra, ""));
let data = tempfile::tempdir().unwrap();
let d = data.path().display().to_string();
let yara = data.path().join("yara");
std::fs::create_dir_all(&yara).unwrap();
std::fs::write(yara.join("eicar.yar"), "rule EICAR_Test_File { strings: $a = \"EICAR-STANDARD-ANTIVIRUS-TEST-FILE\" condition: $a }").unwrap();
let repo = |name: &str| {
format!(
r#"(ron_name: "{name}", ron_repo_type: "raw", ron_upstreams: [], ron_in: None,
ron_out: Some((ron_storage_endpoint: "storage", ron_exposed_endpoint: "door")),
ron_data_dir: Some("{d}/{name}")),"#
)
};
let methods: Vec<String> = ["ida", "bob", "carol"]
.iter()
.map(|p| {
format!(
r#"StaticToken(subject: "{p}", token: Some("{}"))"#,
token(p)
)
})
.collect();
let sandbox = sandbox_ron();
let ron = format!(
r#"(
repositories: [ {incoming} {quarantine} {approved} {production} ],
exposed_endpoints: [ (ron_name: "door", ron_url: "127.0.0.1:{grpc}", ron_http_url: Some("127.0.0.1:{http}")) ],
storage_endpoints: [ (ron_name: "storage", ron_storage_type: "znippy", ron_path: "{d}") ],
audit: (dir: "{d}/audit", enabled: true),
auth: Some((
methods: [ {methods} ],
duties: {{ "ida": [Importer], "bob": [SecurityApprover], "carol": [SecurityApprover] }},
)),
promotion: Some((
ladder_name: "intake",
stages: [
(name: "incoming", repo: "incoming"), (name: "quarantine", repo: "quarantine"),
(name: "approved", repo: "approved"), (name: "production", repo: "production"),
],
intake: Some((
incoming: "incoming", quarantine: "quarantine", approved: "approved", production: "production",
{sandbox}
approvals: {approvals},
scan: {scan},
inspect: (yara: Some((rules: Some("{yara}"), required: true)), {inspect_more}),
)),
)),
pull_through: Some(({pull_through})),
{extra}
)"#,
incoming = repo("incoming"),
quarantine = repo("quarantine"),
approved = repo("approved"),
// HCON-5: an estate build gates reads; the repository clients read is public (as the
// ONE box feeds an estate's `public_repositories`), the intake stages stay gated.
production =
repo("production").replacen("ron_in: None,", "ron_in: None, visibility: Public,", 1),
methods = methods.join(", "),
yara = yara.display(),
);
let mut cfg = tempfile::NamedTempFile::new().unwrap();
cfg.write_all(ron.as_bytes()).unwrap();
let mut holger: Holger = read_ron_config(cfg.path()).map_err(|e| format!("config: {e}"))?;
holger
.instantiate_backends()
.map_err(|e| format!("backends: {e}"))?;
wire_holger(&mut holger).map_err(|e| format!("wire: {e:#}"))?;
let rt = tokio::runtime::Builder::new_multi_thread()
.worker_threads(4)
.enable_all()
.build()
.unwrap();
{
let _in = rt.enter();
holger.start().map_err(|e| format!("start: {e:#}"))?;
}
let r = Running {
holger,
_rt: rt,
_data: data,
_cfg: cfg,
http: format!("127.0.0.1:{http}"),
};
for _ in 0..200 {
if get(&r.http, "/healthz", &[]).status == 200 {
return Ok(r);
}
std::thread::sleep(Duration::from_millis(20));
}
Err("never healthy".into())
}
struct Got {
status: u16,
/// The status line's reason phrase (`Forbidden`, or the refusal Holger puts there).
reason: String,
headers: BTreeMap<String, String>,
body: Vec<u8>,
}
impl Got {
fn text(&self) -> String {
String::from_utf8_lossy(&self.body).to_string()
}
}
/// A plain HTTP/1.1 GET to the Holger door.
fn get(addr: &str, path: &str, headers: &[(&str, &str)]) -> Got {
let Ok(mut s) = TcpStream::connect(addr) else {
return Got {
status: 0,
reason: String::new(),
headers: BTreeMap::new(),
body: Vec::new(),
};
};
let mut req = format!("GET {path} HTTP/1.1\r\nHost: {addr}\r\nConnection: close\r\n");
for (k, v) in headers {
req.push_str(&format!("{k}: {v}\r\n"));
}
req.push_str("\r\n");
s.write_all(req.as_bytes()).unwrap();
let mut all = Vec::new();
let _ = s.read_to_end(&mut all);
let split = all
.windows(4)
.position(|w| w == b"\r\n\r\n")
.map(|i| i + 4)
.unwrap_or(all.len());
let head = String::from_utf8_lossy(&all[..split]).to_string();
let status = head
.split_whitespace()
.nth(1)
.and_then(|s| s.parse().ok())
.unwrap_or(0);
let headers = head
.lines()
.skip(1)
.filter_map(|l| {
l.split_once(':')
.map(|(k, v)| (k.trim().to_ascii_lowercase(), v.trim().to_string()))
})
.collect();
let reason = head
.lines()
.next()
.and_then(|l| l.splitn(3, ' ').nth(2))
.unwrap_or_default()
.to_string();
Got {
status,
reason,
headers,
body: all[split..].to_vec(),
}
}
fn wait_state(chain: &IntakeChain, id: &traits::ArtifactId, want: ArtifactState) {
let t = Instant::now();
while t.elapsed() < Duration::from_secs(60) {
if chain.state(id) == want {
return;
}
std::thread::sleep(Duration::from_millis(50));
}
let jobs = chain.pull().map(|p| p.debug_records()).unwrap_or_default();
panic!(
"{} never reached {want}; it stands at {} ({:#?}); pull records: {jobs}",
id.name,
chain.state(id),
chain.history(id)
);
}
/// A GET once the pull job is done. The chain marks the artifact (Blocked, say) before
/// the pull job itself ends, and while the job still runs the door answers 503; under
/// load that window is seen (ODEN-CI round 2: a 503 where the refusal's 403 was due).
/// Bounded, as [`wait_state`].
fn get_settled(addr: &str, path: &str, accept: &[(&str, &str)]) -> Got {
let t = Instant::now();
loop {
let g = get(addr, path, accept);
if g.status != 503 || t.elapsed() > Duration::from_secs(60) {
return g;
}
std::thread::sleep(Duration::from_millis(50));
}
}
fn wait_for<T>(what: &str, mut f: impl FnMut() -> Option<T>) -> T {
let t = Instant::now();
while t.elapsed() < Duration::from_secs(60) {
if let Some(v) = f() {
return v;
}
std::thread::sleep(Duration::from_millis(50));
}
panic!("never: {what}");
}
fn approve(chain: &IntakeChain, id: &traits::ArtifactId, who: &str) {
chain
.advance(
TransitionRequest::new(id.clone(), ArtifactState::Approved, who)
.with_reason("reviewed"),
)
.unwrap_or_else(|e| panic!("{who} approves {}: {e}", id.name));
}
fn pypi(up: &Upstream, extra: &str) -> String {
format!(
r#"sources: [ (name: "pypi", kind: python, url: "{}/simple/"{extra}) ], retry_after_secs: Some(1),"#,
up.base
)
}
const PEP691: (&str, &str) = ("Accept", "application/vnd.pypi.simple.v1+json");
// ── pip ─────────────────────────────────────────────────────────────────────
#[test]
fn a_pip_pull_is_not_yet_then_waits_for_a_person_then_serves_from_the_sealed_store() {
let up = upstream();
let r = start(&pypi(&up, ""), 1, "Some((required: [vulnerability]))");
let chain = r.chain();
// 1. The first request: not yet, and a pull is on its way.
let first = get(&r.http, "/production/simple/sampleproject/", &[PEP691]);
assert_eq!(first.status, 503, "{}", first.text());
assert!(
first.headers.contains_key("retry-after"),
"{:?}",
first.headers
);
assert!(
first.text().contains("being pulled into the intake chain"),
"{}",
first.text()
);
// 2. The newest release's wheel and sdist land INGESTED as pull-through:pypi
// and walk to POLICY_CHECKED; nobody approves them by default.
let (w, s) = (pip_id("pypi", WHEEL), pip_id("pypi", SDIST));
wait_state(&chain, &w, ArtifactState::PolicyChecked);
wait_state(&chain, &s, ArtifactState::PolicyChecked);
for id in [&w, &s] {
let h = chain.history(id);
let states: Vec<ArtifactState> = h.iter().map(|e| e.to).collect();
assert_eq!(
states,
[
ArtifactState::Ingested,
ArtifactState::Scanned,
ArtifactState::PolicyChecked
],
"{h:#?}"
);
assert!(h.iter().all(|e| e.actor == "pull-through:pypi"), "{h:#?}");
// The door's own inspection ran at ingest, and the scanners at SCANNED.
assert!(
h[0].evidence
.iter()
.any(|e| e.kind == EvidenceKind::Malware && e.source == "holger-inspect"),
"{:?}",
h[0].evidence
);
assert!(
h[1].evidence.iter().any(|e| e.kind == EvidenceKind::Scan),
"{:?}",
h[1].evidence
);
}
assert_eq!(
chain.history(&w)[0].sha256,
sha256_hex(&fixture(&format!("pypi/files/{WHEEL}")).unwrap())
);
// Older releases are not pulled.
assert!(chain
.status(&pip_id("pypi", "sampleproject-3.0.0-py3-none-any.whl"))
.is_none());
// 3. While it waits for a person the page and the file say so, and serve nothing.
let held = get_settled(&r.http, "/production/simple/sampleproject/", &[PEP691]);
// Listed, but yanked with the reason (P2): pip installs none of it unpinned.
assert_eq!(held.status, 200, "{}", held.text());
assert!(
held.text().contains("waiting for approval"),
"{}",
held.text()
);
let file = get_settled(&r.http, &format!("/production/pypi/{WHEEL}"), &[]);
assert_eq!(file.status, 404, "{}", file.text());
// 4. A person approves: the next request walks it on and serves it, from the SEALED store.
approve(&chain, &w, "bob");
approve(&chain, &s, "bob");
let page = get_settled(&r.http, "/production/simple/sampleproject/", &[PEP691]);
assert_eq!(page.status, 200, "{}", page.text());
let doc: serde_json::Value = serde_json::from_slice(&page.body).unwrap();
let files: Vec<&str> = doc["files"]
.as_array()
.unwrap()
.iter()
.filter(|f| f.get("yanked").is_none())
.map(|f| f["filename"].as_str().unwrap())
.collect();
assert_eq!(
files.len(),
2,
"the installable files are the AVAILABLE ones: {doc}"
);
assert!(files.contains(&WHEEL) && files.contains(&SDIST), "{doc}");
let whl = get(&r.http, &format!("/production/pypi/{WHEEL}"), &[]);
assert_eq!(whl.status, 200, "{}", whl.text());
assert_eq!(
whl.headers.get("x-holger-intake-state").map(String::as_str),
Some("AVAILABLE")
);
assert_eq!(whl.body, fixture(&format!("pypi/files/{WHEEL}")).unwrap());
let states: Vec<ArtifactState> = chain.history(&w).iter().map(|e| e.to).collect();
assert_eq!(
*states.last().unwrap(),
ArtifactState::Available,
"{states:?}"
);
// 5. The upstream gave each file once; the PEP 740 attestation came with it.
assert_eq!(up.hits(&format!("/files/{WHEEL}")), 1);
assert_eq!(up.hits(&format!("/integrity/{WHEEL}")), 1);
}
#[test]
fn a_planted_eicar_wheel_is_refused_at_the_door_and_never_served() {
let up = upstream();
let r = start(&pypi(&up, ""), 1, "Some((required: [vulnerability]))");
let chain = r.chain();
let first = get(&r.http, "/production/simple/evil-pkg/", &[PEP691]);
assert_eq!(first.status, 503, "{}", first.text());
// The project page lists the refused file yanked, with the reason (lane HOL-310-CHECK:
// pip prints it; before, the page answered 404 and pip said "from versions: none").
let refused = wait_for("the refusal is answered", || {
let g = get(&r.http, "/production/simple/evil-pkg/", &[PEP691]);
(g.status == 200 && g.text().contains("refused by Holger")).then_some(g)
});
assert!(
refused.text().contains("EICAR_Test_File"),
"{}",
refused.text()
);
// Never stored, never recorded as a coordinate, never served.
let id = pip_id("pypi", "evil_pkg-1.0.0-py3-none-any.whl");
assert!(chain.status(&id).is_none(), "{:#?}", chain.history(&id));
let file = get(
&r.http,
"/production/pypi/evil_pkg-1.0.0-py3-none-any.whl",
&[],
);
// A refused file answers 403 with the reason (lane HOL-310-CHECK; it was 404).
assert_eq!(file.status, 403, "{}", file.text());
assert!(file.text().contains("refused"), "{}", file.text());
}
#[test]
fn auto_approval_is_one_vote_and_only_when_the_floor_and_the_source_say_so() {
// On, with one approver: the pull walks to AVAILABLE with nobody touching it.
let up = upstream();
let auto = r#", auto_approve: when(provenance: off, cve_max: high)"#;
let pt = format!("{} rules: (allow_auto_approve: true),", pypi(&up, auto));
let r = start(&pt, 1, "Some((required: [vulnerability]))");
let chain = r.chain();
assert_eq!(
get(&r.http, "/production/simple/sampleproject/", &[PEP691]).status,
503
);
let w = pip_id("pypi", WHEEL);
wait_state(&chain, &w, ArtifactState::Available);
let h = chain.history(&w);
let approved = h
.iter()
.find(|e| e.to == ArtifactState::Approved)
.expect("an APPROVED entry");
assert_eq!(approved.actor, "auto-approve:pypi");
assert!(
approved.reason.contains("auto-approved")
&& approved.reason.contains("allow_auto_approve: true"),
"{}",
approved.reason
);
assert!(approved
.evidence
.iter()
.any(|e| e.kind == EvidenceKind::Approval
&& e.source == "auto-approve:pypi"
&& e.verdict == EvidenceVerdict::Pass));
let page = get_settled(&r.http, "/production/simple/sampleproject/", &[PEP691]);
assert_eq!(page.status, 200, "{}", page.text());
drop(r);
// On, under two approvers: the automatic vote is one of two; a person casts the other.
let up = upstream();
let pt = format!("{} rules: (allow_auto_approve: true),", pypi(&up, auto));
let r = start(&pt, 2, "Some((required: [vulnerability]))");
let chain = r.chain();
assert_eq!(
get(&r.http, "/production/simple/sampleproject/", &[PEP691]).status,
503
);
let w = pip_id("pypi", WHEEL);
wait_for("the automatic vote", || {
(chain.pending_approvals(&w).len() == 1).then_some(())
});
assert_eq!(chain.state(&w), ArtifactState::PolicyChecked);
assert_eq!(chain.pending_approvals(&w)[0].approver, "auto-approve:pypi");
approve(&chain, &w, "bob");
assert_eq!(chain.state(&w), ArtifactState::Approved);
assert_eq!(
get_settled(&r.http, &format!("/production/pypi/{WHEEL}"), &[]).status,
200
);
drop(r);
// Asked by the source but refused by the floor: the config does not load.
let up = upstream();
let parsed: PullOnly =
ron::from_str(&format!("(pull_through: Some(({})))", pypi(&up, auto))).unwrap();
let e = holger_server_lib::pull_intake::build_sources(&parsed.pull_through.unwrap())
.err()
.expect("refused at load")
.to_string();
assert!(e.contains("allow_auto_approve: false"), "{e}");
}
#[derive(serde::Deserialize)]
struct PullOnly {
pull_through: Option<holger_server_lib::pull_intake::PullThroughSettings>,
}
// ── OCI ─────────────────────────────────────────────────────────────────────
/// Stands in for G3a's verifier: checks that the material is Chainguard's real
/// cosign attachments for the image digest, as fetched with the image.
#[derive(Default)]
struct Recording {
calls: AtomicUsize,
}
impl Verifier for Recording {
fn verify(&self, subject: &[u8], materials: &[Material], _demand: &Demand) -> Outcome {
self.calls.fetch_add(1, Ordering::SeqCst);
let fail = |why: String| Outcome {
verdict: Verdict::Block,
detail: why,
code: Some("test".into()),
verified: 0,
};
let digest = format!("sha256:{}", sha256_hex(subject));
let Some(m) = materials.iter().find(|m| m.name == "cosign.json") else {
return fail(format!("no cosign.json in {materials:?}"));
};
let v: serde_json::Value = serde_json::from_slice(&m.bytes).unwrap();
if v["schema"] != "holger.cosign-attachments/v1" || v["digest"] != digest.as_str() {
return fail(format!("schema/digest: {} {}", v["schema"], v["digest"]));
}
use base64::Engine as _;
let mut n = 0;
for key in ["signatures", "attestations"] {
for l in v[key].as_array().unwrap() {
let blob = base64::engine::general_purpose::STANDARD
.decode(l["blob"].as_str().unwrap())
.unwrap();
if format!("sha256:{}", sha256_hex(&blob)) != l["digest"].as_str().unwrap() {
return fail(format!("{key}: a layer blob does not hash to its digest"));
}
if l["annotations"]["dev.sigstore.cosign/bundle"]
.as_str()
.is_none()
{
return fail(format!("{key}: no rekor bundle annotation"));
}
n += 1;
}
}
if v["signatures"].as_array().unwrap().len() != 1
|| v["attestations"].as_array().unwrap().len() != 3
{
return fail(format!("expected 1 signature and 3 attestations, got {v}"));
}
Outcome {
verdict: Verdict::Pass,
detail: "the test verifier read Chainguard's material".into(),
code: None,
verified: n,
}
}
}
/// G3a's real verifier, counted: Chainguard's material verified offline against
/// Sigstore's public-good trust root (the fixture `holger-provenance` pins).
#[derive(Default)]
struct Real {
calls: AtomicUsize,
inner: holger_server_lib::pull_intake::provenance::HolgerProvenance,
}
impl Real {
fn new() -> Self {
let root = std::fs::read(concat!(
env!("CARGO_MANIFEST_DIR"),
"/../../provenance/tests/fixtures/sigstore/trusted_root.json"
))
.unwrap();
let trust = holger_provenance::TrustRoot::from_trusted_root_json(&root).unwrap();
Self {
calls: AtomicUsize::new(0),
inner: holger_server_lib::pull_intake::provenance::HolgerProvenance::new(Some(
Arc::new(trust),
)),
}
}
}
impl Verifier for Real {
fn verify(&self, subject: &[u8], materials: &[Material], demand: &Demand) -> Outcome {
self.calls.fetch_add(1, Ordering::SeqCst);
self.inner.verify(subject, materials, demand)
}
}
fn cgr(up: &Upstream) -> String {
format!(
r#"sources: [ (name: "cgr", kind: oci, url: "{}", platforms: ["linux/amd64"],
gates: (provenance: Some(require(identity: Some((issuer: "{CHAINGUARD_ISSUER}", subject: Some("{CHAINGUARD_SUBJECT}"))))))) ],
rules: (require_signed: true), retry_after_secs: Some(1),"#,
up.base
)
}
const OCI_ACCEPT: (&str, &str) = (
"Accept",
"application/vnd.oci.image.index.v1+json, application/vnd.oci.image.manifest.v1+json",
);
#[test]
fn a_signed_chainguard_image_is_pulled_whole_with_its_signatures_and_served_after_approval() {
let up = upstream();
let r = start(&cgr(&up), 1, VULN);
let chain = r.chain();
let rec = Arc::new(Real::new());
chain.pull().unwrap().replace_verifier(rec.clone());
let m = "/v2/production/cgr/chainguard/static/manifests/latest";
let first = get(&r.http, m, &[OCI_ACCEPT]);
assert_eq!(first.status, 503, "{}", first.text());
let e: serde_json::Value = serde_json::from_slice(&first.body).unwrap();
assert_eq!(e["errors"][0]["code"], "UNAVAILABLE", "{e}");
assert!(first.headers.contains_key("retry-after"));
let id = oci_id("cgr", "chainguard/static", INDEX);
wait_state(&chain, &id, ArtifactState::PolicyChecked);
assert_eq!(
rec.calls.load(Ordering::SeqCst),
1,
"the verifier saw the image once"
);
let h = chain.history(&id);
assert!(h.iter().all(|e| e.actor == "pull-through:cgr"), "{h:#?}");
let prov = h[1]
.evidence
.iter()
.find(|e| e.kind == EvidenceKind::Provenance)
.expect("a provenance slot");
assert_eq!(
(prov.source.as_str(), prov.verdict),
("holger-provenance", EvidenceVerdict::Pass),
"{prov:?}"
);
// The token dance, the signature and the attestations were fetched with the image.
assert_eq!(up.hits("/token"), 1, "one token for the whole pull");
assert!(
up.hits(&format!(
"/v2/chainguard/static/manifests/sha256-{}.sig",
&INDEX[7..]
)) >= 1
);
assert!(
up.hits(&format!(
"/v2/chainguard/static/manifests/sha256-{}.att",
&INDEX[7..]
)) >= 1
);
// The arm64 platform was not kept.
assert_eq!(up.hits("/v2/chainguard/static/manifests/sha256:15b445a48b1c6e46a9d09463cbfea4f9951bacf55f02b7991e0e99988d91aa7b"), 0);
let held = get_settled(&r.http, m, &[OCI_ACCEPT]);
assert_eq!(held.status, 404, "{}", held.text());
assert!(
held.text().contains("MANIFEST_UNKNOWN") && held.text().contains("waiting for approval"),
"{}",
held.text()
);
approve(&chain, &id, "bob");
let index = get_settled(&r.http, m, &[OCI_ACCEPT]);
assert_eq!(index.status, 200, "{}", index.text());
assert_eq!(
index
.headers
.get("docker-content-digest")
.map(String::as_str),
Some(INDEX)
);
assert_eq!(
index.body,
fixture(&format!(
"oci/chainguard/static/manifests/{}",
INDEX.replace(':', "_")
))
.unwrap()
);
let amd = get(
&r.http,
&format!("/v2/production/cgr/chainguard/static/manifests/{AMD64}"),
&[OCI_ACCEPT],
);
assert_eq!(amd.status, 200, "{}", amd.text());
let mv: serde_json::Value = serde_json::from_slice(&amd.body).unwrap();
for d in std::iter::once(mv["config"]["digest"].as_str().unwrap()).chain(
mv["layers"]
.as_array()
.unwrap()
.iter()
.map(|l| l["digest"].as_str().unwrap()),
) {
let b = get(
&r.http,
&format!("/v2/production/cgr/chainguard/static/blobs/{d}"),
&[],
);
assert_eq!(b.status, 200, "{d}: {}", b.text());
assert_eq!(format!("sha256:{}", sha256_hex(&b.body)), d);
}
let tags = get(
&r.http,
"/v2/production/cgr/chainguard/static/tags/list",
&[],
);
let tv: serde_json::Value = serde_json::from_slice(&tags.body).unwrap();
assert_eq!(tv["tags"], serde_json::json!(["latest"]), "{tv}");
// A blob of no AVAILABLE image is the registry's BLOB_UNKNOWN.
let none = get(
&r.http,
&format!(
"/v2/production/cgr/chainguard/static/blobs/sha256:{}",
"0".repeat(64)
),
&[],
);
assert_eq!(none.status, 404);
assert!(none.text().contains("BLOB_UNKNOWN"));
}
/// ★ HOL-REFERRERS (owner 2026-10-30): **the image's signatures, attestations and
/// referrers come with it and are served with it**, so a client's `cosign verify` and
/// `notation verify` work through Holger: the `.sig` and `.att` tags byte for byte, the
/// OCI 1.1 referrers index (filtered by `artifactType` when asked) with each referrer's
/// manifest and blobs. Short of AVAILABLE none of it is served, and none of it is ever
/// pulled on its own.
#[test]
fn a_pulled_image_brings_its_signatures_and_referrers_and_serves_them_once_available() {
let up = upstream();
let r = start(&cgr(&up), 1, VULN);
let chain = r.chain();
chain
.pull()
.unwrap()
.replace_verifier(Arc::new(Real::new()));
let base = "/v2/production/cgr/chainguard/static";
let sig_tag = format!("sha256-{}.sig", &INDEX[7..]);
let att_tag = format!("sha256-{}.att", &INDEX[7..]);
assert_eq!(
get(&r.http, &format!("{base}/manifests/latest"), &[OCI_ACCEPT]).status,
503
);
let id = oci_id("cgr", "chainguard/static", INDEX);
wait_state(&chain, &id, ArtifactState::PolicyChecked);
assert!(
up.hits(&format!("/v2/chainguard/static/referrers/{INDEX}")) >= 1,
"the referrers were asked for with the image"
);
// Waiting for a person: nothing attached is served, and asking pulls nothing.
let sig = get(&r.http, &format!("{base}/manifests/{sig_tag}"), &[]);
assert_eq!(sig.status, 404, "{}", sig.text());
let empty = get(&r.http, &format!("{base}/referrers/{INDEX}"), &[]);
assert_eq!(empty.status, 200, "{}", empty.text());
let ev: serde_json::Value = serde_json::from_slice(&empty.body).unwrap();
assert_eq!(
ev["manifests"],
serde_json::json!([]),
"not AVAILABLE, no referrers: {ev}"
);
let before = up.hits(&format!("/v2/chainguard/static/manifests/{sig_tag}"));
approve(&chain, &id, "bob");
assert_eq!(
get_settled(&r.http, &format!("{base}/manifests/latest"), &[OCI_ACCEPT]).status,
200
);
// The cosign tag scheme, byte for byte from the sealed layout.
for tag in [&sig_tag, &att_tag] {
let g = get(&r.http, &format!("{base}/manifests/{tag}"), &[]);
assert_eq!(g.status, 200, "{tag}: {}", g.text());
let want = fixture(&format!("oci/chainguard/static/manifests/{tag}")).unwrap();
assert_eq!(g.body, want, "{tag} byte for byte");
assert_eq!(
g.headers.get("docker-content-digest"),
Some(&format!("sha256:{}", sha256_hex(&want)))
);
let mv: serde_json::Value = serde_json::from_slice(&g.body).unwrap();
for l in mv["layers"].as_array().unwrap() {
let d = l["digest"].as_str().unwrap();
let b = get(&r.http, &format!("{base}/blobs/{d}"), &[]);
assert_eq!(
(b.status, format!("sha256:{}", sha256_hex(&b.body))),
(200, d.to_string()),
"{tag}'s layer {d}"
);
}
}
assert_eq!(
up.hits(&format!("/v2/chainguard/static/manifests/{sig_tag}")),
before,
"served from the seal, not pulled again"
);
// OCI 1.1 referrers.
let (sbom_manifest, sbom_layer) = sbom_referrer();
let sbom_digest = format!("sha256:{}", sha256_hex(&sbom_manifest));
let idx = get(&r.http, &format!("{base}/referrers/{INDEX}"), &[]);
assert_eq!(idx.status, 200, "{}", idx.text());
assert_eq!(
idx.headers.get("content-type").map(String::as_str),
Some("application/vnd.oci.image.index.v1+json")
);
let iv: serde_json::Value = serde_json::from_slice(&idx.body).unwrap();
let m = iv["manifests"].as_array().unwrap();
assert_eq!(m.len(), 1, "{iv}");
assert_eq!(
(m[0]["digest"].as_str(), m[0]["artifactType"].as_str()),
(Some(sbom_digest.as_str()), Some(SPDX)),
"{iv}"
);
assert_eq!(
m[0]["annotations"]["org.opencontainers.image.created"],
"2026-10-30T00:00:00Z"
);
let filtered = get(
&r.http,
&format!("{base}/referrers/{INDEX}?artifactType=application%2Fvnd.cncf.notary.signature"),
&[],
);
assert_eq!(
filtered
.headers
.get("oci-filters-applied")
.map(String::as_str),
Some("artifactType")
);
let fv: serde_json::Value = serde_json::from_slice(&filtered.body).unwrap();
assert_eq!(fv["manifests"], serde_json::json!([]));
let rm = get(&r.http, &format!("{base}/manifests/{sbom_digest}"), &[]);
assert_eq!(
(rm.status, rm.body.clone()),
(200, sbom_manifest),
"{}",
rm.text()
);
let rl = get(
&r.http,
&format!("{base}/blobs/sha256:{}", sha256_hex(&sbom_layer)),
&[],
);
assert_eq!((rl.status, rl.body), (200, sbom_layer));
// A platform manifest with no referrers: an empty index.
let none = get(&r.http, &format!("{base}/referrers/{AMD64}"), &[]);
let nv: serde_json::Value = serde_json::from_slice(&none.body).unwrap();
assert_eq!(
(none.status, nv["manifests"].clone()),
(200, serde_json::json!([]))
);
}
/// ★ A29 row 11 (owner): **a container image is NEVER auto-approved**, whatever the floor
/// says. A source that asks for it beside a permissive floor (`allow_auto_approve: true`)
/// is refused by name at load; were it loaded, the image must still wait for a person.
#[test]
fn an_oci_image_is_never_auto_approved_whatever_the_floor_says() {
let up = upstream();
let pt = cgr(&up)
.replace(r#"platforms: ["linux/amd64"],"#, r#"platforms: ["linux/amd64"], auto_approve: when(provenance: require, cve_max: critical),"#)
.replace("rules: (require_signed: true)", "rules: (require_signed: true, allow_auto_approve: true, cve_max: critical)");
assert!(
pt.contains("auto_approve: when") && pt.contains("allow_auto_approve: true"),
"{pt}"
);
let (grpc, http) = (hold_port(), hold_port());
match try_start(grpc, http, &pt, 1, "Some((required: [vulnerability]))", "") {
Err(e) => {
assert!(
e.contains("never auto-approved") && e.contains("cgr"),
"the refusal names the rule and the source: {e}"
);
}
Ok(r) => {
// The config loaded: drive the pull and say how far it got without a person.
let chain = r.chain();
chain
.pull()
.unwrap()
.replace_verifier(Arc::new(Real::new()));
let m = "/v2/production/cgr/chainguard/static/manifests/latest";
assert_eq!(get(&r.http, m, &[OCI_ACCEPT]).status, 503);
let id = oci_id("cgr", "chainguard/static", INDEX);
wait_state(&chain, &id, ArtifactState::PolicyChecked);
std::thread::sleep(std::time::Duration::from_secs(3));
let h = chain.history(&id);
let auto: Vec<_> = h
.iter()
.filter(|e| e.actor.starts_with("auto-approve:"))
.map(|e| format!("{:?} by {}", e.to, e.actor))
.collect();
panic!("an OCI source with auto_approve and a permissive floor LOADED; the image reached {:?} with no person ({auto:?})", chain.state(&id));
}
}
}
#[test]
fn without_the_provenance_verifier_a_required_signature_fails_closed() {
let up = upstream();
let r = start(&cgr(&up), 1, VULN);
let chain = r.chain();
chain.pull().unwrap().replace_verifier(Arc::new(
holger_server_lib::pull_intake::provenance::NotInBuild,
));
let m = "/v2/production/cgr/chainguard/static/manifests/latest";
assert_eq!(get(&r.http, m, &[OCI_ACCEPT]).status, 503);
let id = oci_id("cgr", "chainguard/static", INDEX);
wait_state(&chain, &id, ArtifactState::Blocked);
let last = chain.history(&id).last().cloned().unwrap();
assert!(
last.reason.contains("not in this build") && last.reason.contains("fail closed"),
"{}",
last.reason
);
let denied = get_settled(&r.http, m, &[OCI_ACCEPT]);
assert_eq!(denied.status, 403, "{}", denied.text());
assert!(denied.text().contains("DENIED"), "{}", denied.text());
}
#[test]
fn an_unsigned_image_is_refused_by_the_floor_require_signed() {
let up = upstream();
let r = start(&cgr(&up), 1, VULN);
let chain = r.chain();
let rec = Arc::new(Recording::default());
chain.pull().unwrap().replace_verifier(rec.clone());
let m = "/v2/production/cgr/unsigned/static/manifests/latest";
assert_eq!(get(&r.http, m, &[OCI_ACCEPT]).status, 503);
let id = oci_id("cgr", "unsigned/static", INDEX);
wait_state(&chain, &id, ArtifactState::Blocked);
let last = chain.history(&id).last().cloned().unwrap();
assert!(
last.reason
.contains("no signature or attestation came with it"),
"{}",
last.reason
);
assert_eq!(
rec.calls.load(Ordering::SeqCst),
0,
"nothing to verify: refused before the verifier"
);
assert_eq!(get_settled(&r.http, m, &[OCI_ACCEPT]).status, 403);
}
// ── an invalidation from the upstream (H4) ───────────────────────────────────
//
// What an applied sync manifest does to each package it invalidates
// (`IntakeChain::invalidate`, run by `server_lib::sync`): QUARANTINED at once, the
// sealed bytes kept, every index drops it, and a read is refused with the reason.
const INVALIDATED: &str = "invalidated by upstream 'dmz' (generation 2): cve-after: CVE-2026-0001";
fn invalidate(chain: &IntakeChain, id: &traits::ArtifactId) {
use holger_server_lib::promote::intake::InvalidateOutcome;
let out = chain
.invalidate(id, "holger-sync:dmz", "sync", INVALIDATED)
.unwrap_or_else(|e| panic!("invalidate {}: {e}", id.name));
assert!(matches!(out, InvalidateOutcome::Held(_)), "{out:?}");
assert_eq!(chain.state(id), ArtifactState::Quarantined);
}
#[test]
fn an_invalidated_wheel_drops_from_the_pep503_and_pep691_pages_and_is_refused() {
let up = upstream();
let r = start(&pypi(&up, ""), 1, "Some((required: [vulnerability]))");
let chain = r.chain();
get(&r.http, "/production/simple/sampleproject/", &[PEP691]);
let (w, s) = (pip_id("pypi", WHEEL), pip_id("pypi", SDIST));
wait_state(&chain, &w, ArtifactState::PolicyChecked);
wait_state(&chain, &s, ArtifactState::PolicyChecked);
approve(&chain, &w, "bob");
approve(&chain, &s, "bob");
assert_eq!(
get_settled(&r.http, "/production/simple/sampleproject/", &[PEP691]).status,
200
);
assert_eq!(
get(&r.http, &format!("/production/pypi/{WHEEL}"), &[]).status,
200
);
invalidate(&chain, &w);
let json = get_settled(&r.http, "/production/simple/sampleproject/", &[PEP691]);
assert_eq!(
json.status,
200,
"the sdist is still AVAILABLE: {}",
json.text()
);
assert!(
!json.text().contains(WHEEL),
"PEP 691 drops the wheel: {}",
json.text()
);
assert!(json.text().contains(SDIST), "{}", json.text());
let html = get_settled(&r.http, "/production/simple/sampleproject/", &[]);
assert_eq!(html.status, 200, "{}", html.text());
assert!(
!html.text().contains(WHEEL) && html.text().contains(SDIST),
"PEP 503 drops the wheel: {}",
html.text()
);
let file = get_settled(&r.http, &format!("/production/pypi/{WHEEL}"), &[]);
// A refusal answers 403 (lane HOL-310-CHECK; it was 404).
assert_eq!(file.status, 403, "{}", file.text());
assert!(
file.text().contains("invalidated by upstream 'dmz'"),
"{}",
file.text()
);
}
#[test]
fn an_invalidated_image_drops_from_the_tag_list_and_is_denied() {
let up = upstream();
let r = start(&cgr(&up), 1, VULN);
let chain = r.chain();
chain
.pull()
.unwrap()
.replace_verifier(Arc::new(Real::new()));
let m = "/v2/production/cgr/chainguard/static/manifests/latest";
get(&r.http, m, &[OCI_ACCEPT]);
let id = oci_id("cgr", "chainguard/static", INDEX);
wait_state(&chain, &id, ArtifactState::PolicyChecked);
approve(&chain, &id, "bob");
assert_eq!(get_settled(&r.http, m, &[OCI_ACCEPT]).status, 200);
let tags = "/v2/production/cgr/chainguard/static/tags/list";
assert_eq!(
serde_json::from_slice::<serde_json::Value>(&get(&r.http, tags, &[]).body).unwrap()["tags"],
serde_json::json!(["latest"])
);
invalidate(&chain, &id);
let tv: serde_json::Value = serde_json::from_slice(&get(&r.http, tags, &[]).body).unwrap();
assert_eq!(
tv["tags"],
serde_json::json!([]),
"the tag list drops it: {tv}"
);
let denied = get_settled(&r.http, m, &[OCI_ACCEPT]);
assert_eq!(denied.status, 403, "{}", denied.text());
assert!(
denied.text().contains("DENIED") && denied.text().contains("invalidated"),
"{}",
denied.text()
);
}
// ── the trust bundle the CONFIG names (OX-WIRED3) ─────────────────────────────
//
// The tests above swap in a verifier with `replace_verifier`, so the boot's own path,
// `rules.trust_root` (a signed `holger.sigstore-trust/v1` bundle) verified against
// `promotion.intake.trust_policy` (`ProvenanceScanner::load_trust_root`), never ran. Here
// nothing is swapped: the verifier is the one the boot built from the RON.
/// The signed trust bundle over Sigstore's public-good `trusted_root.json` (the fixture
/// `holger-provenance` pins), the bundle trust policy that trusts its signers, and one that
/// pins other signers. Paths, as the RON names them.
struct TrustFiles {
_dir: tempfile::TempDir,
bundle: String,
policy: String,
stranger: String,
}
fn trust_files() -> TrustFiles {
use holger_provenance::trust::{build_trust_bundle, BundleSigner, TrustBundleOptions};
use update_bundle::policy::{
ApprovedSigner, CertificateValidity, ExpiredPosture, TrustMode, TrustPolicy,
};
use update_bundle::schema::POLICY_SCHEMA;
let mint = |cn: &str| {
let key = rcgen::KeyPair::generate_for(&rcgen::PKCS_ED25519).unwrap();
let mut p = rcgen::CertificateParams::new(vec![]).unwrap();
p.distinguished_name.push(rcgen::DnType::CommonName, cn);
let der = p.self_signed(&key).unwrap().der().to_vec();
(
holger_provenance::crypto::sha256_hex(&der),
BundleSigner {
cert_der: der,
key_pkcs8: key.serialize_der(),
},
)
};
let (fr, reviewer) = mint("trust-reviewer");
let (fo, owner) = mint("trust-owner");
let (fl, release) = mint("trust-release");
let policy = TrustPolicy {
schema_version: POLICY_SCHEMA.to_string(),
trust_mode: Some(TrustMode::PinnedSigners),
trusted_roots: vec![],
certificate_validity: CertificateValidity::Enforce,
approved_signers: vec![
ApprovedSigner {
fingerprint: fr,
subject: None,
roles: vec!["security-reviewer".into()],
},
ApprovedSigner {
fingerprint: fo,
subject: None,
roles: vec!["system-owner".into()],
},
],
required_approval_roles: vec!["security-reviewer".into(), "system-owner".into()],
minimum_distinct_approvers: 2,
trusted_release_signers: vec![fl],
release_signer_must_be_distinct: true,
maximum_bundle_age_secs: None,
require_expiry: true,
on_expired_metadata: Some(ExpiredPosture::Refuse),
require_sbom: false,
require_vex: false,
require_scan: false,
allowed_products: vec![holger_provenance::TRUST_PRODUCT.into()],
allowed_models: vec![],
};
let dir = tempfile::tempdir().unwrap();
let now = chrono::Utc::now();
let root = std::fs::read(concat!(
env!("CARGO_MANIFEST_DIR"),
"/../../provenance/tests/fixtures/sigstore/trusted_root.json"
))
.unwrap();
let bundle = build_trust_bundle(
&root,
TrustBundleOptions {
sequence: 1,
created_at: now - chrono::Duration::minutes(5),
expires_at: Some(now + chrono::Duration::days(90)),
supplier: "holger-sigstore-trust".into(),
producer: "test".into(),
source_uri: Some("https://tuf-repo-cdn.sigstore.dev".into()),
policy: policy.clone(),
approvals: vec![
("security-reviewer".into(), reviewer),
("system-owner".into(), owner),
],
release,
out: dir.path().join("sigstore-trust.holger.znippy"),
apk_keys: vec![],
pgp_keys: vec![],
},
)
.expect("the trust bundle is signed");
let policy_path = dir.path().join("bundle-trust.ron");
std::fs::write(&policy_path, policy.to_ron().unwrap()).unwrap();
let mut other = policy;
other.trusted_release_signers = vec!["00".repeat(32)];
let stranger_path = dir.path().join("stranger-trust.ron");
std::fs::write(&stranger_path, other.to_ron().unwrap()).unwrap();
TrustFiles {
bundle: bundle.display().to_string(),
policy: policy_path.display().to_string(),
stranger: stranger_path.display().to_string(),
_dir: dir,
}
}
/// `cgr` whose floor names the trust bundle, and the gate's `subject` as given.
fn cgr_trusting(up: &Upstream, bundle: &str, subject: &str) -> String {
let pt = cgr(up)
.replace(
"rules: (require_signed: true)",
&format!("rules: (require_signed: true, trust_root: Some(\"{bundle}\"))"),
)
.replace(
&format!("subject: Some(\"{CHAINGUARD_SUBJECT}\")"),
&format!("subject: Some(\"{subject}\")"),
);
assert!(
pt.contains("trust_root: Some(") && pt.contains(subject),
"{pt}"
);
pt
}
/// `promotion.intake.scan` runs the vulnerability match ([`VULN`]), and the intake's bundle trust policy beside it.
fn scan_trusting(policy: &str) -> String {
format!("{VULN}, trust_policy: Some(\"{policy}\")")
}
/// ★ **The trust bundle the config names verifies Chainguard's signed image**, through the
/// pull and the intake with the boot's own verifier: the image walks to POLICY_CHECKED with
/// the provenance slot passed, waits for a person, and is served after the approval. An
/// unsigned image from the same source is held (BLOCKED) and refused at the door.
#[test]
fn a_signed_image_passes_against_the_trust_bundle_the_config_names() {
let t = trust_files();
let up = upstream();
let r = start(
&cgr_trusting(&up, &t.bundle, CHAINGUARD_SUBJECT),
1,
&scan_trusting(&t.policy),
);
let chain = r.chain();
let m = "/v2/production/cgr/chainguard/static/manifests/latest";
assert_eq!(get(&r.http, m, &[OCI_ACCEPT]).status, 503, "not yet");
let id = oci_id("cgr", "chainguard/static", INDEX);
wait_state(&chain, &id, ArtifactState::PolicyChecked);
let h = chain.history(&id);
let prov = h[1]
.evidence
.iter()
.find(|e| e.kind == EvidenceKind::Provenance)
.expect("a provenance slot");
assert_eq!(
(prov.source.as_str(), prov.verdict),
("holger-provenance", EvidenceVerdict::Pass),
"{prov:?}"
);
let held = get_settled(&r.http, m, &[OCI_ACCEPT]);
assert_eq!(held.status, 404, "{}", held.text());
approve(&chain, &id, "bob");
let index = get_settled(&r.http, m, &[OCI_ACCEPT]);
assert_eq!(index.status, 200, "{}", index.text());
assert_eq!(
index
.headers
.get("docker-content-digest")
.map(String::as_str),
Some(INDEX)
);
let unsigned = "/v2/production/cgr/unsigned/static/manifests/latest";
assert_eq!(get(&r.http, unsigned, &[OCI_ACCEPT]).status, 503);
let uid = oci_id("cgr", "unsigned/static", INDEX);
wait_state(&chain, &uid, ArtifactState::Blocked);
let last = chain.history(&uid).last().cloned().unwrap();
assert!(
last.reason
.contains("no signature or attestation came with it"),
"{}",
last.reason
);
assert_eq!(get_settled(&r.http, unsigned, &[OCI_ACCEPT]).status, 403);
}
/// ★ **A signature by someone the gate does not trust is held**, against the same bundle:
/// Chainguard's real signature, under a gate that names another workflow, is BLOCKED by the
/// boot's verifier with the identity refusal named, and never served.
#[test]
fn a_wrongly_signed_image_is_held_against_the_trust_bundle_the_config_names() {
let t = trust_files();
let up = upstream();
let other = "https://github.com/chainguard-images/images/.github/workflows/release.yaml@refs/heads/not-main";
let r = start(
&cgr_trusting(&up, &t.bundle, other),
1,
&scan_trusting(&t.policy),
);
let chain = r.chain();
let m = "/v2/production/cgr/chainguard/static/manifests/latest";
assert_eq!(get(&r.http, m, &[OCI_ACCEPT]).status, 503);
let id = oci_id("cgr", "chainguard/static", INDEX);
wait_state(&chain, &id, ArtifactState::Blocked);
let last = chain.history(&id).last().cloned().unwrap();
assert!(
last.reason.contains("holger-provenance-5"),
"the identity refusal is named: {}",
last.reason
);
let denied = get_settled(&r.http, m, &[OCI_ACCEPT]);
assert_eq!(denied.status, 403, "{}", denied.text());
}
/// ★ **A trust bundle the config cannot trust stops the boot by name**: a bundle trust
/// policy that pins other release signers (`holger-provenance-10`), and a trust root with no
/// bundle trust policy at all.
#[test]
fn a_trust_bundle_the_config_cannot_trust_refuses_the_boot_by_name() {
let t = trust_files();
let up = upstream();
let pt = cgr_trusting(&up, &t.bundle, CHAINGUARD_SUBJECT);
let e = try_start(
hold_port(),
hold_port(),
&pt,
1,
&scan_trusting(&t.stranger),
"",
)
.err()
.expect("a bundle from strangers booted");
assert!(
e.contains("rules.trust_root") && e.contains("holger-provenance-10"),
"{e}"
);
let e = try_start(hold_port(), hold_port(), &pt, 1, VULN, "")
.err()
.expect("an unverified trust root booted");
assert!(e.contains("names no bundle trust policy"), "{e}");
}
/// **A pinned release that is not the newest is installable** (lane HOL-FIELD-TEST P2). In the
/// field, `pip install requests==2.32.3` through Holger said "from versions: none": the project
/// page pulled only the newest release, and the page lists only what is AVAILABLE, so pip never
/// saw a link to the release it was asked for and could not ask for it. Now every file the
/// upstream lists that is not AVAILABLE here is on the page as YANKED (PEP 592), with the reason:
/// an unpinned install skips it (the AVAILABLE set stays what is installable), a pin `==` selects
/// it, and asking for its link pulls exactly that file into the chain, where it waits for people.
#[test]
fn a_pinned_older_release_is_listed_yanked_and_pulled_when_its_link_is_asked() {
let up = upstream();
let r = start(&pypi(&up, ""), 1, "Some((required: [vulnerability]))");
let chain = r.chain();
const OLD: &str = "sampleproject-3.0.0-py3-none-any.whl";
// 1. Once the upstream's page was read, the project page answers with its files.
assert_eq!(
get(&r.http, "/production/simple/sampleproject/", &[PEP691]).status,
503
);
wait_state(&chain, &pip_id("pypi", WHEEL), ArtifactState::PolicyChecked);
let page = wait_for("the project page lists the upstream's files", || {
let g = get(&r.http, "/production/simple/sampleproject/", &[PEP691]);
(g.status == 200).then_some(g)
});
let doc: serde_json::Value = serde_json::from_slice(&page.body).unwrap();
let files = doc["files"].as_array().unwrap();
let old = files
.iter()
.find(|f| f["filename"] == OLD)
.unwrap_or_else(|| panic!("the older release is listed: {doc}"));
assert_eq!(old["url"], format!("../../pypi/{OLD}"), "{old}");
assert!(
old["hashes"]["sha256"]
.as_str()
.is_some_and(|h| h.len() == 64),
"{old}"
);
assert!(
old["yanked"]
.as_str()
.is_some_and(|y| y.contains("not in Holger yet")),
"a file Holger has not pulled is yanked, with the reason: {old}"
);
// The newest release, pulled and waiting for a person, is yanked too, and says why.
let newest = files.iter().find(|f| f["filename"] == WHEEL).unwrap();
assert!(
newest["yanked"]
.as_str()
.is_some_and(|y| y.contains("waiting for approval")),
"{newest}"
);
// The PEP 503 page says the same.
let html = get(&r.http, "/production/simple/sampleproject/", &[]);
assert_eq!(html.status, 200, "{}", html.text());
let line = html
.text()
.lines()
.find(|l| l.contains(OLD))
.map(str::to_string)
.unwrap_or_else(|| panic!("{}", html.text()));
assert!(
line.contains(&format!("href=\"../../pypi/{OLD}#sha256="))
&& line.contains("data-yanked=\""),
"{line}"
);
// 2. Asking for its link pulls exactly that file; it waits for a person.
let id = pip_id("pypi", OLD);
let first = get(&r.http, &format!("/production/pypi/{OLD}"), &[]);
assert_eq!(first.status, 503, "{}", first.text());
wait_state(&chain, &id, ArtifactState::PolicyChecked);
approve(&chain, &id, "bob");
let whl = get_settled(&r.http, &format!("/production/pypi/{OLD}"), &[]);
assert_eq!(whl.status, 200, "{}", whl.text());
assert_eq!(whl.body, fixture(&format!("pypi/files/{OLD}")).unwrap());
// 3. AVAILABLE now: listed as an installable file, no longer yanked.
let page = get(&r.http, "/production/simple/sampleproject/", &[PEP691]);
let doc: serde_json::Value = serde_json::from_slice(&page.body).unwrap();
let old = doc["files"]
.as_array()
.unwrap()
.iter()
.find(|f| f["filename"] == OLD)
.unwrap_or_else(|| panic!("{doc}"));
assert!(old.get("yanked").is_none(), "{old}");
}
/// The intel database a box gets by import, installed by hand from one OSV record.
fn install_intel(dir: &Path, osv: &str) {
use holger_intel::chrono::Utc;
use holger_intel::{AdvisorySet, InstalledIntel, LocalIntelDb};
let advisories = holger_intel::feeds::parse_osv_feed(osv.as_bytes()).unwrap();
let set = AdvisorySet::from_advisories(Utc::now(), vec![], &advisories);
let n = set.advisories.len() as u64;
let db = LocalIntelDb::open(dir);
let bytes = set.to_json_bytes().unwrap();
db.stage_generation(1, &bytes).unwrap();
db.activate(&InstalledIntel {
schema: holger_intel::db::INSTALLED_SCHEMA.into(),
bundle_id: "holger-intel-1".into(),
sequence: 1,
supplier: holger_intel::DEFAULT_SUPPLIER.into(),
snapshot_at: Utc::now(),
imported_at: Utc::now(),
candidate_digest: "00".into(),
release_signer_fingerprint: "00".into(),
advisories_file: LocalIntelDb::generation_file(1),
advisories_sha256: holger_server_lib::security::blob_digest_hex(&bytes),
advisories: n,
})
.unwrap();
}
/// **A pulled package is scanned as the package it is** (lane HOL-FIELD-TEST P11). In the
/// field, `PyYAML-5.3.tar.gz` (two critical CVEs) and `lodash-4.17.15.tgz` came through the
/// pull-through with "raw PyYAML@5.3: 0 finding(s)": the intake repository is `raw`, and only a
/// wheel was taken for a Python package, so an sdist (and a pulled npm tarball, Maven jar or
/// crate) was matched against no ecosystem at all and every known CVE passed. The source a
/// package came through says what it is.
#[test]
fn a_pulled_sdist_is_scanned_as_its_python_package_and_a_known_cve_refuses_it() {
let up = upstream();
let intel = tempfile::tempdir().unwrap();
install_intel(
intel.path(),
r#"{"id": "GHSA-hfld-0001-0001", "modified": "2026-01-01T00:00:00Z", "aliases": ["CVE-2099-0001"],
"summary": "a critical flaw in sampleproject", "database_specific": {"severity": "CRITICAL"},
"affected": [ { "package": { "ecosystem": "PyPI", "name": "sampleproject" },
"ranges": [ { "type": "ECOSYSTEM", "events": [ { "introduced": "0" }, { "fixed": "4.0.1" } ] } ] } ] }"#,
);
let r = start_with(
&pypi(&up, ""),
1,
"Some((required: [vulnerability]))",
&format!(
r#"intel: Some((db_dir: "{}", offline: true)),"#,
intel.path().display()
),
);
let chain = r.chain();
let _ = get(&r.http, "/production/simple/sampleproject/", &[PEP691]);
let (w, s) = (pip_id("pypi", WHEEL), pip_id("pypi", SDIST));
// The wheel was always judged as `pip`; the sdist must be too.
wait_state(&chain, &w, ArtifactState::Blocked);
wait_state(&chain, &s, ArtifactState::Blocked);
let why = format!("{:?}", chain.history(&s));
assert!(
why.contains("GHSA-hfld-0001-0001") || why.contains("CVE-2099-0001"),
"the sdist is refused for the advisory: {why}"
);
}
/// **pip says why Holger refused a file** (lane HOL-310-CHECK). In the 3.1.0 field check,
/// `pip install PyYAML==5.3` through Holger said only "Could not find a version that satisfies
/// the requirement PyYAML==5.3 (from versions: none)": a BLOCKED file was left off the project
/// page, and its link answered 404. The two GHSA ids were only in `holger-cli intake state`.
/// Now a refused file is on the page as YANKED (PEP 592) with "refused by Holger: <the reason>",
/// so a `==` pin selects it and pip prints the reason ("Reason for being yanked: ..."), and its
/// link answers 403 with the same reason. It is still never served, and asking pulls nothing.
#[test]
fn a_refused_pip_file_is_listed_yanked_with_the_reason_and_its_link_answers_403() {
let up = upstream();
let intel = tempfile::tempdir().unwrap();
install_intel(
intel.path(),
r#"{"id": "GHSA-hfld-0002-0002", "modified": "2026-01-01T00:00:00Z", "aliases": ["CVE-2099-0002"],
"summary": "a critical flaw in sampleproject", "database_specific": {"severity": "CRITICAL"},
"affected": [ { "package": { "ecosystem": "PyPI", "name": "sampleproject" },
"ranges": [ { "type": "ECOSYSTEM", "events": [ { "introduced": "0" }, { "fixed": "4.0.1" } ] } ] } ] }"#,
);
let r = start_with(
&pypi(&up, ""),
1,
"Some((required: [vulnerability]))",
&format!(
r#"intel: Some((db_dir: "{}", offline: true)),"#,
intel.path().display()
),
);
let chain = r.chain();
let _ = get(&r.http, "/production/simple/sampleproject/", &[PEP691]);
let s = pip_id("pypi", SDIST);
wait_state(&chain, &s, ArtifactState::Blocked);
// The PEP 691 page lists it yanked, with the reason.
let page = wait_for("the project page answers", || {
let g = get(&r.http, "/production/simple/sampleproject/", &[PEP691]);
(g.status == 200).then_some(g)
});
let doc: serde_json::Value = serde_json::from_slice(&page.body).unwrap();
let f = doc["files"]
.as_array()
.unwrap()
.iter()
.find(|f| f["filename"] == SDIST)
.unwrap_or_else(|| panic!("the refused sdist is listed: {doc}"));
let yanked = f["yanked"].as_str().unwrap_or_default();
assert!(
yanked.contains("refused by Holger") && yanked.contains("GHSA-hfld-0002-0002"),
"the refused file is yanked with its reason: {f}"
);
// The PEP 503 page says the same.
let html = get(&r.http, "/production/simple/sampleproject/", &[]);
let line = html
.text()
.lines()
.find(|l| l.contains(&format!("{SDIST}#sha256=")))
.map(str::to_string)
.unwrap_or_else(|| panic!("{}", html.text()));
assert!(
line.contains("data-yanked=\"refused by Holger") && line.contains("GHSA-hfld-0002-0002"),
"{line}"
);
// Its link: 403 with the reason, never the file, and no new pull.
let hits = up.hits(&format!("/files/{SDIST}"));
let file = get(&r.http, &format!("/production/pypi/{SDIST}"), &[]);
assert_eq!(file.status, 403, "{}", file.text());
assert!(
file.text().contains("GHSA-hfld-0002-0002"),
"{}",
file.text()
);
// pip shows neither the body nor the yanked reason when the download fails; it prints the
// status line ("403 Client Error: <reason phrase> for url"). So the reason phrase and a
// header carry the refusal too, on one line.
assert!(
file.reason.starts_with("Refused by Holger") && file.reason.contains("GHSA-hfld-0002-0002"),
"the status line names the refusal: {:?}",
file.reason
);
assert!(
file.headers
.get("x-holger-refusal")
.is_some_and(|v| v.contains("GHSA-hfld-0002-0002")),
"{:?}",
file.headers
);
assert_eq!(
up.hits(&format!("/files/{SDIST}")),
hits,
"asking pulled it again"
);
assert_eq!(chain.state(&s), ArtifactState::Blocked);
}
/// A stand-in clamd on a unix socket: while `healthy` is false it drops every connection (the
/// sandboxed clamd down, a worker that cannot reach it); then it answers every INSTREAM clean.
fn fake_clamd(sock: &Path, healthy: Arc<std::sync::atomic::AtomicBool>) {
let l = std::os::unix::net::UnixListener::bind(sock).unwrap();
std::thread::spawn(move || {
for conn in l.incoming() {
let Ok(mut c) = conn else { continue };
if !healthy.load(Ordering::SeqCst) {
drop(c);
continue;
}
let mut cmd = Vec::new();
let mut b = [0u8; 1];
while c.read_exact(&mut b).is_ok() && b[0] != 0 {
cmd.push(b[0]);
}
match cmd.as_slice() {
b"zPING" => {
let _ = c.write_all(b"PONG\0");
}
b"zINSTREAM" => {
loop {
let mut len = [0u8; 4];
if c.read_exact(&mut len).is_err() {
break;
}
let n = u32::from_be_bytes(len) as usize;
if n == 0 {
break;
}
let mut chunk = vec![0u8; n];
if c.read_exact(&mut chunk).is_err() {
break;
}
}
let _ = c.write_all(b"stream: OK\0");
}
_ => {
let _ = c.write_all(b"UNKNOWN COMMAND\0");
}
}
}
});
}
/// **A scan that could not run is retried, not a refusal for good** (lane HOL-FIELD-TEST P8). In
/// the field a stale scan worker made every pull fail its INGEST scan, and each file was then
/// recorded "refused at the door" and stayed refused after the worker was rebuilt. A scanner
/// that cannot run judges nothing: the file is not stored and not served (fail closed), and the
/// next request after the retry interval pulls and scans it again.
#[test]
fn a_pull_the_scanners_could_not_judge_is_scanned_again_once_they_can() {
let up = upstream();
let dir = tempfile::tempdir().unwrap();
let sock = dir.path().join("clamd.sock");
let healthy = Arc::new(std::sync::atomic::AtomicBool::new(false));
fake_clamd(&sock, healthy.clone());
let r = start_inspecting(
&pypi(&up, ""),
1,
"Some((required: [vulnerability]))",
"",
&format!(
r#"clamav: Some((socket: Some("{}"), required: true))"#,
sock.display()
),
);
let chain = r.chain();
let id = pip_id("pypi", WHEEL);
let path = format!("/production/pypi/{WHEEL}");
// 1. clamd is down: the pull is not judged, nothing is stored, nothing is served.
let first = get(&r.http, &path, &[]);
assert_eq!(first.status, 503, "{}", first.text());
let held = get_settled(&r.http, &path, &[]);
assert_eq!(held.status, 404, "{}", held.text());
assert!(chain.status(&id).is_none(), "{:#?}", chain.history(&id));
assert!(
!held
.text()
.contains("refused by the intake chain and never served"),
"a scan that could not run is no refusal for good: {}",
held.text()
);
// 2. clamd is back: the next request after the retry interval pulls and scans it again.
healthy.store(true, Ordering::SeqCst);
std::thread::sleep(Duration::from_millis(1100));
let _ = get(&r.http, &path, &[]);
wait_state(&chain, &id, ArtifactState::PolicyChecked);
approve(&chain, &id, "bob");
let whl = get_settled(&r.http, &path, &[]);
assert_eq!(whl.status, 200, "{}", whl.text());
}
/// `dir` made unwritable (`0555`) or writable again (`0755`): a store that cannot be written.
fn writable(dir: &Path, yes: bool) {
use std::os::unix::fs::PermissionsExt as _;
std::fs::create_dir_all(dir).unwrap();
let mode = if yes { 0o755 } else { 0o555 };
std::fs::set_permissions(dir, std::fs::Permissions::from_mode(mode)).unwrap();
}
/// Ask for `path` until `id` stands at `want` (each request may queue the retry).
fn ask_until(r: &Running, path: &str, id: &traits::ArtifactId, want: ArtifactState) {
let chain = r.chain();
let t = Instant::now();
while chain.state(id) != want {
assert!(
t.elapsed() < Duration::from_secs(90),
"{} never reached {want}; it stands at {} ({:#?}); pull records: {}",
id.name,
chain.state(id),
chain.history(id),
chain.pull().map(|p| p.debug_records()).unwrap_or_default()
);
let _ = get(&r.http, path, &[]);
std::thread::sleep(Duration::from_millis(250));
}
}
/// The pull records name a storage failure of `name`.
fn wait_store_failed(r: &Running, name: &str) -> String {
let chain = r.chain();
let t = Instant::now();
loop {
let rec = chain.pull().map(|p| p.debug_records()).unwrap_or_default();
if let Some(l) = rec
.lines()
.find(|l| l.contains("\"store-failed\"") && l.contains(name))
{
return l.to_string();
}
assert!(
t.elapsed() < Duration::from_secs(30),
"no store-failed record of {name}; the pull records: {rec}"
);
std::thread::sleep(Duration::from_millis(50));
}
}
/// **A full disk at ingest is no verdict** (lane HOL-ENOSPC-SETS, field row
/// HOL-PULL-ENOSPC-BLOCK). The incoming store cannot be written: nothing is stored, nothing
/// is served, the pull is recorded `store-failed` with the cause by name, and the next request
/// after the retry interval pulls it again once the store can be written. Never "refused at
/// the door" for good.
#[test]
fn a_pull_the_incoming_store_could_not_take_is_retried_once_it_can() {
let up = upstream();
let r = start(&pypi(&up, ""), 1, "Some((required: [vulnerability]))");
let chain = r.chain();
let id = pip_id("pypi", WHEEL);
let path = format!("/production/pypi/{WHEEL}");
let incoming = r._data.path().join("incoming");
writable(&incoming, false);
let first = get(&r.http, &path, &[]);
assert_eq!(first.status, 503, "{}", first.text());
let rec = wait_store_failed(&r, &id.name);
assert!(
rec.contains("holger-intake-13"),
"the record names the storage failure by its code: {rec}"
);
assert!(chain.status(&id).is_none(), "{:#?}", chain.history(&id));
let held = get(&r.http, &path, &[]);
assert_ne!(held.status, 200, "{}", held.text());
assert!(
!held
.text()
.contains("refused by the intake chain and never served"),
"a store that could not be written is no refusal for good: {}",
held.text()
);
// The disk has room again: the pull is asked again and walks on.
writable(&incoming, true);
ask_until(&r, &path, &id, ArtifactState::PolicyChecked);
approve(&chain, &id, "bob");
let whl = get_settled(&r.http, &path, &[]);
assert_eq!(whl.status, 200, "{}", whl.text());
}
/// **The field case** (oden, 2026-11-01, `/home` at 100 %): the scan-evidence store could not
/// be written at SCANNED and the pulled artifact was BLOCKED for good. A storage failure is no
/// verdict: the artifact stays where it stood (INGESTED, never served), the pull is recorded
/// `store-failed`, and it is walked again once the store can be written.
#[test]
fn a_pull_the_scan_evidence_store_could_not_take_is_held_and_walked_again() {
let up = upstream();
let r = start(&pypi(&up, ""), 1, "Some((required: [vulnerability]))");
let chain = r.chain();
let id = pip_id("pypi", WHEEL);
let path = format!("/production/pypi/{WHEEL}");
let evidence = r._data.path().join("holger-scan-evidence");
writable(&evidence, false);
let first = get(&r.http, &path, &[]);
assert_eq!(first.status, 503, "{}", first.text());
let rec = wait_store_failed(&r, &id.name);
assert!(rec.contains("holger-intake-13"), "{rec}");
assert_eq!(
chain.state(&id),
ArtifactState::Ingested,
"never BLOCKED for a disk: {:#?}",
chain.history(&id)
);
let held = get(&r.http, &path, &[]);
assert_ne!(held.status, 200, "{}", held.text());
writable(&evidence, true);
ask_until(&r, &path, &id, ArtifactState::PolicyChecked);
approve(&chain, &id, "bob");
let whl = get_settled(&r.http, &path, &[]);
assert_eq!(whl.status, 200, "{}", whl.text());
}