crates/heimdall-probe/tests/minimise.rs
// SPDX-License-Identifier: AGPL-3.0-only
//! Merge, minimise, codegen and the drift check on synthetic observations
//! (no kernel needed). Heimdall is a defensive sandbox; these build and
//! check its own profiles.
use heimdall_probe::codegen::{files, to_rust};
use heimdall_probe::observe::RunInfo;
use heimdall_probe::{check, generate, Call, GenOptions, Generated, Observations};
use heimdall_seccomp::{Action, Arch, ArgCmp, Libc, Width};
use proptest::prelude::*;
fn call(s: &str, args: &[(u8, u64)]) -> Call {
Call {
syscall: s.into(),
args: args.to_vec(),
}
}
fn obs() -> Observations {
let mut o = Observations::new("w", Arch::X86_64, Libc::Glibc);
for (name, err) in [("happy", false), ("fuzz_bad_input", true)] {
o.runs.insert(RunInfo {
name: name.into(),
error_path: err,
exit: "exit 0".into(),
});
}
o.record(call("read", &[]), "happy");
o.record(call("read", &[]), "fuzz_bad_input");
o.record(call("openat", &[(2, 0x80000)]), "happy");
o.record(call("openat", &[(2, 0x80241)]), "happy");
o.record(call("socket", &[(0, 1), (1, 0x80001), (2, 0)]), "happy");
o.record(call("socket", &[(0, 10), (1, 2), (2, 17)]), "happy");
o.record(call("mprotect", &[(2, 1)]), "happy");
o.record(call("mprotect", &[(2, 7)]), "happy");
o.record(call("clone", &[(0, 0x3d0f00)]), "happy");
o.record(call("getrusage", &[]), "fuzz_bad_input");
o
}
fn rule<'a>(g: &'a Generated, name: &str) -> &'a heimdall_seccomp::Rule {
g.profile
.rules
.iter()
.find(|r| r.syscall == name)
.unwrap_or_else(|| panic!("no rule for {name}"))
}
#[test]
fn a_call_without_recorded_arguments_is_a_bare_allow() {
let g = generate(&obs(), GenOptions::default());
assert_eq!(rule(&g, "read").when, Vec::<Vec<ArgCmp>>::new());
assert_eq!(rule(&g, "read").action, Action::Allow);
}
#[test]
fn openat_flags_become_a_mask_of_the_union_of_the_observed_bits() {
let g = generate(&obs(), GenOptions::default());
let r = rule(&g, "openat");
assert_eq!(r.when, vec![vec![ArgCmp::only_bits32(2, 0x80241)]]);
let nr = Arch::X86_64.syscall_nr("openat").unwrap();
let kill = Action::KillProcess;
assert_eq!(
g.profile.evaluate(nr, &[0, 0, 0x80000, 0, 0, 0], kill),
Action::Allow
);
assert_eq!(
g.profile.evaluate(nr, &[0, 0, 0x00040, 0, 0, 0], kill),
Action::Allow
);
assert_eq!(
g.profile.evaluate(nr, &[0, 0, 0x00002, 0, 0, 0], kill),
kill,
"O_RDWR never seen"
);
}
#[test]
fn socket_tuples_are_kept_exactly() {
let g = generate(&obs(), GenOptions::default());
let r = rule(&g, "socket");
assert_eq!(r.when.len(), 2);
let nr = Arch::X86_64.syscall_nr("socket").unwrap();
let kill = Action::KillProcess;
assert_eq!(
g.profile.evaluate(nr, &[1, 0x80001, 0, 0, 0, 0], kill),
Action::Allow
);
assert_eq!(
g.profile.evaluate(nr, &[10, 2, 17, 0, 0, 0], kill),
Action::Allow
);
assert_eq!(
g.profile.evaluate(nr, &[10, 1, 0, 0, 0, 0], kill),
kill,
"AF_INET6 stream never seen"
);
assert_eq!(
g.profile.evaluate(nr, &[1, 2, 17, 0, 0, 0], kill),
kill,
"mixed tuple never seen"
);
}
#[test]
fn clone_flags_are_a_64_bit_mask() {
let g = generate(&obs(), GenOptions::default());
assert_eq!(
rule(&g, "clone").when,
vec![vec![ArgCmp::only_bits64(0, 0x3d0f00)]]
);
}
#[test]
fn a_write_exec_mapping_is_reported_and_left_out() {
let g = generate(&obs(), GenOptions::default());
let r = rule(&g, "mprotect");
assert_eq!(r.when, vec![vec![ArgCmp::eq32(2, 1)]]);
assert_eq!(g.review.wx.len(), 1);
assert_eq!(g.review.wx[0].call, "mprotect arg2=0x7");
assert_eq!(g.review.wx[0].runs, vec!["happy"]);
}
#[test]
fn allow_wx_keeps_a_write_exec_mapping() {
let g = generate(
&obs(),
GenOptions {
allow_wx: true,
..GenOptions::default()
},
);
assert_eq!(rule(&g, "mprotect").when.len(), 2);
assert!(g.review.wx.is_empty());
}
#[test]
fn a_call_with_too_many_distinct_values_is_a_reported_bare_allow() {
let mut o = obs();
for i in 0..20 {
o.record(call("ioctl", &[(1, 0x5400 + i)]), "happy");
}
let g = generate(
&o,
GenOptions {
exact_limit: 16,
..GenOptions::default()
},
);
assert!(rule(&g, "ioctl").when.is_empty());
assert!(
g.review
.bare
.iter()
.any(|b| b.starts_with("ioctl: 20 distinct")),
"{:?}",
g.review.bare
);
let g = generate(
&o,
GenOptions {
exact_limit: 20,
..GenOptions::default()
},
);
assert_eq!(rule(&g, "ioctl").when.len(), 20);
}
#[test]
fn near_miss_lists_calls_seen_only_on_error_paths() {
let g = generate(&obs(), GenOptions::default());
assert_eq!(g.review.near_miss.len(), 1);
assert_eq!(g.review.near_miss[0].call, "getrusage");
assert_eq!(g.review.near_miss[0].runs, vec!["fuzz_bad_input"]);
// Kept in the profile: production must not die on its error paths.
rule(&g, "getrusage");
}
#[test]
fn clone3_is_a_fixed_enosys_rule() {
let g = generate(&obs(), GenOptions::default());
assert_eq!(
rule(&g, "clone3").action,
Action::Errno(libc::ENOSYS as u16)
);
assert!(g.review.fixed[0].starts_with("clone3"));
}
#[test]
fn generating_twice_gives_byte_identical_files() {
let a = files(&generate(&obs(), GenOptions::default()));
let b = files(&generate(&obs(), GenOptions::default()));
assert_eq!(a, b);
}
#[test]
fn the_record_order_does_not_change_the_files() {
let o = obs();
let mut rev = Observations::new("w", Arch::X86_64, Libc::Glibc);
rev.runs = o.runs.clone();
for (c, runs) in o.calls.iter().rev() {
for r in runs.iter().rev() {
rev.record(c.clone(), r);
}
}
assert_eq!(
files(&generate(&o, GenOptions::default())),
files(&generate(&rev, GenOptions::default()))
);
}
#[test]
fn merge_is_the_union_of_runs_and_calls() {
let mut a = Observations::new("w", Arch::X86_64, Libc::Glibc);
a.record(call("read", &[]), "one");
let mut b = Observations::new("w", Arch::X86_64, Libc::Glibc);
b.record(call("read", &[]), "two");
b.record(call("write", &[]), "two");
a.merge(&b);
assert_eq!(a.calls.len(), 2);
assert_eq!(a.calls[&call("read", &[])].len(), 2);
}
#[test]
fn observations_json_round_trips() {
let mut o = obs();
o.unknown
.entry("nr 999".into())
.or_default()
.insert("happy".into());
let j = o.to_json();
assert_eq!(Observations::from_json(&j).unwrap(), o);
assert_eq!(Observations::from_json(&j).unwrap().to_json(), j);
}
#[test]
fn generated_json_round_trips_and_reads_as_a_plain_profile() {
let g = generate(&obs(), GenOptions::default());
let j = g.to_json();
assert_eq!(Generated::from_json(&j).unwrap(), g);
assert_eq!(heimdall_seccomp::Profile::from_json(&j).unwrap(), g.profile);
}
#[test]
fn the_rust_file_holds_the_const_table_and_the_review() {
let g = generate(&obs(), GenOptions::default());
let rs = to_rust(&g);
assert!(rs.starts_with(heimdall_probe::codegen::GENERATED_HEADER));
assert!(rs.contains("\n// GENERATED by heimdall-probe"));
assert!(rs.contains("pub const PROFILE: heimdall_seccomp::StaticProfile"));
assert!(rs.contains("syscall: \"openat\""));
assert!(rs.contains("heimdall_seccomp::ArgCmp { arg: 2, width: heimdall_seccomp::Width::W32, mask: 0xfff7fdbe, value: 0x0 }"));
assert!(rs.contains("action: heimdall_seccomp::Action::Errno(38)"));
assert!(
rs.contains("// getrusage (by fuzz_bad_input)"),
"near-miss in the header"
);
assert!(
rs.contains("// mprotect arg2=0x7 (by happy)"),
"W|X in the header"
);
assert!(!rs.contains("//!"), "inner doc comments break include!");
}
#[test]
fn check_passes_when_every_observed_call_is_allowed() {
let o = obs();
let g = generate(&o, GenOptions::default());
let mut o2 = o.clone();
o2.calls
.retain(|c, _| c.syscall != "mprotect" || c.arg(2) != Some(7));
assert_eq!(check(&o2, &g.profile), vec![]);
}
#[test]
fn check_names_a_new_call_and_the_test_that_made_it() {
let g = generate(&obs(), GenOptions::default());
let mut o = obs();
o.record(call("sysinfo", &[]), "happy");
o.record(call("openat", &[(2, 0x2)]), "fuzz_bad_input");
let v = check(&o, &g.profile);
let names: Vec<String> = v.iter().map(|x| x.to_string()).collect();
assert!(
names
.iter()
.any(|s| s.contains("sysinfo") && s.contains("made by test happy")),
"{names:?}"
);
assert!(
names
.iter()
.any(|s| s.contains("openat arg2=0x2") && s.contains("fuzz_bad_input")),
"{names:?}"
);
assert!(
names.iter().any(|s| s.contains("mprotect arg2=0x7")),
"the W|X call is a violation too: {names:?}"
);
}
#[test]
fn check_refuses_unknown_numbers_and_a_wrong_arch() {
let g = generate(&obs(), GenOptions::default());
let mut o = obs();
o.unknown
.entry("nr 4242".into())
.or_default()
.insert("happy".into());
assert!(check(&o, &g.profile).iter().any(|v| v.call == "nr 4242"));
let other = Observations::new("w", Arch::Aarch64, Libc::Glibc);
assert_eq!(check(&other, &g.profile).len(), 1);
}
proptest! {
/// Whatever was observed, the generated profile allows all of it except
/// W|X mappings (generate, then check: no violations but those).
#[test]
fn the_generated_profile_allows_everything_it_was_generated_from(
calls in prop::collection::vec(
(prop::sample::select(&["read", "openat", "socket", "ioctl", "mmap", "mprotect", "clone", "prctl", "fcntl"][..]),
prop::array::uniform3(any::<u64>())), 0..40),
limit in 1usize..20,
) {
let mut o = Observations::new("p", Arch::X86_64, Libc::Musl);
for (name, vals) in &calls {
let args: Vec<(u8, u64)> = heimdall_probe::facts::spec(name)
.map(|s| s.args.iter().zip(vals).map(|(&(i, w), v)| (i, if w == Width::W32 { v & 0xffff_ffff } else { *v })).collect())
.unwrap_or_default();
o.record(Call { syscall: name.to_string(), args }, "t");
}
let g = generate(&o, GenOptions { exact_limit: limit, allow_wx: false });
prop_assert!(g.profile.validate().is_ok());
for v in check(&o, &g.profile) {
prop_assert!(g.review.wx.iter().any(|w| w.call == v.call), "{} is not allowed", v);
}
}
}
#[test]
fn a_generated_profile_is_not_under_heimdalls_agpl_but_mit_0_usable_in_any_repo() {
let rs = to_rust(&generate(&obs(), GenOptions::default()));
let first: Vec<&str> = rs.lines().take(3).collect();
assert_eq!(first[0], "// SPDX-License-Identifier: MIT-0");
assert!(
first[1].contains("generated output of Heimdall"),
"{first:?}"
);
assert!(
first[2].contains("not covered by Heimdall's AGPL"),
"{first:?}"
);
assert!(!rs.contains("AGPL-3.0-only"), "{rs}");
// The JSON forms carry no licence header at all (data only).
let g = generate(&obs(), GenOptions::default());
let (_, json, oci) = heimdall_probe::codegen::files(&g);
assert!(!json.contains("AGPL") && !oci.contains("AGPL"));
}