greeneye: GMMU/FBIF observability, PTE decoder and oracle, MPU/libos simulator wave

Simulator development wave (docs 42-44): GMMU page-table entry decode with its oracle/workflow harness, FBIF observability, and MPU/LIBOS device surfaces in the device model, plus refreshed boot/machine/profile data (tu104-570.144), the qemu-line results, and the new unit tests. Docs 00/06/35-41 updated to match; harness helpers and .gitignore included.

1 files changed, 484 insertions, 0 deletions

diff --git a/debug/risc-v-gsp-rm-vm/tests/test_gmmu_debug.py b/debug/risc-v-gsp-rm-vm/tests/test_gmmu_debug.py
new file mode 100644
index 0000000..d9046be
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/tests/test_gmmu_debug.py
@@ -0,0 +1,484 @@
+"""Tests for the GMMU/FBIF static<->dynamic join (harness/gmmu_debug.py, docs/44).
+
+All fixtures are synthetic: no firmware, no simulator run, no QEMU, no
+hardware.  The end-to-end demo test builds its own fixture in a temporary
+directory and checks the frozen threshold result.
+
+Run::
+
+    python3 -m unittest tests.test_gmmu_debug -v
+"""
+from __future__ import annotations
+
+import contextlib
+import io
+import json
+import os
+import sys
+import tempfile
+import unittest
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+TOOL_ROOT = os.path.abspath(os.path.join(HERE, ".."))
+if TOOL_ROOT not in sys.path:
+    sys.path.insert(0, TOOL_ROOT)
+
+from harness.gmmu_debug import (  # noqa: E402
+    DEBUG_REPORT_FORMAT,
+    HITS_HEADER,
+    MANUAL_STATIC_BASELINE_SECONDS,
+    TRACE_FORMAT,
+    WORKFLOW_THRESHOLD_FRACTION,
+    GmmuInputError,
+    build_debug_report,
+    load_sim_trace,
+    load_symbols,
+    main,
+    run_demo,
+    run_report,
+    write_hits_tsv,
+)
+
+
+def _write_json(path: str, payload) -> str:
+    with open(path, "w", encoding="utf-8") as fh:
+        json.dump(payload, fh)
+    return path
+
+
+def _symbols_json(path: str) -> str:
+    return _write_json(
+        path,
+        {
+            "format": "greeneye-symbols/1",
+            "symbols": [
+                {"addr": 0x4001000, "name": "gmmu_map_alloc", "size": 0x120},
+                {"addr": 0x4002000, "name": "fbif_setup", "size": 0x80},
+            ],
+        },
+    )
+
+
+def _events() -> list[dict]:
+    return [
+        {"seq": 0, "pc": "0x4002040", "kind": "write", "window": "falcon",
+         "offset": "0x624", "name": "FBIF_CTL", "value": "0x1"},
+        {"seq": 1, "pc": "0x4002050", "kind": "read", "window": "falcon",
+         "offset": "0x8", "name": "IRQSTAT", "value": "0x0"},
+        {"seq": 2, "pc": "0x4001010", "kind": "write", "window": "gmmu",
+         "offset": "0x0", "name": "GMMU_PDB_LO", "value": "0x400000"},
+        {"seq": 3, "pc": "0x4001020", "kind": "write", "window": "gmmu",
+         "offset": "0x0", "name": "GMMU_PDB_LO", "value": "0x800000"},
+        {"seq": 4, "pc": "0x4009000", "kind": "write", "window": "gmmu",
+         "offset": "0x10", "name": "GMMU_MODE", "value": "0x7"},
+    ]
+
+
+def _g1_document() -> dict:
+    """The nested shape emitted by sim/devices/observe.py:build_trace."""
+    return {
+        "schema": TRACE_FORMAT,
+        "image_sha256": "ab" * 32,
+        "profile": "tu104-570.144",
+        "profile_name": "fixture",
+        "symbols": "symbols.json",
+        "summary": {"windows": 1, "accesses": 2, "retained_accesses": 2,
+                    "dropped_accesses": 0},
+        "windows": [
+            {
+                "label": "falcon-observe",
+                "base": "0x300110000",
+                "size": "0x1000",
+                "mode": "observe-only",
+                "evidence": "header",
+                "counts": {"r": 1, "w": 1, "served": 0, "declined": 2},
+                "registers": [
+                    {
+                        "offset": "0x624",
+                        "name": "FBIF_CTL",
+                        "access": "rw",
+                        "counts": {"r": 0, "w": 1, "served": 0},
+                        "accesses": [
+                            {"addr": "0x300110624", "offset": "0x624", "size": 4,
+                             "rw": "w", "value": "0x1", "pc": "0x4002040", "vt": 123,
+                             "window": "falcon-observe", "register": "FBIF_CTL",
+                             "served": False, "pc_symbol": "fbif_setup"},
+                        ],
+                    }
+                ],
+                "unclassified": {
+                    "counts": {"r": 1, "w": 0, "served": 0},
+                    "accesses": [
+                        {"addr": "0x300110008", "offset": "0x8", "size": 4,
+                         "rw": "r", "value": None, "pc": "0x4002050", "vt": 123,
+                         "window": "falcon-observe", "register": None,
+                         "served": False, "pc_symbol": "fbif_setup"},
+                    ],
+                },
+            }
+        ],
+    }
+
+
+def _run_main(argv: list[str]) -> tuple[int, str, str]:
+    out, err = io.StringIO(), io.StringIO()
+    with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
+        rc = main(argv)
+    return rc, out.getvalue(), err.getvalue()
+
+
+class TraceIngestTests(unittest.TestCase):
+    def test_trace_tagged_and_aliases(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = _write_json(
+                os.path.join(tmp, "t.json"),
+                {
+                    "format": TRACE_FORMAT,
+                    "events": [
+                        {"pc": 0x4002040, "rw": "w", "block": "falcon",
+                         "reg_offset": 0x624, "register": "FBIF_CTL", "data": 1},
+                    ],
+                },
+            )
+            trace = load_sim_trace(path)
+            self.assertEqual(trace.format, TRACE_FORMAT)
+            obs = trace.observations[0]
+            self.assertEqual((obs.kind, obs.window, obs.offset, obs.name, obs.value),
+                             ("write", "falcon", 0x624, "FBIF_CTL", 1))
+
+    def test_jsonl_fallback(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = os.path.join(tmp, "t.jsonl")
+            with open(path, "w", encoding="utf-8") as fh:
+                fh.write(json.dumps({"format": TRACE_FORMAT}) + "\n")
+                fh.write(json.dumps({"t": "write", "addr": "0x624", "size": 4, "pc": "0x4002040"}) + "\n")
+                fh.write(json.dumps({"t": "summary", "insns": 1}) + "\n")
+            trace = load_sim_trace(path)
+            self.assertEqual(len(trace.observations), 1)
+            self.assertEqual(trace.observations[0].addr, 0x624)
+
+    def test_empty_event_list_is_legal(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = _write_json(os.path.join(tmp, "t.json"), {"format": TRACE_FORMAT, "events": []})
+            trace = load_sim_trace(path)
+            self.assertEqual(trace.observations, [])
+
+    def test_untagged_object_without_events_is_clean_error(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = _write_json(os.path.join(tmp, "t.json"), {"hello": "world"})
+            with self.assertRaises(GmmuInputError):
+                load_sim_trace(path)
+
+    def test_missing_trace_is_clean_error(self):
+        with self.assertRaises(GmmuInputError):
+            load_sim_trace("/does/not/exist.json")
+
+    def test_session_directory_ingest(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            blobs = os.path.join(tmp, "blobs")
+            os.makedirs(blobs)
+            _write_json(os.path.join(tmp, "session.json"), {"format": "greeneye-session/1"})
+            _write_json(
+                os.path.join(blobs, "gmmu-fbif-trace.json"),
+                {"format": TRACE_FORMAT, "events": _events()},
+            )
+            trace = load_sim_trace(tmp)
+            self.assertEqual(len(trace.observations), len(_events()))
+
+    def test_g1_nested_document_flattens(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = _write_json(os.path.join(tmp, "t.json"), _g1_document())
+            trace = load_sim_trace(path)
+            self.assertEqual(trace.format, TRACE_FORMAT)
+            self.assertEqual(len(trace.observations), 2)
+            self.assertEqual(trace.source["profile"], "tu104-570.144")
+            self.assertEqual(trace.source["summary"]["dropped_accesses"], 0)
+            self.assertEqual(trace.source["summary"]["retained_accesses"], 2)
+            write_obs = trace.observations[0]
+            self.assertEqual(write_obs.kind, "write")
+            self.assertEqual(write_obs.name, "FBIF_CTL")
+            self.assertEqual(write_obs.window, "falcon-observe")
+            self.assertEqual(write_obs.offset, 0x624)
+            self.assertEqual(write_obs.addr, 0x300110624)
+            self.assertEqual(write_obs.pc, 0x4002040)
+            self.assertEqual(write_obs.vt, 123)
+            self.assertIs(write_obs.handled, False)
+            self.assertEqual(write_obs.function, "fbif_setup")
+            read_obs = trace.observations[1]
+            self.assertEqual(read_obs.kind, "read")
+            self.assertEqual(read_obs.offset, 0x8)
+            self.assertIsNone(read_obs.value)
+
+    def test_g1_nested_document_report(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = _write_json(os.path.join(tmp, "t.json"), _g1_document())
+            report = build_debug_report(load_sim_trace(path), None)
+            self.assertEqual(report["summary"]["functions"], 1)
+            function = report["functions"][0]
+            self.assertEqual(function["function"], "fbif_setup")
+            registers = {r["register"]: r for r in function["registers"]}
+            self.assertEqual(registers["FBIF_CTL"]["write_count"], 1)
+            self.assertEqual(registers["0x8"]["read_count"], 1)
+
+
+class ObserveProducerIntegrationTests(unittest.TestCase):
+    """Ingest a document produced by the real sim/devices/observe.py module.
+
+    The producer is in-tree, so the import is unconditional: if it breaks,
+    the G1<->G3 interface has regressed and this test must fail, not skip.
+    """
+
+    def test_ingest_real_build_trace_document(self):
+        from sim.devices.observe import ObservedWindow, build_trace
+        window = ObservedWindow(
+            base=0x300110000,
+            size=0x1000,
+            label="integration",
+            registers={"0x624": {"name": "FBIF_CTL", "width": 4, "access": "rw"}},
+        )
+        self.assertFalse(window.write(0x624, 4, 1, pc=0x4002040))
+        self.assertIsNone(window.read(0x8, 4, pc=0x4002050))
+        registry = type("Registry", (), {"devices": [window]})()
+        document = build_trace(
+            registry, image_sha256="cd" * 32, profile="tu104-570.144"
+        )
+        self.assertEqual(document.get("schema"), TRACE_FORMAT)
+        with tempfile.TemporaryDirectory() as tmp:
+            path = _write_json(os.path.join(tmp, "t.json"), document)
+            trace = load_sim_trace(path)
+        self.assertEqual(trace.format, TRACE_FORMAT)
+        self.assertEqual(len(trace.observations), 2)
+        names = {obs.name for obs in trace.observations}
+        self.assertIn("FBIF_CTL", names)
+        write_obs = next(obs for obs in trace.observations if obs.kind == "write")
+        self.assertEqual(write_obs.offset, 0x624)
+        self.assertEqual(write_obs.value, 1)
+
+
+class SymbolLoadingTests(unittest.TestCase):
+    def test_symbols_json_db(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            db = load_symbols(_symbols_json(os.path.join(tmp, "s.json")))
+            self.assertEqual(db.resolve(0x4002040).name, "fbif_setup")
+
+    def test_symbols_plain_map_json(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = _write_json(os.path.join(tmp, "s.json"), {"0x4001000": "gmmu_map_alloc"})
+            db = load_symbols(path)
+            self.assertEqual(db.get("gmmu_map_alloc").addr, 0x4001000)
+
+    def test_symbols_tsv(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = os.path.join(tmp, "s.tsv")
+            with open(path, "w", encoding="utf-8") as fh:
+                fh.write("0x4001000\tgmmu_map_alloc\n0x4002000\tfbif_setup\n")
+            db = load_symbols(path)
+            self.assertEqual(db.resolve(0x4002000).name, "fbif_setup")
+
+    def test_missing_symbols_is_clean_error(self):
+        with self.assertRaises(GmmuInputError):
+            load_symbols("/does/not/exist.tsv")
+
+
+class ReportTests(unittest.TestCase):
+    def _report(self, events=None, symbols=None):
+        with tempfile.TemporaryDirectory() as tmp:
+            trace_path = _write_json(
+                os.path.join(tmp, "t.json"),
+                {"format": TRACE_FORMAT, "events": events if events is not None else _events()},
+            )
+            trace = load_sim_trace(trace_path)
+            return build_debug_report(trace, symbols)
+
+    def test_report_shape_and_resolution(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            symbols = load_symbols(_symbols_json(os.path.join(tmp, "s.json")))
+            report = self._report(symbols=symbols)
+            self.assertEqual(report["format"], DEBUG_REPORT_FORMAT)
+            summary = report["summary"]
+            self.assertEqual(summary["functions"], 3)
+            self.assertEqual(summary["writes"], 4)
+            self.assertEqual(summary["reads"], 1)
+            names = [f["function"] for f in report["functions"]]
+            self.assertEqual(names, ["gmmu_map_alloc", "fbif_setup", "sub_0x4009000"])
+            fbif = report["functions"][1]
+            self.assertEqual(fbif["addr"], 0x4002000)
+            registers = {r["register"]: r for r in fbif["registers"]}
+            self.assertEqual(registers["FBIF_CTL"]["write_count"], 1)
+            self.assertEqual(registers["FBIF_CTL"]["first_pc"], 0x4002040)
+            self.assertEqual(registers["FBIF_CTL"]["last_pc"], 0x4002040)
+            self.assertEqual(registers["IRQSTAT"]["read_count"], 1)
+            unknown = report["functions"][2]
+            self.assertEqual(unknown["function"], "sub_0x4009000")
+            self.assertEqual(unknown["addr"], 0x4009000)
+
+    def test_sequence_order_and_last_pc(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            symbols = load_symbols(_symbols_json(os.path.join(tmp, "s.json")))
+            report = self._report(symbols=symbols)
+            alloc = report["functions"][0]
+            pdb = next(r for r in alloc["registers"] if r["register"] == "GMMU_PDB_LO")
+            self.assertEqual(pdb["write_count"], 2)
+            self.assertEqual(pdb["values"], [0x400000, 0x800000])
+            self.assertEqual(pdb["first_pc"], 0x4001010)
+            self.assertEqual(pdb["last_pc"], 0x4001020)
+
+    def test_explicit_function_field_wins(self):
+        events = [
+            {"seq": 0, "pc": "0x4009000", "kind": "write", "window": "gmmu",
+             "offset": "0x10", "name": "GMMU_MODE", "value": "0x7",
+             "function": "fbif_setup"},
+        ]
+        with tempfile.TemporaryDirectory() as tmp:
+            symbols = load_symbols(_symbols_json(os.path.join(tmp, "s.json")))
+            report = self._report(events=events, symbols=symbols)
+            self.assertEqual(report["functions"][0]["function"], "fbif_setup")
+            self.assertEqual(report["functions"][0]["addr"], 0x4002000)
+
+    def test_no_symbols_gives_sub_labels(self):
+        report = self._report(symbols=None)
+        self.assertTrue(all(f["function"].startswith("sub_") for f in report["functions"]))
+
+    def test_events_without_pc_are_counted_not_attributed(self):
+        events = [{"seq": 0, "kind": "write", "window": "gmmu", "offset": "0x10",
+                   "name": "GMMU_MODE", "value": "0x7"}]
+        report = self._report(events=events, symbols=None)
+        self.assertEqual(report["summary"]["unknown_pc_events"], 1)
+        self.assertEqual(report["summary"]["functions"], 0)
+
+    def test_empty_trace(self):
+        report = self._report(events=[], symbols=None)
+        self.assertEqual(report["summary"]["functions"], 0)
+        self.assertEqual(report["summary"]["registers"], 0)
+        self.assertEqual(report["functions"], [])
+
+
+class HitsTsvTests(unittest.TestCase):
+    def test_hits_tsv_is_markgreeneye_compatible(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            symbols = load_symbols(_symbols_json(os.path.join(tmp, "s.json")))
+            trace = load_sim_trace(
+                _write_json(
+                    os.path.join(tmp, "t.json"),
+                    {"format": TRACE_FORMAT, "events": _events()},
+                )
+            )
+            report = build_debug_report(trace, symbols)
+            hits_path = write_hits_tsv(report, os.path.join(tmp, "greeneye_hits.tsv"))
+            with open(hits_path, "r", encoding="utf-8") as fh:
+                lines = [line.rstrip("\n") for line in fh]
+            self.assertEqual(lines[0], HITS_HEADER)
+            self.assertTrue(lines[0].startswith("addr"))
+            rows = [line.split("\t") for line in lines[1:]]
+            self.assertEqual(len(rows), 3)
+            for cols in rows:
+                self.assertGreaterEqual(len(cols), 2)
+                int(cols[0].replace("0x", ""), 16)
+                self.assertTrue(cols[1].isdigit())
+            first = rows[0]
+            self.assertEqual(first[0], "0x4001000")
+            self.assertEqual(first[2], "gmmu_map_alloc")
+            self.assertIn("write", first[3])
+            addrs = {int(cols[0], 16) for cols in rows}
+            self.assertIn(0x4009000, addrs)
+
+    def test_hits_tsv_empty_report(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            report = {"format": DEBUG_REPORT_FORMAT, "functions": []}
+            hits_path = write_hits_tsv(report, os.path.join(tmp, "greeneye_hits.tsv"))
+            with open(hits_path, "r", encoding="utf-8") as fh:
+                self.assertEqual(fh.read().strip(), HITS_HEADER)
+
+
+class ArtifactTests(unittest.TestCase):
+    def test_run_report_writes_artifacts(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            symbols_path = _symbols_json(os.path.join(tmp, "s.json"))
+            trace_path = _write_json(
+                os.path.join(tmp, "t.json"),
+                {"format": TRACE_FORMAT, "events": _events()},
+            )
+            out = os.path.join(tmp, "out")
+            report = run_report(trace_path, symbols_path, out)
+            for key in ("report_json", "report_md", "hits_tsv", "ghidra_command"):
+                self.assertIn(key, report["artifacts"])
+            self.assertTrue(os.path.isfile(report["artifacts"]["report_json"]))
+            self.assertTrue(os.path.isfile(report["artifacts"]["report_md"]))
+            self.assertTrue(os.path.isfile(report["artifacts"]["hits_tsv"]))
+            self.assertIn("MarkGreenEyeHits.java", report["artifacts"]["ghidra_command"])
+            with open(report["artifacts"]["report_json"], "r", encoding="utf-8") as fh:
+                loaded = json.load(fh)
+            self.assertEqual(loaded["summary"]["functions"], 3)
+            with open(report["artifacts"]["report_md"], "r", encoding="utf-8") as fh:
+                md = fh.read()
+            self.assertIn("Per function", md)
+            self.assertIn("Ghidra annotation", md)
+
+
+class DemoTests(unittest.TestCase):
+    def test_demo_end_to_end_and_frozen_threshold(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            out = io.StringIO()
+            with contextlib.redirect_stdout(out):
+                result = run_demo(tmp)
+            self.assertIn("RESULT_JSON:", out.getvalue())
+            self.assertTrue(result["pass"])
+            self.assertLessEqual(result["measured_seconds"], result["threshold_seconds"])
+            self.assertEqual(
+                result["threshold_seconds"],
+                round(MANUAL_STATIC_BASELINE_SECONDS * WORKFLOW_THRESHOLD_FRACTION, 4),
+            )
+            self.assertIn("estimate", result["manual_baseline_label"])
+            self.assertEqual(result["summary"]["functions"], 4)
+            self.assertEqual(result["summary"]["writes"], 11)
+            self.assertEqual(result["summary"]["reads"], 2)
+            for key in ("report_json", "report_md", "hits_tsv"):
+                self.assertTrue(os.path.isfile(result["artifacts"][key]))
+            self.assertTrue(os.path.isfile(os.path.join(tmp, "demo-result.json")))
+
+
+class CliTests(unittest.TestCase):
+    def test_report_cli(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            symbols_path = _symbols_json(os.path.join(tmp, "s.json"))
+            trace_path = _write_json(
+                os.path.join(tmp, "t.json"),
+                {"format": TRACE_FORMAT, "events": _events()},
+            )
+            out = os.path.join(tmp, "out")
+            rc, stdout, _stderr = _run_main(
+                ["report", "--trace", trace_path, "--symbols", symbols_path, "--out", out]
+            )
+            self.assertEqual(rc, 0)
+            payload = json.loads(stdout[stdout.index("{"):stdout.rindex("}") + 1])
+            self.assertEqual(payload["summary"]["functions"], 3)
+            self.assertTrue(os.path.isfile(os.path.join(out, "greeneye_hits.tsv")))
+
+    def test_report_cli_missing_trace_exits_2(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            rc, _stdout, stderr = _run_main(
+                ["report", "--trace", "/nope.json", "--out", os.path.join(tmp, "out")]
+            )
+            self.assertEqual(rc, 2)
+            self.assertIn("trace not found", stderr)
+
+    def test_empty_trace_cli_writes_report(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            trace_path = _write_json(
+                os.path.join(tmp, "t.json"), {"format": TRACE_FORMAT, "events": []}
+            )
+            out = os.path.join(tmp, "out")
+            rc, stdout, _stderr = _run_main(["report", "--trace", trace_path, "--out", out])
+            self.assertEqual(rc, 0)
+            payload = json.loads(stdout[stdout.index("{"):stdout.rindex("}") + 1])
+            self.assertEqual(payload["summary"]["functions"], 0)
+            self.assertTrue(os.path.isfile(os.path.join(out, "gmmu-debug-report.md")))
+
+    def test_no_command_prints_help(self):
+        rc, stdout, _stderr = _run_main([])
+        self.assertEqual(rc, 2)
+        self.assertIn("usage", stdout.lower())
+
+
+if __name__ == "__main__":
+    unittest.main()