gdb-driver: puppeteer live gdb (plain CLI + GEF) over an RPC socketHEADmain

Embedded gdb-Python plugin exposing a JSONL unix-socket RPC: three tiers of
control (raw tmux keys / semantic verbs / structured introspection), real
stop-event settling, and gef>-prompt command echoing so the tmux pane stays
human-legible. Includes the drive/pane client CLIs, an end-to-end smoke test,
and the TUI-driving design blueprint.

1 files changed, 564 insertions, 0 deletions

diff --git a/driver.py b/driver.py
new file mode 100644
index 0000000..0492393
--- /dev/null
+++ b/driver.py
@@ -0,0 +1,564 @@
+"""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()