mirror of
https://github.com/hakril/PythonForWindows
synced 2026-06-08 14:31:45 +00:00
Compare commits
9 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 4b71ebb174 | |||
| 35e7a8da4a | |||
| c46d76a846 | |||
| bef410a1c3 | |||
| 426d28ded9 | |||
| 5f5ad2531b | |||
| 9aaf2c6f6c | |||
| 8d5bb5c032 | |||
| e4ca049004 |
@@ -26,7 +26,7 @@ jobs:
|
||||
strategy:
|
||||
fail-fast: false
|
||||
matrix:
|
||||
runs-on: [windows-2019, windows-latest]
|
||||
runs-on: [windows-2022, windows-latest]
|
||||
python-version: [2.7, 3.6, 3.11]
|
||||
python-architecture: [x86, x64]
|
||||
include:
|
||||
|
||||
+51
-1
@@ -367,6 +367,54 @@ def test_memory_breakpoint_exec(proc32_64_debug):
|
||||
for i in range(NB_NOP_IN_PAGE + 1):
|
||||
assert TSTBP.DATA[i] == addr + i
|
||||
|
||||
def test_memory_breakpoint_trigger_multipage(proc32_64_debug):
|
||||
"""Check that a memory breakpoint triggering on multiple page on the same instruction restore are correctly restored on all pages"""
|
||||
|
||||
class MultiPageMemBP(windows.debug.MemoryBreakpoint):
|
||||
ALL_TRIGGER_ADDR = []
|
||||
|
||||
def trigger(self, dbg, exc):
|
||||
pc_fault_addr = dbg.current_thread.context.pc
|
||||
self.ALL_TRIGGER_ADDR.append(pc_fault_addr)
|
||||
print(hex(pc_fault_addr))
|
||||
# Stop when we have a breakpoint in the 2nd page of the alloc
|
||||
if shellcodeaddr + 0x1000 <= pc_fault_addr <= shellcodeaddr + 0x2000:
|
||||
dbg.current_process.exit()
|
||||
|
||||
# Trigger a write that will write on both page at once,
|
||||
# Triggering both pages mem-bp on the same instruction.
|
||||
# Then call an instruction at the end of page to see if both page of mem-bp still trigger the BP
|
||||
|
||||
shellcodeaddr = proc32_64_debug.virtual_alloc(0x2000)
|
||||
|
||||
if proc32_64_debug.bitness == 64:
|
||||
shellcode = x64.MultipleInstr()
|
||||
shellcode += x64.Mov("RAX", shellcodeaddr + 0xffe)
|
||||
shellcode += x64.Mov("RCX", 0xc3909090) # Nopnopnopret
|
||||
shellcode += x64.Mov(x64.mem("[RAX]"), "ECX") # Will write on both page at once
|
||||
shellcode += x64.Push("RAX")
|
||||
shellcode += x64.Ret() # Jump on the nop + ret
|
||||
else:
|
||||
shellcode = x86.MultipleInstr()
|
||||
shellcode += x86.Mov("EAX", shellcodeaddr + 0xffe)
|
||||
shellcode += x86.Mov("ECX", 0xc3909090) # Nopnopnopret
|
||||
shellcode += x86.Mov(x86.mem("[EAX]"), "ECX") # Will write on both page at once
|
||||
shellcode += x86.Push("EAX")
|
||||
shellcode += x86.Ret() # Jump on the nop + ret
|
||||
|
||||
d = windows.debug.Debugger(proc32_64_debug)
|
||||
bp = MultiPageMemBP(addr=shellcodeaddr, size=0x2000, events="XW")
|
||||
proc32_64_debug.write_memory(shellcodeaddr, shellcode.get_code())
|
||||
d.add_bp(bp)
|
||||
proc32_64_debug.create_thread(shellcodeaddr, 0)
|
||||
|
||||
d.loop()
|
||||
# Check that the 2 nop at the end of the first page of the membp trigger
|
||||
# If access right where not correctly restored: this would not trigger on the first page
|
||||
assert shellcodeaddr + 0xffe in bp.ALL_TRIGGER_ADDR
|
||||
assert shellcodeaddr + 0xfff in bp.ALL_TRIGGER_ADDR
|
||||
assert shellcodeaddr + 0x1000 in bp.ALL_TRIGGER_ADDR
|
||||
|
||||
|
||||
# breakpoint remove
|
||||
import threading
|
||||
@@ -700,4 +748,6 @@ def test_keyboardinterrupt_when_bp_event(proc32_64_debug, monkeypatch):
|
||||
assert proc32_64_debug.read_memory(addr, len(TEST_CODE)) == TEST_CODE
|
||||
assert bad_thread.context.pc == addr
|
||||
else:
|
||||
raise ValueError("Should have raised")
|
||||
raise ValueError("Should have raised")
|
||||
|
||||
|
||||
|
||||
@@ -236,6 +236,10 @@ def test_assembler():
|
||||
CheckInstr(Jmp)(mem('[EAX]'))
|
||||
CheckInstr(Jmp)(mem('[EAX + 2]'))
|
||||
CheckInstr(Jmp)(mem('[0x12345678]'))
|
||||
# Ljmp testing
|
||||
CheckInstr(Ljmp)(0x33, 0x12345678)
|
||||
CheckInstr(Ljmp, expected_result="ljmp 0x23:0x11223344")("0x23:0x11223344")
|
||||
assert Ljmp(0x33, 0x12345678).get_code() == Ljmp("0x33:0x12345678").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()
|
||||
|
||||
@@ -55,6 +55,7 @@ class MemoryBreakpoint(Breakpoint):
|
||||
self.size = size if size is not None else self.DEFAULT_SIZE
|
||||
events = events if events is not None else self.DEFAULT_EVENTS
|
||||
self.events = set(events)
|
||||
self._reput_pages = [] # The current memory BP page that is passed
|
||||
|
||||
def trigger(self, dbg, exception):
|
||||
"""Called when breakpoint is hit"""
|
||||
@@ -126,14 +127,16 @@ class FunctionParamDumpBPAbstract(object):
|
||||
def extract_arguments_64bits(self, cproc, cthread):
|
||||
x = windows.debug.X64ArgumentRetriever()
|
||||
res = OrderedDict()
|
||||
for i, (name, type) in enumerate(zip(self.target_params, self.target_args)):
|
||||
for i, (name, atype) in enumerate(zip(self.target_params, self.target_args)):
|
||||
value = x.get_arg(i, cproc, cthread)
|
||||
rt = windows.remotectypes.transform_type_to_remote64bits(type)
|
||||
rt = windows.remotectypes.transform_type_to_remote64bits(atype)
|
||||
if issubclass(rt, windows.remotectypes.RemoteValue):
|
||||
t = rt(value, cproc)
|
||||
else:
|
||||
t = rt(value)
|
||||
if not hasattr(t, "contents"):
|
||||
|
||||
if (not isinstance(t, (windows.remotectypes.RemotePtr64, windows.remotectypes.RemotePtr32)) or
|
||||
isinstance(t, (ctypes.c_char_p, ctypes.c_wchar_p))):
|
||||
try:
|
||||
t = t.value
|
||||
except AttributeError:
|
||||
|
||||
@@ -422,8 +422,9 @@ class Debugger(object):
|
||||
return True
|
||||
|
||||
def _restore_breakpoint_MEMBP(self, bp, target):
|
||||
(page_addr, page_prot) = bp._reput_page
|
||||
return target.virtual_protect(page_addr, PAGE_SIZE, page_prot, None)
|
||||
for (page_addr, page_prot) in bp._reput_pages:
|
||||
target.virtual_protect(page_addr, PAGE_SIZE, page_prot, None)
|
||||
del bp._reput_pages[:]
|
||||
|
||||
|
||||
def _remove_breakpoint_MEMBP(self, bp, target):
|
||||
@@ -542,7 +543,7 @@ class Debugger(object):
|
||||
ctx = thread.context
|
||||
ctx.EEFlags.TF = 1
|
||||
thread.set_context(ctx)
|
||||
bp._reput_page = (fault_page, page_prot.value)
|
||||
bp._reput_pages.append((fault_page, page_prot.value))
|
||||
self._breakpoint_to_reput[cp.pid].add(bp)
|
||||
|
||||
# debug event handlers
|
||||
@@ -665,6 +666,7 @@ class Debugger(object):
|
||||
fault_type = exception.ExceptionRecord.ExceptionInformation[0]
|
||||
fault_addr = exception.ExceptionRecord.ExceptionInformation[1]
|
||||
pc_addr = self.current_thread.context.pc
|
||||
dbgprint("Handling access_violation at pc={0:#x} addr={1:#x}".format(pc_addr, fault_addr), "DBG")
|
||||
if fault_addr == pc_addr:
|
||||
fault_type = EXEC
|
||||
event = EVENT_STR[fault_type]
|
||||
@@ -683,11 +685,24 @@ class Debugger(object):
|
||||
self._pass_memory_breakpoint(bp, original_prot, fault_page)
|
||||
return DBG_CONTINUE
|
||||
|
||||
# We may have setup the "EEFlags.TF" ourself if the membreakpoint triggered twice on the same instruction
|
||||
# Ex: write on two pages handled by our breakpoint (unaligned write on 0xfff-0x1000)
|
||||
|
||||
originalctx = self.current_thread.context
|
||||
original_tf = originalctx.EEFlags.TF
|
||||
# Temporary disable EEFlags.TF to see if user callback explicit ask for it
|
||||
originalctx.EEFlags.TF = 0
|
||||
self.current_thread.set_context(originalctx)
|
||||
|
||||
with self.DisabledMemoryBreakpoint():
|
||||
continue_flag = mem_bp.trigger(self, exception)
|
||||
if self._killed_in_action():
|
||||
return continue_flag
|
||||
self._explicit_single_step[self.current_thread.tid] = self.current_thread.context.EEFlags.TF
|
||||
# Update explicit trigger based on new value of EEFlags.TF
|
||||
self._explicit_single_step[self.current_thread.tid] |= self.current_thread.context.EEFlags.TF
|
||||
# Reupdate the real EEFlags.TF based on its current value and the original one
|
||||
if original_tf != 0 and not self.current_thread.context.EEFlags.TF:
|
||||
self.single_step()
|
||||
if self._explicit_single_step[self.current_thread.tid]:
|
||||
dbgprint("Someone ask for an explicit Single step - 5", "DBG")
|
||||
# If BP has not been removed in trigger, pas it
|
||||
|
||||
@@ -400,6 +400,16 @@ class Imm32(object):
|
||||
|
||||
class SegmentSelectorAbsoluteAddr(object):
|
||||
def accept_arg(self, args, instr_state):
|
||||
# Special case ptr 16:32
|
||||
if isinstance(args[0], str) and args[0].count(":") == 1:
|
||||
imm16, imm32 = [int(x, 0) for x in args[0].split(":")]
|
||||
sizess, datass = UImm16().accept_arg([imm16], instr_state)
|
||||
sizeabs, dataabs = Imm32().accept_arg([imm32], instr_state)
|
||||
if sizess is None or sizeabs is None:
|
||||
return None, None
|
||||
# We only consumed 1 args as it was the same string
|
||||
return (1, dataabs + datass)
|
||||
|
||||
sizess, datass = UImm16().accept_arg(args, instr_state)
|
||||
if sizess is None:
|
||||
return None, None
|
||||
@@ -711,6 +721,10 @@ class Jmp(JmpType):
|
||||
(RawBits.from_int(8, 0xff), Slash(4)),
|
||||
(RawBits.from_int(8, 0xea), SegmentSelectorAbsoluteAddr())]
|
||||
|
||||
# Allow a second mnemonic for the longjump
|
||||
class Ljmp(JmpType):
|
||||
encoding = [(RawBits.from_int(8, 0xea), SegmentSelectorAbsoluteAddr())]
|
||||
|
||||
|
||||
class Jz(JmpType):
|
||||
encoding = [(RawBits.from_int(8, 0x74), JmpImm8(2)),
|
||||
|
||||
Reference in New Issue
Block a user