Your browser sucks, JavaScript disabled! Gunnar's console needs JavaScript and WebAssembly. Git over SSH on port 2222 does not.

nordisk/heimdall

guests/static-scan/tests/contract.rs

// SPDX-License-Identifier: AGPL-3.0-only
//! The guest's side of the contract: refusal codes, parameters, the report's
//! subject, data digests, check selection, canonical bytes, determinism and
//! the in-toto wrapping. Heimdall is a defensive sandbox.

mod common;
use common::*;

use heimdall_guest_static_scan::{Outcome, MODGUNN_VERSION, SCANNER};
use heimdall_scan::{codes, in_toto_statement, Check, DataKind, Report, Verdict, PREDICATE_TYPE};

fn full_input() -> Vec<u8> {
    envelope(
        &npm_tgz(NPM_BENIGN),
        &[
            (DataKind::PopularNames, b"lodash\n"),
            (DataKind::InternalNamespaces, b"@acme/\n"),
            (
                DataKind::Registry,
                br#"{"name":"leftpad-ok","versions":{"1.0.0":{}}}"#,
            ),
            (DataKind::Malicious, b"[]"),
        ],
    )
}

const PURL: &str = "pkg:npm/leftpad-ok@1.0.0";

#[test]
fn a_benign_package_with_every_snapshot_is_clean_on_all_five_checks() {
    let r = report(&full_input(), &[("purl", PURL), ("origin", "public")]);
    assert_eq!(r.verdict, Verdict::Clean, "{r:#?}");
    let checks: Vec<Check> = r.checks.iter().map(|c| c.check).collect();
    assert_eq!(checks, Check::ALL.to_vec());
}

#[test]
fn with_no_snapshots_every_data_check_is_unknown_and_the_package_is_unknown() {
    let r = report(
        &envelope(&npm_tgz(NPM_BENIGN), &[]),
        &[("purl", PURL), ("origin", "public")],
    );
    assert_eq!(r.verdict, Verdict::Unknown);
    for c in &r.checks {
        if c.check == Check::Behavior {
            assert_eq!(c.verdict, Verdict::Clean);
        } else {
            assert_unknown(c, "data-missing");
        }
    }
}

#[test]
fn one_finding_anywhere_makes_the_package_findings() {
    let input = envelope(&npm_tgz(NPM_DROPPER), &[]);
    let r = report(&input, &[("purl", "pkg:npm/evil-pkg@1.0.0")]);
    assert_eq!(r.verdict, Verdict::Findings);
}

#[test]
fn the_subject_names_the_purl_and_the_artifact_digest() {
    let art = npm_tgz(NPM_BENIGN);
    let r = report(
        &envelope(&art, &[]),
        &[("purl", PURL), ("checks", "behavior")],
    );
    assert_eq!(r.subject.purl, PURL);
    assert_eq!(r.subject.ecosystem, "npm");
    assert_eq!(r.subject.name, "leftpad-ok");
    assert_eq!(r.subject.version.as_deref(), Some("1.0.0"));
    assert_eq!(r.subject.bytes, art.len() as u64);
    use sha2::Digest;
    let want: String = sha2::Sha256::digest(&art)
        .iter()
        .map(|b| format!("{b:02x}"))
        .collect();
    assert_eq!(r.subject.sha256, want);
    r.check_subject(PURL, &want).unwrap();
    assert!(r.check_subject("pkg:npm/other@1.0.0", &want).is_err());
    assert!(r.check_subject(PURL, &"0".repeat(64)).is_err());
}

#[test]
fn every_snapshot_sent_is_listed_by_digest_in_name_order() {
    let r = report(&full_input(), &[("purl", PURL), ("origin", "public")]);
    let names: Vec<&str> = r.data.iter().map(|d| d.name.as_str()).collect();
    assert_eq!(
        names,
        [
            "internal-namespaces",
            "malicious",
            "popular-names",
            "registry"
        ]
    );
    assert_eq!(r.data[1].bytes, 2);
}

#[test]
fn the_scanner_names_itself_and_the_modgunn_release_that_judged() {
    let r = report(&full_input(), &[("purl", PURL)]);
    assert_eq!(r.scanner.name, SCANNER);
    assert_eq!(r.scanner.version, env!("CARGO_PKG_VERSION"));
    assert_eq!(r.scanner.modgunn, MODGUNN_VERSION);
}

#[test]
fn the_modgunn_version_in_reports_is_the_one_in_the_lock() {
    let lock = std::fs::read_to_string(concat!(env!("CARGO_MANIFEST_DIR"), "/Cargo.lock")).unwrap();
    let at = lock
        .find("name = \"modgunn\"\nversion = \"")
        .expect("modgunn in the lock");
    let v = &lock[at + "name = \"modgunn\"\nversion = \"".len()..];
    assert_eq!(&v[..v.find('"').unwrap()], MODGUNN_VERSION);
}

#[test]
fn the_checks_parameter_selects_checks_and_reports_them_in_contract_order() {
    let r = report(
        &full_input(),
        &[("purl", PURL), ("checks", "malicious,behavior")],
    );
    let checks: Vec<Check> = r.checks.iter().map(|c| c.check).collect();
    assert_eq!(checks, [Check::Behavior, Check::Malicious]);
}

#[test]
fn checks_that_do_not_read_the_archive_never_judge_it() {
    // Garbage artifact, but only the name check runs: it is judged on its own.
    let input = envelope(b"not an archive", &[(DataKind::PopularNames, b"lodash\n")]);
    let r = report(
        &input,
        &[("purl", "pkg:npm/lodashs@1.0.0"), ("checks", "typosquat")],
    );
    assert_eq!(r.verdict, Verdict::Findings);
}

#[test]
fn two_runs_of_the_same_job_give_the_same_bytes() {
    let a = run(&full_input(), &[("purl", PURL)]);
    let b = run(&full_input(), &[("purl", PURL)]);
    match (a, b) {
        (Outcome::Report { json: a, .. }, Outcome::Report { json: b, .. }) => assert_eq!(a, b),
        _ => panic!("refused"),
    }
}

#[test]
fn a_bad_envelope_is_refused_with_heimdall_scan_1() {
    for bad in [&b""[..], b"heimdall-scan-input v2\n", &npm_tgz(NPM_BENIGN)] {
        assert!(refusal(bad, &[("purl", PURL)]).starts_with(&format!("[{}]", codes::BAD_INPUT)));
    }
}

#[test]
fn unknown_repeated_or_malformed_parameters_are_refused_with_heimdall_scan_2() {
    let input = full_input();
    for p in [
        vec![("purl", PURL), ("colour", "blue")],
        vec![("purl", PURL), ("origin", "somewhere")],
        vec![("purl", PURL), ("max-entries", "0")],
        vec![("purl", PURL), ("max-entries", "-1")],
        vec![("purl", PURL), ("max-unpacked-bytes", "99999999999999")],
        vec![("purl", PURL), ("checks", "behavior,behavior")],
        vec![("purl", PURL), ("purl", PURL)],
    ] {
        let why = refusal(&input, &p);
        assert!(
            why.starts_with(&format!("[{}]", codes::BAD_PARAMS)),
            "{p:?}: {why}"
        );
    }
}

#[test]
fn a_missing_or_bad_purl_is_refused_with_heimdall_scan_3() {
    let input = full_input();
    for p in [
        vec![],
        vec![("purl", "npm/x@1")],
        vec![("purl", "pkg:npm")],
        vec![("purl", "pkg:npm/%zz@1")],
    ] {
        let why = refusal(&input, &p);
        assert!(
            why.starts_with(&format!("[{}]", codes::BAD_PURL)),
            "{p:?}: {why}"
        );
    }
}

#[test]
fn an_unknown_check_is_refused_with_heimdall_scan_4() {
    let why = refusal(
        &full_input(),
        &[("purl", PURL), ("checks", "behavior,vibes")],
    );
    assert!(
        why.starts_with(&format!("[{}]", codes::UNKNOWN_CHECK)),
        "{why}"
    );
}

#[test]
fn the_report_wraps_as_an_in_toto_statement_about_the_purl_and_digest() {
    let r = report(&full_input(), &[("purl", PURL)]);
    let s = in_toto_statement(&r);
    assert_eq!(s["_type"], "https://in-toto.io/Statement/v1");
    assert_eq!(s["predicateType"], PREDICATE_TYPE);
    assert_eq!(s["subject"][0]["name"], PURL);
    assert_eq!(
        s["subject"][0]["digest"]["sha256"],
        r.subject.sha256.as_str()
    );
    let back: Report = serde_json::from_value(s["predicate"].clone()).unwrap();
    assert_eq!(back, r);
}

#[test]
fn a_tampered_document_is_refused_by_the_callers_reader() {
    let Outcome::Report { json, .. } = run(
        &envelope(&npm_tgz(NPM_DROPPER), &[]),
        &[("purl", "pkg:npm/evil-pkg@1.0.0")],
    ) else {
        panic!()
    };
    let text = String::from_utf8(json).unwrap();
    // Flip the overall verdict to clean: the fold no longer agrees.
    let forged = text.replace(r#""verdict":"findings"}"#, r#""verdict":"clean"}"#);
    assert_ne!(forged, text);
    assert!(Report::from_canonical_json(forged.as_bytes()).is_err());
    // Re-indented (not canonical) is refused too.
    let v: serde_json::Value = serde_json::from_str(&text).unwrap();
    assert!(
        Report::from_canonical_json(serde_json::to_string_pretty(&v).unwrap().as_bytes()).is_err()
    );
}