re_summary: one call for a function's name, prototype, size, convention, readable status, callers and callees with their status, globals, strings, vtable slots and Mac/Xbox hints (harbor re summary); ghidra_check: confirm the shared Ghidra server serves our image (address range, byte samples, SHA-256 when the imported file is readable) (harbor re ghidra-check)

This commit is contained in:
KeelrLeek committed 2026-10-09 09:51:09 -05:00
1 parent 57fb31f2d6
commit 21b7fedb6f
4 files changed
+366 -1

No files matched your search

+16 -1
View File
@@ -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)
+21
View File
@@ -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',
+230
View File
@@ -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 (<ADDR>_<Name>.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/**/<ADDR>_<Name>.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("<I", addr)
out = []
for lo, hi in tables:
pos = data.find(needle, max(0, lo - self.base), hi - self.base)
while pos >= 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("<I", data, start - 4)[0]) and pos - start < 4 * 400:
start -= 4
table = start + self.base
slot = (pos - start) // 4
name = self.name_of(table)
kind = "vtable" if table in refs or table in mac or (name and "vtable" in name.lower()) \
else "pointer table"
out.append({"table": f"0x{table:X}", "slot": slot, "offset": f"0x{slot * 4:X}", "kind": kind,
**({"table_name": name} if name else {}),
**({"mac_class": mac[table]} if table in mac else {})})
pos = data.find(needle, pos + 1, hi - self.base)
return out
def summary(self, target):
"""Everything a worker needs before starting on a function, in one call: name, prototype, size, convention,
whether it is readable on main, callers and callees (named, with their status on main), globals and strings,
the vtable slots it fills, and the Mac/Xbox name hints."""
addr = self.resolve(target)
func = self.function_at(addr)
if not func:
raise HarborError(f"0x{addr:X} is not inside a known function")
start, size, name = func
d = self._facts()[start]
def fn(a):
status = self.status_of(a)
return {"address": f"0x{a:X}", "name": self.name_of(a), "status": status or "no file",
"readable": status == "readable"}
out = {"address": f"0x{start:X}", "name": name, "size": size, "end": f"0x{start + size:X}"}
if addr != start:
out["asked"] = f"0x{addr:X} (+0x{addr - start:X})"
proto = self._prototypes().get(name) if name else None
out["prototype"] = proto
conv = self.convention(start)
out["convention"] = conv["summary"] if conv else None
if conv:
out["registers_in"], out["ret_pops"] = conv["registers_in"], conv["ret_pops"]
status = self.status_of(start)
out["status"] = status or "no file"
out["readable"] = status == "readable"
path = self._src_index().get(start)
if path and self.src:
out["file"] = str(path.relative_to(self.src.parent)).replace(os.sep, "/")
row = self._csv("names.csv").get(start)
if row and len(row) > 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
+99
View File
@@ -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("<II", 0x401000, 0x401040) # 0x4010C0: a two-slot vtable
(self.d / "image.bin").write_bytes(bytes(image))
(self.d / "sections.json").write_text(json.dumps([
{"name": ".text", "start": 0x401000, "end": 0x401080}, {"name": ".rdata", "start": 0x4010B0,
"end": 0x401100}]), encoding="utf-8")
facts = [json.loads(l) for l in (self.d / "facts.jsonl").read_text(encoding="utf-8").splitlines()]
facts[1]["data"] = ["0x4010C0"] # the constructor installs it
facts[0]["strings"] = [".cue"]
(self.d / "facts.jsonl").write_text("\n".join(json.dumps(f) for f in facts), encoding="utf-8")
src = self.d / "src" / "game"
src.mkdir(parents=True)
(src / "401040_Helper.c").write_text("/* Helper\n * status: readable\n */\nvoid Helper(void) {}\n",
encoding="utf-8")
(self.d / "mac.csv").write_text("pc_address,demangled,confidence\n0x00401000,Level::Get(),high\n",
encoding="utf-8")
r = Reverse({"image": str(self.d / "image.bin"), "base": "0x401000", "facts": str(self.d / "facts.jsonl"),
"annotations": str(self.d / "annotations"), "src": str(self.d / "src"),
"mac_names": str(self.d / "mac.csv")})
s = r.summary("GetLevel")
self.assertEqual((s["name"], s["size"], s["status"], s["readable"]), ("GetLevel", 18, "no file", False))
self.assertIn("plain ret", s["prototype"])
self.assertEqual(s["callees"], [{"address": "0x401040", "name": "Helper", "status": "readable",
"readable": True}])
self.assertEqual(s["globals"], [{"address": "0x401080", "name": "g_root", "type": "int"}])
self.assertEqual(s["strings"], [".cue"])
self.assertEqual(s["vtable_slots"], [{"table": "0x4010C0", "slot": 0, "offset": "0x0", "kind": "vtable"}])
self.assertEqual(s["hints"]["mac_name"]["demangled"], "Level::Get()")
h = r.summary("0x401040")
self.assertTrue(h["readable"])
self.assertEqual([c["name"] for c in h["callers"]], ["GetLevel"])
self.assertEqual(h["vtable_slots"][0]["slot"], 1)
from harbor.reverse import format_summary
self.assertIn("vtable 0x4010C0 slot 1 (+0x4)", format_summary(h))
class GhidraCheck(unittest.TestCase):
"""ghidra_check compares the served program with the project image: range, SHA-256 of the file, byte samples."""
def serve(self, lo, hi, path, image):
import http.server
import threading
class Fake(http.server.BaseHTTPRequestHandler):
def log_message(self, *a):
pass
def do_POST(self):
body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
if "id" not in body:
self.send_response(202)
self.end_headers()
return
params = body.get("params", {})
name, args = params.get("name"), params.get("arguments", {})
text = ""
if name == "get_binary_info":
text = f"Program Information:\nName: image.bin\nImage Base: 00000000\nMin Address: {lo:08x}\n" \
f"Max Address: {hi:08x}\nFunction Count: 2\n"
elif name == "list_binaries":
text = f"Open Programs in Ghidra:\n\n1. image.bin [ACTIVE]\n Executable Path: /{path}\n"
elif name == "get_data_at":
a, n = int(args["address"], 16), int(args["len"])
chunk = image[a - 0x401000:a - 0x401000 + n]
rows = [f"{a + i:x} " + " ".join(f"{b:02x}" for b in chunk[i:i + 16]) + " |....|"
for i in range(0, len(chunk), 16)]
text = f"Hexdump at {a:08x} ({n} bytes):\n\n" + "\n".join(rows) + "\n"
result = {"content": [{"type": "text", "text": text}]} if body["method"] == "tools/call" else {}
data = json.dumps({"jsonrpc": "2.0", "id": body["id"], "result": result}).encode()
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(data)))
self.end_headers()
self.wfile.write(data)
server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Fake)
threading.Thread(target=server.serve_forever, daemon=True).start()
self.addCleanup(server.server_close)
self.addCleanup(server.shutdown)
return f"http://127.0.0.1:{server.server_port}/mcp"
def test_match_and_mismatch(self):
from harbor.reverse import ghidra_check
d = temp_dir()
self.addCleanup(shutil.rmtree, d, True)
image = bytes(range(256)) * 4
(d / "image.bin").write_bytes(image)
config = {"image": str(d / "image.bin"), "base": "0x401000"}
good = ghidra_check(config | {"ghidra_url": self.serve(0x401000, 0x4013FF, (d / "image.bin").as_posix(), image)})
self.assertTrue(good["ok"], good)
self.assertEqual([c["ok"] for c in good["checks"]], [True, True, True])
other = d / "other.bin"
other.write_bytes(image[:-1] + b"\x00")
bad = ghidra_check(config | {"ghidra_url": self.serve(0x401000, 0x4013FE, other.as_posix(),
image[:512] + b"\0" * 512)})
self.assertFalse(bad["ok"])
self.assertEqual([c["ok"] for c in bad["checks"]], [False, False, False])
class GhidraForwarder(unittest.TestCase):
"""The ghidra() tool against a fake streamable-HTTP MCP server: a session, a polled async task, a refusal."""