From 07a031109b614c5a9338caeda5e50a78fcd05c1d Mon Sep 17 00:00:00 2001 From: KeelrLeek <156356974+KeelrLeek@users.noreply.github.com> Date: Mon, 5 Oct 2026 13:45:16 -0500 Subject: [PATCH] re_similar, re_struct_audit; next_functions favours raw functions with a readable near-duplicate --- harbor/cli.py | 11 ++++++++--- harbor/mcp_server.py | 27 ++++++++++++++++++++++++++- harbor/queue.py | 9 +++++++++ harbor/reverse.py | 22 ++++++++++++++++++++++ 4 files changed, 65 insertions(+), 4 deletions(-) diff --git a/harbor/cli.py b/harbor/cli.py index 1c60724..ba4e093 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"]) + p.add_argument("action", choices=["disasm", "xrefs", "bytes", "lookup", "ghidra", "ghidra-tools", "debugger-tools", "match", "similar", "struct-audit"]) 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") @@ -684,7 +684,12 @@ def main(argv=None): elif cmd == "re": from .reverse import Ghidra, Reverse r = None if args.action.startswith(("ghidra", "debugger", "match")) else Reverse(harbor.project.reverse) - if args.action == "match": + if args.action in ("similar", "struct-audit"): + from .reverse import run_toolkit + res = (run_toolkit(harbor.project.reverse, "similar", [args.target]) if args.action == "similar" else + run_toolkit(harbor.project.reverse, "struct_audit", ["--struct", args.target], json_to_file=True)) + out(res, json.dumps(res, indent=1)) + elif args.action == "match": from .reverse import match_functions res = match_functions(harbor.project.reverse, args.target, args.decomp) out(res, res["summary"] + "\n" + "\n".join( @@ -717,7 +722,7 @@ def main(argv=None): from .queue import FunctionQueue q = FunctionQueue(harbor) if cmd == "next-functions": - rows = q.next_functions(who(), args.n) + rows = q.next_functions(who(), args.n, similar_readable=q.similar_readable()) out(rows, "\n".join(f"{r['address']:<10} {r['score']:<7} {(r['name'] or r['file'] or ''):<34} {r['why']}" for r in rows)) elif cmd == "defer": diff --git a/harbor/mcp_server.py b/harbor/mcp_server.py index 75fff7d..30e06e9 100644 --- a/harbor/mcp_server.py +++ b/harbor/mcp_server.py @@ -518,6 +518,30 @@ def re_match(targets: str, decomp: str | None = None) -> dict: raise ToolError(str(error)) from None +@mcp.tool() +def re_similar(target: str, k: int = 10, min_score: float = 0.7) -> dict: + """Functions whose instructions look like this one (normalized: addresses blanked, registers and small + constants kept), best first, with each one's status. A readable near-duplicate is a head start: read it + first. 0.85+ is usually the same code (template copies); 0.75-0.85 the same idiom; check before reusing.""" + try: + from .reverse import run_toolkit + return run_toolkit(harbor().project.reverse, "similar", + [str(target), "--k", str(max(1, min(int(k), 50))), "--min", str(float(min_score))]) + except (HarborError, projects.ProjectError) as error: + raise ToolError(str(error)) from None + + +@mcp.tool() +def re_struct_audit(struct: str) -> list: + """Check a struct's field types against how the game's code accesses it through its globals: width, + signedness (movsx/movzx), float vs int, and unnamed bytes the code uses. Use it before changing types.h.""" + try: + from .reverse import run_toolkit + return run_toolkit(harbor().project.reverse, "struct_audit", ["--struct", str(struct)], json_to_file=True) + except (HarborError, projects.ProjectError) as error: + raise ToolError(str(error)) from None + + def _queue(): from .queue import FunctionQueue return FunctionQueue(harbor()) @@ -529,7 +553,8 @@ def next_functions(n: int = 10) -> list: by someone else, rejected fewer than the attempt cap. Each row says why it ranks where it does (size, how many callees already have names, the convention from the bytes, earlier rejections).""" try: - return _queue().next_functions(me(), max(1, min(int(n), 50))) + q = _queue() + return q.next_functions(me(), max(1, min(int(n), 50)), similar_readable=q.similar_readable()) except (HarborError, projects.ProjectError) as error: raise ToolError(str(error)) from None diff --git a/harbor/queue.py b/harbor/queue.py index 7dbe3cb..a3de3b8 100644 --- a/harbor/queue.py +++ b/harbor/queue.py @@ -103,6 +103,15 @@ class FunctionQueue: return {int(row["address"], 16): row for row in csv.DictReader(f)} return {int(row["address"], 16): row for row in json.loads(Path(path).read_text(encoding="utf-8"))} + def similar_readable(self): + """Raw functions with a readable near-duplicate (from the [reverse] similar_clusters file).""" + path = self.config.get("similar_clusters") + if not path or not Path(path).exists(): + return frozenset() + data = json.loads(Path(path).read_text(encoding="utf-8")) + return frozenset(int(m["address"], 16) for c in data.get("clusters", []) for m in c.get("members", []) + if m.get("best_readable") and m.get("status") != "readable") + # ------------------------------------------------------------------ queue def score(self, addr, facts, named, rejected, similar_readable=frozenset()): d = facts[addr] diff --git a/harbor/reverse.py b/harbor/reverse.py index a343507..b687361 100644 --- a/harbor/reverse.py +++ b/harbor/reverse.py @@ -509,3 +509,25 @@ def match_functions(config, targets, decomp=None, flags=None): errors = [line for line in result.stdout.splitlines() if "compile errors" in line] return {"results": rows, "summary": result.stdout.strip().splitlines()[0] if result.stdout.strip() else "", **({"compile": errors[0]} if errors else {})} + + +def run_toolkit(config, key, argv, json_to_file=False, timeout=900): + """Run the toolkit script configured as [reverse] with argv and return its JSON (stdout, or a temp file + passed as `--json ` when json_to_file).""" + import subprocess + import sys + import tempfile + script = (config or {}).get(key) + if not script or not Path(script).exists(): + raise HarborError(f"this project has no {key} script in its [reverse] table") + with tempfile.TemporaryDirectory(dir=(config or {}).get("debugger_temp")) as tmp: + out = Path(tmp) / "out.json" + command = [sys.executable, "-B", str(script), *argv] + (["--json", str(out)] if json_to_file else ["--json"]) + result = subprocess.run(command, capture_output=True, text=True, encoding="utf-8", errors="replace", + timeout=timeout, cwd=str(Path(script).parent), + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0)) + text = out.read_text(encoding="utf-8") if json_to_file and out.exists() else result.stdout + try: + return json.loads(text) + except ValueError: + raise HarborError(f"{key} failed: " + (result.stdout + result.stderr).strip()[-800:]) from None