server/lib/tests/intake_chain_holds_a_wheel_until_available.rs
//! **A Python wheel uploaded over HTTP PUT sits in INGESTED, is not downloadable,
//! and reaches AVAILABLE only by passing every intake state.**
//!
//! This is the intake chain driven through the real doors of a real server: four
//! `raw` repositories (incoming, quarantine, approved, production) on one HTTP
//! and one gRPC listener, the promotion ladder over exactly those four, and the
//! `intake:` block naming them.
//!
//! * The PUT lands the wheel in INGESTED and nothing serves it: not incoming,
//! not production.
//! * A write into quarantine, approved or production is refused by name, on the
//! HTTP door AND on the gRPC door.
//! * A skip (INGESTED → AVAILABLE, INGESTED → SEALED) is refused by name and
//! moves nothing.
//! * Each transition is walked in turn; production serves 404 until AVAILABLE,
//! and then serves the wheel's exact bytes from the sealed store.
//! * Every transition left an audit record in the server's audit log.
//!
//! The intake chain is the `supply-chain` plane, so this file is gated on it and
//! its arm is in `.nornir/testmatrix-arms.json`.
use std::io::Write;
use traits::test_support::hold_port;
use holger_server_lib::grpc::holger_proto::promotion_service_client::PromotionServiceClient;
use holger_server_lib::grpc::holger_proto::repository_service_client::RepositoryServiceClient;
use holger_server_lib::grpc::holger_proto::{
ArtifactId as ProtoId, PromoteArtifactRequest, PutArtifactRequest,
};
use holger_server_lib::promote::intake::{ArtifactState, IntakeError, TransitionRequest};
use holger_server_lib::{read_ron_config, wire_holger, ArtifactId, Holger};
struct Running {
data: tempfile::TempDir,
_cfg: tempfile::NamedTempFile,
holger: Holger,
base: String,
grpc: String,
}
fn start() -> Running {
let mut last = String::new();
for _ in 0..8 {
let (grpc, http) = (hold_port(), hold_port());
match try_start(grpc, http) {
Ok(r) => return r,
Err(e) => last = e,
}
}
panic!("server never started; last error: {last}");
}
fn try_start(grpc: u16, http: u16) -> Result<Running, String> {
let data = tempfile::tempdir().unwrap();
let d = data.path().display().to_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("{d}/{name}")),"#
)
};
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),
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: (sandbox: off),
incoming: "incoming",
quarantine: "quarantine",
approved: "approved",
production: "production",
)),
)),
)"#,
incoming = repo("incoming"),
quarantine = repo("quarantine"),
approved = repo("approved"),
production = repo("production"),
);
let mut cfg = tempfile::NamedTempFile::new().unwrap();
cfg.write_all(ron.as_bytes()).unwrap();
// The data root is the audit directory's parent: the promotion store and
// the intake record live under `{d}`.
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}"))?;
holger.start().map_err(|e| format!("start: {e}"))?;
Ok(Running {
data,
_cfg: cfg,
holger,
base: format!("http://127.0.0.1:{http}"),
grpc: format!("http://127.0.0.1:{grpc}"),
})
}
async fn wait_healthy(c: &reqwest::Client, base: &str) {
for _ in 0..100 {
if let Ok(r) = c.get(format!("{base}/healthz")).send().await {
if r.status() == 200 {
return;
}
}
tokio::time::sleep(std::time::Duration::from_millis(20)).await;
}
panic!("gateway never became healthy at {base}");
}
const WHEEL: &str = "demo_pkg-1.0.0-py3-none-any.whl";
/// A small but well-formed wheel: a stored-only zip with the dist-info METADATA
/// and WHEEL members a wheel carries.
fn wheel_bytes() -> Vec<u8> {
fn member(out: &mut Vec<u8>, central: &mut Vec<u8>, name: &str, body: &[u8]) {
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(&(name.len() as u16).to_le_bytes());
out.extend_from_slice(&0u16.to_le_bytes());
out.extend_from_slice(name.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(&(name.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(name.as_bytes());
}
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 mut out = Vec::new();
let mut central = Vec::new();
member(
&mut out,
&mut central,
"demo_pkg/__init__.py",
b"__version__ = \"1.0.0\"\n",
);
member(
&mut out,
&mut central,
"demo_pkg-1.0.0.dist-info/METADATA",
b"Metadata-Version: 2.1\nName: demo-pkg\nVersion: 1.0.0\nLicense-Expression: MIT\n",
);
member(
&mut out,
&mut central,
"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",
);
let cd_offset = out.len() as u32;
let cd_len = central.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, 3, 0, 3, 0]);
out.extend_from_slice(&cd_len.to_le_bytes());
out.extend_from_slice(&cd_offset.to_le_bytes());
out.extend_from_slice(&0u16.to_le_bytes());
out
}
fn wheel_id() -> ArtifactId {
ArtifactId {
namespace: None,
name: WHEEL.into(),
version: String::new(),
}
}
#[tokio::test(flavor = "multi_thread")]
async fn a_wheel_put_over_http_reaches_available_only_through_every_state() {
let r = start();
let cl = tls_roots(reqwest::Client::builder())
.build()
.expect("reqwest client");
wait_healthy(&cl, &r.base).await;
let chain = r
.holger
.intake()
.expect("the config declares an intake chain");
let wheel = wheel_bytes();
let url = |repo: &str| format!("{}/{repo}/{WHEEL}", r.base);
// ── UNTRUSTED → INGESTED over HTTP PUT ──
let put = cl
.put(url("incoming"))
.body(wheel.clone())
.send()
.await
.unwrap();
assert_eq!(put.status(), 201, "the PUT is ingested");
assert_eq!(
put.headers()
.get("x-holger-intake-state")
.and_then(|v| v.to_str().ok()),
Some("INGESTED")
);
assert_eq!(chain.state(&wheel_id()), ArtifactState::Ingested);
// ── INGESTED is not downloadable, from any intake repository ──
for repo in ["incoming", "quarantine", "approved", "production"] {
let get = cl.get(url(repo)).send().await.unwrap();
assert_eq!(
get.status(),
404,
"GET /{repo} must not serve an INGESTED wheel"
);
}
// ── nothing is written past the chain: HTTP and gRPC ──
for repo in ["quarantine", "approved", "production"] {
let put = cl.put(url(repo)).body(wheel.clone()).send().await.unwrap();
assert_eq!(put.status(), 403, "PUT /{repo} is refused");
let body = put.text().await.unwrap();
assert!(
body.contains("[holger-intake-11]"),
"refused by name: {body}"
);
}
let mut grpc = None;
for _ in 0..100 {
if let Ok(c) = RepositoryServiceClient::connect(r.grpc.clone()).await {
grpc = Some(c);
break;
}
tokio::time::sleep(std::time::Duration::from_millis(20)).await;
}
let mut grpc = grpc.expect("gRPC came up");
let st = grpc
.put_artifact(PutArtifactRequest {
repository: "production".into(),
id: Some(ProtoId {
namespace: String::new(),
name: WHEEL.into(),
version: String::new(),
}),
data: wheel.clone().into(),
async_job: false,
})
.await
.expect_err("gRPC PutArtifact into production is refused");
assert!(
st.message().contains("[holger-intake-11]"),
"{}",
st.message()
);
let mut promote = PromotionServiceClient::connect(r.grpc.clone())
.await
.unwrap();
let st = promote
.promote_artifact(PromoteArtifactRequest {
from_repo: "incoming".into(),
to_repo: "production".into(),
artifact: Some(ProtoId {
namespace: String::new(),
name: WHEEL.into(),
version: String::new(),
}),
..Default::default()
})
.await
.expect_err("a promote past the chain is refused");
assert!(
st.message().contains("[holger-intake-11]"),
"{}",
st.message()
);
assert_eq!(
chain.state(&wheel_id()),
ArtifactState::Ingested,
"nothing moved"
);
// ── a skip is refused by name and moves nothing ──
for to in [
ArtifactState::Available,
ArtifactState::Sealed,
ArtifactState::Approved,
] {
let err = chain
.advance(TransitionRequest::new(wheel_id(), to, "sec-officer"))
.expect_err("a skip is refused");
assert!(matches!(err, IntakeError::SkipsState { .. }), "{err}");
let line = err.tagged();
assert!(line.starts_with("[holger-intake-1]"), "{line}");
assert!(
line.contains("SCANNED"),
"the refusal names the skipped state: {line}"
);
assert_eq!(chain.state(&wheel_id()), ArtifactState::Ingested);
}
// ── each transition in turn; production serves nothing until AVAILABLE ──
for to in [
ArtifactState::Scanned,
ArtifactState::PolicyChecked,
ArtifactState::Approved,
ArtifactState::Sealed,
ArtifactState::Promoted,
] {
chain
.advance(TransitionRequest::new(wheel_id(), to, "sec-officer"))
.unwrap_or_else(|e| panic!("→ {to}: {e}"));
assert_eq!(chain.state(&wheel_id()), to);
let get = cl.get(url("production")).send().await.unwrap();
assert_eq!(get.status(), 404, "production does not serve a {to} wheel");
}
chain
.advance(TransitionRequest::new(
wheel_id(),
ArtifactState::Available,
"sec-officer",
))
.unwrap();
// ── AVAILABLE: production serves the exact bytes, from the sealed store ──
let get = cl.get(url("production")).send().await.unwrap();
assert_eq!(get.status(), 200);
assert_eq!(
get.headers()
.get("x-holger-intake-state")
.and_then(|v| v.to_str().ok()),
Some("AVAILABLE")
);
assert_eq!(
get.bytes().await.unwrap().as_ref(),
&wheel[..],
"the wheel, byte for byte"
);
for repo in ["incoming", "quarantine", "approved"] {
let get = cl.get(url(repo)).send().await.unwrap();
assert_eq!(
get.status(),
404,
"only production serves, even when AVAILABLE ({repo})"
);
}
// ── every transition is in the server's audit log ──
let events = holger_server_lib::audit::FileAuditLog::read_dir(r.data.path().join("audit"))
.expect("the audit log reads back");
let trail: Vec<String> = events
.iter()
.map(|e| e.detail.clone())
.filter(|d| d.starts_with("intake ") && !d.starts_with("intake refused"))
.collect();
let want = [
"intake UNTRUSTED → INGESTED",
"intake INGESTED → SCANNED",
"intake SCANNED → POLICY_CHECKED",
"intake POLICY_CHECKED → APPROVED",
"intake APPROVED → SEALED",
"intake SEALED → PROMOTED",
"intake PROMOTED → AVAILABLE",
];
assert_eq!(trail.len(), want.len(), "{trail:#?}");
for (got, want) in trail.iter().zip(want) {
assert!(got.starts_with(want), "{got} should start with {want}");
}
assert!(
events
.iter()
.any(|e| e.detail.contains("intake refused: [holger-intake-1]")),
"the refused skips are audited too"
);
}
/// reqwest 0.13 with the ring provider and the bundled roots, as the other door
/// tests build it (the crate links `rustls-no-provider`).
fn tls_roots(b: reqwest::ClientBuilder) -> reqwest::ClientBuilder {
let _ = rustls::crypto::ring::default_provider().install_default();
b.tls_certs_only(
webpki_root_certs::TLS_SERVER_ROOT_CERTS
.iter()
.filter_map(|c| reqwest::Certificate::from_der(c).ok()),
)
}