intel/tests/offline_intel.rs
//! S7: OSV / NVD / KEV → signed intel bundle → (USB) → verify → local advisory
//! database → offline scan. Every test runs on fixture feeds; none opens a socket.
mod common;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use chrono::{Duration, Utc};
use common::*;
use holger_intel::feeds::FeedSpec;
use holger_intel::net::{Fetch, IntelNet, NetError};
use holger_intel::resolve::{FINDING_EXPIRED, FINDING_MISSING, FINDING_STALE};
use holger_intel::{
import, resolve, resolve_local, ImportRequest, IntelScanner, LiveIntel, LocalIntelDb,
Staleness, StalenessRule,
};
use holger_update_bundle::digest::sha256_hex;
use modgunn::core::{ArtifactRef, Decision, Severity};
use modgunn::scan::{Artifact, Policy, Provenance, ProvenancePolicy};
fn scan_policy() -> Policy {
Policy {
version: "intel-test".into(),
provenance: ProvenancePolicy::Ignore,
..Policy::default()
}
}
fn wheel_artifact(dir: &std::path::Path) -> Artifact {
let bytes = std::fs::read(fixture_wheel(dir)).unwrap();
let (name, version) = wheel_coords(&bytes);
Artifact::new(
ArtifactRef {
ecosystem: "pip".into(),
name,
version,
blob_sha256: sha256_hex(&bytes),
},
Some("Apache-2.0".into()),
Provenance::unsigned(),
)
}
fn import_at(
pki: &Pki,
db: &LocalIntelDb,
bundle: &std::path::Path,
first: bool,
) -> anyhow::Result<holger_intel::ImportOutcome> {
let req = ImportRequest {
bundle,
policy: &pki.policy,
db,
first_import: first,
};
import(&req, &Fixed(Utc::now()), &MemAudit::default())
}
#[test]
fn feeds_normalize_into_one_enriched_set() {
let dir = tempfile::tempdir().unwrap();
let set = fixture_set(dir.path(), Utc::now());
assert_eq!(
set.sources.len(),
4,
"osv, nvd, kev and the popular PyPI names are each recorded: {:?}",
set.sources
);
let req = set
.advisories
.iter()
.find(|a| a.package == "requests")
.expect("the OSV record");
assert!(req.aliases.iter().any(|a| a == "CVE-2018-18074"), "{req:?}");
assert!(req.kev, "KEV enriches by alias");
assert!(
req.cvss.starts_with("CVSS:3.1/"),
"NVD enriches by alias: {req:?}"
);
assert!(req.cwe.contains(&"CWE-522".to_string()));
assert_eq!(
req.severity,
Severity::High,
"NVD's stronger severity is taken"
);
let again = fixture_set(dir.path(), set.generated_at);
let json = |s: &holger_intel::AdvisorySet| serde_json::to_vec(&s.advisories).unwrap();
assert_eq!(
json(&set),
json(&again),
"the same feeds give the same advisory bytes"
);
}
#[test]
fn build_then_import_then_scan_finds_the_known_cve_in_a_fixture_wheel() {
let dir = tempfile::tempdir().unwrap();
let pki = Pki::new();
let built = bundle(&pki, dir.path(), 42, Utc::now());
assert!(built.report.approvals.len() == 2 && built.advisories >= 2);
let db = LocalIntelDb::open(dir.path().join("intel-db"));
let out = import_at(&pki, &db, &built.path, true).expect("first import");
assert_eq!(out.record.outcome, "new");
assert_eq!(out.installed.sequence, 42);
assert!(out.stored_bundle.exists() && out.receipt.exists());
let resolved = resolve(
&db,
StalenessRule::default(),
&IntelNet::offline(),
&[],
Utc::now(),
)
.unwrap();
assert_eq!(resolved.evidence.source, "local-intel");
assert_eq!(resolved.evidence.sequence, Some(42));
let date = Utc::now().format("%Y-%m-%d").to_string();
assert_eq!(
resolved.evidence.snapshot_date.as_deref(),
Some(date.as_str())
);
let scanner = IntelScanner::new(scan_policy(), resolved);
let v = scanner.scan(&wheel_artifact(dir.path()));
let hit = v
.verdict
.findings
.iter()
.find(|f| f.id == "PYSEC-2018-28")
.unwrap_or_else(|| panic!("the CVE is found: {:?}", v.verdict));
assert_eq!(
hit.severity,
Severity::Critical,
"a KEV hit escalates to Critical"
);
assert_eq!(v.verdict.decision, Decision::Block);
assert!(
v.verdict.advisory_db_rev.contains("seq 42") && v.verdict.advisory_db_rev.contains(&date),
"the evidence names the bundle sequence and date: {}",
v.verdict.advisory_db_rev
);
assert!(
v.evidence.describe().contains("offline"),
"{}",
v.evidence.describe()
);
}
#[test]
fn a_tampered_bundle_is_refused_and_writes_nothing() {
let dir = tempfile::tempdir().unwrap();
let pki = Pki::new();
let built = bundle(&pki, dir.path(), 7, Utc::now());
// The attack that matters: unpack, drop the advisory that would have
// blocked, repack. Every byte the archive holds is still well-formed.
let mut members = holger_update_bundle::load_bundle(&built.path).unwrap();
let key = format!("payload/{}", holger_intel::ADVISORIES_REL);
let mut set = holger_intel::AdvisorySet::from_json_bytes(&members[&key]).unwrap();
set.advisories.retain(|a| a.package != "requests");
members.insert(key, set.to_json_bytes().unwrap());
let tampered = dir.path().join("tampered.holger.znippy");
holger_update_bundle::pack_bundle(&members, &tampered).unwrap();
let db = LocalIntelDb::open(dir.path().join("intel-db"));
let err = import_at(&pki, &db, &tampered, true).expect_err("a tampered bundle must be refused");
assert!(
format!("{err:#}").contains("verification failed") || format!("{err:#}").contains("intel"),
"{err:#}"
);
assert!(db.installed().unwrap().is_none(), "nothing went live");
// A bundle signed by someone the policy does not trust is refused the same way.
let stranger = Pki::new();
let foreign = bundle(&stranger, dir.path(), 8, Utc::now());
let err =
import_at(&pki, &db, &foreign.path, true).expect_err("an untrusted signer must be refused");
assert!(
format!("{err:#}").contains("verification failed"),
"{err:#}"
);
assert!(db.installed().unwrap().is_none());
}
#[test]
fn an_older_sequence_is_refused_and_the_newer_set_stays_live() {
let dir = tempfile::tempdir().unwrap();
let pki = Pki::new();
let older = bundle(&pki, dir.path(), 10, Utc::now() - Duration::days(1));
let newer = bundle(&pki, dir.path(), 11, Utc::now());
let db = LocalIntelDb::open(dir.path().join("intel-db"));
// No floor yet: refused unless the operator says --first-import.
let err = import_at(&pki, &db, &newer.path, false).expect_err("no floor without first_import");
assert!(
format!("{err:#}").to_lowercase().contains("floor"),
"{err:#}"
);
import_at(&pki, &db, &newer.path, true).expect("first import");
let err = import_at(&pki, &db, &older.path, false).expect_err("rollback must be refused");
assert!(
format!("{err:#}").contains("10") && format!("{err:#}").contains("11"),
"{err:#}"
);
// first_import does not weaken a floor in force.
import_at(&pki, &db, &older.path, true).expect_err("first_import never bypasses a floor");
assert_eq!(db.installed().unwrap().unwrap().sequence, 11);
// The same bundle again is an idempotent re-import, not a replay.
let again = import_at(&pki, &db, &newer.path, false).unwrap();
assert_eq!(again.record.outcome, "idempotent-reimport");
}
#[test]
fn an_audit_failure_refuses_the_import_and_changes_nothing() {
let dir = tempfile::tempdir().unwrap();
let pki = Pki::new();
let first = bundle(&pki, dir.path(), 20, Utc::now());
let second = bundle(&pki, dir.path(), 21, Utc::now());
let db = LocalIntelDb::open(dir.path().join("intel-db"));
fn req<'a>(
b: &'a std::path::Path,
pki: &'a Pki,
db: &'a LocalIntelDb,
first: bool,
) -> ImportRequest<'a> {
ImportRequest {
bundle: b,
policy: &pki.policy,
db,
first_import: first,
}
}
// Off from the start: nothing at all.
let err = import(
&req(&first.path, &pki, &db, true),
&Fixed(Utc::now()),
&OffAudit,
)
.expect_err("audit off");
assert!(format!("{err:#}").contains("audit"), "{err:#}");
assert!(db.installed().unwrap().is_none());
assert!(
!db.state_dir().exists() || std::fs::read_dir(db.state_dir()).unwrap().next().is_none(),
"no floor"
);
import(
&req(&first.path, &pki, &db, true),
&Fixed(Utc::now()),
&MemAudit::default(),
)
.unwrap();
let before = std::fs::read(db.root().join("installed.json")).unwrap();
let err = import(
&req(&second.path, &pki, &db, false),
&Fixed(Utc::now()),
&FailingAudit,
)
.expect_err("audit write fails");
assert!(
format!("{err:#}").contains("audit record could not be written"),
"{err:#}"
);
assert_eq!(
std::fs::read(db.root().join("installed.json")).unwrap(),
before,
"the live set is unchanged"
);
assert!(
!db.root().join(LocalIntelDb::generation_file(21)).exists(),
"the staged generation is undone"
);
assert!(!db
.bundles_dir()
.join(format!("{}.holger.znippy", second.bundle_id))
.exists());
// The floor did not move: the same bundle imports once the sink works.
let ok = import(
&req(&second.path, &pki, &db, false),
&Fixed(Utc::now()),
&MemAudit::default(),
)
.unwrap();
assert_eq!(ok.record.outcome, "new");
}
/// A fetcher that counts every call and never answers: it would only ever be
/// reached by a network call.
struct Tripwire(Arc<AtomicU64>);
impl Fetch for Tripwire {
fn get(&self, url: &str) -> Result<Vec<u8>, NetError> {
self.0.fetch_add(1, Ordering::SeqCst);
Err(NetError::Unavailable {
url: url.into(),
why: "tripwire".into(),
})
}
}
#[test]
fn offline_mode_refuses_the_network() {
let dir = tempfile::tempdir().unwrap();
let pki = Pki::new();
let built = bundle(&pki, dir.path(), 5, Utc::now());
let db = LocalIntelDb::open(dir.path().join("intel-db"));
import_at(&pki, &db, &built.path, true).unwrap();
let feeds = FeedSpec::default_set(&["PyPI".to_string()], 2026);
// The offline gate refuses a URL outright; nothing stands behind it.
let offline = IntelNet::offline();
let err = offline
.get("https://osv-vulnerabilities.storage.googleapis.com/PyPI/all.zip")
.unwrap_err();
assert!(err.is_refused(), "{err}");
assert!(err.to_string().contains("network refused"));
// Resolving offline never asks the gate at all, and reads the local DB.
let r = resolve(&db, StalenessRule::default(), &offline, &feeds, Utc::now()).unwrap();
assert_eq!(r.evidence.source, "local-intel");
assert_eq!(
offline.refused_count(),
1,
"only the explicit call above reached the gate"
);
// Online with no network: the feeds are tried, then the local DB answers.
let calls = Arc::new(AtomicU64::new(0));
let online = IntelNet::online(Box::new(Tripwire(calls.clone())));
let r = resolve(&db, StalenessRule::default(), &online, &feeds, Utc::now()).unwrap();
assert_eq!(
calls.load(Ordering::SeqCst),
1,
"online mode does try the network"
);
assert_eq!(r.evidence.source, "local-intel");
assert!(
r.evidence
.network
.as_deref()
.unwrap_or("")
.contains("no network"),
"{:?}",
r.evidence
);
}
#[test]
fn a_stale_database_warns_then_blocks() {
let dir = tempfile::tempdir().unwrap();
let pki = Pki::new();
let snapshot = Utc::now() - Duration::days(2);
let built = bundle(&pki, dir.path(), 30, snapshot);
let db = LocalIntelDb::open(dir.path().join("intel-db"));
import_at(&pki, &db, &built.path, true).unwrap();
let rule = StalenessRule {
warn_after_days: 7,
block_after_days: 30,
};
rule.validate().unwrap();
assert!(StalenessRule {
warn_after_days: 30,
block_after_days: 7
}
.validate()
.is_err());
// A clean artifact, so only the database's age decides.
let clean = Artifact::new(
ArtifactRef {
ecosystem: "pip".into(),
name: "requests".into(),
version: "2.31.0".into(),
blob_sha256: "00".into(),
},
None,
Provenance::unsigned(),
);
let at = |days: i64| {
let r = resolve_local(&db, rule, snapshot + Duration::days(days), None).unwrap();
IntelScanner::new(scan_policy(), r).scan(&clean)
};
let fresh = at(3);
assert_eq!(fresh.evidence.staleness, Staleness::Fresh);
assert_eq!(
fresh.verdict.decision,
Decision::Pass,
"{:?}",
fresh.verdict
);
let warn = at(8);
assert_eq!(warn.evidence.staleness, Staleness::Warn);
assert_eq!(warn.verdict.decision, Decision::Warn);
let f = warn
.verdict
.findings
.iter()
.find(|f| f.id == FINDING_STALE)
.expect("a stale finding");
assert!(
f.summary.contains("seq 30") && f.summary.contains("8 days"),
"{}",
f.summary
);
let block = at(31);
assert_eq!(block.evidence.staleness, Staleness::Block);
assert_eq!(block.verdict.decision, Decision::Block);
assert!(block
.verdict
.findings
.iter()
.any(|f| f.id == FINDING_EXPIRED));
// No database at all blocks too.
let empty = LocalIntelDb::open(dir.path().join("empty"));
let r = resolve_local(&empty, rule, Utc::now(), None).unwrap();
let v = IntelScanner::new(scan_policy(), r).scan(&clean);
assert_eq!(v.verdict.decision, Decision::Block);
assert!(v.verdict.findings.iter().any(|f| f.id == FINDING_MISSING));
}
#[test]
fn a_database_edited_after_import_is_refused() {
let dir = tempfile::tempdir().unwrap();
let pki = Pki::new();
let built = bundle(&pki, dir.path(), 3, Utc::now());
let db = LocalIntelDb::open(dir.path().join("intel-db"));
import_at(&pki, &db, &built.path, true).unwrap();
let live = db.root().join(LocalIntelDb::generation_file(3));
let mut bytes = std::fs::read(&live).unwrap();
bytes.extend_from_slice(b" ");
std::fs::write(&live, bytes).unwrap();
let err = db.load().expect_err("an edited database is refused");
assert!(
format!("{err:#}").contains("does not match its install record"),
"{err:#}"
);
}
#[test]
fn an_import_goes_live_without_a_restart() {
let dir = tempfile::tempdir().unwrap();
let pki = Pki::new();
let db = LocalIntelDb::open(dir.path().join("intel-db"));
let live = LiveIntel::new(db.clone(), StalenessRule::default());
let (_, e) = live.current(Utc::now()).unwrap();
assert_eq!(e.source, "none");
import_at(
&pki,
&db,
&bundle(&pki, dir.path(), 1, Utc::now()).path,
true,
)
.unwrap();
let (adb, e) = live.current(Utc::now()).unwrap();
assert_eq!(e.sequence, Some(1));
assert!(!adb.advisories.is_empty());
import_at(
&pki,
&db,
&bundle(&pki, dir.path(), 2, Utc::now()).path,
false,
)
.unwrap();
let (_, e) = live.current(Utc::now()).unwrap();
assert_eq!(
e.sequence,
Some(2),
"the second import is what the next scan reads"
);
}
/// SANDBOX-SCAN: the image CVE scanner matches an image's apk packages against
/// the Wolfi and Chainguard advisories, so a default online build fetches them,
/// and a real OSV Wolfi record lands under the `wolfi` label the image's
/// packages carry.
#[test]
fn the_default_feeds_carry_the_image_distributions() {
let defaults = holger_intel::feeds::DEFAULT_ECOSYSTEMS;
for want in ["Wolfi", "Chainguard"] {
assert!(
defaults.contains(&want),
"{want} is not fetched by default: {defaults:?}"
);
}
let rec = std::fs::read_to_string(
std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
.join("../server/lib/tests/fixtures/sandbox/osv-wolfi-openssl.json"),
)
.unwrap();
let advisories = holger_intel::modgunn::osv::parse_doc(&rec).unwrap();
assert!(
advisories
.iter()
.any(|a| a.ecosystem == "wolfi" && a.package == "openssl"),
"{:?}",
advisories
.iter()
.map(|a| (&a.ecosystem, &a.package))
.collect::<Vec<_>>()
);
}
/// HOL-DOGFOOD-8: scans that start together (a pull-through closure: hundreds of crates at once)
/// must share ONE loaded database. Each one that missed the cache used to load its own copy of
/// the whole advisory set (262 717 advisories on the oden dogfood Holger), and the box grew to
/// 416 GB. Every concurrent caller gets the same `Arc`.
#[test]
fn concurrent_scans_share_one_loaded_database() {
let dir = tempfile::tempdir().unwrap();
let pki = Pki::new();
let db = LocalIntelDb::open(dir.path().join("intel-db"));
import_at(
&pki,
&db,
&bundle(&pki, dir.path(), 1, Utc::now()).path,
true,
)
.unwrap();
let live = Arc::new(LiveIntel::new(db, StalenessRule::default()));
let n = 32;
let start = Arc::new(std::sync::Barrier::new(n));
let handles: Vec<_> = (0..n)
.map(|_| {
let (live, start) = (live.clone(), start.clone());
std::thread::spawn(move || {
start.wait();
live.current(Utc::now()).unwrap().0
})
})
.collect();
let dbs: Vec<_> = handles.into_iter().map(|h| h.join().unwrap()).collect();
let distinct = dbs
.iter()
.enumerate()
.filter(|(i, d)| !dbs[..*i].iter().any(|e| Arc::ptr_eq(e, d)))
.count();
assert_eq!(
distinct, 1,
"{n} concurrent scans loaded {distinct} copies of the database"
);
}
/// **The default feeds judge Debian, Alpine and the RPM family** (H32-FAILCLOSED follow-up):
/// a scan with no advisory data for an ecosystem holds every package of it, so a default
/// online build fetches what the chain views and APK sources serve. An OSV bucket name with a
/// space is fetched percent-encoded, and the real records land under the chain's labels (`deb`,
/// `apk`, `rpm`).
#[test]
fn the_default_feeds_carry_debian_alpine_and_the_rpm_family() {
let defaults = holger_intel::feeds::DEFAULT_ECOSYSTEMS;
for want in [
"Debian",
"Alpine",
"AlmaLinux",
"Rocky Linux",
"Red Hat",
"SUSE",
"openSUSE",
] {
assert!(
defaults.contains(&want),
"{want} is not fetched by default: {defaults:?}"
);
}
assert_eq!(
holger_intel::feeds::FeedSpec::osv("Rocky Linux").uri,
format!("{}/Rocky%20Linux/all.zip", holger_intel::feeds::OSV_BUCKET)
);
let fx = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
.join("../server/lib/tests/fixtures/known_bad");
for (file, eco) in [
("RLSA-2023-0946.json", "rpm"),
("ALPINE-CVE-2024-5535.json", "apk"),
("DEBIAN-CVE-2024-5535.json", "deb"),
] {
let rec = std::fs::read_to_string(fx.join(file)).unwrap();
let advisories = holger_intel::modgunn::osv::parse_doc(&rec).unwrap();
assert!(
advisories.iter().any(|a| a.ecosystem == eco),
"{file}: {:?}",
advisories.iter().map(|a| &a.ecosystem).collect::<Vec<_>>()
);
}
}
/// H32-TYPOSQUAT: the popular-names list crosses the airgap inside the signed bundle and is
/// what the box reads after import; before any import there is none (the scan holds).
#[test]
fn the_popular_names_cross_the_airgap_in_the_signed_set() {
let dir = tempfile::tempdir().unwrap();
let pki = Pki::new();
let db = LocalIntelDb::open(dir.path().join("intel-db"));
let live = LiveIntel::new(db.clone(), StalenessRule::default());
let (supply, _) = live.current_supply(Utc::now()).unwrap();
assert!(supply.popular_for("pypi").is_none(), "no import, no list");
import_at(
&pki,
&db,
&bundle(&pki, dir.path(), 7, Utc::now()).path,
true,
)
.unwrap();
let (supply, e) = live.current_supply(Utc::now()).unwrap();
assert_eq!(e.sequence, Some(7));
assert_eq!(
supply.popular_for("pip").unwrap(),
["boto3", "numpy", "requests"]
);
assert!(
supply.popular_for("npm").is_none(),
"only what the feeds carried"
);
}