Extract Rust specific strings from binaries #791

floss/language/rust/extract.py

@@ -8,8 +8,9 @@
 
 import pefile
 import binary2strings as b2s
+
 from floss.results import StaticString, StringEncoding
-from floss.language.utils import find_lea_xrefs, get_struct_string_candidates
+from floss.language.utils import find_lea_xrefs, find_mov_xrefs, find_push_xrefs, get_struct_string_candidates
 
 logger = logging.getLogger(__name__)
 
@@ -68,7 +69,7 @@ def split_strings(static_strings: List[StaticString], address: int) -> None:
             return
 
 
-def extract_rust_strings(sample: str, min_length: int) -> List[StaticString]:
+def extract_rust_strings(sample: pathlib.Path, min_length: int) -> List[StaticString]:
     """
     Extract Rust strings from a sample
     """
@@ -104,9 +105,11 @@ def get_string_blob_strings(pe: pefile.PE, min_length: int) -> Iterable[StaticSt
     static_strings = filter_and_transform_utf8_strings(strings, start_rdata)
 
     struct_string_addrs = map(lambda c: c.address, get_struct_string_candidates(pe))
-    xrefs = find_lea_xrefs(pe)
+    xrefs_lea = find_lea_xrefs(pe)
+    xrefs_push = find_push_xrefs(pe)
+    xrefs_mov = find_mov_xrefs(pe)
 
-    for addr in itertools.chain(struct_string_addrs, xrefs):
+    for addr in itertools.chain(struct_string_addrs, xrefs_lea, xrefs_push, xrefs_mov):
         address = addr - image_base - virtual_address + pointer_to_raw_data
 
         if not (start_rdata <= address < end_rdata):

floss/language/utils.py

@@ -160,6 +160,109 @@ def find_lea_xrefs(pe: pefile.PE) -> Iterable[VA]:
                 yield xref
 
 
+def find_i386_push_xrefs(buf: bytes) -> Iterable[VA]:
+    """
+    scan the given data found at the given base address
+    to find all the 32-bit PUSH instructions,
+    extracting the target virtual address.
+    """
+    push_insn_re = re.compile(
+        rb"""
+        (
+              \x68       # 68 aa aa 00 00       push   0xaaaa
+        )
+        (?P<address>....)
+        """,
+        re.DOTALL + re.VERBOSE,
+    )
+
+    for match in push_insn_re.finditer(buf):
+        address_bytes = match.group("address")
+        address = struct.unpack("<I", address_bytes)[0]
+
+        yield address
+
+
+def find_push_xrefs(pe: pefile.PE) -> Iterable[VA]:
+    """
+    scan the executable sections of the given PE file
+    for PUSH instructions that reference valid memory addresses,
+    yielding the virtual addresses.
+    """
+    low, high = get_image_range(pe)
+
+    for section in pe.sections:
+        if not section.IMAGE_SCN_MEM_EXECUTE:
+            continue
+
+        code = section.get_data()
+
+        if pe.FILE_HEADER.Machine == pefile.MACHINE_TYPE["IMAGE_FILE_MACHINE_AMD64"]:
+            xrefs: Iterable[VA] = []  # no push instructions on amd64
+        elif pe.FILE_HEADER.Machine == pefile.MACHINE_TYPE["IMAGE_FILE_MACHINE_I386"]:
+            xrefs = find_i386_push_xrefs(code)
+        else:
+            raise ValueError("unhandled architecture")
+
+        for xref in xrefs:
+            if low <= xref < high:
+                yield xref
+
+
+def find_i386_mov_xrefs(buf: bytes) -> Iterable[VA]:
+    """
+    scan the given data found at the given base address
+    to find all the 32-bit MOV instructions,
+    extracting the target virtual address.
+    """
+    mov_insn_re = re.compile(
+        rb"""
+        (
+              \xB9       # b9 aa aa 00 00       mov    ecx,0xaaaa
+            | \xBB       # bb aa aa 00 00       mov    ebx,0xaaaa
+            | \xBA       # ba aa aa 00 00       mov    edx,0xaaaa
+            | \xB8       # b8 aa aa 00 00       mov    eax,0xaaaa
+            | \xBE       # be aa aa 00 00       mov    esi,0xaaaa
+            | \xBF       # bf aa aa 00 00       mov    edi,0xaaaa
+        )
+        (?P<address>....)
+        """,
+        re.DOTALL + re.VERBOSE,
+    )
+
+    for match in mov_insn_re.finditer(buf):
+        address_bytes = match.group("address")
+        address = struct.unpack("<I", address_bytes)[0]
+
+        yield address
+
+
+def find_mov_xrefs(pe: pefile.PE) -> Iterable[VA]:
+    """
+    scan the executable sections of the given PE file
+    for MOV instructions that reference valid memory addresses,
+    yielding the virtual addresses.
+    """
+    low, high = get_image_range(pe)
+
+    for section in pe.sections:
+        if not section.IMAGE_SCN_MEM_EXECUTE:
+            continue
+
+        code = section.get_data()
+
+        if pe.FILE_HEADER.Machine == pefile.MACHINE_TYPE["IMAGE_FILE_MACHINE_AMD64"]:
+            xrefs: Iterable[VA] = []  # no mov instructions on amd64
+        elif pe.FILE_HEADER.Machine == pefile.MACHINE_TYPE["IMAGE_FILE_MACHINE_I386"]:
+            xrefs = find_i386_mov_xrefs(code)
+        else:
+            raise ValueError("unhandled architecture")
+
+        for xref in xrefs:
+            if low <= xref < high:
+                yield xref
+
+
 def get_max_section_size(pe: pefile.PE) -> int:
     """get the size of the largest section, as seen on disk."""
     return max(map(lambda s: s.SizeOfRawData, pe.sections))

tests/test_language_extract_rust.py

@@ -1,6 +1,7 @@
 import pathlib
 
 import pytest
+
 from floss.results import StaticString, StringEncoding
 from floss.language.rust.extract import extract_rust_strings
 
@@ -42,8 +43,6 @@ def test_data_string_offset(request, string, offset, encoding, rust_strings):
 @pytest.mark.parametrize(
     "string,offset,encoding,rust_strings",
     [
-        # TODO
-        #  pytest.param("hello world", 0xA03E1, StringEncoding.UTF8, "rust_strings32"),
         # .text:0000000140021155 4C 8D 05 2C DA 09 lea     r8, aAccesserror ; "AccessError"
         # .text:000000014002115C 48 8D 74 24 20    lea     rsi, [rsp+38h+var_18]
         # .text:0000000140021161 41 B9 0B 00 00 00 mov     r9d, 11
@@ -53,3 +52,30 @@ def test_data_string_offset(request, string, offset, encoding, rust_strings):
 )
 def test_lea_mov(request, string, offset, encoding, rust_strings):
     assert StaticString(string=string, offset=offset, encoding=encoding) in request.getfixturevalue(rust_strings)
+
+
+@pytest.mark.parametrize(
+    "string,offset,encoding,rust_strings",
+    [
+        # .text:0041EF8C 68 50 08 4B 00            push    offset unk_4B0850 ; "AccessError"
+        # .text:0041EFB8 68 5B 08 4B 00            push    offset unk_4B085B "already destroyed"
+        pytest.param("AccessError", 0xAE850, StringEncoding.UTF8, "rust_strings32"),
+        pytest.param("already destroyed", 0xAE85B, StringEncoding.UTF8, "rust_strings32"),
+    ],
+)
+def test_push(request, string, offset, encoding, rust_strings):
+    assert StaticString(string=string, offset=offset, encoding=encoding) in request.getfixturevalue(rust_strings)
+
+
+@pytest.mark.parametrize(
+    "string,offset,encoding,rust_strings",
+    [
+        # .text:0046B04A BA 1A 00 00 00                                mov     edx, 1Ah        ; jumptable 0046A19C case 8752
+        # .text:0046B04F B9 A0 C2 4B 00                                mov     ecx, offset unk_4BC2A0
+        # .text:0046B054 E9 93 F8 FF FF                                jmp     loc_46A8EC      ; jumptable 0046A1CA case 0
+        pytest.param("DW_AT_SUN_return_value_ptr", 0xBA2A0, StringEncoding.UTF8, "rust_strings32"),
+        pytest.param("DW_AT_SUN_c_vla", 0xBA2BA, StringEncoding.UTF8, "rust_strings32"),
+    ],
+)
+def test_mov_jmp(request, string, offset, encoding, rust_strings):
+    assert StaticString(string=string, offset=offset, encoding=encoding) in request.getfixturevalue(rust_strings)