path: root/driver.py

"""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()