crates/heimdall-container/tests/kernel.rs
// SPDX-License-Identifier: AGPL-3.0-only
// Copyright (C) 2026 Ignalina ApS and Vetra Automation AB
//! The kernel feature report (spec §10), strict mode and truthful features.
//! Heimdall is a defensive sandbox; these tests check only its own refusal
//! to run with less protection than it says.
use heimdall_container::kernel::{parse_actions, parse_release, KernelReport};
use heimdall_container::*;
use heimdall_core::*;
fn full() -> KernelReport {
KernelReport {
release: "7.0.0-38-generic".into(),
seccomp_filter: true,
user_notif: true,
notif_addfd: true,
landlock_abi: Some(8),
cgroup_v2: true,
cgroup_controllers: vec!["cpu".into(), "memory".into(), "pids".into()],
user_namespaces: true,
}
}
fn cfg() -> HeimdallConfig {
HeimdallConfig::new("/nonexistent/layout", "/nonexistent/work")
}
fn refusal(r: Result<HeimdallRunner, RunError>) -> String {
match r {
Err(RunError::Unsupported { what }) => what,
other => panic!("expected a refusal to start, got {other:?}"),
}
}
#[test]
fn the_release_parses_to_major_minor() {
assert_eq!(parse_release("7.0.0-38-generic"), Some((7, 0)));
assert_eq!(parse_release("5.9.12"), Some((5, 9)));
assert_eq!(parse_release("6.12-rc3"), Some((6, 12)));
assert_eq!(parse_release("garbage"), None);
}
#[test]
fn actions_avail_names_kill_process_and_user_notif() {
assert_eq!(
parse_actions("kill_process kill_thread trap errno user_notif trace log allow\n"),
(true, true)
);
assert_eq!(
parse_actions("kill_thread trap errno trace allow"),
(false, false)
);
assert_eq!(parse_actions(""), (false, false));
}
#[test]
fn the_default_config_is_strict_with_landlock() {
let c = cfg();
assert!(c.strict);
assert_eq!(c.landlock, LandlockMode::Auto);
assert_eq!(c.cgroup_parent, "heimdall");
}
#[test]
fn features_are_heimdall_container_landlock_default_seccomp_broker_image_verified() {
let r = HeimdallRunner::with_kernel(cfg(), BootKey::generate(), full()).unwrap();
assert_eq!(
r.features(),
Features {
runner: "heimdall".into(),
container: true,
seccomp: Protection::Default,
landlock: true,
broker: true,
wasm: false,
image_verified_each_start: true,
}
);
assert_eq!(r.kernel(), &full());
}
#[test]
fn strict_mode_refuses_to_start_without_landlock() {
let k = KernelReport {
landlock_abi: None,
..full()
};
let why = refusal(HeimdallRunner::with_kernel(cfg(), BootKey::generate(), k));
assert!(why.contains("Landlock"), "{why}");
}
#[test]
fn strict_mode_refuses_to_start_with_landlock_switched_off() {
let mut c = cfg();
c.landlock = LandlockMode::Off;
let why = refusal(HeimdallRunner::with_kernel(c, BootKey::generate(), full()));
assert!(why.contains("Landlock"), "{why}");
}
#[test]
fn strict_mode_refuses_without_cgroup_v2_or_a_required_controller_or_user_namespaces() {
let no_v2 = KernelReport {
cgroup_v2: false,
cgroup_controllers: vec![],
..full()
};
assert!(refusal(HeimdallRunner::with_kernel(
cfg(),
BootKey::generate(),
no_v2
))
.contains("cgroup v2"));
for c in ["memory", "pids", "cpu"] {
let k = KernelReport {
cgroup_controllers: full()
.cgroup_controllers
.into_iter()
.filter(|x| x != c)
.collect(),
..full()
};
let why = refusal(HeimdallRunner::with_kernel(cfg(), BootKey::generate(), k));
assert!(why.contains(c), "{why}");
}
let no_userns = KernelReport {
user_namespaces: false,
..full()
};
assert!(refusal(HeimdallRunner::with_kernel(
cfg(),
BootKey::generate(),
no_userns
))
.contains("user namespaces"));
}
#[test]
fn every_missing_feature_is_named_at_once() {
let k = KernelReport {
landlock_abi: None,
user_namespaces: false,
..full()
};
let why = refusal(HeimdallRunner::with_kernel(cfg(), BootKey::generate(), k));
assert!(
why.contains("Landlock") && why.contains("user namespaces"),
"{why}"
);
}
#[test]
fn without_seccomp_it_refuses_to_start_even_when_not_strict() {
let mut c = cfg();
c.strict = false;
let k = KernelReport {
seccomp_filter: false,
..full()
};
assert!(refusal(HeimdallRunner::with_kernel(c, BootKey::generate(), k)).contains("seccomp"));
}
#[test]
fn not_strict_without_landlock_it_starts_and_says_landlock_false() {
let mut c = cfg();
c.strict = false;
let k = KernelReport {
landlock_abi: None,
..full()
};
let r = HeimdallRunner::with_kernel(c, BootKey::generate(), k).unwrap();
assert!(!r.features().landlock);
assert_eq!(r.kernel().landlock_abi, None);
assert!(r.features().container);
}
#[test]
fn not_strict_with_landlock_off_it_says_landlock_false_on_a_landlock_kernel() {
let mut c = cfg();
c.strict = false;
c.landlock = LandlockMode::Off;
let r = HeimdallRunner::with_kernel(c, BootKey::generate(), full()).unwrap();
assert!(!r.features().landlock);
assert_eq!(r.kernel().landlock_abi, Some(8));
}
#[test]
fn landlock_net_needs_abi_4_and_addfd_is_reported() {
assert!(full().landlock_net());
let k = KernelReport {
landlock_abi: Some(3),
..full()
};
assert!(!k.landlock_net());
assert!(full().notif_addfd);
}
#[test]
fn missing_for_strict_is_empty_on_a_full_kernel() {
assert!(full().missing_for_strict().is_empty());
assert!(full().has_controller("memory"));
assert!(!full().has_controller("io"));
}
#[test]
fn probing_this_kernel_names_its_release_and_does_not_panic() {
let k = KernelReport::probe(std::path::Path::new("/sys/fs/cgroup"), "heimdall");
assert!(!k.release.is_empty());
assert!(parse_release(&k.release).is_some(), "{}", k.release);
if k.user_notif {
assert!(k.seccomp_filter);
}
if k.notif_addfd {
assert!(k.user_notif);
}
}
/// ONE kernel probe (KTOOL fix 5): the report's USER_NOTIF and ADDFD are the
/// broker's own probe (`heimdall_broker::Support`), which asks the kernel,
/// never a guess from the release string.
#[test]
fn the_notifier_features_are_the_brokers_probe() {
let root = std::path::Path::new("/sys/fs/cgroup");
let real = heimdall_broker::Support::probe();
let k = KernelReport::probe(root, "heimdall");
assert_eq!(k.user_notif, real.user_notif);
assert_eq!(k.notif_addfd, real.user_notif && real.addfd);
let no_addfd = heimdall_broker::Support {
user_notif: true,
addfd: false,
..real
};
let k = KernelReport::probe_with(root, "heimdall", no_addfd);
assert!(
!k.notif_addfd,
"the broker found no ADDFD, the report must not claim it"
);
assert!(k.user_notif);
let with_addfd = heimdall_broker::Support {
user_notif: true,
addfd: true,
..real
};
assert!(KernelReport::probe_with(root, "heimdall", with_addfd).notif_addfd);
let no_notif = heimdall_broker::Support {
user_notif: false,
addfd: false,
..real
};
let k = KernelReport::probe_with(root, "heimdall", no_notif);
assert!(!k.user_notif && !k.notif_addfd);
}
/// `heimdall kernel needs` (KTOOL fix 7): every check strict mode refuses on
/// has its kconfig lines, and the document validates.
#[test]
fn every_strict_check_has_its_kconfig_lines() {
use heimdall_container::kernel::{strict_kconfig_needs, KernelNeeds, NeedOp, NeededBy};
let needs = strict_kconfig_needs();
let has = |sym: &str| {
needs
.iter()
.any(|n| n.symbol == sym && n.op == NeedOp::Set && n.value == "y")
};
// One report per strict check, each missing only that feature: its lines
// are named, and the line set is exactly what the checks plus the job and
// the broker need.
let cases: [(&str, KernelReport, &[&str]); 7] = [
(
"seccomp_filter",
KernelReport {
seccomp_filter: false,
..full()
},
&["CONFIG_SECCOMP", "CONFIG_SECCOMP_FILTER"],
),
(
"landlock",
KernelReport {
landlock_abi: None,
..full()
},
&[
"CONFIG_MULTIUSER",
"CONFIG_SECURITY",
"CONFIG_SECURITY_LANDLOCK",
],
),
(
"cgroup_v2",
KernelReport {
cgroup_v2: false,
..full()
},
&["CONFIG_CGROUPS"],
),
(
"cgroup.memory",
KernelReport {
cgroup_controllers: vec!["cpu".into(), "pids".into()],
..full()
},
&["CONFIG_MEMCG"],
),
(
"cgroup.pids",
KernelReport {
cgroup_controllers: vec!["cpu".into(), "memory".into()],
..full()
},
&["CONFIG_CGROUP_PIDS"],
),
(
"cgroup.cpu",
KernelReport {
cgroup_controllers: vec!["memory".into(), "pids".into()],
..full()
},
&[
"CONFIG_CGROUP_SCHED",
"CONFIG_FAIR_GROUP_SCHED",
"CONFIG_CFS_BANDWIDTH",
],
),
(
"user_namespaces",
KernelReport {
user_namespaces: false,
..full()
},
&["CONFIG_MULTIUSER", "CONFIG_NAMESPACES", "CONFIG_USER_NS"],
),
];
for (feature, report, syms) in cases {
assert_eq!(report.missing_for_strict().len(), 1, "{feature}");
for s in syms {
assert!(has(s), "{feature}: {s} missing from the needs");
assert!(
needs
.iter()
.any(|n| n.symbol == *s && n.needed_by == NeededBy::Strict),
"{feature}: {s} is a strict need"
);
}
assert!(
needs.iter().any(|n| n.feature == feature),
"{feature} named"
);
}
assert!(needs.iter().any(|n| n.symbol == "CONFIG_LSM"
&& n.op == NeedOp::ListContains
&& n.value == "landlock"));
assert!(needs
.iter()
.any(|n| n.symbol == "CONFIG_CROSS_MEMORY_ATTACH" && n.needed_by == NeededBy::Broker));
for s in [
"CONFIG_PID_NS",
"CONFIG_IPC_NS",
"CONFIG_UTS_NS",
"CONFIG_NET_NS",
] {
assert!(
needs
.iter()
.any(|n| n.symbol == s && n.needed_by == NeededBy::Job),
"{s}"
);
}
assert!(needs.iter().all(|n| !n.reason.is_empty()));
let doc = KernelNeeds::strict();
doc.validate().unwrap();
let back: KernelNeeds = serde_json::from_str(&serde_json::to_string(&doc).unwrap()).unwrap();
assert_eq!(back, doc);
}
/// The validator refuses by name.
#[test]
fn the_kernel_needs_validator_refuses_by_name() {
use heimdall_container::kernel::{KernelNeeds, NeedOp};
let ok = KernelNeeds::strict();
let bad = |f: &dyn Fn(&mut KernelNeeds), word: &str| {
let mut d = ok.clone();
f(&mut d);
let e = d.validate().unwrap_err();
assert!(e.contains(word), "{e} should name {word}");
};
bad(&|d| d.schema = "nornir.kernel-needs/1".into(), "schema");
bad(&|d| d.needs.clear(), "no needs");
bad(&|d| d.needs[0].symbol = "SECCOMP".into(), "CONFIG_");
bad(&|d| d.needs[0].reason.clear(), "reason");
bad(&|d| d.needs[0].value = "m".into(), "value");
bad(
&|d| {
let n = d.needs[0].clone();
d.needs.push(n)
},
"twice",
);
bad(
&|d| {
d.needs[0].op = NeedOp::ListContains;
d.needs[0].value = "a,b".into()
},
"member",
);
}
#[test]
fn the_admitted_features_are_what_the_runner_reports() {
let f = admit(&cfg(), &full()).unwrap();
let r = HeimdallRunner::with_kernel(cfg(), BootKey::generate(), full()).unwrap();
assert_eq!(f, r.features());
}
#[test]
fn the_broker_mode_follows_the_kernel_full_check_only_deny_only_or_off() {
use heimdall_broker::Mode;
use heimdall_container::brokered::mode_for;
let k = full();
assert_eq!(mode_for(BrokerSetting::Auto, &k), Some(Mode::Full));
let no_addfd = KernelReport {
notif_addfd: false,
..full()
};
assert_eq!(
mode_for(BrokerSetting::Auto, &no_addfd),
Some(Mode::CheckOnly)
);
let no_notif = KernelReport {
user_notif: false,
notif_addfd: false,
..full()
};
assert_eq!(
mode_for(BrokerSetting::Auto, &no_notif),
Some(Mode::DenyOnly)
);
assert_eq!(mode_for(BrokerSetting::Off, &k), None);
// Features.broker is true only in full mode; strict mode does not refuse
// a kernel without the broker (spec §10: the fallbacks).
for (k, b) in [(full(), true), (no_addfd, false), (no_notif, false)] {
let r = HeimdallRunner::with_kernel(cfg(), BootKey::generate(), k).unwrap();
assert_eq!(r.features().broker, b);
}
let mut c = cfg();
c.broker = BrokerSetting::Off;
assert!(
!HeimdallRunner::with_kernel(c, BootKey::generate(), full())
.unwrap()
.features()
.broker
);
}
#[test]
fn the_job_broker_policy_is_never_wider_than_its_landlock_plan() {
use heimdall_broker::Access;
let id = JobId::random();
let p = heimdall_container::brokered::policy(id, &["/data".into()], &["/scratch".into()], &[]);
let flags = |w: bool| {
if w {
(libc::O_WRONLY | libc::O_CREAT) as u64
} else {
libc::O_RDONLY as u64
}
};
// Writable: /out, /tmp, /dev/null, policy.write. Readable: never /out.
for (path, read, write) in [
("/out/x", false, true),
("/tmp/x", true, true),
("/work/x", true, true),
("/in/x", true, false),
("/etc/x", true, false),
("/proc/self/status", true, false),
("/data/x", true, false),
("/scratch/x", true, true),
] {
assert_eq!(
p.check_open(path.as_bytes(), flags(false), 0).is_ok(),
read,
"read {path}"
);
assert_eq!(
p.check_open(path.as_bytes(), flags(true), 0o644).is_ok(),
write,
"write {path}"
);
}
assert!(
p.check_open(b"/", (libc::O_RDONLY | libc::O_DIRECTORY) as u64, 0)
.is_err(),
"never / itself"
);
assert!(p.check_open(b"/dev/null", libc::O_WRONLY as u64, 0).is_ok());
assert!(p
.check_open(b"/dev/urandom", libc::O_WRONLY as u64, 0)
.is_err());
assert!(p
.prefixes
.iter()
.all(|x| x.access != Access::WriteOnly || x.path == std::path::Path::new("/out")));
assert!(p.fetch.is_empty());
assert!(p
.check_socket(libc::AF_INET as u64, libc::SOCK_STREAM as u64, 0)
.is_err());
}
#[test]
fn a_fetch_jobs_declared_peers_become_the_broker_policy_and_the_jobs_peer_list() {
use heimdall_container::brokered::{peers_env, policy};
let n = NetAllow {
peers: vec![
NetPeer::new("pypi.org", "10.1.2.3", 443),
NetPeer::new("files.pythonhosted.org", "2001:db8::7", 443),
],
};
let addrs = n.check().unwrap();
let p = policy(JobId::random(), &[], &[], &addrs);
assert_eq!(p.fetch, addrs);
let tcp = libc::SOCK_STREAM as u64;
assert!(p.check_socket(libc::AF_INET as u64, tcp, 0).is_ok());
assert!(
p.check_socket(libc::AF_INET6 as u64, tcp, 0).is_ok(),
"a v6 peer is declared"
);
assert!(p
.check_socket(libc::AF_INET as u64, libc::SOCK_RAW as u64, 0)
.is_err());
let sa = |a: &str| heimdall_broker::policy::sockaddr_bytes(a.parse().unwrap());
assert!(p.check_connect(&sa("10.1.2.3:443")).is_ok());
assert!(p.check_connect(&sa("[2001:db8::7]:443")).is_ok());
assert!(p.check_connect(&sa("10.1.2.3:444")).is_err());
assert!(p.check_connect(&sa("10.1.2.4:443")).is_err());
// The job's peer list: HEIMDALL_NET_PEERS=<the declared peers as JSON>.
let e = peers_env(&n);
let (k, v) = e.split_once('=').unwrap();
assert_eq!(k, "HEIMDALL_NET_PEERS");
let back: Vec<NetPeer> = serde_json::from_str(v).unwrap();
assert_eq!(back, n.peers);
}