Gunnar

nordisk/heimdall

guests/verify-bundle/tests/dsse.rs

// SPDX-License-Identifier: AGPL-3.0-only
//! DSSE envelopes inside a bundle: verified BEFORE the statement is parsed,
//! signed by a pinned signer, bound to a payload of this very bundle.
//!
//! Envelopes are made by znippy's own `dsse_sign` (what Holger's attestation
//! store uses), plus one real SLSA provenance envelope from a dwarves release
//! build that Holger's provenance crate verifies.

mod common;
use common::*;

use base64::Engine as _;
use holger_update_bundle::EvidenceKind;
use rcgen::{CertificateParams, DnType, KeyPair, PKCS_ECDSA_P256_SHA256, PKCS_ED25519};
use znippy_common::sign::{dsse_sign, signer_from_pkcs8, SigAlg};

const DSSE: &str = "heimdall-bundle-4";
const BINDING: &str = "heimdall-bundle-5";
const ENVELOPE_MEDIA: &str = "application/vnd.dsse.envelope.v1+json";
const INTOTO: &str = "application/vnd.in-toto+json";

struct Attester {
    alg: SigAlg,
    key_pkcs8: Vec<u8>,
    cert_der: Vec<u8>,
    keyid: String,
}

fn attester(alg: SigAlg) -> Attester {
    let key = match alg {
        SigAlg::Ed25519 => KeyPair::generate_for(&PKCS_ED25519).unwrap(),
        SigAlg::EcdsaP256 => KeyPair::generate_for(&PKCS_ECDSA_P256_SHA256).unwrap(),
    };
    let mut p = CertificateParams::new(vec![]).unwrap();
    p.distinguished_name.push(DnType::CommonName, "scan-gate");
    let cert = p.self_signed(&key).unwrap();
    let der = cert.der().to_vec();
    Attester {
        alg,
        key_pkcs8: key.serialize_der(),
        keyid: sha256_hex(&der),
        cert_der: der,
    }
}

fn statement_about(sha: &str) -> Vec<u8> {
    serde_json::to_vec(&serde_json::json!({
        "_type": "https://in-toto.io/Statement/v1",
        "subject": [{"name": "firmware.bin", "digest": {"sha256": sha}}],
        "predicateType": "https://holger.rs/scan-evidence/v1",
        "predicate": {"verdict": "clean"}
    }))
    .unwrap()
}

fn envelope(a: &Attester, payload_type: &str, payload: &[u8]) -> Vec<u8> {
    let s = signer_from_pkcs8(a.alg, &a.key_pkcs8, &a.cert_der).unwrap();
    let env = dsse_sign(payload_type, payload, &[s.as_ref()]).unwrap();
    env.to_json().unwrap().into_bytes()
}

fn firmware_sha() -> String {
    sha256_hex(b"firmware-v1")
}

/// A Holger-made bundle carrying `envelopes` (path under evidence/, bytes) and
/// the attester's certificate, as the airgap export lays them out.
fn bundle_with(envelopes: Vec<(&str, Vec<u8>)>, cert: Option<&Attester>) -> Fixture {
    let mut spec = Spec::default();
    for (p, b) in envelopes {
        spec.evidence
            .push((p.into(), b, ENVELOPE_MEDIA.into(), EvidenceKind::Other));
    }
    if let Some(a) = cert {
        spec.evidence.push((
            "attestation-signer.der".into(),
            a.cert_der.clone(),
            "application/pkix-cert".into(),
            EvidenceKind::Other,
        ));
    }
    build(spec)
}

fn pinned(a: &Attester) -> String {
    format!("dsse: (trusted_signers: [\"{}\"])", a.keyid)
}

fn run(fx: &Fixture, extra: &str) -> Outcome {
    verify_bundle(&fx.pack(), &fx.params_with(extra))
}

#[test]
fn an_ed25519_attestation_by_a_pinned_signer_is_verified_and_reported() {
    let a = attester(SigAlg::Ed25519);
    let env = envelope(&a, INTOTO, &statement_about(&firmware_sha()));
    let fx = bundle_with(vec![("attestations/0001.dsse.json", env)], Some(&a));
    let o = run(&fx, &pinned(&a));
    assert!(o.is_accepted(), "{:?}", o.reason());
    let r: serde_json::Value = serde_json::from_slice(&o.output).unwrap();
    assert_eq!(
        r["dsse"][0]["member"],
        "evidence/attestations/0001.dsse.json"
    );
    assert_eq!(
        r["dsse"][0]["predicate_type"],
        "https://holger.rs/scan-evidence/v1"
    );
    assert_eq!(r["dsse"][0]["signers"][0], a.keyid);
    assert_eq!(r["dsse"][0]["subjects_bound"][0], firmware_sha());
}

#[test]
fn a_p256_attestation_by_a_pinned_signer_is_verified() {
    let a = attester(SigAlg::EcdsaP256);
    let env = envelope(&a, INTOTO, &statement_about(&firmware_sha()));
    let fx = bundle_with(vec![("attestations/0001.dsse.json", env)], Some(&a));
    let o = run(&fx, &pinned(&a));
    assert!(o.is_accepted(), "{:?}", o.reason());
}

#[test]
fn a_real_dwarves_slsa_release_provenance_verifies_against_its_pinned_raw_key() {
    let dir = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/slsa-dwarf");
    let jsonl = std::fs::read(dir.join("holger-2.0.0-rebuild.intoto.jsonl")).unwrap();
    let krate = std::fs::read(dir.join("holger-errcode-2.0.0.crate")).unwrap();
    let pem = std::fs::read_to_string(dir.join("release-test-key.pub")).unwrap();
    let mut spec = Spec::default();
    spec.payloads.push((
        "crates/holger-errcode-2.0.0.crate".into(),
        krate,
        "application/gzip".into(),
    ));
    spec.evidence.push((
        "provenance/holger-2.0.0-rebuild.intoto.jsonl".into(),
        jsonl,
        "application/jsonl".into(),
        EvidenceKind::Other,
    ));
    let fx = build(spec);
    let o = run(&fx, &format!("dsse: (keys: [{pem:?}])"));
    assert!(o.is_accepted(), "{:?}", o.reason());
    let r: serde_json::Value = serde_json::from_slice(&o.output).unwrap();
    assert_eq!(
        r["dsse"][0]["predicate_type"],
        "https://slsa.dev/provenance/v1"
    );
    assert_eq!(
        r["dsse"][0]["signers"][0],
        "b7238cc7bd98598c053314a119a98e670bc3ab1243cf3f7bf35817a7e3f4b6a5"
    );
    assert_eq!(
        r["dsse"][0]["subjects_bound"][0],
        "f70dfe827db476c30434625dd8ea2ad8a5581ba1ad0e83b6209ba5ea029b14ba"
    );
    // Without the key pinned, the same real envelope is refused.
    assert_refused(&run(&fx, ""), DSSE, "not a pinned attestation signer");
}

#[test]
fn an_attestation_by_an_unpinned_signer_is_refused() {
    let a = attester(SigAlg::Ed25519);
    let env = envelope(&a, INTOTO, &statement_about(&firmware_sha()));
    let fx = bundle_with(vec![("attestations/0001.dsse.json", env)], Some(&a));
    assert_refused(&run(&fx, ""), DSSE, "not a pinned attestation signer");
    let other = attester(SigAlg::Ed25519);
    assert_refused(
        &run(&fx, &pinned(&other)),
        DSSE,
        "not a pinned attestation signer",
    );
}

#[test]
fn a_wrong_payload_type_is_refused_before_the_payload_is_even_decoded() {
    let a = attester(SigAlg::Ed25519);
    // The payload is not even valid base64-of-JSON material: only the type is read.
    let env = envelope(&a, "application/vnd.evil+json", b"\xff\xfe not json");
    let fx = bundle_with(vec![("attestations/0001.dsse.json", env)], Some(&a));
    assert_refused(&run(&fx, &pinned(&a)), DSSE, "payloadType");
}

#[test]
fn a_flipped_signature_is_refused() {
    let a = attester(SigAlg::Ed25519);
    let env = envelope(&a, INTOTO, &statement_about(&firmware_sha()));
    let mut v: serde_json::Value = serde_json::from_slice(&env).unwrap();
    let b64 = base64::engine::general_purpose::STANDARD;
    let mut sig = b64
        .decode(v["signatures"][0]["sig"].as_str().unwrap())
        .unwrap();
    sig[5] ^= 1;
    v["signatures"][0]["sig"] = b64.encode(sig).into();
    let fx = bundle_with(
        vec![(
            "attestations/0001.dsse.json",
            serde_json::to_vec(&v).unwrap(),
        )],
        Some(&a),
    );
    assert_refused(&run(&fx, &pinned(&a)), DSSE, "does not verify");
}

#[test]
fn the_signature_is_checked_before_the_statement_is_parsed() {
    // Garbage statement AND a broken signature: the refusal is the signature's,
    // so not one byte of the statement was interpreted.
    let a = attester(SigAlg::Ed25519);
    let env = envelope(&a, INTOTO, b"{{{{ this is not json");
    let mut v: serde_json::Value = serde_json::from_slice(&env).unwrap();
    v["payload"] = base64::engine::general_purpose::STANDARD
        .encode(b"{{{{ this is not json either")
        .into();
    let fx = bundle_with(
        vec![(
            "attestations/0001.dsse.json",
            serde_json::to_vec(&v).unwrap(),
        )],
        Some(&a),
    );
    assert_refused(&run(&fx, &pinned(&a)), DSSE, "does not verify");
}

#[test]
fn a_validly_signed_statement_that_is_not_json_is_refused_after_verification() {
    let a = attester(SigAlg::Ed25519);
    let env = envelope(&a, INTOTO, b"{{{{ not json");
    let fx = bundle_with(vec![("attestations/0001.dsse.json", env)], Some(&a));
    assert_refused(&run(&fx, &pinned(&a)), BINDING, "not JSON");
}

#[test]
fn a_signature_without_a_keyid_is_refused() {
    let a = attester(SigAlg::Ed25519);
    let env = envelope(&a, INTOTO, &statement_about(&firmware_sha()));
    let mut v: serde_json::Value = serde_json::from_slice(&env).unwrap();
    v["signatures"][0]["keyid"] = "".into();
    let fx = bundle_with(
        vec![(
            "attestations/0001.dsse.json",
            serde_json::to_vec(&v).unwrap(),
        )],
        Some(&a),
    );
    assert_refused(&run(&fx, &pinned(&a)), DSSE, "no keyid");
}

#[test]
fn an_envelope_with_no_signatures_is_refused() {
    let a = attester(SigAlg::Ed25519);
    let env = envelope(&a, INTOTO, &statement_about(&firmware_sha()));
    let mut v: serde_json::Value = serde_json::from_slice(&env).unwrap();
    v["signatures"] = serde_json::json!([]);
    let fx = bundle_with(
        vec![(
            "attestations/0001.dsse.json",
            serde_json::to_vec(&v).unwrap(),
        )],
        Some(&a),
    );
    assert_refused(&run(&fx, &pinned(&a)), DSSE, "no signature");
}

#[test]
fn an_envelope_with_more_than_sixteen_signatures_is_refused() {
    let a = attester(SigAlg::Ed25519);
    let env = envelope(&a, INTOTO, &statement_about(&firmware_sha()));
    let mut v: serde_json::Value = serde_json::from_slice(&env).unwrap();
    let one = v["signatures"][0].clone();
    v["signatures"] = serde_json::Value::Array(vec![one; 17]);
    let fx = bundle_with(
        vec![(
            "attestations/0001.dsse.json",
            serde_json::to_vec(&v).unwrap(),
        )],
        Some(&a),
    );
    assert_refused(&run(&fx, &pinned(&a)), DSSE, "more than 16");
}

#[test]
fn an_envelope_with_an_unknown_field_is_refused() {
    let a = attester(SigAlg::Ed25519);
    let env = envelope(&a, INTOTO, &statement_about(&firmware_sha()));
    let mut v: serde_json::Value = serde_json::from_slice(&env).unwrap();
    v["payloadTypeOverride"] = "x".into();
    let fx = bundle_with(
        vec![(
            "attestations/0001.dsse.json",
            serde_json::to_vec(&v).unwrap(),
        )],
        Some(&a),
    );
    assert_refused(&run(&fx, &pinned(&a)), DSSE, "envelope JSON");
}

#[test]
fn a_payload_that_is_not_base64_is_refused() {
    let a = attester(SigAlg::Ed25519);
    let env = envelope(&a, INTOTO, &statement_about(&firmware_sha()));
    let mut v: serde_json::Value = serde_json::from_slice(&env).unwrap();
    v["payload"] = "!!!not-base64!!!".into();
    let fx = bundle_with(
        vec![(
            "attestations/0001.dsse.json",
            serde_json::to_vec(&v).unwrap(),
        )],
        Some(&a),
    );
    assert_refused(&run(&fx, &pinned(&a)), DSSE, "not base64");
}

#[test]
fn a_statement_about_some_other_artifact_does_not_bind() {
    let a = attester(SigAlg::Ed25519);
    let env = envelope(
        &a,
        INTOTO,
        &statement_about(&sha256_hex(b"some other file")),
    );
    let fx = bundle_with(vec![("attestations/0001.dsse.json", env)], Some(&a));
    assert_refused(
        &run(&fx, &pinned(&a)),
        BINDING,
        "no subject names a payload",
    );
}

#[test]
fn a_validly_signed_payload_that_is_not_an_in_toto_statement_is_refused() {
    let a = attester(SigAlg::Ed25519);
    let env = envelope(
        &a,
        INTOTO,
        br#"{"_type":"https://evil/Statement","subject":[]}"#,
    );
    let fx = bundle_with(vec![("attestations/0001.dsse.json", env)], Some(&a));
    assert_refused(&run(&fx, &pinned(&a)), BINDING, "not an in-toto Statement");
}

#[test]
fn a_pinned_signer_whose_certificate_is_absent_is_refused() {
    let a = attester(SigAlg::Ed25519);
    let env = envelope(&a, INTOTO, &statement_about(&firmware_sha()));
    let fx = bundle_with(vec![("attestations/0001.dsse.json", env)], None);
    assert_refused(&run(&fx, &pinned(&a)), DSSE, "no certificate for keyid");
    // ...and accepted when the caller supplies the certificate itself.
    let extra = format!(
        "dsse: (trusted_signers: [\"{}\"], certificates: [\"{}\"])",
        a.keyid,
        hex(&a.cert_der)
    );
    assert!(run(&fx, &extra).is_accepted());
}

#[test]
fn an_attestation_certificate_out_of_force_at_now_is_refused() {
    let key = KeyPair::generate_for(&PKCS_ED25519).unwrap();
    let mut p = CertificateParams::new(vec![]).unwrap();
    p.distinguished_name
        .push(DnType::CommonName, "old-scan-gate");
    p.not_before = rcgen::date_time_ymd(2020, 1, 1);
    p.not_after = rcgen::date_time_ymd(2021, 1, 1);
    let der = p.self_signed(&key).unwrap().der().to_vec();
    let a = Attester {
        alg: SigAlg::Ed25519,
        key_pkcs8: key.serialize_der(),
        keyid: sha256_hex(&der),
        cert_der: der,
    };
    let env = envelope(&a, INTOTO, &statement_about(&firmware_sha()));
    let mut fx = bundle_with(vec![("attestations/0001.dsse.json", env)], Some(&a));
    assert_refused(&run(&fx, &pinned(&a)), DSSE, "not in force");
    // The policy's named clock waiver covers attestation certificates too.
    fx.policy.certificate_validity =
        holger_update_bundle::CertificateValidity::IgnoreUnreliableClock;
    assert!(run(&fx, &pinned(&a)).is_accepted());
}

#[test]
fn every_envelope_of_a_jsonl_member_must_verify() {
    let a = attester(SigAlg::Ed25519);
    let good_line = envelope(&a, INTOTO, &statement_about(&firmware_sha()));
    let other = attester(SigAlg::Ed25519);
    let bad_line = envelope(&other, INTOTO, &statement_about(&firmware_sha()));
    let mut both_good = good_line.clone();
    both_good.push(b'\n');
    both_good.extend_from_slice(&good_line);
    let fx = bundle_with(vec![("provenance/rel.intoto.jsonl", both_good)], Some(&a));
    let o = run(&fx, &pinned(&a));
    assert!(o.is_accepted(), "{:?}", o.reason());
    let r: serde_json::Value = serde_json::from_slice(&o.output).unwrap();
    assert_eq!(r["dsse"].as_array().unwrap().len(), 2);

    let mut mixed = good_line.clone();
    mixed.push(b'\n');
    mixed.extend_from_slice(&bad_line);
    let fx = bundle_with(vec![("provenance/rel.intoto.jsonl", mixed)], Some(&a));
    assert_refused(
        &run(&fx, &pinned(&a)),
        DSSE,
        "not a pinned attestation signer",
    );
}

#[test]
fn a_member_recognised_only_by_its_media_type_is_still_verified() {
    let a = attester(SigAlg::Ed25519);
    let other = attester(SigAlg::Ed25519);
    let env = envelope(&other, INTOTO, &statement_about(&firmware_sha()));
    // named like plain JSON, declared as a DSSE envelope in core.json
    let fx = bundle_with(vec![("attestations/0001.json", env)], Some(&a));
    assert_refused(
        &run(&fx, &pinned(&a)),
        DSSE,
        "not a pinned attestation signer",
    );
}

#[test]
fn a_caller_who_requires_an_envelope_refuses_a_bundle_that_has_none() {
    let fx = good();
    assert_refused(
        &run(&fx, "dsse: (require_envelope: true)"),
        DSSE,
        "carries none",
    );
}

#[test]
fn a_pinned_key_that_is_neither_pem_nor_hex_is_a_parameter_refusal() {
    let fx = good();
    assert_refused(
        &run(&fx, "dsse: (keys: [\"nope nope\"])"),
        "heimdall-bundle-1",
        "dsse.keys[0]",
    );
}