server/cli/tests/downstream_serves_the_build_systems.rs
//! **H3: the downstream serves the on-prem build systems** (Holger v2.0, A31 §6).
//!
//! An upstream Holger walks one package of each kind to AVAILABLE with signed scan
//! evidence and two-person approval: a wheel, a crate, a Maven POM and JAR, an OCI
//! image and a Go module. It exports them (`holger export`) and the candidate is
//! approved and release-signed. The DOWNSTREAM is the real `--features downstream`
//! binary (`holger-downstream`, built on its own: no scanner, no upstream code in it).
//!
//! * **USB**: `holger-downstream bundle import` verifies the bundle and lands every
//! package AVAILABLE; the started downstream serves them with each kind's own index:
//! PEP 503 and PEP 691 with PEP 658/714 core metadata, the cargo sparse index (the
//! index line read from the sealed `.crate`), `maven-metadata.xml` and checksums,
//! the OCI distribution API (the tag from the sealed layout), and the GOPROXY
//! protocol. An upload is refused (`holger-downstream-3`) and a package the
//! upstream never sent is "not sent" (`holger-downstream-4`), never fetched.
//! * **The guide's spelling** (F9, the field test of 2026-11-01): the same downstream written
//! with `steps:` and `sources:`, each source with no `url`, serves the image (manifest,
//! blobs, referrers), the Maven and cargo files and the wheel, and dials nothing; its real
//! client arm is `podman pull` (ignored).
//! * **Pull**: the same signed bundle published in the upstream's `bundles`
//! repository reaches a second downstream over its one line out (the `mirror:`
//! block with one Holger source, into its `bundle.inbox`), is verified the same way
//! and served.
//! * **Zero egress** (the airgap rule, at run time): every downstream process runs with
//! a connect trap (`LD_PRELOAD`, a C shim built here with `cc`) that records each
//! `connect`, `sendto` and `getaddrinfo` it makes. The USB downstream dials nothing at
//! all; the pulling one dials its configured upstream and nothing else. The suite's
//! gate also runs this file in a container with `--network none`
//! (`HOLGER_TEST_NETWORK_NONE=1` makes the test check it has no route out).
//! * **Invalidation** (H4 on the downstream): the upstream invalidates a package after
//! the bundle landed, the signed sync manifest goes over on the stick, `sync apply` on
//! the downstream holds it, and the started downstream serves nothing of it.
//! * **Real clients**: the H6 end-to-end gate (`tests/e2e_internet_to_build_systems.rs`,
//! run by `cargo xtask e2e` in its clients image with `--network none`) builds with real
//! `pip`, `uv`, `cargo`, `mvn`, `skopeo` and `go` against this same downstream binary.
//!
//! The downstream binary: `HOLGER_DOWNSTREAM_BIN`, or `holger-downstream` beside the
//! test's `holger` (`cargo build -p holger --no-default-features --features downstream
//! --bin holger-downstream` into the same target directory first).
use std::collections::BTreeMap;
use std::io::{Read, Write};
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Output, Stdio};
use std::sync::Arc;
use std::time::{Duration, Instant};
use server_lib::promote::intake::{ArtifactState, IntakeChain, TransitionRequest};
use server_lib::scanner::{
ScanOutcome, ScanPipeline, ScanSubject, Scanner, ScannerKind, ScannerSet,
};
use server_lib::ArtifactId;
use traits::test_support::hold_port;
const BIN: &str = env!("CARGO_BIN_EXE_holger");
const WHEEL: &str = "demo_pkg-1.0.0-py3-none-any.whl";
const CRATE_ID: &str = "crates/demo-crate/demo-crate-0.1.0.crate";
const POM_ID: &str = "maven/org/demo/demo-lib/1.0.0/demo-lib-1.0.0.pom";
const JAR_ID: &str = "maven/org/demo/demo-lib/1.0.0/demo-lib-1.0.0.jar";
const GO_ZIP_ID: &str = "go/example.com/hello/@v/v1.0.0.zip";
const GO_MOD_ID: &str = "go/example.com/hello/@v/v1.0.0.mod";
const IMAGE: &str = "demo/hello";
const TAG: &str = "1.0";
fn id(name: &str) -> ArtifactId {
ArtifactId {
namespace: None,
name: name.into(),
version: String::new(),
}
}
fn text(o: &Output) -> String {
format!(
"{}{}",
String::from_utf8_lossy(&o.stdout),
String::from_utf8_lossy(&o.stderr)
)
}
fn sha256(b: &[u8]) -> String {
update_bundle::digest::sha256_hex(b)
}
// ── the packages ────────────────────────────────────────────────────────────
const METADATA: &str =
"Metadata-Version: 2.1\nName: demo-pkg\nVersion: 1.0.0\nLicense-Expression: MIT\n";
fn wheel() -> Vec<u8> {
zip_entry::fixture::zip(&[
("demo_pkg/__init__.py", b"__version__ = \"1.0.0\"\n"),
("demo_pkg-1.0.0.dist-info/METADATA", METADATA.as_bytes()),
("demo_pkg-1.0.0.dist-info/WHEEL", b"Wheel-Version: 1.0\nGenerator: holger-test\nRoot-Is-Purelib: true\nTag: py3-none-any\n"),
("demo_pkg-1.0.0.dist-info/RECORD", b""),
])
}
fn tar_gz(entries: &[(&str, &[u8])]) -> (Vec<u8>, Vec<u8>) {
let mut tar = tar::Builder::new(Vec::new());
for (path, body) in entries {
let mut h = tar::Header::new_gnu();
h.set_size(body.len() as u64);
h.set_mode(0o644);
h.set_mtime(0);
h.set_cksum();
tar.append_data(&mut h, path, *body).unwrap();
}
let raw = tar.into_inner().unwrap();
let mut gz = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());
gz.write_all(&raw).unwrap();
(gz.finish().unwrap(), raw)
}
fn dot_crate() -> Vec<u8> {
let manifest = b"[package]\nname = \"demo-crate\"\nversion = \"0.1.0\"\nedition = \"2021\"\ndescription = \"a demo\"\nlicense = \"MIT\"\n\n[features]\ndefault = []\nloud = []\n";
tar_gz(&[
("demo-crate-0.1.0/Cargo.toml", manifest),
(
"demo-crate-0.1.0/src/lib.rs",
b"pub fn greet() -> &'static str { \"hello from holger\" }\n",
),
])
.0
}
fn pom() -> Vec<u8> {
br#"<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0">
<modelVersion>4.0.0</modelVersion>
<groupId>org.demo</groupId>
<artifactId>demo-lib</artifactId>
<version>1.0.0</version>
<packaging>jar</packaging>
</project>
"#
.to_vec()
}
fn jar() -> Vec<u8> {
zip_entry::fixture::zip(&[
(
"META-INF/MANIFEST.MF",
b"Manifest-Version: 1.0\nCreated-By: holger-test\n",
),
("org/demo/Demo.txt", b"demo\n"),
])
}
fn go_zip() -> Vec<u8> {
zip_entry::fixture::zip(&[
(
"example.com/hello@v1.0.0/go.mod",
b"module example.com/hello\n\ngo 1.21\n",
),
(
"example.com/hello@v1.0.0/hello.go",
b"package hello\n\nfunc Hello() string { return \"hello from holger\" }\n",
),
])
}
/// An OCI image (one layer, one file) as the sealed layout the chain carries.
fn image() -> (ArtifactId, Vec<u8>, String) {
let (layer, raw) = tar_gz(&[("hello.txt", b"hello from holger\n")]);
let digest = |b: &[u8]| format!("sha256:{}", sha256(b));
let config = serde_json::to_vec(&serde_json::json!({
"architecture": "amd64", "os": "linux",
"rootfs": { "type": "layers", "diff_ids": [digest(&raw)] },
"config": {}
}))
.unwrap();
let manifest = serde_json::to_vec(&serde_json::json!({
"schemaVersion": 2,
"mediaType": "application/vnd.oci.image.manifest.v1+json",
"config": { "mediaType": "application/vnd.oci.image.config.v1+json", "digest": digest(&config), "size": config.len() },
"layers": [ { "mediaType": "application/vnd.oci.image.layer.v1.tar+gzip", "digest": digest(&layer), "size": layer.len() } ]
}))
.unwrap();
let top = digest(&manifest);
let mut p = server_lib::pull_intake::oci::Pulled {
name: IMAGE.into(),
tag: Some(TAG.into()),
digest: top.clone(),
..Default::default()
};
p.manifests.insert(
top.clone(),
(
"application/vnd.oci.image.manifest.v1+json".into(),
manifest.clone(),
),
);
p.blobs.insert(digest(&config), config);
p.blobs.insert(digest(&layer), layer);
let tar = server_lib::pull_intake::oci::layout_tar(&p).unwrap();
(
server_lib::pull_intake::oci_id("cgr", IMAGE, &top),
with_attachments(tar, &top, manifest.len()),
top,
)
}
/// HOL-REFERRERS: what came WITH the image from its source, in the layout as the
/// pull-through keeps it: a cosign `.sig` tag and an SPDX SBOM referrer, each with
/// its blobs (`holger/media-types.json`: `images`, `attached_tags`, `referrers`).
fn attachments(top: &str, top_size: usize) -> (Vec<u8>, Vec<u8>, Vec<u8>, Vec<u8>) {
let digest = |b: &[u8]| format!("sha256:{}", sha256(b));
let sig_layer =
br#"{"critical":{"type":"cosign container image signature"},"optional":null}"#.to_vec();
let sig = serde_json::to_vec(&serde_json::json!({
"schemaVersion": 2, "mediaType": "application/vnd.oci.image.manifest.v1+json",
"config": { "mediaType": "application/vnd.oci.empty.v1+json", "digest": digest(b"{}"), "size": 2 },
"layers": [ { "mediaType": "application/vnd.dev.cosign.simplesigning.v1+json", "digest": digest(&sig_layer), "size": sig_layer.len(),
"annotations": { "dev.cosignproject.cosign/signature": "MEUCIQ" } } ]
}))
.unwrap();
let sbom_layer = br#"{"spdxVersion":"SPDX-2.3","name":"demo/hello"}"#.to_vec();
let sbom = serde_json::to_vec(&serde_json::json!({
"schemaVersion": 2, "mediaType": "application/vnd.oci.image.manifest.v1+json", "artifactType": "application/spdx+json",
"config": { "mediaType": "application/vnd.oci.empty.v1+json", "digest": digest(b"{}"), "size": 2 },
"layers": [ { "mediaType": "application/spdx+json", "digest": digest(&sbom_layer), "size": sbom_layer.len() } ],
"subject": { "mediaType": "application/vnd.oci.image.manifest.v1+json", "digest": top, "size": top_size }
}))
.unwrap();
(sig, sig_layer, sbom, sbom_layer)
}
fn with_attachments(tar: Vec<u8>, top: &str, top_size: usize) -> Vec<u8> {
use std::io::Read as _;
let digest = |b: &[u8]| format!("sha256:{}", sha256(b));
let (sig, sig_layer, sbom, sbom_layer) = attachments(top, top_size);
let mut files: BTreeMap<String, Vec<u8>> = BTreeMap::new();
let mut ar = tar::Archive::new(tar.as_slice());
for e in ar.entries().unwrap() {
let mut e = e.unwrap();
let path = e.path().unwrap().to_string_lossy().to_string();
let mut b = Vec::new();
e.read_to_end(&mut b).unwrap();
files.insert(path, b);
}
let mut meta: serde_json::Value =
serde_json::from_slice(&files["holger/media-types.json"]).unwrap();
meta["manifests"][digest(&sig)] = "application/vnd.oci.image.manifest.v1+json".into();
meta["manifests"][digest(&sbom)] = "application/vnd.oci.image.manifest.v1+json".into();
meta["images"] = serde_json::json!([top]);
meta["attached_tags"] =
serde_json::json!({ format!("sha256-{}.sig", &top[7..]): digest(&sig) });
meta["referrers"] = serde_json::json!({ top: [ {
"mediaType": "application/vnd.oci.image.manifest.v1+json", "digest": digest(&sbom), "size": sbom.len(), "artifactType": "application/spdx+json"
} ] });
files.insert(
"holger/media-types.json".into(),
serde_json::to_vec_pretty(&meta).unwrap(),
);
for b in [sig, sig_layer, sbom, sbom_layer, b"{}".to_vec()] {
files.insert(format!("blobs/sha256/{}", sha256(&b)), b);
}
let mut out = tar::Builder::new(Vec::new());
for (path, bytes) in &files {
let mut h = tar::Header::new_ustar();
h.set_size(bytes.len() as u64);
h.set_mode(0o644);
h.set_mtime(0);
h.set_entry_type(tar::EntryType::Regular);
out.append_data(&mut h, path, bytes.as_slice()).unwrap();
}
out.into_inner().unwrap()
}
fn packages() -> Vec<(ArtifactId, Vec<u8>)> {
let (img, img_bytes, _) = image();
vec![
(id(WHEEL), wheel()),
(id(CRATE_ID), dot_crate()),
(id(POM_ID), pom()),
(id(JAR_ID), jar()),
(img, img_bytes),
(id(GO_ZIP_ID), go_zip()),
(
id(GO_MOD_ID),
b"module example.com/hello\n\ngo 1.21\n".to_vec(),
),
]
}
// ── the upstream ────────────────────────────────────────────────────────────
/// The upstream's vulnerability scan, stood in by a scanner that reports clean. Its
/// evidence is stored and SIGNED exactly as a real scanner's is; what the downstream
/// verifies is that signature, never the scanner.
struct CleanVulnerabilityScan;
impl Scanner for CleanVulnerabilityScan {
fn id(&self) -> &str {
"upstream-vulnerability-scan"
}
fn version(&self) -> String {
"test 1".into()
}
fn kind(&self) -> ScannerKind {
ScannerKind::Vulnerability
}
fn scan(&self, subject: &ScanSubject<'_>) -> anyhow::Result<ScanOutcome> {
Ok(ScanOutcome::pass(format!(
"{}: no advisory",
subject.file_name
)))
}
}
/// The upstream's SBOM step, stood in the same way: a minimal CycloneDX document,
/// stored and signed with the evidence (the release policy requires one).
struct SbomOfTheUpstream;
impl Scanner for SbomOfTheUpstream {
fn id(&self) -> &str {
"upstream-sbom"
}
fn version(&self) -> String {
"test 1".into()
}
fn kind(&self) -> ScannerKind {
ScannerKind::Sbom
}
fn scan(&self, subject: &ScanSubject<'_>) -> anyhow::Result<ScanOutcome> {
let doc = serde_json::json!({
"bomFormat": "CycloneDX", "specVersion": "1.5", "version": 1,
"metadata": { "component": { "type": "library", "name": subject.file_name } },
"components": []
});
Ok(ScanOutcome::pass(format!("{}: an SBOM", subject.file_name)).with_document(doc))
}
}
fn intake_repos(d: &Path, extra: &str) -> String {
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("{}/{name}")),"#,
d.display()
)
};
format!(
"{} {} {} {} {extra}",
repo("incoming"),
repo("quarantine"),
repo("approved"),
repo("production")
)
}
fn ladder(rules: &str, trust_policy: Option<&Path>) -> String {
let tp = trust_policy
.map(|p| format!(r#"trust_policy: Some("{}"),"#, p.display()))
.unwrap_or_default();
format!(
r#"promotion: Some((
ladder_name: "intake",
stages: [
(name: "incoming", repo: "incoming"),
(name: "quarantine", repo: "quarantine"),
(name: "approved", repo: "approved"),
(name: "production", repo: "production"),
],
intake: Some(({rules}
incoming: "incoming", quarantine: "quarantine",
approved: "approved", production: "production",
{tp}
)),
)),"#
)
}
struct Upstream {
dir: tempfile::TempDir,
config: PathBuf,
pki: bench_scenarios::secure_update::AttestPki,
/// The gRPC port of `start_upstream`'s server (the pull test's line).
grpc: u16,
}
/// The upstream: every package walked to AVAILABLE through its intake chain with
/// signed scan evidence, approved by a person who did not import it.
fn upstream() -> Upstream {
let dir = tempfile::tempdir().unwrap();
let d = dir.path().join("up");
std::fs::create_dir_all(d.join("bundles")).unwrap();
let pki = bench_scenarios::secure_update::mint_attestation_pki(&d.join("pki"), "upstream-scan")
.unwrap();
let grpc = hold_port();
let http = hold_port();
let bundles = format!(
r#"(ron_name: "bundles", ron_repo_type: "raw", ron_upstreams: [], ron_in: None,
ron_out: Some((ron_storage_endpoint: "storage", ron_exposed_endpoint: "door")), ron_data_dir: Some("{}/bundles")),"#,
d.display()
);
let ron = format!(
r#"(
repositories: [ {repos} ],
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),
attestation: Some((
trust_roots: ["{root}"],
signer_key: Some("{key}"),
signer_cert: Some("{cert}"),
signer_alg: Some("ecdsa-p256"),
)),
{ladder}
)"#,
repos = intake_repos(&d, &bundles),
d = d.display(),
root = pki.root_der.display(),
key = pki.key_pkcs8.display(),
cert = pki.cert_der.display(),
ladder = ladder("rules: (sandbox: off),", None),
);
let config = dir.path().join("upstream.ron");
std::fs::write(&config, ron).unwrap();
let mut h = server_lib::read_ron_config(&config).unwrap();
h.instantiate_backends().unwrap();
let audit = h.audit_log().unwrap();
let settings = h
.promotion
.as_ref()
.and_then(|p| p.intake.clone())
.expect("the intake block");
let incoming = h.get_repo("incoming").unwrap().clone();
let pipeline = ScanPipeline::new(
ScannerSet::new()
.with(Arc::new(CleanVulnerabilityScan))
.with(Arc::new(SbomOfTheUpstream))
.require(ScannerKind::Vulnerability)
.require(ScannerKind::Sbom),
h.scan_evidence_store(audit.clone()),
)
.with_attestation(
h.attestation_store().unwrap().expect("a store"),
h.attestation_signer().unwrap().expect("a signer"),
);
let chain = IntakeChain::new(
settings,
h.promotion_engine().unwrap(),
h.data_root().join("holger-prod-store"),
audit,
incoming.clone(),
)
.with_scanner(Arc::new(pipeline));
for (pid, bytes) in packages() {
chain
.ingest("ida", "test", "incoming", &pid, &bytes, |a| {
a.store(&*incoming, &bytes)
})
.unwrap_or_else(|e| panic!("ingest {}: {e}", pid.name));
for (to, who) in [
(ArtifactState::Scanned, "ida"),
(ArtifactState::PolicyChecked, "ida"),
(ArtifactState::Approved, "bob"),
(ArtifactState::Sealed, "bob"),
(ArtifactState::Promoted, "bob"),
(ArtifactState::Available, "bob"),
] {
chain
.advance(TransitionRequest::new(pid.clone(), to, who))
.unwrap_or_else(|e| panic!("{} → {to}: {e}", pid.name));
}
}
Upstream {
dir,
config,
pki,
grpc,
}
}
/// Export every package, approve it twice and release-sign it: `(bundle, policy)`.
fn signed_bundle(up: &Upstream, name: &str) -> (PathBuf, PathBuf) {
let cand = up.dir.path().join(format!("{name}.cand"));
let mut c = Command::new(BIN);
c.args(["export", "--config"])
.arg(&up.config)
.arg("--candidate")
.arg(&cand);
c.args([
"--bundle-id",
name,
"--sequence",
"1",
"--supplier",
"upstream",
"--product",
"on-prem",
"--release-version",
"1",
]);
for (pid, _) in packages() {
c.args(["--artifact", &pid.name]);
}
let out = c.output().unwrap();
assert!(out.status.success(), "export: {}", text(&out));
let members = update_bundle::prepare::read_members_dir(&cand).unwrap();
let bundle = up.dir.path().join(format!("{name}.holger.znippy"));
let policy = bench_scenarios::secure_update::sign_candidate_and_pack(members, &bundle).unwrap();
let policy_path = up.dir.path().join(format!("{name}.policy.ron"));
std::fs::write(&policy_path, policy.to_ron().unwrap()).unwrap();
(bundle, policy_path)
}
// ── the downstream ──────────────────────────────────────────────────────────
/// The real downstream binary (no scanner, no upstream code in it).
fn downstream_bin() -> PathBuf {
if let Ok(p) = std::env::var("HOLGER_DOWNSTREAM_BIN") {
return PathBuf::from(p);
}
let beside = Path::new(BIN).with_file_name("holger-downstream");
assert!(
beside.exists(),
"no downstream binary: build it first with `cargo build --manifest-path holger-downstream/Cargo.toml \
--features sandbox-off --bin holger-downstream` (into this target directory), or set HOLGER_DOWNSTREAM_BIN"
);
beside
}
struct Downstream {
dir: PathBuf,
config: PathBuf,
http: u16,
child: Option<Child>,
}
// ── the connect trap (zero egress at run time) ──────────────────────────────
// The trap is shared with the default build's zero-outbound test (`connect_trap.rs`).
#[path = "support/connect_trap.rs"]
mod connect_trap;
use connect_trap::{dialled, trapped};
/// `--network none` when the gate asks for it: no default route out of this namespace.
fn assert_no_route_out_when_asked() {
if std::env::var("HOLGER_TEST_NETWORK_NONE").as_deref() != Ok("1") {
return;
}
let route = std::fs::read_to_string("/proc/net/route").unwrap_or_default();
let default = route
.lines()
.skip(1)
.any(|l| l.split_whitespace().nth(1) == Some("00000000"));
assert!(
!default,
"HOLGER_TEST_NETWORK_NONE=1 but this namespace has a default route:\n{route}"
);
}
impl Drop for Downstream {
fn drop(&mut self) {
if let Some(mut c) = self.child.take() {
let _ = c.kill();
let _ = c.wait();
}
}
}
/// How a downstream config writes its chain and its serve-only sources.
#[derive(Clone, Copy)]
enum Spelling {
/// The `promotion:` ladder with its `intake:` block, and the `pull_through:` block.
PullThrough,
/// The guide's spelling (README, "Three products by hand"): `steps:`, `rules:` and
/// `sources:`, each source with no `url`.
Sources,
}
/// A downstream config: the intake chain (no scanner), the serve-only sources, the
/// upstream's attestation roots, and (pull mode) the one upstream Holger.
fn downstream(
up: &Upstream,
name: &str,
trust_policy: Option<&Path>,
pull_from: Option<u16>,
) -> Downstream {
downstream_spelled(up, name, trust_policy, pull_from, Spelling::PullThrough)
}
fn downstream_spelled(
up: &Upstream,
name: &str,
trust_policy: Option<&Path>,
pull_from: Option<u16>,
spelling: Spelling,
) -> Downstream {
let dir = up.dir.path().join(name);
std::fs::create_dir_all(&dir).unwrap();
let (http, grpc) = (hold_port(), hold_port());
let inbox = format!(
r#"(ron_name: "inbox", ron_repo_type: "raw", ron_upstreams: [], ron_in: None, ron_out: None, ron_data_dir: Some("{}/bundle.inbox")),"#,
dir.display()
);
let mirror = match pull_from {
Some(port) => format!(
r#"boundary: Fetch,
mirror: Some((ron_state_dir: "{state}", ron_sources: [ (ron_name: "upstream", ron_kind: Holger, ron_endpoint: "http://127.0.0.1:{port}",
ron_credential: Anonymous, ron_every_secs: 1, ron_retry_secs: 1,
ron_repositories: [ (ron_from: "bundles", ron_to: "inbox", ron_unproven: TrustSize) ]) ])),"#,
state = dir.join("holger-mirror").display()
),
None => String::new(),
};
// The binary under test is the toolchain image's: no container engine, so its
// verify-bundle jobs run in-process (`holger-downstream --features sandbox-off`).
let chain = match spelling {
Spelling::PullThrough => format!(
r#"{ladder}
pull_through: Some((sources: [
(name: "crates", kind: crates),
(name: "maven", kind: maven),
(name: "cgr", kind: oci),
(name: "go", kind: go),
])),"#,
ladder = ladder("rules: (sandbox: off),", trust_policy),
),
Spelling::Sources => format!(
r#"steps: ({tp}),
rules: (sandbox: off),
sources: [
(name: "cgr", kind: oci),
(name: "maven", kind: maven),
(name: "crates", kind: crates),
],"#,
tp = trust_policy
.map(|p| format!(r#"trust_policy: Some("{}")"#, p.display()))
.unwrap_or_default(),
),
};
let ron = format!(
r#"(
repositories: [ {repos} ],
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),
attestation: Some((
trust_roots: ["{root}"],
enrolled_signers: ["{cert}"],
)),
{chain}
{mirror}
)"#,
repos = intake_repos(&dir, &inbox),
d = dir.display(),
root = up.pki.root_der.display(),
cert = up.pki.cert_der.display(),
);
let config = up.dir.path().join(format!("{name}.ron"));
std::fs::write(&config, ron).unwrap();
Downstream {
dir,
config,
http,
child: None,
}
}
fn start(bin: &Path, config: &Path, http: u16, log: &Path, connects: Option<&Path>) -> Child {
let log_file = std::fs::File::create(log).unwrap();
let mut c = Command::new(bin);
if let Some(connects) = connects {
trapped(&mut c, connects);
}
let child = c
.args(["start", "--config"])
.arg(config)
.stdout(Stdio::from(log_file.try_clone().unwrap()))
.stderr(Stdio::from(log_file))
.spawn()
.unwrap();
let t = Instant::now();
while std::net::TcpStream::connect(("127.0.0.1", http)).is_err() {
assert!(
t.elapsed() < Duration::from_secs(60),
"the server on :{http} never listened:\n{}",
std::fs::read_to_string(log).unwrap_or_default()
);
std::thread::sleep(Duration::from_millis(100));
}
child
}
impl Downstream {
/// The trap's log of every downstream process of this test.
fn connects(&self) -> PathBuf {
self.dir.join("connects.log")
}
/// The admin's first-import authorization on this side (NCSC gap 2): the bundles of
/// [`signed_bundle`] are vendor `upstream`, product `on-prem`, sequence 1.
fn allow_first_import(&self) {
let out = self.cli(
&[
"bundle",
"allow-first-import",
"--config",
self.config.to_str().unwrap(),
"--vendor",
"upstream",
"--product",
"on-prem",
"--sequence",
"1",
]
.map(std::ffi::OsStr::new),
);
assert!(
out.status.success(),
"the admin's first-import authorization: {}",
text(&out)
);
}
fn import(&self, bundle: &Path, policy: &Path) -> Output {
self.allow_first_import();
trapped(&mut Command::new(downstream_bin()), &self.connects())
.args(["bundle", "import", "--config"])
.arg(&self.config)
.arg("--bundle")
.arg(bundle)
.arg("--policy")
.arg(policy)
.output()
.unwrap()
}
fn start(&mut self) {
self.child = Some(start(
&downstream_bin(),
&self.config,
self.http,
&self.dir.join("server.log"),
Some(&self.connects()),
));
}
fn stop(&mut self) {
if let Some(mut c) = self.child.take() {
let _ = c.kill();
let _ = c.wait();
}
}
/// One CLI verb of the downstream binary, trapped.
fn cli(&self, args: &[&std::ffi::OsStr]) -> Output {
trapped(&mut Command::new(downstream_bin()), &self.connects())
.args(args)
.output()
.unwrap()
}
fn log(&self) -> String {
std::fs::read_to_string(self.dir.join("server.log")).unwrap_or_default()
}
}
// ── HTTP ────────────────────────────────────────────────────────────────────
struct Reply {
status: u16,
headers: BTreeMap<String, String>,
body: Vec<u8>,
}
impl Reply {
fn text(&self) -> String {
String::from_utf8_lossy(&self.body).to_string()
}
}
fn request(port: u16, method: &str, path: &str, headers: &[(&str, &str)], body: &[u8]) -> Reply {
let mut s = std::net::TcpStream::connect(("127.0.0.1", port)).unwrap();
let mut head = format!("{method} {path} HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: close\r\nContent-Length: {}\r\n", body.len());
for (k, v) in headers {
head.push_str(&format!("{k}: {v}\r\n"));
}
head.push_str("\r\n");
s.write_all(head.as_bytes()).unwrap();
s.write_all(body).unwrap();
let mut raw = Vec::new();
s.read_to_end(&mut raw).unwrap();
let split = raw
.windows(4)
.position(|w| w == b"\r\n\r\n")
.expect("an HTTP head");
let head = String::from_utf8_lossy(&raw[..split]).to_string();
let status: u16 = head.split_whitespace().nth(1).unwrap().parse().unwrap();
let headers: BTreeMap<String, String> = head
.lines()
.skip(1)
.filter_map(|l| l.split_once(':'))
.map(|(k, v)| (k.trim().to_ascii_lowercase(), v.trim().to_string()))
.collect();
let mut body = raw[split + 4..].to_vec();
if headers
.get("transfer-encoding")
.is_some_and(|v| v.contains("chunked"))
{
let mut out = Vec::new();
let mut rest = &body[..];
loop {
let nl = rest.windows(2).position(|w| w == b"\r\n").unwrap();
let n = usize::from_str_radix(std::str::from_utf8(&rest[..nl]).unwrap().trim(), 16)
.unwrap();
if n == 0 {
break;
}
out.extend_from_slice(&rest[nl + 2..nl + 2 + n]);
rest = &rest[nl + 2 + n + 2..];
}
body = out;
}
Reply {
status,
headers,
body,
}
}
fn get(port: u16, path: &str) -> Reply {
request(port, "GET", path, &[], &[])
}
/// ★ HOL-REFERRERS: **the downstream serves what came with the image, and the
/// upstream Holger's own evidence about it, as referrers**, so an on-prem cluster's
/// `cosign verify` / `cosign verify-attestation` works with no line out: the `.sig`
/// tag and the SBOM referrer byte for byte, the upstream's signed scan evidence and
/// CycloneDX SBOM as referrers of the image digest (each a Sigstore bundle around the
/// DSSE envelope the bundle carried, signed by the upstream's key) and as layers of
/// the `.att` tag.
fn assert_serves_the_referrers(port: u16, signer_cert_der: &[u8], log: &dyn Fn() -> String) {
use base64::Engine as _;
let (_, _, top) = image();
let digest = |b: &[u8]| format!("sha256:{}", sha256(b));
let base = "/v2/production/cgr/demo/hello";
let (sig, _, sbom, sbom_layer) = attachments(&top, 0);
let g = get(port, &format!("{base}/manifests/sha256-{}.sig", &top[7..]));
assert_eq!(
(g.status, g.body.clone()),
(200, sig),
"the .sig tag, byte for byte: {}\n{}",
g.text(),
log()
);
let idx = get(port, &format!("{base}/referrers/{top}"));
assert_eq!(idx.status, 200, "{}\n{}", idx.text(), log());
let v: serde_json::Value = serde_json::from_slice(&idx.body).unwrap();
let m = v["manifests"].as_array().unwrap();
let sbom_ref = m
.iter()
.find(|d| d["artifactType"] == "application/spdx+json")
.unwrap_or_else(|| panic!("the image's own SBOM referrer: {v}"));
let sbom_digest = sbom_ref["digest"].as_str().unwrap().to_string();
let got = get(port, &format!("{base}/manifests/{sbom_digest}"));
assert_eq!(got.status, 200);
let gv: serde_json::Value = serde_json::from_slice(&got.body).unwrap();
assert_eq!(gv["layers"][0]["digest"], digest(&sbom_layer));
let _ = sbom;
assert_eq!(
get(port, &format!("{base}/blobs/{}", digest(&sbom_layer))).body,
sbom_layer
);
// The upstream's evidence: two referrers (scan evidence, SBOM), each a bundle.
let keyid = sha256(signer_cert_der);
let holger: Vec<&serde_json::Value> = m
.iter()
.filter(|d| d["annotations"]["rs.holger.evidence"] == "holger")
.collect();
let types: std::collections::BTreeSet<&str> = holger
.iter()
.filter_map(|d| d["annotations"]["dev.sigstore.bundle.predicateType"].as_str())
.collect();
assert_eq!(
types,
[
"https://cyclonedx.org/bom",
"https://holger.rs/attestation/scan-evidence/v1"
]
.into_iter()
.collect(),
"{v}"
);
for d in &holger {
assert_eq!(
d["artifactType"],
"application/vnd.dev.sigstore.bundle.v0.3+json"
);
let rm = get(
port,
&format!("{base}/manifests/{}", d["digest"].as_str().unwrap()),
);
assert_eq!(rm.status, 200, "{}", rm.text());
let rv: serde_json::Value = serde_json::from_slice(&rm.body).unwrap();
assert_eq!(rv["subject"]["digest"], top.as_str());
let b = get(
port,
&format!(
"{base}/blobs/{}",
rv["layers"][0]["digest"].as_str().unwrap()
),
);
assert_eq!(b.status, 200);
let bundle: serde_json::Value = serde_json::from_slice(&b.body).unwrap();
assert_eq!(
bundle["verificationMaterial"]["publicKey"]["hint"],
keyid.as_str(),
"signed by the upstream's evidence key"
);
let payload = base64::engine::general_purpose::STANDARD
.decode(bundle["dsseEnvelope"]["payload"].as_str().unwrap())
.unwrap();
let st: serde_json::Value = serde_json::from_slice(&payload).unwrap();
assert!(
st["subject"]
.as_array()
.unwrap()
.iter()
.any(|s| s["digest"]["sha256"] == &top[7..]),
"the statement names the image digest: {st}"
);
}
// The `.att` tag, for older cosign: the upstream's two envelopes as DSSE layers.
let att = get(port, &format!("{base}/manifests/sha256-{}.att", &top[7..]));
assert_eq!(att.status, 200, "{}", att.text());
let av: serde_json::Value = serde_json::from_slice(&att.body).unwrap();
let layers = av["layers"].as_array().unwrap();
assert_eq!(layers.len(), 2, "{av}");
for l in layers {
assert_eq!(l["mediaType"], "application/vnd.dsse.envelope.v1+json");
let b = get(
port,
&format!("{base}/blobs/{}", l["digest"].as_str().unwrap()),
);
assert_eq!(
(b.status, digest(&b.body)),
(200, l["digest"].as_str().unwrap().to_string())
);
}
}
/// ★ Every kind's own index and files, as the build systems read them.
fn assert_serves_every_kind(port: u16, log: &dyn Fn() -> String) {
assert_serves_python(port, log);
assert_serves_cargo(port, log);
assert_serves_maven(port);
assert_serves_the_image(port, log);
assert_serves_go(port);
assert_never_fetches_a_miss(port);
}
/// Python: PEP 503 with PEP 658/714, PEP 691 JSON, the metadata file.
fn assert_serves_python(port: u16, log: &dyn Fn() -> String) {
let page = get(port, "/production/simple/demo-pkg/");
assert_eq!(page.status, 200, "{}\n{}", page.text(), log());
let meta_sha = sha256(METADATA.as_bytes());
assert!(
page.text()
.contains(&format!("#sha256={}", sha256(&wheel()))),
"{}",
page.text()
);
assert!(
page.text()
.contains(&format!("data-core-metadata=\"sha256={meta_sha}\"")),
"PEP 714: {}",
page.text()
);
assert!(
page.text()
.contains(&format!("data-dist-info-metadata=\"sha256={meta_sha}\"")),
"PEP 658: {}",
page.text()
);
let json = request(
port,
"GET",
"/production/simple/demo-pkg/",
&[("Accept", "application/vnd.pypi.simple.v1+json")],
&[],
);
let j: serde_json::Value = serde_json::from_slice(&json.body)
.unwrap_or_else(|e| panic!("PEP 691: {e}: {}", json.text()));
assert_eq!(j["files"][0]["core-metadata"]["sha256"], meta_sha, "{j}");
let meta = get(port, &format!("/production/{WHEEL}.metadata"));
assert_eq!(
(meta.status, meta.text()),
(200, METADATA.to_string()),
"the core metadata file"
);
assert_eq!(get(port, &format!("/production/{WHEEL}")).body, wheel());
}
/// cargo: config.json, the index line from the sealed .crate, the download.
fn assert_serves_cargo(port: u16, log: &dyn Fn() -> String) {
let cfg: serde_json::Value =
serde_json::from_slice(&get(port, "/production/crates/config.json").body).unwrap();
assert!(
cfg["dl"].as_str().unwrap().contains("/production/crates/"),
"{cfg}"
);
let idx = get(port, "/production/crates/de/mo/demo-crate");
assert_eq!(idx.status, 200, "{}\n{}", idx.text(), log());
let line: serde_json::Value = serde_json::from_str(idx.text().lines().next().unwrap()).unwrap();
assert_eq!(
(line["name"].as_str(), line["vers"].as_str()),
(Some("demo-crate"), Some("0.1.0")),
"{line}"
);
assert_eq!(line["cksum"], sha256(&dot_crate()), "{line}");
assert_eq!(line["features"]["loud"], serde_json::json!([]), "{line}");
assert_eq!(
get(port, "/production/crates/demo-crate/demo-crate-0.1.0.crate").body,
dot_crate()
);
}
/// Maven: maven-metadata.xml over the AVAILABLE versions, the files and their checksums.
fn assert_serves_maven(port: u16) {
let md = get(
port,
"/production/maven/org/demo/demo-lib/maven-metadata.xml",
);
assert_eq!(md.status, 200, "{}", md.text());
assert!(
md.text().contains("<version>1.0.0</version>"),
"{}",
md.text()
);
assert_eq!(
get(
port,
"/production/maven/org/demo/demo-lib/1.0.0/demo-lib-1.0.0.jar"
)
.body,
jar()
);
let sha1 = get(
port,
"/production/maven/org/demo/demo-lib/1.0.0/demo-lib-1.0.0.pom.sha1",
);
assert_eq!(sha1.status, 200, "{}", sha1.text());
assert_eq!(sha1.text().trim().len(), 40, "{}", sha1.text());
}
/// OCI: the tag from the sealed layout, the manifest by tag and by digest, the blobs.
fn assert_serves_the_image(port: u16, log: &dyn Fn() -> String) {
let (_, _, image_digest) = image();
let v2 = get(port, "/v2/");
assert_eq!(v2.status, 200, "{}", v2.text());
let tags: serde_json::Value =
serde_json::from_slice(&get(port, "/v2/production/cgr/demo/hello/tags/list").body).unwrap();
assert_eq!(tags["tags"], serde_json::json!([TAG]), "{tags}");
let m = get(port, "/v2/production/cgr/demo/hello/manifests/1.0");
assert_eq!(m.status, 200, "{}\n{}", m.text(), log());
assert_eq!(
m.headers.get("docker-content-digest"),
Some(&image_digest),
"{:?}",
m.headers
);
let manifest: serde_json::Value = serde_json::from_slice(&m.body).unwrap();
let layer = manifest["layers"][0]["digest"].as_str().unwrap();
let blob = get(
port,
&format!("/v2/production/cgr/demo/hello/blobs/{layer}"),
);
assert_eq!(
(blob.status, format!("sha256:{}", sha256(&blob.body))),
(200, layer.to_string())
);
assert_eq!(
get(
port,
&format!("/v2/production/cgr/demo/hello/manifests/{image_digest}")
)
.status,
200
);
}
/// Go: the GOPROXY protocol.
fn assert_serves_go(port: u16) {
assert_eq!(
get(port, "/production/go/example.com/hello/@v/list").text(),
"v1.0.0\n"
);
let info: serde_json::Value =
serde_json::from_slice(&get(port, "/production/go/example.com/hello/@v/v1.0.0.info").body)
.unwrap();
assert_eq!(info["Version"], "v1.0.0", "{info}");
assert!(info["Time"].as_str().unwrap().ends_with('Z'), "{info}");
assert!(get(port, "/production/go/example.com/hello/@v/v1.0.0.mod")
.text()
.starts_with("module example.com/hello"));
assert_eq!(
get(port, "/production/go/example.com/hello/@v/v1.0.0.zip").body,
go_zip()
);
let latest: serde_json::Value =
serde_json::from_slice(&get(port, "/production/go/example.com/hello/@latest").body)
.unwrap();
assert_eq!(latest["Version"], "v1.0.0");
}
/// What the upstream never sent is not here, and is never fetched.
fn assert_never_fetches_a_miss(port: u16) {
let miss = get(port, "/production/crates/se/rd/serde");
assert_eq!(miss.status, 404, "{}", miss.text());
assert!(
miss.text().contains("[holger-downstream-4]"),
"{}",
miss.text()
);
}
/// ★ **USB**: verify, land, serve every kind; no upload.
#[test]
fn a_downstream_imports_the_signed_bundle_and_serves_every_kind_to_its_build_systems() {
let up = upstream();
let (bundle, policy) = signed_bundle(&up, "onprem-1");
let mut ds = downstream(&up, "downstream", None, None);
let out = ds.import(&bundle, &policy);
assert!(
out.status.success(),
"the downstream's import: {}",
text(&out)
);
let t = text(&out);
assert_eq!(
t.matches("LANDED ").count(),
packages().len(),
"every package landed AVAILABLE: {t}"
);
ds.start();
let log = ds.log();
assert!(
log.contains("downstream: zero egress"),
"the downstream says what it may dial:\n{log}"
);
let dsl = || ds.log();
assert_serves_every_kind(ds.http, &dsl);
assert_serves_the_referrers(ds.http, &std::fs::read(&up.pki.cert_der).unwrap(), &dsl);
// An upload is refused: a downstream runs no scanner, so it takes nothing but its upstream's.
let put = request(
ds.http,
"PUT",
"/incoming/smuggled-0.1-py3-none-any.whl",
&[],
&wheel(),
);
assert!(
put.status >= 400,
"an upload must be refused: {} {}",
put.status,
put.text()
);
assert!(
put.text().contains("[holger-downstream-3]"),
"{}",
put.text()
);
// The trail says the scan results are the upstream's, verified, not re-run here.
let records: String = std::fs::read_dir(ds.dir.join("holger-prod-store/lineage/intake"))
.unwrap()
.flatten()
.map(|e| std::fs::read_to_string(e.path()).unwrap_or_default())
.collect();
assert!(
records.contains("upstream-signed-evidence"),
"the landing names whose scan it trusts"
);
assert!(
!records.contains("secure-side-rescan"),
"nothing was scanned here"
);
// ★ Zero egress at run time: the import and the server dialled nothing at all.
assert_no_route_out_when_asked();
ds.stop();
let dials = dialled(&ds.connects());
assert!(
dials.is_empty(),
"a USB downstream with no upstream dials nothing, yet it did:\n{dials:#?}"
);
}
/// The downstream of the F9 tests: the guide's spelling, the bundle imported off the stick.
fn sources_downstream(up: &Upstream, name: &str) -> Downstream {
let (bundle, policy) = signed_bundle(up, &format!("{name}-1"));
let ds = downstream_spelled(up, name, None, None, Spelling::Sources);
let out = ds.import(&bundle, &policy);
assert!(
out.status.success(),
"the downstream's import: {}",
text(&out)
);
let t = text(&out);
assert!(
t.contains("LANDED cgr/demo/hello/sha256-") && t.contains("AVAILABLE"),
"the image landed AVAILABLE: {t}"
);
ds
}
/// ★ F9 (the field test of 2026-11-01): **a downstream written as the guide writes it
/// serves the image its bundle carried.** The guide's RON says `steps:` and `sources:`, not
/// the older `promotion:` ladder and `pull_through:` block, and a downstream's source names
/// no `url`: it serves what its upstream sent and fetches nothing, in the one-way build (no
/// `online`, no fetch code) as in the fetching one. Until then `sources:` refused every
/// `oci`, `maven` and `crates` source in a build without `online,jobs`, so no
/// `/v2/production/<source>/…` view existed and the image that landed AVAILABLE answered
/// `podman pull` with 404.
#[test]
fn a_downstream_written_with_sources_serves_the_image_its_bundle_carried() {
let up = upstream();
let mut ds = sources_downstream(&up, "sources");
ds.start();
let dsl = || ds.log();
assert!(
ds.log().contains("downstream: zero egress"),
"the downstream says what it may dial:\n{}",
ds.log()
);
assert_serves_the_image(ds.http, &dsl);
assert_serves_the_referrers(ds.http, &std::fs::read(&up.pki.cert_der).unwrap(), &dsl);
assert_serves_python(ds.http, &dsl);
assert_serves_maven(ds.http);
assert_serves_cargo(ds.http, &dsl);
assert_never_fetches_a_miss(ds.http);
let miss = get(ds.http, "/v2/production/cgr/demo/never-sent/manifests/1.0");
assert_eq!(miss.status, 404, "{}", miss.text());
assert!(
miss.text().contains("MANIFEST_UNKNOWN") && miss.text().contains("[holger-downstream-4]"),
"an image the upstream never sent is not sent, never fetched: {}",
miss.text()
);
// The same source WITH a `url` fetches, which a downstream never does: its start is refused.
let fetching = ds.dir.join("fetching.ron");
let ron = std::fs::read_to_string(&ds.config).unwrap();
let with_url = ron.replacen(
r#"(name: "cgr", kind: oci)"#,
r#"(name: "cgr", kind: oci, url: "https://cgr.dev")"#,
1,
);
assert_ne!(with_url, ron);
std::fs::write(&fetching, with_url).unwrap();
let out = ds.cli(&[
std::ffi::OsStr::new("start"),
std::ffi::OsStr::new("--config"),
fetching.as_os_str(),
]);
assert!(
!out.status.success() && text(&out).contains("source 'cgr'"),
"a source with a `url` must refuse the downstream's start: {}",
text(&out)
);
// ★ Zero egress at run time: a source with no `url` dials nothing.
assert_no_route_out_when_asked();
ds.stop();
let dials = dialled(&ds.connects());
assert!(
dials.is_empty(),
"a downstream whose sources name no `url` dials nothing, yet it did:\n{dials:#?}"
);
}
/// ★ F9's real client: **`podman pull` of the image from a downstream written with
/// `sources:`**, by tag, into a store of its own; podman holds the digest the upstream sealed.
#[ignore = "needs podman (the full CI set runs it)"]
#[test]
fn real_podman_pulls_the_image_from_the_downstream_written_with_sources() {
let up = upstream();
let mut ds = sources_downstream(&up, "sources-podman");
ds.start();
let (_, _, top) = image();
let store = tempfile::tempdir().unwrap();
let podman = |args: &[&str]| -> (bool, String) {
let o = Command::new("podman")
.arg("--root")
.arg(store.path().join("root"))
.arg("--runroot")
.arg(store.path().join("run"))
.args(["--storage-driver", "vfs"])
.args(args)
.output()
.expect("podman");
(o.status.success(), text(&o))
};
let image_ref = format!("127.0.0.1:{}/production/cgr/{IMAGE}:{TAG}", ds.http);
let (ok, t) = podman(&["pull", "--tls-verify=false", &image_ref]);
assert!(ok, "podman pull from the downstream:\n{t}\n{}", ds.log());
let (ok, digest) = podman(&["image", "inspect", "--format", "{{.Digest}}", &image_ref]);
assert!(ok, "{digest}");
assert_eq!(
digest.trim(),
top,
"the image podman holds is the one sealed upstream"
);
let _ = podman(&["rmi", "--all", "--force"]);
ds.stop();
}
/// ★ **Pull**: the same signed bundle, published by the upstream, comes over the
/// downstream's one line out and is verified and served the same way.
#[ignore = "needs the FETCHING holger-downstream (`boundary: Fetch`; the one-way build refuses it, holger-boundary-2): the suite step downstream:serve runs it after downstream:wg-build"]
#[test]
fn a_downstream_pulls_the_signed_bundle_from_its_one_upstream_and_serves_it() {
let mut up = upstream();
let (bundle, policy) = signed_bundle(&up, "onprem-2");
std::fs::copy(
&bundle,
up.dir.path().join("up/bundles/onprem-2.holger.znippy"),
)
.unwrap();
let up_http = {
let ron = std::fs::read_to_string(&up.config).unwrap();
ron.split("ron_http_url: Some(\"127.0.0.1:")
.nth(1)
.unwrap()
.split('"')
.next()
.unwrap()
.parse::<u16>()
.unwrap()
};
let mut upstream_server = start(
Path::new(BIN),
&up.config,
up_http,
&up.dir.path().join("upstream.log"),
None,
);
let mut ds = downstream(&up, "pulled", Some(&policy), Some(up.grpc));
std::fs::create_dir_all(ds.dir.join("bundle.inbox")).unwrap();
ds.allow_first_import();
ds.start();
let t = Instant::now();
// The bundle has landed when the inbox ARCHIVED it: the sweep moves it there only after
// the whole import returned. Python's page alone answered while the later kinds were still
// landing, and the Maven read raced them (a 404 holger-downstream-4, 1 run in 3 on dwarf).
let archived = || {
std::fs::read_dir(ds.dir.join("bundle.inbox/archive"))
.map(|d| {
d.flatten().any(|e| {
e.file_name()
.to_string_lossy()
.ends_with("onprem-2.holger.znippy")
})
})
.unwrap_or(false)
};
loop {
if archived() {
break;
}
assert!(
t.elapsed() < Duration::from_secs(90),
"the pulled bundle never landed:\n{}",
ds.log()
);
std::thread::sleep(Duration::from_millis(250));
}
assert!(
ds.log()
.contains("downstream: zero egress except its upstream Holger 'upstream'"),
"{}",
ds.log()
);
let dsl = || ds.log();
assert_serves_every_kind(ds.http, &dsl);
assert!(
std::fs::read_dir(ds.dir.join("bundle.inbox/archive"))
.unwrap()
.flatten()
.any(|e| e
.file_name()
.to_string_lossy()
.ends_with("onprem-2.holger.znippy")),
"the pulled bundle was verified and archived by the inbox import"
);
let _ = upstream_server.kill();
let _ = upstream_server.wait();
// ★ Its one line out, and nothing else: every dial went to the configured upstream.
assert_no_route_out_when_asked();
ds.stop();
let dials = dialled(&ds.connects());
let line = format!("inet 127.0.0.1 {}", up.grpc);
assert!(
dials.contains(&line),
"the trap saw the pull itself ({line}): {dials:#?}"
);
let other: Vec<&String> = dials
.iter()
.filter(|d| **d != line && *d != "dns 127.0.0.1")
.collect();
assert!(
other.is_empty(),
"the downstream dialled beyond its one upstream (127.0.0.1:{}):\n{other:#?}",
up.grpc
);
up.grpc = 0;
}
// ── H4 on the downstream: an invalidated package is blocked ─────────────────
/// Put a `sync:` block into a written config.
fn add_sync(config: &Path, block: &str) {
let s = std::fs::read_to_string(config).unwrap();
let i = s.rfind(')').unwrap();
std::fs::write(
config,
format!("{} sync: Some(({block})),\n{}", &s[..i], &s[i..]),
)
.unwrap();
}
/// ★ **Invalidation**: the bundle lands and is served; the upstream then invalidates the
/// wheel; its signed sync manifest goes over on the stick; the downstream's `sync apply`
/// holds it; the started downstream serves nothing of it (every other kind stays), and
/// the ack comes back. No dial on the downstream side at any step.
#[ignore = "needs the FETCHING holger-downstream (`boundary: Fetch`; the one-way build refuses it, holger-boundary-2): the suite step downstream:serve runs it after downstream:wg-build"]
#[test]
fn an_invalidated_package_is_blocked_on_the_downstream() {
use server_lib::sync::protocol::SyncKey;
use std::ffi::OsStr;
let os = |s: &str| OsStr::new(s).to_os_string();
let up = upstream();
let mut ds = downstream(&up, "invalidated", None, None);
let keys = up.dir.path().join("sync-keys");
std::fs::create_dir_all(&keys).unwrap();
let (up_key, down_key) = (keys.join("cloud.key"), keys.join("onprem.key"));
let up_hex = SyncKey::open_or_mint(&up_key).unwrap().public_hex();
let down_hex = SyncKey::open_or_mint(&down_key).unwrap().public_hex();
add_sync(
&up.config,
&format!(
r#"name: "cloud", key: Some("{}"), downstreams: [(name: "onprem", key: "{down_hex}")], ack_deadline_secs: 3600"#,
up_key.display()
),
);
add_sync(
&ds.config,
&format!(
r#"name: "onprem", key: Some("{}"), upstreams: [(name: "cloud", key: "{up_hex}")]"#,
down_key.display()
),
);
// The ack goes back on the stick: a receipt, so this downstream is `boundary: Fetch`.
let s = std::fs::read_to_string(&ds.config).unwrap();
std::fs::write(
&ds.config,
s.replacen("(\n", "(\n boundary: Fetch,\n", 1),
)
.unwrap();
let stick = up.dir.path().join("stick");
std::fs::create_dir_all(&stick).unwrap();
let up_cli = |args: Vec<std::ffi::OsString>| {
let out = Command::new(BIN).args(&args).output().unwrap();
assert!(out.status.success(), "upstream {args:?}: {}", text(&out));
text(&out)
};
// 1. The bundle carries the manifest; the downstream lands it and writes its ack.
let (bundle, policy) = signed_bundle(&up, "onprem-sync");
ds.allow_first_import();
let out = ds.cli(&[
OsStr::new("bundle"),
OsStr::new("import"),
OsStr::new("--config"),
ds.config.as_os_str(),
OsStr::new("--bundle"),
bundle.as_os_str(),
OsStr::new("--policy"),
policy.as_os_str(),
OsStr::new("--ack-out"),
stick.as_os_str(),
]);
let t = text(&out);
assert!(
out.status.success() && t.contains("LANDED") && t.contains("ACK "),
"{t}"
);
ds.start();
assert_eq!(
get(ds.http, &format!("/production/{WHEEL}")).body,
wheel(),
"served before the invalidation"
);
ds.stop();
// 2. The upstream invalidates the wheel; the manifest goes on the stick.
let t = up_cli(vec![
os("sync"),
os("invalidate"),
os("--config"),
up.config.clone().into(),
os("--coordinate"),
os(WHEEL),
os("--reason"),
os("malware"),
os("--detail"),
os("found after its approval"),
]);
assert!(t.contains("INVALIDATED "), "{t}");
let t = up_cli(vec![
os("sync"),
os("manifest"),
os("--config"),
up.config.clone().into(),
os("--out"),
stick.clone().into(),
]);
let manifest = PathBuf::from(
t.lines()
.find_map(|l| l.strip_prefix("MANIFEST "))
.unwrap_or_else(|| panic!("{t}"))
.trim(),
);
// 3. The downstream applies it: the wheel is held, and its ack goes back.
let out = ds.cli(&[
OsStr::new("sync"),
OsStr::new("apply"),
OsStr::new("--config"),
ds.config.as_os_str(),
OsStr::new("--manifest"),
manifest.as_os_str(),
OsStr::new("--ack-out"),
stick.as_os_str(),
]);
let t = text(&out);
assert!(
out.status.success() && t.contains(&format!("HELD {WHEEL}")),
"{t}"
);
// 4. The started downstream serves nothing of it; the other kinds stay.
ds.start();
let file = get(ds.http, &format!("/production/{WHEEL}"));
assert!(
file.status >= 400 && file.body != wheel(),
"an invalidated wheel must not be served: {} {}",
file.status,
file.text()
);
let page = get(ds.http, "/production/simple/demo-pkg/");
assert!(
!page.text().contains(WHEEL),
"the index must not list an invalidated file: {} {}",
page.status,
page.text()
);
assert_eq!(
get(
ds.http,
"/production/crates/demo-crate/demo-crate-0.1.0.crate"
)
.body,
dot_crate(),
"the rest is still served"
);
ds.stop();
// 5. The upstream takes the ack: the invalidation reached this downstream.
let ack = std::fs::read_dir(&stick)
.unwrap()
.flatten()
.map(|e| e.path())
.filter(|p| {
p.file_name()
.unwrap()
.to_string_lossy()
.starts_with("holger-sync-ack-onprem-")
})
.max()
.expect("an ack on the stick");
let t = up_cli(vec![
os("sync"),
os("ack"),
os("--config"),
up.config.clone().into(),
os("--ack"),
ack.into(),
]);
assert!(t.contains("1 invalidation(s) reached"), "{t}");
assert_no_route_out_when_asked();
let dials = dialled(&ds.connects());
assert!(dials.is_empty(), "the USB downstream dialled: {dials:#?}");
}
/// ★ HOL-REFERRERS, **real clients on the downstream**: after the bundle import an
/// on-prem cluster verifies Holger's verdict with no line out. cosign 2 (the `.att`
/// tag) and cosign 3 (the referrer bundle) `verify-attestation --key <the upstream's
/// evidence key>` pass against the downstream, a key that is not the upstream's fails,
/// and `oras discover` lists the image's SBOM referrer and the evidence bundles.
#[ignore = "needs podman and the cosign 2/3 and oras images (the full CI set runs it)"]
#[test]
fn real_cosign_verifies_the_upstreams_evidence_on_the_downstream() {
const COSIGN2: &str = "gcr.io/projectsigstore/cosign:v2.4.3";
const COSIGN3: &str = "ghcr.io/sigstore/cosign/cosign:v3.0.2";
const ORAS: &str = "ghcr.io/oras-project/oras:v1.2.3";
let up = upstream();
let (bundle, policy) = signed_bundle(&up, "onprem-real");
let mut ds = downstream(&up, "downstream-real", None, None);
let out = ds.import(&bundle, &policy);
assert!(
out.status.success(),
"the downstream's import: {}",
text(&out)
);
ds.start();
let (_, _, top) = image();
let o = tempfile::tempdir().unwrap();
std::fs::set_permissions(
o.path(),
std::os::unix::fs::PermissionsExt::from_mode(0o777),
)
.unwrap();
let pk8 = std::fs::read(&up.pki.key_pkcs8).unwrap();
let key = rcgen::KeyPair::try_from(pk8.as_slice()).unwrap();
std::fs::write(o.path().join("evidence.pub"), key.public_key_pem()).unwrap();
let other = rcgen::KeyPair::generate_for(&rcgen::PKCS_ECDSA_P256_SHA256).unwrap();
std::fs::write(o.path().join("other.pub"), other.public_key_pem()).unwrap();
let image_ref = format!("127.0.0.1:{}/production/cgr/demo/hello@{top}", ds.http);
let run = |image: &str, args: &[&str]| -> (bool, String) {
let o2 = Command::new("podman")
.args(["run", "--rm", "--network", "host", "-v"])
.arg(format!("{}:/out:Z", o.path().display()))
.args(["-e", "HOME=/tmp", image])
.args(args)
.output()
.expect("podman");
(
o2.status.success(),
format!(
"{}{}",
String::from_utf8_lossy(&o2.stdout),
String::from_utf8_lossy(&o2.stderr)
),
)
};
for c in [COSIGN2, COSIGN3] {
let (ok, t) = run(
c,
&[
"verify-attestation",
"--key",
"/out/evidence.pub",
"--type",
"https://holger.rs/attestation/scan-evidence/v1",
"--insecure-ignore-tlog=true",
"--allow-http-registry",
&image_ref,
],
);
assert!(
ok,
"{c} verify-attestation of the upstream's scan evidence on the downstream:\n{t}\n{}",
ds.log()
);
let (ok, t) = run(
c,
&[
"verify-attestation",
"--key",
"/out/evidence.pub",
"--type",
"cyclonedx",
"--insecure-ignore-tlog=true",
"--allow-http-registry",
&image_ref,
],
);
assert!(
ok,
"{c} verify-attestation of the upstream's CycloneDX SBOM on the downstream:\n{t}"
);
let (ok, t) = run(
c,
&[
"verify-attestation",
"--key",
"/out/other.pub",
"--type",
"https://holger.rs/attestation/scan-evidence/v1",
"--insecure-ignore-tlog=true",
"--allow-http-registry",
&image_ref,
],
);
assert!(
!ok,
"{c}: a key that is not the upstream's must not verify:\n{t}"
);
}
let (ok, t) = run(
ORAS,
&["discover", "--plain-http", "--format", "json", &image_ref],
);
assert!(
ok && t.contains("application/spdx+json")
&& t.contains("application/vnd.dev.sigstore.bundle.v0.3+json"),
"oras discover on the downstream:\n{t}"
);
ds.stop();
}