server/lib/tests/intake_security_dashboard.rs
//! **The security dashboard's two reads** (lane S9): `ListIntakeArtifacts`
//! and `GetArtifactEvidence`, over one real server's gRPC door.
//!
//! * the queues are the server's: five of them, named and titled by holger,
//! and a row moves from Incoming to Awaiting approval as the chain moves it;
//! * the read duty decides every row: a Developer is shown nothing before
//! AVAILABLE, neither listed nor counted;
//! * the vote count is the server's: one of two approvals shows as 1 of 2;
//! * the evidence of a SCANNED wheel carries its integrity hashes, every
//! scanner's verdict and the SBOM document.
// A server's test, never a gateway build's (lane HOL-CLEANUP): a gateway build is a
// downstream, and it scans, and a downstream runs no scanners (holger-downstream-1).
#![cfg(not(feature = "gateway"))]
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::{
AdvanceArtifactRequest, ArtifactId as ProtoId, GetArtifactEvidenceRequest, IntakeVerb,
ListIntakeArtifactsRequest, ListIntakeArtifactsResponse,
};
use holger_server_lib::{read_ron_config, wire_holger, Holger};
const PRINCIPALS: [&str; 5] = ["ida", "bob", "carol", "aud", "dev"];
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],
}},
)),
promotion: Some((
ladder_name: "intake",
advisory_db: Some("{COVERAGE}"),
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,
scan: Some((required: [])),
)),
)),
)"#,
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 proto_id() -> Option<ProtoId> {
Some(ProtoId {
namespace: String::new(),
name: WHEEL.into(),
version: String::new(),
})
}
async fn client(r: &Running) -> PromotionServiceClient<tonic::transport::Channel> {
let mut last = String::new();
for _ in 0..50 {
match PromotionServiceClient::connect(format!("http://{}", r.grpc)).await {
Ok(c) => return c,
Err(e) => last = e.to_string(),
}
tokio::time::sleep(std::time::Duration::from_millis(20)).await;
}
panic!("the gRPC door was not reached: {last}");
}
fn as_<T>(who: &str, body: T) -> tonic::Request<T> {
let mut r = tonic::Request::new(body);
r.metadata_mut().insert(
"authorization",
format!("Bearer {}", token(who)).parse().unwrap(),
);
r
}
async fn list(
r: &Running,
who: &str,
queue: &str,
) -> Result<ListIntakeArtifactsResponse, tonic::Status> {
client(r)
.await
.list_intake_artifacts(as_(
who,
ListIntakeArtifactsRequest {
queue: queue.into(),
limit: 0,
},
))
.await
.map(|r| r.into_inner())
}
async fn advance(r: &Running, who: &str) -> Result<String, String> {
let req = AdvanceArtifactRequest {
async_job: false,
artifact: proto_id(),
verb: IntakeVerb::Advance as i32,
..Default::default()
};
client(r)
.await
.advance_artifact(as_(who, req))
.await
.map(|r| r.into_inner().state)
.map_err(|s| s.message().to_string())
}
fn counts(l: &ListIntakeArtifactsResponse) -> Vec<(String, u64)> {
l.queues.iter().map(|q| (q.name.clone(), q.count)).collect()
}
fn count(l: &ListIntakeArtifactsResponse, queue: &str) -> u64 {
l.queues
.iter()
.find(|q| q.name == queue)
.map(|q| q.count)
.unwrap_or(u64::MAX)
}
#[tokio::test(flavor = "multi_thread")]
async fn the_dashboard_reads_the_servers_queues_votes_and_evidence() {
let r = start();
let cl = tls_roots(reqwest::Client::builder())
.build()
.expect("reqwest client");
wait_healthy(&cl, &r.base).await;
let _ = &r.holger;
// ── empty: five queues, the server's names and titles, all zero ──
let l = list(&r, "aud", "").await.expect("an auditor lists");
let titles: Vec<(&str, &str)> = l
.queues
.iter()
.map(|q| (q.name.as_str(), q.title.as_str()))
.collect();
assert_eq!(
titles,
vec![
("incoming", "Incoming"),
("quarantined", "Quarantined"),
("awaiting-approval", "Awaiting approval"),
("blocked", "Blocked"),
("available", "Approved / available"),
]
);
assert!(counts(&l).iter().all(|(_, n)| *n == 0), "{l:?}");
assert!(l.artifacts.is_empty(), "no queue asked, no rows");
// ── the Importer ingests: one row in Incoming ──
let put = cl
.put(format!("{}/incoming/{WHEEL}", r.base))
.header("Authorization", format!("Bearer {}", token("ida")))
.body(wheel_bytes())
.send()
.await
.unwrap();
assert_eq!(
put.status(),
201,
"{}",
put.text().await.unwrap_or_default()
);
let l = list(&r, "aud", "incoming").await.expect("an auditor lists");
assert_eq!(count(&l, "incoming"), 1);
assert_eq!(l.artifacts.len(), 1);
let row = &l.artifacts[0];
assert_eq!(
(
row.state.as_str(),
row.importer.as_str(),
row.repository.as_str()
),
("INGESTED", "ida", "incoming")
);
assert_eq!(row.artifact.as_ref().map(|a| a.name.as_str()), Some(WHEEL));
assert_eq!(row.sha256.len(), 64);
// ── a Developer is shown nothing before AVAILABLE: not listed, not counted ──
let l = list(&r, "dev", "incoming")
.await
.expect("a developer may list");
assert_eq!(count(&l, "incoming"), 0);
assert!(l.artifacts.is_empty());
let e = client(&r)
.await
.get_artifact_evidence(as_(
"dev",
GetArtifactEvidenceRequest {
artifact: proto_id(),
},
))
.await
.expect_err("a developer reads no evidence before AVAILABLE");
assert!(
e.message().starts_with("[holger-duty-6]"),
"{}",
e.message()
);
// ── an unknown queue is refused by name ──
let e = list(&r, "aud", "pending").await.expect_err("no such queue");
assert_eq!(e.code(), tonic::Code::InvalidArgument);
assert!(e.message().contains("awaiting-approval"), "{}", e.message());
// ── SCANNED: the evidence is the scanners' ──
assert_eq!(advance(&r, "ida").await.unwrap(), "SCANNED");
let ev = client(&r)
.await
.get_artifact_evidence(as_(
"aud",
GetArtifactEvidenceRequest {
artifact: proto_id(),
},
))
.await
.expect("an auditor reads the evidence")
.into_inner();
assert!(ev.scanned, "{ev:?}");
assert_eq!(ev.state, "SCANNED");
let algs: Vec<&str> = ev.hashes.iter().map(|h| h.alg.as_str()).collect();
assert_eq!(algs, vec!["sha256", "blake3"]);
assert_eq!(
ev.hashes[0].hex, ev.sha256,
"the scanned bytes are the ingested bytes"
);
assert_eq!(ev.file_name, WHEEL);
assert!(ev.size > 0);
let kinds: Vec<&str> = ev.records.iter().map(|r| r.kind.as_str()).collect();
for k in ["vulnerability", "sbom", "malware", "license", "provenance"] {
assert!(kinds.contains(&k), "a slot for {k}: {kinds:?}");
}
let sbom = ev.records.iter().find(|r| r.kind == "sbom").unwrap();
assert!(!sbom.not_configured);
let doc: serde_json::Value =
serde_json::from_str(&sbom.document_json).expect("the SBOM is JSON");
assert_eq!(doc["bomFormat"], "CycloneDX", "{doc}");
let malware = ev.records.iter().find(|r| r.kind == "malware").unwrap();
assert!(
malware.not_configured,
"no malware scanner is installed here, and the record says so"
);
assert!(!ev.verdict.is_empty());
// ── POLICY_CHECKED: Awaiting approval, and the server counts the votes ──
assert_eq!(advance(&r, "ida").await.unwrap(), "POLICY_CHECKED");
let l = list(&r, "aud", "awaiting-approval").await.unwrap();
assert_eq!(
(count(&l, "incoming"), count(&l, "awaiting-approval")),
(0, 1)
);
assert_eq!(
(l.artifacts[0].approvals, l.artifacts[0].approvals_required),
(0, 2)
);
assert_eq!(
advance(&r, "bob").await.unwrap(),
"POLICY_CHECKED",
"one vote is short of two"
);
let l = list(&r, "aud", "awaiting-approval").await.unwrap();
assert_eq!(
(l.artifacts[0].approvals, l.artifacts[0].approvals_required),
(1, 2)
);
let e = advance(&r, "ida")
.await
.expect_err("the importer never approves");
assert!(e.starts_with("[holger-duty-1]"), "{e}");
assert_eq!(advance(&r, "carol").await.unwrap(), "APPROVED");
let l = list(&r, "aud", "available").await.unwrap();
assert_eq!(
(count(&l, "awaiting-approval"), count(&l, "available")),
(0, 1)
);
assert_eq!(l.artifacts[0].state, "APPROVED");
}
/// 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()),
)
}
/// The box's advisory data covers every ecosystem this test takes in (H32-FAILCLOSED: a scan with
/// no advisory data for a package's ecosystem holds it): one record per ecosystem, on no real
/// package.
const COVERAGE: &str = concat!(
env!("CARGO_MANIFEST_DIR"),
"/tests/fixtures/coverage/osv-every-ecosystem.json"
);