mirror of
https://github.com/hakril/PythonForWindows
synced 2026-06-08 14:31:45 +00:00
Testing hook without generated stub + new hook test in new file
This commit is contained in:
Clement Rouault
@@ -8,4 +8,6 @@ Since 0.2:
|
||||
* Object returned by `windows.native_exec.create_function` has an attribute `code_addr` with the address of the executable code
|
||||
* add `windows.winproxy.is_implemented` Ex: windows.winproxy.is_implemented(windows.winproxy.QueryWorkingSetEx)
|
||||
* registry.py handle REG_QWORD manually (_winreg does not)
|
||||
* CurrentProcessReadSyswow doest not use ``current_process.handle`` anymore but ``OpenProcess(current_process.pid)`` (Compat windows10 where method 1 doest not work)
|
||||
* CurrentProcessReadSyswow doest not use ``current_process.handle`` anymore but ``OpenProcess(current_process.pid)`` (Compat windows10 where method 1 doest not work)
|
||||
* Add: system.handles (winobject\handle.py)
|
||||
* You can have multiple execute_python_unsafe at the same time in the same process (didn't know: consequence of new injection code)
|
||||
@@ -14,6 +14,7 @@ TODO:
|
||||
|
||||
- Threading
|
||||
- Quid IAT hook stub ? just einit threads and remove this ?
|
||||
- Continue test with new generate_callback_stub and remove commented code if it works
|
||||
|
||||
- Injection
|
||||
- code generated by generate_python_exec_shellcode_64[32] may be reused
|
||||
|
||||
@@ -115,6 +115,7 @@ def load_dll_in_remote_process(target, dll_name):
|
||||
# Hardcore mode
|
||||
# We don't have k32 or PEB->Ldr
|
||||
# Go inject a GetProcAddress(LoadLib) + LoadLib shellcode :D
|
||||
dbgprint("DLL Via manual getproc / loadlib", "DLLINJECT")
|
||||
if target.bitness == 32:
|
||||
return perform_manual_getproc_loadlib_32(target, dll_name)
|
||||
return perform_manual_getproc_loadlib_64(target, dll_name)
|
||||
|
||||
@@ -271,11 +271,15 @@ def generate_stub_64(callback):
|
||||
def generate_callback_stub(callback, types):
|
||||
func_type = ctypes.WINFUNCTYPE(*types)
|
||||
c_callable = func_type(callback)
|
||||
if windows.current_process.bitness == 32:
|
||||
stub = generate_stub_32(c_callable)
|
||||
else:
|
||||
stub = generate_stub_64(c_callable)
|
||||
stub_addr = allocator.write_code(stub.get_code())
|
||||
|
||||
|
||||
stub = c_callable
|
||||
stub_addr = ctypes.cast(c_callable, ctypes.c_void_p).value
|
||||
# if windows.current_process.bitness == 32:
|
||||
# stub = generate_stub_32(c_callable)
|
||||
# else:
|
||||
# stub = generate_stub_64(c_callable)
|
||||
# stub_addr = allocator.write_code(stub.get_code())
|
||||
generate_callback_stub.l.append((stub, c_callable))
|
||||
return stub_addr
|
||||
|
||||
|
||||
@@ -1,3 +1,8 @@
|
||||
from mytest import WindowsTestCase, WindowsAPITestCase, DebuggerTestCase, NativeUtilsTestCase, SystemTestCase, pop_calc_32, pop_calc_64, Calc32, Calc64
|
||||
|
||||
__all__ = ["SystemTestCase", "WindowsTestCase", "WindowsAPITestCase", "DebuggerTestCase", "NativeUtilsTestCase"]
|
||||
from test_utils import *
|
||||
|
||||
from mytest import WindowsTestCase, WindowsAPITestCase, DebuggerTestCase, NativeUtilsTestCase, SystemTestCase
|
||||
from test_hooks import HookTestCase
|
||||
|
||||
|
||||
__all__ = ["SystemTestCase", "WindowsTestCase", "WindowsAPITestCase", "DebuggerTestCase", "NativeUtilsTestCase", "HookTestCase"]
|
||||
|
||||
+2
-96
@@ -4,70 +4,10 @@ import time
|
||||
import os
|
||||
import textwrap
|
||||
import random
|
||||
from contextlib import contextmanager
|
||||
|
||||
sys.path.append(".")
|
||||
import unittest
|
||||
import windows
|
||||
import windows.debug
|
||||
import windows.native_exec.simple_x86 as x86
|
||||
import windows.native_exec.simple_x64 as x64
|
||||
import windows.native_exec.nativeutils as nativeutils
|
||||
|
||||
from test_utils import *
|
||||
from windows.generated_def.winstructs import *
|
||||
|
||||
from windows.native_exec.nativeutils import GetProcAddress64, GetProcAddress32
|
||||
|
||||
|
||||
is_process_32_bits = windows.current_process.bitness == 32
|
||||
is_process_64_bits = windows.current_process.bitness == 64
|
||||
|
||||
is_windows_32_bits = windows.system.bitness == 32
|
||||
is_windows_64_bits = windows.system.bitness == 64
|
||||
|
||||
windows_32bit_only = unittest.skipIf(not is_windows_32_bits, "Test for 32bits Kernel only")
|
||||
windows_64bit_only = unittest.skipIf(not is_windows_64_bits, "Test for 64bits Kernel only")
|
||||
|
||||
process_32bit_only = unittest.skipIf(not is_process_32_bits, "Test for 32bits process only")
|
||||
process_64bit_only = unittest.skipIf(not is_process_64_bits, "Test for 64bits process only")
|
||||
|
||||
|
||||
if is_windows_32_bits:
|
||||
def pop_calc_32(dwCreationFlags=0):
|
||||
return windows.utils.create_process(r"C:\Windows\system32\calc.exe", dwCreationFlags=dwCreationFlags, show_windows=True)
|
||||
|
||||
def pop_calc_64(dwCreationFlags=0):
|
||||
raise WindowsError("Cannot create calc64 in 32bits system")
|
||||
else:
|
||||
def pop_calc_32(dwCreationFlags=0):
|
||||
return windows.utils.create_process(r"C:\Windows\syswow64\calc.exe", dwCreationFlags=dwCreationFlags, show_windows=True)
|
||||
|
||||
if is_process_32_bits:
|
||||
def pop_calc_64(dwCreationFlags=0):
|
||||
with windows.utils.DisableWow64FsRedirection():
|
||||
return windows.utils.create_process(r"C:\Windows\system32\calc.exe", dwCreationFlags=dwCreationFlags, show_windows=True)
|
||||
else:
|
||||
def pop_calc_64(dwCreationFlags=0):
|
||||
return windows.utils.create_process(r"C:\Windows\system32\calc.exe", dwCreationFlags=dwCreationFlags, show_windows=True)
|
||||
|
||||
|
||||
@contextmanager
|
||||
def Calc64(dwCreationFlags=0, exit_code=0):
|
||||
try:
|
||||
calc = pop_calc_64(dwCreationFlags)
|
||||
yield calc
|
||||
finally:
|
||||
if "calc" in locals():
|
||||
calc.exit(exit_code)
|
||||
|
||||
@contextmanager
|
||||
def Calc32(dwCreationFlags=0, exit_code=0):
|
||||
try:
|
||||
calc = pop_calc_32(dwCreationFlags)
|
||||
yield calc
|
||||
finally:
|
||||
if "calc" in locals():
|
||||
calc.exit(exit_code)
|
||||
|
||||
class SystemTestCase(unittest.TestCase):
|
||||
def test_version(self):
|
||||
@@ -307,40 +247,6 @@ class WindowsTestCase(unittest.TestCase):
|
||||
dword = struct.unpack("<Q", calc.read_memory(data, 8))[0]
|
||||
self.assertEqual(dword, get_current_proc_id)
|
||||
|
||||
def test_self_iat_hook_sucess(self):
|
||||
pythondll_mod = [m for m in windows.current_process.peb.modules if m.name.startswith("python") and m.name.endswith(".dll")][0]
|
||||
RegOpenKeyExA = [n for n in pythondll_mod.pe.imports['advapi32.dll'] if n.name == "RegOpenKeyExA"][0]
|
||||
|
||||
hook_value = []
|
||||
|
||||
@windows.hooks.RegOpenKeyExACallback
|
||||
def open_reg_hook(hKey, lpSubKey, ulOptions, samDesired, phkResult, real_function):
|
||||
hook_value.append((hKey, lpSubKey.value))
|
||||
phkResult[0] = 12345678
|
||||
return 0
|
||||
|
||||
RegOpenKeyExA.set_hook(open_reg_hook)
|
||||
import _winreg
|
||||
open_args = (0x12345678, "MY_KEY_VALUE")
|
||||
k = _winreg.OpenKey(*open_args)
|
||||
self.assertEqual(k.handle, 12345678)
|
||||
self.assertEqual(hook_value[0], open_args)
|
||||
|
||||
def test_self_iat_hook_fail_return(self):
|
||||
pythondll_mod = [m for m in windows.current_process.peb.modules if m.name.startswith("python") and m.name.endswith(".dll")][0]
|
||||
RegOpenKeyExA = [n for n in pythondll_mod.pe.imports['advapi32.dll'] if n.name == "RegOpenKeyExA"][0]
|
||||
|
||||
@windows.hooks.RegOpenKeyExACallback
|
||||
def open_reg_hook_fail(hKey, lpSubKey, ulOptions, samDesired, phkResult, real_function):
|
||||
return 0x11223344
|
||||
|
||||
RegOpenKeyExA.set_hook(open_reg_hook_fail)
|
||||
import _winreg
|
||||
open_args = (0x12345678, "MY_KEY_VALUE")
|
||||
with self.assertRaises(WindowsError) as ar:
|
||||
_winreg.OpenKey(*open_args)
|
||||
self.assertEqual(ar.exception.winerror, 0x11223344)
|
||||
|
||||
def test_thread_exit_value_32(self):
|
||||
with Calc32() as calc:
|
||||
res = calc.execute_python("import time;time.sleep(0.1); 2")
|
||||
@@ -886,7 +792,7 @@ class DebuggerTestCase(unittest.TestCase):
|
||||
code += x86.Pop("ECX")
|
||||
code += x86.Pop("ECX")
|
||||
code += x86.Ret()
|
||||
RemoteManualLoadLibray += GetProcAddress32
|
||||
RemoteManualLoadLibray += nativeutils.GetProcAddress32
|
||||
|
||||
addr = target.virtual_alloc(0x1000)
|
||||
addr2 = addr + len(dll)
|
||||
|
||||
@@ -0,0 +1,122 @@
|
||||
import ctypes
|
||||
import textwrap
|
||||
|
||||
from test_utils import *
|
||||
from windows.generated_def.winstructs import *
|
||||
|
||||
|
||||
class HookTestCase(unittest.TestCase):
|
||||
|
||||
def test_self_iat_hook_success(self):
|
||||
"""Test hook success in single(self) thread"""
|
||||
pythondll_mod = [m for m in windows.current_process.peb.modules if m.name.startswith("python") and m.name.endswith(".dll")][0]
|
||||
RegOpenKeyExA = [n for n in pythondll_mod.pe.imports['advapi32.dll'] if n.name == "RegOpenKeyExA"][0]
|
||||
|
||||
hook_value = []
|
||||
|
||||
@windows.hooks.RegOpenKeyExACallback
|
||||
def open_reg_hook(hKey, lpSubKey, ulOptions, samDesired, phkResult, real_function):
|
||||
hook_value.append((hKey, lpSubKey.value))
|
||||
phkResult[0] = 12345678
|
||||
return 0
|
||||
|
||||
RegOpenKeyExA.set_hook(open_reg_hook)
|
||||
import _winreg
|
||||
open_args = (0x12345678, "MY_KEY_VALUE")
|
||||
k = _winreg.OpenKey(*open_args)
|
||||
self.assertEqual(k.handle, 12345678)
|
||||
self.assertEqual(hook_value[0], open_args)
|
||||
|
||||
def test_self_iat_hook_fail_return(self):
|
||||
"""Test hook fail in single(self) thread"""
|
||||
pythondll_mod = [m for m in windows.current_process.peb.modules if m.name.startswith("python") and m.name.endswith(".dll")][0]
|
||||
RegOpenKeyExA = [n for n in pythondll_mod.pe.imports['advapi32.dll'] if n.name == "RegOpenKeyExA"][0]
|
||||
|
||||
@windows.hooks.RegOpenKeyExACallback
|
||||
def open_reg_hook_fail(hKey, lpSubKey, ulOptions, samDesired, phkResult, real_function):
|
||||
return 0x11223344
|
||||
|
||||
RegOpenKeyExA.set_hook(open_reg_hook_fail)
|
||||
import _winreg
|
||||
open_args = (0x12345678, "MY_KEY_VALUE")
|
||||
with self.assertRaises(WindowsError) as ar:
|
||||
_winreg.OpenKey(*open_args)
|
||||
self.assertEqual(ar.exception.winerror, 0x11223344)
|
||||
|
||||
|
||||
def test_self_iat_hook_multithread(self):
|
||||
"""Test IAT hook in current process with multi thread trigger"""
|
||||
cp = windows.current_process
|
||||
# Might change this to XP compat ?
|
||||
kernelbase_mod = [m for m in cp.peb.modules if m.name == "kernelbase.dll"][0]
|
||||
LdrLoadDll = [n for n in kernelbase_mod.pe.imports['ntdll.dll'] if n.name == "LdrLoadDll"][0]
|
||||
|
||||
calling_thread = set([])
|
||||
@windows.hooks.LdrLoadDllCallback
|
||||
def MyHook(*args, **kwargs):
|
||||
calling_thread.add(windows.current_thread.tid)
|
||||
return kwargs["real_function"]()
|
||||
|
||||
LdrLoadDll.set_hook(MyHook)
|
||||
# Trigger from local thread
|
||||
ctypes.WinDLL("kernel32.dll")
|
||||
self.assertEqual(calling_thread, set([windows.current_thread.tid]))
|
||||
# Trigger from another thread
|
||||
k32 = [m for m in cp.peb.modules if m.name == "kernel32.dll"][0]
|
||||
load_libraryA = k32.pe.exports["LoadLibraryA"]
|
||||
with cp.allocated_memory(0x1000) as addr:
|
||||
cp.write_memory(addr, "DLLNOTFOUND.NOT_A_REAL_DLL" + "\x00")
|
||||
t = cp.create_thread(load_libraryA, addr)
|
||||
self.assertEqual(len(calling_thread), 2)
|
||||
|
||||
def test_remote_iat_hook_32(self):
|
||||
with Calc32() as calc:
|
||||
calc.execute_python("import windows")
|
||||
calc.execute_python("windows.utils.create_console()")
|
||||
|
||||
code = """
|
||||
import windows.generated_def as gdef
|
||||
|
||||
cp = windows.current_process
|
||||
kernelbase_mod = [m for m in cp.peb.modules if m.name == "kernelbase.dll"][0]
|
||||
LdrLoadDll = [n for n in kernelbase_mod.pe.imports['ntdll.dll'] if n.name == "LdrLoadDll"][0]
|
||||
|
||||
calling_thread = set([])
|
||||
hooking_thread = windows.current_thread.tid
|
||||
@windows.hooks.LdrLoadDllCallback
|
||||
def MyHook(*args, **kwargs):
|
||||
calling_thread.add(windows.current_thread.tid)
|
||||
print(windows.current_thread.tid)
|
||||
return kwargs["real_function"]()
|
||||
|
||||
LdrLoadDll.set_hook(MyHook)
|
||||
print("Hooker = " + str(windows.current_thread.tid))
|
||||
import ctypes
|
||||
try:
|
||||
ctypes.WinDLL("NOT_A_REAL_DLL")
|
||||
except WindowsError as e:
|
||||
pass
|
||||
"""
|
||||
calc.execute_python_unsafe(textwrap.dedent(code))
|
||||
# Tricky part: we use an injected thread exit_value to ask stuff about the remote python
|
||||
def remote_ask(request):
|
||||
t = calc.execute_python_unsafe(request)
|
||||
t.wait()
|
||||
return t.exit_code
|
||||
|
||||
self.assertEqual(remote_ask("windows.current_thread.exit(len(calling_thread))"), 1)
|
||||
self.assertEqual(remote_ask("windows.current_thread.exit(calling_thread == set([hooking_thread]))"), 1)
|
||||
|
||||
# Trigger hook from another Python thread
|
||||
calc.execute_python_unsafe("ctypes.WinDLL('ANOTHER_FAKE_DLL')").wait()
|
||||
self.assertEqual(remote_ask("windows.current_thread.exit(len(calling_thread))"), 2)
|
||||
|
||||
# Trigger hook from a NONPython thread
|
||||
k32 = [m for m in calc.peb.modules if m.name == "kernel32.dll"][0]
|
||||
load_libraryA = k32.pe.exports["LoadLibraryA"]
|
||||
with calc.allocated_memory(0x1000) as addr:
|
||||
calc.write_memory(addr, "DLLNOTFOUND.NOT_A_REAL_DLL" + "\x00")
|
||||
t = calc.create_thread(load_libraryA, addr)
|
||||
self.assertEqual(remote_ask("windows.current_thread.exit(len(calling_thread))"), 3)
|
||||
|
||||
|
||||
@@ -0,0 +1,65 @@
|
||||
from contextlib import contextmanager
|
||||
|
||||
import unittest
|
||||
import windows
|
||||
import windows.debug
|
||||
import windows.native_exec.simple_x86 as x86
|
||||
import windows.native_exec.simple_x64 as x64
|
||||
import windows.native_exec.nativeutils as nativeutils
|
||||
|
||||
|
||||
is_process_32_bits = windows.current_process.bitness == 32
|
||||
is_process_64_bits = windows.current_process.bitness == 64
|
||||
|
||||
is_windows_32_bits = windows.system.bitness == 32
|
||||
is_windows_64_bits = windows.system.bitness == 64
|
||||
|
||||
windows_32bit_only = unittest.skipIf(not is_windows_32_bits, "Test for 32bits Kernel only")
|
||||
windows_64bit_only = unittest.skipIf(not is_windows_64_bits, "Test for 64bits Kernel only")
|
||||
|
||||
process_32bit_only = unittest.skipIf(not is_process_32_bits, "Test for 32bits process only")
|
||||
process_64bit_only = unittest.skipIf(not is_process_64_bits, "Test for 64bits process only")
|
||||
|
||||
|
||||
if is_windows_32_bits:
|
||||
def pop_calc_32(dwCreationFlags=0):
|
||||
return windows.utils.create_process(r"C:\Windows\system32\calc.exe", dwCreationFlags=dwCreationFlags, show_windows=True)
|
||||
|
||||
def pop_calc_64(dwCreationFlags=0):
|
||||
raise WindowsError("Cannot create calc64 in 32bits system")
|
||||
else:
|
||||
def pop_calc_32(dwCreationFlags=0):
|
||||
return windows.utils.create_process(r"C:\Windows\syswow64\calc.exe", dwCreationFlags=dwCreationFlags, show_windows=True)
|
||||
|
||||
if is_process_32_bits:
|
||||
def pop_calc_64(dwCreationFlags=0):
|
||||
with windows.utils.DisableWow64FsRedirection():
|
||||
return windows.utils.create_process(r"C:\Windows\system32\calc.exe", dwCreationFlags=dwCreationFlags, show_windows=True)
|
||||
else:
|
||||
def pop_calc_64(dwCreationFlags=0):
|
||||
return windows.utils.create_process(r"C:\Windows\system32\calc.exe", dwCreationFlags=dwCreationFlags, show_windows=True)
|
||||
|
||||
|
||||
@contextmanager
|
||||
def Calc64(dwCreationFlags=0, exit_code=0):
|
||||
try:
|
||||
calc = pop_calc_64(dwCreationFlags)
|
||||
yield calc
|
||||
except Exception as e:
|
||||
print(e)
|
||||
raise
|
||||
finally:
|
||||
if "calc" in locals():
|
||||
calc.exit(exit_code)
|
||||
|
||||
@contextmanager
|
||||
def Calc32(dwCreationFlags=0, exit_code=0):
|
||||
try:
|
||||
calc = pop_calc_32(dwCreationFlags)
|
||||
yield calc
|
||||
except Exception as e:
|
||||
print(e)
|
||||
raise
|
||||
finally:
|
||||
if "calc" in locals():
|
||||
calc.exit(exit_code)
|
||||
Reference in New Issue
Block a user