From af7e64638c4a4c17c8427a4481dff9c821ae5943 Mon Sep 17 00:00:00 2001 From: clement rouault Date: Thu, 29 Jul 2021 20:01:51 +0200 Subject: [PATCH] Fixed pe_parse for non-standard SizeOfOptionalHeader + .NET PE32 mapped in 64b processes --- tests/test_parse_pe.py | 68 +++++++++++++++++++++++++++++++++-- windows/pe_parse.py | 80 ++++++++++++++++++++++++++++-------------- 2 files changed, 119 insertions(+), 29 deletions(-) diff --git a/tests/test_parse_pe.py b/tests/test_parse_pe.py index a2c7ff0..ed89a98 100644 --- a/tests/test_parse_pe.py +++ b/tests/test_parse_pe.py @@ -1,9 +1,11 @@ import pytest -import windows import time -import weakref +import zlib +import ctypes +import windows from .conftest import pop_proc_32, pop_proc_64 +from .pfwtest import * @pytest.fixture(params=[None, pop_proc_32, pop_proc_64], ids=["local-pe", "remote-pe32", "remote-pe64"]) def pe(request): @@ -19,7 +21,6 @@ def pe(request): proc.exit(0) - def test_pe_imports(pe): imports = pe.imports assert imports @@ -38,3 +39,64 @@ def test_pe_sections(pe): assert ".text" in sections_names +# This is a PE build with LIEF & containing a somewhat strange OptionalHeader which is +# not sizeof(IMAGE_OPTIONAL_HEADER32) nor sizeof(IMAGE_OPTIONAL_HEADER64) +# Its sizeof(IMAGE_OPTIONAL_HEADER32)+8 (0xd8) indicating a "hole" between data directory & the sections header +# The building script is build_pe_strange_optionalheader_size.py + +strange_optional_header_size_pe = b"""eJzzjZrAwMzAwMACxP//MzDsYIAABwbC4AMRagYbCHBlYPBhZEYRu8HAxMjNyMnAwMQAwSAgAMUK +II4DhA2UY4VKw2hwQLFAmBBTFSBqQYQAwig4MECyg4ogAGiuMRZxvZLUihIQg5EB7gcGNlQ1CgwM +CXopiSWJQLY2VACsjgNVnQMDwwG9HEMIByQP8wsXhnkPyPTGKBgFo4COoPNp4xtQ2ZDx9T8QiNoY +ODD01qhwZIDkwjNApUEGg4IDQ8B/UQegVOebwxOGFxjg4B9w4JGak5OvUJ5flJOikFaUn6uQqBBQ +WZKRn+eWXxSemZeSX16sUJJaXKIAqjhHwbADGgYIdhOQbWOAXR2ofZEAxDlAXGGAW4yBwT21JLgk +xSMxLyUnFcQPL8osSXXLBHOCUhNToEwGYOpyrUhNZshOLcpLzTE20kvJyaGRJ0cBTgAAMrq98g==""" + + +def test_pe_parsing_strange_optional_header_size(tmp_path, proc32): + pe_path = (tmp_path / "pe_strange_optionalheader_size.exe") + pe_data = zlib.decompress(b64decode(strange_optional_header_size_pe)) + + with pe_path.open("wb") as f: + f.write(pe_data) + + mod = proc32.load_library(str(pe_path)) + # Check imports (This data directory retrieval) works + assert mod.pe.imports + assert set(imp.name for imp in mod.pe.imports["kernel32.dll"]) == {'WriteFile', 'WinExec', 'GetStdHandle', 'ReadFile'} + # Also check that section retrieval works (as its position is based on OptionalHeader Size) + assert set(s.name for s in mod.pe.sections) == {".text", ".data", ".l1"} + +# Make a test from current_process parsing ? +def test_pe_parsing_dotnet_32_process_64(proc64): + # .NET pe32 loadable in 64bit process -> rewrite of the OptionalHeader + mod = proc64.load_library(r"C:\Windows\System32\stordiag.exe") + # It was a PE32 + assert mod.pe.get_NT_HEADER().FileHeader.Machine == gdef.IMAGE_FILE_MACHINE_I386 + # Now Optional Header should be 64b + opt_hdr = mod.pe.get_OptionalHeader() + assert mod.pe.get_NT_HEADER().FileHeader.SizeOfOptionalHeader == ctypes.sizeof(gdef.IMAGE_OPTIONAL_HEADER64) + assert mod.pe.get_OptionalHeader().Magic == gdef.IMAGE_NT_OPTIONAL_HDR64_MAGIC + # Check imports (This data directory retrieval) works + assert mod.pe.imports + assert mod.pe.imports["mscoree.dll"][0].name == "_CorExeMain" + # Also check that section retrieval works (as its position is based on OptionalHeader Size) + assert mod.pe.sections + assert ".text" in set(s.name for s in mod.pe.sections) + +def test_pe_parsing_dotnet_32_current_process_64(proc64): + # .NET pe32 loadable in 64bit process -> rewrite of the OptionalHeader + # So we injecte python code in a the remote proc64 to test the parsing from itself + + PIPE_NAME = "PFW_TEST_Pipe" + rcode = r"""import sys; import windows; import windows.pipe; windows.pipe.send_object("{pipe}", )""" + + mod = proc64.load_library(r"C:\Windows\System32\stordiag.exe") + assert proc64.peb.modules[-1].name == "stordiag.exe" + proc64.execute_python("import sys; import windows; import windows.pipe") + with windows.pipe.create(PIPE_NAME) as np: + proc64.execute_python("""pemod = [x for x in windows.current_process.peb.modules if x.name == 'stordiag.exe'][0].pe""") + rcode = """windows.pipe.send_object("{pipe}", (list(pemod.imports), [sec.name for sec in pemod.sections]))""" + proc64.execute_python(rcode.format(pipe=PIPE_NAME)) + imported_dlls, sections_names = np.recv() + assert imported_dlls == ['mscoree.dll'] + assert ".text" in sections_names diff --git a/windows/pe_parse.py b/windows/pe_parse.py index 15d299d..b677626 100644 --- a/windows/pe_parse.py +++ b/windows/pe_parse.py @@ -80,6 +80,14 @@ class THUNK_DATA(ctypes.Union): ("AddressOfData", PVOID) ] +# Special case for .NET PE32 rewrite as 64b +# We may have a PE in a 64b process with a 32b IAT +class THUNK_DATA_32(ctypes.Union): + _fields_ = [ + ("Ordinal", DWORD), + ("AddressOfData", DWORD) + ] + class IMPORT_BY_NAME(ctypes.Structure): _fields_ = [ ("Hint", WORD), @@ -182,7 +190,6 @@ class IATEntry(ctypes.Structure): def on_destroy(self, *args): # We cannot know if the hook was enabled here.. print("DESTROY: {0} -> ".format(args, self.enabled)) - # import pdb;pdb.set_trace() # print(args[0]()) def remove_hook(self): @@ -200,11 +207,16 @@ class IATEntry(ctypes.Structure): class IMAGE_IMPORT_DESCRIPTOR(IMAGE_IMPORT_DESCRIPTOR): # TODO: use explicite name winstructs.IMAGE_IMPORT_DESCRIPTOR - def get_INT(self): + def get_INT(self, pe): + THUNK_DATA_TYPE = THUNK_DATA if not self.OriginalFirstThunk: return None + # We may have 32bits PE mapped in 32bits process (thanks to .NET PE) + if self.target is None and pe.bitness != windows.current_process.bitness: + assert windows.current_process.bitness == 64 and pe.bitness == 32, "Mapped 64b PE in current process 32b not handled" + THUNK_DATA_TYPE = THUNK_DATA_32 int_addr = self.OriginalFirstThunk + self.baseaddr - int_entry = self.transformers.create_structure_at(THUNK_DATA, int_addr) + int_entry = self.transformers.create_structure_at(THUNK_DATA_TYPE, int_addr) res = [] while int_entry.Ordinal: if int_entry.Ordinal & self.IMAGE_ORDINAL_FLAG: @@ -212,23 +224,24 @@ class IMAGE_IMPORT_DESCRIPTOR(IMAGE_IMPORT_DESCRIPTOR): # TODO: use explicite na else: import_by_name = self.transformers.create_structure_at(IMPORT_BY_NAME, self.baseaddr + int_entry.AddressOfData) name_address = self.baseaddr + int_entry.AddressOfData + type(import_by_name).Name.offset - if self.target is None: - name = get_string(self.target, name_address) - else: - name = get_string(self.target, name_address) + name = get_string(self.target, name_address) res.append((import_by_name.Hint, name)) int_addr += ctypes.sizeof(type(int_entry)) - int_entry = self.transformers.create_structure_at(THUNK_DATA, int_addr) + int_entry = self.transformers.create_structure_at(THUNK_DATA_TYPE, int_addr) return res - def get_IAT(self): + def get_IAT(self, pe): + THUNK_DATA_TYPE = THUNK_DATA + if self.target is None and pe.bitness != windows.current_process.bitness: + assert windows.current_process.bitness == 64 and pe.bitness == 32, "Mapped 64b PE in current process 32b not handled" + THUNK_DATA_TYPE = THUNK_DATA_32 iat_addr = self.FirstThunk + self.baseaddr - iat_entry = self.transformers.create_structure_at(THUNK_DATA, iat_addr) + iat_entry = self.transformers.create_structure_at(THUNK_DATA_TYPE, iat_addr) res = [] while iat_entry.Ordinal: res.append(IATEntry.create(iat_addr, -1, "??", self.target, self.transformers)) iat_addr += ctypes.sizeof(type(iat_entry)) - iat_entry = self.transformers.create_structure_at(THUNK_DATA, iat_addr) + iat_entry = self.transformers.create_structure_at(THUNK_DATA_TYPE, iat_addr) return res @classmethod @@ -314,21 +327,37 @@ class PEFile(object): return self.transformers.create_structure_at(IMAGE_NT_HEADERS32, self.baseaddr + offset) return self.transformers.create_structure_at(IMAGE_NT_HEADERS64, self.baseaddr + offset) + + STANDARD_OPTIONAL_HEADER_TYPE_PER_MAGIC = { + IMAGE_NT_OPTIONAL_HDR32_MAGIC: IMAGE_OPTIONAL_HEADER32, + IMAGE_NT_OPTIONAL_HDR64_MAGIC: IMAGE_OPTIONAL_HEADER64, + } + + STANDARD_OPTIONAL_HEADER_SIZE_PER_MAGIC = ( + (IMAGE_NT_OPTIONAL_HDR32_MAGIC, ctypes.sizeof(IMAGE_OPTIONAL_HEADER32)), + (IMAGE_NT_OPTIONAL_HDR64_MAGIC, ctypes.sizeof(IMAGE_OPTIONAL_HEADER64)), + ) + def get_OptionalHeader(self): - return self.get_NT_HEADER().OptionalHeader + # We can have a 32bits PE with a 64 bits OptionalHeader + # Ex : PE32 .NET that allow to be loaded in 64b process + # See: https://github.com/dotnet/runtime/blob/8bbe33819464216becffb7cf8b7ea8dd3bab5836/src/coreclr/src/vm/peimagelayout.cpp#L599 + # In this case the OptionalHeader is transformed in 64bits & OptionalHeader.Magic is changed accordingly + # So we cannot just rely on get_NT_HEADER() to give us the correct OptionalHeader type. some re-check are required + default_opth = self.get_NT_HEADER().OptionalHeader + # Cannot juste compare types with type(default_opth) as it may be a remoteType + current_opth_infos = (default_opth.Magic, ctypes.sizeof(default_opth)) + if current_opth_infos in self.STANDARD_OPTIONAL_HEADER_SIZE_PER_MAGIC: + # The default OptionalHeader structure match what we expect based on the magic (most of the cases) + return default_opth + # Mismatch -> PE32 remapped as 64b (with OptionalHeader rewrite) + # Remap the correct OptionalHeader + opt_header_real_type = self.STANDARD_OPTIONAL_HEADER_TYPE_PER_MAGIC[default_opth.Magic] + opt_header_addr = default_opth._base_addr if self.target else ctypes.addressof(default_opth) + return self.transformers.create_structure_at(opt_header_real_type, opt_header_addr) def get_DataDirectory(self): - # This won't work if we load a PE32 in a 64bit process - # PE32 .NET... - # return self.get_OptionalHeader().DataDirectory - DataDirectory_type = IMAGE_DATA_DIRECTORY * IMAGE_NUMBEROF_DIRECTORY_ENTRIES - SizeOfOptionalHeader = self.get_NT_HEADER().FileHeader.SizeOfOptionalHeader - if self.target is None: - opt_header_addr = ctypes.addressof(self.get_NT_HEADER().OptionalHeader) - else: - opt_header_addr = self.get_NT_HEADER().OptionalHeader._base_addr - DataDirectory_addr = opt_header_addr + SizeOfOptionalHeader - ctypes.sizeof(DataDirectory_type) - return self.transformers.create_structure_at(DataDirectory_type, DataDirectory_addr) + return self.get_OptionalHeader().DataDirectory def get_IMPORT_DESCRIPTORS(self): @@ -380,7 +409,6 @@ class PEFile(object): export_end = export_start + export_datadir.Size if exp_dir is None: return res - # import pdb;pdb.set_trace() raw_exports = exp_dir.get_exports() for id, rva_addr, rva_name in raw_exports: if export_start <= rva_addr < export_end: @@ -413,8 +441,8 @@ class PEFile(object): :type: {:class:`str` : [:class:`IATEntry`]}""" res = {} for import_descriptor in self.get_IMPORT_DESCRIPTORS(): - INT = import_descriptor.get_INT() - IAT = import_descriptor.get_IAT() + INT = import_descriptor.get_INT(self) + IAT = import_descriptor.get_IAT(self) if INT is not None: for iat_entry, (ord, name) in zip(IAT, INT): # str(name.decode()) -> python2 and python3 compatible for str result