From 9f761e8a9f98170597dd06a0b376d97a9f779ea5 Mon Sep 17 00:00:00 2001 From: Duncan Ogilvie Date: Mon, 13 Mar 2023 00:59:49 +0100 Subject: [PATCH 1/3] Improve syscall exceptions a bit #46 --- src/dumpulator/dumpulator.py | 15 ++++++--------- 1 file changed, 6 insertions(+), 9 deletions(-) diff --git a/src/dumpulator/dumpulator.py b/src/dumpulator/dumpulator.py index 6089da7..bb25cdc 100644 --- a/src/dumpulator/dumpulator.py +++ b/src/dumpulator/dumpulator.py @@ -1072,7 +1072,7 @@ def raise_kill(self, exc=None): self.regs.cip = FORCE_KILL_ADDR self.kill_me = exc if exc is not None: - raise exc + return exc else: self.kill_me = True self._uc.emu_stop() @@ -1501,7 +1501,7 @@ def syscall_arg(index): try: argvalue = argtype(dp.args[i] & 0xFFFFFFFF) except KeyError as x: - raise Exception(f"Unknown enum value {dp.args[i]} for {type(argtype)}") + raise Exception(f"Unknown enum value {dp.args[i]} for {type(argtype)}") from None else: argvalue = argtype(argvalue) args.append(argvalue) @@ -1517,7 +1517,7 @@ def syscall_arg(index): if isinstance(status, ExceptionInfo): print("context switch, stopping emulation") dp.exception = status - dp.raise_kill(UcError(UC_ERR_EXCEPTION)) + raise dp.raise_kill(UcError(UC_ERR_EXCEPTION)) from None else: dp.info(f"status = {status:x}") dp.regs.cax = status @@ -1530,17 +1530,14 @@ def syscall_arg(index): except UcError as err: raise err except Exception as exc: - traceback.print_exc() dp.error(f"Exception thrown during syscall implementation, stopping emulation!") - dp.raise_kill(exc) + raise dp.raise_kill(exc) from None finally: dp.sequence_id += 1 else: - dp.error(f"syscall index: {index:x} -> {name} not implemented!") - dp.raise_kill(NotImplementedError()) + raise dp.raise_kill(NotImplementedError(f"syscall index: {index:x} -> {name} not implemented!")) from None else: - dp.error(f"syscall index {index:x} out of range") - dp.raise_kill(IndexError()) + raise dp.raise_kill(IndexError(f"syscall index {index:x} out of range")) from None def _emulate_unsupported_instruction(dp: Dumpulator, instr: CsInsn): if instr.id == X86_INS_RDRAND: From f42ea3fee738ce3520b8000550ef2cba9bdba18b Mon Sep 17 00:00:00 2001 From: Duncan Ogilvie Date: Mon, 13 Mar 2023 01:01:36 +0100 Subject: [PATCH 2/3] Add ntprimitives.Struct to improve the development experience --- src/dumpulator/native.py | 19 ++- src/dumpulator/ntprimitives.py | 219 +++++++++++++++++++-------------- src/dumpulator/ntstructs.py | 108 +++++++--------- src/dumpulator/ntsyscalls.py | 6 +- 4 files changed, 184 insertions(+), 168 deletions(-) diff --git a/src/dumpulator/native.py b/src/dumpulator/native.py index 411bd3e..f4b4594 100644 --- a/src/dumpulator/native.py +++ b/src/dumpulator/native.py @@ -596,18 +596,13 @@ class SECTION_IMAGE_INFORMATION(ctypes.Structure): ] return SECTION_IMAGE_INFORMATION() -def PROCESS_BASIC_INFORMATION(arch: Architecture): - class PROCESS_BASIC_INFORMATION(ctypes.Structure): - _alignment_ = arch.alignment() - _fields_ = [ - ("ExitStatus", ctypes.c_uint32), - ("PebBaseAddress", arch.ptr_type()), - ("AffinityMask", arch.ptr_type()), - ("BasePriority", ctypes.c_uint32), - ("UniqueProcessId", arch.ptr_type()), - ("InheritedFromUniqueProcessId", arch.ptr_type()), - ] - return PROCESS_BASIC_INFORMATION() +class PROCESS_BASIC_INFORMATION(Struct): + ExitStatus: ULONG + PebBaseAddress: PVOID + AffinityMask: KAFFINITY + BasePriority: KPRIORITY + UniqueProcessId: ULONG_PTR + InheritedFromUniqueProcessId: ULONG_PTR class KEY_VALUE_FULL_INFORMATION(ctypes.Structure): _fields_ = [ diff --git a/src/dumpulator/ntprimitives.py b/src/dumpulator/ntprimitives.py index dabad48..49c574a 100644 --- a/src/dumpulator/ntprimitives.py +++ b/src/dumpulator/ntprimitives.py @@ -1,11 +1,11 @@ import struct import ctypes import typing -from typing import Optional, Annotated, Generic, TypeVar, Type +from typing import Optional, Annotated, Generic, TypeVar, Type, Union, SupportsInt, SupportsBytes from enum import Enum from dataclasses import dataclass -class Architecture(object): +class Architecture: def __init__(self, x64: bool): self._x64 = x64 @@ -16,61 +16,61 @@ def x64(self): def ptr_size(self): return 8 if self._x64 else 4 - def ptr_type(self, t=None): # TODO: implement type + def ptr_type(self): return ctypes.c_uint64 if self._x64 else ctypes.c_uint32 def alignment(self): return 16 if self._x64 else 8 - def read(self, addr: int, size: int) -> bytes: + def read(self, addr: SupportsInt, size: int) -> bytes: raise NotImplementedError() - def write(self, addr: int, data: bytes): + def write(self, addr: SupportsInt, data: Union[SupportsBytes, bytes]): raise NotImplementedError() - def read_char(self, addr: int) -> int: + def read_char(self, addr: SupportsInt) -> int: return struct.unpack(" int: + def read_short(self, addr: SupportsInt) -> int: return struct.unpack(" int: + def read_long(self, addr: SupportsInt) -> int: return struct.unpack(" int: + def read_byte(self, addr: SupportsInt) -> int: return struct.unpack(" int: + def read_ushort(self, addr: SupportsInt) -> int: return struct.unpack(" int: + def read_ulong(self, addr: SupportsInt) -> int: return struct.unpack(" int: + def read_ptr(self, addr: SupportsInt) -> int: return struct.unpack(" str: + def read_str(self, addr: SupportsInt, encoding="utf-8") -> str: # TODO: safely read the memory data = self.read(addr, 512) @@ -86,10 +86,8 @@ def read_str(self, addr: int, encoding="utf-8") -> str: return data.decode(encoding) - T = TypeVar("T") - class P(Generic[T]): _ptr_ = True @@ -113,24 +111,34 @@ def type(self) -> Type[T]: def is_ptr(cls, tv): return hasattr(tv, "_ptr_") - def read(self, size) -> bytes: + def read(self, size: int) -> bytes: return self.arch.read(self.ptr, size) - def write(self, data: bytes): + def write(self, data: Union[SupportsBytes, bytes]): self.arch.write(self.ptr, data) - def __getitem__(self, index): + def __getitem__(self, index) -> T: ptype = self.type if ptype is None: - return self.arch.read_ptr(self.ptr + index * self.arch.ptr_size()) + raise TypeError(f"No type associated with pointer") + + if P.is_ptr(ptype): + ptr = self.ptr + index * self.arch.ptr_size() + return ptype(self.arch, self.arch.read_ptr(ptr)) else: - assert index == 0 # TODO: sizeof() not yet implemented - sizeof = self.arch.ptr_size() - ptr = self.ptr + index * sizeof - if P.is_ptr(ptype): - return ptype(self.arch, self.arch.read_ptr(ptr)) + size = Struct.sizeof(ptype, self.arch) + ptr = self.ptr + index * size + if issubclass(ptype, Struct): + return ptype(self.arch, ptr) else: - return ptype(PVOID(self.arch, ptr)) + ctype = Struct.translate_ctype(self.arch.ptr_type(), ptype) + size = ctypes.sizeof(ctype) + data = self.arch.read(ptr, size) + value = ctype.from_buffer(data) + return ptype(value) + + def deref(self) -> T: + return self[0] def __int__(self): return self.ptr @@ -144,10 +152,10 @@ def __ne__(self, other): def __str__(self): return hex(self.ptr) - def read_byte_str(self, size): + def read_byte_str(self, size: int): return bytes(self.read(size)) - def read_str(self, size, encoding="utf8"): + def read_str(self, size: int, encoding="utf8"): return self.read(size).decode(encoding) def read_unicode_str(self): @@ -158,44 +166,89 @@ def read_unicode_str(self): def read_ptr(self): return self.arch.read_ptr(self.ptr) - def write_ptr(self, value: int): - return self.arch.write_ptr(self.ptr, value) + def write_ptr(self, value: typing.SupportsInt): + return self.arch.write_ptr(self.ptr, int(value)) - def write_ulong(self, value: int): - return self.arch.write_ulong(self.ptr, value) + def write_ulong(self, value: typing.SupportsInt): + return self.arch.write_ulong(self.ptr, int(value)) def read_ulong(self): return self.arch.read_ulong(self.ptr) - def deref(self): - return self[0] - class PVOID(P): pass +# TODO: find a way to show the fields in the PyCharm debugger (properties?) class Struct: - def __init__(self, arch: Architecture): + def __init__(self, arch: Architecture, ptr: int = 0): self._hints = typing.get_type_hints(self) + # TODO: allow 'binding' the pointer + self._ptr = ptr self._arch = arch fields = [] for name, t in self._hints.items(): - fields.append((name, self._translate_ctype(name, t))) - self._ctype = Struct.create_type( + ctype = Struct.translate_ctype(arch.ptr_type(), t) + if ctype is None: + raise TypeError(f"Unsupported native type {t.__name__} for member {self.__class__.__name__}{name}") + fields.append((name, ctype)) + self._ctype = Struct._create_type( self.__class__.__name__ + "_ctype", ctypes.Structure, _fields_=fields, _alignment_=arch.alignment() ) - self._cself = self._ctype() + if ptr != 0: + data = arch.read(ptr, Struct.sizeof(self)) + self._cself = self._ctype.from_buffer_copy(data) + else: + self._cself = self._ctype() + # Add properties to visualize things in the debugger + for name in self._hints: + object.__setattr__(self, name, property(lambda s: getattr(s, name))) + + @classmethod + def sizeof(cls, value, arch: Optional[Architecture] = None) -> int: + if P.is_ptr(value): + if arch is None: + ctype = value.arch.ptr_type() + else: + ctype = arch.ptr_type() + elif isinstance(value, Struct): + ctype = value._ctype + elif issubclass(value, Struct): + if arch is None: + raise TypeError("No architecture passed") + ctype = value(arch)._ctype + elif isinstance(value, Int): + if arch is None: + raise TypeError("No architecture passed") + ctype = Struct.translate_ctype(arch.ptr_type(), type(value)) + elif issubclass(value, Int): + if arch is None: + raise TypeError("No architecture passed") + ctype = Struct.translate_ctype(arch.ptr_type(), value) + else: + raise NotImplementedError() + assert ctype is not None + return ctypes.sizeof(ctype) + + @classmethod + def bytes(cls, value: "Struct") -> bytes: + assert isinstance(value, Struct) + return bytes(value) + + def __bytes__(self) -> bytes: + return bytes(self._cself) # https://stackoverflow.com/questions/28552433/dynamically-create-ctypes-in-python @staticmethod def _create_type(name, *bases, **attrs): return type(name, bases, attrs) - def _translate_ctype(self, name: str, t: type): - if t is P: - return self._arch.ptr_type() + @staticmethod + def translate_ctype(ptr_type, t: type): + if P.is_ptr(t): + return ptr_type elif issubclass(t, Enum): return ctypes.c_uint32 elif issubclass(t, UCHAR): @@ -211,56 +264,41 @@ def _translate_ctype(self, name: str, t: type): elif issubclass(t, LONG): return ctypes.c_int32 elif issubclass(t, ULONG_PTR): - return self._arch.ptr_type() + return ptr_type else: - raise TypeError(f"Unsupported native type {t.__name__} for member {self.__class__.__name__}{name}") + return None - def __getattribute__(self, name): - if name.startswith("_"): + def __getattribute__(self, name: str): + if name.startswith("__"): + return object.__getattribute__(self, name) + elif name != "_hints" and name in self._hints: + # Proxy the ctypes fields + atype = self._hints[name] + avalue = getattr(self._cself, name) + if P.is_ptr(atype): + return atype(self._arch, avalue) + elif issubclass(atype, Enum): + return atype(avalue) + else: + return atype(avalue) + else: return object.__getattribute__(self, name) - if name not in self._hints: - raise AttributeError(f"Attribute not found: {self.__class__.__name__}.{name}") - return getattr(self._cself, name) - def __setattr__(self, name, value): - if name.startswith("_"): + def __setattr__(self, name: str, value): + if name.startswith("__"): + object.__setattr__(self, name, value) + elif name != "_hints" and name in self._hints: + # TODO: support assigning pointers properly + # TODO: support assigning enums properly + setattr(self._cself, name, int(value)) + elif name.startswith("_") or name in self.__dict__: object.__setattr__(self, name, value) else: - if name not in self._hints: - raise AttributeError(f"Attribute not found: {self.__class__.__name__}.{name}") - return setattr(self._cself, name, value) - -# Note: this is very WIP -class ArchStream: - def __init__(self, ptr: PVOID): - self.ptr = ptr - self.pos = 0 - - @property - def x64(self): - return self.ptr.arch.ptr_size() == 8 - - def skip(self, size): - self.pos += size - - def read(self, size): - data = self.ptr.arch.read(self.ptr.ptr + self.pos, size) - self.pos += size - return data - - def read_ushort(self): - return struct.unpack(" P[T]: - ptr = struct.unpack(" bytes: + encoded = s.encode("utf-16-le") + b"\0\0" + if ptr.arch.x64: + data = struct.pack(" bytes: - encoded = s.encode("utf-16-le") + b"\0\0" - if ptr.arch.x64: - data = struct.pack(" Date: Mon, 13 Mar 2023 01:07:02 +0100 Subject: [PATCH 3/3] Cache dependencies --- .github/workflows/build.yml | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 2dd0527..59d6012 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -19,7 +19,9 @@ jobs: with: python-version: '3.9' architecture: 'x64' - + cache: 'pip' + cache-dependency-path: 'setup.cfg' + - name: Python setup run: | python setup.py develop