|
# ReactOS GDB debug helper for QEMU |
|
# Usage: gdb -x .gdbinit |
|
# Then: ros-connect (connects to :1234, loads symbols) |
|
# Or: ros-connect :5678 (custom port) |
|
# |
|
# Non-aggressive: does NOT auto-connect or auto-load. |
|
# All commands are registered immediately and work once connected. |
|
|
|
set architecture i386:x86-64 |
|
set disassembly-flavor intel |
|
set pagination off |
|
set confirm off |
|
|
|
python |
|
import gdb |
|
import struct |
|
import os |
|
import re as _re |
|
|
|
KSEG0_BASE = 0xFFFFF80000000000 |
|
|
|
def _detect_build_dir(): |
|
env = os.environ.get('ROS_BUILD_DIR') |
|
if env and os.path.isdir(env): |
|
return env |
|
cwd = os.getcwd() |
|
# Try cwd itself first (gdb may already be inside the build dir) |
|
if os.path.isfile(os.path.join(cwd, 'ntoskrnl', 'ntoskrnl.exe')): |
|
return cwd |
|
for cand in ('build_nt62', 'build', 'build_nt61', 'build_nt6', 'build_nt52'): |
|
p = os.path.join(cwd, cand) |
|
if os.path.isfile(os.path.join(p, 'ntoskrnl', 'ntoskrnl.exe')): |
|
return p |
|
# Last resort: walk up looking for build_nt62/ntoskrnl |
|
parent = os.path.dirname(cwd) |
|
while parent and parent != '/': |
|
for cand in ('build_nt62', 'build', 'build_nt61', 'build_nt6'): |
|
p = os.path.join(parent, cand) |
|
if os.path.isdir(os.path.join(p, 'ntoskrnl')): |
|
return p |
|
parent = os.path.dirname(parent) |
|
return os.path.join(cwd, 'build') |
|
|
|
BUILD_DIR = _detect_build_dir() |
|
|
|
# ============================================================ |
|
# Memory read helpers |
|
# ============================================================ |
|
|
|
def read_mem(addr, size): |
|
try: |
|
inferior = gdb.selected_inferior() |
|
return bytes(inferior.read_memory(addr, size)) |
|
except: |
|
return None |
|
|
|
def read_u8(addr): |
|
data = read_mem(addr, 1) |
|
return data[0] if data else None |
|
|
|
def read_u16(addr): |
|
data = read_mem(addr, 2) |
|
return struct.unpack('<H', data)[0] if data else None |
|
|
|
def read_i16(addr): |
|
data = read_mem(addr, 2) |
|
return struct.unpack('<h', data)[0] if data else None |
|
|
|
def read_u32(addr): |
|
data = read_mem(addr, 4) |
|
return struct.unpack('<I', data)[0] if data else None |
|
|
|
def read_u64(addr): |
|
data = read_mem(addr, 8) |
|
return struct.unpack('<Q', data)[0] if data else None |
|
|
|
def read_unicode_string(addr): |
|
length = read_u16(addr) |
|
buf_ptr = read_u64(addr + 8) |
|
if length is None or buf_ptr is None or length == 0 or length > 512: |
|
return None |
|
data = read_mem(buf_ptr, length) |
|
if data is None: |
|
return None |
|
try: |
|
return data.decode('utf-16-le') |
|
except: |
|
return None |
|
|
|
def fmt_ptr(value): |
|
return f"0x{value:016x}" if value is not None else "?" |
|
|
|
def read_ascii_string(addr, max_len=16): |
|
data = read_mem(addr, max_len) |
|
if not data: |
|
return None |
|
data = data.split(b"\x00", 1)[0] |
|
try: |
|
return data.decode("ascii", errors="replace") |
|
except: |
|
return None |
|
|
|
# ============================================================ |
|
# PE helpers |
|
# ============================================================ |
|
|
|
def find_pe_base_from_rip(): |
|
try: |
|
rip = int(gdb.parse_and_eval("$rip")) |
|
except: |
|
return None |
|
if rip < KSEG0_BASE: |
|
return None |
|
page = rip & ~0xFFF |
|
for _ in range(4096): |
|
sig = read_u16(page) |
|
if sig == 0x5A4D: |
|
pe_off = read_u32(page + 0x3C) |
|
if pe_off and pe_off < 0x1000: |
|
pe_sig = read_u32(page + pe_off) |
|
if pe_sig == 0x00004550: |
|
return page |
|
page -= 0x1000 |
|
return None |
|
|
|
def get_pe_all_sections(base): |
|
pe_off = read_u32(base + 0x3C) |
|
if not pe_off: |
|
return [] |
|
num_sections = read_u16(base + pe_off + 6) |
|
size_opt = read_u16(base + pe_off + 20) |
|
if num_sections is None or size_opt is None: |
|
return [] |
|
sec_start = base + pe_off + 24 + size_opt |
|
sections = [] |
|
for i in range(num_sections): |
|
sec = sec_start + i * 40 |
|
sec_name = read_mem(sec, 8) |
|
if sec_name is None: |
|
continue |
|
sec_name = sec_name.rstrip(b'\x00').decode('ascii', errors='replace') |
|
rva = read_u32(sec + 12) |
|
if rva: |
|
sections.append((sec_name, rva)) |
|
return sections |
|
|
|
def load_pe_symbols(base, name, filepath): |
|
sections = get_pe_all_sections(base) |
|
text_rva = None |
|
extra_sections = [] |
|
for sname, rva in sections: |
|
if sname == '.text': |
|
text_rva = rva |
|
elif not sname.startswith('.debug'): |
|
extra_sections.append((sname, rva)) |
|
if text_rva is None: |
|
text_rva = 0x1000 |
|
text_addr = base + text_rva |
|
section_args = ' '.join(f'-s {sname} 0x{base + rva:x}' for sname, rva in extra_sections) |
|
cmd = f'add-symbol-file {filepath} 0x{text_addr:x} {section_args}' |
|
try: |
|
gdb.execute(cmd, to_string=True) |
|
print(f" Loaded {name} @ 0x{base:x} (.text=0x{text_addr:x})") |
|
return True |
|
except Exception as e: |
|
print(f" Failed to load {name}: {e}") |
|
return False |
|
|
|
def addr_to_sym(addr): |
|
try: |
|
sym = gdb.execute(f"info symbol 0x{addr:x}", to_string=True).strip() |
|
if "No symbol" not in sym: |
|
parts = sym.split(" in section ") |
|
return parts[0] |
|
except: |
|
pass |
|
return None |
|
|
|
# ============================================================ |
|
# CPU mode detection and context helpers |
|
# ============================================================ |
|
|
|
def detect_cpu_context(): |
|
"""Detect current execution context. |
|
Returns dict with 'mode' ('kernel', 'user64', 'compat32', 'real', 'unknown'), |
|
'cs', 'rip', 'rsp', and 'can_read_kernel' flag.""" |
|
try: |
|
cs = int(gdb.parse_and_eval("$cs")) |
|
rip = int(gdb.parse_and_eval("$rip")) |
|
rsp = int(gdb.parse_and_eval("$rsp")) |
|
except: |
|
return {'mode': 'unknown', 'cs': 0, 'rip': 0, 'rsp': 0, 'can_read_kernel': False} |
|
|
|
if cs == 0x10: |
|
return {'mode': 'kernel', 'cs': cs, 'rip': rip, 'rsp': rsp, 'can_read_kernel': True} |
|
elif cs == 0x33: |
|
return {'mode': 'user64', 'cs': cs, 'rip': rip, 'rsp': rsp, 'can_read_kernel': True} |
|
elif cs == 0x23: |
|
return {'mode': 'compat32', 'cs': cs, 'rip': rip, 'rsp': rsp, 'can_read_kernel': True} |
|
else: |
|
# Check if real mode |
|
try: |
|
cr0 = int(gdb.parse_and_eval("$cr0")) |
|
if not (cr0 & 1): |
|
return {'mode': 'real', 'cs': cs, 'rip': rip, 'rsp': rsp, 'can_read_kernel': False} |
|
except: |
|
pass |
|
return {'mode': 'unknown', 'cs': cs, 'rip': rip, 'rsp': rsp, 'can_read_kernel': False} |
|
|
|
|
|
class ReactosWhere(gdb.Command): |
|
"""Show current CPU context: kernel, user64, compat32, or real mode. |
|
If in user mode, shows how to access kernel structures. |
|
Usage: ros-where""" |
|
|
|
def __init__(self): |
|
super().__init__("ros-where", gdb.COMMAND_USER) |
|
|
|
def invoke(self, arg, from_tty): |
|
ctx = detect_cpu_context() |
|
mode_desc = { |
|
'kernel': 'Kernel mode (ring 0)', |
|
'user64': 'User mode 64-bit', |
|
'compat32': 'User mode 32-bit compat (WoW64!)', |
|
'real': 'Real mode (bootloader)', |
|
'unknown': 'Unknown', |
|
} |
|
print(f"Mode: {mode_desc.get(ctx['mode'], ctx['mode'])}") |
|
print(f"CS: 0x{ctx['cs']:x}") |
|
print(f"RIP: 0x{ctx['rip']:x}") |
|
print(f"RSP: 0x{ctx['rsp']:x}") |
|
|
|
if ctx['mode'] == 'compat32': |
|
print(f"\n ** 32-bit compat mode detected (WoW64 code running)") |
|
print(f" ** EIP=0x{ctx['rip'] & 0xFFFFFFFF:x} ESP=0x{ctx['rsp'] & 0xFFFFFFFF:x}") |
|
# Try to read GS base (KPCR) — works even in user mode via QEMU gdbstub |
|
try: |
|
gsbase = int(gdb.parse_and_eval("$gs_base")) |
|
print(f" ** GS base (KPCR) = 0x{gsbase:x}") |
|
except: |
|
pass |
|
print(f" ** Kernel structures are still accessible via QEMU gdbstub") |
|
print(f" ** Use ros-load-symbols, ros-lsmod etc. normally") |
|
|
|
elif ctx['mode'] == 'user64': |
|
print(f"\n ** 64-bit user mode") |
|
# Try to identify the module |
|
sym = addr_to_sym(ctx['rip']) |
|
if sym: |
|
print(f" ** RIP is in: {sym}") |
|
print(f" ** Kernel structures are accessible via QEMU gdbstub") |
|
|
|
elif ctx['mode'] == 'kernel': |
|
sym = addr_to_sym(ctx['rip']) |
|
if sym: |
|
print(f" ** RIP is in: {sym}") |
|
|
|
# Check if we can read kernel memory (QEMU gdbstub gives full access) |
|
if ctx['can_read_kernel']: |
|
test = read_u16(KSEG0_BASE + 0x400000) |
|
if test == 0x5A4D: |
|
print(f"\n Kernel memory accessible (ntoskrnl MZ header verified)") |
|
else: |
|
print(f"\n WARNING: Cannot read ntoskrnl at expected location") |
|
|
|
ReactosWhere() |
|
|
|
|
|
# ============================================================ |
|
# KTHREAD / KTRAP_FRAME offsets (ReactOS amd64) |
|
# ============================================================ |
|
|
|
KTHREAD_INITIAL_STACK = 0x28 |
|
KTHREAD_KERNEL_STACK = 0x58 |
|
KTHREAD_TRAP_FRAME = 0x90 |
|
KTHREAD_APC_STATE = 0x98 |
|
KTHREAD_WAIT_BLOCK_LIST = 0x0D0 |
|
KTHREAD_STATE = 0x184 |
|
KTHREAD_KERNEL_APC_DISABLE = 0x1E4 |
|
KTHREAD_SPECIAL_APC_DISABLE = 0x1E6 |
|
KTHREAD_WAIT_REASON = 0x283 |
|
|
|
EPROCESS_UNIQUE_PROCESS_ID = 0x0F8 |
|
EPROCESS_ACTIVE_PROCESS_LINKS = 0x100 |
|
EPROCESS_IMAGE_FILE_NAME = 0x260 |
|
EPROCESS_THREAD_LIST_HEAD = 0x288 |
|
|
|
ETHREAD_CID = 0x398 |
|
ETHREAD_THREAD_LIST_ENTRY = 0x408 |
|
|
|
KAPC_STATE_KERNEL_LIST = 0x00 |
|
KAPC_STATE_USER_LIST = 0x10 |
|
KAPC_STATE_PROCESS = 0x20 |
|
KAPC_STATE_IN_PROGRESS = 0x28 |
|
KAPC_STATE_KAPC_PENDING = 0x29 |
|
KAPC_STATE_UAPC_PENDING = 0x2A |
|
|
|
KAPC_APC_LIST_ENTRY = 0x10 |
|
KAPC_KERNEL_ROUTINE = 0x20 |
|
KAPC_NORMAL_ROUTINE = 0x30 |
|
|
|
KWAIT_BLOCK_THREAD = 0x18 |
|
KWAIT_BLOCK_OBJECT = 0x20 |
|
|
|
TF_RAX = 0x030 |
|
TF_RCX = 0x038 |
|
TF_RDX = 0x040 |
|
TF_R8 = 0x048 |
|
TF_R9 = 0x050 |
|
TF_R10 = 0x058 |
|
TF_R11 = 0x060 |
|
TF_GSBASE = 0x068 |
|
TF_FAULT_ADDR = 0x0D0 |
|
TF_DR0 = 0x0D8 |
|
TF_SEG_DS = 0x130 |
|
TF_SEG_ES = 0x132 |
|
TF_SEG_FS = 0x134 |
|
TF_SEG_GS = 0x136 |
|
TF_TRAP_LINK = 0x138 |
|
TF_RBX = 0x140 |
|
TF_RDI = 0x148 |
|
TF_RSI = 0x150 |
|
TF_RBP = 0x158 |
|
TF_ERROR_CODE = 0x160 |
|
TF_RIP = 0x168 |
|
TF_SEG_CS = 0x170 |
|
TF_EFLAGS = 0x178 |
|
TF_RSP = 0x180 |
|
TF_SEG_SS = 0x188 |
|
TF_SIZE = 0x190 |
|
TF_PREV_IRQL = 0x029 |
|
TF_PREV_MODE = 0x028 |
|
|
|
POOL_BLOCK_SIZE = 16 |
|
|
|
# ============================================================ |
|
# ros-connect: connect to QEMU and optionally load symbols |
|
# ============================================================ |
|
|
|
class ReactosConnect(gdb.Command): |
|
"""Connect to QEMU gdbstub and load ntoskrnl symbols. |
|
Usage: ros-connect [host:port] |
|
Default: ros-connect :1234""" |
|
|
|
def __init__(self): |
|
super().__init__("ros-connect", gdb.COMMAND_USER) |
|
|
|
def invoke(self, arg, from_tty): |
|
target = arg.strip() if arg.strip() else ":1234" |
|
try: |
|
gdb.execute(f"target remote {target}") |
|
except Exception as e: |
|
print(f"Connection failed: {e}") |
|
return |
|
|
|
# Detect mode |
|
try: |
|
cr0 = int(gdb.parse_and_eval("$cr0")) |
|
efer = int(gdb.parse_and_eval("$efer")) |
|
pe = cr0 & 1 |
|
lma = efer & 0x400 |
|
if not pe: |
|
print("CPU mode: real (16-bit)") |
|
return |
|
elif lma: |
|
print("CPU mode: long (64-bit)") |
|
else: |
|
print("CPU mode: protected (32-bit)") |
|
return |
|
except: |
|
print("Cannot detect CPU mode") |
|
|
|
# Auto-load ntoskrnl |
|
gdb.execute("ros-load-symbols") |
|
|
|
ReactosConnect() |
|
|
|
|
|
# ============================================================ |
|
# ros-load-symbols |
|
# ============================================================ |
|
|
|
def _pe_export_name(base): |
|
"""Read the PE export DLL name at the given mapped base, or None.""" |
|
try: |
|
pe_off = read_u32(base + 0x3C) |
|
if not pe_off or pe_off >= 0x1000: |
|
return None |
|
if read_u32(base + pe_off) != 0x00004550: |
|
return None |
|
# ImageBase=24 bytes from PE+24, Magic 0x20b for PE32+; assume PE32+ for amd64. |
|
# Data directory 0 (export) is at OptionalHeader+112 on PE32+. |
|
opt_off = base + pe_off + 24 |
|
exp_rva = read_u32(opt_off + 112) |
|
if not exp_rva: |
|
return None |
|
name_rva = read_u32(base + exp_rva + 0xc) |
|
if not name_rva: |
|
return None |
|
# Read up to 64 bytes of the DLL name. |
|
infer = gdb.selected_inferior() |
|
raw = bytes(infer.read_memory(base + name_rva, 64)) |
|
nul = raw.find(b'\x00') |
|
if nul >= 0: |
|
raw = raw[:nul] |
|
return raw.decode('ascii', errors='replace').lower() |
|
except Exception: |
|
return None |
|
|
|
|
|
def _find_kernel_pe_base(): |
|
"""Scan the kernel mapping region for an ntoskrnl.exe/ntkrnlmp.exe PE. |
|
|
|
FreeLdr maps the kernel image somewhere in [KSEG0_BASE, KSEG0_BASE+~4MB); |
|
the actual address depends on the PE size and previous allocations |
|
(e.g. it differs between ntoskrnl.exe and ntkrnlmp.exe). Try the canonical |
|
0x400000 base first, then scan.""" |
|
cands = [KSEG0_BASE + 0x400000] |
|
# Walk down/up in 4 KiB pages, ntkrnlmp is typically below 0x400000. |
|
for off in range(0x100000, 0x800000, 0x1000): |
|
cands.append(KSEG0_BASE + off) |
|
seen = set() |
|
for base in cands: |
|
if base in seen: |
|
continue |
|
seen.add(base) |
|
try: |
|
if read_u16(base) != 0x5A4D: |
|
continue |
|
except Exception: |
|
continue |
|
name = _pe_export_name(base) |
|
if name and (name == 'ntoskrnl.exe' or name == 'ntkrnlmp.exe'): |
|
return base, name |
|
return None, None |
|
|
|
|
|
class ReactosLoadSymbols(gdb.Command): |
|
"""Load ReactOS kernel symbols. Finds the kernel base by scanning.""" |
|
|
|
def __init__(self): |
|
super().__init__("ros-load-symbols", gdb.COMMAND_USER) |
|
|
|
def invoke(self, arg, from_tty): |
|
ntos_path = os.path.join(BUILD_DIR, 'ntoskrnl', 'ntoskrnl.exe') |
|
ntmp_path = os.path.join(BUILD_DIR, 'ntoskrnl', 'ntkrnlmp', 'ntkrnlmp.exe') |
|
base, name = _find_kernel_pe_base() |
|
if base is None: |
|
# Last-ditch fallback: probe near the current RIP. |
|
base = find_pe_base_from_rip() |
|
if base is not None: |
|
name = _pe_export_name(base) or 'ntoskrnl.exe' |
|
if base is None: |
|
print(" Could not find ntoskrnl/ntkrnlmp") |
|
return |
|
if name == 'ntkrnlmp.exe' and os.path.exists(ntmp_path): |
|
path = ntmp_path |
|
label = 'ntkrnlmp.exe' |
|
elif os.path.exists(ntos_path): |
|
path = ntos_path |
|
label = 'ntoskrnl.exe' |
|
else: |
|
print(f"ERROR: neither {ntos_path} nor {ntmp_path} present") |
|
return |
|
print(f" {label} @ 0x{base:x}") |
|
load_pe_symbols(base, label, path) |
|
|
|
ReactosLoadSymbols() |
|
|
|
|
|
class ReactosLoadAt(gdb.Command): |
|
"""Load ntoskrnl symbols at a given base. Usage: ros-load-at <addr>""" |
|
def __init__(self): |
|
super().__init__("ros-load-at", gdb.COMMAND_USER) |
|
def invoke(self, arg, from_tty): |
|
if not arg.strip(): |
|
print("Usage: ros-load-at <base-address>") |
|
return |
|
base = int(arg.strip(), 0) |
|
ntos_path = os.path.join(BUILD_DIR, 'ntoskrnl', 'ntoskrnl.exe') |
|
if not os.path.exists(ntos_path): |
|
print(f"ERROR: {ntos_path} not found") |
|
return |
|
load_pe_symbols(base, "ntoskrnl.exe", ntos_path) |
|
|
|
ReactosLoadAt() |
|
|
|
|
|
class ReactosLoadModule(gdb.Command): |
|
"""Load symbols for a module. Usage: ros-load-module <base> <build-path>""" |
|
def __init__(self): |
|
super().__init__("ros-load-module", gdb.COMMAND_USER) |
|
def invoke(self, arg, from_tty): |
|
parts = arg.strip().split(None, 1) |
|
if len(parts) < 2: |
|
print("Usage: ros-load-module <base-addr> <build/relative/path>") |
|
return |
|
base = int(parts[0], 0) |
|
relpath = parts[1] |
|
filepath = os.path.join(BUILD_DIR, relpath) |
|
if not os.path.exists(filepath): |
|
print(f"ERROR: {filepath} not found") |
|
return |
|
load_pe_symbols(base, os.path.basename(relpath), filepath) |
|
|
|
ReactosLoadModule() |
|
|
|
|
|
# ============================================================ |
|
# ros-lsmod |
|
# ============================================================ |
|
|
|
MODULE_PATHS = { |
|
'ntoskrnl.exe': 'ntoskrnl/ntoskrnl.exe', |
|
'hal.dll': 'hal/halx86/hal.dll', |
|
'kdcom.dll': 'drivers/base/kdcom/kdcom.dll', |
|
'bootvid.dll': 'drivers/base/bootvid/bootvid.dll', |
|
'ntfs.sys': 'drivers/filesystems/ntfs/ntfs.sys', |
|
'win32k.sys': 'win32ss/win32k.sys', |
|
'scsiport.sys': 'drivers/storage/port/scsiport/scsiport.sys', |
|
} |
|
|
|
class ReactosModuleList(gdb.Command): |
|
"""Walk PsLoadedModuleList. Usage: ros-lsmod [--load]""" |
|
def __init__(self): |
|
super().__init__("ros-lsmod", gdb.COMMAND_USER) |
|
def invoke(self, arg, from_tty): |
|
do_load = '--load' in arg |
|
try: |
|
list_head = int(gdb.parse_and_eval("(unsigned long long)&PsLoadedModuleList")) |
|
except: |
|
print("PsLoadedModuleList not found. Load ntoskrnl symbols first.") |
|
return |
|
flink = read_u64(list_head) |
|
if flink is None: |
|
print("Cannot read PsLoadedModuleList") |
|
return |
|
print(f"{'Base':>20s} {'Size':>10s} Name") |
|
print(f"{'----':>20s} {'----':>10s} ----") |
|
entry_addr = flink |
|
count = 0 |
|
while entry_addr != list_head and count < 256: |
|
dll_base = read_u64(entry_addr + 0x30) |
|
size = read_u32(entry_addr + 0x40) |
|
name = read_unicode_string(entry_addr + 0x58) |
|
if dll_base is None: |
|
break |
|
name_str = name if name else "<unknown>" |
|
size_str = f"0x{size:x}" if size else "?" |
|
print(f" 0x{dll_base:016x} {size_str:>10s} {name_str}") |
|
if do_load and name: |
|
lname = name.lower() |
|
if lname in MODULE_PATHS: |
|
fpath = os.path.join(BUILD_DIR, MODULE_PATHS[lname]) |
|
if os.path.exists(fpath): |
|
load_pe_symbols(dll_base, name, fpath) |
|
entry_addr = read_u64(entry_addr) |
|
if entry_addr is None: |
|
break |
|
count += 1 |
|
print(f"\n{count} modules loaded") |
|
|
|
ReactosModuleList() |
|
|
|
|
|
# ============================================================ |
|
# ros-addr2mod |
|
# ============================================================ |
|
|
|
class ReactosFindModuleByAddr(gdb.Command): |
|
"""Find which module an address belongs to. Usage: ros-addr2mod <addr>""" |
|
def __init__(self): |
|
super().__init__("ros-addr2mod", gdb.COMMAND_USER) |
|
def invoke(self, arg, from_tty): |
|
if not arg.strip(): |
|
print("Usage: ros-addr2mod <address>") |
|
return |
|
target = int(arg.strip(), 0) |
|
try: |
|
list_head = int(gdb.parse_and_eval("(unsigned long long)&PsLoadedModuleList")) |
|
except: |
|
print("PsLoadedModuleList not found.") |
|
return |
|
flink = read_u64(list_head) |
|
if flink is None: |
|
return |
|
entry_addr = flink |
|
count = 0 |
|
while entry_addr != list_head and count < 256: |
|
dll_base = read_u64(entry_addr + 0x30) |
|
size = read_u32(entry_addr + 0x40) |
|
name = read_unicode_string(entry_addr + 0x58) |
|
if dll_base and size and dll_base <= target < dll_base + size: |
|
offset = target - dll_base |
|
name_str = name if name else "<unknown>" |
|
print(f"0x{target:x} => {name_str} + 0x{offset:x} (base=0x{dll_base:x})") |
|
return |
|
entry_addr = read_u64(entry_addr) |
|
if entry_addr is None: |
|
break |
|
count += 1 |
|
print(f"0x{target:x} not found in any loaded module") |
|
|
|
ReactosFindModuleByAddr() |
|
|
|
|
|
# ============================================================ |
|
# ros-thread, ros-waiters, ros-irp |
|
# ============================================================ |
|
|
|
def dump_thread_stack(thread_addr): |
|
init_stack = read_u64(thread_addr + KTHREAD_INITIAL_STACK) |
|
kern_stack = read_u64(thread_addr + KTHREAD_KERNEL_STACK) |
|
if not init_stack or not kern_stack: |
|
print(f" Cannot read stack pointers") |
|
return |
|
print(f" KernelStack = 0x{kern_stack:x}") |
|
print(f" InitialStack = 0x{init_stack:x}") |
|
low = min(kern_stack, init_stack) |
|
high = max(kern_stack, init_stack) |
|
scan_len = min(high - low, 0x8000) |
|
print(f" Stack span = {high - low} bytes\n") |
|
for off in range(0, scan_len, 8): |
|
addr = low + off |
|
val = read_u64(addr) |
|
if val is None: |
|
continue |
|
if val >= 0xFFFFF80000000000 and val < 0xFFFFFFFFFFC00000: |
|
sym = addr_to_sym(val) |
|
if sym: |
|
print(f" [0x{addr:x}] 0x{val:x} {sym}") |
|
|
|
class ReactosThreadInfo(gdb.Command): |
|
"""Inspect a kernel thread. Usage: ros-thread <addr>""" |
|
def __init__(self): |
|
super().__init__("ros-thread", gdb.COMMAND_USER) |
|
def invoke(self, arg, from_tty): |
|
if not arg.strip(): |
|
print("Usage: ros-thread <kthread-address>") |
|
return |
|
thread = int(arg.strip(), 0) |
|
print(f"=== KTHREAD 0x{thread:x} ===\n") |
|
init_stack = read_u64(thread + KTHREAD_INITIAL_STACK) |
|
kern_stack = read_u64(thread + KTHREAD_KERNEL_STACK) |
|
print(f"InitialStack = 0x{init_stack:x}" if init_stack else "InitialStack = ?") |
|
print(f"KernelStack = 0x{kern_stack:x}" if kern_stack else "KernelStack = ?") |
|
apc_base = thread + KTHREAD_APC_STATE |
|
process = read_u64(apc_base + KAPC_STATE_PROCESS) |
|
kapc_disable = read_i16(thread + KTHREAD_KERNEL_APC_DISABLE) |
|
special_disable = read_i16(thread + KTHREAD_SPECIAL_APC_DISABLE) |
|
kapc_pending = read_u8(apc_base + KAPC_STATE_KAPC_PENDING) |
|
print(f"\nAPC State:") |
|
print(f" KernelApcDisable = {kapc_disable}") |
|
print(f" SpecialApcDisable = {special_disable}") |
|
print(f" KernelApcPending = {kapc_pending}") |
|
print(f" Process = 0x{process:x}" if process else " Process = ?") |
|
print(f"\nStack trace:") |
|
dump_thread_stack(thread) |
|
|
|
ReactosThreadInfo() |
|
|
|
|
|
class ReactosFindWaiters(gdb.Command): |
|
"""Find threads waiting on a dispatcher object. Usage: ros-waiters <addr>""" |
|
def __init__(self): |
|
super().__init__("ros-waiters", gdb.COMMAND_USER) |
|
def invoke(self, arg, from_tty): |
|
if not arg.strip(): |
|
print("Usage: ros-waiters <event-address>") |
|
return |
|
event = int(arg.strip(), 0) |
|
hdr = read_mem(event, 4) |
|
if not hdr: |
|
print("Cannot read object header") |
|
return |
|
obj_type = hdr[0] |
|
signal_state = read_u32(event + 4) |
|
wait_list_head = event + 8 |
|
flink = read_u64(wait_list_head) |
|
type_names = {0: "NotificationEvent", 1: "SynchronizationEvent", |
|
2: "Mutant", 3: "Process", 5: "Thread", |
|
8: "NotificationTimer", 9: "SynchronizationTimer", |
|
29: "Semaphore"} |
|
print(f"Type={type_names.get(obj_type, f'?({obj_type})')} SignalState={signal_state}") |
|
if not flink or flink == wait_list_head: |
|
print("No waiters") |
|
return |
|
entry = flink |
|
count = 0 |
|
while entry != wait_list_head and count < 32: |
|
thread_ptr = read_u64(entry + KWAIT_BLOCK_THREAD) |
|
print(f" Waiter: Thread=0x{thread_ptr:x}" if thread_ptr else " Waiter: Thread=NULL") |
|
entry = read_u64(entry) |
|
if entry is None: |
|
break |
|
count += 1 |
|
print(f"{count} waiter(s)") |
|
|
|
ReactosFindWaiters() |
|
|
|
|
|
class ReactosIrpInfo(gdb.Command): |
|
"""Inspect an IRP. Usage: ros-irp <addr>""" |
|
def __init__(self): |
|
super().__init__("ros-irp", gdb.COMMAND_USER) |
|
def invoke(self, arg, from_tty): |
|
if not arg.strip(): |
|
print("Usage: ros-irp <irp-address>") |
|
return |
|
irp = int(arg.strip(), 0) |
|
irp_type = read_u16(irp) |
|
flags = read_u32(irp + 0x10) |
|
user_event = read_u64(irp + 0x50) |
|
user_buffer = read_u64(irp + 0x68) |
|
thread = read_u64(irp + 0x78) |
|
print(f"IRP 0x{irp:x}: Type=0x{irp_type:x} Flags=0x{flags:x}") |
|
print(f" UserEvent=0x{user_event:x} UserBuffer=0x{user_buffer:x} Thread=0x{thread:x}") |
|
|
|
ReactosIrpInfo() |
|
|
|
|
|
# ============================================================ |
|
# ros-trapframes — also accepts CS=0x23 (compat mode) |
|
# ============================================================ |
|
|
|
class ReactosFindTrapFrames(gdb.Command): |
|
"""Scan stack for KTRAP_FRAMEs. Usage: ros-trapframes [lo] [hi]""" |
|
def __init__(self): |
|
super().__init__("ros-trapframes", gdb.COMMAND_USER) |
|
def invoke(self, arg, from_tty): |
|
args = arg.strip().split() |
|
if len(args) >= 2: |
|
low = int(args[0], 0) |
|
high = int(args[1], 0) |
|
elif len(args) == 1: |
|
low = int(args[0], 0) |
|
high = low + 0x2000 |
|
else: |
|
try: |
|
low = int(gdb.parse_and_eval("$rsp")) |
|
except: |
|
print("Cannot read RSP") |
|
return |
|
high = low + 0x4000 |
|
found = 0 |
|
for pos in range(low, high - TF_SIZE, 8): |
|
cs_data = read_mem(pos + TF_SEG_CS, 2) |
|
if cs_data is None: |
|
continue |
|
seg_cs = struct.unpack('<H', cs_data)[0] |
|
# Accept kernel (0x10), user64 (0x33), and compat32 (0x23) |
|
if seg_cs not in (0x10, 0x33, 0x23): |
|
continue |
|
rip = read_u64(pos + TF_RIP) |
|
if rip is None: |
|
continue |
|
# For user/compat mode, RIP can be low; for kernel, must be KSEG0 |
|
if seg_cs == 0x10 and not (0xFFFFF80000000000 <= rip <= 0xFFFFFFFFFFC00000): |
|
continue |
|
if seg_cs in (0x33, 0x23) and rip > 0x800000000000: |
|
continue # user addresses should be low |
|
rsp = read_u64(pos + TF_RSP) or 0 |
|
error_code = read_u64(pos + TF_ERROR_CODE) or 0 |
|
rax = read_u64(pos + TF_RAX) or 0 |
|
rcx = read_u64(pos + TF_RCX) or 0 |
|
sym = addr_to_sym(rip) |
|
sym_str = f" ({sym})" if sym else "" |
|
cs_name = {0x10: "kernel", 0x33: "user64", 0x23: "compat32"}.get(seg_cs, "?") |
|
found += 1 |
|
print(f"KTRAP_FRAME @ 0x{pos:x}:") |
|
print(f" Rip = 0x{rip:x}{sym_str}") |
|
print(f" CS = 0x{seg_cs:x} ({cs_name})") |
|
print(f" Rsp = 0x{rsp:x}") |
|
print(f" ErrorCode = 0x{error_code:x}") |
|
print(f" Rax=0x{rax:x} Rcx=0x{rcx:x}") |
|
print() |
|
if found == 0: |
|
print("No KTRAP_FRAMEs found.") |
|
else: |
|
print(f"{found} trap frame(s) found.") |
|
|
|
ReactosFindTrapFrames() |
|
|
|
|
|
# ============================================================ |
|
# ros-callchain |
|
# ============================================================ |
|
|
|
class ReactosCallChain(gdb.Command): |
|
"""Heuristic stack scan for return addresses. Usage: ros-callchain [rsp] [depth]""" |
|
def __init__(self): |
|
super().__init__("ros-callchain", gdb.COMMAND_USER) |
|
def invoke(self, arg, from_tty): |
|
args = arg.strip().split() |
|
try: |
|
rsp = int(args[0], 0) if args else int(gdb.parse_and_eval("$rsp")) |
|
except: |
|
print("Cannot read RSP") |
|
return |
|
depth = int(args[1], 0) if len(args) >= 2 else 0x400 |
|
# Build module ranges |
|
ranges = [] |
|
try: |
|
list_head = int(gdb.parse_and_eval("(unsigned long long)&PsLoadedModuleList")) |
|
flink = read_u64(list_head) |
|
entry = flink |
|
count = 0 |
|
while entry and entry != list_head and count < 256: |
|
dll_base = read_u64(entry + 0x30) |
|
size = read_u32(entry + 0x40) |
|
name = read_unicode_string(entry + 0x58) |
|
if dll_base and size: |
|
ranges.append((dll_base, dll_base + size, name or "?")) |
|
entry = read_u64(entry) |
|
count += 1 |
|
except: |
|
pass |
|
print(f"Scanning 0x{rsp:x} - 0x{rsp + depth:x}:") |
|
for off in range(0, depth, 8): |
|
val = read_u64(rsp + off) |
|
if val is None: |
|
continue |
|
mod = None |
|
for lo, hi, n in ranges: |
|
if lo <= val < hi: |
|
mod = n |
|
break |
|
if mod is None: |
|
if not ranges and 0xFFFFF80000000000 <= val <= 0xFFFFF8FFFFFFFFFF: |
|
sym = addr_to_sym(val) |
|
if sym: |
|
print(f" [RSP+0x{off:03x}] 0x{val:x} {sym}") |
|
continue |
|
sym = addr_to_sym(val) |
|
sym_str = sym if sym else f"{mod}+?" |
|
print(f" [RSP+0x{off:03x}] 0x{val:x} {sym_str}") |
|
|
|
ReactosCallChain() |
|
|
|
|
|
# ============================================================ |
|
# ros-kdb-bt |
|
# ============================================================ |
|
|
|
# Symbols that belong to the kdb / RtlAssert / int-0x2c preamble. |
|
# These are skipped when walking from the parked KDB stack back to |
|
# whoever called RtlAssert. Match by symbol prefix (left-anchored). |
|
_KDB_PREAMBLE_PREFIXES = ( |
|
"Cp16550", "CpGetByte", "CpPutByte", |
|
"KdPortGetByte", "KdPortPutByte", |
|
"KdbpTryGetChar", "KdbpGetChar", "KdpReadTermKey", |
|
"KdIoReadLine", "KdIoPrint", "KdIo", |
|
"KdReceivePacket", "KdSendPacket", "KdpSendString", |
|
"KdbgReceivePacket", "KdbgSendPacket", |
|
"KdpPrompt", "KdpPromptString", "KdpTrap", "KdpReport", |
|
"KdpGetContext", "KdpSetContext", |
|
"KdpSendWaitContinue", "KdpReportExceptionStateChange", |
|
"KdEnterDebugger", "KdExitDebugger", |
|
"KdDebuggerNotPresent", "KdpContext", |
|
"KiDispatchException", "InternalDispatchException", |
|
"KiDebugServiceTrap", "KiSystemCall64", |
|
"DebugService", "DbgPrompt", "DbgPrint", "DbgUserBreakPoint", |
|
"RtlAssert", "vDbgPrintEx", "TraceDataBuffer", |
|
"__FUNCTION__", "MmSysPteIndex", # arg / rodata neighbours |
|
"KiInitialPcr", "KiProcessorBlock", # neighbour symbols at +/- offsets |
|
"KiP0DoubleFault", # double-fault stack data |
|
) |
|
|
|
def _is_preamble_sym(sym): |
|
if not sym: |
|
return True |
|
bare = sym.split(" + ", 1)[0] |
|
return any(bare.startswith(p) for p in _KDB_PREAMBLE_PREFIXES) |
|
|
|
def _stack_kernel_returns(rsp, depth_bytes, ranges): |
|
"""Yield (offset, value, sym, module) for every plausible kernel-text |
|
return-address slot in [rsp, rsp+depth_bytes).""" |
|
for off in range(0, depth_bytes, 8): |
|
val = read_u64(rsp + off) |
|
if val is None: |
|
continue |
|
# Plausible kernel-text addresses: ntoskrnl 0xfffff800.. + drivers 0xfffff880.. |
|
if not (0xFFFFF80000000000 <= val <= 0xFFFFFFFFFFFFFFFF): |
|
continue |
|
mod = None |
|
for lo, hi, n in ranges: |
|
if lo <= val < hi: |
|
mod = n |
|
break |
|
if mod is None: |
|
# Accept ntoskrnl-range even without module table (early boot etc). |
|
if not (0xFFFFF80000000000 <= val <= 0xFFFFF8FFFFFFFFFF): |
|
continue |
|
sym = addr_to_sym(val) |
|
if not sym: |
|
continue |
|
yield off, val, sym, mod |
|
|
|
def _build_module_ranges(): |
|
ranges = [] |
|
try: |
|
list_head = int(gdb.parse_and_eval("(unsigned long long)&PsLoadedModuleList")) |
|
flink = read_u64(list_head) |
|
entry = flink |
|
count = 0 |
|
while entry and entry != list_head and count < 256: |
|
dll_base = read_u64(entry + 0x30) |
|
size = read_u32(entry + 0x40) |
|
name = read_unicode_string(entry + 0x58) |
|
if dll_base and size: |
|
ranges.append((dll_base, dll_base + size, name or "?")) |
|
entry = read_u64(entry) |
|
count += 1 |
|
except Exception: |
|
pass |
|
return ranges |
|
|
|
def _thread_stack_has_kdb(rsp, ranges, scan_bytes=0x2000): |
|
"""Return True if this thread's stack contains a frame in KdpPrompt/ |
|
KdReceivePacket/KdpTrap path — i.e. it's the CPU parked in kdb.""" |
|
needles = ("KdpPrompt", "KdReceivePacket", "KdpTrap", "KdIoReadLine", |
|
"KdbpTryGetChar", "Cp16550ReadLsr") |
|
seen = 0 |
|
for off, val, sym, mod in _stack_kernel_returns(rsp, scan_bytes, ranges): |
|
bare = sym.split(" + ", 1)[0] |
|
if any(bare.startswith(n) for n in needles): |
|
seen += 1 |
|
if seen >= 2: |
|
return True |
|
return False |
|
|
|
def _looks_like_trap_frame(addr): |
|
"""Heuristic: at `addr` does memory match a KTRAP_FRAME shape with a |
|
plausible kernel/user RIP and CS in {0x10, 0x33, 0x23, 0x1b}?""" |
|
if not addr or addr < 0xFFFFF80000000000: |
|
return False |
|
rip = read_u64(addr + TF_RIP) |
|
cs = read_u64(addr + TF_SEG_CS) |
|
rsp = read_u64(addr + TF_RSP) |
|
if rip is None or cs is None or rsp is None: |
|
return False |
|
cs &= 0xFFFF |
|
if cs not in (0x10, 0x33, 0x23, 0x1b): |
|
return False |
|
# RIP must look like code (kernel or 64-bit user). |
|
if not ((0xFFFFF80000000000 <= rip <= 0xFFFFFFFFFFFFFFFF) or |
|
(0x0000000000400000 <= rip <= 0x00007FFFFFFFFFFF)): |
|
return False |
|
return True |
|
|
|
def _find_assert_trapframe(kdb_rsp, ranges, scan_bytes=0x4000): |
|
"""Scan the kdb-side stack for a pointer to a KTRAP_FRAME whose RIP |
|
is in the DebugService / RtlAssert / KiBreakpointTrap preamble. Return |
|
the trap frame address, or None.""" |
|
for off in range(0, scan_bytes, 8): |
|
val = read_u64(kdb_rsp + off) |
|
if val is None or val < 0xFFFFF80000000000: |
|
continue |
|
if not _looks_like_trap_frame(val): |
|
continue |
|
rip = read_u64(val + TF_RIP) |
|
sym = addr_to_sym(rip) or "" |
|
bare = sym.split(" + ", 1)[0] |
|
if bare.startswith(("DebugService", "RtlAssert", "DbgPrompt", |
|
"KiBreakpointTrap", "DbgUserBreakPoint", |
|
"DbgPrint")): |
|
return val |
|
return None |
|
|
|
# CONTEXT (amd64) field offsets used to read the asserter's RSP/RIP. |
|
CTX_RSP = 0x98 |
|
CTX_RIP = 0xF8 |
|
|
|
def _get_kdbg_context_rsp(): |
|
"""Return (rsp, rip) from the global KdbgKdContext CONTEXT structure |
|
(the asserter's saved register state at the trap), or (None, None).""" |
|
try: |
|
ctx_addr = int(gdb.parse_and_eval("(unsigned long long)&KdbgKdContext")) |
|
except Exception: |
|
return None, None |
|
rsp = read_u64(ctx_addr + CTX_RSP) |
|
rip = read_u64(ctx_addr + CTX_RIP) |
|
return rsp, rip |
|
|
|
def _gdb_bt_lines(depth): |
|
try: |
|
out = gdb.execute(f"bt {depth}", to_string=True) |
|
except Exception as e: |
|
return [f"(bt failed: {e})"] |
|
return [ln.rstrip() for ln in out.splitlines() if ln.strip()] |
|
|
|
_BT_PREAMBLE_FUNCS = ( |
|
"Cp16550ReadLsr", "Uart16550GetByte", "CpGetByte", "CpPutByte", |
|
"KdPortGetByteEx", "KdbpTryGetCharSerial", "KdpReadTermKey", |
|
"KdIoReadLine", "KdReceivePacket", "KdbgReceivePacket", |
|
"KdpPromptString", "KdpPrompt", "KdpTrap", "KdpReport", |
|
"KdpPrintString", |
|
"KiDispatchException", "InternalDispatchException", |
|
"KiDebugServiceTrap", "DebugService", |
|
"DbgPrompt", "DbgPrint", "DbgUserBreakPoint", |
|
"RtlAssert", "vDbgPrintEx", |
|
"KiProcessorFreezeHandler", "KiNmiInterruptHandler", |
|
"KiNmiInterruptWithEf", "KiNmiInterrupt", |
|
) |
|
|
|
def _bt_line_is_preamble(line): |
|
# Match "#N 0x... in FunctionName (...)" or "#N FunctionName (...)" |
|
m = _re.search(r'(?:in\s+)([A-Za-z_][A-Za-z0-9_]*)', line) |
|
if not m: |
|
# Also match "#N FunctionName (" (no 'in' prefix when frame has no addr). |
|
m = _re.search(r'^#\d+\s+([A-Za-z_][A-Za-z0-9_]*)\s*\(', line) |
|
if not m: |
|
return False |
|
return m.group(1) in _BT_PREAMBLE_FUNCS |
|
|
|
class ReactosKdbBacktrace(gdb.Command): |
|
"""Show the asserting callchain from a kdb-reached assert, stripping |
|
the RtlAssert / DbgPrompt / int 0x2c preamble. |
|
|
|
Usage: ros-kdb-bt [rsp] [depth] |
|
|
|
With no args, auto-selects the kdb-parked CPU, runs `bt 60` on it, |
|
and prints the frames past the kdb preamble. With [rsp] supplied, |
|
falls back to the heuristic stack scan used by ros-callchain. |
|
Run after `ros-connect` while target is halted at a kdb prompt.""" |
|
def __init__(self): |
|
super().__init__("ros-kdb-bt", gdb.COMMAND_USER) |
|
def invoke(self, arg, from_tty): |
|
args = arg.strip().split() |
|
ranges = _build_module_ranges() |
|
rsp = None |
|
if args: |
|
try: |
|
rsp = int(args[0], 0) |
|
except Exception: |
|
print(f"Bad rsp arg: {args[0]}") |
|
return |
|
depth = int(args[1], 0) if len(args) >= 2 else 0x4000 |
|
|
|
if rsp is None: |
|
# Auto-find the kdb-parked thread, then chase the trap frame |
|
# off the dedicated KdbStack back to the asserter's stack. |
|
saved = gdb.selected_thread() |
|
kdb_thread = None |
|
kdb_rsp = None |
|
for inf in gdb.inferiors(): |
|
for thr in inf.threads(): |
|
try: |
|
thr.switch() |
|
cur_rsp = int(gdb.parse_and_eval("(unsigned long long)$rsp")) |
|
if _thread_stack_has_kdb(cur_rsp, ranges): |
|
kdb_thread = thr |
|
kdb_rsp = cur_rsp |
|
break |
|
except Exception: |
|
continue |
|
if kdb_thread: |
|
break |
|
if kdb_thread is None: |
|
print("Could not locate a CPU parked in kdb. Pass [rsp] explicitly.") |
|
if saved: |
|
saved.switch() |
|
return |
|
# GDB's unwinder has CFI for the kernel and can walk across |
|
# the kdb stack swap. Use `bt` and filter the preamble. |
|
print(f"Parked CPU: thread #{kdb_thread.num} RSP=0x{kdb_rsp:x}") |
|
lines = _gdb_bt_lines(60) |
|
# Strip the leading kdb-prompt preamble frames. |
|
start = 0 |
|
while start < len(lines) and _bt_line_is_preamble(lines[start]): |
|
start += 1 |
|
# GDB's unwinder stops at KdbpStackSwitchAndCall (asm stub |
|
# that swaps to the asserter's process stack). The next |
|
# frame line shows the asserter's saved RSP as if it were |
|
# a RIP. Find it and continue with a heuristic scan from |
|
# the asserter's stack. |
|
swap_idx = next((i for i, ln in enumerate(lines) |
|
if "KdbpStackSwitchAndCall" in ln), None) |
|
asserter_rsp = None |
|
if swap_idx is not None and swap_idx + 1 < len(lines): |
|
m = _re.search(r'#\d+\s+(0x[0-9a-fA-F]+)\s+in\s+\?\?', |
|
lines[swap_idx + 1]) |
|
if m: |
|
cand = int(m.group(1), 0) |
|
if 0xFFFFF80000000000 <= cand < 0xFFFFFFFFFFFFFFFF: |
|
asserter_rsp = cand |
|
print(f"Stripped {start} kdb/assert preamble frames.") |
|
print("=== Asserting callchain (gdb-unwound, innermost first) ===") |
|
for i, ln in enumerate(lines[start:start+30]): |
|
ln2 = _re.sub(r'^#\d+\s+', f'#{i:<2} ', ln) |
|
print(" " + ln2) |
|
if len(lines) - start > 30: |
|
print(f" ... ({len(lines) - start - 30} more, raise [depth] or use plain `bt`)") |
|
|
|
# Continue past the stack swap into the asserter's frames. |
|
if asserter_rsp: |
|
print("") |
|
print(f"=== Asserter's stack (RSP=0x{asserter_rsp:x}, heuristic) ===") |
|
frames = list(_stack_kernel_returns(asserter_rsp, depth, ranges)) |
|
chain = [] |
|
for off, val, sym, mod in frames: |
|
if chain and chain[-1][2] == sym: |
|
continue |
|
# Skip the kdb stack-swap helpers themselves. |
|
bare = sym.split(" + ", 1)[0] |
|
if bare.startswith(("KdbpStackSwitchAndCall", |
|
"KdbpCallMainLoop", |
|
"KdbEnterDebuggerException", |
|
"KdbpCliMainLoop", "KdbPrompt", |
|
"KdbPromptString", |
|
"Kdbg", "KdbpAttachToThread", |
|
"KdbpStallExecutionProcessor")): |
|
continue |
|
chain.append((off, val, sym, mod)) |
|
# Now strip any residual RtlAssert/DbgPrompt/etc preamble. |
|
istart = 0 |
|
while istart < len(chain) and _is_preamble_sym(chain[istart][2]): |
|
istart += 1 |
|
printed = 0 |
|
for off, val, sym, mod in chain[istart:]: |
|
tag = f"[{mod}]" if mod else "" |
|
print(f" [RSP+0x{off:04x}] 0x{val:x} {sym} {tag}") |
|
printed += 1 |
|
if printed >= 60: |
|
print(" ... (truncated)") |
|
break |
|
return |
|
|
|
# Explicit rsp supplied — fall back to heuristic stack scan. |
|
# Walk: collect every kernel return on the stack in order, then |
|
# strip the leading preamble run. The first non-preamble symbol |
|
# is the asserting function. |
|
frames = list(_stack_kernel_returns(rsp, depth, ranges)) |
|
# Dedupe consecutive duplicates that come from compiler-emitted |
|
# double-pushes of return addresses (RtlAssert + 39 / + 188 etc). |
|
chain = [] |
|
for off, val, sym, mod in frames: |
|
if chain and chain[-1][2] == sym: |
|
continue |
|
chain.append((off, val, sym, mod)) |
|
# Find first non-preamble entry. |
|
start = 0 |
|
while start < len(chain) and _is_preamble_sym(chain[start][2]): |
|
start += 1 |
|
if start == 0: |
|
print("(no preamble frames matched — output is raw)") |
|
else: |
|
print(f"Stripped {start} kdb/assert preamble frames.") |
|
print(f"Scanning 0x{rsp:x} - 0x{rsp + depth:x} ({len(chain)} kernel-text slots)") |
|
# Print the cleaned chain. |
|
print("=== Asserting callchain (innermost first) ===") |
|
printed = 0 |
|
for off, val, sym, mod in chain[start:]: |
|
tag = f"[{mod}]" if mod else "" |
|
print(f" [RSP+0x{off:04x}] 0x{val:x} {sym} {tag}") |
|
printed += 1 |
|
if printed >= 30: |
|
print(" ... (truncated; pass larger [depth] for more)") |
|
break |
|
|
|
ReactosKdbBacktrace() |
|
|
|
|
|
# ============================================================ |
|
# ros-findmod |
|
# ============================================================ |
|
|
|
class ReactosFindModule(gdb.Command): |
|
"""Find module base by scanning for MZ/PE. Usage: ros-findmod <addr>""" |
|
def __init__(self): |
|
super().__init__("ros-findmod", gdb.COMMAND_USER) |
|
def invoke(self, arg, from_tty): |
|
if not arg.strip(): |
|
print("Usage: ros-findmod <address>") |
|
return |
|
target = int(arg.strip(), 0) |
|
page = target & ~0xFFF |
|
for i in range(4096): |
|
sig = read_u16(page) |
|
if sig == 0x5A4D: |
|
pe_off = read_u32(page + 0x3C) |
|
if pe_off and pe_off < 0x1000: |
|
pe_sig = read_u32(page + pe_off) |
|
if pe_sig == 0x00004550: |
|
print(f"Module base: 0x{page:x}") |
|
sections = get_pe_all_sections(page) |
|
for sname, rva in sections: |
|
print(f" {sname:8s} @ 0x{page + rva:x}") |
|
return |
|
page -= 0x1000 |
|
print(f"No MZ/PE found scanning back from 0x{target:x}") |
|
|
|
ReactosFindModule() |
|
|
|
|
|
# ============================================================ |
|
# ros-pslist, ros-findproc |
|
# ============================================================ |
|
|
|
def _walk_eprocesses(max_count=128): |
|
try: |
|
head_addr = int(gdb.parse_and_eval("(unsigned long long)&PsActiveProcessHead")) |
|
except: |
|
print("PsActiveProcessHead not found.") |
|
return |
|
flink = read_u64(head_addr) |
|
if flink is None or flink == 0: |
|
return |
|
seen = set() |
|
e = flink |
|
while e and e != head_addr and e not in seen and len(seen) < max_count: |
|
seen.add(e) |
|
eproc = e - EPROCESS_ACTIVE_PROCESS_LINKS |
|
pid = read_u64(eproc + EPROCESS_UNIQUE_PROCESS_ID) |
|
name = read_ascii_string(eproc + EPROCESS_IMAGE_FILE_NAME, 16) |
|
yield (eproc, pid, name) |
|
nxt = read_u64(e) |
|
if nxt is None: |
|
break |
|
e = nxt |
|
|
|
def _walk_process_threads(eproc, max_count=256): |
|
list_head = eproc + EPROCESS_THREAD_LIST_HEAD |
|
entry = read_u64(list_head) |
|
seen = set() |
|
count = 0 |
|
while entry and entry != list_head and entry not in seen and count < max_count: |
|
seen.add(entry) |
|
thread = entry - ETHREAD_THREAD_LIST_ENTRY |
|
yield thread |
|
entry = read_u64(entry) |
|
count += 1 |
|
|
|
class ReactosProcessList(gdb.Command): |
|
"""List active processes. Usage: ros-pslist""" |
|
def __init__(self): |
|
super().__init__("ros-pslist", gdb.COMMAND_USER) |
|
def invoke(self, arg, from_tty): |
|
count = 0 |
|
for eproc, pid, name in _walk_eprocesses(): |
|
count += 1 |
|
shown = name if name else "<no name>" |
|
tcount = sum(1 for _ in _walk_process_threads(eproc)) |
|
print(f" {fmt_ptr(eproc)} pid={fmt_ptr(pid)} threads={tcount:2d} {shown}") |
|
print(f"\n{count} active processes") |
|
|
|
ReactosProcessList() |
|
|
|
class ReactosFindProcess(gdb.Command): |
|
"""Find process by name. Usage: ros-findproc <name>""" |
|
def __init__(self): |
|
super().__init__("ros-findproc", gdb.COMMAND_USER) |
|
def invoke(self, arg, from_tty): |
|
needle = arg.strip().lower() |
|
if not needle: |
|
print("Usage: ros-findproc <name>") |
|
return |
|
for eproc, pid, name in _walk_eprocesses(): |
|
if name and needle in name.lower(): |
|
print(f" {fmt_ptr(eproc)} pid={fmt_ptr(pid)} {name}") |
|
|
|
class ReactosProcessThreads(gdb.Command): |
|
"""List threads in an EPROCESS. Usage: ros-threads <eprocess-address>""" |
|
def __init__(self): |
|
super().__init__("ros-threads", gdb.COMMAND_USER) |
|
def invoke(self, arg, from_tty): |
|
if not arg.strip(): |
|
print("Usage: ros-threads <eprocess-address>") |
|
return |
|
eproc = int(arg.strip(), 0) |
|
name = read_ascii_string(eproc + EPROCESS_IMAGE_FILE_NAME, 16) or "<no name>" |
|
print(f"=== EPROCESS {fmt_ptr(eproc)} {name} ===") |
|
for thread in _walk_process_threads(eproc): |
|
init_stack = read_u64(thread + KTHREAD_INITIAL_STACK) |
|
kern_stack = read_u64(thread + KTHREAD_KERNEL_STACK) |
|
state = read_u8(thread + KTHREAD_STATE) |
|
wait_reason = read_u8(thread + KTHREAD_WAIT_REASON) |
|
pid = read_u64(thread + ETHREAD_CID) |
|
tid = read_u64(thread + ETHREAD_CID + 8) |
|
print(f" ETHREAD {fmt_ptr(thread)} pid={fmt_ptr(pid)} tid={fmt_ptr(tid)} " |
|
f"state={state} wait={wait_reason} " |
|
f"KernelStack={fmt_ptr(kern_stack)} InitialStack={fmt_ptr(init_stack)}") |
|
dump_thread_stack(thread) |
|
print("") |
|
|
|
ReactosProcessThreads() |
|
|
|
ReactosFindProcess() |
|
|
|
|
|
# ============================================================ |
|
# ros-gdt — dump GDT entries (useful for WoW64 debugging) |
|
# ============================================================ |
|
|
|
# ============================================================ |
|
# Physical memory + page table walker |
|
# ============================================================ |
|
|
|
def read_phys(paddr, size=8): |
|
"""Read physical memory via QEMU monitor xp command.""" |
|
try: |
|
if size == 8: |
|
out = gdb.execute(f"monitor xp/1gx 0x{paddr:x}", to_string=True) |
|
elif size == 4: |
|
out = gdb.execute(f"monitor xp/1wx 0x{paddr:x}", to_string=True) |
|
else: |
|
out = gdb.execute(f"monitor xp/{size}bx 0x{paddr:x}", to_string=True) |
|
# Parse byte values |
|
parts = out.split(':')[1].strip().split() |
|
return bytes(int(x, 16) for x in parts[:size]) |
|
# Parse: "addr: 0xvalue" |
|
val_str = out.split(':')[1].strip() |
|
return int(val_str, 16) |
|
except Exception as e: |
|
return None |
|
|
|
def virt_to_phys(cr3, vaddr): |
|
"""Walk x86-64 4-level page tables to translate virtual to physical. |
|
Returns (phys_addr, page_size) or (None, None) on failure.""" |
|
pml4_base = cr3 & 0xFFFFFFFFF000 |
|
pml4_idx = (vaddr >> 39) & 0x1FF |
|
pdpt_idx = (vaddr >> 30) & 0x1FF |
|
pd_idx = (vaddr >> 21) & 0x1FF |
|
pt_idx = (vaddr >> 12) & 0x1FF |
|
offset = vaddr & 0xFFF |
|
|
|
# PML4 |
|
pml4e = read_phys(pml4_base + pml4_idx * 8) |
|
if pml4e is None or not (pml4e & 1): |
|
return None, None # not present |
|
|
|
# PDPT |
|
pdpt_base = pml4e & 0xFFFFFFFFF000 |
|
pdpte = read_phys(pdpt_base + pdpt_idx * 8) |
|
if pdpte is None or not (pdpte & 1): |
|
return None, None |
|
if pdpte & 0x80: # 1GB page |
|
phys = (pdpte & 0xFFFFFC0000000) | (vaddr & 0x3FFFFFFF) |
|
return phys, 0x40000000 |
|
|
|
# PD |
|
pd_base = pdpte & 0xFFFFFFFFF000 |
|
pde = read_phys(pd_base + pd_idx * 8) |
|
if pde is None or not (pde & 1): |
|
return None, None |
|
if pde & 0x80: # 2MB page |
|
phys = (pde & 0xFFFFFFFE00000) | (vaddr & 0x1FFFFF) |
|
return phys, 0x200000 |
|
|
|
# PT |
|
pt_base = pde & 0xFFFFFFFFF000 |
|
pte = read_phys(pt_base + pt_idx * 8) |
|
if pte is None or not (pte & 1): |
|
return None, None |
|
phys = (pte & 0xFFFFFFFFF000) | offset |
|
return phys, 0x1000 |
|
|
|
def read_virt_via_cr3(cr3, vaddr, size=8): |
|
"""Read virtual memory from a specific address space (CR3).""" |
|
phys, _ = virt_to_phys(cr3, vaddr) |
|
if phys is None: |
|
return None |
|
return read_phys(phys, size) |
|
|
|
|
|
class ReactosReadVirt(gdb.Command): |
|
"""Read virtual memory from any process's address space. |
|
Usage: ros-xp <cr3> <vaddr> [count] |
|
Reads 8-byte values at vaddr using the given CR3 page tables. |
|
Works even when the target process is not current.""" |
|
|
|
def __init__(self): |
|
super().__init__("ros-xp", gdb.COMMAND_USER) |
|
|
|
def invoke(self, arg, from_tty): |
|
args = arg.strip().split() |
|
if len(args) < 2: |
|
print("Usage: ros-xp <cr3> <vaddr> [count]") |
|
print(" cr3: page table base (from EPROCESS or trap frame)") |
|
print(" vaddr: virtual address to read") |
|
print(" count: number of 8-byte values (default 8)") |
|
return |
|
cr3 = int(args[0], 0) |
|
vaddr = int(args[1], 0) |
|
count = int(args[2], 0) if len(args) >= 3 else 8 |
|
|
|
for i in range(count): |
|
addr = vaddr + i * 8 |
|
val = read_virt_via_cr3(cr3, addr) |
|
if val is None: |
|
print(f" 0x{addr:x}: <not mapped>") |
|
else: |
|
# Try to show as both hex and potential ASCII |
|
print(f" 0x{addr:x}: 0x{val:016x}") |
|
|
|
ReactosReadVirt() |
|
|
|
|
|
class ReactosTranslate(gdb.Command): |
|
"""Translate virtual address using page tables. |
|
Usage: ros-translate <cr3> <vaddr> |
|
Shows PML4/PDPT/PD/PT entries and final physical address.""" |
|
|
|
def __init__(self): |
|
super().__init__("ros-translate", gdb.COMMAND_USER) |
|
|
|
def invoke(self, arg, from_tty): |
|
args = arg.strip().split() |
|
if len(args) < 2: |
|
print("Usage: ros-translate <cr3> <vaddr>") |
|
return |
|
cr3 = int(args[0], 0) |
|
vaddr = int(args[1], 0) |
|
|
|
pml4_base = cr3 & 0xFFFFFFFFF000 |
|
pml4_idx = (vaddr >> 39) & 0x1FF |
|
pdpt_idx = (vaddr >> 30) & 0x1FF |
|
pd_idx = (vaddr >> 21) & 0x1FF |
|
pt_idx = (vaddr >> 12) & 0x1FF |
|
offset = vaddr & 0xFFF |
|
|
|
print(f"VA 0x{vaddr:x} with CR3=0x{cr3:x}") |
|
print(f" Indices: PML4={pml4_idx} PDPT={pdpt_idx} PD={pd_idx} PT={pt_idx} Off=0x{offset:x}") |
|
|
|
pml4e = read_phys(pml4_base + pml4_idx * 8) |
|
if pml4e is None: |
|
print(f" PML4[{pml4_idx}]: <read failed>") |
|
return |
|
p = "P" if pml4e & 1 else "NP" |
|
rw = "RW" if pml4e & 2 else "RO" |
|
us = "User" if pml4e & 4 else "Kern" |
|
nx = "NX" if pml4e & (1 << 63) else "" |
|
print(f" PML4[{pml4_idx}] = 0x{pml4e:016x} [{p} {rw} {us} {nx}]") |
|
if not (pml4e & 1): |
|
print(" -> NOT PRESENT") |
|
return |
|
|
|
pdpt_base = pml4e & 0xFFFFFFFFF000 |
|
pdpte = read_phys(pdpt_base + pdpt_idx * 8) |
|
p = "P" if pdpte & 1 else "NP" |
|
rw = "RW" if pdpte & 2 else "RO" |
|
us = "User" if pdpte & 4 else "Kern" |
|
ps = "1GB" if pdpte & 0x80 else "" |
|
print(f" PDPT[{pdpt_idx}] = 0x{pdpte:016x} [{p} {rw} {us} {ps}]") |
|
if not (pdpte & 1): |
|
print(" -> NOT PRESENT") |
|
return |
|
if pdpte & 0x80: |
|
phys = (pdpte & 0xFFFFFC0000000) | (vaddr & 0x3FFFFFFF) |
|
print(f" -> 1GB page, phys = 0x{phys:x}") |
|
return |
|
|
|
pd_base = pdpte & 0xFFFFFFFFF000 |
|
pde = read_phys(pd_base + pd_idx * 8) |
|
p = "P" if pde & 1 else "NP" |
|
rw = "RW" if pde & 2 else "RO" |
|
us = "User" if pde & 4 else "Kern" |
|
ps = "2MB" if pde & 0x80 else "" |
|
print(f" PD[{pd_idx}] = 0x{pde:016x} [{p} {rw} {us} {ps}]") |
|
if not (pde & 1): |
|
print(" -> NOT PRESENT") |
|
return |
|
if pde & 0x80: |
|
phys = (pde & 0xFFFFFFFE00000) | (vaddr & 0x1FFFFF) |
|
print(f" -> 2MB page, phys = 0x{phys:x}") |
|
return |
|
|
|
pt_base = pde & 0xFFFFFFFFF000 |
|
pte = read_phys(pt_base + pt_idx * 8) |
|
p = "P" if pte & 1 else "NP" |
|
rw = "RW" if pte & 2 else "RO" |
|
us = "User" if pte & 4 else "Kern" |
|
nx = "NX" if pte & (1 << 63) else "" |
|
print(f" PT[{pt_idx}] = 0x{pte:016x} [{p} {rw} {us} {nx}]") |
|
if not (pte & 1): |
|
print(" -> NOT PRESENT") |
|
return |
|
phys = (pte & 0xFFFFFFFFF000) | offset |
|
print(f" -> phys = 0x{phys:x}") |
|
|
|
# Read the value at that physical address |
|
val = read_phys(phys) |
|
if val is not None: |
|
print(f" -> value = 0x{val:016x}") |
|
|
|
ReactosTranslate() |
|
|
|
|
|
class ReactosDumpGdt(gdb.Command): |
|
"""Dump GDT entries. Usage: ros-gdt [base] [limit] |
|
If no args, reads GDTR from QEMU monitor socket at /tmp/wow64.sock |
|
or tries known ReactOS default.""" |
|
def __init__(self): |
|
super().__init__("ros-gdt", gdb.COMMAND_USER) |
|
def invoke(self, arg, from_tty): |
|
args = arg.strip().split() |
|
if len(args) >= 2: |
|
gdtbase = int(args[0], 0) |
|
gdtlimit = int(args[1], 0) |
|
else: |
|
# Try to get from QEMU monitor |
|
gdtbase = None |
|
import subprocess |
|
for sock in ['/tmp/wow64.sock', '/tmp/reactos-monitor.sock']: |
|
try: |
|
result = subprocess.run( |
|
['socat', '-', f'UNIX-CONNECT:{sock}'], |
|
input=b'info registers\n', capture_output=True, timeout=2) |
|
for line in result.stdout.decode(errors='replace').split('\n'): |
|
if 'GDT=' in line: |
|
parts = line.split('GDT=')[1].strip().split() |
|
gdtbase = int(parts[0], 16) |
|
gdtlimit = int(parts[1], 16) |
|
break |
|
if gdtbase: |
|
break |
|
except: |
|
pass |
|
if gdtbase is None: |
|
# Try known ReactOS default |
|
gdtbase = 0xfffff80000099000 |
|
gdtlimit = 0x7ff |
|
print(f"Using default GDT base 0x{gdtbase:x}") |
|
|
|
NAMES = { |
|
0x00: 'NULL', 0x08: '???', 0x10: 'R0_CODE', 0x18: 'R0_DATA', |
|
0x20: 'R3_CMCODE', 0x28: 'R3_DATA', 0x30: 'R3_CODE', |
|
0x40: 'TSS', 0x50: 'R3_CMTEB', |
|
} |
|
print(f"GDT base=0x{gdtbase:x} limit=0x{gdtlimit:x}\n") |
|
for sel in range(0, min(gdtlimit + 1, 0x80), 8): |
|
raw = read_mem(gdtbase + sel, 8) |
|
if raw is None: |
|
continue |
|
w0, w1 = struct.unpack('<II', raw) |
|
if w0 == 0 and w1 == 0 and sel != 0: |
|
continue |
|
base = ((w0 >> 16) & 0xFFFF) | ((w1 & 0xFF) << 16) | (((w1 >> 24) & 0xFF) << 24) |
|
limit = (w0 & 0xFFFF) | (((w1 >> 16) & 0xF) << 16) |
|
type_ = (w1 >> 8) & 0xF |
|
s = (w1 >> 12) & 1 |
|
dpl = (w1 >> 13) & 3 |
|
p = (w1 >> 15) & 1 |
|
l = (w1 >> 21) & 1 |
|
db = (w1 >> 22) & 1 |
|
g = (w1 >> 23) & 1 |
|
name = NAMES.get(sel, '') |
|
mode = "long64" if l else ("compat32" if db else "16bit") |
|
print(f" 0x{sel:02x} {name:12s} base=0x{base:08x} lim=0x{limit:05x} " |
|
f"type={type_:x} DPL={dpl} P={p} {mode} G={g}") |
|
|
|
ReactosDumpGdt() |
|
|
|
|
|
# ============================================================ |
|
# ros-fault — dump faulting context from KTRAP_FRAME in RBP |
|
# ============================================================ |
|
|
|
KPCR_CURRENT_THREAD = 0x188 |
|
|
|
def _get_current_process_name(): |
|
"""Get the current process name from EPROCESS.""" |
|
try: |
|
gs_base = int(gdb.parse_and_eval("$gs_base")) & 0xFFFFFFFFFFFFFFFF |
|
kthread = read_u64(gs_base + KPCR_CURRENT_THREAD) |
|
if not kthread: |
|
return None |
|
eproc = read_u64(kthread + KTHREAD_APC_STATE + KAPC_STATE_PROCESS) |
|
if not eproc: |
|
return None |
|
blk = read_mem(eproc, 0x400) |
|
if blk: |
|
hits = _re.findall(rb"[A-Za-z0-9_.\\-]{2,20}\.exe", blk) |
|
if hits: |
|
return hits[0].decode("ascii") |
|
except: |
|
pass |
|
return None |
|
|
|
def _dump_trap_frame(tf, source=""): |
|
"""Dump a KTRAP_FRAME. Returns False if tf is not a valid trap frame.""" |
|
rip = read_u64(tf + TF_RIP) |
|
cs = read_u16(tf + TF_SEG_CS) |
|
if cs is None or cs not in (0x10, 0x23, 0x33): |
|
return False |
|
rsp = read_u64(tf + TF_RSP) |
|
ss = read_u16(tf + TF_SEG_SS) |
|
err = read_u64(tf + TF_ERROR_CODE) |
|
rax = read_u64(tf + TF_RAX) |
|
rcx = read_u64(tf + TF_RCX) |
|
rdx = read_u64(tf + TF_RDX) |
|
r8 = read_u64(tf + TF_R8) |
|
r9 = read_u64(tf + TF_R9) |
|
rbx = read_u64(tf + TF_RBX) |
|
rdi = read_u64(tf + TF_RDI) |
|
rsi = read_u64(tf + TF_RSI) |
|
rbp_val = read_u64(tf + TF_RBP) |
|
eflags = read_u64(tf + TF_EFLAGS) |
|
cr2 = read_u64(tf + TF_FAULT_ADDR) |
|
ds = read_u16(tf + TF_SEG_DS) |
|
es = read_u16(tf + TF_SEG_ES) |
|
fs = read_u16(tf + TF_SEG_FS) |
|
gs_seg = read_u16(tf + TF_SEG_GS) |
|
|
|
cs_name = {0x10: "kernel", 0x33: "user64", 0x23: "compat32"}.get(cs, "?") |
|
src_str = f" (from {source})" if source else "" |
|
|
|
print(f"=== KTRAP_FRAME @ 0x{tf:x}{src_str} ===\n") |
|
|
|
# Process name |
|
proc_name = _get_current_process_name() |
|
if proc_name: |
|
print(f" Process: {proc_name}") |
|
|
|
print(f" RIP = 0x{rip:x} (CS=0x{cs:x} {cs_name})" if rip is not None else " RIP = ?") |
|
print(f" RSP = 0x{rsp:x} (SS=0x{ss:x})" if rsp is not None else " RSP = ?") |
|
print(f" EFLAGS = 0x{eflags:x}" if eflags is not None else " EFLAGS = ?") |
|
print(f" Error = 0x{err:x}" if err is not None else " Error = ?") |
|
|
|
if err is not None and err != 0: |
|
ext = err & 1 |
|
tbl = (err >> 1) & 3 |
|
idx = (err >> 3) & 0x1FFF |
|
tbl_name = ["GDT", "IDT", "LDT", "IDT"][tbl] |
|
print(f" -> selector index={idx} table={tbl_name} ext={ext}") |
|
|
|
if cr2: |
|
print(f" CR2 = 0x{cr2:x}") |
|
|
|
def _p(name, val): |
|
return f"0x{val:016x}" if val is not None else "?" |
|
print(f"\n RAX={_p('rax',rax)} RBX={_p('rbx',rbx)}") |
|
print(f" RCX={_p('rcx',rcx)} RDX={_p('rdx',rdx)}") |
|
print(f" RSI={_p('rsi',rsi)} RDI={_p('rdi',rdi)}") |
|
print(f" RBP={_p('rbp',rbp_val)} R8 ={_p('r8',r8)}") |
|
print(f" R9 ={_p('r9',r9)}") |
|
if ds is not None: |
|
print(f" DS=0x{ds:x} ES=0x{es:x} FS=0x{fs:x} GS=0x{gs_seg:x}") |
|
|
|
if rip: |
|
sym = addr_to_sym(rip) |
|
if sym: |
|
print(f"\n Symbol: {sym}") |
|
|
|
if rip: |
|
print(f"\n Code at 0x{rip:x}:") |
|
try: |
|
if cs == 0x23: |
|
gdb.execute("set architecture i386") |
|
disas = gdb.execute(f"x/5i 0x{rip:x}", to_string=True) |
|
print(disas.rstrip()) |
|
except Exception as e: |
|
print(f" (cannot disassemble: {e})") |
|
finally: |
|
if cs == 0x23: |
|
try: gdb.execute("set architecture i386:x86-64") |
|
except: pass |
|
|
|
if rsp: |
|
print(f"\n Stack at RSP=0x{rsp:x}:") |
|
try: |
|
for i in range(8): |
|
addr = rsp + i * 8 |
|
val = read_u64(addr) |
|
if val is not None: |
|
sym = addr_to_sym(val) |
|
sym_str = f" {sym}" if sym else "" |
|
print(f" [+0x{i*8:02x}] 0x{val:016x}{sym_str}") |
|
except: |
|
print(f" (cannot read)") |
|
|
|
try: |
|
cr3 = int(gdb.parse_and_eval("$cr3")) |
|
print(f"\n CR3 = 0x{cr3:x}") |
|
except: |
|
pass |
|
|
|
return True |
|
|
|
|
|
class ReactosFaultContext(gdb.Command): |
|
"""Dump faulting/caller context from KTRAP_FRAME. |
|
Auto-detects source: tries RBP (trap handler), then KTHREAD.TrapFrame. |
|
Usage: ros-fault [trapframe_addr]""" |
|
|
|
def __init__(self): |
|
super().__init__("ros-fault", gdb.COMMAND_USER) |
|
|
|
def invoke(self, arg, from_tty): |
|
if arg.strip(): |
|
tf = int(arg.strip(), 0) |
|
if not _dump_trap_frame(tf, "explicit"): |
|
print(f"0x{tf:x} does not look like a valid KTRAP_FRAME") |
|
return |
|
|
|
# Try RBP (valid in trap handlers after EnterTrap) |
|
try: |
|
rbp = int(gdb.parse_and_eval("$rbp")) & 0xFFFFFFFFFFFFFFFF |
|
if rbp > 0xFFFFF00000000000: # kernel stack range |
|
cs_test = read_u16(rbp + TF_SEG_CS) |
|
if cs_test in (0x10, 0x23, 0x33): |
|
if _dump_trap_frame(rbp, "RBP"): |
|
return |
|
except: |
|
pass |
|
|
|
# Try KTHREAD.TrapFrame |
|
try: |
|
gs_base = int(gdb.parse_and_eval("$gs_base")) & 0xFFFFFFFFFFFFFFFF |
|
kthread = read_u64(gs_base + KPCR_CURRENT_THREAD) |
|
if kthread: |
|
thread_tf = read_u64(kthread + KTHREAD_TRAP_FRAME) |
|
if thread_tf: |
|
print(f"KTHREAD=0x{kthread:x}") |
|
if _dump_trap_frame(thread_tf, "KTHREAD.TrapFrame"): |
|
return |
|
except: |
|
pass |
|
|
|
print("No valid KTRAP_FRAME found (not in trap handler, no active syscall)") |
|
|
|
ReactosFaultContext() |
|
|
|
|
|
# ============================================================ |
|
# ros-frame-regs — recover pushed non-volatile registers |
|
# ============================================================ |
|
|
|
class ReactosFrameRegs(gdb.Command): |
|
"""Recover pushed non-volatile registers from a function's stack frame. |
|
Usage: ros-frame-regs <rsp_at_fault> [prologue_pattern] |
|
Patterns: cc-release, cc-uninit, ntfs-release |
|
Or raw: "r14,rdi,rsi,rbx:0x38" """ |
|
|
|
PATTERNS = { |
|
'cc-release': (['r14', 'rdi', 'rsi', 'rbx'], 0x38), |
|
'cc-uninit': (['r14', 'rbp', 'rdi', 'rsi', 'rbx'], 0x20), |
|
'ntfs-release': (['rdi', 'rsi', 'rbx'], 0x20), |
|
} |
|
|
|
def __init__(self): |
|
super().__init__("ros-frame-regs", gdb.COMMAND_USER) |
|
|
|
def invoke(self, arg, from_tty): |
|
args = arg.strip().split() |
|
if len(args) < 1: |
|
print("Usage: ros-frame-regs <rsp> [pattern|reg,reg:sub_size]") |
|
print("Patterns:", ', '.join(self.PATTERNS.keys())) |
|
return |
|
rsp = int(args[0], 0) |
|
if len(args) >= 2: |
|
pat_name = args[1] |
|
if pat_name in self.PATTERNS: |
|
regs, sub_size = self.PATTERNS[pat_name] |
|
elif ':' in pat_name: |
|
parts = pat_name.split(':') |
|
regs = parts[0].split(',') |
|
sub_size = int(parts[1], 0) |
|
else: |
|
print(f"Unknown pattern '{pat_name}'") |
|
return |
|
else: |
|
regs, sub_size = ['rbx'], 0x20 |
|
base = rsp + sub_size |
|
print(f"Stack frame (sub $0x{sub_size:x}, {len(regs)} pushes):") |
|
for i, reg in enumerate(reversed(regs)): |
|
addr = base + i * 8 |
|
val = read_u64(addr) |
|
val_str = f"0x{val:x}" if val is not None else "?" |
|
sym = addr_to_sym(val) if val and val > 0xFFFFF80000000000 else None |
|
sym_str = f" ({sym})" if sym else "" |
|
print(f" saved_{reg:4s} @ 0x{addr:x} = {val_str}{sym_str}") |
|
ret_addr_loc = base + len(regs) * 8 |
|
ret_val = read_u64(ret_addr_loc) |
|
if ret_val: |
|
sym = addr_to_sym(ret_val) |
|
sym_str = f" ({sym})" if sym else "" |
|
print(f" return @ 0x{ret_addr_loc:x} = 0x{ret_val:x}{sym_str}") |
|
|
|
ReactosFrameRegs() |
|
|
|
|
|
# ============================================================ |
|
# ros-verify — compare loaded binary vs build |
|
# ============================================================ |
|
|
|
class ReactosVerifyBinary(gdb.Command): |
|
"""Compare loaded module .text vs build on disk. |
|
Usage: ros-verify <module_base> <build_path>""" |
|
|
|
def __init__(self): |
|
super().__init__("ros-verify", gdb.COMMAND_USER) |
|
|
|
def invoke(self, arg, from_tty): |
|
import hashlib |
|
args = arg.strip().split(None, 1) |
|
if len(args) < 2: |
|
print("Usage: ros-verify <module_base> <build_path>") |
|
return |
|
base = int(args[0], 0) |
|
rel_path = args[1] |
|
filepath = os.path.join(BUILD_DIR, rel_path) if not os.path.isabs(rel_path) else rel_path |
|
if not os.path.exists(filepath): |
|
print(f"ERROR: {filepath} not found") |
|
return |
|
sections = get_pe_all_sections(base) |
|
text_rva = None |
|
for sname, rva in sections: |
|
if sname == '.text': |
|
text_rva = rva |
|
break |
|
if text_rva is None: |
|
print("ERROR: No .text section found") |
|
return |
|
pe_off = read_u32(base + 0x3C) |
|
num_sections = read_u16(base + pe_off + 6) |
|
size_opt = read_u16(base + pe_off + 20) |
|
sec_start = base + pe_off + 24 + size_opt |
|
text_size = None |
|
for i in range(num_sections): |
|
sec = sec_start + i * 40 |
|
sec_name = read_mem(sec, 8) |
|
if sec_name and sec_name.rstrip(b'\x00') == b'.text': |
|
text_size = read_u32(sec + 8) |
|
break |
|
if text_size is None or text_size == 0: |
|
print("ERROR: Cannot determine .text size") |
|
return |
|
print(f"Comparing .text (RVA=0x{text_rva:x}, size=0x{text_size:x})...") |
|
loaded_data = bytearray() |
|
chunk = 4096 |
|
for off in range(0, text_size, chunk): |
|
sz = min(chunk, text_size - off) |
|
d = read_mem(base + text_rva + off, sz) |
|
if d is None: |
|
print(f" ERROR: Memory read failed at base+0x{text_rva + off:x}") |
|
return |
|
loaded_data.extend(d) |
|
loaded_hash = hashlib.sha256(bytes(loaded_data)).hexdigest() |
|
with open(filepath, 'rb') as f: |
|
pe_data = f.read(0x400) |
|
disk_pe_off = struct.unpack_from('<I', pe_data, 0x3C)[0] |
|
disk_num_sec = struct.unpack_from('<H', pe_data, disk_pe_off + 6)[0] |
|
disk_opt_size = struct.unpack_from('<H', pe_data, disk_pe_off + 20)[0] |
|
disk_sec_start = disk_pe_off + 24 + disk_opt_size |
|
for i in range(disk_num_sec): |
|
sec_off = disk_sec_start + i * 40 |
|
if sec_off + 40 > len(pe_data): |
|
f.seek(0) |
|
pe_data = f.read(sec_off + 40) |
|
sec_name = pe_data[sec_off:sec_off + 8].rstrip(b'\x00') |
|
if sec_name == b'.text': |
|
disk_vsize = struct.unpack_from('<I', pe_data, sec_off + 8)[0] |
|
disk_raw_off = struct.unpack_from('<I', pe_data, sec_off + 20)[0] |
|
f.seek(disk_raw_off) |
|
disk_text = f.read(min(disk_vsize, text_size)) |
|
break |
|
else: |
|
print(" ERROR: .text not found in build file") |
|
return |
|
build_hash = hashlib.sha256(disk_text).hexdigest() |
|
if loaded_hash == build_hash: |
|
print(f" MATCH: {loaded_hash[:16]}...") |
|
else: |
|
print(f" MISMATCH!") |
|
print(f" Loaded: {loaded_hash[:32]}...") |
|
print(f" Build: {build_hash[:32]}...") |
|
|
|
ReactosVerifyBinary() |
|
|
|
|
|
# ============================================================ |
|
# Pool debugging helpers |
|
# ============================================================ |
|
|
|
def pool_tag_str(raw_bytes): |
|
return ''.join(chr(b) if 32 <= b < 127 else '.' for b in raw_bytes) |
|
|
|
def parse_pool_header(addr): |
|
data = read_mem(addr, POOL_BLOCK_SIZE) |
|
if data is None: |
|
return None |
|
return { |
|
'addr': addr, |
|
'prev_size': data[0], 'pool_index': data[1], |
|
'block_size': data[2], 'pool_type': data[3], |
|
'tag': pool_tag_str(data[4:8]), 'tag_bytes': data[4:8], |
|
'actual_size': data[2] * POOL_BLOCK_SIZE, |
|
'free': data[3] == 0, |
|
} |
|
|
|
def is_valid_pool_tag(tag_bytes): |
|
for b in tag_bytes: |
|
if b == 0: continue |
|
if b < 0x20 or b > 0x7e: return False |
|
return True |
|
|
|
def is_pool_page(page_addr): |
|
data = read_mem(page_addr, 0x1000) |
|
if data is None: return False |
|
if data[0] != 0 or data[2] == 0 or data[1] > 16: return False |
|
if not is_valid_pool_tag(data[4:8]): return False |
|
pos, prev_bs = 0, 0 |
|
for idx in range(5): |
|
if pos >= 0x1000: break |
|
bs = data[pos + 2] |
|
if bs == 0: return pos == 0 |
|
if idx > 0 and data[pos] != prev_bs: return False |
|
if not is_valid_pool_tag(data[pos + 4:pos + 8]): return False |
|
prev_bs = bs |
|
pos += bs * POOL_BLOCK_SIZE |
|
return True |
|
|
|
def walk_pool_page(page_addr): |
|
blocks, errors = [], [] |
|
pos, prev_bs, idx = 0, 0, 0 |
|
while pos < 0x1000: |
|
hdr = parse_pool_header(page_addr + pos) |
|
if hdr is None: |
|
errors.append(f"Cannot read header at +0x{pos:03x}") |
|
break |
|
hdr['offset'] = pos |
|
hdr['index'] = idx |
|
if idx > 0 and hdr['prev_size'] != prev_bs: |
|
errors.append(f"Block #{idx} at +0x{pos:03x}: PrevSize={hdr['prev_size']} != {prev_bs}") |
|
elif idx == 0 and hdr['prev_size'] != 0: |
|
errors.append(f"Block #0: PrevSize={hdr['prev_size']} != 0") |
|
if hdr['block_size'] == 0: |
|
errors.append(f"Block #{idx}: BlockSize=0") |
|
blocks.append(hdr) |
|
break |
|
blocks.append(hdr) |
|
prev_bs = hdr['block_size'] |
|
pos += hdr['actual_size'] |
|
idx += 1 |
|
if idx > 256: |
|
errors.append("Too many blocks (>256)") |
|
break |
|
if pos != 0x1000 and not errors: |
|
errors.append(f"Blocks end at +0x{pos:03x}, expected +0x1000") |
|
return blocks, errors |
|
|
|
|
|
class ReactosPoolPage(gdb.Command): |
|
"""Walk pool blocks on a page. Usage: ros-pool-page <addr>""" |
|
def __init__(self): |
|
super().__init__("ros-pool-page", gdb.COMMAND_USER) |
|
def invoke(self, arg, from_tty): |
|
if not arg.strip(): |
|
print("Usage: ros-pool-page <address>") |
|
return |
|
addr = int(arg.strip(), 0) |
|
page = addr & ~0xFFF |
|
print(f"=== Pool page 0x{page:x} ===\n") |
|
blocks, errors = walk_pool_page(page) |
|
for b in blocks: |
|
status = "FREE" if b['free'] else f"pt={b['pool_type']}" |
|
marker = " <<<" if page + b['offset'] <= addr < page + b['offset'] + b['actual_size'] else "" |
|
print(f" #{b['index']:3d} +0x{b['offset']:03x} [{b['tag']:4s}] " |
|
f"bs={b['block_size']:3d} (0x{b['actual_size']:03x}) ps={b['prev_size']:3d} {status}{marker}") |
|
if errors: |
|
print(f"\n*** {len(errors)} error(s):") |
|
for e in errors: print(f" {e}") |
|
else: |
|
print(f"\n{len(blocks)} blocks, no errors.") |
|
|
|
ReactosPoolPage() |
|
|
|
|
|
class ReactosPoolBlock(gdb.Command): |
|
"""Inspect a pool block. Usage: ros-pool-block <addr>""" |
|
def __init__(self): |
|
super().__init__("ros-pool-block", gdb.COMMAND_USER) |
|
def invoke(self, arg, from_tty): |
|
if not arg.strip(): |
|
print("Usage: ros-pool-block <address>") |
|
return |
|
data_addr = int(arg.strip(), 0) |
|
hdr_addr = data_addr - POOL_BLOCK_SIZE |
|
hdr = parse_pool_header(hdr_addr) |
|
if hdr is None: |
|
print(f"Cannot read pool header at 0x{hdr_addr:x}") |
|
return |
|
print(f"Pool block 0x{data_addr:x}: tag='{hdr['tag']}' bs={hdr['block_size']} " |
|
f"(0x{hdr['actual_size']:x}) {'FREE' if hdr['free'] else 'alloc'}") |
|
data_size = min(hdr['actual_size'] - POOL_BLOCK_SIZE, 256) |
|
if data_size > 0: |
|
data = read_mem(data_addr, data_size) |
|
if data: |
|
for i in range(0, len(data), 16): |
|
chunk = data[i:i+16] |
|
hex_part = ' '.join(f'{b:02x}' for b in chunk).ljust(47) |
|
ascii_part = ''.join(chr(b) if 32 <= b < 127 else '.' for b in chunk) |
|
print(f" +0x{i:03x}: {hex_part} {ascii_part}") |
|
|
|
ReactosPoolBlock() |
|
|
|
|
|
class ReactosPoolScan(gdb.Command): |
|
"""Scan pool pages for corruption. Usage: ros-pool-scan <start> [num_pages]""" |
|
def __init__(self): |
|
super().__init__("ros-pool-scan", gdb.COMMAND_USER) |
|
def invoke(self, arg, from_tty): |
|
args = arg.strip().split() |
|
if len(args) < 1: |
|
print("Usage: ros-pool-scan <start_page> [num_pages]") |
|
return |
|
start = int(args[0], 0) & ~0xFFF |
|
num = int(args[1], 0) if len(args) >= 2 else 256 |
|
print(f"Scanning {num} pages from 0x{start:x}...") |
|
corrupt, ok, skip = 0, 0, 0 |
|
for i in range(num): |
|
page = start + i * 0x1000 |
|
if not is_pool_page(page): |
|
skip += 1 |
|
continue |
|
blocks, errors = walk_pool_page(page) |
|
if errors: |
|
corrupt += 1 |
|
print(f"\n CORRUPT: 0x{page:x}") |
|
for e in errors: print(f" {e}") |
|
else: |
|
ok += 1 |
|
print(f"\nDone: {ok} OK, {corrupt} CORRUPT, {skip} skipped") |
|
|
|
ReactosPoolScan() |
|
|
|
|
|
class ReactosPoolFind(gdb.Command): |
|
"""Find pool blocks by tag. Usage: ros-pool-find <tag> <start> [num_pages]""" |
|
def __init__(self): |
|
super().__init__("ros-pool-find", gdb.COMMAND_USER) |
|
def invoke(self, arg, from_tty): |
|
args = arg.strip().split() |
|
if len(args) < 2: |
|
print("Usage: ros-pool-find <tag> <start_page> [num_pages]") |
|
return |
|
tag = args[0][:4].ljust(4) |
|
start = int(args[1], 0) & ~0xFFF |
|
num = int(args[2], 0) if len(args) >= 3 else 256 |
|
found = 0 |
|
for i in range(num): |
|
page = start + i * 0x1000 |
|
if not is_pool_page(page): continue |
|
blocks, _ = walk_pool_page(page) |
|
for b in blocks: |
|
if b['tag'] == tag: |
|
found += 1 |
|
print(f" 0x{page + b['offset'] + POOL_BLOCK_SIZE:x} " |
|
f"bs={b['block_size']:3d} {'FREE' if b['free'] else 'alloc'}") |
|
print(f"\n{found} block(s) found.") |
|
|
|
ReactosPoolFind() |
|
|
|
|
|
class ReactosPoolCrash(gdb.Command): |
|
"""Auto-analyze pool corruption from crash stack. Usage: ros-pool-crash""" |
|
def __init__(self): |
|
super().__init__("ros-pool-crash", gdb.COMMAND_USER) |
|
def invoke(self, arg, from_tty): |
|
try: |
|
rsp = int(gdb.parse_and_eval("$rsp")) |
|
except: |
|
print("Cannot read RSP") |
|
return |
|
print("=== Pool crash analysis ===\n") |
|
pool_addrs = set() |
|
for off in range(0, 0x800, 8): |
|
val = read_u64(rsp + off) |
|
if val and 0xFFFFFA8000000000 <= val < 0xFFFFFA8100000000: |
|
pool_addrs.add(val) |
|
if not pool_addrs: |
|
print("No pool addresses found on stack.") |
|
return |
|
pages = sorted(set(a & ~0xFFF for a in pool_addrs)) |
|
for page in pages: |
|
print(f"\n--- Page 0x{page:x} ---") |
|
blocks, errors = walk_pool_page(page) |
|
page_refs = [a for a in pool_addrs if a & ~0xFFF == page] |
|
for b in blocks: |
|
status = "FREE" if b['free'] else f"pt={b['pool_type']}" |
|
blk_start = page + b['offset'] |
|
refs = [a for a in page_refs if blk_start <= a < blk_start + b['actual_size']] |
|
marker = f" <<< {', '.join(f'0x{a:x}' for a in refs)}" if refs else "" |
|
print(f" #{b['index']:3d} +0x{b['offset']:03x} [{b['tag']:4s}] " |
|
f"bs={b['block_size']:3d} ps={b['prev_size']:3d} {status}{marker}") |
|
if errors: |
|
print(f"\n *** CORRUPTION:") |
|
for e in errors: print(f" {e}") |
|
print(f"\n--- Call chain ---") |
|
for off in range(0, 0x600, 8): |
|
val = read_u64(rsp + off) |
|
if val and KSEG0_BASE <= val <= 0xFFFFFFFFFFC00000: |
|
sym = addr_to_sym(val) |
|
if sym: |
|
print(f" [RSP+0x{off:03x}] {sym}") |
|
|
|
ReactosPoolCrash() |
|
|
|
|
|
# ============================================================ |
|
# ros-rmap-arm — arm hbreaks for the MM rmap leak diagnostic |
|
# ============================================================ |
|
# Sets four hardware breakpoints (KVM gdbstub limit) on: |
|
# MmInsertRmap, MmDeleteRmap, MmCleanProcessAddressSpace, |
|
# procsup.c:1378 (the "Might leak resources" early-return path). |
|
# Each emits a single printf line to gdb's logging file. Use with the helper |
|
# script /tmp/rmap_diag2.gdb (which sets up logging) or after manually |
|
# enabling: set logging file <path>; set logging on. |
|
# |
|
# Addresses are computed from ntoskrnl runtime base via known RVAs. |
|
# To override the base, pass it as an argument: ros-rmap-arm 0xfffff80000400000 |
|
class ReactosRmapArm(gdb.Command): |
|
"""Arm rmap-leak diagnostic breakpoints on ntoskrnl. |
|
Usage: ros-rmap-arm [ntoskrnl_runtime_base] |
|
Default base is 0xFFFFF80000400000 (NT 6.2 amd64 ROS canonical). |
|
Each BP emits an RMAP_* line via printf with no Python side-effects, so |
|
gdb 17's known internal-error from python-in-bp-commands is avoided.""" |
|
|
|
# RVAs from x86_64-w64-mingw32-nm of build_nt62/ntoskrnl/ntoskrnl.exe |
|
RVA_INSERT = 0xF7525 # MmInsertRmap entry |
|
RVA_DELETE = 0xF77AD # MmDeleteRmap entry |
|
RVA_CLEAN = 0xCB713 # MmCleanProcessAddressSpace entry |
|
RVA_LEAK = 0xCB745 # procsup.c:1378 ("Might leak resources" early-return) |
|
|
|
def __init__(self): |
|
super().__init__("ros-rmap-arm", gdb.COMMAND_USER) |
|
|
|
def invoke(self, arg, from_tty): |
|
a = arg.strip() |
|
base = int(a, 0) if a else 0xFFFFF80000400000 |
|
cmds = [ |
|
f"hbreak *0x{base + self.RVA_INSERT:x}", |
|
"commands\nsilent\nprintf \"RMAP_INS p=%lx eproc=%lx va=%lx caller=%lx\\n\", $rcx, $rdx, $r8, *(unsigned long*)$rsp\ncontinue\nend", |
|
f"hbreak *0x{base + self.RVA_DELETE:x}", |
|
"commands\nsilent\nprintf \"RMAP_DEL p=%lx eproc=%lx va=%lx caller=%lx\\n\", $rcx, $rdx, $r8, *(unsigned long*)$rsp\ncontinue\nend", |
|
f"hbreak *0x{base + self.RVA_CLEAN:x}", |
|
"commands\nsilent\nprintf \"RMAP_CLEAN_ENTRY eproc=%lx flags2=0x%x rsp=%lx\\n\", $rcx, *(unsigned char*)($rcx+0x389), $rsp\ncontinue\nend", |
|
f"hbreak *0x{base + self.RVA_LEAK:x}", |
|
"commands\nsilent\nprintf \"RMAP_LEAK_EARLY_RETURN eproc=%lx asi=%d image=%s\\n\", $rbx, (*(unsigned char*)($rbx+0x389)>>2)&3, (char*)($rbx+0x2e0)\nbacktrace 25\nprintf \"RMAP_LEAK_END\\n\"\ncontinue\nend", |
|
] |
|
for c in cmds: |
|
try: |
|
gdb.execute(c, to_string=False) |
|
except Exception as e: |
|
print(f" failed: {c.splitlines()[0]}: {e}") |
|
return |
|
print(f"ros-rmap-arm: 4 hbreaks armed at base 0x{base:x}") |
|
print(" Set 'set logging file <path>' / 'set logging on' before continue.") |
|
|
|
ReactosRmapArm() |
|
|
|
|
|
# ============================================================ |
|
# ARM3 PTE / PFN / proto-array helpers |
|
# ============================================================ |
|
# Source-of-truth: |
|
# sdk/include/ndk/amd64/mmtypes.h - MMPTE union (Hard/Soft/Trans/Proto/Subsect/List) |
|
# ntoskrnl/include/internal/mm.h - MMPFN |
|
# sdk/include/ndk/mmtypes.h - CONTROL_AREA, SUBSECTION |
|
# sdk/include/xdk/amd64/mm.h - PTE_BASE = 0xFFFFF68000000000 |
|
# ntoskrnl/include/internal/amd64/mm.h - MiAddressToPte = PTE_BASE + (va>>12)*8 |
|
# = PTE_BASE + ((va>>9) & 0xFFFFFFFFF8) |
|
# |
|
# AMD64 MMPTE bit layout (NTDDI_LONGHORN+): |
|
# bit 0 : Valid (Hard) |
|
# bit 1 : Dirty1 (or Write) (Hard) |
|
# bit 2 : Owner (Hard) |
|
# bit 10 : Prototype |
|
# bit 11 : Transition (Hard reserved0; Soft.Transition) |
|
# For SOFTWARE PTEs (Valid=0): |
|
# bit 0 : Valid (always 0 if soft) |
|
# bits 1..4 : PageFileLow (4 bits) [Soft] |
|
# bits 5..9 : Protection (5 bits) [Soft/Trans/Proto] |
|
# bit 10 : Prototype [Soft/Trans/Proto] |
|
# bit 11 : Transition [Soft/Trans] |
|
# bits 32..63 : PageFileHigh (32 bits) [Soft] |
|
# Proto form (Valid=0, Prototype=1, Transition=0): |
|
# bits 16..63 : ProtoAddress (signed 48) |
|
# Subsection form (Valid=0, Prototype=1, but used inside subsection PTEs): |
|
# bits 16..63 : SubsectionAddress (signed 48) |
|
# Trans form (Valid=0, Prototype=0, Transition=1): |
|
# bits 12..47 : PageFrameNumber (36 bits, NTDDI_LONGHORN+) |
|
# |
|
# Bit positions verified against MMPTE_HARDWARE / MMPTE_SOFTWARE / MMPTE_PROTOTYPE |
|
# in sdk/include/ndk/amd64/mmtypes.h. |
|
|
|
PTE_BASE_AMD64 = 0xFFFFF68000000000 |
|
PTE_TOP_AMD64 = 0xFFFFF6FFFFFFFFFF |
|
|
|
# MMPFN field offsets (amd64, MI_TRACE_PFNS off — ROS default) |
|
PFN_U1 = 0x00 # 8 bytes (Flink/WsIndex/Event/...) |
|
PFN_PTE_ADDRESS = 0x08 # PMMPTE |
|
PFN_U2 = 0x10 # 8 bytes (Blink/ShareCount) |
|
PFN_REFERENCE_COUNT = 0x18 # USHORT |
|
PFN_E1 = 0x1A # USHORT MMPFNENTRY (PageLocation, PrototypePte, ...) |
|
PFN_USED_PT_ENTRIES = 0x1C # ULONG (_WIN64) |
|
PFN_ORIGINAL_PTE = 0x20 # MMPTE / RmapListHead |
|
PFN_U4 = 0x28 # 8 bytes (PteFrame:57 | InPageError:1 | VerifierAlloc:1 | AweAlloc:1 | Priority:3 | MustBeCached:1) |
|
SIZEOF_MMPFN = 0x60 # heuristic — covers u1..u4 + UsedPageTableEntries + Wsle/NextLRU/PreviousLRU |
|
# We only ever index the array; sizeof is read from struct via gdb if available. |
|
|
|
# CONTROL_AREA / SUBSECTION offsets (amd64) |
|
CONTROL_AREA_SEGMENT = 0x00 |
|
CONTROL_AREA_NUM_PFN_REFS = 0x1C |
|
CONTROL_AREA_NUM_VIEWS = 0x20 |
|
CONTROL_AREA_FLAGS = 0x2C |
|
CONTROL_AREA_FILE_POINTER = 0x30 |
|
SIZEOF_CONTROL_AREA = 0x50 # SUBSECTION starts at ControlArea + 1 |
|
|
|
SUBSECTION_CONTROL_AREA = 0x00 |
|
SUBSECTION_FLAGS = 0x08 |
|
SUBSECTION_STARTING_SECTOR = 0x0C |
|
SUBSECTION_NUM_FULL_SECTORS = 0x10 |
|
SUBSECTION_BASE = 0x18 # PMMPTE (proto array) |
|
SUBSECTION_UNUSED_PTES = 0x20 |
|
SUBSECTION_PTES_IN = 0x24 |
|
SUBSECTION_NEXT = 0x28 |
|
|
|
|
|
def mi_address_to_pte(va): |
|
"""MiAddressToPte (amd64): PTE_BASE + ((va >> 12) * 8) with the 9-bit |
|
self-mapped masking from _MiAddressToPte in ntoskrnl/include/internal/amd64/mm.h.""" |
|
offset = (va >> (12 - 3)) & (0xFFFFFFFFF << 3) |
|
return PTE_BASE_AMD64 + offset |
|
|
|
|
|
def _decode_pte_value(val): |
|
"""Return list of (form_name, dict_of_fields). The 'one' form the kernel |
|
will interpret depends on Valid/Prototype/Transition triplet — we mark it |
|
with 'active': True.""" |
|
valid = val & 1 |
|
prototype = (val >> 10) & 1 |
|
transition = (val >> 11) & 1 |
|
forms = [] |
|
|
|
# Hard.Valid form (only if Valid == 1) |
|
hard = { |
|
'PFN' : (val >> 12) & 0xFFFFFFFFF, # 36 bits at NTDDI_LONGHORN+ |
|
'Write' : (val >> 1) & 1, # Dirty1/Write |
|
'Owner' : (val >> 2) & 1, |
|
'WriteThrough': (val >> 3) & 1, |
|
'CacheDisable': (val >> 4) & 1, |
|
'Accessed' : (val >> 5) & 1, |
|
'Dirty' : (val >> 6) & 1, |
|
'LargePage' : (val >> 7) & 1, |
|
'Global' : (val >> 8) & 1, |
|
'CopyOnWrite' : (val >> 9) & 1, |
|
'Prototype' : prototype, |
|
'NoExecute' : (val >> 63) & 1, |
|
} |
|
forms.append(('Hard.Valid', hard, bool(valid))) |
|
|
|
# Soft form (Valid=0, Prototype=0, Transition=0) |
|
soft = { |
|
'PageFileLow' : (val >> 1) & 0xF, |
|
'Protection' : (val >> 5) & 0x1F, |
|
'Prototype' : prototype, |
|
'Transition' : transition, |
|
'UsedPTEs' : (val >> 12) & 0x3FF, |
|
'PageFileHigh': (val >> 32) & 0xFFFFFFFF, |
|
} |
|
soft_active = (not valid) and (not prototype) and (not transition) |
|
forms.append(('Soft.PageFile', soft, soft_active)) |
|
|
|
# Trans form (Valid=0, Prototype=0, Transition=1) |
|
trans = { |
|
'PFN' : (val >> 12) & 0xFFFFFFFFF, # 36 bits at NTDDI_LONGHORN+ |
|
'Protection' : (val >> 5) & 0x1F, |
|
'Write' : (val >> 1) & 1, |
|
'Owner' : (val >> 2) & 1, |
|
'WriteThrough': (val >> 3) & 1, |
|
'CacheDisable': (val >> 4) & 1, |
|
} |
|
trans_active = (not valid) and (not prototype) and bool(transition) |
|
forms.append(('Soft.Transition', trans, trans_active)) |
|
|
|
# Proto form (Valid=0, Prototype=1) |
|
# ProtoAddress is signed 48-bit at bits 16..63 |
|
proto_raw = (val >> 16) & 0xFFFFFFFFFFFF |
|
if proto_raw & (1 << 47): |
|
proto_addr = proto_raw | 0xFFFF000000000000 |
|
else: |
|
proto_addr = proto_raw |
|
proto = { |
|
'ProtoAddress': proto_addr, |
|
'Protection' : (val >> 11) & 0x1F, # MMPTE_PROTOTYPE.Protection at bits 11..15 |
|
'ReadOnly' : (val >> 8) & 1, |
|
} |
|
proto_active = (not valid) and bool(prototype) and (not transition) |
|
forms.append(('Soft.Prototype', proto, proto_active)) |
|
|
|
# Subsection form (Valid=0, Prototype=1) — same encoding as Proto but |
|
# SubsectionAddress lives at bits 16..63 also; we annotate. |
|
sub = { |
|
'SubsectionAddress': proto_addr, |
|
'Protection' : (val >> 5) & 0x1F, |
|
} |
|
forms.append(('Subsection', sub, False)) # never auto-active; same triplet as Proto |
|
|
|
# List form (free PTE lists) |
|
list_form = { |
|
'OneEntry' : (val >> 1) & 1, |
|
'NextEntry': (val >> 32) & 0xFFFFFFFF, |
|
} |
|
forms.append(('List', list_form, False)) |
|
|
|
return valid, prototype, transition, forms |
|
|
|
|
|
def _classify_pte(val): |
|
"""Return short tag of the *active* form.""" |
|
valid = val & 1 |
|
prototype = (val >> 10) & 1 |
|
transition = (val >> 11) & 1 |
|
if val == 0: |
|
return 'ZERO' |
|
if valid: |
|
return 'HARD.VALID' |
|
if prototype and not transition: |
|
return 'PROTO' |
|
if (not prototype) and transition: |
|
return 'TRANSITION' |
|
if (not prototype) and (not transition): |
|
# Soft: distinguish "all-zero except protection" vs real pagefile entry |
|
if ((val >> 32) & 0xFFFFFFFF) == 0 and ((val >> 1) & 0xF) == 0: |
|
return 'SOFT.ZERO' |
|
return 'SOFT.PAGEFILE' |
|
return 'UNKNOWN' |
|
|
|
|
|
def _print_pte(pte_addr, val, indent=' '): |
|
valid, proto, trans, forms = _decode_pte_value(val) |
|
tag = _classify_pte(val) |
|
print(f"{indent}PTE @ 0x{pte_addr:x} = 0x{val:016x} [{tag}]") |
|
print(f"{indent} Valid={valid} Prototype={proto} Transition={trans}") |
|
for name, fields, active in forms: |
|
if name in ('Subsection', 'List') and not active: |
|
continue |
|
if name == 'Hard.Valid' and not valid: |
|
continue # don't spam Hard fields when not valid |
|
marker = ' <== ACTIVE' if active else '' |
|
kv = ' '.join(f"{k}=0x{v:x}" if isinstance(v, int) and v > 9 else f"{k}={v}" |
|
for k, v in fields.items()) |
|
print(f"{indent} {name:18s} {kv}{marker}") |
|
|
|
|
|
class ReactosPte(gdb.Command): |
|
"""Decode a single 64-bit PTE at <pte_addr> (the address of the PTE cell). |
|
Usage: ros-pte <pte_addr>""" |
|
|
|
def __init__(self): |
|
super().__init__("ros-pte", gdb.COMMAND_USER) |
|
|
|
def invoke(self, arg, from_tty): |
|
if not arg.strip(): |
|
print("Usage: ros-pte <pte_address>") |
|
return |
|
try: |
|
pte_addr = int(arg.strip(), 0) |
|
except Exception: |
|
print(f"Bad address: {arg.strip()}") |
|
return |
|
val = read_u64(pte_addr) |
|
if val is None: |
|
print(f"Cannot read PTE at 0x{pte_addr:x}") |
|
return |
|
_print_pte(pte_addr, val, indent='') |
|
|
|
ReactosPte() |
|
|
|
|
|
class ReactosPteVa(gdb.Command): |
|
"""Find PTE that maps <va> via PTE_BASE self-map and decode it. |
|
Usage: ros-pte-va <va>""" |
|
|
|
def __init__(self): |
|
super().__init__("ros-pte-va", gdb.COMMAND_USER) |
|
|
|
def invoke(self, arg, from_tty): |
|
if not arg.strip(): |
|
print("Usage: ros-pte-va <virtual_address>") |
|
return |
|
try: |
|
va = int(arg.strip(), 0) |
|
except Exception: |
|
print(f"Bad address: {arg.strip()}") |
|
return |
|
pte_addr = mi_address_to_pte(va) |
|
print(f"VA 0x{va:x} -> PTE cell @ 0x{pte_addr:x}") |
|
val = read_u64(pte_addr) |
|
if val is None: |
|
print(f" Cannot read PTE (PT not present?)") |
|
return |
|
_print_pte(pte_addr, val, indent=' ') |
|
|
|
|
|
ReactosPteVa() |
|
|
|
|
|
def _pfn_database_base(): |
|
"""Resolve MmPfnDatabase global. Falls back to canonical 0xFFFFFA8000000000.""" |
|
try: |
|
addr = int(gdb.parse_and_eval("(unsigned long long)MmPfnDatabase")) |
|
if addr: |
|
return addr |
|
except Exception: |
|
pass |
|
return 0xFFFFFA8000000000 |
|
|
|
|
|
def _pfn_entry_size(): |
|
"""Try to derive sizeof(MMPFN) via gdb type info; fall back to SIZEOF_MMPFN.""" |
|
try: |
|
t = gdb.lookup_type("struct _MMPFN") |
|
return int(t.sizeof) |
|
except Exception: |
|
return SIZEOF_MMPFN |
|
|
|
|
|
def _pfn_addr_from_arg(s): |
|
"""Accept either a raw PFN number (small) or an MMPFN struct address (large).""" |
|
n = int(s, 0) |
|
pfn_base = _pfn_database_base() |
|
pfn_size = _pfn_entry_size() |
|
if n < 0x100000000: # treat <4G as a PFN index |
|
return pfn_base + n * pfn_size, n, pfn_size |
|
# else treat as struct address |
|
if n >= pfn_base and (n - pfn_base) % pfn_size == 0: |
|
return n, (n - pfn_base) // pfn_size, pfn_size |
|
# arbitrary address; return as-is, no PFN index |
|
return n, None, pfn_size |
|
|
|
|
|
PAGE_LOCATION_NAMES = { |
|
0: 'Zeroed', 1: 'Free', 2: 'Standby', 3: 'Modified', |
|
4: 'ModifiedNoWrite', 5: 'Bad', 6: 'Active', 7: 'Transition', |
|
} |
|
|
|
|
|
class ReactosPfn(gdb.Command): |
|
"""Decode an MMPFN entry. Arg may be a PFN number or struct address. |
|
Usage: ros-pfn <pfn_or_addr>""" |
|
|
|
def __init__(self): |
|
super().__init__("ros-pfn", gdb.COMMAND_USER) |
|
|
|
def invoke(self, arg, from_tty): |
|
if not arg.strip(): |
|
print("Usage: ros-pfn <pfn-number-or-address>") |
|
return |
|
try: |
|
entry_addr, pfn_idx, pfn_size = _pfn_addr_from_arg(arg.strip()) |
|
except Exception as e: |
|
print(f"Bad arg: {e}") |
|
return |
|
u1 = read_u64(entry_addr + PFN_U1) |
|
pte = read_u64(entry_addr + PFN_PTE_ADDRESS) |
|
u2 = read_u64(entry_addr + PFN_U2) |
|
rcnt = read_u16(entry_addr + PFN_REFERENCE_COUNT) |
|
e1 = read_u16(entry_addr + PFN_E1) |
|
upte = read_u32(entry_addr + PFN_USED_PT_ENTRIES) |
|
orig = read_u64(entry_addr + PFN_ORIGINAL_PTE) |
|
u4 = read_u64(entry_addr + PFN_U4) |
|
if u1 is None or pte is None: |
|
print(f"Cannot read MMPFN at 0x{entry_addr:x}") |
|
return |
|
idx_str = f"PFN={pfn_idx} (0x{pfn_idx:x})" if pfn_idx is not None else "PFN=?" |
|
print(f"=== MMPFN @ 0x{entry_addr:x} {idx_str} sizeof=0x{pfn_size:x} ===") |
|
print(f" u1 = 0x{u1:016x} (Flink/WsIndex/Event/RmapListHead-HACK)") |
|
print(f" PteAddress = 0x{pte:016x}") |
|
print(f" u2 (ShareCount) = 0x{u2:016x} ({u2})") |
|
print(f" ReferenceCount = {rcnt}") |
|
if e1 is not None: |
|
modified = e1 & 1 |
|
read_in_progress = (e1 >> 1) & 1 |
|
write_in_prog = (e1 >> 2) & 1 |
|
prototype_pte = (e1 >> 3) & 1 |
|
page_color = (e1 >> 4) & 0xF |
|
page_location = (e1 >> 8) & 7 |
|
removal_req = (e1 >> 11) & 1 |
|
cache_attr = (e1 >> 12) & 3 |
|
rom = (e1 >> 14) & 1 |
|
parity = (e1 >> 15) & 1 |
|
loc_name = PAGE_LOCATION_NAMES.get(page_location, '?') |
|
print(f" u3.e1 (0x{e1:04x}):") |
|
print(f" PageLocation = {page_location} ({loc_name})") |
|
print(f" PrototypePte = {prototype_pte}") |
|
print(f" Modified = {modified}") |
|
print(f" ReadInProgress = {read_in_progress} (StartOfAllocation)") |
|
print(f" WriteInProgress = {write_in_prog} (EndOfAllocation)") |
|
print(f" PageColor = {page_color}") |
|
print(f" RemovalRequested = {removal_req}") |
|
print(f" CacheAttribute = {cache_attr}") |
|
print(f" Rom={rom} ParityError={parity}") |
|
if upte is not None: |
|
print(f" UsedPageTableEntries = {upte}") |
|
if u4 is not None: |
|
pte_frame = u4 & ((1 << 57) - 1) |
|
in_page_err = (u4 >> 57) & 1 |
|
verifier = (u4 >> 58) & 1 |
|
awe = (u4 >> 59) & 1 |
|
priority = (u4 >> 60) & 7 |
|
must_cached = (u4 >> 63) & 1 |
|
print(f" u4 (0x{u4:016x}):") |
|
print(f" PteFrame={pte_frame:#x} InPageError={in_page_err} VerifierAlloc={verifier} " |
|
f"AweAlloc={awe} Priority={priority} MustBeCached={must_cached}") |
|
if orig is not None: |
|
print(f" OriginalPte:") |
|
_print_pte(entry_addr + PFN_ORIGINAL_PTE, orig, indent=' ') |
|
|
|
ReactosPfn() |
|
|
|
|
|
# ros-rmap — walk the per-PFN rmap chain (ROS-managed MM) |
|
# |
|
# The rmap is stored at MMPFN.OriginalPte (overlaid as RmapListHead per |
|
# ntoskrnl/include/internal/mm.h ~L430). Each MM_RMAP_ENTRY: |
|
# +0x00 Next (PMM_RMAP_ENTRY) |
|
# +0x08 Process (PEPROCESS) |
|
# +0x10 Address (PVOID) |
|
# size = 0x18. |
|
|
|
def _eproc_name(eproc_addr): |
|
"""Best-effort: scan EPROCESS bytes for first *.exe name.""" |
|
if not eproc_addr: |
|
return "" |
|
blk = read_mem(eproc_addr, 0x400) |
|
if not blk: |
|
return "" |
|
hits = _re.findall(rb"[A-Za-z0-9_.\-]{2,20}\.exe", blk) |
|
if hits: |
|
try: return hits[0].decode("ascii") |
|
except: return "" |
|
return "" |
|
|
|
|
|
class ReactosRmap(gdb.Command): |
|
"""Walk the rmap (reverse-mapping) chain for a PFN. Each entry shows |
|
which (Process, VA) currently has the page mapped via the ROS-managed |
|
SSE-resident path. |
|
Usage: ros-rmap <pfn-or-MMPFN-addr> [max=64]""" |
|
|
|
def __init__(self): |
|
super().__init__("ros-rmap", gdb.COMMAND_USER) |
|
|
|
def invoke(self, arg, from_tty): |
|
args = arg.strip().split() |
|
if not args: |
|
print("Usage: ros-rmap <pfn-or-addr> [max=64]") |
|
return |
|
try: |
|
entry_addr, pfn_idx, _ = _pfn_addr_from_arg(args[0]) |
|
except Exception as e: |
|
print(f"Bad arg: {e}") |
|
return |
|
max_n = int(args[1], 0) if len(args) > 1 else 64 |
|
head = read_u64(entry_addr + PFN_ORIGINAL_PTE) |
|
if head is None: |
|
print(f"Cannot read MMPFN at 0x{entry_addr:x}") |
|
return |
|
idx_str = f"PFN={pfn_idx} (0x{pfn_idx:x})" if pfn_idx is not None else "PFN=?" |
|
print(f"=== rmap chain @ MMPFN 0x{entry_addr:x} {idx_str} ===") |
|
print(f" RmapListHead = 0x{head:016x}") |
|
if head == 0: |
|
print(" (empty)") |
|
return |
|
seen = set() |
|
cur = head |
|
i = 0 |
|
while cur and cur not in seen and i < max_n: |
|
seen.add(cur) |
|
nxt = read_u64(cur + 0x00) |
|
proc = read_u64(cur + 0x08) |
|
va = read_u64(cur + 0x10) |
|
if nxt is None or proc is None or va is None: |
|
print(f" [{i:3d}] @0x{cur:016x} <unreadable>") |
|
break |
|
name = _eproc_name(proc) if proc else "" |
|
tag = "" |
|
# Special segment-rmap encoding: bit0 of Address indicates "segment rmap" |
|
if va & 1: |
|
tag = " [SEGMENT-RMAP]" |
|
print(f" [{i:3d}] @0x{cur:016x} Process=0x{proc:016x} ({name}) Address=0x{va:016x}{tag} Next=0x{nxt:016x}") |
|
cur = nxt |
|
i += 1 |
|
if cur in seen: |
|
print(f" *** loop detected at 0x{cur:x}") |
|
elif i >= max_n: |
|
print(f" ... truncated at {max_n} entries") |
|
|
|
ReactosRmap() |
|
|
|
|
|
class ReactosProtoWalk(gdb.Command): |
|
"""Walk the proto-PTE array of a CONTROL_AREA's first SUBSECTION. |
|
Flags any Hard.Valid proto as a MiSegmentDelete invariant violation. |
|
Usage: ros-proto-walk <ControlArea_addr> [max_ptes=32]""" |
|
|
|
def __init__(self): |
|
super().__init__("ros-proto-walk", gdb.COMMAND_USER) |
|
|
|
def invoke(self, arg, from_tty): |
|
args = arg.strip().split() |
|
if len(args) < 1: |
|
print("Usage: ros-proto-walk <ControlArea_addr> [max_ptes=32]") |
|
return |
|
try: |
|
ca = int(args[0], 0) |
|
except Exception: |
|
print(f"Bad ControlArea: {args[0]}") |
|
return |
|
max_ptes = int(args[1], 0) if len(args) >= 2 else 32 |
|
|
|
# Read CONTROL_AREA header |
|
seg = read_u64(ca + CONTROL_AREA_SEGMENT) |
|
npfn_refs = read_u32(ca + CONTROL_AREA_NUM_PFN_REFS) |
|
nviews = read_u32(ca + CONTROL_AREA_NUM_VIEWS) |
|
flags = read_u32(ca + CONTROL_AREA_FLAGS) |
|
file_obj = read_u64(ca + CONTROL_AREA_FILE_POINTER) |
|
if seg is None: |
|
print(f"Cannot read CONTROL_AREA at 0x{ca:x}") |
|
return |
|
print(f"=== CONTROL_AREA @ 0x{ca:x} ===") |
|
print(f" Segment = 0x{seg:x}") |
|
print(f" NumberOfPfnRefs = {npfn_refs}") |
|
print(f" NumberOfMappedViews= {nviews}") |
|
print(f" Flags (LongFlags) = 0x{flags:x}") |
|
print(f" FilePointer = 0x{file_obj:x}") |
|
|
|
# First subsection lives at ControlArea + sizeof(CONTROL_AREA) |
|
sub = ca + SIZEOF_CONTROL_AREA |
|
sub_idx = 0 |
|
violations = 0 |
|
total_walked = 0 |
|
while sub and sub_idx < 16: |
|
sub_ca = read_u64(sub + SUBSECTION_CONTROL_AREA) |
|
sub_flags = read_u32(sub + SUBSECTION_FLAGS) |
|
sub_base = read_u64(sub + SUBSECTION_BASE) |
|
sub_unused = read_u32(sub + SUBSECTION_UNUSED_PTES) |
|
sub_count = read_u32(sub + SUBSECTION_PTES_IN) |
|
sub_next = read_u64(sub + SUBSECTION_NEXT) |
|
if sub_base is None or sub_count is None: |
|
print(f"\n Cannot read SUBSECTION at 0x{sub:x}") |
|
break |
|
if sub_ca != ca and sub_idx == 0: |
|
# In ROS the embedded subsection's ControlArea backref must equal ca |
|
print(f" WARN: SUBSECTION[0].ControlArea=0x{sub_ca:x} != 0x{ca:x}") |
|
print(f"\n --- SUBSECTION[{sub_idx}] @ 0x{sub:x} ---") |
|
print(f" ControlArea = 0x{sub_ca:x}") |
|
print(f" SubsectionFlags = 0x{sub_flags:x}") |
|
print(f" SubsectionBase = 0x{sub_base:x} (proto array)") |
|
print(f" PtesInSubsection = {sub_count} UnusedPtes={sub_unused}") |
|
print(f" NextSubsection = 0x{sub_next:x}") |
|
|
|
limit = min(sub_count or 0, max_ptes) |
|
print(f"\n Walking first {limit} of {sub_count} proto-PTEs:") |
|
print(f" Invariant (MiSegmentDelete): each proto must be ZERO / TRANSITION / SOFT.PAGEFILE") |
|
print(f" {'idx':>5s} {'addr':>18s} {'value':>18s} classification") |
|
for i in range(limit): |
|
pte_addr = sub_base + i * 8 |
|
v = read_u64(pte_addr) |
|
if v is None: |
|
print(f" {i:5d} 0x{pte_addr:016x} <unreadable>") |
|
continue |
|
tag = _classify_pte(v) |
|
marker = '' |
|
if tag == 'HARD.VALID': |
|
marker = ' *** VIOLATION (proto must not be Hard.Valid)' |
|
violations += 1 |
|
print(f" {i:5d} 0x{pte_addr:016x} 0x{v:016x} {tag}{marker}") |
|
total_walked += 1 |
|
|
|
if not sub_next or sub_idx >= 0: |
|
# Default: only walk first subsection unless caller asks for more. |
|
# We still report NextSubsection above. |
|
break |
|
sub = sub_next |
|
sub_idx += 1 |
|
|
|
print(f"\n Summary: walked {total_walked} proto-PTEs, {violations} VIOLATION(s)") |
|
|
|
ReactosProtoWalk() |
|
|
|
|
|
class ReactosMappingWalk(gdb.Command): |
|
"""Walk PTEs for a contiguous VA range (e.g. an MmMapViewInSystemSpace view). |
|
Usage: ros-mapping-walk <base_va> <num_pages>""" |
|
|
|
def __init__(self): |
|
super().__init__("ros-mapping-walk", gdb.COMMAND_USER) |
|
|
|
def invoke(self, arg, from_tty): |
|
args = arg.strip().split() |
|
if len(args) < 2: |
|
print("Usage: ros-mapping-walk <base_va> <num_pages>") |
|
return |
|
try: |
|
base = int(args[0], 0) & ~0xFFF |
|
n = int(args[1], 0) |
|
except Exception: |
|
print("Bad arguments") |
|
return |
|
if n <= 0 or n > 4096: |
|
print(f"num_pages must be 1..4096 (got {n})") |
|
return |
|
print(f"VA range 0x{base:x} - 0x{base + n*0x1000 - 1:x} ({n} pages)") |
|
print(f" {'idx':>5s} {'va':>18s} {'pte':>18s} {'value':>18s} class") |
|
cnt_hard = cnt_proto = cnt_trans = cnt_soft = cnt_zero = cnt_unread = 0 |
|
for i in range(n): |
|
va = base + i * 0x1000 |
|
pte_addr = mi_address_to_pte(va) |
|
v = read_u64(pte_addr) |
|
if v is None: |
|
cnt_unread += 1 |
|
print(f" {i:5d} 0x{va:016x} 0x{pte_addr:016x} {'<unreadable>':>18s}") |
|
continue |
|
tag = _classify_pte(v) |
|
if tag == 'HARD.VALID': cnt_hard += 1 |
|
elif tag == 'PROTO': cnt_proto += 1 |
|
elif tag == 'TRANSITION': cnt_trans += 1 |
|
elif tag.startswith('SOFT'): cnt_soft += 1 |
|
elif tag == 'ZERO': cnt_zero += 1 |
|
print(f" {i:5d} 0x{va:016x} 0x{pte_addr:016x} 0x{v:016x} {tag}") |
|
print(f"\n Summary: hard={cnt_hard} proto={cnt_proto} transition={cnt_trans} " |
|
f"soft={cnt_soft} zero={cnt_zero} unreadable={cnt_unread}") |
|
|
|
ReactosMappingWalk() |
|
|
|
|
|
# ============================================================ |
|
# MM struct dumpers — SECTION/SEGMENT/CONTROL_AREA/SUBSECTION, |
|
# SECTION_OBJECT_POINTERS, SHARED_CACHE_MAP, |
|
# ROS_VACB, MMVAD, MMSUPPORT, MM ranges. |
|
# ============================================================ |
|
# |
|
# Source-of-truth offsets (amd64, NTDDI_WIN8 — the active build target): |
|
# sdk/include/ndk/mmtypes.h : _SECTION (l.809), _SEGMENT (l.407), |
|
# _CONTROL_AREA (l.519), _SUBSECTION (l.571), |
|
# _MMSECTION_FLAGS (l.461), _MMVAD (l.720), |
|
# _MMVAD_SHORT (l.788), _MMVAD_FLAGS (l.686), |
|
# _MM_AVL_TABLE (l.660), _MMSUPPORT (l.922) |
|
# sdk/include/xdk/iotypes.h : _SECTION_OBJECT_POINTERS (l.1794) |
|
# sdk/include/ndk/cctypes.h : _PRIVATE_CACHE_MAP (l.67) |
|
# ntoskrnl/include/internal/cc.h: _ROS_SHARED_CACHE_MAP (l.170), |
|
# _ROS_VACB (l.207) |
|
# ntoskrnl/mm/ARM3/miarm.h : MmSystemRangeStart, Mm{Paged,NonPaged}PoolStart, |
|
# MmSessionBase, MmSystemCache{Start,End,Ws}, |
|
# MiSystemViewStart |
|
# sdk/include/ndk/pstypes.h : EPROCESS.VadRoot (l.1768) — last fields, runtime lookup |
|
|
|
# --- _SECTION layout (amd64) --- |
|
# MMADDRESS_NODE (5 ptrs/ULONG_PTR = 0x28) + Segment(8) + SizeOfSection(8) + |
|
# u.LongFlags(4) + InitialPageProtection(4). |
|
SECTION_SEGMENT = 0x28 |
|
SECTION_SIZEOF_SECTION = 0x30 |
|
SECTION_FLAGS = 0x38 |
|
SECTION_INITIAL_PROTECTION = 0x3C |
|
|
|
# --- _SEGMENT layout (amd64) --- |
|
# ControlArea(8) TotalNumberOfPtes(4) NonExtendedPtes(4) Spare0(4) |
|
# pad(4) SizeOfSegment(8) SegmentPteTemplate(8) NumberOfCommittedPages(4) |
|
# pad(4) ExtendInfo(8) SegmentFlags(4) pad(4) BasedAddress(8) u1(8) u2(8) |
|
# PrototypePte(8) ThePtes[1](8) |
|
SEGMENT_CONTROL_AREA = 0x00 |
|
SEGMENT_TOTAL_PTES = 0x08 |
|
SEGMENT_NON_EXTENDED_PTES = 0x0C |
|
SEGMENT_SIZE_OF_SEGMENT = 0x18 |
|
SEGMENT_PTE_TEMPLATE = 0x20 |
|
SEGMENT_COMMITTED_PAGES = 0x28 |
|
SEGMENT_FLAGS_OFF = 0x38 |
|
SEGMENT_BASED_ADDRESS = 0x40 |
|
SEGMENT_PROTOTYPE_PTE = 0x60 |
|
|
|
# --- _MMSECTION_FLAGS bit table (sdk/include/ndk/mmtypes.h:461) --- |
|
_MMSECTION_FLAG_BITS = [ |
|
'BeingDeleted','BeingCreated','BeingPurged','NoModifiedWriting', |
|
'FailAllIo','Image','Based','File','Networked','NoCache', |
|
'PhysicalMemory','CopyOnWrite','Reserve','Commit','FloppyMedia', |
|
'WasPurged','UserReference','GlobalMemory','DeleteOnClose', |
|
'FilePointerNull','DebugSymbolsLoaded','SetMappedFileIoComplete', |
|
'CollidedFlush','NoChange','filler0','ImageMappedInSystemSpace', |
|
'UserWritable','Accessed','GlobalOnlyPerSession','Rom', |
|
'WriteCombined','filler', |
|
] |
|
|
|
def _decode_section_flags(val): |
|
if val is None: |
|
return '<unreadable>' |
|
set_bits = [name for i, name in enumerate(_MMSECTION_FLAG_BITS) |
|
if (val >> i) & 1 and not name.startswith('filler')] |
|
return ', '.join(set_bits) if set_bits else '(none)' |
|
|
|
|
|
# --- _CONTROL_AREA full layout (amd64, NT62) --- |
|
# 0x00 Segment, 0x08 DereferenceList(16), 0x18 NumberOfSectionReferences(4), |
|
# 0x1C NumberOfPfnReferences(4), 0x20 NumberOfMappedViews(4), |
|
# 0x24 NumberOfSystemCacheViews(4), 0x28 NumberOfUserReferences(4), |
|
# 0x2C u.LongFlags(4), 0x30 FilePointer(8), 0x38 WaitingForDeletion(8), |
|
# 0x40 ModifiedWriteCount(2), 0x42 FlushInProgressCount(2), |
|
# 0x44 WritableUserReferences(4), 0x48 QuadwordPad(4) [+pad to 0x50]. |
|
CA_SEGMENT = 0x00 |
|
CA_DEREF_LIST = 0x08 |
|
CA_NUM_SEC_REFS = 0x18 |
|
CA_NUM_PFN_REFS = 0x1C |
|
CA_NUM_MAPPED = 0x20 |
|
CA_NUM_SYS_CACHE = 0x24 |
|
CA_NUM_USER_REFS = 0x28 |
|
CA_LONG_FLAGS = 0x2C |
|
CA_FILE_POINTER = 0x30 |
|
CA_WAITING_DEL = 0x38 |
|
CA_MODIFIED_WC = 0x40 |
|
CA_FLUSH_IP = 0x42 |
|
CA_WRITABLE_UREFS = 0x44 |
|
|
|
# --- _SECTION_OBJECT_POINTERS --- |
|
SOP_DATA_SECTION = 0x00 |
|
SOP_SHARED_CACHE = 0x08 |
|
SOP_IMAGE_SECTION = 0x10 |
|
|
|
# --- _ROS_SHARED_CACHE_MAP layout (amd64, computed from cc.h:170) --- |
|
# NodeTypeCode(2)+NodeByteSize(2)+OpenCount(4)=8; |
|
# FileSize@0x08(8); BcbList@0x10(16); SectionSize@0x20(8); |
|
# ValidDataLength@0x28(8); FileObject@0x30(8); DirtyPages@0x38(4); |
|
# SharedCacheMapLinks@0x40(16); Flags@0x50(4); pad(4); Section@0x58(8); |
|
# CreateEvent@0x60(8); Callbacks@0x68(8); LazyWriteContext@0x70(8); |
|
# PrivateList@0x78(16); DirtyPageThreshold@0x88(4); pad(4); |
|
# BcbSpinLock@0x90(8); PrivateCacheMap@0x98(0x68); |
|
# CacheMapVacbListHead@0x100(16); ... |
|
SCM_NODE_TYPE = 0x00 |
|
SCM_NODE_BYTE_SIZE = 0x02 |
|
SCM_OPEN_COUNT = 0x04 |
|
SCM_FILE_SIZE = 0x08 |
|
SCM_BCB_LIST = 0x10 |
|
SCM_SECTION_SIZE = 0x20 |
|
SCM_VALID_DATA_LEN = 0x28 |
|
SCM_FILE_OBJECT = 0x30 |
|
SCM_DIRTY_PAGES = 0x38 |
|
SCM_LINKS = 0x40 |
|
SCM_FLAGS = 0x50 |
|
SCM_SECTION = 0x58 |
|
SCM_CREATE_EVENT = 0x60 |
|
SCM_CALLBACKS = 0x68 |
|
SCM_LAZY_WRITE_CTX = 0x70 |
|
SCM_PRIVATE_LIST = 0x78 |
|
SCM_DIRTY_THRESHOLD = 0x88 |
|
SCM_BCB_SPINLOCK = 0x90 |
|
SCM_PRIVATE_CACHEMAP = 0x98 |
|
SCM_VACB_LIST_HEAD = 0x100 |
|
|
|
# --- _ROS_VACB layout (amd64, computed from cc.h:207) --- |
|
# BaseAddress(8)@0; Dirty(1)@8; PageOut(1)@9; pad(2); MappedCount(4)@0xC; |
|
# CacheMapVacbListEntry@0x10(16); DirtyVacbListEntry@0x20(16); |
|
# VacbLruListEntry@0x30(16); FileOffset@0x40(8); ReferenceCount@0x48(4); |
|
# pad(4); SharedCacheMap@0x50(8). |
|
VACB_BASE_ADDRESS = 0x00 |
|
VACB_DIRTY = 0x08 |
|
VACB_PAGE_OUT = 0x09 |
|
VACB_MAPPED_COUNT = 0x0C |
|
VACB_LIST_ENTRY = 0x10 # CacheMapVacbListEntry (Flink) |
|
VACB_DIRTY_LIST_ENTRY = 0x20 |
|
VACB_LRU_LIST_ENTRY = 0x30 |
|
VACB_FILE_OFFSET = 0x40 |
|
VACB_REFCOUNT = 0x48 |
|
VACB_SHARED_CACHE_MAP = 0x50 |
|
|
|
# --- _MMVAD_SHORT layout (amd64, mmtypes.h:788) --- |
|
# u1(Parent or Balance)(8)@0; LeftChild(8)@8; RightChild(8)@0x10; |
|
# StartingVpn(8)@0x18; EndingVpn(8)@0x20; u.LongFlags(8)@0x28. |
|
VAD_PARENT = 0x00 |
|
VAD_LEFT = 0x08 |
|
VAD_RIGHT = 0x10 |
|
VAD_STARTING_VPN = 0x18 |
|
VAD_ENDING_VPN = 0x20 |
|
VAD_LONG_FLAGS = 0x28 |
|
# Long _MMVAD also has at +0x30 ControlArea, +0x38 FirstPrototypePte, |
|
# +0x40 LastContiguousPte, +0x48 LongFlags2. |
|
VAD_CONTROL_AREA = 0x30 |
|
VAD_FIRST_PROTO_PTE = 0x38 |
|
VAD_LAST_CONTIG_PTE = 0x40 |
|
|
|
VAD_TYPE_NAMES = { |
|
0: 'VadNone', 1: 'VadDevicePhysicalMemory', 2: 'VadImageMap', |
|
3: 'VadAwe', 4: 'VadWriteWatch', 5: 'VadLargePages', |
|
6: 'VadRotatePhysical', 7: 'VadLargePageSection', |
|
} |
|
|
|
|
|
def _decode_mmvad_flags(longflags): |
|
"""Decode MMVAD_FLAGS (amd64). See mmtypes.h:686. |
|
CommitCharge:51 NoChange:1 VadType:3 MemCommit:1 Protection:5 Spare:2 PrivateMemory:1""" |
|
if longflags is None: |
|
return {} |
|
return { |
|
'CommitCharge' : longflags & ((1 << 51) - 1), |
|
'NoChange' : (longflags >> 51) & 1, |
|
'VadType' : (longflags >> 52) & 7, |
|
'MemCommit' : (longflags >> 55) & 1, |
|
'Protection' : (longflags >> 56) & 0x1F, |
|
'PrivateMemory' : (longflags >> 63) & 1, |
|
} |
|
|
|
|
|
def _try_eval_addr(expr): |
|
"""Resolve a kernel symbol or expression to an integer address; None on failure.""" |
|
try: |
|
return int(gdb.parse_and_eval(expr)) & 0xFFFFFFFFFFFFFFFF |
|
except Exception: |
|
return None |
|
|
|
|
|
def _maybe_lookup_field_offset(typename, fieldname): |
|
"""Return offset of fieldname in typename via gdb type info; None if unavailable.""" |
|
try: |
|
t = gdb.lookup_type(typename) |
|
for f in t.fields(): |
|
if f.name == fieldname: |
|
return f.bitpos // 8 |
|
except Exception: |
|
pass |
|
return None |
|
|
|
|
|
def _read_unicode_string_at(addr): |
|
"""UNICODE_STRING { USHORT Length; USHORT MaxLen; ULONG pad; PWSTR Buffer; }""" |
|
if not addr: |
|
return None |
|
return read_unicode_string(addr) |
|
|
|
|
|
# ============================================================ |
|
# ros-section <addr> |
|
# ============================================================ |
|
class ReactosSection(gdb.Command): |
|
"""Dump _SECTION (the user object). Auto-chains into Segment. |
|
Usage: ros-section <addr>""" |
|
|
|
def __init__(self): |
|
super().__init__("ros-section", gdb.COMMAND_USER) |
|
|
|
def invoke(self, arg, from_tty): |
|
if not arg.strip(): |
|
print("Usage: ros-section <SECTION_addr>") |
|
return |
|
try: |
|
addr = int(arg.strip(), 0) |
|
except Exception: |
|
print(f"Bad address: {arg.strip()}") |
|
return |
|
try: |
|
seg = read_u64(addr + SECTION_SEGMENT) |
|
size = read_u64(addr + SECTION_SIZEOF_SECTION) |
|
flags = read_u32(addr + SECTION_FLAGS) |
|
init_protect = read_u32(addr + SECTION_INITIAL_PROTECTION) |
|
if seg is None: |
|
print(f"Cannot read SECTION at 0x{addr:x}") |
|
return |
|
print(f"=== Section @ 0x{addr:016x} ===") |
|
print(f" Segment = 0x{seg:016x}") |
|
if size is not None: |
|
print(f" SizeOfSection = 0x{size:x} ({size} bytes)") |
|
print(f" InitialPageProtection = 0x{init_protect:x}") |
|
print(f" u.LongFlags = 0x{flags:x}") |
|
print(f" flags set : {_decode_section_flags(flags)}") |
|
print(f" Use: ros-segment 0x{seg:x}") |
|
except Exception as e: |
|
print(f"ros-section error: {e}") |
|
|
|
ReactosSection() |
|
|
|
|
|
# ============================================================ |
|
# ros-segment <addr> |
|
# ============================================================ |
|
class ReactosSegment(gdb.Command): |
|
"""Dump _SEGMENT. |
|
Usage: ros-segment <addr>""" |
|
|
|
def __init__(self): |
|
super().__init__("ros-segment", gdb.COMMAND_USER) |
|
|
|
def invoke(self, arg, from_tty): |
|
if not arg.strip(): |
|
print("Usage: ros-segment <SEGMENT_addr>") |
|
return |
|
try: |
|
addr = int(arg.strip(), 0) |
|
except Exception: |
|
print(f"Bad address: {arg.strip()}") |
|
return |
|
try: |
|
ca = read_u64(addr + SEGMENT_CONTROL_AREA) |
|
total_ptes = read_u32(addr + SEGMENT_TOTAL_PTES) |
|
non_ext_ptes = read_u32(addr + SEGMENT_NON_EXTENDED_PTES) |
|
seg_size = read_u64(addr + SEGMENT_SIZE_OF_SEGMENT) |
|
pte_template = read_u64(addr + SEGMENT_PTE_TEMPLATE) |
|
committed = read_u32(addr + SEGMENT_COMMITTED_PAGES) |
|
seg_flags = read_u32(addr + SEGMENT_FLAGS_OFF) |
|
based_addr = read_u64(addr + SEGMENT_BASED_ADDRESS) |
|
proto_pte = read_u64(addr + SEGMENT_PROTOTYPE_PTE) |
|
if ca is None: |
|
print(f"Cannot read SEGMENT at 0x{addr:x}") |
|
return |
|
print(f"=== Segment @ 0x{addr:016x} ===") |
|
print(f" ControlArea = 0x{ca:016x}") |
|
print(f" TotalNumberOfPtes = {total_ptes}") |
|
print(f" NonExtendedPtes = {non_ext_ptes}") |
|
print(f" SizeOfSegment = 0x{seg_size:x} ({seg_size} bytes)") |
|
print(f" SegmentPteTemplate = 0x{pte_template:016x}") |
|
_print_pte(addr + SEGMENT_PTE_TEMPLATE, pte_template, indent=' ') |
|
print(f" NumberOfCommittedPages= {committed}") |
|
print(f" SegmentFlags = 0x{seg_flags:x}") |
|
print(f" BasedAddress = 0x{based_addr:016x}") |
|
print(f" PrototypePte = 0x{proto_pte:016x} (head of proto array)") |
|
print(f" Use: ros-controlarea 0x{ca:x}") |
|
print(f" ros-proto-walk 0x{ca:x}") |
|
except Exception as e: |
|
print(f"ros-segment error: {e}") |
|
|
|
ReactosSegment() |
|
|
|
|
|
# ============================================================ |
|
# ros-controlarea <addr> |
|
# ============================================================ |
|
class ReactosControlArea(gdb.Command): |
|
"""Dump _CONTROL_AREA. Auto-chains into the embedded SUBSECTION. |
|
Usage: ros-controlarea <addr>""" |
|
|
|
def __init__(self): |
|
super().__init__("ros-controlarea", gdb.COMMAND_USER) |
|
|
|
def invoke(self, arg, from_tty): |
|
if not arg.strip(): |
|
print("Usage: ros-controlarea <CONTROL_AREA_addr>") |
|
return |
|
try: |
|
ca = int(arg.strip(), 0) |
|
except Exception: |
|
print(f"Bad address: {arg.strip()}") |
|
return |
|
try: |
|
seg = read_u64(ca + CA_SEGMENT) |
|
n_sec_refs = read_u32(ca + CA_NUM_SEC_REFS) |
|
n_pfn_refs = read_u32(ca + CA_NUM_PFN_REFS) |
|
n_mapped = read_u32(ca + CA_NUM_MAPPED) |
|
n_sys_cache = read_u32(ca + CA_NUM_SYS_CACHE) |
|
n_user_refs = read_u32(ca + CA_NUM_USER_REFS) |
|
flags = read_u32(ca + CA_LONG_FLAGS) |
|
file_obj = read_u64(ca + CA_FILE_POINTER) |
|
waiting_del = read_u64(ca + CA_WAITING_DEL) |
|
mod_wc = read_u16(ca + CA_MODIFIED_WC) |
|
flush_ip = read_u16(ca + CA_FLUSH_IP) |
|
writable_ur = read_u32(ca + CA_WRITABLE_UREFS) |
|
if seg is None: |
|
print(f"Cannot read CONTROL_AREA at 0x{ca:x}") |
|
return |
|
print(f"=== ControlArea @ 0x{ca:016x} ===") |
|
print(f" Segment = 0x{seg:016x}") |
|
print(f" NumberOfSectionReferences= {n_sec_refs}") |
|
print(f" NumberOfPfnReferences = {n_pfn_refs}") |
|
print(f" NumberOfMappedViews = {n_mapped}") |
|
print(f" NumberOfSystemCacheViews = {n_sys_cache}") |
|
print(f" NumberOfUserReferences = {n_user_refs}") |
|
print(f" WritableUserReferences = {writable_ur}") |
|
print(f" ModifiedWriteCount = {mod_wc}") |
|
print(f" FlushInProgressCount = {flush_ip}") |
|
print(f" WaitingForDeletion = 0x{waiting_del:016x}") |
|
print(f" FilePointer = 0x{file_obj:016x}") |
|
# Try to peek at FILE_OBJECT.FileName (offset 0x58 amd64) if non-NULL |
|
if file_obj: |
|
try: |
|
fname_off = _maybe_lookup_field_offset("struct _FILE_OBJECT", |
|
"FileName") |
|
if fname_off is None: |
|
fname_off = 0x58 # amd64 default |
|
name = _read_unicode_string_at(file_obj + fname_off) |
|
if name: |
|
print(f" FilePointer->FileName = \"{name}\"") |
|
read_off = _maybe_lookup_field_offset("struct _FILE_OBJECT", |
|
"ReadAccess") |
|
if read_off is not None: |
|
ra = read_u8(file_obj + read_off) |
|
wa = read_u8(file_obj + read_off + 1) |
|
if ra is not None: |
|
print(f" FilePointer->ReadAccess={ra} WriteAccess={wa}") |
|
except Exception: |
|
pass |
|
print(f" u.LongFlags = 0x{flags:x}") |
|
print(f" flags set : {_decode_section_flags(flags)}") |
|
sub = ca + 0x50 # SIZEOF_CONTROL_AREA — first SUBSECTION immediately follows |
|
print(f" embedded SUBSECTION at = 0x{sub:016x}") |
|
print(f" Use: ros-subsection 0x{sub:x}") |
|
print(f" ros-segment 0x{seg:x}") |
|
print(f" ros-proto-walk 0x{ca:x}") |
|
except Exception as e: |
|
print(f"ros-controlarea error: {e}") |
|
|
|
ReactosControlArea() |
|
|
|
|
|
# ============================================================ |
|
# ros-subsection <addr> |
|
# ============================================================ |
|
class ReactosSubsection(gdb.Command): |
|
"""Dump _SUBSECTION (mmtypes.h:571). |
|
Usage: ros-subsection <addr>""" |
|
|
|
def __init__(self): |
|
super().__init__("ros-subsection", gdb.COMMAND_USER) |
|
|
|
def invoke(self, arg, from_tty): |
|
if not arg.strip(): |
|
print("Usage: ros-subsection <SUBSECTION_addr>") |
|
return |
|
try: |
|
sub = int(arg.strip(), 0) |
|
except Exception: |
|
print(f"Bad address: {arg.strip()}") |
|
return |
|
try: |
|
ca = read_u64(sub + SUBSECTION_CONTROL_AREA) |
|
sub_flags = read_u32(sub + SUBSECTION_FLAGS) |
|
start_sec = read_u32(sub + SUBSECTION_STARTING_SECTOR) |
|
full_secs = read_u32(sub + SUBSECTION_NUM_FULL_SECTORS) |
|
sub_base = read_u64(sub + SUBSECTION_BASE) |
|
unused = read_u32(sub + SUBSECTION_UNUSED_PTES) |
|
count = read_u32(sub + SUBSECTION_PTES_IN) |
|
nxt = read_u64(sub + SUBSECTION_NEXT) |
|
if ca is None: |
|
print(f"Cannot read SUBSECTION at 0x{sub:x}") |
|
return |
|
print(f"=== Subsection @ 0x{sub:016x} ===") |
|
print(f" ControlArea = 0x{ca:016x}") |
|
print(f" NextSubsection = 0x{nxt:016x}") |
|
print(f" SubsectionBase = 0x{sub_base:016x} (proto-PTE array)") |
|
print(f" PtesInSubsection = {count}") |
|
print(f" UnusedPtes = {unused}") |
|
print(f" StartingSector = {start_sec}") |
|
print(f" NumberOfFullSectors = {full_secs}") |
|
print(f" SubsectionFlags = 0x{sub_flags:x}") |
|
# Decode MMSUBSECTION_FLAGS bits (mmtypes.h:497) |
|
ro = sub_flags & 1 |
|
rw = (sub_flags >> 1) & 1 |
|
stat = (sub_flags >> 2) & 1 |
|
gmem = (sub_flags >> 3) & 1 |
|
prot = (sub_flags >> 4) & 0x1F |
|
print(f" ReadOnly={ro} ReadWrite={rw} Static={stat} GlobalMemory={gmem}" |
|
f" Protection=0x{prot:x}") |
|
print(f" Use: ros-proto-walk 0x{ca:x} {count}") |
|
print(f" ros-controlarea 0x{ca:x}") |
|
except Exception as e: |
|
print(f"ros-subsection error: {e}") |
|
|
|
ReactosSubsection() |
|
|
|
|
|
# ============================================================ |
|
# ros-secptr <addr> |
|
# ============================================================ |
|
class ReactosSecPtr(gdb.Command): |
|
"""Dump _SECTION_OBJECT_POINTERS (xdk/iotypes.h:1794). |
|
Auto-chains into ros-cachemap if SharedCacheMap is non-NULL. |
|
Usage: ros-secptr <addr>""" |
|
|
|
def __init__(self): |
|
super().__init__("ros-secptr", gdb.COMMAND_USER) |
|
|
|
def invoke(self, arg, from_tty): |
|
if not arg.strip(): |
|
print("Usage: ros-secptr <SECTION_OBJECT_POINTERS_addr>") |
|
return |
|
try: |
|
sop = int(arg.strip(), 0) |
|
except Exception: |
|
print(f"Bad address: {arg.strip()}") |
|
return |
|
try: |
|
data_sect = read_u64(sop + SOP_DATA_SECTION) |
|
shared_cm = read_u64(sop + SOP_SHARED_CACHE) |
|
image_sect = read_u64(sop + SOP_IMAGE_SECTION) |
|
if data_sect is None: |
|
print(f"Cannot read SECTION_OBJECT_POINTERS at 0x{sop:x}") |
|
return |
|
print(f"=== SectionObjectPointers @ 0x{sop:016x} ===") |
|
print(f" DataSectionObject = 0x{data_sect:016x} (CONTROL_AREA*)") |
|
print(f" SharedCacheMap = 0x{shared_cm:016x} (ROS_SHARED_CACHE_MAP*)") |
|
print(f" ImageSectionObject = 0x{image_sect:016x} (CONTROL_AREA*)") |
|
if data_sect: |
|
print(f" Use: ros-controlarea 0x{data_sect:x}") |
|
if image_sect and image_sect != data_sect: |
|
print(f" ros-controlarea 0x{image_sect:x}") |
|
if shared_cm: |
|
print(f" ros-cachemap 0x{shared_cm:x}") |
|
except Exception as e: |
|
print(f"ros-secptr error: {e}") |
|
|
|
ReactosSecPtr() |
|
|
|
|
|
# ============================================================ |
|
# ros-cachemap <addr> |
|
# ============================================================ |
|
class ReactosCacheMap(gdb.Command): |
|
"""Dump _ROS_SHARED_CACHE_MAP (cc.h:170). |
|
Usage: ros-cachemap <addr>""" |
|
|
|
def __init__(self): |
|
super().__init__("ros-cachemap", gdb.COMMAND_USER) |
|
|
|
def invoke(self, arg, from_tty): |
|
if not arg.strip(): |
|
print("Usage: ros-cachemap <ROS_SHARED_CACHE_MAP_addr>") |
|
return |
|
try: |
|
scm = int(arg.strip(), 0) |
|
except Exception: |
|
print(f"Bad address: {arg.strip()}") |
|
return |
|
try: |
|
ntype = read_u16(scm + SCM_NODE_TYPE) |
|
nbsize = read_u16(scm + SCM_NODE_BYTE_SIZE) |
|
opencount = read_u32(scm + SCM_OPEN_COUNT) |
|
file_size = read_u64(scm + SCM_FILE_SIZE) |
|
sect_size = read_u64(scm + SCM_SECTION_SIZE) |
|
valid_dl = read_u64(scm + SCM_VALID_DATA_LEN) |
|
file_obj = read_u64(scm + SCM_FILE_OBJECT) |
|
dirty = read_u32(scm + SCM_DIRTY_PAGES) |
|
flags = read_u32(scm + SCM_FLAGS) |
|
section = read_u64(scm + SCM_SECTION) |
|
bcb_lock = read_u64(scm + SCM_BCB_SPINLOCK) |
|
vacb_flink= read_u64(scm + SCM_VACB_LIST_HEAD) |
|
vacb_blink= read_u64(scm + SCM_VACB_LIST_HEAD + 8) |
|
link_f = read_u64(scm + SCM_LINKS) |
|
link_b = read_u64(scm + SCM_LINKS + 8) |
|
if file_obj is None: |
|
print(f"Cannot read ROS_SHARED_CACHE_MAP at 0x{scm:x}") |
|
return |
|
print(f"=== SharedCacheMap @ 0x{scm:016x} ===") |
|
print(f" NodeTypeCode/ByteSize = 0x{ntype:04x} / 0x{nbsize:04x}") |
|
print(f" OpenCount = {opencount}") |
|
print(f" FileSize = 0x{file_size:x}") |
|
print(f" SectionSize = 0x{sect_size:x}") |
|
print(f" ValidDataLength = 0x{valid_dl:x}") |
|
print(f" FileObject = 0x{file_obj:016x}") |
|
if file_obj: |
|
try: |
|
fname_off = _maybe_lookup_field_offset("struct _FILE_OBJECT", |
|
"FileName") or 0x58 |
|
name = _read_unicode_string_at(file_obj + fname_off) |
|
if name: |
|
print(f" FileObject->FileName = \"{name}\"") |
|
except Exception: |
|
pass |
|
print(f" DirtyPages = {dirty}") |
|
print(f" Flags = 0x{flags:x}") |
|
fbits = [] |
|
if flags & 0x1: fbits.append('READAHEAD_DISABLED') |
|
if flags & 0x2: fbits.append('WRITEBEHIND_DISABLED') |
|
if flags & 0x4: fbits.append('IN_CREATION') |
|
if flags & 0x8: fbits.append('IN_LAZYWRITE') |
|
if fbits: |
|
print(f" flags set : {', '.join(fbits)}") |
|
print(f" Section = 0x{section:016x}") |
|
print(f" BcbSpinLock = 0x{bcb_lock:x}") |
|
print(f" CleanSharedCacheMapLink= F=0x{link_f:x} B=0x{link_b:x}") |
|
print(f" CacheMapVacbListHead = F=0x{vacb_flink:016x} B=0x{vacb_blink:016x}") |
|
empty = (vacb_flink == scm + SCM_VACB_LIST_HEAD) |
|
if empty: |
|
print(f" (VACB list is empty — list head points to itself)") |
|
else: |
|
print(f" Use: ros-vacb-list 0x{scm:x}") |
|
print(f" ros-section 0x{section:x}") |
|
except Exception as e: |
|
print(f"ros-cachemap error: {e}") |
|
|
|
ReactosCacheMap() |
|
|
|
|
|
# ============================================================ |
|
# ros-vacb <addr> |
|
# ============================================================ |
|
class ReactosVacb(gdb.Command): |
|
"""Dump one _ROS_VACB (cc.h:207). |
|
Usage: ros-vacb <addr>""" |
|
|
|
def __init__(self): |
|
super().__init__("ros-vacb", gdb.COMMAND_USER) |
|
|
|
def invoke(self, arg, from_tty): |
|
if not arg.strip(): |
|
print("Usage: ros-vacb <ROS_VACB_addr>") |
|
return |
|
try: |
|
v = int(arg.strip(), 0) |
|
except Exception: |
|
print(f"Bad address: {arg.strip()}") |
|
return |
|
try: |
|
base = read_u64(v + VACB_BASE_ADDRESS) |
|
dirty = read_u8(v + VACB_DIRTY) |
|
pageout = read_u8(v + VACB_PAGE_OUT) |
|
mapped = read_u32(v + VACB_MAPPED_COUNT) |
|
le_f = read_u64(v + VACB_LIST_ENTRY) |
|
le_b = read_u64(v + VACB_LIST_ENTRY + 8) |
|
foffset = read_u64(v + VACB_FILE_OFFSET) |
|
refcnt = read_u32(v + VACB_REFCOUNT) |
|
scm = read_u64(v + VACB_SHARED_CACHE_MAP) |
|
if base is None: |
|
print(f"Cannot read ROS_VACB at 0x{v:x}") |
|
return |
|
print(f"=== Vacb @ 0x{v:016x} ===") |
|
print(f" BaseAddress = 0x{base:016x}") |
|
print(f" FileOffset = 0x{foffset:x}") |
|
print(f" ReferenceCount = {refcnt}") |
|
print(f" MappedCount = {mapped}") |
|
print(f" Dirty = {dirty}") |
|
print(f" PageOut = {pageout}") |
|
print(f" SharedCacheMap = 0x{scm:016x}") |
|
print(f" CacheMapVacbListEntry = F=0x{le_f:x} B=0x{le_b:x}") |
|
if scm: |
|
print(f" Use: ros-cachemap 0x{scm:x}") |
|
print(f" ros-mapping-walk 0x{base:x} <num_pages>") |
|
except Exception as e: |
|
print(f"ros-vacb error: {e}") |
|
|
|
ReactosVacb() |
|
|
|
|
|
# ============================================================ |
|
# ros-vacb-list <SharedCacheMap_addr> [max=16] |
|
# ============================================================ |
|
class ReactosVacbList(gdb.Command): |
|
"""Walk CacheMapVacbListHead embedded in a ROS_SHARED_CACHE_MAP. |
|
Usage: ros-vacb-list <SharedCacheMap_addr> [max=16]""" |
|
|
|
def __init__(self): |
|
super().__init__("ros-vacb-list", gdb.COMMAND_USER) |
|
|
|
def invoke(self, arg, from_tty): |
|
args = arg.strip().split() |
|
if len(args) < 1: |
|
print("Usage: ros-vacb-list <SharedCacheMap_addr> [max=16]") |
|
return |
|
try: |
|
scm = int(args[0], 0) |
|
except Exception: |
|
print(f"Bad address: {args[0]}") |
|
return |
|
max_walk = int(args[1], 0) if len(args) >= 2 else 16 |
|
head = scm + SCM_VACB_LIST_HEAD |
|
try: |
|
cur = read_u64(head) |
|
if cur is None: |
|
print(f"Cannot read list head at 0x{head:x}") |
|
return |
|
print(f"=== VACB list of SharedCacheMap @ 0x{scm:016x} ===") |
|
print(f" ListHead @ 0x{head:016x} -> Flink=0x{cur:x}") |
|
print(f" {'#':>3s} {'vacb':>18s} {'base':>18s} {'fileoff':>14s} " |
|
f"{'ref':>4s} {'map':>4s} {'D':>1s} {'PO':>2s}") |
|
i = 0 |
|
visited = set() |
|
while cur and cur != head and i < max_walk: |
|
if cur in visited: |
|
print(f" ** loop detected at 0x{cur:x}") |
|
break |
|
visited.add(cur) |
|
# cur points at the embedded LIST_ENTRY (CacheMapVacbListEntry) |
|
vacb = cur - VACB_LIST_ENTRY |
|
base = read_u64(vacb + VACB_BASE_ADDRESS) |
|
foff = read_u64(vacb + VACB_FILE_OFFSET) |
|
refcnt = read_u32(vacb + VACB_REFCOUNT) |
|
mapped = read_u32(vacb + VACB_MAPPED_COUNT) |
|
dirty = read_u8(vacb + VACB_DIRTY) |
|
pageout = read_u8(vacb + VACB_PAGE_OUT) |
|
if base is None: |
|
print(f" {i:3d} 0x{vacb:016x} <unreadable>") |
|
break |
|
print(f" {i:3d} 0x{vacb:016x} 0x{base:016x} 0x{foff:>12x} " |
|
f"{refcnt:>4d} {mapped:>4d} {dirty:>1d} {pageout:>2d}") |
|
nxt = read_u64(cur) |
|
if nxt is None: |
|
break |
|
cur = nxt |
|
i += 1 |
|
if i == 0 and cur == head: |
|
print(" (list is empty)") |
|
elif i >= max_walk: |
|
print(f" ... truncated at {max_walk} entries") |
|
except Exception as e: |
|
print(f"ros-vacb-list error: {e}") |
|
|
|
ReactosVacbList() |
|
|
|
|
|
# ============================================================ |
|
# ros-vad <addr> |
|
# ============================================================ |
|
class ReactosVad(gdb.Command): |
|
"""Dump _MMVAD_SHORT and (if applicable) _MMVAD long-form fields. |
|
Usage: ros-vad <vad_addr>""" |
|
|
|
def __init__(self): |
|
super().__init__("ros-vad", gdb.COMMAND_USER) |
|
|
|
def invoke(self, arg, from_tty): |
|
if not arg.strip(): |
|
print("Usage: ros-vad <vad_addr>") |
|
return |
|
try: |
|
v = int(arg.strip(), 0) |
|
except Exception: |
|
print(f"Bad address: {arg.strip()}") |
|
return |
|
try: |
|
parent_raw = read_u64(v + VAD_PARENT) |
|
left = read_u64(v + VAD_LEFT) |
|
right = read_u64(v + VAD_RIGHT) |
|
start_vpn = read_u64(v + VAD_STARTING_VPN) |
|
end_vpn = read_u64(v + VAD_ENDING_VPN) |
|
longflags = read_u64(v + VAD_LONG_FLAGS) |
|
if parent_raw is None: |
|
print(f"Cannot read VAD at 0x{v:x}") |
|
return |
|
balance = parent_raw & 3 |
|
parent = parent_raw & ~3 |
|
flags = _decode_mmvad_flags(longflags) |
|
vt_name = VAD_TYPE_NAMES.get(flags.get('VadType'), '?') |
|
va_lo = (start_vpn or 0) << 12 |
|
va_hi = (((end_vpn or 0) + 1) << 12) - 1 |
|
print(f"=== Vad @ 0x{v:016x} ===") |
|
print(f" Parent (u1) = 0x{parent:016x} Balance={balance}") |
|
print(f" LeftChild = 0x{left:016x}") |
|
print(f" RightChild = 0x{right:016x}") |
|
print(f" StartingVpn = 0x{start_vpn:x} EndingVpn=0x{end_vpn:x}") |
|
print(f" VA range = 0x{va_lo:x} .. 0x{va_hi:x}") |
|
print(f" u.LongFlags = 0x{longflags:x}") |
|
print(f" CommitCharge = 0x{flags['CommitCharge']:x}") |
|
print(f" Protection = 0x{flags['Protection']:x}") |
|
print(f" VadType = {flags['VadType']} ({vt_name})") |
|
print(f" NoChange = {flags['NoChange']}") |
|
print(f" MemCommit = {flags['MemCommit']}") |
|
print(f" PrivateMemory = {flags['PrivateMemory']}") |
|
# If not PrivateMemory, this is a long VAD with ControlArea + protos. |
|
if flags['PrivateMemory'] == 0: |
|
ca = read_u64(v + VAD_CONTROL_AREA) |
|
first_proto= read_u64(v + VAD_FIRST_PROTO_PTE) |
|
last_contig= read_u64(v + VAD_LAST_CONTIG_PTE) |
|
if ca is not None: |
|
print(f" ControlArea = 0x{ca:016x}") |
|
print(f" FirstPrototypePte = 0x{first_proto:016x}") |
|
print(f" LastContiguousPte = 0x{last_contig:016x}") |
|
if ca: |
|
print(f" Use: ros-controlarea 0x{ca:x}") |
|
except Exception as e: |
|
print(f"ros-vad error: {e}") |
|
|
|
ReactosVad() |
|
|
|
|
|
# ============================================================ |
|
# ros-vad-tree <eprocess_or_avltable_addr> [max=64] |
|
# ============================================================ |
|
def _walk_vad_tree(node, out, depth, max_nodes): |
|
if not node or len(out) >= max_nodes: |
|
return |
|
if depth > 64: |
|
return |
|
out.append(node) |
|
left = read_u64(node + VAD_LEFT) |
|
right = read_u64(node + VAD_RIGHT) |
|
if left: |
|
_walk_vad_tree(left, out, depth + 1, max_nodes) |
|
if right: |
|
_walk_vad_tree(right, out, depth + 1, max_nodes) |
|
|
|
|
|
class ReactosVadTree(gdb.Command): |
|
"""Walk an EPROCESS's VAD AVL tree (or any _MM_AVL_TABLE root). |
|
Usage: ros-vad-tree <eprocess_or_table_addr> [max=64]""" |
|
|
|
def __init__(self): |
|
super().__init__("ros-vad-tree", gdb.COMMAND_USER) |
|
|
|
def invoke(self, arg, from_tty): |
|
args = arg.strip().split() |
|
if len(args) < 1: |
|
print("Usage: ros-vad-tree <eprocess_or_table_addr> [max=64]") |
|
return |
|
try: |
|
target = int(args[0], 0) |
|
except Exception: |
|
print(f"Bad address: {args[0]}") |
|
return |
|
max_nodes = int(args[1], 0) if len(args) >= 2 else 64 |
|
try: |
|
# Try to interpret as EPROCESS by computing VadRoot offset via gdb types. |
|
vad_root_off = _maybe_lookup_field_offset("struct _EPROCESS", "VadRoot") |
|
avl = None |
|
if vad_root_off is not None: |
|
avl = target + vad_root_off |
|
print(f" Treating 0x{target:x} as EPROCESS, VadRoot @ +0x{vad_root_off:x}" |
|
f" -> 0x{avl:x}") |
|
else: |
|
avl = target |
|
print(f" Treating 0x{target:x} as raw _MM_AVL_TABLE") |
|
# _MM_AVL_TABLE.BalancedRoot is a _MMADDRESS_NODE; the actual tree root |
|
# is BalancedRoot.RightChild (per ARM3 convention). |
|
balanced_right = read_u64(avl + 0x10) # MMADDRESS_NODE.RightChild |
|
if balanced_right is None: |
|
print(f" Cannot read BalancedRoot at 0x{avl:x}") |
|
return |
|
if not balanced_right: |
|
print(f" Tree is empty (BalancedRoot.RightChild == NULL)") |
|
return |
|
print(f" BalancedRoot.RightChild = 0x{balanced_right:x}") |
|
nodes = [] |
|
_walk_vad_tree(balanced_right, nodes, 0, max_nodes) |
|
print(f" walked {len(nodes)} VAD node(s):") |
|
print(f" {'idx':>4s} {'vad':>18s} {'va_start':>18s} " |
|
f"{'va_end':>18s} type protect") |
|
for i, n in enumerate(nodes): |
|
sv = read_u64(n + VAD_STARTING_VPN) or 0 |
|
ev = read_u64(n + VAD_ENDING_VPN) or 0 |
|
lf = read_u64(n + VAD_LONG_FLAGS) or 0 |
|
fl = _decode_mmvad_flags(lf) |
|
print(f" {i:4d} 0x{n:016x} 0x{sv<<12:016x} " |
|
f"0x{((ev+1)<<12)-1:016x} " |
|
f"{fl['VadType']} 0x{fl['Protection']:x}") |
|
except Exception as e: |
|
print(f"ros-vad-tree error: {e}") |
|
|
|
ReactosVadTree() |
|
|
|
|
|
# ============================================================ |
|
# ros-mmsupport <process_addr> |
|
# ============================================================ |
|
class ReactosMmSupport(gdb.Command): |
|
"""Dump _MMSUPPORT (mmtypes.h:922). |
|
Pass a pointer directly to MMSUPPORT, or an EPROCESS (we'll find Vm via gdb types). |
|
Usage: ros-mmsupport <addr>""" |
|
|
|
def __init__(self): |
|
super().__init__("ros-mmsupport", gdb.COMMAND_USER) |
|
|
|
def invoke(self, arg, from_tty): |
|
if not arg.strip(): |
|
print("Usage: ros-mmsupport <addr>") |
|
return |
|
try: |
|
target = int(arg.strip(), 0) |
|
except Exception: |
|
print(f"Bad address: {arg.strip()}") |
|
return |
|
try: |
|
# If it looks like an EPROCESS, locate the Vm field. |
|
vm_off = _maybe_lookup_field_offset("struct _EPROCESS", "Vm") |
|
mms = target |
|
if vm_off is not None: |
|
# Heuristic: if &EPROCESS->Vm is readable as MMSUPPORT, prefer it. |
|
mms = target + vm_off |
|
print(f" Treating 0x{target:x} as EPROCESS, Vm @ +0x{vm_off:x}" |
|
f" -> 0x{mms:x}") |
|
# NTDDI_LONGHORN+ layout. WorkingSetExpansionLinks(16) + LastTrimpStamp(2) |
|
# + NextPageColor(2) + Flags(4) + PageFaultCount(4) + PeakWorkingSetSize(4) |
|
# + Spare0(4) + MinimumWorkingSetSize(4) + MaximumWorkingSetSize(4) |
|
# = 0x10 + 4 + 4 + 4 + 4 + 4 + 4 = 0x28. Then VmWorkingSetList(8) @ 0x28. |
|
# Then Claim(4)@0x30, Spare(4)@0x34, WorkingSetPrivateSize(4)@0x38, |
|
# WorkingSetSizeOverhead(4)@0x3C, WorkingSetSize(4)@0x40. |
|
wsel_f = read_u64(mms + 0x00) |
|
wsel_b = read_u64(mms + 0x08) |
|
flags = read_u32(mms + 0x14) |
|
pfcount = read_u32(mms + 0x18) |
|
peak_ws = read_u32(mms + 0x1C) |
|
min_ws = read_u32(mms + 0x24) |
|
max_ws = read_u32(mms + 0x28) |
|
wsl = read_u64(mms + 0x30) |
|
ws_size = read_u32(mms + 0x48) |
|
if wsel_f is None: |
|
print(f"Cannot read MMSUPPORT at 0x{mms:x}") |
|
return |
|
print(f"=== MmSupport @ 0x{mms:016x} ===") |
|
print(f" WorkingSetExpansionLinks F=0x{wsel_f:x} B=0x{wsel_b:x}") |
|
print(f" Flags = 0x{flags:x}") |
|
ssp = flags & 1 |
|
bt = (flags >> 1) & 1 |
|
sl = (flags >> 2) & 1 |
|
mp = (flags >> 8) & 0xFF |
|
print(f" SessionSpace={ssp} BeingTrimmed={bt} SessionLeader={sl}" |
|
f" MemoryPriority={mp}") |
|
print(f" PageFaultCount = {pfcount}") |
|
print(f" PeakWorkingSetSize = {peak_ws}") |
|
print(f" Min/MaxWorkingSetSize = {min_ws} / {max_ws}") |
|
print(f" WorkingSetSize = {ws_size}") |
|
print(f" VmWorkingSetList = 0x{wsl:016x}") |
|
except Exception as e: |
|
print(f"ros-mmsupport error: {e}") |
|
|
|
ReactosMmSupport() |
|
|
|
|
|
# ============================================================ |
|
# ros-mmranges |
|
# ============================================================ |
|
class ReactosMmRanges(gdb.Command): |
|
"""One-screen map of ARM3 virtual ranges. Resolves symbols at runtime. |
|
Usage: ros-mmranges""" |
|
|
|
def __init__(self): |
|
super().__init__("ros-mmranges", gdb.COMMAND_USER) |
|
|
|
def invoke(self, arg, from_tty): |
|
print("=== ARM3 virtual address map (resolved from ntoskrnl symbols) ===") |
|
|
|
def show(label, expr, deref=True): |
|
addr = _try_eval_addr(expr) |
|
if addr is None: |
|
print(f" {label:30s} = <symbol not loaded>") |
|
return |
|
if deref: |
|
# globals like MmPagedPoolStart are PVOID — read the pointer's value |
|
val = read_u64(addr) |
|
if val is not None: |
|
print(f" {label:30s} = 0x{val:016x}") |
|
else: |
|
print(f" {label:30s} @ 0x{addr:x} <unreadable>") |
|
else: |
|
print(f" {label:30s} = 0x{addr:016x}") |
|
|
|
print(f" {'PTE_BASE (amd64)':30s} = 0x{PTE_BASE_AMD64:016x}") |
|
print(f" {'PTE_TOP (amd64)':30s} = 0x{PTE_TOP_AMD64:016x}") |
|
# User-mode ceiling |
|
show('MmHighestUserAddress', '&MmHighestUserAddress', deref=True) |
|
show('MmUserProbeAddress', '&MmUserProbeAddress', deref=True) |
|
# System range / pools |
|
show('MmSystemRangeStart', '&MmSystemRangeStart', deref=True) |
|
show('MmPagedPoolStart', '&MmPagedPoolStart', deref=True) |
|
show('MmPagedPoolEnd', '&MmPagedPoolEnd', deref=True) |
|
show('MmNonPagedPoolStart', '&MmNonPagedPoolStart', deref=True) |
|
show('MmNonPagedPoolEnd', '&MmNonPagedPoolEnd', deref=True) |
|
show('MmNonPagedSystemStart', '&MmNonPagedSystemStart', deref=True) |
|
# System cache |
|
show('MmSystemCacheStart', '&MmSystemCacheStart', deref=True) |
|
show('MmSystemCacheEnd', '&MmSystemCacheEnd', deref=True) |
|
# Misc ARM3 ranges |
|
show('MiSystemViewStart', '&MiSystemViewStart', deref=True) |
|
show('MmSessionBase', '&MmSessionBase', deref=True) |
|
# Useful neighbours |
|
show('MmPfnDatabase', '&MmPfnDatabase', deref=True) |
|
show('PsIdleProcess', '&PsIdleProcess', deref=True) |
|
show('MmSystemCacheWs (addr)', '&MmSystemCacheWs', deref=False) |
|
# KUSER_SHARED_DATA — fixed VA on amd64 |
|
print(f" {'KUSER_SHARED_DATA (kernel)':30s} = 0xFFFFF78000000000") |
|
print(f" {'KUSER_SHARED_DATA (user)':30s} = 0x000000007FFE0000") |
|
|
|
ReactosMmRanges() |
|
|
|
|
|
# ============================================================ |
|
# Locks / Sync : ros-eresource, ros-pushlock, ros-dispobj |
|
# ============================================================ |
|
# |
|
# Source-of-truth offsets (amd64): |
|
# sdk/include/xdk/extypes.h:231 : _ERESOURCE |
|
# sdk/include/ndk/extypes.h:624 : _EX_PUSH_LOCK |
|
# sdk/include/xdk/ketypes.h:792,863.. : DISPATCHER_HEADER, KEVENT, KSEMAPHORE |
|
# sdk/include/ndk/ketypes.h:405 : KOBJECTS enum |
|
# commit 308dbcd42ab + sdk/include/ndk/pstypes.h:1550 : |
|
# ETHREAD tail HeldPushLocks[8] + HeldPushLockCount (DBG only). |
|
|
|
# _ERESOURCE field offsets (amd64). LIST_ENTRY=16, ULONG_PTR pad rules. |
|
ERES_SYS_LIST = 0x00 # LIST_ENTRY |
|
ERES_OWNER_TABLE = 0x10 # POWNER_ENTRY |
|
ERES_ACTIVE_COUNT = 0x18 # SHORT |
|
ERES_FLAG = 0x1A # USHORT |
|
ERES_SHARED_WAITERS = 0x20 # PKSEMAPHORE |
|
ERES_EXCL_WAITERS = 0x28 # PKEVENT |
|
ERES_OWNER_ENTRY = 0x30 # OWNER_ENTRY (8 + 4 + 4pad = 16) |
|
ERES_ACTIVE_ENTRIES = 0x40 # ULONG |
|
ERES_CONTENTION = 0x44 # ULONG |
|
ERES_NUM_SHARED = 0x48 # ULONG |
|
ERES_NUM_EXCL = 0x4C # ULONG |
|
ERES_RESERVED2 = 0x50 # PVOID |
|
ERES_ADDRESS = 0x58 # PVOID / CreatorBackTraceIndex |
|
ERES_SPINLOCK = 0x60 # KSPIN_LOCK |
|
|
|
# OWNER_ENTRY: OwnerThread(8) + DUMMYUNION(4) + pad(4) |
|
OWN_OWNER_THREAD = 0x00 |
|
OWN_OWNER_COUNT = 0x08 # bottom 30 bits |
|
|
|
class ReactosEResource(gdb.Command): |
|
"""Decode an ERESOURCE. Source: sdk/include/xdk/extypes.h:231. |
|
Usage: ros-eresource <addr>""" |
|
def __init__(self): |
|
super().__init__("ros-eresource", gdb.COMMAND_USER) |
|
def invoke(self, arg, from_tty): |
|
if not arg.strip(): |
|
print("Usage: ros-eresource <ERESOURCE_addr>") |
|
return |
|
try: |
|
addr = int(arg.strip(), 0) |
|
except Exception: |
|
print(f"Bad address: {arg.strip()}") |
|
return |
|
try: |
|
owner_table = read_u64(addr + ERES_OWNER_TABLE) |
|
active = read_i16(addr + ERES_ACTIVE_COUNT) |
|
flag = read_u16(addr + ERES_FLAG) |
|
shared_w = read_u64(addr + ERES_SHARED_WAITERS) |
|
excl_w = read_u64(addr + ERES_EXCL_WAITERS) |
|
owner_thr = read_u64(addr + ERES_OWNER_ENTRY + OWN_OWNER_THREAD) |
|
own_word = read_u32(addr + ERES_OWNER_ENTRY + OWN_OWNER_COUNT) |
|
active_e = read_u32(addr + ERES_ACTIVE_ENTRIES) |
|
contention = read_u32(addr + ERES_CONTENTION) |
|
num_shared = read_u32(addr + ERES_NUM_SHARED) |
|
num_excl = read_u32(addr + ERES_NUM_EXCL) |
|
spinlock = read_u64(addr + ERES_SPINLOCK) |
|
except Exception as e: |
|
print(f"ros-eresource read error: {e}") |
|
return |
|
if active is None: |
|
print(f"Cannot read ERESOURCE at 0x{addr:x}") |
|
return |
|
print(f"=== ERESOURCE @ 0x{addr:016x} ===") |
|
print(f" ActiveCount = {active}") |
|
flag_bits = [] |
|
if flag & 0x10: flag_bits.append('NeverExclusive') |
|
if flag & 0x20: flag_bits.append('ReleaseByOtherThread') |
|
if flag & 0x80: flag_bits.append('OwnedExclusive') |
|
flag_s = ', '.join(flag_bits) if flag_bits else '(none)' |
|
print(f" Flag = 0x{flag:x} [{flag_s}]") |
|
print(f" ContentionCount = {contention}") |
|
print(f" NumberOfSharedWaiters = {num_shared}") |
|
print(f" NumberOfExclusiveWaiters = {num_excl}") |
|
print(f" ActiveEntries = {active_e}") |
|
print(f" OwnerTable = 0x{owner_table:016x}") |
|
print(f" SharedWaiters (KSEM*) = 0x{shared_w:016x}") |
|
print(f" ExclusiveWaiters (KEVT*) = 0x{excl_w:016x}") |
|
own_count = own_word & 0x3FFFFFFF if own_word is not None else 0 |
|
own_io = (own_word >> 0) & 1 if own_word is not None else 0 |
|
own_ref = (own_word >> 1) & 1 if own_word is not None else 0 |
|
print(f" OwnerEntry.OwnerThread = 0x{owner_thr:016x}") |
|
print(f" OwnerEntry.OwnerCount = {own_count} IoBoosted={own_io} OwnerRefd={own_ref}") |
|
print(f" SpinLock = 0x{spinlock:x}") |
|
if owner_thr: |
|
print(f" Use: ros-thread 0x{owner_thr:x}") |
|
if shared_w: |
|
print(f" Use: ros-dispobj 0x{shared_w:x} # KSEMAPHORE for shared waiters") |
|
if excl_w: |
|
print(f" Use: ros-dispobj 0x{excl_w:x} # KEVENT for exclusive waiters") |
|
|
|
ReactosEResource() |
|
|
|
|
|
class ReactosPushLock(gdb.Command): |
|
"""Decode an EX_PUSH_LOCK (8 bytes). Source: sdk/include/ndk/extypes.h:624. |
|
Optional 2nd arg = ETHREAD address: walk DBG HeldPushLocks[] tail array |
|
(commit 308dbcd42ab; sdk/include/ndk/pstypes.h:1550). |
|
Usage: ros-pushlock <addr> [ethread_addr]""" |
|
def __init__(self): |
|
super().__init__("ros-pushlock", gdb.COMMAND_USER) |
|
def invoke(self, arg, from_tty): |
|
args = arg.strip().split() |
|
if not args: |
|
print("Usage: ros-pushlock <addr> [ethread_addr]") |
|
return |
|
try: |
|
addr = int(args[0], 0) |
|
except Exception: |
|
print(f"Bad address: {args[0]}") |
|
return |
|
val = read_u64(addr) |
|
if val is None: |
|
print(f"Cannot read EX_PUSH_LOCK at 0x{addr:x}") |
|
return |
|
locked = val & 1 |
|
waiting = (val >> 1) & 1 |
|
waking = (val >> 2) & 1 |
|
multi = (val >> 3) & 1 |
|
shared = (val >> 4) |
|
print(f"=== EX_PUSH_LOCK @ 0x{addr:016x} = 0x{val:016x} ===") |
|
print(f" Locked = {locked}") |
|
print(f" Waiting = {waiting}") |
|
print(f" Waking = {waking}") |
|
print(f" MultipleShared = {multi}") |
|
print(f" Shared (count or wait-block ptr) = 0x{shared:x}") |
|
if waiting and shared: |
|
wb_ptr = val & ~0xF |
|
print(f" Likely first wait-block @ 0x{wb_ptr:x}") |
|
if len(args) >= 2: |
|
try: |
|
ethread = int(args[1], 0) |
|
except Exception: |
|
print(f"Bad ethread: {args[1]}") |
|
return |
|
# ETHREAD HeldPushLocks/HeldPushLockCount live at the tail (DBG-only). |
|
# We can't compute offset without symbols; try gdb type lookup first. |
|
off_arr = _maybe_lookup_field_offset("struct _ETHREAD", "HeldPushLocks") |
|
off_cnt = _maybe_lookup_field_offset("struct _ETHREAD", "HeldPushLockCount") |
|
if off_arr is None or off_cnt is None: |
|
print(f" (DBG ETHREAD HeldPushLocks not present — release build or symbols missing)") |
|
return |
|
cnt = read_u8(ethread + off_cnt) or 0 |
|
print(f" ETHREAD 0x{ethread:x} HeldPushLockCount = {cnt}") |
|
for i in range(min(cnt, 8)): |
|
p = read_u64(ethread + off_arr + i*8) |
|
marker = ' <== this lock' if p == addr else '' |
|
print(f" HeldPushLocks[{i}] = 0x{p:016x}{marker}") |
|
|
|
ReactosPushLock() |
|
|
|
|
|
# DISPATCHER_HEADER (amd64): Type@0, Size/etc@1..3, SignalState@4, WaitListHead@8 (16). |
|
DISP_TYPE = 0x00 |
|
DISP_SIGNAL = 0x04 |
|
DISP_WAITLIST = 0x08 |
|
|
|
_KOBJECT_NAMES = { |
|
0:'EventNotification', 1:'EventSync', 2:'Mutant', 3:'Process', 4:'Queue', |
|
5:'Semaphore', 6:'Thread', 7:'Gate', 8:'TimerNotification', 9:'TimerSync', |
|
18:'Apc', 19:'Dpc', 20:'DeviceQueue', 21:'EventPair', 22:'Interrupt', |
|
23:'Profile', |
|
} |
|
|
|
class ReactosDispObj(gdb.Command): |
|
"""Decode any dispatcher object via DISPATCHER_HEADER + walk WaitListHead. |
|
Source: sdk/include/xdk/ketypes.h:792 ; KOBJECTS at ndk/ketypes.h:405. |
|
KWAIT_BLOCK layout (NTDDI_WIN8) at xdk/ketypes.h:507. |
|
Usage: ros-dispobj <addr>""" |
|
def __init__(self): |
|
super().__init__("ros-dispobj", gdb.COMMAND_USER) |
|
def invoke(self, arg, from_tty): |
|
if not arg.strip(): |
|
print("Usage: ros-dispobj <addr>") |
|
return |
|
try: |
|
addr = int(arg.strip(), 0) |
|
except Exception: |
|
print(f"Bad address: {arg.strip()}") |
|
return |
|
try: |
|
t = read_u8(addr + DISP_TYPE) |
|
b1 = read_u8(addr + 1) |
|
sz = read_u8(addr + 2) |
|
inserted = read_u8(addr + 3) |
|
sig = read_u32(addr + DISP_SIGNAL) |
|
flink = read_u64(addr + DISP_WAITLIST) |
|
blink = read_u64(addr + DISP_WAITLIST + 8) |
|
except Exception as e: |
|
print(f"ros-dispobj read error: {e}") |
|
return |
|
if t is None: |
|
print(f"Cannot read DISPATCHER_HEADER at 0x{addr:x}") |
|
return |
|
# KOBJECTS encoded only in low 5 bits on some builds; check both |
|
kname = _KOBJECT_NAMES.get(t, _KOBJECT_NAMES.get(t & 0x1F, f'?({t})')) |
|
print(f"=== DispatcherHeader @ 0x{addr:016x} ===") |
|
print(f" Type = 0x{t:02x} [{kname}]") |
|
print(f" Abandoned/Sig= 0x{b1:02x}") |
|
print(f" Size/Hand = 0x{sz:02x}") |
|
print(f" Inserted/DBG = 0x{inserted:02x}") |
|
print(f" SignalState = {sig if sig is not None and sig < 0x80000000 else sig}") |
|
print(f" WaitListHead F=0x{flink:016x} B=0x{blink:016x}") |
|
head = addr + DISP_WAITLIST |
|
if not flink or flink == head: |
|
print(" (no waiters)") |
|
return |
|
cur = flink |
|
seen = set() |
|
i = 0 |
|
# KWAIT_BLOCK NTDDI_WIN8: WaitListEntry@0, WaitType@0x10, BlockState@0x11, |
|
# WaitKey@0x12, SpareLong@0x14 (_WIN64), Thread@0x18, Object@0x20. |
|
while cur and cur != head and cur not in seen and i < 16: |
|
seen.add(cur) |
|
wb = cur # KWAIT_BLOCK starts at WaitListEntry |
|
wait_type = read_u8(wb + 0x10) |
|
wkey = read_u16(wb + 0x12) |
|
thr = read_u64(wb + 0x18) |
|
obj = read_u64(wb + 0x20) |
|
print(f" [{i:2d}] WaitBlock@0x{wb:x} Thread=0x{thr:x} Obj=0x{obj:x}" |
|
f" WaitType={wait_type} Key={wkey}") |
|
cur = read_u64(wb) |
|
if cur is None: |
|
break |
|
i += 1 |
|
if i and thr: |
|
print(f" Use: ros-thread 0x{thr:x}") |
|
|
|
ReactosDispObj() |
|
|
|
|
|
# ============================================================ |
|
# Heap : ros-heap, ros-heap-freelist |
|
# ============================================================ |
|
# |
|
# Source: sdk/lib/rtl/heap.h:222 (_HEAP), :129 (_HEAP_FREE_ENTRY). |
|
# RtlInitializeHeap (heap.c:117..225) sets: |
|
# FreeHints : PLIST_ENTRY array, size DeCommitFreeBlockThreshold |
|
# FreeHintBitmap: RTL_BITMAP whose Buffer = &FreeHints[DeCommitFreeBlockThreshold] |
|
# Layout (amd64) — the critical fields: |
|
HEAP_FLAGS = 0x70 |
|
HEAP_FORCE_FLAGS = 0x74 |
|
HEAP_VIRT_ALLOC_THRESHOLD = 0x9C |
|
HEAP_SIGNATURE = 0xA0 |
|
HEAP_DECOMMIT_THRESHOLD = 0xB8 # SIZE_T (in HEAP_ENTRY units) |
|
HEAP_TOTAL_FREE = 0xC8 |
|
HEAP_PROCHEAPS_INDEX = 0xD8 # USHORT |
|
# FreeLists / FreeHintBitmap / FreeHints offsets vary by HEAP_SEGMENTS |
|
# constant + 64-byte counters block. The HEAP_SEGMENTS=64 case yields: |
|
# FreeLists @ 0x368 (16), LockVariable @ 0x378 ... after Counters/Tuning, |
|
# FreeHintBitmap and FreeHints[] follow the heap header. We use gdb type |
|
# lookup at runtime; fall back to fixed offsets that match heap.h:222. |
|
HEAP_FREELISTS_FALLBACK = 0x368 |
|
HEAP_LOCK_VAR_FALLBACK = 0x378 |
|
|
|
# HEAP_FREE_ENTRY (amd64): HEAP_COMMON_ENTRY (16) + LIST_ENTRY FreeList (16) = 32 |
|
HFE_PREVIOUS_BLOCK_PRIV = 0x00 # encoded |
|
HFE_SIZE = 0x08 # USHORT |
|
HFE_FLAGS = 0x0A # UCHAR |
|
HFE_SMALL_TAG_INDEX = 0x0B # UCHAR |
|
HFE_PREV_SIZE = 0x0C # USHORT |
|
HFE_SEGMENT_OFFSET = 0x0E # UCHAR |
|
HFE_UNUSED_BYTES = 0x0F # UCHAR |
|
HFE_LIST = 0x10 # LIST_ENTRY (Flink, Blink) |
|
|
|
|
|
def _heap_field(addr, name, fallback): |
|
"""Try to use real heap struct from gdb; fall back to fixed offset.""" |
|
off = _maybe_lookup_field_offset("struct _HEAP", name) |
|
if off is None: |
|
return read_u64(addr + fallback) if name != 'Flags' and name != 'ForceFlags' \ |
|
else read_u32(addr + fallback) |
|
# Read by best size guess: pointers/SIZE_T => 8, ULONG => 4 |
|
if name in ('Flags','ForceFlags','Signature','VirtualMemoryThreshold'): |
|
return read_u32(addr + off) |
|
return read_u64(addr + off) |
|
|
|
|
|
class ReactosHeap(gdb.Command): |
|
"""Dump RTL _HEAP header + summary of FreeHintBitmap. |
|
Source: sdk/lib/rtl/heap.h:222, sdk/lib/rtl/heap.c:117..225. |
|
Usage: ros-heap <heap_addr>""" |
|
def __init__(self): |
|
super().__init__("ros-heap", gdb.COMMAND_USER) |
|
def invoke(self, arg, from_tty): |
|
if not arg.strip(): |
|
print("Usage: ros-heap <heap_addr>") |
|
return |
|
try: |
|
h = int(arg.strip(), 0) |
|
except Exception: |
|
print(f"Bad address: {arg.strip()}") |
|
return |
|
try: |
|
sig = read_u32(h + HEAP_SIGNATURE) |
|
flags = read_u32(h + HEAP_FLAGS) |
|
force = read_u32(h + HEAP_FORCE_FLAGS) |
|
vthr = read_u32(h + HEAP_VIRT_ALLOC_THRESHOLD) |
|
decom = read_u64(h + HEAP_DECOMMIT_THRESHOLD) |
|
tfree = read_u64(h + HEAP_TOTAL_FREE) |
|
phidx = read_u16(h + HEAP_PROCHEAPS_INDEX) |
|
except Exception as e: |
|
print(f"ros-heap read error: {e}") |
|
return |
|
if sig is None: |
|
print(f"Cannot read HEAP at 0x{h:x}") |
|
return |
|
print(f"=== _HEAP @ 0x{h:016x} ===") |
|
print(f" Signature = 0x{sig:08x} (expected 0xEEFFEEFF)") |
|
print(f" Flags = 0x{flags:08x}") |
|
print(f" ForceFlags = 0x{force:08x}") |
|
print(f" VirtualMemoryThreshold = 0x{vthr:x} (HEAP_ENTRYs)") |
|
print(f" DeCommitFreeBlockThr = 0x{decom:x} (HEAP_ENTRYs)") |
|
print(f" TotalFreeSize = 0x{tfree:x} (HEAP_ENTRYs)") |
|
print(f" ProcessHeapsListIndex = {phidx}") |
|
# Try gdb to find FreeLists, FreeHintBitmap, FreeHints |
|
fl_off = _maybe_lookup_field_offset("struct _HEAP", "FreeLists") |
|
fh_off = _maybe_lookup_field_offset("struct _HEAP", "FreeHints") |
|
fhb_off = _maybe_lookup_field_offset("struct _HEAP", "FreeHintBitmap") |
|
if fl_off is None: fl_off = HEAP_FREELISTS_FALLBACK |
|
flink = read_u64(h + fl_off) |
|
blink = read_u64(h + fl_off + 8) |
|
print(f" FreeLists @ +0x{fl_off:x} F=0x{flink:016x} B=0x{blink:016x}") |
|
if fhb_off is not None: |
|
bm_size = read_u32(h + fhb_off + 0x00) # SizeOfBitMap |
|
bm_buf = read_u64(h + fhb_off + 0x08) # Buffer |
|
if bm_size is not None and bm_buf: |
|
set_bits = [] |
|
# Walk up to ceil(bm_size/32) ULONGs; collect first 10 set bits |
|
ulongs = (bm_size + 31) // 32 if bm_size else 0 |
|
for u in range(min(ulongs, 1024)): |
|
word = read_u32(bm_buf + u*4) |
|
if not word: |
|
continue |
|
for bit in range(32): |
|
if word & (1 << bit): |
|
set_bits.append(u*32 + bit) |
|
if len(set_bits) >= 10: |
|
break |
|
if len(set_bits) >= 10: |
|
break |
|
print(f" FreeHintBitmap SizeOfBitMap={bm_size} Buffer=0x{bm_buf:x}") |
|
if fh_off is not None and set_bits: |
|
print(f" First {len(set_bits)} set FreeHint bits:") |
|
for b in set_bits: |
|
ptr = read_u64(h + fh_off + b*8) |
|
print(f" [bit {b:4d}] FreeHints[{b}] = 0x{ptr:016x}") |
|
elif not set_bits: |
|
print(f" FreeHintBitmap: (no bits set)") |
|
print(f" Use: ros-heap-freelist 0x{h:x} [size_class]") |
|
|
|
ReactosHeap() |
|
|
|
|
|
class ReactosHeapFreeList(gdb.Command): |
|
"""Walk a heap free list. size_class=0 walks Heap->FreeLists anchor; |
|
nonzero walks Heap->FreeHints[size_class]. |
|
Source: sdk/lib/rtl/heap.h:129 (_HEAP_FREE_ENTRY). |
|
Usage: ros-heap-freelist <heap_addr> [size_class=0] [max=64]""" |
|
def __init__(self): |
|
super().__init__("ros-heap-freelist", gdb.COMMAND_USER) |
|
def invoke(self, arg, from_tty): |
|
args = arg.strip().split() |
|
if not args: |
|
print("Usage: ros-heap-freelist <heap_addr> [size_class=0] [max=64]") |
|
return |
|
try: |
|
h = int(args[0], 0) |
|
except Exception: |
|
print(f"Bad address: {args[0]}") |
|
return |
|
size_class = int(args[1], 0) if len(args) > 1 else 0 |
|
max_n = int(args[2], 0) if len(args) > 2 else 64 |
|
fl_off = _maybe_lookup_field_offset("struct _HEAP", "FreeLists") \ |
|
or HEAP_FREELISTS_FALLBACK |
|
fh_off = _maybe_lookup_field_offset("struct _HEAP", "FreeHints") |
|
if size_class == 0: |
|
head = h + fl_off |
|
print(f"=== Heap 0x{h:x} FreeLists anchor @ 0x{head:x} ===") |
|
cur = read_u64(head) |
|
if cur is None: |
|
print(" (unreadable)") |
|
return |
|
else: |
|
if fh_off is None: |
|
print(" Cannot resolve _HEAP.FreeHints offset; pass size_class=0") |
|
return |
|
slot = h + fh_off + size_class*8 |
|
head = slot |
|
cur = read_u64(slot) |
|
print(f"=== Heap 0x{h:x} FreeHints[{size_class}] @ 0x{slot:x} -> 0x{cur:x} ===") |
|
if not cur: |
|
print(" (empty)") |
|
return |
|
seen = set() |
|
i = 0 |
|
while cur and cur != head and cur not in seen and i < max_n: |
|
seen.add(cur) |
|
# FreeList LIST_ENTRY is at offset HFE_LIST inside HEAP_FREE_ENTRY. |
|
entry = cur - HFE_LIST # cur points at LIST_ENTRY |
|
sz = read_u16(entry + HFE_SIZE) |
|
psz = read_u16(entry + HFE_PREV_SIZE) |
|
fl = read_u8(entry + HFE_FLAGS) or 0 |
|
segoff = read_u8(entry + HFE_SEGMENT_OFFSET) or 0 |
|
f = read_u64(cur + 0) |
|
b = read_u64(cur + 8) |
|
print(f" [{i:3d}] entry@0x{entry:x} Size={sz} PrevSize={psz} " |
|
f"Flags=0x{fl:02x} SegOff={segoff} F=0x{f:x} B=0x{b:x}") |
|
cur = f |
|
i += 1 |
|
if cur in seen: |
|
print(f" *** loop detected at 0x{cur:x}") |
|
elif i >= max_n: |
|
print(f" ... truncated at {max_n}") |
|
|
|
ReactosHeapFreeList() |
|
|
|
|
|
# ============================================================ |
|
# I/O : ros-fileobj, ros-vpb, ros-devobj, ros-drvobj |
|
# ============================================================ |
|
# |
|
# Source: sdk/include/xdk/iotypes.h:1998 (_FILE_OBJECT), |
|
# :189 (_VPB), :288 (_DEVICE_OBJECT), :2307 (_DRIVER_OBJECT). |
|
|
|
# FILE_OBJECT (amd64) — Type(2)/Size(2)+pad(4)=8 |
|
FO_DEVICE_OBJECT = 0x08 |
|
FO_VPB = 0x10 |
|
FO_FS_CONTEXT = 0x18 |
|
FO_FS_CONTEXT2 = 0x20 |
|
FO_SECTION_PTR = 0x28 |
|
FO_PRIVATE_CACHEMAP = 0x30 |
|
FO_FINAL_STATUS = 0x38 |
|
FO_RELATED_FO = 0x40 |
|
FO_LOCK_OPERATION = 0x48 # 8 BOOLEANs in a row |
|
FO_FLAGS = 0x50 |
|
FO_FILENAME = 0x58 # UNICODE_STRING (16) |
|
FO_CURBYTE_OFFSET = 0x68 |
|
|
|
class ReactosFileObj(gdb.Command): |
|
"""Decode a FILE_OBJECT. Source: sdk/include/xdk/iotypes.h:1998. |
|
Usage: ros-fileobj <addr>""" |
|
def __init__(self): |
|
super().__init__("ros-fileobj", gdb.COMMAND_USER) |
|
def invoke(self, arg, from_tty): |
|
if not arg.strip(): |
|
print("Usage: ros-fileobj <FILE_OBJECT_addr>") |
|
return |
|
try: |
|
f = int(arg.strip(), 0) |
|
except Exception: |
|
print(f"Bad address: {arg.strip()}") |
|
return |
|
try: |
|
t = read_i16(f + 0) |
|
sz = read_i16(f + 2) |
|
dev = read_u64(f + FO_DEVICE_OBJECT) |
|
vpb = read_u64(f + FO_VPB) |
|
fc1 = read_u64(f + FO_FS_CONTEXT) |
|
fc2 = read_u64(f + FO_FS_CONTEXT2) |
|
sop = read_u64(f + FO_SECTION_PTR) |
|
pcm = read_u64(f + FO_PRIVATE_CACHEMAP) |
|
fst = read_u32(f + FO_FINAL_STATUS) |
|
rfo = read_u64(f + FO_RELATED_FO) |
|
booleans = read_mem(f + FO_LOCK_OPERATION, 8) |
|
flags = read_u32(f + FO_FLAGS) |
|
name = _read_unicode_string_at(f + FO_FILENAME) or '' |
|
except Exception as e: |
|
print(f"ros-fileobj read error: {e}") |
|
return |
|
if t is None: |
|
print(f"Cannot read FILE_OBJECT at 0x{f:x}") |
|
return |
|
print(f"=== FILE_OBJECT @ 0x{f:016x} ===") |
|
print(f" Type={t} Size={sz}") |
|
print(f" DeviceObject = 0x{dev:016x}") |
|
print(f" Vpb = 0x{vpb:016x}") |
|
print(f" FsContext = 0x{fc1:016x}") |
|
print(f" FsContext2 = 0x{fc2:016x}") |
|
print(f" SectionObjectPointer = 0x{sop:016x}") |
|
print(f" PrivateCacheMap = 0x{pcm:016x}") |
|
print(f" FinalStatus = 0x{fst:08x}") |
|
print(f" RelatedFileObject = 0x{rfo:016x}") |
|
print(f" Flags = 0x{flags:08x}") |
|
if booleans: |
|
lo, dp, ra, wa, da, sr, sw, sd = booleans[:8] |
|
print(f" LockOp={lo} DeletePend={dp} ReadAcc={ra} WriteAcc={wa} " |
|
f"DelAcc={da} SharedR={sr} SharedW={sw} SharedD={sd}") |
|
if name: |
|
display = name[:80] + ('...' if len(name) > 80 else '') |
|
print(f" FileName = \"{display}\"") |
|
if dev: print(f" Use: ros-devobj 0x{dev:x}") |
|
if vpb: print(f" Use: ros-vpb 0x{vpb:x}") |
|
if sop: print(f" Use: ros-secptr 0x{sop:x}") |
|
|
|
ReactosFileObj() |
|
|
|
|
|
# VPB (amd64): Type(2)+Size(2)+Flags(2)+VolLabelLen(2)=8 |
|
VPB_FLAGS = 0x04 |
|
VPB_VOL_LABEL_LEN = 0x06 |
|
VPB_DEVICE_OBJECT = 0x08 |
|
VPB_REAL_DEVICE = 0x10 |
|
VPB_SERIAL_NUMBER = 0x18 |
|
VPB_REFERENCE_COUNT = 0x1C |
|
VPB_VOLUME_LABEL = 0x20 |
|
|
|
class ReactosVpb(gdb.Command): |
|
"""Decode a VPB. Source: sdk/include/xdk/iotypes.h:189. |
|
Usage: ros-vpb <addr>""" |
|
def __init__(self): |
|
super().__init__("ros-vpb", gdb.COMMAND_USER) |
|
def invoke(self, arg, from_tty): |
|
if not arg.strip(): |
|
print("Usage: ros-vpb <VPB_addr>") |
|
return |
|
try: |
|
v = int(arg.strip(), 0) |
|
except Exception: |
|
print(f"Bad address: {arg.strip()}") |
|
return |
|
try: |
|
t = read_i16(v + 0) |
|
sz = read_i16(v + 2) |
|
flags = read_u16(v + VPB_FLAGS) |
|
vlen = read_u16(v + VPB_VOL_LABEL_LEN) |
|
dev = read_u64(v + VPB_DEVICE_OBJECT) |
|
rdev = read_u64(v + VPB_REAL_DEVICE) |
|
serial = read_u32(v + VPB_SERIAL_NUMBER) |
|
refc = read_u32(v + VPB_REFERENCE_COUNT) |
|
label_buf = read_mem(v + VPB_VOLUME_LABEL, min(vlen or 0, 64)) |
|
except Exception as e: |
|
print(f"ros-vpb read error: {e}") |
|
return |
|
if t is None: |
|
print(f"Cannot read VPB at 0x{v:x}") |
|
return |
|
print(f"=== VPB @ 0x{v:016x} ===") |
|
print(f" Type={t} Size={sz} RefCount={refc} Serial=0x{serial:08x}") |
|
flag_bits = [] |
|
if flags & 0x0001: flag_bits.append('Mounted') |
|
if flags & 0x0002: flag_bits.append('Locked') |
|
if flags & 0x0004: flag_bits.append('ShutdownDev') |
|
if flags & 0x0008: flag_bits.append('IsRemoved') |
|
if flags & 0x0010: flag_bits.append('RawMount') |
|
if flags & 0x0020: flag_bits.append('RemoveMounted') |
|
print(f" Flags = 0x{flags:04x} [{', '.join(flag_bits) or '(none)'}]") |
|
label = '' |
|
if label_buf: |
|
try: label = label_buf.decode('utf-16-le').rstrip('\x00') |
|
except: pass |
|
print(f" VolumeLabel ({vlen} bytes) = \"{label}\"") |
|
print(f" DeviceObject (FS volume) = 0x{dev:016x}") |
|
print(f" RealDevice = 0x{rdev:016x}") |
|
if dev: print(f" Use: ros-devobj 0x{dev:x}") |
|
if rdev: print(f" Use: ros-devobj 0x{rdev:x}") |
|
|
|
ReactosVpb() |
|
|
|
|
|
# DEVICE_OBJECT (amd64, DECLSPEC_ALIGN(16)): Type(2)+Size(2)+RefCnt(4)=8 |
|
DO_DRIVER_OBJECT = 0x08 |
|
DO_NEXT_DEVICE = 0x10 |
|
DO_ATTACHED_DEVICE = 0x18 |
|
DO_CURRENT_IRP = 0x20 |
|
DO_TIMER = 0x28 |
|
DO_FLAGS = 0x30 |
|
DO_CHARACTERISTICS = 0x34 |
|
DO_VPB = 0x38 |
|
DO_DEVICE_EXTENSION = 0x40 |
|
DO_DEVICE_TYPE = 0x48 |
|
DO_STACK_SIZE = 0x4C # CCHAR |
|
|
|
class ReactosDevObj(gdb.Command): |
|
"""Decode a DEVICE_OBJECT. Source: sdk/include/xdk/iotypes.h:288. |
|
Usage: ros-devobj <addr>""" |
|
def __init__(self): |
|
super().__init__("ros-devobj", gdb.COMMAND_USER) |
|
def invoke(self, arg, from_tty): |
|
if not arg.strip(): |
|
print("Usage: ros-devobj <DEVICE_OBJECT_addr>") |
|
return |
|
try: |
|
d = int(arg.strip(), 0) |
|
except Exception: |
|
print(f"Bad address: {arg.strip()}") |
|
return |
|
try: |
|
t = read_i16(d + 0) |
|
sz = read_u16(d + 2) |
|
refc = read_u32(d + 4) |
|
drv = read_u64(d + DO_DRIVER_OBJECT) |
|
nxt = read_u64(d + DO_NEXT_DEVICE) |
|
att = read_u64(d + DO_ATTACHED_DEVICE) |
|
cur = read_u64(d + DO_CURRENT_IRP) |
|
flags = read_u32(d + DO_FLAGS) |
|
chars = read_u32(d + DO_CHARACTERISTICS) |
|
vpb = read_u64(d + DO_VPB) |
|
ext = read_u64(d + DO_DEVICE_EXTENSION) |
|
dtype = read_u32(d + DO_DEVICE_TYPE) |
|
stack = read_u8(d + DO_STACK_SIZE) |
|
except Exception as e: |
|
print(f"ros-devobj read error: {e}") |
|
return |
|
if t is None: |
|
print(f"Cannot read DEVICE_OBJECT at 0x{d:x}") |
|
return |
|
print(f"=== DEVICE_OBJECT @ 0x{d:016x} ===") |
|
print(f" Type={t} Size={sz} RefCount={refc} StackSize={stack}") |
|
print(f" DriverObject = 0x{drv:016x}") |
|
print(f" NextDevice = 0x{nxt:016x}") |
|
print(f" AttachedDevice = 0x{att:016x}") |
|
print(f" CurrentIrp = 0x{cur:016x}") |
|
print(f" Flags = 0x{flags:08x}") |
|
print(f" Characteristics = 0x{chars:08x}") |
|
print(f" Vpb = 0x{vpb:016x}") |
|
print(f" DeviceExtension = 0x{ext:016x}") |
|
print(f" DeviceType = 0x{dtype:08x}") |
|
if drv: print(f" Use: ros-drvobj 0x{drv:x}") |
|
if vpb: print(f" Use: ros-vpb 0x{vpb:x}") |
|
if att: print(f" Use: ros-devobj 0x{att:x}") |
|
if nxt: print(f" Use: ros-devobj 0x{nxt:x}") |
|
|
|
ReactosDevObj() |
|
|
|
|
|
# DRIVER_OBJECT (amd64): Type(2)+Size(2)+pad(4)=8 |
|
DRO_DEVICE_OBJECT = 0x08 |
|
DRO_FLAGS = 0x10 |
|
DRO_DRIVER_START = 0x18 |
|
DRO_DRIVER_SIZE = 0x20 |
|
DRO_DRIVER_SECTION = 0x28 |
|
DRO_DRIVER_EXTENSION = 0x30 |
|
DRO_DRIVER_NAME = 0x38 # UNICODE_STRING(16) |
|
DRO_HW_DATABASE = 0x48 # PUNICODE_STRING |
|
DRO_FAST_IO_DISPATCH = 0x50 |
|
DRO_DRIVER_INIT = 0x58 |
|
DRO_DRIVER_STARTIO = 0x60 |
|
DRO_DRIVER_UNLOAD = 0x68 |
|
DRO_MAJOR_FUNCTION = 0x70 # PDRIVER_DISPATCH[28] |
|
|
|
_IRP_MJ_NAMES = [ |
|
'CREATE','CREATE_NAMED_PIPE','CLOSE','READ','WRITE','QUERY_INFORMATION', |
|
'SET_INFORMATION','QUERY_EA','SET_EA','FLUSH_BUFFERS','QUERY_VOLUME_INFO', |
|
'SET_VOLUME_INFO','DIRECTORY_CONTROL','FILE_SYSTEM_CONTROL','DEVICE_CONTROL', |
|
'INTERNAL_DEVICE_CONTROL','SHUTDOWN','LOCK_CONTROL','CLEANUP','CREATE_MAILSLOT', |
|
'QUERY_SECURITY','SET_SECURITY','POWER','SYSTEM_CONTROL','DEVICE_CHANGE', |
|
'QUERY_QUOTA','SET_QUOTA','PNP', |
|
] |
|
|
|
def _resolve_symbol(addr): |
|
if not addr: |
|
return '' |
|
try: |
|
out = gdb.execute(f"info symbol 0x{addr:x}", to_string=True).strip() |
|
return out.split(' in section')[0] |
|
except Exception: |
|
return '' |
|
|
|
class ReactosDrvObj(gdb.Command): |
|
"""Decode a DRIVER_OBJECT. Source: sdk/include/xdk/iotypes.h:2307. |
|
Resolves MajorFunction[] non-default entries via 'info symbol'. |
|
Usage: ros-drvobj <addr>""" |
|
def __init__(self): |
|
super().__init__("ros-drvobj", gdb.COMMAND_USER) |
|
def invoke(self, arg, from_tty): |
|
if not arg.strip(): |
|
print("Usage: ros-drvobj <DRIVER_OBJECT_addr>") |
|
return |
|
try: |
|
d = int(arg.strip(), 0) |
|
except Exception: |
|
print(f"Bad address: {arg.strip()}") |
|
return |
|
try: |
|
t = read_i16(d + 0); sz = read_i16(d + 2) |
|
dev = read_u64(d + DRO_DEVICE_OBJECT) |
|
flags = read_u32(d + DRO_FLAGS) |
|
dst = read_u64(d + DRO_DRIVER_START) |
|
dsz = read_u32(d + DRO_DRIVER_SIZE) |
|
sec = read_u64(d + DRO_DRIVER_SECTION) |
|
ext = read_u64(d + DRO_DRIVER_EXTENSION) |
|
name = _read_unicode_string_at(d + DRO_DRIVER_NAME) or '' |
|
fast = read_u64(d + DRO_FAST_IO_DISPATCH) |
|
init_ = read_u64(d + DRO_DRIVER_INIT) |
|
stio = read_u64(d + DRO_DRIVER_STARTIO) |
|
unl = read_u64(d + DRO_DRIVER_UNLOAD) |
|
except Exception as e: |
|
print(f"ros-drvobj read error: {e}") |
|
return |
|
if t is None: |
|
print(f"Cannot read DRIVER_OBJECT at 0x{d:x}") |
|
return |
|
print(f"=== DRIVER_OBJECT @ 0x{d:016x} ===") |
|
print(f" Type={t} Size={sz}") |
|
print(f" DriverName = \"{name}\"") |
|
print(f" DeviceObject = 0x{dev:016x}") |
|
print(f" Flags = 0x{flags:08x}") |
|
print(f" DriverStart/Size = 0x{dst:016x} / 0x{dsz:x}") |
|
print(f" DriverSection = 0x{sec:016x}") |
|
print(f" DriverExtension = 0x{ext:016x}") |
|
print(f" FastIoDispatch = 0x{fast:016x}") |
|
for label, p in (('DriverInit',init_),('DriverStartIo',stio),('DriverUnload',unl)): |
|
sym = _resolve_symbol(p) if p else '' |
|
print(f" {label:16s} = 0x{p:016x} {sym}") |
|
# Resolve MajorFunction[] — flag entries that don't point to the default |
|
default = read_u64(d + DRO_MAJOR_FUNCTION) # MJ_CREATE — usually IopInvalidDeviceRequest |
|
print(f" MajorFunction[CREATE default] = 0x{default:016x} {_resolve_symbol(default)}") |
|
for i, mname in enumerate(_IRP_MJ_NAMES): |
|
p = read_u64(d + DRO_MAJOR_FUNCTION + i*8) |
|
if p and p != default: |
|
sym = _resolve_symbol(p) |
|
print(f" [{i:2d}] MJ_{mname:22s} 0x{p:016x} {sym}") |
|
if dev: print(f" Use: ros-devobj 0x{dev:x}") |
|
|
|
ReactosDrvObj() |
|
|
|
|
|
# ============================================================ |
|
# Object manager : ros-obj, ros-handles |
|
# ============================================================ |
|
# |
|
# Source: sdk/include/ndk/obtypes.h:111 (OBJECT_TO_OBJECT_HEADER), |
|
# :430 (_OBJECT_HEADER_NAME_INFO), :485 (_OBJECT_HEADER), |
|
# :379 (_OBJECT_TYPE). |
|
|
|
# OBJECT_HEADER (amd64): PointerCount(8) + HandleCount(8) + Type(8) + |
|
# NameInfoOff(1)+HandleInfoOff(1)+QuotaInfoOff(1)+Flags(1) + pad(4) |
|
# + ObjectCreateInfo(8) + SecurityDescriptor(8) + Body |
|
OH_POINTER_COUNT = 0x00 |
|
OH_HANDLE_COUNT = 0x08 |
|
OH_TYPE = 0x10 |
|
OH_NAME_INFO_OFFSET = 0x18 |
|
OH_HANDLE_INFO_OFFSET = 0x19 |
|
OH_QUOTA_INFO_OFFSET = 0x1A |
|
OH_FLAGS = 0x1B |
|
OH_BODY = 0x30 # QUAD-aligned |
|
|
|
# OBJECT_HEADER_NAME_INFO: Directory(8) + Name UNICODE_STRING(16) |
|
OHNI_DIRECTORY = 0x00 |
|
OHNI_NAME = 0x08 |
|
|
|
# OBJECT_TYPE.Name offset (amd64): ERESOURCE(0x68) + LIST_ENTRY(0x10) = 0x78 |
|
OT_NAME = 0x78 |
|
|
|
class ReactosObj(gdb.Command): |
|
"""Walk back from an object body to its OBJECT_HEADER and decode it. |
|
Resolves Type->Name and (if present) the OBJECT_HEADER_NAME_INFO Name. |
|
Source: sdk/include/ndk/obtypes.h:111, :430, :485. |
|
Usage: ros-obj <object_body_addr>""" |
|
def __init__(self): |
|
super().__init__("ros-obj", gdb.COMMAND_USER) |
|
def invoke(self, arg, from_tty): |
|
if not arg.strip(): |
|
print("Usage: ros-obj <object_addr>") |
|
return |
|
try: |
|
body = int(arg.strip(), 0) |
|
except Exception: |
|
print(f"Bad address: {arg.strip()}") |
|
return |
|
hdr = body - OH_BODY |
|
try: |
|
pc = read_u64(hdr + OH_POINTER_COUNT) |
|
hc = read_u64(hdr + OH_HANDLE_COUNT) |
|
otp = read_u64(hdr + OH_TYPE) |
|
ni_o = read_u8(hdr + OH_NAME_INFO_OFFSET) |
|
hi_o = read_u8(hdr + OH_HANDLE_INFO_OFFSET) |
|
qi_o = read_u8(hdr + OH_QUOTA_INFO_OFFSET) |
|
flg = read_u8(hdr + OH_FLAGS) |
|
except Exception as e: |
|
print(f"ros-obj read error: {e}") |
|
return |
|
if pc is None: |
|
print(f"Cannot read OBJECT_HEADER at 0x{hdr:x}") |
|
return |
|
# Sign-extend PointerCount/HandleCount to int64 for sane print |
|
if pc & (1 << 63): pc -= 1 << 64 |
|
if hc & (1 << 63): hc -= 1 << 64 |
|
print(f"=== Object @ 0x{body:016x} (header @ 0x{hdr:x}) ===") |
|
print(f" PointerCount = {pc}") |
|
print(f" HandleCount = {hc}") |
|
print(f" Type = 0x{otp:016x}") |
|
print(f" NameInfoOffset = {ni_o}") |
|
print(f" HandleInfoOffset = {hi_o}") |
|
print(f" QuotaInfoOffset = {qi_o}") |
|
flags_bits = [] |
|
if flg & 0x01: flags_bits.append('NewObject') |
|
if flg & 0x02: flags_bits.append('KernelObject') |
|
if flg & 0x04: flags_bits.append('CreatorInfo') |
|
if flg & 0x08: flags_bits.append('Exclusive') |
|
if flg & 0x10: flags_bits.append('Permanent') |
|
if flg & 0x20: flags_bits.append('SecurityInfo') |
|
if flg & 0x40: flags_bits.append('SingleProcess') |
|
if flg & 0x80: flags_bits.append('DeferDelete') |
|
print(f" Flags = 0x{flg:02x} [{', '.join(flags_bits) or '(none)'}]") |
|
if otp: |
|
tname = _read_unicode_string_at(otp + OT_NAME) |
|
if tname: |
|
print(f" Type->Name = \"{tname}\"") |
|
if ni_o: |
|
ni_addr = hdr - ni_o |
|
ni_name = _read_unicode_string_at(ni_addr + OHNI_NAME) |
|
ni_dir = read_u64(ni_addr + OHNI_DIRECTORY) |
|
if ni_name is not None: |
|
print(f" ObjName = \"{ni_name}\" Directory=0x{ni_dir:x}") |
|
|
|
ReactosObj() |
|
|
|
|
|
class ReactosHandles(gdb.Command): |
|
"""Walk EPROCESS->ObjectTable and dump in-use handles. |
|
Source: sdk/include/ndk/extypes.h:787 (_HANDLE_TABLE), |
|
ntoskrnl/ex/handle.c (TableCode levels: 0=low,1=mid,2=high). |
|
Usage: ros-handles <eprocess_addr> [max=64]""" |
|
def __init__(self): |
|
super().__init__("ros-handles", gdb.COMMAND_USER) |
|
def invoke(self, arg, from_tty): |
|
args = arg.strip().split() |
|
if not args: |
|
print("Usage: ros-handles <eprocess_addr> [max=64]") |
|
return |
|
try: |
|
eproc = int(args[0], 0) |
|
except Exception: |
|
print(f"Bad address: {args[0]}") |
|
return |
|
max_n = int(args[1], 0) if len(args) > 1 else 64 |
|
ot_off = _maybe_lookup_field_offset("struct _EPROCESS", "ObjectTable") |
|
if ot_off is None: |
|
print(" (cannot resolve EPROCESS.ObjectTable offset; symbols missing)") |
|
return |
|
ht = read_u64(eproc + ot_off) |
|
if not ht: |
|
print(f" EPROCESS->ObjectTable = NULL (terminating?)") |
|
return |
|
tc = read_u64(ht) # TableCode at offset 0 NTDDI_WINXP+ |
|
hc = read_u32(ht + 0x70) # HandleCount approximate |
|
if tc is None: |
|
print(f"Cannot read HANDLE_TABLE at 0x{ht:x}") |
|
return |
|
level = tc & 3 |
|
base = tc & ~3 |
|
print(f"=== HANDLE_TABLE @ 0x{ht:016x} (process 0x{eproc:x}) ===") |
|
print(f" TableCode = 0x{tc:x} Level={level} Base=0x{base:x} HandleCount~{hc}") |
|
# On amd64: HANDLE_TABLE_ENTRY = 16. LOW_LEVEL_ENTRIES = 4096/16 = 256. |
|
LOW = 256 |
|
MID = 512 # 4096/8 |
|
TAGBITS = 2 |
|
shown = 0 |
|
# Yield (handle_index, entry_addr) for each entry |
|
def iter_low(low_table_base, base_index): |
|
for i in range(LOW): |
|
yield base_index + i, low_table_base + i*16 |
|
def iter_mid(mid_table, base_index): |
|
for j in range(MID): |
|
low = read_u64(mid_table + j*8) |
|
if not low: |
|
continue |
|
yield from iter_low(low, base_index + j*LOW) |
|
if level == 0: |
|
it = iter_low(base, 0) |
|
elif level == 1: |
|
it = iter_mid(base, 0) |
|
else: |
|
def iter_high(): |
|
for k in range(MID): |
|
mid = read_u64(base + k*8) |
|
if not mid: |
|
continue |
|
yield from iter_mid(mid, k*MID*LOW) |
|
it = iter_high() |
|
printed_first_obj_chain = 0 |
|
for idx, ea in it: |
|
if shown >= max_n: |
|
print(f" ... truncated at {max_n}") |
|
break |
|
obj_attr = read_u64(ea + 0) |
|
grant = read_u32(ea + 8) |
|
if obj_attr is None or obj_attr == 0: |
|
continue |
|
# NextFreeTableEntry: low bits == 0, high == small int? we just skip 0. |
|
# Mask the attribute bits (bottom 3) to recover object pointer. |
|
obj = obj_attr & ~0x7 |
|
handle = (idx << TAGBITS) |
|
print(f" [{shown:3d}] Handle=0x{handle:04x} Object=0x{obj:016x}" |
|
f" Granted=0x{grant:08x}") |
|
if printed_first_obj_chain < 3 and obj: |
|
# Auto-resolve type name only (cheap) |
|
hdr = obj - OH_BODY |
|
otp = read_u64(hdr + OH_TYPE) |
|
if otp: |
|
tname = _read_unicode_string_at(otp + OT_NAME) |
|
if tname: |
|
print(f" Type=\"{tname}\"") |
|
printed_first_obj_chain += 1 |
|
shown += 1 |
|
|
|
ReactosHandles() |
|
|
|
|
|
# ============================================================ |
|
# CPU state : ros-prcb |
|
# ============================================================ |
|
# |
|
# Source: sdk/include/ndk/amd64/ketypes.h:653 (_KPRCB), :960 (_KIPCR). |
|
|
|
class ReactosPrcb(gdb.Command): |
|
"""Read KPCR via $gs_base, follow to KPRCB, dump key scheduler fields. |
|
Source: sdk/include/ndk/amd64/ketypes.h:960 (KIPCR), :653 (KPRCB). |
|
Usage: ros-prcb""" |
|
def __init__(self): |
|
super().__init__("ros-prcb", gdb.COMMAND_USER) |
|
def invoke(self, arg, from_tty): |
|
try: |
|
gs_base = int(gdb.parse_and_eval("$gs_base")) & 0xFFFFFFFFFFFFFFFF |
|
except Exception: |
|
try: |
|
gs_base = int(gdb.parse_and_eval("$gs.base")) & 0xFFFFFFFFFFFFFFFF |
|
except Exception: |
|
print("ros-prcb: cannot read $gs_base / $gs.base") |
|
return |
|
# KIPCR layout: NtTib union puts Self at offset 0x18 (NtTib.SubSystemTib) |
|
# CurrentPrcb at offset 0x20. Irql at 0x50. Prcb (embedded) at 0x180. |
|
try: |
|
self_pcr = read_u64(gs_base + 0x18) |
|
cur_prcb_p = read_u64(gs_base + 0x20) |
|
irql = read_u8(gs_base + 0x50) |
|
except Exception as e: |
|
print(f"ros-prcb read KPCR error: {e}") |
|
return |
|
prcb = gs_base + 0x180 # embedded Prcb |
|
if cur_prcb_p and cur_prcb_p != prcb: |
|
print(f" (note: CurrentPrcb=0x{cur_prcb_p:x} differs from embedded 0x{prcb:x})") |
|
prcb = cur_prcb_p |
|
elif cur_prcb_p is None: |
|
print(f" (note: CurrentPrcb unreadable; using embedded 0x{prcb:x})") |
|
try: |
|
mxcsr = read_u32(prcb + 0x00) |
|
number = read_u16(prcb + 0x04) |
|
int_req = read_u8(prcb + 0x06) |
|
idle_h = read_u8(prcb + 0x07) |
|
cur_thr = read_u64(prcb + 0x08) |
|
nxt_thr = read_u64(prcb + 0x10) |
|
idl_thr = read_u64(prcb + 0x18) |
|
except Exception as e: |
|
print(f"ros-prcb read KPRCB error: {e}") |
|
return |
|
print(f"=== KPCR @ 0x{gs_base:016x} ===") |
|
print(f" Self = {fmt_ptr(self_pcr)}") |
|
print(f" CurrentPrcb = {fmt_ptr(cur_prcb_p)}") |
|
print(f" Irql = 0x{irql:x}" if irql is not None else " Irql = ?") |
|
print(f"=== KPRCB @ 0x{prcb:016x} ===") |
|
print(f" MxCsr = 0x{mxcsr:08x}" if mxcsr is not None else " MxCsr = ?") |
|
print(f" Number = {number}" if number is not None else " Number = ?") |
|
print(f" InterruptRequest= {int_req if int_req is not None else '?'} IdleHalt={idle_h if idle_h is not None else '?'}") |
|
print(f" CurrentThread = {fmt_ptr(cur_thr)}") |
|
print(f" NextThread = {fmt_ptr(nxt_thr)}") |
|
print(f" IdleThread = {fmt_ptr(idl_thr)}") |
|
# BuildType/CpuStepping/Model — offsets vary; use gdb if symbols loaded. |
|
bt_off = _maybe_lookup_field_offset("struct _KPRCB", "BuildType") |
|
cs_off = _maybe_lookup_field_offset("struct _KPRCB", "CpuStepping") |
|
cm_off = _maybe_lookup_field_offset("struct _KPRCB", "CpuModel") |
|
if bt_off is not None: |
|
bt = read_u8(prcb + bt_off) |
|
print(f" BuildType = 0x{bt:x} (CHK={bt&1} MP={bool(bt&2)})") |
|
if cs_off is not None and cm_off is not None: |
|
cs = read_u8(prcb + cs_off); cm = read_u8(prcb + cm_off) |
|
print(f" CpuStepping/Model = 0x{cs:x} / 0x{cm:x}") |
|
# DpcData[2] — offset varies; try gdb |
|
dpc_off = _maybe_lookup_field_offset("struct _KPRCB", "DpcData") |
|
if dpc_off is not None: |
|
# KDPC_DATA: DpcListHead(16) DpcLock(8) DpcQueueDepth(LONG) DpcCount(LONG) |
|
for i in range(2): |
|
base = prcb + dpc_off + i*0x20 |
|
qd = read_u32(base + 0x18) |
|
cnt = read_u32(base + 0x1C) |
|
print(f" DpcData[{i}] DpcQueueDepth={qd} DpcCount={cnt}") |
|
if cur_thr: print(f" Use: ros-thread 0x{cur_thr:x}") |
|
|
|
ReactosPrcb() |
|
|
|
|
|
# ============================================================ |
|
# Pagefile / Working set : ros-pagefile, ros-wsl |
|
# ============================================================ |
|
# |
|
# Source: ntoskrnl/include/internal/mm.h:518 (_MMPAGING_FILE), |
|
# :29 (MmNumberOfPagingFiles), :532 (MmPagingFile[]). |
|
# sdk/include/ndk/mmtypes.h:869 (_MMWSL). |
|
|
|
# MMPAGING_FILE (amd64) |
|
PF_SIZE = 0x00 |
|
PF_MAX = 0x08 |
|
PF_MIN = 0x10 |
|
PF_FREESPACE = 0x18 |
|
PF_CUR_USAGE = 0x20 |
|
PF_FILE_OBJECT = 0x28 |
|
PF_PAGEFILE_NAME = 0x30 # UNICODE_STRING(16) |
|
PF_BITMAP = 0x40 |
|
PF_FILE_HANDLE = 0x48 |
|
|
|
class ReactosPagefile(gdb.Command): |
|
"""Dump MmPagingFile[] entries. Source: ntoskrnl/include/internal/mm.h:518. |
|
Usage: ros-pagefile [idx]""" |
|
def __init__(self): |
|
super().__init__("ros-pagefile", gdb.COMMAND_USER) |
|
def invoke(self, arg, from_tty): |
|
# Resolve MmPagingFile[] base and MmNumberOfPagingFiles |
|
try: |
|
arr = int(gdb.parse_and_eval("&MmPagingFile")) |
|
except Exception: |
|
print("ros-pagefile: MmPagingFile symbol unavailable") |
|
return |
|
try: |
|
n_addr = int(gdb.parse_and_eval("&MmNumberOfPagingFiles")) |
|
n = read_u32(n_addr) or 0 |
|
except Exception: |
|
n = 0 |
|
idx = None |
|
if arg.strip(): |
|
try: |
|
idx = int(arg.strip(), 0) |
|
except Exception: |
|
print(f"Bad idx: {arg.strip()}") |
|
return |
|
print(f"=== MmPagingFile[] MmNumberOfPagingFiles={n} ===") |
|
rng = range(idx, idx+1) if idx is not None else range(min(n or 16, 16)) |
|
for i in rng: |
|
pf = read_u64(arr + i*8) |
|
if pf is None: |
|
continue |
|
if pf == 0: |
|
if idx is not None: |
|
print(f" [{i}] (NULL)") |
|
continue |
|
sz = read_u64(pf + PF_SIZE) |
|
mxs = read_u64(pf + PF_MAX) |
|
mns = read_u64(pf + PF_MIN) |
|
free = read_u64(pf + PF_FREESPACE) |
|
cur = read_u64(pf + PF_CUR_USAGE) |
|
fo = read_u64(pf + PF_FILE_OBJECT) |
|
name = _read_unicode_string_at(pf + PF_PAGEFILE_NAME) or '' |
|
bm = read_u64(pf + PF_BITMAP) |
|
fh = read_u64(pf + PF_FILE_HANDLE) |
|
print(f" [{i}] MMPAGING_FILE @ 0x{pf:016x}") |
|
print(f" Name = \"{name}\"") |
|
print(f" Size/Min/Max = {sz} / {mns} / {mxs} pages") |
|
print(f" FreeSpace = {free} CurrentUsage={cur}") |
|
print(f" FileObject = 0x{fo:016x}") |
|
print(f" Bitmap = 0x{bm:016x}") |
|
print(f" FileHandle = 0x{fh:016x}") |
|
if fo: print(f" Use: ros-fileobj 0x{fo:x}") |
|
|
|
ReactosPagefile() |
|
|
|
|
|
class ReactosWsl(gdb.Command): |
|
"""Find Process->Vm.VmWorkingSetList (MMWSL) and dump it. |
|
Source: sdk/include/ndk/mmtypes.h:869, :943 (MMSUPPORT.VmWorkingSetList). |
|
Usage: ros-wsl <eprocess_addr>""" |
|
def __init__(self): |
|
super().__init__("ros-wsl", gdb.COMMAND_USER) |
|
def invoke(self, arg, from_tty): |
|
if not arg.strip(): |
|
print("Usage: ros-wsl <eprocess_addr>") |
|
return |
|
try: |
|
eproc = int(arg.strip(), 0) |
|
except Exception: |
|
print(f"Bad address: {arg.strip()}") |
|
return |
|
vm_off = _maybe_lookup_field_offset("struct _EPROCESS", "Vm") |
|
if vm_off is None: |
|
print(" (cannot resolve EPROCESS.Vm offset; symbols missing)") |
|
return |
|
# MMSUPPORT.VmWorkingSetList @ +0x30 (verified by ros-mmsupport above) |
|
wsl = read_u64(eproc + vm_off + 0x30) |
|
if not wsl: |
|
print(f" EPROCESS 0x{eproc:x} VmWorkingSetList = NULL") |
|
return |
|
try: |
|
ff = read_u32(wsl + 0x00) |
|
fdyn = read_u32(wsl + 0x04) |
|
last = read_u32(wsl + 0x08) |
|
nxt = read_u32(wsl + 0x0C) |
|
wsle = read_u64(wsl + 0x10) |
|
last_in = read_u32(wsl + 0x18) |
|
non_dir = read_u32(wsl + 0x1C) |
|
ht = read_u64(wsl + 0x20) |
|
ht_size = read_u32(wsl + 0x28) |
|
committed = read_u32(wsl + 0x2C) |
|
ht_start = read_u64(wsl + 0x30) |
|
ht_high = read_u64(wsl + 0x38) |
|
except Exception as e: |
|
print(f"ros-wsl read error: {e}") |
|
return |
|
if ff is None: |
|
print(f"Cannot read MMWSL at 0x{wsl:x}") |
|
return |
|
print(f"=== MMWSL @ 0x{wsl:016x} (EPROCESS 0x{eproc:x}) ===") |
|
print(f" FirstFree = {ff}") |
|
print(f" FirstDynamic = {fdyn}") |
|
print(f" LastEntry = {last}") |
|
print(f" NextSlot = {nxt}") |
|
print(f" Wsle (PMMWSLE) = 0x{wsle:016x}") |
|
print(f" LastInitializedWsle = {last_in}") |
|
print(f" NonDirectCount = {non_dir}") |
|
print(f" HashTable = 0x{ht:016x} (size={ht_size})") |
|
print(f" CommittedPageTables = {committed}") |
|
print(f" HashTableStart = 0x{ht_start:016x}") |
|
print(f" HighestPermittedHash = 0x{ht_high:016x}") |
|
# AddressCreationLock — chain to ros-pushlock if locatable |
|
acl_off = _maybe_lookup_field_offset("struct _EPROCESS", "AddressCreationLock") |
|
if acl_off is not None: |
|
print(f" Use: ros-pushlock 0x{eproc + acl_off:x} 0x<ethread>" |
|
f" # AddressCreationLock") |
|
|
|
ReactosWsl() |
|
|
|
|
|
class ReactosTriage(gdb.Command): |
|
"""Universal triage dump for any kdb-reached assertion/bugcheck. |
|
Runs: ros-where, ros-prcb, regs, bt, ros-fault, ros-pslist, ros-mmranges, |
|
then prints class-specific follow-up hints. Suitable for blind invocation |
|
by triage.sh on any asserted state, regardless of bug class.""" |
|
def __init__(self): |
|
super().__init__("ros-triage", gdb.COMMAND_USER) |
|
def invoke(self, arg, from_tty): |
|
cmds = [ |
|
("=== CPU mode ===", "ros-where"), |
|
("=== KPRCB ===", "ros-prcb"), |
|
("=== regs ===", "info reg"), |
|
("=== stack ===", "bt 30"), |
|
("=== ros-fault ===", "ros-fault"), |
|
("=== process list ===", "ros-pslist"), |
|
("=== ARM3 VA map ===", "ros-mmranges"), |
|
] |
|
for header, cmd in cmds: |
|
print(header) |
|
try: |
|
gdb.execute(cmd) |
|
except Exception as e: |
|
print(f" [{cmd} failed: {e}]") |
|
print("") |
|
print("=== Follow-up dumpers (run manually after classification) ===") |
|
print(" heap.c:1185 (freelist) : ros-heap <heap>; check rsi/rdi for heap ptr") |
|
print(" heap.c:448 (bitmap) : ros-heap-freelist <heap> <size>") |
|
print(" Cc 0x23 (MmMapView) : ros-secptr <fileobj>; ros-cachemap <scm>;") |
|
print(" ros-section <section>") |
|
print(" 0x1A Bad PTE : ros-pte-va <va>; ros-pfn <pfn>;") |
|
print(" ros-translate <cr3> <va>") |
|
print(" Rmap dup (sec.c:1967) : grep 'Rmap trace match' /tmp/repro26-com1.log") |
|
print(" (MmDumpRmapTrace already in-kernel)") |
|
print(" Heap dup/coalesce : grep 'Heap trace match' /tmp/repro26-com1.log") |
|
print(" (RtlpDumpHeapTrace already in-kernel)") |
|
print(" Unknown : ros-trapframes; ros-callchain $rsp") |
|
|
|
ReactosTriage() |
|
|
|
|
|
# ============================================================ |
|
# Done — print available commands |
|
# ============================================================ |
|
|
|
print("") |
|
print("=== ReactOS GDB helpers loaded (non-aggressive) ===") |
|
print("Connect with: ros-connect [:port]") |
|
print("") |
|
print("Commands (after connecting):") |
|
print(" ros-where Show CPU context (kernel/user64/compat32)") |
|
print(" ros-load-symbols Auto-find ntoskrnl and load symbols") |
|
print(" ros-load-at <addr> Load ntoskrnl at a known base") |
|
print(" ros-load-module <a> <p> Load symbols for any module") |
|
print(" ros-lsmod [--load] List/load kernel modules") |
|
print(" ros-addr2mod <addr> Find module owning an address") |
|
print(" ros-thread <addr> Inspect KTHREAD") |
|
print(" ros-waiters <addr> Find waiters on dispatcher object") |
|
print(" ros-irp <addr> Inspect IRP") |
|
print(" ros-fault [tf_addr] Dump faulting context from trap frame (RBP)") |
|
print(" ros-trapframes [lo] [hi] Scan for KTRAP_FRAMEs (incl CS=0x23)") |
|
print(" ros-frame-regs <rsp> [pat] Recover pushed non-volatile regs") |
|
print(" ros-callchain [rsp] [n] Heuristic stack scan") |
|
print(" ros-kdb-bt [rsp] [n] Asserting callchain past kdb/RtlAssert preamble") |
|
print(" ros-findmod <addr> Find module base by MZ scan") |
|
print(" ros-verify <base> <path> Compare loaded vs build binary") |
|
print(" ros-pslist List active processes") |
|
print(" ros-findproc <name> Find process by name") |
|
print(" ros-threads <eproc> List threads and stack symbols for a process") |
|
print(" ros-gdt [base] [limit] Dump GDT entries") |
|
print(" ros-xp <cr3> <va> [n] Read virtual mem from any process (via page tables)") |
|
print(" ros-translate <cr3> <va> Walk page tables, show PML4/PDPT/PD/PT entries") |
|
print("Pool debugging:") |
|
print(" ros-pool-page <addr> Walk pool blocks on a page") |
|
print(" ros-pool-block <addr> Inspect a single pool block") |
|
print(" ros-pool-scan <start> [n] Scan for pool corruption") |
|
print(" ros-pool-find <tag> <start> [n] Find blocks by tag") |
|
print(" ros-pool-crash Auto-analyze pool crash") |
|
print("MM diagnostics:") |
|
print(" ros-rmap-arm [base] Arm 4 hbreaks for MM rmap leak diagnostic") |
|
print("ARM3 PTE / PFN:") |
|
print(" ros-pte <pte_addr> Decode 64-bit PTE at cell address (Hard/Soft/Trans/Proto)") |
|
print(" ros-pte-va <va> Find self-mapped PTE for VA and decode it") |
|
print(" ros-pfn <pfn-or-addr> Decode MMPFN: ShareCount, e1, OriginalPte, u4.PteFrame") |
|
print(" ros-rmap <pfn-or-addr> Walk per-PFN rmap chain: each (Process, Address) entry") |
|
print(" ros-proto-walk <ca> [max] Walk CONTROL_AREA's proto array, flag Hard.Valid violations") |
|
print(" ros-mapping-walk <va> <n> ros-pte-va for n pages of a contiguous VA range") |
|
print("MM struct dumpers:") |
|
print(" ros-section <addr> Decode _SECTION; chains to ros-segment") |
|
print(" ros-segment <addr> Decode _SEGMENT; suggests ros-controlarea / ros-proto-walk") |
|
print(" ros-controlarea <addr> Decode _CONTROL_AREA + flags; chains to embedded SUBSECTION") |
|
print(" ros-subsection <addr> Decode _SUBSECTION; suggests ros-proto-walk") |
|
print(" ros-secptr <addr> Decode _SECTION_OBJECT_POINTERS; chains to cachemap/CA") |
|
print(" ros-cachemap <addr> Decode _ROS_SHARED_CACHE_MAP; suggests ros-vacb-list") |
|
print(" ros-vacb-list <scm> [max] Walk CacheMapVacbListHead, list each ROS_VACB") |
|
print(" ros-vacb <addr> Decode one _ROS_VACB") |
|
print(" ros-vad <addr> Decode _MMVAD/_MMVAD_SHORT (VA range, flags, CA)") |
|
print(" ros-vad-tree <eproc> [max] Walk EPROCESS->VadRoot AVL tree, dump each VAD") |
|
print(" ros-mmsupport <eproc> Dump _MMSUPPORT (WS sizes, fault count, flags)") |
|
print(" ros-mmranges One-screen ARM3 VA map (pools/cache/PTE_BASE/etc.)") |
|
print("Locks/Sync:") |
|
print(" ros-eresource <addr> Decode ERESOURCE (active/contention/owner/waiters)") |
|
print(" ros-pushlock <addr> [eth] Decode EX_PUSH_LOCK; optional DBG HeldPushLocks walk") |
|
print(" ros-dispobj <addr> Decode DISPATCHER_HEADER + walk WaitListHead/KWAIT_BLOCKs") |
|
print("Heap (RTL user-mode):") |
|
print(" ros-heap <addr> _HEAP header + FreeHintBitmap summary") |
|
print(" ros-heap-freelist <h> [sz] Walk Heap->FreeLists or FreeHints[size_class]") |
|
print("I/O:") |
|
print(" ros-fileobj <addr> Decode FILE_OBJECT (FileName, flags, chains)") |
|
print(" ros-vpb <addr> Decode VPB (volume label, refcount, flags)") |
|
print(" ros-devobj <addr> Decode DEVICE_OBJECT (driver, stack, vpb, ext)") |
|
print(" ros-drvobj <addr> Decode DRIVER_OBJECT + resolve MajorFunction[] symbols") |
|
print("Object manager:") |
|
print(" ros-obj <addr> Walk to OBJECT_HEADER; resolve Type->Name + ObjName") |
|
print(" ros-handles <eproc> [max] Walk EPROCESS->ObjectTable, dump in-use handles") |
|
print("CPU state:") |
|
print(" ros-prcb Read $gs_base KPCR/KPRCB (cur/idle thread, irql, dpc)") |
|
print("Pagefile / Working set:") |
|
print(" ros-pagefile [idx] Dump MmPagingFile[idx] or summarize all") |
|
print(" ros-wsl <eproc> Dump EPROCESS->Vm.VmWorkingSetList (MMWSL)") |
|
print("Triage:") |
|
print(" ros-triage Universal dump for any kdb-reached assert") |
|
print("") |
|
|
|
end |