guests/verify-bundle/tests/split_wasm.rs
// SPDX-License-Identifier: AGPL-3.0-only
//! The split guest in the real host (docs/GATLING-WASM.md): the VERIFIER
//! component, and N isolated DECODER instances the host drives with gatling.
//! Heimdall is a defensive sandbox; the lying decoders here are test-only
//! builds that prove the host and the verifier catch them, and the limit
//! tests prove each decoder instance is held by its own limits.
#![cfg(feature = "openzl")]
mod common;
use common::guests::{component, core_module, evil_component, imports, loaded, Build, Lab};
use common::zwriter::RawArchive;
use common::*;
use std::time::{Duration, Instant};
use heimdall_wasm::{codes, GuestOutcome, RunLimits, Verdict};
fn job_params(fx: &Fixture) -> heimdall_core::Params {
let mut p = heimdall_core::Params::new();
p.insert("now".into(), t0().to_rfc3339());
p.insert("policy".into(), fx.policy.to_ron().unwrap());
p
}
fn files(o: &GuestOutcome) -> Vec<(String, Vec<u8>)> {
o.files
.iter()
.map(|f| (f.name.clone(), f.bytes.clone()))
.collect()
}
/// A bundle with many OpenZL chunks (4 KiB each): several gatling units.
fn many_chunk_bundle() -> (Fixture, Vec<u8>) {
let fx = build(Spec {
payloads: vec![(
"firmware.bin".into(),
(0..1_200_000u32)
.map(|i| (i % 251) as u8 ^ (i >> 9) as u8)
.collect(),
"application/octet-stream".into(),
)],
..Spec::default()
});
let a = RawArchive::from_members_openzl(&fx.members, 4096).encode();
(fx, a)
}
fn limits(decoders: usize) -> RunLimits {
RunLimits {
decoders,
..RunLimits::default()
}
}
// ---------------------------------------------------------------- imports
#[test]
fn the_verifier_module_imports_only_emit_file_decode_chunks_and_digest_members_and_links_no_c() {
let mut i = imports(core_module(Build::Verifier));
i.sort();
assert_eq!(
i,
vec![
"$root::decode-chunks",
"$root::digest-members",
"$root::emit-file"
]
);
// No OpenZL C in the verifier: it is about 1 MB smaller than the
// single-instance guest that links the decoder.
let (v, j) = (
core_module(Build::Verifier).len(),
core_module(Build::Job).len(),
);
assert!(
v + 700_000 < j,
"verifier {v} B, single-instance guest {j} B"
);
}
#[test]
fn the_decoder_module_imports_nothing_at_all() {
assert_eq!(imports(core_module(Build::Decoder)), Vec::<String>::new());
assert_eq!(
imports(core_module(Build::EvilDecoder)),
Vec::<String>::new()
);
}
#[test]
fn the_host_refuses_a_verifier_loaded_without_its_decoder() {
let lab = Lab::new();
let p = lab.signed("verifier", component(Build::Verifier));
let e = lab.host.load(&p).err().expect("refused");
assert_eq!(e.code(), codes::LOAD_IMPORT_REFUSED, "{e}");
assert!(e.to_string().contains("decode-chunks"), "{e}");
}
#[test]
fn the_host_refuses_a_decoder_that_imports_anything_or_lacks_decode_chunks() {
let lab = Lab::new();
for (name, b, code) in [
("job", Build::Job, codes::LOAD_IMPORT_REFUSED),
("verifier", Build::Verifier, codes::LOAD_IMPORT_REFUSED),
] {
let p = lab.signed(name, component(b));
let e = lab.host.load_decoder(&p).err().expect("refused");
assert_eq!(e.code(), code, "{name}: {e}");
}
// A decoder is not a job: loaded as one it has no `run`.
let p = lab.signed("decoder", component(Build::Decoder));
let e = lab.host.load(&p).err().expect("refused");
assert_eq!(e.code(), codes::LOAD_MISSING_EXPORT, "{e}");
}
#[test]
fn an_unsigned_decoder_is_never_loaded() {
let lab = Lab::new();
let p = lab.signed("decoder", component(Build::Decoder));
std::fs::remove_file(lab.dir.path().join("decoder.wasm.p7s")).unwrap();
let e = lab.host.load_decoder(&p).err().expect("refused");
assert_eq!(e.code(), codes::LOAD_UNTRUSTED, "{e}");
}
// ---------------------------------------------------------------- verdicts
#[test]
fn a_many_chunk_openzl_bundle_is_accepted_by_the_split_guest_with_the_single_instance_files() {
let (fx, a) = many_chunk_bundle();
let l = loaded();
let single = l.job.run(&a, &job_params(&fx), &RunLimits::default());
let split = l.split.run(&a, &job_params(&fx), &limits(8));
assert!(
matches!(split.verdict, Verdict::Accepted { .. }),
"{}",
split.verdict
);
assert_eq!(split.verdict, single.verdict);
assert_eq!(files(&split), files(&single));
}
#[test]
fn the_split_guest_gives_the_same_verdict_files_and_fuel_for_every_decoder_count_1_to_16() {
let (fx, a) = many_chunk_bundle();
let mut tampered = RawArchive::from_members_openzl(&fx.members, 4096);
let n = tampered.rows.len();
// A bit flip in a late chunk's frame: refused the same way for every N.
let r = tampered.rows[n - 2].clone();
tampered.blobs[r.blob_offset as usize + 7] ^= 0x40;
let tampered = tampered.encode();
let l = loaded();
for archive in [&a, &tampered] {
let first = l.split.run(archive, &job_params(&fx), &limits(1));
for n in 2..=16 {
let o = l.split.run(archive, &job_params(&fx), &limits(n));
assert_eq!(o.verdict, first.verdict, "N={n}");
assert_eq!(files(&o), files(&first), "N={n}");
assert_eq!(
o.fuel_used, first.fuel_used,
"N={n}: fuel is per unit, not per worker"
);
}
}
}
#[test]
fn the_seed_corpus_gets_the_same_verdicts_from_both_guests() {
let fx = good();
let dir = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("fuzz/seeds");
let l = loaded();
let mut n = 0;
for sub in ["archive", "bundle"] {
for e in std::fs::read_dir(dir.join(sub)).unwrap() {
let bytes = std::fs::read(e.unwrap().path()).unwrap();
let s = l.job.run(&bytes, &job_params(&fx), &RunLimits::default());
let t = l.split.run(&bytes, &job_params(&fx), &RunLimits::default());
assert_eq!(s.verdict, t.verdict);
n += 1;
}
}
assert!(n >= 8);
}
// ---------------------------------------------------------------- liars
fn evil_run(mode: &[u8; 8], l: &RunLimits) -> GuestOutcome {
evil_run_timed(mode, l).0
}
/// The run and how long the run alone took (loading is not counted).
fn evil_run_timed(mode: &[u8; 8], l: &RunLimits) -> (GuestOutcome, Duration) {
let lab = Lab::new();
let d = lab.decoder_from("evil", &evil_component(mode));
let g = lab.split_with(d);
let fx = good();
let (a, p) = (pack_holger(&fx.members), job_params(&fx));
let t = Instant::now();
let o = g.run(&a, &p, l);
(o, t.elapsed())
}
fn assert_code(o: &GuestOutcome, code: &str, needle: &str) {
assert_eq!(o.verdict.code(), Some(code), "{}", o.verdict);
let r = o.verdict.reason().unwrap();
assert!(r.contains(needle), "{r}");
assert!(o.files.is_empty());
}
#[test]
fn the_evil_decoder_build_is_honest_in_its_honest_mode() {
let o = evil_run(b"honest__", &RunLimits::default());
assert!(
matches!(o.verdict, Verdict::Accepted { .. }),
"{}",
o.verdict
);
}
#[test]
fn a_decoder_instance_returning_wrong_bytes_under_the_right_digest_is_refused_by_the_host() {
let o = evil_run(b"flipbyte", &RunLimits::default());
assert_code(
&o,
codes::DECODER_MISMATCH,
"is not the BLAKE3 of the bytes returned",
);
}
#[test]
fn a_decoder_instance_returning_a_wrong_digest_is_refused_by_the_host() {
let o = evil_run(b"baddigst", &RunLimits::default());
assert_code(
&o,
codes::DECODER_MISMATCH,
"is not the BLAKE3 of the bytes returned",
);
}
#[test]
fn a_decoder_instance_returning_a_short_chunk_is_refused_by_the_host() {
let o = evil_run(b"shortone", &RunLimits::default());
assert_code(&o, codes::DECODER_MISMATCH, "declared");
}
#[test]
fn a_decoder_instance_dropping_an_answer_is_refused_by_the_host() {
let o = evil_run(b"dropone_", &RunLimits::default());
assert_code(&o, codes::DECODER_MISMATCH, "answers for");
}
#[test]
fn a_decoder_instance_lying_consistently_is_caught_by_the_verifier_with_holgers_code() {
let o = evil_run(b"consistn", &RunLimits::default());
assert_code(
&o,
codes::GUEST_REFUSED,
"does not match its blake3 checksum",
);
assert!(o
.verdict
.reason()
.unwrap()
.starts_with(&format!("[{}]", holger_errcode::UPDATE_CONTAINER.code)));
}
#[test]
fn a_trapping_decoder_instance_ends_the_job_as_a_trap_naming_the_decoder() {
let o = evil_run(b"trap____", &RunLimits::default());
assert_code(&o, codes::TRAPPED, "decoder instance");
}
// ---------------------------------------------------------------- limits
#[test]
fn a_spinning_decoder_instance_runs_out_of_its_own_fuel() {
let l = RunLimits {
fuel: 200_000_000,
..RunLimits::default()
};
let o = evil_run(b"spin____", &l);
assert_code(&o, codes::OUT_OF_FUEL, "decoder instance");
assert!(o.verdict.reason().unwrap().contains("200000000"));
}
#[test]
fn a_spinning_decoder_instance_with_unmetered_fuel_is_stopped_by_the_jobs_wall_clock() {
let l = RunLimits {
fuel: u64::MAX,
timeout: Duration::from_millis(1500),
..RunLimits::default()
};
let (o, took) = evil_run_timed(b"spin____", &l);
assert_code(&o, codes::TIMEOUT, "decoder instance");
assert!(took < Duration::from_secs(10), "{took:?}");
}
#[test]
fn a_memory_hogging_decoder_instance_hits_its_own_memory_limit() {
let l = RunLimits {
memory_bytes: 96 << 20,
..RunLimits::default()
};
let o = evil_run(b"hog_____", &l);
assert_code(&o, codes::MEMORY_LIMIT, "decoder instance");
assert!(o
.verdict
.reason()
.unwrap()
.contains(&format!("limit {}", 96 << 20)));
}
#[test]
fn each_decoder_instance_has_its_own_fuel_so_the_units_together_may_burn_more_than_one_budget() {
let (fx, _) = many_chunk_bundle();
let ra = RawArchive::from_members_openzl(&fx.members, 4096);
let frames: Vec<(&[u8], u64)> = ra
.rows
.iter()
.filter(|r| r.compressed)
.map(|r| {
(
&ra.blobs[r.blob_offset as usize..(r.blob_offset + r.blob_size) as usize],
r.uncompressed_size,
)
})
.collect();
let d = loaded().split.decoder().unwrap();
let all = d.decode(&frames, &limits(4));
assert!(all.result.is_ok());
assert!(all.units >= 4, "{} units", all.units);
// Half of what all units burn together is still enough for each one.
let per_instance = RunLimits {
fuel: all.fuel_used / 2,
decoders: 4,
..RunLimits::default()
};
let r = d.decode(&frames, &per_instance);
assert!(r.result.is_ok(), "{:?}", r.result.err());
assert!(r.fuel_used > per_instance.fuel);
// And a budget below one unit's need stops it, by name.
let starved = RunLimits {
fuel: 1000,
..per_instance
};
let e = d.decode(&frames, &starved).result.unwrap_err();
assert_eq!(e.code, codes::OUT_OF_FUEL);
assert!(
e.reason.starts_with("decoder instance for chunks 0.."),
"{}",
e.reason
);
}
#[test]
fn the_decoder_api_decodes_frames_to_the_original_bytes_under_any_worker_count() {
let l = loaded();
let d = l.split.decoder().unwrap();
let pieces: Vec<Vec<u8>> = (0..150u32)
.map(|i| (0..(3000 + i * 7)).map(|j| (j % 13 + i) as u8).collect())
.collect();
let frames: Vec<Vec<u8>> = pieces.iter().map(|p| zwriter::openzl_frame(p)).collect();
let refs: Vec<(&[u8], u64)> = frames
.iter()
.zip(&pieces)
.map(|(f, p)| (&f[..], p.len() as u64))
.collect();
let first = d.decode(&refs, &limits(1));
for n in [1, 3, 8, 16] {
let r = d.decode(&refs, &limits(n));
assert_eq!(r, first, "N={n}");
let answers = r.result.unwrap();
for (a, p) in answers.iter().zip(&pieces) {
let a = a.as_ref().unwrap();
assert_eq!(&a.bytes, p);
assert_eq!(a.digest, blake3::hash(p).as_bytes().to_vec());
}
}
assert_eq!(first.units, 3, "150 chunks = units of 64, 64, 22");
}
#[test]
fn the_fuzz_found_allocation_frame_is_a_per_chunk_refusal_in_a_decoder_instance() {
let l = loaded();
let d = l.split.decoder().unwrap();
let frame = include_bytes!("../fuzz/seeds/openzl_frame/regress-constant-stream-alloc");
let r = d.decode(&[(&frame[..], 0)], &RunLimits::default());
let answers = r.result.expect("an answer, not a trap");
let why = answers[0].as_ref().expect_err("refused");
assert!(why.contains("Allocation"), "{why}");
}
#[test]
fn frames_whose_graph_asks_for_hundreds_of_megabytes_hit_the_decoder_instances_own_memory_limit() {
// Two finds of the decoder_instance fuzz run (2026-10-31): small frames
// whose OpenZL graph grows the instance's memory to ~274 MB and ~2 GB.
// Contained by the instance's own limit, named, never a host fault.
let d = loaded().split.decoder().unwrap();
let dir = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("fuzz/seeds/openzl_frame");
for name in [
"regress-wasm-internal-alloc-274mb",
"regress-wasm-internal-alloc-2gb",
] {
let frame = std::fs::read(dir.join(name)).unwrap();
let size = heimdall_guest_verify_bundle::zl::frame_size(&frame).unwrap();
let e = d
.decode(&[(&frame[..], size)], &RunLimits::default())
.result
.expect_err(name);
assert_eq!(e.code, codes::MEMORY_LIMIT, "{name}: {}", e.reason);
assert!(e
.reason
.starts_with("decoder instance for chunks 0..1: memory limit"));
}
}
// ---------------------------------------------------------------- measure
fn word_salad(len: usize, seed: u64) -> Vec<u8> {
const W: [&str; 16] = [
"drive ", "motor ", "torque ", "ACS880 ", "R4 ", "fault ", "0x3F ", "speed=", "1450 ",
"rpm\n", "param ", "ok ", "warn ", "id=", "42 ", "ramp ",
];
let mut x = 0x2545_f491_4f6c_dd1du64 ^ seed;
let mut v = Vec::with_capacity(len + 16);
while v.len() < len {
x ^= x << 13;
x ^= x >> 7;
x ^= x << 17;
v.extend_from_slice(W[(x % 16) as usize].as_bytes());
v.push((x >> 40) as u8);
}
v.truncate(len);
v
}
fn cpu_time() -> Duration {
// utime + stime of this process, in clock ticks (100 Hz on Linux).
let s = std::fs::read_to_string("/proc/self/stat").unwrap();
let rest = &s[s.rfind(')').unwrap() + 2..];
let f: Vec<&str> = rest.split(' ').collect();
let t: u64 = f[11].parse::<u64>().unwrap() + f[12].parse::<u64>().unwrap();
Duration::from_millis(t * 10)
}
/// Not a pass/fail test: wall and CPU time of the single-instance guest
/// versus the split guest at N = 1, 8, 16, on 8 MB, 64 MB and 192 MB of
/// OpenZL members in 1 MiB chunks (docs/GATLING-WASM.md).
/// `cargo test --release --test split_wasm -- --ignored --nocapture measure`
#[test]
#[ignore]
fn measure_split_versus_single_instance() {
let l = loaded();
let big = RunLimits {
fuel: u64::MAX / 4,
timeout: Duration::from_secs(900),
memory_bytes: 2 << 30,
max_file_bytes: 1 << 30,
max_emit_bytes: 2 << 30,
..RunLimits::default()
};
for (mb, members) in [(8usize, 1usize), (64, 1), (192, 6)] {
let each = mb * 1_000_000 / members;
let fx = build(Spec {
payloads: (0..members)
.map(|i| {
(
format!("part{i}.bin"),
word_salad(each, i as u64),
"application/octet-stream".into(),
)
})
.collect(),
..Spec::default()
});
let a = RawArchive::from_members_openzl(&fx.members, 1 << 20).encode();
let p = job_params(&fx);
let timed = |what: &str, f: &dyn Fn() -> GuestOutcome| {
let _ = f();
let n = if mb > 64 { 2 } else { 3 };
let (t0, c0) = (Instant::now(), cpu_time());
let mut fuel = 0;
for _ in 0..n {
let o = f();
assert!(
matches!(o.verdict, Verdict::Accepted { .. }),
"{what}: {}",
o.verdict
);
fuel = o.fuel_used;
}
let (wall, cpu) = (t0.elapsed() / n, (cpu_time() - c0) / n);
println!(
"MEASURE {mb} MB ({} B on disk): {what}: wall {:?}, cpu {:?}, fuel {fuel}",
a.len(),
wall,
cpu
);
};
timed("single-instance", &|| l.job.run(&a, &p, &big));
for n in [1usize, 8, 16] {
let lim = RunLimits {
decoders: n,
..big.clone()
};
timed(&format!("split N={n}"), &|| l.split.run(&a, &p, &lim));
}
// One 1 MiB unit's fuel (decode + BLAKE3 in one decoder instance).
let ra = RawArchive::from_members_openzl(&fx.members, 1 << 20);
let r = ra
.rows
.iter()
.filter(|r| r.compressed)
.max_by_key(|r| r.uncompressed_size)
.unwrap();
let frame = &ra.blobs[r.blob_offset as usize..(r.blob_offset + r.blob_size) as usize];
let one = l
.split
.decoder()
.unwrap()
.decode(&[(frame, r.uncompressed_size)], &big);
println!(
"MEASURE fuel for one {} B chunk in a decoder instance: {}",
r.uncompressed_size, one.fuel_used
);
}
}
/// Not a pass/fail test: replays every file in `$HEIMDALL_REPLAY_DIR` (fuzz
/// slow units, for example) through one decoder instance, as the
/// decoder_instance fuzz target does, and prints what each one costs.
/// `HEIMDALL_REPLAY_DIR=... cargo test --release --test split_wasm -- --ignored --nocapture replay`
#[test]
#[ignore]
fn replay_frames_through_a_decoder_instance() {
let Ok(dir) = std::env::var("HEIMDALL_REPLAY_DIR") else {
return;
};
let d = loaded().split.decoder().unwrap();
let limits = RunLimits {
decoders: 1,
fuel: 2_000_000_000,
timeout: Duration::from_secs(20),
..RunLimits::default()
};
for e in std::fs::read_dir(dir).unwrap() {
let p = e.unwrap().path();
let data = std::fs::read(&p).unwrap();
let Ok(size) = heimdall_guest_verify_bundle::zl::frame_size(&data) else {
continue;
};
let t = Instant::now();
let r = d.decode(&[(&data[..], size)], &limits);
let what = match &r.result {
Ok(a) => match &a[0] {
Ok(_) => "decoded".to_string(),
Err(why) => format!("refused: {why}"),
},
Err(f) => format!("{}: {}", f.code, f.reason),
};
println!(
"REPLAY {} frame {} B declares {size} B: {:?}, fuel {}: {what}",
p.file_name().unwrap().to_string_lossy(),
data.len(),
t.elapsed(),
r.fuel_used
);
}
}