bootintel-detectors 0.13.0

//! The Rust half of the cross-implementation boot-chain check.
//!
//! The same verdict logic ships twice: here, and in Python in the bootintel.com
//! engine. The reason it ships twice is that a consultancy under an NDA cannot
//! upload a capture and still needs the answer, so `bootintel` decides
//! offline. The risk is that two implementations of one ruleset drift, which
//! would be worse than a bug: the CLI and the dashboard would tell one
//! operator two different things about one board.
//!
//! Neither side is the reference. `tests/fixtures/boot_chain/expect.txt` is,
//! and the engine repo holds a byte-identical copy that its own test
//! (`api/tests/test_boot_chain_parity.py`) asserts against. The file is
//! generated by `scripts/gen-boot-chain-expect.sh` in bootintel.com, which
//! writes both copies in one go. Do not hand-edit it to make this pass: the
//! only thing that makes green meaningful here is that the other repo is
//! reading the same bytes.

use std::fs;
use std::path::PathBuf;

use bootintel_detectors::{boot_chain, os_hardening};

fn fixtures() -> PathBuf {
    PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/boot_chain")
}

/// FNV-1a, matching the generator. Not a security property: it detects a
/// fixture log edited without regenerating the expectation. Chosen because it
/// is eight lines in both languages and adds no dependency to a crate that is
/// deliberately regex-only.
fn fnv1a64(data: &[u8]) -> String {
    let mut h: u64 = 0xcbf2_9ce4_8422_2325;
    for byte in data {
        h = (h ^ u64::from(*byte)).wrapping_mul(0x100_0000_01b3);
    }
    format!("{h:016x}")
}

fn field(out: &mut Vec<String>, indent: &str, key: &str, value: &str) {
    if value.is_empty() {
        out.push(format!("{indent}{key}"));
    } else {
        out.push(format!("{indent}{key} {value}"));
    }
}

/// Byte-for-byte the format the Python renderer emits. A plain line format
/// rather than JSON so it diffs cleanly in review, and so this crate needs no
/// serde: it stays serde-free on purpose to keep the CLI binary small.
fn render(name: &str, log: &str) -> Vec<String> {
    let a = boot_chain::assess(log);
    let (session, integrity, verdicts) = (&a.session, &a.integrity, &a.verdicts);
    let mut out = vec![format!(
        "## fixture {name} fnv1a64={}",
        fnv1a64(log.as_bytes())
    )];
    field(
        &mut out,
        "  ",
        "reached",
        if session.reached { "true" } else { "false" },
    );
    field(&mut out, "  ", "evidence", &session.evidence);
    let bytes = match (session.env_used_bytes, session.env_total_bytes) {
        (Some(used), Some(total)) if total > 0 => format!("{used}/{total}"),
        _ => String::new(),
    };
    field(&mut out, "  ", "env_bytes", &bytes);
    // Same field order as `analysis_engine/parity_render.py`, and absent fields
    // are omitted rather than rendered empty: "the capture said nothing about
    // this" is a different claim from "this is off".
    let mut integ = |k: &str, val: Option<&str>| {
        if let Some(x) = val {
            field(&mut out, "  ", "integrity", &format!("{k}={x}"));
        }
    };
    integ("image_check", integrity.image_check.as_deref());
    integ(
        "image_check_result",
        integrity.image_check_result.as_deref(),
    );
    integ(
        "image_check_evidence",
        integrity.image_check_evidence.as_deref(),
    );
    let algos = integrity.image_hash_algorithms.join(", ");
    integ(
        "image_hash_algorithms",
        if algos.is_empty() {
            None
        } else {
            Some(algos.as_str())
        },
    );
    integ(
        "image_signature_checked",
        if integrity.image_signature_checked {
            Some("true")
        } else {
            None
        },
    );
    integ(
        "image_signature_evidence",
        integrity.image_signature_evidence.as_deref(),
    );
    integ(
        "image_check_failed",
        integrity.image_check_failed.as_deref(),
    );
    integ("hab_fuse", integrity.hab_fuse.as_deref());
    integ("hab_evidence", integrity.hab_evidence.as_deref());
    integ(
        "ubifs_unauthenticated",
        integrity.ubifs_unauthenticated.as_deref(),
    );
    integ("env_crc_failed", integrity.env_crc_failed.as_deref());
    // Hardening, in the same field order as `analysis_engine/parity_render.py`.
    // `mac_modules` renders even when empty, because "an LSM line was seen and
    // none of them provide mandatory access control" is a real finding and a
    // different claim from "no LSM line was seen at all". It is therefore keyed
    // off `lsm` being present, which is how the engine's dict distinguishes them.
    let h = os_hardening::parse(log);
    if let Some(m) = &h.mem_auto_init {
        field(
            &mut out,
            "  ",
            "hardening",
            &format!(
                "mem_auto_init=stack:{} heap_alloc:{} heap_free:{}",
                m.stack, m.heap_alloc, m.heap_free
            ),
        );
    }
    let mut hard = |k: &str, val: Option<&str>| {
        if let Some(x) = val {
            field(&mut out, "  ", "hardening", &format!("{k}={x}"));
        }
    };
    hard("kaslr", h.kaslr.as_deref());
    hard("kaslr_reason", h.kaslr_reason.as_deref());
    if !h.lsm.is_empty() {
        hard("lsm", Some(h.lsm.join(", ").as_str()));
        hard("mac_modules", Some(h.mac_modules.join(", ").as_str()));
    }
    hard("selinux", h.selinux.as_deref());
    hard("apparmor", h.apparmor.as_deref());
    hard(
        "ignored_kernel_parameters",
        h.ignored_kernel_parameters.as_deref(),
    );
    for (key, value) in &session.bdinfo {
        field(&mut out, "  ", "bdinfo", &format!("{key}={value}"));
    }
    if let Some(dev) = &session.mtd_device {
        field(&mut out, "  ", "mtd_device", dev);
    }
    for part in &session.mtd_partitions {
        field(
            &mut out,
            "  ",
            "mtd_part",
            &format!(
                "{} size=0x{:x} offset=0x{:x} ro={}",
                part.name, part.size, part.offset, part.read_only
            ),
        );
    }
    for (key, value) in &session.env {
        field(&mut out, "  ", "env", &format!("{key}={value}"));
    }
    for verdict in verdicts {
        out.push("  verdict".to_string());
        field(&mut out, "    ", "title", &verdict.title);
        field(&mut out, "    ", "state", &verdict.state);
        field(&mut out, "    ", "severity", &verdict.severity);
        field(&mut out, "    ", "evidence", &verdict.evidence);
        field(&mut out, "    ", "detail", &verdict.detail);
        field(
            &mut out,
            "    ",
            "remediation",
            verdict.remediation.as_deref().unwrap_or(""),
        );
    }
    out
}

fn expectation() -> String {
    fs::read_to_string(fixtures().join("expect.txt")).expect("expect.txt is committed")
}

fn declared_fixtures(expected: &str) -> Vec<String> {
    expected
        .lines()
        .filter_map(|l| l.strip_prefix("## fixture "))
        .map(|rest| {
            rest.split_whitespace()
                .next()
                .unwrap_or_default()
                .to_string()
        })
        .collect()
}

#[test]
fn the_cli_reproduces_the_shared_expectation() {
    let expected = expectation();
    let mut lines: Vec<String> = expected
        .lines()
        .filter(|l| l.starts_with('#') && !l.starts_with("##"))
        .map(str::to_string)
        .collect();
    for name in declared_fixtures(&expected) {
        let log = fs::read_to_string(fixtures().join(&name)).expect("fixture is committed");
        lines.extend(render(&name, &log));
    }
    let actual = lines.join("\n") + "\n";
    if actual != expected {
        // Show the first divergence rather than two 200-line blobs.
        let (want, got): (Vec<&str>, Vec<&str>) =
            (expected.lines().collect(), actual.lines().collect());
        let at = want
            .iter()
            .zip(got.iter())
            .position(|(a, b)| a != b)
            .unwrap_or(want.len().min(got.len()));
        panic!(
            "Rust output no longer matches the shared expectation at line {}:\n  \
             expect.txt: {:?}\n  this crate: {:?}\n\nIf the rule changed, regenerate BOTH \
             copies with scripts/gen-boot-chain-expect.sh in bootintel.com. If it did not, \
             the two implementations have drifted.",
            at + 1,
            want.get(at).unwrap_or(&"<end of file>"),
            got.get(at).unwrap_or(&"<end of file>"),
        );
    }
}

#[test]
fn every_fixture_is_pinned_to_its_recorded_hash() {
    let expected = expectation();
    let pins: Vec<&str> = expected
        .lines()
        .filter_map(|l| l.strip_prefix("## fixture "))
        .collect();
    assert!(!pins.is_empty(), "expectation file declares no fixtures");
    for line in pins {
        let mut parts = line.split_whitespace();
        let name = parts.next().expect("fixture name");
        let pin = parts.next().expect("fixture pin");
        let bytes = fs::read(fixtures().join(name)).expect("fixture is committed");
        assert_eq!(
            pin,
            format!("fnv1a64={}", fnv1a64(&bytes)),
            "{name} changed since expect.txt was generated"
        );
    }
}

/// The guard that matters most, and the mirror of the CVE over-matching bug:
/// `KEY=value` lines are everywhere in a boot log, so a careless parser would
/// invent an environment out of a kernel command line and report confident
/// verdicts about a device nobody ever talked to.
#[test]
fn a_plain_boot_log_yields_no_session_and_no_verdict() {
    let plain = "U-Boot 2020.10 (Sep 17 2023 - 11:38:21 +0000)\n\
                 Hit any key to stop autoboot:  2\n\
                 Booting from flash...\n\
                 [    0.000000] Kernel command line: console=ttyS0 root=/dev/mtdblock2\n\
                 CONFIG_FOO=bar\n\
                 PATH=/usr/bin:/bin\n\
                 bootcmd=this is not really an environment\n\
                 [    1.000000] procd: - init -\n";
    let a = boot_chain::assess(plain);
    assert!(
        !a.session.reached,
        "claimed a shell on a log with no session"
    );
    assert!(
        a.session.env.is_empty(),
        "invented an environment: {:?}",
        a.session.env
    );
    assert!(a.verdicts.is_empty(), "produced verdicts with no session");
}

#[test]
fn an_unterminated_run_of_key_values_is_discarded() {
    // Same shape as a real dump but with no `Environment size:` line, so
    // nothing proves it was one. Absence of proof is not a verdict.
    let log = "bootdelay=3\nbootcmd=bootm 0x82000000\nverify=no\n";
    let a = boot_chain::assess(log);
    assert!(a.session.env.is_empty());
    assert!(a.verdicts.is_empty());
}

/// Absence is reported as unknown, never as good news: U-Boot prints only what
/// is set, so a missing `bootdelay` means the compiled-in default applies and
/// cannot be read from a capture.
#[test]
fn a_missing_variable_is_unknown_rather_than_hardened() {
    let log = "=> printenv\nbootcmd=run sfboot\nEnvironment size: 20/65532 bytes\n";
    let verdicts = boot_chain::assess(log).verdicts;
    let delay = verdicts
        .iter()
        .find(|v| v.title == "Autoboot delay")
        .expect("an absent bootdelay still gets a verdict");
    assert_eq!(delay.state, "unknown");
    assert_eq!(delay.evidence, "bootdelay absent");
    assert!(verdicts.iter().all(|v| v.state != "hardened"));
}

#[test]
fn a_wrapped_value_is_rejoined_rather_than_ending_the_dump() {
    let log = fs::read_to_string(fixtures().join("wrapped-env.log")).expect("fixture");
    let session = boot_chain::assess(&log).session;
    let upfw = session
        .env
        .get("upfw")
        .expect("the wrapped variable survived");
    assert!(
        upfw.ends_with("${filesize};"),
        "continuation line was dropped: {upfw}"
    );
    // And the variable AFTER the wrap is still there, which is what a
    // boundary-on-continuation bug would have eaten.
    assert!(session.env.contains_key("upbootldr"));
}

#[test]
fn every_verdict_names_the_variable_it_read() {
    // A consultant has to defend the answer in a client report rather than
    // quote a tool, so an unevidenced verdict is a broken one.
    for name in declared_fixtures(&expectation()) {
        let log = fs::read_to_string(fixtures().join(&name)).expect("fixture");
        for verdict in boot_chain::assess(&log).verdicts {
            assert!(
                !verdict.evidence.trim().is_empty(),
                "{name}: {:?} has no evidence",
                verdict.title
            );
            assert!(
                ["confirmed", "exposed", "hardened", "unknown"].contains(&verdict.state.as_str()),
                "{name}: {:?} has state {:?}",
                verdict.title,
                verdict.state
            );
        }
    }
}

/// A checksum is not a signature, and the difference is the whole reason this
/// module reads the boot output rather than trusting `verify=`.
#[test]
fn a_passing_checksum_is_reported_as_integrity_not_authenticity() {
    let log = "## Booting kernel from Legacy Image at 82000000 ...\n\
                  Verifying Checksum ... OK\n\
               => printenv\nbootcmd=bootm 0x82000000\n\
               Environment size: 20/65532 bytes\n";
    let a = boot_chain::assess(log);
    assert_eq!(a.integrity.image_check.as_deref(), Some("uimage_crc"));
    assert_eq!(a.integrity.image_check_result.as_deref(), Some("passed"));
    let v = a
        .verdicts
        .iter()
        .find(|v| v.title == "Image verification")
        .expect("a verdict about verification");
    // `confirmed`, not `hardened`: the check happened, and it does not protect
    // against someone who can rewrite the image and its checksum.
    assert_eq!(v.state, "confirmed");
    assert!(v.detail.contains("not a signature"), "{}", v.detail);
    assert!(!a.integrity.image_signature_checked);
}

/// U-Boot prints `sha256,rsa2048:dev+ OK` when a signature node was verified,
/// and `sha256+ OK` when only a digest was. Reporting the second as a signature
/// would tell a client authenticity was established when it was not. No corpus
/// log exercises either, so these are the documented formats.
#[test]
fn only_a_signature_algorithm_establishes_authenticity() {
    let unsigned = boot_chain::parse_integrity("   Verifying Hash Integrity ... sha256+ OK\n");
    assert_eq!(unsigned.image_hash_algorithms, ["sha256"]);
    assert!(
        !unsigned.image_signature_checked,
        "an unsigned FIT hash was reported as a verified signature"
    );

    let signed =
        boot_chain::parse_integrity("   Verifying Hash Integrity ... sha256,rsa2048:dev+ OK\n");
    assert_eq!(signed.image_hash_algorithms, ["sha256", "rsa2048:dev"]);
    assert!(signed.image_signature_checked);
}

/// The false positive that reached production on the engine side: a substring
/// match on "Bad Magic Number" catches a filesystem complaint and reports a
/// failed image verification on a device whose bootloader said no such thing.
#[test]
fn a_filesystem_complaint_is_not_an_image_check_failure() {
    let fsck = boot_chain::parse_integrity(
        "fsck.ext2: Bad magic number in super-block while trying to open /dev/pramdisk0\n",
    );
    assert_eq!(fsck.image_check_failed, None);

    // And a test harness quoting U-Boot's error strings in a list, which is what
    // bootintel-7 actually contains.
    let harness = boot_chain::parse_integrity(
        "bootloader-commands: Wait for prompt ['=>', 'Bad Linux ARM64 Image magic!', 'TIMEOUT']\n",
    );
    assert_eq!(harness.image_check_failed, None);

    // U-Boot's own standalone line still registers.
    for line in ["   Bad Magic Number\n", "Bad Header Checksum.\n"] {
        assert!(
            boot_chain::parse_integrity(line)
                .image_check_failed
                .is_some(),
            "{line:?} was not recognised"
        );
    }
}

/// The HAB fuse is the anchor: while it is unblown the boot ROM runs unsigned
/// images whatever the bootloader prints afterwards. It must not depend on
/// having captured a printenv, because bootintel-7 does not have one.
#[test]
fn the_secure_boot_anchor_does_not_need_an_environment() {
    let log = "Normal Boot\nhab fuse not enabled\nu-boot=> boot\n";
    let a = boot_chain::assess(log);
    assert!(a.session.env.is_empty(), "this capture has no environment");
    let anchor = a
        .verdicts
        .iter()
        .find(|v| v.title == "Secure boot anchor")
        .expect("the anchor verdict survives a missing environment");
    assert_eq!(anchor.state, "exposed");
    assert_eq!(anchor.severity, "high");
}

/// A capture can carry both `verify=no` and an observed passing check. Emitting
/// one verdict for each produced two contradictory answers to one question,
/// which is a bug this codebase has already fixed once in another form.
#[test]
fn a_config_and_observation_conflict_is_one_verdict_that_names_it() {
    let log = "   Verifying Checksum ... OK\n=> printenv\nverify=no\n\
               Environment size: 20/65532 bytes\n";
    let entries: Vec<_> = boot_chain::assess(log)
        .verdicts
        .into_iter()
        .filter(|v| v.title == "Image verification")
        .collect();
    assert_eq!(
        entries.len(),
        1,
        "two verdicts for one question: {entries:?}"
    );
    assert!(
        entries[0].detail.contains("verify=no"),
        "the conflict is not explained: {}",
        entries[0].detail
    );
}

/// "The check ran" is a different claim from "the check passed".
#[test]
fn a_check_with_no_captured_result_is_unknown_rather_than_passing() {
    let log = "   Verifying Checksum ...\n=> printenv\nbootcmd=bootm 0x82000000\n\
               Environment size: 20/65532 bytes\n";
    let a = boot_chain::assess(log);
    assert_eq!(
        a.integrity.image_check_result.as_deref(),
        Some("not_captured")
    );
    let v = a
        .verdicts
        .iter()
        .find(|v| v.title == "Image verification")
        .expect("a verdict");
    assert_eq!(v.state, "unknown");
}

/// bdinfo is recognised by its own shape, because the command that produced it
/// cannot be relied on: bootintel-20 prints a full dump after `Boot-> bdinfo`,
/// and `Boot->` is not U-Boot's default prompt. A command-gated parser read that
/// entire dump as nothing.
#[test]
fn a_bdinfo_dump_behind_a_vendor_prompt_is_still_read() {
    let log = "Boot-> bdinfo\n\
               boot_params = 0x87D2EFB0\n\
               memstart    = 0x80000000\n\
               flashsize   = 0x01000000\n\
               ethaddr     = 00:1F:45:F2:B7:3B\n";
    let s = boot_chain::parse_session(log);
    assert_eq!(
        s.bdinfo.get("boot_params").map(String::as_str),
        Some("0x87D2EFB0")
    );
    assert_eq!(
        s.bdinfo.get("memstart").map(String::as_str),
        Some("0x80000000")
    );
    assert_eq!(s.bdinfo.len(), 4);
    assert!(s.reached, "a bdinfo dump proves someone was at the prompt");
}

/// The discriminator between board info and an environment variable is the
/// whitespace around the `=`. Without it, every environment dump containing
/// baudrate or ethaddr would also be recorded as board info.
#[test]
fn an_environment_line_is_not_board_info() {
    let log = "=> printenv\nbaudrate=115200\nethaddr=00:11:22:33:44:55\n\
               Environment size: 40/65532 bytes\n";
    let s = boot_chain::parse_session(log);
    assert!(
        s.bdinfo.is_empty(),
        "read the environment as board info: {:?}",
        s.bdinfo
    );
    assert_eq!(s.env.len(), 2);
}

/// `size` and `start` are too generic to be board-info keys on their own.
/// bootintel-5 prints an MTD table in a vendor format whose rows are exactly
/// `size = 0x180000`, and treating that as bdinfo was a live false positive.
#[test]
fn a_vendor_mtd_table_is_not_board_info() {
    let log = "mtd_part[0]:\nname = KERNEL\nsize = 0x180000\noffset = 0x37000\n";
    let s = boot_chain::parse_session(log);
    assert!(s.bdinfo.is_empty(), "{:?}", s.bdinfo);
}

/// U-Boot prints `device nor0 <spi0.0>, # parts = 4`: the bracketed chip id is
/// optional and the space in `# parts` is real. The engine's first pattern
/// required neither and so never matched, recording the partitions but not the
/// device they belong to.
#[test]
fn an_mtdparts_dump_is_parsed_including_its_device() {
    let log = "=> mtdparts\n\n\
               device nor0 <spi0.0>, # parts = 4\n \
               #: name                size            offset          mask_flags\n \
               0: u-boot              0x00020000      0x00000000      1\n \
               1: kernel              0x00100000      0x00020000      0\n";
    let s = boot_chain::parse_session(log);
    assert_eq!(s.mtd_device.as_deref(), Some("nor0"));
    assert_eq!(s.mtd_partitions.len(), 2);
    assert_eq!(s.mtd_partitions[0].name, "u-boot");
    assert_eq!(s.mtd_partitions[0].size, 0x20000);
    assert!(
        s.mtd_partitions[0].read_only,
        "mask_flags 1 means read-only"
    );
    assert!(!s.mtd_partitions[1].read_only);
}

#[test]
fn a_device_line_without_the_bracketed_chip_id_also_parses() {
    let s = boot_chain::parse_session("device nand0, #parts = 2\n");
    assert_eq!(s.mtd_device.as_deref(), Some("nand0"));
}