server/cli/tests/intel_offline.rs
//! **S7, through the real binary**: `intel build` (from fixture feeds, no
//! network) → `intel import` (verify, floor, fail-closed audit) → `intel scan`
//! finds the known CVE in a fixture wheel, offline, and names the bundle's
//! sequence and date. No test here opens a socket.
use std::path::{Path, PathBuf};
use std::process::{Command, Output};
use rcgen::{CertificateParams, DnType, KeyPair, PKCS_ED25519};
use update_bundle::digest::sha256_hex;
use update_bundle::policy::{
ApprovedSigner, CertificateValidity, ExpiredPosture, TrustMode, TrustPolicy,
};
use update_bundle::schema::POLICY_SCHEMA;
const BIN: &str = env!("CARGO_BIN_EXE_holger");
const FIXTURES: &str = concat!(env!("CARGO_MANIFEST_DIR"), "/../../intel/tests/fixtures");
struct Site {
dir: tempfile::TempDir,
signers: Vec<(String, PathBuf, PathBuf)>,
release: (PathBuf, PathBuf),
policy: PathBuf,
}
fn mint(dir: &Path, cn: &str) -> (String, PathBuf, PathBuf) {
let key = KeyPair::generate_for(&PKCS_ED25519).unwrap();
let mut params = CertificateParams::new(vec![]).unwrap();
params.distinguished_name.push(DnType::CommonName, cn);
let cert = params.self_signed(&key).unwrap();
let (c, k) = (dir.join(format!("{cn}.crt")), dir.join(format!("{cn}.key")));
std::fs::write(&c, cert.der()).unwrap();
std::fs::write(&k, key.serialize_der()).unwrap();
(sha256_hex(cert.der()), c, k)
}
fn site() -> Site {
let dir = tempfile::tempdir().unwrap();
let pki = dir.path().join("pki");
std::fs::create_dir_all(&pki).unwrap();
let reviewer = mint(&pki, "security-reviewer");
let owner = mint(&pki, "system-owner");
let release = mint(&pki, "release");
let policy = TrustPolicy {
schema_version: POLICY_SCHEMA.into(),
trust_mode: Some(TrustMode::PinnedSigners),
trusted_roots: vec![],
certificate_validity: CertificateValidity::Enforce,
approved_signers: vec![
ApprovedSigner {
fingerprint: reviewer.0.clone(),
subject: None,
roles: vec!["security-reviewer".into()],
},
ApprovedSigner {
fingerprint: owner.0.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.0.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/v1".into()],
allowed_models: vec![],
};
let policy_path = dir.path().join("intel-trust.ron");
std::fs::write(&policy_path, policy.to_ron().unwrap()).unwrap();
let feeds = dir.path().join("feeds");
std::fs::create_dir_all(&feeds).unwrap();
for (from, to) in [
("PYSEC-2018-28.json", "osv-PyPI.json"),
("nvd-recent.json", "nvd-recent.json"),
("kev.json", "kev.json"),
] {
std::fs::copy(Path::new(FIXTURES).join(from), feeds.join(to)).unwrap();
}
let wheels = dir.path().join("wheels");
std::fs::create_dir_all(&wheels).unwrap();
let meta = b"Metadata-Version: 2.1\nName: requests\nVersion: 2.19.0\nLicense: Apache-2.0\n";
std::fs::write(
wheels.join("requests-2.19.0-py2.py3-none-any.whl"),
zip_entry::fixture::zip(&[
("requests-2.19.0.dist-info/METADATA", meta),
("requests/__init__.py", b""),
]),
)
.unwrap();
Site {
signers: vec![reviewer, owner],
release: (release.1, release.2),
policy: policy_path,
dir,
}
}
impl Site {
fn p(&self, rel: &str) -> PathBuf {
self.dir.path().join(rel)
}
/// A config with an `intel:` block and the given audit block.
fn config(&self, name: &str, audit: &str, audit_is_a_file: bool) -> PathBuf {
let audit_dir = self.p(&format!("{name}-audit"));
if audit_is_a_file {
std::fs::write(&audit_dir, b"not a directory").unwrap();
} else {
std::fs::create_dir_all(&audit_dir).unwrap();
}
let cfg = self.p(&format!("{name}.ron"));
std::fs::write(
&cfg,
format!(
r#"(
exposed_endpoints: [],
storage_endpoints: [],
repositories: [],
audit: {},
intel: Some((
db_dir: "{}",
offline: true,
warn_after_days: 7,
block_after_days: 30,
trust_policy: Some("{}"),
)),
)
"#,
audit.replace("{}", &audit_dir.display().to_string()),
self.p(&format!("{name}-intel-db")).display(),
self.policy.display(),
),
)
.unwrap();
cfg
}
fn build(&self, sequence: u64) -> (Output, PathBuf) {
let out = self.p(&format!("intel-{sequence}.holger.znippy"));
let mut c = Command::new(BIN);
c.args(["intel", "build", "--offline", "--feeds-dir"])
.arg(self.p("feeds"))
.arg("--policy")
.arg(&self.policy)
.arg("--release-cert")
.arg(&self.release.0)
.arg("--release-key")
.arg(&self.release.1)
.args(["--sequence", &sequence.to_string(), "--out"])
.arg(&out);
for (i, (_, crt, key)) in self.signers.iter().enumerate() {
let role = if i == 0 {
"security-reviewer"
} else {
"system-owner"
};
c.arg("--approve")
.arg(format!("{role}={},{}", crt.display(), key.display()));
}
(c.output().unwrap(), out)
}
}
fn run(args: &[&str], paths: &[&Path]) -> Output {
let mut c = Command::new(BIN);
c.args(args);
for p in paths {
c.arg(p);
}
c.env_remove("ESTATE_CLOCK")
.env_remove("ESTATE_CLOCK_KEY")
.output()
.unwrap()
}
fn text(o: &Output) -> String {
format!(
"{}\n{}",
String::from_utf8_lossy(&o.stdout),
String::from_utf8_lossy(&o.stderr)
)
}
fn import(cfg: &Path, bundle: &Path, first: bool) -> Output {
let mut c = Command::new(BIN);
c.args(["intel", "import"])
.arg(bundle)
.arg("--config")
.arg(cfg);
if first {
allow_first_import(BIN, cfg, bundle, None);
}
c.env_remove("ESTATE_CLOCK")
.env_remove("ESTATE_CLOCK_KEY")
.output()
.unwrap()
}
#[test]
fn build_import_then_an_offline_scan_finds_the_cve_in_a_fixture_wheel() {
let s = site();
let (b, bundle) = s.build(100);
assert!(b.status.success(), "{}", text(&b));
assert!(text(&b).contains("BUILT holger-intel-100-"), "{}", text(&b));
let cfg = s.config("ok", "( dir: \"{}\", enabled: true )", false);
let i = import(&cfg, &bundle, true);
assert!(i.status.success(), "{}", text(&i));
assert!(
text(&i).contains("IMPORTED (new) holger-intel-100-"),
"{}",
text(&i)
);
let audit: String = std::fs::read_dir(s.p("ok-audit"))
.unwrap()
.flatten()
.filter_map(|e| std::fs::read_to_string(e.path()).ok())
.collect();
assert!(
audit.contains("intel-import") && audit.contains("seq=100"),
"the import is audited: {audit:?}"
);
let wheels = s.p("wheels");
let scan = run(
&["intel", "scan", "--ecosystem", "pip", "--config"],
&[&cfg, Path::new("--path"), &wheels],
);
let t = text(&scan);
assert!(!scan.status.success(), "a KEV-listed CVE blocks:\n{t}");
assert!(
t.contains("PYSEC-2018-28") && t.contains("[BLOCK] requests 2.19.0"),
"{t}"
);
let date = chrono::Utc::now().format("%Y-%m-%d").to_string();
assert!(
t.contains(&format!("seq 100 dated {date}")),
"the evidence names sequence and date:\n{t}"
);
assert!(t.contains("offline"), "{t}");
let st = run(&["intel", "status", "--config"], &[&cfg]);
assert!(
st.status.success() && text(&st).contains("seq 100"),
"{}",
text(&st)
);
}
#[test]
fn an_older_sequence_and_a_tampered_bundle_are_refused_by_the_binary() {
let s = site();
let (_, newer) = s.build(200);
let (_, older) = s.build(150);
let cfg = s.config("roll", "( dir: \"{}\", enabled: true )", false);
assert!(import(&cfg, &newer, true).status.success());
let o = import(&cfg, &older, false);
assert!(
!o.status.success(),
"an older sequence went through:\n{}",
text(&o)
);
assert!(text(&o).contains("sequence rollback"), "{}", text(&o));
let mut members = update_bundle::load_bundle(&newer).unwrap();
let key = "payload/advisories.json".to_string();
let mut v: serde_json::Value = serde_json::from_slice(&members[&key]).unwrap();
v["advisories"] = serde_json::json!([]);
members.insert(key, serde_json::to_vec(&v).unwrap());
let forged = s.p("forged.holger.znippy");
update_bundle::pack_bundle(&members, &forged).unwrap();
let t = import(&cfg, &forged, false);
assert!(
!t.status.success(),
"a tampered bundle went through:\n{}",
text(&t)
);
assert!(text(&t).contains("verification failed"), "{}", text(&t));
}
#[test]
fn an_intel_import_with_a_broken_or_disabled_audit_sink_is_refused_and_writes_nothing() {
let s = site();
let (_, bundle) = s.build(300);
for (name, block, as_file) in [
("broken", "( dir: \"{}\", enabled: true )", true),
("off", "( dir: \"{}\", enabled: false )", false),
] {
let cfg = s.config(name, block, as_file);
// With or without an admin's first-import attempt (refused itself: it cannot be
// audited), the import's refusal is the audit's, never the absent floor's.
for granted in [true, false] {
let o = import(&cfg, &bundle, granted);
assert!(
!o.status.success(),
"{name}: an import with no audit record went through:\n{}",
text(&o)
);
assert!(
text(&o).to_lowercase().contains("audit"),
"{name} (grant asked: {granted}): {}",
text(&o)
);
assert!(
!text(&o).contains("[holger-import-9]"),
"{name}: the audit is judged before the floor: {}",
text(&o)
);
}
let db = s.p(&format!("{name}-intel-db"));
assert!(
!db.join("installed.json").exists(),
"{name}: nothing went live"
);
assert!(
!db.join("state").exists()
|| std::fs::read_dir(db.join("state"))
.unwrap()
.next()
.is_none(),
"{name}: no floor"
);
}
}
#[test]
fn intel_build_in_offline_mode_refuses_the_network() {
let s = site();
let out = s.p("never.holger.znippy");
let o = Command::new(BIN)
.args([
"intel",
"build",
"--offline",
"--ecosystem",
"PyPI",
"--policy",
])
.arg(&s.policy)
.arg("--release-cert")
.arg(&s.release.0)
.arg("--release-key")
.arg(&s.release.1)
.arg("--out")
.arg(&out)
.output()
.unwrap();
let t = text(&o);
assert!(!o.status.success(), "{t}");
assert!(
t.contains("network refused"),
"the refusal names the network: {t}"
);
assert!(!out.exists());
}
/// The admin's first-import authorization on the secure side (lane HOL-CDS, NCSC gap 2), for
/// the vendor, product and sequence of `bundle` (verified under `policy`). Its refusal (a
/// floor that already stands) is the import's business, so it is not asserted here.
fn allow_first_import(bin: &str, config: &Path, bundle: &Path, policy: Option<&Path>) {
let mut c = Command::new(bin);
c.args(["bundle", "allow-first-import", "--config"])
.arg(config)
.arg("--from-bundle")
.arg(bundle);
if let Some(p) = policy {
c.arg("--policy").arg(p);
}
let _ = c
.env_remove("ESTATE_CLOCK")
.env_remove("ESTATE_CLOCK_KEY")
.output();
}