diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/plugins')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/plugins/README.md | 6 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/plugins/mcp_server.py | 1155 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/plugins/rmharness_plugin.py | 120 |
3 files changed, 1278 insertions, 3 deletions
diff --git a/debug/risc-v-gsp-rm-vm/plugins/README.md b/debug/risc-v-gsp-rm-vm/plugins/README.md index aa9bc72..7fe69da 100644 --- a/debug/risc-v-gsp-rm-vm/plugins/README.md +++ b/debug/risc-v-gsp-rm-vm/plugins/README.md @@ -27,8 +27,8 @@ Types of plugins that fit green-eye: recorded events. 3. **Virtualization-environment updates** — replay/record wrappers, QEMU argument tweaks, device-model shims. -4. **AI/harness integration** — export sessions to analysis services; - **MCP servers are planned as a plugin** (see `docs/08-ROADMAP.md`), not - part of v0.1. +4. **AI/harness integration** — export sessions to analysis services; the + **MCP server is implemented as a bundled plugin** (`mcp_server.py`, see + `docs/20-MCP-SERVER.md`), stdlib-only and read-only by default. The example plugin is `example_plugin.py`. diff --git a/debug/risc-v-gsp-rm-vm/plugins/mcp_server.py b/debug/risc-v-gsp-rm-vm/plugins/mcp_server.py new file mode 100644 index 0000000..b5e8f90 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/plugins/mcp_server.py @@ -0,0 +1,1155 @@ +"""green-eye MCP server plugin (roadmap R3). + +A stdlib-only Model Context Protocol server (JSON-RPC 2.0) that exposes the +*active* green-eye session as MCP tools. Two ways to run it: + +* **plugin mode** - load it into a live session and serve that session:: + + ./greeneye.sh attach --plugin plugins/mcp_server.py + greeneye> mcp-serve [--allow-write] [--allow-inject] + +* **standalone mode** - launch its own session and serve it:: + + python3 plugins/mcp_server.py --stdio [--kernel <elf>] + [--backend qemu|native] [--record] [--plugins PATH] + +Transport is newline-delimited JSON-RPC over stdin/stdout by default; an +optional TCP mode binds to loopback only (``--tcp 127.0.0.1:PORT``). stdout +carries protocol frames only - all logs go to stderr. + +Safety: read/observe tools are always available; memory writes and remote +calls require ``--allow-write`` and service injection requires +``--allow-inject``. Capabilities owned by other components (the QEMU-only +harnesses, the native trace) are probed defensively and return a structured +``not_available`` tool error when absent. See ``docs/20-MCP-SERVER.md``. +""" +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import socket +import subprocess +import sys +from typing import Iterable + +_TOOL_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if _TOOL_ROOT not in sys.path: # `python3 plugins/mcp_server.py` must import greeneye + sys.path.insert(0, _TOOL_ROOT) + +from greeneye.errors import GreenEyeError # noqa: E402 + +SERVER_NAME = "greeneye-mcp" +JSONRPC_VERSION = "2.0" +SUPPORTED_PROTOCOLS = ("2024-11-05", "2025-03-26", "2025-06-18") +DEFAULT_PROTOCOL = "2024-11-05" + +PARSE_ERROR = -32700 +INVALID_REQUEST = -32600 +METHOD_NOT_FOUND = -32601 +INVALID_PARAMS = -32602 + +_DUMPS = os.path.join(_TOOL_ROOT, "dumps") +_PLUGINS_DIR = os.path.join(_TOOL_ROOT, "plugins") +_DEMO_ELF = os.path.join(_TOOL_ROOT, "firmware", "uproc_demo", "build", "uproc_demo.elf") +_FWIMAGE = os.path.join( + os.path.dirname(_TOOL_ROOT), "firmware", "57X-GSP", "570.144", "gsp-570.144.fwimage.bin" +) +MUTATION_SCRIPT = os.path.join(_TOOL_ROOT, "harness", "mutation_harness.py") +MATRIX_SCRIPT = os.path.join(_TOOL_ROOT, "harness", "portability_matrix.py") +LOOPBACK_HOSTS = ("127.0.0.1", "localhost", "::1") + +#: Per-response size caps so a single tool call can never flood a client. +MAX_READ_BYTES = 1 << 20 +MAX_HEXDUMP_BYTES = 1 << 20 +MAX_SNAPSHOT_BYTES = 1 << 24 + + +class McpError(Exception): + """A JSON-RPC level error (unknown method/tool, bad params).""" + + def __init__(self, code: int, message: str, data=None): + super().__init__(message) + self.code = code + self.message = message + self.data = data + + +class ToolError(Exception): + """A tool-execution error returned inside a tools/call result (isError).""" + + def __init__(self, message: str, *, code: str = "error", **details): + super().__init__(message) + self.payload = {"error": code, "message": message, **details} + + +def _unavailable(message: str, **details) -> ToolError: + return ToolError(message, code="not_available", **details) + + +# --------------------------------------------------------------------------- +# tool catalogue +# --------------------------------------------------------------------------- + +def _address(description: str) -> dict: + return {"type": ["string", "integer"], "description": description} + + +def _integer(description: str, default=None, minimum=None, maximum=None) -> dict: + schema: dict = {"type": "integer", "description": description} + if default is not None: + schema["default"] = default + if minimum is not None: + schema["minimum"] = minimum + if maximum is not None: + schema["maximum"] = maximum + return schema + + +def _number(description: str, default=None) -> dict: + schema: dict = {"type": "number", "description": description} + if default is not None: + schema["default"] = default + return schema + + +def _string(description: str, default=None) -> dict: + schema: dict = {"type": "string", "description": description} + if default is not None: + schema["default"] = default + return schema + + +def _boolean(description: str, default=None) -> dict: + schema: dict = {"type": "boolean", "description": description} + if default is not None: + schema["default"] = default + return schema + + +def _tool(name: str, description: str, properties: dict | None = None, required: Iterable[str] = ()) -> dict: + schema: dict = {"type": "object", "properties": properties or {}} + if required: + schema["required"] = list(required) + return {"name": name, "description": description, "inputSchema": schema} + + +TOOLS: list[dict] = [ + _tool("session_info", "Active session state: backend, PC, symbols, recording, breakpoints/watchpoints, enabled capabilities."), + _tool( + "symbols_resolve", + "Resolve a symbol name, 'symbol+0xNN' expression or hex address against the session symbol DB.", + {"target": _address("symbol name, symbol+offset or address")}, + ("target",), + ), + _tool( + "bp_set", + "Set a software or hardware breakpoint at a symbol/address; hits are recorded in the session.", + {"target": _address("symbol or address"), "kind": {"type": "string", "enum": ["sw", "hw"], "default": "sw"}}, + ("target",), + ), + _tool("bp_clear", "Clear every breakpoint at a symbol/address.", {"target": _address("symbol or address")}, ("target",)), + _tool( + "watch_set", + "Set a read/write/access watchpoint; hits are recorded in the session.", + { + "target": _address("symbol or address"), + "size": _integer("watched region size in bytes", 8, 1, 4096), + "kind": {"type": "string", "enum": ["w", "r", "a"], "default": "w"}, + }, + ("target",), + ), + _tool( + "run", + "Resume and service hooks/watchpoints for a wall-clock budget (bounded; default 1.0s).", + { + "duration": _number("wall-clock budget in seconds", 1.0), + "max_stops": _integer("stop budget (optional)", None, 1), + "timeout": _number("alias for duration when duration is null"), + }, + ), + _tool( + "run_until", + "Run until execution reaches a target address/symbol, servicing intervening stops.", + {"target": _address("symbol or address"), "timeout": _number("seconds to wait", 5.0)}, + ("target",), + ), + _tool("step", "Single-step the target n times, dispatching hooks and watchpoints.", {"count": _integer("instructions", 1, 1, 10000)}), + _tool( + "read_mem", + "Read memory and return lowercase hex.", + {"addr": _address("address or symbol"), "size": _integer("bytes to read", 64, 1, MAX_READ_BYTES)}, + ("addr",), + ), + _tool( + "write_mem", + "Write hex bytes to target memory. Requires --allow-write at server startup.", + {"addr": _address("address or symbol"), "hex": _string("hex bytes, e.g. '01 02 ff'")}, + ("addr", "hex"), + ), + _tool( + "hexdump", + "Symbol-annotated hexdump of target memory.", + {"addr": _address("address or symbol"), "size": _integer("bytes", 64, 1, MAX_HEXDUMP_BYTES)}, + ("addr",), + ), + _tool( + "snapshot", + "Capture a memory region into the recording (zlib + sha256) and return its digest.", + {"addr": _address("address or symbol"), "size": _integer("bytes", None, 1, MAX_SNAPSHOT_BYTES), "label": _string("snapshot label", "mcp")}, + ("addr", "size"), + ), + _tool( + "call_function", + "Remote-call a target function (Frida-like). Requires --allow-write at server startup.", + {"target": _address("symbol or address"), "args": {"type": "array", "items": {"type": "integer"}, "description": "a0..a7"}, "timeout": _number("seconds", 5.0)}, + ("target",), + ), + _tool( + "inject_service", + "Load and invoke an injected RV64 service blob. Requires --allow-inject at server startup.", + { + "service": _string("path to the service ELF"), + "slot": _integer("demo handler slot", 6, 0, 7), + "load_addr": _address("load address (optional)"), + "cmd": _integer("service command", 1, 0), + "arg0": _integer("argument 0", 0, 0), + "arg1": _integer("argument 1", 0, 0), + "path": {"type": "string", "enum": ["remote", "firmware", "both"], "default": "remote"}, + "settle": _number("seconds to settle before the firmware mailbox call", 0.25), + }, + ("service",), + ), + _tool( + "trace_start", + "Start the native exec/access trace (native backend, recording session only).", + { + "exec": _boolean("record exec events", True), + "start": _address("only PCs >= start"), + "end": _address("only PCs <= end"), + "max_events": _integer("ring budget (0 = unlimited)"), + }, + ), + _tool("trace_stop", "Stop the native trace, flush it to blobs/trace.jsonl and return event counts."), + _tool("analyze_current", "Render the symbol-annotated summary of the active recording.", {"max_rows": _integer("table row cap", 40, 1, 1000)}), + _tool( + "compare_sessions", + "A/B two recorded session directories at the symbolic event level.", + {"a": _string("session A directory"), "b": _string("session B directory"), "symbols": _boolean("resolve with the active symbol DB", True), "max_rows": _integer("table row cap", 40, 1, 1000)}, + ("a", "b"), + ), + _tool( + "export_hits", + "Export breakpoint/watchpoint hits as Ghidra's greeneye_hits.tsv.", + {"session": _string("session directory (default: active recording)"), "out": _string("output TSV path")}, + ), + _tool( + "probe_firmware", + "Run the QEMU probe CLI on a firmware image and report the first divergence (QEMU required).", + { + "elf": _string("firmware ELF (default: the project fwimage)"), + "instructions": _integer("step budget (0 = wall-clock duration)", 2000, 0), + "duration": _number("wall-clock seconds when instructions=0", 1.0), + "no_relocate": _boolean("do not relocate to 0x80000000", False), + "symbols": _string("green-eye symbol DB JSON"), + "timeout": _number("subprocess timeout seconds", 300.0), + }, + ), + _tool( + "mutation_run", + "Run harness/mutation_harness.py (QEMU-only) and return the outcome-class counts.", + { + "kernel": _string("demo/re-hosted ELF (default: active kernel or bundled demo)"), + "out": _string("output directory (default: dumps/mcp-mutation)"), + "iterations": _integer("mutation count", 10, 1, 10000), + "seed": _integer("deterministic seed", 1), + "timeout": _number("observation budget per mutation", 0.25), + "settle": _number("settle budget per attempt", 0.2), + "memory_region": _string("optional RAM region ADDR:SIZE"), + "allow_foreign_elf": _boolean("permit a non-demo ELF", False), + }, + ), + _tool( + "portability_matrix_run", + "Run harness/portability_matrix.py (QEMU-only) over one or more firmware ELFs.", + { + "targets": {"type": "array", "items": {"type": "string"}, "description": "firmware ELF paths"}, + "out": _string("output directory (default: dumps/mcp-matrix)"), + "probe_instructions": _integer("probe step budget", 2000, 0), + "scenario_duration": _number("per-phase wall budget", 0.4), + "symbols": _string("green-eye symbol DB JSON"), + "max_rows": _integer("report row cap", 40, 1, 1000), + }, + ("targets",), + ), + _tool("plugins_list", "List bundled plugins, GREENEYE_PLUGINS and plugin-provided REPL commands."), + _tool("close", "Stop serving MCP; in standalone mode also close the launched session."), +] + +_TOOL_NAMES = {tool["name"] for tool in TOOLS} + + +# --------------------------------------------------------------------------- +# server +# --------------------------------------------------------------------------- + + +class GreenEyeMCPServer: + """JSON-RPC 2.0 MCP server over newline-delimited streams.""" + + def __init__( + self, + session, + *, + allow_write: bool = False, + allow_inject: bool = False, + api=None, + owns_session: bool = False, + input_stream=None, + output_stream=None, + log_stream=None, + ): + self.session = session + self.allow_write = bool(allow_write) + self.allow_inject = bool(allow_inject) + self.api = api # GreenEyeAPI when loaded as a plugin (plugins_list) + self.owns_session = bool(owns_session) + self._input = input_stream if input_stream is not None else sys.stdin + self._output = output_stream if output_stream is not None else sys.stdout + self._log_stream = log_stream if log_stream is not None else sys.stderr + self._stop = False + self.session_closed = False + self._initialized = False + + # -- transport --------------------------------------------------------- + + def serve_stdio(self) -> None: + self._log(f"serving {self.session.name!r} over stdio (write={self.allow_write}, inject={self.allow_inject})") + self.serve_stream(self._input, self._output) + self._log("stdio server stopped") + + def serve_stream(self, stream_in, stream_out) -> None: + for raw in stream_in: + if self._stop: + break + if isinstance(raw, bytes): + raw = raw.decode("utf-8", "replace") + self.process_line(raw, stream_out) + + def serve_tcp(self, host: str, port: int) -> None: + if host not in LOOPBACK_HOSTS: + raise ValueError(f"refusing non-loopback bind {host!r}; MCP is local-only") + server = socket.socket(socket.AF_INET if ":" not in host else socket.AF_INET6, socket.SOCK_STREAM) + server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + server.bind((host, port)) + server.listen(2) + bound = server.getsockname() + self._log(f"serving {self.session.name!r} on {bound[0]}:{bound[1]} (loopback only)") + try: + while not self._stop: + try: + conn, _addr = server.accept() + except OSError: + break + with conn: + reader = conn.makefile("r", encoding="utf-8", newline="\n") + writer = conn.makefile("w", encoding="utf-8", newline="\n") + try: + self.serve_stream(reader, writer) + finally: + reader.close() + writer.close() + finally: + server.close() + self._log("tcp server stopped") + + def process_line(self, line: str, stream_out=None) -> None: + stream = stream_out if stream_out is not None else self._output + line = line.strip() + if not line: + return + try: + message = json.loads(line) + except json.JSONDecodeError as exc: + self._write(stream, self._error(None, PARSE_ERROR, f"parse error: {exc}")) + return + try: + response = self.handle_message(message) + except McpError as exc: # defensive: protocol bugs must not kill the loop + self._log(f"protocol error: {exc}") + response = self._error(message.get("id") if isinstance(message, dict) else None, exc.code, exc.message, exc.data) + if response is not None: + self._write(stream, response) + + def _write(self, stream, payload: dict) -> None: + stream.write(json.dumps(payload, separators=(",", ":"), default=str) + "\n") + stream.flush() + + # -- JSON-RPC ---------------------------------------------------------- + + @staticmethod + def _result(request_id, result: dict) -> dict: + return {"jsonrpc": JSONRPC_VERSION, "id": request_id, "result": result} + + @staticmethod + def _error(request_id, code: int, message: str, data=None) -> dict: + payload = {"jsonrpc": JSONRPC_VERSION, "id": request_id, "error": {"code": code, "message": message}} + if data is not None: + payload["error"]["data"] = data + return payload + + def handle_message(self, message) -> dict | None: + if not isinstance(message, dict): + return self._error(None, INVALID_REQUEST, "message must be a JSON object") + method = message.get("method") + request_id = message.get("id") + if not isinstance(method, str): + return self._error(request_id, INVALID_REQUEST, "missing method") + params = message.get("params") + if params is None: + params = {} + if not isinstance(params, dict): + return self._error(request_id, INVALID_PARAMS, "params must be an object") + + if method.startswith("notifications/") or "id" not in message: + return None # notifications never get a response + + if method == "initialize": + return self._result(request_id, self._initialize(params)) + if method == "ping": + return self._result(request_id, {}) + if method == "tools/list": + return self._result(request_id, {"tools": TOOLS}) + if method == "tools/call": + try: + result = self._dispatch_tool(params) + except McpError as exc: + return self._error(request_id, exc.code, exc.message, exc.data) + return self._result(request_id, result) + return self._error(request_id, METHOD_NOT_FOUND, f"unknown method {method!r}", {"method": method}) + + def _initialize(self, params: dict) -> dict: + requested = params.get("protocolVersion") + version = requested if requested in SUPPORTED_PROTOCOLS else DEFAULT_PROTOCOL + self._initialized = True + return { + "protocolVersion": version, + "capabilities": {"tools": {"listChanged": False}}, + "serverInfo": {"name": SERVER_NAME, "version": _version()}, + "instructions": ( + "green-eye session server. Read/observe tools are always on; writes need " + "--allow-write and injection needs --allow-inject. Never claim timing, cycles " + "or hardware behaviour; sessions are the evidence." + ), + } + + # -- tool dispatch ----------------------------------------------------- + + def _dispatch_tool(self, params: dict) -> dict: + name = params.get("name") + if not isinstance(name, str): + raise McpError(INVALID_PARAMS, "tools/call requires a tool name") + handler = getattr(self, f"tool_{name}", None) + if handler is None or name not in _TOOL_NAMES: + raise McpError(INVALID_PARAMS, f"unknown tool {name!r}", {"tool": name}) + arguments = params.get("arguments") + if arguments is None: + arguments = {} + if not isinstance(arguments, dict): + raise McpError(INVALID_PARAMS, "tool arguments must be an object") + try: + result = handler(arguments) + except ToolError as exc: + return _tool_error(exc.payload) + except GreenEyeError as exc: + return _tool_error({"error": "greeneye", "message": str(exc)}) + except Exception as exc: # one bad call must never take the server down + self._log(f"tool {name} failed: {exc!r}") + return _tool_error({"error": "internal", "message": f"{type(exc).__name__}: {exc}"}) + return {"content": [{"type": "text", "text": json.dumps(result, sort_keys=True, default=str)}]} + + def _require_capability(self, name: str) -> None: + if name == "write" and not self.allow_write: + raise ToolError( + "target writes and remote calls are disabled; restart the server with --allow-write", + code="write_disabled", + ) + if name == "inject" and not self.allow_inject: + raise ToolError( + "service injection is disabled; restart the server with --allow-inject", + code="inject_disabled", + ) + + def _annotate(self, addr: int) -> str: + return self.session.symbols.annotate(addr) + + def _session_kernel(self) -> str | None: + target = self.session.target + kernel = getattr(target, "kernel", None) + if kernel: + return str(kernel) + return getattr(self.session.image, "path", None) + + # -- tools: session ---------------------------------------------------- + + def tool_session_info(self, args: dict) -> dict: + session = self.session + backend = "native" if hasattr(session.client, "trace") else ("qemu" if session.target is not None else "custom") + return { + "name": session.name, + "backend": backend, + "pc": session.pc, + "pc_symbol": self._annotate(session.pc), + "symbols": len(session.symbols), + "recording": getattr(session.recorder, "dir", None), + "kernel": self._session_kernel(), + "closed": self.session_closed, + "capabilities": { + "write": self.allow_write, + "inject": self.allow_inject, + "trace": backend == "native", + }, + "breakpoints": [ + {"addr": bp.addr, "label": bp.label, "kind": bp.kind, "hits": bp.hits} + for bp in session._bps.values() # noqa: SLF001 - the MCP server is a privileged client + ], + "watches": [ + {"addr": wp.addr, "size": wp.size, "kind": wp.kind, "hits": wp.hits} + for wp in session._watches.values() # noqa: SLF001 + ], + "tools": len(TOOLS), + } + + def tool_symbols_resolve(self, args: dict) -> dict: + addr = self.session.resolve(_require(args, "target")) + sym = self.session.symbols.resolve(addr) + return {"address": addr, "symbol": sym.name if sym else None, "annotation": self._annotate(addr)} + + # -- tools: instrumentation ------------------------------------------- + + def tool_bp_set(self, args: dict) -> dict: + target = _require(args, "target") + kind = args.get("kind", "sw") + if kind not in ("sw", "hw"): + raise ToolError("kind must be 'sw' or 'hw'", code="invalid_arguments") + handle = self.session.hook(target, kind=kind, listener="mcp") + return {"addr": handle.addr, "label": handle.label, "kind": kind} + + def tool_bp_clear(self, args: dict) -> dict: + addr = self.session.resolve(_require(args, "target")) + removed = 0 + for bp in list(self.session._bps.values()): # noqa: SLF001 + if bp.addr == addr: + self.session.remove_breakpoint(bp) + removed += 1 + return {"addr": addr, "cleared": removed} + + def tool_watch_set(self, args: dict) -> dict: + kind = args.get("kind", "w") + if kind not in ("w", "r", "a"): + raise ToolError("kind must be 'w', 'r' or 'a'", code="invalid_arguments") + wp = self.session.watch( + _require(args, "target"), + size=_as_int(args.get("size", 8), "size"), + kind=kind, + listener="mcp", + ) + return {"addr": wp.addr, "size": wp.size, "kind": wp.kind, "label": wp.label} + + # -- tools: execution -------------------------------------------------- + + def tool_run(self, args: dict) -> dict: + duration = args.get("duration", 1.0) + timeout = args.get("timeout") + if duration is None: + duration = timeout if timeout is not None else 1.0 + max_stops = args.get("max_stops") + stops = self.session.run( + duration=float(duration), + max_stops=_as_int(max_stops, "max_stops") if max_stops is not None else None, + timeout=float(timeout) if timeout is not None else None, + ) + return {"stops": stops, "pc": self.session.pc, "pc_symbol": self._annotate(self.session.pc)} + + def tool_run_until(self, args: dict) -> dict: + hit = self.session.run_until(_require(args, "target"), timeout=float(args.get("timeout", 5.0))) + return { + "addr": hit.addr, + "kind": hit.kind, + "symbol": hit.symbol.name if hit.symbol else None, + "pc": hit.pc, + "pc_symbol": self._annotate(hit.pc), + } + + def tool_step(self, args: dict) -> dict: + count = _as_int(args.get("count", 1), "count") + if not 1 <= count <= 10000: + raise ToolError("count must be in 1..10000", code="invalid_arguments") + hits = self.session.step(count) + return { + "steps": len(hits), + "pc": self.session.pc, + "pc_symbol": self._annotate(self.session.pc), + "hits": [ + {"kind": hit.kind, "addr": hit.addr, "symbol": hit.symbol.name if hit.symbol else None} + for hit in hits + ], + } + + # -- tools: memory ----------------------------------------------------- + + def tool_read_mem(self, args: dict) -> dict: + addr = self.session.resolve(_require(args, "addr")) + size = _bounded(_as_int(args.get("size", 64), "size"), 1, MAX_READ_BYTES, "size") + data = self.session.memory.read(addr, size) + return {"addr": addr, "size": len(data), "hex": data.hex(), "annotation": self._annotate(addr)} + + def tool_write_mem(self, args: dict) -> dict: + self._require_capability("write") + addr = self.session.resolve(_require(args, "addr")) + data = _hex_bytes(_require(args, "hex")) + self.session.memory.write(addr, data) + return {"addr": addr, "written": len(data), "hex": data.hex(), "annotation": self._annotate(addr)} + + def tool_hexdump(self, args: dict) -> dict: + addr = self.session.resolve(_require(args, "addr")) + size = _bounded(_as_int(args.get("size", 64), "size"), 1, MAX_HEXDUMP_BYTES, "size") + return {"addr": addr, "size": size, "text": self.session.memory.hexdump(addr, size)} + + def tool_snapshot(self, args: dict) -> dict: + addr = self.session.resolve(_require(args, "addr")) + size = _bounded(_as_int(_require(args, "size"), "size"), 1, MAX_SNAPSHOT_BYTES, "size") + label = str(args.get("label", "mcp")) + data = self.session.memory.snapshot(addr, size, name=label) + return { + "label": label, + "addr": addr, + "size": len(data), + "sha256": hashlib.sha256(data).hexdigest(), + "recording": getattr(self.session.recorder, "dir", None), + } + + def tool_call_function(self, args: dict) -> dict: + self._require_capability("write") + target = _require(args, "target") + call_args = args.get("args") or [] + if not isinstance(call_args, list): + raise ToolError("args must be an array of integers", code="invalid_arguments") + if len(call_args) > 8: + raise ToolError("at most 8 arguments (a0..a7)", code="invalid_arguments") + values = [_as_int(value, "args[]") for value in call_args] + retval = self.session.call(target, values, timeout=float(args.get("timeout", 5.0))) + return {"retval": retval, "retval_hex": hex(retval), "target": str(target)} + + def tool_inject_service(self, args: dict) -> dict: + self._require_capability("inject") + service_path = str(_require(args, "service")) + if not os.path.isfile(service_path): + raise ToolError( + f"service not found: {service_path} (build it with `make sdk`)", code="invalid_arguments" + ) + try: + from greeneye.inject import invoke_remote, invoke_via_firmware, load_service + except Exception as exc: + raise _unavailable(f"injection support unavailable: {exc}") from exc + load_addr = args.get("load_addr") + service = load_service( + self.session, + service_path, + load_addr=_as_int(load_addr, "load_addr") if load_addr is not None else None, + slot=_as_int(args.get("slot", 6), "slot"), + ) + cmd = _as_int(args.get("cmd", 1), "cmd") + a0 = _as_int(args.get("arg0", 0), "arg0") + a1 = _as_int(args.get("arg1", 0), "arg1") + mode = args.get("path", "remote") + if mode not in ("remote", "firmware", "both"): + raise ToolError("path must be 'remote', 'firmware' or 'both'", code="invalid_arguments") + result = { + "service": { + "name": service.name, + "load_addr": service.load_addr, + "entry_addr": service.entry_addr, + "slot": service.slot, + "size": service.size, + }, + "cmd": cmd, + "args": [a0, a1], + } + if mode in ("remote", "both"): + result["remote"] = invoke_remote(self.session, service, cmd, a0, a1) + if mode in ("firmware", "both"): + self.session.run(duration=float(args.get("settle", 0.25))) + result["firmware"] = invoke_via_firmware(self.session, service, cmd, a0, a1) + return result + + # -- tools: native trace ---------------------------------------------- + + def tool_trace_start(self, args: dict) -> dict: + if not hasattr(self.session.client, "trace"): + raise _unavailable("trace start/stop is native-backend only (start the server with --backend native)") + if getattr(self.session.recorder, "dir", None) is None: + raise _unavailable("trace_start requires a recording session (start the server/session with --record)") + if getattr(self.session.client, "trace", None) is not None: + raise ToolError("a trace is already active; call trace_stop first", code="invalid_arguments") + try: + from sim.trace import make_trace + except Exception as exc: + raise _unavailable(f"native trace module unavailable: {exc}") from exc + spec: dict = {"exec": bool(args.get("exec", True))} + for key in ("start", "end", "max_events"): + if args.get(key) is not None: + spec[key] = _as_int(args[key], key) + tracer = make_trace(spec) + self.session.client.trace = tracer + self.session.client._recorder = self.session.recorder # noqa: SLF001 - SimClient boundary + self.session.recorder.meta["trace"] = tracer.config() + return {"tracing": True, "config": tracer.config()} + + def tool_trace_stop(self, args: dict) -> dict: + tracer = getattr(self.session.client, "trace", None) + if tracer is None: + raise ToolError("no active trace", code="invalid_arguments") + self.session.client.trace = None + blob = tracer.flush(self.session.recorder) + return {"tracing": False, "blob": blob, "counts": tracer.counts()} + + # -- tools: analysis --------------------------------------------------- + + def tool_analyze_current(self, args: dict) -> dict: + directory = getattr(self.session.recorder, "dir", None) + if not directory: + raise _unavailable("the session is not recording; start it with --record") + try: + from greeneye.analyze import load_events, load_meta, load_trace, render_summary_from + except Exception as exc: + raise _unavailable(f"analysis support unavailable: {exc}") from exc + # An active recording has no session.json until close(), so merge the + # live recorder metadata instead of requiring the closed-session files. + if os.path.isfile(os.path.join(directory, "session.json")): + meta = load_meta(directory) + else: + meta = dict(getattr(self.session.recorder, "meta", {})) + meta.setdefault("format", "greeneye-session/1") + meta.setdefault("name", self.session.name) + meta.setdefault("dir", directory) + events = load_events(directory) + trace = load_trace(directory, meta) + report = render_summary_from( + meta, events, symbols=self.session.symbols, max_rows=_as_int(args.get("max_rows", 40), "max_rows"), trace=trace + ) + return {"session": directory, "report": report} + + def tool_compare_sessions(self, args: dict) -> dict: + a = str(_require(args, "a")) + b = str(_require(args, "b")) + for path in (a, b): + if not os.path.isdir(path): + raise ToolError(f"session directory not found: {path}", code="invalid_arguments") + try: + from greeneye.analyze import compare_sessions + except Exception as exc: + raise _unavailable(f"analysis support unavailable: {exc}") from exc + symbols = self.session.symbols if args.get("symbols", True) else None + report = compare_sessions(a, b, symbols=symbols, max_rows=_as_int(args.get("max_rows", 40), "max_rows")) + return {"a": a, "b": b, "report": report} + + def tool_export_hits(self, args: dict) -> dict: + directory = args.get("session") or getattr(self.session.recorder, "dir", None) + if not directory: + raise _unavailable("no session directory given and the active session is not recording") + if not os.path.isdir(str(directory)): + raise ToolError(f"session directory not found: {directory}", code="invalid_arguments") + out = args.get("out") or os.path.join(str(directory), "greeneye_hits.tsv") + try: + from greeneye.ghidra import export_hits + except Exception as exc: + raise _unavailable(f"Ghidra export unavailable: {exc}") from exc + path = export_hits(str(directory), str(out), symbols=self.session.symbols) + return {"session": str(directory), "path": path} + + # -- tools: external harnesses (defensive) ----------------------------- + + def tool_probe_firmware(self, args: dict) -> dict: + elf = args.get("elf") or (_FWIMAGE if os.path.isfile(_FWIMAGE) else None) + if not elf: + raise _unavailable(f"default fwimage not found and no elf argument given ({_FWIMAGE})") + if not os.path.isfile(str(elf)): + raise ToolError(f"firmware not found: {elf}", code="invalid_arguments") + cmd = [sys.executable, "-m", "greeneye", "probe", str(elf), "--record"] + if args.get("instructions") is not None: + cmd += ["--instructions", str(_as_int(args["instructions"], "instructions"))] + elif args.get("duration") is not None: + cmd += ["--duration", str(float(args["duration"]))] + if args.get("no_relocate"): + cmd.append("--no-relocate") + if args.get("symbols"): + cmd += ["--symbols", str(args["symbols"])] + rc, stdout, stderr = self._run_cli(cmd, timeout=float(args.get("timeout", 300.0))) + if rc != 0: + raise ToolError( + f"probe failed (rc={rc})", code="probe_failed", elf=str(elf), stdout=stdout[-4000:], stderr=stderr[-4000:] + ) + return {"elf": str(elf), "returncode": rc, "stdout": stdout[-20000:], "stderr": stderr[-4000:]} + + def tool_mutation_run(self, args: dict) -> dict: + if not os.path.isfile(MUTATION_SCRIPT): + raise _unavailable(f"mutation harness not present: {MUTATION_SCRIPT}") + kernel = args.get("kernel") or self._session_kernel() or _DEMO_ELF + if not os.path.isfile(str(kernel)): + raise ToolError(f"kernel not found: {kernel}", code="invalid_arguments") + out = str(args.get("out") or os.path.join(_DUMPS, "mcp-mutation")) + cmd = [ + sys.executable, + MUTATION_SCRIPT, + "--kernel", + str(kernel), + "--out", + out, + "--iterations", + str(_as_int(args.get("iterations", 10), "iterations")), + "--seed", + str(_as_int(args.get("seed", 1), "seed")), + "--timeout", + str(float(args.get("timeout", 0.25))), + "--settle", + str(float(args.get("settle", 0.2))), + ] + if args.get("memory_region"): + cmd += ["--memory-region", str(args["memory_region"])] + if args.get("allow_foreign_elf"): + cmd.append("--allow-foreign-elf") + rc, stdout, stderr = self._run_cli(cmd, timeout=float(args.get("subprocess_timeout", 1800.0))) + outcomes = _count_findings(os.path.join(out, "findings.jsonl")) + result = { + "kernel": str(kernel), + "out": out, + "returncode": rc, + "outcomes": outcomes, + "findings": sum(outcomes.values()), + "stdout": stdout[-4000:], + "stderr": stderr[-4000:], + } + if rc != 0: + raise ToolError("mutation harness failed", code="harness_failed", **result) + return result + + def tool_portability_matrix_run(self, args: dict) -> dict: + if not os.path.isfile(MATRIX_SCRIPT): + raise _unavailable(f"portability matrix not present: {MATRIX_SCRIPT}") + targets = args.get("targets") + if not isinstance(targets, list) or not targets: + raise ToolError("targets must be a non-empty array of ELF paths", code="invalid_arguments") + for target in targets: + if not os.path.isfile(str(target)): + raise ToolError(f"target not found: {target}", code="invalid_arguments") + out = str(args.get("out") or os.path.join(_DUMPS, "mcp-matrix")) + cmd = [sys.executable, MATRIX_SCRIPT, "--out", out] + for target in targets: + cmd += ["--target", str(target)] + cmd += [ + "--probe-instructions", + str(_as_int(args.get("probe_instructions", 2000), "probe_instructions")), + "--scenario-duration", + str(float(args.get("scenario_duration", 0.4))), + "--max-rows", + str(_as_int(args.get("max_rows", 40), "max_rows")), + ] + if args.get("symbols"): + cmd += ["--symbols", str(args["symbols"])] + rc, stdout, stderr = self._run_cli(cmd, timeout=float(args.get("subprocess_timeout", 3600.0))) + report_path = os.path.join(out, "portability-report.json") + summary = None + if os.path.isfile(report_path): + try: + with open(report_path, "r", encoding="utf-8") as fh: + report = json.load(fh) + summary = { + "targets": [ + {"name": t.get("name"), "status": t.get("status"), "sha256": (t.get("sha256") or "")[:16]} + for t in report.get("targets", []) + ], + "comparisons": len(report.get("comparisons", [])), + } + except (OSError, ValueError) as exc: + summary = {"parse_error": str(exc)} + result = { + "targets": [str(target) for target in targets], + "out": out, + "returncode": rc, + "report": report_path if os.path.isfile(report_path) else None, + "summary": summary, + "stdout": stdout[-4000:], + "stderr": stderr[-4000:], + } + if rc != 0: + raise ToolError("portability matrix failed", code="harness_failed", **result) + return result + + def tool_plugins_list(self, args: dict) -> dict: + bundled: list[str] = [] + if os.path.isdir(_PLUGINS_DIR): + bundled = sorted( + name + for name in os.listdir(_PLUGINS_DIR) + if name.endswith(".py") and not name.startswith("_") + ) + commands = sorted(self.api.commands) if self.api is not None else [] + return { + "bundled": bundled, + "GREENEYE_PLUGINS": os.environ.get("GREENEYE_PLUGINS"), + "commands": commands, + "mcp_plugin": "mcp_server.py" in bundled, + } + + def tool_close(self, args: dict) -> dict: + if self.owns_session and not self.session_closed: + self.session.close() + self.session_closed = True + self._stop = True + return {"closed": True, "session_closed": self.session_closed} + + # -- helpers ----------------------------------------------------------- + + def _run_cli(self, cmd: list[str], timeout: float) -> tuple[int, str, str]: + env = dict(os.environ) + env.setdefault("GREENEYE_DUMPS", _DUMPS) + try: + completed = subprocess.run( + cmd, + cwd=_TOOL_ROOT, + env=env, + capture_output=True, + text=True, + timeout=timeout, + ) + except FileNotFoundError as exc: + raise _unavailable(f"cannot run {cmd[0]}: {exc}") from exc + except subprocess.TimeoutExpired as exc: + raise ToolError(f"command timed out after {timeout}s", code="timeout", cmd=cmd) from exc + return completed.returncode, completed.stdout or "", completed.stderr or "" + + def _log(self, message: str) -> None: + try: + print(f"[greeneye-mcp] {message}", file=self._log_stream, flush=True) + except Exception: # a broken log stream must not break serving + pass + + +# --------------------------------------------------------------------------- +# small argument helpers +# --------------------------------------------------------------------------- + + +def _require(args: dict, key: str): + if key not in args or args[key] is None: + raise ToolError(f"missing required argument {key!r}", code="invalid_arguments") + return args[key] + + +def _as_int(value, name: str) -> int: + if isinstance(value, bool): + raise ToolError(f"{name} must be an integer", code="invalid_arguments") + if isinstance(value, int): + return int(value) + if isinstance(value, str): + try: + return int(value, 0) + except ValueError as exc: + raise ToolError(f"{name}: cannot parse integer from {value!r}", code="invalid_arguments") from exc + raise ToolError(f"{name} must be an integer", code="invalid_arguments") + + +def _bounded(value: int, minimum: int, maximum: int, name: str) -> int: + if not minimum <= value <= maximum: + raise ToolError(f"{name} must be in {minimum}..{maximum}", code="invalid_arguments") + return value + + +def _hex_bytes(text: str) -> bytes: + cleaned = str(text).strip().replace(" ", "").replace(":", "").replace(",", "") + if cleaned.startswith("0x") or cleaned.startswith("0X"): + cleaned = cleaned[2:] + try: + return bytes.fromhex(cleaned) + except ValueError as exc: + raise ToolError(f"bad hex data: {exc}", code="invalid_arguments") from exc + + +def _tool_error(payload: dict) -> dict: + return { + "content": [{"type": "text", "text": json.dumps(payload, sort_keys=True, default=str)}], + "isError": True, + } + + +def _count_findings(path: str) -> dict: + counts: dict[str, int] = {} + if not os.path.isfile(path): + return counts + try: + with open(path, "r", encoding="utf-8") as fh: + for line in fh: + line = line.strip() + if not line: + continue + try: + outcome = json.loads(line).get("outcome", "?") + except ValueError: + continue + counts[outcome] = counts.get(outcome, 0) + 1 + except OSError: + return counts + return counts + + +def _version() -> str: + try: + import greeneye + + return greeneye.__version__ + except Exception: # pragma: no cover - only on a broken install + return "unknown" + + +# --------------------------------------------------------------------------- +# plugin entry point +# --------------------------------------------------------------------------- + + +def build_serve_parser(prog: str = "mcp-serve") -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog=prog, + description="serve the active green-eye session over MCP (JSON-RPC over stdio or loopback TCP)", + ) + parser.add_argument("--stdio", action="store_true", help="serve over stdin/stdout (default)") + parser.add_argument("--tcp", metavar="HOST:PORT", type=parse_tcp, help="serve over loopback TCP instead") + parser.add_argument("--allow-write", action="store_true", help="enable memory writes and remote calls") + parser.add_argument("--allow-inject", action="store_true", help="enable service injection") + return parser + + +def register(ge) -> None: + """green-eye plugin hook: add the ``mcp-serve`` command.""" + + def mcp_serve(session, argv: list[str]) -> int: + parser = build_serve_parser() + try: + options = parser.parse_args(argv) + except SystemExit as exc: # --help must not kill the REPL + return int(exc.code or 0) + server = GreenEyeMCPServer( + session, + allow_write=options.allow_write, + allow_inject=options.allow_inject, + api=ge, + owns_session=False, + ) + ge.log( + "mcp server", + phase="start", + transport=f"tcp:{options.tcp[0]}:{options.tcp[1]}" if options.tcp else "stdio", + allow_write=options.allow_write, + allow_inject=options.allow_inject, + ) + try: + if options.tcp: + server.serve_tcp(*options.tcp) + else: + server.serve_stdio() + finally: + if getattr(session.recorder, "dir", None): + ge.log("mcp server", phase="stop", requests="see events") + return 0 + + ge.add_command( + "mcp-serve", + mcp_serve, + help="serve this live session over MCP (stdlib JSON-RPC; --allow-write/--allow-inject to enable writes)", + ) + + +# --------------------------------------------------------------------------- +# standalone entry point +# --------------------------------------------------------------------------- + + +def parse_tcp(text: str) -> tuple[str, int]: + host, _, port_text = text.rpartition(":") + host = host or "127.0.0.1" + if host not in LOOPBACK_HOSTS: + raise argparse.ArgumentTypeError(f"MCP is local-only; {host!r} is not loopback (use 127.0.0.1)") + try: + port = int(port_text) + except ValueError as exc: + raise argparse.ArgumentTypeError(f"expected HOST:PORT, got {text!r}") from exc + if not 0 <= port <= 65535: + raise argparse.ArgumentTypeError(f"port out of range: {port}") + return host, port + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(prog="mcp_server", description="green-eye MCP server (roadmap R3)") + transport = parser.add_mutually_exclusive_group() + transport.add_argument("--stdio", action="store_true", help="serve over stdin/stdout (default)") + transport.add_argument("--tcp", metavar="HOST:PORT", type=parse_tcp, help="serve over loopback TCP instead") + parser.add_argument("--kernel", help="firmware ELF to launch (default: bundled demo when built)") + parser.add_argument("--backend", choices=("qemu", "native"), default="qemu", help="execution backend") + parser.add_argument("--record", nargs="?", const=True, default=None, help="record to a path (or dumps/)") + parser.add_argument("--name", default="mcp", help="session name") + parser.add_argument("--plugins", action="append", default=[], help="extra plugin path (repeatable)") + parser.add_argument("--allow-write", action="store_true", help="enable memory writes and remote calls") + parser.add_argument("--allow-inject", action="store_true", help="enable service injection") + return parser + + +def _launch_standalone(args) -> tuple[object, object]: + import greeneye as ge + from greeneye.plugins import PluginManager + + os.environ.setdefault("GREENEYE_DUMPS", _DUMPS) + kernel = args.kernel + if kernel and not os.path.isfile(kernel): + raise GreenEyeError(f"kernel not found: {kernel}") + if not kernel and os.path.isfile(_DEMO_ELF): + kernel = _DEMO_ELF + session = ge.launch(kernel=kernel, backend=args.backend, record=args.record, name=args.name) + manager = PluginManager(session) + manager.load_all(args.plugins) + return session, manager.api + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + try: + session, api = _launch_standalone(args) + except Exception as exc: + print(f"greeneye-mcp error: {exc}", file=sys.stderr) + return 2 + server = GreenEyeMCPServer( + session, + allow_write=args.allow_write, + allow_inject=args.allow_inject, + api=api, + owns_session=True, + ) + try: + if args.tcp: + server.serve_tcp(*args.tcp) + else: + server.serve_stdio() + except KeyboardInterrupt: + print("[greeneye-mcp] interrupted", file=sys.stderr) + finally: + if not server.session_closed: + try: + session.close() + except Exception: # closing is best-effort on shutdown + pass + return 0 + + +if __name__ == "__main__": # pragma: no cover - process entry point + raise SystemExit(main()) diff --git a/debug/risc-v-gsp-rm-vm/plugins/rmharness_plugin.py b/debug/risc-v-gsp-rm-vm/plugins/rmharness_plugin.py new file mode 100644 index 0000000..085751c --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/plugins/rmharness_plugin.py @@ -0,0 +1,120 @@ +"""green-eye plugin: RM host-oracle harness commands. + +Registers REPL commands that run the read-only Open-IOV RM tools through the +``rmharness`` wrapper scripts and record each run into the active session: + + greeneye> rmtool gpu list + greeneye> rmtool mem stats --format=json + greeneye> rmtop --once + greeneye> rmq + +Every run is recorded as a ``rm_harness_run`` event with a captured +stdout+stderr blob (via ``ge.recorder.blob``); write/ownership operations are +refused and recorded as ``rm_harness_skip``. The plugin is stdlib-only and +never touches kernel modules, VFIO, mdevs or the live GPU state. +""" +from __future__ import annotations + +import hashlib +import os +import subprocess +import sys +import time + +_TOOL_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +_HARNESS = os.path.join(_TOOL_ROOT, "rmharness") +_WRAPPERS = { + "rmtool": os.path.join(_HARNESS, "rmtool.sh"), + "rmtop": os.path.join(_HARNESS, "rmtop.sh"), + "rmq": os.path.join(_HARNESS, "rmq.sh"), +} +#: arguments that would name a write/ownership operation on any tool. +_BLOCKED = frozenset({ + "alloc", "allocate", "map", "unmap", "pin", "free", "create", + "destroy", "submit", "bind", "unbind", "reset", "start", "stop", + "load", "unload", "reboot", "write", "poke", "set", +}) + + +def _blocked_arg(args): + for arg in args: + token = arg.split("=", 1)[0].lstrip("-").lower() + if token in _BLOCKED: + return arg + return None + + +def _record(ge, tool, args, proc, duration): + stdout = proc.stdout or "" + stderr = proc.stderr or "" + payload = (stdout + stderr).encode("utf-8", "replace") + blob = None + try: + blob = ge.recorder.blob( + f"rmharness/{tool}-{time.time_ns()}.txt", payload) + except Exception: + blob = None + fields = dict( + plugin="rmharness", + tool=tool, + argv=list(args), + exit=proc.returncode, + duration_s=round(duration, 6), + stdout_sha256=hashlib.sha256( + stdout.encode("utf-8", "replace")).hexdigest(), + stderr_sha256=hashlib.sha256( + stderr.encode("utf-8", "replace")).hexdigest(), + stdout_bytes=len(stdout.encode("utf-8", "replace")), + stderr_bytes=len(stderr.encode("utf-8", "replace")), + blob=blob, + readonly=True, + ) + ge.recorder.event("rm_harness_run", **fields) + + +def _run(ge, tool, args): + wrapper = _WRAPPERS[tool] + if not os.path.isfile(wrapper): + print(f" rmharness: wrapper missing: {wrapper}", file=sys.stderr) + ge.recorder.event("rm_harness_error", plugin="rmharness", + tool=tool, error="wrapper missing") + return 2 + bad = _blocked_arg(args) + if bad is not None: + print(f" rmharness: refused '{bad}': read-only policy " + f"(see docs/17-RM-HARNESS.md)", file=sys.stderr) + ge.recorder.event("rm_harness_skip", plugin="rmharness", tool=tool, + argv=list(args), reason=f"blocked argument {bad!r}") + return 2 + start = time.monotonic() + proc = subprocess.run([wrapper, *args], capture_output=True, text=True) + duration = time.monotonic() - start + if proc.stdout: + print(proc.stdout, end="") + if proc.stderr: + print(proc.stderr, end="", file=sys.stderr) + _record(ge, tool, args, proc, duration) + directory = getattr(ge.recorder, "dir", None) + if directory: + print(f" rmharness: recorded {tool} run -> {directory}") + return proc.returncode + + +def register(ge): + ge.log("rmharness plugin loaded; tools=" + ",".join(sorted(_WRAPPERS))) + + def cmd_rmtool(session, args): + return _run(ge, "rmtool", args) + + def cmd_rmtop(session, args): + return _run(ge, "rmtop", args) + + def cmd_rmq(session, args): + return _run(ge, "rmq", args) + + ge.add_command("rmtool", cmd_rmtool, + help="rmtool <args>: read-only RM oracle (gpu/engine/mem)") + ge.add_command("rmtop", cmd_rmtop, + help="rmtop <args>: dashboard; use --once for capture") + ge.add_command("rmq", cmd_rmq, + help="rmq: existing tools/rmq host oracle (read-only)") |