server/inspect/tests/inspect.rs
//! The inspector against real hostile archives, built in the test.
//!
//! Every fixture is generated here, small in memory and never written to disk.
//! The only "malware" is the EICAR test string, assembled from two halves so
//! this source file itself is not a signature hit.
use std::io::{Cursor, Write};
use std::sync::Arc;
use holger_inspect::malware::{
ClamAv, ClamAvConfig, MalwareScanner, ScanOutcome, Yara, YaraConfig,
};
use holger_inspect::{
inspect, ExecPolicy, FindingKind, Format, Inspector, Limits, Rule, Severity, Verdict,
};
use zip::write::SimpleFileOptions;
use zip::CompressionMethod;
const MIB: u64 = 1 << 20;
fn eicar() -> Vec<u8> {
let a = "X5O!P%@AP[4\\PZX54(P^)7CC)7}$";
let b = "EICAR-STANDARD-ANTIVIRUS-TEST-FILE!$H+H*";
format!("{a}{b}").into_bytes()
}
const EICAR_RULE: &str = r#"
rule EICAR_Test_File {
meta: description = "the EICAR anti-virus test string"
strings: $a = "EICAR-STANDARD-ANTIVIRUS-TEST-FILE" $b = "X5O!P%@AP"
condition: all of them
}
"#;
fn deflated() -> SimpleFileOptions {
SimpleFileOptions::default().compression_method(CompressionMethod::Deflated)
}
/// A zip of (name, bytes) members.
fn zip_of(members: &[(&str, &[u8])]) -> Vec<u8> {
let mut w = zip::ZipWriter::new(Cursor::new(Vec::new()));
for (name, body) in members {
w.start_file(*name, deflated()).unwrap();
w.write_all(body).unwrap();
}
w.finish().unwrap().into_inner()
}
/// A zip with one member of `len` zero bytes, written in chunks.
fn zeros_zip(name: &str, len: u64) -> Vec<u8> {
let mut w = zip::ZipWriter::new(Cursor::new(Vec::new()));
w.start_file(name, deflated().large_file(len > u32::MAX as u64))
.unwrap();
let chunk = vec![0u8; MIB as usize];
for _ in 0..len / MIB {
w.write_all(&chunk).unwrap();
}
w.finish().unwrap().into_inner()
}
fn wheel(extra: &[(&str, &[u8])]) -> Vec<u8> {
let mut m: Vec<(&str, &[u8])> = vec![
("demo/__init__.py", b"print('demo')\n"),
(
"demo-1.0.dist-info/METADATA",
b"Metadata-Version: 2.1\nName: demo\nVersion: 1.0\n",
),
(
"demo-1.0.dist-info/WHEEL",
b"Wheel-Version: 1.0\nTag: py3-none-any\n",
),
];
m.extend_from_slice(extra);
zip_of(&m)
}
fn elf(e_type: u8) -> Vec<u8> {
let mut b = vec![0u8; 128];
b[..4].copy_from_slice(b"\x7fELF");
b[4] = 2; // 64-bit
b[5] = 1; // little-endian
b[6] = 1;
b[16] = e_type;
b[18] = 0x3e; // x86-64
b
}
fn tar_header(name: &[u8], kind: tar::EntryType, link: &[u8], size: u64) -> tar::Header {
let mut h = tar::Header::new_ustar();
h.as_old_mut().name[..name.len()].copy_from_slice(name);
h.as_old_mut().linkname[..link.len()].copy_from_slice(link);
h.set_entry_type(kind);
h.set_size(size);
h.set_mode(0o644);
h.set_cksum();
h
}
fn gz(bytes: &[u8]) -> Vec<u8> {
let mut e = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());
e.write_all(bytes).unwrap();
e.finish().unwrap()
}
fn blocks(f: &holger_inspect::Findings, kind: FindingKind) -> bool {
f.has(kind, Severity::Block)
}
// ── a clean artifact ────────────────────────────────────────────────────────
#[test]
fn a_clean_wheel_passes_and_every_member_is_counted() {
let f = Inspector::default().inspect(&wheel(&[]), Format::Wheel, "demo-1.0-py3-none-any.whl");
assert_eq!(f.verdict(), Verdict::Pass, "{:#?}", f.findings);
assert_eq!(f.entries, 3);
assert!(f.expanded_bytes > 0);
}
#[test]
fn a_clean_sdist_tar_gz_is_one_level() {
let mut b = tar::Builder::new(Vec::new());
let body = b"from setuptools import setup\nsetup()\n";
let h = tar_header(
b"demo-1.0/setup.py",
tar::EntryType::Regular,
b"",
body.len() as u64,
);
b.append(&h, &body[..]).unwrap();
let sdist = gz(&b.into_inner().unwrap());
let f = Inspector::default().inspect(
&sdist,
Format::declared("demo-1.0.tar.gz", "pip"),
"demo-1.0.tar.gz",
);
assert_eq!(f.format, Format::Sdist);
assert_eq!(f.verdict(), Verdict::Pass, "{:#?}", f.findings);
assert_eq!((f.entries, f.max_depth_seen), (1, 0));
}
// ── zip bombs ───────────────────────────────────────────────────────────────
#[test]
fn a_real_zip_bomb_is_refused_under_the_default_limits() {
// 32 MiB of zeros deflate to a few tens of KiB: about 1000:1.
let bomb = zeros_zip("demo/data.bin", 32 * MIB);
assert!(
(bomb.len() as u64) < MIB,
"small on disk: {} bytes",
bomb.len()
);
let f = inspect(&bomb, Format::Wheel);
assert!(blocks(&f, FindingKind::ZipBomb), "{:#?}", f.findings);
assert_eq!(f.verdict(), Verdict::Fail);
// Refused at the ratio cap (16 MiB floor), not after the whole 32 MiB.
assert!(
f.expanded_bytes < 17 * MIB,
"stopped early: {}",
f.expanded_bytes
);
}
#[test]
fn a_gzip_bomb_in_a_crate_is_refused_and_the_floor_is_configurable() {
let mut b = tar::Builder::new(Vec::new());
let body = vec![0u8; 4 * MIB as usize];
let h = tar_header(
b"demo-1.0/zeros",
tar::EntryType::Regular,
b"",
body.len() as u64,
);
b.append(&h, &body[..]).unwrap();
let krate = gz(&b.into_inner().unwrap());
// Under the default 16 MiB floor, 4 MiB is not judged by its ratio.
let f = Inspector::default().inspect(&krate, Format::Crate, "demo-1.0.crate");
assert!(
!f.has(FindingKind::ZipBomb, Severity::Info),
"{:#?}",
f.findings
);
let tight = Inspector::new(Limits {
ratio_floor_bytes: MIB,
..Limits::default()
});
let f = tight.inspect(&krate, Format::Crate, "demo-1.0.crate");
assert!(blocks(&f, FindingKind::ZipBomb), "{:#?}", f.findings);
}
#[test]
fn the_total_expansion_limit_holds_across_members() {
let half = vec![7u8; 512 * 1024];
let z = zip_of(&[("a.bin", &half), ("b.bin", &half), ("c.bin", &half)]);
let insp = Inspector::new(Limits {
max_total_expanded: MIB,
..Limits::default()
});
let f = insp.inspect(&z, Format::Generic, "data.zip");
assert!(blocks(&f, FindingKind::ExpansionLimit), "{:#?}", f.findings);
assert!(
Inspector::default()
.inspect(&z, Format::Generic, "data.zip")
.verdict()
== Verdict::Pass
);
}
#[test]
fn the_member_count_limit_holds() {
let z = zip_of(&[("a", b"1"), ("b", b"2"), ("c", b"3")]);
let f = Inspector::new(Limits {
max_entries: 2,
..Limits::default()
})
.inspect(&z, Format::Generic, "x.zip");
assert!(blocks(&f, FindingKind::TooManyEntries), "{:#?}", f.findings);
}
// ── nesting ─────────────────────────────────────────────────────────────────
fn nested(levels: usize) -> Vec<u8> {
let mut inner = zip_of(&[("payload.txt", b"hello")]);
for i in 1..levels {
let name = format!("level{i}.zip");
inner = zip_of(&[(name.as_str(), &inner)]);
}
inner
}
#[test]
fn nested_archives_past_the_depth_limit_are_refused() {
// The artifact plus 3 nested archives is the default limit and passes.
let ok = Inspector::default().inspect(&nested(4), Format::Generic, "outer.zip");
assert_eq!(ok.verdict(), Verdict::Pass, "{:#?}", ok.findings);
assert_eq!(ok.max_depth_seen, 3);
// One more level is refused.
let deep = Inspector::default().inspect(&nested(5), Format::Generic, "outer.zip");
assert!(
blocks(&deep, FindingKind::NestingTooDeep),
"{:#?}",
deep.findings
);
// And the limit is configurable.
let strict = Inspector::new(Limits {
max_depth: 1,
..Limits::default()
});
assert!(blocks(
&strict.inspect(&nested(3), Format::Generic, "outer.zip"),
FindingKind::NestingTooDeep
));
}
// ── traversal ───────────────────────────────────────────────────────────────
#[test]
fn a_traversal_entry_is_refused() {
let w = wheel(&[("../../etc/cron.d/evil", b"* * * * * root sh\n")]);
let f = inspect(&w, Format::Wheel);
assert!(blocks(&f, FindingKind::PathTraversal), "{:#?}", f.findings);
let abs = zip_of(&[("/etc/passwd", b"root::0:0::/:/bin/sh\n")]);
assert!(blocks(
&inspect(&abs, Format::Wheel),
FindingKind::AbsolutePath
));
}
#[test]
fn a_tar_traversal_entry_and_an_escaping_symlink_are_refused() {
let mut b = tar::Builder::new(Vec::new());
let body = b"x";
b.append(
&tar_header(b"pkg/../../x", tar::EntryType::Regular, b"", 1),
&body[..],
)
.unwrap();
b.append(
&tar_header(b"pkg/link", tar::EntryType::Symlink, b"../../etc/shadow", 0),
&b""[..],
)
.unwrap();
b.append(
&tar_header(b"pkg/ok", tar::EntryType::Symlink, b"../pkg/x", 0),
&b""[..],
)
.unwrap();
b.append(
&tar_header(b"pkg/hard", tar::EntryType::Link, b"/etc/passwd", 0),
&b""[..],
)
.unwrap();
let t = gz(&b.into_inner().unwrap());
let f = Inspector::default().inspect(&t, Format::Npm, "demo-1.0.tgz");
assert!(blocks(&f, FindingKind::PathTraversal), "{:#?}", f.findings);
let escapes: Vec<_> = f
.findings
.iter()
.filter(|x| x.kind == FindingKind::SymlinkEscape)
.map(|x| x.path.clone())
.collect();
assert_eq!(
escapes,
vec![
"demo-1.0.tgz!pkg/link".to_string(),
"demo-1.0.tgz!pkg/hard".to_string()
],
"{:#?}",
f.findings
);
}
#[test]
fn a_zip_symlink_that_escapes_is_refused() {
let mut w = zip::ZipWriter::new(Cursor::new(Vec::new()));
w.start_file("demo/__init__.py", deflated()).unwrap();
w.write_all(b"").unwrap();
w.add_symlink(
"demo/creds",
"../../../root/.ssh/id_ed25519",
SimpleFileOptions::default(),
)
.unwrap();
w.add_symlink("demo/alias", "__init__.py", SimpleFileOptions::default())
.unwrap();
let z = w.finish().unwrap().into_inner();
let f = inspect(&z, Format::Wheel);
let escapes: Vec<_> = f
.findings
.iter()
.filter(|x| x.kind == FindingKind::SymlinkEscape)
.collect();
assert_eq!(escapes.len(), 1, "{:#?}", f.findings);
assert!(escapes[0].path.ends_with("demo/creds"));
}
// ── magic bytes against the declared format ─────────────────────────────────
#[test]
fn a_png_renamed_whl_is_refused() {
let png = b"\x89PNG\r\n\x1a\n\0\0\0\rIHDR\0\0\0\x01\0\0\0\x01\x08\x06\0\0\0\x1f\x15\xc4\x89";
let f = Inspector::default().inspect(
png,
Format::declared("demo-1.0-py3-none-any.whl", "pip"),
"demo-1.0-py3-none-any.whl",
);
assert_eq!(f.format, Format::Wheel);
assert!(blocks(&f, FindingKind::MimeMismatch), "{:#?}", f.findings);
assert!(
f.findings[0].detail.contains("PNG image"),
"{}",
f.findings[0]
);
}
#[test]
fn a_member_whose_name_hides_a_binary_is_refused() {
let w = wheel(&[("demo/logo.png", &elf(2))]);
let f = inspect(&w, Format::Wheel);
assert!(blocks(&f, FindingKind::MimeMismatch), "{:#?}", f.findings);
}
// ── executables and scripts, per format ─────────────────────────────────────
#[test]
fn a_wheel_with_an_elf_executable_is_refused() {
let w = wheel(&[("demo/helper", &elf(2))]);
let f = inspect(&w, Format::Wheel);
assert!(blocks(&f, FindingKind::Executable), "{:#?}", f.findings);
}
#[test]
fn an_extension_module_passes_by_default_and_blocks_where_the_policy_forbids_it() {
let w = wheel(&[("demo/_speedups.cpython-312-x86_64-linux-gnu.so", &elf(3))]);
assert_eq!(inspect(&w, Format::Wheel).verdict(), Verdict::Pass);
let pure_only = ExecPolicy {
libraries: Rule::Block,
..ExecPolicy::default_for(Format::Wheel)
};
let f = Inspector::default()
.with_policy(Format::Wheel, pure_only)
.inspect(&w, Format::Wheel, "demo.whl");
assert!(blocks(&f, FindingKind::Executable), "{:#?}", f.findings);
}
#[test]
fn scripts_follow_the_format_policy() {
let w = wheel(&[
(
"demo-1.0.data/scripts/demo",
b"#!/usr/bin/env python\nprint(1)\n",
),
("demo/tool.sh", b"#!/bin/sh\nrm -rf /\n"),
]);
let f = inspect(&w, Format::Wheel);
let scripts: Vec<_> = f
.findings
.iter()
.filter(|x| x.kind == FindingKind::Script)
.collect();
assert_eq!(
scripts.len(),
1,
"only the one outside .data/scripts: {:#?}",
f.findings
);
assert_eq!(scripts[0].severity, Severity::Warn);
// A helm chart is text: an ELF in it blocks.
let mut b = tar::Builder::new(Vec::new());
let bin = elf(2);
b.append(
&tar_header(
b"chart/files/run",
tar::EntryType::Regular,
b"",
bin.len() as u64,
),
&bin[..],
)
.unwrap();
let chart = gz(&b.into_inner().unwrap());
assert!(blocks(
&Inspector::default().inspect(&chart, Format::Helm, "chart-1.0.tgz"),
FindingKind::Executable
));
}
// ── the malware hook ────────────────────────────────────────────────────────
#[test]
fn the_yara_adapter_detects_eicar_with_a_test_rule() {
let yara = Yara::from_source(EICAR_RULE, true);
assert_eq!(yara.engine_name(), Ok("built-in"));
assert_eq!(
yara.scan(&eicar(), "eicar.com"),
ScanOutcome::Detected(vec!["EICAR_Test_File".into()])
);
assert_eq!(yara.scan(b"just text", "x"), ScanOutcome::Clean);
// Inside a wheel, the member is found and named.
let insp = Inspector::default().with_scanner(Arc::new(yara));
let w = wheel(&[("demo/eicar.txt", &eicar())]);
let f = insp.inspect(&w, Format::Wheel, "demo.whl");
let hit = f
.findings
.iter()
.find(|x| x.kind == FindingKind::Malware)
.expect("detected");
assert_eq!(hit.path, "demo.whl!demo/eicar.txt");
assert_eq!(f.verdict(), Verdict::Fail);
assert_eq!(f.scanners, vec!["yara".to_string()]);
assert_eq!(
insp.inspect(&wheel(&[]), Format::Wheel, "demo.whl")
.verdict(),
Verdict::Pass
);
}
#[test]
fn yara_rules_on_disk_load_from_a_directory() {
let dir = tempfile::tempdir().unwrap();
std::fs::write(dir.path().join("eicar.yar"), EICAR_RULE).unwrap();
let y = Yara::from_config(&YaraConfig {
rules: Some(dir.path().display().to_string()),
..Default::default()
});
assert!(matches!(y.scan(&eicar(), "e"), ScanOutcome::Detected(_)));
}
#[test]
fn a_required_scanner_that_is_not_configured_blocks() {
let optional =
Inspector::default().with_scanner(Arc::new(Yara::from_config(&YaraConfig::default())));
let f = optional.inspect(&wheel(&[]), Format::Wheel, "demo.whl");
assert_eq!(f.verdict(), Verdict::Pass, "optional: info only");
assert!(f.has(FindingKind::ScannerNotConfigured, Severity::Info));
let required = YaraConfig {
required: true,
..Default::default()
};
let f = Inspector::default()
.with_scanner(Arc::new(Yara::from_config(&required)))
.inspect(&wheel(&[]), Format::Wheel, "d.whl");
assert!(
blocks(&f, FindingKind::ScannerNotConfigured),
"{:#?}",
f.findings
);
// A rule the built-in matcher cannot run, and no yara binary: not configured.
let y = Yara::from_source("rule R { strings: $a = /x+/ condition: $a }", true);
assert!(y.engine_name().is_err());
let clam = ClamAv::from_config(&ClamAvConfig {
socket: Some("/nonexistent/holger-test/clamd.ctl".into()),
required: true,
..Default::default()
});
assert!(matches!(
clam.scan(b"x", "x"),
ScanOutcome::NotConfigured(_)
));
let f = Inspector::default().with_scanner(Arc::new(clam)).inspect(
&wheel(&[]),
Format::Wheel,
"d.whl",
);
assert!(blocks(&f, FindingKind::ScannerNotConfigured));
}
/// A stand-in clamd that speaks INSTREAM, so the adapter's protocol is tested
/// on every host. It answers FOUND when the stream holds the EICAR marker.
#[cfg(unix)]
#[test]
fn the_clamav_adapter_speaks_instream_to_clamd() {
use std::io::Read;
use std::os::unix::net::UnixListener;
let dir = tempfile::tempdir().unwrap();
let sock = dir.path().join("clamd.ctl");
let listener = UnixListener::bind(&sock).unwrap();
let server = std::thread::spawn(move || {
for _ in 0..2 {
let (mut s, _) = listener.accept().unwrap();
let mut cmd = [0u8; 10];
s.read_exact(&mut cmd).unwrap();
assert_eq!(&cmd, b"zINSTREAM\0");
let mut data = Vec::new();
loop {
let mut len = [0u8; 4];
s.read_exact(&mut len).unwrap();
let n = u32::from_be_bytes(len) as usize;
if n == 0 {
break;
}
let mut chunk = vec![0u8; n];
s.read_exact(&mut chunk).unwrap();
data.extend_from_slice(&chunk);
}
let found = data.windows(9).any(|w| w == b"EICAR-STA");
let reply: &[u8] = if found {
b"stream: Eicar-Test-Signature FOUND\0"
} else {
b"stream: OK\0"
};
s.write_all(reply).unwrap();
}
});
let clam = ClamAv::from_config(&ClamAvConfig {
socket: Some(sock.display().to_string()),
..Default::default()
});
assert_eq!(
clam.scan(&eicar(), "eicar.com"),
ScanOutcome::Detected(vec!["Eicar-Test-Signature".into()])
);
assert_eq!(clam.scan(b"clean", "clean"), ScanOutcome::Clean);
server.join().unwrap();
}
/// The real engine, when this host has one.
#[test]
fn the_clamav_adapter_detects_eicar_with_a_real_clamd() {
let clam = ClamAv::from_config(&ClamAvConfig::default());
match clam.engine() {
Err(why) => {
eprintln!("SKIP the_clamav_adapter_detects_eicar_with_a_real_clamd: ClamAV is absent on this host ({why})");
}
Ok(engine) => {
let got = clam.scan(&eicar(), "eicar.com");
assert!(
matches!(got, ScanOutcome::Detected(_)),
"{engine:?} answered {got:?}"
);
assert_eq!(clam.scan(b"a clean file", "clean.txt"), ScanOutcome::Clean);
}
}
}
// ── every container is opened or blocked (lane HOL-NCSC, NCSC P2) ──────────
fn tar_of(members: &[(&[u8], &[u8])]) -> Vec<u8> {
let mut b = tar::Builder::new(Vec::new());
for (name, body) in members {
let h = tar_header(name, tar::EntryType::Regular, b"", body.len() as u64);
b.append(&h, *body).unwrap();
}
b.into_inner().unwrap()
}
fn evil_tar() -> Vec<u8> {
tar_of(&[
(b"usr/share/doc/demo/README", b"demo\n"),
(b"../../etc/cron.d/evil", b"* * * * * root sh\n"),
])
}
fn clean_tar() -> Vec<u8> {
tar_of(&[(b"usr/share/doc/demo/README", b"demo\n")])
}
/// A real xz stream (liblzma, the reference encoder; a test-only dependency): lzma-rs's own
/// encoder writes uncompressed LZMA2 chunks, which would make no bomb.
fn xz(bytes: &[u8]) -> Vec<u8> {
liblzma::encode_all(Cursor::new(bytes), 6).unwrap()
}
fn zst(bytes: &[u8]) -> Vec<u8> {
zstd::encode_all(Cursor::new(bytes), 3).unwrap()
}
fn bz2(bytes: &[u8]) -> Vec<u8> {
let mut e = bzip2::write::BzEncoder::new(Vec::new(), bzip2::Compression::default());
e.write_all(bytes).unwrap();
e.finish().unwrap()
}
/// A `!<arch>` archive (a .deb is one).
fn ar_of(members: &[(&str, &[u8])]) -> Vec<u8> {
let mut out = b"!<arch>\n".to_vec();
for (name, body) in members {
out.extend_from_slice(
format!(
"{:<16}{:<12}{:<6}{:<6}{:<8}{:<10}`\n",
name,
0,
0,
0,
100644,
body.len()
)
.as_bytes(),
);
out.extend_from_slice(body);
if body.len() % 2 == 1 {
out.push(b'\n');
}
}
out
}
/// A cpio `newc` stream (an rpm's payload): `(name, mode, body)`.
fn cpio(members: &[(&str, u32, &[u8])]) -> Vec<u8> {
let mut out = Vec::new();
let mut add = |name: &str, mode: u32, data: &[u8]| {
let fields = [
0,
mode,
0,
0,
1,
0,
data.len() as u32,
0,
0,
0,
0,
name.len() as u32 + 1,
0,
];
out.extend_from_slice(b"070701");
for f in fields {
out.extend_from_slice(format!("{f:08x}").as_bytes());
}
out.extend_from_slice(name.as_bytes());
out.push(0);
while out.len() % 4 != 0 {
out.push(0);
}
out.extend_from_slice(data);
while out.len() % 4 != 0 {
out.push(0);
}
};
for (n, m, d) in members {
add(n, *m, d);
}
add("TRAILER!!!", 0, b"");
out
}
/// An rpm: the lead, an (empty) signature header, an (empty) header, then `payload`.
fn rpm_of(payload: &[u8]) -> Vec<u8> {
let mut b = vec![0u8; 96];
b[..4].copy_from_slice(&[0xed, 0xab, 0xee, 0xdb]);
b[4] = 3;
for _ in 0..2 {
b.extend_from_slice(&[0x8e, 0xad, 0xe8, 0x01, 0, 0, 0, 0]);
b.extend_from_slice(&0u32.to_be_bytes());
b.extend_from_slice(&0u32.to_be_bytes());
}
b.extend_from_slice(payload);
b
}
/// Bytes a compressor cannot shrink (xorshift).
fn noise(len: usize) -> Vec<u8> {
let mut x: u64 = 0x9e37_79b9_7f4a_7c15;
(0..len)
.map(|_| {
x ^= x << 13;
x ^= x >> 7;
x ^= x << 17;
x as u8
})
.collect()
}
#[test]
fn every_compressed_tar_is_unpacked_and_its_members_inspected() {
for (ext, wrap) in [
("xz", xz as fn(&[u8]) -> Vec<u8>),
("zst", zst),
("bz2", bz2),
] {
let name = format!("data.tar.{ext}");
let f = Inspector::default().inspect(&wrap(&evil_tar()), Format::Generic, &name);
assert!(
blocks(&f, FindingKind::PathTraversal),
"{name}: {:#?}",
f.findings
);
let f = Inspector::default().inspect(&wrap(&clean_tar()), Format::Generic, &name);
assert_eq!(f.verdict(), Verdict::Pass, "{name}: {:#?}", f.findings);
assert_eq!(
(f.entries, f.max_depth_seen),
(1, 0),
"{name}: a compressed tar is one level"
);
}
// A compressed single file is a member one level down, inspected like any member.
let f = Inspector::default().inspect(&xz(&elf(2)), Format::Wheel, "tool.xz");
assert!(blocks(&f, FindingKind::Executable), "{:#?}", f.findings);
}
#[test]
fn a_deb_is_opened_down_to_its_data_members() {
let deb = |data: Vec<u8>| {
ar_of(&[
("debian-binary", b"2.0\n"),
("control.tar.xz", &xz(&clean_tar())),
("data.tar.zst", &zst(&data)),
])
};
let f = Inspector::default().inspect(&deb(clean_tar()), Format::Deb, "demo_1.0_amd64.deb");
assert_eq!(f.verdict(), Verdict::Pass, "{:#?}", f.findings);
assert!(
f.entries >= 5,
"the ar members and both tars' members are counted: {}",
f.entries
);
let f = Inspector::default().inspect(&deb(evil_tar()), Format::Deb, "demo_1.0_amd64.deb");
assert!(blocks(&f, FindingKind::PathTraversal), "{:#?}", f.findings);
let f = Inspector::default().inspect(&ar_of(&[("../evil", b"x")]), Format::Generic, "lib.a");
assert!(blocks(&f, FindingKind::PathTraversal), "{:#?}", f.findings);
}
#[test]
fn an_rpm_payload_is_opened_and_its_cpio_members_inspected() {
let clean = cpio(&[("./usr/share/doc/demo/README", 0o100644, b"demo\n")]);
let evil = cpio(&[(
"./usr/share/../../../etc/cron.d/evil",
0o100644,
b"* * * * * root sh\n",
)]);
let link = cpio(&[("./usr/lib/demo/shadow", 0o120777, b"/etc/shadow")]);
for (how, wrap) in [
("gzip", gz as fn(&[u8]) -> Vec<u8>),
("xz", xz),
("zstd", zst),
("bzip2", bz2),
] {
let f = Inspector::default().inspect(
&rpm_of(&wrap(&clean)),
Format::Rpm,
"demo-1.0-1.x86_64.rpm",
);
assert_eq!(f.verdict(), Verdict::Pass, "{how}: {:#?}", f.findings);
assert_eq!(f.entries, 1, "{how}");
let f = Inspector::default().inspect(
&rpm_of(&wrap(&evil)),
Format::Rpm,
"demo-1.0-1.x86_64.rpm",
);
assert!(
blocks(&f, FindingKind::PathTraversal),
"{how}: {:#?}",
f.findings
);
// A package payload is a root file system: an absolute symlink resolves in the
// installed root, so it is recorded (as an image layer's is); outside a package
// format it is refused.
let f = Inspector::default().inspect(
&rpm_of(&wrap(&link)),
Format::Rpm,
"demo-1.0-1.x86_64.rpm",
);
assert!(
f.has(FindingKind::SymlinkEscape, Severity::Info) && f.verdict() == Verdict::Pass,
"{how}: {:#?}",
f.findings
);
let f = Inspector::default().inspect(
&rpm_of(&wrap(&link)),
Format::Generic,
"demo-1.0-1.x86_64.rpm",
);
assert!(
blocks(&f, FindingKind::SymlinkEscape),
"{how}: {:#?}",
f.findings
);
}
// A payload in no format the walk opens is blocked, never passed.
let f = Inspector::default().inspect(
&rpm_of(b"\x00\x01\x02 not a payload"),
Format::Rpm,
"demo-1.0-1.x86_64.rpm",
);
assert_eq!(f.verdict(), Verdict::Fail, "{:#?}", f.findings);
}
#[test]
fn a_conda_package_is_opened_both_ways() {
let conda = zip_of(&[
("metadata.json", b"{\"conda_pkg_format_version\": 2}"),
("pkg-demo-1.0-0.tar.zst", &zst(&evil_tar())),
("info-demo-1.0-0.tar.zst", &zst(&clean_tar())),
]);
let f = Inspector::default().inspect(&conda, Format::Conda, "demo-1.0-0.conda");
assert!(blocks(&f, FindingKind::PathTraversal), "{:#?}", f.findings);
let f = Inspector::default().inspect(&bz2(&evil_tar()), Format::Conda, "demo-1.0-0.tar.bz2");
assert!(blocks(&f, FindingKind::PathTraversal), "{:#?}", f.findings);
let f = Inspector::default().inspect(&bz2(&clean_tar()), Format::Conda, "demo-1.0-0.tar.bz2");
assert_eq!(f.verdict(), Verdict::Pass, "{:#?}", f.findings);
}
#[test]
fn a_container_that_cannot_be_unpacked_blocks() {
let sevenz = [&[b'7', b'z', 0xbc, 0xaf, 0x27, 0x1c][..], &[0u8; 64][..]].concat();
let f = Inspector::default().inspect(&sevenz, Format::Generic, "data.7z");
assert!(blocks(&f, FindingKind::NotUnpacked), "{:#?}", f.findings);
assert_eq!(f.verdict(), Verdict::Fail);
let w = wheel(&[("demo/data.7z", &sevenz)]);
assert!(blocks(
&Inspector::default().inspect(&w, Format::Wheel, "demo-1.0-py3-none-any.whl"),
FindingKind::NotUnpacked
));
}
#[test]
fn the_new_decoders_hold_the_bomb_limits() {
let zeros = tar_of(&[(b"demo/zeros", &vec![0u8; 8 * MIB as usize])]);
let tight = Inspector::new(Limits {
ratio_floor_bytes: MIB,
..Limits::default()
});
for (ext, wrap) in [
("xz", xz as fn(&[u8]) -> Vec<u8>),
("zst", zst),
("bz2", bz2),
] {
let f = tight.inspect(&wrap(&zeros), Format::Generic, &format!("bomb.tar.{ext}"));
assert!(blocks(&f, FindingKind::ZipBomb), "{ext}: {:#?}", f.findings);
let f = tight.inspect(
&rpm_of(&wrap(&cpio(&[(
"./zeros",
0o100644,
&vec![0u8; 8 * MIB as usize],
)]))),
Format::Rpm,
"bomb.rpm",
);
assert!(
blocks(&f, FindingKind::ZipBomb),
"rpm {ext}: {:#?}",
f.findings
);
}
// xz is decoded in memory, so it is held to max_nested_bytes too.
let big = tar_of(&[(b"demo/noise", &noise(2 * MIB as usize))]);
let small = Inspector::new(Limits {
max_nested_bytes: MIB,
..Limits::default()
});
let f = small.inspect(&xz(&big), Format::Generic, "big.tar.xz");
assert!(blocks(&f, FindingKind::NestedTooLarge), "{:#?}", f.findings);
}
// ── Windows script launchers in Python packages (lane HOL-310-CHECK) ───────
/// A Windows PE executable of `len` bytes (MZ, `e_lfanew` 0x40, `PE\0\0`, no DLL flag), the rest
/// noise so it does not compress like a bomb.
fn pe_exe(len: usize) -> Vec<u8> {
let mut b = noise(len.max(0x80));
b[..2].copy_from_slice(b"MZ");
b[0x3c..0x40].copy_from_slice(&0x40u32.to_le_bytes());
b[0x40..0x44].copy_from_slice(b"PE\0\0");
// COFF characteristics (at PE + 22): executable image, 32-bit machine; no 0x2000 (DLL).
b[0x40 + 22..0x40 + 24].copy_from_slice(&0x0102u16.to_le_bytes());
b
}
fn sdist_of(members: &[(&str, &[u8])]) -> Vec<u8> {
let mut b = tar::Builder::new(Vec::new());
for (name, body) in members {
let h = tar_header(
name.as_bytes(),
tar::EntryType::Regular,
b"",
body.len() as u64,
);
b.append(&h, *body).unwrap();
}
gz(&b.into_inner().unwrap())
}
fn executables(f: &holger_inspect::Findings) -> Vec<(Severity, String)> {
f.findings
.iter()
.filter(|x| x.kind == FindingKind::Executable)
.map(|x| (x.severity, format!("{}: {}", x.path, x.detail)))
.collect()
}
/// **setuptools and pip were BLOCKED as sdists and as wheels** (lane HOL-310-CHECK, the Holger
/// 3.1.0 field check): setuptools carries `setuptools/{cli,gui}{,-32,-64,-arm64}.exe` (8 files,
/// 11 to 14 KiB) and pip carries distlib's `pip/_vendor/distlib/{t,w}{32,64,64-arm}.exe` (6 files,
/// 90 to 183 KiB). They are the Windows console-script launchers: a small PE `.exe` that pip or
/// setuptools copies next to a script on a WINDOWS install; nothing runs them on a build host or
/// a Linux client. Blocking them blocked every sdist build (its build backend is setuptools) and
/// pip itself. The rule, for wheels and sdists: a PE executable named `*.exe` of at most
/// 256 KiB warns (the malware scanners still read it); a bigger one, an ELF or Mach-O
/// executable, or a PE under any other name still blocks.
#[test]
fn a_windows_launcher_stub_in_a_python_package_warns_and_does_not_block() {
let launcher = pe_exe(14 * 1024);
let sdist = sdist_of(&[
(
"demo-1.0/setup.py",
b"from setuptools import setup\nsetup()\n",
),
("demo-1.0/demo/cli-64.exe", &launcher),
]);
let f = Inspector::default().inspect(&sdist, Format::Sdist, "demo-1.0.tar.gz");
assert_ne!(f.verdict(), Verdict::Fail, "{:#?}", f.findings);
let ex = executables(&f);
assert!(
ex.len() == 1 && ex[0].0 == Severity::Warn && ex[0].1.contains("Windows script launcher"),
"one warning that names it a launcher: {ex:#?}"
);
let whl = wheel(&[("demo/_vendor/t64.exe", &pe_exe(180 * 1024))]);
let f = Inspector::default().inspect(&whl, Format::Wheel, "demo-1.0-py3-none-any.whl");
assert_ne!(f.verdict(), Verdict::Fail, "{:#?}", f.findings);
assert!(
executables(&f).iter().all(|(s, _)| *s == Severity::Warn),
"{:#?}",
f.findings
);
}
#[test]
fn a_real_executable_in_a_python_package_still_blocks() {
// Bigger than any launcher stub.
let big = sdist_of(&[("demo-1.0/demo/setup-helper.exe", &pe_exe(300 * 1024))]);
let f = Inspector::default().inspect(&big, Format::Sdist, "demo-1.0.tar.gz");
assert!(blocks(&f, FindingKind::Executable), "{:#?}", f.findings);
// A PE that is not named like a launcher.
let hidden = sdist_of(&[("demo-1.0/demo/helper", &pe_exe(14 * 1024))]);
let f = Inspector::default().inspect(&hidden, Format::Sdist, "demo-1.0.tar.gz");
assert!(blocks(&f, FindingKind::Executable), "{:#?}", f.findings);
// A native Linux executable, whatever its name.
let elf_exe = sdist_of(&[("demo-1.0/demo/cli-64.exe", &elf(2))]);
let f = Inspector::default().inspect(&elf_exe, Format::Sdist, "demo-1.0.tar.gz");
assert!(blocks(&f, FindingKind::Executable), "{:#?}", f.findings);
let whl = wheel(&[("demo/run.exe", &pe_exe(300 * 1024))]);
assert!(blocks(
&inspect(&whl, Format::Wheel),
FindingKind::Executable
));
// Other formats keep their rule: a launcher-sized PE in a crate or a helm chart blocks.
let chart = sdist_of(&[("chart/files/cli-64.exe", &pe_exe(14 * 1024))]);
let f = Inspector::default().inspect(&chart, Format::Helm, "chart-1.0.tgz");
assert!(blocks(&f, FindingKind::Executable), "{:#?}", f.findings);
}
/// **Only a known launcher's name is excused** (SEC-REVIEW 2026-11-01 W1). The excuse took any
/// PE named `*.exe` of at most 256 KiB at any path, so an attacker's own wheel carried a
/// 256 KiB PE payload as a WARN. The names setuptools and distlib ship are excused; any other
/// `*.exe` follows `executables` (block).
#[test]
fn only_a_known_launcher_name_is_excused() {
for name in [
"demo/payload.exe",
"demo/_vendor/evil-64.exe",
"demo/cli-64.exe.exe",
] {
let whl = wheel(&[(name, &pe_exe(200 * 1024))]);
let f = Inspector::default().inspect(&whl, Format::Wheel, "demo-1.0-py3-none-any.whl");
assert!(
blocks(&f, FindingKind::Executable),
"{name}: {:#?}",
f.findings
);
}
for name in [
"setuptools/cli-arm64.exe",
"setuptools/gui-32.exe",
"pip/_vendor/distlib/w64-arm.exe",
"pip/_vendor/distlib/t32.exe",
] {
let whl = wheel(&[(name, &pe_exe(20 * 1024))]);
let f = Inspector::default().inspect(&whl, Format::Wheel, "demo-1.0-py3-none-any.whl");
assert_ne!(f.verdict(), Verdict::Fail, "{name}: {:#?}", f.findings);
}
}
#[test]
fn an_operator_policy_that_names_no_launcher_rule_blocks_launchers_too() {
// An explicit `policies: { sdist: (...) }` is the operator's whole rule: no launcher allowance.
let strict: ExecPolicy =
ron::from_str("(executables: block, libraries: block, scripts: allow)").unwrap();
let sdist = sdist_of(&[("demo-1.0/demo/cli-64.exe", &pe_exe(14 * 1024))]);
let f = Inspector::default()
.with_policy(Format::Sdist, strict)
.inspect(&sdist, Format::Sdist, "demo-1.0.tar.gz");
assert!(blocks(&f, FindingKind::Executable), "{:#?}", f.findings);
// ... and one that names it gets it.
let warn: ExecPolicy = ron::from_str(
"(executables: block, libraries: block, scripts: allow, windows_launchers: Some(warn))",
)
.unwrap();
let f = Inspector::default()
.with_policy(Format::Sdist, warn)
.inspect(&sdist, Format::Sdist, "demo-1.0.tar.gz");
assert_ne!(f.verdict(), Verdict::Fail, "{:#?}", f.findings);
}