"""GMMU/FBIF observer tests: no-fake-read, no semantic change, trace schema. Every assertion is tied to the evidence documented in ``sim/devices/observe.py`` and ``docs/42-GMMU-FBIF-OBSERVABILITY.md``. The real-firmware runs are bounded (<= 20k instructions) and assert observed numbers only: adding observer windows must not change instructions, stop PC/reason, unknown MMIO, unhandled-in-window counts or served accesses. """ import json import os import sys import tempfile import unittest sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from sim.devices import ( # noqa: E402 ObservedWindow, PmcBlock, TRACE_SCHEMA, build_trace, collect_observations, load_profile, parse_window_spec, registry_from_profile, windows_from_section, ) from sim.devices.base import DeviceRegistry # noqa: E402 from sim.machine import Machine # noqa: E402 TOOL_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) DEFAULT_IMAGE = os.path.normpath( os.path.join(TOOL_ROOT, "..", "..", "firmware", "57X-GSP", "570.144", "gsp-570.144.fwimage.bin")) DEFAULT_PROFILE = os.path.join(TOOL_ROOT, "sim", "profiles", "tu104-570.144.json") PMC_BASE = 0x2000000000000000 PMC_BOOT_42 = PMC_BASE + 0xA00 class ObserveWindowUnitTests(unittest.TestCase): def test_observe_only_declines_and_records(self): reg = DeviceRegistry() dev = ObservedWindow(base=0x1000, size=0x100, label="unit", mode="observe-only", registers={0x10: "REG_A"}) reg.add(dev) value = reg.read(0x1010, 4, pc=0x4000) self.assertEqual(value, (False, 0)) self.assertEqual(reg.unhandled, 1) self.assertEqual(dev.counts["r"], 1) self.assertEqual(dev.counts["declined"], 1) entry = dev.accesses[0] self.assertEqual(entry["addr"], 0x1010) self.assertEqual(entry["offset"], 0x10) self.assertEqual(entry["size"], 4) self.assertEqual(entry["rw"], "r") self.assertIsNone(entry["value"]) # never invent a read value self.assertFalse(entry["served"]) self.assertEqual(entry["pc"], 0x4000) self.assertEqual(entry["register"], "REG_A") self.assertEqual(entry["window"], "unit") self.assertIsNone(entry["vt"]) # no clock supplied self.assertFalse(reg.write(0x1010, 4, 0xAB, pc=0x4004)) self.assertEqual(dev.accesses[1]["value"], 0xAB) self.assertFalse(dev.accesses[1]["served"]) def test_storage_reads_back_written_values_only(self): dev = ObservedWindow(base=0x2000, size=0x100, label="st", mode="storage", registers={ 0x10: {"name": "SPARE", "access": "rw"}, 0x14: {"name": "RO", "access": "ro"}, }) reg = DeviceRegistry() reg.add(dev) # Never-written storage register: declined, no invented reset value. self.assertEqual(reg.read(0x2010, 4), (False, 0)) self.assertTrue(reg.write(0x2010, 4, 0xCAFE)) self.assertEqual(reg.read(0x2010, 4), (True, 0xCAFE)) # Read-only register: writes declined, reads have no value source. self.assertFalse(reg.write(0x2014, 4, 1)) self.assertEqual(reg.read(0x2014, 4), (False, 0)) # Undeclared offset inside the window stays declined. self.assertEqual(reg.read(0x2018, 4), (False, 0)) # Wrong width is not served. self.assertEqual(reg.read(0x2010, 8), (False, 0)) self.assertEqual(dev.written[0x10], 0xCAFE) self.assertEqual(dev.counts["served"], 2) # one write + one read-back def test_storage_field_decode_in_trace(self): dev = ObservedWindow(base=0x3000, size=0x100, label="fields", mode="storage", registers={0x10: {"name": "CFG", "fields": {"LOW": [3, 0], "HIGH": [7, 4]}}}) reg = DeviceRegistry() reg.add(dev) reg.write(0x3010, 4, 0x0053, pc=0x100) reg.read(0x3010, 4, pc=0x104) doc = build_trace(reg, image_sha256="0" * 64, profile="p.json", profile_name="p") win = doc["windows"][0] self.assertEqual(win["label"], "fields") self.assertEqual(win["mode"], "storage") reg_entry = win["registers"][0] self.assertEqual(reg_entry["name"], "CFG") self.assertEqual(reg_entry["counts"], {"r": 1, "w": 1, "served": 2}) write, read = reg_entry["accesses"] self.assertEqual(write["value"], "0x53") self.assertEqual(write["decoded"], {"LOW": 3, "HIGH": 5}) self.assertEqual(read["value"], "0x53") def test_windows_from_section_expands_count_and_stride(self): section = { "mode": "observe-only", "registers": {"0x0": "REG0"}, "windows": [ {"base": "0x1000", "size": "0x100", "label": "single"}, {"base": "0x2000", "size": "0x100", "label": "many", "count": 3, "stride": "0x1000"}, {"base": "0x3000", "size": "0x100", "label": "override", "mode": "storage"}, ], } windows = windows_from_section(section) self.assertEqual([w["label"] for w in windows], ["single", "many-0", "many-1", "many-2", "override"]) self.assertEqual([w["base"] for w in windows], [0x1000, 0x2000, 0x3000, 0x4000, 0x3000]) self.assertEqual(windows[1]["registers"], {"0x0": "REG0"}) self.assertEqual(windows[-1]["mode"], "storage") def test_parse_window_spec(self): win = parse_window_spec("0x2000000000100000:0x1000:adhoc") self.assertEqual((win["base"], win["size"], win["label"], win["mode"]), (0x2000000000100000, 0x1000, "adhoc", "observe-only")) win = parse_window_spec("0x1000:0x100:st:storage") self.assertEqual(win["mode"], "storage") with self.assertRaises(ValueError): parse_window_spec("0x1000") with self.assertRaises(ValueError): parse_window_spec("0x1000:0x100:bad:wrong-mode") def test_state_round_trip(self): dev = ObservedWindow(base=0x4000, size=0x100, label="rt", mode="storage", registers={0x10: "SPARE"}) dev.write(0x10, 4, 0x1234) snap = dev.state() other = ObservedWindow(base=0x4000, size=0x100, label="rt", mode="storage", registers={0x10: "SPARE"}) other.load_state(snap) self.assertEqual(other.state(), dev.state()) self.assertEqual(other.written[0x10], 0x1234) def test_access_log_cap_is_reported_not_silent(self): dev = ObservedWindow(base=0x6000, size=0x100, label="cap", max_access_log=2, registers={0x10: "REG_A"}) for value in (1, 2, 3, 4): dev.write(0x10, 4, value) doc = build_trace(type("Registry", (), {"devices": [dev]})()) self.assertEqual(dev.counts["w"], 4) win = doc["windows"][0] self.assertEqual(win["counts"]["w"], 4) self.assertEqual(win["retained_accesses"], 2) self.assertEqual(win["dropped_accesses"], 2) self.assertEqual(win["registers"][0]["counts"]["w"], 2) summary = doc["summary"] self.assertEqual(summary["accesses"], 4) self.assertEqual(summary["retained_accesses"], 2) self.assertEqual(summary["dropped_accesses"], 2) class NoSemanticChangeTests(unittest.TestCase): def test_observe_only_matches_no_window_lenient(self): reg_a = DeviceRegistry() reg_a.add(ObservedWindow(base=0x50000000, size=0x1000, label="obs")) reg_b = DeviceRegistry() a = Machine(memory_mb=1, ram_base=0, devices=reg_a, unknown_mmio="lenient") b = Machine(memory_mb=1, ram_base=0, devices=reg_b, unknown_mmio="lenient") self.assertEqual(a._mmio_load(0x50000010, 4, pc=0x40), b._mmio_load(0x50000010, 4, pc=0x40)) self.assertEqual(a.mmio_unknown, b.mmio_unknown) self.assertEqual(a.mmio_log, b.mmio_log) a._mmio_store(0x50000010, 4, 7, pc=0x44) b._mmio_store(0x50000010, 4, 7, pc=0x44) self.assertEqual(a.mmio_unknown, b.mmio_unknown) self.assertEqual(a.mmio_log, b.mmio_log) def test_observe_only_matches_no_window_strict(self): from sim.machine import Fault reg_a = DeviceRegistry() reg_a.add(ObservedWindow(base=0x50000000, size=0x1000, label="obs")) a = Machine(memory_mb=1, ram_base=0, devices=reg_a, unknown_mmio="fault") b = Machine(memory_mb=1, ram_base=0, devices=DeviceRegistry(), unknown_mmio="fault") with self.assertRaises(Fault) as ca: a._mmio_load(0x50000010, 4, pc=0x40) with self.assertRaises(Fault) as cb: b._mmio_load(0x50000010, 4, pc=0x40) self.assertEqual(ca.exception.cause, cb.exception.cause) self.assertEqual(ca.exception.tval, cb.exception.tval) def test_observer_does_not_shadow_a_modelled_window(self): reg = DeviceRegistry() pmc = PmcBlock(chip_id=0x164, base=PMC_BASE, size=0x1000) reg.add(pmc) obs = ObservedWindow(base=PMC_BOOT_42, size=0x10, label="overlay") reg.add(obs) machine = Machine(memory_mb=1, ram_base=0, devices=reg, unknown_mmio="lenient") value = machine._mmio_load(PMC_BOOT_42, 4, pc=0x400) self.assertEqual(value, pmc.read(0xA00, 4)) self.assertEqual(reg.unhandled, 0) self.assertEqual(len(obs.accesses), 1) self.assertFalse(obs.accesses[0]["served"]) self.assertIsNone(obs.accesses[0]["value"]) def test_gmmu_wpr_overlap_leaves_underlying_path_identical(self): # gmmu-wpr-lock (profile) overlaps the fwsec WPR2 window; the # observer must record and decline, leaving whatever the underlying # path does byte-for-byte identical to a run without the observer. profile = load_profile(DEFAULT_PROFILE) addr = 0x20000000001FA824 def probe(p): machine = Machine(memory_mb=16, ram_base=0, unknown_mmio="lenient") reg = registry_from_profile(p, machine=machine) machine.devices = reg value = machine._mmio_load(addr, 4, pc=0x40) observations = collect_observations(reg) return (machine.mmio_unknown, reg.unhandled, value, {d.label: d.counts for d in observations}) with_observers = probe(profile) stripped = json.loads(json.dumps(profile)) stripped["devices"].pop("gmmu", None) stripped["devices"].pop("fbif", None) without_observers = probe(stripped) self.assertEqual(with_observers[:3], without_observers[:3]) self.assertEqual(with_observers[3]["gmmu-wpr-lock"]["declined"], 1) self.assertEqual(without_observers[3], {}) class ProfileAndBootTests(unittest.TestCase): def test_profile_declares_gmmu_fbif_observe_windows(self): profile = load_profile(DEFAULT_PROFILE) self.assertIn("gmmu", profile["devices"]) self.assertIn("fbif", profile["devices"]) machine = Machine(memory_mb=16, ram_base=0, unknown_mmio="lenient") reg = registry_from_profile(profile, machine=machine) observers = collect_observations(reg) self.assertEqual(len(observers), 23) by_label = {d.label: d for d in observers} self.assertEqual(by_label["gmmu-pri-mmu"].base, 0x2000000000100C80) self.assertEqual(by_label["gmmu-pri-mmu"].size, 0x280) self.assertEqual(by_label["gmmu-wpr-lock"].base, 0x20000000001FA7D8) self.assertEqual(by_label["gmmu-hsmmu"].base, 0x20000000001FAC00) self.assertEqual(by_label["fbpa"].base, 0x20000000009A0000) self.assertEqual(by_label["fbpa-partition-15"].base, 0x2000000000900000 + 15 * 0x4000) self.assertEqual(by_label["fbpa-mc-2"].base, 0x2000000000980000 + 2 * 0x4000) for dev in observers: self.assertEqual(dev.mode, "observe-only") self.assertEqual(by_label["gmmu-pri-mmu"].registers[0x0]["name"], "NV_PFB_PRI_MMU_CTRL") self.assertEqual(by_label["fbpa"].registers[0x50]["name"], "NV_PFB_FBPA_SPARE") def test_b2_numbers_unchanged_by_observer_windows(self): from sim.gsp_boot import run_boot r = run_boot(image=DEFAULT_IMAGE, profile_path=DEFAULT_PROFILE, instructions=20_000, record=False, use_shim=True) self.assertEqual(r.rung, "B2") self.assertEqual(r.instructions, 20_000) self.assertEqual(r.stop_reason, "budget") self.assertEqual(r.mmio_unknown, 0) self.assertEqual(r.device_unhandled, 0) self.assertEqual(r.device_accesses, 15) # The run reaches the WPR2 reads inside the gmmu-wpr-lock observer # window (offsets 0x4C/0x50); the observer records them without # serving, and the fwsec device still answers. self.assertIsNotNone(r.trace_summary) self.assertEqual(r.trace_summary["windows"], 23) self.assertEqual(r.trace_summary["accesses"], 2) self.assertEqual(r.trace_summary["served"], 0) # Same run with the observer sections removed: identical execution. with open(DEFAULT_PROFILE, "r", encoding="utf-8") as fh: profile = json.load(fh) del profile["devices"]["gmmu"] del profile["devices"]["fbif"] with tempfile.TemporaryDirectory() as tmp: path = os.path.join(tmp, "no-observe.json") with open(path, "w", encoding="utf-8") as fh: json.dump(profile, fh) base = run_boot(image=DEFAULT_IMAGE, profile_path=path, instructions=20_000, record=False, use_shim=True) self.assertEqual(base.instructions, r.instructions) self.assertEqual(base.stop_reason, r.stop_reason) self.assertEqual(base.stop_pc, r.stop_pc) self.assertEqual(base.mmio_unknown, r.mmio_unknown) self.assertEqual(base.device_unhandled, r.device_unhandled) self.assertEqual(base.device_accesses, r.device_accesses) self.assertEqual(base.registers, r.registers) def test_boot_trace_schema_and_determinism(self): from sim.gsp_boot import run_boot with tempfile.TemporaryDirectory() as tmp: path = os.path.join(tmp, "trace-dir") os.makedirs(path) r = run_boot(image=DEFAULT_IMAGE, profile_path=DEFAULT_PROFILE, instructions=20_000, record=False, use_shim=True, trace_out=path) self.assertEqual(r.trace_path, os.path.join(path, "gmmu-fbif-trace.json")) with open(r.trace_path, "r", encoding="utf-8") as fh: doc = json.load(fh) self.assertEqual(doc["schema"], TRACE_SCHEMA) self.assertEqual(doc["image_sha256"], r.image_sha256) self.assertEqual(set(doc["summary"]), {"windows", "windows_touched", "accesses", "retained_accesses", "dropped_accesses", "reads", "writes", "served", "registers_touched"}) self.assertEqual(doc["summary"]["accesses"], 2) self.assertEqual(doc["summary"]["retained_accesses"], 2) self.assertEqual(doc["summary"]["dropped_accesses"], 0) labels = [w["label"] for w in doc["windows"]] self.assertIn("gmmu-pri-mmu", labels) self.assertIn("fbpa", labels) for win in doc["windows"]: self.assertEqual(set(win), {"label", "base", "size", "mode", "evidence", "counts", "dropped_accesses", "retained_accesses", "registers", "unclassified"}) self.assertEqual(win["mode"], "observe-only") self.assertEqual(win["dropped_accesses"], 0) self.assertTrue(win["base"].startswith("0x")) r2 = run_boot(image=DEFAULT_IMAGE, profile_path=DEFAULT_PROFILE, instructions=20_000, record=False, use_shim=True) self.assertEqual(r.device_state_sha256, r2.device_state_sha256) def test_runtime_window_records_without_faking_and_resolves_symbols(self): from sim.gsp_boot import run_boot symbols = {"format": "greeneye-symbols/1", "count": 1, "symbols": [{"addr": "0x4009014", "name": "boot_probe", "size": 64}]} with tempfile.TemporaryDirectory() as tmp: sym_path = os.path.join(tmp, "symbols.json") trace_path = os.path.join(tmp, "trace.json") with open(sym_path, "w", encoding="utf-8") as fh: json.dump(symbols, fh) r = run_boot( image=DEFAULT_IMAGE, profile_path=DEFAULT_PROFILE, instructions=20_000, record=False, use_shim=True, observe_windows=["0x%x:0x10:pmc-probe" % PMC_BOOT_42], symbols_path=sym_path, trace_out=trace_path) # Two gmmu-wpr-lock reads (the WPR2 pair) plus the two BOOT_42 # reads inside the runtime PMC probe window. self.assertEqual(r.trace_summary["accesses"], 4) self.assertEqual(r.trace_summary["windows"], 24) with open(trace_path, "r", encoding="utf-8") as fh: doc = json.load(fh) self.assertEqual(doc["symbols"], sym_path) win = [w for w in doc["windows"] if w["label"] == "pmc-probe"][0] entry = win["unclassified"]["accesses"][0] self.assertEqual(entry["addr"], "0x%x" % PMC_BOOT_42) self.assertEqual(entry["rw"], "r") self.assertIsNone(entry["value"]) # observe-only: no fake read self.assertFalse(entry["served"]) self.assertEqual(entry["pc_symbol"], "boot_probe") self.assertIsInstance(entry["vt"], int) if __name__ == "__main__": unittest.main()