mirror of
https://github.com/hakril/PythonForWindows
synced 2026-06-08 14:31:45 +00:00
Fixed pe_parse for non-standard SizeOfOptionalHeader + .NET PE32 mapped in 64b processes
This commit is contained in:
clement rouault
+65
-3
@@ -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
|
||||
|
||||
+54
-26
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user