From feb76b48c355c7d74dddfd001b502199fb01bcd7 Mon Sep 17 00:00:00 2001 From: Clement Rouault Date: Wed, 26 Apr 2017 11:07:52 +0200 Subject: [PATCH] unit tests now check for uncollectable-object (gc.garbage) --- TODO | 6 ++-- windows/debug/debugger.py | 3 +- windows/test/__init__.py | 2 +- windows/test/mytest.py | 63 +++++++++++++++++++++++++++++++---- windows/test/test_crypto.py | 4 +++ windows/test/test_debugger.py | 34 ++++++++++++++----- windows/test/test_hooks.py | 6 +++- windows/test/test_syswow.py | 4 +++ windows/test/test_utils.py | 15 +++++++++ 9 files changed, 115 insertions(+), 22 deletions(-) diff --git a/TODO b/TODO index 44de29d..8bb2d03 100644 --- a/TODO +++ b/TODO @@ -11,10 +11,10 @@ TODO: - Test !! (bp, BP_HX, bp on only on process, bp_hx on only one thread..) - test breakpoint with specific target - -Debugger + - Rethink/adapt Debugger._explicit_single_step - Does not handle case where EEFlags.TF was by the debugge before trigering the exception - Should set the flag explicitly in single_step ? and not just use EEFlags.TF ? + - Does not handle case where EEFlags.TF was by the debugge before trigering the exception + - Should set the flag explicitly in single_step ? and not just use EEFlags.TF ? - _handle_load_dll - error in keys of 'self._module_by_process[self.current_process.pid]' - if I have a ntdll32 and ntdll64: both would have the same name in the list.. diff --git a/windows/debug/debugger.py b/windows/debug/debugger.py index 7d30e1a..70a89a3 100644 --- a/windows/debug/debugger.py +++ b/windows/debug/debugger.py @@ -695,9 +695,8 @@ class Debugger(object): self.current_thread = None self.current_process = None - if cpid == self.target.pid: + if self.target and cpid == self.target.pid: self.target = None - return retvalue def _handle_create_thread(self, debug_event): diff --git a/windows/test/__init__.py b/windows/test/__init__.py index 57d22ea..9313cb7 100644 --- a/windows/test/__init__.py +++ b/windows/test/__init__.py @@ -1,7 +1,7 @@ from test_utils import * -from mytest import WindowsTestCase, WindowsAPITestCase, NativeUtilsTestCase, SystemTestCase +from mytest import WindowsTestCase, WindowsAPITestCase, NativeUtilsTestCase, SystemTestCase, GeneratedCodeTestCase from test_hooks import HookTestCase from test_debugger import DebuggerTestCase from test_syswow import SyswowTestCase diff --git a/windows/test/mytest.py b/windows/test/mytest.py index aa2e7aa..a0c9535 100644 --- a/windows/test/mytest.py +++ b/windows/test/mytest.py @@ -11,39 +11,49 @@ from windows.generated_def.winstructs import * class SystemTestCase(unittest.TestCase): + @check_for_gc_garbage def test_version(self): return windows.system.version + @check_for_gc_garbage def test_version_name(self): - return windows.system.version_name + return windows.system.version_name + @check_for_gc_garbage def test_computer_name(self): return windows.system.computer_name + @check_for_gc_garbage def test_services(self): return windows.system.services + @check_for_gc_garbage def test_logicaldrives(self): return windows.system.logicaldrives + @check_for_gc_garbage def test_processes(self): return windows.system.processes + @check_for_gc_garbage def test_threads(self): return windows.system.threads + @check_for_gc_garbage def test_wmi(self): return windows.system.wmi.select("Win32_Process", "*") + @check_for_gc_garbage def test_processes(self): procs = windows.system.processes self.assertIn(windows.current_process.pid, [p.pid for p in procs]) class WindowsTestCase(unittest.TestCase): - def setUp(self): - pass + # def setUp(self): + # pass + @check_for_gc_garbage def test_pop_calc_32(self): with Calc32() as calc: self.assertEqual(calc.bitness, 32) @@ -53,12 +63,15 @@ class WindowsTestCase(unittest.TestCase): with Calc64() as calc: self.assertEqual(calc.bitness, 64) + @check_for_gc_garbage def test_get_current_process_peb(self): return windows.current_process.peb + @check_for_gc_garbage def test_get_current_process_modules(self): self.assertIn("python", windows.current_process.peb.modules[0].name) + @check_for_gc_garbage def test_local_process_pe_imports(self): python_module = windows.current_process.peb.modules[0] imp = python_module.pe.imports @@ -67,6 +80,7 @@ class WindowsTestCase(unittest.TestCase): k32_base = windows.winproxy.LoadLibraryA("kernel32.dll") self.assertEqual(windows.winproxy.GetProcAddress(k32_base, "GetCurrentProcessId"), current_proc_id_iat.value) + @check_for_gc_garbage def test_local_process_pe_exports(self): mods = [m for m in windows.current_process.peb.modules if m.name == "kernel32.dll"] self.assertTrue(mods, 'Could not find "kernel32.dll" in current process modules') @@ -75,6 +89,7 @@ class WindowsTestCase(unittest.TestCase): k32_base = windows.winproxy.LoadLibraryA("kernel32.dll") self.assertEqual(windows.winproxy.GetProcAddress(k32_base, "GetCurrentProcessId"), get_current_proc_id) + @check_for_gc_garbage def test_local_process_pe_sections(self): mods = [m for m in windows.current_process.peb.modules if m.name == "kernel32.dll"] self.assertTrue(mods, 'Could not find "kernel32.dll" in current process modules') @@ -86,18 +101,20 @@ class WindowsTestCase(unittest.TestCase): sections[0].size # Read / write - + @check_for_gc_garbage def test_read_memory_32(self): with Calc32() as calc: k32 = [m for m in calc.peb.modules if m.name == "kernel32.dll"][0] self.assertEqual(calc.read_memory(k32.baseaddr, 2), "MZ") @windows_64bit_only + @check_for_gc_garbage def test_read_memory_64(self): with Calc64() as calc: k32 = [m for m in calc.peb.modules if m.name == "kernel32.dll"][0] self.assertEqual(calc.read_memory(k32.baseaddr, 2), "MZ") + @check_for_gc_garbage def test_write_memory_32(self): with Calc32() as calc: k32 = [m for m in calc.peb.modules if m.name == "kernel32.dll"][0] @@ -106,6 +123,7 @@ class WindowsTestCase(unittest.TestCase): self.assertEqual(calc.read_memory(k32.baseaddr, 2), "XD") @windows_64bit_only + @check_for_gc_garbage def test_write_memory_64(self): with Calc64() as calc: k32 = [m for m in calc.peb.modules if m.name == "kernel32.dll"][0] @@ -113,7 +131,7 @@ class WindowsTestCase(unittest.TestCase): calc.write_memory(k32.baseaddr, "XD") self.assertEqual(calc.read_memory(k32.baseaddr, 2), "XD") - + @check_for_gc_garbage def test_read_string(self): test_string = "TEST_STRING" string_to_write = test_string + "\x00" @@ -122,6 +140,7 @@ class WindowsTestCase(unittest.TestCase): calc.write_memory(addr, string_to_write) self.assertEqual(calc.read_string(addr), test_string) + @check_for_gc_garbage def test_read_string_end_page(self): test_string = "TEST_STRING" string_to_write = test_string + "\x00" @@ -130,6 +149,7 @@ class WindowsTestCase(unittest.TestCase): calc.write_memory(addr, string_to_write) self.assertEqual(calc.read_string(addr), test_string) + @check_for_gc_garbage def test_read_wstring(self): test_string = "TEST_STRING" string_to_write = test_string + "\x00" @@ -138,7 +158,7 @@ class WindowsTestCase(unittest.TestCase): calc.write_memory(addr, "\x00".join(string_to_write)) self.assertEqual(calc.read_wstring(addr), test_string) - + @check_for_gc_garbage def test_read_wstring_end_page(self): test_string = "TEST_STRING" string_to_write = test_string + "\x00" @@ -149,6 +169,7 @@ class WindowsTestCase(unittest.TestCase): self.assertEqual(calc.read_wstring(addr), test_string) # Native execution + @check_for_gc_garbage def test_execute_to_32(self): with Calc32() as calc: data = calc.virtual_alloc(0x1000) @@ -162,6 +183,7 @@ class WindowsTestCase(unittest.TestCase): self.assertEqual(dword, 0x42424242) @windows_64bit_only + @check_for_gc_garbage def test_execute_to_64(self): with Calc64() as calc: data = calc.virtual_alloc(0x1000) @@ -176,6 +198,7 @@ class WindowsTestCase(unittest.TestCase): # Python execution @windows_64bit_only + @check_for_gc_garbage def test_execute_python_to_64(self): with Calc64() as calc: data = calc.virtual_alloc(0x1000) @@ -184,6 +207,7 @@ class WindowsTestCase(unittest.TestCase): dword = struct.unpack(" not found".format(page_target)) @windows_64bit_only + @check_for_gc_garbage def test_get_working_set_64(self): with Calc64() as calc: k32 = [m for m in calc.peb.modules if m.name == "kernel32.dll"][0] @@ -445,6 +488,7 @@ class WindowsTestCase(unittest.TestCase): else: raise ValueError("query_working_set page info for <0x{0:x}> not found".format(page_target)) + @check_for_gc_garbage def test_get_working_setex_32(self): with Calc32() as calc: k32 = [m for m in calc.peb.modules if m.name == "kernel32.dll"][0] @@ -474,6 +518,7 @@ class WindowsTestCase(unittest.TestCase): raise ValueError("query_working_set page info for <0x{0:x}> not found".format(page_target)) @windows_64bit_only + @check_for_gc_garbage def test_get_working_setex_64(self): with Calc64() as calc: k32 = [m for m in calc.peb.modules if m.name == "kernel32.dll"][0] @@ -504,6 +549,7 @@ class WindowsTestCase(unittest.TestCase): else: raise ValueError("query_working_set page info for <0x{0:x}> not found".format(page_target)) + @check_for_gc_garbage def test_mapped_filename_32(self): with Calc32() as calc: k32 = [m for m in calc.peb.modules if m.name == "kernel32.dll"][0] @@ -511,18 +557,21 @@ class WindowsTestCase(unittest.TestCase): self.assertTrue(mapped_filname.endswith("kernel32.dll")) @windows_64bit_only + @check_for_gc_garbage def test_mapped_filename_64(self): with Calc64() as calc: k32 = [m for m in calc.peb.modules if m.name == "kernel32.dll"][0] mapped_filname = calc.get_mapped_filename(k32.baseaddr) self.assertTrue(mapped_filname.endswith("kernel32.dll")) + @check_for_gc_garbage def test_thread_teb_base_32(self): with Calc32() as calc: t = calc.threads[0] self.assertNotEqual(t.teb_base, 0) @windows_64bit_only + @check_for_gc_garbage def test_thread_teb_base_64(self): with Calc64() as calc: t = calc.threads[0] diff --git a/windows/test/test_crypto.py b/windows/test/test_crypto.py index 9c69e1f..a5712b8 100644 --- a/windows/test/test_crypto.py +++ b/windows/test/test_crypto.py @@ -59,11 +59,13 @@ class CryptoTestCase(unittest.TestCase): cls.raw_cert = TEST_CERT.decode("base64") cls.raw_pfx = TEST_PFX.decode("base64") + @check_for_gc_garbage def test_certificate(self): cert = windows.crypto.CertificateContext.from_buffer(self.raw_cert) self.assertEqual(cert.serial, '1b 8e 94 cb 0b 3e eb b6 41 39 f3 c9 09 b1 6b 46') self.assertEqual(cert.name, 'PythonForWindowsTest') + @check_for_gc_garbage def test_pfx(self): pfx = windows.crypto.import_pfx(self.raw_pfx, TEST_PFX_PASSWORD) orig_cert = windows.crypto.CertificateContext.from_buffer(self.raw_cert) @@ -72,10 +74,12 @@ class CryptoTestCase(unittest.TestCase): # Test cert comparaison self.assertEqual(certs[0], orig_cert) + @check_for_gc_garbage def test_open_pfx_bad_password(self): with self.assertRaises(WindowsError) as ar: pfx = windows.crypto.import_pfx(self.raw_pfx, "BadPassword") + @check_for_gc_garbage def test_encrypt_decrypt(self): message_to_encrypt = "Testing message \xff\x01" cert = windows.crypto.CertificateContext.from_buffer(self.raw_cert) diff --git a/windows/test/test_debugger.py b/windows/test/test_debugger.py index 5ee7b81..2928bd7 100644 --- a/windows/test/test_debugger.py +++ b/windows/test/test_debugger.py @@ -5,9 +5,11 @@ import threading import os class DebuggerTestCase(unittest.TestCase): + @check_for_gc_garbage def debuggable_calc_32(self): return windows.utils.create_process(r"C:\python27\python.exe", dwCreationFlags=DEBUG_PROCESS | CREATE_NEW_CONSOLE, show_windows=True) + @check_for_gc_garbage def test_init_breakpoint_callback(self): """Checking that the initial breakpoint call `on_exception`""" TEST_CASE = self @@ -20,6 +22,7 @@ class DebuggerTestCase(unittest.TestCase): d = MyDbg(calc) d.loop() + @check_for_gc_garbage def test_simple_standard_breakpoint(self): """Check that a standard Breakpoint method `trigger` is called with the correct informations""" TEST_CASE = self @@ -74,6 +77,7 @@ class DebuggerTestCase(unittest.TestCase): # d.add_bp(TSTBP(LdrLoadDll32)) # d.loop() + @check_for_gc_garbage def test_simple_hwx_breakpoint(self): """Test that simple HXBP are trigger""" TEST_CASE = self @@ -98,6 +102,7 @@ class DebuggerTestCase(unittest.TestCase): d.add_bp(TSTBP(LdrLoadDll32)) d.loop() + @check_for_gc_garbage def test_multiple_hwx_breakpoint(self): """Checking that multiple succesives HXBP are properly triggered""" TEST_CASE = self @@ -130,6 +135,7 @@ class DebuggerTestCase(unittest.TestCase): # Used to verif we actually called the Breakpoints TEST_CASE.assertEqual(data[0], 4) + @check_for_gc_garbage def test_four_hwx_breakpoint_fail(self): """Check that setting 4HXBP in the same thread fails""" TEST_CASE = self @@ -160,6 +166,7 @@ class DebuggerTestCase(unittest.TestCase): # Used to verif we actually NOT called the Breakpoints TEST_CASE.assertEqual(data[0], 0) + @check_for_gc_garbage def test_hwx_breakpoint_are_on_all_thread(self): """Checking that HXBP without target are set on all threads""" TEST_CASE = self @@ -196,6 +203,7 @@ class DebuggerTestCase(unittest.TestCase): # Used to verif we actually called the Breakpoints TEST_CASE.assertEqual(data[0], 2) + @check_for_gc_garbage def test_simple_breakpoint_name_addr(self): """Check breakpoint address resolution for format dll!api""" TEST_CASE = self @@ -219,6 +227,7 @@ class DebuggerTestCase(unittest.TestCase): d.loop() TEST_CASE.assertEqual(data[0], 1) + @check_for_gc_garbage def test_simple_hardware_breakpoint_name_addr(self): """Check HXBP address resolution for format dll!api""" TEST_CASE = self @@ -242,6 +251,7 @@ class DebuggerTestCase(unittest.TestCase): d.loop() TEST_CASE.assertEqual(data[0], 1) + @check_for_gc_garbage def perform_manual_getproc_loadlib_32(self, target, dll_name): dll = "KERNEL32.DLL\x00".encode("utf-16-le") api = "LoadLibraryA\x00" @@ -273,7 +283,7 @@ class DebuggerTestCase(unittest.TestCase): t = target.execute(RemoteManualLoadLibray.get_code(), addr4) return t - + @check_for_gc_garbage def test_hardware_breakpoint_name_addr(self): """Check that name addr in HXBP are trigger in all threads""" TEST_CASE = self @@ -288,7 +298,7 @@ class DebuggerTestCase(unittest.TestCase): data[0] += 1 if data[0] == 1: # Perform a loaddll in a new thread :) - # See if it's trigger a bp + # See if it triggers a bp t = TEST_CASE.perform_manual_getproc_loadlib_32(dbg.current_process, "wintrust.dll") self.new_thread = t if hasattr(self, "new_thread") and dbg.current_thread.tid == self.new_thread.tid: @@ -303,6 +313,7 @@ class DebuggerTestCase(unittest.TestCase): d.loop() #TEST_CASE.assertEqual(data[0], 1) + @check_for_gc_garbage def test_single_step(self): """Check that BP/dbg can trigger single step and that instruction follows""" TEST_CASE = self @@ -337,7 +348,7 @@ class DebuggerTestCase(unittest.TestCase): for i in range(NB_SINGLE_STEP): TEST_CASE.assertEqual(data[i], addr + 1 + i) - + @check_for_gc_garbage def test_single_step_hxbp(self): """Check that HXBPBP/dbg can trigger single step""" TEST_CASE = self @@ -372,7 +383,7 @@ class DebuggerTestCase(unittest.TestCase): for i in range(NB_SINGLE_STEP): TEST_CASE.assertEqual(data[i], addr + 1 + i) - + @check_for_gc_garbage def test_memory_breakpoint_write(self): """Check MemoryBP WRITE""" TEST_CASE = self @@ -416,6 +427,7 @@ class DebuggerTestCase(unittest.TestCase): # Used to verif we actually called the Breakpoints for the good addresses TEST_CASE.assertEqual(store_data[0], 2) + @check_for_gc_garbage def test_memory_breakpoint_exec(self): """Check MemoryBP EXEC""" TEST_CASE = self @@ -444,7 +456,7 @@ class DebuggerTestCase(unittest.TestCase): for i in range(NB_NOP_IN_PAGE + 1): TEST_CASE.assertEqual(data[i], addr + i) - + @check_for_gc_garbage def test_standard_breakpoint_self_remove(self): TEST_CASE = self data = [] @@ -471,6 +483,7 @@ class DebuggerTestCase(unittest.TestCase): d.loop() TEST_CASE.assertEqual(data, [u"FILENAME1", u"FILENAME2"]) + @check_for_gc_garbage def test_standard_breakpoint_remove(self): TEST_CASE = self data = [] @@ -497,6 +510,7 @@ class DebuggerTestCase(unittest.TestCase): d.loop() TEST_CASE.assertEqual(data, [u"FILENAME1", u"FILENAME2"]) + @check_for_gc_garbage def test_hxbp_breakpoint_remove(self): TEST_CASE = self data = [] @@ -523,6 +537,7 @@ class DebuggerTestCase(unittest.TestCase): d.loop() TEST_CASE.assertEqual(data, [u"FILENAME1", u"FILENAME2"]) + @check_for_gc_garbage def test_hxbp_breakpoint_self_remove(self): TEST_CASE = self data = [] @@ -550,7 +565,7 @@ class DebuggerTestCase(unittest.TestCase): d.loop() TEST_CASE.assertEqual(data, [u"FILENAME1", u"FILENAME2"]) - + @check_for_gc_garbage def test_mem_breakpoint_remove(self): TEST_CASE = self data = [] @@ -585,6 +600,7 @@ class DebuggerTestCase(unittest.TestCase): d.loop() TEST_CASE.assertEqual(data, [data_addr, data_addr + 4]) + @check_for_gc_garbage def test_mem_breakpoint_self_remove(self): TEST_CASE = self data = [] @@ -620,7 +636,7 @@ class DebuggerTestCase(unittest.TestCase): d.loop() TEST_CASE.assertEqual(data, [data_addr, data_addr + 4]) - + @check_for_gc_garbage def test_read_write_bp_same_page(self): TEST_CASE = self data = [] @@ -670,6 +686,7 @@ class DebuggerTestCase(unittest.TestCase): TEST_CASE.assertEqual(data, expected_result) + @check_for_gc_garbage def test_exe_in_module_list(self): class MyDbg(windows.debug.Debugger): def on_exception(self, exception): @@ -683,7 +700,7 @@ class DebuggerTestCase(unittest.TestCase): d = MyDbg(calc) d.loop() - + @check_for_gc_garbage def test_exe_in_module_list(self): class MyDbg(windows.debug.Debugger): def on_exception(self, exception): @@ -697,6 +714,7 @@ class DebuggerTestCase(unittest.TestCase): d = MyDbg(calc) d.loop() + @check_for_gc_garbage def test_bp_exe_by_name(self): NBCALL = [0] TEST_CASE = self diff --git a/windows/test/test_hooks.py b/windows/test/test_hooks.py index 07fd1c4..d227da4 100644 --- a/windows/test/test_hooks.py +++ b/windows/test/test_hooks.py @@ -6,7 +6,7 @@ from windows.generated_def.winstructs import * class HookTestCase(unittest.TestCase): - + @check_for_gc_garbage 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] @@ -29,6 +29,7 @@ class HookTestCase(unittest.TestCase): # Remove the hook x.disable() + @check_for_gc_garbage 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] @@ -46,6 +47,7 @@ class HookTestCase(unittest.TestCase): self.assertEqual(ar.exception.winerror, 0x11223344) x.disable() + @check_for_gc_garbage def test_self_iat_hook_multithread(self): """Test IAT hook in current process with multi thread trigger""" cp = windows.current_process @@ -73,6 +75,7 @@ class HookTestCase(unittest.TestCase): self.assertEqual(len(calling_thread), 2) x.disable() + @check_for_gc_garbage def test_remote_iat_hook_32(self): with Calc32() as calc: calc.execute_python("import windows") @@ -124,6 +127,7 @@ class HookTestCase(unittest.TestCase): t.wait() self.assertEqual(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") diff --git a/windows/test/test_syswow.py b/windows/test/test_syswow.py index fc19024..28c39e9 100644 --- a/windows/test/test_syswow.py +++ b/windows/test/test_syswow.py @@ -8,6 +8,7 @@ from windows.generated_def.winstructs import * class SyswowTestCase(unittest.TestCase): @windows_64bit_only @process_32bit_only + @check_for_gc_garbage 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) @@ -15,6 +16,7 @@ class SyswowTestCase(unittest.TestCase): @windows_64bit_only @process_32bit_only + @check_for_gc_garbage def test_self_pebsyswow(self): peb64 = windows.current_process.peb_syswow modules_names = [m.name for m in peb64.modules] @@ -24,6 +26,7 @@ class SyswowTestCase(unittest.TestCase): self.assertIn("Wow64LdrpInitialize", wow64.pe.exports) @windows_64bit_only + @check_for_gc_garbage def test_remote_pebsyswow(self): with Calc32() as calc: peb64 = calc.peb_syswow @@ -34,6 +37,7 @@ class SyswowTestCase(unittest.TestCase): self.assertIn("Wow64LdrpInitialize", wow64.pe.exports) @windows_64bit_only + @check_for_gc_garbage def test_getset_syswow_context(self): with Calc32() as calc: addr = calc.virtual_alloc(0x1000) diff --git a/windows/test/test_utils.py b/windows/test/test_utils.py index e48fb43..401d252 100644 --- a/windows/test/test_utils.py +++ b/windows/test/test_utils.py @@ -8,6 +8,8 @@ import windows.native_exec.simple_x64 as x64 import windows.native_exec.nativeutils as nativeutils from windows.generated_def import CREATE_NEW_CONSOLE +import gc + is_process_32_bits = windows.current_process.bitness == 32 is_process_64_bits = windows.current_process.bitness == 64 @@ -75,6 +77,19 @@ def Calc32(dwCreationFlags=DEFAULT_CREATION_FLAGS, exit_code=0): calc.exit(exit_code) +def check_for_gc_garbage(f): + def wrapper(testcase, *args, **kwargs): + garbage_before = set(gc.garbage) + res = f(testcase, *args, **kwargs) + gc.collect() + new_garbage = set(gc.garbage) - garbage_before + testcase.assertFalse(new_garbage, "Test generated uncollectable object ({0})".format(new_garbage)) + return res + return wrapper + + + + def print_call(f): def wrapper(*args, **kwargs): res = f(*args, **kwargs)