path: root/debug/risc-v-gsp-rm-vm/tests/test_gmmu_fbif_trace.py

"""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()