pytest integration finished + moved tests/ to top dir

This commit is contained in:
hakril
2017-09-10 18:57:59 +02:00
parent 492b6f4f02
commit b3117ce4e3
25 changed files with 383 additions and 2371 deletions
-2
View File
@@ -1,2 +0,0 @@
[pytest]
usefixtures = check_for_handle_leak_final
+4 -1
View File
@@ -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("")
View File
@@ -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"""
@@ -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):
+274
View File
@@ -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()
@@ -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()
@@ -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
-256
View File
@@ -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"
-13
View File
@@ -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"]
-763
View File
@@ -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("<I", calc.read_memory(data, 4))[0]
self.assertEqual(dword, 0x42424242)
@windows_64bit_only
@check_for_gc_garbage
def test_execute_to_64(self):
with Calc64() as calc:
data = calc.virtual_alloc(0x1000)
shellcode = x64.MultipleInstr()
shellcode += x64.Mov('RAX', 0x4242424243434343)
shellcode += x64.Mov(x64.create_displacement(disp=data), 'RAX')
shellcode += x64.Ret()
calc.execute(shellcode.get_code())
time.sleep(0.1)
dword = struct.unpack("<Q", calc.read_memory(data, 8))[0]
self.assertEqual(dword, 0x4242424243434343)
# Python execution
@windows_64bit_only
@check_for_gc_garbage
def test_execute_python_to_64(self):
with Calc64() as calc:
data = calc.virtual_alloc(0x1000)
calc.execute_python('import ctypes; ctypes.c_uint.from_address({0}).value = 0x42424242'.format(data))
#time.sleep(0.1)
dword = struct.unpack("<I", calc.read_memory(data, 4))[0]
self.assertEqual(dword, 0x42424242)
@check_for_gc_garbage
def test_execute_python_to_32(self):
with Calc32() as calc:
data = calc.virtual_alloc(0x1000)
calc.execute_python('import ctypes; ctypes.c_uint.from_address({0}).value = 0x42424242'.format(data))
#time.sleep(0.1)
dword = struct.unpack("<I", calc.read_memory(data, 4))[0]
self.assertEqual(dword, 0x42424242)
@check_for_gc_garbage
def test_execute_python_to_32_suspended(self):
with Calc32(dwCreationFlags=CREATE_SUSPENDED) as calc:
data = calc.virtual_alloc(0x1000)
calc.execute_python('import ctypes; ctypes.c_uint.from_address({0}).value = 0x42424242'.format(data))
dword = struct.unpack("<I", calc.read_memory(data, 4))[0]
self.assertEqual(dword, 0x42424242)
# Check calc32 is still suspended:
# 1 thread | except windows 10 that pop threads
# main thread suspend count == 1
self.assertEqual(calc.threads[0].suspend(), 1)
if not is_windows_10:
self.assertEqual(len(calc.threads), 1)
@windows_64bit_only
@check_for_gc_garbage
def test_execute_python_to_64_suspended(self):
with Calc64(dwCreationFlags=CREATE_SUSPENDED) as calc:
data = calc.virtual_alloc(0x1000)
calc.execute_python('import ctypes; ctypes.c_uint.from_address({0}).value = 0x42424242'.format(data))
dword = struct.unpack("<I", calc.read_memory(data, 4))[0]
self.assertEqual(dword, 0x42424242)
# Check calc32 is still suspended:
# 1 thread | except windows 10 that pop threads
# main thread suspend count == 1
self.assertEqual(calc.threads[0].suspend(), 1)
if not is_windows_10:
self.assertEqual(len(calc.threads), 1)
@check_for_gc_garbage
def test_parse_remote_32_peb(self):
with Calc32() as calc:
# Wait for PEB initialization
# Yeah a don't know but on 32bits system the parsing might begin before
# InMemoryOrderModuleList is setup..
import time; time.sleep(0.1)
self.assertEqual(calc.peb.modules[0].name, test_binary_name)
@windows_64bit_only
@check_for_gc_garbage
def test_parse_remote_64_peb(self):
with Calc64() as calc:
self.assertEqual(calc.peb.modules[0].name, test_binary_name)
@check_for_gc_garbage
def test_parse_remote_32_pe(self):
with Calc32() as calc:
# Wait for PEB initialization
# Yeah a don't know but on 32bits system the parsing might begin before
# InMemoryOrderModuleList is setup..
import time; time.sleep(0.1)
mods = [m for m in calc.peb.modules if m.name == "kernel32.dll"]
self.assertTrue(mods, 'Could not find "kernel32.dll" in calc32')
k32 = mods[0]
mods[0].pe.sections[0].name # Just see if it's parse
self.assertEqual(mods[0].pe.export_name.lower(), "kernel32.dll")
get_current_proc_id = k32.pe.exports['GetCurrentProcessId']
# TODO: check get_current_proc_id value (but we cannot do 64->32 injection for now)
#if is_process_64_bits:
# raise NotImplementedError("Python execution 64->32")
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("<I", calc.read_memory(data, 4))[0]
self.assertEqual(dword, get_current_proc_id)
@windows_64bit_only
@check_for_gc_garbage
def test_parse_remote_64_pe(self):
with Calc64() as calc:
mods = [m for m in calc.peb.modules if m.name == "kernel32.dll"]
self.assertTrue(mods, 'Could not find "kernel32.dll" in calc32')
k32 = mods[0]
mods[0].pe.sections[0].name
self.assertEqual(mods[0].pe.export_name.lower(), "kernel32.dll")
get_current_proc_id = k32.pe.exports['GetCurrentProcessId']
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_ulonglong.from_address({1}).value = GetCurrentProcessId
""".format(os.getcwd(), data)
calc.execute_python(textwrap.dedent(remote_python_code))
#time.sleep(0.5)
dword = struct.unpack("<Q", calc.read_memory(data, 8))[0]
self.assertEqual(dword, get_current_proc_id)
@check_for_gc_garbage
def test_remote_peb_exe_32(self):
with Calc32() as calc:
exe = calc.peb.exe
exe_by_module = calc.peb.modules[0].pe
self.assertEqual(exe.baseaddr, exe_by_module.baseaddr)
self.assertEqual(exe.bitness, exe_by_module.bitness)
@windows_64bit_only
@check_for_gc_garbage
def test_remote_peb_exe_64(self):
with Calc64() as calc:
exe = calc.peb.exe
exe_by_module = calc.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_thread_exit_value_32(self):
with Calc32() as calc:
res = calc.execute_python("import time;time.sleep(0.1); 2")
self.assertEqual(res, True)
with self.assertRaises(windows.injection.RemotePythonError) as ar:
t = calc.execute_python("import time;time.sleep(0.1); raise ValueError('BYE')")
@windows_64bit_only
@check_for_gc_garbage
def test_thread_exit_value_64(self):
with Calc64() as calc:
res = calc.execute_python("import time;time.sleep(0.1); 2")
self.assertEqual(res, True)
with self.assertRaises(windows.injection.RemotePythonError) as ar:
t = calc.execute_python("import time;time.sleep(0.1); raise ValueError('BYE')")
@check_for_gc_garbage
def test_thread_start_address_32(self):
with Calc32() as calc:
t = calc.threads[0]
t.start_address # No better idea right now that checking for crash/exception
@windows_64bit_only
@check_for_gc_garbage
def test_thread_start_address_64(self):
with Calc64() as calc:
t = calc.threads[0]
t.start_address # No better idea right now that checking for crash/exception
@check_for_gc_garbage
def test_get_context_address_32(self):
with Calc32() as calc:
code = x86.MultipleInstr()
code += x86.Mov("EAX", 0x42424242)
code += x86.Label(":LOOP")
code += x86.Jmp(":LOOP")
t = calc.execute(code.get_code())
time.sleep(0.5)
cont = t.context
self.assertEqual(cont.Eax, 0x42424242)
@windows_64bit_only
@check_for_gc_garbage
def test_get_context_address_64(self):
with Calc64() as calc:
code = x64.MultipleInstr()
code += x64.Mov("RAX", 0x4242424243434343)
code += x64.Label(":LOOP")
code += x64.Jmp(":LOOP")
t = calc.execute(code.get_code())
time.sleep(0.5)
cont = t.context
self.assertEqual(cont.Rax, 0x4242424243434343)
@check_for_gc_garbage
def test_process_is_exit(self):
with Calc32(exit_code=42) as calc:
self.assertEqual(calc.is_exit, False)
# out of context manager: process is exit
self.assertEqual(calc.exit_code, 42)
self.assertEqual(calc.is_exit, True)
@check_for_gc_garbage
def test_set_thread_context_32(self):
code = x86.MultipleInstr()
code += x86.Label(":LOOP")
code += x86.Jmp(":LOOP")
data_len = len(code.get_code())
code += x86.Ret()
with Calc32() as calc:
t = calc.execute(code.get_code())
time.sleep(0.1)
self.assertEqual(calc.is_exit, False)
t.suspend()
ctx = t.context
ctx.Eip += data_len
ctx.Eax = 0x11223344
t.set_context(ctx)
t.resume()
time.sleep(0.1)
self.assertEqual(t.exit_code, 0x11223344)
@windows_64bit_only
@check_for_gc_garbage
def test_set_thread_context_64(self):
code = x64.MultipleInstr()
code += x64.Label(":LOOP")
code += x64.Jmp(":LOOP")
data_len = len(code.get_code())
code += x64.Ret()
with Calc64() as calc:
t = calc.execute(code.get_code())
time.sleep(0.1)
self.assertEqual(calc.is_exit, False)
t.suspend()
ctx = t.context
ctx.Rip += data_len
ctx.Rax = 0x11223344
t.set_context(ctx)
t.resume()
time.sleep(0.1)
self.assertEqual(t.exit_code, 0x11223344)
@check_for_gc_garbage
def test_load_library_32(self):
DLL = "wintrust.dll"
with Calc32() as calc:
calc.load_library(DLL)
self.assertIn(DLL, [m.name for m in calc.peb.modules])
@windows_64bit_only
@check_for_gc_garbage
def test_load_library_64(self):
DLL = "wintrust.dll"
with Calc64() as calc:
calc.load_library(DLL)
self.assertIn(DLL, [m.name for m in calc.peb.modules])
@check_for_gc_garbage
def test_token_info(self):
token = windows.current_process.token
self.assertIsInstance(token.computername, basestring)
self.assertIsInstance(token.username, basestring)
self.assertIsInstance(token.integrity, (int, long))
self.assertIsInstance(token.is_elevated, (bool))
@check_for_gc_garbage
def test_get_working_set_32(self):
with Calc32() 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))
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)
-100
View File
@@ -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)
-746
View File
@@ -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)
-183
View File
@@ -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)
-74
View File
@@ -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))
-132
View File
@@ -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)