First try at single_step and memory breakpoints in Debugger + test

TODO

@@ -20,6 +20,10 @@ TODO:
     - Add test for debugger with breakpoint that add another breakpoint on trigger
 
     - Some test/doc on windows.system.handles
+    - Some test/doc on debugger and MemBP
+
+    - registry
+        - test !
 
 
 Documentation

samples/test_code.py

@@ -52,7 +52,7 @@ def hexdump(string, start_addr=0):
 
 class CodeTesteur(dbg.Debugger):
     def __init__(self, process, code, register_start={}):
-        super(CodeTesteur, self).__init__(process, already_debuggable=True)
+        super(CodeTesteur, self).__init__(process)
 
         self.initial_code = code
         code += "\xcc"

windows/debug.py

@@ -20,6 +20,7 @@ from windows.winobject.exception import VectoredException
 
 STANDARD_BP = "BP"
 HARDWARE_EXEC_BP = "HXBP"
+MEMORY_BREAKPOINT = "MEMBP"
 
 class DEBUG_EVENT(DEBUG_EVENT):
     KNOWN_EVENT_CODE = dict((x,x) for x in [EXCEPTION_DEBUG_EVENT,
@@ -62,13 +63,14 @@ class Debugger(object):
 
         self._explicit_single_step = {}
 
+        self._watched_memory = []
+
+
     @classmethod
     def attach(cls, target):
         winproxy.DebugActiveProcess(target.pid)
         return cls(target)
 
-
-
     def _init_dispatch_handlers(self):
         dbg_evt_dispatch = {}
         dbg_evt_dispatch[EXCEPTION_DEBUG_EVENT] = self._handle_exception
@@ -138,7 +140,7 @@ class Debugger(object):
     def _setup_breakpoint(self, bp, target):
         _setup_method = getattr(self, "_setup_breakpoint_" + bp.type)
         if target is None:
-            if bp.type == STANDARD_BP: #TODO: better..
+            if bp.type in [STANDARD_BP, MEMORY_BREAKPOINT]: #TODO: better..
                 targets = self.processes.values()
             else:
                 targets = self.threads.values()
@@ -181,6 +183,16 @@ class Debugger(object):
         self.breakpoints[target.owner.pid][addr] = bp
         return True
 
+    def _setup_breakpoint_MEMBP(self, bp, target):
+        addr = self._resolve(bp.addr, target)
+        if addr is None:
+            return False
+        old_prot = DWORD()
+        target.virtual_protect(addr, bp.size, bp.protect, old_prot)
+        self._watched_memory.append((bp, addr, addr + bp.size, old_prot.value))
+        # TODO: watch for overlap with other MEM breakpoints
+        return True
+
     def _setup_pending_breakpoints_new_process(self, new_process):
         for bp in self._pending_breakpoints_new[None]:
             if bp.apply_to_target(new_process): #BP for thread or process ?
@@ -243,14 +255,107 @@ class Debugger(object):
         process.write_memory(addr, self._memory_save[process.pid][addr])
         regs = thread.context
         regs.EFlags |= (1 << 8)
-        #regs.pc -= 1 # Done at 269 before dispatch
+        #regs.pc -= 1 # Done in _handle_exception_breakpoint before dispatch
         thread.set_context(regs)
-        self._breakpoint_to_reput[thread.tid] = addr #Register pending breakpoint for next single step
+        bp = self.breakpoints[self.current_process.pid][addr]
+        self._breakpoint_to_reput[thread.tid].append(bp) #Register pending breakpoint for next single step
+
+    def _pass_memory_breakpoint(self, bp, begin, end, original_prot):
+        cp = self.current_process
+        cp.virtual_protect(begin, bp.size, original_prot, None)
+        thread = self.current_thread
+        ctx = thread.context
+        ctx.EEFlags.TF = 1
+        thread.set_context(ctx)
+        self._breakpoint_to_reput[thread.tid].append(bp)
 
     # debug event handlers
     def _handle_unknown_debug_event(self, debug_event):
         raise NotImplementedError("dwDebugEventCode = {0}".format(debug_event.dwDebugEventCode))
 
+
+    def _handle_exception_breakpoint(self, exception, excp_addr):
+        if excp_addr in self.breakpoints[self.current_process.pid]:
+            thread = self.current_thread
+            ctx = thread.context
+            ctx.pc -= 1
+            thread.set_context(ctx)
+            continue_flag = self._dispatch_breakpoint(exception, excp_addr)
+            self._explicit_single_step[self.current_thread.tid] = self.current_thread.context.EEFlags.TF
+            self._pass_breakpoint(excp_addr)
+            return continue_flag
+        return self.on_exception(exception)
+
+    # TODO: mov me
+    def _restore_breakpoints(self):
+        for bp in self._breakpoint_to_reput[self.current_thread.tid]:
+            #print("TODO: restore {0}".format(bp))
+            if bp.type == HARDWARE_EXEC_BP:
+                raise NotImplementedError("Why is this here ? we use RF flags to pass HXBP")
+            #print("[RST] Restoring <{0}>".format(bp))
+            self._setup_breakpoint(bp, self.current_process)
+        del self._breakpoint_to_reput[self.current_thread.tid][:]
+        return
+
+
+    def _handle_exception_singlestep(self, exception, excp_addr):
+        if self.current_thread.tid in self._breakpoint_to_reput and self._breakpoint_to_reput[self.current_thread.tid]:
+            self._restore_breakpoints()
+            if self._explicit_single_step[self.current_thread.tid]:
+                self.on_single_step(exception) # TODO: default implem / dispatcher ?
+            self._explicit_single_step[self.current_thread.tid] = self.current_thread.context.EEFlags.TF
+            return DBG_CONTINUE
+        elif excp_addr in self.breakpoints[self.current_process.pid]:
+            # Verif that's not a standard BP ?
+            bp = self.breakpoints[self.current_process.pid][excp_addr]
+            bp.trigger(self, exception)
+            ctx = self.current_thread.context
+            self._explicit_single_step[self.current_thread.tid] = ctx.EEFlags.TF
+            ctx.EEFlags.RF = 1
+            self.current_thread.set_context(ctx)
+            return DBG_CONTINUE
+        elif self._explicit_single_step[self.current_thread.tid]:
+            continue_flag = self.on_single_step(exception)
+            return continue_flag # TODO: default implem / dispatcher ?
+        else:
+            continue_flag = self.on_exception(exception)
+            self._explicit_single_step[self.current_thread.tid] = self.current_thread.context.EEFlags.TF
+            return continue_flag
+
+    def _handle_exception_access_violation(self, exception, excp_addr):
+        READ = 0
+        WRITE = 1
+        EXEC = 2
+
+        fault_type = exception.ExceptionRecord.ExceptionInformation[0]
+        fault_addr = exception.ExceptionRecord.ExceptionInformation[1]
+        pc_addr = self.current_thread.context.pc
+        if fault_addr == pc_addr:
+            fault_type = EXEC
+
+        #print("FAULT AT {0:#x} ({1})".format(fault_addr, fault_type))
+        for bp, begin, end, original_prot in self._watched_memory:
+            if begin <= fault_addr < end:
+                ## Reject bad EXCEPTION ?
+                #if fault_type == EXEC and bp.PROTECT not in [PAGE_EXECUTE, PAGE_EXECUTE_READ, PAGE_EXECUTE_READWRITE]:
+                #    break
+                #if fault_type == READ and bp.PROTECT not in [PAGE_EXECUTE, PAGE_EXECUTE_READ, PAGE_EXECUTE_READWRITE]:
+                #    break
+                #if fault_type == EXEC and bp.PROTECT not in [PAGE_EXECUTE, PAGE_EXECUTE_READ, PAGE_EXECUTE_READWRITE]:
+                #    break
+
+
+
+                #print("BP MEM TRIGGER {0}".format(bp))
+                continue_flag = bp.trigger(self, exception)
+                self._explicit_single_step[self.current_thread.tid] = self.current_thread.context.EEFlags.TF
+                self._pass_memory_breakpoint(bp, begin, end, original_prot)
+                return continue_flag
+        else:
+            self.on_exception(exception)
+
+
+    # TODO: self._explicit_single_step setup by single_step() ? check at the end ? finally ?
     def _handle_exception(self, debug_event):
         """Handle EXCEPTION_DEBUG_EVENT"""
         exception = debug_event.u.Exception
@@ -263,40 +368,19 @@ class Debugger(object):
 
         excp_code = exception.ExceptionRecord.ExceptionCode
         excp_addr = exception.ExceptionRecord.ExceptionAddress
+
+        #print("[DBG] Got a <{0}> in <{1}>".format(excp_code, self.current_thread.tid))
+
         if excp_code in [EXCEPTION_BREAKPOINT, STATUS_WX86_BREAKPOINT] and excp_addr in self.breakpoints[self.current_process.pid]:
-            thread = self.current_thread
-            ctx = thread.context
-            ctx.pc -= 1
-            thread.set_context(ctx)
-            continue_flag = self._dispatch_breakpoint(exception, excp_addr)
-            self._explicit_single_step[self.current_thread.tid] = self.current_thread.context.EEFlags.TF
-            self._pass_breakpoint(excp_addr)
-            return continue_flag
+            return self._handle_exception_breakpoint(exception, excp_addr)
         elif excp_code in [EXCEPTION_SINGLE_STEP, STATUS_WX86_SINGLE_STEP]:
-            if self.current_thread.tid in self._breakpoint_to_reput:
-                addr = self._breakpoint_to_reput[self.current_thread.tid]
-                del self._breakpoint_to_reput[self.current_thread.tid]
-                # Re-put the breakpoint
-                self.current_process.write_memory(addr, "\xcc")
-                if self._explicit_single_step[self.current_thread.tid]:
-                    self.on_single_step(exception) # TODO: default implem / dispatcher ?
-                self._explicit_single_step[self.current_thread.tid] = self.current_thread.context.EEFlags.TF
-                return DBG_CONTINUE
-            elif excp_addr in self.breakpoints[self.current_process.pid]:
-                # Verif that's not a standard BP ?
-                bp = self.breakpoints[self.current_process.pid][excp_addr]
-                # TODO: What to do if explicit single step ?
-                bp.trigger(self, exception)
-                ctx = self.current_thread.context
-                ctx.EEFlags.RF = 1
-                self.current_thread.set_context(ctx)
-                return DBG_CONTINUE
-            elif self._explicit_single_step[self.current_thread.tid]:
-                return self.on_single_step(exception) # TODO: default implem / dispatcher ?
-            else:
-                return self.on_exception(exception)
-        else: # Do not trigger self.on_exception if breakpoint was registered
-            return self.on_exception(exception)
+            return self._handle_exception_singlestep(exception, excp_addr)
+        elif excp_code in [EXCEPTION_ACCESS_VIOLATION]:
+            return self._handle_exception_access_violation(exception, excp_addr)
+        else:
+            continue_flag = self.on_exception(exception)
+            self._explicit_single_step[self.current_thread.tid] = self.current_thread.context.EEFlags.TF
+            return continue_flag
 
 
     def _get_loaded_dll(self, load_dll):
@@ -327,6 +411,8 @@ class Debugger(object):
         self.current_process = WinProcess._from_handle(create_process.hProcess)
         self.current_thread = WinThread._from_handle(create_process.hThread)
         self.threads[self.current_thread.tid] = self.current_thread
+        self._explicit_single_step[self.current_thread.tid] = False
+        self._breakpoint_to_reput[self.current_thread.tid] = []
         self.processes[self.current_process.pid] = self.current_process
         self.breakpoints[self.current_process.pid] = {}
         self._module_by_process[self.current_process.pid] = {}
@@ -342,6 +428,8 @@ class Debugger(object):
         exit_process = debug_event.u.ExitProcess
         retvalue = self.on_exit_process(exit_process)
         del self.threads[self.current_thread.tid]
+        del self._explicit_single_step[self.current_thread.tid]
+        del self._breakpoint_to_reput[self.current_thread.tid]
         del self.processes[self.current_process.pid]
         # Hack IT, ContinueDebugEvent will close the HANDLE for us
         # Should we make another handle instead ?
@@ -356,6 +444,7 @@ class Debugger(object):
         self.current_thread = WinThread._from_handle(create_thread.hThread)
         self.threads[self.current_thread.tid] = self.current_thread
         self._explicit_single_step[self.current_thread.tid] = False
+        self._breakpoint_to_reput[self.current_thread.tid] = []
         self._setup_pending_breakpoints_new_thread(self.current_thread)
         return self.on_create_thread(create_thread)
 
@@ -367,6 +456,7 @@ class Debugger(object):
         retvalue = self.on_exit_thread(exit_thread)
         del self.threads[self.current_thread.tid]
         del self._explicit_single_step[self.current_thread.tid]
+        del self._breakpoint_to_reput[self.current_thread.tid]
         # Hack IT, ContinueDebugEvent will close the HANDLE for us
         # Should we make another handle instead ?
         dbgprint("Removing handle {0} for {1} (will be closed by continueDebugEvent".format(hex(self.current_thread._handle), self.current_thread), "HANDLE")
@@ -450,8 +540,6 @@ class Debugger(object):
         ctx.EEFlags.TF = 1
         t.set_context(ctx)
 
-
-
     # Public callback
     def on_exception(self, exception):
         """Called on exception event other that known breakpoint. ``exception`` is one of the following type:
@@ -466,6 +554,9 @@ class Debugger(object):
             return DBG_EXCEPTION_NOT_HANDLED
         return DBG_CONTINUE
 
+    def on_single_step(self, exception):
+        raise NotImplementedError("Debugger that explicitly single step should implement <on_single_step>")
+
     def on_create_process(self, create_process):
         """Called on create_process event (for param type see https://msdn.microsoft.com/en-us/library/windows/desktop/ms679286(v=vs.85).aspx)"""
         pass
@@ -535,6 +626,21 @@ class HXBreakpoint(Breakpoint):
     def apply_to_target(self, target):
         return isinstance(target, WinThread)
 
+class MemoryBreakpoint(Breakpoint):
+    type = MEMORY_BREAKPOINT
+
+    DEFAULT_PROTECT = PAGE_READONLY
+    DEFAULT_SIZE = 0x1000
+    def __init__(self, addr, size=None, prot=None):
+        super(MemoryBreakpoint, self).__init__(addr)
+        self.size = size if size is not None else self.DEFAULT_SIZE
+        self.protect = size if prot is not None else self.DEFAULT_PROTECT
+
+
+    def trigger(self, dbg, exception):
+        """Called when breakpoint is hit"""
+        pass
+
 
 class LocalDebugger(object):
     """A debugger interface around :func:`AddVectoredExceptionHandler`"""

windows/test/__init__.py

@@ -1,8 +1,9 @@
 
 from test_utils import *
 
-from mytest import WindowsTestCase, WindowsAPITestCase, DebuggerTestCase, NativeUtilsTestCase, SystemTestCase
+from mytest import WindowsTestCase, WindowsAPITestCase, NativeUtilsTestCase, SystemTestCase
 from test_hooks import HookTestCase
+from test_debugger import DebuggerTestCase
 
 
 __all__ = ["SystemTestCase", "WindowsTestCase", "WindowsAPITestCase", "DebuggerTestCase", "NativeUtilsTestCase", "HookTestCase"]

windows/test/mytest.py

@@ -496,7 +496,6 @@ class WindowsAPITestCase(unittest.TestCase):
             windows.winproxy.CreateFileA("NONEXISTFILE.FILE")
 
 class NativeUtilsTestCase(unittest.TestCase):
-
     @process_64bit_only
     def test_strlenw64(self):
         strlenw64 = windows.native_exec.create_function(nativeutils.StrlenW64.get_code(), [UINT, LPCWSTR])
@@ -553,296 +552,11 @@ class NativeUtilsTestCase(unittest.TestCase):
         self.assertEqual(getprocaddr32("KERNEL32.DLL", "YOLOAPI"), 0xffffffff)
 
 
-class DebuggerTestCase(unittest.TestCase):
-
-    def debuggable_calc_32(self):
-        return windows.utils.create_process(r"C:\python27\python.exe", dwCreationFlags=DEBUG_PROCESS | CREATE_NEW_CONSOLE, show_windows=True)
-
-    def test_init_breakpoint_callback(self):
-        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, already_debuggable=True)
-        d.loop()
-
-    def test_simple_standard_breakpoint(self):
-        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 - 1, 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, already_debuggable=True)
-        d.add_bp(TSTBP(LdrLoadDll32))
-        d.loop()
-
-    def test_standard_breakpoint_multiple_threads(self):
-        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 - 1, self.addr)
-                data[0] += 1
-                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, already_debuggable=True)
-        calc.execute("\xc3")
-        calc.execute("\xc3")
-        calc.execute("\xc3")
-        d.add_bp(TSTBP(LdrLoadDll32))
-        d.loop()
-
-    def test_simple_hwx_breakpoint(self):
-        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, already_debuggable=True)
-        d.add_bp(TSTBP(LdrLoadDll32))
-        d.loop()
-
-    def test_multiple_hwx_breakpoint(self):
-        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, already_debuggable=True)
-        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)
-
-    def test_four_hwx_breakpoint_fail(self):
-        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, already_debuggable=True)
-        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)
-
-    def test_hwx_breakpoint_are_on_all_thread(self):
-        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:
-                    d.current_process.exit()
-
-        calc = pop_calc_32(dwCreationFlags=DEBUG_PROCESS)
-        d = MyDbg(calc, already_debuggable=True)
-        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], 1)
-
-    def test_simple_breakpoint_name_addr(self):
-        TEST_CASE = self
-        data = [0]
-        class TSTBP(windows.debug.Breakpoint):
-            def trigger(self, dbg, exc):
-                addr = exc.ExceptionRecord.ExceptionAddress
-                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 - 1, addr)
-                data[0] += 1
-                d.current_process.exit()
-
-        calc = pop_calc_32(dwCreationFlags=DEBUG_PROCESS)
-
-        d = windows.debug.Debugger(calc, already_debuggable=True)
-        d.add_bp(TSTBP("ntdll.dll!LdrLoadDll"))
-        d.loop()
-        TEST_CASE.assertEqual(data[0], 1)
-
-    def test_simple_hardware_breakpoint_name_addr(self):
-        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
-                d.current_process.exit()
-
-        calc = pop_calc_32(dwCreationFlags=DEBUG_PROCESS)
-        d = windows.debug.Debugger(calc, already_debuggable=True)
-        d.add_bp(TSTBP("ntdll.dll!LdrLoadDll"))
-        d.loop()
-        TEST_CASE.assertEqual(data[0], 1)
-
-    def perform_manual_getproc_loadlib_32_yolo(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
-
-
-    def test_hardware_breakpoint_name_addr(self):
-        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's trigger a bp
-                    t = TEST_CASE.perform_manual_getproc_loadlib_32_yolo(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, already_debuggable=True)
-        d.add_bp(TSTBP("ntdll.dll!LdrLoadDll"))
-        # Code that will load wintrust !
-        d.loop()
-        #TEST_CASE.assertEqual(data[0], 1)
-
-
 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(DebuggerTestCase))
     alltests.addTest(unittest.makeSuite(NativeUtilsTestCase))
     alltests.debug()
     tester = unittest.TextTestRunner(verbosity=2)

windows/test/test_debugger.py

@@ -0,0 +1,441 @@
+from test_utils import *
+from windows.generated_def.winstructs import *
+
+class DebuggerTestCase(unittest.TestCase):
+    def debuggable_calc_32(self):
+        return windows.utils.create_process(r"C:\python27\python.exe", dwCreationFlags=DEBUG_PROCESS | CREATE_NEW_CONSOLE, show_windows=True)
+
+    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()
+
+    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()
+
+    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()
+
+    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)
+
+    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)
+
+    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)
+
+    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.dll!LdrLoadDll"))
+        d.loop()
+        TEST_CASE.assertEqual(data[0], 1)
+
+    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.dll!LdrLoadDll"))
+        d.loop()
+        TEST_CASE.assertEqual(data[0], 1)
+
+    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
+
+
+    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's trigger 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.dll!LdrLoadDll"))
+        # Code that will load wintrust !
+        d.loop()
+        #TEST_CASE.assertEqual(data[0], 1)
+
+    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)
+
+
+    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)
+
+
+    def test_memory_breakpoint_write(self):
+        """Check MemoryBP WRITE"""
+
+        TEST_CASE = self
+        store_data = [0]
+        class TSTBP(windows.debug.MemoryBreakpoint):
+            DEFAULT_PROTECT = PAGE_READONLY
+            """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
+                TEST_CASE.assertEqual(fault_addr, data + eax)
+                store_data[0] += 1
+                if store_data[0] == 2:
+                    dbg.current_process.exit()
+                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.Ret()
+
+        calc.write_memory(addr, injected.get_code())
+        d.add_bp(TSTBP(data, size=0x1000))
+        calc.create_thread(addr, 0)
+        d.loop()
+        # Used to verif we actually called the Breakpoints
+        TEST_CASE.assertEqual(store_data[0], 2)
+
+    def test_memory_breakpoint_exec(self):
+        """Check that HXBPBP/dbg can trigger single step"""
+        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
+            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)
+
+if __name__ == '__main__':
+    alltests = unittest.TestSuite()
+    alltests.addTest(unittest.makeSuite(DebuggerTestCase))
+    alltests.debug()
+    tester = unittest.TextTestRunner(verbosity=2)
+    tester.run(alltests)

windows/winobject/process.py

@@ -329,13 +329,13 @@ class Process(AutoHandle):
     def virtual_protected(self, addr, size, protect):
         """A context manager for local virtual_protect (old Protection are restored at exit)"""
         old_protect = DWORD()
-        self.low_virtual_protect(addr, size, protect, old_protect)
+        self.virtual_protect(addr, size, protect, old_protect)
         try:
             yield addr
         finally:
-             self.low_virtual_protect(addr, size, old_protect.value, old_protect)
+             self.virtual_protect(addr, size, old_protect.value, old_protect)
 
-    def low_virtual_protect(self, addr, size, protect, old_protect):
+    def virtual_protect(self, addr, size, protect, old_protect):
         if windows.current_process.bitness == 32 and self.bitness == 64:
             #addr = (addr >> 12) << 12
             #addr = ULONG64(addr)