Exploring remotectypes

This commit is contained in:
Clement Rouault
2015-09-03 12:56:32 +02:00
parent bd855ce7d5
commit a133254c6b
5 changed files with 507 additions and 21 deletions
+41 -4
View File
@@ -10,6 +10,7 @@ import windows.generated_def.winfuncs as winfuncs
kernel32 = ctypes.windll.kernel32
advapi32 = ctypes.windll.Advapi32
iphlpapi = ctypes.windll.iphlpapi
ntdll = ctypes.windll.ntdll
class Kernel32Error(WindowsError):
def __new__(cls, func_name):
@@ -114,6 +115,31 @@ class Advapi32Proxy(ApiProxy):
class IphlpapiProxy(ApiProxy):
APIDLL = iphlpapi
default_error_check = staticmethod(iphlpapi_error_check)
class NtdllProxy(ApiProxy):
APIDLL = ntdll
default_error_check = staticmethod(kernel32_error_check)
class OptionalExport(object):
"""used 'around' a Proxy decorator
Should be used for export that are not available everywhere (ntdll internals | 32/64 bits stuff)
If the export is not found the function will be None
Example:
@OptionalExport(NtdllProxy('NtWow64ReadVirtualMemory64'))
def NtWow64ReadVirtualMemory64(...)
...
"""
def __init__(self, subdecorator):
self.subdecorator = subdecorator
def __call__(self, f):
try:
return self.subdecorator(f)
except AttributeError as e:
print("NOT FOUND")
print(e)
return None
def TransparentApiProxy(APIDLL, func_name, error_check):
"""Create a ctypes function for 'func_name' with no python arg pre-check"""
@@ -138,6 +164,7 @@ class NeededParameterType(object):
def __repr__(self):
return "NeededParameter"
NeededParameter = NeededParameterType()
@@ -313,6 +340,18 @@ def RemoveVectoredExceptionHandler(Handler):
@Kernel32Proxy("WaitForSingleObject", kernel32_zero_check)
def WaitForSingleObject(hHandle, dwMilliseconds=INFINITE):
return WaitForSingleObject.ctypes_function(hHandle, dwMilliseconds)
@Kernel32Proxy("DeviceIoControl")
def DeviceIoControl(hDevice, dwIoControlCode, lpInBuffer, nInBufferSize=None, lpOutBuffer=NeededParameter, nOutBufferSize=None, lpBytesReturned=None, lpOverlapped=None):
if nInBufferSize is None:
nInBufferSize = len(lpInBuffer)
if nOutBufferSize is None:
nOutBufferSize = len(lpOutBuffer)
if lpBytesReturned is None:
# Some windows check 0 / others does not
lpBytesReturned = ctypes.byref(DWORD())
return DeviceIoControl.ctypes_function(hDevice, dwIoControlCode, lpInBuffer, nInBufferSize, lpOutBuffer, nOutBufferSize, lpBytesReturned, lpOverlapped)
###### ADVAPI32 ########
@@ -342,14 +381,12 @@ def AdjustTokenPrivileges(TokenHandle, DisableAllPrivileges=False, NewState=Need
SetTcpEntry = TransparentIphlpapiProxy('SetTcpEntry')
@IphlpapiProxy('GetExtendedTcpTable')
@OptionalExport(IphlpapiProxy('GetExtendedTcpTable'))
def GetExtendedTcpTable(pTcpTable, pdwSize=None, bOrder=True, ulAf=NeededParameter, TableClass=TCP_TABLE_OWNER_PID_ALL, Reserved=0):
if pdwSize is None:
ctypes.sizeof(pTcpTable)
return GetExtendedTcpTable.ctypes_function(pTcpTable, pdwSize, bOrder, ulAf, TableClass, Reserved)
# Design 2
# Design 2 should use the automatics args
+309
View File
@@ -0,0 +1,309 @@
import _ctypes
import ctypes
import ctypes.wintypes
import itertools
from _ctypes import _SimpleCData
class DummyTarget(object):
def read_memory(self, addr, size):
#print("read_memory at {0} size {1}".format(addr, size))
return (ctypes.c_char * size).from_address(addr)[:]
# 64bits pointeurs and long
### Utils ####
def is_pointer(x):
return isinstance(x, _ctypes._Pointer)
def is_pointer_type(x):
return issubclass(x, _ctypes._Pointer)
def is_array(x):
return isinstance(x, _ctypes.Array)
def is_array_type(x):
return issubclass(x, _ctypes.Array)
def is_structure_type(x):
return issubclass(x, ctypes.Structure)
def is_union_type(x):
return issubclass(x, ctypes.Union)
#### My types ####
## 64bits pointer types ##
# I know direct inheritance from _SimpleCData seems bad
# But it seems to be the only way to have the normal
# ctypes.Structure way of working (need to investigate)
class c_void_p64(_SimpleCData):
_type_ = "Q"
class c_char_p64(_SimpleCData):
_type_ = "Q"
class c_wchar_p64(_SimpleCData):
_type_ = "Q"
# standard type translation
# don't know how to handle size_t since it's non-distinguable from c_ulong
# maybe force import before ctypes and modif stuff into ctypes ?
## Remote Value
# Used by the RemoteStructure to access the target memory
class RemoteValue(object):
@classmethod
def from_buffer_with_target(cls, buffer, offset=0, target=None):
x = cls.from_buffer(buffer)
x.target = target
return x
class RemotePtr(RemoteValue):
@property
def raw_value(self):
return ctypes.cast(self, ctypes.c_void_p).value
class RemoteCCharP(RemotePtr, ctypes.c_char_p):
@property
def value(self):
base = self.raw_value
res = []
for i in itertools.count():
x = self.target.read_memory(base + (i * 0x100), 0x100)
if "\x00" in x:
res.append(x.split("\x00", 1)[0])
break
res.append(x)
return "".join(res)
class RemoteWCharP(RemotePtr, ctypes.c_char_p):
@property
def value(self):
base = self.raw_value
res = []
for i in itertools.count():
x = self.target.read_memory(base + (i * 0x100), 0x100)
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")])
break
res.extend(x)
return "".join(res).decode('utf16')
class RemoteStructurePointer(RemotePtr, ctypes.c_void_p):
@classmethod
def from_buffer_with_target_and_ptr_type(cls, buffer, offset=0, target=None, ptr_type=None):
x = cls.from_buffer(buffer)
x.target = target
x.real_pointer_type = ptr_type
return x
@property
def contents(self):
#print(self.real_pointer_type)
remote_pointed_type = RemoteStructure.from_structure(self.real_pointer_type._type_)
return remote_pointed_type(self.raw_value, self.target)
def __repr__(self):
return "<RemoteStructurePointer to {0}>".format(self.real_pointer_type._type_.__name__)
# 64bits pointers
class RemotePtr64(RemoteValue):
def __init__(self, value, target):
self.target = target
super(RemotePtr64, self).__init__(value)
@property
def raw_value(self):
# Bypass our own 'value' implementation
# Even if we are a subclass of c_ulonglong
my_addr = ctypes.addressof(self)
return ctypes.c_ulonglong.from_address(my_addr).value
return ctypes.cast(self, ctypes.c_ulonglong).value
#def from_value(self):
class Remote_c_void_p64(RemotePtr64, c_void_p64):
pass
# base explanation:
# RemotePtr64 for the good `raw_value` implem
# RemoteCCharP for the good `value` implem
# c_char_p64 for the good _type_ (ctypes size)
class Remote_c_char_p64(c_char_p64, RemotePtr64, RemoteCCharP):
def __repr__(self):
return "<Remote_c_char_p64({0})>".format(self.raw_value)
class Remote_w_char_p64(c_wchar_p64, RemotePtr64, RemoteWCharP):
def __repr__(self):
return "<Remote_c_char_p64({0})>".format(self.raw_value)
class RemoteStructurePointer64(Remote_c_void_p64):
@property
def raw_value(self):
return self.value
@classmethod
def from_buffer_with_target_and_ptr_type(cls, buffer, offset=0, target=None, ptr_type=None):
x = cls.from_buffer(buffer)
x.target = target
x.real_pointer_type = ptr_type
return x
@property
def contents(self):
remote_pointed_type = transform_type_to_remote64bits(self.real_pointer_type._sub_ctypes_)
return remote_pointed_type(self.raw_value, self.target)
type_32_64_translation_table = {
ctypes.c_void_p : Remote_c_void_p64,
ctypes.c_char_p : Remote_c_char_p64,
ctypes.c_wchar_p: Remote_w_char_p64,
#ctypes.wintypes.WPARAM : ctypes.c_ulonglong,
#ctypes.wintypes.LPARAM : ctypes.c_longlong,
#ctypes.c_size_t : c_size_t64, # don't know how to handle size_t size it's non-distinguable from c_ulong
#ctypes.c_ssize_t: c_ssize_t64
}
class RemoteStructureUnion(object):
"""Target is a process object"""
_reserved_name = ["_target", "_fields_", "_fields_dict_", "_base_addr", "_get_field_by_name",
"_get_field_descrptor_by_name", "_handle_field_getattr", "_field_type_to_remote_type", "__getattribute__", "_fields_"]
_field_type_to_remote_type = {
ctypes.c_char_p : RemoteCCharP,
ctypes.c_wchar_p : RemoteWCharP,
Remote_c_void_p64 : Remote_c_void_p64,
Remote_c_char_p64 : Remote_c_char_p64,
Remote_w_char_p64 : Remote_w_char_p64}
def __init__(self, base_addr, target):
if type(base_addr) not in (int, long):
import pdb;pdb.set_trace()
self._target = target
self._base_addr = base_addr
self._fields_dict_ = dict(self._fields_)
def _get_field_by_name(self, fieldname):
try:
return self._fields_dict_[fieldname]
except KeyError:
raise AttributeError(fieldname + "is not a field of {0}".format(type(self)))
def _get_field_descrptor_by_name(self, fieldname):
return getattr(type(self), fieldname) # ctypes metaclass fill this for us
def _handle_field_getattr(self, ftype, fosset, fsize):
s = self._target.read_memory(self._base_addr + fosset, fsize)
if ftype in self._field_type_to_remote_type:
return self._field_type_to_remote_type[ftype].from_buffer_with_target(bytearray(s), target=self._target).value
if issubclass(ftype, _ctypes._Pointer): # Pointer
return RemoteStructurePointer.from_buffer_with_target_and_ptr_type(bytearray(s), target=self._target, ptr_type=ftype)
if issubclass(ftype, RemotePtr64): # Pointer to remote64 bits process
return RemoteStructurePointer64.from_buffer_with_target_and_ptr_type(bytearray(s), target=self._target, ptr_type=ftype)
if issubclass(ftype, RemoteStructureUnion): # Structure already transfomed in remote
return ftype(self._base_addr + fosset, self._target)
if issubclass(ftype, ctypes.Structure): # Structure that must be transfomed
return RemoteStructure.from_structure(ftype)(self._base_addr + fosset, self._target)
if issubclass(ftype, ctypes.Union): # Structure that must be transfomed
return RemoteUnion.from_structure(ftype)(self._base_addr + fosset, self._target)
if issubclass(ftype, _ctypes.Array):
return ftype.from_buffer(bytearray(s))
# Normal types
# Follow the ctypes usage: if it's not directly inherited from _SimpleCData
# We do not apply the .value
# Seems weird but it's mandatory AND useful :D (in pe_parse)
if _SimpleCData not in ftype.__bases__:
return ftype.from_buffer(bytearray(s))
return ftype.from_buffer(bytearray(s)).value
def __getattribute__(self, fieldname):
if fieldname in type(self)._reserved_name: #Prevent recursion !
return super(RemoteStructureUnion, self).__getattribute__(fieldname)
try:
t = self._get_field_by_name(fieldname)
except AttributeError as e: # Not a real attribute
return super(RemoteStructureUnion, self).__getattribute__(fieldname)
descr = self._get_field_descrptor_by_name(fieldname)
return self._handle_field_getattr(t, descr.offset, descr.size)
@classmethod
def from_structure(cls, structcls):
class MyStruct(cls, structcls): # inherit of structcls to keep property (see winobject.LoadedModule)
_fields_ = structcls._fields_
MyStruct.__name__ = "Remote" + structcls.__name__
return MyStruct
@classmethod
def from_fields(cls, fields, base_cls=None):
bases = [cls]
if base_cls:
bases.append(base_cls)
# inherit of structcls to keep property (see winobject.LoadedModule)
RemoteStruct = type("RemoteStruct", tuple(bases), {"_fields_" : fields})
if base_cls:
RemoteStruct.__name__ = "Remote" + base_cls.__name__
return RemoteStruct
class RemoteStructure(RemoteStructureUnion, ctypes.Structure):
pass
class RemoteUnion(RemoteStructureUnion, ctypes.Union):
pass
remote_struct = RemoteStructure.from_structure
if ctypes.sizeof(ctypes.c_void_p) == 4:
# ctypes 32 -> 64 methods
def MakePtr(type):
class PointerToStruct64(Remote_c_void_p64):
_sub_ctypes_ = (type)
return PointerToStruct64
def transform_structure_to_remote64bits(structcls):
"""Create a remote structure for a 64bits target process"""
new_fields = []
for fname, ftype in structcls._fields_:
ftype = transform_type_to_remote64bits(ftype)
new_fields.append((fname, ftype))
return RemoteStructure.from_fields(new_fields, base_cls=structcls)
def transform_union_to_remote64bits(structcls):
"""Create a remote structure for a 64bits target process"""
new_fields = []
for fname, ftype in structcls._fields_:
ftype = transform_type_to_remote64bits(ftype)
new_fields.append((fname, ftype))
return RemoteUnion.from_fields(new_fields, base_cls=structcls)
def transform_type_to_remote64bits(ftype):
if is_pointer_type(ftype):
return MakePtr(ftype._type_)
if is_array_type(ftype):
return (transform_type_to_remote64bits(ftype._type_) * ftype._length_)
if is_structure_type(ftype):
return transform_structure_to_remote64bits(ftype)
if is_union_type(ftype):
return transform_union_to_remote64bits(ftype)
# Normal types
return type_32_64_translation_table.get(ftype, ftype)
def dump_type(t):
fields_name = t._fields_
for fn in fields_name:
print getattr(t, fn[0])
+3 -3
View File
@@ -38,9 +38,9 @@ NtCreateThreadStub = Pretty_NtCreateThreadStub.replace(" ", "").replace("\n", ""
def genere_return_32bits_stub(ret_addr):
ret_32b = x64.MultipleInstr()
ret_32b += x64.Mov_RCX_X((CS_32bits << 32) + ret_addr)
ret_32b += x64.Push_RCX()
ret_32b += x64.Retf()
ret_32b += x64.Mov('RCX', (CS_32bits << 32) + ret_addr)
ret_32b += x64.Push('RCX')
ret_32b += x64.Retf32() #32 bits return addr
return ret_32b.get_code()
# The format of a jump to 64bits mode
+19 -4
View File
@@ -3,7 +3,9 @@ import msvcrt
import os
import copy
import sys
import code
import windows
from . import k32testing as kernel32proxy
from .generated_def import windef
from .generated_def.winstructs import *
@@ -17,10 +19,12 @@ def swallow_ctypes_copy(ctypes_object):
def get_func_addr(dll_name, func_name):
dll = ctypes.WinDLL(dll_name)
return kernel32proxy.GetProcAddress(dll._handle, func_name)
modules = windows.current_process.peb.modules
if not dll_name.lower().endswith(".dll"):
dll_name += ".dll"
mod = [x for x in modules if x.name == dll_name][0]
return mod.pe.exports[func_name]
def is_wow_64(hProcess):
try:
fnIsWow64Process = get_func_addr("kernel32.dll", "IsWow64Process")
@@ -61,6 +65,17 @@ def create_console():
import os
#os.dup2(console_stderr.fileno(), 2)
sys.stderr = console_stderr
class FixedInteractiveConsole(code.InteractiveConsole):
def raw_input(self, prompt=">>>"):
sys.stdout.write(prompt)
return raw_input("")
def pop_shell():
create_console()
FixedInteractiveConsole(locals()).interact()
class VirtualProtected(object):
"""A context manager usable like `VirtualProtect` that will restore the old protection at exit
+135 -10
View File
@@ -2,11 +2,16 @@ import ctypes
import os
import codecs
import copy
import time
import struct
import windows
import windows.syswow64
import windows.k32testing as kernel32proxy
import windows.injection as injection
import windows.native_exec as native_exec
import windows.native_exec.simple_x86 as x86
import windows.native_exec.simple_x64 as x64
from . import utils
@@ -375,9 +380,27 @@ class WinProcess(PROCESSENTRY32, Process):
def read_memory(self, addr, size):
"""Read `size` from `addr`"""
print("Read on page {0}".format(hex(addr & 0xfffffffffffff000)))
buffer = ctypes.create_string_buffer(size)
self.low_read_memory(addr, ctypes.byref(buffer), size)
return buffer[:]
#Simple cache test
real_read = read_memory
def read_memory(self, addr, size):
"""Cached version for test"""
if not hasattr(self, "_cache_cache"):
self._cache_cache = {}
page_addr = addr & 0xfffffffffffff000
if page_addr in self._cache_cache:
print("CACHED Read on page {0}".format(hex(page_addr)))
page_data = self._cache_cache[page_addr]
return page_data[addr & 0xfff: (addr & 0xfff) + size]
else:
page_data = self.real_read(page_addr, 0x1000)
self._cache_cache[page_addr] = page_data
return page_data[addr & 0xfff: (addr & 0xfff) + size]
def read_memory_into(self, addr, struct):
"""Read a :mod:`ctypes` struct from `addr`"""
@@ -400,8 +423,42 @@ class WinProcess(PROCESSENTRY32, Process):
def execute_python(self, pycode):
"""Execute Python code into the remote process"""
return injection.execute_python_code(self, pycode)
def get_peb_addr(self):
get_peb_32_code = codecs.decode(b'64a130000000', 'hex')
get_peb_64_code = codecs.decode(b"65488B042560000000", 'hex')
dest = self.virtual_alloc(0x1000)
if self.bitness == 32:
get_peb_code = get_peb_32_code
store_peb = x86.MultipleInstr()
store_peb += x86.Mov(x86.create_displacement(disp=dest), 'EAX')
store_peb += x86.Ret()
get_peb_code += store_peb.get_code()
self.write_memory(dest, "\x00" * 4)
self.write_memory(dest + 4, get_peb_code)
self.create_thread(dest + 4, 0)
time.sleep(0.01)
peb_addr = struct.unpack("<I", self.read_memory(dest, 4))[0]
return peb_addr
else:
get_peb_code = get_peb_64_code
store_peb = x64.MultipleInstr()
store_peb += x64.Mov(x64.create_displacement(disp=dest), 'RAX')
store_peb += x64.Ret()
get_peb_code += store_peb.get_code()
self.write_memory(dest, "\x00" * 8)
self.write_memory(dest + 8, get_peb_code)
self.create_thread(dest + 8, 0)
time.sleep(0.01)
peb_addr = struct.unpack("<Q", self.read_memory(dest, 8))[0]
return peb_addr
def peb(self):
if windows.current_process.bitness == 32 and self.bitness == 64:
return RemotePEB64(self.get_peb_addr(), self)
return RemotePEB(self.get_peb_addr(), self)
class LoadedModule(LDR_DATA_TABLE_ENTRY):
"""An entry in the PEB Ldr list"""
@property
@@ -418,7 +475,7 @@ class LoadedModule(LDR_DATA_TABLE_ENTRY):
:type: str
"""
return self.BaseDllName.Buffer
return str(self.BaseDllName.Buffer).lower()
@property
def fullname(self):
@@ -449,17 +506,26 @@ class LIST_ENTRY_PTR(PVOID):
def TO_LDR_ENTRY(self):
return LDR_DATA_TABLE_ENTRY.from_address(self.value - sizeof(PVOID) * 2)
class PEB(PEB):
def transform_ctypes_fields(struct, replacement):
return [(name, replacement.get(name, type)) for name, type in struct._fields_]
class RTL_USER_PROCESS_PARAMETERS(Structure):
_fields_ = transform_ctypes_fields(RTL_USER_PROCESS_PARAMETERS, # The one in generated_def
{"ImagePathName" : WinUnicodeString,
"CommandLine" : WinUnicodeString})
class PEB(Structure):
"""The PEB (Process Environment Block) of the current process"""
_fields_ = transform_ctypes_fields(PEB, # The one in generated_def
{"ProcessParameters" : POINTER(RTL_USER_PROCESS_PARAMETERS)})
@property
def imagepath(self):
"""The ImagePathName of the PEB
:type: :class:`WinUnicodeString`
"""
raw_imagepath = self.ProcessParameters.contents.ImagePathName
return WinUnicodeString.from_address(ctypes.addressof(raw_imagepath))
return self.ProcessParameters.contents.ImagePathName
@property
def commandline(self):
@@ -468,8 +534,7 @@ class PEB(PEB):
:type: :class:`WinUnicodeString`
"""
# This or changing the __repr__ of LSA_UNICODE_STRING
raw_cmd = self.ProcessParameters.contents.CommandLine
return WinUnicodeString.from_address(ctypes.addressof(raw_cmd))
return self.ProcessParameters.contents.CommandLine
@property
def modules(self):
@@ -484,4 +549,64 @@ class PEB(PEB):
res.append(current_dll)
list_entry_ptr = ctypes.cast(current_dll.InMemoryOrderLinks.Flink, LIST_ENTRY_PTR)
current_dll = list_entry_ptr.TO_LDR_ENTRY()
return [LoadedModule.from_address(addressof(LDR)) for LDR in res]
return [LoadedModule.from_address(addressof(LDR)) for LDR in res]
import windows.remotectypes as rctypes
class RemotePEB(rctypes.RemoteStructure.from_structure(PEB)):
RemoteLoadedModule = rctypes.RemoteStructure.from_structure(LoadedModule)
def ptr_flink_to_remote_module(self, ptr_value):
return self.RemoteLoadedModule(ptr_value - ctypes.sizeof(ctypes.c_void_p) * 2, self._target)
@property
def modules(self):
"""The loaded modules present in the PEB
:type: [:class:`LoadedModule`] -- List of loaded modules
"""
res = []
list_entry_ptr = self.Ldr.contents.InMemoryOrderModuleList.Flink.raw_value
current_dll = self.ptr_flink_to_remote_module(list_entry_ptr)
while current_dll.DllBase:
res.append(current_dll)
list_entry_ptr = current_dll.InMemoryOrderLinks.Flink.raw_value
current_dll = self.ptr_flink_to_remote_module(list_entry_ptr)
return res
if CurrentProcess().bitness == 32:
class RemoteLoadedModule64(rctypes.transform_type_to_remote64bits(LoadedModule)):
@property
def pe(self):
"""A PE representation of the module
:type: :class:`windows.pe_parse.PEFile`
"""
return pe_parse.PEFile(self.baseaddr, target=self._target)
class RemotePEB64(rctypes.transform_type_to_remote64bits(PEB)):
#RemoteLoadedModule64 = rctypes.transform_type_to_remote64bits(LoadedModule)
def ptr_flink_to_remote_module(self, ptr_value):
return RemoteLoadedModule64(ptr_value - ctypes.sizeof(rctypes.c_void_p64) * 2, self._target)
@property
def modules(self):
"""The loaded modules present in the PEB
:type: [:class:`LoadedModule`] -- List of loaded modules
"""
res = []
list_entry_ptr = self.Ldr.contents.InMemoryOrderModuleList.Flink.raw_value
current_dll = self.ptr_flink_to_remote_module(list_entry_ptr)
while current_dll.DllBase:
res.append(current_dll)
list_entry_ptr = current_dll.InMemoryOrderLinks.Flink.raw_value
current_dll = self.ptr_flink_to_remote_module(list_entry_ptr)
return res