server/lib/tests/an_online_estate_refreshes_its_intel.rs
//! **An online estate fetches its own intel** (lane W4, `intel.refresh`).
//!
//! A sold estate is online (`intel.offline: false`), and its intake chain requires
//! the vulnerability scan at SCANNED. Until an advisory database is installed,
//! every artifact blocks there. The `intel.refresh` jera job fixes that: once at
//! boot (and daily) the box fetches OSV, NVD and KEV — here from a local fixture
//! feed server named by `intel.feed_mirror` — and lands the set through the import
//! door's verify, floor and audit path. A wheel then passes SCANNED.
//!
//! An offline config (`intel.offline: true`) queues no refresh and dials nothing,
//! not even when the refresh is asked for by hand.
use std::io::{Read, Write};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use holger_server_lib::promote::intake::{ArtifactState, TransitionRequest};
use holger_server_lib::{read_ron_config, wire_holger, ArtifactId, Holger};
const FIXTURES: &str = concat!(env!("CARGO_MANIFEST_DIR"), "/../../intel/tests/fixtures");
/// A plain-http feed mirror on loopback, counting every request it answers.
struct FeedServer {
base: String,
hits: Arc<AtomicUsize>,
}
fn feed_server() -> FeedServer {
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let base = format!("http://{}", listener.local_addr().unwrap());
let hits = Arc::new(AtomicUsize::new(0));
let files: Vec<(&str, Vec<u8>)> = vec![
(
"index.txt",
b"osv-PyPI.json\nnvd-recent.json\nkev.json\n".to_vec(),
),
(
"osv-PyPI.json",
std::fs::read(Path::new(FIXTURES).join("PYSEC-2018-28.json")).unwrap(),
),
(
"nvd-recent.json",
std::fs::read(Path::new(FIXTURES).join("nvd-recent.json")).unwrap(),
),
(
"kev.json",
std::fs::read(Path::new(FIXTURES).join("kev.json")).unwrap(),
),
];
let counter = Arc::clone(&hits);
std::thread::spawn(move || {
for stream in listener.incoming() {
let Ok(mut s) = stream else { continue };
counter.fetch_add(1, Ordering::SeqCst);
let mut buf = [0u8; 4096];
let n = s.read(&mut buf).unwrap_or(0);
let head = String::from_utf8_lossy(&buf[..n]).to_string();
let path = head
.split_whitespace()
.nth(1)
.unwrap_or("/")
.trim_start_matches('/')
.to_string();
let reply = match files.iter().find(|(name, _)| *name == path) {
Some((_, body)) => {
let mut r = format!(
"HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
body.len()
)
.into_bytes();
r.extend_from_slice(body);
r
}
None => b"HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"
.to_vec(),
};
let _ = s.write_all(&reply);
}
});
FeedServer { base, hits }
}
struct Running {
_dir: tempfile::TempDir,
data: PathBuf,
holger: Holger,
}
fn start(offline: bool, mirror: &str) -> Running {
let mut last = String::new();
for _ in 0..8 {
match try_start(offline, mirror) {
Ok(r) => return r,
Err(e) => last = e,
}
}
panic!("server never started: {last}");
}
fn try_start(offline: bool, mirror: &str) -> Result<Running, String> {
let (grpc, http) = {
let a = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let b = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
(
a.local_addr().unwrap().port(),
b.local_addr().unwrap().port(),
)
};
let dir = tempfile::tempdir().unwrap();
let data = dir.path().join("data");
std::fs::create_dir_all(&data).unwrap();
let d = data.display().to_string();
let repo = |name: &str| {
format!(
r#"(ron_name: "{name}", ron_repo_type: "raw", ron_upstreams: [], ron_in: None,
ron_out: Some((ron_storage_endpoint: "storage", ron_exposed_endpoint: "door")),
ron_data_dir: Some("{d}/{name}")),"#
)
};
let ron = format!(
r#"(
repositories: [ {} {} {} {} ],
exposed_endpoints: [ (ron_name: "door", ron_url: "127.0.0.1:{grpc}", ron_http_url: Some("127.0.0.1:{http}")) ],
storage_endpoints: [ (ron_name: "storage", ron_storage_type: "znippy", ron_path: "{d}") ],
audit: (dir: "{d}/audit", enabled: true),
intel: Some((db_dir: "{d}/intel", offline: {offline}, feed_mirror: Some("{mirror}"))),
promotion: Some((
ladder_name: "intake",
stages: [
(name: "incoming", repo: "incoming"),
(name: "quarantine", repo: "quarantine"),
(name: "approved", repo: "approved"),
(name: "production", repo: "production"),
],
intake: Some((rules: (sandbox: off),
incoming: "incoming", quarantine: "quarantine", approved: "approved", production: "production",
scan: Some((required: [vulnerability])),
)),
)),
)"#,
repo("incoming"),
repo("quarantine"),
repo("approved"),
repo("production"),
);
let cfg = dir.path().join("holger.ron");
std::fs::write(&cfg, ron).unwrap();
let mut holger: Holger = read_ron_config(&cfg).map_err(|e| format!("config: {e}"))?;
holger
.instantiate_backends()
.map_err(|e| format!("backends: {e}"))?;
wire_holger(&mut holger).map_err(|e| format!("wire: {e}"))?;
holger.start().map_err(|e| format!("start: {e}"))?;
Ok(Running {
_dir: dir,
data,
holger,
})
}
fn wheel(version: &str) -> Vec<u8> {
let meta =
format!("Metadata-Version: 2.1\nName: requests\nVersion: {version}\nLicense: Apache-2.0\n");
zip_entry::fixture::zip(&[
(
&format!("requests-{version}.dist-info/METADATA"),
meta.as_bytes(),
),
("requests/__init__.py", b""),
])
}
fn scan(r: &Running, version: &str) -> Result<(), String> {
let chain = r.holger.intake().expect("an intake chain");
let incoming = r.holger.get_repo("incoming").unwrap().clone();
let id = ArtifactId {
namespace: None,
name: format!("requests-{version}-py2.py3-none-any.whl"),
version: String::new(),
};
let bytes = wheel(version);
chain
.ingest("imp", "test", "incoming", &id, &bytes, || {
incoming.put(&id, &bytes)
})
.map_err(|e| e.to_string())?;
chain
.advance(TransitionRequest::new(id, ArtifactState::Scanned, "imp"))
.map(|_| ())
.map_err(|e| e.to_string())
}
fn refresh_jobs(r: &Running) -> Vec<holger_server_lib::jobs::JobView> {
let q = r.holger.job_queue_cell.get().expect("the job queue");
q.list(100)
.unwrap()
.into_iter()
.filter(|j| j.kind == holger_server_lib::jobs::kind::INTEL_REFRESH)
.collect()
}
#[tokio::test(flavor = "multi_thread")]
async fn an_online_estates_first_boot_fetches_imports_and_a_wheel_then_passes_scanned() {
let feeds = feed_server();
let r = start(false, &feeds.base);
let installed = r.data.join("intel/installed.json");
for _ in 0..300 {
if installed.exists() {
break;
}
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
}
assert!(
installed.exists(),
"the boot refresh installed a database: jobs {:?}",
refresh_jobs(&r)
);
assert!(
feeds.hits.load(Ordering::SeqCst) >= 4,
"the index and the three feeds were fetched"
);
let jobs = refresh_jobs(&r);
assert!(
jobs.iter()
.any(|j| j.status == "done" && j.target == "intel.refresh"),
"a jera job did it: {jobs:?}"
);
// Through the same audit-first import a carried bundle takes.
let audit: String = std::fs::read_dir(r.data.join("audit"))
.unwrap()
.flatten()
.map(|e| std::fs::read_to_string(e.path()).unwrap_or_default())
.collect();
assert!(
audit.contains("intel-import") && audit.contains("intel-refresh"),
"audited first: {audit}"
);
// A fixed requests passes SCANNED; the vulnerable one is held by the advisory.
scan(&r, "2.31.0").expect("a clean wheel passes SCANNED");
let held = scan(&r, "2.19.0").expect_err("the fetched advisory blocks the vulnerable wheel");
assert!(held.contains("vulnerability"), "{held}");
}
#[tokio::test(flavor = "multi_thread")]
async fn an_offline_config_never_dials_out() {
let feeds = feed_server();
let r = start(true, &feeds.base);
tokio::time::sleep(std::time::Duration::from_secs(3)).await;
assert!(
refresh_jobs(&r).is_empty(),
"an offline box queues no refresh"
);
assert!(!r.data.join("intel/installed.json").exists());
// Asked by hand, with a real fetcher: refused by name, nothing dialled.
let door = r.holger.import_door().expect("the import door");
let e = door
.refresh_intel(&holger_server_lib::holger_intel::IntelNet::http())
.expect_err("offline");
assert!(e.tagged().starts_with("[holger-import-7]"), "{e}");
assert_eq!(
feeds.hits.load(Ordering::SeqCst),
0,
"an offline config never dials the feed server"
);
// And the scan still blocks without a database (never a pass).
let held = scan(&r, "2.31.0").expect_err("no database, no pass");
assert!(held.contains("vulnerability"), "{held}");
}