249 lines
13 KiB
Python
249 lines
13 KiB
Python
"""Which functions to decompile next, which to set aside, and how far the project is, per function.
|
|
|
|
- next_functions(agent): the functions in the agent's bands that are not decompiled, not deferred and not claimed by
|
|
someone else, easiest first. Ease is scored from the bytes and the team's records: small functions, callees that
|
|
already have names, stack-only conventions, game code, few earlier rejections and a readable near-duplicate rank
|
|
higher. A function rejected `attempt_cap` times or more is left out (defer it or ask the coordinator).
|
|
- defer(agent, address, reason, evidence): set a function aside with a reason slug; it becomes a note on fn:0x...
|
|
(`DEFERRED <reason>: <evidence>`) so everyone sees it, and the queue skips it until the note is replaced.
|
|
- progress(): every function in a tier, best first: exact (byte match), verified (behaves the same in the emulator),
|
|
readable, deferred, raw, plus library and engine functions counted apart.
|
|
|
|
Configured from the project's [reverse] table (facts, annotations, src, conventions) plus optional result files:
|
|
match_results (match71.py --json), diffrun_results (diffrun.py --json), library_results (libsweep csv)."""
|
|
import csv
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
|
|
from .reverse import Reverse
|
|
from .util import HarborError
|
|
|
|
DEFER_REASONS = ("register-args", "needs-type", "callee-unnamed", "inline", "x87", "layout", "too-large",
|
|
"engine-template", "data-not-code", "other")
|
|
READABLE = " * status: readable"
|
|
TIER_ORDER = ("exact", "verified", "readable", "deferred", "raw")
|
|
|
|
|
|
class FunctionQueue:
|
|
def __init__(self, harbor):
|
|
self.h = harbor
|
|
self.config = harbor.project.reverse or {}
|
|
self.r = Reverse(self.config)
|
|
self.cap = int(self.config.get("attempt_cap", 2))
|
|
self.min_size = int(self.config.get("queue_min_size", 8))
|
|
# [[lo, hi, "label"], ...] ranked low, e.g. the compiler-generated static initializers at the end of .text
|
|
self.low_priority = [(int(str(lo), 0), int(str(hi), 0), label)
|
|
for lo, hi, label in self.config.get("low_priority_ranges", [])]
|
|
|
|
# ------------------------------------------------------------------ records
|
|
def statuses(self):
|
|
"""address -> (status, file name) from the headers of src/**/<ADDR>_<Name>.c."""
|
|
out = {}
|
|
src = self.r.src
|
|
if not src or not src.exists():
|
|
return out
|
|
for path in src.rglob("*.c"):
|
|
m = re.match(r"([0-9A-F]{6,8})_", path.name)
|
|
if not m:
|
|
continue
|
|
with path.open(encoding="utf-8", errors="replace") as f:
|
|
head = [next(f, "") for _ in range(8)]
|
|
status = "readable" if any(line.startswith(READABLE) for line in head) else \
|
|
next((line.split("status:", 1)[1].strip() for line in head if "status:" in line), "raw")
|
|
out[int(m.group(1), 16)] = (status, path.name)
|
|
return out
|
|
|
|
def rejections(self):
|
|
"""address -> how many review items on fn:0x... were rejected or sent back."""
|
|
out = {}
|
|
rows = self.h.conn.execute(
|
|
"SELECT i.scope, COUNT(*) AS n FROM item_events e JOIN items i ON i.id = e.item_id "
|
|
"WHERE e.to_state IN ('rejected', 'changes_requested') AND i.scope LIKE 'fn:0x%' GROUP BY i.scope")
|
|
for row in rows:
|
|
try:
|
|
out[int(row["scope"][3:], 16)] = row["n"]
|
|
except ValueError:
|
|
pass
|
|
return out
|
|
|
|
def deferrals(self):
|
|
"""address -> (reason, evidence, author) for live DEFERRED notes."""
|
|
out = {}
|
|
for row in self.h.conn.execute("SELECT scope, body, author FROM notes WHERE replaced_by IS NULL "
|
|
"AND body LIKE 'DEFERRED %' AND scope LIKE 'fn:0x%'"):
|
|
m = re.match(r"DEFERRED ([\w-]+): (.*)", row["body"], re.S)
|
|
if m:
|
|
out[int(row["scope"][3:], 16)] = (m.group(1), m.group(2).strip(), row["author"])
|
|
return out
|
|
|
|
def bands(self, agent):
|
|
return [(row["lo"], row["hi"]) for row in self.h.conn.execute(
|
|
"SELECT lo, hi FROM claims WHERE kind='band' AND agent=? AND released_at IS NULL", (agent,))
|
|
if row["lo"] is not None]
|
|
|
|
def claimed_by_others(self, agent):
|
|
return [(row["lo"], row["hi"]) for row in self.h.conn.execute(
|
|
"SELECT lo, hi FROM claims WHERE kind IN ('draft','edit','naming') AND agent<>? AND released_at IS NULL",
|
|
(agent,)) if row["lo"] is not None]
|
|
|
|
def _named(self):
|
|
names = self.r._csv("names.csv")
|
|
return {a for a, row in names.items() if len(row) > 1 and row[1] and not re.match(r"(FUN|sub|thunk)_", row[1])}
|
|
|
|
def _category(self, addr):
|
|
row = self.r._csv("names.csv").get(addr)
|
|
if row and len(row) > 2 and row[2]:
|
|
return row[2]
|
|
return (self.r._facts().get(addr) or {}).get("category", "unsorted")
|
|
|
|
def _results(self, key):
|
|
path = self.config.get(key)
|
|
if not path or not Path(path).exists():
|
|
return {}
|
|
if str(path).endswith(".csv"):
|
|
with open(path, encoding="utf-8") as f:
|
|
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")
|
|
|
|
def library_ranges(self):
|
|
"""[(start, end)] of statically linked library blocks (the [reverse] library_ranges csv)."""
|
|
path = self.config.get("library_ranges")
|
|
if not path or not Path(path).exists():
|
|
return []
|
|
with open(path, encoding="utf-8") as f:
|
|
return [(int(r["start"], 16), int(r["end"], 16)) for r in csv.DictReader(f)
|
|
if r.get("verdict", "").startswith("D3DX") or "stubs" in r.get("verdict", "")]
|
|
# the D3DX9 block, and the compiler's import thunks and exception-unwind funclets: nothing to decompile
|
|
|
|
def is_library(self, addr, library, ranges):
|
|
return addr in library or any(lo <= addr < hi for lo, hi in ranges)
|
|
|
|
# ------------------------------------------------------------------ queue
|
|
def score(self, addr, facts, named, rejected, similar_readable=frozenset(), with_convention=True):
|
|
d = facts[addr]
|
|
size = int(d.get("size") or 0)
|
|
calls = [int(c, 16) if isinstance(c, str) else c for c in d.get("calls", [])]
|
|
named_share = (sum(1 for c in calls if c in named) / len(calls)) if calls else 1.0
|
|
score = 1.0 / (1.0 + size / 250.0) # small first
|
|
score *= 0.5 + 0.5 * named_share # callees with names make readable code possible
|
|
conv = self.r.convention(addr) if with_convention and size <= 4000 else None
|
|
if conv and conv["summary"].startswith("custom"):
|
|
score *= 0.75 # register arguments: prototypes.h work too
|
|
if self._category(addr) not in ("game", "unsorted"):
|
|
score *= 0.6
|
|
score *= 0.5 ** rejected.get(addr, 0)
|
|
low = next((label for lo, hi, label in self.low_priority if lo <= addr < hi), None)
|
|
if low:
|
|
score *= 0.2 # e.g. the compiler's static initializers
|
|
reasons_low = [low]
|
|
else:
|
|
reasons_low = []
|
|
if addr in similar_readable:
|
|
score *= 1.5 # a readable near-duplicate to start from
|
|
reasons = [f"{size} bytes", f"{round(named_share * 100)}% of {len(calls)} callees named"]
|
|
if conv:
|
|
reasons.append(conv["summary"])
|
|
if rejected.get(addr):
|
|
reasons.append(f"rejected {rejected[addr]}x")
|
|
if addr in similar_readable:
|
|
reasons.append("has a readable near-duplicate")
|
|
reasons += reasons_low
|
|
return score, reasons
|
|
|
|
def next_functions(self, agent, n=10, similar_readable=frozenset()):
|
|
bands = self.bands(agent)
|
|
if not bands:
|
|
raise HarborError(f"{agent} has no band; ask the coordinator for one")
|
|
taken = self.claimed_by_others(agent)
|
|
statuses, deferred, rejected = self.statuses(), self.deferrals(), self.rejections()
|
|
facts, named = self.r._facts(), self._named()
|
|
library = self._results("library_results")
|
|
ranges = self.library_ranges()
|
|
ranked = []
|
|
for addr in facts:
|
|
if not any(lo <= addr <= hi for lo, hi in bands) or any(lo <= addr <= hi for lo, hi in taken):
|
|
continue
|
|
if statuses.get(addr, ("raw",))[0] == "readable" or addr in deferred or self.is_library(addr, library, ranges):
|
|
continue
|
|
if rejected.get(addr, 0) >= self.cap:
|
|
continue
|
|
if self._category(addr) in ("library", "platform"):
|
|
continue
|
|
if int(facts[addr].get("size") or 0) < self.min_size or \
|
|
re.match(r"(nullsub|thunk)_", self.r.name_of(addr) or "") or \
|
|
re.match(r"thunk_", (statuses.get(addr) or (None, ""))[1].split("_", 1)[-1]):
|
|
continue # stubs and thunks are named in bulk, not decompiled
|
|
head = self.r.read(addr, 10) if int(facts[addr].get("size") or 0) == 10 else b""
|
|
if head[:1] and 0xB8 <= head[0] <= 0xBF and head[5:6] == b"\xe9":
|
|
continue # mov reg, imm32; jmp: an at-exit destructor thunk or an SEH handler stub (compiler's)
|
|
score, reasons = self.score(addr, facts, named, rejected, similar_readable, with_convention=False)
|
|
ranked.append((score, addr, reasons))
|
|
ranked.sort(reverse=True)
|
|
# the convention read from the bytes only lowers a score, and it is the slow part: read it for the head only
|
|
head = [(*self.score(a, facts, named, rejected, similar_readable), a) for _, a, _ in ranked[:max(4 * n, 40)]]
|
|
ranked = sorted(((s, a, why) for s, why, a in head), reverse=True)
|
|
return [{"address": f"0x{a:X}", "name": self.r.name_of(a), "score": round(s, 4), "why": ", ".join(why),
|
|
"file": statuses.get(a, (None, None))[1]} for s, a, why in ranked[:n]]
|
|
|
|
def defer(self, agent, address, reason, evidence):
|
|
if reason not in DEFER_REASONS:
|
|
raise HarborError("reason must be one of " + ", ".join(DEFER_REASONS))
|
|
if not (evidence or "").strip():
|
|
raise HarborError("say what you found (evidence), so the next person does not start from zero")
|
|
addr = self.r.resolve(address)
|
|
func = self.r.function_at(addr)
|
|
start = func[0] if func else addr
|
|
return self.h.note(agent, f"fn:0x{start:X}", f"DEFERRED {reason}: {evidence.strip()}")
|
|
|
|
# ------------------------------------------------------------------ progress
|
|
def progress(self):
|
|
facts, statuses, deferred = self.r._facts(), self.statuses(), self.deferrals()
|
|
exact = {a for a, row in self._results("match_results").items()
|
|
if str(row.get("status")) in ("match", "exact", "exact-relaxed")}
|
|
# behaves the same in the emulator, with at least one trial that ran to the end in both (not 'weak')
|
|
verified = {a for a, row in self._results("diffrun_results").items()
|
|
if str(row.get("verdict")) == "same" and not row.get("weak")}
|
|
library = set(self._results("library_results"))
|
|
ranges = self.library_ranges()
|
|
tiers = {t: {"functions": 0, "bytes": 0} for t in TIER_ORDER + ("library", "engine")}
|
|
for addr, d in facts.items():
|
|
size = int(d.get("size") or 0)
|
|
category = self._category(addr)
|
|
readable = statuses.get(addr, ("raw",))[0] == "readable"
|
|
if self.is_library(addr, library, ranges) or category in ("library", "platform", "crt"):
|
|
tier = "library"
|
|
elif category == "engine" and not readable:
|
|
tier = "engine"
|
|
elif addr in exact:
|
|
tier = "exact"
|
|
elif addr in verified:
|
|
tier = "verified"
|
|
elif readable:
|
|
tier = "readable"
|
|
elif addr in deferred:
|
|
tier = "deferred"
|
|
else:
|
|
tier = "raw"
|
|
tiers[tier]["functions"] += 1
|
|
tiers[tier]["bytes"] += size
|
|
sources = {k: bool(self.config.get(k) and Path(self.config[k]).exists())
|
|
for k in ("match_results", "diffrun_results", "library_results")}
|
|
return {"tiers": tiers, "deferred_reasons": _count(r for r, _, _ in deferred.values()), "sources": sources}
|
|
|
|
|
|
def _count(values):
|
|
out = {}
|
|
for v in values:
|
|
out[v] = out.get(v, 0) + 1
|
|
return out
|