"""gdb-driver: embedded RPC plugin (Milestone 1 skeleton). Load inside gdb: gdb -q -x driver.py --args ./prog ... Architecture (see TUI_DRIVING_BLUEPRINT.md): - A background thread owns a unix-socket JSONL server. - gdb's Python API is MAIN-THREAD-ONLY, so any handler that touches gdb state is marshalled onto gdb's thread via gdb.post_event() and we block on a threading.Event for the result. (Proven: post_event fires while gdb sits idle at the prompt.) - Wire protocol: {"id":N,"method":str,"params":{}}\\n -> {"id":N,"result":...} | {"id":N,"error":{"message":...}} Milestone 1 surface: ping / methods / quit (lifecycle only). """ import gdb import json import os import socket import stat import subprocess import sys import threading import traceback PROTO = 1 MAIN_THREAD_TIMEOUT = 30.0 # seconds to wait for a main-thread callback # -------------------------------------------------------------------------- # socket path resolution # -------------------------------------------------------------------------- def _runtime_dir(): base = os.environ.get("XDG_RUNTIME_DIR") or f"/tmp/gdb-driver-{os.getuid()}" d = os.path.join(base, "gdb-driver") os.makedirs(d, mode=0o700, exist_ok=True) return d def _sock_path(): p = os.environ.get("GDB_DRIVER_SOCK") if p: return p return os.path.join(_runtime_dir(), f"{os.getpid()}.sock") # -------------------------------------------------------------------------- # main-thread marshalling # -------------------------------------------------------------------------- def on_main(fn, timeout=MAIN_THREAD_TIMEOUT): """Run fn() on gdb's main thread; return its value or raise its exception.""" box = {} done = threading.Event() def cb(): try: box["ok"] = fn() except Exception as e: # noqa: BLE001 - propagate to caller box["err"] = e finally: done.set() gdb.post_event(cb) if not done.wait(timeout): raise TimeoutError(f"main-thread callback timed out after {timeout}s") if "err" in box: raise box["err"] return box.get("ok") # -------------------------------------------------------------------------- # adapter (grows per milestone; M1 = lifecycle) # -------------------------------------------------------------------------- # gdb prints exactly ONE prompt at startup (before the interactive loop) and # does NOT reprint after a programmatic gdb.execute(). So: consume that one # pending prompt for the first driven command, then render our own prompt for # every command after -- giving a consistent 'gef> cmd' line every time. _PROMPT_PENDING = True def _current_prompt(): """The live (GEF-colored) prompt string, matching what a human sees.""" try: hook = getattr(gdb, "prompt_hook", None) if hook: p = hook(lambda: "") if p: return p except Exception: # noqa: BLE001 pass try: return gdb.parameter("prompt") or "(gdb) " except Exception: # noqa: BLE001 return "(gdb) " def echo_cmd(cmdline): """Make a programmatically-driven command look hand-typed in the pane.""" global _PROMPT_PENDING try: prompt = "" if _PROMPT_PENDING else _current_prompt() _PROMPT_PENDING = False gdb.write(prompt + cmdline + "\n", gdb.STDOUT) gdb.flush() except Exception: # noqa: BLE001 - echoing must never break driving pass def _sal_of(frame): """(file, line) for a frame, or (None, None).""" try: sal = frame.find_sal() if sal and sal.symtab: return sal.symtab.filename, sal.line except gdb.error: pass return None, None def _sym_for(pc): """'func+0xoff' for a pc via the (minimal) symbol table, else None. Works for no-debug binaries where find_sal() has no file/line.""" try: out = gdb.execute(f"info symbol {pc:#x}", to_string=True).strip() except gdb.error: return None if not out or out.startswith("No symbol"): return None # 'add + 10 in section .text of /path' -> 'add+0x10' part = out.split(" in section")[0].strip() if " + " in part: name, off = part.split(" + ", 1) try: return f"{name}+{int(off):#x}" except ValueError: return part.replace(" + ", "+") return part def _require_frame(): """Return the selected frame or raise a clean 'not ready' error.""" try: return gdb.selected_frame() except gdb.error: raise RuntimeError("not ready: no stack frame (inferior not stopped)") class GdbAdapter: def __init__(self): self.ready = False self.pane = os.environ.get("TMUX_PANE") # --- method table -------------------------------------------------- def methods(self): return [ {"name": "ping", "tier": "lifecycle", "desc": "liveness; {ok,proto,ready,pid}"}, {"name": "methods", "tier": "lifecycle", "desc": "this verb table (self-documentation)"}, {"name": "quit", "tier": "lifecycle", "desc": "reply, then gdb quit"}, {"name": "state", "tier": "read", "desc": "{status,pc,function,file,line,thread}"}, {"name": "regs", "tier": "read", "desc": "general-purpose registers {name:hex}"}, {"name": "bt", "tier": "read", "desc": "backtrace [{level,pc,func,file,line}]; param n"}, {"name": "mem", "tier": "read", "desc": "read memory; params addr(expr), len, [fmt]"}, {"name": "screen", "tier": "read", "desc": "tmux capture-pane of the gdb pane; param color"}, {"name": "breakpoints", "tier": "read", "desc": "list breakpoints [{num,type,loc,enabled,hits}]"}, {"name": "run/start/cont/next/step/finish/until", "tier": "semantic", "desc": "run-control; settles at next stop"}, {"name": "break/tbreak/delete/watch", "tier": "semantic", "desc": "breakpoint mgmt"}, {"name": "print/x/set/frame/up/down", "tier": "semantic", "desc": "eval / examine / assign / frame nav"}, {"name": "cmd", "tier": "semantic", "desc": "run ANY gdb/GEF command; param line. escape hatch " "(use run-control verbs for continue/step -- cmd does " "not settle on stop)"}, ] # --- lifecycle ----------------------------------------------------- def ping(self): return {"ok": True, "proto": PROTO, "ready": self.ready, "pid": os.getpid()} # --- read tier (all run on gdb's main thread) ---------------------- def state(self): inf = gdb.selected_inferior() if inf is None or not inf.is_valid() or inf.pid == 0: return {"status": "no-inferior"} try: thr = gdb.selected_thread() tnum = thr.num if thr else None except gdb.error: tnum = None try: fr = gdb.selected_frame() except gdb.error: return {"status": "running", "thread": tnum} pc = int(fr.pc()) f, ln = _sal_of(fr) loc = f"{f}:{ln}" if f else (_sym_for(pc) or hex(pc)) return {"status": "stopped", "pc": hex(pc), "function": fr.name(), "file": f, "line": ln, "loc": loc, "thread": tnum} def regs(self): fr = _require_frame() arch = fr.architecture() out = {} for rd in arch.registers("general"): try: val = fr.read_register(rd) # mask to the register's OWN width, not a hardcoded 64 bits -- # otherwise a 32-bit reg with the high bit set sign-extends to # a bogus 0xffffffffXXXXXXXX value. size = val.type.sizeof or 8 out[rd.name] = hex(int(val) & ((1 << (8 * size)) - 1)) except (gdb.error, ValueError, gdb.MemoryError): out[rd.name] = str(val) return out def bt(self, n=None): _require_frame() frames = [] fr = gdb.newest_frame() lvl = 0 while fr is not None and (n is None or lvl < n): pc = int(fr.pc()) f, ln = _sal_of(fr) loc = f"{f}:{ln}" if f else (_sym_for(pc) or hex(pc)) frames.append({"level": lvl, "pc": hex(pc), "func": fr.name(), "file": f, "line": ln, "loc": loc}) try: fr = fr.older() except gdb.error: break lvl += 1 return frames def mem(self, addr, len=64, fmt="xb"): inf = gdb.selected_inferior() if inf is None or inf.pid == 0: raise RuntimeError("not ready: no inferior") base = int(gdb.parse_and_eval(str(addr))) length = int(len) raw = bytes(inf.read_memory(base, length)) lines = [] for off in range(0, length, 16): chunk = raw[off:off + 16] hexs = " ".join(f"{b:02x}" for b in chunk) asc = "".join(chr(b) if 32 <= b < 127 else "." for b in chunk) lines.append(f"{base + off:#018x} {hexs:<47} {asc}") return {"addr": hex(base), "len": length, "hex": raw.hex(), "dump": "\n".join(lines)} def breakpoints(self): out = [] for b in gdb.breakpoints(): out.append({"num": b.number, "type": b.type, "loc": b.location, "expr": b.expression, "enabled": b.enabled, "hits": b.hit_count, "temporary": b.temporary}) return out # --- mutating tier ------------------------------------------------- # run-control: execute WITHOUT to_string so GEF's context renders in the # viewer pane; gdb.execute blocks until the next stop (sync all-stop mode) # => settled the moment it returns. Return fresh structured state. def _run(self, cmdline): echo_cmd(cmdline) gdb.execute(cmdline, to_string=False) return {"cmd": cmdline, "state": self.state()} # quick command: capture output for the agent AND replay it into the pane # so the livestream shows both the command and its result. def _cmd(self, cmdline): echo_cmd(cmdline) out = gdb.execute(cmdline, to_string=True) if out: gdb.write(out if out.endswith("\n") else out + "\n") gdb.flush() return {"cmd": cmdline, "output": out.rstrip("\n"), "state": self.state()} # NB: run-control verbs are NOT here -- they need stop-event settle and # are orchestrated from the socket thread (see run_control_dispatch). def brk(self, loc): return self._cmd(f"break {loc}") def tbreak(self, loc): return self._cmd(f"tbreak {loc}") def delete(self, n=None): return self._cmd("delete" + (f" {n}" if n is not None else "")) def watch(self, expr): return self._cmd(f"watch {expr}") def eval(self, expr): return self._cmd(f"print {expr}") def examine(self, fmt, addr): return self._cmd(f"x/{fmt} {addr}") def setvar(self, expr): return self._cmd(f"set var {expr}") def cmd(self, line): return self._cmd(line) def frame(self, n): return self._run(f"frame {int(n)}") def up(self, n=1): return self._run(f"up {int(n)}") def down(self, n=1): return self._run(f"down {int(n)}") # --- screen (does NOT touch gdb state; runs on socket thread) ------- def screen(self, color=False): if not self.pane: raise RuntimeError("no TMUX_PANE (not spawned inside tmux?)") cmd = ["tmux", "capture-pane", "-p", "-t", self.pane] if color: cmd.insert(2, "-e") text = subprocess.check_output(cmd, text=True) return {"pane": self.pane, "text": text} ADAPTER = GdbAdapter() # read verbs (touch gdb state; quick) _READ = {"state", "regs", "bt", "mem", "breakpoints"} # mutating verbs: wire name -> adapter attribute _MUT = { "break": "brk", "tbreak": "tbreak", "delete": "delete", "watch": "watch", "print": "eval", "x": "examine", "set": "setvar", "frame": "frame", "up": "up", "down": "down", "cmd": "cmd", } # run-control verbs may execute arbitrary inferior code -> generous timeout _RUNCONTROL = {"run", "start", "cont", "continue", "next", "step", "nexti", "stepi", "finish", "until"} RUNCONTROL_TIMEOUT = 120.0 def _rc_cmdline(method, params): if method == "run": a = params.get("args") return "run" + (f" {a}" if a else "") if method == "start": return "start" if method in ("cont", "continue"): return "continue" if method in ("next", "step", "nexti", "stepi"): return f"{method} {int(params.get('n', 1))}" if method == "finish": return "finish" if method == "until": loc = params.get("loc") return "until" + (f" {loc}" if loc else "") raise ValueError(f"not a run-control verb: {method}") def run_control_dispatch(method, params): """Orchestrated on the SOCKET thread. gdb.execute() run-control returns async under post_event, so we settle on the real stop/exited event. The main thread must stay free to *deliver* that event, hence we only marshal the execute() + state read onto it, and wait here.""" cmdline = _rc_cmdline(method, params) settled = threading.Event() err = {} def on_stop(_evt): settled.set() def on_exit(_evt): settled.set() def arm_and_go(): # arm listeners then fire the command, all on the main thread so the # ordering is race-free even if execute() blocks until stop (sync mode) gdb.events.stop.connect(on_stop) gdb.events.exited.connect(on_exit) echo_cmd(cmdline) try: gdb.execute(cmdline, to_string=False) except gdb.error as e: err["e"] = str(e) settled.set() gdb.post_event(arm_and_go) ok = settled.wait(RUNCONTROL_TIMEOUT) # tear down listeners on the main thread def disarm(): try: gdb.events.stop.disconnect(on_stop) except Exception: # noqa: BLE001 pass try: gdb.events.exited.disconnect(on_exit) except Exception: # noqa: BLE001 pass try: on_main(disarm, timeout=10) except Exception: # noqa: BLE001 pass if "e" in err: msg = err["e"] # a program that exits/finishes during the command is a normal # outcome, not an error: settle to the (no-inferior) state + note. low = msg.lower() if "exited" in low or "not being run" in low: return {"cmd": cmdline, "state": on_main(ADAPTER.state), "note": msg} raise RuntimeError(msg) if not ok: raise TimeoutError( f"run-control {cmdline!r} did not settle in {RUNCONTROL_TIMEOUT}s") return {"cmd": cmdline, "state": on_main(ADAPTER.state)} def dispatch(method, params): """Return a result dict/value, or raise. Runs on the socket thread.""" if method == "ping": return ADAPTER.ping() if method == "methods": return ADAPTER.methods() if method == "quit": # scheduled by the server after the reply is flushed return {"ok": True, "bye": True} if method == "screen": return ADAPTER.screen(**params) if method in _READ: fn = getattr(ADAPTER, method) return on_main(lambda: fn(**params)) if method in _RUNCONTROL: return run_control_dispatch(method, params) if method in _MUT: fn = getattr(ADAPTER, _MUT[method]) return on_main(lambda: fn(**params)) raise ValueError(f"unknown method: {method}") # -------------------------------------------------------------------------- # server (background thread) # -------------------------------------------------------------------------- class Server(threading.Thread): daemon = True def __init__(self, path): super().__init__(name="gdb-driver-rpc") self.path = path self._sock = None self._busy = threading.Lock() # single-driver gate def run(self): try: if os.path.exists(self.path): os.unlink(self.path) s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) s.bind(self.path) os.chmod(self.path, stat.S_IRUSR | stat.S_IWUSR) # 0600 s.listen(4) self._sock = s ADAPTER.ready = True # NB: log to a file, not the pane -- writing here would land on # gdb's startup prompt line and corrupt the hand-typed look. try: with open(self.path + ".log", "a") as lf: lf.write(f"listening on {self.path}\n") except OSError: pass while True: conn, _ = s.accept() # handle each connection concurrently so the single-driver # gate can *refuse* a second client instead of queuing it threading.Thread(target=self._serve, args=(conn,), daemon=True).start() except Exception: # noqa: BLE001 sys.stderr.write("[gdb-driver] server crashed:\n") traceback.print_exc() def _serve(self, conn): try: self._loop(conn) finally: conn.close() def _loop(self, conn): f = conn.makefile("rwb") for raw in f: raw = raw.strip() if not raw: continue rid = None try: msg = json.loads(raw) rid = msg.get("id") method = msg["method"] params = msg.get("params") or {} except Exception as e: # noqa: BLE001 - malformed frame f.write(_enc({"id": rid, "error": {"message": str(e)}})) f.flush() continue # single-driver gate is PER-REQUEST: it guards against two clients # dispatching concurrently, but is not held during the idle # readline wait (which would race a fast sequential client's # teardown and spuriously refuse it). if not self._busy.acquire(blocking=False): f.write(_enc({"id": rid, "error": { "message": "busy: another request is in flight"}})) f.flush() continue try: resp = {"id": rid, "result": dispatch(method, params)} except Exception as e: # noqa: BLE001 - errors are data resp = {"id": rid, "error": {"message": str(e), "type": type(e).__name__}} finally: # release BEFORE writing: the write flush lets the client fire # its next request, which (on its own thread) must not see the # gate still held (that race caused spurious 'busy'). self._busy.release() f.write(_enc(resp)) f.flush() if method == "quit": gdb.post_event(lambda: gdb.execute("quit")) return def _enc(obj): return (json.dumps(obj, separators=(",", ":")) + "\n").encode() # -------------------------------------------------------------------------- # bootstrap # -------------------------------------------------------------------------- def _init(): # make gdb non-interactive-safe for programmatic driving for cmd in ("set pagination off", "set confirm off"): try: gdb.execute(cmd, to_string=True) except gdb.error: pass path = _sock_path() Server(path).start() _init()