provenance/src/verify.rs
//! ★ The verifier: every candidate checked in a fixed order, and one decision
//! under the effective policy.
//!
//! Per candidate, in this order (the first failure is its refusal):
//!
//! 1. **Binding.** The statement's subject digest is the artifact's sha256 (or
//! sha512, as npm writes it); a message signature's digest is the artifact's;
//! a cosign image signature names the manifest digest. A valid signature over
//! OTHER bytes is never evidence for these.
//! 2. **Signature.** Over `PAE(payloadType, payload)` for DSSE (byte-exact,
//! `znippy_common::sign::dsse_pae`), over the artifact or the simple-signing
//! payload otherwise. A changed byte fails here.
//! 3. **Signer.** A key: one of the policy's `keys`. A certificate (keyless): an
//! allowed identity (OIDC issuer + subject), a Rekor entry that verifies
//! against the pinned log and records THIS signature by THIS certificate, the
//! certificate valid at the logged time, issued by a pinned Fulcio CA valid
//! then, code-signing, with an SCT from a pinned CT log.
//! 4. **Provenance.** For SLSA v1 / v0.2 and PyPI publish statements: the builder
//! is allowed and its SLSA build level reaches `slsa_min`.
use serde::{Deserialize, Serialize};
use crate::cert;
use crate::crypto;
use crate::error::{ProvenanceError, Result};
use crate::material::{self, Candidate, Signed, Signer};
use crate::policy::{EffectivePolicy, Mode};
use crate::statement::{self, ArtifactDigests, SlsaSummary, Statement};
use crate::tlog::{self, Binding, TlogCheck};
use crate::trust::TrustRoot;
/// The schema of the evidence document.
pub const EVIDENCE_SCHEMA: &str = "holger.provenance-evidence/v1";
/// The SLSA build level a verified, signed provenance shows when the policy
/// gives its builder no level: signed provenance from a named builder (L2).
pub const DEFAULT_SIGNED_LEVEL: u8 = 2;
/// What this verifier never checks, because it is offline by design.
pub const NOT_CHECKED_OFFLINE: &[&str] = &[
"no online lookup: a Rekor entry is checked against the pinned log key only (its SET and/or its inclusion proof against a signed checkpoint); no consistency proof between checkpoints",
"no revocation: Fulcio certificates live ten minutes and Sigstore publishes no CRL",
"no RFC 3161 timestamp: the signing time is the Rekor entry's integratedTime",
"the trust root is as fresh as the last signed trust bundle carried over the airgap",
];
/// Who signed a verified candidate.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum SignerInfo {
Key {
id: String,
alg: String,
},
Keyless {
issuer: String,
subject: String,
ca: String,
},
}
/// One candidate, as the evidence records it.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct CandidateReport {
pub origin: String,
#[serde(default, skip_serializing_if = "String::is_empty")]
pub predicate_type: String,
/// `true` when every check passed.
pub verified: bool,
/// The refusal's code (`holger-provenance-<n>`), when refused.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub code: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub error: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub signer: Option<SignerInfo>,
/// A provenance statement (SLSA or PyPI publish) about these bytes.
pub is_provenance: bool,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub slsa: Option<SlsaSummary>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub builder_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub slsa_level: Option<u8>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub tlog: Vec<TlogCheck>,
/// What was checked, in order.
pub checked: Vec<String>,
/// What was not checked for this candidate, and why.
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub not_checked: Vec<String>,
}
/// The decision.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum Verdict {
Pass,
Warn,
Block,
}
/// ★ The evidence about one artifact.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ProvenanceReport {
pub schema: String,
pub verdict: Verdict,
/// The deciding refusal's code, when the verdict is not `pass`.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub code: Option<String>,
pub detail: String,
pub digests: ArtifactDigests,
pub provenance: Mode,
pub signed: Mode,
pub policy: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub trust_root_sha256: Option<String>,
pub candidates: Vec<CandidateReport>,
/// Material that could not be read at all, with its refusal.
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub unreadable: Vec<String>,
pub not_checked: Vec<String>,
}
impl ProvenanceReport {
/// The verified provenance candidates.
pub fn verified_provenance(&self) -> impl Iterator<Item = &CandidateReport> {
self.candidates
.iter()
.filter(|c| c.verified && c.is_provenance)
}
}
struct Ctx<'a> {
artifact: &'a [u8],
digests: &'a ArtifactDigests,
policy: &'a EffectivePolicy,
trust: Option<&'a TrustRoot>,
}
/// The bytes a signature covers, and what it binds.
struct Bound {
message: Vec<u8>,
payload: Vec<u8>,
statement: Option<Statement>,
signatures: Vec<(String, Vec<u8>)>,
}
fn bind(c: &Candidate, cx: &Ctx<'_>, r: &mut CandidateReport) -> Result<Bound> {
match &c.signed {
Signed::Dsse {
payload_type,
payload,
signatures,
} => {
if payload_type != statement::PAYLOAD_TYPE_INTOTO {
return Err(ProvenanceError::Malformed(format!(
"payloadType {payload_type:?} is not {}; DSSE binds the type into the signed bytes",
statement::PAYLOAD_TYPE_INTOTO
)));
}
let st = Statement::parse(payload)?;
r.predicate_type = st.predicate_type.clone();
if !st.names(cx.digests) {
return Err(ProvenanceError::SubjectMismatch(format!(
"the statement names {:?}, not sha256:{}",
st.describe_subjects(),
cx.digests.sha256
)));
}
r.checked.push(format!(
"subject digest is the artifact's ({})",
st.statement_type
));
Ok(Bound {
message: znippy_common::sign::dsse_pae(payload_type, payload),
payload: payload.clone(),
statement: Some(st),
signatures: signatures.clone(),
})
}
Signed::Artifact { sha256, signature } => {
if let Some(d) = sha256 {
if *d != cx.digests.sha256 {
return Err(ProvenanceError::SubjectMismatch(format!(
"the message digest is sha256:{d}, not sha256:{}",
cx.digests.sha256
)));
}
}
r.predicate_type = "message-signature".into();
r.checked.push("message digest is the artifact's".into());
Ok(Bound {
message: cx.artifact.to_vec(),
payload: cx.artifact.to_vec(),
statement: None,
signatures: vec![(String::new(), signature.clone())],
})
}
Signed::SimpleSigning { payload, signature } => {
let v: serde_json::Value = serde_json::from_slice(payload).map_err(|e| {
ProvenanceError::Malformed(format!("the cosign simple-signing payload: {e}"))
})?;
let named = v
.pointer("/critical/image/docker-manifest-digest")
.and_then(|x| x.as_str())
.unwrap_or("");
let want = format!("sha256:{}", cx.digests.sha256);
if named != want {
return Err(ProvenanceError::SubjectMismatch(format!(
"the image signature names {named:?}, not {want}"
)));
}
r.predicate_type = "cosign container image signature".into();
r.checked
.push("the signed manifest digest is the artifact's".into());
Ok(Bound {
message: payload.clone(),
payload: payload.clone(),
statement: None,
signatures: vec![(String::new(), signature.clone())],
})
}
Signed::NpmRegistry {
name,
version,
integrity,
signatures,
} => {
use base64::Engine as _;
let sha512 = hex::decode(&cx.digests.sha512).unwrap_or_default();
let want = format!(
"sha512-{}",
base64::engine::general_purpose::STANDARD.encode(sha512)
);
if *integrity != want {
return Err(ProvenanceError::SubjectMismatch(format!(
"the registry signed integrity {integrity:?}, the tarball is {want:?}"
)));
}
r.predicate_type = "npm registry signature".into();
r.checked
.push("the signed integrity is the tarball's sha512".into());
let message = format!("{name}@{version}:{integrity}").into_bytes();
Ok(Bound {
message: message.clone(),
payload: message,
statement: None,
signatures: signatures.clone(),
})
}
}
}
fn verify_key_signed(
c: &Candidate,
b: &Bound,
hint: &str,
cx: &Ctx<'_>,
r: &mut CandidateReport,
) -> Result<Vec<u8>> {
let mut tried = 0;
for (keyid, sig) in &b.signatures {
let want = if keyid.is_empty() {
hint
} else {
keyid.as_str()
};
// An OpenPGP key verifies `.asc` files only ([`verify_pgp`]).
for k in cx.policy.keys.iter().filter(|k| {
k.alg != crate::policy::KeyAlg::Openpgp && (want.is_empty() || k.id == want)
}) {
tried += 1;
let key = crypto::PublicKey::from_text(&k.public_key)
.map_err(|e| ProvenanceError::PolicyInvalid(format!("key {:?}: {e}", k.id)))?;
if key.verify(&b.message, sig) {
r.checked.push(format!(
"signature verifies with allowed key {:?} ({})",
k.id,
key.alg()
));
r.signer = Some(SignerInfo::Key {
id: k.id.clone(),
alg: key.alg().into(),
});
let spki = crypto::pem_or_b64(&k.public_key)?;
let _ = c;
return Ok(spki);
}
}
}
if tried == 0 {
let ids: Vec<&str> = b.signatures.iter().map(|(k, _)| k.as_str()).collect();
return Err(ProvenanceError::SignerUntrusted(format!(
"no allowed key for keyid {ids:?} (hint {hint:?})"
)));
}
Err(ProvenanceError::SignatureInvalid(format!(
"the signature does not verify with any of the {tried} allowed key(s) it names"
)))
}
fn check_tlog(
c: &Candidate,
b: &Bound,
sig: &[u8],
verifier_der: &[u8],
trust: &TrustRoot,
r: &mut CandidateReport,
) -> Result<i64> {
let mut last = None;
for e in &c.tlog {
let checked = tlog::verify_entry(e, trust).and_then(|chk| {
tlog::check_binding(
e,
&Binding {
payload: &b.payload,
envelope: c.envelope_bytes.as_deref(),
signature: sig,
verifier_der,
},
)?;
Ok(chk)
});
match checked {
Ok(chk) => {
r.checked.push(format!(
"Rekor entry {} in {} verifies ({}{}) and records this signature",
chk.log_index,
chk.log,
if chk.set_verified { "SET" } else { "" },
if chk.inclusion_verified {
if chk.set_verified {
" + inclusion proof"
} else {
"inclusion proof"
}
} else {
""
},
));
let t = chk.integrated_time;
r.tlog.push(chk);
return Ok(t);
}
Err(e) => last = Some(e),
}
}
Err(last.unwrap_or_else(|| {
ProvenanceError::TlogUnverified(
"a keyless signature carries no transparency-log entry".into(),
)
}))
}
fn verify_keyless(
c: &Candidate,
b: &Bound,
leaf: &[u8],
cx: &Ctx<'_>,
r: &mut CandidateReport,
) -> Result<()> {
let key = cert::public_key(leaf)?;
let sig = b
.signatures
.iter()
.map(|(_, s)| s)
.find(|s| key.verify(&b.message, s))
.ok_or_else(|| {
ProvenanceError::SignatureInvalid(
"the signature does not verify with the certificate's key".into(),
)
})?
.clone();
r.checked.push(format!(
"signature verifies with the certificate's key ({})",
key.alg()
));
let id = cert::identity(leaf)?;
let (issuer, subject) = (
id.issuer.unwrap_or_default(),
id.subject.unwrap_or_default(),
);
if !cx
.policy
.identities
.iter()
.any(|i| i.matches(&issuer, &subject))
{
return Err(ProvenanceError::IdentityMismatch(format!(
"issuer {issuer:?} subject {subject:?} is not one of the {} allowed identit{}",
cx.policy.identities.len(),
if cx.policy.identities.len() == 1 {
"y"
} else {
"ies"
}
)));
}
r.checked
.push(format!("identity {issuer} / {subject} is allowed"));
let trust = cx.trust.ok_or_else(|| {
ProvenanceError::TrustRoot(
"a keyless signature needs the pinned Sigstore trust root, and none is configured"
.into(),
)
})?;
let t = check_tlog(c, b, &sig, leaf, trust, r)?;
let (nb, na) = cert::validity(leaf)?;
if t < nb || t > na {
return Err(ProvenanceError::CertExpired(format!(
"logged at {t}, the certificate is valid {nb}..{na}"
)));
}
r.checked
.push("the certificate was valid at the logged signing time".into());
let ca = trust
.cas
.iter()
.find(|ca| ca.valid.covers(t) && cert::issued_by(leaf, &ca.chain[0]).unwrap_or(false))
.ok_or_else(|| {
ProvenanceError::SignerUntrusted(
"the certificate was not issued by a pinned Fulcio CA valid at the signing time"
.into(),
)
})?;
r.checked
.push(format!("issued by the pinned CA {}", ca.uri));
if !cert::allows_code_signing(leaf)? {
return Err(ProvenanceError::SignerUntrusted(
"the certificate does not allow code signing".into(),
));
}
if trust.ctlogs.is_empty() {
r.not_checked
.push("SCT: the trust root pins no CT log".into());
} else {
let issuer_spki = cert::spki_der(&ca.chain[0])?;
let scts = cert::scts(leaf)?;
let ok = scts.iter().any(|s| {
trust.ctlog(&s.log_id).is_some_and(|log| {
cert::sct_signed_bytes(s, leaf, &issuer_spki)
.map(|m| log.key.verify(&m, &s.signature))
.unwrap_or(false)
})
});
if !ok {
return Err(ProvenanceError::SignerUntrusted(format!(
"none of the certificate's {} SCT(s) verifies with a pinned CT log",
scts.len()
)));
}
r.checked
.push("an embedded SCT verifies with a pinned CT log".into());
}
r.signer = Some(SignerInfo::Keyless {
issuer,
subject,
ca: ca.uri.clone(),
});
Ok(())
}
fn judge_provenance(b: &Bound, cx: &Ctx<'_>, r: &mut CandidateReport) -> Result<()> {
let Some(st) = b.statement.as_ref().filter(|s| s.is_provenance()) else {
return Ok(());
};
r.is_provenance = true;
let slsa = st.slsa();
let builder = match &slsa {
Some(s) => s.builder_id.clone(),
// A PyPI publish attestation names its builder through the certificate:
// the Trusted Publisher workflow that signed it.
None => match &r.signer {
Some(SignerInfo::Keyless { subject, .. }) => subject.clone(),
Some(SignerInfo::Key { id, .. }) => id.clone(),
None => String::new(),
},
};
r.slsa = slsa;
r.builder_id = Some(builder.clone());
let rule = cx.policy.builders.iter().find(|b| b.id == builder);
if !cx.policy.builders.is_empty() && rule.is_none() {
return Err(ProvenanceError::BuilderNotAllowed(format!(
"builder {builder:?} is not one of the {} allowed builder(s)",
cx.policy.builders.len()
)));
}
let level = rule.and_then(|b| b.level).unwrap_or(DEFAULT_SIGNED_LEVEL);
r.slsa_level = Some(level);
if level < cx.policy.slsa_min {
return Err(ProvenanceError::SlsaLevelTooLow(format!(
"builder {builder:?} is SLSA build level {level}, below the minimum {}",
cx.policy.slsa_min
)));
}
r.checked
.push(format!("builder {builder:?} at SLSA build level {level}"));
Ok(())
}
fn verify_one(c: &Candidate, cx: &Ctx<'_>) -> CandidateReport {
let mut r = CandidateReport {
origin: c.origin.clone(),
predicate_type: String::new(),
verified: false,
code: None,
error: None,
signer: None,
is_provenance: false,
slsa: None,
builder_id: None,
slsa_level: None,
tlog: Vec::new(),
checked: Vec::new(),
not_checked: Vec::new(),
};
let out = (|| -> Result<()> {
let b = bind(c, cx, &mut r)?;
match &c.signer {
Signer::Key { hint } => {
let spki = verify_key_signed(c, &b, hint, cx, &mut r)?;
if !c.tlog.is_empty() {
match cx.trust {
Some(trust) => {
let sig = b
.signatures
.first()
.map(|(_, s)| s.clone())
.unwrap_or_default();
check_tlog(c, &b, &sig, &spki, trust, &mut r)?;
}
None => r
.not_checked
.push("its Rekor entry: no trust root is configured".into()),
}
}
}
Signer::Cert { leaf, .. } => verify_keyless(c, &b, leaf, cx, &mut r)?,
}
judge_provenance(&b, cx, &mut r)
})();
match out {
Ok(()) => r.verified = true,
Err(e) => {
r.code = Some(e.code_str().to_string());
r.error = Some(e.tagged());
}
}
r
}
/// An APK package against one `APKINDEX.tar.gz` ([`crate::apk`]): the index
/// signature with a pinned apk key (narrowed by the policy's `apk_keys`), the
/// control hash the index names, the data hash `.PKGINFO` names. A verified index
/// entry is the distributor's signed statement of where these bytes come from, so
/// it counts as provenance; it names no SLSA build level.
fn verify_apk(name: &str, bytes: &[u8], cx: &Ctx<'_>) -> CandidateReport {
let mut r = CandidateReport {
origin: format!("apk-index:{name}"),
predicate_type: "apk-index-signature".into(),
verified: false,
code: None,
error: None,
signer: None,
is_provenance: false,
slsa: None,
builder_id: None,
slsa_level: None,
tlog: Vec::new(),
checked: Vec::new(),
not_checked: Vec::new(),
};
let out = (|| -> Result<()> {
let pinned: Vec<crate::apk::ApkKey> = cx
.trust
.map(|t| {
t.apk_keys
.iter()
.filter(|k| {
cx.policy.apk_keys.is_empty() || cx.policy.apk_keys.contains(&k.name)
})
.cloned()
.collect()
})
.unwrap_or_default();
if pinned.is_empty() {
return Err(ProvenanceError::SignerUntrusted(
if cx.trust.is_none_or(|t| t.apk_keys.is_empty()) {
"no apk key is pinned: the trust bundle carries none".into()
} else {
format!(
"none of the pinned apk keys is one the policy allows ({:?})",
cx.policy.apk_keys
)
},
));
}
let idx = crate::apk::verify_index(bytes, &pinned)?;
r.checked.push(format!(
"the APK index signature verifies with the pinned key {:?} ({})",
idx.key, idx.alg
));
r.signer = Some(SignerInfo::Key {
id: idx.key.clone(),
alg: idx.alg.into(),
});
let p = crate::apk::verify_package(cx.artifact, &idx)?;
r.checked.push(format!(
"the signed index names {} {} with this control segment's hash and this size",
p.name, p.version
));
r.checked.push(format!(
".PKGINFO's datahash is the data segment's sha256 ({})",
p.datahash
));
r.is_provenance = true;
let builder = format!("apk-key:{}", idx.key);
r.builder_id = Some(builder.clone());
if !cx.policy.builders.is_empty() && !cx.policy.builders.iter().any(|b| b.id == builder) {
return Err(ProvenanceError::BuilderNotAllowed(format!(
"builder {builder:?} is not one of the {} allowed builder(s)",
cx.policy.builders.len()
)));
}
if cx.policy.slsa_min > 0 {
return Err(ProvenanceError::SlsaLevelTooLow(format!(
"a signed APK index shows no SLSA build level, and the minimum is {}",
cx.policy.slsa_min
)));
}
r.not_checked.push("the package's own .SIGN segment, when it has one: the signed index already names its control hash".into());
Ok(())
})();
match out {
Ok(()) => r.verified = true,
Err(e) => {
r.code = Some(e.code_str().to_string());
r.error = Some(e.tagged());
}
}
r
}
/// A detached OpenPGP signature (Maven Central's `.asc`, [`crate::pgp`]) against
/// the pinned OpenPGP keys: the trust bundle's (narrowed by the policy's
/// `pgp_keys`) and the policy's `keys` of `alg: openpgp`. A signature by an
/// unpinned key is unverified (`holger-provenance-4`). Like a signed APK index, a
/// verified signature by a pinned release key is the publisher's statement that
/// these are its bytes, so it counts as provenance with no SLSA build level.
fn verify_pgp(name: &str, asc: &[u8], cx: &Ctx<'_>) -> CandidateReport {
let mut r = CandidateReport {
origin: format!("pgp-signature:{name}"),
predicate_type: "openpgp-detached-signature".into(),
verified: false,
code: None,
error: None,
signer: None,
is_provenance: false,
slsa: None,
builder_id: None,
slsa_level: None,
tlog: Vec::new(),
checked: Vec::new(),
not_checked: Vec::new(),
};
let out = (|| -> Result<()> {
if !crate::pgp::BUILT {
// No verifier, so no pin can be honoured: unverified, by name, never a pass.
return Err(ProvenanceError::SignerUntrusted(format!(
"unverified: {}",
crate::pgp::not_built("a detached OpenPGP signature (.asc)")
)));
}
let mut pinned: Vec<crate::pgp::PgpKey> = cx
.trust
.map(|t| {
t.pgp_keys
.iter()
.filter(|k| {
cx.policy.pgp_keys.is_empty() || cx.policy.pgp_keys.contains(&k.name)
})
.cloned()
.collect()
})
.unwrap_or_default();
for k in cx
.policy
.keys
.iter()
.filter(|k| k.alg == crate::policy::KeyAlg::Openpgp)
{
pinned.push(
crate::pgp::PgpKey::from_armored(&k.id, k.public_key.as_bytes()).map_err(|e| {
ProvenanceError::PolicyInvalid(format!("key {:?} (alg openpgp): {e}", k.id))
})?,
);
}
if pinned.is_empty() {
return Err(ProvenanceError::SignerUntrusted(
if cx.trust.is_none_or(|t| t.pgp_keys.is_empty()) {
"unverified: no OpenPGP key is pinned (the trust bundle carries none and no floor key is alg openpgp)".into()
} else {
format!("unverified: none of the pinned OpenPGP keys is one the policy allows ({:?})", cx.policy.pgp_keys)
},
));
}
let v = crate::pgp::verify_detached(cx.artifact, asc, &pinned)?;
r.checked.push(format!(
"the OpenPGP signature verifies over these bytes with the pinned key {:?} ({}, signer {})",
v.key, v.alg, v.signer_fingerprint
));
r.signer = Some(SignerInfo::Key {
id: v.key.clone(),
alg: v.alg.clone(),
});
r.is_provenance = true;
let builder = format!("pgp-key:{}", v.key);
r.builder_id = Some(builder.clone());
if !cx.policy.builders.is_empty() && !cx.policy.builders.iter().any(|b| b.id == builder) {
return Err(ProvenanceError::BuilderNotAllowed(format!(
"builder {builder:?} is not one of the {} allowed builder(s)",
cx.policy.builders.len()
)));
}
if cx.policy.slsa_min > 0 {
return Err(ProvenanceError::SlsaLevelTooLow(format!(
"an OpenPGP signature shows no SLSA build level, and the minimum is {}",
cx.policy.slsa_min
)));
}
r.not_checked.push(
"the pinned key's expiry and any revocation published after it was pinned: the key is trusted as the trust bundle pins it".into(),
);
Ok(())
})();
match out {
Ok(()) => r.verified = true,
Err(e) => {
r.code = Some(e.code_str().to_string());
r.error = Some(e.tagged());
}
}
r
}
/// The refusal that decides, when nothing passed: a refused provenance candidate
/// first, then any refused candidate, then "nothing was offered".
fn deciding(
cands: &[CandidateReport],
want_provenance: bool,
unreadable: &[ProvenanceError],
) -> ProvenanceError {
let refused = |c: &&CandidateReport| !c.verified;
let pick = cands
.iter()
.filter(refused)
.find(|c| {
c.is_provenance
|| c.predicate_type.contains("provenance")
|| c.predicate_type.contains("attestations/publish")
})
.or_else(|| cands.iter().find(refused));
if let Some(c) = pick {
let msg = c.error.clone().unwrap_or_default();
let code = c.code.clone().unwrap_or_default();
return error_from(&code, &format!("{}: {msg}", c.origin));
}
if let Some(e) = unreadable.first() {
return e.clone();
}
if want_provenance {
if cands.iter().any(|c| c.verified) {
ProvenanceError::Missing(
"the bytes are signed, but no signed provenance statement names them".into(),
)
} else {
ProvenanceError::Missing("no attestation names these bytes".into())
}
} else {
ProvenanceError::Unsigned("no signature or attestation names these bytes".into())
}
}
fn error_from(code: &str, msg: &str) -> ProvenanceError {
let m = msg.to_string();
match code {
"holger-provenance-2" => ProvenanceError::SubjectMismatch(m),
"holger-provenance-3" => ProvenanceError::SignatureInvalid(m),
"holger-provenance-4" => ProvenanceError::SignerUntrusted(m),
"holger-provenance-5" => ProvenanceError::IdentityMismatch(m),
"holger-provenance-6" => ProvenanceError::BuilderNotAllowed(m),
"holger-provenance-7" => ProvenanceError::SlsaLevelTooLow(m),
"holger-provenance-8" => ProvenanceError::TlogUnverified(m),
"holger-provenance-9" => ProvenanceError::Malformed(m),
"holger-provenance-10" => ProvenanceError::TrustRoot(m),
"holger-provenance-11" => ProvenanceError::PolicyInvalid(m),
"holger-provenance-12" => ProvenanceError::CertExpired(m),
"holger-provenance-13" => ProvenanceError::Unsigned(m),
_ => ProvenanceError::Missing(m),
}
}
/// ★ Verify every piece of material about `artifact` and decide under `policy`.
///
/// `materials` are `(name, bytes)` files as they travelled with the artifact.
/// Nothing here reads the network or the clock: the signing time is the log's.
pub fn verify_artifact(
artifact: &[u8],
materials: &[(String, Vec<u8>)],
policy: &EffectivePolicy,
trust: Option<&TrustRoot>,
) -> ProvenanceReport {
let digests = ArtifactDigests::of(artifact);
let mut report = ProvenanceReport {
schema: EVIDENCE_SCHEMA.to_string(),
verdict: Verdict::Pass,
code: None,
detail: String::new(),
digests: digests.clone(),
provenance: policy.provenance,
signed: policy.signed,
policy: policy.summary(),
trust_root_sha256: trust.map(|t| t.digest.clone()),
candidates: Vec::new(),
unreadable: Vec::new(),
not_checked: NOT_CHECKED_OFFLINE.iter().map(|s| s.to_string()).collect(),
};
if policy.is_off() {
report.detail = "provenance is off for this source; nothing was checked".into();
return report;
}
let cx = Ctx {
artifact,
digests: &digests,
policy,
trust,
};
let mut unreadable = Vec::new();
for (name, bytes) in materials {
if crate::apk::is_index(name, bytes) {
report.candidates.push(verify_apk(name, bytes, &cx));
continue;
}
if crate::pgp::is_signature(name, bytes) {
report.candidates.push(verify_pgp(name, bytes, &cx));
continue;
}
match material::parse(name, bytes) {
Ok(cands) => report
.candidates
.extend(cands.iter().map(|c| verify_one(c, &cx))),
Err(e) => {
report.unreadable.push(format!("{name}: {}", e.tagged()));
unreadable.push(e);
}
}
}
let any_verified = report.candidates.iter().any(|c| c.verified);
let prov_ok = report.verified_provenance().count();
let fail = |want_prov: bool| deciding(&report.candidates, want_prov, &unreadable);
let (verdict, err) = match (policy.provenance, policy.signed) {
(Mode::Require, _) if prov_ok == 0 => (Verdict::Block, Some(fail(true))),
(_, Mode::Require) if !any_verified => (Verdict::Block, Some(fail(false))),
(Mode::Prefer, _) if prov_ok == 0 => (Verdict::Warn, Some(fail(true))),
(_, Mode::Prefer) if !any_verified => (Verdict::Warn, Some(fail(false))),
_ => (Verdict::Pass, None),
};
report.verdict = verdict;
match err {
Some(e) => {
report.code = Some(e.code_str().to_string());
report.detail = format!(
"{} under provenance {} / signed {}: {}",
if verdict == Verdict::Block {
"refused"
} else {
"warned"
},
policy.provenance.as_str(),
policy.signed.as_str(),
e.tagged()
);
}
None => {
let n = report.candidates.iter().filter(|c| c.verified).count();
report.detail = format!(
"{n} of {} candidate(s) verified offline, {prov_ok} of them provenance",
report.candidates.len()
);
}
}
report
}