server/lib/tests/intake_serves_pip.rs
//! **pip through the intake chain** (lane W1): a wheel uploaded the way twine
//! uploads it lands in INGESTED, is walked to AVAILABLE with S4's
//! `AdvanceArtifact` and two approvers, and only then does the production
//! repository's PEP 503 / PEP 691 simple index list it, with the sha256 of the
//! sealed bytes.
//!
//! One real server: four `raw` intake repositories, static-token principals with
//! duties, and `approvals: 2`. The index is S5's (`IntakeChain::simple_index`);
//! W1 adds the twine upload door and the PEP 691 JSON form.
//!
//! * `a_twine_upload_is_listed_only_at_available` runs on every plain
//! `cargo test`. It drives the HTTP doors with reqwest: the twine form POST
//! (`/incoming/legacy/`) and a PUT (`/incoming/<file>`) are
//! both INGESTED; `/production/simple/` and `/production/simple/<pkg>/` answer
//! 404 for the project at every state short of AVAILABLE; at AVAILABLE the
//! HTML and the JSON page list the wheel with its sha256, the link serves the
//! exact bytes, and the second wheel, still INGESTED, is never listed.
//! * `real_pip_installs_and_imports_from_the_intake_index` is the measurement:
//! the same walk, then a REAL `pip install --index-url <holger>/production/simple/`
//! in the official python image under `podman run --network host`, and an
//! import of the installed package. It is `#[ignore]`d for a plain
//! `cargo test` (it needs podman and an image) and reached by its arm in
//! `.nornir/testmatrix-arms.json` (`ignored: true`). No podman FAILS by name;
//! it never skips.
use std::io::Write;
use std::path::{Path, PathBuf};
use std::process::Command;
use traits::test_support::hold_port;
use holger_server_lib::grpc::holger_proto::{
AdvanceArtifactRequest, ArtifactId as ProtoId, IntakeVerb,
};
use holger_server_lib::promote::duties::client;
use holger_server_lib::promote::intake::ArtifactState;
use holger_server_lib::{read_ron_config, wire_holger, ArtifactId, Holger};
use sha2::Digest as _;
const PRINCIPALS: [&str; 4] = ["ida", "bob", "carol", "adm"];
const PYTHON_IMAGE: &str = "docker.io/library/python:3.13-slim-trixie";
const WHEEL: &str = "demo_pkg-1.0.0-py3-none-any.whl";
const WHEEL_NEXT: &str = "demo_pkg-1.1.0-py3-none-any.whl";
fn token(who: &str) -> String {
format!("{who}-token-of-sufficient-length-for-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 methods: Vec<String> = PRINCIPALS
.iter()
.map(|p| {
format!(
r#"StaticToken(subject: "{p}", token: Some("{}"))"#,
token(p)
)
})
.collect();
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],
"adm": [RepositoryAdmin],
}},
)),
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",
approvals: 2,
)),
)),
)"#,
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(", "),
);
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}"))?;
holger.start().map_err(|e| format!("start: {e}"))?;
Ok(Running {
_data: data,
_cfg: cfg,
holger,
base: format!("http://127.0.0.1:{http}"),
grpc: format!("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}");
}
/// A small wheel pip installs: a stored-only zip with the package, its
/// METADATA, WHEEL and a RECORD that lists every member with its digest.
fn wheel_bytes(version: &str) -> 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
}
fn record_line(name: &str, body: &[u8]) -> String {
use base64::Engine as _;
let digest = sha2::Sha256::digest(body);
let b64 = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(digest);
format!("{name},sha256={b64},{}\n", body.len())
}
let info = format!("demo_pkg-{version}.dist-info");
let members: Vec<(String, Vec<u8>)> = vec![
("demo_pkg/__init__.py".into(), format!("__version__ = \"{version}\"\n").into_bytes()),
(
format!("{info}/METADATA"),
format!("Metadata-Version: 2.1\nName: demo-pkg\nVersion: {version}\nLicense-Expression: MIT\n").into_bytes(),
),
(
format!("{info}/WHEEL"),
b"Wheel-Version: 1.0\nGenerator: holger-test\nRoot-Is-Purelib: true\nTag: py3-none-any\n".to_vec(),
),
];
let mut record = String::new();
for (n, b) in &members {
record.push_str(&record_line(n, b));
}
record.push_str(&format!("{info}/RECORD,,\n"));
let mut all = members;
all.push((format!("{info}/RECORD"), record.into_bytes()));
let mut out = Vec::new();
let mut central = Vec::new();
for (n, b) in &all {
member(&mut out, &mut central, n, b);
}
let cd_offset = out.len() as u32;
let cd_len = central.len() as u32;
let count = all.len() as u16;
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(&count.to_le_bytes());
out.extend_from_slice(&count.to_le_bytes());
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 sha256_hex(b: &[u8]) -> String {
sha2::Sha256::digest(b)
.iter()
.map(|x| format!("{x:02x}"))
.collect()
}
/// A dist coordinate: the file name, no namespace, no version (the same shape a
/// raw repository gives a top-level file).
fn dist_id(file: &str) -> ArtifactId {
ArtifactId {
namespace: None,
name: file.into(),
version: String::new(),
}
}
/// The multipart body `twine upload` sends to `/legacy/`.
fn twine_form(file: &str, version: &str, content: &[u8]) -> (String, Vec<u8>) {
let boundary = "----holger-w1-twine-boundary";
let mut body = Vec::new();
let field = |body: &mut Vec<u8>, name: &str, value: &str| {
body.extend_from_slice(format!("--{boundary}\r\nContent-Disposition: form-data; name=\"{name}\"\r\n\r\n{value}\r\n").as_bytes());
};
field(&mut body, ":action", "file_upload");
field(&mut body, "protocol_version", "1");
field(&mut body, "metadata_version", "2.1");
field(&mut body, "name", "demo-pkg");
field(&mut body, "version", version);
field(&mut body, "filetype", "bdist_wheel");
field(&mut body, "pyversion", "py3");
field(&mut body, "sha256_digest", &sha256_hex(content));
body.extend_from_slice(
format!(
"--{boundary}\r\nContent-Disposition: form-data; name=\"content\"; filename=\"{file}\"\r\nContent-Type: application/octet-stream\r\n\r\n"
)
.as_bytes(),
);
body.extend_from_slice(content);
body.extend_from_slice(format!("\r\n--{boundary}--\r\n").as_bytes());
(format!("multipart/form-data; boundary={boundary}"), body)
}
async fn advance(r: &Running, who: &str, file: &str) -> Result<String, String> {
let req = AdvanceArtifactRequest {
artifact: Some(ProtoId {
namespace: String::new(),
name: file.into(),
version: String::new(),
}),
verb: IntakeVerb::Advance as i32,
..Default::default()
};
let mut last = String::new();
for _ in 0..50 {
match client::advance(&r.grpc, Some(&token(who)), req.clone()).await {
Err(e) if e.contains("was not reached") => {
last = e;
tokio::time::sleep(std::time::Duration::from_millis(20)).await;
}
other => return other.map(|a| a.state),
}
}
Err(last)
}
const JSON_ACCEPT: &str = "application/vnd.pypi.simple.v1+json";
/// The project is NOT on the production index: 404 on its page (HTML and JSON),
/// and the root does not name it.
async fn not_listed(cl: &reqwest::Client, base: &str, why: &str) {
let root = cl
.get(format!("{base}/production/simple/"))
.send()
.await
.unwrap();
assert_eq!(root.status(), 200, "the root index answers ({why})");
let html = root.text().await.unwrap();
assert!(
!html.contains("demo-pkg"),
"the root must not name demo-pkg ({why}): {html}"
);
for accept in ["text/html", JSON_ACCEPT] {
let page = cl
.get(format!("{base}/production/simple/demo-pkg/"))
.header("Accept", accept)
.send()
.await
.unwrap();
assert_eq!(
page.status(),
404,
"the project page is 404 ({why}, {accept})"
);
}
}
/// Upload both wheels (twine POST and PUT), then walk the first to AVAILABLE,
/// asserting at every state that the index does not list it yet.
async fn upload_and_walk(r: &Running, cl: &reqwest::Client) -> Vec<u8> {
let chain = r
.holger
.intake()
.expect("the config declares an intake chain");
let wheel = wheel_bytes("1.0.0");
let next = wheel_bytes("1.1.0");
// ── the twine upload: POST /incoming/legacy/ with Basic __token__ ──
let (ctype, form) = twine_form(WHEEL, "1.0.0", &wheel);
let post = cl
.post(format!("{}/incoming/legacy/", r.base))
.basic_auth("__token__", Some(token("ida")))
.header("Content-Type", ctype)
.body(form)
.send()
.await
.unwrap();
let status = post.status();
let state = post
.headers()
.get("x-holger-intake-state")
.and_then(|v| v.to_str().ok())
.map(str::to_owned);
let sum = post
.headers()
.get("x-checksum-sha256")
.and_then(|v| v.to_str().ok())
.map(str::to_owned);
assert!(
status.is_success(),
"the twine POST is ingested: {status} {}",
post.text().await.unwrap_or_default()
);
assert_eq!(state.as_deref(), Some("INGESTED"));
assert_eq!(
sum.as_deref(),
Some(sha256_hex(&wheel).as_str()),
"the digest is the wheel's, not the form's"
);
assert_eq!(chain.state(&dist_id(WHEEL)), ArtifactState::Ingested);
assert_eq!(chain.importer(&dist_id(WHEEL)).as_deref(), Some("ida"));
// ── the plain PUT of the second wheel: INGESTED, and stays there ──
let put = cl
.put(format!("{}/incoming/{WHEEL_NEXT}", r.base))
.bearer_auth(token("ida"))
.body(next.clone())
.send()
.await
.unwrap();
assert_eq!(
put.status(),
201,
"the PUT is ingested: {}",
put.text().await.unwrap_or_default()
);
assert_eq!(chain.state(&dist_id(WHEEL_NEXT)), ArtifactState::Ingested);
// ── nothing is written past the chain ──
let put = cl
.put(format!("{}/production/{WHEEL}", r.base))
.bearer_auth(token("adm"))
.body(wheel.clone())
.send()
.await
.unwrap();
assert_eq!(put.status(), 403, "a write into production is refused");
not_listed(cl, &r.base, "INGESTED").await;
// ── the walk: ida scans and policy-checks, bob and carol approve, adm seals ──
for (who, want) in [("ida", "SCANNED"), ("ida", "POLICY_CHECKED")] {
assert_eq!(
advance(r, who, WHEEL)
.await
.unwrap_or_else(|e| panic!("{who} → {want}: {e}")),
want
);
not_listed(cl, &r.base, want).await;
}
let e = advance(r, "ida", WHEEL)
.await
.expect_err("the importer never approves her own import");
assert!(e.starts_with("[holger-duty-"), "{e}");
assert_eq!(
advance(r, "bob", WHEEL).await.expect("bob approves"),
"POLICY_CHECKED",
"one of two approvals"
);
not_listed(cl, &r.base, "one approval").await;
assert_eq!(
advance(r, "carol", WHEEL).await.expect("carol approves"),
"APPROVED"
);
not_listed(cl, &r.base, "APPROVED").await;
for (who, want) in [("adm", "SEALED"), ("bob", "PROMOTED")] {
assert_eq!(
advance(r, who, WHEEL)
.await
.unwrap_or_else(|e| panic!("{who} → {want}: {e}")),
want
);
not_listed(cl, &r.base, want).await;
let get = cl
.get(format!("{}/production/{WHEEL}", r.base))
.send()
.await
.unwrap();
assert_eq!(get.status(), 404, "the file is not downloadable at {want}");
}
assert_eq!(
advance(r, "adm", WHEEL)
.await
.expect("adm makes it AVAILABLE"),
"AVAILABLE"
);
assert_eq!(chain.state(&dist_id(WHEEL)), ArtifactState::Available);
assert_eq!(chain.state(&dist_id(WHEEL_NEXT)), ArtifactState::Ingested);
wheel
}
#[tokio::test(flavor = "multi_thread")]
async fn a_twine_upload_is_listed_only_at_available() {
let r = start();
let cl = tls_roots(reqwest::Client::builder())
.build()
.expect("reqwest client");
wait_healthy(&cl, &r.base).await;
let wheel = upload_and_walk(&r, &cl).await;
let sha = sha256_hex(&wheel);
// ── PEP 503: the root names the project, the page links the wheel with its sha256 ──
let root = cl
.get(format!("{}/production/simple/", r.base))
.send()
.await
.unwrap();
assert_eq!(root.status(), 200);
assert!(root
.text()
.await
.unwrap()
.contains("<a href=\"demo-pkg/\">demo-pkg</a>"));
let explicit = cl
.get(format!("{}/production/simple/demo-pkg/json", r.base))
.send()
.await
.unwrap();
assert_eq!(
explicit
.headers()
.get("content-type")
.unwrap()
.to_str()
.unwrap(),
JSON_ACCEPT,
"simple/<project>/json is the JSON form"
);
let page = cl
.get(format!("{}/production/simple/demo-pkg/", r.base))
.header("Accept", "text/html")
.send()
.await
.unwrap();
assert_eq!(page.status(), 200);
assert!(page
.headers()
.get("content-type")
.unwrap()
.to_str()
.unwrap()
.starts_with("text/html"));
let html = page.text().await.unwrap();
let href = format!("../../{WHEEL}#sha256={sha}");
assert!(
html.contains(&href),
"the page links the wheel with the sealed sha256: {html}"
);
assert!(
!html.contains(WHEEL_NEXT),
"an INGESTED wheel is never listed: {html}"
);
// A non-normalized project name reaches the same page.
let odd = cl
.get(format!("{}/production/simple/Demo_Pkg/", r.base))
.send()
.await
.unwrap();
assert_eq!(odd.status(), 200);
assert!(odd.text().await.unwrap().contains(&href));
// ── PEP 691: the JSON page, negotiated by Accept ──
let page = cl
.get(format!("{}/production/simple/demo-pkg/", r.base))
.header("Accept", JSON_ACCEPT)
.send()
.await
.unwrap();
assert_eq!(page.status(), 200);
assert_eq!(
page.headers()
.get("content-type")
.unwrap()
.to_str()
.unwrap(),
JSON_ACCEPT
);
let json: serde_json::Value = page.json().await.unwrap();
assert_eq!(json["meta"]["api-version"], "1.0");
assert_eq!(json["name"], "demo-pkg");
let files = json["files"].as_array().expect("files");
assert_eq!(files.len(), 1, "only the AVAILABLE wheel: {json}");
assert_eq!(files[0]["filename"], WHEEL);
assert_eq!(files[0]["hashes"]["sha256"], sha.as_str());
let root = cl
.get(format!("{}/production/simple/", r.base))
.header("Accept", JSON_ACCEPT)
.send()
.await
.unwrap();
let root: serde_json::Value = root.json().await.unwrap();
assert_eq!(root["projects"][0]["name"], "demo-pkg");
// ── the link serves the exact sealed bytes ──
let url = files[0]["url"].as_str().unwrap();
assert_eq!(
url,
format!("../../{WHEEL}"),
"relative to the project page, as PEP 691 allows"
);
let get = cl
.get(format!("{}/production/{WHEEL}", r.base))
.send()
.await
.unwrap();
assert_eq!(get.status(), 200);
let got = get.bytes().await.unwrap();
assert_eq!(&got[..], &wheel[..], "the wheel, byte for byte");
assert_eq!(sha256_hex(&got), sha);
let get = cl
.get(format!("{}/production/{WHEEL_NEXT}", r.base))
.send()
.await
.unwrap();
assert_eq!(get.status(), 404, "the INGESTED wheel is not downloadable");
// Signed in: the intake stages stay gated on an estate build (HCON-5), and this asks
// whether they serve an index, not whether they are gated.
for repo in ["incoming", "quarantine", "approved"] {
let get = cl
.get(format!("{}/{repo}/simple/demo-pkg/", r.base))
.bearer_auth(token("adm"))
.send()
.await
.unwrap();
assert_eq!(
get.status(),
404,
"only production serves an index ({repo})"
);
}
}
// ───────────────────────────── the real pip ─────────────────────────────
/// `podman` on `PATH`, or a failure that says so. Never a skip.
fn podman() -> PathBuf {
let path = std::env::var_os("PATH").unwrap_or_default();
for dir in std::env::split_paths(&path) {
let candidate = dir.join("podman");
if candidate.is_file() {
return candidate;
}
}
panic!(
"podman not on PATH — this test drives the REAL pip in the official python \
image and proves nothing without a container runtime; do not make it skip"
);
}
/// The image is present, or is pulled once. A failed pull is a failure.
fn ensure_image(podman: &Path, image: &str) {
let exists = Command::new(podman)
.args(["image", "exists", image])
.status()
.expect("podman runs");
if exists.success() {
return;
}
let pull = Command::new(podman)
.args(["pull", image])
.output()
.expect("podman runs");
assert!(
pull.status.success(),
"image not pullable: {image}\n{}",
String::from_utf8_lossy(&pull.stderr)
);
}
#[tokio::test(flavor = "multi_thread")]
#[ignore = "needs podman and the python image; reached by its arm in .nornir/testmatrix-arms.json"]
async fn real_pip_installs_and_imports_from_the_intake_index() {
let podman = podman();
ensure_image(&podman, PYTHON_IMAGE);
let r = start();
let cl = tls_roots(reqwest::Client::builder())
.build()
.expect("reqwest client");
wait_healthy(&cl, &r.base).await;
upload_and_walk(&r, &cl).await;
let index = format!("{}/production/simple/", r.base);
let script = format!(
"set -e; pip install --no-cache-dir --disable-pip-version-check --index-url {index} demo-pkg; \
python -c 'import demo_pkg; print(\"imported demo_pkg\", demo_pkg.__version__)'"
);
let base = r.base.clone();
let out = tokio::task::spawn_blocking(move || {
Command::new(&podman)
.args([
"run",
"--rm",
"--network",
"host",
PYTHON_IMAGE,
"sh",
"-c",
&script,
])
.output()
.expect("podman run")
})
.await
.unwrap();
let stdout = String::from_utf8_lossy(&out.stdout);
let stderr = String::from_utf8_lossy(&out.stderr);
assert!(
out.status.success(),
"pip install against {base} failed\nstdout:\n{stdout}\nstderr:\n{stderr}"
);
assert!(
stdout.contains("Successfully installed demo-pkg-1.0.0"),
"pip installed the AVAILABLE 1.0.0, not the INGESTED 1.1.0:\n{stdout}"
);
assert!(
stdout.contains("imported demo_pkg 1.0.0"),
"the package imports:\n{stdout}"
);
println!("{stdout}");
}
/// 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()),
)
}