| # |
| # GDB debugging support |
| # |
| # Copyright 2012 Red Hat, Inc. and/or its affiliates |
| # |
| # Authors: |
| # Avi Kivity <avi@redhat.com> |
| # |
| # This work is licensed under the terms of the GNU GPL, version 2 |
| # or later. See the COPYING file in the top-level directory. |
| |
| import atexit |
| import gdb |
| import os |
| import pty |
| import re |
| import struct |
| import textwrap |
| |
| from collections import OrderedDict |
| from copy import deepcopy |
| |
| VOID_PTR = gdb.lookup_type('void').pointer() |
| |
| # Registers in the same order they're present in ELF coredump file. |
| # See asm/ptrace.h |
| PT_REGS = ['r15', 'r14', 'r13', 'r12', 'rbp', 'rbx', 'r11', 'r10', 'r9', |
| 'r8', 'rax', 'rcx', 'rdx', 'rsi', 'rdi', 'orig_rax', 'rip', 'cs', |
| 'eflags', 'rsp', 'ss'] |
| |
| coredump = None |
| |
| |
| class Coredump: |
| _ptregs_suff = '.ptregs' |
| |
| def __init__(self, coredump, executable): |
| atexit.register(self._cleanup) |
| |
| self.coredump = coredump |
| self.executable = executable |
| self._ptregs_blob = coredump + self._ptregs_suff |
| self._dirty = False |
| |
| with open(coredump, 'rb') as f: |
| while f.read(4) != b'CORE': |
| pass |
| gdb.write(f'core file {coredump}: found "CORE" at 0x{f.tell():x}\n') |
| |
| # Looking for struct elf_prstatus and pr_reg field in it (an array |
| # of general purpose registers). See sys/procfs.h. |
| |
| # lseek(f.fileno(), 4, SEEK_CUR): go to elf_prstatus |
| f.seek(4, 1) |
| |
| # lseek(f.fileno(), 112, SEEK_CUR): |
| # offsetof(struct elf_prstatus, pr_reg) |
| f.seek(112, 1) |
| |
| self._ptregs_offset = f.tell() |
| |
| # If binary blob with the name /path/to/coredump + '.ptregs' |
| # exists, that means proper cleanup didn't happen during previous |
| # GDB session with the same coredump, and registers in the dump |
| # itself might've remained patched. Thus we restore original |
| # registers values from this blob |
| if os.path.exists(self._ptregs_blob): |
| with open(self._ptregs_blob, 'rb') as b: |
| orig_ptregs_bytes = b.read() |
| self._dirty = True |
| else: |
| orig_ptregs_bytes = f.read(len(PT_REGS) * 8) |
| |
| values = struct.unpack(f"={len(PT_REGS)}q", orig_ptregs_bytes) |
| self._orig_ptregs = OrderedDict(zip(PT_REGS, values)) |
| |
| if not os.path.exists(self._ptregs_blob): |
| gdb.write(f'saving original pt_regs in {self._ptregs_blob}\n') |
| with open(self._ptregs_blob, 'wb') as b: |
| b.write(orig_ptregs_bytes) |
| |
| gdb.write('\n') |
| |
| def patch_regs(self, regs): |
| # Set dirty flag early on to make sure regs are restored upon cleanup |
| self._dirty = True |
| |
| gdb.write(f'patching core file {self.coredump}\n') |
| patched_ptregs = deepcopy(self._orig_ptregs) |
| int_regs = {k: int(v) for k, v in regs.items()} |
| patched_ptregs.update(int_regs) |
| |
| with open(self.coredump, 'ab') as f: |
| gdb.write(f'assume pt_regs at 0x{self._ptregs_offset:x}\n') |
| f.seek(self._ptregs_offset, 0) |
| gdb.write('writing regs:\n') |
| for reg in self._orig_ptregs.keys(): |
| if reg in int_regs: |
| gdb.write(f" {reg}: {int_regs[reg]:#16x}\n") |
| f.write(struct.pack(f"={len(PT_REGS)}q", *patched_ptregs.values())) |
| |
| gdb.write('\n') |
| |
| def restore_regs(self): |
| if not self._dirty: |
| return |
| |
| gdb.write(f'\nrestoring original regs in core file {self.coredump}\n') |
| with open(self.coredump, 'ab') as f: |
| gdb.write(f'assume pt_regs at 0x{self._ptregs_offset:x}\n') |
| f.seek(self._ptregs_offset, 0) |
| f.write(struct.pack(f"={len(PT_REGS)}q", |
| *self._orig_ptregs.values())) |
| |
| self._dirty = False |
| gdb.write('\n') |
| |
| def _cleanup(self): |
| if os.path.exists(self._ptregs_blob): |
| self.restore_regs() |
| gdb.write(f'\nremoving saved pt_regs file {self._ptregs_blob}\n') |
| os.unlink(self._ptregs_blob) |
| |
| |
| def pthread_self(): |
| '''Fetch the base address of TLS.''' |
| return gdb.parse_and_eval("$fs_base") |
| |
| def get_glibc_pointer_guard(): |
| '''Fetch glibc pointer guard value''' |
| fs_base = pthread_self() |
| return gdb.parse_and_eval('*(uint64_t*)((uint64_t)%s + 0x30)' % fs_base) |
| |
| def glibc_ptr_demangle(val, pointer_guard): |
| '''Undo effect of glibc's PTR_MANGLE()''' |
| return gdb.parse_and_eval('(((uint64_t)%s >> 0x11) | ((uint64_t)%s << (64 - 0x11))) ^ (uint64_t)%s' % (val, val, pointer_guard)) |
| |
| def get_jmpbuf_regs(jmpbuf): |
| JB_RBX = 0 |
| JB_RBP = 1 |
| JB_R12 = 2 |
| JB_R13 = 3 |
| JB_R14 = 4 |
| JB_R15 = 5 |
| JB_RSP = 6 |
| JB_PC = 7 |
| |
| pointer_guard = get_glibc_pointer_guard() |
| return {'rbx': jmpbuf[JB_RBX], |
| 'rbp': glibc_ptr_demangle(jmpbuf[JB_RBP], pointer_guard), |
| 'rsp': glibc_ptr_demangle(jmpbuf[JB_RSP], pointer_guard), |
| 'r12': jmpbuf[JB_R12], |
| 'r13': jmpbuf[JB_R13], |
| 'r14': jmpbuf[JB_R14], |
| 'r15': jmpbuf[JB_R15], |
| 'rip': glibc_ptr_demangle(jmpbuf[JB_PC], pointer_guard) } |
| |
| def symbol_lookup(addr): |
| # Example: "__clone3 + 44 in section .text of /lib64/libc.so.6" |
| result = gdb.execute(f"info symbol {hex(addr)}", to_string=True).strip() |
| try: |
| if "+" in result: |
| (func, result) = result.split(" + ") |
| (offset, result) = result.split(" in ") |
| else: |
| offset = "0" |
| (func, result) = result.split(" in ") |
| func_str = f"{func}<+{offset}> ()" |
| except: |
| return f"??? ({result})" |
| |
| # Example: Line 321 of "../util/coroutine-ucontext.c" starts at address |
| # 0x55cf3894d993 <qemu_coroutine_switch+99> and ends at 0x55cf3894d9ab |
| # <qemu_coroutine_switch+123>. |
| result = gdb.execute(f"info line *{hex(addr)}", to_string=True).strip() |
| if not result.startswith("Line "): |
| return func_str |
| result = result[5:] |
| |
| try: |
| result = result.split(" starts ")[0] |
| (line, path) = result.split(" of ") |
| path = path.replace("\"", "") |
| except: |
| return func_str |
| |
| return f"{func_str} at {path}:{line}" |
| |
| def run_with_pty(cmd): |
| # Create a PTY pair |
| master_fd, slave_fd = pty.openpty() |
| |
| pid = os.fork() |
| if pid == 0: # Child |
| os.close(master_fd) |
| # Attach stdin/stdout/stderr to the PTY slave side |
| os.dup2(slave_fd, 0) |
| os.dup2(slave_fd, 1) |
| os.dup2(slave_fd, 2) |
| os.close(slave_fd) |
| os.execvp("gdb", cmd) # Runs gdb and doesn't return |
| |
| # Parent |
| os.close(slave_fd) |
| |
| output = bytearray() |
| try: |
| while True: |
| data = os.read(master_fd, 65536) |
| if not data: |
| break |
| output.extend(data) |
| except OSError: # in case subprocess exits and we get EBADF on read() |
| pass |
| finally: |
| try: |
| os.close(master_fd) |
| except OSError: # in case we get EBADF on close() |
| pass |
| |
| # Wait for child to finish (reap zombie) |
| os.waitpid(pid, 0) |
| |
| return output.decode('utf-8') |
| |
| def dump_backtrace_patched(regs): |
| cmd = ['gdb', '-batch', |
| '-ex', 'set debuginfod enabled off', |
| '-ex', 'set complaints 0', |
| '-ex', 'set style enabled on', |
| '-ex', 'python print("----split----")', |
| '-ex', 'bt', coredump.executable, coredump.coredump] |
| |
| coredump.patch_regs(regs) |
| out = run_with_pty(cmd).split('----split----')[1] |
| gdb.write(out) |
| |
| def dump_backtrace(regs): |
| ''' |
| Backtrace dump with raw registers, mimic GDB command 'bt'. |
| ''' |
| # Here only rbp and rip that matter.. |
| rbp = regs['rbp'] |
| rip = regs['rip'] |
| i = 0 |
| |
| while rbp: |
| # For all return addresses on stack, we want to look up symbol/line |
| # on the CALL command, because the return address is the next |
| # instruction instead of the CALL. Here -1 would work for any |
| # sized CALL instruction. |
| print(f"#{i} {hex(rip)} in {symbol_lookup(rip if i == 0 else rip-1)}") |
| rip = gdb.parse_and_eval(f"*(uint64_t *)(uint64_t)({hex(rbp)} + 8)") |
| rbp = gdb.parse_and_eval(f"*(uint64_t *)(uint64_t)({hex(rbp)})") |
| i += 1 |
| |
| def dump_backtrace_live(regs): |
| ''' |
| Backtrace dump with gdb's 'bt' command, only usable in a live session. |
| ''' |
| old = dict() |
| |
| # remember current stack frame and select the topmost |
| # so that register modifications don't wreck it |
| selected_frame = gdb.selected_frame() |
| gdb.newest_frame().select() |
| |
| for i in regs: |
| old[i] = gdb.parse_and_eval('(uint64_t)$%s' % i) |
| |
| for i in regs: |
| gdb.execute('set $%s = %s' % (i, regs[i])) |
| |
| gdb.execute('bt') |
| |
| for i in regs: |
| gdb.execute('set $%s = %s' % (i, old[i])) |
| |
| selected_frame.select() |
| |
| def bt_jmpbuf(jmpbuf, detailed=False): |
| '''Backtrace a jmpbuf''' |
| regs = get_jmpbuf_regs(jmpbuf) |
| try: |
| # This reuses gdb's "bt" command, which can be slightly prettier |
| # but only works with live sessions. |
| dump_backtrace_live(regs) |
| except: |
| if detailed: |
| # Obtain detailed trace by patching regs in copied coredump |
| dump_backtrace_patched(regs) |
| else: |
| # If above doesn't work, fallback to poor man's unwind |
| dump_backtrace(regs) |
| |
| def co_cast(co): |
| return co.cast(gdb.lookup_type('CoroutineUContext').pointer()) |
| |
| def coroutine_to_jmpbuf(co): |
| coroutine_pointer = co_cast(co) |
| return coroutine_pointer['env']['__jmpbuf'] |
| |
| def init_coredump(): |
| global coredump |
| |
| files = gdb.execute('info files', False, True).split('\n') |
| |
| if not 'core dump' in files[1]: |
| return False |
| |
| core_path = re.search("`(.*)'", files[2]).group(1) |
| exec_path = re.match('^Symbols from "(.*)".$', files[0]).group(1) |
| |
| if coredump is None: |
| coredump = Coredump(core_path, exec_path) |
| |
| return True |
| |
| class CoroutineCommand(gdb.Command): |
| __doc__ = textwrap.dedent("""\ |
| Display coroutine backtrace |
| |
| Usage: qemu coroutine COROPTR [--detailed] |
| Show backtrace for a coroutine specified by COROPTR |
| |
| --detailed obtain detailed trace by copying coredump, patching |
| regs in it, and runing gdb subprocess to get |
| backtrace from the patched coredump |
| """) |
| |
| def __init__(self): |
| gdb.Command.__init__(self, 'qemu coroutine', gdb.COMMAND_DATA, |
| gdb.COMPLETE_NONE) |
| |
| def _usage(self): |
| gdb.write('usage: qemu coroutine <coroutine-pointer> [--detailed]\n') |
| return |
| |
| def invoke(self, arg, from_tty): |
| argv = gdb.string_to_argv(arg) |
| argc = len(argv) |
| if argc == 0 or argc > 2 or (argc == 2 and argv[1] != '--detailed'): |
| return self._usage() |
| detailed = True if argc == 2 else False |
| |
| is_coredump = init_coredump() |
| if detailed and not is_coredump: |
| gdb.write('--detailed is only valid when debugging core dumps\n') |
| return |
| |
| try: |
| bt_jmpbuf(coroutine_to_jmpbuf(gdb.parse_and_eval(argv[0])), |
| detailed=detailed) |
| finally: |
| coredump.restore_regs() |
| |
| class CoroutineBt(gdb.Command): |
| __doc__ = textwrap.dedent("""\ |
| Display backtrace including coroutine switches |
| |
| Usage: qemu bt [--detailed] |
| |
| --detailed obtain detailed trace by copying coredump, patching |
| regs in it, and runing gdb subprocess to get |
| backtrace from the patched coredump |
| """) |
| |
| def __init__(self): |
| gdb.Command.__init__(self, 'qemu bt', gdb.COMMAND_STACK, |
| gdb.COMPLETE_NONE) |
| |
| def _usage(self): |
| gdb.write('usage: qemu bt [--detailed]\n') |
| return |
| |
| def invoke(self, arg, from_tty): |
| argv = gdb.string_to_argv(arg) |
| argc = len(argv) |
| if argc > 1 or (argc == 1 and argv[0] != '--detailed'): |
| return self._usage() |
| detailed = True if argc == 1 else False |
| |
| is_coredump = init_coredump() |
| if detailed and not is_coredump: |
| gdb.write('--detailed is only valid when debugging core dumps\n') |
| return |
| |
| gdb.execute("bt") |
| |
| try: |
| # This only works with a live session |
| co_ptr = gdb.parse_and_eval("qemu_coroutine_self()") |
| except: |
| # Fallback to use hard-coded ucontext vars if it's coredump |
| co_ptr = gdb.parse_and_eval("co_tls_current") |
| |
| if co_ptr == False: |
| return |
| |
| try: |
| while True: |
| co = co_cast(co_ptr) |
| co_ptr = co["base"]["caller"] |
| if co_ptr == 0: |
| break |
| gdb.write("\nCoroutine at " + str(co_ptr) + ":\n") |
| bt_jmpbuf(coroutine_to_jmpbuf(co_ptr), detailed=detailed) |
| finally: |
| coredump.restore_regs() |
| |
| class CoroutineSPFunction(gdb.Function): |
| def __init__(self): |
| gdb.Function.__init__(self, 'qemu_coroutine_sp') |
| |
| def invoke(self, addr): |
| return get_jmpbuf_regs(coroutine_to_jmpbuf(addr))['rsp'].cast(VOID_PTR) |
| |
| class CoroutinePCFunction(gdb.Function): |
| def __init__(self): |
| gdb.Function.__init__(self, 'qemu_coroutine_pc') |
| |
| def invoke(self, addr): |
| return get_jmpbuf_regs(coroutine_to_jmpbuf(addr))['rip'].cast(VOID_PTR) |
