bench-scenarios/src/intake.rs
//! **The intake chain, measured per package.** Owner order 2026-10-23: "the bench
//! for testing verifying packages and all".
//!
//! Holger is a security gateway, and the thing it sells is the walk an artifact
//! takes from the wire to a download (`server_lib::promote::intake`). These arms
//! put a number on every step of that walk, per package type, with REAL public
//! packages (`fixtures/intake/fixtures.json`):
//!
//! | arm | what it times | the code it drives |
//! |---|---|---|
//! | `hash` | sha256 and blake3 by size | `security::blob_digest_hex`, `blake3::hash` |
//! | `inspect` | zip-bomb, nesting, MIME and archive expansion | `holger_inspect::Inspector` |
//! | `malware` | the inspector with no engine, YARA and ClamAV | `holger_inspect::malware` |
//! | `cve` | one advisory lookup against the real OSV dump | `scanner::osv::OsvVulnScanner` |
//! | `license` | the SPDX licence check | `scanner::license::LicenseScanner` |
//! | `sbom` | the CycloneDX SBOM generator | `scanner::wheel::WheelSbomScanner` |
//! | `provenance` | DSSE PAE + ed25519/P-256, in-toto, Fulcio chain, Rekor proof | `znippy_common::sign`, `attest::Statement` |
//! | `policy` | the promotion gates over one package | `promote::gate::catalog_from_config` |
//! | `seal` | seal into the production store | `promote::PromotionEngine::seal` |
//! | `e2e` | INGESTED → AVAILABLE for one package, and N in parallel | `promote::intake::IntakeChain` |
//! | `fetch` | a cold read out of a znippy archive on a (slow) disk | `RustRepoZnippy::with_archive` |
//!
//! Every arm has an ANSWER floor: a check that the step did its job (EICAR is
//! found, a known-vulnerable version is flagged, a real Chainguard signature
//! verifies, the Rekor proof reaches the logged root). A step that got faster by
//! not working is red, not green.
//!
//! **The lanes it follows.** The `provenance` arm runs `holger-provenance` (lane
//! G3a) over that crate's real fixtures, beside the primitives; the `policy` arm
//! runs the RON `rules` floor and a source's `gates` (lane G0, `sources.rs`)
//! beside the gate catalog. Both landed on main on 2026-10-23.
use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use std::sync::{Arc, OnceLock};
use std::time::{Duration, Instant};
use anyhow::{anyhow, bail, ensure, Context, Result};
use serde_json::{json, Map, Value};
use server_lib::scanner::{ScanSubject, ScanVerdict, Scanner};
use traits::{ArtifactId, RepositoryBackendTrait};
use crate::Scale;
pub type Metrics = Map<String, Value>;
// ─────────────────────────────── fixtures ───────────────────────────────
/// One real package from the manifest, with its bytes.
#[derive(Debug, Clone)]
pub struct Package {
pub key: String,
pub file: String,
pub ecosystem: String,
pub name: String,
pub version: String,
pub bytes: Arc<Vec<u8>>,
}
/// The loaded fixture set.
pub struct Fixtures {
pub committed: PathBuf,
pub cache: PathBuf,
pub manifest: Value,
pub packages: Vec<Package>,
/// Manifest keys with no file in the committed dir or the cache.
pub missing: Vec<String>,
}
/// The committed fixture directory (`bench-scenarios/fixtures/intake`).
pub fn committed_dir() -> PathBuf {
std::env::var_os("HOLGER_BENCH_FIXTURE_DIR")
.map(PathBuf::from)
.unwrap_or_else(|| PathBuf::from(concat!(env!("CARGO_MANIFEST_DIR"), "/fixtures/intake")))
}
/// The cache the large, fetched fixtures live in. `HOLGER_BENCH_FIXTURES`, else
/// `<workspace>/target/bench-fixtures`.
pub fn cache_dir() -> PathBuf {
std::env::var_os("HOLGER_BENCH_FIXTURES")
.map(PathBuf::from)
.unwrap_or_else(|| {
PathBuf::from(concat!(
env!("CARGO_MANIFEST_DIR"),
"/../target/bench-fixtures"
))
})
}
fn fetch_allowed() -> bool {
std::env::var("HOLGER_BENCH_FETCH")
.map(|v| v == "1")
.unwrap_or(false)
}
fn sha256_hex(b: &[u8]) -> String {
server_lib::security::blob_digest_hex(b)
}
fn http_get(url: &str, bearer: Option<&str>) -> Result<Vec<u8>> {
let mut req = ureq::get(url);
if let Some(t) = bearer {
req = req.header("Authorization", &format!("Bearer {t}"));
}
let mut resp = req.call().with_context(|| format!("GET {url}"))?;
resp.body_mut()
.with_config()
.limit(4 << 30)
.read_to_vec()
.with_context(|| format!("read {url}"))
}
/// Fetch one manifest entry into the cache (the online side only).
fn fetch_entry(entry: &Value, dest: &Path) -> Result<()> {
let bytes = if let Some(oci) = entry.get("oci") {
let registry = oci["registry"].as_str().unwrap_or("cgr.dev");
let repo = oci["repository"]
.as_str()
.ok_or_else(|| anyhow!("oci.repository"))?;
let digest = oci["digest"]
.as_str()
.ok_or_else(|| anyhow!("oci.digest"))?;
let tok: Value = serde_json::from_slice(&http_get(
&format!("https://{registry}/token?scope=repository:{repo}:pull"),
None,
)?)?;
let tok = tok["token"]
.as_str()
.ok_or_else(|| anyhow!("{registry} gave no token"))?
.to_string();
http_get(
&format!("https://{registry}/v2/{repo}/blobs/{digest}"),
Some(&tok),
)?
} else {
let url = entry["url"]
.as_str()
.ok_or_else(|| anyhow!("no url for {entry}"))?;
http_get(url, None)?
};
if let Some(want) = entry.get("sha256").and_then(Value::as_str) {
let got = sha256_hex(&bytes);
ensure!(
got == want,
"fixture {} is sha256 {got}, the manifest pins {want}",
dest.display()
);
}
if let Some(p) = dest.parent() {
std::fs::create_dir_all(p)?;
}
let tmp = dest.with_extension("part");
std::fs::write(&tmp, &bytes)?;
std::fs::rename(&tmp, dest)?;
Ok(())
}
fn locate(entry: &Value, committed: &Path, cache: &Path) -> Result<Option<PathBuf>> {
let file = entry["file"]
.as_str()
.ok_or_else(|| anyhow!("fixture without a file: {entry}"))?;
let c = committed.join(file);
if c.is_file() {
return Ok(Some(c));
}
let k = cache.join(file);
if k.is_file() {
return Ok(Some(k));
}
if entry
.get("committed")
.and_then(Value::as_bool)
.unwrap_or(false)
{
bail!("the committed fixture {} is missing", c.display());
}
if fetch_allowed() {
fetch_entry(entry, &k)?;
return Ok(Some(k));
}
Ok(None)
}
/// Load the manifest and every package it names that is present. With
/// `HOLGER_BENCH_FETCH=1`, missing large fixtures are fetched first.
pub fn load() -> Result<Fixtures> {
let committed = committed_dir();
let cache = cache_dir();
let manifest: Value = serde_json::from_slice(
&std::fs::read(committed.join("fixtures.json"))
.with_context(|| format!("{}/fixtures.json", committed.display()))?,
)?;
let mut packages = Vec::new();
let mut missing = Vec::new();
for e in manifest["packages"].as_array().into_iter().flatten() {
let key = e["key"].as_str().unwrap_or("?").to_string();
// A wheel committed as its members (text), made here (OpenSSF Scorecard counts a
// committed .whl as a binary artifact).
let made = e
.get("members")
.and_then(Value::as_str)
.map(|d| committed.join(d));
let found = match &made {
Some(d) => Some(d.clone()),
None => locate(e, &committed, &cache)?,
};
match found {
None => missing.push(key),
Some(path) => {
let bytes = if made.is_some() {
wheel_from_members(&path)?
} else {
std::fs::read(&path)?
};
if let Some(want) = e.get("sha256").and_then(Value::as_str) {
let got = sha256_hex(&bytes);
ensure!(
got == want,
"fixture {} is sha256 {got}, the manifest pins {want}",
path.display()
);
}
packages.push(Package {
key,
file: e["file"].as_str().unwrap_or_default().to_string(),
ecosystem: e["ecosystem"].as_str().unwrap_or("raw").to_string(),
name: e["name"].as_str().unwrap_or_default().to_string(),
version: e["version"].as_str().unwrap_or_default().to_string(),
bytes: Arc::new(bytes),
});
}
}
}
for e in manifest["advisories"].as_array().into_iter().flatten() {
if locate(e, &committed, &cache)?.is_none() {
missing.push(e["key"].as_str().unwrap_or("?").to_string());
}
}
Ok(Fixtures {
committed,
cache,
manifest,
packages,
missing,
})
}
/// Fetch every fixture that is not present yet (the online side). Returns the
/// paths it holds afterwards.
pub fn fetch_all() -> Result<Vec<PathBuf>> {
let committed = committed_dir();
let cache = cache_dir();
let manifest: Value = serde_json::from_slice(&std::fs::read(committed.join("fixtures.json"))?)?;
let mut out = Vec::new();
for list in ["packages", "advisories"] {
for e in manifest[list].as_array().into_iter().flatten() {
if let Some(d) = e.get("members").and_then(Value::as_str) {
out.push(committed.join(d));
continue;
}
let file = e["file"].as_str().unwrap_or_default();
let c = committed.join(file);
if c.is_file() {
out.push(c);
continue;
}
let k = cache.join(file);
if !k.is_file() {
eprintln!("bench fixtures: fetching {file}");
fetch_entry(e, &k)?;
}
out.push(k);
}
}
Ok(out)
}
static FIXTURES: OnceLock<std::result::Result<Fixtures, String>> = OnceLock::new();
/// The process-wide fixture set (loaded once, shared by every arm).
pub fn fixtures() -> Result<&'static Fixtures> {
FIXTURES
.get_or_init(|| load().map_err(|e| format!("{e:#}")))
.as_ref()
.map_err(|e| anyhow!("bench fixtures: {e}"))
}
pub(crate) fn base_metrics(fx: &Fixtures) -> Metrics {
let mut m = Metrics::new();
m.insert("skipped".into(), json!(0));
m.insert("fixtures_present".into(), json!(fx.packages.len()));
m.insert("fixtures_missing".into(), json!(fx.missing.len()));
if !fx.missing.is_empty() {
m.insert("fixtures_missing_keys".into(), json!(fx.missing.join(",")));
}
m
}
// ─────────────────────────────── timing ────────────────────────────────
/// Run `f` until `budget` has passed (at least `min_iters` times), and return
/// the per-call latencies.
pub(crate) fn sample(
min_iters: usize,
budget: Duration,
mut f: impl FnMut() -> Result<()>,
) -> Result<Vec<Duration>> {
let mut out = Vec::new();
let start = Instant::now();
while out.len() < min_iters || (start.elapsed() < budget && out.len() < 100_000) {
let t = Instant::now();
f()?;
out.push(t.elapsed());
}
Ok(out)
}
fn pct(lat: &mut [Duration], p: f64) -> f64 {
if lat.is_empty() {
return 0.0;
}
lat.sort();
let i = ((lat.len() as f64 - 1.0) * p).round() as usize;
lat[i.min(lat.len() - 1)].as_secs_f64() * 1e3
}
fn mean_ms(lat: &[Duration]) -> f64 {
if lat.is_empty() {
return 0.0;
}
lat.iter().map(Duration::as_secs_f64).sum::<f64>() / lat.len() as f64 * 1e3
}
/// `<prefix>_ms` (mean), `<prefix>_p50_ms`, `<prefix>_p99_ms` and, with a
/// size, `<prefix>_mbs`.
pub(crate) fn put_lat(m: &mut Metrics, prefix: &str, mut lat: Vec<Duration>, bytes: Option<usize>) {
let mean = mean_ms(&lat);
m.insert(format!("{prefix}_ms"), json!(mean));
m.insert(format!("{prefix}_p50_ms"), json!(pct(&mut lat, 0.50)));
m.insert(format!("{prefix}_p99_ms"), json!(pct(&mut lat, 0.99)));
if let Some(b) = bytes {
if mean > 0.0 {
m.insert(
format!("{prefix}_mbs"),
json!(b as f64 / 1_048_576.0 / (mean / 1e3)),
);
}
}
}
pub(crate) fn budget(scale: &Scale) -> Duration {
// quick: a third of a second per measured cell; full and huge: a second.
Duration::from_millis(if scale.bench_secs <= 2 { 300 } else { 1000 })
}
/// Peak resident set of this process (`VmHWM`), in MiB.
pub fn peak_rss_mb() -> f64 {
std::fs::read_to_string("/proc/self/status")
.ok()
.and_then(|s| {
s.lines()
.find(|l| l.starts_with("VmHWM:"))
.and_then(|l| l.split_whitespace().nth(1))
.and_then(|kb| kb.parse::<f64>().ok())
})
.map(|kb| kb / 1024.0)
.unwrap_or(0.0)
}
/// Reset `VmHWM` to the current RSS (Linux `clear_refs` 5), so the next peak
/// read belongs to the next measurement only.
pub fn reset_peak_rss() {
let _ = std::fs::write("/proc/self/clear_refs", "5");
}
// ──────────────────────────────── hash ─────────────────────────────────
/// sha256 (Holger's `blob_digest_hex`) and blake3 by input size.
pub fn hash_arm(scale: &Scale) -> Result<Metrics> {
let mut m = Metrics::new();
m.insert("skipped".into(), json!(0));
let b = budget(scale);
for (label, size) in [
("4k", 4usize << 10),
("64k", 64 << 10),
("1m", 1 << 20),
("16m", 16 << 20),
("256m", 256 << 20),
] {
let data: Vec<u8> = (0..size)
.map(|i| (i as u8).wrapping_mul(131).wrapping_add(7))
.collect();
let mut sha = String::new();
let lat = sample(3, b, || {
sha = std::hint::black_box(sha256_hex(&data));
Ok(())
})?;
put_lat(&mut m, &format!("sha256_{label}"), lat, Some(size));
let lat = sample(3, b, || {
std::hint::black_box(blake3::hash(&data));
Ok(())
})?;
put_lat(&mut m, &format!("blake3_{label}"), lat, Some(size));
ensure!(sha.len() == 64, "hash: sha256 hex is {} chars", sha.len());
}
// ANSWER floor: the spec's empty-input digest.
ensure!(
sha256_hex(b"") == "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
"hash: sha256 of the empty input is wrong"
);
m.insert(
"ops_per_sec".into(),
json!(m["sha256_1m_mbs"].as_f64().unwrap_or(0.0)),
);
Ok(m)
}
// ─────────────────────────────── inspect ───────────────────────────────
pub(crate) fn zip_of(members: &[(&str, &[u8])], deflate: bool) -> Result<Vec<u8>> {
use std::io::Write;
let mut w = zip::ZipWriter::new(std::io::Cursor::new(Vec::new()));
let method = if deflate {
zip::CompressionMethod::Deflated
} else {
zip::CompressionMethod::Stored
};
let opt = zip::write::SimpleFileOptions::default().compression_method(method);
for (name, body) in members {
w.start_file(*name, opt)?;
w.write_all(body)?;
}
Ok(w.finish()?.into_inner())
}
/// **A wheel made from its members**, committed as text under `dir` (a real package's
/// files, unpacked once): deflated, the members in path order with the `.dist-info/`
/// directory last and its `RECORD` at the very end, every timestamp 1980-01-01, so the
/// same members always make the same bytes. The `RECORD` is upstream's: it names the
/// sha256 of every other member, and the tests hold the made wheel to it.
pub fn wheel_from_members(dir: &Path) -> Result<Vec<u8>> {
fn walk(base: &Path, d: &Path, out: &mut Vec<(String, Vec<u8>)>) -> Result<()> {
let mut entries: Vec<_> = std::fs::read_dir(d)
.with_context(|| format!("{}", d.display()))?
.collect::<std::io::Result<_>>()?;
entries.sort_by_key(|e| e.file_name());
for e in entries {
let p = e.path();
if p.is_dir() {
walk(base, &p, out)?;
} else {
let name = p
.strip_prefix(base)
.expect("below base")
.to_string_lossy()
.replace('\\', "/");
out.push((name, std::fs::read(&p)?));
}
}
Ok(())
}
let mut members = Vec::new();
walk(dir, dir, &mut members)?;
ensure!(
!members.is_empty(),
"no wheel members under {}",
dir.display()
);
let rank = |n: &str| match (n.contains(".dist-info/"), n.ends_with(".dist-info/RECORD")) {
(_, true) => 2,
(true, false) => 1,
_ => 0,
};
members.sort_by(|a, b| (rank(&a.0), &a.0).cmp(&(rank(&b.0), &b.0)));
use std::io::Write;
let mut w = zip::ZipWriter::new(std::io::Cursor::new(Vec::new()));
let opt = zip::write::SimpleFileOptions::default()
.compression_method(zip::CompressionMethod::Deflated)
.last_modified_time(zip::DateTime::default())
.unix_permissions(0o644);
for (name, body) in &members {
w.start_file(name.as_str(), opt)?;
w.write_all(body)?;
}
Ok(w.finish()?.into_inner())
}
/// A wheel-shaped zip whose one member deflates 1 GiB of zeros: the classic
/// bomb. The inspector must refuse it, and fast.
fn zip_bomb() -> Result<Vec<u8>> {
use std::io::Write;
let mut w = zip::ZipWriter::new(std::io::Cursor::new(Vec::new()));
let opt = zip::write::SimpleFileOptions::default()
.compression_method(zip::CompressionMethod::Deflated)
.large_file(true);
w.start_file("bomb-1.0.dist-info/METADATA", opt)?;
w.write_all(b"Metadata-Version: 2.1\nName: bomb\nVersion: 1.0\n")?;
w.start_file("bomb/zeros.bin", opt)?;
let chunk = vec![0u8; 1 << 20];
for _ in 0..1024 {
w.write_all(&chunk)?;
}
Ok(w.finish()?.into_inner())
}
/// A zip nested `depth` levels deep (zip in zip in zip …).
fn nested_zip(depth: usize) -> Result<Vec<u8>> {
let mut inner = zip_of(&[("payload.txt", b"hello from the bottom")], true)?;
for d in 0..depth {
let name = format!("level{d}.zip");
inner = zip_of(&[(name.as_str(), &inner)], false)?;
}
Ok(inner)
}
fn inspector() -> holger_inspect::Inspector {
holger_inspect::Inspector::from_config(&holger_inspect::InspectConfig::default())
}
fn format_of(p: &Package) -> holger_inspect::Format {
holger_inspect::Format::declared(&p.file, &p.ecosystem)
}
/// The inspector over every real package, plus a zip bomb and deep nesting.
pub fn inspect_arm(scale: &Scale) -> Result<Metrics> {
let fx = fixtures()?;
let mut m = base_metrics(fx);
let insp = inspector();
let b = budget(scale);
for p in &fx.packages {
let fmt = format_of(p);
let f = insp.inspect(&p.bytes, fmt, &p.file);
let lat = sample(2, b, || {
std::hint::black_box(insp.inspect(&p.bytes, fmt, &p.file));
Ok(())
})?;
put_lat(
&mut m,
&format!("inspect_{}", p.key),
lat,
Some(p.bytes.len()),
);
m.insert(format!("inspect_{}_entries", p.key), json!(f.entries));
m.insert(
format!("inspect_{}_expanded_mb", p.key),
json!(f.expanded_bytes as f64 / 1_048_576.0),
);
m.insert(
format!("inspect_{}_verdict", p.key),
json!(format!("{:?}", f.verdict()).to_lowercase()),
);
m.insert(format!("inspect_{}_format", p.key), json!(fmt.name()));
m.insert(format!("size_{}_bytes", p.key), json!(p.bytes.len()));
}
// The bomb: refused, and the time to the refusal.
let bomb = zip_bomb()?;
let fmt = holger_inspect::Format::Wheel;
let f = insp.inspect(&bomb, fmt, "bomb-1.0-py3-none-any.whl");
ensure!(
f.has(
holger_inspect::FindingKind::ZipBomb,
holger_inspect::Severity::Block
) || f.has(
holger_inspect::FindingKind::ExpansionLimit,
holger_inspect::Severity::Block
) || f.has(
holger_inspect::FindingKind::EntryTooLarge,
holger_inspect::Severity::Block
),
"inspect: the 1 GiB zip bomb was not refused: {}",
f.summary()
);
let lat = sample(2, b, || {
std::hint::black_box(insp.inspect(&bomb, fmt, "bomb-1.0-py3-none-any.whl"));
Ok(())
})?;
put_lat(&mut m, "inspect_zipbomb", lat, None);
m.insert("inspect_zipbomb_bytes".into(), json!(bomb.len()));
// Nesting one past the default limit: refused by name.
let deep = nested_zip(holger_inspect::Limits::DEFAULT_MAX_DEPTH as usize + 2)?;
let f = insp.inspect(&deep, holger_inspect::Format::Generic, "deep.zip");
ensure!(
f.has(
holger_inspect::FindingKind::NestingTooDeep,
holger_inspect::Severity::Warn
),
"inspect: a zip nested {} deep was not flagged: {}",
holger_inspect::Limits::DEFAULT_MAX_DEPTH + 2,
f.summary()
);
let lat = sample(2, b, || {
std::hint::black_box(insp.inspect(&deep, holger_inspect::Format::Generic, "deep.zip"));
Ok(())
})?;
put_lat(&mut m, "inspect_nested", lat, None);
if let Some(v) = m.get("inspect_wheel_s_ms").and_then(Value::as_f64) {
m.insert(
"ops_per_sec".into(),
json!(if v > 0.0 { 1e3 / v } else { 0.0 }),
);
}
Ok(m)
}
// ─────────────────────────────── malware ───────────────────────────────
/// The EICAR test string, split so this source file is not itself a detection.
pub(crate) fn eicar() -> Vec<u8> {
let mut s = b"X5O!P%@AP[4\\PZX54(P^)7CC)7}$".to_vec();
s.extend_from_slice(b"EICAR-STANDARD-ANTIVIRUS-TEST-FILE!$H+H*");
s
}
/// A small, realistic rule set the built-in matcher runs: EICAR plus the shapes
/// a supply-chain rule pack looks for in Python, npm and shell payloads.
pub const YARA_RULES: &str = r#"
rule EICAR_Test_File { strings: $a = "EICAR-STANDARD-ANTIVIRUS-TEST-FILE" condition: $a }
rule Py_Exec_Base64 { strings: $a = "exec(base64.b64decode(" $b = "exec(__import__('base64')" condition: any of them }
rule Py_Install_Hook_Download { strings: $a = "urllib.request.urlopen(" $b = "cmdclass" $c = "os.system(" condition: all of them }
rule Npm_Preinstall_Curl { strings: $a = "\"preinstall\"" $b = "curl " $c = "| sh" condition: all of them }
rule Discord_Webhook_Exfil { strings: $a = "discord.com/api/webhooks/" nocase condition: $a }
rule Reverse_Shell { strings: $a = "/dev/tcp/" $b = "bash -i" condition: all of them }
rule Crypto_Miner { strings: $a = "stratum+tcp://" nocase $b = "xmrig" nocase condition: any of them }
rule PE_In_Package { strings: $mz = { 4D 5A 90 00 03 00 00 00 } condition: $mz }
"#;
/// The clamd endpoint (`HOLGER_BENCH_CLAMD=host:port` or a socket path).
fn clamd() -> Option<String> {
std::env::var("HOLGER_BENCH_CLAMD")
.ok()
.filter(|s| !s.trim().is_empty())
}
/// The inspector with no engine, with built-in YARA, and with ClamAV.
pub fn malware_arm(scale: &Scale) -> Result<Metrics> {
use holger_inspect::malware::{ClamAv, ClamAvConfig, MalwareScanner, ScanOutcome, Yara};
let fx = fixtures()?;
let mut m = base_metrics(fx);
let b = budget(scale);
let bare = inspector();
let yara: Arc<dyn MalwareScanner> = Arc::new(Yara::from_source(YARA_RULES, true));
ensure!(
Yara::from_source(YARA_RULES, true).engine_name().is_ok(),
"malware: the bench rule set does not fit the built-in YARA matcher"
);
let with_yara = inspector().with_scanner(yara.clone());
let clam: Option<Arc<dyn MalwareScanner>> = clamd().map(|s| {
Arc::new(ClamAv::from_config(&ClamAvConfig {
socket: Some(s),
required: true,
..ClamAvConfig::default()
})) as Arc<dyn MalwareScanner>
});
// ANSWER floors: each engine finds EICAR inside a wheel.
let infected = zip_of(&[("evil/__init__.py", &eicar())], true)?;
let f = with_yara.inspect(
&infected,
holger_inspect::Format::Wheel,
"evil-1.0-py3-none-any.whl",
);
ensure!(
f.has(
holger_inspect::FindingKind::Malware,
holger_inspect::Severity::Block
),
"malware: built-in YARA missed EICAR in a wheel: {}",
f.summary()
);
m.insert("yara_engine".into(), json!("built-in"));
m.insert("clamav_available".into(), json!(clam.is_some() as u8));
let with_clam = clam.as_ref().map(|c| inspector().with_scanner(c.clone()));
if let (Some(c), Some(insp)) = (&clam, &with_clam) {
match c.scan(&eicar(), "eicar.com") {
ScanOutcome::Detected(_) => {}
other => bail!(
"malware: ClamAV at {} did not detect EICAR: {other:?}",
clamd().unwrap_or_default()
),
}
let f = insp.inspect(
&infected,
holger_inspect::Format::Wheel,
"evil-1.0-py3-none-any.whl",
);
ensure!(
f.has(
holger_inspect::FindingKind::Malware,
holger_inspect::Severity::Block
),
"malware: ClamAV missed EICAR in a wheel: {}",
f.summary()
);
}
for p in &fx.packages {
let fmt = format_of(p);
let lat = sample(2, b, || {
std::hint::black_box(bare.inspect(&p.bytes, fmt, &p.file));
Ok(())
})?;
put_lat(&mut m, &format!("none_{}", p.key), lat, Some(p.bytes.len()));
let f = with_yara.inspect(&p.bytes, fmt, &p.file);
m.insert(
format!("yara_{}_hits", p.key),
json!(f
.findings
.iter()
.filter(|x| x.kind == holger_inspect::FindingKind::Malware)
.count()),
);
let lat = sample(2, b, || {
std::hint::black_box(with_yara.inspect(&p.bytes, fmt, &p.file));
Ok(())
})?;
put_lat(&mut m, &format!("yara_{}", p.key), lat, Some(p.bytes.len()));
if let Some(insp) = &with_clam {
// clamd caches a clean verdict by the bytes' hash, so every call gets
// bytes of its own (`variant`); the same bytes twice time the cache.
let mut lat = Vec::new();
let start = Instant::now();
let mut k = 0u64;
while lat.len() < 2 || (start.elapsed() < b && lat.len() < 200) {
k += 1;
let bytes = variant(&p.bytes, 9_000_000 + k);
let t = Instant::now();
std::hint::black_box(insp.inspect(&bytes, fmt, &p.file));
lat.push(t.elapsed());
}
put_lat(
&mut m,
&format!("clamav_{}", p.key),
lat,
Some(p.bytes.len()),
);
}
}
if let Some(v) = m.get("yara_wheel_s_ms").and_then(Value::as_f64) {
m.insert(
"ops_per_sec".into(),
json!(if v > 0.0 { 1e3 / v } else { 0.0 }),
);
}
Ok(m)
}
// ──────────────────────────────── CVE ──────────────────────────────────
/// The real OSV advisories (PyPI + npm + crates.io), loaded once.
struct Advisories {
db: server_lib::security::AdvisoryDb,
load_secs: f64,
by_feed: BTreeMap<String, usize>,
}
static ADVISORIES: OnceLock<std::result::Result<Advisories, String>> = OnceLock::new();
fn read_osv_zip(path: &Path) -> Result<Vec<modgunn::scan::Advisory>> {
let f = std::fs::File::open(path)?;
let mut z = zip::ZipArchive::new(f)?;
let mut out = Vec::new();
for i in 0..z.len() {
let mut e = z.by_index(i)?;
if !e.name().map(|n| n.ends_with(".json")).unwrap_or(false) {
continue;
}
let mut s = String::new();
std::io::Read::read_to_string(&mut e, &mut s)?;
if let Ok(v) = modgunn::osv::parse_doc(&s) {
out.extend(v);
}
}
Ok(out)
}
fn advisories() -> Result<&'static Advisories> {
ADVISORIES
.get_or_init(|| {
(|| -> Result<Advisories> {
let fx = fixtures()?;
let t = Instant::now();
let mut all = Vec::new();
let mut by_feed = BTreeMap::new();
for e in fx.manifest["advisories"].as_array().into_iter().flatten() {
let key = e["key"].as_str().unwrap_or("?").to_string();
let Some(path) = locate(e, &fx.committed, &fx.cache)? else {
continue;
};
let v = read_osv_zip(&path).with_context(|| format!("{}", path.display()))?;
by_feed.insert(key, v.len());
all.extend(v);
}
let db = server_lib::security::AdvisoryDb::new("bench-osv-real", all);
Ok(Advisories {
db,
load_secs: t.elapsed().as_secs_f64(),
by_feed,
})
})()
.map_err(|e| format!("{e:#}"))
})
.as_ref()
.map_err(|e| anyhow!("advisories: {e}"))
}
/// One advisory lookup per package against the real OSV dumps, plus the
/// known-vulnerable twins that must be flagged.
pub fn cve_arm(scale: &Scale) -> Result<Metrics> {
use server_lib::scanner::osv::OsvVulnScanner;
let fx = fixtures()?;
let mut m = base_metrics(fx);
let adv = advisories()?;
let total: usize = adv.by_feed.values().sum();
if total == 0 {
m.insert("skipped".into(), json!(1));
m.insert(
"why".into(),
json!("no OSV dump in the fixture cache (cargo xtask bench --fetch-fixtures)"),
);
return Ok(m);
}
m.insert("advisories".into(), json!(total));
for (k, n) in &adv.by_feed {
m.insert(format!("advisories_{k}"), json!(n));
}
m.insert("db_load_secs".into(), json!(adv.load_secs));
let t = Instant::now();
let scanner = OsvVulnScanner::new(adv.db.clone());
m.insert("db_clone_secs".into(), json!(t.elapsed().as_secs_f64()));
let b = budget(scale);
for p in &fx.packages {
if p.ecosystem == "oci" {
continue; // an image layer has no single package coordinate
}
let subject =
ScanSubject::new(&p.file, &p.ecosystem, &p.bytes).with_coords(&p.name, &p.version);
let out = scanner.scan(&subject)?;
m.insert(format!("cve_{}_findings", p.key), json!(out.findings.len()));
m.insert(
format!("cve_{}_verdict", p.key),
json!(out.verdict.as_str()),
);
let lat = sample(3, b, || {
std::hint::black_box(scanner.scan(&subject)?);
Ok(())
})?;
put_lat(&mut m, &format!("cve_{}", p.key), lat, None);
}
// ANSWER floors: versions with published advisories are flagged.
for (eco, name, ver, key) in [
("pip", "requests", "2.31.0", "requests_2_31_0"),
("npm", "lodash", "4.17.20", "lodash_4_17_20"),
("cargo", "tokio", "1.18.0", "tokio_1_18_0"),
] {
let file = format!("{name}-{ver}");
let s = ScanSubject::new(&file, eco, b"").with_coords(name, ver);
let out = scanner.scan(&s)?;
m.insert(
format!("cve_floor_{key}_findings"),
json!(out.findings.len()),
);
ensure!(
!out.findings.is_empty(),
"cve: {eco} {name} {ver} has published advisories and the lookup found none ({})",
out.detail
);
}
if let Some(v) = m.get("cve_wheel_s_ms").and_then(Value::as_f64) {
m.insert(
"ops_per_sec".into(),
json!(if v > 0.0 { 1e3 / v } else { 0.0 }),
);
}
Ok(m)
}
// ────────────────────────────── license ────────────────────────────────
fn license_policy() -> server_lib::scanner::license::LicensePolicy {
use server_lib::scanner::license::{LicensePolicy, UnknownLicense};
LicensePolicy::new(
vec![],
vec![
"AGPL-3.0-only".into(),
"GPL-3.0-only".into(),
"SSPL-1.0".into(),
],
UnknownLicense::Hold,
)
}
/// The SPDX licence check per package.
pub fn license_arm(scale: &Scale) -> Result<Metrics> {
use server_lib::scanner::license::LicenseScanner;
let fx = fixtures()?;
let mut m = base_metrics(fx);
let sc = LicenseScanner::new(license_policy());
let b = budget(scale);
for p in &fx.packages {
let j = sc.judge(&p.ecosystem, &p.bytes);
let out = j.outcome();
m.insert(
format!("license_{}_verdict", p.key),
json!(out.verdict.as_str()),
);
m.insert(format!("license_{}_hold", p.key), json!(out.hold as u8));
let lat = sample(2, b, || {
std::hint::black_box(sc.judge(&p.ecosystem, &p.bytes));
Ok(())
})?;
put_lat(
&mut m,
&format!("license_{}", p.key),
lat,
Some(p.bytes.len()),
);
}
// ANSWER floor: requests declares Apache-2.0 and the policy allows it.
if let Some(p) = fx.packages.iter().find(|p| p.key == "wheel_s") {
let v = sc.judge(&p.ecosystem, &p.bytes).outcome().verdict;
ensure!(
v == ScanVerdict::Pass,
"license: requests (Apache-2.0) did not pass: {v}"
);
}
if let Some(v) = m.get("license_wheel_s_ms").and_then(Value::as_f64) {
m.insert(
"ops_per_sec".into(),
json!(if v > 0.0 { 1e3 / v } else { 0.0 }),
);
}
Ok(m)
}
// ──────────────────────────────── SBOM ─────────────────────────────────
/// The CycloneDX SBOM per wheel (the only generator Holger ships today).
pub fn sbom_arm(scale: &Scale) -> Result<Metrics> {
use server_lib::scanner::wheel::WheelSbomScanner;
let fx = fixtures()?;
let mut m = base_metrics(fx);
let b = budget(scale);
let mut generated = 0;
for p in &fx.packages {
let subject = ScanSubject::new(&p.file, &p.ecosystem, &p.bytes);
let out = WheelSbomScanner.scan(&subject)?;
let has_doc = out.document.is_some();
m.insert(format!("sbom_{}_generated", p.key), json!(has_doc as u8));
if !has_doc {
continue;
}
generated += 1;
let doc = serde_json::to_vec(out.document.as_ref().unwrap())?;
m.insert(format!("sbom_{}_doc_bytes", p.key), json!(doc.len()));
let lat = sample(2, b, || {
std::hint::black_box(WheelSbomScanner.scan(&subject)?);
Ok(())
})?;
put_lat(&mut m, &format!("sbom_{}", p.key), lat, Some(p.bytes.len()));
}
m.insert("sbom_generators".into(), json!("wheel"));
m.insert("sbom_generated".into(), json!(generated));
ensure!(
generated > 0 || !fx.packages.iter().any(|p| p.file.ends_with(".whl")),
"sbom: no wheel produced a CycloneDX document"
);
if let Some(v) = m.get("sbom_wheel_s_ms").and_then(Value::as_f64) {
m.insert(
"ops_per_sec".into(),
json!(if v > 0.0 { 1e3 / v } else { 0.0 }),
);
}
Ok(m)
}
// ───────────────────────────── provenance ──────────────────────────────
fn pem_blocks(text: &str) -> Vec<Vec<u8>> {
let mut out = Vec::new();
let mut cur = String::new();
let mut inside = false;
for line in text.lines() {
if line.starts_with("-----BEGIN") {
inside = true;
cur.clear();
} else if line.starts_with("-----END") {
inside = false;
if let Some(d) = znippy_common::sign::base64_decode(&cur) {
out.push(d);
}
} else if inside {
cur.push_str(line.trim());
}
}
out
}
fn sha256(b: &[u8]) -> Vec<u8> {
use sha2::Digest;
sha2::Sha256::digest(b).to_vec()
}
fn node(l: &[u8], r: &[u8]) -> Vec<u8> {
let mut v = Vec::with_capacity(65);
v.push(1u8);
v.extend_from_slice(l);
v.extend_from_slice(r);
sha256(&v)
}
/// RFC 9162 §2.1.3.2: the root an inclusion proof reaches. A bench-side reference
/// until `holger-provenance` (lane G3a) lands on main with its own.
pub fn merkle_root_from_proof(
index: u64,
size: u64,
leaf: &[u8],
proof: &[Vec<u8>],
) -> Option<Vec<u8>> {
if index >= size {
return None;
}
let (mut f, mut s) = (index, size - 1);
let mut r = leaf.to_vec();
for p in proof {
if s == 0 {
return None;
}
if f & 1 == 1 || f == s {
r = node(p, &r);
if f & 1 == 0 {
while f & 1 == 0 && f != 0 {
f >>= 1;
s >>= 1;
}
}
} else {
r = node(&r, p);
}
f >>= 1;
s >>= 1;
}
(s == 0).then_some(r)
}
fn hex_decode(s: &str) -> Option<Vec<u8>> {
(0..s.len())
.step_by(2)
.map(|i| u8::from_str_radix(s.get(i..i + 2)?, 16).ok())
.collect()
}
/// One Rekor entry's inclusion proof, parsed.
struct RekorProof {
leaf: Vec<u8>,
index: u64,
size: u64,
root: Vec<u8>,
hashes: Vec<Vec<u8>>,
checkpoint: String,
}
fn rekor_proof(entry: &Value) -> Result<RekorProof> {
let v = entry
.as_object()
.and_then(|o| o.values().next())
.ok_or_else(|| anyhow!("rekor: empty entry"))?;
let body = znippy_common::sign::base64_decode(v["body"].as_str().unwrap_or_default())
.ok_or_else(|| anyhow!("rekor: body is not base64"))?;
let mut leaf_in = vec![0u8];
leaf_in.extend_from_slice(&body);
let p = &v["verification"]["inclusionProof"];
Ok(RekorProof {
leaf: sha256(&leaf_in),
index: p["logIndex"]
.as_u64()
.ok_or_else(|| anyhow!("rekor: logIndex"))?,
size: p["treeSize"]
.as_u64()
.ok_or_else(|| anyhow!("rekor: treeSize"))?,
root: hex_decode(p["rootHash"].as_str().unwrap_or_default())
.ok_or_else(|| anyhow!("rekor: rootHash"))?,
hashes: p["hashes"]
.as_array()
.into_iter()
.flatten()
.map(|h| {
hex_decode(h.as_str().unwrap_or_default()).ok_or_else(|| anyhow!("rekor: hash"))
})
.collect::<Result<_>>()?,
checkpoint: p["checkpoint"].as_str().unwrap_or_default().to_string(),
})
}
/// Verify a Rekor checkpoint (a signed note) with the log's P-256 key. Returns
/// the root hash the note commits to.
fn verify_checkpoint(note: &str, spki_der: &[u8]) -> Result<Vec<u8>> {
let (body, sigs) = note
.split_once("\n\n")
.ok_or_else(|| anyhow!("checkpoint: no signature block"))?;
let signed = format!("{body}\n");
let line = sigs
.lines()
.next()
.ok_or_else(|| anyhow!("checkpoint: empty signature block"))?;
let b64 = line.rsplit(' ').next().unwrap_or_default();
let raw = znippy_common::sign::base64_decode(b64)
.ok_or_else(|| anyhow!("checkpoint: signature is not base64"))?;
ensure!(raw.len() > 4, "checkpoint: short signature");
// The SPKI's last 65 bytes are the uncompressed P-256 point.
ensure!(spki_der.len() >= 65, "checkpoint: short key");
let point = &spki_der[spki_der.len() - 65..];
let key =
ring::signature::UnparsedPublicKey::new(&ring::signature::ECDSA_P256_SHA256_ASN1, point);
key.verify(signed.as_bytes(), &raw[4..])
.map_err(|_| anyhow!("checkpoint: the signature does not verify"))?;
let root_b64 = body
.lines()
.nth(2)
.ok_or_else(|| anyhow!("checkpoint: no root line"))?;
znippy_common::sign::base64_decode(root_b64)
.ok_or_else(|| anyhow!("checkpoint: root is not base64"))
}
/// DSSE PAE + signature verify (ed25519 and P-256) through the znippy verifier
/// Holger's attest store uses, an in-toto parse of a real SLSA statement, the
/// Chainguard envelope against its Fulcio certificate and chain, and the
/// Rekor inclusion proof and checkpoint.
pub fn provenance_arm(scale: &Scale) -> Result<Metrics> {
use znippy_common::sign::{
dev, dsse_sign, dsse_verify, ArchiveSigner, CertStore, DsseEnvelope,
};
let fx = fixtures()?;
let mut m = base_metrics(fx);
let b = budget(scale);
m.insert("holger_provenance_crate".into(), json!(1));
holger_provenance_cases(&mut m, b)?;
let pdir = fx.committed.join(
fx.manifest["provenance"]["dir"]
.as_str()
.unwrap_or("chainguard-python"),
);
let att = &fx.manifest["provenance"]["attestation"];
let envelope_text =
std::fs::read_to_string(pdir.join(att["envelope"].as_str().unwrap_or("att-0.payload")))?;
let annotations: Value = serde_json::from_slice(&std::fs::read(
pdir.join(
att["annotations"]
.as_str()
.unwrap_or("att-0.annotations.json"),
),
)?)?;
// 1. Holger's own signers: PAE + sign + verify, ed25519 and P-256.
let env = DsseEnvelope::from_json(&envelope_text)?;
let statement = env.payload.clone();
let (ca_key, ca_der) = dev::mint_ca("holger-bench-ca")?;
let roots = CertStore::from_der_certs(&[ca_der.clone()])?;
let ed = dev::new_ed25519_signer(&ca_key, &ca_der, "bench-ed25519")?;
let p256 = dev::new_p256_signer(&ca_key, &ca_der, "bench-p256")?;
for (label, signer) in [
("ed25519", &ed as &dyn ArchiveSigner),
("p256", &p256 as &dyn ArchiveSigner),
] {
let lat = sample(3, b, || {
std::hint::black_box(dsse_sign(
"application/vnd.in-toto+json",
&statement,
&[signer],
)?);
Ok(())
})?;
put_lat(&mut m, &format!("dsse_sign_{label}"), lat, None);
let signed = dsse_sign("application/vnd.in-toto+json", &statement, &[signer])?;
let certs = vec![signer.certificate_der().to_vec()];
ensure!(
!dsse_verify(&signed, &certs, &roots)?.is_empty(),
"provenance: own {label} envelope did not verify"
);
let mut bad = signed.clone();
bad.payload[0] ^= 1;
ensure!(
dsse_verify(&bad, &certs, &roots).is_err(),
"provenance: a tampered {label} envelope verified"
);
let lat = sample(3, b, || {
std::hint::black_box(dsse_verify(&signed, &certs, &roots)?);
Ok(())
})?;
put_lat(&mut m, &format!("dsse_verify_{label}"), lat, None);
}
let lat = sample(3, b, || {
std::hint::black_box(env.pae());
Ok(())
})?;
put_lat(&mut m, "dsse_pae", lat, None);
// 2. in-toto: Holger's statement parser over the real SLSA v1 statement.
// FINDING (2026-10-23): Chainguard's cosign attestations are in-toto
// Statement v0.1, and `attest::Statement::parse` accepted v1 only. FIXED
// 2026-10-24 (lane G3): v0.1 is accepted; `intoto_v01_accepted` stays as the
// metric (1 from now on). The v1-typed copy is what is timed, as before.
let v01 = server_lib::attest::Statement::parse(&statement);
m.insert("intoto_v01_accepted".into(), json!(v01.is_ok() as u8));
if let Err(e) = &v01 {
m.insert("intoto_v01_refusal".into(), json!(e.to_string()));
}
let statement = {
let mut v: Value = serde_json::from_slice(&statement)?;
m.insert("intoto_type".into(), v["_type"].clone());
v["_type"] = json!("https://in-toto.io/Statement/v1");
serde_json::to_vec(&v)?
};
let st = server_lib::attest::Statement::parse(&statement)?;
ensure!(
st.predicate_type == "https://slsa.dev/provenance/v1",
"provenance: the Chainguard statement is {:?}",
st.predicate_type
);
m.insert("intoto_statement_bytes".into(), json!(statement.len()));
let lat = sample(3, b, || {
std::hint::black_box(server_lib::attest::Statement::parse(&statement)?);
Ok(())
})?;
put_lat(&mut m, "intoto_parse", lat, None);
// 3. The real Chainguard envelope: P-256 over PAE with the Fulcio leaf, and
// the leaf chained to the Fulcio intermediate and root it carries. FIXED
// 2026-10-24 (znippy-common 0.10.1): the chain walks through intermediates
// and takes Fulcio's ECDSA SHA-384 / P-384 links; `chainguard_dsse_fulcio_verified` is 1.
let leaf = pem_blocks(
annotations["dev.sigstore.cosign/certificate"]
.as_str()
.unwrap_or_default(),
);
let chain = pem_blocks(
annotations["dev.sigstore.cosign/chain"]
.as_str()
.unwrap_or_default(),
);
ensure!(
leaf.len() == 1 && !chain.is_empty(),
"provenance: the annotations carry no Fulcio certificate and chain"
);
let fulcio = CertStore::from_der_certs(&chain)?;
let chainguard = dsse_verify(&env, &leaf, &fulcio);
m.insert(
"chainguard_dsse_fulcio_verified".into(),
json!(chainguard.is_ok() as u8),
);
if let Err(e) = &chainguard {
m.insert(
"chainguard_dsse_fulcio_error".into(),
json!(format!("{e:#}")),
);
}
// The signature alone, P-256 over PAE with the leaf key (ring), so the
// number exists whatever the chain walk says.
let leaf_point = {
let (_, cert) =
x509_parser::parse_x509_certificate(&leaf[0]).map_err(|e| anyhow!("leaf: {e}"))?;
cert.public_key().subject_public_key.data.to_vec()
};
let pae = env.pae();
let sig = env
.signatures
.first()
.ok_or_else(|| anyhow!("provenance: the envelope has no signature"))?
.sig
.clone();
let key = ring::signature::UnparsedPublicKey::new(
&ring::signature::ECDSA_P256_SHA256_ASN1,
leaf_point,
);
ensure!(
key.verify(&pae, &sig).is_ok(),
"provenance: Chainguard's P-256 signature over PAE does not verify"
);
let lat = sample(3, b, || {
key.verify(&pae, &sig).map_err(|_| anyhow!("p256"))?;
Ok(())
})?;
put_lat(&mut m, "chainguard_p256_verify", lat, None);
if chainguard.is_ok() {
let lat = sample(3, b, || {
std::hint::black_box(dsse_verify(&env, &leaf, &fulcio)?);
Ok(())
})?;
put_lat(&mut m, "chainguard_dsse_fulcio", lat, None);
}
// The Fulcio chain, one signature per link (ring, P-384 for Sigstore's CA).
{
let mut links = Vec::new();
let all: Vec<Vec<u8>> = leaf.iter().chain(chain.iter()).cloned().collect();
for w in all.windows(2) {
let (_, child) =
x509_parser::parse_x509_certificate(&w[0]).map_err(|e| anyhow!("cert: {e}"))?;
let (_, parent) =
x509_parser::parse_x509_certificate(&w[1]).map_err(|e| anyhow!("cert: {e}"))?;
let alg = match parent.public_key().subject_public_key.data.len() {
97 => &ring::signature::ECDSA_P384_SHA384_ASN1,
_ => &ring::signature::ECDSA_P256_SHA256_ASN1,
};
let alg = if child.signature_algorithm.algorithm.to_id_string() == "1.2.840.10045.4.3.2"
{
&ring::signature::ECDSA_P256_SHA256_ASN1
} else if child.signature_algorithm.algorithm.to_id_string() == "1.2.840.10045.4.3.3" {
if parent.public_key().subject_public_key.data.len() == 97 {
&ring::signature::ECDSA_P384_SHA384_ASN1
} else {
&ring::signature::ECDSA_P256_SHA384_ASN1
}
} else {
alg
};
links.push((
child.tbs_certificate.as_ref().to_vec(),
child.signature_value.data.to_vec(),
parent.public_key().subject_public_key.data.to_vec(),
alg,
));
}
for (tbs, sig, key, alg) in &links {
ring::signature::UnparsedPublicKey::new(*alg, key)
.verify(tbs, sig)
.map_err(|_| anyhow!("provenance: a Fulcio chain link does not verify"))?;
}
m.insert("fulcio_chain_links".into(), json!(links.len()));
let lat = sample(3, b, || {
for (tbs, sig, key, alg) in &links {
ring::signature::UnparsedPublicKey::new(*alg, key)
.verify(tbs, sig)
.map_err(|_| anyhow!("link"))?;
}
Ok(())
})?;
put_lat(&mut m, "fulcio_chain", lat, None);
}
// 4. Rekor: the offline inclusion proof and the signed checkpoint.
let rekor: Value = serde_json::from_slice(&std::fs::read(
pdir.join(att["rekor"].as_str().unwrap_or("rekor-3416929314.json")),
)?)?;
let proof = rekor_proof(&rekor)?;
let root = merkle_root_from_proof(proof.index, proof.size, &proof.leaf, &proof.hashes)
.ok_or_else(|| anyhow!("provenance: the inclusion proof is malformed"))?;
ensure!(
root == proof.root,
"provenance: the Rekor inclusion proof does not reach the logged root"
);
m.insert("rekor_proof_hashes".into(), json!(proof.hashes.len()));
m.insert("rekor_tree_size".into(), json!(proof.size));
let lat = sample(3, b, || {
let r = merkle_root_from_proof(proof.index, proof.size, &proof.leaf, &proof.hashes)
.ok_or_else(|| anyhow!("proof"))?;
ensure!(r == proof.root, "proof");
Ok(())
})?;
put_lat(&mut m, "rekor_inclusion", lat, None);
let pubkey = pem_blocks(&std::fs::read_to_string(
pdir.join(
fx.manifest["provenance"]["rekor_public_key"]
.as_str()
.unwrap_or("rekor-pubkey.pem"),
),
)?);
let spki = pubkey
.first()
.ok_or_else(|| anyhow!("provenance: no Rekor key"))?;
let cp_root = verify_checkpoint(&proof.checkpoint, spki)?;
ensure!(
cp_root == proof.root,
"provenance: the checkpoint commits to another root"
);
let lat = sample(3, b, || {
std::hint::black_box(verify_checkpoint(&proof.checkpoint, spki)?);
Ok(())
})?;
put_lat(&mut m, "rekor_checkpoint", lat, None);
// The whole offline check for one image signature, as one number.
let total = [
"chainguard_p256_verify_ms",
"fulcio_chain_ms",
"rekor_inclusion_ms",
"rekor_checkpoint_ms",
"intoto_parse_ms",
]
.iter()
.filter_map(|k| m.get(*k).and_then(Value::as_f64))
.sum::<f64>();
m.insert("primitives_total_ms".into(), json!(total));
// The headline: one Chainguard image (four candidates) through holger-provenance.
let total = m
.get("prov_chainguard_image_ms")
.and_then(Value::as_f64)
.unwrap_or(total);
m.insert(
"ops_per_sec".into(),
json!(if total > 0.0 { 1e3 / total } else { 0.0 }),
);
Ok(m)
}
/// `holger_provenance::verify_artifact` over that crate's own real, public
/// fixtures (`provenance/tests/fixtures`, fetched 2026-10-23 by lane G3a): a
/// Chainguard image (one cosign signature and three attestations), a PyPI PEP 740
/// attestation and an npm provenance plus the registry's publish attestation, all
/// verified offline against Sigstore's pinned public-good trust root.
fn holger_provenance_cases(m: &mut Metrics, b: Duration) -> Result<()> {
use holger_provenance::{
effective, verify_artifact, Identity, KeyAlg, ProvenanceFloor, ProvenanceGate, TrustRoot,
TrustedKey, Verdict,
};
let fx = PathBuf::from(concat!(
env!("CARGO_MANIFEST_DIR"),
"/../provenance/tests/fixtures"
));
let read =
|p: &str| std::fs::read(fx.join(p)).with_context(|| format!("provenance fixture {p}"));
const GH: &str = "https://token.actions.githubusercontent.com";
let root_json = read("sigstore/trusted_root.json")?;
let lat = sample(3, b, || {
std::hint::black_box(
TrustRoot::from_trusted_root_json(&root_json).map_err(|e| anyhow!("{e}"))?,
);
Ok(())
})?;
put_lat(m, "prov_trust_root_load", lat, None);
let trust = TrustRoot::from_trusted_root_json(&root_json).map_err(|e| anyhow!("{e}"))?;
let req = |identity: Option<Identity>, slsa: Option<u8>| ProvenanceGate::Require {
identity,
identities: vec![],
builders: vec![],
keys: vec![],
apk_keys: vec![],
pgp_keys: vec![],
slsa_min: slsa,
};
let npm_keys: Value = serde_json::from_slice(&read("npm/npm-keys.json")?)?;
let npm_key = npm_keys["keys"]
.as_array()
.and_then(|a| a.iter().find(|k| k["expires"].is_null()))
.ok_or_else(|| anyhow!("npm-keys.json has no current key"))?;
let cases = vec![
(
"chainguard_image",
read("chainguard/static-latest.index.json")?,
("chainguard/static-latest.cosign.json", read("chainguard/static-latest.cosign.json")?),
effective(
&ProvenanceFloor { require_signed: true, ..Default::default() },
&req(
Some(Identity::exact(GH, "https://github.com/chainguard-images/images/.github/workflows/release.yaml@refs/heads/main")),
Some(2),
),
),
),
(
"pypi_pep740",
read("pypi/sampleproject-4.0.0.tar.gz")?,
(
"pypi/sampleproject-4.0.0.tar.gz.provenance.json",
read("pypi/sampleproject-4.0.0.tar.gz.provenance.json")?,
),
effective(
&ProvenanceFloor::default(),
&req(Some(Identity::exact(GH, "https://github.com/pypa/sampleproject/.github/workflows/release.yml@refs/heads/main")), Some(2)),
),
),
(
"npm_provenance",
read("npm/sigstore-5.0.0.tgz")?,
("npm/sigstore-5.0.0.attestations.json", read("npm/sigstore-5.0.0.attestations.json")?),
effective(
&ProvenanceFloor {
identities: vec![Identity::exact(GH, "https://github.com/sigstore/sigstore-js/.github/workflows/release.yml@refs/heads/main")],
keys: vec![TrustedKey {
id: npm_key["keyid"].as_str().unwrap_or_default().into(),
alg: KeyAlg::EcdsaP256,
public_key: npm_key["key"].as_str().unwrap_or_default().into(),
}],
builders: vec!["https://github.com/actions/runner/github-hosted".into()],
..Default::default()
},
&req(None, Some(2)),
),
),
];
for (label, artifact, (name, bytes), policy) in &cases {
let materials = vec![(name.to_string(), bytes.clone())];
let r = verify_artifact(artifact, &materials, policy, Some(&trust));
// ANSWER floor: every real case verifies, every candidate in it.
ensure!(
r.verdict == Verdict::Pass,
"provenance: {label} did not verify: {} {:?}",
r.detail,
r.code
);
ensure!(
r.candidates.iter().all(|c| c.verified),
"provenance: a {label} candidate did not verify"
);
m.insert(
format!("prov_{label}_candidates"),
json!(r.candidates.len()),
);
let lat = sample(3, b, || {
std::hint::black_box(verify_artifact(artifact, &materials, policy, Some(&trust)));
Ok(())
})?;
put_lat(m, &format!("prov_{label}"), lat, None);
// And other bytes are refused, never passed.
let other = verify_artifact(b"other bytes", &materials, policy, Some(&trust));
ensure!(
other.verdict == Verdict::Block,
"provenance: {label}'s material vouched for other bytes"
);
}
Ok(())
}
// ──────────────────────────────── policy ───────────────────────────────
/// The promotion gates over one package: the OSV gate (the real advisories),
/// the provenance gate and the attested gate, as `catalog_from_config` builds
/// them.
pub fn policy_arm(scale: &Scale) -> Result<Metrics> {
use server_lib::attest::ArtifactDigest;
use server_lib::promote::gate::{catalog_from_config, ProvenancePolicy};
use server_lib::promote::{PromotedArtifact, PromotionContext};
let fx = fixtures()?;
let mut m = base_metrics(fx);
m.insert("source_gates".into(), json!(1));
let b = budget(scale);
source_gates(&mut m, b, &fx.packages)?;
let db = advisories().map(|a| a.db.clone()).unwrap_or_default();
m.insert("advisories".into(), json!(db.advisories.len()));
for (label, db) in [
("real_db", db),
("empty_db", server_lib::security::AdvisoryDb::default()),
] {
let cat = catalog_from_config(
db,
license_policy(),
Some("https://slsa.dev/provenance/v1".into()),
ProvenancePolicy::Warn,
0,
);
let names: Vec<String> = cat.available().iter().map(|s| s.to_string()).collect();
m.insert("gates".into(), json!(names.join(",")));
let gates = cat.resolve(&names).map_err(|e| anyhow!("{e}"))?;
for p in &fx.packages {
let id = ArtifactId {
namespace: None,
name: p.name.clone(),
version: p.version.clone(),
};
let set = [PromotedArtifact {
id: &id,
ecosystem: &p.ecosystem,
digest: ArtifactDigest::of_sha256(&p.bytes),
bytes: &p.bytes,
attestations: &[],
}];
let ctx = PromotionContext {
principal: "bench",
from_repo: "approved",
to_repo: "production",
hop: None,
set: &set,
};
let d = gates.evaluate(&ctx);
m.insert(
format!("policy_{}_{}_summary", label, p.key),
json!(d.summary()),
);
let lat = sample(3, b, || {
std::hint::black_box(gates.evaluate(&ctx));
Ok(())
})?;
put_lat(&mut m, &format!("policy_{label}_{}", p.key), lat, None);
}
}
if let Some(v) = m.get("policy_real_db_wheel_s_ms").and_then(Value::as_f64) {
m.insert(
"ops_per_sec".into(),
json!(if v > 0.0 { 1e3 / v } else { 0.0 }),
);
}
Ok(m)
}
/// The RON `rules` floor and one source's `gates` (lane G0, `sources.rs`): the
/// effective policy an artifact gets (floor, then the source's tightening), and
/// what that policy refuses at SCANNED from the package's real scan records.
fn source_gates(m: &mut Metrics, b: Duration, packages: &[Package]) -> Result<()> {
use server_lib::scanner::license::LicenseScanner;
use server_lib::scanner::{FindingSeverity, ScannerSet};
use server_lib::sources::{CveGate, EffectivePolicy, Gates, LicenseGate, Rules};
let floor = Rules {
require_signed: false,
require_provenance: false,
deny_licenses: vec!["AGPL-3.0-only".into()],
..Rules::default()
};
let gates = Gates {
license: Some(LicenseGate {
deny: vec!["GPL-3.0-only".into()],
}),
cve: Some(CveGate {
max: FindingSeverity::Medium,
}),
..Gates::default()
};
let lat = sample(3, b, || {
std::hint::black_box(EffectivePolicy::floor("bench", &floor).tighten(
&gates,
"source 'bench'",
&floor,
)?);
Ok(())
})?;
put_lat(m, "sources_effective_policy", lat, None);
let policy =
EffectivePolicy::floor("bench", &floor).tighten(&gates, "source 'bench'", &floor)?;
m.insert("sources_policy".into(), json!(policy.summary()));
let db = advisories().map(|a| a.db.clone()).unwrap_or_default();
let set = ScannerSet::new()
.with(Arc::new(server_lib::scanner::osv::OsvVulnScanner::new(db)))
.with(Arc::new(LicenseScanner::new(license_policy())))
.with(Arc::new(server_lib::scanner::wheel::WheelSbomScanner));
let lic = LicenseScanner::new(license_policy());
for p in packages {
let subject =
ScanSubject::new(&p.file, &p.ecosystem, &p.bytes).with_coords(&p.name, &p.version);
let ev = set.scan(&subject);
let judged = lic.judge(&p.ecosystem, &p.bytes);
let refused = policy.refusals(&ev.records, Some(&judged));
m.insert(format!("sources_{}_refusals", p.key), json!(refused.len()));
let lat = sample(3, b, || {
std::hint::black_box(policy.refusals(&ev.records, Some(&judged)));
Ok(())
})?;
put_lat(m, &format!("sources_{}", p.key), lat, None);
}
Ok(())
}
// ───────────────────────────────── seal ────────────────────────────────
/// Seal into the production store: blake3 content address, blob, create-once ref.
pub fn seal_arm(scale: &Scale) -> Result<Metrics> {
use server_lib::promote::PromotionEngine;
let fx = fixtures()?;
let mut m = base_metrics(fx);
let dir = tempfile::tempdir()?;
let engine = PromotionEngine::new(dir.path());
let b = budget(scale);
// Every seal gets bytes of its own (`variant`): the store is content-addressed,
// so sealing the same bytes twice only writes a ref and would time nothing.
let mut n = 0u64;
for p in &fx.packages {
let mut lat = Vec::new();
let start = Instant::now();
while lat.len() < 3 || (start.elapsed() < b && lat.len() < 2000) {
n += 1;
let bytes = variant(&p.bytes, 7_000_000 + n);
let id = ArtifactId {
namespace: None,
name: format!("seal-{n}/{}", p.file),
version: String::new(),
};
let t = Instant::now();
engine
.seal(
"approved",
"production",
&id,
&bytes,
"bench@holger",
1_700_000_000 + n as i64,
)
.map_err(|e| anyhow!("seal: {e}"))?;
lat.push(t.elapsed());
}
put_lat(&mut m, &format!("seal_{}", p.key), lat, Some(p.bytes.len()));
}
if let Some(v) = m.get("seal_wheel_s_ms").and_then(Value::as_f64) {
m.insert(
"ops_per_sec".into(),
json!(if v > 0.0 { 1e3 / v } else { 0.0 }),
);
}
Ok(m)
}
// ────────────────────────────── end to end ─────────────────────────────
/// The same package with different bytes, still a valid archive: a zip gains an
/// end-of-central-directory comment, a gzip stream a different MTIME. Distinct
/// bytes mean distinct digests, so no scan evidence is reused between copies.
pub fn variant(bytes: &[u8], i: u64) -> Vec<u8> {
let mut v = bytes.to_vec();
if v.starts_with(&[0x1f, 0x8b]) && v.len() > 8 {
v[4..8].copy_from_slice(&(i as u32 ^ 0x5a5a_0000).to_le_bytes());
} else if v.starts_with(b"PK") && v.len() >= 22 {
let eocd = v.len() - 22;
if v[eocd..eocd + 4] == [0x50, 0x4b, 0x05, 0x06] && v[eocd + 20..eocd + 22] == [0, 0] {
v[eocd + 20..eocd + 22].copy_from_slice(&8u16.to_le_bytes());
v.extend_from_slice(&i.to_le_bytes());
}
} else {
v.extend_from_slice(&i.to_le_bytes());
}
v
}
/// A full intake chain, wired the way `Holger::intake_scan_pipeline` wires it:
/// the inspector at ingest, then SCANNED with inspection, YARA (and ClamAV when
/// `HOLGER_BENCH_CLAMD` names one), OSV over the real advisories, the licence
/// scanner, the wheel SBOM and the model scanner; POLICY_CHECKED with the gate
/// catalog; one approval; SEAL; PROMOTED; AVAILABLE. A file audit log, as the
/// server writes.
pub struct Chain {
pub _dir: tempfile::TempDir,
pub chain: server_lib::promote::intake::IntakeChain,
pub incoming: Arc<dyn RepositoryBackendTrait>,
}
/// `worker`: `None` runs every scanner in this process (`sandbox: off`, a bench
/// build is a test build); `Some` wires the chain as the server does with the
/// scan sandbox: the INGEST walk and the in-worker scanners in the container,
/// the vulnerability match from the worker's report, ClamAV streamed to clamd.
pub fn build_chain(
root: Option<&Path>,
worker: Option<Arc<server_lib::sandbox::ScanWorker>>,
) -> Result<Chain> {
use holger_inspect::malware::{ClamAvConfig, YaraConfig};
use server_lib::promote::gate::{catalog_from_config, ProvenancePolicy};
use server_lib::promote::intake::{
ArtifactState, IntakeChain, IntakeScanSettings, IntakeSettings, ResolvedPolicy,
};
use server_lib::promote::{PromotionEngine, Stage, StageLadder};
use server_lib::scanner::{EvidenceStore, ScanPipeline, ScannerKind, ScannerSet};
let dir = match root {
Some(r) => tempfile::tempdir_in(r)?,
None => tempfile::tempdir()?,
};
let rules = dir.path().join("bench.yar");
std::fs::write(&rules, YARA_RULES)?;
let inspect = holger_inspect::InspectConfig {
clamav: clamd().map(|s| ClamAvConfig {
socket: Some(s),
required: true,
..ClamAvConfig::default()
}),
yara: Some(YaraConfig {
rules: Some(rules.display().to_string()),
binary: None,
required: true,
}),
..Default::default()
};
let mut required = vec![
ScannerKind::Inspection,
ScannerKind::Malware,
ScannerKind::Vulnerability,
ScannerKind::License,
];
required.sort();
let settings = IntakeSettings {
incoming: "incoming".into(),
quarantine: "quarantine".into(),
approved: "approved".into(),
production: "production".into(),
transitions: vec![],
inspect: inspect.clone(),
secrets: None,
// The arms time the scanners' code in this process (a bench build is a test build).
rules: server_lib::sandbox::IntakeRules {
sandbox: if worker.is_some() {
server_lib::sandbox::SandboxRule::Required
} else {
server_lib::sandbox::SandboxRule::Off
},
},
sandbox: None,
scan: Some(IntakeScanSettings {
required: required.clone(),
provenance: None,
crate_lock_advisories: Default::default(),
}),
approvals: 1,
trust_policy: None,
npm: None,
oci_push: Vec::new(),
unscanned: Vec::new(),
cargo: None,
bundles: None,
bundles_trust_policy: None,
};
let audit: Arc<dyn server_lib::audit::AuditLog> = Arc::new(
server_lib::audit::FileAuditLog::new(dir.path().join("audit"))
.map_err(|e| anyhow!("audit: {e}"))?,
);
let incoming: Arc<dyn RepositoryBackendTrait> = Arc::new(server_lib::raw::RawBackend::new(
"incoming".into(),
dir.path().join("incoming"),
));
let prod_root = dir.path().join("prod");
let ladder = StageLadder::new(
settings
.repos()
.iter()
.map(|r| Stage::new(*r, *r))
.collect(),
);
let engine = PromotionEngine::new(&prod_root)
.with_served_repositories(settings.repos())
.with_ladder("intake", ladder);
let db = advisories().map(|a| a.db.clone()).unwrap_or_default();
let vuln = Arc::new(server_lib::scanner::osv::OsvVulnScanner::new(db.clone()));
let mut set = match &worker {
None => {
let set = ScannerSet::new()
.with(vuln)
.with(Arc::new(server_lib::scanner::wheel::WheelSbomScanner))
.with(Arc::new(server_lib::scanner::model::ModelScanner::new()))
.with(Arc::new(server_lib::scanner::license::LicenseScanner::new(
license_policy(),
)));
server_lib::scanner::inspect::install(set, &inspect)
}
Some(w) => {
// As `Holger::intake_scan_pipeline` wires it with the sandbox.
let mut set = ScannerSet::new().with(Arc::new(
server_lib::sandbox::VulnFromWorker::new(vuln, Arc::clone(w)),
));
for sc in server_lib::sandbox::WorkerScanner::all(w) {
set = set.with(sc);
}
if let Some(c) = &inspect.clamav {
let clam = holger_inspect::malware::ClamAv::from_config(c);
set = set.with(Arc::new(server_lib::sandbox::ClamStream::new(Arc::new(
clam,
))));
}
set
}
};
for k in &required {
set = set.require(*k);
}
let pipeline = Arc::new(ScanPipeline::new(
set,
EvidenceStore::new(dir.path().join("evidence"), audit.clone()),
));
let cat = catalog_from_config(
db,
license_policy(),
Some("https://slsa.dev/provenance/v1".into()),
ProvenancePolicy::Warn,
0,
);
// `attested` is left out: a package fetched from PyPI, npm or crates.io carries
// no SLSA statement in this box's evidence plane, so that gate refuses every
// one of them (the `policy` arm still times it).
let names: Vec<String> = cat
.available()
.iter()
.filter(|n| **n != "attested")
.map(|s| s.to_string())
.collect();
let gates = cat.resolve(&names).map_err(|e| anyhow!("{e}"))?;
let mut chain = IntakeChain::new(settings, engine, &prod_root, audit, incoming.clone())
.with_scanner(pipeline)
.with_policy(
ArtifactState::PolicyChecked,
ResolvedPolicy {
gates,
require_evidence: vec![],
},
);
if let Some(w) = worker {
chain = chain.with_worker(w);
}
Ok(Chain {
_dir: dir,
chain,
incoming,
})
}
/// Where one package ended.
#[derive(Debug, Clone)]
pub struct Walk {
pub latency: Duration,
pub reached: String,
pub refused: Option<String>,
}
/// Walk one artifact from UNTRUSTED to AVAILABLE (or to the state that refused it).
pub fn walk_one(c: &Chain, name: &str, bytes: &[u8]) -> Walk {
use server_lib::promote::intake::{ArtifactState::*, TransitionRequest};
let t = Instant::now();
let id = ArtifactId {
namespace: None,
name: name.to_string(),
version: String::new(),
};
let inc = c.incoming.clone();
let put = id.clone();
if let Err(e) = c.chain.ingest(
"importer",
"127.0.0.1:1",
"incoming",
&id,
bytes,
move || inc.put(&put, bytes),
) {
return Walk {
latency: t.elapsed(),
reached: "UNTRUSTED".into(),
refused: Some(e.tagged()),
};
}
let mut reached = "INGESTED".to_string();
for to in [
Scanned,
PolicyChecked,
Approved,
Sealed,
Promoted,
Available,
] {
let actor = if to == Approved {
"approver"
} else {
"officer"
};
match c
.chain
.advance(TransitionRequest::new(id.clone(), to, actor))
{
Ok(_) => reached = to.name().to_string(),
Err(e) => {
return Walk {
latency: t.elapsed(),
reached,
refused: Some(e.tagged()),
}
}
}
}
Walk {
latency: t.elapsed(),
reached,
refused: None,
}
}
fn e2e_count(scale: &Scale) -> usize {
std::env::var("HOLGER_BENCH_E2E_N")
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(if scale.bench_secs <= 2 { 64 } else { 512 })
}
/// One package end to end per type, then N packages in parallel: packages per
/// second, p50/p99 latency and peak RSS.
pub fn e2e_arm(scale: &Scale) -> Result<Metrics> {
let fx = fixtures()?;
let mut m = base_metrics(fx);
let _ = advisories(); // load outside every clock
let c = build_chain(None, None)?;
let mut serial = 0u64;
for p in &fx.packages {
let mut lats = Vec::new();
let mut last = None;
for _ in 0..3 {
serial += 1;
let w = walk_one(
&c,
&format!("one-{serial}/{}", p.file),
&variant(&p.bytes, serial),
);
lats.push(w.latency);
last = Some(w);
}
let w = last.unwrap();
put_lat(&mut m, &format!("e2e_{}", p.key), lats, Some(p.bytes.len()));
m.insert(format!("e2e_{}_reached", p.key), json!(w.reached));
if let Some(r) = &w.refused {
m.insert(format!("e2e_{}_refused", p.key), json!(r));
}
}
// ANSWER floor: the small wheel walks the whole chain.
if fx.packages.iter().any(|p| p.key == "wheel_s") {
ensure!(
m.get("e2e_wheel_s_reached").and_then(Value::as_str) == Some("AVAILABLE"),
"e2e: requests did not reach AVAILABLE: {:?}",
m.get("e2e_wheel_s_refused")
);
}
// N in parallel, the package types mixed round-robin.
let n = e2e_count(scale);
let corpus: Vec<(String, Vec<u8>)> = (0..n)
.map(|i| {
let p = &fx.packages[i % fx.packages.len()];
(
format!("par-{i}/{}", p.file),
variant(&p.bytes, 1_000_000 + i as u64),
)
})
.collect();
let bytes: usize = corpus.iter().map(|(_, b)| b.len()).sum();
let par = build_chain(None, None)?;
reset_peak_rss();
let rss0 = peak_rss_mb();
let t = Instant::now();
let walks = gatling::gatling_forkjoin::gatling_for_each(n, 0, |i| {
walk_one(&par, &corpus[i].0, &corpus[i].1)
});
let elapsed = t.elapsed().as_secs_f64().max(f64::MIN_POSITIVE);
let rss1 = peak_rss_mb();
let available = walks.iter().filter(|w| w.reached == "AVAILABLE").count();
let mut refusals: BTreeMap<String, usize> = BTreeMap::new();
for w in walks.iter().filter(|w| w.refused.is_some()) {
*refusals.entry(w.reached.clone()).or_default() += 1;
}
let mut lat: Vec<Duration> = walks.iter().map(|w| w.latency).collect();
m.insert("par_packages".into(), json!(n));
m.insert("par_workers".into(), json!(crate::cores()));
m.insert("par_available".into(), json!(available));
m.insert("par_refused_at".into(), json!(refusals));
m.insert("par_packages_per_sec".into(), json!(n as f64 / elapsed));
m.insert(
"par_mb_per_sec".into(),
json!(bytes as f64 / 1_048_576.0 / elapsed),
);
m.insert("par_p50_ms".into(), json!(pct(&mut lat, 0.50)));
m.insert("par_p99_ms".into(), json!(pct(&mut lat, 0.99)));
m.insert("par_seconds".into(), json!(elapsed));
m.insert("par_peak_rss_mb".into(), json!(rss1));
m.insert("par_rss_before_mb".into(), json!(rss0));
m.insert("ops_per_sec".into(), json!(n as f64 / elapsed));
Ok(m)
}
// ─────────────────────────────── fetch ─────────────────────────────────
/// Drop a file's pages from the page cache, so the next read goes to the disk.
fn drop_cache(path: &Path) {
#[cfg(unix)]
{
use std::os::unix::io::AsRawFd;
if let Ok(f) = std::fs::File::open(path) {
let _ = f.sync_all();
// SAFETY: a valid fd for the whole call; advice only.
unsafe {
libc::posix_fadvise(f.as_raw_fd(), 0, 0, libc::POSIX_FADV_DONTNEED);
}
}
}
}
fn drop_cache_tree(dir: &Path) {
if let Ok(rd) = std::fs::read_dir(dir) {
for e in rd.flatten() {
let p = e.path();
if p.is_dir() {
drop_cache_tree(&p);
} else {
drop_cache(&p);
}
}
}
}
/// Every real package stored in one znippy archive on the disk under
/// `HOLGER_BENCH_SLOW_DIR` (a slow device; default: the temp dir), then fetched
/// cold (page cache dropped, archive re-opened) and warm.
pub fn fetch_arm(scale: &Scale) -> Result<Metrics> {
use repository_znippy_rust::RustRepoZnippy;
let fx = fixtures()?;
let mut m = base_metrics(fx);
let root = std::env::var_os("HOLGER_BENCH_SLOW_DIR").map(PathBuf::from);
m.insert(
"disk".into(),
json!(std::env::var("HOLGER_BENCH_DISK").unwrap_or_else(|_| "default temp dir".into())),
);
let dir = match &root {
Some(r) => tempfile::tempdir_in(r)?,
None => tempfile::tempdir()?,
};
let archive = dir.path().join("fetch.znippy");
let w = RustRepoZnippy::with_writable_archive("fetch".into(), archive.clone());
let ids: Vec<(ArtifactId, &Package)> = fx
.packages
.iter()
.map(|p| {
(
ArtifactId {
namespace: None,
name: p.name.replace('/', "-"),
version: format!("{}-{}", p.version.len(), p.key),
},
p,
)
})
.collect();
let t = Instant::now();
for (id, p) in &ids {
w.put(id, &p.bytes)
.with_context(|| format!("fetch: put {}", p.key))?;
}
m.insert("store_secs".into(), json!(t.elapsed().as_secs_f64()));
drop(w);
let rounds = if scale.bench_secs <= 2 { 5 } else { 15 };
for (id, p) in &ids {
let mut cold = Vec::new();
let mut warm = Vec::new();
for _ in 0..rounds {
drop_cache_tree(dir.path());
let t = Instant::now();
let r = RustRepoZnippy::with_writable_archive("fetch".into(), archive.clone());
let got = r
.fetch(id)?
.ok_or_else(|| anyhow!("fetch: {} is not in the archive", p.key))?;
cold.push(t.elapsed());
ensure!(
got.len() == p.bytes.len() && got[..] == p.bytes[..],
"fetch: {} came back different",
p.key
);
let t = Instant::now();
std::hint::black_box(r.fetch(id)?);
warm.push(t.elapsed());
}
put_lat(
&mut m,
&format!("cold_{}", p.key),
cold,
Some(p.bytes.len()),
);
put_lat(
&mut m,
&format!("warm_{}", p.key),
warm,
Some(p.bytes.len()),
);
}
m.insert(
"archive_bytes".into(),
json!(std::fs::metadata(&archive).map(|x| x.len()).unwrap_or(0)),
);
if let Some(v) = m.get("cold_wheel_s_ms").and_then(Value::as_f64) {
m.insert(
"ops_per_sec".into(),
json!(if v > 0.0 { 1e3 / v } else { 0.0 }),
);
}
Ok(m)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn the_merkle_proof_reaches_the_root_of_a_small_tree() {
// A 5-leaf tree built by hand, RFC 9162 shape.
let leaves: Vec<Vec<u8>> = (0..5u8)
.map(|i| {
let mut v = vec![0u8];
v.push(i);
sha256(&v)
})
.collect();
let n01 = node(&leaves[0], &leaves[1]);
let n23 = node(&leaves[2], &leaves[3]);
let n0123 = node(&n01, &n23);
let root = node(&n0123, &leaves[4]);
assert_eq!(
merkle_root_from_proof(4, 5, &leaves[4], &[n0123.clone()]),
Some(root.clone())
);
assert_eq!(
merkle_root_from_proof(
2,
5,
&leaves[2],
&[leaves[3].clone(), n01.clone(), leaves[4].clone()]
),
Some(root.clone())
);
assert_ne!(
merkle_root_from_proof(
2,
5,
&leaves[3],
&[leaves[3].clone(), n01, leaves[4].clone()]
),
Some(root)
);
}
#[test]
fn a_variant_is_a_different_digest_and_still_the_same_archive() {
let z = zip_of(&[("a/b.txt", b"hello")], true).unwrap();
let v = variant(&z, 7);
assert_ne!(sha256(&z), sha256(&v));
let mut a = zip::ZipArchive::new(std::io::Cursor::new(v)).unwrap();
assert_eq!(a.len(), 1);
assert_eq!(a.by_index(0).unwrap().name().unwrap(), "a/b.txt");
let mut gz = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::fast());
std::io::Write::write_all(&mut gz, b"payload").unwrap();
let g = gz.finish().unwrap();
let gv = variant(&g, 9);
assert_ne!(g, gv);
let mut out = String::new();
std::io::Read::read_to_string(&mut flate2::read::GzDecoder::new(&gv[..]), &mut out)
.unwrap();
assert_eq!(out, "payload");
}
/// RFC 4648 base64url without padding (a wheel RECORD's digest form).
fn b64url(b: &[u8]) -> String {
const A: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_";
let mut o = String::new();
for c in b.chunks(3) {
let n = c
.iter()
.enumerate()
.fold(0u32, |n, (i, x)| n | (u32::from(*x) << (16 - 8 * i)));
for i in 0..=c.len() {
o.push(A[((n >> (18 - 6 * i)) & 63) as usize] as char);
}
}
o
}
#[test]
fn the_wheel_made_from_its_members_is_the_one_its_record_names_and_always_the_same() {
let dir = committed_dir().join("requests-2.32.3-py3-none-any");
let w = wheel_from_members(&dir).unwrap();
assert_eq!(
w,
wheel_from_members(&dir).unwrap(),
"the same members make the same bytes"
);
let mut z = zip::ZipArchive::new(std::io::Cursor::new(&w[..])).unwrap();
let names: Vec<String> = z.file_names().map(|n| n.unwrap().into_owned()).collect();
assert_eq!(
names.last().map(String::as_str),
Some("requests-2.32.3.dist-info/RECORD")
);
let mut record = String::new();
std::io::Read::read_to_string(
&mut z.by_name("requests-2.32.3.dist-info/RECORD").unwrap(),
&mut record,
)
.unwrap();
let mut checked = 0;
for line in record.lines() {
let mut f = line.split(',');
let (name, hash) = (f.next().unwrap(), f.next().unwrap_or(""));
let Some(want) = hash.strip_prefix("sha256=") else {
continue;
};
let mut body = Vec::new();
std::io::Read::read_to_end(
&mut z.by_name(name).unwrap_or_else(|e| panic!("{name}: {e}")),
&mut body,
)
.unwrap();
let hex = sha256_hex(&body);
let raw: Vec<u8> = (0..hex.len())
.step_by(2)
.map(|i| u8::from_str_radix(&hex[i..i + 2], 16).unwrap())
.collect();
assert_eq!(
b64url(&raw),
want,
"{name} is not the member upstream's RECORD names"
);
checked += 1;
}
assert_eq!(
checked,
names.len() - 1,
"every member but RECORD is named with its sha256"
);
}
#[test]
fn the_committed_fixtures_load_and_match_their_pins() {
let fx = load().unwrap();
for k in ["wheel_s", "npm_s", "crate_s"] {
assert!(
fx.packages.iter().any(|p| p.key == k),
"{k} is committed and must load"
);
}
}
#[test]
fn the_bench_rule_set_runs_on_the_built_in_matcher_and_finds_eicar() {
use holger_inspect::malware::Yara;
let y = Yara::from_source(YARA_RULES, true);
assert_eq!(y.engine_name().unwrap(), "built-in");
let insp = inspector().with_scanner(Arc::new(y));
let z = zip_of(&[("evil/__init__.py", &eicar())], true).unwrap();
let f = insp.inspect(
&z,
holger_inspect::Format::Wheel,
"evil-1.0-py3-none-any.whl",
);
assert!(
f.has(
holger_inspect::FindingKind::Malware,
holger_inspect::Severity::Block
),
"{}",
f.summary()
);
}
#[test]
fn the_chainguard_rekor_proof_and_checkpoint_verify_offline() {
let fx = load().unwrap();
let pdir = fx.committed.join("chainguard-python");
let rekor: Value =
serde_json::from_slice(&std::fs::read(pdir.join("rekor-3416929314.json")).unwrap())
.unwrap();
let p = rekor_proof(&rekor).unwrap();
assert_eq!(
merkle_root_from_proof(p.index, p.size, &p.leaf, &p.hashes),
Some(p.root.clone())
);
let key = pem_blocks(&std::fs::read_to_string(pdir.join("rekor-pubkey.pem")).unwrap());
assert_eq!(verify_checkpoint(&p.checkpoint, &key[0]).unwrap(), p.root);
let mut bad = p.checkpoint.clone();
bad = bad.replacen(&p.size.to_string(), &(p.size + 1).to_string(), 1);
assert!(
verify_checkpoint(&bad, &key[0]).is_err(),
"a changed tree size must not verify"
);
}
}