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

nordisk/holger

provenance/tests/common/mod.rs

//! A Sigstore in miniature, minted in the test: a Fulcio-like CA (P-384 root
//! issuing P-256 leaves with the Fulcio identity extensions and an embedded SCT),
//! a Rekor-like log (P-256 key, SETs, RFC 6962 inclusion proofs and signed
//! checkpoints) and a CT log. Plus plain signing keys. No key is committed.

#![allow(dead_code)]

use holger_provenance::{crypto, der, tlog};
use rcgen::{
    BasicConstraints, CertificateParams, CustomExtension, DnType, ExtendedKeyUsagePurpose, IsCa,
    Issuer, KeyPair, PublicKeyData, SanType, SerialNumber, SigningKey, PKCS_ECDSA_P256_SHA256,
    PKCS_ECDSA_P384_SHA384, PKCS_ED25519,
};
use serde_json::{json, Value};

pub const IN_TOTO: &str = "application/vnd.in-toto+json";

/// Unix time of 2026-10-23T12:00:00Z; the minted leaves are valid that whole day.
pub const SIGNED_AT: i64 = 1_792_756_800;

pub fn b64(b: &[u8]) -> String {
    crypto::b64_encode(b)
}

pub fn spki(k: &KeyPair) -> Vec<u8> {
    k.subject_public_key_info()
}

pub fn pem_of_der(label: &str, der: &[u8]) -> String {
    let body = b64(der);
    let lines: Vec<&str> = body
        .as_bytes()
        .chunks(64)
        .map(|c| std::str::from_utf8(c).unwrap())
        .collect();
    format!(
        "-----BEGIN {label}-----\n{}\n-----END {label}-----\n",
        lines.join("\n")
    )
}

/// An in-toto statement about `bytes`.
pub fn statement(name: &str, bytes: &[u8], predicate_type: &str, predicate: Value) -> Vec<u8> {
    serde_json::to_vec(&json!({
        "_type": "https://in-toto.io/Statement/v1",
        "subject": [{"name": name, "digest": {"sha256": crypto::sha256_hex(bytes)}}],
        "predicateType": predicate_type,
        "predicate": predicate,
    }))
    .unwrap()
}

/// A SLSA v1 predicate.
pub fn slsa(builder: &str) -> Value {
    json!({
        "buildDefinition": {
            "buildType": "https://holger.rs/demo-build/v1",
            "externalParameters": {"repository": "https://git.gunnar.rs/nordisk/holger", "ref": "refs/heads/main"},
            "resolvedDependencies": [{"uri": "git+https://git.gunnar.rs/nordisk/holger@refs/heads/main", "digest": {"gitCommit": "cc40724c"}}]
        },
        "runDetails": {"builder": {"id": builder}, "metadata": {"invocationId": "demo-run-1"}}
    })
}

pub fn pae(payload: &[u8]) -> Vec<u8> {
    znippy_common::sign::dsse_pae(IN_TOTO, payload)
}

/// A plain signing key with a DSSE keyid.
pub struct Key {
    pub id: String,
    pub pair: KeyPair,
    pub alg: holger_provenance::KeyAlg,
}

impl Key {
    pub fn ed25519(id: &str) -> Self {
        Self {
            id: id.into(),
            pair: KeyPair::generate_for(&PKCS_ED25519).unwrap(),
            alg: holger_provenance::KeyAlg::Ed25519,
        }
    }
    pub fn p256(id: &str) -> Self {
        Self {
            id: id.into(),
            pair: KeyPair::generate_for(&PKCS_ECDSA_P256_SHA256).unwrap(),
            alg: holger_provenance::KeyAlg::EcdsaP256,
        }
    }
    pub fn trusted(&self) -> holger_provenance::TrustedKey {
        holger_provenance::TrustedKey {
            id: self.id.clone(),
            alg: self.alg,
            public_key: self.pair.public_key_pem(),
        }
    }
    pub fn sign(&self, msg: &[u8]) -> Vec<u8> {
        self.pair.sign(msg).unwrap()
    }
    /// A DSSE envelope over `payload`, keyid = this key's id.
    pub fn envelope(&self, payload: &[u8]) -> Vec<u8> {
        serde_json::to_vec(&json!({
            "payloadType": IN_TOTO,
            "payload": b64(payload),
            "signatures": [{"keyid": self.id, "sig": b64(&self.sign(&pae(payload)))}],
        }))
        .unwrap()
    }
}

/// RFC 6962 Merkle tree hash.
fn mth(leaves: &[Vec<u8>]) -> Vec<u8> {
    use sha2::{Digest, Sha256};
    match leaves.len() {
        0 => Sha256::digest([]).to_vec(),
        1 => leaves[0].clone(),
        n => {
            let k = n.next_power_of_two() / 2;
            let k = if k == n { n / 2 } else { k };
            let mut d = Sha256::new();
            d.update([1u8]);
            d.update(mth(&leaves[..k]));
            d.update(mth(&leaves[k..]));
            d.finalize().to_vec()
        }
    }
}

/// RFC 6962 audit path for leaf `m`.
fn path(m: usize, leaves: &[Vec<u8>]) -> Vec<Vec<u8>> {
    let n = leaves.len();
    if n <= 1 {
        return vec![];
    }
    let k = n.next_power_of_two() / 2;
    let k = if k == n { n / 2 } else { k };
    if m < k {
        let mut p = path(m, &leaves[..k]);
        p.push(mth(&leaves[k..]));
        p
    } else {
        let mut p = path(m - k, &leaves[k..]);
        p.push(mth(&leaves[..k]));
        p
    }
}

pub struct MiniSigstore {
    pub root_key: KeyPair,
    pub root_params: CertificateParams,
    pub root_der: Vec<u8>,
    pub rekor: KeyPair,
    pub ct: KeyPair,
}

impl MiniSigstore {
    pub fn new() -> Self {
        let root_key = KeyPair::generate_for(&PKCS_ECDSA_P384_SHA384).unwrap();
        let mut root_params = CertificateParams::new(vec![]).unwrap();
        root_params
            .distinguished_name
            .push(DnType::OrganizationName, "sigstore.test");
        root_params
            .distinguished_name
            .push(DnType::CommonName, "mini-fulcio");
        root_params.is_ca = IsCa::Ca(BasicConstraints::Unconstrained);
        root_params.not_before = rcgen::date_time_ymd(2020, 1, 1);
        root_params.not_after = rcgen::date_time_ymd(2040, 1, 1);
        let root_der = root_params.self_signed(&root_key).unwrap().der().to_vec();
        Self {
            root_key,
            root_params,
            root_der,
            rekor: KeyPair::generate_for(&PKCS_ECDSA_P256_SHA256).unwrap(),
            ct: KeyPair::generate_for(&PKCS_ECDSA_P256_SHA256).unwrap(),
        }
    }

    pub fn rekor_id(&self) -> Vec<u8> {
        crypto::key_id(&spki(&self.rekor)).to_vec()
    }

    /// The `trusted_root.json` that pins this mini Sigstore.
    pub fn trusted_root_json(&self) -> Vec<u8> {
        let key = |k: &KeyPair, url: &str| {
            json!({
                "baseUrl": url,
                "hashAlgorithm": "SHA2_256",
                "publicKey": {"rawBytes": b64(&spki(k)), "keyDetails": "PKIX_ECDSA_P256_SHA_256", "validFor": {"start": "2020-01-01T00:00:00Z"}},
                "logId": {"keyId": b64(&crypto::key_id(&spki(k)))}
            })
        };
        serde_json::to_vec(&json!({
            "mediaType": "application/vnd.dev.sigstore.trustedroot+json;version=0.1",
            "tlogs": [key(&self.rekor, "https://rekor.test")],
            "certificateAuthorities": [{
                "subject": {"organization": "sigstore.test", "commonName": "mini-fulcio"},
                "uri": "https://fulcio.test",
                "certChain": {"certificates": [{"rawBytes": b64(&self.root_der)}]},
                "validFor": {"start": "2020-01-01T00:00:00Z"}
            }],
            "ctlogs": [key(&self.ct, "https://ctfe.test")],
            "timestampAuthorities": []
        }))
        .unwrap()
    }

    /// A Fulcio-like leaf for (`issuer`, `subject`), valid on 2026-10-23 (or the
    /// day given), with an SCT from the mini CT log.
    pub fn leaf(&self, issuer: &str, subject: &str, day: (i32, u8, u8)) -> (KeyPair, Vec<u8>) {
        let key = KeyPair::generate_for(&PKCS_ECDSA_P256_SHA256).unwrap();
        let mut params = CertificateParams::new(vec![]).unwrap();
        params.serial_number = Some(SerialNumber::from(0x5157_0001u64));
        params.subject_alt_names = vec![SanType::URI(subject.try_into().unwrap())];
        params.extended_key_usages = vec![ExtendedKeyUsagePurpose::CodeSigning];
        params.not_before = rcgen::date_time_ymd(day.0, day.1, day.2);
        params.not_after = rcgen::date_time_ymd(day.0, day.1, day.2 + 1);
        params.custom_extensions = vec![
            CustomExtension::from_oid_content(
                &[1, 3, 6, 1, 4, 1, 57264, 1, 8],
                der::encode(0x0c, issuer.as_bytes()),
            ),
            // The deprecated raw-string issuer Fulcio still writes beside it; cosign 2 reads
            // only this one (measured with cosign 2.4.3: "with issuer " empty without it).
            CustomExtension::from_oid_content(
                &[1, 3, 6, 1, 4, 1, 57264, 1, 1],
                issuer.as_bytes().to_vec(),
            ),
        ];
        let ca = Issuer::from_params(&self.root_params, &self.root_key);
        let precert = params.signed_by(&key, &ca).unwrap().der().to_vec();
        // The CT log signs the TBS without the SCT extension; add the SCT and sign again.
        let ts: u64 = (SIGNED_AT as u64 - 60) * 1000;
        let sct_tmpl = holger_provenance::cert::Sct {
            log_id: crypto::key_id(&spki(&self.ct)),
            timestamp_ms: ts,
            extensions: vec![],
            hash_alg: 4,
            sig_alg: 3,
            signature: vec![],
        };
        let signed = holger_provenance::cert::sct_signed_bytes(
            &sct_tmpl,
            &precert,
            &self.root_key.subject_public_key_info(),
        )
        .unwrap();
        let sig = self.ct.sign(&signed).unwrap();
        let mut one = vec![0u8];
        one.extend_from_slice(&sct_tmpl.log_id);
        one.extend_from_slice(&ts.to_be_bytes());
        one.extend_from_slice(&[0, 0, 4, 3]);
        one.extend_from_slice(&(sig.len() as u16).to_be_bytes());
        one.extend_from_slice(&sig);
        let mut list = (one.len() as u16).to_be_bytes().to_vec();
        list.extend_from_slice(&one);
        let mut tls = (list.len() as u16).to_be_bytes().to_vec();
        tls.extend_from_slice(&list);
        params
            .custom_extensions
            .push(CustomExtension::from_oid_content(
                &[1, 3, 6, 1, 4, 1, 11129, 2, 4, 2],
                der::encode(0x04, &tls),
            ));
        let der = params.signed_by(&key, &ca).unwrap().der().to_vec();
        (key, der)
    }

    /// A Rekor `dsse` entry for this signature, with a SET and an inclusion proof
    /// (leaf 2 of a 5-leaf tree) under a signed checkpoint, in bundle JSON.
    pub fn tlog_entry(
        &self,
        payload: &[u8],
        sig: &[u8],
        cert_der: &[u8],
        integrated_time: i64,
    ) -> Value {
        let body = serde_json::to_vec(&json!({
            "apiVersion": "0.0.1", "kind": "dsse",
            "spec": {
                "payloadHash": {"algorithm": "sha256", "value": crypto::sha256_hex(payload)},
                "signatures": [{"signature": b64(sig), "verifier": b64(pem_of_der("CERTIFICATE", cert_der).as_bytes())}]
            }
        }))
        .unwrap();
        let mut leaves: Vec<Vec<u8>> = (0..5u8).map(|i| tlog::leaf_hash(&[i])).collect();
        leaves[2] = tlog::leaf_hash(&body);
        let root = mth(&leaves);
        let hashes: Vec<String> = path(2, &leaves).iter().map(|h| b64(h)).collect();
        let note = format!("rekor.test - 1\n5\n{}\n", b64(&root));
        let mut nsig = self.rekor_id()[..4].to_vec();
        nsig.extend(self.rekor.sign(note.as_bytes()).unwrap());
        let checkpoint = format!("{note}\n\u{2014} rekor.test {}\n", b64(&nsig));
        let entry = holger_provenance::material::TlogEntry {
            log_index: 4242,
            log_id: self.rekor_id(),
            integrated_time,
            kind: "dsse".into(),
            version: "0.0.1".into(),
            body: body.clone(),
            set: None,
            proof: None,
        };
        let set = self.rekor.sign(&tlog::set_payload(&entry)).unwrap();
        json!({
            "logIndex": "4242",
            "logId": {"keyId": b64(&self.rekor_id())},
            "kindVersion": {"kind": "dsse", "version": "0.0.1"},
            "integratedTime": integrated_time.to_string(),
            "inclusionPromise": {"signedEntryTimestamp": b64(&set)},
            "inclusionProof": {"logIndex": "2", "rootHash": b64(&root), "treeSize": "5", "hashes": hashes, "checkpoint": {"envelope": checkpoint}},
            "canonicalizedBody": b64(&body)
        })
    }

    /// A Sigstore bundle v0.3: a DSSE envelope over `payload`, signed keyless as
    /// (`issuer`, `subject`), logged at `integrated_time`.
    pub fn bundle(
        &self,
        payload: &[u8],
        issuer: &str,
        subject: &str,
        integrated_time: i64,
    ) -> Vec<u8> {
        let (key, cert) = self.leaf(issuer, subject, (2026, 10, 23));
        let sig = key.sign(&pae(payload)).unwrap();
        let entry = self.tlog_entry(payload, &sig, &cert, integrated_time);
        serde_json::to_vec(&json!({
            "mediaType": "application/vnd.dev.sigstore.bundle.v0.3+json",
            "verificationMaterial": {"certificate": {"rawBytes": b64(&cert)}, "tlogEntries": [entry]},
            "dsseEnvelope": {"payload": b64(payload), "payloadType": IN_TOTO, "signatures": [{"sig": b64(&sig), "keyid": ""}]}
        }))
        .unwrap()
    }

    /// A PEP 740 provenance object, as PyPI's Integrity API (or Chainguard's
    /// Python index) answers it, with the publisher block Chainguard documents.
    pub fn pep740(&self, statement: &[u8], issuer: &str, subject: &str) -> Vec<u8> {
        let (key, cert) = self.leaf(issuer, subject, (2026, 10, 23));
        let sig = key.sign(&pae(statement)).unwrap();
        let entry = self.tlog_entry(statement, &sig, &cert, SIGNED_AT);
        serde_json::to_vec(&json!({
            "version": 1,
            "attestation_bundles": [{
                "publisher": {"kind": "URL", "environment": "chainguard", "issuer": issuer, "identity": subject,
                              "repository": "chainguard-dev/ecosystems-wheel-rebuilder", "workflow": "python-build-versions.yaml"},
                "attestations": [{
                    "version": 1,
                    "verification_material": {"certificate": b64(&cert), "transparency_entries": [entry]},
                    "envelope": {"statement": b64(statement), "signature": b64(&sig)}
                }]
            }]
        }))
        .unwrap()
    }
}