diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/greeneye/cli.py')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/greeneye/cli.py | 688 |
1 files changed, 688 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/cli.py b/debug/risc-v-gsp-rm-vm/greeneye/cli.py new file mode 100644 index 0000000..5bfcee2 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/greeneye/cli.py @@ -0,0 +1,688 @@ +"""green-eye command line interface. + +Commands: + info environment/tooling check + demo boot the bundled uproc demo and run a canned instrumentation pass + run run a Python instrumentation script against a target + attach interactive REPL against a QEMU gdbstub + probe run an arbitrary firmware (default: the real GSP fwimage) for N + instructions and report the first divergence + record boot a firmware and record a memory-activity session + analyze render a recorded session to a symbol-annotated report + compare A/B two recorded sessions at the symbolic level + symbols build/inspect a green-eye symbol database from Ghidra exports + plugins list discoverable plugins +""" +from __future__ import annotations + +import argparse +import cmd +import os +import sys +import tempfile +import time +from pathlib import Path + +import greeneye as ge +from . import analyze as _analyze +from . import ghidra as _ghidra +from .elf import load_elf +from .errors import GreenEyeError +from .plugins import PluginManager +from .structs import schema_from_json +from .symbols import SymbolDB +from .target import QemuTarget + +TOOL_ROOT = Path(__file__).resolve().parent.parent +DUMPS = TOOL_ROOT / "dumps" +DEMO_ELF = TOOL_ROOT / "firmware" / "uproc_demo" / "build" / "uproc_demo.elf" +DEMO_SCHEMA = TOOL_ROOT / "firmware" / "uproc_demo" / "gsp_ctrl.schema.json" +DEFAULT_FWIMAGE = TOOL_ROOT.parents[1] / "firmware" / "57X-GSP" / "570.144" / "gsp-570.144.fwimage.bin" + +RPC_HANDLERS = ["rpc_ping", "rpc_echo", "rpc_fill", "rpc_sum", "rpc_state", "rpc_fault"] + + +# --------------------------------------------------------------------------- +# helpers +# --------------------------------------------------------------------------- + + +def _ensure_dumps_env() -> None: + os.environ.setdefault("GREENEYE_DUMPS", str(DUMPS)) + + +def _build_demo(quiet: bool = False) -> Path: + import subprocess + + fw_dir = TOOL_ROOT / "firmware" / "uproc_demo" + result = subprocess.run( + ["make", "-C", str(fw_dir)], + capture_output=quiet, + text=True, + ) + if result.returncode != 0: + output = (result.stdout or "") + (result.stderr or "") + raise GreenEyeError(f"demo firmware build failed:\n{output}") + if not DEMO_ELF.exists(): + raise GreenEyeError(f"demo build produced no {DEMO_ELF}") + return DEMO_ELF + + +def _load_schema(path: Path): + import json + + with open(path, "r", encoding="utf-8") as fh: + return schema_from_json(json.load(fh)) + + +def _start_session( + *, + kernel: str | None, + port: int | None, + record, + name: str, + plugins: list[str] | None = None, +): + if port: + db = SymbolDB() + if kernel: + db.load_elf(kernel) + session = ge.attach(port=port, record=record, symbols=db, name=name) + else: + session = ge.launch(kernel=kernel, record=record, name=name) + manager = PluginManager(session) + manager.load_all(plugins) + return session, manager + + +def _print_session_dir(session) -> None: + directory = getattr(session.recorder, "dir", None) + if directory: + print(f"[green-eye] session: {directory}") + + +# --------------------------------------------------------------------------- +# commands +# --------------------------------------------------------------------------- + + +def cmd_info(args) -> int: + from .target import DEFAULT_QEMU + + print(f"green-eye {ge.__version__}") + print(f"tool root: {TOOL_ROOT}") + qemu = os.environ.get("GREENEYE_QEMU", DEFAULT_QEMU) + import shutil + + resolved = shutil.which(qemu) if os.path.sep not in qemu else (qemu if os.path.exists(qemu) else None) + print(f"qemu: {resolved or 'NOT FOUND (install qemu-system-misc)'}") + if resolved: + import subprocess + + version = subprocess.run([resolved, "--version"], capture_output=True, text=True).stdout.strip() + print(f"qemu version: {version}") + for label, path in [ + ("demo firmware", DEMO_ELF), + ("demo schema", DEMO_SCHEMA), + ("default fwimage", DEFAULT_FWIMAGE), + ]: + print(f"{label}: {path} ({'present' if Path(path).exists() else 'missing'})") + cuda_names = Path(TOOL_ROOT).parents[1] / "userspace" / "cuda" / "570.144" / "annotations" / "names.tsv" + print(f"CUDA Ghidra names.tsv: {cuda_names} ({'present' if cuda_names.exists() else 'missing'})") + plugins_dir = TOOL_ROOT / "plugins" + plugins = sorted(p.name for p in plugins_dir.glob("*.py") if not p.name.startswith("_")) + print(f"bundled plugins: {', '.join(plugins) or 'none'}") + return 0 + + +def cmd_demo(args) -> int: + _ensure_dumps_env() + elf = args.kernel or (_build_demo() if not args.no_build else DEMO_ELF) + if not Path(elf).exists(): + raise GreenEyeError(f"demo firmware not built: {elf}") + schema = _load_schema(DEMO_SCHEMA) + record = True if args.record is None else args.record + session, _manager = _start_session(kernel=str(elf), port=None, record=record, name="demo") + hits: list[str] = [] + try: + ctrl = session.symbols.lookup("gsp_ctrl") + for name in RPC_HANDLERS: + sym = session.symbols.get(name) + if sym is not None: + session.hook( + sym.addr, + on_enter=lambda hit: hits.append(f"enter {hit.label()} a0={hit.arg(0):#x}"), + on_leave=lambda hit: hits.append(f"leave {hit.label()} -> {hit.retval:#x}"), + ) + session.watch(ctrl + 16, size=8, kind="w", listener="rpc_count") + print(f"[demo] ctrl={ctrl:#x} schema={schema.name} size={schema.size:#x}") + print("[demo] running the firmware self-test ...") + session.run(duration=args.duration) + banner = session.target.read_uart() if session.target else "" + print("[demo] uart:", banner.strip().splitlines()[0] if banner.strip() else "<empty>") + before = session.snapshot(ctrl, schema.size, "ctrl-after-selftest") + + print("[demo] host stimulus: PING") + session.memory.write_u64(ctrl + 32, 1) + session.memory.write_u64(ctrl + 40, 0x1234) + session.run(duration=0.5) + after = session.snapshot(ctrl, schema.size, "ctrl-after-ping") + + values = schema.decode(after, base=ctrl) + print("[demo] control block after stimulus:") + from .structs import render + + print(render(schema, values)) + diffs = ge.Memory.diff(before, after, base=ctrl) + if diffs: + print("[demo] snapshot diff (self-test -> ping):") + print(ge.Memory.render_diff(diffs, session.symbols, max_lines=16)) + + print("[demo] host stimulus: FAULT (ebreak) to demonstrate trap capture") + session.memory.write_u64(ctrl + 32, 6) + session.run(duration=1.0, max_stops=1) + pc = session.pc + sym = session.symbols.resolve(pc) + print(f"[demo] stopped at {pc:#x} {sym.name if sym else '<unknown>'} (ebreak)") + session.step() # step past the trap + print("[demo] stepped past the trap; hits:") + for line in hits[:24]: + print(" ", line) + print(f"[demo] total hook lines: {len(hits)}") + _print_session_dir(session) + finally: + session.close() + return 0 + + +def cmd_run(args) -> int: + _ensure_dumps_env() + script = Path(args.script) + if not script.exists(): + raise GreenEyeError(f"script not found: {script}") + kernel = args.kernel + if not kernel and not args.port: + kernel = str(_build_demo()) + session, manager = _start_session( + kernel=kernel, port=args.port, record=args.record or True, name=script.stem, plugins=args.plugin + ) + namespace = { + "ge": ge, + "session": session, + "s": session, + "memory": session.memory, + "symbols": session.symbols, + "recorder": session.recorder, + "greeneye": ge, + } + try: + source = script.read_text(encoding="utf-8") + exec(compile(source, str(script), "exec"), namespace) + return 0 + except SystemExit as exc: + return int(exc.code or 0) + finally: + session.close() + + +# -- interactive REPL ------------------------------------------------------- + + +class GreenEyeRepl(cmd.Cmd): + intro = "green-eye interactive instrumentation. Type 'help' for commands." + prompt = "greeneye> " + + def __init__(self, session, manager): + super().__init__() + self.session = session + self.manager = manager + self._listener_hooks = [] + + # utility + def _print_hit(self, hit): + print(" " + str(hit)) + + def _say(self, text): + print(text) + + # commands + def do_regs(self, arg): + """regs -- dump all registers.""" + regs = self.session.read_regs() + for name, value in regs.items(): + print(f" {name:<5} = {value:#018x}") + + def do_reg(self, arg): + """reg <name> [value] -- read or write one register.""" + parts = arg.split() + if not parts: + return self._say("usage: reg <name> [value]") + if len(parts) == 1: + print(f" {parts[0]} = {self.session.client.read_reg(parts[0]):#x}") + else: + self.session.client.write_reg(parts[0], int(parts[1], 0)) + self._say("ok") + + def do_x(self, arg): + """x <addr> [len] -- hexdump memory (annotated).""" + parts = arg.split() + if not parts: + return self._say("usage: x <addr> [len]") + addr = self.session.resolve(parts[0]) + length = int(parts[1], 0) if len(parts) > 1 else 64 + print(self.session.memory.hexdump(addr, length)) + + def do_mem(self, arg): + """mem <addr> <len> -- raw memory read.""" + addr, length = arg.split() + print(self.session.memory.read(self.session.resolve(addr), int(length, 0)).hex()) + + def do_w(self, arg): + """w <addr> <hexbytes> -- write memory.""" + addr, data = arg.split() + self.session.memory.write(self.session.resolve(addr), bytes.fromhex(data)) + self._say("ok") + + def do_wstr(self, arg): + """wstr <addr> <text> -- write a string plus NUL.""" + addr, _, text = arg.partition(" ") + self.session.memory.write(self.session.resolve(addr), text.encode() + b"\x00") + self._say("ok") + + def do_bp(self, arg): + """bp <addr|symbol> [sw|hw] -- set a breakpoint that prints on hit.""" + parts = arg.split() + if not parts: + return self._say("usage: bp <addr|symbol> [sw|hw]") + kind = parts[1] if len(parts) > 1 else "sw" + handle = self.session.hook(parts[0], on_enter=self._print_hit, kind=kind, listener="repl") + self._listener_hooks.append(handle) + self._say(f"breakpoint @ {handle.addr:#x} ({handle.label})") + + def do_hook(self, arg): + """hook <addr|symbol> -- alias for bp with enter/leave printing.""" + parts = arg.split() + if not parts: + return self._say("usage: hook <addr|symbol>") + handle = self.session.hook( + parts[0], on_enter=self._print_hit, on_leave=self._print_hit, listener="repl" + ) + self._listener_hooks.append(handle) + self._say(f"hooked {handle.label}") + + def do_watch(self, arg): + """watch <addr|symbol> [size] [w|r|a] -- set a watchpoint.""" + parts = arg.split() + if not parts: + return self._say("usage: watch <addr|symbol> [size] [w|r|a]") + size = int(parts[1], 0) if len(parts) > 1 else 8 + kind = parts[2] if len(parts) > 2 else "w" + wp = self.session.watch(parts[0], size=size, kind=kind, callback=self._print_hit, listener="repl") + self._say(f"watch {kind}@{wp.addr:#x} size={size}") + + def do_s(self, arg): + """s [n] -- step n instructions (default 1).""" + count = int(arg) if arg.strip() else 1 + for hit in self.session.step(count): + self._print_hit(hit) + + def do_c(self, arg): + """c [seconds] -- continue and service events for N seconds (default 1).""" + seconds = float(arg) if arg.strip() else 1.0 + stops = self.session.run(duration=seconds) + self._say(f"resumed for {seconds}s, serviced {stops} stops") + + def do_events(self, arg): + """events [n] -- show the last n recorded events (default 20).""" + from .record import load_events + + directory = getattr(self.session.recorder, "dir", None) + if not directory: + return self._say("this session is not recording") + events = load_events(directory) + for event in events[-int(arg or 20) :]: + print(" " + str(event)) + + def do_snap(self, arg): + """snap <label> <addr> <len> -- snapshot a memory region into the session.""" + parts = arg.split() + if len(parts) < 3: + return self._say("usage: snap <label> <addr> <len>") + data = self.session.memory.snapshot(self.session.resolve(parts[1]), int(parts[2], 0), name=parts[0]) + self._say(f"snapshot {parts[0]}: {len(data)} bytes") + + def do_sym(self, arg): + """sym <addr|name> -- resolve a symbol or an address.""" + target = arg.strip() + if not target: + return self._say("usage: sym <addr|name>") + try: + print(" " + self.session.symbols.annotate(self.session.resolve(target))) + except GreenEyeError as exc: + self._say(f"error: {exc}") + + def do_note(self, arg): + """note <text> -- add a note to the session recording.""" + self.session.note(arg) + self._say("noted") + + def do_quit(self, arg): + """quit -- close the session and exit.""" + return True + + do_EOF = do_quit + + def onecmd(self, line): + name = line.split()[0] if line.split() else "" + if name in self.manager.api.commands: + fn, _help = self.manager.api.commands[name] + try: + fn(self.session, line.split()[1:]) + except Exception as exc: + self._say(f"plugin error: {exc}") + return False + if name and name not in self.get_names() and not hasattr(self, f"do_{name}"): + return super().onecmd(line) + return super().onecmd(line) + + def get_names(self): + names = super().get_names() + names += [f"do_{cmd}" for cmd in self.manager.api.commands] + return names + + def default(self, line): + self._say(f"unknown command: {line.split()[0] if line.split() else line!r}") + + +def cmd_attach(args) -> int: + _ensure_dumps_env() + kernel = args.kernel + if kernel: + kernel = str(Path(kernel)) + elif not args.port: + kernel = str(_build_demo()) + session, manager = _start_session( + kernel=kernel, port=args.port, record=args.record, name="repl", plugins=args.plugin + ) + try: + GreenEyeRepl(session, manager).cmdloop() + finally: + session.close() + return 0 + + +def cmd_launch(args) -> int: + """Start a target, print the gdbstub endpoint, and keep QEMU alive.""" + _ensure_dumps_env() + kernel = args.kernel or str(_build_demo()) + if not Path(kernel).exists(): + raise GreenEyeError(f"kernel not found: {kernel}") + session = ge.launch(kernel=str(kernel), record=args.record, name="launch", keep_alive=True) + port = session.target.port + print(f"[green-eye] target running: {kernel}") + print(f"[green-eye] gdbstub: 127.0.0.1:{port}") + print(f"[green-eye] attach: ./greeneye.sh attach --port {port} --kernel {kernel}") + print("[green-eye] Ctrl-C stops waiting; QEMU keeps running for attach.") + try: + while True: + time.sleep(1) + except KeyboardInterrupt: + print("[green-eye] detached (QEMU still running)") + return 0 + + +def cmd_record(args) -> int: + _ensure_dumps_env() + elf = args.kernel or str(_build_demo()) + session, _manager = _start_session(kernel=str(elf), port=None, record=args.record or True, name=args.name) + try: + for spec in args.watch or []: + addr_or_sym, size = (spec.split(":", 1) + ["8"])[:2] + session.watch(addr_or_sym, size=int(size, 0)) + for spec in args.snapshot or []: + addr_or_sym, size = spec.split(":", 1) + session.snapshot(session.resolve(addr_or_sym), int(size, 0), name=addr_or_sym) + session.run(duration=args.duration) + print(f"[record] serviced run for {args.duration}s") + _print_session_dir(session) + finally: + session.close() + return 0 + + +def cmd_probe(args) -> int: + _ensure_dumps_env() + elf = Path(args.elf) + if not elf.exists(): + raise GreenEyeError(f"firmware not found: {elf}") + image = load_elf(str(elf)) + print(f"[probe] {image.summary()}") + relocate = not args.no_relocate + bias = (0x80000000 - image.min_vaddr) if relocate else 0 + kernel_path = str(elf) + if relocate: + from .elf import write_flat_kernel + + out = Path(tempfile.mkdtemp(prefix="greeneye-probe-")) / (elf.name + ".flat.bin") + kernel_path, entry_addr = write_flat_kernel(str(elf), str(out), load_addr=0x80000000) + size = Path(kernel_path).stat().st_size + print( + f"[probe] flat kernel {kernel_path} ({size} bytes); " + f"load 0x80000000 entry {entry_addr:#x} (original {image.entry:#x})" + ) + target = QemuTarget( + kernel=kernel_path, + dlog="guest_errors,unimp,int", + memory_mb=max(256, (image.max_vaddr - image.min_vaddr) // (1024 * 1024) + 64), + ) + target.start(paused=True) + from .rsp import RspClient + + client = RspClient(target.host, target.port) + client.connect() + client.sync() + symbols = SymbolDB() + if args.symbols: + symbols.load_json(args.symbols) + session = ge.Session( + client=client, + target=target, + symbols=symbols, + record=args.record or True, + name=f"probe-{elf.stem}", + meta={"firmware": str(elf), "entry": image.entry, "bias": bias, "kernel": kernel_path}, + ) + session.load_image(image, bias=bias) + try: + if args.instructions: + print(f"[probe] stepping {args.instructions} instructions ...") + session.step(args.instructions) + else: + print(f"[probe] running for {args.duration}s ...") + session.run(duration=args.duration) + pc = session.pc + sym = session.symbols.resolve(pc) + print(f"[probe] stopped at {pc:#x} {sym.name if sym else '<unknown>'}") + from .analyze import describe_fault, parse_qemu_faults + + qlog = target.read_qemu_log() + faults = parse_qemu_faults(qlog) + if faults: + print("[probe] " + describe_fault(faults[0], bias=bias)) + print(f"[probe] total exceptions: {len(faults)}") + else: + print("[probe] no exception recorded (execution stayed in mapped memory)") + tail = qlog.strip().splitlines() + if tail: + print("[probe] qemu log tail:") + for line in tail[-6:]: + print(" ", line) + _print_session_dir(session) + finally: + session.close() + return 0 + + +def cmd_analyze(args) -> int: + symbols = None + if args.symbols: + symbols = SymbolDB() + symbols.load_json(args.symbols) + report = _analyze.render_summary(args.session, symbols=symbols) + if args.out: + Path(args.out).write_text(report, encoding="utf-8") + print(f"wrote {args.out}") + else: + print(report) + return 0 + + +def cmd_compare(args) -> int: + symbols = SymbolDB() if args.symbols else None + report = _analyze.compare_sessions(args.a, args.b, symbols=symbols) + print(report) + return 0 + + +def cmd_symbols(args) -> int: + if args.symbols_command == "export": + db = _ghidra.build_symbol_db( + names_tsv=args.names, + index_tsv=args.index, + exports_tsv=args.exports, + callgraph_tsv=args.callgraph, + elf=args.elf, + bias=args.bias, + ) + db.save_json(args.out, meta={"bias": args.bias}) + print(f"wrote {args.out}: {len(db)} symbols") + return 0 + if args.symbols_command == "resolve": + db = SymbolDB() + if args.db: + db.load_json(args.db) + print(db.annotate(int(args.addr, 0))) + return 0 + return 1 + + +def cmd_plugins(args) -> int: + plugins_dir = TOOL_ROOT / "plugins" + print("bundled plugins:") + for path in sorted(plugins_dir.glob("*.py")): + if path.name.startswith("_"): + continue + print(f" {path.name}") + env = os.environ.get("GREENEYE_PLUGINS") + if env: + print(f"GREENEYE_PLUGINS: {env}") + return 0 + + +# --------------------------------------------------------------------------- +# parser +# --------------------------------------------------------------------------- + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(prog="greeneye", description=__doc__.splitlines()[0]) + parser.add_argument("--version", action="version", version=f"green-eye {ge.__version__}") + sub = parser.add_subparsers(dest="command", required=True) + + sub.add_parser("info", help="environment/tooling check").set_defaults(fn=cmd_info) + + demo = sub.add_parser("demo", help="boot the demo uproc and instrument it") + demo.add_argument("--kernel", help="firmware ELF (default: built demo)") + demo.add_argument("--no-build", action="store_true", help="do not rebuild the demo") + demo.add_argument("--record", nargs="?", const=True, default=None, help="record to a path (or default dumps/)") + demo.add_argument("--duration", type=float, default=1.0) + demo.set_defaults(fn=cmd_demo) + + run = sub.add_parser("run", help="run a Python instrumentation script") + run.add_argument("script") + run.add_argument("--kernel") + run.add_argument("--port", type=int) + run.add_argument("--record", nargs="?", const=True, default=None) + run.add_argument("--plugin", action="append", default=[]) + run.set_defaults(fn=cmd_run) + + attach = sub.add_parser("attach", help="interactive REPL over a gdbstub (launches if no --port)") + attach.add_argument("--port", type=int, help="attach to an existing gdbstub instead of launching") + attach.add_argument("--kernel", help="firmware ELF (default: built demo)") + attach.add_argument("--record", nargs="?", const=True, default=None) + attach.add_argument("--plugin", action="append", default=[]) + attach.set_defaults(fn=cmd_attach) + + launch = sub.add_parser("launch", help="start a target and keep it alive for attach") + launch.add_argument("--kernel", help="firmware ELF (default: built demo)") + launch.add_argument("--record", nargs="?", const=True, default=None) + launch.set_defaults(fn=cmd_launch) + + probe = sub.add_parser("probe", help="run a firmware image and report first divergence") + probe.add_argument("elf", nargs="?", default=str(DEFAULT_FWIMAGE)) + probe.add_argument("--no-relocate", action="store_true") + probe.add_argument("--instructions", type=int, default=0) + probe.add_argument("--duration", type=float, default=1.0) + probe.add_argument("--symbols", help="green-eye symbol DB JSON to annotate the stop") + probe.add_argument("--record", nargs="?", const=True, default=None) + probe.set_defaults(fn=cmd_probe) + + record = sub.add_parser("record", help="boot and record a memory-activity session") + record.add_argument("--kernel") + record.add_argument("--name", default="record") + record.add_argument("--duration", type=float, default=2.0) + record.add_argument("--watch", action="append", help="addr|symbol[:size]") + record.add_argument("--snapshot", action="append", help="addr|symbol:len") + record.add_argument("--record", nargs="?", const=True, default=None) + record.set_defaults(fn=cmd_record) + + analyze = sub.add_parser("analyze", help="render a recorded session report") + analyze.add_argument("session") + analyze.add_argument("--symbols") + analyze.add_argument("--out") + analyze.set_defaults(fn=cmd_analyze) + + compare = sub.add_parser("compare", help="A/B two recorded sessions") + compare.add_argument("a") + compare.add_argument("b") + compare.add_argument("--symbols", action="store_true") + compare.set_defaults(fn=cmd_compare) + + symbols = sub.add_parser("symbols", help="build/inspect a symbol database") + symbols_sub = symbols.add_subparsers(dest="symbols_command", required=True) + exp = symbols_sub.add_parser("export") + exp.add_argument("--names") + exp.add_argument("--index") + exp.add_argument("--exports") + exp.add_argument("--callgraph") + exp.add_argument("--elf") + exp.add_argument("--bias", type=lambda v: int(v, 0), default=0) + exp.add_argument("-o", "--out", required=True) + exp.set_defaults(fn=cmd_symbols) + res = symbols_sub.add_parser("resolve") + res.add_argument("addr") + res.add_argument("--db") + res.set_defaults(fn=cmd_symbols) + + plugins = sub.add_parser("plugins", help="list discoverable plugins") + plugins.set_defaults(fn=cmd_plugins) + + return parser + + +def main(argv: list[str] | None = None) -> int: + _ensure_dumps_env() + parser = build_parser() + args = parser.parse_args(argv) + try: + return args.fn(args) + except GreenEyeError as exc: + print(f"green-eye error: {exc}", file=sys.stderr) + return 2 + except KeyboardInterrupt: + print("interrupted", file=sys.stderr) + return 130 + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) |