from collections import defaultdict
from contextlib import contextmanager
import windows
import windows.winobject.exception as winexception
from windows import winproxy
from windows.generated_def import windef
from windows.generated_def.winstructs import *
from .breakpoints import *
[docs]class LocalDebugger(object):
"""A debugger interface around :func:`AddVectoredExceptionHandler`.
Handle:
* Standard BP (int3)
* Hardware-Exec BP (DrX)"""
def __init__(self):
self.breakpoints = {}
self._memory_save = {}
self._reput_breakpoint = {}
self._hxbp_breakpoint = defaultdict(dict)
self.callback_vectored = winexception.VectoredException(self.callback)
winproxy.AddVectoredExceptionHandler(0, self.callback_vectored)
self.setup_hxbp_callback_vectored = winexception.VectoredException(self.setup_hxbp_callback)
self.hxbp_info = None
self.code = windows.native_exec.create_function("\xcc\xc3", [PVOID])
self.veh_depth = 0
self.current_exception = None
self.exceptions_stack = [None]
@contextmanager
def NewCurrentException(self, exc):
try:
self.exceptions_stack.append(exc)
self.current_exception = exc
self.veh_depth += 1
yield exc
finally:
self.exceptions_stack.pop()
self.current_exception = self.exceptions_stack[-1]
self.veh_depth -= 1
[docs] def get_exception_code(self):
"""Return ExceptionCode of current exception"""
return self.current_exception[0].ExceptionRecord[0].ExceptionCode
[docs] def get_exception_context(self):
"""Return context of current exception"""
return self.current_exception[0].ContextRecord[0]
[docs] def single_step(self):
"""Make the current thread to single step"""
self.get_exception_context().EEFlags.TF = 1
return windef.EXCEPTION_CONTINUE_EXECUTION
def _pass_breakpoint(self, addr, single_step):
with windows.utils.VirtualProtected(addr, 1, PAGE_EXECUTE_READWRITE):
windows.current_process.write_memory(addr, self._memory_save[addr])
self._reput_breakpoint[windows.current_thread.tid] = self.breakpoints[addr], single_step
return self.single_step()
def callback(self, exc):
with self.NewCurrentException(exc):
return self.handle_exception(exc)
def handle_exception(self, exc):
exp_code = self.get_exception_code()
context = self.get_exception_context()
exp_addr = context.pc
if exp_code == EXCEPTION_BREAKPOINT and exp_addr in self.breakpoints:
res = self.breakpoints[exp_addr].trigger(self, exc)
single_step = self.get_exception_context().EEFlags.TF # single step activated by breakpoint
if exp_addr in self.breakpoints: # Breakpoint deleted itself ?
return self._pass_breakpoint(exp_addr, single_step)
return EXCEPTION_CONTINUE_EXECUTION
if exp_code == EXCEPTION_SINGLE_STEP and windows.current_thread.tid in self._reput_breakpoint:
bp, single_step = self._reput_breakpoint[windows.current_thread.tid]
self._memory_save[bp.addr] = windows.current_process.read_memory(bp.addr, 1)
with windows.utils.VirtualProtected(bp.addr, 1, PAGE_EXECUTE_READWRITE):
windows.current_process.write_memory(bp.addr, "\xcc")
del self._reput_breakpoint[windows.current_thread.tid]
if single_step:
return self.on_exception(exc)
return windef.EXCEPTION_CONTINUE_EXECUTION
elif exp_code == EXCEPTION_SINGLE_STEP and exp_addr in self._hxbp_breakpoint[windows.current_thread.tid]:
res = self._hxbp_breakpoint[windows.current_thread.tid][exp_addr].trigger(self, exc)
context.EEFlags.RF = 1
return EXCEPTION_CONTINUE_EXECUTION
return self.on_exception(exc)
[docs] def on_exception(self, exc):
"""Called on exception"""
if not self.get_exception_code() in winexception.exception_name_by_value:
return windef.EXCEPTION_CONTINUE_SEARCH
return windef.EXCEPTION_CONTINUE_EXECUTION
[docs] def del_bp(self, bp):
"""Delete a breakpoint"""
if bp.type == STANDARD_BP:
with windows.utils.VirtualProtected(bp.addr, 1, PAGE_EXECUTE_READWRITE):
windows.current_process.write_memory(bp.addr, self._memory_save[bp.addr])
del self._memory_save[bp.addr]
del self.breakpoints[bp.addr]
return
if bp.type == HARDWARE_EXEC_BP:
threads_by_tid = {t.tid: t for t in windows.current_process.threads}
for tid in self._hxbp_breakpoint:
if bp.addr in self._hxbp_breakpoint[tid] and self._hxbp_breakpoint[tid][bp.addr] == bp:
if tid == windows.current_thread.tid:
self.remove_hxbp_self_thread(bp.addr)
else:
self.remove_hxbp_other_thread(bp.addr, threads_by_tid[tid])
del self._hxbp_breakpoint[tid][bp.addr]
return
raise NotImplementedError("Unknow BP type {0}".format(bp.type))
[docs] def add_bp(self, bp, targets=None):
"""Add a breakpoint, bp is a "class:`Breakpoint`
If the ``bp`` type is ``STANDARD_BP``, target must be None.
If the ``bp`` type is ``HARDWARE_EXEC_BP``, target can be None (all threads), or some threads of the process
"""
if bp.type == HARDWARE_EXEC_BP:
return self.add_bp_hxbp(bp, targets)
if bp.type != STANDARD_BP:
raise NotImplementedError("Unknow BP type {0}".format(bp.type))
if targets is not None:
raise ValueError("LocalDebugger: STANDARD_BP doest not support targets {0}".format(targets))
self.breakpoints[bp.addr] = bp
self._memory_save[bp.addr] = windows.current_process.read_memory(bp.addr, 1)
with windows.utils.VirtualProtected(bp.addr, 1, PAGE_EXECUTE_READWRITE):
windows.current_process.write_memory(bp.addr, "\xcc")
return
def add_bp_hxbp(self, bp, targets=None):
if bp.type != HARDWARE_EXEC_BP:
raise NotImplementedError("Add non standard-BP in LocalDebugger")
if targets is None:
targets = windows.current_process.threads
for thread in targets:
if thread.owner.pid != windows.current_process.pid:
raise ValueError("Cannot add HXBP to target in remote process {0}".format(thread))
if thread.tid == windows.current_thread.tid:
self.setup_hxbp_self_thread(bp.addr)
else:
self.setup_hxbp_other_thread(bp.addr, thread)
self._hxbp_breakpoint[thread.tid][bp.addr] = bp
def setup_hxbp_callback(self, exc):
with self.NewCurrentException(exc):
exp_code = self.get_exception_code()
context = self.get_exception_context()
exp_addr = context.pc
hxbp_used = self.setup_hxbp_in_context(context, self.data)
windows.current_process.write_memory(exp_addr, "\x90")
# Raising in the VEH is a bad idea..
# So better give the information to triggerer..
if hxbp_used is not None:
self.get_exception_context().Eax = exp_addr
else:
self.get_exception_context().Eax = 0
return windef.EXCEPTION_CONTINUE_EXECUTION
def remove_hxbp_callback(self, exc):
with self.NewCurrentException(exc):
exp_code = self.get_exception_code()
context = self.get_exception_context()
exp_addr = context.pc
hxbp_used = self.remove_hxbp_in_context(context, self.data)
windows.current_process.write_memory(exp_addr, "\x90")
# Raising in the VEH is a bad idea..
# So better give the information to triggerer..
if hxbp_used is not None:
self.get_exception_context().Eax = exp_addr
else:
self.get_exception_context().Eax = 0
return windef.EXCEPTION_CONTINUE_EXECUTION
def setup_hxbp_in_context(self, context, addr):
for i in range(4):
is_used = getattr(context.EDr7, "L" + str(i))
empty_drx = str(i)
if not is_used:
context.EDr7.GE = 1
context.EDr7.LE = 1
setattr(context.EDr7, "L" + empty_drx, 1)
setattr(context, "Dr" + empty_drx, addr)
return i
return None
def remove_hxbp_in_context(self, context, addr):
for i in range(4):
target_drx = str(i)
is_used = getattr(context.EDr7, "L" + str(i))
draddr = getattr(context, "Dr" + target_drx)
if is_used and draddr == addr:
setattr(context.EDr7, "L" + target_drx, 0)
setattr(context, "Dr" + target_drx, 0)
return i
return None
def setup_hxbp_self_thread(self, addr):
if self.current_exception is not None:
x = self.setup_hxbp_in_context(self.get_exception_context(), addr)
if x is None:
raise ValueError("Could not setup HXBP")
return
self.data = addr
with winexception.VectoredExceptionHandler(1, self.setup_hxbp_callback):
x = self.code()
if x is None:
raise ValueError("Could not setup HXBP")
windows.current_process.write_memory(x, "\xcc")
return
def setup_hxbp_other_thread(self, addr, thread):
thread.suspend()
ctx = thread.context
x = self.setup_hxbp_in_context(ctx, addr)
if x is None:
raise ValueError("Could not setup HXBP in {0}".format(thread))
thread.set_context(ctx)
thread.resume()
def remove_hxbp_self_thread(self, addr):
if self.current_exception is not None:
x = self.remove_hxbp_in_context(self.get_exception_context(), addr)
if x is None:
raise ValueError("Could not setup HXBP")
return
self.data = addr
with winexception.VectoredExceptionHandler(1, self.remove_hxbp_callback):
x = self.code()
if x is None:
raise ValueError("Could not remove HXBP")
windows.current_process.write_memory(x, "\xcc")
return
def remove_hxbp_other_thread(self, addr, thread):
thread.suspend()
ctx = thread.context
x = self.remove_hxbp_in_context(ctx, addr)
if x is None:
raise ValueError("Could not setup HXBP in {0}".format(thread))
thread.set_context(ctx)
thread.resume()