diff --git a/harbor/cli.py b/harbor/cli.py index fc74c06..4a1c1bf 100644 --- a/harbor/cli.py +++ b/harbor/cli.py @@ -255,7 +255,7 @@ def build_parser(): p = sub.add_parser("who", help="who has worked on a scope (notes, reviews, claims, messages)") p.add_argument("scope") p = sub.add_parser("re", help="reverse-engineering lookups from the project's image and annotations") - p.add_argument("action", choices=["disasm", "xrefs", "bytes", "lookup", "ghidra", "ghidra-tools", "debugger-tools", "match", "similar", "struct-audit", "behaviour"]) + p.add_argument("action", choices=["disasm", "xrefs", "bytes", "lookup", "summary", "ghidra", "ghidra-tools", "ghidra-check", "debugger-tools", "match", "similar", "struct-audit", "behaviour"]) p.add_argument("target", nargs="?", default="", help="an address (0x4AD850) or a name; for ghidra, the tool") p.add_argument("arguments", nargs="?", default="{}", help="ghidra: the tool's arguments as JSON") p.add_argument("--decomp", help="match: the checkout or worktree to compile from") @@ -722,6 +722,21 @@ def main(argv=None): elif args.action == "ghidra-tools": res = Ghidra(harbor.project.reverse).tools() out(res, "\n".join(f"{t['name']:<26} {t['description'].splitlines()[0][:90]}" for t in res)) + elif args.action == "ghidra-check": + from .reverse import ghidra_check + res = ghidra_check(harbor.project.reverse) + mark = lambda ok: "ok " if ok else ("?? " if ok is None else "FAIL") + out(res, "\n".join( + [f"{'Ghidra serves our program' if res['ok'] else 'Ghidra does NOT serve our program'}: " + f"{res['program']} ({res['functions_in_ghidra']} functions); image {res['image_base']}, " + f"{res['image_size']} bytes, sha256 {res['image_sha256'][:16]}..."] + + [f" {mark(c['ok'])} {c['check']}: " + (f"expected {c['expected']}, Ghidra {c['ghidra']}" if "ghidra" in c + else ("differs at " + ", ".join(c["differ_at"]) if c.get("differ_at") else "all equal")) + for c in res["checks"]])) + elif args.action == "summary": + from .reverse import format_summary + res = r.summary(args.target) + out(res, format_summary(res)) elif args.action == "ghidra": text = Ghidra(harbor.project.reverse).call(args.target, json.loads(args.arguments)) out({"result": text}, text) diff --git a/harbor/mcp_server.py b/harbor/mcp_server.py index 139af26..f677085 100644 --- a/harbor/mcp_server.py +++ b/harbor/mcp_server.py @@ -38,6 +38,8 @@ review queue, presence and stop flags. Protocol: 10. Before your session ends or your allowance runs out, write handoff(): finished, in flight, next. 11. To read the original machine code use re_disassemble, re_xrefs, re_bytes and re_lookup: they read the project's own image with the team's current names, so do not write your own disassembler or offset maths. + re_summary(address) gives the overview in one call: name, prototype, size, convention, readable status, + callers and callees with their status, globals, strings, vtable slots and the Mac/Xbox hints. For Ghidra's view (decompiler, p-code, stack frames, struct field references, types) use ghidra_tools() once, then ghidra(tool, arguments): one shared read-only Ghidra session holding the team's annotations. To check a value in the running game (x87 control word, `this`, struct fields), use debugger_tools() and @@ -563,6 +565,14 @@ def re_lookup(target: str) -> dict: return _re(lambda r: r.lookup(target)) +@mcp.tool +def re_summary(target: str) -> dict: + """One call before starting on a function: its current name and prototype, size and calling convention (read + from the bytes), whether it is readable on main, its callers and callees with their names and whether each is + readable on main, the globals and strings it uses, the vtable slots it fills, and the Mac/Xbox name hints.""" + return _re(lambda r: r.summary(target)) + + _ghidra = {} @@ -692,6 +702,17 @@ def ghidra_tools() -> list: return _gh(lambda g: g.tools()) +@mcp.tool() +def ghidra_check() -> dict: + """Is the shared Ghidra session serving our program? Compares its address range with the project image's base + and size, the SHA-256 of the file it imported with ours, and sample bytes; ok is false when any differs.""" + try: + from .reverse import ghidra_check as check + return check(harbor().project.reverse) + except (HarborError, projects.ProjectError) as error: + raise ToolError(str(error)) from None + + @mcp.tool() def ghidra(tool: str, arguments: dict | None = None) -> str: """Run one tool in the shared read-only Ghidra session, e.g. ghidra('get_code', {'function': 'GetFogLevel', diff --git a/harbor/reverse.py b/harbor/reverse.py index f7ebc14..cee68c8 100644 --- a/harbor/reverse.py +++ b/harbor/reverse.py @@ -7,6 +7,8 @@ Configured per project in a [reverse] table: imports = 'R:/.../imports.json' optional: [{module, name, slot}] annotations = 'R:/.../annotations' names.csv, functions.csv, globals.csv, prototypes.h src = 'R:/.../src' optional: the decompiled C files (_.c) + sections = 'R:/.../sections.json' optional: [{name, start, end}] (default: sections.json beside the image), + used to find the vtables and pointer tables a function sits in Every answer comes from those files, read fresh when they change, so the agents see the team's current names. Nothing here writes anything.""" @@ -42,6 +44,7 @@ class Reverse: self.conventions_path = Path(config["conventions"]) if config.get("conventions") else None self.source_files = Path(config["source_files"]) if config.get("source_files") else None self.mac_names = Path(config["mac_names"]) if config.get("mac_names") else None + self.sections_path = Path(config["sections"]) if config.get("sections") else None self._cache = {} # ----------------------------------------------------------------- loading (cached by file mtime) @@ -491,6 +494,176 @@ class Reverse: break return out + # ----------------------------------------------------------------- one-call summary + def _src_index(self): + """address -> path of src/**/_.c (one directory walk, reused for 30 seconds).""" + import time + hit = self._cache.get("src_index") + if hit and time.monotonic() - hit[0] < 30: + return hit[1] + out = {} + if self.src and self.src.exists(): + for folder, _, files in os.walk(self.src): + for name in files: + m = re.match(r"([0-9A-Fa-f]{6,8})_.*\.c$", name) + if m: + out.setdefault(int(m.group(1), 16), Path(folder) / name) + self._cache["src_index"] = (time.monotonic(), out) + return out + + def status_of(self, addr): + """The decompiled file's status on main ('readable', 'raw', ...), or None when there is no file.""" + path = self._src_index().get(addr) + if not path: + return None + try: + with path.open(encoding="utf-8", errors="replace") as f: + head = [next(f, "") for _ in range(8)] + except OSError: + return None + return next((l.split("status:", 1)[1].strip() for l in head if "status:" in l), "raw") + + def _sections(self): + """[(name, start, end)] from sections.json beside the image (or [reverse] sections), else [].""" + path = self.sections_path or self.image.with_name("sections.json") + return self._load("sections", path, lambda p: [ + (s["name"], _int(s["start"]), _int(s["end"])) for s in json.loads(p.read_text(encoding="utf-8"))] if p else []) + + def _code_data_refs(self): + facts = self._facts() + key = ("data_refs", id(facts)) + if key not in self._cache: + self._cache[key] = {_int(a) for d in facts.values() for a in d.get("data", [])} + return self._cache[key] + + def _mac_vtables(self): + return self._load("mac_vtables", self.mac_names.with_name("vtables.csv") if self.mac_names else None, + lambda p: {int(r["pc_vtable"], 16): r.get("class") for r in + csv.DictReader(p.read_text(encoding="utf-8").splitlines()) + if (r.get("pc_vtable") or "").startswith("0x")} if p else {}) + + def vtable_slots(self, addr): + """Where the function's address sits in a table of code pointers outside .text: the table start (walked back + over function addresses to the entry the code references), the slot index, and what the table is.""" + data = self._bytes() + facts = self._facts() + sections = self._sections() + text = next(((lo, hi) for name, lo, hi in sections if name == ".text"), None) + tables = [(lo, hi) for name, lo, hi in sections if name in (".rdata", ".data")] + if not tables: # no section list: everything past the last function + end = max((a + int(d.get("size") or 0) for a, d in facts.items()), default=self.base) + tables = [(end, self.base + len(data))] + refs, mac = self._code_data_refs(), self._mac_vtables() + is_code = (lambda v: v in facts) if facts else (lambda v: text and text[0] <= v < text[1]) + needle = struct.pack("= 0 and len(out) < 40: + if pos % 4 == 0: + start = pos + while start - 4 >= lo - self.base and (start + self.base) not in refs and \ + (start + self.base) not in mac and \ + is_code(struct.unpack_from(" 3 and row[3]: + out["note"] = row[3] if len(row[3]) < 600 else row[3][:600] + "..." + callers = sorted(set(self._callers().get(start, []))) + callees = list(dict.fromkeys(_int(a) for a in d.get("calls", []))) + out["callers"] = [fn(a) for a in callers] + out["callees"] = [fn(a) for a in callees] + if d.get("imports"): + out["imports"] = d["imports"] + glob = self._csv("globals.csv") + out["globals"] = [{"address": f"0x{a:X}", "name": self.name_of(a), + **({"type": glob[a][2]} if a in glob and len(glob[a]) > 2 and glob[a][2] else {})} + for a in dict.fromkeys(_int(x) for x in d.get("data", []))] + out["strings"] = d.get("strings", []) + out["vtable_slots"] = self.vtable_slots(start) + out["hints"] = self.origins(start) + out["counts"] = {"callers": len(callers), "callers_readable": sum(c["readable"] for c in out["callers"]), + "callees": len(callees), "callees_readable": sum(c["readable"] for c in out["callees"])} + return out + + +def format_summary(s): + """Reverse.summary() as text for the command line.""" + mark = lambda f: f"{f['address']} {f['name'] or '(unnamed)'} [{'readable' if f['readable'] else f['status']}]" + lines = [f"{s['name'] or '(unnamed)'} {s['address']}..{s['end']} ({s['size']} bytes)" + + (f", asked {s['asked']}" if s.get("asked") else ""), + f"status on main: {s['status']}" + (f" ({s['file']})" if s.get("file") else ""), + f"convention: {s['convention'] or 'unknown'}", + "prototype: " + (s["prototype"] or "none in prototypes.h")] + c = s["counts"] + lines.append(f"callers ({c['callers']}, {c['callers_readable']} readable):") + lines += [f" {mark(f)}" for f in s["callers"]] or [" none"] + lines.append(f"callees ({c['callees']}, {c['callees_readable']} readable):") + lines += [f" {mark(f)}" for f in s["callees"]] or [" none"] + if s.get("imports"): + lines.append("imports: " + ", ".join(s["imports"])) + lines.append(f"globals ({len(s['globals'])}):") + lines += [f" {g['address']} {g['name'] or '(unnamed)'}" + (f" : {g['type']}" if g.get("type") else "") + for g in s["globals"]] or [" none"] + if s["strings"]: + lines.append("strings: " + ", ".join(repr(x) for x in s["strings"])) + lines.append("vtable slots:" + ("" if s["vtable_slots"] else " none")) + lines += [f" {v['kind']} {v['table']}" + (f" {v['table_name']}" if v.get("table_name") else "") + + (f" (Mac class {v['mac_class']})" if v.get("mac_class") else "") + f" slot {v['slot']} (+{v['offset']})" + for v in s["vtable_slots"]] + h = s["hints"] + if h.get("mac_name"): + lines.append(f"Mac name: {h['mac_name'].get('demangled')} ({h['mac_name'].get('confidence')})") + if h.get("source_file"): + sf = h["source_file"] + lines.append(f"Xbox source file: {sf.get('source_file')}" + (f" lines {sf['lines']}" if sf.get("lines") else "") + + f" ({sf.get('confidence')})") + if not h: + lines.append("Mac/Xbox hints: none") + return "\n".join(lines) + class Ghidra: """A small client for the shared, read-only Ghidra MCP server (GhidrAssistMCP over streamable HTTP), so every @@ -572,6 +745,63 @@ class Ghidra: return re.sub(r"^\[Context\] Operating on: [^\n]*\n+", "", text) +def _hexdump_bytes(text): + """The bytes of a GhidrAssistMCP get_data_at hexdump ('401000 53 55 8b ... |SU.|').""" + out = bytearray() + for line in text.splitlines(): + m = re.match(r"^\s*[0-9A-Fa-f]{4,16}\s{2}((?:[0-9A-Fa-f]{2}\s{1,2})+)", line + " ") + if m: + out += bytes.fromhex("".join(m.group(1).split())) + return bytes(out) + + +def ghidra_check(config, samples=16, sample_len=128): + """Is the shared Ghidra server serving our program? Read-only: compares the program's address range with the + project image (base and size), the SHA-256 of the file Ghidra imported (when that path is readable here) with + the image's, and sample bytes read through Ghidra (get_data_at) with the image at spread-out addresses.""" + import hashlib + reverse = Reverse(config) + image = reverse._bytes() + if not image: + raise HarborError(f"the project image {reverse.image} is missing") + base = reverse.base + g = Ghidra(config) + info = g.call("get_binary_info", {}) + binaries = g.call("list_binaries", {}) + field = lambda label, text: (re.search(rf"^\s*{label}:\s*(.+?)\s*$", text, re.M) or [None, None])[1] + lo, hi = field("Min Address", info), field("Max Address", info) + checks = [] + want_lo, want_hi = base, base + len(image) - 1 + got = (int(lo, 16), int(hi, 16)) if lo and hi else None + checks.append({"check": "address range", "ok": got == (want_lo, want_hi), + "expected": f"0x{want_lo:X}..0x{want_hi:X} ({len(image)} bytes)", + "ghidra": f"0x{got[0]:X}..0x{got[1]:X} ({got[1] - got[0] + 1} bytes)" if got else info.strip()[:200]}) + sha = hashlib.sha256(image).hexdigest() + exe = field("Executable Path", binaries) + local = Path(exe[1:] if exe and re.match(r"^/[A-Za-z]:/", exe) else exe) if exe else None + if local and local.is_file(): + theirs = hashlib.sha256(local.read_bytes()).hexdigest() + checks.append({"check": "sha256 of the imported file", "ok": theirs == sha, "expected": sha, + "ghidra": theirs, "file": str(local)}) + else: + checks.append({"check": "sha256 of the imported file", "ok": None, "expected": sha, + "ghidra": f"not readable here ({exe or 'no path reported'})"}) + step = max(sample_len, (len(image) - sample_len) // max(1, samples - 1)) & ~0xF + bad = [] + for i in range(samples): + addr = base + min(i * step, len(image) - sample_len) + theirs = _hexdump_bytes(g.call("get_data_at", {"address": f"0x{addr:X}", "len": sample_len})) + ours = image[addr - base:addr - base + sample_len] + if theirs != ours: + bad.append(f"0x{addr:X}") + checks.append({"check": f"{samples} byte samples of {sample_len}", "ok": not bad, + **({"differ_at": bad} if bad else {})}) + ok = all(c["ok"] is not False for c in checks) + return {"ok": ok, "program": field("Name", info), "image": str(reverse.image), "image_sha256": sha, + "image_base": f"0x{base:X}", "image_size": len(image), "functions_in_ghidra": field("Function Count", info), + "checks": checks} + + class Debugger: """A per-agent x64dbg session through the x64dbg-automate MCP server (stdio), started on first use and kept for the life of this Harbor process, so breakpoints and the paused game survive between calls. Configured by diff --git a/tests/test_reverse.py b/tests/test_reverse.py index caa32b8..d79483b 100644 --- a/tests/test_reverse.py +++ b/tests/test_reverse.py @@ -58,6 +58,105 @@ class ReverseTools(unittest.TestCase): with self.assertRaises(HarborError): Reverse({}) + def test_summary_names_status_vtables_and_hints(self): + image = bytearray((self.d / "image.bin").read_bytes()) + image[0xC0:0xC8] = struct.pack("