From 3f6df88f954a3f08d5326167f9e14812f495224a Mon Sep 17 00:00:00 2001 From: KeelrLeek <156356974+KeelrLeek@users.noreply.github.com> Date: Mon, 5 Oct 2026 11:16:56 -0500 Subject: [PATCH] Add shared reverse-engineering lookups (re_disassemble, re_xrefs, re_bytes, re_lookup) Read the project's flat image, facts.jsonl and annotations with the team's current names; exposed as MCP tools and as 'harbor re'. Configured by a [reverse] table in the project file. --- harbor/cli.py | 21 ++++ harbor/mcp_server.py | 50 +++++++- harbor/projects.py | 4 +- harbor/reverse.py | 278 ++++++++++++++++++++++++++++++++++++++++++ tests/test_reverse.py | 59 +++++++++ 5 files changed, 410 insertions(+), 2 deletions(-) create mode 100644 harbor/reverse.py create mode 100644 tests/test_reverse.py diff --git a/harbor/cli.py b/harbor/cli.py index 1408831..310840d 100644 --- a/harbor/cli.py +++ b/harbor/cli.py @@ -254,6 +254,12 @@ def build_parser(): p.add_argument("--file") 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"]) + p.add_argument("target", help="an address (0x4AD850) or a name") + p.add_argument("--max-bytes", type=int, dest="max_bytes") + p.add_argument("--count", type=int, default=16) + p.add_argument("--as", dest="as_type", default="hex", choices=["hex", "int32", "uint32", "float", "double", "string"]) p = sub.add_parser("feed", help="the latest tool calls of the team (or one agent), from the hooks") p.add_argument("--agent", dest="filter_agent") p.add_argument("--limit", type=int, default=30) @@ -666,6 +672,21 @@ def main(argv=None): text = Path(args.file).read_text(encoding="utf-8") if args.file else " ".join(args.text) r = harbor.summarize(who(), args.thread, text) out(r, f"summary #{r['id']} pinned to {r['thread']}") + elif cmd == "re": + from .reverse import Reverse + r = Reverse(harbor.project.reverse) + if args.action == "disasm": + text = r.disassemble(args.target, args.max_bytes) + out({"disassembly": text}, text) + elif args.action == "xrefs": + res = r.xrefs(args.target) + out(res, json.dumps(res, indent=1)) + elif args.action == "bytes": + res = r.bytes_at(args.target, args.count, args.as_type) + out(res, json.dumps(res, indent=1)) + else: + res = r.lookup(args.target) + out(res, json.dumps(res, indent=1)) elif cmd == "who": rows = harbor.experts(args.scope) out(rows, "\n".join(f"{e['agent']:<9} " + ", ".join(f"{v} {k}" for k, v in e["detail"].items()) for e in rows) diff --git a/harbor/mcp_server.py b/harbor/mcp_server.py index e4c0e4d..479aad7 100644 --- a/harbor/mcp_server.py +++ b/harbor/mcp_server.py @@ -33,7 +33,9 @@ review queue, presence and stop flags. Protocol: 7. Put finished work in the review queue with submit(); do not announce commits in chat. A coordinator lands it. 8. When you are waiting on someone, use wait_for_messages() instead of polling or sleeping. 9. A correction is send(..., corrects=). -10. Before your session ends or your allowance runs out, write handoff(): finished, in flight, next.""" +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.""" KindName = Literal["note", "question", "answer", "request", "finding", "retraction", "decision", "status", "handoff"] Importance = Literal["normal", "high", "urgent"] @@ -436,6 +438,52 @@ def resume(target: str, relay_of: int | None = None, reason: str | None = None) return run(do) +_reverse = {} + + +def _re(fn): + """Run a reverse-engineering lookup against the project's [reverse] files (read-only, no Harbor lock needed).""" + try: + h = harbor() + r = _reverse.get(h.project.name) + if r is None: + from .reverse import Reverse + r = _reverse[h.project.name] = Reverse(h.project.reverse) + return fn(r) + except (HarborError, projects.ProjectError) as error: + raise ToolError(str(error)) from None + + +@mcp.tool +def re_disassemble(target: str, max_bytes: int | None = None) -> str: + """Disassemble the whole function at an address or name (or max_bytes from the address), from the project's own + image of the executable, with call targets, imports and known globals named from the team's annotations. Use + this instead of writing your own disassembler script.""" + return _re(lambda r: r.disassemble(target, max_bytes)) + + +@mcp.tool +def re_xrefs(target: str) -> dict: + """Callers and callees of the function at an address or name, plus the globals, imports and strings it uses, + each with its current name.""" + return _re(lambda r: r.xrefs(target)) + + +@mcp.tool +def re_bytes(target: str, count: int = 16, as_type: Literal["hex", "int32", "uint32", "float", "double", + "string"] = "hex") -> dict: + """Bytes at an address or name, read as hex, int32, uint32, float, double or a C string (constants, tables, + initial values of globals).""" + return _re(lambda r: r.bytes_at(target, count, as_type)) + + +@mcp.tool +def re_lookup(target: str) -> dict: + """What the team has recorded about an address or name: the names.csv, functions.csv and globals.csv rows, the + prototype, the containing function, and the decompiled file with its status.""" + return _re(lambda r: r.lookup(target)) + + def _report_in(): """Tell Harbor this agent's app has its tools running: at start, then every 5 minutes while the app keeps the server open (updates the connection only, not the agent's last-seen time).""" diff --git a/harbor/projects.py b/harbor/projects.py index 80fb24d..bea984a 100644 --- a/harbor/projects.py +++ b/harbor/projects.py @@ -94,6 +94,7 @@ class Project: policy: dict = field(default_factory=dict) autopilot: bool = False # running agents keep working instead of ending their turns (hooks) shared_files: tuple = () # glob patterns (repo-relative) every agent edits and the integrator merges: not collisions + reverse: dict = field(default_factory=dict) # [reverse]: image, base, facts, imports, annotations, src (harbor.reverse) @property def names(self): @@ -207,7 +208,8 @@ def project_from_dict(data, source=None, home=None): source=source, git_main=str(data.get("git", {}).get("main", "main")), templates=_templates(data.get("templates", {}), names), policy=_policy(data.get("policy"), names), autopilot=bool(data.get("autopilot", False)), - shared_files=tuple(str(p).replace("\\", "/") for p in data.get("shared_files", []))) + shared_files=tuple(str(p).replace("\\", "/") for p in data.get("shared_files", [])), + reverse=dict(data.get("reverse", {}))) def load_file(path, home=None): diff --git a/harbor/reverse.py b/harbor/reverse.py new file mode 100644 index 0000000..ef91bf6 --- /dev/null +++ b/harbor/reverse.py @@ -0,0 +1,278 @@ +"""Shared reverse-engineering lookups for the agents, backed by the project's own image and annotations. + +Configured per project in a [reverse] table: + image = 'R:/.../image.bin' flat image of the executable, the first byte at `base` + base = 0x401000 + facts = 'R:/.../facts.jsonl' one JSON object per function: address, size, name, calls, data, ... + 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) + +Every answer comes from those files, read fresh when they change, so the agents see the team's current names. +Nothing here writes anything.""" +import csv +import json +import os +import re +import struct +from pathlib import Path + +from .util import HarborError + +MAX_BYTES = 0x4000 + + +def _int(value): + if isinstance(value, int): + return value + text = str(value).strip() + return int(text, 16) if text.lower().startswith("0x") or re.fullmatch(r"[0-9A-Fa-f]{6,8}", text) else int(text) + + +class Reverse: + def __init__(self, config): + if not config or not config.get("image"): + raise HarborError("this project has no [reverse] table (image, base, facts, annotations)") + self.image = Path(config["image"]) + self.base = _int(config.get("base", 0x401000)) + self.facts_path = Path(config["facts"]) if config.get("facts") else None + self.imports_path = Path(config["imports"]) if config.get("imports") else None + self.annotations = Path(config["annotations"]) if config.get("annotations") else None + self.src = Path(config["src"]) if config.get("src") else None + self._cache = {} + + # ----------------------------------------------------------------- loading (cached by file mtime) + def _load(self, key, path, loader): + if path is None or not path.exists(): + return loader(None) + stamp = path.stat().st_mtime + hit = self._cache.get(key) + if hit and hit[0] == stamp: + return hit[1] + value = loader(path) + self._cache[key] = (stamp, value) + return value + + def _bytes(self): + return self._load("image", self.image, lambda p: p.read_bytes() if p else b"") + + def _facts(self): + def load(p): + out = {} + if p: + for line in p.open(encoding="utf-8"): + if line.strip(): + d = json.loads(line) + out[_int(d["address"])] = d + return out + return self._load("facts", self.facts_path, load) + + def _callers(self): + facts = self._facts() + key = ("callers", id(facts)) + if key not in self._cache: + callers = {} + for addr, d in facts.items(): + for c in d.get("calls", []): + callers.setdefault(_int(c), []).append(addr) + self._cache[key] = callers + return self._cache[key] + + def _imports(self): + return self._load("imports", self.imports_path, lambda p: { + int(d["slot"]): f"{d['module']}!{d['name']}" for d in json.load(p.open(encoding="utf-8"))} if p else {}) + + def _csv(self, name): + path = self.annotations / name if self.annotations else None + + def load(p): + rows = {} + if p: + for row in csv.reader(p.open(encoding="utf-8")): + if row and row[0].lower().startswith("0x"): + rows[_int(row[0])] = row + return rows + return self._load(name, path, load) + + def _prototypes(self): + path = self.annotations / "prototypes.h" if self.annotations else None + + def load(p): + out = {} + if p: + lines = p.read_text(encoding="utf-8").split("\n") + i = 0 + while i < len(lines): + line = lines[i] + m = re.match(r"^[A-Za-z_][^(;]*?\b([A-Za-z_]\w*)\s*\(", line) if not line.startswith( + ("typedef", "#", "/", " ")) else None + j = i + 1 + while m and j < len(lines) and lines[j].startswith(" ") and not re.match(r"^[A-Za-z_]", lines[j]): + j += 1 + if m: + out[m.group(1)] = "\n".join(lines[i:j]) + i = j if m else i + 1 + return out + return self._load("prototypes", path, load) + + # ----------------------------------------------------------------- names + def function_at(self, addr): + """The function containing addr: (start, size, name), from facts.jsonl, named from names.csv.""" + facts = self._facts() + if addr in facts: + start = addr + else: + starts = [a for a in facts if a <= addr < a + int(facts[a].get("size") or 0)] + if not starts: + return None + start = max(starts) + d = facts[start] + return start, int(d.get("size") or 0), self.name_of(start) or d.get("name") + + def name_of(self, addr): + row = self._csv("names.csv").get(addr) + if row and len(row) > 1 and row[1]: + return row[1] + g = self._csv("globals.csv").get(addr) + if g and len(g) > 1 and g[1]: + return g[1] + imp = self._imports().get(addr) + if imp: + return imp + d = self._facts().get(addr) + return d.get("name") if d else None + + def address_of(self, name): + for table in ("names.csv", "globals.csv"): + for addr, row in self._csv(table).items(): + if len(row) > 1 and row[1] == name: + return addr + for addr, d in self._facts().items(): + if d.get("name") == name: + return addr + return None + + def resolve(self, target): + """'0x4AD850', '4AD850', or a name -> address.""" + text = str(target).strip() + if re.fullmatch(r"(0x)?[0-9A-Fa-f]{5,8}", text): + return int(text, 16) + addr = self.address_of(text) + if addr is None: + raise HarborError(f"{target!r} is neither an address nor a known name") + return addr + + # ----------------------------------------------------------------- tools + def read(self, addr, count): + data = self._bytes() + offset = addr - self.base + if offset < 0 or offset >= len(data): + raise HarborError(f"0x{addr:X} is outside the image (0x{self.base:X}..0x{self.base + len(data):X})") + return data[offset:offset + count] + + def disassemble(self, target, max_bytes=None): + """The whole function (or max_bytes from an address) with call targets, jumps into other functions, + import slots and known globals named.""" + try: + import capstone + except ImportError: + raise HarborError("capstone is not installed for this Python (py -3.12 -m pip install capstone)") from None + addr = self.resolve(target) + func = self.function_at(addr) + if max_bytes: + start, size = addr, min(int(max_bytes), MAX_BYTES) + elif func: + start, size = func[0], func[1] + else: + start, size = addr, 0x100 + if size > MAX_BYTES: + raise HarborError(f"the function is {size} bytes; pass max_bytes (at most {MAX_BYTES}) and walk it in parts") + code = self.read(start, size) + md = capstone.Cs(capstone.CS_ARCH_X86, capstone.CS_MODE_32) + lines = [] + if func and not max_bytes: + lines.append(f"; {func[2] or 'unnamed'} 0x{func[0]:X}..0x{func[0] + func[1]:X} ({func[1]} bytes)") + for ins in md.disasm(code, start): + note = self._annotate(ins.op_str, ins.mnemonic, start, start + size) + lines.append(f"{ins.address:08X} {ins.mnemonic:7} {ins.op_str}{(' ; ' + note) if note else ''}") + return "\n".join(lines) + + def _annotate(self, op_str, mnemonic, lo, hi): + notes = [] + constant = re.search(r"(dword|qword) ptr \[0x([0-9a-f]{6,8})\]", op_str) if mnemonic.startswith("f") else None + if constant: # an x87 constant: show its value as the FPU reads it + size = 4 if constant.group(1) == "dword" else 8 + try: + raw = self.read(int(constant.group(2), 16), size) + value = struct.unpack("