fix: use UC_HOOK_MEM_READ for peripheral MMIO instead of UC_HOOK_MEM_READ_UNMAPPED

The unmapped hook only fires once per page — after the first read maps
the page, subsequent reads go to stale mapped memory. Now peripheral
region is mapped as R+W and every read/write is intercepted via
UC_HOOK_MEM_READ/WRITE hooks on the address range.

Results: vuln-firmware.elf now shows 962+ PIP reads per execution
(was 1 before). Fix also resolves circular import in fuzzing/__init__.py
and corrects vector table section (address 0) being skipped in load_elf.

1550 tests passing.

Author: RTOSploit

rtosploit/fuzzing/__init__.py

@@ -7,11 +7,19 @@
 from rtosploit.fuzzing.fuzz_input import FuzzInputStream, InputExhausted
 from rtosploit.fuzzing.input_injector import FuzzableInput, InputInjector
 from rtosploit.fuzzing.mutator import Mutator
-from rtosploit.fuzzing.unicorn_worker import (
-    UnicornFuzzEngine,
-    UnicornFuzzStats,
-    UnicornFuzzWorker,
-)
+
+# Unicorn imports are lazy to avoid circular dependency:
+# fuzzing/__init__ -> unicorn_worker -> unicorn_engine -> fuzzing/execution -> fuzzing/__init__
+
+
+def _get_unicorn_classes():
+    from rtosploit.fuzzing.unicorn_worker import (
+        UnicornFuzzEngine,
+        UnicornFuzzStats,
+        UnicornFuzzWorker,
+    )
+    return UnicornFuzzEngine, UnicornFuzzStats, UnicornFuzzWorker
+
 
 __all__ = [
     "CorpusManager",
@@ -26,8 +34,5 @@
     "InputInjector",
     "Mutator",
     "StopReason",
-    "UnicornFuzzEngine",
-    "UnicornFuzzStats",
-    "UnicornFuzzWorker",
     "make_result",
 ]

rtosploit/peripherals/unicorn_engine.py

@@ -201,17 +201,35 @@ def setup(self) -> None:
             # Raw binary: write entire firmware data to flash base
             uc.mem_write(self._firmware.base_address, self._firmware.data)
 
-        # 4. Peripheral region (0x40000000-0x5FFFFFFF): intentionally NOT mapped
-        # MMIO accesses trigger unmapped hooks -> PIP/SVD/fallback
+        # 4. Map peripheral region (0x40000000-0x5FFFFFFF) as R+W
+        # We map it so reads don't trigger UC_HOOK_MEM_READ_UNMAPPED (which
+        # only fires once per page). Instead, we use UC_HOOK_MEM_READ on the
+        # range to intercept EVERY read for PIP/SVD routing.
+        periph_size = _PERIPH_END - _PERIPH_START
+        uc.mem_map(_PERIPH_START, periph_size, 3)  # UC_PROT_READ | UC_PROT_WRITE
+        for page in range(_PERIPH_START, _PERIPH_END, 0x1000):
+            self._mapped_pages.add(page)
 
         # 5. Map system registers region (0xE0000000-0xE00FFFFF) with R+W
         uc.mem_map(_SYSTEM_REG_START, _SYSTEM_REG_END - _SYSTEM_REG_START, 3)
         for page in range(_SYSTEM_REG_START, _SYSTEM_REG_END, 0x1000):
             self._mapped_pages.add(page)
 
         # Set up hooks
-        uc.hook_add(UC_HOOK_MEM_READ_UNMAPPED, self._hook_mem_read_unmapped)
-        uc.hook_add(UC_HOOK_MEM_WRITE_UNMAPPED, self._hook_mem_write_unmapped)
+        # Use UC_HOOK_MEM_READ on peripheral range to intercept EVERY MMIO read
+        # (not just unmapped — the region is mapped but we override values via hook)
+        uc.hook_add(UC_HOOK_MEM_READ, self._hook_periph_read,
+                     begin=_PERIPH_START, end=_PERIPH_END - 1)
+        uc.hook_add(UC_HOOK_MEM_WRITE, self._hook_periph_write,
+                     begin=_PERIPH_START, end=_PERIPH_END - 1)
+        # System register reads/writes
+        uc.hook_add(UC_HOOK_MEM_READ, self._hook_sysreg_read,
+                     begin=_SYSTEM_REG_START, end=_SYSTEM_REG_END - 1)
+        uc.hook_add(UC_HOOK_MEM_WRITE, self._hook_sysreg_write,
+                     begin=_SYSTEM_REG_START, end=_SYSTEM_REG_END - 1)
+        # Unmapped access outside peripheral/system = crash
+        uc.hook_add(UC_HOOK_MEM_READ_UNMAPPED, self._hook_unmapped_access)
+        uc.hook_add(UC_HOOK_MEM_WRITE_UNMAPPED, self._hook_unmapped_access)
         uc.hook_add(UC_HOOK_BLOCK, self._hook_block)
         uc.hook_add(UC_HOOK_CODE, self._hook_code)
 
@@ -384,88 +402,54 @@ def restore_snapshot(self, snapshot: UnicornSnapshot) -> None:
     # MMIO hooks with PIP routing
     # ------------------------------------------------------------------
 
-    def _map_page_if_needed(self, uc, address: int) -> bool:
-        """Map a 4KB page if not already mapped. Returns True if mapped."""
-        page_base = address & ~0xFFF
-        if page_base in self._mapped_pages:
-            return True
-        try:
-            uc.mem_map(page_base, 0x1000, 3)  # R+W
-            self._mapped_pages.add(page_base)
-            return True
-        except Exception:
-            return False
+    def _hook_periph_read(self, uc, access, address, size, value, user_data):
+        """Intercept EVERY read in the peripheral MMIO range.
 
-    def _hook_mem_read_unmapped(self, uc, access, address, size, value, user_data):
-        """Handle unmapped memory reads with PIP routing.
-
-        Routing:
-        - Peripheral range (0x40000000-0x5FFFFFFF): SVD -> PIP -> fallback
-        - System registers (0xE0000000-0xE00FFFFF): system reg handler
-        - Other: crash (unmapped non-peripheral access)
+        Routes through CompositeMMIOHandler (SVD -> PIP -> fallback).
+        The peripheral region is mapped as R+W so this hook fires on every
+        access, not just the first (unlike UC_HOOK_MEM_READ_UNMAPPED).
         """
-        if _PERIPH_START <= address < _PERIPH_END:
-            # Peripheral MMIO range -> route through composite handler
-            try:
-                result = self._mmio_handler.read(address, size)
-            except InputExhausted:
-                self._stopped = True
-                self._stop_reason = "input_exhausted"
-                self._stop_reason_enum = StopReason.INPUT_EXHAUSTED
-                uc.emu_stop()
-                return False
-            self._map_page_if_needed(uc, address)
-            uc.mem_write(address, struct.pack("<I", result & 0xFFFFFFFF)[:size])
-            return True
-
-        elif _SYSTEM_REG_START <= address < _SYSTEM_REG_END:
-            # System registers -> composite handler routes to CortexMSystemRegisters
+        try:
             result = self._mmio_handler.read(address, size)
-            self._map_page_if_needed(uc, address)
-            uc.mem_write(address, struct.pack("<I", result & 0xFFFFFFFF)[:size])
-            return True
-
-        else:
-            # Non-peripheral unmapped access -> crash
+        except InputExhausted:
             self._stopped = True
-            self._stop_reason = "unmapped_access"
-            self._stop_reason_enum = StopReason.UNMAPPED_ACCESS
-            self._crash_address = address
-            self._crash_type = f"unmapped_read at 0x{address:08X}"
-            logger.debug("Unmapped non-peripheral read at 0x%08X", address)
+            self._stop_reason = "input_exhausted"
+            self._stop_reason_enum = StopReason.INPUT_EXHAUSTED
             uc.emu_stop()
-            return False
+            return
+        # Write the PIP/SVD result into mapped memory so the CPU reads it
+        uc.mem_write(address, struct.pack("<I", result & 0xFFFFFFFF)[:size])
 
-    def _hook_mem_write_unmapped(self, uc, access, address, size, value, user_data):
-        """Handle unmapped memory writes with PIP routing."""
-        if _PERIPH_START <= address < _PERIPH_END:
-            # Peripheral MMIO write
-            try:
-                self._mmio_handler.write(address, value, size)
-            except InputExhausted:
-                self._stopped = True
-                self._stop_reason = "input_exhausted"
-                self._stop_reason_enum = StopReason.INPUT_EXHAUSTED
-                uc.emu_stop()
-                return False
-            self._map_page_if_needed(uc, address)
-            return True
-
-        elif _SYSTEM_REG_START <= address < _SYSTEM_REG_END:
+    def _hook_periph_write(self, uc, access, address, size, value, user_data):
+        """Intercept EVERY write in the peripheral MMIO range."""
+        try:
             self._mmio_handler.write(address, value, size)
-            self._map_page_if_needed(uc, address)
-            return True
-
-        else:
-            # Non-peripheral unmapped access -> crash
+        except InputExhausted:
             self._stopped = True
-            self._stop_reason = "unmapped_access"
-            self._stop_reason_enum = StopReason.UNMAPPED_ACCESS
-            self._crash_address = address
-            self._crash_type = f"unmapped_write at 0x{address:08X}"
-            logger.debug("Unmapped non-peripheral write at 0x%08X", address)
+            self._stop_reason = "input_exhausted"
+            self._stop_reason_enum = StopReason.INPUT_EXHAUSTED
             uc.emu_stop()
-            return False
+
+    def _hook_sysreg_read(self, uc, access, address, size, value, user_data):
+        """Intercept reads in the system register range (0xE0000000+)."""
+        result = self._mmio_handler.read(address, size)
+        uc.mem_write(address, struct.pack("<I", result & 0xFFFFFFFF)[:size])
+
+    def _hook_sysreg_write(self, uc, access, address, size, value, user_data):
+        """Intercept writes in the system register range."""
+        self._mmio_handler.write(address, value, size)
+
+    def _hook_unmapped_access(self, uc, access, address, size, value, user_data):
+        """Handle unmapped access outside peripheral/system ranges = crash."""
+        is_write = access in (2, 3, 4, 5)  # UC_MEM_WRITE variants
+        self._stopped = True
+        self._stop_reason = "unmapped_access"
+        self._stop_reason_enum = StopReason.UNMAPPED_ACCESS
+        self._crash_address = address
+        self._crash_type = f"unmapped_{'write' if is_write else 'read'} at 0x{address:08X}"
+        logger.debug("Unmapped access at 0x%08X (crash)", address)
+        uc.emu_stop()
+        return False
 
     # ------------------------------------------------------------------
     # Block hook (coverage + interrupts)

rtosploit/utils/binary.py

@@ -225,9 +225,12 @@ def load_elf(path: Path) -> FirmwareImage:
                     base_address = addr
 
         for section in elf.iter_sections():
-            if section.header.sh_addr == 0:
-                continue
             flags = section.header.sh_flags
+            # Skip sections without ALLOC flag (debug, metadata, comments)
+            # Do NOT skip address-0 sections — Cortex-M vector table lives there
+            if not (flags & 0x2):  # SHF_ALLOC = 0x2
+                if section.header.sh_addr == 0:
+                    continue
             perms = ""
             if flags & 0x4:
                 perms += "r"

tests/unit/test_unicorn_engine.py

@@ -96,37 +96,35 @@ def test_setup_loads_firmware_sections(self) -> None:
         assert engine.execution_count == 0
 
     def test_mmio_read_routes_to_handler(self) -> None:
-        """Unmapped MMIO reads are routed to the CompositeMMIOHandler."""
+        """Peripheral MMIO reads are routed to the CompositeMMIOHandler."""
         from rtosploit.peripherals.models.mmio_fallback import CompositeMMIOHandler
 
         handler = CompositeMMIOHandler()
         fw = _make_firmware(with_sections=False)
         engine = UnicornRehostEngine(fw, mmio_handler=handler)
         engine.setup()
 
-        # Directly invoke the hook to verify routing
-        # Peripheral region (0x40000000+) is unmapped on purpose
-        uc_mock = MagicMock()
-        result = engine._hook_mem_read_unmapped(
+        # Directly invoke the peripheral read hook
+        uc_mock = engine._uc  # Use the real Unicorn instance
+        engine._hook_periph_read(
             uc_mock, None, 0x40000000, 4, 0, None,
         )
-        assert result is True
         # The handler should have been called (fallback returns 1 for first read)
+        assert handler.get_coverage_stats()["total"] > 0
 
     def test_mmio_write_routes_to_handler(self) -> None:
-        """Unmapped MMIO writes are routed to the CompositeMMIOHandler."""
+        """Peripheral MMIO writes are routed to the CompositeMMIOHandler."""
         from rtosploit.peripherals.models.mmio_fallback import CompositeMMIOHandler
 
         handler = CompositeMMIOHandler()
         fw = _make_firmware(with_sections=False)
         engine = UnicornRehostEngine(fw, mmio_handler=handler)
         engine.setup()
 
-        uc_mock = MagicMock()
-        result = engine._hook_mem_write_unmapped(
-            uc_mock, None, 0x40000000, 4, 0xDEADBEEF, None,
+        engine._hook_periph_write(
+            engine._uc, None, 0x40000000, 4, 0xDEADBEEF, None,
         )
-        assert result is True
+        assert handler.get_coverage_stats()["total"] > 0
 
     def test_hal_hook_registration(self) -> None:
         """add_hal_hook stores the handler for the given address."""

tests/unit/test_unicorn_pip_integration.py

@@ -226,14 +226,14 @@ def test_unmapped_non_peripheral_access_is_crash(self) -> None:
         engine = UnicornRehostEngine(fw, mmio_handler=handler)
         engine.setup()
 
-        # Directly test the hook with a non-peripheral address
+        # Directly test the unmapped access hook with a non-peripheral address
         uc_mock = MagicMock()
-        result = engine._hook_mem_read_unmapped(
-            uc_mock, None, 0x00000004, 4, 0, None,
+        result = engine._hook_unmapped_access(
+            uc_mock, 0, 0x70000004, 4, 0, None,  # Address outside all mapped regions
         )
         assert result is False
         assert engine._stop_reason_enum == StopReason.UNMAPPED_ACCESS
-        assert engine._crash_address == 0x00000004
+        assert engine._crash_address == 0x70000004
 
     def test_peripheral_read_routes_through_pip(self) -> None:
         """Peripheral MMIO reads route through PIP when configured."""
@@ -245,12 +245,10 @@ def test_peripheral_read_routes_through_pip(self) -> None:
         # Set up fuzz input with enough data for PIP
         engine.set_fuzz_input(b"\xFF" * 64)
 
-        # Directly test hook with a peripheral address
-        uc_mock = MagicMock()
-        result = engine._hook_mem_read_unmapped(
-            uc_mock, None, 0x40000100, 4, 0, None,
+        # Directly test peripheral read hook
+        engine._hook_periph_read(
+            engine._uc, None, 0x40000100, 4, 0, None,
         )
-        assert result is True
         # PIP should have been invoked (pip_handled counter incremented)
         stats = handler.get_coverage_stats()
         assert stats["pip_handled"] > 0