path: root/debug/risc-v-gsp-rm-vm/greeneye/gdbridge.py

"""Ghidra control bridge: let GhidraScripts drive a risc-v-gsp-rm-debugger session.

A tiny JSON-lines server over localhost.  Each request is one line::

    {"id": 1, "cmd": "run_until", "args": {"target": "0x800000ce", "timeout": 5}}

and each response is one line::

    {"id": 1, "ok": true, "result": {"addr": 2147483854}}

This is the transport behind ``ghidra/GreenEyeControl.java``: a Ghidra user
can "execute until here", set/clear breakpoints, read memory and export hit
lists without leaving the decompiler.  See ``docs/13-GHIDRA-CONTROL.md``.
"""
from __future__ import annotations

import json
import socket
import threading

from .errors import GreenEyeError

#: The bridge is unauthenticated and write-capable; it must never be exposed.
LOOPBACK_HOSTS = ("127.0.0.1", "localhost", "::1")


class GhidraBridge:
    def __init__(self, session, host: str = "127.0.0.1", port: int = 0):
        self.session = session
        self.host = host
        self.port = port
        self._server: socket.socket | None = None
        self._thread: threading.Thread | None = None
        self._clients: list[threading.Thread] = []
        self._stop = threading.Event()

    # -- lifecycle ---------------------------------------------------------

    def start(self) -> int:
        if self.host not in LOOPBACK_HOSTS:
            raise ValueError(f"refusing non-loopback bind {self.host!r}; the Ghidra bridge is local-only")
        server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        server.bind((self.host, self.port))
        server.listen(4)
        self.port = server.getsockname()[1]
        self._server = server
        self._thread = threading.Thread(target=self._accept_loop, name="greeneye-ghidra", daemon=True)
        self._thread.start()
        return self.port

    def stop(self) -> None:
        self._stop.set()
        server = self._server
        self._server = None
        if server is not None:
            # shutdown() wakes a thread blocked in accept(); close() alone does
            # not, which left the listener up after a <close> command.
            try:
                server.shutdown(socket.SHUT_RDWR)
            except OSError:
                pass
            try:
                server.close()
            except OSError:
                pass
        if self._thread is not None and self._thread.is_alive() and threading.current_thread() is not self._thread:
            self._thread.join(timeout=2.0)

    def serve_forever(self) -> None:
        if self._thread is None:
            self.start()
        try:
            while not self._stop.is_set() and (self._thread is not None and self._thread.is_alive()):
                self._thread.join(timeout=0.5)
        except KeyboardInterrupt:
            pass
        finally:
            self.stop()

    def _accept_loop(self) -> None:
        assert self._server is not None
        while not self._stop.is_set():
            try:
                client, _addr = self._server.accept()
            except OSError:
                return
            handler = threading.Thread(target=self._client_loop, args=(client,), daemon=True)
            handler.start()
            self._clients.append(handler)

    def _client_loop(self, client: socket.socket) -> None:
        with client:
            stream = client.makefile("rwb")
            for line in stream:
                line = line.strip()
                if not line:
                    continue
                try:
                    request = json.loads(line)
                except json.JSONDecodeError as exc:
                    response = {"id": None, "ok": False, "error": f"bad json: {exc}"}
                else:
                    response = self._dispatch(request)
                stream.write((json.dumps(response) + "\n").encode("utf-8"))
                stream.flush()
                if response.get("close"):
                    self.stop()
                    return

    # -- command dispatch --------------------------------------------------

    def _dispatch(self, request: dict) -> dict:
        request_id = request.get("id")
        cmd = request.get("cmd", "")
        args = request.get("args") or {}
        try:
            handler = getattr(self, f"cmd_{cmd}", None)
            if handler is None:
                raise GreenEyeError(f"unknown bridge command {cmd!r}")
            result = handler(**args)
            response = {"id": request_id, "ok": True, "result": result}
            if cmd == "close":
                response["close"] = True
            return response
        except Exception as exc:  # never kill the bridge on a bad request
            return {"id": request_id, "ok": False, "error": str(exc)}

    # -- commands ----------------------------------------------------------

    def cmd_ping(self) -> str:
        return "pong"

    def cmd_info(self) -> dict:
        return {
            "name": self.session.name,
            "pc": self.session.pc,
            "symbols": len(self.session.symbols),
            "recording": getattr(self.session.recorder, "dir", None),
        }

    def cmd_resolve(self, name: str) -> int:
        return self.session.resolve(name)

    def cmd_bp_set(self, target, kind: str = "sw") -> int:
        handle = self.session.hook(target, kind=kind, listener="ghidra")
        return handle.addr

    def cmd_bp_clear(self, addr) -> bool:
        addr = self.session.resolve(addr) if isinstance(addr, str) else int(addr)
        for _key, bp in list(self.session._bps.items()):  # noqa: SLF001 - bridge is a privileged client
            if bp.addr == addr:
                self.session.remove_breakpoint(bp)
        return True

    def cmd_bp_list(self) -> list[dict]:
        return [
            {"addr": bp.addr, "label": bp.label, "hits": bp.hits, "kind": bp.kind}
            for bp in self.session._bps.values()  # noqa: SLF001
        ]

    def cmd_watch_set(self, target, size: int = 8, kind: str = "w") -> int:
        wp = self.session.watch(target, size=size, kind=kind, listener="ghidra")
        return wp.addr

    def cmd_run_until(self, target, timeout: float = 5.0) -> dict:
        hit = self.session.run_until(target, timeout=timeout)
        return {"addr": hit.addr, "symbol": hit.symbol.name if hit.symbol else None, "kind": hit.kind}

    def cmd_run(self, duration: float = 1.0, max_stops: int | None = None) -> int:
        return self.session.run(duration=duration, max_stops=max_stops)

    def cmd_step(self, count: int = 1) -> list[int]:
        return [hit.addr for hit in self.session.step(count)]

    def cmd_read_mem(self, addr, size: int = 64) -> str:
        real = self.session.resolve(addr) if isinstance(addr, str) else int(addr)
        return self.session.memory.read(real, size).hex()

    def cmd_write_mem(self, addr, hex: str) -> bool:
        real = self.session.resolve(addr) if isinstance(addr, str) else int(addr)
        self.session.memory.write(real, bytes.fromhex(hex))
        return True

    def cmd_read_reg(self, name: str) -> int:
        return self.session.client.read_reg(name)

    def cmd_write_reg(self, name: str, value) -> bool:
        self.session.client.write_reg(name, int(value))
        return True

    def cmd_call(self, target, args: list | None = None, timeout: float = 5.0) -> int:
        return self.session.call(target, tuple(args or ()), timeout=timeout)

    def cmd_snapshot(self, addr, size: int, label: str = "ghidra") -> str:
        real = self.session.resolve(addr) if isinstance(addr, str) else int(addr)
        self.session.memory.snapshot(real, size, name=label)
        return label

    def cmd_note(self, text: str) -> bool:
        self.session.note(text)
        return True

    def cmd_hits(self) -> list[dict]:
        from .record import load_events

        directory = getattr(self.session.recorder, "dir", None)
        if not directory:
            return []
        counts: dict[int, int] = {}
        for event in load_events(directory):
            if event.get("type") == "bp_hit":
                addr = event.get("pc")
                if isinstance(addr, int):
                    counts[addr] = counts.get(addr, 0) + 1
        return [
            {
                "addr": addr,
                "hits": hits,
                "symbol": (self.session.symbols.resolve(addr).name if self.session.symbols.resolve(addr) else None),
            }
            for addr, hits in sorted(counts.items(), key=lambda item: -item[1])
        ]

    def cmd_export_hits(self, path: str) -> str:
        from .ghidra import export_hits

        directory = getattr(self.session.recorder, "dir", None)
        if not directory:
            raise GreenEyeError("session is not recording; hits cannot be exported")
        return export_hits(directory, path, symbols=self.session.symbols)

    def cmd_close(self) -> bool:
        return True