server/lib/tests/what_c_downloaded_is_answered.rs
//! ★ **"What did C download" is answerable** (stinky SB #19): the audit query
//! came back as the minimal one — every DOWNLOAD row of the one audit log, by
//! who and by repository, in a time range — over gRPC
//! `AuditService.QueryDownloads` and HTTP `GET /-/audit/downloads`.
//!
//! * C reads, D reads, an admin writes: the answer for C is C's downloads only;
//! by repository it is everybody's downloads from it; a time range narrows.
//! * Each row names its `seq` and chain `hash`, and the chain verifies.
//! * Only an Auditor or an Admin reads it (`holger-audit-10`).
use std::io::Write;
use holger_server_lib::audit::trail::client;
use holger_server_lib::grpc::holger_proto::QueryDownloadsRequest;
use holger_server_lib::{read_ron_config, wire_holger, Holger};
use traits::test_support::hold_port;
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 try_start(grpc: u16, http: u16) -> Result<Running, String> {
let data = tempfile::tempdir().unwrap();
let d = data.path().display().to_string();
let methods: Vec<String> = ["c", "d", "aud", "adm"]
.iter()
.map(|p| {
format!(
r#"StaticToken(subject: "{p}", token: Some("{}"))"#,
token(p)
)
})
.collect();
let ron = format!(
r#"(
repositories: [
(ron_name: "crates", 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}/crates")),
(ron_name: "pypi", 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}/pypi")),
],
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} ],
roles: {{ "adm": Admin, "c": Reader, "d": Reader, "aud": Reader }},
duties: {{ "aud": [Auditor], "adm": [RepositoryAdmin] }},
require_auth_for_reads: true,
)),
)"#,
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: holger,
base: format!("http://127.0.0.1:{http}"),
grpc: format!("127.0.0.1:{grpc}"),
})
}
fn start() -> Running {
let mut last = String::new();
for _ in 0..8 {
match try_start(hold_port(), hold_port()) {
Ok(r) => return r,
Err(e) => last = e,
}
}
panic!("server never started; last error: {last}");
}
async fn downloads(
r: &Running,
who: &str,
q: QueryDownloadsRequest,
) -> Result<holger_server_lib::grpc::holger_proto::QueryDownloadsResponse, String> {
let mut last = String::new();
for _ in 0..100 {
match client::downloads(&r.grpc, Some(&token(who)), q.clone()).await {
Ok(t) => return Ok(t),
Err(e) if e.contains("was not reached") => {
last = e;
tokio::time::sleep(std::time::Duration::from_millis(20)).await;
}
Err(e) => return Err(e),
}
}
Err(last)
}
fn now_nanos() -> i64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos() as i64
}
#[tokio::test(flavor = "multi_thread")]
async fn what_c_downloaded_is_answered_by_ident_by_repository_and_by_time() {
let r = start();
let cl = reqwest::Client::new();
for _ in 0..100 {
if cl
.get(format!("{}/healthz", r.base))
.send()
.await
.map(|x| x.status() == 200)
.unwrap_or(false)
{
break;
}
tokio::time::sleep(std::time::Duration::from_millis(20)).await;
}
for (repo, name) in [
("crates", "a-1.tar"),
("crates", "b-1.tar"),
("pypi", "w.whl"),
] {
let put = cl
.put(format!("{}/{repo}/{name}", r.base))
.bearer_auth(token("adm"))
.body(name.as_bytes().to_vec())
.send()
.await
.unwrap();
assert!(put.status().is_success(), "{repo}/{name}: {}", put.status());
}
let get = |who: &'static str, path: &'static str| {
let (cl, base) = (cl.clone(), r.base.clone());
async move {
let g = cl
.get(format!("{base}/{path}"))
.bearer_auth(token(who))
.send()
.await
.unwrap();
assert!(
g.status().is_success(),
"{who} reads {path}: {}",
g.status()
);
}
};
get("c", "crates/a-1.tar").await;
get("d", "crates/b-1.tar").await;
get("c", "pypi/w.whl").await;
let mid = now_nanos();
tokio::time::sleep(std::time::Duration::from_millis(5)).await;
get("c", "crates/b-1.tar").await;
// By ident: C's downloads, nobody else's, no upload.
let c = downloads(
&r,
"aud",
QueryDownloadsRequest {
ident: "c".into(),
..Default::default()
},
)
.await
.expect("an Auditor reads");
let mut seen: Vec<(String, String)> = c
.rows
.iter()
.map(|x| (x.repo.clone(), x.artifact.clone()))
.collect();
seen.sort();
assert_eq!(c.matched, 3, "{:?}", c.rows);
assert!(
seen.iter()
.any(|(repo, a)| repo == "crates" && a.contains("a-1.tar")),
"{seen:?}"
);
assert!(
seen.iter()
.any(|(repo, a)| repo == "pypi" && a.contains("w.whl")),
"{seen:?}"
);
assert!(
c.rows
.iter()
.all(|x| x.ident == "c" && !x.hash.is_empty() && x.status == 200),
"every row is C's and names its chain hash: {:?}",
c.rows
);
// By repository: every download from it.
let crates = downloads(
&r,
"adm",
QueryDownloadsRequest {
repo: "crates".into(),
..Default::default()
},
)
.await
.expect("an Admin reads");
let mut who: Vec<&str> = crates.rows.iter().map(|x| x.ident.as_str()).collect();
who.sort();
assert_eq!(who, ["c", "c", "d"], "{:?}", crates.rows);
// By time: only what C downloaded after `mid`.
let late = downloads(
&r,
"aud",
QueryDownloadsRequest {
ident: "c".into(),
since_nanos: mid,
..Default::default()
},
)
.await
.unwrap();
assert_eq!(late.rows.len(), 1, "{:?}", late.rows);
assert!(late.rows[0].artifact.contains("b-1.tar"), "{:?}", late.rows);
// The same answer over HTTP.
let h = cl
.get(format!("{}/-/audit/downloads?ident=c&repo=crates", r.base))
.bearer_auth(token("aud"))
.send()
.await
.unwrap();
assert_eq!(h.status(), 200);
let v: serde_json::Value = h.json().await.unwrap();
assert_eq!(v["matched"], 2, "{v}");
assert!(
v["rows"]
.as_array()
.unwrap()
.iter()
.all(|x| x["ident"] == "c" && x["repo"] == "crates"),
"{v}"
);
// Only an Auditor or an Admin reads it.
let refused = downloads(
&r,
"c",
QueryDownloadsRequest {
ident: "c".into(),
..Default::default()
},
)
.await
.expect_err("a reader does not read the audit");
assert!(refused.contains("holger-audit-10"), "{refused}");
let h = cl
.get(format!("{}/-/audit/downloads?ident=c", r.base))
.bearer_auth(token("d"))
.send()
.await
.unwrap();
assert_eq!(h.status(), 403);
}