mirror of
https://github.com/hakril/PythonForWindows
synced 2026-06-08 14:31:45 +00:00
handle code now use gdef + added some handle tests
This commit is contained in:
hakril
@@ -0,0 +1,35 @@
|
||||
import windows
|
||||
|
||||
def test_handle_process_id():
|
||||
handle_with_process = [h for h in windows.system.handles if h.dwProcessId]
|
||||
handle = handle_with_process[-1]
|
||||
proc = handle.process
|
||||
assert proc.pid == handle.dwProcessId == handle.pid
|
||||
|
||||
|
||||
def test_local_handle_type():
|
||||
t = windows.current_process.threads[0]
|
||||
th = t.handle
|
||||
hobj = [h for h in windows.current_process.handles if h.value == th][0]
|
||||
assert hobj.type == "Thread"
|
||||
assert hobj.name == ""
|
||||
assert hobj.infos
|
||||
|
||||
PIPE_NAME = "PFW_Test_handle_Pipe"
|
||||
TEST_FILE_FOR_HANDLE = r"C:\Windows\explorer.exe"
|
||||
|
||||
def test_remote_handle_type_and_name(proc32_64):
|
||||
# tmpfile
|
||||
proc32_64.execute_python("import windows")
|
||||
# A filename that a normal process should not have a handle on (to be sur)
|
||||
proc32_64.execute_python(r"""f = open(r"{filename}")""".format(filename=TEST_FILE_FOR_HANDLE))
|
||||
proc32_64.execute_python(r"""h = windows.utils.get_handle_from_file(f)""")
|
||||
with windows.pipe.create(PIPE_NAME) as np:
|
||||
proc32_64.execute_python("""windows.pipe.send_object("{pipe}", h)""".format(pipe=PIPE_NAME))
|
||||
file_handle_vlue = np.recv()
|
||||
remote_handle = [x for x in proc32_64.handles if x.value == file_handle_vlue][0]
|
||||
assert remote_handle.pid == proc32_64.pid
|
||||
assert remote_handle.type == "File"
|
||||
assert remote_handle.name.startswith("\Device\HarddiskVolume")
|
||||
assert remote_handle.name.endswith(TEST_FILE_FOR_HANDLE[2:]) # Remove volume letter
|
||||
assert remote_handle.infos
|
||||
+93
-37
@@ -4,15 +4,11 @@ import ctypes
|
||||
import windows
|
||||
from windows import winproxy
|
||||
from windows.generated_def import windef
|
||||
from windows.generated_def.winstructs import *
|
||||
|
||||
# Remove this ?
|
||||
class EPUBLIC_OBJECT_TYPE_INFORMATION(PUBLIC_OBJECT_TYPE_INFORMATION):
|
||||
pass
|
||||
import windows.generated_def as gdef
|
||||
|
||||
current_process_pid = os.getpid()
|
||||
|
||||
class Handle(SYSTEM_HANDLE):
|
||||
class BaseSystemHandle(object):
|
||||
"""A handle of the system"""
|
||||
@windows.utils.fixedpropety
|
||||
def process(self):
|
||||
@@ -23,6 +19,15 @@ class Handle(SYSTEM_HANDLE):
|
||||
# return [p for p in windows.system.processes if p.pid == self.dwProcessId][0]
|
||||
return windows.WinProcess(pid=self.dwProcessId)
|
||||
|
||||
@property
|
||||
def pid(self):
|
||||
return self.dwProcessId
|
||||
|
||||
@property
|
||||
def value(self):
|
||||
return self.wValue
|
||||
|
||||
|
||||
@windows.utils.fixedpropety
|
||||
def name(self):
|
||||
"""The name of the handle
|
||||
@@ -44,35 +49,33 @@ class Handle(SYSTEM_HANDLE):
|
||||
|
||||
def _get_object_name(self):
|
||||
lh = self.local_handle
|
||||
size_needed = DWORD()
|
||||
size_needed = gdef.DWORD()
|
||||
yyy = ctypes.c_buffer(0x1000)
|
||||
winproxy.NtQueryObject(lh, ObjectNameInformation, ctypes.byref(yyy), ctypes.sizeof(yyy), ctypes.byref(size_needed))
|
||||
return LSA_UNICODE_STRING.from_buffer_copy(yyy[:size_needed.value]).str
|
||||
winproxy.NtQueryObject(lh, gdef.ObjectNameInformation, ctypes.byref(yyy), ctypes.sizeof(yyy), ctypes.byref(size_needed))
|
||||
return gdef.LSA_UNICODE_STRING.from_buffer_copy(yyy[:size_needed.value]).str
|
||||
|
||||
def _get_object_type(self):
|
||||
lh = self.local_handle
|
||||
xxx = EPUBLIC_OBJECT_TYPE_INFORMATION()
|
||||
size_needed = DWORD()
|
||||
xxx = gdef.PUBLIC_OBJECT_TYPE_INFORMATION()
|
||||
size_needed = gdef.DWORD()
|
||||
try:
|
||||
winproxy.NtQueryObject(lh, ObjectTypeInformation, ctypes.byref(xxx), ctypes.sizeof(xxx), ctypes.byref(size_needed))
|
||||
winproxy.NtQueryObject(lh, gdef.ObjectTypeInformation, ctypes.byref(xxx), ctypes.sizeof(xxx), ctypes.byref(size_needed))
|
||||
except WindowsError as e:
|
||||
if e.code != STATUS_INFO_LENGTH_MISMATCH:
|
||||
# print("ERROR WITH {0:x}".format(lh))
|
||||
if e.code != gdef.STATUS_INFO_LENGTH_MISMATCH:
|
||||
raise
|
||||
size = size_needed.value
|
||||
buffer = ctypes.c_buffer(size)
|
||||
winproxy.NtQueryObject(lh, ObjectTypeInformation, buffer, size, ctypes.byref(size_needed))
|
||||
xxx = EPUBLIC_OBJECT_TYPE_INFORMATION.from_buffer_copy(buffer)
|
||||
winproxy.NtQueryObject(lh, gdef.ObjectTypeInformation, buffer, size, ctypes.byref(size_needed))
|
||||
xxx = gdef.PUBLIC_OBJECT_TYPE_INFORMATION.from_buffer_copy(buffer)
|
||||
return xxx.TypeName.str
|
||||
|
||||
def _get_object_basic_infos(self):
|
||||
pass
|
||||
lh = self.local_handle
|
||||
size_needed = DWORD()
|
||||
basic_infos = PUBLIC_OBJECT_BASIC_INFORMATION()
|
||||
winproxy.NtQueryObject(lh, ObjectBasicInformation, ctypes.byref(basic_infos), ctypes.sizeof(basic_infos), ctypes.byref(size_needed))
|
||||
size_needed = gdef.DWORD()
|
||||
basic_infos = gdef.PUBLIC_OBJECT_BASIC_INFORMATION()
|
||||
winproxy.NtQueryObject(lh, gdef.ObjectBasicInformation, ctypes.byref(basic_infos), ctypes.sizeof(basic_infos), ctypes.byref(size_needed))
|
||||
return basic_infos
|
||||
#PUBLIC_OBJECT_BASIC_INFORMATION
|
||||
|
||||
@windows.utils.fixedpropety
|
||||
def local_handle(self):
|
||||
@@ -81,8 +84,8 @@ class Handle(SYSTEM_HANDLE):
|
||||
:type: :class:`int`"""
|
||||
if self.dwProcessId == windows.current_process.pid:
|
||||
return self.wValue
|
||||
res = HANDLE()
|
||||
winproxy.DuplicateHandle(self.process.handle, self.wValue, windows.current_process.handle, ctypes.byref(res), dwOptions=DUPLICATE_SAME_ACCESS)
|
||||
res = gdef.HANDLE()
|
||||
winproxy.DuplicateHandle(self.process.handle, self.wValue, windows.current_process.handle, ctypes.byref(res), dwOptions=gdef.DUPLICATE_SAME_ACCESS)
|
||||
return res.value
|
||||
|
||||
def description(self):
|
||||
@@ -113,25 +116,78 @@ class Handle(SYSTEM_HANDLE):
|
||||
if hasattr(self, "_local_handle"):
|
||||
return winproxy.CloseHandle(self._local_handle)
|
||||
|
||||
class Handle(gdef.SYSTEM_HANDLE, BaseSystemHandle):
|
||||
pass
|
||||
|
||||
class HandleWow64(gdef.SYSTEM_HANDLE64, BaseSystemHandle):
|
||||
pass # For wow64 process
|
||||
|
||||
def enumerate_handles():
|
||||
size_needed = ULONG()
|
||||
size = 0x1000
|
||||
buffer = ctypes.c_buffer(size)
|
||||
|
||||
if windows.current_process.is_wow_64:
|
||||
return enumerate_handles_syswow64()
|
||||
size_needed = gdef.ULONG()
|
||||
# Should at least be sizeof(gdef.SYSTEM_HANDLE_INFORMATION)
|
||||
tmp_buffer = windows.utils.BUFFER(gdef.SYSTEM_HANDLE_INFORMATION)()
|
||||
try:
|
||||
winproxy.NtQuerySystemInformation(16, buffer, size, ReturnLength=ctypes.byref(size_needed))
|
||||
winproxy.NtQuerySystemInformation(gdef.SystemHandleInformation, tmp_buffer, tmp_buffer.real_size, ReturnLength=ctypes.byref(size_needed))
|
||||
except WindowsError as e:
|
||||
pass
|
||||
size = size_needed.value + 0x1000 # In case we have some more handle created
|
||||
buf = windows.utils.BUFFER(gdef.SYSTEM_HANDLE_INFORMATION)(size=size)
|
||||
size_needed.value = 0
|
||||
winproxy.NtQuerySystemInformation(gdef.SystemHandleInformation, buf, buf.real_size, ReturnLength=ctypes.byref(size_needed))
|
||||
handle_array = windows.utils.resized_array(buf[0].Handles, buf[0].HandleCount, Handle)
|
||||
return list(handle_array)
|
||||
|
||||
size = size_needed.value + 0x1000
|
||||
buffer = ctypes.c_buffer(size)
|
||||
winproxy.NtQuerySystemInformation(16, buffer, size, ReturnLength=ctypes.byref(size_needed))
|
||||
x = SYSTEM_HANDLE_INFORMATION.from_buffer(buffer)
|
||||
class _GENERATED_SYSTEM_HANDLE_INFORMATION(ctypes.Structure):
|
||||
_fields_ = [
|
||||
("HandleCount", ULONG),
|
||||
("Handles", Handle * x.HandleCount),
|
||||
]
|
||||
return list(_GENERATED_SYSTEM_HANDLE_INFORMATION.from_buffer_copy(buffer[:size_needed.value]).Handles)
|
||||
|
||||
def enumerate_handles_syswow64():
|
||||
size_needed = gdef.ULONG()
|
||||
# Should at least be sizeof(gdef.SYSTEM_HANDLE_INFORMATION)
|
||||
tmp_buffer = windows.utils.BUFFER(gdef.SYSTEM_HANDLE_INFORMATION64)()
|
||||
try:
|
||||
windows.syswow64.NtQuerySystemInformation_32_to_64(gdef.SystemHandleInformation, tmp_buffer, tmp_buffer.real_size, ReturnLength=ctypes.byref(size_needed))
|
||||
except WindowsError as e:
|
||||
pass
|
||||
size = size_needed.value + 0x1000 # In case we have some more handle created
|
||||
buf = windows.utils.BUFFER(gdef.SYSTEM_HANDLE_INFORMATION64)(size=size)
|
||||
size_needed.value = 0
|
||||
windows.syswow64.NtQuerySystemInformation_32_to_64(gdef.SystemHandleInformation, buf, buf.real_size, ReturnLength=ctypes.byref(size_needed))
|
||||
handle_array = windows.utils.resized_array(buf[0].Handles, buf[0].HandleCount, HandleWow64)
|
||||
return list(handle_array)
|
||||
|
||||
|
||||
def enumerate_type():
|
||||
"WIP: DO NOT USE"
|
||||
size_needed = DWORD()
|
||||
fsize = 8
|
||||
fbuffer = ctypes.c_buffer(fsize)
|
||||
try:
|
||||
winproxy.NtQueryObject(None, gdef.ObjectTypesInformation, fbuffer, fsize, ctypes.byref(size_needed))
|
||||
except WindowsError as e:
|
||||
if e.code != STATUS_INFO_LENGTH_MISMATCH:
|
||||
raise
|
||||
else:
|
||||
# We had enought memory ?
|
||||
return
|
||||
|
||||
# Looks like the Wow64 syscall emulation is broken :D
|
||||
# It write AFTER the buffer if we are a wow64 process :D
|
||||
# So better allocate a standalone buffer (triggering a ACCESS_VIOLATION) that corrupting the heap
|
||||
# This is a worst case scenario, as we allocation more space it should not happen !
|
||||
size = size_needed.value + 0x200
|
||||
size_needed.value = 0
|
||||
|
||||
with windows.current_process.allocated_memory(size, gdef.PAGE_READWRITE) as buffer_base:
|
||||
winproxy.NtQueryObject(None, gdef.ObjectTypesInformation, buffer_base, size, ctypes.byref(size_needed))
|
||||
# Cache some exceptions ?
|
||||
# Parse the buffer data in-place as string are addr-dependant
|
||||
types_info = gdef.OBJECT_TYPES_INFORMATION.from_address(buffer_base)
|
||||
offset = ctypes.sizeof(gdef.PVOID) # Looks like the size of the struct is PTR aligned as the struct is follower by other stuff
|
||||
for i in range(types_info.NumberOfTypes):
|
||||
info = gdef.PUBLIC_OBJECT_TYPE_INFORMATION.from_address(buffer_base + offset)
|
||||
yield info
|
||||
offset += ctypes.sizeof(gdef.PUBLIC_OBJECT_TYPE_INFORMATION) + info.TypeName.MaximumLength
|
||||
if offset % ctypes.sizeof(gdef.PVOID):
|
||||
offset += ctypes.sizeof(gdef.PVOID) - (offset % ctypes.sizeof(gdef.PVOID))
|
||||
# End-of ctx-manager
|
||||
return
|
||||
|
||||
Reference in New Issue
Block a user