server/lib/src/scanner/provenance.rs
//! The PROVENANCE scanner (lane G3a): `holger_provenance` behind the [`Scanner`] trait.
//!
//! It fills the [`ScannerKind::Provenance`] slot at the SCANNED transition, so
//! its evidence is what the POLICY_CHECKED step and the gates read. Per artifact
//! it takes the effective policy (the `rules` floor plus the artifact's source
//! gate, [`holger_provenance::effective`]), reads the provenance material that
//! travelled with the bytes from the [`ProvenanceInbox`], verifies it OFFLINE and
//! answers:
//!
//! * `require` and nothing verifies: `block`, the refusal named
//! (`holger-provenance-<n>`);
//! * `prefer` and nothing verifies: `warn`, recorded;
//! * `off`: `pass`, and the record says nothing was checked.
//!
//! The whole [`holger_provenance::ProvenanceReport`] (every candidate, what was
//! and was not checked, the SLSA builder, build type, invocation and resolved
//! dependencies) is the record's document, so it is in the per-artifact
//! evidence, the audit trail and GetArtifactEvidence.
use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use crate::lock::RwLockOrRecover as _;
use holger_provenance::{
check_floor, check_tightens, effective, verify_artifact, ProvenanceFloor, ProvenanceGate,
ProvenanceReport, TrustRoot, Verdict,
};
use serde::{Deserialize, Serialize};
use super::{FindingSeverity, ScanFinding, ScanOutcome, ScanSubject, Scanner, ScannerKind};
/// The scanner id in the evidence. Stable: lane G1 and the gates key on it.
pub const SCANNER_ID: &str = holger_provenance::SCANNER_ID;
/// The `provenance:` key inside `promotion.intake.scan`, which the `sources:` /
/// `rules:` loader (lane G0+G2) lowers into.
///
/// ```ron
/// scan: Some((
/// required: [provenance],
/// provenance: Some((
/// rules: (require_signed: true, identities: [...], trust_root: Some("…")),
/// sources: {
/// "chainguard": require(identity: (issuer: "https://token.actions.githubusercontent.com",
/// subject: "https://github.com/chainguard-images/images/.github/workflows/release.yaml@refs/heads/main"),
/// slsa_min: 2),
/// "pypi": prefer,
/// },
/// )),
/// )),
/// ```
#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ProvenanceScanSettings {
/// The global floor (the provenance fields of `rules:`).
#[serde(default)]
pub rules: ProvenanceFloor,
/// Each source's (or repository's) `gates.provenance`, by name.
#[serde(default)]
pub sources: BTreeMap<String, ProvenanceGate>,
/// The gate for an artifact whose source names no gate. Default `off` (the floor holds).
#[serde(default)]
pub default: ProvenanceGate,
/// Where the provenance material for an artifact waits, by its sha256.
/// Default `<data_root>/holger-provenance`.
#[serde(default)]
pub inbox: Option<String>,
/// The `kind: apk` sources (lane G3): an artifact from one of them also gets the
/// source's signed APK indexes as material, `<inbox>/apk/<source>/*`
/// (`APKINDEX.tar.gz`, one per repository and arch).
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub apk_sources: Vec<String>,
}
/// Where an artifact's provenance material comes from.
pub trait MaterialSource: Send + Sync {
/// Every file of material about the bytes with this sha256, as `(name, bytes)`.
fn materials(
&self,
sha256_hex: &str,
file_name: &str,
) -> anyhow::Result<Vec<(String, Vec<u8>)>>;
/// The material every artifact of an APK source shares: its signed indexes.
fn apk_indexes(&self, _source: &str) -> anyhow::Result<Vec<(String, Vec<u8>)>> {
Ok(Vec::new())
}
}
/// Every regular file directly in `dir`, by name (none when it does not exist).
fn files_in(dir: &Path) -> anyhow::Result<Vec<(String, Vec<u8>)>> {
let rd = match std::fs::read_dir(dir) {
Ok(rd) => rd,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
Err(e) => return Err(anyhow::anyhow!("reading {}: {e}", dir.display())),
};
let mut out = Vec::new();
for e in rd.flatten() {
let name = e.file_name().to_string_lossy().into_owned();
if name.starts_with('.') || !e.path().is_file() {
continue;
}
out.push((name, std::fs::read(e.path())?));
}
out.sort_by(|a, b| a.0.cmp(&b.0));
Ok(out)
}
/// ★ The provenance inbox: `<root>/<sha256>/<file>`. The online side (or the
/// importer, or the demo) drops the attestations that came with an artifact
/// here, under the artifact's sha256: a `.intoto.jsonl`, a Sigstore bundle, a
/// PEP 740 provenance, npm's attestations, cosign's attachments.
#[derive(Debug, Clone)]
pub struct ProvenanceInbox {
root: PathBuf,
}
impl ProvenanceInbox {
pub fn new(root: impl Into<PathBuf>) -> Self {
Self { root: root.into() }
}
pub fn root(&self) -> &Path {
&self.root
}
fn dir(&self, sha256_hex: &str) -> anyhow::Result<PathBuf> {
if sha256_hex.len() != 64
|| !sha256_hex
.bytes()
.all(|b| b.is_ascii_hexdigit() && !b.is_ascii_uppercase())
{
anyhow::bail!("{sha256_hex:?} is not a lowercase sha256");
}
Ok(self.root.join(sha256_hex))
}
/// File one piece of material about `sha256_hex` under `name`.
pub fn put(&self, sha256_hex: &str, name: &str, bytes: &[u8]) -> anyhow::Result<PathBuf> {
let safe = !name.is_empty()
&& name.len() <= 200
&& !name.starts_with('.')
&& name
.bytes()
.all(|b| b.is_ascii_alphanumeric() || b"._-@+".contains(&b));
if !safe {
anyhow::bail!("{name:?} is not a safe material file name");
}
let dir = self.dir(sha256_hex)?;
std::fs::create_dir_all(&dir)?;
let path = dir.join(name);
let tmp = dir.join(format!(".{name}.tmp"));
std::fs::write(&tmp, bytes)?;
std::fs::rename(&tmp, &path)?;
Ok(path)
}
}
impl MaterialSource for ProvenanceInbox {
fn materials(
&self,
sha256_hex: &str,
_file_name: &str,
) -> anyhow::Result<Vec<(String, Vec<u8>)>> {
files_in(&self.dir(sha256_hex)?)
}
fn apk_indexes(&self, source: &str) -> anyhow::Result<Vec<(String, Vec<u8>)>> {
if source.is_empty() || source.contains('/') || source.starts_with('.') {
anyhow::bail!("{source:?} is not a source name");
}
files_in(&self.apk_dir(source))
}
}
impl ProvenanceInbox {
/// Where an APK source's signed indexes wait: `<root>/apk/<source>/`.
pub fn apk_dir(&self, source: &str) -> PathBuf {
self.root.join("apk").join(source)
}
}
/// **Where `verify_artifact` runs.** Unset, in this process (a test build); set,
/// in the scan sandbox (`crate::sandbox`): the DSSE, Sigstore, DER/X.509 and
/// Rekor parsing of untrusted material never runs in the server.
pub trait RemoteVerify: Send + Sync {
fn verify(
&self,
artifact: &[u8],
materials: &[(String, Vec<u8>)],
policy: &holger_provenance::EffectivePolicy,
) -> ProvenanceReport;
}
/// ★ The provenance scanner.
pub struct ProvenanceScanner {
settings: ProvenanceScanSettings,
materials: Arc<dyn MaterialSource>,
/// The trust root, the version that names it, and the sandboxed verifier that holds it:
/// replaced whole when a signed refresh of the root is imported (lane HOL-OCI-EVIDENCE).
loaded: std::sync::RwLock<Loaded>,
}
struct Loaded {
trust: Option<Arc<TrustRoot>>,
version: String,
remote: Option<Arc<dyn RemoteVerify>>,
}
/// The scanner's version: its policy and the trust root it reads (stored evidence made
/// under another root is stale).
fn version_of(settings: &ProvenanceScanSettings, trust: Option<&Arc<TrustRoot>>) -> String {
let policy = serde_json::to_vec(settings).unwrap_or_default();
format!(
"holger-server-lib {} over holger-provenance; policy blake3:{}; trust root {}",
env!("CARGO_PKG_VERSION"),
&blake3::hash(&policy).to_hex()[..16],
trust
.map(|t| {
// A root that pins only apk keys has no trusted_root.json digest.
let root = t
.digest
.get(..16)
.map(|d| format!("sha256:{d}"))
.unwrap_or_else(|| "apk keys only".into());
let apk: Vec<&str> = t.apk_keys.iter().map(|k| k.name.as_str()).collect();
let pgp: Vec<String> = t
.pgp_keys
.iter()
.map(|k| format!("{} {}", k.name, k.fingerprint))
.collect();
let root = if apk.is_empty() {
root
} else {
format!("{root}; apk keys {apk:?}")
};
if pgp.is_empty() {
root
} else {
format!("{root}; pgp keys {pgp:?}")
}
})
.unwrap_or_else(|| "none".into()),
)
}
impl ProvenanceScanner {
/// Build it. Every source gate is checked against the floor first: a gate
/// that loosens the floor refuses the boot by name (`holger-provenance-11`).
pub fn new(
settings: ProvenanceScanSettings,
trust: Option<Arc<TrustRoot>>,
materials: Arc<dyn MaterialSource>,
) -> anyhow::Result<Self> {
// The default gate unset (`off`) means "the floor holds"; a NAMED gate is
// what may only tighten it.
let default = (settings.default != ProvenanceGate::Off)
.then(|| ("<default>".to_string(), settings.default.clone()));
// The floor on its own, so a pin it holds is read even with no source (T9b: a
// key of `alg: openpgp` in a build without `maven-pgp` is refused here by name).
check_floor(&settings.rules).map_err(|e| {
anyhow::anyhow!("config `promotion.intake.scan.provenance`: {}", e.tagged())
})?;
for (name, gate) in settings
.sources
.iter()
.map(|(n, g)| (n.clone(), g.clone()))
.chain(default)
{
check_tightens(&settings.rules, &gate, &name).map_err(|e| {
anyhow::anyhow!("config `promotion.intake.scan.provenance`: {}", e.tagged())
})?;
}
let version = version_of(&settings, trust.as_ref());
Ok(Self {
settings,
materials,
loaded: std::sync::RwLock::new(Loaded {
trust,
version,
remote: None,
}),
})
}
/// Load the trust root a floor names, verified as a signed `holger.sigstore-trust/v1`
/// bundle against `bundle_policy` (the intake's bundle trust policy), at the estate's time.
#[cfg(feature = "import-door")]
pub fn load_trust_root(
floor: &ProvenanceFloor,
bundle_policy: Option<&str>,
) -> anyhow::Result<Option<Arc<TrustRoot>>> {
match Self::load_trust_root_json(floor, bundle_policy)? {
None => Ok(None),
Some(json) => {
let apk = Self::load_apk_keys(floor, bundle_policy)?;
let pgp = Self::load_pgp_keys(floor, bundle_policy)?;
let root = TrustRoot::from_trusted_root_json(&json)
.and_then(|r| r.with_apk_keys(&apk))
.and_then(|r| r.with_pgp_keys(&pgp))
.map_err(|e| anyhow::anyhow!("config `rules.trust_root`: {}", e.tagged()))?;
Ok(Some(Arc::new(root)))
}
}
}
/// The verified `trusted_root.json` the floor's trust bundle carries (the
/// same checks as [`load_trust_root`](Self::load_trust_root)); the scan
/// sandbox hands these bytes to the worker.
#[cfg(feature = "import-door")]
pub fn load_trust_root_json(
floor: &ProvenanceFloor,
bundle_policy: Option<&str>,
) -> anyhow::Result<Option<Vec<u8>>> {
let Some(path) = floor.trust_root.as_deref().filter(|p| !p.trim().is_empty()) else {
return Ok(None);
};
if empty_slot(path) {
log::warn!(
"provenance: the Sigstore trust root slot {path} is empty (this medium seeded none): keyless signatures verify \
against nothing until a medium seeds it or a signed holger.sigstore-trust/v1 bundle is imported"
);
return Ok(None);
}
let Some(policy_path) = bundle_policy else {
anyhow::bail!(
"config `rules.trust_root: {path:?}` names a Sigstore trust bundle, and `promotion.intake.trust_policy` \
names no bundle trust policy to verify it against; an unverified trust root is not trusted"
);
};
let text = std::fs::read_to_string(policy_path)
.map_err(|e| anyhow::anyhow!("trust policy {policy_path}: {e}"))?;
let policy = update_bundle::TrustPolicy::from_ron(&text)
.map_err(|e| anyhow::anyhow!("trust policy {policy_path}: {e}"))?;
let json = TrustRoot::verified_json_from_bundle(
Path::new(path),
&policy,
&crate::import_door::EstateTime,
)
.map_err(|e| anyhow::anyhow!("config `rules.trust_root`: {}", e.tagged()))?;
Ok(Some(json))
}
/// The pinned APK keys (`(name, PEM)`) the floor's trust bundle carries, from
/// the same verified bundle as the root; the scan sandbox hands them to the worker.
#[cfg(feature = "import-door")]
pub fn load_apk_keys(
floor: &ProvenanceFloor,
bundle_policy: Option<&str>,
) -> anyhow::Result<Vec<(String, Vec<u8>)>> {
let Some(path) = floor.trust_root.as_deref().filter(|p| !p.trim().is_empty()) else {
return Ok(Vec::new());
};
if empty_slot(path) {
return Ok(Vec::new());
}
let Some(policy_path) = bundle_policy else {
return Ok(Vec::new());
};
let text = std::fs::read_to_string(policy_path)
.map_err(|e| anyhow::anyhow!("trust policy {policy_path}: {e}"))?;
let policy = update_bundle::TrustPolicy::from_ron(&text)
.map_err(|e| anyhow::anyhow!("trust policy {policy_path}: {e}"))?;
let (_, apk) = TrustRoot::verified_from_bundle(
Path::new(path),
&policy,
&crate::import_door::EstateTime,
)
.map_err(|e| anyhow::anyhow!("config `rules.trust_root`: {}", e.tagged()))?;
for (name, pem) in &apk {
holger_provenance::apk::ApkKey::from_pem(name, pem)
.map_err(|e| anyhow::anyhow!("config `rules.trust_root`: {}", e.tagged()))?;
}
Ok(apk)
}
/// The pinned OpenPGP keys (`(name, armored key)`, Maven Central's `.asc`
/// signers) the floor's trust bundle carries, from the same verified bundle as
/// the root; each is read and its bindings checked here, at boot.
#[cfg(feature = "import-door")]
pub fn load_pgp_keys(
floor: &ProvenanceFloor,
bundle_policy: Option<&str>,
) -> anyhow::Result<Vec<(String, Vec<u8>)>> {
let Some(path) = floor.trust_root.as_deref().filter(|p| !p.trim().is_empty()) else {
return Ok(Vec::new());
};
if empty_slot(path) {
return Ok(Vec::new());
}
let Some(policy_path) = bundle_policy else {
return Ok(Vec::new());
};
let text = std::fs::read_to_string(policy_path)
.map_err(|e| anyhow::anyhow!("trust policy {policy_path}: {e}"))?;
let policy = update_bundle::TrustPolicy::from_ron(&text)
.map_err(|e| anyhow::anyhow!("trust policy {policy_path}: {e}"))?;
let v =
TrustRoot::verified_bundle(Path::new(path), &policy, &crate::import_door::EstateTime)
.map_err(|e| anyhow::anyhow!("config `rules.trust_root`: {}", e.tagged()))?;
for (name, armored) in &v.pgp_keys {
holger_provenance::pgp::PgpKey::from_armored(name, armored)
.map_err(|e| anyhow::anyhow!("config `rules.trust_root`: {}", e.tagged()))?;
}
Ok(v.pgp_keys)
}
/// Verify in the scan sandbox from now on.
pub fn with_remote(self, remote: Arc<dyn RemoteVerify>) -> Self {
self.loaded.write_or_recover().remote = Some(remote);
self
}
/// The trust bundle the floor names (`rules.trust_root`), when it names one.
pub fn trust_root_path(&self) -> Option<&str> {
self.settings
.rules
.trust_root
.as_deref()
.filter(|p| !p.trim().is_empty())
}
/// The floor (its trust root path among it).
pub fn floor(&self) -> &ProvenanceFloor {
&self.settings.rules
}
/// **Read a refreshed trust root from now on** (lane HOL-OCI-EVIDENCE): an imported,
/// verified `holger.sigstore-trust/v1` bundle replaces the root, the sandboxed verifier
/// that holds its bytes, and the version (so earlier evidence is stale). Returns the
/// new version.
pub fn reload_trust(
&self,
trust: Option<Arc<TrustRoot>>,
remote: Option<Arc<dyn RemoteVerify>>,
) -> String {
let version = version_of(&self.settings, trust.as_ref());
let mut l = self.loaded.write_or_recover();
l.trust = trust;
if remote.is_some() {
l.remote = remote;
}
l.version = version.clone();
version
}
/// Does `verify_artifact` run in the sandbox?
pub fn is_remote(&self) -> bool {
self.loaded.read_or_recover().remote.is_some()
}
fn verify(
&self,
artifact: &[u8],
materials: &[(String, Vec<u8>)],
policy: &holger_provenance::EffectivePolicy,
) -> ProvenanceReport {
let (remote, trust) = {
let l = self.loaded.read_or_recover();
(l.remote.clone(), l.trust.clone())
};
match &remote {
Some(r) => r.verify(artifact, materials, policy),
None => verify_artifact(artifact, materials, policy, trust.as_deref()),
}
}
/// The gate for a subject: by its source, then its repository, then the default.
pub fn gate_for(&self, subject: &ScanSubject<'_>) -> (String, &ProvenanceGate) {
for key in [subject.source, subject.repo].into_iter().flatten() {
if let Some(g) = self.settings.sources.get(key) {
return (key.to_string(), g);
}
}
("<default>".to_string(), &self.settings.default)
}
/// The material this scanner's source holds about the bytes with `sha256_hex`,
/// from `source` (an APK source adds its signed indexes).
pub fn materials_for(
&self,
sha256_hex: &str,
file_name: &str,
source: Option<&str>,
) -> anyhow::Result<Vec<(String, Vec<u8>)>> {
let mut m = self.materials.materials(sha256_hex, file_name)?;
if let Some(src) = source.filter(|s| self.settings.apk_sources.iter().any(|a| a == s)) {
for (name, bytes) in self.materials.apk_indexes(src)? {
if !m.iter().any(|(n, _)| *n == name) {
m.push((name, bytes));
}
}
}
Ok(m)
}
/// Verify one subject against its own material and return the full report.
pub fn report(&self, subject: &ScanSubject<'_>) -> anyhow::Result<(String, ProvenanceReport)> {
let (by, gate) = self.gate_for(subject);
let policy = effective(&self.settings.rules, gate);
let materials = if policy.is_off() {
Vec::new()
} else {
self.materials_for(&subject.sha256_hex(), subject.file_name, subject.source)?
};
let image = signed_bytes_of(subject);
Ok((
by,
self.verify(
image.as_deref().unwrap_or(subject.bytes),
&materials,
&policy,
),
))
}
/// Verify one subject against material handed in (the airgap import's
/// carried files), under this side's own policy and trust root.
pub fn report_with(
&self,
subject: &ScanSubject<'_>,
materials: &[(String, Vec<u8>)],
) -> (String, ProvenanceReport) {
let (by, gate) = self.gate_for(subject);
let policy = effective(&self.settings.rules, gate);
let image = signed_bytes_of(subject);
(
by,
self.verify(
image.as_deref().unwrap_or(subject.bytes),
materials,
&policy,
),
)
}
}
impl Scanner for ProvenanceScanner {
fn id(&self) -> &str {
SCANNER_ID
}
fn version(&self) -> String {
self.loaded.read_or_recover().version.clone()
}
fn kind(&self) -> ScannerKind {
ScannerKind::Provenance
}
fn scan(&self, subject: &ScanSubject<'_>) -> anyhow::Result<ScanOutcome> {
let (by, report) = self.report(subject)?;
let mut findings: Vec<ScanFinding> = report
.candidates
.iter()
.filter(|c| !c.verified)
.map(|c| {
ScanFinding::new(
c.code.clone().unwrap_or_default(),
FindingSeverity::High,
format!("{}: {}", c.origin, c.error.clone().unwrap_or_default()),
)
})
.collect();
if let (Some(code), true) = (&report.code, findings.is_empty()) {
findings.push(ScanFinding::new(
code.clone(),
FindingSeverity::High,
report.detail.clone(),
));
}
let detail = format!("source {by}: {}", report.detail);
let doc = serde_json::to_value(&report)?;
Ok(match report.verdict {
Verdict::Pass => ScanOutcome {
findings,
..ScanOutcome::pass(detail)
},
Verdict::Warn => ScanOutcome::warn(findings, detail),
Verdict::Block => ScanOutcome::block(findings, detail),
}
.with_document(doc))
}
}
/// **The estate's trust root slot, empty** (lane HOL-OCI-EVIDENCE): the floor names the bundle
/// a medium seeds under `<data_root>/holger-trust/` (`crate::sigstore_trust`), and this medium
/// seeded none. The box boots and verifies no keyless signature; any other named root that is
/// missing still refuses the boot.
fn empty_slot(path: &str) -> bool {
let p = Path::new(path);
!p.exists()
&& p.file_name()
.is_some_and(|n| n == crate::sigstore_trust::BUNDLE_FILE)
&& p.parent()
.and_then(|d| d.file_name())
.is_some_and(|d| d == crate::sigstore_trust::DIR)
}
/// **What an image's signatures are about** (lane HOL-OCI-EVIDENCE): cosign signs an image's
/// MANIFEST (its digest is the signature's `docker-manifest-digest`), not the layout tar the
/// chain carries, so the subject of an image layout (`.oci.tar`, pulled through or pushed to
/// a scanned registry) is its top manifest. `None` for anything else, or a layout that does
/// not read (the bytes then stand, and nothing verifies against them).
fn signed_bytes_of(subject: &ScanSubject<'_>) -> Option<Vec<u8>> {
if !subject.file_name.ends_with(".oci.tar") {
return None;
}
let l = crate::pull_intake::oci::Layout::read(subject.bytes).ok()?;
l.manifest(&l.digest).map(|(_, m)| m.to_vec())
}
/// The provenance report a stored evidence record carries, when it has one.
pub fn report_of(evidence: &super::Evidence) -> Option<ProvenanceReport> {
evidence
.records_of(ScannerKind::Provenance)
.find(|r| r.scanner_id == SCANNER_ID)
.and_then(|r| r.document.clone())
.and_then(|d| serde_json::from_value(d).ok())
}
#[cfg(test)]
mod apk_source_tests {
use super::*;
const FX: &str = concat!(
env!("CARGO_MANIFEST_DIR"),
"/../../provenance/tests/fixtures"
);
fn fx(p: &str) -> Vec<u8> {
let b = std::fs::read(format!("{FX}/{p}")).unwrap_or_else(|e| panic!("{p}: {e}"));
assert!(
!b.starts_with(b"version https://git-lfs"),
"{p} is an LFS pointer: run `git lfs pull`"
);
b
}
#[test]
fn an_apk_source_reads_its_signed_index_from_the_inbox_apk_dir() {
let dir = tempfile::tempdir().unwrap();
let inbox = ProvenanceInbox::new(dir.path());
std::fs::create_dir_all(inbox.apk_dir("wolfi")).unwrap();
std::fs::write(
inbox.apk_dir("wolfi").join("APKINDEX.tar.gz"),
fx("wolfi/stage3/APKINDEX.tar.gz"),
)
.unwrap();
let settings = ProvenanceScanSettings {
rules: ProvenanceFloor {
require_provenance: true,
..Default::default()
},
apk_sources: vec!["wolfi".into()],
..Default::default()
};
let trust = TrustRoot::default()
.with_apk_keys(&[(
"wolfi-signing.rsa.pub".into(),
fx("wolfi/wolfi-signing.rsa.pub"),
)])
.unwrap();
let s = ProvenanceScanner::new(settings, Some(Arc::new(trust)), Arc::new(inbox)).unwrap();
let apk = fx("wolfi/stage3/libffi-3.4.2-r0.apk");
let subject = ScanSubject::new("libffi-3.4.2-r0.apk", "raw", &apk);
let (_, r) = s.report(&subject.with_source("wolfi")).unwrap();
assert_eq!(r.verdict, Verdict::Pass, "{}", r.detail);
// Another source does not get the Wolfi index.
let (_, r) = s.report(&subject.with_source("team")).unwrap();
assert_eq!(
r.code.as_deref(),
Some("holger-provenance-1"),
"{}",
r.detail
);
}
/// T9: a Maven upload whose `.asc` waits in the inbox beside it, against the
/// OpenPGP keys the trust root pins, narrowed per source by `require(pgp_keys)`.
#[test]
fn a_maven_pom_and_its_asc_verify_against_the_pinned_pgp_keys_per_source() {
let povirk = "povirk-BDB5FA4FE719D787FB3D3197F6D4A1D411E9D1AE.asc";
let ceki = "ceki-60200AC4AE761F1614D6C46766D68DAA073BE985.asc";
let pom = fx("maven/guava-33.3.1-jre.pom");
let dir = tempfile::tempdir().unwrap();
let inbox = ProvenanceInbox::new(dir.path());
let sha = holger_provenance::ArtifactDigests::of(&pom).sha256;
inbox
.put(
&sha,
"guava-33.3.1-jre.pom.asc",
&fx("maven/guava-33.3.1-jre.pom.asc"),
)
.unwrap();
let only = |k: &str| ProvenanceGate::Require {
identity: None,
identities: vec![],
builders: vec![],
keys: vec![],
apk_keys: vec![],
pgp_keys: vec![k.to_string()],
slsa_min: None,
};
let settings = ProvenanceScanSettings {
rules: ProvenanceFloor {
require_provenance: true,
..Default::default()
},
sources: [
("central".to_string(), only(povirk)),
("slf4j".to_string(), only(ceki)),
]
.into_iter()
.collect(),
..Default::default()
};
let keys: Vec<(String, Vec<u8>)> = [povirk, ceki]
.iter()
.map(|k| (k.to_string(), fx(&format!("maven/{k}"))))
.collect();
if !holger_provenance::pgp::BUILT {
// T9b: without `maven-pgp`, the same sources and keys are refused at load by name.
let e = TrustRoot::default().with_pgp_keys(&keys).unwrap_err();
assert!(e.tagged().contains("[provenance-pgp-not-built]"), "{e}");
let e = format!(
"{:#}",
ProvenanceScanner::new(settings, None, Arc::new(inbox))
.err()
.expect("refused")
);
assert!(
e.contains("[provenance-pgp-not-built]") && e.contains("\"central\""),
"{e}"
);
return;
}
let trust = TrustRoot::default().with_pgp_keys(&keys).unwrap();
let s = ProvenanceScanner::new(settings, Some(Arc::new(trust)), Arc::new(inbox)).unwrap();
let subject = ScanSubject::new("guava-33.3.1-jre.pom", "raw", &pom);
let (_, r) = s.report(&subject.with_source("central")).unwrap();
assert_eq!(r.verdict, Verdict::Pass, "{}", r.detail);
// A source pinned to another signer: unverified, refused by name.
let (_, r) = s.report(&subject.with_source("slf4j")).unwrap();
assert_eq!(
(r.verdict, r.code.as_deref()),
(Verdict::Block, Some("holger-provenance-4")),
"{}",
r.detail
);
assert!(s.version().contains("pgp keys"), "{}", s.version());
}
/// T9b: a floor key of `alg: openpgp` in a build without OpenPGP (`maven-pgp`) is
/// refused when the scanner is built, by name, even with no source to check it.
#[test]
fn a_floor_openpgp_key_without_the_feature_is_refused_by_name_at_load() {
let povirk = "povirk-BDB5FA4FE719D787FB3D3197F6D4A1D411E9D1AE.asc";
let settings = ProvenanceScanSettings {
rules: ProvenanceFloor {
keys: vec![holger_provenance::TrustedKey {
id: "guava-release".into(),
alg: holger_provenance::KeyAlg::Openpgp,
public_key: String::from_utf8(fx(&format!("maven/{povirk}"))).unwrap(),
}],
..Default::default()
},
..Default::default()
};
let dir = tempfile::tempdir().unwrap();
let r = ProvenanceScanner::new(settings, None, Arc::new(ProvenanceInbox::new(dir.path())));
if holger_provenance::pgp::BUILT {
r.unwrap();
} else {
let e = format!("{:#}", r.err().expect("refused at load"));
assert!(
e.contains("[provenance-pgp-not-built]") && e.contains("`maven-pgp`"),
"{e}"
);
}
}
}