server/lib/src/security_view/tests.rs
//! The Security page's words (lane HOL-SEC): what holger says, decided here, never the console.
use super::*;
fn id(name: &str, version: &str) -> traits::ArtifactId {
traits::ArtifactId {
namespace: None,
name: name.into(),
version: version.into(),
}
}
fn item(name: &str, state: &str, reason: &str) -> Item {
Item {
id: id(name, "1.0.0"),
state: state.into(),
held_from: (state == "QUARANTINED").then(|| "SCANNED".to_string()),
actor: "approver-1".into(),
reason: reason.into(),
importer: "importer-1".into(),
votes: u32::from(state == "POLICY_CHECKED"),
}
}
fn event(secs: i64, action: &str, trail: &str, status: u16, detail: &str) -> Event {
Event {
ts_nanos: secs * 1_000_000_000,
ident: "importer-1".into(),
action: action.into(),
trail: trail.into(),
repo: "incoming".into(),
artifact: "evil-2.0.0.whl".into(),
status,
detail: detail.into(),
}
}
/// A gateway as the estate ships it: a chain, its scanners in containers, two sources, one
/// package at each interesting place, gated reads, a downstream, and some audit events.
fn gateway() -> Facts {
let slot = |kind: &str, required: bool, installed: &[&str]| ScannerSlot {
kind: kind.into(),
required,
installed: installed.iter().map(|s| s.to_string()).collect(),
};
Facts {
chain: Some(Chain {
approvals: 2,
sandbox: Sandbox::Container("crun (rootless, no network)".into()),
scanners: vec![
slot("inspection", true, &["holger-inspect 1"]),
slot("malware", true, &["clamav 1.4", "yara 4"]),
slot("vulnerability", true, &["modgunn-osv 2 db 7"]),
slot("provenance", true, &["holger-provenance 1"]),
slot("secrets", false, &[]),
slot("sbom", true, &[]),
],
sources: vec![
SourceFact {
name: "chainguard".into(),
kind: "pull".into(),
speaks: "oci".into(),
checks: Some("Signature required.".into()),
},
SourceFact {
name: "crates-io".into(),
kind: "proxy".into(),
speaks: "crates".into(),
checks: None,
},
],
items: vec![
item("wait-pkg", "POLICY_CHECKED", ""),
item("held-pkg", "QUARANTINED", "invalidated upstream: malware"),
item("bad-pkg", "BLOCKED", "CVE-2026-1234 critical"),
item("good-pkg", "AVAILABLE", ""),
item("new-pkg", "INGESTED", ""),
],
runner: None,
}),
access: Access {
gated: true,
front_open: true,
public: vec!["docs".into()],
repos: vec![
Repo {
name: "docs".into(),
public: true,
..Default::default()
},
Repo {
name: "production".into(),
..Default::default()
},
Repo {
name: "acme".into(),
owner_only: true,
gated: true,
..Default::default()
},
],
},
sync: Some(SyncFacts {
name: "cloud".into(),
key: "ab12cd34ef56ab12cd34ef56ab12cd34".into(),
downstreams: vec![SyncPeer {
name: "lab-holger".into(),
key: "99887766554433221100".into(),
state: "behind".into(),
headline: "1 generation not acknowledged".into(),
}],
upstreams: Vec::new(),
}),
events: Ok(vec![
event(1_000, "download", "", 200, ""),
event(2_000, "promote", "approved", 200, "vote 1 of 2"),
event(
3_000,
"promote",
"",
403,
"intake refused: [holger-gate-3] the source denies GPL-3.0",
),
event(4_000, "list", "", 200, ""),
]),
}
}
fn section<'a>(r: &'a pb::SecuritySummaryReply, key: &str) -> &'a pb::SecuritySection {
r.sections
.iter()
.find(|s| s.key == key)
.unwrap_or_else(|| panic!("no section {key}: {r:#?}"))
}
fn row<'a>(s: &'a pb::SecuritySection, name: &str) -> &'a pb::SecurityRow {
s.rows
.iter()
.find(|r| r.name == name)
.unwrap_or_else(|| panic!("section {} has no row {name}: {s:#?}", s.key))
}
#[test]
fn the_page_is_the_intake_chain_in_its_order_and_names_no_machinery() {
let r = summary(&gateway());
let keys: Vec<&str> = r.sections.iter().map(|s| s.key.as_str()).collect();
assert_eq!(
keys,
[
SECTION_CHAIN,
SECTION_SOURCES,
SECTION_SCANNERS,
SECTION_APPROVALS,
SECTION_HELD,
SECTION_ACCESS,
SECTION_SYNC,
SECTION_EVENTS
]
);
assert!(
r.headline.contains("1 package waits for approval, 2 held")
&& r.headline.contains("Scans run in containers"),
"{}",
r.headline
);
let all = format!("{r:?}").to_lowercase();
for machinery in [
"twin",
"overlay",
"wireguard",
"lag_entries",
"fetched_total",
"consumers",
] {
assert!(
!all.contains(machinery),
"the Security page speaks of the machinery ({machinery})"
);
}
}
#[test]
fn the_chain_counts_each_step_and_says_who_approves() {
let r = summary(&gateway());
let c = section(&r, SECTION_CHAIN);
assert_eq!(row(c, "In").state, "1");
assert_eq!(row(c, "Policy checked").state, "1");
assert_eq!(row(c, "Served").state, "1");
assert!(row(c, "Approved")
.detail
.contains("2 different people approve it. The importer cannot approve."));
}
#[test]
fn each_source_says_what_is_checked_or_that_it_skips_the_chain() {
let r = summary(&gateway());
let s = section(&r, SECTION_SOURCES);
assert_eq!(row(s, "chainguard").tone, TONE_GOOD);
assert!(row(s, "chainguard").detail.contains("Signature required"));
assert_eq!(row(s, "crates-io").tone, TONE_WARN);
assert!(row(s, "crates-io")
.detail
.contains("skips the intake chain, so no scanner sees it"));
}
#[test]
fn the_scanners_say_where_they_run_and_what_each_one_checks() {
let r = summary(&gateway());
let s = section(&r, SECTION_SCANNERS);
assert!(
s.summary.contains("runs in its own container (crun"),
"{}",
s.summary
);
let p = row(s, "Provenance");
assert_eq!(p.state, "required");
for w in [
"cosign keyless",
"DSSE",
"Chainguard",
"holger-provenance 1",
] {
assert!(
p.detail.contains(w),
"provenance: {w} missing in {}",
p.detail
);
}
assert!(
row(s, "Malware").detail.contains("ClamAV and YARA")
&& row(s, "Malware").detail.contains("clamav 1.4, yara 4")
);
assert!(row(s, "CVEs").detail.contains("container images"));
assert_eq!(row(s, "Secrets").state, "off");
assert_eq!(
(row(s, "SBOM").state.as_str(), row(s, "SBOM").tone.as_str()),
("missing", TONE_BAD)
);
}
#[test]
fn what_waits_for_approval_shows_its_votes_and_links_the_artifact() {
let r = summary(&gateway());
let a = section(&r, SECTION_APPROVALS);
assert_eq!(a.rows.len(), 1);
let w = row(a, "wait-pkg@1.0.0");
assert_eq!(w.state, "1 of 2");
assert!(w.detail.contains("Imported by importer-1"));
assert_eq!(
w.artifact.as_ref().map(|a| a.name.as_str()),
Some("wait-pkg")
);
}
#[test]
fn what_is_held_or_refused_says_why_by_name() {
let r = summary(&gateway());
let h = section(&r, SECTION_HELD);
let q = row(h, "held-pkg@1.0.0");
assert_eq!(q.state, "quarantined");
assert!(
q.detail.contains("invalidated upstream: malware") && q.detail.contains("held at scanned"),
"{}",
q.detail
);
let b = row(h, "bad-pkg@1.0.0");
assert_eq!((b.state.as_str(), b.tone.as_str()), ("blocked", TONE_BAD));
assert!(b.detail.contains("CVE-2026-1234"));
let refused = row(h, "evil-2.0.0.whl");
assert_eq!(refused.state, "refused");
assert!(
refused
.detail
.starts_with("[holger-gate-3] the source denies GPL-3.0"),
"{}",
refused.detail
);
}
#[test]
fn who_can_read_is_the_access_block_and_each_repository() {
let r = summary(&gateway());
let a = section(&r, SECTION_ACCESS);
assert_eq!(a.summary, "Reading packages needs a sign-in. Public, readable with no sign-in: docs. The front page and the sign-up are open.");
assert_eq!(row(a, "docs").state, "public");
assert_eq!(row(a, "production").state, "sign-in");
assert_eq!(row(a, "acme").state, "owner only");
let mut open = gateway();
open.access.gated = false;
open.access.public.clear();
assert_eq!(
row(section(&summary(&open), SECTION_ACCESS), "production").state,
"open"
);
}
#[test]
fn the_sync_names_the_signing_key_and_each_downstream() {
let r = summary(&gateway());
let s = section(&r, SECTION_SYNC);
assert!(
s.summary
.contains("signed with its sync key ab12cd34ef56ab12…"),
"{}",
s.summary
);
let d = row(s, "lab-holger");
assert_eq!((d.state.as_str(), d.tone.as_str()), ("behind", TONE_WARN));
assert!(
d.detail.contains("1 generation not acknowledged.")
&& d.detail.contains("9988776655443322…"),
"{}",
d.detail
);
let none = summary(&Facts {
sync: None,
..gateway()
});
assert_eq!(
section(&none, SECTION_SYNC).summary,
"This Holger syncs with no other Holger."
);
}
#[test]
fn the_events_are_the_newest_security_events_and_no_reads() {
let r = summary(&gateway());
let e = section(&r, SECTION_EVENTS);
let names: Vec<(&str, &str)> = e
.rows
.iter()
.map(|r| (r.name.as_str(), r.state.as_str()))
.collect();
assert_eq!(names, [("promote", "refused"), ("approved", "done")]);
assert!(
e.rows[1].detail.starts_with(
"1970-01-01 00:33 UTC by importer-1 on incoming/evil-2.0.0.whl. vote 1 of 2"
),
"{}",
e.rows[1].detail
);
let off = summary(&Facts {
events: Err("audit is off on this Holger".into()),
..gateway()
});
assert!(section(&off, SECTION_EVENTS)
.summary
.contains("audit is off"));
}
#[test]
fn a_holger_with_no_chain_says_it_checks_nothing() {
let r = summary(&Facts {
chain: None,
..gateway()
});
assert!(r.headline.contains("no intake chain"), "{}", r.headline);
let keys: Vec<&str> = r.sections.iter().map(|s| s.key.as_str()).collect();
assert_eq!(
keys,
[SECTION_HELD, SECTION_ACCESS, SECTION_SYNC, SECTION_EVENTS]
);
}
#[test]
fn a_key_is_shown_short() {
assert_eq!(short_key("0123456789abcdef0123"), "0123456789abcdef…");
assert_eq!(short_key("abc"), "abc");
}
#[cfg(feature = "intake")]
#[test]
fn a_source_policy_reads_as_its_checks() {
let rules = crate::sources::Rules {
require_signed: true,
require_provenance: true,
identities: vec![crate::sources::Identity::exact(
"https://token.actions.githubusercontent.com",
"https://github.com/chainguard-images/images/.github/workflows/release.yaml@refs/heads/main",
)],
..Default::default()
};
let p = crate::sources::EffectivePolicy::floor("chainguard", &rules);
let w = policy_words(&p, 2);
assert!(
w.starts_with("Signature required, signed by https://github.com/chainguard-images/"),
"{w}"
);
assert!(
w.contains("refuses a CVE above") && w.ends_with("2 people approve."),
"{w}"
);
}
#[cfg(feature = "intake")]
#[test]
fn the_page_says_what_the_job_runner_has_and_that_it_is_not_heimdall() {
let mut f = gateway();
let features =
holger_job_runner::ContainerRunner::features_of(holger_sandbox::EngineKind::Crun);
f.chain.as_mut().unwrap().runner = Some(RunnerFacts::of(&features, "ed25519:00112233aabbccdd"));
let r = summary(&f);
let s = r
.sections
.iter()
.find(|s| s.key == SECTION_RUNNER)
.expect("the runner section");
assert!(
s.summary.contains("holger-crun") && s.summary.contains("not Heimdall"),
"{}",
s.summary
);
assert!(
s.summary.contains("ed25519:00112233aabbccdd"),
"the pinned key: {}",
s.summary
);
let state = |n: &str| {
s.rows
.iter()
.find(|r| r.name == n)
.map(|r| (r.state.clone(), r.tone.clone()))
.unwrap_or_else(|| panic!("{n}"))
};
assert_eq!(state("container"), ("yes".into(), TONE_GOOD.into()));
for missing in ["seccomp", "landlock", "broker", "wasm"] {
assert_eq!(state(missing), ("no".into(), TONE_WARN.into()), "{missing}");
}
// No intake chain, no runner section.
let mut none = gateway();
none.chain.as_mut().unwrap().runner = None;
assert!(!summary(&none)
.sections
.iter()
.any(|s| s.key == SECTION_RUNNER));
}
/// Lane HOL-HEIMDALL-02: under Heimdall's runner the page says so, shows what Heimdall
/// has (its fixed seccomp profile, Landlock, the image verified at every start) and what
/// it got from the kernel; a runner that fell back says why.
#[cfg(feature = "intake")]
#[test]
fn the_page_shows_heimdalls_features_its_kernel_report_and_any_fallback() {
use crate::job_runner::heimdall_core::{Features, KernelFeatures, Protection};
let heimdall = Features {
runner: "heimdall".into(),
container: true,
seccomp: Protection::Default,
landlock: true,
broker: false,
wasm: false,
image_verified_each_start: true,
};
let kernel = KernelFeatures {
release: "7.0.0-34-generic".into(),
seccomp_filter: true,
user_notif: true,
notif_addfd: true,
landlock_abi: Some(6),
cgroup_v2: true,
cgroup_controllers: vec!["cpu".into(), "memory".into(), "pids".into()],
user_namespaces: true,
};
let mut f = gateway();
f.chain.as_mut().unwrap().runner = Some(
RunnerFacts::of(&heimdall, "ed25519:00112233aabbccdd")
.with_kernel(Some(&kernel))
.with_fallback(None),
);
let r = summary(&f);
let s = r.sections.iter().find(|s| s.key == SECTION_RUNNER).unwrap();
assert!(
s.summary.contains("in heimdall,") && !s.summary.contains("not Heimdall"),
"{}",
s.summary
);
let row = |n: &str| {
s.rows
.iter()
.find(|r| r.name == n)
.unwrap_or_else(|| panic!("{n} in {:?}", s.rows))
.clone()
};
for has in ["container", "seccomp", "landlock", "image"] {
assert_eq!(row(has).state, "yes", "{has}");
}
assert!(
row("seccomp").detail.contains("Heimdall's fixed"),
"{:?}",
row("seccomp")
);
assert_eq!(row("broker").state, "no");
assert_eq!(row("kernel").state, "7.0.0-34-generic");
assert_eq!(row("kernel: Landlock").state, "ABI 6");
assert_eq!(row("kernel: seccomp filters").state, "yes");
assert_eq!(row("kernel: cgroup v2").state, "yes");
assert!(row("kernel: cgroup v2")
.detail
.contains("cpu, memory, pids"));
assert_eq!(row("kernel: user namespaces").state, "yes");
// Fallen back to youki-plain: the page says the runner, that it is not Heimdall, and why.
let plain = Features {
runner: "youki-plain".into(),
landlock: false,
..heimdall
};
let mut f = gateway();
f.chain.as_mut().unwrap().runner = Some(
RunnerFacts::of(&plain, "ed25519:00")
.with_kernel(None)
.with_fallback(Some(
"heimdall refuses to start in strict mode: missing Landlock",
)),
);
let r = summary(&f);
let s = r.sections.iter().find(|s| s.key == SECTION_RUNNER).unwrap();
assert!(
s.summary.contains("youki-plain")
&& s.summary.contains("not Heimdall")
&& s.summary.contains("missing Landlock"),
"{}",
s.summary
);
assert!(!s.rows.iter().any(|r| r.name.starts_with("kernel")));
assert_eq!(
s.rows.iter().find(|r| r.name == "seccomp").unwrap().state,
"no",
"youki-plain's profile is moby's, not Heimdall's"
);
}