crates/heimdall-scan/tests/contract.rs
// SPDX-License-Identifier: AGPL-3.0-only
//! The static-scanner contract on its own: the envelope, purls, the findings
//! document's rules and the in-toto wrapping. Heimdall is a defensive sandbox.
use heimdall_scan::input::{InputError, MAGIC};
use heimdall_scan::report::unknown;
use heimdall_scan::*;
use proptest::prelude::*;
fn sample() -> ScanInput {
ScanInput::new(b"artifact bytes".to_vec())
.with(DataKind::Registry, b"{}".to_vec())
.with(DataKind::PopularNames, b"lodash\n".to_vec())
}
#[test]
fn an_envelope_round_trips_and_borrows_the_input() {
let s = sample();
let b = s.encode();
let r = ScanInputRef::decode(&b).unwrap();
assert_eq!(r.artifact, b"artifact bytes");
assert_eq!(r.data[&DataKind::Registry], b"{}");
assert_eq!(r.data[&DataKind::PopularNames], b"lodash\n");
assert_eq!(r.data.len(), 2);
}
#[test]
fn equal_inputs_encode_to_equal_bytes_whatever_the_insert_order() {
let a = ScanInput::new(vec![1])
.with(DataKind::Malicious, vec![2])
.with(DataKind::PopularNames, vec![3]);
let b = ScanInput::new(vec![1])
.with(DataKind::PopularNames, vec![3])
.with(DataKind::Malicious, vec![2]);
assert_eq!(a.encode(), b.encode());
}
#[test]
fn an_empty_artifact_is_a_valid_envelope() {
let b = ScanInput::new(Vec::new()).encode();
assert_eq!(ScanInputRef::decode(&b).unwrap().artifact, b"");
}
fn raw(sections: &[(&[u8], &[u8])], count: u32) -> Vec<u8> {
let mut out = MAGIC.to_vec();
out.extend_from_slice(&count.to_be_bytes());
for (n, b) in sections {
out.push(n.len() as u8);
out.extend_from_slice(n);
out.extend_from_slice(&(b.len() as u64).to_be_bytes());
out.extend_from_slice(b);
}
out
}
#[test]
fn the_decoder_refuses_every_malformed_envelope() {
assert_eq!(ScanInputRef::decode(b"").unwrap_err(), InputError::BadMagic);
assert_eq!(
ScanInputRef::decode(b"heimdall-scan-input v2\n\0\0\0\x01").unwrap_err(),
InputError::BadMagic
);
assert_eq!(
ScanInputRef::decode(&raw(&[], 0)).unwrap_err(),
InputError::SectionCount(0)
);
assert_eq!(
ScanInputRef::decode(&raw(&[], 17)).unwrap_err(),
InputError::SectionCount(17)
);
assert_eq!(
ScanInputRef::decode(&raw(&[(b"registry", b"x")], 1)).unwrap_err(),
InputError::NoArtifact
);
assert_eq!(
ScanInputRef::decode(&raw(&[(b"artifact", b"x"), (b"artifact", b"y")], 2)).unwrap_err(),
InputError::DuplicateSection("artifact".into())
);
assert_eq!(
ScanInputRef::decode(&raw(
&[
(b"artifact", b"x"),
(b"registry", b"1"),
(b"registry", b"2")
],
3
))
.unwrap_err(),
InputError::DuplicateSection("registry".into())
);
assert_eq!(
ScanInputRef::decode(&raw(&[(b"artifact", b"x"), (b"secrets", b"y")], 2)).unwrap_err(),
InputError::UnknownSection("secrets".into())
);
assert!(matches!(
ScanInputRef::decode(&raw(&[(b"artifact", b"x"), (b"\xff", b"y")], 2)).unwrap_err(),
InputError::UnknownSection(_)
));
let mut trailing = raw(&[(b"artifact", b"x")], 1);
trailing.push(0);
assert_eq!(
ScanInputRef::decode(&trailing).unwrap_err(),
InputError::TrailingBytes(1)
);
assert!(matches!(
ScanInputRef::decode(&raw(&[(b"artifact", b"x")], 2)).unwrap_err(),
InputError::Truncated(_)
));
}
#[test]
fn a_section_length_larger_than_the_input_is_truncated_not_an_allocation() {
let mut b = MAGIC.to_vec();
b.extend_from_slice(&1u32.to_be_bytes());
b.push(8);
b.extend_from_slice(b"artifact");
b.extend_from_slice(&u64::MAX.to_be_bytes());
assert_eq!(
ScanInputRef::decode(&b).unwrap_err(),
InputError::Truncated("section body")
);
}
#[test]
fn data_kind_names_round_trip_and_are_unique() {
let mut seen = std::collections::BTreeSet::new();
for k in DataKind::ALL {
assert_eq!(DataKind::from_name(k.name()), Some(k));
assert!(seen.insert(k.name()));
assert_ne!(k.name(), heimdall_scan::input::ARTIFACT);
}
}
#[test]
fn purls_of_every_scanned_ecosystem_parse_to_modgunn_names() {
let cases = [
("pkg:npm/lodash@4.17.15", "npm", "lodash", Some("4.17.15")),
(
"pkg:npm/%40acme/billing@1.0.0",
"npm",
"@acme/billing",
Some("1.0.0"),
),
(
"pkg:npm/@acme/billing@1.0.0",
"npm",
"@acme/billing",
Some("1.0.0"),
),
("pkg:npm/@acme/billing", "npm", "@acme/billing", None),
("pkg:pypi/PyYAML@5.3", "pypi", "pyyaml", Some("5.3")),
(
"pkg:pypi/Zope.Interface@6.0",
"pypi",
"zope-interface",
Some("6.0"),
),
(
"pkg:cargo/smallvec@1.6.0",
"cargo",
"smallvec",
Some("1.6.0"),
),
(
"pkg:golang/github.com/acme/widget@v1.2.0",
"go",
"github.com/acme/widget",
Some("v1.2.0"),
),
(
"pkg:maven/org.apache.logging.log4j/log4j-core@2.14.1",
"maven",
"org.apache.logging.log4j:log4j-core",
Some("2.14.1"),
),
(
"pkg:npm/lodash@4.17.15?repository_url=https://r.example#lib/x",
"npm",
"lodash",
Some("4.17.15"),
),
("pkg:NPM/lodash", "npm", "lodash", None),
];
for (s, eco, name, ver) in cases {
let p = Purl::parse(s).unwrap_or_else(|e| panic!("{s}: {e}"));
assert_eq!(p.ecosystem(), eco, "{s}");
assert_eq!(p.package_name(), name, "{s}");
assert_eq!(p.version.as_deref(), ver, "{s}");
}
let q = Purl::parse("pkg:npm/lodash@4.17.15?repository_url=https://r.example#lib/x").unwrap();
assert_eq!(q.qualifiers["repository_url"], "https://r.example");
assert_eq!(q.subpath.as_deref(), Some("lib/x"));
}
#[test]
fn bad_purls_are_refused() {
for s in [
"",
"npm/lodash",
"pkg:",
"pkg:npm",
"pkg:npm/",
"pkg:/lodash",
"pkg:n%70m/x",
"pkg:npm/%zz",
"pkg:npm/%ff%fe",
"pkg:npm/a%00b",
"pkg:npm/a b",
"pkg:npm/x?novalue",
"pkg:1npm/x",
] {
assert!(Purl::parse(s).is_err(), "{s:?} parsed");
}
assert_eq!(
Purl::parse(&format!("pkg:npm/{}", "a".repeat(5000))),
Err(PurlError::TooLong)
);
}
#[test]
fn a_purl_prints_back_to_an_equal_purl() {
for s in [
"pkg:npm/%40acme/billing@1.0.0",
"pkg:golang/github.com/acme/widget@v1.2.0",
"pkg:cargo/x@1.0.0?a=b#c/d",
] {
let p = Purl::parse(s).unwrap();
assert_eq!(Purl::parse(&p.to_string()).unwrap(), p, "{s}");
}
}
fn report(checks: Vec<CheckResult>) -> Report {
Report {
scanner: ScannerId {
name: "s".into(),
version: "0".into(),
modgunn: "0.0.8".into(),
},
subject: Subject {
purl: "pkg:npm/x@1".into(),
ecosystem: "npm".into(),
name: "x".into(),
version: Some("1".into()),
sha256: "ab".repeat(32),
bytes: 3,
},
data: vec![],
verdict: Report::fold(&checks),
checks,
}
}
fn finding() -> Finding {
Finding {
id: "X".into(),
severity: Severity::High,
summary: "x".into(),
path: None,
}
}
fn why() -> Option<UnknownReason> {
Some(UnknownReason {
code: unknown::DATA_MISSING.into(),
detail: "d".into(),
})
}
#[test]
fn a_check_verdict_follows_from_its_contents() {
assert_eq!(
CheckResult::new(Check::Behavior, vec![], None).verdict,
Verdict::Clean
);
assert_eq!(
CheckResult::new(Check::Behavior, vec![], why()).verdict,
Verdict::Unknown
);
assert_eq!(
CheckResult::new(Check::Behavior, vec![finding()], why()).verdict,
Verdict::Findings
);
assert_eq!(
CheckResult::new(Check::Behavior, vec![finding()], None).verdict,
Verdict::Findings
);
}
#[test]
fn the_overall_verdict_is_findings_over_unknown_over_clean() {
let c = |f: bool, u: bool, k| {
CheckResult::new(
k,
if f { vec![finding()] } else { vec![] },
if u { why() } else { None },
)
};
assert_eq!(
report(vec![c(false, false, Check::Behavior)]).verdict,
Verdict::Clean
);
assert_eq!(
report(vec![
c(false, false, Check::Behavior),
c(false, true, Check::Typosquat)
])
.verdict,
Verdict::Unknown
);
assert_eq!(
report(vec![
c(false, true, Check::Behavior),
c(true, false, Check::Typosquat)
])
.verdict,
Verdict::Findings
);
}
#[test]
fn validate_refuses_every_inconsistent_document() {
let good = report(vec![
CheckResult::new(Check::Behavior, vec![], None),
CheckResult::new(Check::Registry, vec![], why()),
]);
good.validate().unwrap();
let mut r = good.clone();
r.checks.clear();
r.verdict = Verdict::Clean;
assert_eq!(r.validate(), Err(ReportError::NoChecks));
let mut r = good.clone();
r.checks.swap(0, 1);
assert_eq!(r.validate(), Err(ReportError::ChecksOutOfOrder));
let mut r = good.clone();
r.checks.push(r.checks[1].clone());
assert_eq!(r.validate(), Err(ReportError::ChecksOutOfOrder));
let mut r = good.clone();
r.checks[1].verdict = Verdict::Clean; // unknown reason, but says clean
r.verdict = Verdict::Clean;
assert_eq!(
r.validate(),
Err(ReportError::CheckVerdict(Check::Registry))
);
let mut r = good.clone();
r.checks[1].unknown.as_mut().unwrap().code = "felt-fine".into();
assert!(matches!(
r.validate(),
Err(ReportError::UnknownCode(Check::Registry, _))
));
let mut r = good.clone();
r.verdict = Verdict::Clean;
assert_eq!(r.validate(), Err(ReportError::Overall));
let mut r = good.clone();
r.data = vec![
DataDigest {
name: "registry".into(),
sha256: "0".into(),
bytes: 0,
},
DataDigest {
name: "malicious".into(),
sha256: "0".into(),
bytes: 0,
},
];
assert_eq!(r.validate(), Err(ReportError::DataOrder));
}
#[test]
fn the_canonical_json_round_trips_and_nothing_else_reads() {
let r = report(vec![CheckResult::new(
Check::Behavior,
vec![finding()],
why(),
)]);
let b = r.to_canonical_json();
assert_eq!(Report::from_canonical_json(&b).unwrap(), r);
let text = String::from_utf8(b.clone()).unwrap();
assert!(!text.contains('\n') && !text.contains(": "));
// An extra field is refused (deny_unknown_fields).
let extra = text.replacen('{', r#"{"extra":1,"#, 1);
assert!(Report::from_canonical_json(extra.as_bytes()).is_err());
// Field order is part of the encoding.
let v: serde_json::Value = serde_json::from_slice(&b).unwrap();
assert!(
Report::from_canonical_json(v.to_string().as_bytes()).is_err()
|| v.to_string().as_bytes() == b.as_slice()
);
}
#[test]
fn the_json_names_are_the_documented_ones() {
let r = report(vec![CheckResult::new(
Check::DependencyConfusion,
vec![finding()],
why(),
)]);
let v: serde_json::Value = serde_json::from_slice(&r.to_canonical_json()).unwrap();
assert_eq!(v["checks"][0]["check"], "dependency-confusion");
assert_eq!(v["checks"][0]["verdict"], "findings");
assert_eq!(v["checks"][0]["findings"][0]["severity"], "high");
assert_eq!(v["checks"][0]["unknown"]["code"], "data-missing");
assert_eq!(v["verdict"], "findings");
assert!(v["checks"][0]["findings"][0].get("path").is_none());
}
#[test]
fn the_statement_is_in_toto_v1_about_the_purl_and_sha256() {
let r = report(vec![CheckResult::new(Check::Behavior, vec![], None)]);
let s = in_toto_statement(&r);
assert_eq!(s["_type"], STATEMENT_TYPE);
assert_eq!(s["predicateType"], PREDICATE_TYPE);
assert_eq!(s["subject"][0]["name"], "pkg:npm/x@1");
assert_eq!(s["subject"][0]["digest"]["sha256"], "ab".repeat(32));
assert_eq!(
serde_json::from_value::<Report>(s["predicate"].clone()).unwrap(),
r
);
}
#[test]
fn guest_refusal_codes_are_unique_and_prefixed() {
let mut seen = std::collections::BTreeSet::new();
for c in codes::ALL {
assert!(c.starts_with("heimdall-scan-"));
assert!(seen.insert(c));
}
let mut seen = std::collections::BTreeSet::new();
for c in unknown::ALL {
assert!(seen.insert(c));
}
}
#[test]
fn check_names_round_trip_and_name_their_snapshot() {
for c in Check::ALL {
assert_eq!(Check::from_name(c.name()), Some(c));
}
assert_eq!(Check::Behavior.needs(), None);
assert_eq!(Check::Typosquat.needs(), Some(DataKind::PopularNames));
assert_eq!(Check::Malicious.needs(), Some(DataKind::Malicious));
}
proptest! {
#[test]
fn any_bytes_never_panic_the_envelope_decoder(b in proptest::collection::vec(any::<u8>(), 0..512)) {
let _ = ScanInputRef::decode(&b);
let mut with_magic = MAGIC.to_vec();
with_magic.extend_from_slice(&b);
let _ = ScanInputRef::decode(&with_magic);
}
#[test]
fn any_envelope_round_trips(
art in proptest::collection::vec(any::<u8>(), 0..256),
data in proptest::collection::btree_map(0usize..4, proptest::collection::vec(any::<u8>(), 0..64), 0..4),
) {
let mut s = ScanInput::new(art.clone());
for (k, v) in &data {
s = s.with(DataKind::ALL[*k], v.clone());
}
let b = s.encode();
let r = ScanInputRef::decode(&b).unwrap();
prop_assert_eq!(r.artifact, art.as_slice());
prop_assert_eq!(r.data.len(), data.len());
// Every proper prefix is refused, never accepted as a shorter input.
for cut in [0, b.len() / 2, b.len() - 1] {
prop_assert!(ScanInputRef::decode(&b[..cut]).is_err());
}
}
#[test]
fn any_string_never_panics_the_purl_parser(s in "\\PC{0,64}") {
let _ = Purl::parse(&s);
let _ = Purl::parse(&format!("pkg:npm/{s}"));
}
#[test]
fn pep503_is_idempotent(s in "[A-Za-z0-9._-]{0,32}") {
let once = heimdall_scan::purl::pep503(&s);
prop_assert_eq!(heimdall_scan::purl::pep503(&once), once.clone());
}
}