forked from Keelrleek/Harbor
252 lines
13 KiB
Python
252 lines
13 KiB
Python
"""The shared reverse-engineering lookups, against a small synthetic image and annotation set."""
|
|
import json
|
|
import shutil
|
|
import struct
|
|
import unittest
|
|
|
|
from helpers import temp_dir
|
|
from harbor.reverse import Reverse
|
|
from harbor.util import HarborError
|
|
|
|
|
|
class ReverseTools(unittest.TestCase):
|
|
def setUp(self):
|
|
self.d = temp_dir()
|
|
base = 0x401000
|
|
image = bytearray(0x100)
|
|
# 0x401000: mov ecx,[0x401080]; call 0x401040; fmul dword [0x401090]; ret
|
|
code = bytes.fromhex("8b0d80104000") + b"\xe8" + struct.pack("<i", 0x401040 - 0x40100B) \
|
|
+ bytes.fromhex("d80d90104000") + b"\xc3"
|
|
image[0:len(code)] = code
|
|
image[0x40] = 0xC3 # 0x401040: ret
|
|
image[0x90:0x94] = struct.pack("<f", 100.0) # 0x401090: 100.0f
|
|
image[0xA0:0xA5] = b".cue\0"
|
|
(self.d / "image.bin").write_bytes(bytes(image))
|
|
facts = [{"address": "0x401000", "size": len(code), "name": "FUN_00401000", "calls": ["0x401040"],
|
|
"data": ["0x401080"], "imports": [], "strings": []},
|
|
{"address": "0x401040", "size": 1, "name": "FUN_00401040", "calls": [], "data": [],
|
|
"imports": [], "strings": []}]
|
|
(self.d / "facts.jsonl").write_text("\n".join(json.dumps(f) for f in facts), encoding="utf-8")
|
|
ann = self.d / "annotations"
|
|
ann.mkdir()
|
|
(ann / "names.csv").write_text("address,name,category,note\n0x00401000,GetLevel,game,\"Reads it.\"\n"
|
|
"0x00401040,Helper,game,x\n", encoding="utf-8")
|
|
(ann / "globals.csv").write_text("address,name,type,note\n0x00401080,g_root,int,x\n", encoding="utf-8")
|
|
(ann / "functions.csv").write_text("address,signature,note\n", encoding="utf-8")
|
|
(ann / "prototypes.h").write_text("int GetLevel(void); /* plain ret */\n", encoding="utf-8")
|
|
self.r = Reverse({"image": str(self.d / "image.bin"), "base": "0x401000", "facts": str(self.d / "facts.jsonl"),
|
|
"annotations": str(ann)})
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.d, ignore_errors=True)
|
|
|
|
def test_disassembly_names_calls_globals_and_constants(self):
|
|
text = self.r.disassemble("GetLevel")
|
|
self.assertIn("0x401080 g_root", text)
|
|
self.assertIn("0x401040 Helper", text)
|
|
self.assertIn("= 100.0f", text)
|
|
|
|
def test_xrefs_bytes_and_lookup(self):
|
|
x = self.r.xrefs("0x401040")
|
|
self.assertEqual([c["name"] for c in x["callers"]], ["GetLevel"])
|
|
self.assertEqual(self.r.bytes_at("0x401090", 4, "float")["float"], ["100.0"])
|
|
self.assertEqual(self.r.bytes_at("0x4010A0", as_type="string")["string"], ".cue")
|
|
info = self.r.lookup("GetLevel")
|
|
self.assertIn("plain ret", info["prototype"])
|
|
with self.assertRaises(HarborError):
|
|
self.r.bytes_at("0x500000")
|
|
with self.assertRaises(HarborError):
|
|
Reverse({})
|
|
|
|
def test_summary_names_status_vtables_and_hints(self):
|
|
image = bytearray((self.d / "image.bin").read_bytes())
|
|
image[0xC0:0xC8] = struct.pack("<II", 0x401000, 0x401040) # 0x4010C0: a two-slot vtable
|
|
(self.d / "image.bin").write_bytes(bytes(image))
|
|
(self.d / "sections.json").write_text(json.dumps([
|
|
{"name": ".text", "start": 0x401000, "end": 0x401080}, {"name": ".rdata", "start": 0x4010B0,
|
|
"end": 0x401100}]), encoding="utf-8")
|
|
facts = [json.loads(l) for l in (self.d / "facts.jsonl").read_text(encoding="utf-8").splitlines()]
|
|
facts[1]["data"] = ["0x4010C0"] # the constructor installs it
|
|
facts[0]["strings"] = [".cue"]
|
|
(self.d / "facts.jsonl").write_text("\n".join(json.dumps(f) for f in facts), encoding="utf-8")
|
|
src = self.d / "src" / "game"
|
|
src.mkdir(parents=True)
|
|
(src / "401040_Helper.c").write_text("/* Helper\n * status: readable\n */\nvoid Helper(void) {}\n",
|
|
encoding="utf-8")
|
|
(self.d / "mac.csv").write_text("pc_address,demangled,confidence\n0x00401000,Level::Get(),high\n",
|
|
encoding="utf-8")
|
|
r = Reverse({"image": str(self.d / "image.bin"), "base": "0x401000", "facts": str(self.d / "facts.jsonl"),
|
|
"annotations": str(self.d / "annotations"), "src": str(self.d / "src"),
|
|
"mac_names": str(self.d / "mac.csv")})
|
|
s = r.summary("GetLevel")
|
|
self.assertEqual((s["name"], s["size"], s["status"], s["readable"]), ("GetLevel", 18, "no file", False))
|
|
self.assertIn("plain ret", s["prototype"])
|
|
self.assertEqual(s["callees"], [{"address": "0x401040", "name": "Helper", "status": "readable",
|
|
"readable": True}])
|
|
self.assertEqual(s["globals"], [{"address": "0x401080", "name": "g_root", "type": "int"}])
|
|
self.assertEqual(s["strings"], [".cue"])
|
|
self.assertEqual(s["vtable_slots"], [{"table": "0x4010C0", "slot": 0, "offset": "0x0", "kind": "vtable"}])
|
|
self.assertEqual(s["hints"]["mac_name"]["demangled"], "Level::Get()")
|
|
h = r.summary("0x401040")
|
|
self.assertTrue(h["readable"])
|
|
self.assertEqual([c["name"] for c in h["callers"]], ["GetLevel"])
|
|
self.assertEqual(h["vtable_slots"][0]["slot"], 1)
|
|
from harbor.reverse import format_summary
|
|
self.assertIn("vtable 0x4010C0 slot 1 (+0x4)", format_summary(h))
|
|
|
|
|
|
class GhidraCheck(unittest.TestCase):
|
|
"""ghidra_check compares the served program with the project image: range, SHA-256 of the file, byte samples."""
|
|
|
|
def serve(self, lo, hi, path, image):
|
|
import http.server
|
|
import threading
|
|
|
|
class Fake(http.server.BaseHTTPRequestHandler):
|
|
def log_message(self, *a):
|
|
pass
|
|
|
|
def do_POST(self):
|
|
body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
|
|
if "id" not in body:
|
|
self.send_response(202)
|
|
self.end_headers()
|
|
return
|
|
params = body.get("params", {})
|
|
name, args = params.get("name"), params.get("arguments", {})
|
|
text = ""
|
|
if name == "get_binary_info":
|
|
text = f"Program Information:\nName: image.bin\nImage Base: 00000000\nMin Address: {lo:08x}\n" \
|
|
f"Max Address: {hi:08x}\nFunction Count: 2\n"
|
|
elif name == "list_binaries":
|
|
text = f"Open Programs in Ghidra:\n\n1. image.bin [ACTIVE]\n Executable Path: /{path}\n"
|
|
elif name == "get_data_at":
|
|
a, n = int(args["address"], 16), int(args["len"])
|
|
chunk = image[a - 0x401000:a - 0x401000 + n]
|
|
rows = [f"{a + i:x} " + " ".join(f"{b:02x}" for b in chunk[i:i + 16]) + " |....|"
|
|
for i in range(0, len(chunk), 16)]
|
|
text = f"Hexdump at {a:08x} ({n} bytes):\n\n" + "\n".join(rows) + "\n"
|
|
result = {"content": [{"type": "text", "text": text}]} if body["method"] == "tools/call" else {}
|
|
data = json.dumps({"jsonrpc": "2.0", "id": body["id"], "result": result}).encode()
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(data)))
|
|
self.end_headers()
|
|
self.wfile.write(data)
|
|
|
|
server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Fake)
|
|
threading.Thread(target=server.serve_forever, daemon=True).start()
|
|
self.addCleanup(server.server_close)
|
|
self.addCleanup(server.shutdown)
|
|
return f"http://127.0.0.1:{server.server_port}/mcp"
|
|
|
|
def test_match_and_mismatch(self):
|
|
from harbor.reverse import ghidra_check
|
|
d = temp_dir()
|
|
self.addCleanup(shutil.rmtree, d, True)
|
|
image = bytes(range(256)) * 4
|
|
(d / "image.bin").write_bytes(image)
|
|
config = {"image": str(d / "image.bin"), "base": "0x401000"}
|
|
good = ghidra_check(config | {"ghidra_url": self.serve(0x401000, 0x4013FF, (d / "image.bin").as_posix(), image)})
|
|
self.assertTrue(good["ok"], good)
|
|
self.assertEqual([c["ok"] for c in good["checks"]], [True, True, True])
|
|
other = d / "other.bin"
|
|
other.write_bytes(image[:-1] + b"\x00")
|
|
bad = ghidra_check(config | {"ghidra_url": self.serve(0x401000, 0x4013FE, other.as_posix(),
|
|
image[:512] + b"\0" * 512)})
|
|
self.assertFalse(bad["ok"])
|
|
self.assertEqual([c["ok"] for c in bad["checks"]], [False, False, False])
|
|
|
|
|
|
class GhidraForwarder(unittest.TestCase):
|
|
"""The ghidra() tool against a fake streamable-HTTP MCP server: a session, a polled async task, a refusal."""
|
|
|
|
def setUp(self):
|
|
import http.server
|
|
import threading
|
|
calls = self.calls = []
|
|
|
|
class Fake(http.server.BaseHTTPRequestHandler):
|
|
def log_message(self, *a):
|
|
pass
|
|
|
|
def do_POST(self):
|
|
body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
|
|
calls.append((body["method"], body.get("params", {}).get("name"), self.headers.get("Mcp-Session-Id")))
|
|
if "id" not in body:
|
|
self.send_response(202)
|
|
self.end_headers()
|
|
return
|
|
name = body.get("params", {}).get("name")
|
|
if body["method"] == "initialize":
|
|
result = {"protocolVersion": "2025-06-18"}
|
|
elif body["method"] == "tools/list":
|
|
result = {"tools": [{"name": "get_code", "description": "Code",
|
|
"inputSchema": {"properties": {"function": {}}, "required": ["function"]}}]}
|
|
elif name == "get_code":
|
|
result = {"content": [{"type": "text", "text": "Task submitted for async execution.\n\n"
|
|
"Task ID: 11111111-2222-3333-4444-555555555555\nStatus: RUNNING"}]}
|
|
elif name == "get_task_status":
|
|
result = {"content": [{"type": "text", "text": "[Context] Operating on: image.bin\n\nint F(void)"}]}
|
|
else:
|
|
result = {"isError": True, "content": [{"type": "text", "text": "Refused: read-only"}]}
|
|
data = ("data: " + json.dumps({"jsonrpc": "2.0", "id": body["id"], "result": result}) + "\n\n").encode()
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "text/event-stream")
|
|
self.send_header("Mcp-Session-Id", "s1")
|
|
self.send_header("Content-Length", str(len(data)))
|
|
self.end_headers()
|
|
self.wfile.write(data)
|
|
|
|
self.server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Fake)
|
|
threading.Thread(target=self.server.serve_forever, daemon=True).start()
|
|
from harbor.reverse import Ghidra
|
|
self.g = Ghidra({"ghidra_url": f"http://127.0.0.1:{self.server.server_port}/mcp"})
|
|
|
|
def tearDown(self):
|
|
self.server.shutdown()
|
|
self.server.server_close()
|
|
|
|
def test_tools_call_and_refusal(self):
|
|
self.assertEqual(self.g.tools()[0]["required"], ["function"])
|
|
self.assertEqual(self.g.call("get_code", {"function": "F"}), "int F(void)")
|
|
self.assertIn(("tools/call", "get_task_status", "s1"), self.calls)
|
|
with self.assertRaises(HarborError):
|
|
self.g.call("rename_symbol", {})
|
|
with self.assertRaises(HarborError):
|
|
from harbor.reverse import Ghidra
|
|
Ghidra({"ghidra_url": "http://127.0.0.1:9/mcp"}).tools()
|
|
|
|
|
|
class DebuggerForwarder(unittest.TestCase):
|
|
"""debugger() keeps one stdio MCP server per Harbor process and skips its notifications."""
|
|
|
|
def test_call_and_error(self):
|
|
import sys
|
|
from pathlib import Path
|
|
from harbor.reverse import Debugger
|
|
d = Debugger({"debugger_command": [sys.executable, "-B", str(Path(__file__).with_name("fake_stdio_mcp.py"))]})
|
|
try:
|
|
self.assertEqual(d.tools()[0]["name"], "echo")
|
|
self.assertEqual(d.call("echo", {"x": 1}), 'echo {"x": 1}')
|
|
with self.assertRaises(HarborError):
|
|
d.call("terminate_session")
|
|
finally:
|
|
d.close()
|
|
with self.assertRaises(HarborError):
|
|
Debugger({})
|
|
|
|
|
|
class MatchFunctions(unittest.TestCase):
|
|
def test_runs_the_matcher_and_reads_its_results(self):
|
|
from pathlib import Path
|
|
from harbor.reverse import match_functions
|
|
config = {"matcher": str(Path(__file__).with_name("fake_matcher.py")), "match_flags": "-TC -O2"}
|
|
res = match_functions(config, "0x412F70, 0x404810")
|
|
self.assertEqual([r["address"] for r in res["results"]], ["0x412F70", "0x404810"])
|
|
self.assertIn("-TC -O2", res["summary"])
|
|
with self.assertRaises(HarborError):
|
|
match_functions(config, "")
|
|
with self.assertRaises(HarborError):
|
|
match_functions({}, "0x412F70")
|