diff --git a/cle/backends/cgc/backedcgc.py b/cle/backends/cgc/backedcgc.py index d5f0fba50..b8c60e34c 100644 --- a/cle/backends/cgc/backedcgc.py +++ b/cle/backends/cgc/backedcgc.py @@ -1,5 +1,6 @@ from __future__ import annotations +from cle.address_translator import AT from cle.backends.backend import register_backend from cle.backends.region import Segment @@ -7,11 +8,17 @@ 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) diff --git a/cle/memory.py b/cle/memory.py index e2de3f139..c2e20dac5 100644 --- a/cle/memory.py +++ b/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 diff --git a/tests/test_backedcgc.py b/tests/test_backedcgc.py new file mode 100644 index 000000000..ed4e46b10 --- /dev/null +++ b/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() == {} diff --git a/tests/test_clemory.py b/tests/test_clemory.py index 432dfc16b..235462fbe 100644 --- a/tests/test_clemory.py +++ b/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():