Fix Clemory backer removal and the BackedCGC backend

cle/backends/cgc/backedcgc.py

@@ -1,17 +1,24 @@
 from __future__ import annotations
 
+from cle.address_translator import AT
 from cle.backends.backend import register_backend
 from cle.backends.region import Segment
 
 from .cgc import CGC
 
 
 class FakeSegment(Segment):
+    """
+    A segment covering memory that the memory backer supplies but the file does not map, such as the stack and heap of
+    a process dump. It is readable and writable by Region's defaults, and never executable.
+    """
+
     def __init__(self, start, size):
         super().__init__(0, start, 0, size)
-        self.is_readable = True
-        self.is_writable = True
-        self.is_executable = False
+
+    @property
+    def is_executable(self) -> bool:
+        return False
 
 
 class BackedCGC(CGC):
@@ -38,13 +45,16 @@ def __init__(
                                         actual memory content as data.
         :param register_backer:         A dict of all register contents. EIP will be used as the entry point of this
                                         executable.
+        :param writes_backer:           A list of the sizes of the writes the process made to stdout.
         :param permissions_map:         A dict of memory region to permission flags
         :param current_allocation_base: An integer representing the current address of the top of the CGC heap.
         """
         super().__init__(*args, **kwargs)
 
-        self.memory_backer = memory_backer
-        self.register_backer = register_backer
+        # Both backers are optional mappings. Normalizing them here keeps the rest of the backend, and
+        # thread_registers(), from having to special-case a missing one.
+        self.memory_backer = memory_backer if memory_backer is not None else {}
+        self.register_backer = register_backer if register_backer is not None else {}
         self.writes_backer = writes_backer
         self.permissions_map = permissions_map
         self.current_allocation_base = current_allocation_base
@@ -56,25 +66,31 @@ def __init__(
         else:
             raise ValueError("Couldn't find executable segment?")
 
-        for start, _ in self.memory._backers:
-            if start != exec_seg_addr:
+        # Segment vaddrs and the backers the caller passes in are linked addresses, but this object's memory is keyed
+        # relative to its base, so every address has to be translated before it reaches self.memory.
+        exec_seg_rva = AT.from_lva(exec_seg_addr, self).to_rva()
+
+        # The dump replaces everything the file mapped except the code itself. remove_backer() mutates the backer
+        # list, so walk a copy of it.
+        for start, _ in list(self.memory._backers):
+            if start != exec_seg_rva:
                 self.memory.remove_backer(start)
 
         for start, data in sorted(self.memory_backer.items()):
-            existing_seg = self.find_segment_containing(start)
-            if existing_seg is None:  # this is the text or data segment
-                new_seg = FakeSegment(start, len(data))
-                self.segments.append(new_seg)
+            if self.find_segment_containing(start) is None:
+                # A region the process had mapped but the file does not describe, such as the stack or the heap.
+                self.segments.append(FakeSegment(start, len(data)))
 
             if start == exec_seg_addr:
                 continue
 
-            if start in self.memory:
-                raise ValueError("IF THIS GETS THROWN I'M GONNA JUMP OUT THE WINDOW")
+            relative_start = AT.from_lva(start, self).to_rva()
+            if relative_start in self.memory:
+                raise ValueError(f"Memory backer at {start:#x} overlaps memory already loaded from the file")
 
-            self.memory.add_backer(start, data)
+            self.memory.add_backer(relative_start, data)
 
-        if self.register_backer is not None and "eip" in self.register_backer:
+        if "eip" in self.register_backer:
             self._entry = self.register_backer["eip"]
 
     @staticmethod
@@ -86,7 +102,8 @@ def threads(self):
         return [0]
 
     def thread_registers(self, thread=None):
-        return self.register_backer.items()
+        # Backend.thread_registers is documented to return a mapping, and angr's SimOS iterates it with .items().
+        return self.register_backer
 
 
 register_backend("backedcgc", BackedCGC)

cle/memory.py

@@ -277,7 +277,9 @@ def __repr__(self) -> str:
         return f"<{self.__class__.__name__} [{hex(self.min_addr)}:{hex(self.max_addr)}]>"
 
     def remove_backer(self, start):
-        backer_idx = bisect.bisect(self._backers, start, key=lambda x: x[0])
+        # bisect_left, not bisect_right: the backer starting exactly at `start` is the one to remove, and bisect_right
+        # would land on the backer after it.
+        backer_idx = bisect.bisect_left(self._backers, start, key=lambda x: x[0])
 
         if len(self._backers) <= backer_idx or self._backers[backer_idx][0] != start:
             raise ValueError("Can't find backer to remove")
@@ -482,6 +484,13 @@ def _update_min_max(self):
         Update the three properties of Clemory: consecutive, min_addr, and max_addr.
         """
 
+        if not self._backers:
+            # Removing the last backer leaves the same empty memory a freshly constructed Clemory has.
+            self.consecutive = True
+            self.min_addr = 0
+            self.max_addr = 0
+            return
+
         is_consecutive = True
         next_start = None
         min_addr, max_addr = None, None

tests/test_backedcgc.py

@@ -0,0 +1,62 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+import cle
+
+TEST_BASE = Path(__file__).resolve().parents[2] / "binaries" / "tests"
+
+# A CGC binary with four loaded segments, linked at 0x60006c4 and with its first executable segment at 0x8048000. The
+# base and that segment sitting at different addresses is what makes the object-relative addressing visible.
+BINARY = TEST_BASE / "i386" / "patchrex" / "memory_scanner"
+EXEC_SEG_ADDR = 0x8048000
+DROPPED_SEG_ADDRS = (0x60006C4, 0x804F348, 0x8063000)
+
+# The instruction pointer the dump was taken at. The file's own entry point is 0x6000703, so a loaded object only
+# reports this one if the register backer reached it.
+DUMP_EIP = 0x8048100
+FILE_ENTRY = 0x6000703
+
+# A region a CGC process dump carries that the file itself does not map.
+STACK_ADDR = 0xBAAAB000
+STACK_DATA = b"\x11" * 0x1000
+
+
+def load(**main_opts) -> cle.Loader:
+    return cle.Loader(BINARY, auto_load_libs=False, main_opts={"backend": "backedcgc", **main_opts})
+
+
+def test_backed_cgc_keeps_the_code_and_maps_the_dump():
+    loader = load(
+        memory_backer={STACK_ADDR: STACK_DATA},
+        register_backer={"eip": DUMP_EIP, "esp": STACK_ADDR},
+    )
+    obj = loader.main_object
+    assert isinstance(obj, cle.BackedCGC)
+
+    # The executable segment loaded from the file survives; it is the one backer the dump does not replace.
+    assert loader.memory.load(EXEC_SEG_ADDR, 4) == b"\x7fCGC"
+
+    # Every other segment the file mapped is dropped, since the dump is the authority on non-code memory. Dropping
+    # them has to survive the backer list shrinking as it goes.
+    for addr in DROPPED_SEG_ADDRS:
+        with pytest.raises(KeyError):
+            loader.memory.load(addr, 4)
+
+    # The dump lands where the dump says it lives, not at that address offset by the object's base.
+    assert loader.memory.load(STACK_ADDR, 4) == STACK_DATA[:4]
+
+    assert obj.entry == DUMP_EIP
+    assert obj.thread_registers() == {"eip": DUMP_EIP, "esp": STACK_ADDR}
+
+
+def test_backed_cgc_loads_without_either_backer():
+    # Both backers are optional, so the backend has to load with neither of them and fall back to the file.
+    loader = load()
+    obj = loader.main_object
+
+    assert loader.memory.load(EXEC_SEG_ADDR, 4) == b"\x7fCGC"
+    assert obj.entry == FILE_ENTRY
+    assert obj.thread_registers() == {}

tests/test_clemory.py

@@ -4,7 +4,9 @@
 import timeit
 import unittest
 
+import archinfo
 import cffi
+import pytest
 
 import cle
 
@@ -118,6 +120,55 @@ def test_clemory_contains():
     assert clemory.consecutive is True
 
 
+def test_remove_backer():
+    clemory = cle.Clemory(archinfo.ArchAMD64(), root=True)
+    clemory.add_backer(0, b"A")
+    clemory.add_backer(10, b"BB")
+    clemory.add_backer(20, b"CCC")
+
+    # The search used to bisect right, landing one past the backer being removed, so no removal ever found its target.
+    clemory.remove_backer(0)
+    assert [start for start, _ in clemory.backers()] == [10, 20]
+    clemory.remove_backer(20)
+    assert list(clemory.backers()) == [(10, bytearray(b"BB"))]
+    assert clemory.min_addr == 10
+    assert clemory.max_addr == 12
+
+    # Only the address a backer starts at identifies it.
+    with pytest.raises(ValueError):
+        clemory.remove_backer(11)
+
+    # Emptying a clemory leaves it in the state a freshly constructed one is in.
+    clemory.remove_backer(10)
+    assert not list(clemory.backers())
+    assert clemory.min_addr == 0
+    assert clemory.max_addr == 0
+    assert clemory.consecutive is True
+    assert 10 not in clemory
+
+
+def test_split_backer():
+    clemory = cle.Clemory(archinfo.ArchAMD64(), root=True)
+    clemory.add_backer(0, b"ABCDEFGH")
+
+    # Splitting removes the backer and re-adds the two halves, so it only works once removal does.
+    clemory.split_backer(4)
+
+    assert list(clemory.backers()) == [(0, bytearray(b"ABCD")), (4, bytearray(b"EFGH"))]
+    assert clemory.load(0, 8) == b"ABCDEFGH"
+
+
+def test_add_backer_overwrite():
+    clemory = cle.Clemory(archinfo.ArchAMD64(), root=True)
+    clemory.add_backer(0, b"ABCDEFGH")
+
+    # Overwriting splits the backer around the new data and drops what the new data replaces, both of which need
+    # removal to work.
+    clemory.add_backer(2, b"xy", overwrite=True)
+
+    assert clemory.load(0, 8) == b"ABxyEFGH"
+
+
 def main():
     g = globals()
     for func_name, func in g.items():