KeelrLeek
3f6df88f95
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.
279 lines
12 KiB
Python
279 lines
12 KiB
Python
"""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 (<ADDR>_<Name>.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("<f" if size == 4 else "<d", raw)[0]
|
|
notes.append(f"= {value!r}{'f' if size == 4 else ''}")
|
|
except HarborError:
|
|
pass
|
|
for m in re.finditer(r"0x([0-9a-f]{6,8})", op_str):
|
|
value = int(m.group(1), 16)
|
|
if lo <= value < hi and not mnemonic.startswith("call"):
|
|
continue # a jump inside this function needs no name
|
|
name = self.name_of(value)
|
|
if name:
|
|
notes.append(f"0x{value:X} {name}")
|
|
elif mnemonic.startswith("call") or mnemonic.startswith("j"):
|
|
func = self.function_at(value)
|
|
if func and func[0] != value:
|
|
notes.append(f"0x{value:X} inside {func[2] or hex(func[0])}+0x{value - func[0]:X}")
|
|
return ", ".join(notes)
|
|
|
|
def xrefs(self, target):
|
|
"""Callers and callees of a function, and the data it touches, named."""
|
|
addr = self.resolve(target)
|
|
func = self.function_at(addr)
|
|
if not func:
|
|
raise HarborError(f"0x{addr:X} is not inside a known function")
|
|
d = self._facts()[func[0]]
|
|
named = lambda a: {"address": f"0x{a:X}", "name": self.name_of(a)}
|
|
return {"function": named(func[0]) | {"size": func[1]},
|
|
"callers": [named(a) for a in sorted(set(self._callers().get(func[0], [])))],
|
|
"calls": [named(_int(a)) for a in d.get("calls", [])],
|
|
"data": [named(_int(a)) for a in d.get("data", [])],
|
|
"imports": d.get("imports", []), "strings": d.get("strings", [])}
|
|
|
|
def bytes_at(self, target, count=16, as_type="hex"):
|
|
"""Raw bytes at an address, also read as hex, int32, uint32, float, double or a C string."""
|
|
addr = self.resolve(target)
|
|
count = max(1, min(int(count), 4096))
|
|
raw = self.read(addr, count if as_type != "string" else 4096)
|
|
if as_type == "string":
|
|
text = raw.split(b"\0", 1)[0]
|
|
return {"address": f"0x{addr:X}", "string": text.decode("latin-1"), "length": len(text)}
|
|
if as_type in ("int32", "uint32", "float"):
|
|
fmt = {"int32": "i", "uint32": "I", "float": "f"}[as_type]
|
|
values = list(struct.unpack_from(f"<{len(raw) // 4}{fmt}", raw))
|
|
return {"address": f"0x{addr:X}", as_type: [repr(v) if as_type == "float" else v for v in values],
|
|
"hex": raw[:len(values) * 4].hex(" ")}
|
|
if as_type == "double":
|
|
values = list(struct.unpack_from(f"<{len(raw) // 8}d", raw))
|
|
return {"address": f"0x{addr:X}", "double": [repr(v) for v in values], "hex": raw[:len(values) * 8].hex(" ")}
|
|
return {"address": f"0x{addr:X}", "hex": raw.hex(" "), "name": self.name_of(addr)}
|
|
|
|
def lookup(self, target):
|
|
"""Everything the team has written about an address or name: names.csv, functions.csv and globals.csv rows,
|
|
the prototype, the decompiled file and its status."""
|
|
addr = self.resolve(target)
|
|
out = {"address": f"0x{addr:X}"}
|
|
func = self.function_at(addr)
|
|
if func:
|
|
out["function"] = {"start": f"0x{func[0]:X}", "size": func[1], "name": func[2]}
|
|
for table, key in (("names.csv", "names_row"), ("functions.csv", "functions_row"), ("globals.csv", "globals_row")):
|
|
row = self._csv(table).get(addr)
|
|
if row:
|
|
out[key] = ",".join(row) if len(",".join(row)) < 4000 else ",".join(row)[:4000] + "..."
|
|
name = self.name_of(addr)
|
|
if name and name in self._prototypes():
|
|
out["prototype"] = self._prototypes()[name]
|
|
if self.src and self.src.exists() and func:
|
|
prefix = f"{func[0]:06X}_"
|
|
for path in self.src.rglob(f"{prefix}*.c"):
|
|
head = path.read_text(encoding="utf-8", errors="replace").split("\n")[:6]
|
|
status = next((l.split("status:", 1)[1].strip() for l in head if "status:" in l), "?")
|
|
out["file"] = {"path": str(path.relative_to(self.src.parent)).replace(os.sep, "/"), "status": status}
|
|
break
|
|
return out
|