provenance/src/statement.rs
//! in-toto Statements and the SLSA provenance predicate.
use std::collections::BTreeMap;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use crate::error::{ProvenanceError, Result};
/// in-toto Statement v1.
pub const STATEMENT_V1: &str = "https://in-toto.io/Statement/v1";
/// in-toto Statement v0.1 (cosign still writes it; Chainguard's images carry it).
pub const STATEMENT_V01: &str = "https://in-toto.io/Statement/v0.1";
/// The DSSE payload type of an in-toto statement.
pub const PAYLOAD_TYPE_INTOTO: &str = "application/vnd.in-toto+json";
/// SLSA Provenance v1.
pub const SLSA_PROVENANCE_V1: &str = "https://slsa.dev/provenance/v1";
/// SLSA Provenance v0.2.
pub const SLSA_PROVENANCE_V02: &str = "https://slsa.dev/provenance/v0.2";
/// PyPI's PEP 740 publish attestation.
pub const PYPI_PUBLISH_V1: &str = "https://docs.pypi.org/attestations/publish/v1";
/// npm's registry publish attestation.
pub const NPM_PUBLISH_V01: &str = "https://github.com/npm/attestation/tree/main/specs/publish/v0.1";
/// The artifact's digests, lowercase hex.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ArtifactDigests {
pub sha256: String,
pub sha512: String,
}
impl ArtifactDigests {
pub fn of(bytes: &[u8]) -> Self {
use sha2::{Digest, Sha256, Sha512};
Self {
sha256: hex::encode(Sha256::digest(bytes)),
sha512: hex::encode(Sha512::digest(bytes)),
}
}
}
/// A parsed statement.
#[derive(Debug, Clone)]
pub struct Statement {
pub statement_type: String,
pub subjects: Vec<(String, BTreeMap<String, String>)>,
pub predicate_type: String,
pub predicate: Value,
}
impl Statement {
/// Parse a statement payload (v1 or v0.1).
pub fn parse(payload: &[u8]) -> Result<Self> {
let bad = |w: String| ProvenanceError::Malformed(format!("in-toto statement: {w}"));
let v: Value =
serde_json::from_slice(payload).map_err(|e| bad(format!("not JSON: {e}")))?;
let t = v.get("_type").and_then(Value::as_str).unwrap_or("");
if t != STATEMENT_V1 && t != STATEMENT_V01 {
return Err(bad(format!(
"_type {t:?} is not an in-toto Statement v1 or v0.1"
)));
}
let mut subjects = Vec::new();
for s in v
.get("subject")
.and_then(Value::as_array)
.into_iter()
.flatten()
{
let name = s
.get("name")
.and_then(Value::as_str)
.unwrap_or("")
.to_string();
let digest = s
.get("digest")
.and_then(Value::as_object)
.map(|m| {
m.iter()
.filter_map(|(k, v)| {
v.as_str()
.map(|h| (k.to_ascii_lowercase(), h.to_ascii_lowercase()))
})
.collect()
})
.unwrap_or_default();
subjects.push((name, digest));
}
if subjects.is_empty() {
return Err(bad("no subject".into()));
}
let predicate_type = v
.get("predicateType")
.and_then(Value::as_str)
.ok_or_else(|| bad("no predicateType".into()))?
.to_string();
Ok(Self {
statement_type: t.to_string(),
subjects,
predicate_type,
predicate: v.get("predicate").cloned().unwrap_or(Value::Null),
})
}
/// Does a subject name these bytes (its sha256, or its sha512)?
pub fn names(&self, d: &ArtifactDigests) -> bool {
self.subjects.iter().any(|(_, m)| {
m.get("sha256").is_some_and(|h| *h == d.sha256)
|| m.get("sha512").is_some_and(|h| *h == d.sha512)
})
}
/// The subjects, for a refusal.
pub fn describe_subjects(&self) -> Vec<String> {
self.subjects
.iter()
.map(|(n, m)| {
let ds: Vec<String> = m.iter().map(|(a, h)| format!("{a}:{h}")).collect();
format!("{n} [{}]", ds.join(", "))
})
.collect()
}
/// Is this a provenance statement (SLSA v1 or v0.2, or a PyPI publish attestation)?
pub fn is_provenance(&self) -> bool {
matches!(
self.predicate_type.as_str(),
SLSA_PROVENANCE_V1 | SLSA_PROVENANCE_V02 | PYPI_PUBLISH_V1
)
}
/// The SLSA provenance, read into the evidence.
pub fn slsa(&self) -> Option<SlsaSummary> {
let p = &self.predicate;
let s = |v: Option<&Value>| v.and_then(Value::as_str).unwrap_or("").to_string();
match self.predicate_type.as_str() {
SLSA_PROVENANCE_V1 => Some(SlsaSummary {
predicate_type: self.predicate_type.clone(),
builder_id: s(p.pointer("/runDetails/builder/id")),
build_type: s(p.pointer("/buildDefinition/buildType")),
invocation_id: s(p.pointer("/runDetails/metadata/invocationId")),
external_parameters: p
.pointer("/buildDefinition/externalParameters")
.cloned()
.unwrap_or(Value::Null),
internal_parameters: p
.pointer("/buildDefinition/internalParameters")
.cloned()
.unwrap_or(Value::Null),
resolved_dependencies: p
.pointer("/buildDefinition/resolvedDependencies")
.and_then(Value::as_array)
.cloned()
.unwrap_or_default(),
}),
SLSA_PROVENANCE_V02 => Some(SlsaSummary {
predicate_type: self.predicate_type.clone(),
builder_id: s(p.pointer("/builder/id")),
build_type: s(p.get("buildType")),
invocation_id: s(p.pointer("/metadata/buildInvocationId")),
external_parameters: p.get("invocation").cloned().unwrap_or(Value::Null),
internal_parameters: Value::Null,
resolved_dependencies: p
.get("materials")
.and_then(Value::as_array)
.cloned()
.unwrap_or_default(),
}),
_ => None,
}
}
}
/// What the evidence keeps of a SLSA provenance predicate.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct SlsaSummary {
pub predicate_type: String,
pub builder_id: String,
pub build_type: String,
/// The invocation id (v1 `runDetails.metadata.invocationId`).
#[serde(default, skip_serializing_if = "String::is_empty")]
pub invocation_id: String,
/// v1 `externalParameters`, or v0.2 `invocation`.
#[serde(default, skip_serializing_if = "Value::is_null")]
pub external_parameters: Value,
#[serde(default, skip_serializing_if = "Value::is_null")]
pub internal_parameters: Value,
/// v1 `resolvedDependencies`, or v0.2 `materials`.
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub resolved_dependencies: Vec<Value>,
}