diff options
Diffstat (limited to 'driver.py')
| -rw-r--r-- | driver.py | 564 |
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() |