Your browser sucks, JavaScript disabled! Gunnar's console needs JavaScript and WebAssembly. Git over SSH on port 2222 does not.

nordisk/holger

provenance/src/policy.rs

//! The trust policy, as the RON writes it.
//!
//! Two layers. The global `rules:` block is the FLOOR ([`ProvenanceFloor`]): it
//! holds for every source. A source's `gates.provenance` ([`ProvenanceGate`]) may
//! only TIGHTEN it. [`effective`] folds the two into the [`EffectivePolicy`] the
//! verifier reads, and [`check_tightens`] refuses, by name, a gate that would
//! loosen the floor.
//!
//! ```ron
//! rules: (
//!     require_signed: true,
//!     require_provenance: false,
//!     slsa_min: 0,
//!     identities: [
//!         (issuer: "https://token.actions.githubusercontent.com",
//!          subject: "https://github.com/chainguard-images/images/.github/workflows/release.yaml@refs/heads/main"),
//!         (issuer: "https://issuer.enforce.dev", subject_regex: "^https://issuer\\.enforce\\.dev/.+$"),
//!     ],
//!     builders: ["https://github.com/chainguard-dev/terraform-provider-apko", (id: "https://github.com/actions/runner/github-hosted", level: 3)],
//!     keys: [(id: "demo-release", alg: ed25519, public_key: "-----BEGIN PUBLIC KEY-----\n...")],
//!     trust_root: Some("/var/lib/holger/sigstore-trust.holger.znippy"),
//! ),
//! sources: [
//!     (name: "chainguard", kind: "oci", url: "https://cgr.dev/chainguard",
//!      gates: (provenance: 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))),
//!     (name: "pypi", kind: "pip", url: "https://pypi.org/simple", gates: (provenance: prefer)),
//! ],
//! ```

use serde::{Deserialize, Deserializer, Serialize};

use crate::error::{ProvenanceError, Result};

/// Deserialize a present field as `Some(T)`, so the RON writes `slsa_min: 2`
/// and not `slsa_min: Some(2)`.
fn some<'de, D: Deserializer<'de>, T: Deserialize<'de>>(
    d: D,
) -> std::result::Result<Option<T>, D::Error> {
    T::deserialize(d).map(Some)
}

/// One allowed keyless signing identity: the certificate's OIDC issuer, and its
/// subject (the SAN: a workflow URI or an e-mail), exactly or by an anchored regex.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Identity {
    pub issuer: String,
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub subject: String,
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub subject_regex: String,
}

impl Identity {
    /// An exact identity.
    pub fn exact(issuer: impl Into<String>, subject: impl Into<String>) -> Self {
        Self {
            issuer: issuer.into(),
            subject: subject.into(),
            subject_regex: String::new(),
        }
    }

    /// An identity whose subject matches `regex` (the whole subject).
    pub fn regex(issuer: impl Into<String>, regex: impl Into<String>) -> Self {
        Self {
            issuer: issuer.into(),
            subject: String::new(),
            subject_regex: regex.into(),
        }
    }

    fn compiled(&self) -> Result<Option<regex::Regex>> {
        if self.subject_regex.is_empty() {
            return Ok(None);
        }
        let anchored = format!(
            "^(?:{})$",
            self.subject_regex
                .trim_start_matches('^')
                .trim_end_matches('$')
        );
        regex::Regex::new(&anchored).map(Some).map_err(|e| {
            ProvenanceError::PolicyInvalid(format!(
                "identity subject_regex {:?}: {e}",
                self.subject_regex
            ))
        })
    }

    /// Refuse an identity that names no subject, or both kinds.
    pub fn validate(&self) -> Result<()> {
        if self.issuer.trim().is_empty() {
            return Err(ProvenanceError::PolicyInvalid(
                "an identity names no issuer".into(),
            ));
        }
        match (self.subject.is_empty(), self.subject_regex.is_empty()) {
            (true, true) => Err(ProvenanceError::PolicyInvalid(format!(
                "the identity for issuer {:?} names no subject; an issuer alone would trust every account it has",
                self.issuer
            ))),
            (false, false) => Err(ProvenanceError::PolicyInvalid(format!(
                "the identity for issuer {:?} names both subject and subject_regex; use one",
                self.issuer
            ))),
            _ => self.compiled().map(|_| ()),
        }
    }

    /// Does a certificate with this issuer and subject match?
    pub fn matches(&self, issuer: &str, subject: &str) -> bool {
        if self.issuer != issuer {
            return false;
        }
        match self.compiled() {
            Ok(Some(re)) => re.is_match(subject),
            Ok(None) => self.subject == subject,
            Err(_) => false,
        }
    }

    /// Is `other` at least as narrow as `self`?
    fn covers(&self, other: &Identity) -> bool {
        if self == other {
            return true;
        }
        self.issuer == other.issuer
            && other.subject_regex.is_empty()
            && self.matches(&other.issuer, &other.subject)
    }
}

impl std::fmt::Display for Identity {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        if self.subject_regex.is_empty() {
            write!(f, "{} / {}", self.issuer, self.subject)
        } else {
            write!(f, "{} / ~{}", self.issuer, self.subject_regex)
        }
    }
}

/// One allowed SLSA builder. The RON writes a bare id, or `(id, level)` to give
/// the builder's SLSA build level (1 to 4).
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct BuilderRule {
    pub id: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub level: Option<u8>,
}

impl<'de> Deserialize<'de> for BuilderRule {
    fn deserialize<D: Deserializer<'de>>(d: D) -> std::result::Result<Self, D::Error> {
        #[derive(Deserialize)]
        #[serde(deny_unknown_fields)]
        struct Full {
            id: String,
            #[serde(default, deserialize_with = "some")]
            level: Option<u8>,
        }
        #[derive(Deserialize)]
        #[serde(untagged)]
        enum Either {
            Id(String),
            Full(Full),
        }
        Ok(match Either::deserialize(d)? {
            Either::Id(id) => BuilderRule { id, level: None },
            Either::Full(f) => BuilderRule {
                id: f.id,
                level: f.level,
            },
        })
    }
}

impl From<&str> for BuilderRule {
    fn from(id: &str) -> Self {
        Self {
            id: id.to_string(),
            level: None,
        }
    }
}

/// The signature algorithm of a configured key.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum KeyAlg {
    Ed25519,
    EcdsaP256,
    /// An OpenPGP public key (an armored `PUBLIC KEY BLOCK` in `public_key`); it
    /// verifies detached `.asc` signatures (Maven Central), never DSSE.
    Openpgp,
}

/// One allowed signing key: a DSSE `keyid` and the public key, as a PEM
/// `PUBLIC KEY` block or the bare base64 of its SubjectPublicKeyInfo DER.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct TrustedKey {
    pub id: String,
    pub alg: KeyAlg,
    pub public_key: String,
}

/// The provenance half of the global `rules:` floor. Every source gets at least this.
#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ProvenanceFloor {
    /// Nothing unsigned comes in, from any source.
    #[serde(default)]
    pub require_signed: bool,
    /// Nothing without verified provenance comes in, from any source.
    #[serde(default)]
    pub require_provenance: bool,
    /// The least SLSA build level provenance must show (0 = no minimum).
    #[serde(default)]
    pub slsa_min: u8,
    /// The keyless identities every source may be signed by.
    #[serde(default)]
    pub identities: Vec<Identity>,
    /// The SLSA builders every source may be built by. Empty = any builder.
    #[serde(default)]
    pub builders: Vec<BuilderRule>,
    /// The signing keys every source may be signed with.
    #[serde(default)]
    pub keys: Vec<TrustedKey>,
    /// The signed Sigstore trust bundle (`holger.sigstore-trust/v1`), for keyless.
    #[serde(default)]
    pub trust_root: Option<String>,
}

/// A source's `gates.provenance`.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
// Strict, as every tier 1 block is (G0): a typo such as `slsa: 2` is refused by name.
#[serde(rename_all = "snake_case", deny_unknown_fields)]
#[derive(Default)]
pub enum ProvenanceGate {
    /// No valid provenance, refused by name.
    Require {
        #[serde(
            default,
            deserialize_with = "some",
            skip_serializing_if = "Option::is_none"
        )]
        identity: Option<Identity>,
        #[serde(default, skip_serializing_if = "Vec::is_empty")]
        identities: Vec<Identity>,
        #[serde(default, skip_serializing_if = "Vec::is_empty")]
        builders: Vec<BuilderRule>,
        /// Key ids from the floor's `keys`.
        #[serde(default, skip_serializing_if = "Vec::is_empty")]
        keys: Vec<String>,
        /// APK signing keys (names the trust bundle pins, `wolfi-signing.rsa.pub`)
        /// that may sign this source's `APKINDEX.tar.gz`. Empty = any pinned apk key.
        #[serde(default, skip_serializing_if = "Vec::is_empty")]
        apk_keys: Vec<String>,
        /// OpenPGP keys (names the trust bundle pins as `pgp-keys/<name>`) that may
        /// sign this source's `.asc` files. Empty = any pinned pgp key.
        #[serde(default, skip_serializing_if = "Vec::is_empty")]
        pgp_keys: Vec<String>,
        #[serde(
            default,
            deserialize_with = "some",
            skip_serializing_if = "Option::is_none"
        )]
        slsa_min: Option<u8>,
    },
    /// Checked, recorded, and warned about when it is missing or does not verify.
    Prefer,
    /// Not checked for this source (the floor still holds).
    #[default]
    Off,
}

/// How strictly one property is demanded.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Mode {
    Off,
    Prefer,
    Require,
}

impl Mode {
    pub fn as_str(self) -> &'static str {
        match self {
            Mode::Off => "off",
            Mode::Prefer => "prefer",
            Mode::Require => "require",
        }
    }
}

/// The policy the verifier reads for one artifact: the floor plus its source's gate.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct EffectivePolicy {
    /// How strictly verified provenance is demanded.
    pub provenance: Mode,
    /// How strictly a verified signature (of any kind) is demanded.
    pub signed: Mode,
    pub slsa_min: u8,
    pub identities: Vec<Identity>,
    pub builders: Vec<BuilderRule>,
    pub keys: Vec<TrustedKey>,
    /// The APK keys (by name) an APK index may be signed with; empty = any key the
    /// trust bundle pins.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub apk_keys: Vec<String>,
    /// The OpenPGP keys (by name) the trust bundle pins that an `.asc` may be
    /// signed with; empty = any pinned pgp key. Floor `keys` of `alg: openpgp`
    /// (in `keys`) count as well.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub pgp_keys: Vec<String>,
    pub trust_root: Option<String>,
}

impl EffectivePolicy {
    /// Does anything need checking at all?
    pub fn is_off(&self) -> bool {
        self.provenance == Mode::Off && self.signed == Mode::Off
    }

    /// One line for the evidence record.
    pub fn summary(&self) -> String {
        format!(
            "provenance {}, signed {}, slsa_min {}, {} identit{}, {} builder(s), {} key(s)",
            self.provenance.as_str(),
            self.signed.as_str(),
            self.slsa_min,
            self.identities.len(),
            if self.identities.len() == 1 {
                "y"
            } else {
                "ies"
            },
            self.builders.len(),
            self.keys.len()
        )
    }
}

fn gate_lists(gate: &ProvenanceGate) -> (Vec<Identity>, Vec<BuilderRule>, Vec<String>, Option<u8>) {
    match gate {
        ProvenanceGate::Require {
            identity,
            identities,
            builders,
            keys,
            slsa_min,
            ..
        } => {
            let mut ids: Vec<Identity> = identity.iter().cloned().collect();
            ids.extend(identities.iter().cloned());
            (ids, builders.clone(), keys.clone(), *slsa_min)
        }
        _ => (Vec::new(), Vec::new(), Vec::new(), None),
    }
}

fn gate_apk_keys(gate: &ProvenanceGate) -> Vec<String> {
    match gate {
        ProvenanceGate::Require { apk_keys, .. } => apk_keys.clone(),
        _ => Vec::new(),
    }
}

fn gate_pgp_keys(gate: &ProvenanceGate) -> Vec<String> {
    match gate {
        ProvenanceGate::Require { pgp_keys, .. } => pgp_keys.clone(),
        _ => Vec::new(),
    }
}

/// ★ Refuse a floor that cannot mean what it says, by name, before any source is read:
/// an identity that does not validate, or a key of `alg: openpgp` that does not read
/// (in a build without the `pgp` feature, every such key: `[provenance-pgp-not-built]`).
pub fn check_floor(floor: &ProvenanceFloor) -> Result<()> {
    for i in &floor.identities {
        i.validate()?;
    }
    for k in floor.keys.iter().filter(|k| k.alg == KeyAlg::Openpgp) {
        crate::pgp::PgpKey::from_armored(&k.id, k.public_key.as_bytes()).map_err(|e| {
            ProvenanceError::PolicyInvalid(format!("floor key {:?} (alg openpgp): {e}", k.id))
        })?;
    }
    Ok(())
}

/// ★ Refuse a source gate that would loosen the floor, by name.
///
/// * `off` or `prefer` while the floor says `require_provenance: true`;
/// * a lower `slsa_min` than the floor's;
/// * an identity, builder or key the floor's non-empty list does not cover;
/// * a key id the floor does not define;
/// * a `require` with no identity and no key at all, which could never pass;
/// * a `require(pgp_keys: [...])` in a build without the `pgp` feature, and anything
///   [`check_floor`] refuses.
pub fn check_tightens(floor: &ProvenanceFloor, gate: &ProvenanceGate, source: &str) -> Result<()> {
    check_floor(floor)?;
    let invalid = |why: String| {
        Err(ProvenanceError::PolicyInvalid(format!(
            "source {source:?}: {why}"
        )))
    };
    if !crate::pgp::BUILT && !gate_pgp_keys(gate).is_empty() {
        return invalid(crate::pgp::not_built("`require(pgp_keys: [...])`"));
    }
    if floor.require_provenance && !matches!(gate, ProvenanceGate::Require { .. }) {
        return invalid(format!(
            "`gates.provenance: {}` would loosen the floor `require_provenance: true`; a source may only tighten it",
            match gate {
                ProvenanceGate::Prefer => "prefer",
                _ => "off",
            }
        ));
    }
    let (ids, builders, keys, slsa) = gate_lists(gate);
    if let Some(s) = slsa {
        if s < floor.slsa_min {
            return invalid(format!(
                "slsa_min {s} is below the floor's {}",
                floor.slsa_min
            ));
        }
        if s > 4 {
            return invalid(format!("slsa_min {s} is not a SLSA build level (0 to 4)"));
        }
    }
    for i in &ids {
        i.validate()?;
        if !floor.identities.is_empty() && !floor.identities.iter().any(|f| f.covers(i)) {
            return invalid(format!(
                "identity {i} is not one the floor's `identities` allow"
            ));
        }
    }
    for b in &builders {
        if !floor.builders.is_empty() && !floor.builders.iter().any(|f| f.id == b.id) {
            return invalid(format!(
                "builder {:?} is not one the floor's `builders` allow",
                b.id
            ));
        }
    }
    for k in &keys {
        if !floor.keys.iter().any(|f| &f.id == k) {
            return invalid(format!("key {k:?} is not defined in the floor's `keys`"));
        }
    }
    if let ProvenanceGate::Require { .. } = gate {
        let eff = effective(floor, gate);
        if eff.identities.is_empty()
            && eff.keys.is_empty()
            && eff.apk_keys.is_empty()
            && eff.pgp_keys.is_empty()
        {
            return invalid(
                "`require` with no identity and no key could never pass: name the identity, the key, the apk key or the pgp key that signs this source"
                    .into(),
            );
        }
    }
    Ok(())
}

/// ★ The floor plus a source's gate. Never looser than the floor, whatever the gate says.
pub fn effective(floor: &ProvenanceFloor, gate: &ProvenanceGate) -> EffectivePolicy {
    let gate_mode = match gate {
        ProvenanceGate::Require { .. } => Mode::Require,
        ProvenanceGate::Prefer => Mode::Prefer,
        ProvenanceGate::Off => Mode::Off,
    };
    let floor_mode = if floor.require_provenance {
        Mode::Require
    } else {
        Mode::Off
    };
    let provenance = gate_mode.max(floor_mode);
    let signed = if floor.require_signed || provenance == Mode::Require {
        Mode::Require
    } else {
        provenance
    };
    let (ids, builders, keys, slsa) = gate_lists(gate);
    let identities = if ids.is_empty() {
        floor.identities.clone()
    } else {
        ids
    };
    let builders = if builders.is_empty() {
        floor.builders.clone()
    } else {
        // The gate names a subset; the floor's level for a builder stands unless the gate gives one.
        builders
            .into_iter()
            .map(|b| BuilderRule {
                level: b.level.or_else(|| {
                    floor
                        .builders
                        .iter()
                        .find(|f| f.id == b.id)
                        .and_then(|f| f.level)
                }),
                id: b.id,
            })
            .collect()
    };
    let keys = if keys.is_empty() {
        floor.keys.clone()
    } else {
        floor
            .keys
            .iter()
            .filter(|k| keys.contains(&k.id))
            .cloned()
            .collect()
    };
    EffectivePolicy {
        provenance,
        signed,
        slsa_min: slsa.unwrap_or(0).max(floor.slsa_min),
        identities,
        builders,
        keys,
        apk_keys: gate_apk_keys(gate),
        pgp_keys: gate_pgp_keys(gate),
        trust_root: floor.trust_root.clone(),
    }
}