admin/hakril-PythonForWindows
1
0
Fork 0
mirror of https://github.com/hakril/PythonForWindows synced 2026-06-08 14:31:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

Improve debugger handling of memBP for future use + fix MemBP prot init + add failing test of multitple BP with diff prot

Browse Source
This commit is contained in:
Clement Rouault
2016-07-13 18:23:26 +02:00
parent 82621b881a
commit 19205f73a1
4 changed files with 85 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
TODO
+1
View File
@@ -9,6 +9,7 @@ TODO:
- Test !! (bp, BP_HX, bp on only on process, bp_hx on only one thread..)
- test breakpoint with specific target
- Add test for debugger with breakpoint that add another breakpoint on trigger
- Handle MemBP of multiple types of the same page..
- remotectypes
- pretty sur I can get rid of PointerToStruct64/PointerToStruct32
windows/debug/breakpoints.py
+1 -1
View File
@@ -49,7 +49,7 @@ class MemoryBreakpoint(Breakpoint):
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
self.protect = prot if prot is not None else self.DEFAULT_PROTECT
def trigger(self, dbg, exception):
windows/debug/debugger.py
+34 -34
View File
@@ -1,5 +1,5 @@
import os.path
from collections import defaultdict
from collections import defaultdict, namedtuple
from contextlib import contextmanager
import windows
@@ -30,6 +30,8 @@ class DEBUG_EVENT(DEBUG_EVENT):
def code(self):
return self.KNOWN_EVENT_CODE.get(self.dwDebugEventCode, self.dwDebugEventCode)
WatchedPage = namedtuple('WatchedPage', ["original_prot", "bps"])
class Debugger(object):
"""A debugger based on standard Win32 API. Handle standard (int3) and Hardware-Exec Breakpoints"""
@@ -241,12 +243,15 @@ class Debugger(object):
old_prot = DWORD()
vprot_begin = affected_pages[0]
vprot_size = PAGE_SIZE * len(affected_pages)
print("[VP] {0:#x} {1:#x} {2}".format(vprot_begin, vprot_size, bp.protect))
target.virtual_protect(vprot_begin, vprot_size, bp.protect, old_prot)
bp._old_prot = old_prot.value
#self._virtual_protected_memory[vprot_begin] = (vprot_size, bp.protect, old_prot)
cp_watch_page = self._watched_pages[self.current_process.pid]
for page_addr in affected_pages:
cp_watch_page[page_addr].append(bp)
if page_addr not in cp_watch_page:
cp_watch_page[page_addr] = WatchedPage(old_prot, [])
cp_watch_page[page_addr].bps.append(bp)
# TODO: watch for overlap with other MEM breakpoints
return True
@@ -263,8 +268,8 @@ class Debugger(object):
cp_watch_page = self._watched_pages[self.current_process.pid]
for page_addr in affected_pages:
cp_watch_page[page_addr].remove(bp)
if not cp_watch_page[page_addr]:
cp_watch_page[page_addr].bps.remove(bp)
if not cp_watch_page[page_addr].bps:
del cp_watch_page[page_addr]
else:
raise NotImplementedError("Removing MemBP on page with multiple MemBP <need to reajust page prot")
@@ -339,9 +344,9 @@ class Debugger(object):
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, fault_page):
def _pass_memory_breakpoint(self, bp, page_protect, fault_page):
cp = self.current_process
cp.virtual_protect(fault_page, PAGE_SIZE, bp._old_prot, None)
cp.virtual_protect(fault_page, PAGE_SIZE, page_protect, None)
thread = self.current_thread
ctx = thread.context
ctx.EEFlags.TF = 1
@@ -393,55 +398,50 @@ class Debugger(object):
self._explicit_single_step[self.current_thread.tid] = self.current_thread.context.EEFlags.TF
return continue_flag
# === Testing PAGE_NOACCESS(0x1L) ===
# exception: access violation reading 0x00470000
# exception: access violation writing 0x00470000
# === Testing PAGE_READONLY(0x2L) ===
# exception: access violation writing 0x00470000
# === Testing PAGE_READWRITE(0x4L) ===
# === Testing PAGE_EXECUTE(0x10L) ===
# exception: access violation writing 0x00470000
# === Testing PAGE_EXECUTE_READ(0x20L) ===
# exception: access violation writing 0x00470000
# === Testing PAGE_EXECUTE_READWRITE(0x40L) ===
def _handle_exception_access_violation(self, exception, excp_addr):
READ = 0
WRITE = 1
EXEC = 2
fault_type = exception.ExceptionRecord.ExceptionInformation[0]
#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
#if fault_addr == pc_addr:
# fault_type = EXEC
fault_page = (fault_addr >> 12) << 12
cp_watch_page = self._watched_pages[self.current_process.pid]
mem_bp = self.get_memory_breakpoint_at(fault_addr, self.current_process)
if mem_bp is False: # No BP on this page
return self.on_exception(exception)
original_prot = cp_watch_page[fault_page].original_prot
if mem_bp is None: # Page as MEMBP but None handle this address
# This hack is bad, find a BP on the page to restore original access..
# TODO: stock original page protection elsewhere ?
bp = self._watched_pages[self.current_process.pid][fault_page][0]
self._pass_memory_breakpoint(bp, fault_page)
bp = cp_watch_page[fault_page].bps[-1]
self._pass_memory_breakpoint(bp, original_prot, fault_page)
return DBG_CONTINUE
continue_flag = mem_bp.trigger(self, exception)
self._explicit_single_step[self.current_thread.tid] = self.current_thread.context.EEFlags.TF
# If BP has not been removed in trigger, pas it
if mem_bp in self._watched_pages[self.current_process.pid][fault_page]:
self._pass_memory_breakpoint(mem_bp, fault_page)
if fault_page in cp_watch_page and mem_bp in cp_watch_page[fault_page].bps:
self._pass_memory_breakpoint(mem_bp, original_prot, fault_page)
return continue_flag
#for bp, vprot_begin, vprot_end, original_prot in self._watched_memory:
# if vprot_begin <= fault_addr < vprot_end:
# # It's the page for this MEMBP that triggeed the BP
# if bp._addr <= fault_addr < bp._addr + bp.size:
# # In the real range of our memBP
# continue_flag = bp.trigger(self, exception)
# self._explicit_single_step[self.current_thread.tid] = self.current_thread.context.EEFlags.TF
# #if excp_addr in self.breakpoints[self.current_process.pid]:
# else:
# #self._explicit_single_step[self.current_thread.tid] = self.current_thread.context.EEFlags.TF
# continue_flag = DBG_CONTINUE
#
# if bp in [x[0] for x in self._watched_memory]:
# self._pass_memory_breakpoint(bp, vprot_begin, vprot_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):
@@ -505,7 +505,7 @@ class Debugger(object):
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._watched_pages[self.current_process.pid] = defaultdict(list)
self._watched_pages[self.current_process.pid] = {} #defaultdict(list)
self.breakpoints[self.current_process.pid] = {}
self._module_by_process[self.current_process.pid] = {}
self._update_debugger_state(debug_event)
@@ -666,7 +666,7 @@ class Debugger(object):
if fault_page not in self._watched_pages[process.pid]:
return False
for bp in self._watched_pages[process.pid][fault_page]:
for bp in self._watched_pages[process.pid][fault_page].bps:
if bp._addr <= addr < bp._addr + bp.size:
return bp
return None
windows/test/test_debugger.py
+49
View File
@@ -614,6 +614,55 @@ class DebuggerTestCase(unittest.TestCase):
d.loop()
TEST_CASE.assertEqual(data, [data_addr, data_addr + 4])
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
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, prot=PAGE_READONLY, size=0x500)
the_read_bp = MemBP(data_addr, prot=PAGE_NOACCESS, size=0x500)
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)
if __name__ == '__main__':
alltests = unittest.TestSuite()
alltests.addTest(unittest.makeSuite(DebuggerTestCase))