intel/tests/common/mod.rs
//! Shared test kit: in-process PKI, the fixture feeds, a fixture wheel, audit
//! doubles and a fixed clock. No test here opens a socket.
#![allow(dead_code)]
use std::cell::RefCell;
use std::path::{Path, PathBuf};
use chrono::{DateTime, Duration, Utc};
use holger_intel::feeds::{normalize, FeedSpec, RawFeed};
use holger_intel::{
build_bundle, AdvisorySet, BuildOptions, BuiltBundle, ImportAudit, ImportRecord, SignerFiles,
};
use holger_update_bundle::digest::sha256_hex;
use holger_update_bundle::policy::{
ApprovedSigner, CertificateValidity, ExpiredPosture, TrustMode, TrustPolicy,
};
use holger_update_bundle::schema::POLICY_SCHEMA;
use holger_update_bundle::time::{TimeError, TrustedTime};
use rcgen::{CertificateParams, DnType, KeyPair, PKCS_ED25519};
pub const FIXTURES: &str = concat!(env!("CARGO_MANIFEST_DIR"), "/tests/fixtures");
pub struct Signer {
pub files: SignerFiles,
pub fingerprint: String,
}
pub fn mint(cn: &str) -> Signer {
let key = KeyPair::generate_for(&PKCS_ED25519).expect("ed25519 key");
let mut params = CertificateParams::new(vec![]).expect("params");
params.distinguished_name.push(DnType::CommonName, cn);
let cert = params.self_signed(&key).expect("self-sign");
let der = cert.der().to_vec();
Signer {
fingerprint: sha256_hex(&der),
files: SignerFiles {
cert_der: der,
key_pkcs8: key.serialize_der(),
},
}
}
fn copy(s: &Signer) -> SignerFiles {
SignerFiles {
cert_der: s.files.cert_der.clone(),
key_pkcs8: s.files.key_pkcs8.clone(),
}
}
/// The intel publisher: two approvers and a distinct release signer, pinned.
pub struct Pki {
pub reviewer: Signer,
pub owner: Signer,
pub release: Signer,
pub policy: TrustPolicy,
}
impl Pki {
pub fn new() -> Self {
let reviewer = mint("intel-reviewer");
let owner = mint("intel-owner");
let release = mint("intel-release");
let policy = TrustPolicy {
schema_version: POLICY_SCHEMA.to_string(),
trust_mode: Some(TrustMode::PinnedSigners),
trusted_roots: vec![],
certificate_validity: CertificateValidity::Enforce,
approved_signers: vec![
ApprovedSigner {
fingerprint: reviewer.fingerprint.clone(),
subject: None,
roles: vec!["security-reviewer".into()],
},
ApprovedSigner {
fingerprint: owner.fingerprint.clone(),
subject: None,
roles: vec!["system-owner".into()],
},
],
required_approval_roles: vec!["security-reviewer".into(), "system-owner".into()],
minimum_distinct_approvers: 2,
trusted_release_signers: vec![release.fingerprint.clone()],
release_signer_must_be_distinct: true,
maximum_bundle_age_secs: None,
require_expiry: true,
on_expired_metadata: Some(ExpiredPosture::Refuse),
require_sbom: false,
require_vex: false,
require_scan: false,
allowed_products: vec![holger_intel::INTEL_PRODUCT.into()],
allowed_models: vec![],
};
Self {
reviewer,
owner,
release,
policy,
}
}
pub fn options(&self, sequence: u64, created_at: DateTime<Utc>, out: PathBuf) -> BuildOptions {
BuildOptions {
sequence,
created_at,
expires_at: Some(Utc::now() + Duration::days(365)),
supplier: holger_intel::DEFAULT_SUPPLIER.into(),
producer: "intel-test".into(),
policy: self.policy.clone(),
approvals: vec![
("security-reviewer".into(), copy(&self.reviewer)),
("system-owner".into(), copy(&self.owner)),
],
release: copy(&self.release),
out,
}
}
}
/// The fixture feeds in a directory, named the way `FeedSpec::from_dir`
/// reads them; the OSV records go in as a real `all.zip` dump.
pub fn feeds_dir(dir: &Path) -> PathBuf {
let feeds = dir.join("feeds");
std::fs::create_dir_all(&feeds).unwrap();
let rec = std::fs::read(Path::new(FIXTURES).join("PYSEC-2018-28.json")).unwrap();
let other = std::fs::read(Path::new(FIXTURES).join("GHSA-unrelated.json")).unwrap();
let zip = zip_entry::fixture::zip(&[
("PYSEC-2018-28.json", &rec),
("GHSA-0000-unrelated.json", &other),
]);
std::fs::write(feeds.join("osv-PyPI.zip"), zip).unwrap();
std::fs::copy(
Path::new(FIXTURES).join("nvd-recent.json"),
feeds.join("nvd-recent.json"),
)
.unwrap();
std::fs::copy(Path::new(FIXTURES).join("kev.json"), feeds.join("kev.json")).unwrap();
// H32-TYPOSQUAT: the popular PyPI names the look-alike check reads.
std::fs::copy(
Path::new(FIXTURES).join("popular-pypi.json"),
feeds.join("popular-pypi.json"),
)
.unwrap();
feeds
}
pub fn fixture_set(dir: &Path, generated_at: DateTime<Utc>) -> AdvisorySet {
let specs = FeedSpec::from_dir(&feeds_dir(dir)).unwrap();
let raw: Vec<RawFeed> = specs
.into_iter()
.map(|s| RawFeed::read_file(s).unwrap())
.collect();
normalize(&raw, generated_at).unwrap()
}
/// Build a signed intel bundle at `sequence`, snapshot dated `created_at`.
pub fn bundle(pki: &Pki, dir: &Path, sequence: u64, created_at: DateTime<Utc>) -> BuiltBundle {
let set = fixture_set(dir, created_at);
let out = dir.join(format!("intel-{sequence}.holger.znippy"));
build_bundle(&set, pki.options(sequence, created_at, out)).unwrap()
}
/// A wheel for requests 2.19.0, which CVE-2018-18074 affects.
pub fn fixture_wheel(dir: &Path) -> PathBuf {
let metadata = b"Metadata-Version: 2.1\nName: requests\nVersion: 2.19.0\nLicense: Apache-2.0\n";
let wheel = zip_entry::fixture::zip(&[
("requests-2.19.0.dist-info/METADATA", metadata),
("requests/__init__.py", b"__version__ = '2.19.0'\n"),
]);
let p = dir.join("requests-2.19.0-py2.py3-none-any.whl");
std::fs::write(&p, wheel).unwrap();
p
}
/// `(name, version)` from a wheel's own METADATA.
pub fn wheel_coords(wheel: &[u8]) -> (String, String) {
let (_, meta) = zip_entry::find_member(wheel, |n| n.ends_with(".dist-info/METADATA"))
.unwrap()
.unwrap();
let text = String::from_utf8(meta).unwrap();
let field = |k: &str| {
text.lines()
.find_map(|l| l.strip_prefix(k))
.unwrap()
.trim()
.to_string()
};
(field("Name:"), field("Version:"))
}
/// An audit sink that records into memory.
#[derive(Default)]
pub struct MemAudit {
pub records: RefCell<Vec<ImportRecord>>,
}
impl ImportAudit for MemAudit {
fn ready(&self) -> anyhow::Result<()> {
Ok(())
}
fn record(&self, r: &ImportRecord) -> anyhow::Result<()> {
self.records.borrow_mut().push(r.clone());
Ok(())
}
}
/// A sink that opens and then fails to write, like a full volume.
pub struct FailingAudit;
impl ImportAudit for FailingAudit {
fn ready(&self) -> anyhow::Result<()> {
Ok(())
}
fn record(&self, _: &ImportRecord) -> anyhow::Result<()> {
anyhow::bail!("No space left on device")
}
}
/// A sink that is switched off.
pub struct OffAudit;
impl ImportAudit for OffAudit {
fn ready(&self) -> anyhow::Result<()> {
anyhow::bail!("audit is off")
}
fn record(&self, _: &ImportRecord) -> anyhow::Result<()> {
unreachable!("a sink that is not ready is never written")
}
}
/// A clock that is trusted and says `0`.
pub struct Fixed(pub DateTime<Utc>);
impl TrustedTime for Fixed {
fn source_name(&self) -> &'static str {
"test-clock"
}
fn is_trusted(&self) -> bool {
true
}
fn now(&self) -> Result<DateTime<Utc>, TimeError> {
Ok(self.0)
}
}