From 6a74c83670d8226629110a48d47806a108b1c336 Mon Sep 17 00:00:00 2001 From: hakril Date: Fri, 7 Feb 2020 23:06:26 +0100 Subject: [PATCH] test_process pass for python3 + added some pe_parse test --- tests/test_parse_pe.py | 40 ++++ windows/pe_parse.py | 395 ++++----------------------------- windows/winobject/exception.py | 2 +- windows/winobject/process.py | 25 ++- 4 files changed, 99 insertions(+), 363 deletions(-) create mode 100644 tests/test_parse_pe.py diff --git a/tests/test_parse_pe.py b/tests/test_parse_pe.py new file mode 100644 index 0000000..a2c7ff0 --- /dev/null +++ b/tests/test_parse_pe.py @@ -0,0 +1,40 @@ +import pytest +import windows +import time +import weakref + +from .conftest import pop_proc_32, pop_proc_64 + +@pytest.fixture(params=[None, pop_proc_32, pop_proc_64], ids=["local-pe", "remote-pe32", "remote-pe64"]) +def pe(request): + # Pe will be kernelbase.dll + if request.param is None: + yield windows.current_process.peb.modules[2].pe + return + + pop_proc = request.param + proc = pop_proc() + time.sleep(0.01) + yield proc.peb.modules[2].pe + proc.exit(0) + + + +def test_pe_imports(pe): + imports = pe.imports + assert imports + assert "ntdll.dll" in imports + assert "NtCreateFile" in [x.name for x in imports["ntdll.dll"]] + +def test_pe_exports(pe): + exports = pe.exports + assert "CreateFileA" in pe.exports + +def test_pe_exports_name(pe): + assert "kernel" in pe.export_name.lower() + +def test_pe_sections(pe): + sections_names = [s.name for s in pe.sections] + assert ".text" in sections_names + + diff --git a/windows/pe_parse.py b/windows/pe_parse.py index ac8d62a..15d299d 100644 --- a/windows/pe_parse.py +++ b/windows/pe_parse.py @@ -46,345 +46,6 @@ def get_pe_bitness(baseaddr, target): raise ValueError("Unknow PE target machine <0x{0:x}>".format(machine)) -# def GetPEFile(baseaddr, target=None, force_bitness=None): -# """Returns a :class:`PEFile` to explore a PE loaded at `baseaddr` in process `target`. -# -# :rtype: :class:`PEFile` -# -# .. note:: -# -# If target is ``None`` it refers to the curent process -# """ -# proc_bitness = windows.current_process.bitness -# -# if force_bitness is None: -# targetedbitness = get_pe_bitness(baseaddr, target) -# else: -# targetedbitness = force_bitness -# -# transformers = get_structure_transformer_for_target(target, targetedbitness) -# ctypes_structure_transformer, create_structure_at = transformers -# -# if targetedbitness == 32: -# IMAGE_ORDINAL_FLAG = IMAGE_ORDINAL_FLAG32 -# else: -# IMAGE_ORDINAL_FLAG = IMAGE_ORDINAL_FLAG64 -# -# def get_string(addr): -# if target is None: -# return ctypes.c_char_p(addr).value -# return target.read_string(addr) -# -# class RVA(DWORD): -# @property -# def addr(self): -# return baseaddr + self.value -# -# def __repr__(self): -# return "".format(self.value, hex(self.addr)) -# -# class StringRVa(RVA): -# if target is None: -# @property -# def str(self): -# return get_string(self.addr).decode() -# else: -# @property -# def str(self): -# return get_string(self.addr).decode() -# -# def __repr__(self): -# return "".format(self.value, self.str) -# -# def __int__(self): -# return self.value -# -# class IMPORT_BY_NAME(ctypes.Structure): -# _fields_ = [ -# ("Hint", WORD), -# ("Name", BYTE) -# ] -# -# class THUNK_DATA(ctypes.Union): -# _fields_ = [ -# ("Ordinal", PVOID), -# ("AddressOfData", PVOID) -# ] -# -# class IATEntry(ctypes.Structure): -# """Represent an entry in the IAT of a module -# Can be used to get resolved value and setup hook -# """ -# _fields_ = [ -# ("value", PVOID)] -# -# -# -# @classmethod -# def create(cls, addr, ord, name): -# self = create_structure_at(cls, addr) -# self.addr = addr -# self.ord = ord -# self.name = name -# self.hook = None -# self.nonhookvalue = self.value -# return self -# -# def __repr__(self): -# return '<{0} "{1}" ordinal {2}>'.format(self.__class__.__name__, self.name, self.ord) -# -# def set_hook(self, callback, types=None): -# """Setup a hook on the entry and return it. -# You MUST keep a reference to the hook while the hook is enabled. -# -# :param callback: the hook -# -# .. note:: -# -# see :ref:`hook_protocol` -# -# :rtype: :class:`windows.hooks.IATHook` -# -# .. warning:: -# -# This works only for PEFile with the current process as target. -# """ -# if target is not None: -# raise NotImplementedError("Setting hook in remote process (use python code injection)") -# -# hook = hooks.IATHook(self, callback, types) -# self.hook = hook -# hook.enable() -# return hook -# -# def remove_hook(self): -# """Remove the hook on the entry""" -# if self.hook is None: -# return False -# self.hook.disable() -# self.hook = None -# return True -# -# class PEFile(object): -# """Represent a PE loaded in a process (current or remote)""" -# #def __del__(self): -# # print("YOLODEL") -# # import pprint -# # pprint.pprint(vars(self)) -# -# def __init__(self): -# self.baseaddr = baseaddr -# self.bitness = targetedbitness -# -# def get_DOS_HEADER(self): -# return create_structure_at(IMAGE_DOS_HEADER, baseaddr) -# -# def get_NT_HEADER(self): -# return self.get_DOS_HEADER().get_NT_HEADER() -# -# def get_OptionalHeader(self): -# return self.get_NT_HEADER().OptionalHeader -# -# 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 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 create_structure_at(DataDirectory_type, DataDirectory_addr) -# -# -# def get_IMPORT_DESCRIPTORS(self): -# import_datadir = self.get_DataDirectory()[IMAGE_DIRECTORY_ENTRY_IMPORT] -# if import_datadir.VirtualAddress == 0: -# return [] -# import_descriptor_addr = RVA(import_datadir.VirtualAddress).addr -# current_import_descriptor = create_structure_at(self.IMAGE_IMPORT_DESCRIPTOR, import_descriptor_addr) -# res = [] -# while current_import_descriptor.FirstThunk: -# res.append(current_import_descriptor) -# import_descriptor_addr += ctypes.sizeof(self.IMAGE_IMPORT_DESCRIPTOR) -# current_import_descriptor = create_structure_at(self.IMAGE_IMPORT_DESCRIPTOR, import_descriptor_addr) -# return res -# -# def get_EXPORT_DIRECTORY(self): -# export_directory_rva = self.get_DataDirectory()[IMAGE_DIRECTORY_ENTRY_EXPORT].VirtualAddress -# if export_directory_rva == 0: -# return None -# export_directory_addr = baseaddr + export_directory_rva -# return create_structure_at(self._IMAGE_EXPORT_DIRECTORY, export_directory_addr) -# -# class PESection((IMAGE_SECTION_HEADER)): -# if target is None: -# @property -# def name(self): -# return get_string(ctypes.addressof(self.Name))[:8] -# else: -# @property -# def name(self): -# return get_string(self._base_addr)[:8] -# -# @property -# def start(self): -# return baseaddr + self.VirtualAddress -# -# @property -# def size(self): -# return self.VirtualSize -# -# def __repr__(self): -# return "".format(self.name) -# -# @utils.fixedpropety -# def sections(self): -# nt_header = self.get_NT_HEADER() -# nb_section = nt_header.FileHeader.NumberOfSections -# SizeOfOptionalHeader = self.get_NT_HEADER().FileHeader.SizeOfOptionalHeader -# if target is None: -# opt_header_addr = ctypes.addressof(self.get_NT_HEADER().OptionalHeader) -# else: -# opt_header_addr = self.get_NT_HEADER().OptionalHeader._base_addr -# base_section = opt_header_addr + SizeOfOptionalHeader -# sections_array = create_structure_at((self.PESection * nb_section), base_section) -# return list(sections_array) -# -# @utils.fixedpropety -# def exports(self): -# """The exports of the PE in a dict. Keys are ordinal (:class:`int`) and name (:class:`str`). -# The values are the addresses of the exports. -# -# :type: {(:class:`int` or :class:`str`) : :class:`int`}""" -# res = {} -# exp_dir = self.get_EXPORT_DIRECTORY() -# if exp_dir is None: -# return res -# raw_exports = exp_dir.get_exports() -# for id, rva_addr, rva_name in raw_exports: -# res[id] = rva_addr.addr -# if rva_name is not None: -# res[rva_name.str] = rva_addr.addr -# return res -# -# @utils.fixedpropety -# def export_name(self): -# """The Name attribute of the ``EXPORT_DIRECTORY``""" -# try: -# return self.get_EXPORT_DIRECTORY().Name.str -# except AttributeError: -# return None -# -# # TODO: get imports by parsing other modules exports if no INT -# @utils.fixedpropety -# def imports(self): -# """The imports of the PE in a dict. -# Keys are the names of DLL to import from and values are :class:`list` -# of :class:`IATEntry` -# -# :type: {:class:`str` : [:class:`IATEntry`]}""" -# res = {} -# for import_descriptor in self.get_IMPORT_DESCRIPTORS(): -# INT = import_descriptor.get_INT() -# IAT = import_descriptor.get_IAT() -# if INT is not None: -# for iat_entry, (ord, name) in zip(IAT, INT): -# # str(name.decode()) -> python2 and python3 compatible for str result -# iat_entry.ord = ord -# iat_entry.name = str(name.decode()) if name else "" -# res.setdefault(import_descriptor.Name.str.lower(), []).extend(IAT) -# return res -# -# # Will be usable as `self.IMAGE_IMPORT_DESCRIPTOR` -# class IMAGE_IMPORT_DESCRIPTOR(ctypes.Structure): -# _fields_ = transform_ctypes_fields(IMAGE_IMPORT_DESCRIPTOR, {"Name": StringRVa, "OriginalFirstThunk": RVA, "FirstThunk": RVA}) -# -# def get_INT(self): -# if not self.OriginalFirstThunk.value: -# return None -# int_addr = self.OriginalFirstThunk.addr -# int_entry = create_structure_at(THUNK_DATA, int_addr) -# res = [] -# while int_entry.Ordinal: -# if int_entry.Ordinal & IMAGE_ORDINAL_FLAG: -# res += [(int_entry.Ordinal & 0x7fffffff, None)] -# else: -# import_by_name = create_structure_at(IMPORT_BY_NAME, baseaddr + int_entry.AddressOfData) -# name_address = baseaddr + int_entry.AddressOfData + type(import_by_name).Name.offset -# if target is None: -# name = get_string(name_address) -# else: -# name = get_string(name_address).decode() -# res.append((import_by_name.Hint, name)) -# int_addr += ctypes.sizeof(type(int_entry)) -# int_entry = create_structure_at(THUNK_DATA, int_addr) -# return res -# -# def get_IAT(self): -# iat_addr = self.FirstThunk.addr -# iat_entry = create_structure_at(THUNK_DATA, iat_addr) -# res = [] -# while iat_entry.Ordinal: -# res.append(IATEntry.create(iat_addr, -1, "??")) -# iat_addr += ctypes.sizeof(type(iat_entry)) -# iat_entry = create_structure_at(THUNK_DATA, iat_addr) -# return res -# -# # Will be usable as `self._IMAGE_EXPORT_DIRECTORY` -# class _IMAGE_EXPORT_DIRECTORY(ctypes.Structure): -# _fields_ = transform_ctypes_fields(IMAGE_EXPORT_DIRECTORY, {"Name": StringRVa, "AddressOfFunctions": RVA, "AddressOfNames": RVA, "AddressOfNameOrdinals": RVA}) -# -# def get_exports(self): -# NameOrdinals = create_structure_at((WORD * self.NumberOfNames), self.AddressOfNameOrdinals.addr) -# NameOrdinals = list(NameOrdinals) -# Functions = create_structure_at((RVA * self.NumberOfFunctions), self.AddressOfFunctions.addr) -# Names = create_structure_at((StringRVa * self.NumberOfNames), self.AddressOfNames.addr) -# res = [] -# for nb, func in enumerate(Functions): -# if nb in NameOrdinals: -# name = Names[NameOrdinals.index(nb)] -# else: -# name = None -# res.append((nb, func, name)) -# return res -# -# current_pe = PEFile() -# -# class IMAGE_DOS_HEADER(ctypes.Structure): -# _fields_ = [ -# ("e_magic", CHAR * 2), -# ("e_cblp", WORD), -# ("e_cp", WORD), -# ("e_crlc", WORD), -# ("e_cparhdr", WORD), -# ("e_minalloc", WORD), -# ("e_maxalloc", WORD), -# ("e_ss", WORD), -# ("e_sp", WORD), -# ("e_csum", WORD), -# ("e_ip", WORD), -# ("e_cs", WORD), -# ("e_lfarlc", WORD), -# ("e_ovno", WORD), -# ("e_res", WORD * 4), -# ("e_oemid", WORD), -# ("e_oeminfo", WORD), -# ("e_res2", WORD * 10), -# ("e_lfanew", DWORD), -# ] -# -# def get_NT_HEADER(self): -# if targetedbitness == 32: -# return create_structure_at(IMAGE_NT_HEADERS32, baseaddr + self.e_lfanew) -# return create_structure_at(IMAGE_NT_HEADERS64, baseaddr + self.e_lfanew) -# return current_pe - - - - ## == PEPARSE V2 == @@ -398,7 +59,7 @@ def GetPEFile(baseaddr, target=None, force_bitness=None): .. note:: - If target is ``None`` it refers to the curent process + If target is ``None`` it refers to the current process """ proc_bitness = windows.current_process.bitness @@ -410,7 +71,6 @@ def GetPEFile(baseaddr, target=None, force_bitness=None): transformers = get_structure_transformer_for_target(target, targetedbitness) #ctypes_structure_transformer, create_structure_at = transformers transfor_funcs = CtypesStructureTransformers(*transformers) # TODO: rename - return PEFile(target, baseaddr, targetedbitness, transfor_funcs) @@ -429,7 +89,7 @@ class IMPORT_BY_NAME(ctypes.Structure): def get_string(target, addr): if target is None: - return ctypes.c_char_p(addr).value + return ctypes.c_char_p(addr).value.decode("latin1") return target.read_string(addr) @@ -438,8 +98,11 @@ class PESection(IMAGE_SECTION_HEADER): @property def name(self): if self.target is None: - return get_string(self.target, ctypes.addressof(self.Name))[:8] - return get_string(self.target, self._base_addr)[:8] + name = get_string(self.target, ctypes.addressof(self.Name))[:8] + else: + name = get_string(self.target, self._base_addr)[:8] + # Decode as UTF-8 as the MS doc say ? + return name @property def start(self): @@ -461,6 +124,14 @@ class PESection(IMAGE_SECTION_HEADER): return self +class IATPtr(PVOID): + @classmethod + def from_iatentry(cls, iat_entry): + self = cls.from_address(iat_entry.addr) + self.addr = iat_entry.addr + self.nonhookvalue = iat_entry.nonhookvalue + return self + class IATEntry(ctypes.Structure): """Represent an entry in the IAT of a module Can be used to get resolved value and setup hook @@ -468,8 +139,6 @@ class IATEntry(ctypes.Structure): _fields_ = [ ("value", PVOID)] - - @classmethod def create(cls, addr, ord, name, target, transformers): self = transformers.create_structure_at(cls, addr) @@ -504,10 +173,18 @@ class IATEntry(ctypes.Structure): raise NotImplementedError("Setting hook in remote process (use python code injection)") hook = hooks.IATHook(self, callback, types) + import weakref + self.whook = weakref.ref(hook, self.on_destroy) self.hook = hook hook.enable() return hook + 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): """Remove the hook on the entry""" if self.hook is None: @@ -516,6 +193,11 @@ class IATEntry(ctypes.Structure): self.hook = None return True + # def __del__(self): + # print(self.hook) + # if self.hook: + # print("LOL BYE {0}".format(self.hook)) + class IMAGE_IMPORT_DESCRIPTOR(IMAGE_IMPORT_DESCRIPTOR): # TODO: use explicite name winstructs.IMAGE_IMPORT_DESCRIPTOR def get_INT(self): @@ -533,7 +215,7 @@ class IMAGE_IMPORT_DESCRIPTOR(IMAGE_IMPORT_DESCRIPTOR): # TODO: use explicite na if self.target is None: name = get_string(self.target, name_address) else: - name = get_string(self.target, name_address).decode() + 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) @@ -569,6 +251,9 @@ class IMAGE_EXPORT_DIRECTORY(IMAGE_EXPORT_DIRECTORY): # TODO: use explicite name func += self.baseaddr if nb in NameOrdinals: name = get_string(self.target, Names[NameOrdinals.index(nb)] + self.baseaddr) + # Export name should be ascii + # Decode from ascii or return bytes ? + # https://docs.microsoft.com/en-us/windows/win32/debug/pe-format#export-address-table else: name = None res.append((nb, func, name)) @@ -620,11 +305,6 @@ class PEFile(object): else: self.IMAGE_ORDINAL_FLAG = IMAGE_ORDINAL_FLAG64 - def get_string(self, addr): - if self.target is None: - return ctypes.c_char_p(addr).value - return self.target.read_string(addr) - def get_DOS_HEADER(self): return self.transformers.create_structure_at(IMAGE_DOS_HEADER, self.baseaddr) @@ -695,10 +375,19 @@ class PEFile(object): :type: {(:class:`int` or :class:`str`) : :class:`int`}""" res = {} exp_dir = self.get_EXPORT_DIRECTORY() + export_datadir = self.get_DataDirectory()[IMAGE_DIRECTORY_ENTRY_EXPORT] + export_start = self.baseaddr + export_datadir.VirtualAddress + 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: + # Export proxy... + # Contains the string to another Dll.Function + rva_addr = get_string(self.target, rva_addr) # Put the string proxy instead + res[id] = rva_addr if rva_name is not None: res[rva_name] = rva_addr @@ -730,7 +419,7 @@ class PEFile(object): for iat_entry, (ord, name) in zip(IAT, INT): # str(name.decode()) -> python2 and python3 compatible for str result iat_entry.ord = ord - iat_entry.name = str(name.decode()) if name else "" + iat_entry.name = str(name) if name else "" name = get_string(self.target, self.baseaddr + import_descriptor.Name) res.setdefault(name.lower(), []).extend(IAT) return res \ No newline at end of file diff --git a/windows/winobject/exception.py b/windows/winobject/exception.py index 5c7efa4..0556abf 100644 --- a/windows/winobject/exception.py +++ b/windows/winobject/exception.py @@ -286,7 +286,7 @@ class ECONTEXT64(ECONTEXTBase, CONTEXT64): temporary workaround or horrible hack ? choose your side """ size = ctypes.sizeof(cls) - nb_qword = (size + 8) / ctypes.sizeof(ULONGLONG) + nb_qword = int((size + 8) / ctypes.sizeof(ULONGLONG)) buffer = (nb_qword * ULONGLONG)() struct_address = ctypes.addressof(buffer) if (struct_address & 0xf) not in [0, 8]: diff --git a/windows/winobject/process.py b/windows/winobject/process.py index 4a8d38a..7c97fb6 100644 --- a/windows/winobject/process.py +++ b/windows/winobject/process.py @@ -29,6 +29,7 @@ from windows.winobject import apisetmap from windows.winobject import token from windows import security +from windows.pycompat import raw_encode, raw_decode TimeInfo = namedtuple("TimeInfo", ["creation", "exit", "kernel", "user"]) """Time information about a process""" @@ -356,11 +357,11 @@ class Process(utils.AutoHandle): read_size = read_size / 2 continue readden += read_size - if "\x00" in x: - res.append(x.split("\x00", 1)[0]) + if b"\x00" in x: + res.append(x.split(b"\x00", 1)[0]) break res.append(x) - return "".join(res) + return b"".join(res).decode("ascii") def read_wstring(self, addr): """Read a windows UTF16 string at ``addr``""" @@ -380,12 +381,16 @@ class Process(utils.AutoHandle): read_size = read_size / 2 continue readden += read_size - utf16_chars = ["".join(c) for c in zip(*[iter(x)] * 2)] - if "\x00\x00" in utf16_chars: - res.extend(utf16_chars[:utf16_chars.index("\x00\x00")]) + # Bytearray will work on py2 & py3 + # Py2: bytearray((0, 0)) == b"\x00\x00" + # Py2: bytearray((0, 0)) == b"\x00\x00" + utf16_chars = [bytearray(c) for c in zip(*[iter(x)] * 2)] + if b"\x00\x00" in utf16_chars: + # Translate bytearray to str/bytes for both py2 & py3 + res.extend(bytes(x) for x in utf16_chars[:utf16_chars.index(b"\x00\x00")]) break res.extend(x) - return "".join(res).decode('utf16') + return b"".join(res).decode('utf16') def write_byte(self, addr, byte): """write a byte at ``addr``""" @@ -667,6 +672,7 @@ class CurrentProcess(Process): def write_memory(self, addr, data): """Write data at addr""" + data = raw_encode(data) # buffertype = (c_char * len(data)).from_address(addr) # buffertype[:len(data)] = data ctypes.memmove(addr, data, len(data)) @@ -1002,13 +1008,13 @@ class WinProcess(Process): pass return '<{0} "{1}" pid {2} at {3}>'.format(self.__class__.__name__, exe_name, self.pid, hex(id(self))) - def virtual_alloc(self, size, prot=PAGE_EXECUTE_READWRITE): + def virtual_alloc(self, size, prot=PAGE_EXECUTE_READWRITE, addr=None): """Allocate memory in the process :return: The address of the allocated memory :rtype: :class:`int` """ - return winproxy.VirtualAllocEx(self.handle, dwSize=size, flProtect=prot) + return winproxy.VirtualAllocEx(self.handle, lpAddress=addr, dwSize=size, flProtect=prot) def virtual_free(self, addr): """Free memory in the process by virtual_alloc""" @@ -1016,6 +1022,7 @@ class WinProcess(Process): def write_memory(self, addr, data): """Write `data` at `addr`""" + data = raw_encode(data) if windows.current_process.bitness == 32 and self.bitness == 64: if not winproxy.is_implemented(winproxy.NtWow64WriteVirtualMemory64): raise ValueError("NtWow64WriteVirtualMemory64 non available in ntdll: cannot write into 64bits processus")