201 lines
8.4 KiB
Python
201 lines
8.4 KiB
Python
"""The web board's Decomp view: progress tiers, a function inspector, each agent's next functions, deferrals and the
|
|
state of the reverse-engineering tools. Everything here reads; nothing writes.
|
|
|
|
It runs on its own Harbor connection with its own lock and short caches, because the answers take seconds (they scan
|
|
src/ and score whole bands) and must not hold up the chat."""
|
|
import json
|
|
import socket
|
|
import subprocess
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
|
|
from .core import Harbor
|
|
from .queue import FunctionQueue
|
|
from .reverse import run_toolkit
|
|
from .util import HarborError
|
|
|
|
PROGRESS_TTL = 120
|
|
QUEUE_TTL = 600
|
|
|
|
|
|
class DecompAPI:
|
|
def __init__(self, project):
|
|
self.project = project
|
|
self.lock = threading.Lock()
|
|
self.cache = {}
|
|
self._local = threading.local()
|
|
|
|
@property
|
|
def h(self):
|
|
# SQLite connections belong to the thread that opened them, and the server answers on many threads
|
|
if getattr(self._local, "h", None) is None:
|
|
self._local.h = Harbor(self.project)
|
|
return self._local.h
|
|
|
|
def _cached(self, key, ttl, fn):
|
|
hit = self.cache.get(key)
|
|
if hit and time.monotonic() - hit[0] < ttl:
|
|
return hit[1]
|
|
value = fn()
|
|
self.cache[key] = (time.monotonic(), value)
|
|
return value
|
|
|
|
def _queue(self):
|
|
return FunctionQueue(self.h)
|
|
|
|
# ------------------------------------------------------------------ answers
|
|
def progress(self):
|
|
with self.lock:
|
|
return self._cached("progress", PROGRESS_TTL, lambda: self._queue().progress())
|
|
|
|
def queues(self, n=6):
|
|
def build():
|
|
q = self._queue()
|
|
similar = q.similar_readable()
|
|
out = []
|
|
for agent in self.project.ai_names:
|
|
if not q.bands(agent):
|
|
continue
|
|
try:
|
|
rows = q.next_functions(agent, n, similar_readable=similar)
|
|
except HarborError as error:
|
|
rows, note = [], str(error)
|
|
else:
|
|
note = None
|
|
out.append({"agent": agent, "bands": [[f"0x{lo:X}", f"0x{hi:X}"] for lo, hi in q.bands(agent)],
|
|
"next": rows, **({"note": note} if note else {})})
|
|
return out
|
|
with self.lock:
|
|
return self._cached("queues", QUEUE_TTL, build)
|
|
|
|
def deferred(self):
|
|
with self.lock:
|
|
q = self._queue()
|
|
return [{"address": f"0x{a:X}", "name": q.r.name_of(a), "reason": reason, "evidence": evidence,
|
|
"author": author} for a, (reason, evidence, author) in sorted(q.deferrals().items())]
|
|
|
|
def function(self, target):
|
|
"""Everything about one function, for the inspector."""
|
|
with self.lock:
|
|
q = self._queue()
|
|
r = q.r
|
|
info = r.lookup(target)
|
|
addr = int(info.get("function", {}).get("start", info["address"]), 16)
|
|
statuses = self._cached("statuses", PROGRESS_TTL, q.statuses)
|
|
status, filename = statuses.get(addr, ("none", None))
|
|
info["status"] = status
|
|
if filename:
|
|
info["src_file"] = filename
|
|
for key, label in (("match_results", "match"), ("diffrun_results", "behaviour")):
|
|
row = q._results(key).get(addr)
|
|
if row:
|
|
info[label] = {k: row[k] for k in ("status", "verdict", "similarity", "first_difference",
|
|
"registers_in", "ours_registers_in", "reason", "weak")
|
|
if k in row}
|
|
deferral = q.deferrals().get(addr)
|
|
if deferral:
|
|
info["deferred"] = {"reason": deferral[0], "evidence": deferral[1], "by": deferral[2]}
|
|
info["category"] = q._category(addr)
|
|
library = q._results("library_results")
|
|
if q.is_library(addr, library, q.library_ranges()):
|
|
info["library"] = True
|
|
try:
|
|
info["disassembly"] = r.disassemble(f"0x{addr:X}") if int(info.get("function", {}).get("size", 0)) <= 0x1800 \
|
|
else r.disassemble(f"0x{addr:X}", 0x600) + "\n; ... (first 0x600 bytes of a large function)"
|
|
except HarborError as error:
|
|
info["disassembly"] = f"; {error}"
|
|
try:
|
|
similar = run_toolkit(self.project.reverse, "similar", [f"0x{addr:X}", "--k", "6"], timeout=60)
|
|
info["similar"] = similar.get("results", [])
|
|
except (HarborError, subprocess.TimeoutExpired, OSError):
|
|
info["similar"] = []
|
|
return info
|
|
|
|
def tools(self):
|
|
"""Is each reverse-engineering tool in place, and how fresh are its results?"""
|
|
config = self.project.reverse or {}
|
|
out = []
|
|
|
|
def fresh(path):
|
|
p = Path(path)
|
|
if not p.exists():
|
|
return None
|
|
return round((time.time() - p.stat().st_mtime) / 60)
|
|
|
|
url = config.get("ghidra_url", "")
|
|
port = int(url.rsplit(":", 1)[1].split("/")[0]) if url.count(":") >= 2 else 0
|
|
up = False
|
|
if port:
|
|
try:
|
|
with socket.create_connection(("127.0.0.1", port), timeout=1):
|
|
up = True
|
|
except OSError:
|
|
up = False
|
|
decomp = Path(config.get("annotations", "")).parent if config.get("annotations") else None
|
|
head = applied = None
|
|
if decomp and (decomp / ".git").exists():
|
|
try:
|
|
head = subprocess.run(["git", "-C", str(decomp), "rev-parse", "HEAD"], capture_output=True, text=True,
|
|
timeout=10).stdout.strip() or None
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
head = None
|
|
applied_file = Path(config.get("debugger_temp", "")).parent / "ghidra-mcp" / "applied-commit.txt" \
|
|
if config.get("debugger_temp") else None
|
|
if applied_file and applied_file.exists():
|
|
applied = applied_file.read_text(encoding="utf-8").strip()
|
|
out.append({"tool": "Shared Ghidra session", "ok": up and (not head or head == applied),
|
|
"detail": ("up" if up else "down") + (f"; names from main {applied[:8]}" if applied else "")
|
|
+ ("" if not head or head == applied else f"; main is at {head[:8]}: run ghidra-serve.ps1 -IfStale")})
|
|
compiler = Path(r"C:\Program Files (x86)\Microsoft Visual C++ Toolkit 2003\bin\cl.exe")
|
|
out.append({"tool": "Game compiler (Visual C++ 2003 Toolkit)", "ok": compiler.exists(),
|
|
"detail": "cl 13.10.3052 installed" if compiler.exists() else "not found"})
|
|
if config.get("debugger_path"):
|
|
out.append({"tool": "Debugger (x32dbg + automation plugin)", "ok": Path(config["debugger_path"]).exists(),
|
|
"detail": config["debugger_path"]})
|
|
for key, label in (("match_results", "Whole-program match results"), ("diffrun_results", "Emulator results"),
|
|
("library_results", "Library sweep"), ("similar_clusters", "Near-duplicate clusters"),
|
|
("source_files", "Source files from the Xbox build"), ("mac_names", "Names from the Mac build")):
|
|
if config.get(key):
|
|
age = fresh(config[key])
|
|
out.append({"tool": label, "ok": age is not None,
|
|
"detail": "missing" if age is None else f"updated {age} min ago"})
|
|
return out
|
|
|
|
|
|
_APIS = {}
|
|
_LOCK = threading.Lock()
|
|
|
|
|
|
def api_for(project):
|
|
with _LOCK:
|
|
if project.name not in _APIS:
|
|
_APIS[project.name] = DecompAPI(project)
|
|
return _APIS[project.name]
|
|
|
|
|
|
def handle(project, path, query):
|
|
"""GET /api/decomp/<what>: returns (status, body)."""
|
|
api = api_for(project)
|
|
what = path.rsplit("/", 1)[-1]
|
|
if not (project.reverse or {}).get("facts"):
|
|
return 404, {"error": "this project has no [reverse] table"}
|
|
if what == "progress":
|
|
return 200, api.progress()
|
|
if what == "queues":
|
|
return 200, api.queues()
|
|
if what == "deferred":
|
|
return 200, api.deferred()
|
|
if what == "function":
|
|
target = (query.get("target") or "").strip()
|
|
if not target:
|
|
return 400, {"error": "give an address or a name"}
|
|
return 200, api.function(target)
|
|
if what == "tools":
|
|
return 200, api.tools()
|
|
return 404, {"error": "not found"}
|
|
|
|
|
|
def _json_safe(value):
|
|
return json.loads(json.dumps(value, default=str))
|