From b3117ce4e36400c562c49382cf30869916c61663 Mon Sep 17 00:00:00 2001 From: hakril Date: Sun, 10 Sep 2017 18:57:59 +0200 Subject: [PATCH] pytest integration finished + moved tests/ to top dir --- pytesting/pytest.ini | 2 - {pytesting => tests}/__init__.py | 0 {pytesting => tests}/conftest.py | 5 +- {pytesting => tests}/dbg_injection.py | 0 {pytesting => tests}/pfwtest.py | 0 tests/pytest.ini | 0 {pytesting => tests}/test_crypto.py | 0 {pytesting => tests}/test_debugger.py | 10 +- {pytesting => tests}/test_generated_code.py | 0 {pytesting => tests}/test_hook.py | 0 {pytesting => tests}/test_native_utils.py | 0 {pytesting => tests}/test_process.py | 6 - tests/test_simple_x64.py | 274 +++++++ .../native_exec => tests}/test_simple_x86.py | 189 +++-- {pytesting => tests}/test_system.py | 1 - {pytesting => tests}/test_syswow.py | 0 {pytesting => tests}/test_winproxy.py | 0 windows/native_exec/test_simple_x64.py | 256 ------ windows/test/__init__.py | 13 - windows/test/mytest.py | 763 ------------------ windows/test/test_crypto.py | 100 --- windows/test/test_debugger.py | 746 ----------------- windows/test/test_hooks.py | 183 ----- windows/test/test_syswow.py | 74 -- windows/test/test_utils.py | 132 --- 25 files changed, 383 insertions(+), 2371 deletions(-) delete mode 100644 pytesting/pytest.ini rename {pytesting => tests}/__init__.py (100%) rename {pytesting => tests}/conftest.py (97%) rename {pytesting => tests}/dbg_injection.py (100%) rename {pytesting => tests}/pfwtest.py (100%) create mode 100644 tests/pytest.ini rename {pytesting => tests}/test_crypto.py (100%) rename {pytesting => tests}/test_debugger.py (98%) rename {pytesting => tests}/test_generated_code.py (100%) rename {pytesting => tests}/test_hook.py (100%) rename {pytesting => tests}/test_native_utils.py (100%) rename {pytesting => tests}/test_process.py (99%) create mode 100644 tests/test_simple_x64.py rename {windows/native_exec => tests}/test_simple_x86.py (53%) rename {pytesting => tests}/test_system.py (97%) rename {pytesting => tests}/test_syswow.py (100%) rename {pytesting => tests}/test_winproxy.py (100%) delete mode 100644 windows/native_exec/test_simple_x64.py delete mode 100644 windows/test/__init__.py delete mode 100644 windows/test/mytest.py delete mode 100644 windows/test/test_crypto.py delete mode 100644 windows/test/test_debugger.py delete mode 100644 windows/test/test_hooks.py delete mode 100644 windows/test/test_syswow.py delete mode 100644 windows/test/test_utils.py diff --git a/pytesting/pytest.ini b/pytesting/pytest.ini deleted file mode 100644 index 8c12560..0000000 --- a/pytesting/pytest.ini +++ /dev/null @@ -1,2 +0,0 @@ -[pytest] -usefixtures = check_for_handle_leak_final \ No newline at end of file diff --git a/pytesting/__init__.py b/tests/__init__.py similarity index 100% rename from pytesting/__init__.py rename to tests/__init__.py diff --git a/pytesting/conftest.py b/tests/conftest.py similarity index 97% rename from pytesting/conftest.py rename to tests/conftest.py index e3f1a30..b6d050f 100644 --- a/pytesting/conftest.py +++ b/tests/conftest.py @@ -176,6 +176,9 @@ def pytest_terminal_summary(terminalreporter, exitstatus): for item in items: descr = item.description() if descr is None: - descr = item.name + try: + descr = item.name + except Exception as e: + descr = repr(e) terminalreporter.write_line(" * <{0}>".format(descr) , Purple=True, bold=True) terminalreporter.write_line("") \ No newline at end of file diff --git a/pytesting/dbg_injection.py b/tests/dbg_injection.py similarity index 100% rename from pytesting/dbg_injection.py rename to tests/dbg_injection.py diff --git a/pytesting/pfwtest.py b/tests/pfwtest.py similarity index 100% rename from pytesting/pfwtest.py rename to tests/pfwtest.py diff --git a/tests/pytest.ini b/tests/pytest.ini new file mode 100644 index 0000000..e69de29 diff --git a/pytesting/test_crypto.py b/tests/test_crypto.py similarity index 100% rename from pytesting/test_crypto.py rename to tests/test_crypto.py diff --git a/pytesting/test_debugger.py b/tests/test_debugger.py similarity index 98% rename from pytesting/test_debugger.py rename to tests/test_debugger.py index f31eacf..50ef689 100644 --- a/pytesting/test_debugger.py +++ b/tests/test_debugger.py @@ -11,8 +11,6 @@ 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', "check_for_handle_leak") - 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) @@ -43,7 +41,6 @@ def get_debug_process_ndll(proc): 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): @@ -58,7 +55,6 @@ def test_simple_standard_breakpoint(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""" @@ -106,8 +102,7 @@ def test_multiple_hwx_breakpoint(proc32_64_debug): # Used to verif we actually called the Breakpoints assert TSTBP.COUNTER == 4 -# @check_for_gc_garbage -# @check_for_handle_leak + def test_four_hwx_breakpoint_fail(proc32_64_debug): """Check that setting 4HXBP in the same thread fails""" # print("test_four_hwx_breakpoint_fail {0}".format(proc32_64_debug)) @@ -167,8 +162,7 @@ def test_hwx_breakpoint_are_on_all_thread(proc32_64_debug): # Used to verif we actually called the Breakpoints assert TSTBP.COUNTER == 2 -# @check_for_handle_leak -# @check_for_gc_garbage + @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""" diff --git a/pytesting/test_generated_code.py b/tests/test_generated_code.py similarity index 100% rename from pytesting/test_generated_code.py rename to tests/test_generated_code.py diff --git a/pytesting/test_hook.py b/tests/test_hook.py similarity index 100% rename from pytesting/test_hook.py rename to tests/test_hook.py diff --git a/pytesting/test_native_utils.py b/tests/test_native_utils.py similarity index 100% rename from pytesting/test_native_utils.py rename to tests/test_native_utils.py diff --git a/pytesting/test_process.py b/tests/test_process.py similarity index 99% rename from pytesting/test_process.py rename to tests/test_process.py index a8d183a..8ad98cd 100644 --- a/pytesting/test_process.py +++ b/tests/test_process.py @@ -63,13 +63,7 @@ class TestCurrentProcessWithCheckGarbage(object): assert isinstance(token.integrity, (int, long)) assert isinstance(token.is_elevated, (bool)) -@check_for_handle_leak -def test_yolo(proc32_64): - print(proc32_64) - print(proc32_64.handle) - -@check_for_handle_leak @check_for_gc_garbage class TestProcessWithCheckGarbage(object): def test_pop_proc_32(self, proc32): diff --git a/tests/test_simple_x64.py b/tests/test_simple_x64.py new file mode 100644 index 0000000..130fbb2 --- /dev/null +++ b/tests/test_simple_x64.py @@ -0,0 +1,274 @@ +try: + import capstone +except ImportError as e: + capstone = None +import pytest + +import windows.native_exec.simple_x64 as x64 +from windows.native_exec.simple_x64 import * +del Test # Prevent pytest warning + +if capstone: + disassembleur = capstone.Cs(capstone.CS_ARCH_X86, capstone.CS_MODE_64) + disassembleur.detail = True + + +def disas(x): + return list(disassembleur.disasm(x, 0)) + +mnemonic_name_exception = {'movabs': 'mov'} + + + + +class CheckInstr(object): + def __init__(self, instr_to_test, expected_result=None, immediat_accepted=None, must_fail=None, debug=False): + self.instr_to_test = instr_to_test + self.immediat_accepted = immediat_accepted + self.expected_result = expected_result + self.must_fail = must_fail + self.debug = debug + + def __call__(self, *args): + try: + if self.debug: + import pdb;pdb.set_trace() + pdb.DONE = True + x64.DEBUG = self.debug + res = bytes(self.instr_to_test(*args).get_code()) + if self.debug: + print(repr(res)) + except ValueError as e: + if self.must_fail == True: + return True + else: + raise + else: + if self.must_fail: + raise ValueError("Instruction did not failed as expected") + capres_list = disas(res) + if len(capres_list) != 1: + raise AssertionError("Trying to disas an instruction resulted in multiple disassembled instrs") + capres = capres_list[0] + print("{0} {1}".format(capres.mnemonic, capres.op_str)) + if self.expected_result is not None: + if "{0} {1}".format(capres.mnemonic, capres.op_str) == self.expected_result: + return True + else: + raise AssertionError("Expected result <{0}> got <{1}>".format(self.expected_result, "{0} {1}".format(capres.mnemonic, capres.op_str))) + if len(res) != len(capres.bytes): + print("<{0}> vs <{1}>".format(repr(res), repr(capres.bytes))) + raise AssertionError("Not all bytes have been used by the disassembler") + self.compare_mnemo(capres) + self.compare_args(args, capres) + + def compare_mnemo(self, capres): + expected = self.instr_to_test.__name__.lower() + cap_mnemo = mnemonic_name_exception.get(str(capres.mnemonic), str(capres.mnemonic)) + if expected != cap_mnemo: + raise AssertionError("Expected menmo {0} got {1}".format(expected, str(capres.mnemonic))) + return True + + def compare_args(self, args, capres): + capres_op = list(capres.operands) + if len(args) != len(capres_op): + raise AssertionError("Expected {0} operands got {1}".format(len(args), len(capres_op))) + for op_args, cap_op in zip(args, capres_op): + if isinstance(op_args, str): # Register + if cap_op.type != capstone.x86.X86_OP_REG: + raise AssertionError("Expected args {0} operands got {1}".format(op_args, capres_op)) + if op_args.lower() != capres.reg_name(cap_op.reg).lower(): + raise AssertionError("Expected register <{0}> got {1}".format(op_args.lower(), capres.reg_name(cap_op.reg).lower())) + elif isinstance(op_args, (int, long)): + if (op_args != cap_op.imm) and not (self.immediat_accepted and self.immediat_accepted == cap_op.imm): + raise AssertionError("Expected Immediat <{0}> got {1}".format(op_args, cap_op.imm)) + elif isinstance(op_args, mem_access): + self.compare_mem_access(op_args, capres, cap_op) + else: + raise ValueError("Unknow argument {0} of type {1}".format(op_args, type(op_args))) + + def compare_mem_access(self, memaccess, capres, cap_op): + if cap_op.type != capstone.x86.X86_OP_MEM: + raise AssertionError("Expected Memaccess <{0}> got {1}".format(memaccess, cap_op)) + if memaccess.prefix is not None and capres.prefix[1] != x64_segment_selectors[memaccess.prefix].PREFIX_VALUE: + try: + get_prefix = [n for n, x in x64_segment_selectors.items() if x.PREFIX_VALUE == capres.prefix[1]][0] + except IndexError: + get_prefix = None + raise AssertionError("Expected Segment overide <{0}> got {1}".format(memaccess.prefix, get_prefix)) + cap_mem = cap_op.mem + if memaccess.base is None and cap_mem.base != capstone.x86.X86_REG_INVALID: + raise AssertionError("Unexpected memaccess base <{0}>".format(capres.reg_name(cap_mem.base))) + if memaccess.base is not None and capres.reg_name(cap_mem.base) != memaccess.base.lower(): + raise AssertionError("Expected mem.base {0} got {1}".format(memaccess.base.lower(), capres.reg_name(cap_mem.base))) + if memaccess.index is None and cap_mem.index != capstone.x86.X86_REG_INVALID: + raise AssertionError("Unexpected memaccess index <{0}>".format(capres.reg_name(cap_mem.base))) + if memaccess.index is not None and capres.reg_name(cap_mem.index) != memaccess.index.lower(): + raise AssertionError("Expected mem.index {0} got {1}".format(memaccess.index.lower(), capres.reg_name(cap_mem.index))) + if memaccess.scale != cap_mem.scale and not (memaccess.scale is None and cap_mem.scale == 1): + raise AssertionError("Expected mem.scale {0} got {1}".format(memaccess.scale, cap_mem.scale)) + if memaccess.disp != cap_mem.disp: + raise AssertionError("Expected mem.disp {0} got {1}".format(memaccess.disp, cap_mem.disp)) + + +def test_assembler(): + CheckInstr(Add)('RAX', 'RSP') + CheckInstr(Add)('RAX', mem('[RCX]')) + CheckInstr(Add)('RAX', mem('[RDI + 0x10]')) + CheckInstr(Add)('RAX', mem('[RSI + 0x7fffffff]')) + CheckInstr(Add)('RAX', mem('[RSI + -0x1]')) + CheckInstr(Add)('RAX', mem('[0x10]')) + CheckInstr(Add)('RAX', mem('fs:[0x10]')) + CheckInstr(Add)('RAX', mem('[RSI + RDI * 2]')) + CheckInstr(Add)('RAX', mem('[RSI + RDI * 2 + 0x10]')) + CheckInstr(Add)('RAX', mem('gs:[RSI + RDI * 2 + 0x10]')) + CheckInstr(Add)('RAX', mem('[R15 * 8 + 0x10]')) + CheckInstr(Add)('RAX', mem('[R9 + R8 * 2 + 0x7fffffff]')) + CheckInstr(Add)('RAX', mem('[R9 + R8 * 2 + -0x80000000]')) + CheckInstr(Add)('RAX', mem('[-1]')) + CheckInstr(Add)('RAX', mem('[0x7fffffff]')) + CheckInstr(Add)('RAX', -1) + + + CheckInstr(Sub)('RCX', 'RSP') + CheckInstr(Sub)('RCX', mem('[RSP]')) + + CheckInstr(Xor)('R15', mem('[RAX + R8 * 2 + 0x11223344]')) + CheckInstr(Xor)('RAX', 'RAX') + CheckInstr(Cmp)('RAX', -1) + #CheckInstr(Cmp, immediat_accepted=-1)('RAX', 0xffffffff) + CheckInstr(Lea)('RAX', mem('[RAX + 1]')) + CheckInstr(Lea)('RAX', mem('fs:[RAX + 1]')) + CheckInstr(Mov)('RAX', mem('[0x1122334455667788]')) + CheckInstr(Mov)('RAX', mem('gs:[0x1122334455667788]')) + CheckInstr(Mov)('RAX', mem('gs:[0x60]')) + CheckInstr(Mov)('RCX', 0x1122334455667788) + CheckInstr(Mov)('RCX', -1) + CheckInstr(Mov)('RCX', -0x1000) + CheckInstr(Mov)('RCX', 0xffffffff) + CheckInstr(Mov)('RAX', 0xffffffff) + CheckInstr(Mov)('R8', 0x1122334455667788) + CheckInstr(Mov)('RCX', -1) + CheckInstr(Mov, immediat_accepted=-1)('RCX', 0xffffffffffffffff) + CheckInstr(Mov)(mem('gs:[0x1122334455667788]'), 'RAX') + CheckInstr(Mov)(mem('[RAX]'), 0x11223344) + CheckInstr(Mov)(mem('[EAX]'), 0x11223344) + CheckInstr(Mov)(mem('[RBX]'), 0x11223344) + CheckInstr(Mov)("R12", mem("[RAX]")) + CheckInstr(Mov)("RAX", mem("[R12]")) + CheckInstr(Mov)("RAX", mem("[RAX + R12]")) + CheckInstr(Mov)("RAX", mem("[R12 + R12]")) + CheckInstr(Mov)("RAX", mem("[R12 + R15]")) + + CheckInstr(Mov)("RAX", mem("[R10]")) + CheckInstr(Mov)("RAX", mem("[R11]")) + CheckInstr(Mov)("RAX", mem("[R12]")) + CheckInstr(Mov)("RAX", mem("[R13]")) + CheckInstr(Mov)("RAX", mem("[R14]")) + CheckInstr(Mov)("RAX", mem("[R15]")) + + #CheckInstr(Mov)("RSI", mem("[R12]")) + + CheckInstr(And)('RCX', 'RBX') + CheckInstr(And)('RAX', 0x11223344) + CheckInstr(And)('EAX', 0x11223344) + CheckInstr(And)('EAX', 0xffffffff) + CheckInstr(And)('RAX', mem('[RAX + 1]')) + CheckInstr(And)(mem('[RAX + 1]'), 'R8') + CheckInstr(And)(mem('[EAX + 1]'), 'R8') + CheckInstr(And)(mem('[RAX + 1]'), 'EAX') + + CheckInstr(Or)('RCX', 'RBX') + CheckInstr(Or)('RAX', 0x11223344) + CheckInstr(Or)('RAX', mem('[RAX + 1]')) + CheckInstr(Or)(mem('[RAX + 1]'), 'R8') + CheckInstr(Or)(mem('[EAX + 1]'), 'R8') + CheckInstr(Or)(mem('[RAX + 1]'), 'EAX') + + CheckInstr(Shr)('RAX', 8) + CheckInstr(Shr)('R15', 0x12) + CheckInstr(Shl)('RAX', 8) + CheckInstr(Shl)('R15', 0x12) + + # I really don't know why it's the inverse + # But I don't care, it's Test dude.. + CheckInstr(x64.Test, expected_result="test r11, rax")('RAX', 'R11') + CheckInstr(x64.Test, expected_result="test edi, eax")('EAX', 'EDI') + CheckInstr(x64.Test)('RCX', 'RCX') + + CheckInstr(x64.Test)(mem('[RDI + 0x100]'), 'RCX') + + assert x64.Test(mem('[RDI + 0x100]'), 'RCX').get_code() == x64.Test('RCX', mem('[RDI + 0x100]')).get_code() + + + CheckInstr(Push)('RAX') + assert len(Push("RAX").get_code()) == 1 + CheckInstr(Push)('R15') + CheckInstr(Push)(0x42) + CheckInstr(Push)(-1) + CheckInstr(Push)(mem("[ECX]")) + CheckInstr(Push)(mem("[RCX]")) + + + CheckInstr(Pop)('RAX') + assert len(Pop("RAX").get_code()) == 1 + + + CheckInstr(Call)('RAX') + CheckInstr(Call)(mem('[RAX + RCX * 8]')) + CheckInstr(Cpuid)() + CheckInstr(Xchg)('RAX', 'RSP') + assert Xchg('RAX', 'RCX').get_code() == Xchg('RCX', 'RAX').get_code() + + # 32 / 64 bits register mixing + CheckInstr(Mov)('ECX', 'EBX') + CheckInstr(Mov)('RCX', mem('[EBX]')) + CheckInstr(Mov)('ECX', mem('[RBX]')) + CheckInstr(Mov)('ECX', mem('[EBX]')) + CheckInstr(Mov)('RCX', mem('[EBX + EBX]')) + CheckInstr(Mov)('RCX', mem('[ESP + EBX + 0x10]')) + CheckInstr(Mov)('ECX', mem('[ESP + EBX + 0x10]')) + CheckInstr(Mov)('ECX', mem('[RBX + RCX + 0x10]')) + + CheckInstr(Mov)(mem('[RBX + RCX + 0x10]'), 'ECX') + CheckInstr(Mov)(mem('[EBX + ECX + 0x10]'), 'ECX') + CheckInstr(Mov)(mem('[EBX + ECX + 0x10]'), 'R8') + + CheckInstr(Not)('RAX') + CheckInstr(Not)(mem('[RAX]')) + + + CheckInstr(ScasB, expected_result="scasb al, byte ptr [rdi]")() + CheckInstr(ScasW, expected_result="scasw ax, word ptr [rdi]")() + CheckInstr(ScasD, expected_result="scasd eax, dword ptr [rdi]")() + CheckInstr(ScasQ, expected_result="scasq rax, qword ptr [rdi]")() + + CheckInstr(CmpsB, expected_result="cmpsb byte ptr [rsi], byte ptr [rdi]")() + CheckInstr(CmpsW, expected_result="cmpsw word ptr [rsi], word ptr [rdi]")() + CheckInstr(CmpsD, expected_result="cmpsd dword ptr [rsi], dword ptr [rdi]")() + CheckInstr(CmpsQ, expected_result="cmpsq qword ptr [rsi], qword ptr [rdi]")() + + + + CheckInstr(Mov, must_fail=True)('RCX', 'ECX') + CheckInstr(Mov, must_fail=True)('RCX', mem('[ECX + RCX]')) + CheckInstr(Mov, must_fail=True)('RCX', mem('[RBX + ECX]')) + CheckInstr(Mov, must_fail=True)('ECX', mem('[ECX + RCX]')) + CheckInstr(Mov, must_fail=True)('ECX', mem('[RBX + ECX]')) + CheckInstr(Add, must_fail=True)('RAX', 0xffffffff) + + + code = MultipleInstr() + code += Nop() + code += Rep + Nop() + code += Ret() + print(repr(code.get_code())) + assert code.get_code() == "\x90\xf3\x90\xc3" + +if capstone is None: + test_assembler = pytest.mark.skip("Capstone not installed")(test_assembler) + +# pytestmark = pytest.mark.skip("YOLO") + +if __name__ == "__main__": + test_assembler() diff --git a/windows/native_exec/test_simple_x86.py b/tests/test_simple_x86.py similarity index 53% rename from windows/native_exec/test_simple_x86.py rename to tests/test_simple_x86.py index d1db925..d446fb3 100644 --- a/windows/native_exec/test_simple_x86.py +++ b/tests/test_simple_x86.py @@ -1,15 +1,25 @@ -import capstone -from simple_x86 import * +try: + import capstone +except ImportError as e: + capstone = None -disassembleur = capstone.Cs(capstone.CS_ARCH_X86, capstone.CS_MODE_32) -disassembleur.detail = True +import pytest + +import windows.native_exec.simple_x86 as x86 +from windows.native_exec.simple_x86 import * +del Test # Prevent pytest warning + +if capstone: + disassembleur = capstone.Cs(capstone.CS_ARCH_X86, capstone.CS_MODE_32) + disassembleur.detail = True def disas(x): return list(disassembleur.disasm(x, 0)) -class TestInstr(object): + +class CheckInstr(object): def __init__(self, instr_to_test, immediat_accepted=None, expected_result=None, debug=False): self.instr_to_test = instr_to_test self.expected_result = expected_result @@ -85,104 +95,111 @@ class TestInstr(object): raise AssertionError("Expected mem.disp {0} got {1}".format(memaccess.disp, cap_mem.disp)) -TestInstr(Mov)('EAX', 'CR3') -TestInstr(Mov)('EDX', 'CR0') -TestInstr(Mov)('EDI', 'CR7') +def test_assembler(): + CheckInstr(Mov)('EAX', 'CR3') + CheckInstr(Mov)('EDX', 'CR0') + CheckInstr(Mov)('EDI', 'CR7') -TestInstr(Mov)('CR3', 'EAX') -TestInstr(Mov)('CR0', 'EDX') -TestInstr(Mov)('CR7', 'EDI') + CheckInstr(Mov)('CR3', 'EAX') + CheckInstr(Mov)('CR0', 'EDX') + CheckInstr(Mov)('CR7', 'EDI') -TestInstr(Mov)('EAX', 'ESP') -TestInstr(Mov)('ECX', mem('[EAX]')) -TestInstr(Mov)('EDX', mem('[ECX + 0x10]')) -TestInstr(Mov)('EDX', mem('[EDI * 8 + 0xffff]')) -TestInstr(Mov)('EDX', mem('[0x11223344]')) -TestInstr(Mov)('EDX', mem('[ESP + EBP * 2 + 0x223344]')) -TestInstr(Mov)(mem('[EBP + EBP * 2 + 0x223344]'), 'ESP') -TestInstr(Mov)('ESI', mem('[ESI + EDI * 1]')) -TestInstr(Mov)('EAX', mem('fs:[0x30]')) -TestInstr(Mov)('EDI', mem('gs:[EAX + ECX * 4]')) -TestInstr(Mov)('AX', 'AX') -TestInstr(Mov)('SI', 'DI') -TestInstr(Mov)('AX', 'AX') -TestInstr(Mov)('AX', mem('fs:[0x30]')) -TestInstr(Mov)('AX', mem('fs:[EAX + 0x30]')) -TestInstr(Mov)('AX', mem('fs:[EAX + ECX * 4+0x30]')) -TestInstr(Add)('EAX', 8) -TestInstr(Add)('EAX', 0xffffffff) -TestInstr(Add)("ECX", mem("[EAX + 0xff]")) -TestInstr(Add)("ECX", mem("[EAX + 0xffffffff]")) + CheckInstr(Mov)('EAX', 'ESP') + CheckInstr(Mov)('ECX', mem('[EAX]')) + CheckInstr(Mov)('EDX', mem('[ECX + 0x10]')) + CheckInstr(Mov)('EDX', mem('[EDI * 8 + 0xffff]')) + CheckInstr(Mov)('EDX', mem('[0x11223344]')) + CheckInstr(Mov)('EDX', mem('[ESP + EBP * 2 + 0x223344]')) + CheckInstr(Mov)(mem('[EBP + EBP * 2 + 0x223344]'), 'ESP') + CheckInstr(Mov)('ESI', mem('[ESI + EDI * 1]')) + CheckInstr(Mov)('EAX', mem('fs:[0x30]')) + CheckInstr(Mov)('EDI', mem('gs:[EAX + ECX * 4]')) + CheckInstr(Mov)('AX', 'AX') + CheckInstr(Mov)('SI', 'DI') + CheckInstr(Mov)('AX', 'AX') + CheckInstr(Mov)('AX', mem('fs:[0x30]')) + CheckInstr(Mov)('AX', mem('fs:[EAX + 0x30]')) + CheckInstr(Mov)('AX', mem('fs:[EAX + ECX * 4+0x30]')) + CheckInstr(Add)('EAX', 8) + CheckInstr(Add)('EAX', 0xffffffff) + CheckInstr(Add)("ECX", mem("[EAX + 0xff]")) + CheckInstr(Add)("ECX", mem("[EAX + 0xffffffff]")) -TestInstr(Add)(mem('[EAX]'), 10) -TestInstr(Mov)('EAX', mem('fs:[0xfffc]')) -TestInstr(Mov)(mem('fs:[0xfffc]'), 0) + CheckInstr(Add)(mem('[EAX]'), 10) + CheckInstr(Mov)('EAX', mem('fs:[0xfffc]')) + CheckInstr(Mov)(mem('fs:[0xfffc]'), 0) -TestInstr(Push)('ECX') -TestInstr(Push)(mem('[ECX + 8]')) + CheckInstr(Push)('ECX') + CheckInstr(Push)(mem('[ECX + 8]')) -TestInstr(Sub)('ECX', 'ESP') -TestInstr(Sub)('ECX', mem('[ESP]')) + CheckInstr(Sub)('ECX', 'ESP') + CheckInstr(Sub)('ECX', mem('[ESP]')) -TestInstr(Inc)('EAX') -TestInstr(Inc)(mem('[0x42424242]')) -TestInstr(Lea)('EAX', mem('[EAX + 1]')) -TestInstr(Lea)('ECX', mem('[EDI + -0xff]')) -TestInstr(Call)('EAX') -TestInstr(Call)(mem('[EAX + ECX * 8]')) -TestInstr(Cpuid)() -TestInstr(Movsb, expected_result='movsb byte ptr es:[edi], byte ptr [esi]')() -TestInstr(Movsd, expected_result='movsd dword ptr es:[edi], dword ptr [esi]')() -TestInstr(Xchg)('EAX', 'ESP') + CheckInstr(Inc)('EAX') + CheckInstr(Inc)(mem('[0x42424242]')) + CheckInstr(Lea)('EAX', mem('[EAX + 1]')) + CheckInstr(Lea)('ECX', mem('[EDI + -0xff]')) + CheckInstr(Call)('EAX') + CheckInstr(Call)(mem('[EAX + ECX * 8]')) + CheckInstr(Cpuid)() + CheckInstr(Movsb, expected_result='movsb byte ptr es:[edi], byte ptr [esi]')() + CheckInstr(Movsd, expected_result='movsd dword ptr es:[edi], dword ptr [esi]')() + CheckInstr(Xchg)('EAX', 'ESP') -TestInstr(Rol)('EAX', 7) -TestInstr(Rol)('ECX', 0) + CheckInstr(Rol)('EAX', 7) + CheckInstr(Rol)('ECX', 0) -TestInstr(Ror)('ECX', 0) -TestInstr(Ror)('EDI', 7) -TestInstr(Ror)('EDI', -128) + CheckInstr(Ror)('ECX', 0) + CheckInstr(Ror)('EDI', 7) + CheckInstr(Ror)('EDI', -128) -TestInstr(Cmp, immediat_accepted=0xffffffff)('EAX', -1) -TestInstr(Cmp)('EAX', 0xffffffff) + CheckInstr(Cmp, immediat_accepted=0xffffffff)('EAX', -1) + CheckInstr(Cmp)('EAX', 0xffffffff) -TestInstr(And)('ECX', 'EBX') -TestInstr(And)('EAX', 0x11223344) -TestInstr(And)('EAX', mem('[EAX + 1]')) -TestInstr(And)(mem('[EAX + EAX]'), 'EDX') + CheckInstr(And)('ECX', 'EBX') + CheckInstr(And)('EAX', 0x11223344) + CheckInstr(And)('EAX', mem('[EAX + 1]')) + CheckInstr(And)(mem('[EAX + EAX]'), 'EDX') -TestInstr(Or)('ECX', 'EBX') -TestInstr(Or)('EAX', 0x11223344) -TestInstr(Or)('EAX', mem('[EAX + 1]')) -TestInstr(Or)(mem('[EAX + EAX]'), 'EDX') + CheckInstr(Or)('ECX', 'EBX') + CheckInstr(Or)('EAX', 0x11223344) + CheckInstr(Or)('EAX', mem('[EAX + 1]')) + CheckInstr(Or)(mem('[EAX + EAX]'), 'EDX') -TestInstr(Shr)('EAX', 8) -TestInstr(Shr)('EDX', 0x12) -TestInstr(Shl)('EAX', 8) -TestInstr(Shl)('EDX', 0x12) + CheckInstr(Shr)('EAX', 8) + CheckInstr(Shr)('EDX', 0x12) + CheckInstr(Shl)('EAX', 8) + CheckInstr(Shl)('EDX', 0x12) -TestInstr(Not)('EAX') -TestInstr(Not)(mem('[EAX]')) + CheckInstr(Not)('EAX') + CheckInstr(Not)(mem('[EAX]')) -TestInstr(ScasB, expected_result="scasb al, byte ptr es:[edi]")() -TestInstr(ScasW, expected_result="scasw ax, word ptr es:[edi]")() -TestInstr(ScasD, expected_result="scasd eax, dword ptr es:[edi]")() + CheckInstr(ScasB, expected_result="scasb al, byte ptr es:[edi]")() + CheckInstr(ScasW, expected_result="scasw ax, word ptr es:[edi]")() + CheckInstr(ScasD, expected_result="scasd eax, dword ptr es:[edi]")() -TestInstr(CmpsB, expected_result="cmpsb byte ptr [esi], byte ptr es:[edi]")() -TestInstr(CmpsW, expected_result="cmpsw word ptr [esi], word ptr es:[edi]")() -TestInstr(CmpsD, expected_result="cmpsd dword ptr [esi], dword ptr es:[edi]")() + CheckInstr(CmpsB, expected_result="cmpsb byte ptr [esi], byte ptr es:[edi]")() + CheckInstr(CmpsW, expected_result="cmpsw word ptr [esi], word ptr es:[edi]")() + CheckInstr(CmpsD, expected_result="cmpsd dword ptr [esi], dword ptr es:[edi]")() -TestInstr(Test)('EAX', 'EAX') -TestInstr(Test, expected_result="test edi, ecx")('ECX', 'EDI') + CheckInstr(x86.Test)('EAX', 'EAX') + CheckInstr(x86.Test, expected_result="test edi, ecx")('ECX', 'EDI') -TestInstr(Test)(mem('[ECX + 0x100]'), 'ECX') + CheckInstr(x86.Test)(mem('[ECX + 0x100]'), 'ECX') -assert Test(mem('[ECX + 0x100]'), 'ECX').get_code() == Test('ECX', mem('[ECX + 0x100]')).get_code() -assert Xchg('EAX', 'ECX').get_code() == Xchg('ECX', 'EAX').get_code() + assert x86.Test(mem('[ECX + 0x100]'), 'ECX').get_code() == x86.Test('ECX', mem('[ECX + 0x100]')).get_code() + assert Xchg('EAX', 'ECX').get_code() == Xchg('ECX', 'EAX').get_code() -code = MultipleInstr() -code += Nop() -code += Rep + Nop() -code += Ret() -print(repr(code.get_code())) -assert code.get_code() == "\x90\xf3\x90\xc3" + code = MultipleInstr() + code += Nop() + code += Rep + Nop() + code += Ret() + print(repr(code.get_code())) + assert code.get_code() == "\x90\xf3\x90\xc3" + +if capstone is None: + test_assembler = pytest.mark.skip("Capstone not installed")(test_assembler) + +if __name__ == "__main__": + test_assembler() diff --git a/pytesting/test_system.py b/tests/test_system.py similarity index 97% rename from pytesting/test_system.py rename to tests/test_system.py index 7e285b0..075be45 100644 --- a/pytesting/test_system.py +++ b/tests/test_system.py @@ -25,7 +25,6 @@ class TestSystemWithCheckGarbage(object): @check_for_gc_garbage -@check_for_handle_leak class TestSystemWithCheckGarbageAndHandleLeak(object): def test_threads(self): return windows.system.threads diff --git a/pytesting/test_syswow.py b/tests/test_syswow.py similarity index 100% rename from pytesting/test_syswow.py rename to tests/test_syswow.py diff --git a/pytesting/test_winproxy.py b/tests/test_winproxy.py similarity index 100% rename from pytesting/test_winproxy.py rename to tests/test_winproxy.py diff --git a/windows/native_exec/test_simple_x64.py b/windows/native_exec/test_simple_x64.py deleted file mode 100644 index 128b5e1..0000000 --- a/windows/native_exec/test_simple_x64.py +++ /dev/null @@ -1,256 +0,0 @@ -import capstone -import simple_x64 as x64 -from simple_x64 import * - -disassembleur = capstone.Cs(capstone.CS_ARCH_X86, capstone.CS_MODE_64) -disassembleur.detail = True - - -def disas(x): - return list(disassembleur.disasm(x, 0)) - -mnemonic_name_exception = {'movabs': 'mov'} - - -class TestInstr(object): - def __init__(self, instr_to_test, expected_result=None, immediat_accepted=None, must_fail=None, debug=False): - self.instr_to_test = instr_to_test - self.immediat_accepted = immediat_accepted - self.expected_result = expected_result - self.must_fail = must_fail - self.debug = debug - - def __call__(self, *args): - try: - if self.debug: - import pdb;pdb.set_trace() - pdb.DONE = True - x64.DEBUG = self.debug - res = bytes(self.instr_to_test(*args).get_code()) - if self.debug: - print(repr(res)) - except ValueError as e: - if self.must_fail == True: - return True - else: - raise - else: - if self.must_fail: - raise ValueError("Instruction did not failed as expected") - capres_list = disas(res) - if len(capres_list) != 1: - raise AssertionError("Trying to disas an instruction resulted in multiple disassembled instrs") - capres = capres_list[0] - print("{0} {1}".format(capres.mnemonic, capres.op_str)) - if self.expected_result is not None: - if "{0} {1}".format(capres.mnemonic, capres.op_str) == self.expected_result: - return True - else: - raise AssertionError("Expected result <{0}> got <{1}>".format(self.expected_result, "{0} {1}".format(capres.mnemonic, capres.op_str))) - if len(res) != len(capres.bytes): - print("<{0}> vs <{1}>".format(repr(res), repr(capres.bytes))) - raise AssertionError("Not all bytes have been used by the disassembler") - self.compare_mnemo(capres) - self.compare_args(args, capres) - - def compare_mnemo(self, capres): - expected = self.instr_to_test.__name__.lower() - cap_mnemo = mnemonic_name_exception.get(str(capres.mnemonic), str(capres.mnemonic)) - if expected != cap_mnemo: - raise AssertionError("Expected menmo {0} got {1}".format(expected, str(capres.mnemonic))) - return True - - def compare_args(self, args, capres): - capres_op = list(capres.operands) - if len(args) != len(capres_op): - raise AssertionError("Expected {0} operands got {1}".format(len(args), len(capres_op))) - for op_args, cap_op in zip(args, capres_op): - if isinstance(op_args, str): # Register - if cap_op.type != capstone.x86.X86_OP_REG: - raise AssertionError("Expected args {0} operands got {1}".format(op_args, capres_op)) - if op_args.lower() != capres.reg_name(cap_op.reg).lower(): - raise AssertionError("Expected register <{0}> got {1}".format(op_args.lower(), capres.reg_name(cap_op.reg).lower())) - elif isinstance(op_args, (int, long)): - if (op_args != cap_op.imm) and not (self.immediat_accepted and self.immediat_accepted == cap_op.imm): - raise AssertionError("Expected Immediat <{0}> got {1}".format(op_args, cap_op.imm)) - elif isinstance(op_args, mem_access): - self.compare_mem_access(op_args, capres, cap_op) - else: - raise ValueError("Unknow argument {0} of type {1}".format(op_args, type(op_args))) - - def compare_mem_access(self, memaccess, capres, cap_op): - if cap_op.type != capstone.x86.X86_OP_MEM: - raise AssertionError("Expected Memaccess <{0}> got {1}".format(memaccess, cap_op)) - if memaccess.prefix is not None and capres.prefix[1] != x64_segment_selectors[memaccess.prefix].PREFIX_VALUE: - try: - get_prefix = [n for n, x in x64_segment_selectors.items() if x.PREFIX_VALUE == capres.prefix[1]][0] - except IndexError: - get_prefix = None - raise AssertionError("Expected Segment overide <{0}> got {1}".format(memaccess.prefix, get_prefix)) - cap_mem = cap_op.mem - if memaccess.base is None and cap_mem.base != capstone.x86.X86_REG_INVALID: - raise AssertionError("Unexpected memaccess base <{0}>".format(capres.reg_name(cap_mem.base))) - if memaccess.base is not None and capres.reg_name(cap_mem.base) != memaccess.base.lower(): - raise AssertionError("Expected mem.base {0} got {1}".format(memaccess.base.lower(), capres.reg_name(cap_mem.base))) - if memaccess.index is None and cap_mem.index != capstone.x86.X86_REG_INVALID: - raise AssertionError("Unexpected memaccess index <{0}>".format(capres.reg_name(cap_mem.base))) - if memaccess.index is not None and capres.reg_name(cap_mem.index) != memaccess.index.lower(): - raise AssertionError("Expected mem.index {0} got {1}".format(memaccess.index.lower(), capres.reg_name(cap_mem.index))) - if memaccess.scale != cap_mem.scale and not (memaccess.scale is None and cap_mem.scale == 1): - raise AssertionError("Expected mem.scale {0} got {1}".format(memaccess.scale, cap_mem.scale)) - if memaccess.disp != cap_mem.disp: - raise AssertionError("Expected mem.disp {0} got {1}".format(memaccess.disp, cap_mem.disp)) - - -TestInstr(Add)('RAX', 'RSP') -TestInstr(Add)('RAX', mem('[RCX]')) -TestInstr(Add)('RAX', mem('[RDI + 0x10]')) -TestInstr(Add)('RAX', mem('[RSI + 0x7fffffff]')) -TestInstr(Add)('RAX', mem('[RSI + -0x1]')) -TestInstr(Add)('RAX', mem('[0x10]')) -TestInstr(Add)('RAX', mem('fs:[0x10]')) -TestInstr(Add)('RAX', mem('[RSI + RDI * 2]')) -TestInstr(Add)('RAX', mem('[RSI + RDI * 2 + 0x10]')) -TestInstr(Add)('RAX', mem('gs:[RSI + RDI * 2 + 0x10]')) -TestInstr(Add)('RAX', mem('[R15 * 8 + 0x10]')) -TestInstr(Add)('RAX', mem('[R9 + R8 * 2 + 0x7fffffff]')) -TestInstr(Add)('RAX', mem('[R9 + R8 * 2 + -0x80000000]')) -TestInstr(Add)('RAX', mem('[-1]')) -TestInstr(Add)('RAX', mem('[0x7fffffff]')) -TestInstr(Add)('RAX', -1) - - -TestInstr(Sub)('RCX', 'RSP') -TestInstr(Sub)('RCX', mem('[RSP]')) - -TestInstr(Xor)('R15', mem('[RAX + R8 * 2 + 0x11223344]')) -TestInstr(Xor)('RAX', 'RAX') -TestInstr(Cmp)('RAX', -1) -#TestInstr(Cmp, immediat_accepted=-1)('RAX', 0xffffffff) -TestInstr(Lea)('RAX', mem('[RAX + 1]')) -TestInstr(Lea)('RAX', mem('fs:[RAX + 1]')) -TestInstr(Mov)('RAX', mem('[0x1122334455667788]')) -TestInstr(Mov)('RAX', mem('gs:[0x1122334455667788]')) -TestInstr(Mov)('RAX', mem('gs:[0x60]')) -TestInstr(Mov)('RCX', 0x1122334455667788) -TestInstr(Mov)('RCX', -1) -TestInstr(Mov)('RCX', -0x1000) -TestInstr(Mov)('RCX', 0xffffffff) -TestInstr(Mov)('RAX', 0xffffffff) -TestInstr(Mov)('R8', 0x1122334455667788) -TestInstr(Mov)('RCX', -1) -TestInstr(Mov, immediat_accepted=-1)('RCX', 0xffffffffffffffff) -TestInstr(Mov)(mem('gs:[0x1122334455667788]'), 'RAX') -TestInstr(Mov)(mem('[RAX]'), 0x11223344) -TestInstr(Mov)(mem('[EAX]'), 0x11223344) -TestInstr(Mov)(mem('[RBX]'), 0x11223344) -TestInstr(Mov)("R12", mem("[RAX]")) -TestInstr(Mov)("RAX", mem("[R12]")) -TestInstr(Mov)("RAX", mem("[RAX + R12]")) -TestInstr(Mov)("RAX", mem("[R12 + R12]")) -TestInstr(Mov)("RAX", mem("[R12 + R15]")) - -TestInstr(Mov)("RAX", mem("[R10]")) -TestInstr(Mov)("RAX", mem("[R11]")) -TestInstr(Mov)("RAX", mem("[R12]")) -TestInstr(Mov)("RAX", mem("[R13]")) -TestInstr(Mov)("RAX", mem("[R14]")) -TestInstr(Mov)("RAX", mem("[R15]")) - -#TestInstr(Mov)("RSI", mem("[R12]")) - -TestInstr(And)('RCX', 'RBX') -TestInstr(And)('RAX', 0x11223344) -TestInstr(And)('EAX', 0x11223344) -TestInstr(And)('EAX', 0xffffffff) -TestInstr(And)('RAX', mem('[RAX + 1]')) -TestInstr(And)(mem('[RAX + 1]'), 'R8') -TestInstr(And)(mem('[EAX + 1]'), 'R8') -TestInstr(And)(mem('[RAX + 1]'), 'EAX') - -TestInstr(Or)('RCX', 'RBX') -TestInstr(Or)('RAX', 0x11223344) -TestInstr(Or)('RAX', mem('[RAX + 1]')) -TestInstr(Or)(mem('[RAX + 1]'), 'R8') -TestInstr(Or)(mem('[EAX + 1]'), 'R8') -TestInstr(Or)(mem('[RAX + 1]'), 'EAX') - -TestInstr(Shr)('RAX', 8) -TestInstr(Shr)('R15', 0x12) -TestInstr(Shl)('RAX', 8) -TestInstr(Shl)('R15', 0x12) - -# I really don't know why it's the inverse -# But I don't care, it's Test dude.. -TestInstr(Test, expected_result="test r11, rax")('RAX', 'R11') -TestInstr(Test, expected_result="test edi, eax")('EAX', 'EDI') -TestInstr(Test)('RCX', 'RCX') - -TestInstr(Test)(mem('[RDI + 0x100]'), 'RCX') - -assert Test(mem('[RDI + 0x100]'), 'RCX').get_code() == Test('RCX', mem('[RDI + 0x100]')).get_code() - - -TestInstr(Push)('RAX') -assert len(Push("RAX").get_code()) == 1 -TestInstr(Push)('R15') -TestInstr(Push)(0x42) -TestInstr(Push)(-1) -TestInstr(Push)(mem("[ECX]")) -TestInstr(Push)(mem("[RCX]")) - - -TestInstr(Pop)('RAX') -assert len(Pop("RAX").get_code()) == 1 - - -TestInstr(Call)('RAX') -TestInstr(Call)(mem('[RAX + RCX * 8]')) -TestInstr(Cpuid)() -TestInstr(Xchg)('RAX', 'RSP') -assert Xchg('RAX', 'RCX').get_code() == Xchg('RCX', 'RAX').get_code() - -# 32 / 64 bits register mixing -TestInstr(Mov)('ECX', 'EBX') -TestInstr(Mov)('RCX', mem('[EBX]')) -TestInstr(Mov)('ECX', mem('[RBX]')) -TestInstr(Mov)('ECX', mem('[EBX]')) -TestInstr(Mov)('RCX', mem('[EBX + EBX]')) -TestInstr(Mov)('RCX', mem('[ESP + EBX + 0x10]')) -TestInstr(Mov)('ECX', mem('[ESP + EBX + 0x10]')) -TestInstr(Mov)('ECX', mem('[RBX + RCX + 0x10]')) - -TestInstr(Mov)(mem('[RBX + RCX + 0x10]'), 'ECX') -TestInstr(Mov)(mem('[EBX + ECX + 0x10]'), 'ECX') -TestInstr(Mov)(mem('[EBX + ECX + 0x10]'), 'R8') - -TestInstr(Not)('RAX') -TestInstr(Not)(mem('[RAX]')) - - -TestInstr(ScasB, expected_result="scasb al, byte ptr [rdi]")() -TestInstr(ScasW, expected_result="scasw ax, word ptr [rdi]")() -TestInstr(ScasD, expected_result="scasd eax, dword ptr [rdi]")() -TestInstr(ScasQ, expected_result="scasq rax, qword ptr [rdi]")() - -TestInstr(CmpsB, expected_result="cmpsb byte ptr [rsi], byte ptr [rdi]")() -TestInstr(CmpsW, expected_result="cmpsw word ptr [rsi], word ptr [rdi]")() -TestInstr(CmpsD, expected_result="cmpsd dword ptr [rsi], dword ptr [rdi]")() -TestInstr(CmpsQ, expected_result="cmpsq qword ptr [rsi], qword ptr [rdi]")() - - - -TestInstr(Mov, must_fail=True)('RCX', 'ECX') -TestInstr(Mov, must_fail=True)('RCX', mem('[ECX + RCX]')) -TestInstr(Mov, must_fail=True)('RCX', mem('[RBX + ECX]')) -TestInstr(Mov, must_fail=True)('ECX', mem('[ECX + RCX]')) -TestInstr(Mov, must_fail=True)('ECX', mem('[RBX + ECX]')) -TestInstr(Add, must_fail=True)('RAX', 0xffffffff) - - -code = MultipleInstr() -code += Nop() -code += Rep + Nop() -code += Ret() -print(repr(code.get_code())) -assert code.get_code() == "\x90\xf3\x90\xc3" diff --git a/windows/test/__init__.py b/windows/test/__init__.py deleted file mode 100644 index 9313cb7..0000000 --- a/windows/test/__init__.py +++ /dev/null @@ -1,13 +0,0 @@ - -from test_utils import * - -from mytest import WindowsTestCase, WindowsAPITestCase, NativeUtilsTestCase, SystemTestCase, GeneratedCodeTestCase -from test_hooks import HookTestCase -from test_debugger import DebuggerTestCase -from test_syswow import SyswowTestCase -from test_crypto import CryptoTestCase - - -__all__ = ["SystemTestCase", "WindowsTestCase", "WindowsAPITestCase", - "DebuggerTestCase", "NativeUtilsTestCase", "HookTestCase", "SyswowTestCase", - "CryptoTestCase"] diff --git a/windows/test/mytest.py b/windows/test/mytest.py deleted file mode 100644 index 1dfebb5..0000000 --- a/windows/test/mytest.py +++ /dev/null @@ -1,763 +0,0 @@ -import sys -import struct -import time -import os -import textwrap -import random -import pickle - -from test_utils import * -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 - - @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 - @check_for_handle_leak - 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 - @check_for_handle_leak - 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 - - @check_for_gc_garbage - def test_limited_handle_query(self): - #if len(smss_list) != 1: - # raise ValueError("Not just one smss.exe: {0}".format(smss_list)) - - class CustomTestRaise(ValueError): - pass - - def custom_get_handle(): - raise CustomTestRaise("_get_handle() should not be called during this test") - - with Calc32() as calc: - save_handle = calc._handle - del calc._handle - calc._get_handle = custom_get_handle - # Check that trying to get a handle raise 'CustomTestRaise' - with self.assertRaises(CustomTestRaise): - calc.handle - # List of attributes that only require PROCESS_QUERY_INFOR - calc.bitness - calc.time_info - # Re-set the handle to be able to kill it - calc._handle = save_handle - - @check_for_gc_garbage - def test_current_process_threads(self): - # Had a bug with WinThread and CurrentProcess.name (which was non-existant) - self.assertTrue([repr(t) for t in windows.current_process.threads]) - - @check_for_gc_garbage - def test_pop_calc_32(self): - with Calc32() as calc: - self.assertEqual(calc.bitness, 32) - - @windows_64bit_only - def test_pop_calc_64(self): - with Calc64() as calc: - self.assertEqual(calc.bitness, 64) - - @check_for_gc_garbage - def test_current_process_ppid(self): - myself = [p for p in windows.system.processes if p.pid == windows.current_process.pid][0] - self.assertEqual(myself.ppid, windows.current_process.ppid) - - @check_for_gc_garbage - def test_process_ppid_32(self): - with Calc32() as calc: - self.assertEqual(calc.ppid, windows.current_process.pid) - - @windows_64bit_only - @check_for_gc_garbage - def test_process_ppid_64(self): - with Calc64() as calc: - self.assertEqual(calc.ppid, windows.current_process.pid) - - @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_get_current_process_exe(self): - exe = windows.current_process.peb.exe - exe_by_module = windows.current_process.peb.modules[0].pe - self.assertEqual(exe.baseaddr, exe_by_module.baseaddr) - self.assertEqual(exe.bitness, exe_by_module.bitness) - - - @check_for_gc_garbage - def test_local_process_pe_imports(self): - python_module = windows.current_process.peb.modules[0] - imp = python_module.pe.imports - self.assertIn("kernel32.dll", 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") - 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') - k32 = mods[0] - get_current_proc_id = k32.pe.exports['GetCurrentProcessId'] - 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') - k32 = mods[0] - sections = k32.pe.sections - all_sections_name = [s.name for s in sections] - self.assertIn(".text", all_sections_name) - sections[0].start - 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] - with calc.virtual_protected(k32.baseaddr, 2, PAGE_EXECUTE_READWRITE): - calc.write_memory(k32.baseaddr, "XD") - 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] - with calc.virtual_protected(k32.baseaddr, 2, PAGE_EXECUTE_READWRITE): - 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" - with Calc32() as calc: - addr = calc.virtual_alloc(0x1000) - 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" - with Calc32() as calc: - addr = calc.virtual_alloc(0x1000) + 0x1000 - len(string_to_write) - 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" - with Calc32() as calc: - addr = calc.virtual_alloc(0x1000) - 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" - with Calc32() as calc: - # Setup string addr at end of page - addr = calc.virtual_alloc(0x1000) + 0x1000 - 26 - calc.write_memory(addr, "\x00".join(string_to_write)) - 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) - shellcode = x86.MultipleInstr() - shellcode += x86.Mov('EAX', 0x42424242) - shellcode += x86.Mov(x86.create_displacement(disp=data), 'EAX') - shellcode += x86.Ret() - calc.execute(shellcode.get_code()) - time.sleep(0.1) - dword = struct.unpack("32 injection for now) - #if is_process_64_bits: - # raise NotImplementedError("Python execution 64->32") - data = calc.virtual_alloc(0x1000) - 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_uint.from_address({1}).value = GetCurrentProcessId - """.format(os.getcwd(), data) - calc.execute_python(textwrap.dedent(remote_python_code)) - #time.sleep(0.5) - dword = struct.unpack("> 12 - for page_info in calc.query_working_set(): - if page_info.virtualpage == page_target: - self.assertEqual(page_info.shared, True) - break - else: - raise ValueError("query_working_set page info for <0x{0:x}> not found".format(page_target)) - data = calc.write_memory(api_addr, data) - for page_info in calc.query_working_set(): - if page_info.virtualpage == page_target: - self.assertEqual(page_info.shared, False) - break - else: - 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_set_64(self): - with Calc64() as calc: - k32 = [m for m in calc.peb.modules if m.name == "kernel32.dll"][0] - api_addr = k32.pe.exports["CreateFileA"] - data = calc.read_memory(api_addr, 5) - page_target = api_addr >> 12 - for page_info in calc.query_working_set(): - if page_info.virtualpage == page_target: - self.assertEqual(page_info.shared, True) - break - else: - raise ValueError("query_working_set page info for <0x{0:x}> not found".format(page_target)) - with calc.virtual_protected(api_addr, 5, PAGE_EXECUTE_READWRITE): - data = calc.write_memory(api_addr, data) - for page_info in calc.query_working_set(): - if page_info.virtualpage == page_target: - self.assertEqual(page_info.shared, False) - break - 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] - - 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 = calc.read_memory(api_addr, 5) - page_target = (api_addr >> 12) << 12 - - for page_info in calc.query_working_setex(pages): - self.assertIn(page_info.VirtualAddress, pages) - if page_info.VirtualAddress == page_target: - self.assertEqual(page_info.VirtualAttributes.shared, True) - break - else: - raise ValueError("query_working_set page info for <0x{0:x}> not found".format(page_target)) - with calc.virtual_protected(api_addr, 5, PAGE_EXECUTE_READWRITE): - data = calc.write_memory(api_addr, data) - for page_info in calc.query_working_setex(pages): - self.assertIn(page_info.VirtualAddress, pages) - if page_info.VirtualAddress == page_target: - self.assertEqual(page_info.VirtualAttributes.shared, False) - break - else: - 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] - - 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 = calc.read_memory(api_addr, 5) - page_target = (api_addr >> 12) << 12 - - for page_info in calc.query_working_setex(pages): - self.assertIn(page_info.VirtualAddress, pages) - if page_info.VirtualAddress == page_target: - self.assertEqual(page_info.VirtualAttributes.shared, True) - break - else: - raise ValueError("query_working_set page info for <0x{0:x}> not found".format(page_target)) - - with calc.virtual_protected(api_addr, 5, PAGE_EXECUTE_READWRITE): - data = calc.write_memory(api_addr, data) - for page_info in calc.query_working_setex(pages): - self.assertIn(page_info.VirtualAddress, pages) - if page_info.VirtualAddress == page_target: - self.assertEqual(page_info.VirtualAttributes.shared, False) - break - 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] - mapped_filname = calc.get_mapped_filename(k32.baseaddr) - 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] - self.assertNotEqual(t.teb_base, 0) - - @check_for_gc_garbage - def test_thread_owner_from_tid_32(self): - with Calc32() as calc: - thread = calc.threads[0] - tst_thread = windows.winobject.process.WinThread(tid=thread.tid) - self.assertEqual(thread.owner_pid, tst_thread.owner_pid) - self.assertEqual(thread.owner.name, tst_thread.owner.name) - - @windows_64bit_only - @check_for_gc_garbage - def test_thread_owner_from_tid_64(self): - with Calc64() as calc: - thread = calc.threads[0] - tst_thread = windows.winobject.process.WinThread(tid=thread.tid) - self.assertEqual(thread.owner_pid, tst_thread.owner_pid) - self.assertEqual(thread.owner.name, tst_thread.owner.name) - - -class WindowsAPITestCase(unittest.TestCase): - def test_createfileA_fail(self): - with self.assertRaises(WindowsError) as ar: - windows.winproxy.CreateFileA("NONEXISTFILE.FILE") - -class GeneratedCodeTestCase(unittest.TestCase): - def test_str_flags_value(self): - self.assertEqual(windows.generated_def.MS_ENHANCED_PROV, windows.generated_def.MS_ENHANCED_PROV_A) - - def _test_pickle_unpickle(self, obj, protocol=0): - pickled = pickle.dumps(obj, protocol) - unpickled = pickle.loads(pickled) - self.assertEqual(unpickled, obj) - - def test_long_flag_picke_v0(self): - self._test_pickle_unpickle(windows.generated_def.PAGE_EXECUTE_READWRITE, 0) - - def test_long_flag_picke_v1(self): - self._test_pickle_unpickle(windows.generated_def.PAGE_EXECUTE_READWRITE, 1) - - def test_long_flag_picke_v2(self): - self._test_pickle_unpickle(windows.generated_def.PAGE_EXECUTE_READWRITE, 2) - - def test_str_flag_picke_v0(self): - self._test_pickle_unpickle(windows.generated_def.szOID_RSA, 0) - - def test_str_flag_picke_v1(self): - self._test_pickle_unpickle(windows.generated_def.szOID_RSA, 1) - - def test_str_flag_picke_v2(self): - self._test_pickle_unpickle(windows.generated_def.szOID_RSA, 2) - -class NativeUtilsTestCase(unittest.TestCase): - @process_64bit_only - def test_strlenw64(self): - strlenw64 = windows.native_exec.create_function(nativeutils.StrlenW64.get_code(), [UINT, LPCWSTR]) - self.assertEqual(strlenw64("YOLO"), 4) - self.assertEqual(strlenw64(""), 0) - - @process_64bit_only - def test_strlena64(self): - strlena64 = windows.native_exec.create_function(nativeutils.StrlenA64.get_code(), [UINT, LPCSTR]) - self.assertEqual(strlena64("YOLO"), 4) - self.assertEqual(strlena64(""), 0) - - @process_64bit_only - def test_getprocaddr64(self): - getprocaddr64 = windows.native_exec.create_function(nativeutils.GetProcAddress64.get_code(), [ULONG64, LPCWSTR, 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 - self.assertEqual((name, hex(addr)), (name, hex(compute_addr))) - - self.assertEqual(getprocaddr64("YOLO.DLL", "whatever"), 0xfffffffffffffffe) - self.assertEqual(getprocaddr64("KERNEL32.DLL", "YOLOAPI"), 0xffffffffffffffff) - - @process_32bit_only - def test_strlenw32(self): - strlenw32 = windows.native_exec.create_function(nativeutils.StrlenW32.get_code(), [UINT, LPCWSTR]) - self.assertEqual(strlenw32("YOLO"), 4) - self.assertEqual(strlenw32(""), 0) - - @process_32bit_only - def test_strlena32(self): - strlena32 = windows.native_exec.create_function(nativeutils.StrlenA32.get_code(), [UINT, LPCSTR]) - self.assertEqual(strlena32("YOLO"), 4) - self.assertEqual(strlena32(""), 0) - - @process_32bit_only - def test_getprocaddr32(self): - getprocaddr32 = windows.native_exec.create_function(nativeutils.GetProcAddress32.get_code(), [UINT, LPCWSTR, 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 - self.assertEqual((name, hex(addr)), (name, hex(compute_addr))) - - self.assertEqual(getprocaddr32("YOLO.DLL", "whatever"), 0xfffffffe) - self.assertEqual(getprocaddr32("KERNEL32.DLL", "YOLOAPI"), 0xffffffff) - - -if __name__ == '__main__': - alltests = unittest.TestSuite() - alltests.addTest(unittest.makeSuite(SystemTestCase)) - alltests.addTest(unittest.makeSuite(WindowsTestCase)) - alltests.addTest(unittest.makeSuite(WindowsAPITestCase)) - alltests.addTest(unittest.makeSuite(NativeUtilsTestCase)) - alltests.addTest(unittest.makeSuite(GeneratedCodeTestCase)) - alltests.debug() - tester = unittest.TextTestRunner(verbosity=2) - tester.run(alltests) diff --git a/windows/test/test_crypto.py b/windows/test/test_crypto.py deleted file mode 100644 index a5712b8..0000000 --- a/windows/test/test_crypto.py +++ /dev/null @@ -1,100 +0,0 @@ -import windows.crypto -import time -from test_utils import * -from windows.generated_def.winstructs import * - - - -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== -""" - - -class CryptoTestCase(unittest.TestCase): - @classmethod - def setUpClass(cls): - 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) - certs = pfx.certs - self.assertEqual(len(certs), 1) - # 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) - # 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 - self.assertNotIn(message_to_encrypt, res) - - # Open pfx and decrypt - pfx = windows.crypto.import_pfx(self.raw_pfx, TEST_PFX_PASSWORD) - decrypt = windows.crypto.decrypt(pfx, res) - decrypt2 = windows.crypto.decrypt(pfx, res2) - - self.assertEqual(message_to_encrypt, decrypt) - self.assertEqual(decrypt, decrypt2) - - diff --git a/windows/test/test_debugger.py b/windows/test/test_debugger.py deleted file mode 100644 index 2928bd7..0000000 --- a/windows/test/test_debugger.py +++ /dev/null @@ -1,746 +0,0 @@ -from test_utils import * -from windows.generated_def.winstructs import * - -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 - class MyDbg(windows.debug.Debugger): - def on_exception(self, exception): - TEST_CASE.assertEqual(exception.ExceptionRecord.ExceptionCode, EXCEPTION_BREAKPOINT) - self.current_process.exit() - - calc = pop_calc_32(dwCreationFlags=DEBUG_PROCESS) - 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 - - class TSTBP(windows.debug.Breakpoint): - def trigger(self, dbg, exc): - TEST_CASE.assertEqual(dbg.current_process.pid, calc.pid) - TEST_CASE.assertEqual(dbg.current_process.read_memory(self.addr, 1), "\xcc") - TEST_CASE.assertEqual(dbg.current_thread.context.pc, self.addr) - d.current_process.exit() - - calc = pop_calc_32(dwCreationFlags=DEBUG_PROCESS) - - if windows.current_process.bitness == 32: - LdrLoadDll32 = windows.current_process.peb.modules[1].pe.exports["LdrLoadDll"] - else: - calcref = pop_calc_32() - LdrLoadDll32 = calcref.peb.modules[1].pe.exports["LdrLoadDll"] - calcref.exit() - - d = windows.debug.Debugger(calc) - d.add_bp(TSTBP(LdrLoadDll32)) - d.loop() - - #def test_standard_breakpoint_multiple_threads(self): - # """Check standard BP trigger by multiples threads""" - # TEST_CASE = self - # data = [0] - # - # class TSTBP(windows.debug.Breakpoint): - # def trigger(self, dbg, exc): - # TEST_CASE.assertEqual(dbg.current_process.pid, calc.pid) - # TEST_CASE.assertEqual(dbg.current_process.read_memory(self.addr, 1), "\xcc") - # TEST_CASE.assertEqual(dbg.current_thread.context.pc, self.addr) - # data[0] += 1 - # print("POUET <{0}>".format(dbg.current_thread.tid)) - # d.current_process.exit() - # - # calc = pop_calc_32(dwCreationFlags=DEBUG_PROCESS) - # - # if windows.current_process.bitness == 32: - # LdrLoadDll32 = windows.current_process.peb.modules[1].pe.exports["LdrLoadDll"] - # else: - # calcref = pop_calc_32() - # LdrLoadDll32 = calcref.peb.modules[1].pe.exports["LdrLoadDll"] - # calcref.exit() - # - # d = windows.debug.Debugger(calc) - # calc.execute("\xc3") - # calc.execute("\xc3") - # calc.execute("\xc3") - # 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 - - class TSTBP(windows.debug.HXBreakpoint): - def trigger(self, dbg, exc): - TEST_CASE.assertEqual(dbg.current_process.pid, calc.pid) - TEST_CASE.assertEqual(dbg.current_thread.context.pc, self.addr) - TEST_CASE.assertNotEqual(dbg.current_thread.context.Dr7, 0) - d.current_process.exit() - - calc = pop_calc_32(dwCreationFlags=DEBUG_PROCESS) - - if windows.current_process.bitness == 32: - LdrLoadDll32 = windows.current_process.peb.modules[1].pe.exports["LdrLoadDll"] - else: - calcref = pop_calc_32() - LdrLoadDll32 = calcref.peb.modules[1].pe.exports["LdrLoadDll"] - calcref.exit() - - d = windows.debug.Debugger(calc) - 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 - data = [0] - class TSTBP(windows.debug.HXBreakpoint): - def __init__(self, addr, expec_before): - self.addr = addr - self.expec_before = expec_before - - def trigger(self, dbg, exc): - TEST_CASE.assertEqual(dbg.current_process.pid, calc.pid) - TEST_CASE.assertEqual(dbg.current_thread.context.pc, self.addr) - TEST_CASE.assertNotEqual(dbg.current_thread.context.Dr7, 0) - TEST_CASE.assertEqual(data[0], self.expec_before) - TEST_CASE.assertNotEqual(dbg.current_process.read_memory(self.addr, 1), "\xcc") - data[0] += 1 - if data[0] == 4: - d.current_process.exit() - - calc = pop_calc_32(dwCreationFlags=DEBUG_PROCESS) - d = windows.debug.Debugger(calc) - addr = calc.virtual_alloc(0x1000) - calc.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)) - calc.create_thread(addr, 0) - d.loop() - # 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 - data = [0] - - 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") - - calc = pop_calc_32(dwCreationFlags=DEBUG_PROCESS) - d = windows.debug.Debugger(calc) - addr = calc.virtual_alloc(0x1000) - calc.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)) - - calc.create_thread(addr, 0) - with self.assertRaises(ValueError) as e: - d.loop() - self.assertIn("DRx", e.exception.message) - # 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 - data = [0] - - class MyDbg(windows.debug.Debugger): - def on_create_thread(self, exception): - # Check that later created thread have their HWX breakpoint :) - TEST_CASE.assertNotEqual(self.current_thread.context.Dr7, 0) - - class TSTBP(windows.debug.HXBreakpoint): - def __init__(self, addr, expec_before): - self.addr = addr - self.expec_before = expec_before - - def trigger(self, dbg, exc): - TEST_CASE.assertNotEqual(len(dbg.current_process.threads), 1) - #for t in dbg.current_process.threads: - # TEST_CASE.assertNotEqual(t.context.Dr7, 0) - if data[0] == 0: #First time we got it ! create new thread - data[0] = 1 - calc.create_thread(addr, 0) - else: - data[0] += 1 - d.current_process.exit() - - calc = pop_calc_32(dwCreationFlags=DEBUG_PROCESS) - d = MyDbg(calc) - addr = calc.virtual_alloc(0x1000) - calc.write_memory(addr, "\x90" * 2 + "\xc3") - d.add_bp(TSTBP(addr, 0)) - calc.create_thread(addr, 0) - d.loop() - # 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 - data = [0] - class TSTBP(windows.debug.Breakpoint): - def trigger(self, dbg, exc): - addr = exc.ExceptionRecord.ExceptionAddress - LdrLoadDlladdr = dbg.current_process.peb.modules[1].pe.exports["LdrLoadDll"] - - TEST_CASE.assertEqual(dbg.current_process.pid, calc.pid) - TEST_CASE.assertEqual(dbg.current_process.read_memory(addr, 1), "\xcc") - TEST_CASE.assertEqual(dbg.current_thread.context.pc, addr) - TEST_CASE.assertEqual(LdrLoadDlladdr, addr) - data[0] += 1 - d.current_process.exit() - - calc = pop_calc_32(dwCreationFlags=DEBUG_PROCESS) - - d = windows.debug.Debugger(calc) - d.add_bp(TSTBP("ntdll!LdrLoadDll")) - 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 - data = [0] - class TSTBP(windows.debug.HXBreakpoint): - def trigger(self, dbg, exc): - addr = exc.ExceptionRecord.ExceptionAddress - LdrLoadDlladdr = dbg.current_process.peb.modules[1].pe.exports["LdrLoadDll"] - - TEST_CASE.assertEqual(dbg.current_process.pid, calc.pid) - TEST_CASE.assertEqual(dbg.current_thread.context.pc, addr) - TEST_CASE.assertEqual(LdrLoadDlladdr, addr) - TEST_CASE.assertNotEqual(dbg.current_thread.context.Dr7, 0) - TEST_CASE.assertNotEqual(dbg.current_process.read_memory(addr, 1), "\xcc") - data[0] += 1 - d.current_process.exit() - - calc = pop_calc_32(dwCreationFlags=DEBUG_PROCESS) - d = windows.debug.Debugger(calc) - d.add_bp(TSTBP("ntdll!LdrLoadDll")) - 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" - 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 - - @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 - data = [0] - class TSTBP(windows.debug.HXBreakpoint): - def trigger(self, dbg, exc): - addr = exc.ExceptionRecord.ExceptionAddress - TEST_CASE.assertEqual(dbg.current_process.pid, calc.pid) - TEST_CASE.assertEqual(dbg.current_thread.context.pc, dbg._resolve(self.addr, dbg.current_process)) - TEST_CASE.assertNotEqual(dbg.current_thread.context.Dr7, 0) - TEST_CASE.assertNotEqual(dbg.current_process.read_memory(addr, 1), "\xcc") - data[0] += 1 - if data[0] == 1: - # Perform a loaddll in a new thread :) - # 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: - for t in dbg.current_process.threads: - TEST_CASE.assertNotEqual(t.context.Dr7, 0) - d.current_process.exit() - - calc = pop_calc_32(dwCreationFlags=DEBUG_PROCESS) - d = windows.debug.Debugger(calc) - d.add_bp(TSTBP("ntdll!LdrLoadDll")) - # Code that will load wintrust ! - 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 - NB_SINGLE_STEP = 3 - data = [] - - class MyDbg(windows.debug.Debugger): - def on_single_step(self, exception): - # Check that later created thread have their HWX breakpoint :) - addr = exception.ExceptionRecord.ExceptionAddress - TEST_CASE.assertEqual(self.current_thread.context.pc, addr) - if len(data) < NB_SINGLE_STEP: - 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() - - calc = pop_calc_32(dwCreationFlags=DEBUG_PROCESS) - d = MyDbg(calc) - addr = calc.virtual_alloc(0x1000) - calc.write_memory(addr, "\x90" * 3 + "\xc3") - d.add_bp(TSTBP(addr)) - calc.create_thread(addr, 0) - d.loop() - # Used to verif we actually called the Breakpoints - TEST_CASE.assertEqual(len(data), NB_SINGLE_STEP) - 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 - NB_SINGLE_STEP = 3 - data = [] - - class MyDbg(windows.debug.Debugger): - def on_single_step(self, exception): - # Check that later created thread have their HWX breakpoint :) - addr = exception.ExceptionRecord.ExceptionAddress - TEST_CASE.assertEqual(self.current_thread.context.pc, addr) - if len(data) < NB_SINGLE_STEP: - data.append(addr) - return self.single_step() - self.current_process.exit() - return - - class TSTBP(windows.debug.HXBreakpoint): - """Check that BP/dbg can trigger single step and that instruction follows""" - def trigger(self, dbg, exc): - return dbg.single_step() - - calc = pop_calc_32(dwCreationFlags=DEBUG_PROCESS) - d = MyDbg(calc) - addr = calc.virtual_alloc(0x1000) - calc.write_memory(addr, "\x90" * 3 + "\xc3") - d.add_bp(TSTBP(addr)) - calc.create_thread(addr, 0) - d.loop() - # Used to verif we actually called the Breakpoints - TEST_CASE.assertEqual(len(data), NB_SINGLE_STEP) - 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 - store_data = [0] - class TSTBP(windows.debug.MemoryBreakpoint): - #DEFAULT_PROTECT = PAGE_READONLY - #DEFAULT_PROTECT = PAGE_READONLY - DEFAULT_EVENTS = "W" - """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.Eax - if eax == 42: - dbg.current_process.exit() - return - TEST_CASE.assertEqual(fault_addr, data + eax) - store_data[0] += 1 - return - - calc = pop_calc_32(dwCreationFlags=DEBUG_PROCESS) - d = windows.debug.Debugger(calc) - addr = calc.virtual_alloc(0x1000) - data = calc.virtual_alloc(0x1000) - - injected = x86.MultipleInstr() - injected += x86.Mov("EAX", 0) - injected += x86.Mov(x86.deref(data), "EAX") - injected += x86.Add("EAX", 4) - injected += x86.Mov(x86.deref(data + 4), "EAX") - injected += x86.Add("EAX", 4) - # This one should NOT trigger the MemBP of size 8 - injected += x86.Mov(x86.deref(data + 8), "EAX") - injected += x86.Mov("EAX", 42) - injected += x86.Mov(x86.deref(data), "EAX") - injected += x86.Ret() - - calc.write_memory(addr, injected.get_code()) - d.add_bp(TSTBP(data, size=0x8)) - calc.create_thread(addr, 0) - d.loop() - # 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 - NB_NOP_IN_PAGE = 3 - data = [] - - class TSTBP(windows.debug.MemoryBreakpoint): - """Check that BP/dbg can trigger single step and that instruction follows""" - #DEFAULT_PROTECT = PAGE_NOACCESS - DEFAULT_EVENTS = "X" - def trigger(self, dbg, exc): - fault_addr = exc.ExceptionRecord.ExceptionInformation[1] - data.append(fault_addr) - if len(data) == NB_NOP_IN_PAGE + 1: - dbg.current_process.exit() - - calc = pop_calc_32(dwCreationFlags=DEBUG_PROCESS) - d = windows.debug.Debugger(calc) - addr = calc.virtual_alloc(0x1000) - calc.write_memory(addr, "\x90" * NB_NOP_IN_PAGE + "\xc3") - d.add_bp(TSTBP(addr, size=0x1000)) - calc.create_thread(addr, 0) - d.loop() - # Used to verif we actually called the Breakpoints - TEST_CASE.assertEqual(len(data), NB_NOP_IN_PAGE + 1) - 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 = [] - - def do_check(): - calc.execute_python_unsafe("open(u'FILENAME1')").wait() - calc.execute_python_unsafe("open(u'FILENAME2')").wait() - calc.execute_python_unsafe("open(u'FILENAME3')").wait() - calc.exit() - - class TSTBP(windows.debug.Breakpoint): - def trigger(self, dbg, exc): - addr = exc.ExceptionRecord.ExceptionAddress - ctx = dbg.current_thread.context - filename = dbg.current_process.read_wstring(dbg.current_process.read_ptr(ctx.sp + 0x4)) - data.append(filename) - if filename == u"FILENAME2": - dbg.del_bp(self) - - calc = pop_calc_32(dwCreationFlags=DEBUG_PROCESS) - d = windows.debug.Debugger(calc) - d.add_bp(TSTBP("kernel32!CreateFileW")) - threading.Thread(target=do_check).start() - d.loop() - TEST_CASE.assertEqual(data, [u"FILENAME1", u"FILENAME2"]) - - @check_for_gc_garbage - def test_standard_breakpoint_remove(self): - TEST_CASE = self - data = [] - - def do_check(): - calc.execute_python_unsafe("open(u'FILENAME1')").wait() - calc.execute_python_unsafe("open(u'FILENAME2')").wait() - d.del_bp(the_bp) - calc.execute_python_unsafe("open(u'FILENAME3')").wait() - calc.exit() - - class TSTBP(windows.debug.Breakpoint): - def trigger(self, dbg, exc): - addr = exc.ExceptionRecord.ExceptionAddress - ctx = dbg.current_thread.context - filename = dbg.current_process.read_wstring(dbg.current_process.read_ptr(ctx.sp + 0x4)) - data.append(filename) - - calc = pop_calc_32(dwCreationFlags=DEBUG_PROCESS) - d = windows.debug.Debugger(calc) - the_bp = TSTBP("kernel32!CreateFileW") - d.add_bp(the_bp) - threading.Thread(target=do_check).start() - d.loop() - TEST_CASE.assertEqual(data, [u"FILENAME1", u"FILENAME2"]) - - @check_for_gc_garbage - def test_hxbp_breakpoint_remove(self): - TEST_CASE = self - data = [] - - def do_check(): - calc.execute_python_unsafe("open(u'FILENAME1')").wait() - calc.execute_python_unsafe("open(u'FILENAME2')").wait() - d.del_bp(the_bp) - calc.execute_python_unsafe("open(u'FILENAME3')").wait() - calc.exit() - - class TSTBP(windows.debug.HXBreakpoint): - def trigger(self, dbg, exc): - addr = exc.ExceptionRecord.ExceptionAddress - ctx = dbg.current_thread.context - filename = dbg.current_process.read_wstring(dbg.current_process.read_ptr(ctx.sp + 0x4)) - data.append(filename) - - calc = pop_calc_32(dwCreationFlags=DEBUG_PROCESS) - d = windows.debug.Debugger(calc) - the_bp = TSTBP("kernel32!CreateFileW") - d.add_bp(the_bp) - threading.Thread(target=do_check).start() - 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 = [] - - def do_check(): - calc.execute_python_unsafe("open(u'FILENAME1')").wait() - calc.execute_python_unsafe("open(u'FILENAME2')").wait() - calc.execute_python_unsafe("open(u'FILENAME3')").wait() - calc.exit() - - class TSTBP(windows.debug.HXBreakpoint): - def trigger(self, dbg, exc): - addr = exc.ExceptionRecord.ExceptionAddress - ctx = dbg.current_thread.context - filename = dbg.current_process.read_wstring(dbg.current_process.read_ptr(ctx.sp + 0x4)) - data.append(filename) - if filename == u"FILENAME2": - #import pdb;pdb.set_trace() - dbg.del_bp(self) - - calc = pop_calc_32(dwCreationFlags=DEBUG_PROCESS) - d = windows.debug.Debugger(calc) - d.add_bp(TSTBP("kernel32!CreateFileW")) - threading.Thread(target=do_check).start() - d.loop() - TEST_CASE.assertEqual(data, [u"FILENAME1", u"FILENAME2"]) - - @check_for_gc_garbage - def test_mem_breakpoint_remove(self): - TEST_CASE = self - data = [] - - def generate_read_at(addr): - res = x86.MultipleInstr() - res += x86.Mov("EAX", x86.deref(addr)) - res += x86.Ret() - return res.get_code() - - def do_check(): - calc.execute(generate_read_at(data_addr)).wait() - calc.execute(generate_read_at(data_addr + 4)).wait() - d.del_bp(the_bp) - calc.execute(generate_read_at(data_addr + 8)).wait() - calc.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) - - calc = pop_calc_32(dwCreationFlags=DEBUG_PROCESS) - d = windows.debug.Debugger(calc) - data_addr = calc.virtual_alloc(0x1000) - the_bp = TSTBP(data_addr, size=0x1000) - d.add_bp(the_bp) - threading.Thread(target=do_check).start() - 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 = [] - - def generate_read_at(addr): - res = x86.MultipleInstr() - res += x86.Mov("EAX", x86.deref(addr)) - res += x86.Ret() - return res.get_code() - - def do_check(): - calc.execute(generate_read_at(data_addr)).wait() - calc.execute(generate_read_at(data_addr + 4)).wait() - calc.execute(generate_read_at(data_addr + 8)).wait() - calc.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) - - calc = pop_calc_32(dwCreationFlags=DEBUG_PROCESS) - d = windows.debug.Debugger(calc) - data_addr = calc.virtual_alloc(0x1000) - the_bp = TSTBP(data_addr, size=0x1000) - d.add_bp(the_bp) - threading.Thread(target=do_check).start() - 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 = [] - - def generate_read_at(addr): - res = x86.MultipleInstr() - res += x86.Mov("EAX", x86.deref(addr)) - res += x86.Ret() - return res.get_code() - - def generate_write_at(addr): - res = x86.MultipleInstr() - res += x86.Mov(x86.deref(addr), "EAX") - res += x86.Ret() - return res.get_code() - - def do_check(): - calc.execute(generate_read_at(data_addr)).wait() - calc.execute(generate_write_at(data_addr + 4)).wait() - calc.execute(generate_read_at(data_addr + 0x500)).wait() - calc.execute(generate_write_at(data_addr + 0x504)).wait() - calc.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)) - - calc = pop_calc_32(dwCreationFlags=DEBUG_PROCESS) - d = windows.debug.Debugger(calc) - data_addr = calc.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)] - - - 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): - exe_name = self.current_process.peb.modules[0].name - this_process_modules = self._module_by_process[self.current_process.pid] - TEST_CASE.assertIn(exe_name, this_process_modules.keys()) - self.current_process.exit() - - TEST_CASE = self - calc = pop_calc_32(dwCreationFlags=DEBUG_PROCESS) - 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): - exename = os.path.basename(calc.peb.imagepath.str) - this_process_modules = self._module_by_process[self.current_process.pid] - TEST_CASE.assertIn(exename, this_process_modules.keys()) - self.current_process.exit() - - TEST_CASE = self - calc = pop_calc_32(dwCreationFlags=DEBUG_PROCESS) - d = MyDbg(calc) - d.loop() - - @check_for_gc_garbage - def test_bp_exe_by_name(self): - NBCALL = [0] - TEST_CASE = self - CALC_ALIVE = True - class TSTBP(windows.debug.Breakpoint): - def trigger(self, dbg, exc): - NBCALL[0] += 1 - TEST_CASE.assertEqual(NBCALL[0], 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, calc.exit).start() - - calc = pop_calc_32(dwCreationFlags=DEBUG_PROCESS) - exepe = windows.pe_parse.GetPEFile(calc.peb.ImageBaseAddress, calc) - entrypoint = exepe.get_OptionalHeader().AddressOfEntryPoint - exename = os.path.basename(calc.peb.imagepath.str) - d = windows.debug.Debugger(calc) - # 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() - self.assertEqual(NBCALL[0], 1) - -if __name__ == '__main__': - alltests = unittest.TestSuite() - alltests.addTest(unittest.makeSuite(DebuggerTestCase)) - alltests.debug() - tester = unittest.TextTestRunner(verbosity=2) - tester.run(alltests) \ No newline at end of file diff --git a/windows/test/test_hooks.py b/windows/test/test_hooks.py deleted file mode 100644 index 1546719..0000000 --- a/windows/test/test_hooks.py +++ /dev/null @@ -1,183 +0,0 @@ -import ctypes -import textwrap - -from test_utils import * -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] - 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) - self.assertEqual(k.handle, 12345678) - self.assertEqual(hook_value[0], open_args) - # 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] - 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 self.assertRaises(WindowsError) as ar: - _winreg.OpenKey(*open_args) - 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 - # 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") - 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) - t.wait() - 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") - 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"]() - - 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() - result = t.exit_code - if result > 100: - import pdb;pdb.set_trace() - return result - - 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) - - @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) diff --git a/windows/test/test_syswow.py b/windows/test/test_syswow.py deleted file mode 100644 index 28c39e9..0000000 --- a/windows/test/test_syswow.py +++ /dev/null @@ -1,74 +0,0 @@ -import windows -import time -import textwrap -from test_utils import * -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) - self.assertEqual(res, 0x4242424242424242) - - @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] - self.assertIn("wow64.dll", modules_names) - # Parsing - wow64 = [m for m in peb64.modules if m.name == "wow64.dll"][0] - 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 - modules_names = [m.name for m in peb64.modules] - self.assertIn("wow64.dll", modules_names) - # Parsing - wow64 = [m for m in peb64.modules if m.name == "wow64.dll"][0] - 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) - - 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 = calc.execute_python_unsafe(textwrap.dedent(remote_python_code)) - # Wait for python execution - while calc.read_qword(addr) != 0x8877665544332211: - pass - ctx = t.context_syswow - # Check the get context - self.assertEqual(ctx.R11, 0x1122334455667788) - self.assertEqual(calc.read_memory(ctx.Rip, 2), x64.assemble("label :loop; jmp :loop")) - t.suspend() - calc.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() - self.assertEqual(RETURN_VALUE, calc.read_qword(addr)) - - diff --git a/windows/test/test_utils.py b/windows/test/test_utils.py deleted file mode 100644 index c42a2c8..0000000 --- a/windows/test/test_utils.py +++ /dev/null @@ -1,132 +0,0 @@ -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 -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 - -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 = 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 windows.system.version[0] < 10: - test_binary_name = "calc.exe" -else: - test_binary_name = "cmd.exe" - test_binary_name = "notepad.exe" - -DEFAULT_CREATION_FLAGS = CREATE_NEW_CONSOLE - -if is_windows_32_bits: - def pop_calc_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_calc_64(dwCreationFlags=DEFAULT_CREATION_FLAGS): - raise WindowsError("Cannot create calc64 in 32bits system") -else: - def pop_calc_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_calc_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_calc_64(dwCreationFlags=DEFAULT_CREATION_FLAGS): - return windows.utils.create_process(r"C:\Windows\system32\{0}".format(test_binary_name), dwCreationFlags=dwCreationFlags, show_windows=True) - - -@contextmanager -def Calc64(dwCreationFlags=DEFAULT_CREATION_FLAGS, 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=DEFAULT_CREATION_FLAGS, 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) - - -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 check_for_handle_leak(f): - def wrapper(testcase, *args, **kwargs): - current_process_hdebugger.refresh_handles() - res = f(testcase, *args, **kwargs) - leaked_handles = current_process_hdebugger.get_new_handle() - testcase.assertFalse(leaked_handles, "Test Leaked <{0}> handles of types ({1})".format(len(leaked_handles), set(h.type for h in leaked_handles))) - return res - return wrapper - -def print_call(f): - def wrapper(*args, **kwargs): - res = f(*args, **kwargs) - print("Call to <{0}>({1}) returned <{2}>".format(f.func_name, (args, kwargs), res)) - return res - return wrapper - - -class HandleDebugger(object): - def __init__(self, pid): - self.pid = pid - self.handles = 0 - - 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): - nh = self.get_handles() - handle_diff = set(h.wValue for h in nh) - set(h.wValue for h in self.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)