From 89e486cd9557c08673944e27b612347bac9dcc7c Mon Sep 17 00:00:00 2001 From: Clement Rouault Date: Wed, 6 Sep 2017 18:42:54 +0200 Subject: [PATCH] Working on rewriting of tests with pytest --- pytesting/__init__.py | 0 pytesting/conftest.py | 133 ++++++++ pytesting/dbg_injection.py | 75 ++++ pytesting/pfwtest.py | 27 ++ pytesting/pytest.ini | 2 + pytesting/test_crypto.py | 104 ++++++ pytesting/test_debugger.py | 567 +++++++++++++++++++++++++++++++ pytesting/test_generated_code.py | 41 +++ pytesting/test_hook.py | 190 +++++++++++ pytesting/test_native_utils.py | 64 ++++ pytesting/test_process.py | 366 ++++++++++++++++++++ pytesting/test_system.py | 35 ++ pytesting/test_syswow.py | 73 ++++ pytesting/test_winproxy.py | 13 + 14 files changed, 1690 insertions(+) create mode 100644 pytesting/__init__.py create mode 100644 pytesting/conftest.py create mode 100644 pytesting/dbg_injection.py create mode 100644 pytesting/pfwtest.py create mode 100644 pytesting/pytest.ini create mode 100644 pytesting/test_crypto.py create mode 100644 pytesting/test_debugger.py create mode 100644 pytesting/test_generated_code.py create mode 100644 pytesting/test_hook.py create mode 100644 pytesting/test_native_utils.py create mode 100644 pytesting/test_process.py create mode 100644 pytesting/test_system.py create mode 100644 pytesting/test_syswow.py create mode 100644 pytesting/test_winproxy.py diff --git a/pytesting/__init__.py b/pytesting/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/pytesting/conftest.py b/pytesting/conftest.py new file mode 100644 index 0000000..0edeb06 --- /dev/null +++ b/pytesting/conftest.py @@ -0,0 +1,133 @@ +import gc +import pytest + +import windows +import windows.generated_def as gdef + +from pfwtest import is_windows_32_bits, is_process_32_bits, test_binary_name, DEFAULT_CREATION_FLAGS + + +if is_windows_32_bits: + def pop_proc_32(dwCreationFlags=DEFAULT_CREATION_FLAGS): + return windows.utils.create_process(r"C:\Windows\system32\{0}".format(test_binary_name), dwCreationFlags=dwCreationFlags, show_windows=True) + + def pop_proc_64(dwCreationFlags=DEFAULT_CREATION_FLAGS): + raise WindowsError("Cannot create calc64 in 32bits system") +else: + def pop_proc_32(dwCreationFlags=DEFAULT_CREATION_FLAGS): + return windows.utils.create_process(r"C:\Windows\syswow64\{0}".format(test_binary_name), dwCreationFlags=dwCreationFlags, show_windows=True) + + if is_process_32_bits: + def pop_proc_64(dwCreationFlags=DEFAULT_CREATION_FLAGS): + with windows.utils.DisableWow64FsRedirection(): + return windows.utils.create_process(r"C:\Windows\system32\{0}".format(test_binary_name), dwCreationFlags=dwCreationFlags, show_windows=True) + else: + def pop_proc_64(dwCreationFlags=DEFAULT_CREATION_FLAGS): + return windows.utils.create_process(r"C:\Windows\system32\{0}".format(test_binary_name), dwCreationFlags=dwCreationFlags, show_windows=True) + + +import sys + +def generate_pop_and_exit_fixtures(proc_popers, ids=[], dwCreationFlags=DEFAULT_CREATION_FLAGS): + @pytest.fixture(params=proc_popers, ids=ids) + def pop_and_exit_process(request): + proc_poper = request.param + proc = proc_poper(dwCreationFlags=dwCreationFlags) + yield proc # provide the fixture value + try: + print("EXIT PROC <{0}>".format(sys.getrefcount(proc))) + # if sys.getrefcount(proc) > 5: + # import pdb;pdb.set_trace() + proc.exit(0) + except WindowsError as e: + if not proc.is_exit: + raise + # import pdb;pdb.set_trace() + # proc.__del__() + del proc + return pop_and_exit_process + +proc32 = generate_pop_and_exit_fixtures([pop_proc_32], ids=["proc32"]) +proc64 = generate_pop_and_exit_fixtures([pop_proc_64], ids=["proc64"]) +if is_windows_32_bits: + proc32_64 = generate_pop_and_exit_fixtures([pop_proc_32], ids=["proc32"]) + proc32_64_suspended = generate_pop_and_exit_fixtures([pop_proc_32], ids=["proc32"], + dwCreationFlags=gdef.CREATE_SUSPENDED) +else: + proc32_64 = generate_pop_and_exit_fixtures([pop_proc_32, pop_proc_64], ids=["proc32", "proc64"]) + proc32_64_suspended = generate_pop_and_exit_fixtures([pop_proc_32, pop_proc_64], ids=["proc32", "proc64"], + dwCreationFlags=gdef.CREATE_SUSPENDED) + + +@pytest.fixture() +def check_for_gc_garbage(request): + garbage_before = set(gc.garbage) + yield + gc.collect() + new_garbage = set(gc.garbage) - garbage_before + assert not new_garbage, "Test generated uncollectable object ({0})".format(new_garbage) + +class HandleDebugger(object): + def __init__(self, pid): + self.pid = pid + self.handles = [] + + def refresh_handles(self): + self.handles = self.get_handles() + + def get_handles(self): + tpid = self.pid + return [h for h in windows.system.handles if h.dwProcessId == tpid] + + def get_new_handle(self, old_handles=None): + nh = self.get_handles() + if old_handles is None: + old_handles = self.handles + handle_diff = set(h.wValue for h in nh) - set(h.wValue for h in old_handles) + return [h for h in nh if h.wValue in handle_diff] + + def handles_types(self, hlist): + return set(h.type for h in hlist) + + def print_new_handle_type(self): + print(self.handles_types(self.get_new_handle())) + + +current_process_hdebugger = HandleDebugger(windows.current_process.pid) +current_process_hdebugger.refresh_handles() + +# TST = current_process_hdebugger.refresh_handles() + +RESULT = {} + +@pytest.fixture() +def check_for_handle_leak(request): + # current_process_hdebugger.refresh_handles() + yield + # leaked_handles = current_process_hdebugger.get_new_handle() + # for lh in leaked_handles: + # RESULT[lh.wValue] = request.function.__name__ + print("HANDLE LEAK SAVE") + # assert not leaked_handles, "Test Leaked <{0}> handles of types ({1})".format(len(leaked_handles), set(h.type for h in leaked_handles)) + + +@pytest.fixture(scope='session') +def check_for_handle_leak_final(request): + # x = current_process_hdebugger.get_handles() + print("CHECK HANDLE FINAL :D") + # current_process_hdebugger.refresh_handles() + yield + # leaked_handles = current_process_hdebugger.get_new_handle(x) + # import pdb;pdb.set_trace() + # print(leaked_handles) + + # leaked_handles = current_process_hdebugger.get_new_handle() + # import pdb;pdb.set_trace() + # assert not leaked_handles, "Test Leaked <{0}> handles of types ({1})".format(len(leaked_handles), set(h.type for h in leaked_handles)) + + +def pytest_unconfigure(*args, **kwargs): + import pdb;pdb.set_trace() + print(TST) + +# pytestmark = pytest.mark.usefixtures('check_for_handle_leak_final') \ No newline at end of file diff --git a/pytesting/dbg_injection.py b/pytesting/dbg_injection.py new file mode 100644 index 0000000..38c6d79 --- /dev/null +++ b/pytesting/dbg_injection.py @@ -0,0 +1,75 @@ +import windows.native_exec.simple_x86 as x86 +import windows.native_exec.simple_x64 as x64 +from windows.native_exec import nativeutils + +def perform_manual_getproc_loadlib_32_for_dbg(target, dll_name): + dll = "KERNEL32.DLL\x00".encode("utf-16-le") + api = "LoadLibraryA\x00" + dll_to_load = dll_name + "\x00" + + RemoteManualLoadLibray = x86.MultipleInstr() + code = RemoteManualLoadLibray + code += x86.Mov("ECX", x86.mem("[ESP + 4]")) + code += x86.Push(x86.mem("[ECX + 4]")) + code += x86.Push(x86.mem("[ECX]")) + code += x86.Call(":FUNC_GETPROCADDRESS32") + code += x86.Push(x86.mem("[ECX + 8]")) + code += x86.Call("EAX") # LoadLibrary + code += x86.Pop("ECX") + code += x86.Pop("ECX") + code += x86.Ret() + RemoteManualLoadLibray += nativeutils.GetProcAddress32 + + addr = target.virtual_alloc(0x1000) + addr2 = addr + len(dll) + addr3 = addr2 + len(api) + addr4 = addr3 + len(dll_to_load) + target.write_memory(addr, dll) + target.write_memory(addr2, api) + target.write_memory(addr3, dll_to_load) + target.write_qword(addr4, addr) + target.write_qword(addr4 + 4, addr2) + target.write_qword(addr4 + 0x8, addr3) + t = target.execute(RemoteManualLoadLibray.get_code(), addr4) + return t + +def perform_manual_getproc_loadlib_64_for_dbg(target, dll_name): + dll = "KERNEL32.DLL\x00".encode("utf-16-le") + api = "LoadLibraryA\x00" + dll_to_load = dll_name + "\x00" + + RemoteManualLoadLibray = x64.MultipleInstr() + code = RemoteManualLoadLibray + code += x64.Mov("R15", "RCX") + code += x64.Mov("RCX", x64.mem("[R15 + 0]")) + code += x64.Mov("RDX", x64.mem("[R15 + 8]")) + code += x64.Call(":FUNC_GETPROCADDRESS64") + code += x64.Mov("RCX", x64.mem("[R15 + 0x10]")) + code += x64.Push("RCX") + code += x64.Push("RCX") + code += x64.Push("RCX") + code += x64.Call("RAX") # LoadLibrary + code += x64.Pop("RCX") + code += x64.Pop("RCX") + code += x64.Pop("RCX") + code += x64.Ret() + RemoteManualLoadLibray += nativeutils.GetProcAddress64 + + addr = target.virtual_alloc(0x1000) + addr2 = addr + len(dll) + addr3 = addr2 + len(api) + addr4 = addr3 + len(dll_to_load) + + target.write_memory(addr, dll) + target.write_memory(addr2, api) + target.write_memory(addr3, dll_to_load) + target.write_qword(addr4, addr) + target.write_qword(addr4 + 8, addr2) + target.write_qword(addr4 + 0x10, addr3) + t = target.execute(RemoteManualLoadLibray.get_code(), addr4) + return t + +def perform_manual_getproc_loadlib_for_dbg(target, *args, **kwargs): + if target.bitness == 32: + return perform_manual_getproc_loadlib_32_for_dbg(target, *args, **kwargs) + return perform_manual_getproc_loadlib_64_for_dbg(target, *args, **kwargs) \ No newline at end of file diff --git a/pytesting/pfwtest.py b/pytesting/pfwtest.py new file mode 100644 index 0000000..267bd45 --- /dev/null +++ b/pytesting/pfwtest.py @@ -0,0 +1,27 @@ +import pytest +import windows + +import windows.generated_def as gdef + +is_process_32_bits = windows.current_process.bitness == 32 +is_process_64_bits = windows.current_process.bitness == 64 +is_process_syswow = windows.current_process.is_wow_64 + +is_windows_32_bits = windows.system.bitness == 32 +is_windows_64_bits = windows.system.bitness == 64 + +is_windows_10 = (windows.system.version[0] == 10) + +windows_32bit_only = pytest.mark.skipif(not is_windows_32_bits, reason="Test for 32bits Kernel only") +windows_64bit_only = pytest.mark.skipif(not is_windows_64_bits, reason="Test for 64bits Kernel only") + +process_32bit_only = pytest.mark.skipif(not is_process_32_bits, reason="Test for 32bits process only") +process_64bit_only = pytest.mark.skipif(not is_process_64_bits, reason="Test for 64bits process only") +process_syswow_only = pytest.mark.skipif(not is_process_syswow, reason="Test for syswow process only") + + +check_for_gc_garbage = pytest.mark.usefixtures("check_for_gc_garbage") +check_for_handle_leak = pytest.mark.usefixtures("check_for_handle_leak") + +test_binary_name = "notepad.exe" +DEFAULT_CREATION_FLAGS = gdef.CREATE_NEW_CONSOLE \ No newline at end of file diff --git a/pytesting/pytest.ini b/pytesting/pytest.ini new file mode 100644 index 0000000..8c12560 --- /dev/null +++ b/pytesting/pytest.ini @@ -0,0 +1,2 @@ +[pytest] +usefixtures = check_for_handle_leak_final \ No newline at end of file diff --git a/pytesting/test_crypto.py b/pytesting/test_crypto.py new file mode 100644 index 0000000..33eb64a --- /dev/null +++ b/pytesting/test_crypto.py @@ -0,0 +1,104 @@ +import pytest + +import windows.crypto +import windows.generated_def as gdef + +from pfwtest import * + +pytestmark = pytest.mark.usefixtures('check_for_gc_garbage') + +TEST_CERT = """ +MIIBwTCCASqgAwIBAgIQG46Uyws+67ZBOfPJCbFrRjANBgkqhkiG9w0BAQsFADAfMR0wGwYDVQQD +ExRQeXRob25Gb3JXaW5kb3dzVGVzdDAeFw0xNzA0MTIxNDM5MjNaFw0xODA0MTIyMDM5MjNaMB8x +HTAbBgNVBAMTFFB5dGhvbkZvcldpbmRvd3NUZXN0MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKB +gQCRHwC/sRfXh5pc4poc85aidrudbPdya+0OeonQlf1JQ1ekf7KSfADV5FLkSQu2BzgBK9DIWTGX +XknBJIzZF03UZsVg5D67V2mnSClXucc0cGFcK4pDDt0tHeabA2GPinVe7Z6qDT4ZxPR8lKaXDdV2 +Pg2hTdcGSpqaltHxph7G/QIDAQABMA0GCSqGSIb3DQEBCwUAA4GBACcQFdOlVjYICOIyAXowQaEN +qcLpN1iWoL9UijNhTY37+U5+ycFT8QksT3Xmh9lEIqXMh121uViy2P/3p+Ek31AN9bB+BhWIM6PQ +gy+ApYDdSwTtWFARSrMqk7rRHUveYEfMw72yaOWDxCzcopEuADKrrYEute4CzZuXF9PbbgK6""" + +## Cert info: +# Name: PythonForWindowsTest +# Serial: '1b 8e 94 cb 0b 3e eb b6 41 39 f3 c9 09 b1 6b 46' + +TEST_PFX_PASSWORD = "TestPassword" + +TEST_PFX = """ +MIIGMwIBAzCCBe8GCSqGSIb3DQEHAaCCBeAEggXcMIIF2DCCA7AGCSqGSIb3DQEHAaCCA6EEggOd +MIIDmTCCA5UGCyqGSIb3DQEMCgECoIICtjCCArIwHAYKKoZIhvcNAQwBAzAOBAhoE8r3qUJeTQIC +B9AEggKQT7jm7ppgH64scyJ3cFW50BurqpMPtxgYyYCCtjdmHMlLPbUoujXOZVYi3seAEERE51BS +TXUi5ydHpY8cZ104nU4iEuJBAc+TZ7NQSTkjLKwAY1r1jrIikkQEmewLVlWQnj9dvCwD3lNkGXG8 +zJdWusta5Lw1Hz5ftsRXvN9UAvH8gxYviVRVmkZA33rI/BiyPZCulu2EBC0MeDBQHLLONup2xVGy ++YgU4Uf7khJIftWCgdrkyJIaMuB7vGUl014ZBV+XWaox+bS71qFQXUP2WnyTeeBVIaTJtggk+80X +fStWwvvzl02LTwGV3kJqWbazPlJkevfRQ7DNh1xa42eO57YEcEl3sR00anFWbL3J/I0bHb5XWY/e +8DYuMgIlat5gub8CTO2IViu6TexXFMXLxZdWAYvJ8ivc/q7mA/JcDJQlNnGof2Z6jY8ykWYloL/R +XMn2LeGqrql/guyRQcDrZu0LGX4sDG0aP9dbjk5fQpXSif1RUY4/T3HYeL0+1zu86ZKwVIIX5YfT +MLheIUGaXy/UJk361vAFKJBERGv1uufnqBxH0r1bRoytOaZr1niEA04u+VJa0DXOZzKBwxNhQRom +x4ffrsP2VnoJX+wnfYhPOjkiPiHyhswheG0VITTkqD+2uF54M5X2LLdzQuJpu0MZ5HOAHck/ZEpa +xV7h+kNse4p7y17b12H6tJNtVoJOlqP0Ujugc7vh4h8ZaPkSqVSV1nEvHzXx0c7gf038jv1+8WlN +4EgHp09FKU7sbSgcPY9jltElgaAr6J8a+rDGtk+055UeUYxM43U8naBiEOL77LP9FA0y8hKLKlJz +0GBCp4bJrLuZJenXHVb1Zme2EXO0jnQ9nB9OEyI3NpYTbZQxgcswEwYJKoZIhvcNAQkVMQYEBAEA +AAAwRwYJKoZIhvcNAQkUMToeOABQAHkAdABoAG8AbgBGAG8AcgBXAGkAbgBkAG8AdwBzAFQATQBQ +AEMAbwBuAHQAYQBpAG4AZQByMGsGCSsGAQQBgjcRATFeHlwATQBpAGMAcgBvAHMAbwBmAHQAIABF +AG4AaABhAG4AYwBlAGQAIABDAHIAeQBwAHQAbwBnAHIAYQBwAGgAaQBjACAAUAByAG8AdgBpAGQA +ZQByACAAdgAxAC4AMDCCAiAGCSqGSIb3DQEHAaCCAhEEggINMIICCTCCAgUGCyqGSIb3DQEMCgED +oIIB3TCCAdkGCiqGSIb3DQEJFgGgggHJBIIBxTCCAcEwggEqoAMCAQICEBuOlMsLPuu2QTnzyQmx +a0YwDQYJKoZIhvcNAQELBQAwHzEdMBsGA1UEAxMUUHl0aG9uRm9yV2luZG93c1Rlc3QwHhcNMTcw +NDEyMTQzOTIzWhcNMTgwNDEyMjAzOTIzWjAfMR0wGwYDVQQDExRQeXRob25Gb3JXaW5kb3dzVGVz +dDCBnzANBgkqhkiG9w0BAQEFAAOBjQAwgYkCgYEAkR8Av7EX14eaXOKaHPOWona7nWz3cmvtDnqJ +0JX9SUNXpH+yknwA1eRS5EkLtgc4ASvQyFkxl15JwSSM2RdN1GbFYOQ+u1dpp0gpV7nHNHBhXCuK +Qw7dLR3mmwNhj4p1Xu2eqg0+GcT0fJSmlw3Vdj4NoU3XBkqampbR8aYexv0CAwEAATANBgkqhkiG +9w0BAQsFAAOBgQAnEBXTpVY2CAjiMgF6MEGhDanC6TdYlqC/VIozYU2N+/lOfsnBU/EJLE915ofZ +RCKlzIddtblYstj/96fhJN9QDfWwfgYViDOj0IMvgKWA3UsE7VhQEUqzKpO60R1L3mBHzMO9smjl +g8Qs3KKRLgAyq62BLrXuAs2blxfT224CujEVMBMGCSqGSIb3DQEJFTEGBAQBAAAAMDswHzAHBgUr +DgMCGgQU70h/rEXLQOberGvgJenggoWU5poEFCfdE1wNK1M38Yp3+qfjEqNIJGCPAgIH0A== +""" + +@pytest.fixture() +def rawcert(): + return TEST_CERT.decode("base64") + + +@pytest.fixture() +def rawpfx(): + return TEST_PFX.decode("base64") + + +def test_certificate(rawcert): + cert = windows.crypto.CertificateContext.from_buffer(rawcert) + assert cert.serial == '1b 8e 94 cb 0b 3e eb b6 41 39 f3 c9 09 b1 6b 46' + assert cert.name == 'PythonForWindowsTest' + + +def test_pfx(rawcert, rawpfx): + pfx = windows.crypto.import_pfx(rawpfx, TEST_PFX_PASSWORD) + orig_cert = windows.crypto.CertificateContext.from_buffer(rawcert) + certs = pfx.certs + assert len(certs) == 1 + # Test cert comparaison + assert certs[0] == orig_cert + + +def test_open_pfx_bad_password(rawpfx): + with pytest.raises(WindowsError) as ar: + pfx = windows.crypto.import_pfx(rawpfx, "BadPassword") + + +def test_encrypt_decrypt(rawcert, rawpfx): + message_to_encrypt = "Testing message \xff\x01" + cert = windows.crypto.CertificateContext.from_buffer(rawcert) + # encrypt should accept a cert or iterable of cert + res = windows.crypto.encrypt(cert, message_to_encrypt) + res2 = windows.crypto.encrypt([cert], message_to_encrypt) + del cert + assert message_to_encrypt not in res + + # Open pfx and decrypt + pfx = windows.crypto.import_pfx(rawpfx, TEST_PFX_PASSWORD) + decrypt = windows.crypto.decrypt(pfx, res) + decrypt2 = windows.crypto.decrypt(pfx, res2) + + assert message_to_encrypt == decrypt + assert decrypt == decrypt2 + + diff --git a/pytesting/test_debugger.py b/pytesting/test_debugger.py new file mode 100644 index 0000000..23936c9 --- /dev/null +++ b/pytesting/test_debugger.py @@ -0,0 +1,567 @@ +import pytest +import textwrap +import ctypes +import os + +import windows +import windows.generated_def as gdef +import windows.native_exec.simple_x86 as x86 +import windows.native_exec.simple_x64 as x64 + +from conftest import generate_pop_and_exit_fixtures, pop_proc_32, pop_proc_64 +from pfwtest import * + +pytestmark = pytest.mark.usefixtures('check_for_gc_garbage') + +proc32_debug = generate_pop_and_exit_fixtures([pop_proc_32], ids=["proc32dbg"], dwCreationFlags=gdef.DEBUG_PROCESS) +proc64_debug = generate_pop_and_exit_fixtures([pop_proc_64], ids=["proc64dbg"], dwCreationFlags=gdef.DEBUG_PROCESS) + +if is_process_64_bits: + proc32_64_debug = generate_pop_and_exit_fixtures([pop_proc_32, pop_proc_64], ids=["proc32dbg", "proc64dbg"], + dwCreationFlags=gdef.DEBUG_PROCESS) +else: + # proc32_64_debug = proc32_debug + no_dbg_64_from_32 = lambda *x, **kwargs: pytest.skip("Cannot debug a proc64 from a 32b process") + proc32_64_debug = generate_pop_and_exit_fixtures([pop_proc_32, no_dbg_64_from_32], ids=["proc32dbg", "proc64dbg"], dwCreationFlags=gdef.DEBUG_PROCESS) + +yolo = generate_pop_and_exit_fixtures([pop_proc_32, pop_proc_64], ids=["proc32dbg", "proc64dbg"], dwCreationFlags=gdef.CREATE_SUSPENDED) + + +def test_init_breakpoint_callback(proc32_64_debug): + """Checking that the initial breakpoint call `on_exception`""" + class MyDbg(windows.debug.Debugger): + def on_exception(self, exception): + assert exception.ExceptionRecord.ExceptionCode == gdef.EXCEPTION_BREAKPOINT + self.current_process.exit() + + d = MyDbg(proc32_64_debug) + d.loop() + + +def get_debug_process_ndll(proc): + proc_pc = proc.threads[0].context.pc + ntdll_addr = proc.query_memory(proc_pc).AllocationBase + return windows.pe_parse.GetPEFile(ntdll_addr, target=proc) + +@check_for_handle_leak +def test_simple_standard_breakpoint(proc32_64_debug): + """Check that a standard Breakpoint method `trigger` is called with the correct informations""" + class TSTBP(windows.debug.Breakpoint): + def trigger(self, dbg, exc): + assert dbg.current_process.pid == proc32_64_debug.pid + assert dbg.current_process.read_memory(self.addr, 1) == "\xcc" + assert dbg.current_thread.context.pc == self.addr + d.current_process.exit() + + LdrLoadDll = get_debug_process_ndll(proc32_64_debug).exports["LdrLoadDll"] + d = windows.debug.Debugger(proc32_64_debug) + d.add_bp(TSTBP(LdrLoadDll)) + d.loop() + +@check_for_handle_leak +def test_simple_hwx_breakpoint(proc32_64_debug): + """Test that simple HXBP are trigger""" + + class TSTBP(windows.debug.HXBreakpoint): + def trigger(self, dbg, exc): + assert dbg.current_process.pid == proc32_64_debug.pid + assert dbg.current_thread.context.pc == self.addr + assert dbg.current_thread.context.Dr7 != 0 + d.current_process.exit() + + LdrLoadDll = get_debug_process_ndll(proc32_64_debug).exports["LdrLoadDll"] + d = windows.debug.Debugger(proc32_64_debug) + d.add_bp(TSTBP(LdrLoadDll)) + d.loop() + + + +def test_multiple_hwx_breakpoint(proc32_64_debug): + """Checking that multiple succesives HXBP are properly triggered""" + class TSTBP(windows.debug.HXBreakpoint): + COUNTER = 0 + def __init__(self, addr, expec_before): + self.addr = addr + self.expec_before = expec_before + + def trigger(self, dbg, exc): + assert dbg.current_process.pid == proc32_64_debug.pid + assert dbg.current_thread.context.pc == self.addr + assert dbg.current_thread.context.Dr7 != 0 + assert TSTBP.COUNTER == self.expec_before + assert dbg.current_process.read_memory(self.addr, 1) != "\xcc" + TSTBP.COUNTER += 1 + if TSTBP.COUNTER == 4: + d.current_process.exit() + + d = windows.debug.Debugger(proc32_64_debug) + addr = proc32_64_debug.virtual_alloc(0x1000) + proc32_64_debug.write_memory(addr, "\x90" * 8) + d.add_bp(TSTBP(addr, 0)) + d.add_bp(TSTBP(addr + 1, 1)) + d.add_bp(TSTBP(addr + 2, 2)) + d.add_bp(TSTBP(addr + 3, 3)) + proc32_64_debug.create_thread(addr, 0) + d.loop() + # Used to verif we actually called the Breakpoints + assert TSTBP.COUNTER == 4 + + +def test_four_hwx_breakpoint_fail(proc32_64_debug): + """Check that setting 4HXBP in the same thread fails""" + class TSTBP(windows.debug.HXBreakpoint): + def __init__(self, addr, expec_before): + self.addr = addr + self.expec_before = expec_before + + def trigger(self, dbg, exc): + raise NotImplementedError("Should fail before") + + d = windows.debug.Debugger(proc32_64_debug) + addr = proc32_64_debug.virtual_alloc(0x1000) + proc32_64_debug.write_memory(addr, "\x90" * 8 + "\xc3") + d.add_bp(TSTBP(addr, 0)) + d.add_bp(TSTBP(addr + 1, 1)) + d.add_bp(TSTBP(addr + 2, 2)) + d.add_bp(TSTBP(addr + 3, 3)) + d.add_bp(TSTBP(addr + 4, 4)) + + proc32_64_debug.create_thread(addr, 0) + with pytest.raises(ValueError) as e: + d.loop() + assert "DRx" in e.value.message + + +def test_hwx_breakpoint_are_on_all_thread(proc32_64_debug): + """Checking that HXBP without target are set on all threads""" + class MyDbg(windows.debug.Debugger): + def on_create_thread(self, exception): + # Check that later created thread have their HWX breakpoint :) + assert self.current_thread.context.Dr7 != 0 + + class TSTBP(windows.debug.HXBreakpoint): + COUNTER = 0 + def __init__(self, addr, expec_before): + self.addr = addr + self.expec_before = expec_before + + def trigger(self, dbg, exc): + assert len(dbg.current_process.threads) != 1 + #for t in dbg.current_process.threads: + # TEST_CASE.assertNotEqual(t.context.Dr7, 0) + if TSTBP.COUNTER == 0: #First time we got it ! create new thread + TSTBP.COUNTER = 1 + dbg.current_process.create_thread(addr, 0) + else: + TSTBP.COUNTER += 1 + d.current_process.exit() + + d = MyDbg(proc32_64_debug) + addr = proc32_64_debug.virtual_alloc(0x1000) + proc32_64_debug.write_memory(addr, "\x90" * 2 + "\xc3") + d.add_bp(TSTBP(addr, 0)) + proc32_64_debug.create_thread(addr, 0) + d.loop() + # Used to verif we actually called the Breakpoints + assert TSTBP.COUNTER == 2 + +@check_for_handle_leak +@pytest.mark.parametrize("bptype", [windows.debug.Breakpoint, windows.debug.HXBreakpoint]) +def test_simple_breakpoint_name_addr(proc32_64_debug, bptype): + """Check breakpoint address resolution for format dll!api""" + class TSTBP(bptype): + COUNTER = 0 + def trigger(self, dbg, exc): + addr = exc.ExceptionRecord.ExceptionAddress + LdrLoadDlladdr = dbg.current_process.peb.modules[1].pe.exports["LdrLoadDll"] + assert dbg.current_process.pid == proc32_64_debug.pid + assert dbg.current_thread.context.pc == addr + assert LdrLoadDlladdr == addr + TSTBP.COUNTER += 1 + d.current_process.exit() + + d = windows.debug.Debugger(proc32_64_debug) + d.add_bp(TSTBP("ntdll!LdrLoadDll")) + d.loop() + assert TSTBP.COUNTER == 1 + +import dbg_injection + +def test_hardware_breakpoint_name_addr(proc32_64_debug): + """Check that name addr in HXBP are trigger in all threads""" + class TSTBP(windows.debug.HXBreakpoint): + COUNTER = 0 + def trigger(self, dbg, exc): + addr = exc.ExceptionRecord.ExceptionAddress + assert dbg.current_process.pid == proc32_64_debug.pid + assert dbg.current_thread.context.pc == dbg._resolve(self.addr, dbg.current_process) + TSTBP.COUNTER += 1 + if TSTBP.COUNTER == 1: + # Perform a loaddll in a new thread :) + # See if it triggers a bp + t = dbg_injection.perform_manual_getproc_loadlib_for_dbg(dbg.current_process, "wintrust.dll") + self.new_thread = t + if hasattr(self, "new_thread") and dbg.current_thread.tid == self.new_thread.tid: + for t in dbg.current_process.threads: + assert t.context.Dr7 != 0 + d.current_process.exit() + + d = windows.debug.Debugger(proc32_64_debug) + d.add_bp(TSTBP("ntdll!LdrLoadDll")) + # Code that will load wintrust ! + d.loop() + + +def test_single_step(proc32_64_debug): + """Check that BP/dbg can trigger single step and that instruction follows""" + NB_SINGLE_STEP = 3 + class MyDbg(windows.debug.Debugger): + DATA = [] + def on_single_step(self, exception): + # Check that later created thread have their HWX breakpoint :) + addr = exception.ExceptionRecord.ExceptionAddress + assert self.current_thread.context.pc == addr + if len(MyDbg.DATA) < NB_SINGLE_STEP: + MyDbg.DATA.append(addr) + return self.single_step() + self.current_process.exit() + return + + class TSTBP(windows.debug.Breakpoint): + """Check that BP/dbg can trigger single step and that instruction follows""" + def trigger(self, dbg, exc): + return dbg.single_step() + + d = MyDbg(proc32_64_debug) + addr = proc32_64_debug.virtual_alloc(0x1000) + proc32_64_debug.write_memory(addr, "\x90" * 3 + "\xc3") + d.add_bp(TSTBP(addr)) + proc32_64_debug.create_thread(addr, 0) + d.loop() + # Used to verif we actually called the Breakpoints + assert len(MyDbg.DATA) == NB_SINGLE_STEP + for i in range(NB_SINGLE_STEP): + assert MyDbg.DATA[i] == addr + 1 + i + +@pytest.mark.parametrize("bptype", [windows.debug.Breakpoint, windows.debug.HXBreakpoint]) +def test_single_step_from_bp(proc32_64_debug, bptype): + """Check that HXBPBP/dbg can trigger single step""" + NB_SINGLE_STEP = 3 + class MyDbg(windows.debug.Debugger): + DATA = [] + def on_single_step(self, exception): + # Check that later created thread have their HWX breakpoint :) + addr = exception.ExceptionRecord.ExceptionAddress + assert self.current_thread.context.pc == addr + if len(MyDbg.DATA) < NB_SINGLE_STEP: + MyDbg.DATA.append(addr) + return self.single_step() + self.current_process.exit() + return + + # class TSTBP(windows.debug.HXBreakpoint): + class TSTBP(bptype): + """Check that BP/dbg can trigger single step and that instruction follows""" + def trigger(self, dbg, exc): + return dbg.single_step() + + d = MyDbg(proc32_64_debug) + addr = proc32_64_debug.virtual_alloc(0x1000) + proc32_64_debug.write_memory(addr, "\x90" * 3 + "\xc3") + d.add_bp(TSTBP(addr)) + proc32_64_debug.create_thread(addr, 0) + d.loop() + # Used to verif we actually called the Breakpoints + assert len(MyDbg.DATA) == NB_SINGLE_STEP + for i in range(NB_SINGLE_STEP): + assert MyDbg.DATA[i] == addr + 1 + i + + +# MEMBP + + +def test_memory_breakpoint_write(proc32_64_debug): + """Check MemoryBP WRITE""" + class TSTBP(windows.debug.MemoryBreakpoint): + #DEFAULT_PROTECT = PAGE_READONLY + #DEFAULT_PROTECT = PAGE_READONLY + DEFAULT_EVENTS = "W" + COUNTER = 0 + """Check that BP/dbg can trigger single step and that instruction follows""" + def trigger(self, dbg, exc): + fault_addr = exc.ExceptionRecord.ExceptionInformation[1] + eax = dbg.current_thread.context.func_result # Rax | Eax + if eax == 42: + dbg.current_process.exit() + return + assert fault_addr == data + eax + TSTBP.COUNTER += 1 + return + + if proc32_64_debug.bitness == 32: + asm, reg = (x86, "EAX") + else: + asm, reg = (x64, "RAX") + + d = windows.debug.Debugger(proc32_64_debug) + addr = proc32_64_debug.virtual_alloc(0x1000) + data = proc32_64_debug.virtual_alloc(0x1000) + + injected = asm.MultipleInstr() + injected += asm.Mov(reg, 0) + injected += asm.Mov(asm.deref(data), reg) + injected += asm.Add(reg, 4) + injected += asm.Mov(asm.deref(data + 4), reg) + injected += asm.Add(reg, 4) + # This one should NOT trigger the MemBP of size 8 + injected += asm.Mov(asm.deref(data + 8), reg) + injected += asm.Mov(reg, 42) + injected += asm.Mov(asm.deref(data), reg) + injected += asm.Ret() + + proc32_64_debug.write_memory(addr, injected.get_code()) + d.add_bp(TSTBP(data, size=0x8)) + proc32_64_debug.create_thread(addr, 0) + d.loop() + # Used to verif we actually called the Breakpoints for the good addresses + assert TSTBP.COUNTER == 2 + + +def test_memory_breakpoint_exec(proc32_64_debug): + """Check MemoryBP EXEC""" + NB_NOP_IN_PAGE = 3 + + class TSTBP(windows.debug.MemoryBreakpoint): + """Check that BP/dbg can trigger single step and that instruction follows""" + #DEFAULT_PROTECT = PAGE_NOACCESS + DEFAULT_EVENTS = "X" + DATA = [] + def trigger(self, dbg, exc): + fault_addr = exc.ExceptionRecord.ExceptionInformation[1] + TSTBP.DATA.append(fault_addr) + if len(TSTBP.DATA) == NB_NOP_IN_PAGE + 1: + dbg.current_process.exit() + + d = windows.debug.Debugger(proc32_64_debug) + addr = proc32_64_debug.virtual_alloc(0x1000) + proc32_64_debug.write_memory(addr, "\x90" * NB_NOP_IN_PAGE + "\xc3") + d.add_bp(TSTBP(addr, size=0x1000)) + proc32_64_debug.create_thread(addr, 0) + d.loop() + # Used to verif we actually called the Breakpoints + assert len(TSTBP.DATA) == NB_NOP_IN_PAGE + 1 + for i in range(NB_NOP_IN_PAGE + 1): + assert TSTBP.DATA[i] == addr + i + + +# breakpoint remove +import threading +@pytest.mark.parametrize("bptype", [windows.debug.FunctionParamDumpHXBP, windows.debug.FunctionParamDumpBP]) +def test_standard_breakpoint_self_remove(proc32_64_debug, bptype): + data = [] + + def do_check(): + proc32_64_debug.execute_python_unsafe("open(u'FILENAME1')").wait() + proc32_64_debug.execute_python_unsafe("open(u'FILENAME2')").wait() + proc32_64_debug.execute_python_unsafe("open(u'FILENAME3')").wait() + proc32_64_debug.exit() + + class TSTBP(bptype): + TARGET = windows.winproxy.CreateFileW + def trigger(self, dbg, exc): + addr = exc.ExceptionRecord.ExceptionAddress + ctx = dbg.current_thread.context + filename = self.extract_arguments(dbg.current_process, dbg.current_thread)["lpFileName"] + data.append(filename) + if filename == u"FILENAME2": + dbg.del_bp(self) + + d = windows.debug.Debugger(proc32_64_debug) + d.add_bp(TSTBP("kernel32!CreateFileW")) + threading.Thread(target=do_check).start() + d.loop() + assert data == [u"FILENAME1", u"FILENAME2"] + +@pytest.mark.parametrize("bptype", [windows.debug.FunctionParamDumpHXBP, windows.debug.FunctionParamDumpBP]) +def test_standard_breakpoint_remove(proc32_64_debug, bptype): + data = [] + + def do_check(): + proc32_64_debug.execute_python_unsafe("open(u'FILENAME1')").wait() + proc32_64_debug.execute_python_unsafe("open(u'FILENAME2')").wait() + d.del_bp(the_bp) + proc32_64_debug.execute_python_unsafe("open(u'FILENAME3')").wait() + proc32_64_debug.exit() + + class TSTBP(bptype): + TARGET = windows.winproxy.CreateFileW + def trigger(self, dbg, exc): + addr = exc.ExceptionRecord.ExceptionAddress + ctx = dbg.current_thread.context + filename = self.extract_arguments(dbg.current_process, dbg.current_thread)["lpFileName"] + data.append(filename) + + d = windows.debug.Debugger(proc32_64_debug) + the_bp = TSTBP("kernel32!CreateFileW") + d.add_bp(the_bp) + threading.Thread(target=do_check).start() + d.loop() + assert data == [u"FILENAME1", u"FILENAME2"] + + +def get_generate_read_at_for_proc(target): + if target.bitness == 32: + def generate_read_at(addr): + res = x86.MultipleInstr() + res += x86.Mov("EAX", x86.deref(addr)) + res += x86.Ret() + return res.get_code() + else: + def generate_read_at(addr): + res = x64.MultipleInstr() + res += x64.Mov("RAX", x64.deref(addr)) + res += x64.Ret() + return res.get_code() + return generate_read_at + +def get_generate_write_at_for_proc(target): + if target.bitness == 32: + def generate_write_at(addr): + res = x86.MultipleInstr() + res += x86.Mov(x86.deref(addr), "EAX") + res += x86.Ret() + return res.get_code() + else: + def generate_write_at(addr): + res = x64.MultipleInstr() + res += x64.Mov(x64.deref(addr), "RAX") + res += x64.Ret() + return res.get_code() + return generate_write_at + +def test_mem_breakpoint_remove(proc32_64_debug): + data = [] + generate_read_at = get_generate_read_at_for_proc(proc32_64_debug) + + def do_check(): + proc32_64_debug.execute(generate_read_at(data_addr)).wait() + proc32_64_debug.execute(generate_read_at(data_addr + 4)).wait() + d.del_bp(the_bp) + proc32_64_debug.execute(generate_read_at(data_addr + 8)).wait() + proc32_64_debug.exit() + + class TSTBP(windows.debug.MemoryBreakpoint): + #DEFAULT_PROTECT = PAGE_NOACCESS + DEFAULT_EVENTS = "RWX" + def trigger(self, dbg, exc): + addr = exc.ExceptionRecord.ExceptionAddress + fault_addr = exc.ExceptionRecord.ExceptionInformation[1] + data.append(fault_addr) + + d = windows.debug.Debugger(proc32_64_debug) + data_addr = proc32_64_debug.virtual_alloc(0x1000) + the_bp = TSTBP(data_addr, size=0x1000) + d.add_bp(the_bp) + threading.Thread(target=do_check).start() + d.loop() + assert data == [data_addr, data_addr + 4] + + +def test_mem_breakpoint_self_remove(proc32_64_debug): + data = [] + generate_read_at = get_generate_read_at_for_proc(proc32_64_debug) + + def do_check(): + proc32_64_debug.execute(generate_read_at(data_addr)).wait() + proc32_64_debug.execute(generate_read_at(data_addr + 4)).wait() + proc32_64_debug.execute(generate_read_at(data_addr + 8)).wait() + proc32_64_debug.exit() + + class TSTBP(windows.debug.MemoryBreakpoint): + #DEFAULT_PROTECT = PAGE_NOACCESS + DEFAULT_EVENTS = "RWX" + def trigger(self, dbg, exc): + addr = exc.ExceptionRecord.ExceptionAddress + fault_addr = exc.ExceptionRecord.ExceptionInformation[1] + data.append(fault_addr) + if fault_addr == data_addr + 4: + dbg.del_bp(self) + + d = windows.debug.Debugger(proc32_64_debug) + data_addr = proc32_64_debug.virtual_alloc(0x1000) + the_bp = TSTBP(data_addr, size=0x1000) + d.add_bp(the_bp) + threading.Thread(target=do_check).start() + d.loop() + assert data == [data_addr, data_addr + 4] + + + +def test_read_write_bp_same_page(proc32_64_debug): + data = [] + generate_read_at = get_generate_read_at_for_proc(proc32_64_debug) + generate_write_at = get_generate_write_at_for_proc(proc32_64_debug) + + def do_check(): + proc32_64_debug.execute(generate_read_at(data_addr)).wait() + proc32_64_debug.execute(generate_write_at(data_addr + 4)).wait() + proc32_64_debug.execute(generate_read_at(data_addr + 0x500)).wait() + proc32_64_debug.execute(generate_write_at(data_addr + 0x504)).wait() + proc32_64_debug.exit() + + class MemBP(windows.debug.MemoryBreakpoint): + #DEFAULT_PROTECT = PAGE_NOACCESS + DEFAULT_EVENTS = "RWX" + def trigger(self, dbg, exc): + addr = exc.ExceptionRecord.ExceptionAddress + fault_addr = exc.ExceptionRecord.ExceptionInformation[1] + #print("Got <{0:#x}> <{1}>".format(fault_addr, exc.ExceptionRecord.ExceptionInformation[0])) + data.append((self, fault_addr)) + + d = windows.debug.Debugger(proc32_64_debug) + data_addr = proc32_64_debug.virtual_alloc(0x1000) + the_write_bp = MemBP(data_addr + 0x500, size=0x500, events="W") + the_read_bp = MemBP(data_addr, size=0x500, events="RW") + d.add_bp(the_write_bp) + d.add_bp(the_read_bp) + threading.Thread(target=do_check).start() + d.loop() + + # generate_read_at (data_addr + 0x500)) (write_bp (PAGE_READONLY)) should not be triggered + expected_result = [(the_read_bp, data_addr), (the_read_bp, data_addr + 4), + (the_write_bp, data_addr + 0x504)] + + assert data == expected_result + + +def test_exe_in_module_list(proc32_64_debug): + class MyDbg(windows.debug.Debugger): + def on_exception(self, exception): + exename = os.path.basename(proc32_64_debug.peb.imagepath.str) + this_process_modules = self._module_by_process[self.current_process.pid] + assert exename and exename in this_process_modules.keys() + self.current_process.exit() + + d = MyDbg(proc32_64_debug) + d.loop() + + +def test_bp_exe_by_name(proc32_64_debug): + class TSTBP(windows.debug.Breakpoint): + COUNTER = 0 + def trigger(self, dbg, exc): + TSTBP.COUNTER += 1 + assert TSTBP.COUNTER == 1 + # Kill the target in 0.5s + # It's not too long + # It's long enought to get trigger being recalled if implem is broken + threading.Timer(0.5, proc32_64_debug.exit).start() + + exepe = proc32_64_debug.peb.exe + entrypoint = exepe.get_OptionalHeader().AddressOfEntryPoint + exename = os.path.basename(proc32_64_debug.peb.imagepath.str) + d = windows.debug.Debugger(proc32_64_debug) + # The goal is to test bp of format 'exename!offset' so we craft a string based on the entrypoint + d.add_bp(TSTBP("{name}!{offset}".format(name=exename, offset=entrypoint))) + d.loop() + assert TSTBP.COUNTER == 1 \ No newline at end of file diff --git a/pytesting/test_generated_code.py b/pytesting/test_generated_code.py new file mode 100644 index 0000000..79aeb92 --- /dev/null +++ b/pytesting/test_generated_code.py @@ -0,0 +1,41 @@ +import pytest +import pickle + +import windows +import windows.generated_def as gdef + +def _test_pickle_unpickle(obj, protocol=0): + pickled = pickle.dumps(obj, protocol) + unpickled = pickle.loads(pickled) + assert unpickled == obj + +def test_str_flags_value(): + assert gdef.MS_ENHANCED_PROV == gdef.MS_ENHANCED_PROV_A + +def test_long_flag_pickle_v0(): + _test_pickle_unpickle(gdef.PAGE_EXECUTE_READWRITE, 0) + +def test_long_flag_pickle_v1(): + _test_pickle_unpickle(gdef.PAGE_EXECUTE_READWRITE, 1) + +def test_long_flag_pickle_v2(): + _test_pickle_unpickle(gdef.PAGE_EXECUTE_READWRITE, 2) + +def test_str_flag_pickle_v0(): + _test_pickle_unpickle(gdef.szOID_RSA, 0) + +def test_str_flag_pickle_v1(): + _test_pickle_unpickle(gdef.szOID_RSA, 1) + +def test_str_flag_pickle_v2(): + _test_pickle_unpickle(gdef.szOID_RSA, 2) + +def test_enum_value_pickle_v0(): + _test_pickle_unpickle(gdef.SystemBasicInformation, 0) + +def test_enum_value_pickle_v1(): + _test_pickle_unpickle(gdef.SystemBasicInformation, 1) + +def test_enum_value_pickle_v2(): + _test_pickle_unpickle(gdef.SystemBasicInformation, 2) + diff --git a/pytesting/test_hook.py b/pytesting/test_hook.py new file mode 100644 index 0000000..aebf1ee --- /dev/null +++ b/pytesting/test_hook.py @@ -0,0 +1,190 @@ +import pytest +import textwrap +import ctypes + +import windows +import windows.generated_def as gdef +import windows.native_exec.simple_x86 as x86 +import windows.native_exec.simple_x64 as x64 + + +from pfwtest import * + +pytestmark = pytest.mark.usefixtures('check_for_gc_garbage') + + +def test_self_iat_hook_success(): + """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 + + x = RegOpenKeyExA.set_hook(open_reg_hook) + import _winreg + open_args = (0x12345678, "MY_KEY_VALUE") + k = _winreg.OpenKey(*open_args) + assert k.handle == 12345678 + assert hook_value[0] == open_args + # Remove the hook + x.disable() + +def test_self_iat_hook_fail_return(): + """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 + + x = RegOpenKeyExA.set_hook(open_reg_hook_fail) + import _winreg + open_args = (0x12345678, "MY_KEY_VALUE") + with pytest.raises(WindowsError) as ar: + _winreg.OpenKey(*open_args) + assert ar.value.winerror == 0x11223344 + x.disable() + + +def test_self_iat_hook_multithread(): + """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"]() + + x = LdrLoadDll.set_hook(MyHook) + # Trigger from local thread + ctypes.WinDLL("kernel32.dll") + assert 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) + t.wait() + assert len(calling_thread) == 2 + x.disable() + +@check_for_gc_garbage +def test_remote_iat_hook(proc32_64): + proc32_64.execute_python("import windows") + proc32_64.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"]() + + x = 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 + """ + proc32_64.execute_python(textwrap.dedent(code)) + # Tricky part: we use an injected thread exit_value to ask stuff about the remote python + def remote_ask(request): + t = proc32_64.execute_python_unsafe(request) + t.wait() + result = t.exit_code + if result > 100: + import pdb;pdb.set_trace() + return result + + assert remote_ask("windows.current_thread.exit(len(calling_thread))") == 1 + assert remote_ask("windows.current_thread.exit(calling_thread == set([hooking_thread]))") == 1 + + # Trigger hook from another Python thread + proc32_64.execute_python_unsafe("ctypes.WinDLL('ANOTHER_FAKE_DLL')").wait() + assert remote_ask("windows.current_thread.exit(len(calling_thread))") == 2 + + # Trigger hook from a NONPython thread + k32 = [m for m in proc32_64.peb.modules if m.name == "kernel32.dll"][0] + load_libraryA = k32.pe.exports["LoadLibraryA"] + with proc32_64.allocated_memory(0x1000) as addr: + proc32_64.write_memory(addr, "DLLNOTFOUND.NOT_A_REAL_DLL" + "\x00") + t = proc32_64.create_thread(load_libraryA, addr) + t.wait() + assert remote_ask("windows.current_thread.exit(len(calling_thread))") == 3 + +#@check_for_gc_garbage +#def test_remote_iat_hook_64(self): +# with Calc64() 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.Callback(*[gdef.PVOID] * 5) +# def MyHook(*args, **kwargs): +# calling_thread.add(windows.current_thread.tid) +# print(windows.current_thread.tid) +# return kwargs["real_function"]() +# +# x = 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(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) +# t.wait() +# self.assertEqual(remote_ask("windows.current_thread.exit(len(calling_thread))"), 3) + + +# TODO: test new hook API diff --git a/pytesting/test_native_utils.py b/pytesting/test_native_utils.py new file mode 100644 index 0000000..a448520 --- /dev/null +++ b/pytesting/test_native_utils.py @@ -0,0 +1,64 @@ +import pytest + +import windows +import windows.generated_def as gdef + +from windows.native_exec import nativeutils + +from pfwtest import * + +@check_for_gc_garbage +class TestNativeUtils(object): + @process_64bit_only + def test_strlenw64(self): + strlenw64 = windows.native_exec.create_function(nativeutils.StrlenW64.get_code(), [gdef.UINT, gdef.LPCWSTR]) + assert strlenw64("YOLO") == 4 + assert strlenw64("") == 0 + + @process_64bit_only + def test_strlena64(self): + strlena64 = windows.native_exec.create_function(nativeutils.StrlenA64.get_code(), [gdef.UINT, gdef.LPCSTR]) + assert strlena64("YOLO") == 4 + assert strlena64("") == 0 + + @process_64bit_only + def test_getprocaddr64(self): + getprocaddr64 = windows.native_exec.create_function(nativeutils.GetProcAddress64.get_code(), [gdef.ULONG64, gdef.LPCWSTR, gdef.LPCSTR]) + k32 = [mod for mod in windows.current_process.peb.modules if mod.name == "kernel32.dll"][0] + exports = [(x,y) for x,y in k32.pe.exports.items() if isinstance(x, basestring)] + + for name, addr in exports: + name = name.encode() + compute_addr = getprocaddr64("KERNEL32.DLL", name) + # Put name in test to know which function caused the assert fails + assert (name, hex(addr)) == (name, hex(compute_addr)) + + assert getprocaddr64("YOLO.DLL", "whatever") == 0xfffffffffffffffe + assert getprocaddr64("KERNEL32.DLL", "YOLOAPI") == 0xffffffffffffffff + + @process_32bit_only + def test_strlenw32(self): + strlenw32 = windows.native_exec.create_function(nativeutils.StrlenW32.get_code(), [gdef.UINT, gdef.LPCWSTR]) + assert strlenw32("YOLO") == 4 + assert strlenw32("") == 0 + + @process_32bit_only + def test_strlena32(self): + strlena32 = windows.native_exec.create_function(nativeutils.StrlenA32.get_code(), [gdef.UINT, gdef.LPCSTR]) + assert strlena32("YOLO") == 4 + assert strlena32("") == 0 + + @process_32bit_only + def test_getprocaddr32(self): + getprocaddr32 = windows.native_exec.create_function(nativeutils.GetProcAddress32.get_code(), [gdef.UINT, gdef.LPCWSTR, gdef.LPCSTR]) + k32 = [mod for mod in windows.current_process.peb.modules if mod.name == "kernel32.dll"][0] + exports = [(x,y) for x,y in k32.pe.exports.items() if isinstance(x, basestring)] + + for name, addr in exports: + name = name.encode() + compute_addr = getprocaddr32("KERNEL32.DLL", name) + # Put name in test to know which function caused the assert fails + assert (name, hex(addr)) == (name, hex(compute_addr)) + + assert getprocaddr32("YOLO.DLL", "whatever") == 0xfffffffe + assert getprocaddr32("KERNEL32.DLL", "YOLOAPI") == 0xffffffff \ No newline at end of file diff --git a/pytesting/test_process.py b/pytesting/test_process.py new file mode 100644 index 0000000..60af33e --- /dev/null +++ b/pytesting/test_process.py @@ -0,0 +1,366 @@ +import pytest + +import os +import time +import struct +import textwrap + +import windows +import windows.generated_def as gdef +import windows.native_exec.simple_x86 as x86 +import windows.native_exec.simple_x64 as x64 + +from pfwtest import * + +@check_for_gc_garbage +class TestCurrentProcessWithCheckGarbage(object): + def test_current_process_ppid(self): + myself = [p for p in windows.system.processes if p.pid == windows.current_process.pid][0] + assert myself.ppid == windows.current_process.ppid + + def test_get_current_process_peb(self): + return windows.current_process.peb + + def test_get_current_process_modules(self): + assert "python" in windows.current_process.peb.modules[0].name + + def test_get_current_process_exe(self): + exe = windows.current_process.peb.exe + exe_by_module = windows.current_process.peb.modules[0].pe + exe.baseaddr == exe_by_module.baseaddr + exe.bitness == exe_by_module.bitness + + def test_current_process_pe_imports(self): + python_module = windows.current_process.peb.modules[0] + imp = python_module.pe.imports + assert "kernel32.dll" in imp.keys(), 'Kernel32.dll not in python imports' + current_proc_id_iat = [f for f in imp["kernel32.dll"] if f.name == "GetCurrentProcessId"][0] + k32_base = windows.winproxy.LoadLibraryA("kernel32.dll") + assert windows.winproxy.GetProcAddress(k32_base, "GetCurrentProcessId") == current_proc_id_iat.value + + def test_current_process_pe_exports(self): + mods = [m for m in windows.current_process.peb.modules if m.name == "kernel32.dll"] + assert mods, 'Could not find "kernel32.dll" in current process modules' + k32 = mods[0] + get_current_proc_id = k32.pe.exports['GetCurrentProcessId'] + k32_base = windows.winproxy.LoadLibraryA("kernel32.dll") + assert windows.winproxy.GetProcAddress(k32_base, "GetCurrentProcessId") == get_current_proc_id + + def test_local_process_pe_sections(self): + mods = [m for m in windows.current_process.peb.modules if m.name == "kernel32.dll"] + assert mods, 'Could not find "kernel32.dll" in current process modules' + k32 = mods[0] + sections = k32.pe.sections + all_sections_name = [s.name for s in sections] + assert ".text" in all_sections_name + sections[0].start + sections[0].size + + def test_token_info(self): + token = windows.current_process.token + assert isinstance(token.computername, basestring) + assert isinstance(token.username, basestring) + assert isinstance(token.integrity, (int, long)) + assert isinstance(token.is_elevated, (bool)) + + + +@check_for_gc_garbage +class TestProcessWithCheckGarbage(object): + def test_pop_proc_32(self, proc32): + assert proc32.bitness == 32 + + @windows_64bit_only + def test_pop_proc_64(self, proc64): + assert proc64.bitness == 64 + + def test_process_ppid(self, proc32_64): + assert proc32_64.ppid == windows.current_process.pid + + # Test process read/write + + def test_read_memory(self, proc32_64): + k32 = [m for m in proc32_64.peb.modules if m.name == "kernel32.dll"][0] + assert proc32_64.read_memory(k32.baseaddr, 2), "MZ" + + def test_write_memory(self, proc32_64): + k32 = [m for m in proc32_64.peb.modules if m.name == "kernel32.dll"][0] + with proc32_64.virtual_protected(k32.baseaddr, 2, gdef.PAGE_EXECUTE_READWRITE): + proc32_64.write_memory(k32.baseaddr, "XD") + assert proc32_64.read_memory(k32.baseaddr, 2) == "XD" + + def test_read_string(self, proc32_64): + test_string = "TEST_STRING" + string_to_write = test_string + "\x00" + with proc32_64.allocated_memory(0x1000) as addr: + proc32_64.write_memory(addr, string_to_write) + assert proc32_64.read_string(addr) == test_string + + def test_read_string_end_page(self, proc32_64): + test_string = "TEST_STRING" + string_to_write = test_string + "\x00" + with proc32_64.allocated_memory(0x1000) as addr: + waddr = addr + 0x1000 - len(string_to_write) + proc32_64.write_memory(addr, string_to_write) + assert proc32_64.read_string(addr) == test_string + + + def test_wread_string(self, proc32_64): + test_string = "TEST_STRING" + string_to_write = test_string + "\x00" + with proc32_64.allocated_memory(0x1000) as addr: + proc32_64.write_memory(addr, "\x00".join(string_to_write)) + assert proc32_64.read_wstring(addr) == test_string + + def test_read_wstring_end_page(self, proc32_64): + test_string = "TEST_STRING" + string_to_write = test_string + "\x00" + with proc32_64.allocated_memory(0x1000) as addr: + waddr = addr + 0x1000 - len(string_to_write) + proc32_64.write_memory(addr, "\x00".join(string_to_write)) + assert proc32_64.read_wstring(addr) == test_string + + # Test native execution + + def test_execute_to_proc32(self, proc32): + with proc32.allocated_memory(0x1000) as addr: + shellcode = x86.MultipleInstr() + shellcode += x86.Mov('EAX', 0x42424242) + shellcode += x86.Mov(x86.create_displacement(disp=addr), 'EAX') + shellcode += x86.Ret() + proc32.execute(shellcode.get_code()) + time.sleep(0.1) + dword = proc32.read_dword(addr) + assert dword, 0x42424242 + + @windows_64bit_only + def test_execute_to_64(self, proc64): + with proc64.allocated_memory(0x1000) as addr: + shellcode = x64.MultipleInstr() + shellcode += x64.Mov('RAX', 0x4242424243434343) + shellcode += x64.Mov(x64.create_displacement(disp=addr), 'RAX') + shellcode += x64.Ret() + proc64.execute(shellcode.get_code()) + time.sleep(0.1) + qword = proc64.read_qword(addr) + assert qword == 0x4242424243434343 + + # Python execution + + def test_execute_python(self, proc32_64): + with proc32_64.allocated_memory(0x1000) as addr: + proc32_64.execute_python('import ctypes; ctypes.c_uint.from_address({0}).value = 0x42424242'.format(addr)) + dword = proc32_64.read_dword(addr) + assert dword == 0x42424242 + + + def test_execute_python_suspended(self, proc32_64_suspended): + proc = proc32_64_suspended + with proc.allocated_memory(0x1000) as addr: + proc.execute_python('import ctypes; ctypes.c_uint.from_address({0}).value = 0x42424242'.format(addr)) + dword = proc.read_dword(addr) + assert dword == 0x42424242 + # Check calc32 is still suspended: + # 1 thread | except windows 10 that pop threads + # main thread suspend count == 1 + assert proc.threads[0].suspend() == 1 + if not is_windows_10: + assert len(proc.threads) == 1 + + + # Remote structure parsing + + def test_parse_remote_peb(self, proc32_64): + # Wait for PEB initialization + # Yeah a don't know but on 32bits system the parsing might begin before + # InMemoryOrderModuleList is setup.. + import time; time.sleep(0.1) + assert proc32_64.peb.modules[0].name == test_binary_name + + + def test_parse_remote_pe(self, proc32_64): + # Wait for PEB initialization + # Yeah a don't know but on 32bits system the parsing might begin before + # InMemoryOrderModuleList is setup.. + import time; time.sleep(0.1) + mods = [m for m in proc32_64.peb.modules if m.name == "kernel32.dll"] + assert mods, 'Could not find "kernel32.dll" in calc32' + k32 = mods[0] + mods[0].pe.sections[0].name # Just see if it's parse + assert mods[0].pe.export_name.lower() == "kernel32.dll" + get_current_proc_id = k32.pe.exports['GetCurrentProcessId'] + # TODO: check get_current_proc_id value (but we cannot do 64->32 injection for now) + #if is_process_64_bits: + # raise NotImplementedError("Python execution 64->32") + with proc32_64.allocated_memory(0x1000) as addr: + remote_python_code = """ + import ctypes + import windows + # windows.utils.create_console() # remove comment for debug + k32 = [m for m in windows.current_process.peb.modules if m.name == "kernel32.dll"][0] + GetCurrentProcessId = k32.pe.exports['GetCurrentProcessId'] + ctypes.c_void_p.from_address({1}).value = GetCurrentProcessId + """.format(os.getcwd(), addr) + x = proc32_64.execute_python(textwrap.dedent(remote_python_code)) + dword = proc32_64.read_ptr(addr) + assert dword == get_current_proc_id + + + def test_remote_peb_exe(self, proc32_64): + exe = proc32_64.peb.exe + exe_by_module = proc32_64.peb.modules[0].pe + assert exe.baseaddr == exe_by_module.baseaddr + assert exe.bitness == exe_by_module.bitness + + def test_execute_python_raises(self, proc32_64): + res = proc32_64.execute_python("import time;time.sleep(0.1); 2") + assert res == True + with pytest.raises(windows.injection.RemotePythonError) as ar: + t = proc32_64.execute_python("import time;time.sleep(0.1); raise ValueError('BYE')") + + def test_thread_start_address(self, proc32_64): + t = proc32_64.threads[0] + t.start_address # No better idea right now that checking for crash/exception + + + def test_get_context_address_32(self, proc32): + code = x86.MultipleInstr() + code += x86.Mov("EAX", 0x42424242) + code += x86.Label(":LOOP") + code += x86.Jmp(":LOOP") + t = proc32.execute(code.get_code()) + time.sleep(0.5) + cont = t.context + assert cont.Eax == 0x42424242 + + @windows_64bit_only + def test_get_context_address_64(self, proc64): + code = x64.MultipleInstr() + code += x64.Mov("RAX", 0x4242424243434343) + code += x64.Label(":LOOP") + code += x64.Jmp(":LOOP") + t = proc64.execute(code.get_code()) + time.sleep(0.5) + cont = t.context + assert cont.Rax == 0x4242424243434343 + + + def test_process_is_exit(self, proc32_64): + assert proc32_64.is_exit == False + proc32_64.exit(42) + assert proc32_64.exit_code == 42 + assert proc32_64.is_exit == True + + + def test_set_thread_context_32(self, proc32): + code = x86.MultipleInstr() + code += x86.Label(":LOOP") + code += x86.Jmp(":LOOP") + data_len = len(code.get_code()) + code += x86.Ret() + + t = proc32.execute(code.get_code()) + time.sleep(0.1) + assert proc32.is_exit == False + t.suspend() + ctx = t.context + ctx.Eip += data_len + ctx.Eax = 0x11223344 + t.set_context(ctx) + t.resume() + time.sleep(0.1) + assert t.exit_code == 0x11223344 + + + @windows_64bit_only + def test_set_thread_context_64(self, proc64): + code = x64.MultipleInstr() + code += x64.Label(":LOOP") + code += x64.Jmp(":LOOP") + data_len = len(code.get_code()) + code += x64.Ret() + t = proc64.execute(code.get_code()) + time.sleep(0.1) + assert proc64.is_exit == False + t.suspend() + ctx = t.context + ctx.Rip += data_len + ctx.Rax = 0x11223344 + t.set_context(ctx) + t.resume() + time.sleep(0.1) + assert t.exit_code == 0x11223344 + + + def test_load_library(self, proc32_64): + DLL = "wintrust.dll" + proc32_64.load_library(DLL) + assert DLL in [m.name for m in proc32_64.peb.modules] + + + + def test_get_working_set(self, proc32_64): + k32 = [m for m in proc32_64.peb.modules if m.name == "kernel32.dll"][0] + api_addr = k32.pe.exports["CreateFileA"] + data = proc32_64.read_memory(api_addr, 5) + page_target = api_addr >> 12 + for page_info in proc32_64.query_working_set(): + if page_info.virtualpage == page_target: + assert page_info.shared == True + break + else: + raise ValueError("query_working_set page info for <0x{0:x}> not found".format(page_target)) + with proc32_64.virtual_protected(api_addr, 5, gdef.PAGE_EXECUTE_READWRITE): + data = proc32_64.write_memory(api_addr, data) + for page_info in proc32_64.query_working_set(): + if page_info.virtualpage == page_target: + assert page_info.shared == False + break + else: + raise ValueError("query_working_set page info for <0x{0:x}> not found".format(page_target)) + + + def test_get_working_setex(self, proc32_64): + k32 = [m for m in proc32_64.peb.modules if m.name == "kernel32.dll"][0] + + text = [s for s in k32.pe.sections if s.name == ".text"][0] + pages = [text.start + off for off in range(0, text.size, 0x1000)] + + api_addr = k32.pe.exports["CreateFileA"] + data = proc32_64.read_memory(api_addr, 5) + page_target = (api_addr >> 12) << 12 + + for page_info in proc32_64.query_working_setex(pages): + assert page_info.VirtualAddress in pages + if page_info.VirtualAddress == page_target: + assert page_info.VirtualAttributes.shared == True + break + else: + raise ValueError("query_working_set page info for <0x{0:x}> not found".format(page_target)) + with proc32_64.virtual_protected(api_addr, 5, gdef.PAGE_EXECUTE_READWRITE): + data = proc32_64.write_memory(api_addr, data) + for page_info in proc32_64.query_working_setex(pages): + assert page_info.VirtualAddress in pages + if page_info.VirtualAddress == page_target: + assert page_info.VirtualAttributes.shared == False + break + else: + raise ValueError("query_working_set page info for <0x{0:x}> not found".format(page_target)) + + + def test_mapped_filename(self, proc32_64): + k32 = [m for m in proc32_64.peb.modules if m.name == "kernel32.dll"][0] + mapped_filname = proc32_64.get_mapped_filename(k32.baseaddr) + assert mapped_filname.endswith("kernel32.dll") + + + def test_thread_teb_base(self, proc32_64): + t = proc32_64.threads[0] + assert t.teb_base != 0 + + + def test_thread_owner_from_tid(self, proc32_64): + thread = proc32_64.threads[0] + tst_thread = windows.winobject.process.WinThread(tid=thread.tid) + assert thread.owner_pid == tst_thread.owner_pid + assert thread.owner.name == tst_thread.owner.name \ No newline at end of file diff --git a/pytesting/test_system.py b/pytesting/test_system.py new file mode 100644 index 0000000..7e285b0 --- /dev/null +++ b/pytesting/test_system.py @@ -0,0 +1,35 @@ +import pytest +import windows + +from pfwtest import * + +@check_for_gc_garbage +class TestSystemWithCheckGarbage(object): + def test_version(self): + return windows.system.version + + def test_version_name(self): + return windows.system.version_name + + def test_computer_name(self): + return windows.system.computer_name + + def test_services(self): + return windows.system.services + + def test_logicaldrives(self): + return windows.system.logicaldrives + + def test_wmi(self): + return windows.system.wmi.select("Win32_Process", "*") + + +@check_for_gc_garbage +@check_for_handle_leak +class TestSystemWithCheckGarbageAndHandleLeak(object): + def test_threads(self): + return windows.system.threads + + def test_processes(self): + procs = windows.system.processes + assert windows.current_process.pid in [p.pid for p in procs] \ No newline at end of file diff --git a/pytesting/test_syswow.py b/pytesting/test_syswow.py new file mode 100644 index 0000000..f1d18d8 --- /dev/null +++ b/pytesting/test_syswow.py @@ -0,0 +1,73 @@ +import pytest +import textwrap + +import windows +import windows.generated_def as gdef +import windows.native_exec.simple_x86 as x86 +import windows.native_exec.simple_x64 as x64 + +from pfwtest import * + +pytestmark = pytest.mark.usefixtures('check_for_gc_garbage') + + + +@process_syswow_only +class TestSyswowCurrentProcess(object): + def test_exec_syswow(self): + x64_code = x64.assemble("mov rax, 0x4040404040404040; mov r11, 0x0202020202020202; add rax, r11; ret") + res = windows.syswow64.execute_64bits_code_from_syswow(x64_code) + assert res == 0x4242424242424242 + + def test_self_pebsyswow(self): + peb64 = windows.current_process.peb_syswow + modules_names = [m.name for m in peb64.modules] + assert "wow64.dll" in modules_names + # Parsing + wow64 = [m for m in peb64.modules if m.name == "wow64.dll"][0] + assert "Wow64LdrpInitialize" in wow64.pe.exports + + +@windows_64bit_only +class TestSyswowRemoteProcess(object): + def test_remote_pebsyswow(self, proc32): + peb64 = proc32.peb_syswow + modules_names = [m.name for m in peb64.modules] + assert "wow64.dll" in modules_names + # Parsing + wow64 = [m for m in peb64.modules if m.name == "wow64.dll"][0] + assert "Wow64LdrpInitialize" in wow64.pe.exports + + + def test_getset_syswow_context(self, proc32): + addr = proc32.virtual_alloc(0x1000) + remote_python_code = """ + import windows + import windows.native_exec.simple_x64 as x64 + windows.utils.create_console() + x64_code = x64.assemble("mov r11, 0x1122334455667788; mov rax, 0x8877665544332211; mov [{0}], rax ;label :loop; jmp :loop; nop; nop; ret") + res = windows.syswow64.execute_64bits_code_from_syswow(x64_code) + print("res = {{0}}".format(hex(res))) + windows.current_process.write_qword({0}, res) + """.format(addr) + + t = proc32.execute_python_unsafe(textwrap.dedent(remote_python_code)) + # Wait for python execution + while proc32.read_qword(addr) != 0x8877665544332211: + pass + ctx = t.context_syswow + # Check the get context + assert ctx.R11 == 0x1122334455667788 + assert proc32.read_memory(ctx.Rip, 2) == x64.assemble("label :loop; jmp :loop") + t.suspend() + proc32.write_memory(ctx.Rip, "\x90\x90") + # Check the set context + RETURN_VALUE = 0x4041424344454647 + ctx.Rax = RETURN_VALUE + ctx.Rip += 2 + t.set_syswow_context(ctx) + t.resume() + t.wait() + assert RETURN_VALUE == proc32.read_qword(addr) + + diff --git a/pytesting/test_winproxy.py b/pytesting/test_winproxy.py new file mode 100644 index 0000000..1c93b4a --- /dev/null +++ b/pytesting/test_winproxy.py @@ -0,0 +1,13 @@ +import pytest + +import windows +import windows.generated_def as gdef + +from pfwtest import * + +pytestmark = pytest.mark.usefixtures('check_for_gc_garbage') + + +def test_createfileA_fail(): + with pytest.raises(WindowsError) as ar: + windows.winproxy.CreateFileA("NONEXISTFILE.FILE") \ No newline at end of file