state_plugins: Map the heap region lazily and grow it on demand. (#6715)

angr/sim_state.py

@@ -26,6 +26,7 @@ if TYPE_CHECKING:
     from angr.simos.javavm import SimJavaVM
 
     from .state_plugins.callstack import CallStack
+    from .state_plugins.heap.heap_base import SimHeapBase
     from .state_plugins.history import SimStateHistory
     from .state_plugins.inspect import SimInspector
     from .state_plugins.jni_references import SimStateJNIReferences
@@ -90,6 +91,7 @@ class SimState[IPTypeConc, IPTypeSym](PluginHub[SimStatePlugin]):
     inspect: SimInspector
     jni_references: SimStateJNIReferences
     scratch: SimStateScratch
+    heap: SimHeapBase
 
     def __init__(
         self,

angr/state_plugins/gdb.py

@@ -66,7 +66,11 @@ class GDB(SimStatePlugin):
         """
         # We set the heap at the same addresses as the gdb session to avoid pointer corruption.
         data = self._read_data(heap_dump)
-        self.state.heap.heap_location = heap_base + len(data)
+        if hasattr(self.state.heap, "heap_location"):
+            self.state.heap.heap_location = heap_base + len(data)  # type: ignore[assignment]
+        # the heap region is mapped lazily; moving the break by hand has to grow it too. This is a no-op unless
+        # the dump was taken at the same base as the heap plugin is configured for.
+        self.state.heap._ensure_mapped(heap_base + len(data))  # pylint: disable=protected-access
         addr = heap_base
         l.info("Set heap from 0x%x to %#x", addr, addr + len(data))
         # FIXME: we should probably make we don't overwrite other stuff loaded there

angr/state_plugins/heap/heap_base.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
+import contextlib
 import logging
+from typing import Self
 
 import angr.sim_options as opts
 from angr.errors import SimMemoryError
@@ -12,6 +14,12 @@ l = logging.getLogger("angr.state_plugins.heap.heap_base")
 DEFAULT_HEAP_LOCATION = 0xC0000000
 DEFAULT_HEAP_SIZE = 0x00800000
 
+# The number of bytes the heap region ``init_state`` maps up front.
+HEAP_INITIAL_MAPPED_SIZE = 0x2000
+
+# The factor by which the mapped extent of the heap grows every time an allocation runs past its end.
+HEAP_MAPPING_GROWTH_FACTOR = 2
+
 
 class SimHeapBase(SimStatePlugin):
     """
@@ -21,7 +29,8 @@ class SimHeapBase(SimStatePlugin):
     implementation, which is based on the original libc SimProcedure implementations.
 
     :ivar heap_base: the address of the base of the heap in memory
-    :ivar heap_size: the total size of the main memory region managed by the heap in memory
+    :ivar heap_size: the maximum size of the main memory region managed by the heap in memory. Only the part of it
+                     that has actually been handed out is mapped; see ``_ensure_mapped``.
     :ivar mmap_base: the address of the region from which large mmap allocations will be made
     """
 
@@ -32,13 +41,84 @@ class SimHeapBase(SimStatePlugin):
         self.heap_size = heap_size if heap_size is not None else DEFAULT_HEAP_SIZE
         self.mmap_base = self.heap_base + self.heap_size * 2
 
-    def copy(self, memo):
-        o = super().copy(memo)
+        # The exclusive end address of the part of the heap region that this plugin has mapped into the memory
+        # plugin. ``None`` means that this plugin does not manage the heap mapping.
+        self._mapped_end: int | None = None
+
+    def copy(self, memo) -> Self:
+        o: SimHeapBase = super().copy(memo)
         o.heap_base = self.heap_base
         o.heap_size = self.heap_size
         o.mmap_base = self.mmap_base
+        o._mapped_end = self._mapped_end  # pylint:disable=protected-access
         return o
 
+    def __setstate__(self, state):
+        if "_mapped_end" not in state:
+            # unpickling a state from before the heap mapping became lazy: back then init_state() mapped the
+            # entire region up front, so that is what the mapped extent has to be
+            state = dict(state)
+            state["_mapped_end"] = state["heap_base"] + state["heap_size"]
+        self.__dict__.update(state)
+
+    def _combine_mapped_end(self, others):
+        """
+        Reconcile the mapped extent of this heap with the heaps it is being merged with.
+        """
+        ends = [self._mapped_end, *(o._mapped_end for o in others)]  # pylint: disable=protected-access
+        self._mapped_end = None if any(e is None for e in ends) else min(ends)
+
+    def _map_range(self, start, end):
+        """
+        Map ``[start, end)`` as read/write, tolerating pages that happen to be mapped already.
+
+        :param start:   the first address to map (rounded down to a page boundary by the memory plugin)
+        :param end:     the exclusive end of the range to map
+        """
+        try:
+            self.state.memory.map_region(start, end - start, 3)
+        except SimMemoryError:
+            # something in this range is mapped already (an mmap, a gdb heap dump, a merge with a state that had
+            # grown further, ...). Redo it page by page and skip whatever exists.
+            page_size = getattr(self.state.memory, "page_size", 0x1000)
+            for page_addr in range(start - (start % page_size), end, page_size):
+                with contextlib.suppress(SimMemoryError):
+                    self.state.memory.map_region(page_addr, page_size, 3)
+
+    def _init_mapping(self):
+        """
+        Map the initial slice of the heap region. Called by ``init_state`` once it has determined that this plugin
+        owns the mapping of the region.
+        """
+        self._mapped_end = self.heap_base
+        self._ensure_mapped(self.heap_base + min(HEAP_INITIAL_MAPPED_SIZE, self.heap_size))
+
+    def _ensure_mapped(self, addr):
+        """
+        Make sure every byte of the heap region below ``addr`` is backed by a mapped page, growing the mapped
+        extent geometrically.
+
+        :param addr:    the exclusive end of the heap range that is about to be used
+        """
+        mapped_end = self._mapped_end
+        if mapped_end is None:
+            # this plugin does not manage the mapping of the heap region
+            return
+
+        region_end = self.heap_base + self.heap_size
+        addr = min(addr, region_end)
+        if addr <= mapped_end:
+            return
+
+        size = max(mapped_end - self.heap_base, HEAP_INITIAL_MAPPED_SIZE)
+        while self.heap_base + size < addr:
+            size *= HEAP_MAPPING_GROWTH_FACTOR
+        new_end = min(self.heap_base + size, region_end)
+
+        l.debug("Growing the mapped heap region to %#x (%d bytes)", new_end, new_end - self.heap_base)
+        self._map_range(mapped_end, new_end)
+        self._mapped_end = new_end
+
     def _conc_alloc_size(self, sim_size):
         """
         Concretizes a size argument, if necessary, to something that makes sense when allocating space. Here we just
@@ -124,4 +204,4 @@ class SimHeapBase(SimStatePlugin):
             self.state.memory.permissions(self.heap_base)
         except SimMemoryError:
             l.debug("Mapping base heap region")
-            self.state.memory.map_region(self.heap_base, self.heap_size, 3)
+            self._init_mapping()

angr/state_plugins/heap/heap_brk.py

@@ -1,11 +1,13 @@
 from __future__ import annotations
 
 import logging
+from typing import Self
 
 import claripy
 
 from angr.errors import SimSolverError
 from angr.sim_state import SimState
+from angr.state_plugins.libc import SimStateLibc
 from angr.state_plugins.plugin import SimStatePlugin
 
 from . import SimHeapBase
@@ -36,7 +38,7 @@ class SimHeapBrk(SimHeapBase):
         self.heap_location = self.heap_base
 
     @SimStatePlugin.memo
-    def copy(self, memo):
+    def copy(self, memo) -> Self:
         o = super().copy(memo)
         o.heap_location = self.heap_location
         return o
@@ -52,8 +54,12 @@ class SimHeapBrk(SimHeapBase):
         size = self._conc_alloc_size(sim_size)
         while size % 16 != 0:
             size += 1
+        assert isinstance(self.state.heap, SimHeapBrk)
         addr = self.state.heap.heap_location
         self.state.heap.heap_location += size
+        # the heap region is mapped lazily, so bumping the break may run past the end of what is mapped. Grow the
+        # mapping before handing the space out; this is the only place in this heap where the break moves up.
+        self.state.heap._ensure_mapped(self.state.heap.heap_location)  # pylint: disable=protected-access
         l.debug("Allocating %d bytes at address %#08x", size, addr)
         return addr
 
@@ -62,6 +68,11 @@ class SimHeapBrk(SimHeapBase):
         The memory release primitive for this heap implementation. Decreases the position of the break to deallocate
         space. Guards against releasing beyond the initial heap base.
 
+        The mapped extent of the heap is deliberately not shrunk here. Unmapping the released pages would discard
+        their contents, so a release followed by an allocation would hand back blank memory instead of the old
+        bytes, which is a behavior change; and since the extent only ever grows to the high-water mark of the
+        break, keeping it costs nothing beyond what the program already asked for.
+
         :param sim_size: a size specifying how much to decrease the break pointer by (may be symbolic or not)
         """
         requested = self._conc_alloc_size(sim_size)
@@ -78,6 +89,7 @@ class SimHeapBrk(SimHeapBase):
 
     def _calloc(self, sim_nmemb, sim_size):
         plugin = self.state.get_plugin("libc")
+        assert isinstance(plugin, SimStateLibc)
 
         if self.state.solver.symbolic(sim_nmemb):
             # TODO: find a better way
@@ -98,6 +110,7 @@ class SimHeapBrk(SimHeapBase):
         ):
             final_size = plugin.max_variable_size
 
+        assert isinstance(self.state.heap, SimHeapBrk)
         addr = self.state.heap.allocate(final_size)
         v = claripy.BVV(0, final_size * 8)
         self.state.memory.store(addr, v)
@@ -105,6 +118,7 @@ class SimHeapBrk(SimHeapBase):
 
     def _realloc(self, ptr, size):
         if size.symbolic:
+            assert isinstance(self.state.libc, SimStateLibc)
             try:
                 size_int = self.state.solver.max(size, extra_constraints=(size < self.state.libc.max_variable_size,))
             except SimSolverError:
@@ -113,6 +127,7 @@ class SimHeapBrk(SimHeapBase):
         else:
             size_int = self.state.solver.eval(size)
 
+        assert isinstance(self.state.heap, SimHeapBrk)
         addr = self.state.heap.allocate(size_int)
 
         if self.state.solver.eval(ptr) != 0:
@@ -122,6 +137,7 @@ class SimHeapBrk(SimHeapBase):
         return addr
 
     def _combine(self, others):
+        self._combine_mapped_end(others)
         new_heap_location = max(o.heap_location for o in others)
         if self.heap_location != new_heap_location:
             self.heap_location = new_heap_location

angr/state_plugins/heap/heap_ptmalloc.py

@@ -367,6 +367,12 @@ class SimHeapPTMalloc(SimHeapFreelist):
         while chunk is None:
             free_size = free_chunk.get_size()
             free_size = concretize(free_size, self.state.solver, sym_free_size_handler)
+            if free_size >= size:
+                # This chunk is about to be carved up. The heap region is mapped lazily, so make sure the chunk
+                # itself, plus the metadata of whatever chunk ends up following it, is backed before anything
+                # writes to it. This is the only place where this heap reaches above the chunks it already
+                # handed out; free()/realloc() only ever touch chunks that malloc() has already mapped.
+                self._ensure_mapped(free_chunk.base + size + 2 * self._chunk_min_size)
             if free_size < size:
                 # Chunk is too small to be used; move to the next or fail
                 fwd = free_chunk.fwd_chunk()
@@ -582,13 +588,33 @@ class SimHeapPTMalloc(SimHeapFreelist):
         if any(o._chunk_align_mask != self._chunk_align_mask for o in others):
             raise SimMergeError("Cannot merge heaps with different chunk alignments")
 
+        self._combine_mapped_end(others)
+
         return False
 
     def merge(self, others, merge_conditions, common_ancestor=None):  # pylint:disable=unused-argument
         return self._combine(others)
 
+    def _map_final_page(self):
+        """
+        This heap keeps the usage information of the real final chunk in the last word of the heap region (see
+        ``_set_final_freeness``), which is written by ``init_state`` and read or written by every operation that
+        touches the last chunk. The region is mapped lazily and will not normally grow that far, so map the single
+        page that holds that word up front.
+        """
+        if self._mapped_end is None:
+            return
+        page_size = getattr(self.state.memory, "page_size", 0x1000)
+        region_end = self.heap_base + self.heap_size
+        final_page = (region_end - 1) - ((region_end - 1) % page_size)
+        if final_page < self._mapped_end:
+            # already covered by the initial mapping
+            return
+        self._map_range(final_page, region_end)
+
     def init_state(self):
         super().init_state()
+        self._map_final_page()
 
         self._chunk_size_t_size = self.state.arch.bytes
         self._chunk_min_size = 4 * self._chunk_size_t_size

tests/state_plugins/test_heap_lazy_mapping.py

@@ -0,0 +1,441 @@
+#!/usr/bin/env python3
+# pylint: disable=missing-class-docstring,no-self-use,protected-access
+from __future__ import annotations
+
+__package__ = __package__ or "tests.state_plugins"  # pylint:disable=redefined-builtin
+
+import os
+import pickle
+import unittest
+
+import claripy
+
+import angr
+from angr import SimHeapBrk, SimHeapPTMalloc, SimState
+from angr.state_plugins.heap.heap_base import HEAP_INITIAL_MAPPED_SIZE, HEAP_MAPPING_GROWTH_FACTOR
+from tests.common import bin_location
+
+gdb_data_location = os.path.join(bin_location, "tests_data", "test_gdb_plugin")
+
+# unconstrained reads out of a fresh heap are noisy and (for SimHeapPTMalloc) slow; nothing here depends on the
+# filler, so pin it to zero
+ZERO_FILL = {angr.options.ZERO_FILL_UNCONSTRAINED_MEMORY, angr.options.ZERO_FILL_UNCONSTRAINED_REGISTERS}
+
+
+def make_state(heap=None, add_options=None):
+    """
+    A bare SimState is enough for everything here and keeps the tests fast: the heap plugin only ever talks to
+    ``state.memory``.
+    """
+    plugins = {"heap": heap} if heap is not None else None
+    return SimState(arch="AMD64", plugins=plugins, add_options=ZERO_FILL | (add_options or set()))
+
+
+def resident_heap_pages(state):
+    """
+    The number of pages the memory plugin has actually materialized inside the heap region.
+
+    This deliberately reaches into ``memory._pages``: the whole point of the lazy mapping is that page objects do
+    not exist, and only the memory object knows that. ``_mapped_end`` alone would not catch a regression that
+    re-introduces eager mapping somewhere else.
+    """
+    page_size = state.memory.page_size
+    lo = state.heap.heap_base // page_size
+    hi = (state.heap.heap_base + state.heap.heap_size) // page_size
+    return sum(1 for pageno, page in state.memory._pages.items() if page is not None and lo <= pageno < hi)
+
+
+def pattern(i, n):
+    """A deterministic n-byte pattern that differs for every i."""
+    return bytes((i * 7 + j * 31 + 0x41) & 0xFF for j in range(n))
+
+
+def store_pattern(state, addr, i, n):
+    state.memory.store(addr, claripy.BVV(pattern(i, n), n * 8))
+
+
+def load_pattern(state, addr, n):
+    return state.solver.eval(state.memory.load(addr, n), cast_to=bytes)
+
+
+class TestHeapLazyMapping(unittest.TestCase):
+    """
+    The heap region is mapped lazily and grown on demand; ``heap_size`` is the maximum extent, not the amount that
+    is mapped.
+    """
+
+    def test_blank_state_maps_only_the_initial_extent(self):
+        # THE regression guard: before the heap mapping became lazy, init_state() mapped all of heap_size (2048
+        # UltraPages, ~17 MB of Python objects) for every single state, including the throwaway states CFG
+        # recovery makes by the thousand.
+        state = make_state()
+        page_size = state.memory.page_size
+
+        assert state.heap._mapped_end == state.heap.heap_base + HEAP_INITIAL_MAPPED_SIZE
+        assert resident_heap_pages(state) == HEAP_INITIAL_MAPPED_SIZE // page_size
+        # a bare SimState maps nothing but the heap, so the heap pages are all the pages there are
+        assert len(state.memory._pages) == HEAP_INITIAL_MAPPED_SIZE // page_size
+
+    def test_project_blank_state_maps_only_the_initial_extent(self):
+        # same guard, but through the path everything else in angr actually uses
+        proj = angr.load_shellcode(b"\x90" * 16, "amd64")
+        state = proj.factory.blank_state()
+
+        assert state.heap._mapped_end == state.heap.heap_base + HEAP_INITIAL_MAPPED_SIZE
+        assert resident_heap_pages(state) == HEAP_INITIAL_MAPPED_SIZE // state.memory.page_size
+
+    def test_growth_is_geometric(self):
+        state = make_state()
+        base = state.heap.heap_base
+        page_size = state.memory.page_size
+
+        extents = []
+        for _ in range(12):
+            state.heap._malloc(0x1000)
+            extents.append(state.heap._mapped_end - base)
+            # the mapped extent must always cover everything that has been handed out
+            assert state.heap._mapped_end >= state.heap.heap_location
+            assert resident_heap_pages(state) == (state.heap._mapped_end - base) // page_size
+
+        # 0x1000 handed out at a time against an 0x2000 initial extent that doubles every time it is overrun
+        expected = [0x2000, 0x2000, 0x4000, 0x4000, 0x8000, 0x8000, 0x8000, 0x8000, 0x10000, 0x10000, 0x10000, 0x10000]
+        assert extents == expected
+        # ...which is exactly HEAP_INITIAL_MAPPED_SIZE scaled by powers of the growth factor
+        for extent in extents:
+            size = HEAP_INITIAL_MAPPED_SIZE
+            while size < extent:
+                size *= HEAP_MAPPING_GROWTH_FACTOR
+            assert size == extent
+
+    def test_growth_uses_a_logarithmic_number_of_mapping_operations(self):
+        # geometric growth is what keeps a program that really does use megabytes of heap from paying a mapping
+        # operation per allocation
+        state = make_state()
+        calls = []
+        real_map_range = state.heap._map_range
+        state.heap._map_range = lambda start, end: (calls.append((start, end)), real_map_range(start, end))[1]
+
+        for _ in range(4096):
+            state.heap._malloc(0x400)
+
+        assert state.heap._mapped_end == state.heap.heap_base + 4096 * 0x400
+        assert 1 <= len(calls) <= 16, f"{len(calls)} mapping operations for 4096 allocations is not logarithmic"
+
+    def test_growth_never_maps_past_heap_size(self):
+        state = make_state(SimHeapBrk(heap_base=0xD0000000, heap_size=0x10000))
+        region_end = state.heap.heap_base + state.heap.heap_size
+
+        state.heap._malloc(0x100000)  # sixteen times the whole region
+        assert state.heap._mapped_end == region_end
+        assert resident_heap_pages(state) == state.heap.heap_size // state.memory.page_size
+
+        state.heap._malloc(0x100000)
+        assert state.heap._mapped_end == region_end
+
+    def test_allocation_landing_exactly_on_a_growth_boundary(self):
+        state = make_state()
+        base = state.heap.heap_base
+
+        # exactly fills the initial extent: nothing has run past the end yet, so nothing grows
+        first = state.heap._malloc(HEAP_INITIAL_MAPPED_SIZE)
+        assert first == base
+        assert state.heap._mapped_end == base + HEAP_INITIAL_MAPPED_SIZE
+
+        # the last byte of the initial extent must be usable
+        store_pattern(state, base + HEAP_INITIAL_MAPPED_SIZE - 16, 1, 16)
+        assert load_pattern(state, base + HEAP_INITIAL_MAPPED_SIZE - 16, 16) == pattern(1, 16)
+
+        # ...and the very next byte is the one that triggers the growth
+        second = state.heap._malloc(16)
+        assert second == base + HEAP_INITIAL_MAPPED_SIZE
+        assert state.heap._mapped_end == base + HEAP_INITIAL_MAPPED_SIZE * HEAP_MAPPING_GROWTH_FACTOR
+        store_pattern(state, second, 2, 16)
+        assert load_pattern(state, second, 16) == pattern(2, 16)
+
+    def test_many_small_allocations_keep_their_data(self):
+        state = make_state()
+        addrs = [state.heap._malloc(1) for _ in range(20000)]
+        for i, addr in enumerate(addrs):
+            state.memory.store(addr, claripy.BVV(i & 0xFF, 8))
+
+        assert state.heap._mapped_end >= state.heap.heap_location
+        bad = [i for i, addr in enumerate(addrs) if state.solver.eval(state.memory.load(addr, 1)) != (i & 0xFF)]
+        assert not bad, f"{len(bad)} of {len(addrs)} one-byte allocations lost their data"
+
+    def test_one_large_allocation_maps_everything_it_needs(self):
+        state = make_state()
+        size = 4 * 1024 * 1024
+        addr = state.heap._malloc(size)
+
+        assert state.heap._mapped_end == state.heap.heap_base + size
+        assert resident_heap_pages(state) == size // state.memory.page_size
+
+        # every page of it has to be usable, not just the first one
+        for off in range(0, size, 0x1000):
+            state.memory.store(addr + off, claripy.BVV(off, 32))
+        bad = [off for off in range(0, size, 0x1000) if state.solver.eval(state.memory.load(addr + off, 4)) != off]
+        assert not bad, f"{len(bad)} pages of the large allocation lost their data"
+
+    def test_symbolic_size_allocation_keeps_its_data(self):
+        state = make_state()
+        state.libc.max_variable_size = 0x8000  # otherwise a symbolic size is clamped to 128 bytes and maps nothing
+        size = claripy.BVS("size", 64)
+        state.solver.add(size.UGE(0x1000), size.ULE(0x5000))
+
+        addr = state.heap._malloc(size)
+        # the symbolic size is maximized, so the whole 0x5000 has to be backed
+        assert state.heap._mapped_end >= addr + 0x5000
+        for i, off in enumerate((0, 0x1000, 0x2FFF, 0x5000 - 64)):
+            store_pattern(state, addr + off, i, 64)
+        for i, off in enumerate((0, 0x1000, 0x2FFF, 0x5000 - 64)):
+            assert load_pattern(state, addr + off, 64) == pattern(i, 64)
+
+    def test_release_does_not_shrink_and_does_not_lose_data(self):
+        # deliberate: release() does not unmap, because unmapping would discard the page contents and a release
+        # followed by an allocation would hand back blank memory instead of the old bytes
+        state = make_state()
+        first = state.heap._malloc(0x4000)
+        store_pattern(state, first + 0x3000, 5, 64)
+        extent_before = state.heap._mapped_end
+
+        state.heap.release(0x4000)
+        assert state.heap.heap_location == state.heap.heap_base
+        assert state.heap._mapped_end == extent_before
+        assert resident_heap_pages(state) == (extent_before - state.heap.heap_base) // state.memory.page_size
+
+        second = state.heap._malloc(0x4000)
+        assert second == first
+        assert load_pattern(state, second + 0x3000, 64) == pattern(5, 64)
+
+        # over-releasing is clamped at the base and still does not unmap anything
+        state.heap.release(0x1000000)
+        assert state.heap.heap_location == state.heap.heap_base
+        assert state.heap._mapped_end == extent_before
+
+    def test_copy_carries_the_mapped_extent(self):
+        state = make_state()
+        state.heap._malloc(0x9000)
+        copy = state.copy()
+        assert copy.heap._mapped_end == state.heap._mapped_end
+        assert resident_heap_pages(copy) == resident_heap_pages(state)
+
+    def test_divergent_copies_do_not_interfere(self):
+        state = make_state()
+        common = []
+        for i in range(64):
+            addr = state.heap._malloc(64)
+            store_pattern(state, addr, i, 64)
+            common.append((addr, i))
+
+        first, second = state.copy(), state.copy()
+        first_allocs, second_allocs = [], []
+        for i in range(64):
+            addr = first.heap._malloc(0x200)
+            store_pattern(first, addr, i + 1000, 64)
+            first_allocs.append((addr, i + 1000))
+
+            addr = second.heap._malloc(0x2000)
+            store_pattern(second, addr, i + 5000, 64)
+            second_allocs.append((addr, i + 5000))
+
+        # the two copies really did grow to different extents, so this is exercising something
+        assert first.heap._mapped_end < second.heap._mapped_end
+        for copy, allocs in ((first, first_allocs), (second, second_allocs)):
+            for addr, i in common + allocs:
+                assert load_pattern(copy, addr, 64) == pattern(i, 64)
+
+    def test_merge_takes_the_minimum_mapped_extent(self):
+        # understating the extent is safe (growing it skips pages that already exist); overstating it could leave
+        # a hole in the middle of the region that nothing would ever map
+        state = make_state()
+        state.heap._malloc(64)
+        first, second = state.copy(), state.copy()
+        first.heap._malloc(0x10000)
+        second.heap._malloc(0x40)
+        assert first.heap._mapped_end > second.heap._mapped_end
+
+        merged, _, _ = first.merge(second)
+        assert merged.heap._mapped_end == min(first.heap._mapped_end, second.heap._mapped_end)
+
+        # NOTE: SimHeapBrk._combine takes max() over `others` *excluding self*, so the merged break drops to the
+        # other state's value instead of the high-water mark. That is a pre-existing angr bug, unrelated to (and
+        # unchanged by) the lazy mapping; assert what actually happens rather than what should.
+        assert merged.heap.heap_location == second.heap.heap_location
+
+        # whatever the extent ended up being, allocating out of the merged state still has to work
+        addr = merged.heap._malloc(0x20000)
+        assert addr >= merged.heap.heap_base
+        store_pattern(merged, addr, 7, 64)
+        store_pattern(merged, addr + 0x20000 - 64, 8, 64)
+        assert load_pattern(merged, addr, 64) == pattern(7, 64)
+        assert load_pattern(merged, addr + 0x20000 - 64, 64) == pattern(8, 64)
+        assert merged.heap._mapped_end >= merged.heap.heap_location
+
+    def test_merge_with_an_unmanaged_extent_stays_unmanaged(self):
+        # bookkeeping-only path, so it is tested directly: if any side does not manage the mapping, neither does
+        # the result
+        first, second = SimHeapBrk(), SimHeapBrk()
+        first._mapped_end = first.heap_base + 0x4000
+        second._mapped_end = None
+        first._combine_mapped_end([second])
+        assert first._mapped_end is None
+
+    def test_unpickling_a_state_from_before_lazy_mapping(self):
+        # states pickled before this change have no _mapped_end at all; back then init_state() mapped the whole
+        # region, so that is the extent they have to come back with
+        heap = SimHeapBrk()
+        old_style = dict(heap.__dict__)
+        del old_style["_mapped_end"]
+
+        restored = SimHeapBrk.__new__(SimHeapBrk)
+        restored.__setstate__(old_style)
+        assert restored._mapped_end == restored.heap_base + restored.heap_size
+
+        # and a state pickled *now* round-trips its actual extent
+        state = make_state()
+        state.heap._malloc(0x5000)
+        extent = state.heap._mapped_end
+        assert pickle.loads(pickle.dumps(state)).heap._mapped_end == extent
+
+    def test_abstract_memory_does_not_manage_the_mapping(self):
+        state = SimState(arch="AMD64", add_options={angr.options.ABSTRACT_MEMORY})
+        assert state.heap._mapped_end is None
+        # ...and _ensure_mapped stays a no-op forever after
+        state.heap._ensure_mapped(state.heap.heap_base + 0x100000)
+        assert state.heap._mapped_end is None
+
+    def test_a_preexisting_mapping_is_left_alone(self):
+        state = make_state()
+        state.memory.map_region(0x50000000, 0x2000, 3)
+        pages_before = len(state.memory._pages)
+
+        # somebody else owns this region, so the heap plugin must not map or grow anything in it
+        heap = SimHeapBrk(heap_base=0x50000000, heap_size=0x100000)
+        state.register_plugin("heap", heap)
+        assert heap._mapped_end is None
+        assert len(state.memory._pages) == pages_before
+
+        heap._ensure_mapped(0x50000000 + 0x80000)
+        assert heap._mapped_end is None
+        assert len(state.memory._pages) == pages_before
+
+    def test_gdb_set_heap_grows_the_extent(self):
+        state = make_state()
+        base = state.heap.heap_base
+        state.gdb.set_heap(os.path.join(gdb_data_location, "heap"), heap_base=base)
+
+        dumped = os.path.getsize(os.path.join(gdb_data_location, "heap"))
+        assert state.heap.heap_location == base + dumped
+        assert state.heap._mapped_end >= base + dumped
+        assert resident_heap_pages(state) == (state.heap._mapped_end - base) // state.memory.page_size
+
+        # the tail of the dump has to have landed in memory, not in a page that was never mapped
+        with open(os.path.join(gdb_data_location, "heap"), "rb") as f:
+            f.seek(dumped - 16)
+            tail = f.read(16)
+        assert load_pattern(state, base + dumped - 16, 16) == tail
+
+    def test_strict_page_access_now_rejects_writes_past_the_allocations(self):
+        # DELIBERATE NARROWING: with the whole region mapped up front, a program could scribble anywhere in the
+        # 8 MB heap under STRICT_PAGE_ACCESS and get away with it. Now only what has been handed out (rounded up
+        # to the mapped extent) is backed, so writing past it faults. Pinned here so the change is visible.
+        state = make_state(add_options={angr.options.STRICT_PAGE_ACCESS})
+        addr = state.heap._malloc(64)
+        beyond = state.heap.heap_base + HEAP_INITIAL_MAPPED_SIZE
+        assert state.heap._mapped_end == beyond
+
+        # inside the allocation, and anywhere inside the mapped extent, is still fine
+        store_pattern(state, addr, 3, 64)
+        assert load_pattern(state, addr, 64) == pattern(3, 64)
+        state.memory.store(beyond - 1, claripy.BVV(0x5A, 8))
+
+        with self.assertRaises(angr.errors.SimSegfaultException):
+            state.memory.store(beyond, claripy.BVV(0x5A, 8))
+
+    def test_writes_past_the_extent_still_work_without_strict_page_access(self):
+        # the narrowing above is confined to STRICT_PAGE_ACCESS; the default memory still auto-maps
+        state = make_state()
+        beyond = state.heap.heap_base + HEAP_INITIAL_MAPPED_SIZE
+        state.memory.store(beyond, claripy.BVV(0x5A, 8))
+        assert state.solver.eval(state.memory.load(beyond, 1)) == 0x5A
+
+
+class TestHeapPTMallocLazyMapping(unittest.TestCase):
+    def test_final_word_page_is_mapped_at_init(self):
+        # this heap keeps the last chunk's usage flag in the final word of the region, which init_state writes, so
+        # that one page is mapped up front even though the extent stops far short of it
+        state = make_state(SimHeapPTMalloc())
+        page_size = state.memory.page_size
+        region_end = state.heap.heap_base + state.heap.heap_size
+        final_pageno = (region_end - 1) // page_size
+
+        assert state.heap._mapped_end == state.heap.heap_base + HEAP_INITIAL_MAPPED_SIZE
+        assert state.memory._pages.get(final_pageno) is not None
+        # the initial extent plus the single final page, and nothing else
+        assert resident_heap_pages(state) == HEAP_INITIAL_MAPPED_SIZE // page_size + 1
+
+    def test_malloc_free_and_reuse_across_a_growth_boundary(self):
+        state = make_state(SimHeapPTMalloc())
+        base = state.heap.heap_base
+
+        first = state.heap.malloc(0x1000)
+        second = state.heap.malloc(0x2000)  # runs past the initial extent
+        third = state.heap.malloc(0x4000)  # ...and past the next one too
+        assert first and second and third
+        assert state.heap._mapped_end > base + HEAP_INITIAL_MAPPED_SIZE
+        assert state.heap._mapped_end >= third + 0x4000
+
+        allocs = {first: 0x1000, second: 0x2000, third: 0x4000}
+        for i, (addr, size) in enumerate(allocs.items()):
+            store_pattern(state, addr, i, 64)
+            store_pattern(state, addr + size - 64, i + 100, 64)
+        chunks_before = [(c.base, c.is_free()) for c in state.heap.chunks()]
+
+        # free the middle chunk and hand the same space back out
+        state.heap.free(second)
+        again = state.heap.malloc(0x2000)
+        assert again == second
+        store_pattern(state, again, 50, 64)
+
+        # the untouched allocations on either side of the freed one kept their contents
+        for i, (addr, size) in enumerate(allocs.items()):
+            if addr == second:
+                continue
+            assert load_pattern(state, addr, 64) == pattern(i, 64)
+            assert load_pattern(state, addr + size - 64, 64) == pattern(i + 100, 64)
+        assert load_pattern(state, again, 64) == pattern(50, 64)
+
+        # the chunk walk has to still terminate and describe the same heap
+        chunks_after = [(c.base, c.is_free()) for c in state.heap.chunks()]
+        assert chunks_before
+        assert [c[0] for c in chunks_after] == [c[0] for c in chunks_before]
+
+    def test_large_allocation_maps_the_whole_region(self):
+        state = make_state(SimHeapPTMalloc())
+        addr = state.heap.malloc(4 * 1024 * 1024)
+        assert addr
+        # the chunk plus the following chunk's metadata runs past half the region, so the extent doubles to cover
+        # all of it
+        assert state.heap._mapped_end == state.heap.heap_base + state.heap.heap_size
+
+        store_pattern(state, addr, 11, 64)
+        store_pattern(state, addr + 4 * 1024 * 1024 - 64, 12, 64)
+        assert load_pattern(state, addr, 64) == pattern(11, 64)
+        assert load_pattern(state, addr + 4 * 1024 * 1024 - 64, 64) == pattern(12, 64)
+
+    def test_many_small_allocations_keep_their_data(self):
+        state = make_state(SimHeapPTMalloc())
+        allocs = []
+        for i in range(200):
+            addr = state.heap.malloc(0x100)
+            assert addr
+            store_pattern(state, addr, i, 64)
+            allocs.append((addr, i))
+
+        assert state.heap._mapped_end > state.heap.heap_base + HEAP_INITIAL_MAPPED_SIZE
+        for addr, i in allocs:
+            assert load_pattern(state, addr, 64) == pattern(i, 64)
+
+
+if __name__ == "__main__":
+    unittest.main()