server/lib/tests/intake_separation_of_duties.rs
//! **Separation of duties on the intake chain, over the wire verb.**
//!
//! One real server (four `raw` intake repositories, an HTTP and a gRPC door),
//! static-token principals with DUTIES from the same `auth` block as the roles,
//! and two-person approval (`approvals: 2`):
//!
//! * `ida` holds Importer AND SecurityApprover: she ingests over HTTP PUT and
//! still can never approve her own import (`holger-duty-1`);
//! * `aud` (Auditor) moves nothing (`holger-duty-2`), on gRPC and on HTTP, and
//! reads the state and its evidence;
//! * `dev` (Developer) cannot read the artifact before it is AVAILABLE
//! (`holger-duty-6`), and can after;
//! * `bob`'s approval alone is not enough (1 of 2), twice still counts once
//! (`holger-duty-3`), and `carol`'s second, distinct approval makes it
//! APPROVED with two attested approval slots;
//! * `adm` (RepositoryAdmin) cannot approve, and walks it on to AVAILABLE.
use std::io::Write;
use traits::test_support::hold_port;
use holger_server_lib::grpc::holger_proto::{
AdvanceArtifactRequest, ArtifactId as ProtoId, GetArtifactStateRequest, 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};
const PRINCIPALS: [&str; 6] = ["ida", "bob", "carol", "aud", "dev", "adm"];
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, SecurityApprover],
"bob": [SecurityApprover],
"carol": [SecurityApprover],
"aud": [Auditor],
"dev": [Developer],
"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"),
production = repo("production"),
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}");
}
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(),
}
}
fn proto_id() -> Option<ProtoId> {
Some(ProtoId {
namespace: String::new(),
name: WHEEL.into(),
version: String::new(),
})
}
async fn advance(
r: &Running,
who: &str,
verb: IntakeVerb,
) -> Result<holger_server_lib::grpc::holger_proto::AdvanceArtifactResponse, String> {
let req = AdvanceArtifactRequest {
artifact: proto_id(),
verb: verb 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,
}
}
Err(last)
}
async fn state(
r: &Running,
who: &str,
) -> Result<holger_server_lib::grpc::holger_proto::GetArtifactStateResponse, String> {
client::state(
&r.grpc,
Some(&token(who)),
GetArtifactStateRequest {
artifact: proto_id(),
},
)
.await
}
fn code(e: &str, want: &str) {
assert!(e.starts_with(&format!("[{want}]")), "want {want}, got: {e}");
}
#[tokio::test(flavor = "multi_thread")]
async fn a_wheel_needs_two_distinct_approvers_who_never_include_its_importer() {
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);
let bearer = |who: &str| format!("Bearer {}", token(who));
// ── the Importer ingests ──
let put = cl
.put(url("incoming"))
.header("Authorization", bearer("ida"))
.body(wheel.clone())
.send()
.await
.unwrap();
assert_eq!(
put.status(),
201,
"an Importer ingests: {}",
put.text().await.unwrap_or_default()
);
assert_eq!(chain.state(&wheel_id()), ArtifactState::Ingested);
assert_eq!(chain.importer(&wheel_id()).as_deref(), Some("ida"));
let put = cl
.put(url("incoming"))
.header("Authorization", bearer("aud"))
.body(wheel.clone())
.send()
.await
.unwrap();
assert_eq!(put.status(), 403, "an Auditor does not ingest");
// ── an Auditor advances nothing: gRPC and HTTP ──
for verb in [
IntakeVerb::Advance,
IntakeVerb::Quarantine,
IntakeVerb::Block,
] {
code(
&advance(&r, "aud", verb).await.expect_err("auditor refused"),
"holger-duty-2",
);
}
let post = cl
.post(format!(
"{}/-/intake/advance?name={WHEEL}&verb=advance",
r.base
))
.header("Authorization", bearer("aud"))
.send()
.await
.unwrap();
assert_eq!(post.status(), 403);
assert!(post.text().await.unwrap().starts_with("[holger-duty-2]"));
code(
&advance(&r, "dev", IntakeVerb::Advance)
.await
.expect_err("developer refused"),
"holger-duty-2",
);
assert_eq!(
chain.state(&wheel_id()),
ArtifactState::Ingested,
"nothing moved"
);
// ── a Developer cannot read a non-AVAILABLE artifact ──
code(
&state(&r, "dev").await.expect_err("developer refused"),
"holger-duty-6",
);
let get = cl
.get(format!("{}/-/intake/state?name={WHEEL}", r.base))
.header("Authorization", bearer("dev"))
.send()
.await
.unwrap();
assert_eq!(get.status(), 403);
for repo in ["incoming", "production"] {
let get = cl
.get(url(repo))
.header("Authorization", bearer("dev"))
.send()
.await
.unwrap();
assert_eq!(
get.status(),
404,
"a Developer downloads nothing before AVAILABLE ({repo})"
);
}
// ── the Importer moves it to SCANNED and POLICY_CHECKED ──
for want in [ArtifactState::Scanned, ArtifactState::PolicyChecked] {
let a = advance(&r, "ida", IntakeVerb::Advance)
.await
.unwrap_or_else(|e| panic!("→ {want}: {e}"));
assert_eq!(a.state, want.name());
}
// ── the importer never approves her own import, though she holds both duties ──
code(
&advance(&r, "ida", IntakeVerb::Advance)
.await
.expect_err("own import"),
"holger-duty-1",
);
code(
&advance(&r, "adm", IntakeVerb::Advance)
.await
.expect_err("an admin never approves"),
"holger-duty-2",
);
// ── two-person approval: one is not enough, the same one twice counts once ──
let one = advance(&r, "bob", IntakeVerb::Advance)
.await
.expect("bob approves");
assert_eq!(
(one.state.as_str(), one.approvals, one.approvals_required),
("POLICY_CHECKED", 1, 2)
);
assert_eq!(
chain.state(&wheel_id()),
ArtifactState::PolicyChecked,
"one approver is not enough"
);
code(
&advance(&r, "bob", IntakeVerb::Advance)
.await
.expect_err("twice"),
"holger-duty-3",
);
let two = advance(&r, "carol", IntakeVerb::Advance)
.await
.expect("carol approves");
assert_eq!(
(two.state.as_str(), two.approvals, two.approvals_required),
("APPROVED", 2, 2)
);
let who: Vec<(&str, &str)> = two
.evidence
.iter()
.map(|e| (e.kind.as_str(), e.source.as_str()))
.collect();
assert_eq!(who, vec![("approval", "bob"), ("approval", "carol")]);
assert!(
two.evidence
.iter()
.all(|e| e.reference.starts_with("sha256:")),
"attested: {:?}",
two.evidence
);
// ── on to AVAILABLE; the Repository Admin seals onward ──
for (who, want) in [("adm", "SEALED"), ("bob", "PROMOTED"), ("adm", "AVAILABLE")] {
let a = advance(&r, who, IntakeVerb::Advance)
.await
.unwrap_or_else(|e| panic!("{who} → {want}: {e}"));
assert_eq!(a.state, want);
}
// ── the Developer reads and downloads it now ──
let s = state(&r, "dev").await.expect("AVAILABLE is readable");
assert_eq!(s.state, "AVAILABLE");
let get = cl
.get(url("production"))
.header("Authorization", bearer("dev"))
.send()
.await
.unwrap();
assert_eq!(get.status(), 200);
assert_eq!(get.bytes().await.unwrap().as_ref(), &wheel[..]);
// ── the Auditor reads the whole trail and its evidence ──
let s = state(&r, "aud").await.expect("an auditor reads");
assert_eq!(s.importer, "ida");
let approved = s
.history
.iter()
.find(|t| t.to_state == "APPROVED")
.expect("the APPROVED entry");
assert_eq!(
approved.evidence.len(),
2,
"both attested approvals in the slot"
);
let actors: Vec<&str> = s.history.iter().map(|t| t.actor.as_str()).collect();
assert_eq!(
actors,
vec!["ida", "ida", "ida", "carol", "adm", "bob", "adm"]
);
code(
&advance(&r, "aud", IntakeVerb::Block)
.await
.expect_err("still nothing"),
"holger-duty-2",
);
}
/// 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()),
)
}