From 8b9371af0d707825ca64f9ea52ea136524de3ac4 Mon Sep 17 00:00:00 2001 From: branover Date: Sat, 15 Aug 2026 10:42:28 -0400 Subject: [PATCH 1/2] fix: normalize new Ghidra PIE imports --- docs/mcp.md | 14 +- src/hexgraph/agent/agent_tools.py | 77 +++- src/hexgraph/agent/mcp_catalog.py | 2 +- src/hexgraph/agent/vr_skill.py | 10 +- src/hexgraph/engine/observations.py | 37 +- src/hexgraph/engine/re/emulation.py | 3 +- src/hexgraph/engine/re/ghidra.py | 12 +- src/hexgraph/engine/re/taint.py | 6 +- src/hexgraph/sandbox/decompiler.py | 9 +- .../sandbox/probes/ghidra_bridge_probe.py | 8 +- src/hexgraph/sandbox/probes/ghidra_probe.py | 41 +- src/hexgraph/sandbox/probes/pyghidra_lib.py | 368 +++++++++++++++++- tests/test_decompiler_fallback.py | 13 + tests/test_ghidra_address_mapping.py | 273 +++++++++++++ tests/test_observations.py | 25 ++ tests/test_pyghidra_bridge.py | 15 + tests/test_re_script.py | 9 +- 17 files changed, 866 insertions(+), 56 deletions(-) create mode 100644 tests/test_ghidra_address_mapping.py diff --git a/docs/mcp.md b/docs/mcp.md index 443a69bf..a49c8cb8 100644 --- a/docs/mcp.md +++ b/docs/mcp.md @@ -283,7 +283,9 @@ already state, and `meta_get_schemas` spells out in its `substrate_vs_graph` and `re_script(target, script=…)` runs an agent-supplied **Python 3** script in the sandbox against the same warm program, opened **read-only** (`getReadOnlyDomainObject` — it queries everything but cannot mutate or save the project, and never executes the target). The namespace exposes `program`/`currentProgram`, a `flat` - `FlatProgramAPI`, `monitor`, and `out_path`; results come back as JSON written to `out_path` (or an assigned + `FlatProgramAPI`, `monitor`, `out_path`, the reported `address_mapping`, and + `file_offset_to_address(value)` / `address_to_file_offset(value)`; results come back as JSON written to + `out_path` (or an assigned `result`), the script body is capped at 64 KiB and delivered off the argv, and long output truncates to `max_chars` (recover it with `obs_get`). It's **warm-only** (run `re_analyze` first), **Ghidra-only**, and requires the persistent bridge to be stopped as described above; this conflict is detected before @@ -295,6 +297,16 @@ already state, and `meta_get_schemas` spells out in its `substrate_vs_graph` and gated behind **`features.ghidra.scripting`** — OFF by default, which HIDES the tool from the list until an operator enables it. Full Ghidra-API reach (data-flow slices, stack-frame layout, dispatch-table recovery, BSim/FunctionID naming, info-leak hunts) without a bespoke tool per query, and no sandbox boundary relaxed. + + New HexGraph Ghidra projects use ELF **virtual addresses**, matching radare2 and debugger virtual/module + offsets, not raw file offsets. On a new cold import HexGraph normalizes Ghidra's default PIE image base (Ghidra + 12.x commonly adds `0x100000`) back to the ELF's preferred load base *before* analysis. Every cold import + validates that base plus the ELF entry point and each `PT_LOAD` at its declared virtual addresses. Existing + warm projects are never rebased, rejected, or rebuilt: HexGraph preserves their analysis and returns a + top-level warning stating the actual image base and load bias whenever one differs from the ELF preference. + Every managed-project Ghidra result includes `address_mapping`, with the image base/load bias and each segment's file offset, virtual + address, and delta. Only when a source really reports a raw file offset, translate its `p_offset` coordinate + with `file_offset_to_address(...)` before calling Ghidra APIs. - **Enrichment of existing objects is automatic and free.** When a call recovers something unambiguous about an object that is *already* a node, a function's recovered prototype and address, the `is_sink` tag on a dangerous import, the call sites on an existing `calls` edge, HexGraph attaches it in place diff --git a/src/hexgraph/agent/agent_tools.py b/src/hexgraph/agent/agent_tools.py index 6447f973..d1ed2d09 100644 --- a/src/hexgraph/agent/agent_tools.py +++ b/src/hexgraph/agent/agent_tools.py @@ -448,6 +448,11 @@ class ToolContext: "fails before opening Ghidra with re_bridge_stop → retry → re_bridge_start remediation. " "CANCELLATION: the next call automatically reaps a prior one-shot project probe only when its " "launcher is positively gone; live or uncertain owners are preserved and reported. " + "ADDRESS SPACE: new PIE imports are normalized to the ELF preferred base. Existing warm " + "projects are preserved even when they use Ghidra's legacy load bias; every result warns with " + "their actual image base and load bias. The script namespace includes " + "address_mapping, file_offset_to_address(value), and address_to_file_offset(value), so use " + "file_offset_to_address only when a source reports a raw file offset. " "WARM-ONLY: errors → run re_analyze first if there's no warm project (never runs a cold " "analysis). CONTRACT (same as the built-in postScripts): your script receives " "out_path = getScriptArgs()[0] and MUST write its JSON result there; HexGraph reads it back. " @@ -584,6 +589,15 @@ def _callee_names(callees) -> list[str]: return out +def _address_mapping_warning(out: dict | None) -> str: + """User-visible warning for a preserved warm Ghidra project's coordinate system.""" + if not isinstance(out, dict): + return "" + mapping = out.get("address_mapping") + warning = out.get("warning") or (mapping.get("warning") if isinstance(mapping, dict) else None) + return f"// WARNING: {warning}\n" if warning else "" + + def _format_decomp(out: dict, label: str, *, limit: int = _MAX) -> str: """Render a focused decompile (decompile_function / decompile_at) result as text: the resolved name+address, callees, pseudocode, and any not-yet-promoted callees. @@ -591,6 +605,7 @@ def _format_decomp(out: dict, label: str, *, limit: int = _MAX) -> str: Truncates the body to `limit` chars with the ACTIONABLE marker (recovery knobs + sizes) rather than a bare `…[truncated]`, so a head-truncation can't silently hide a tail sink — the marker names both the max_chars re-call and the get_observation full-body path.""" + address_warning = _address_mapping_warning(out) focus = out.get("focus") if not focus: defined = out.get("functions", []) or [] @@ -605,7 +620,7 @@ def _format_decomp(out: dict, label: str, *, limit: int = _MAX) -> str: # simply not code). Dumping the function-name list can't help an address lookup — point at the # raw-disassembly / deeper-analysis escape hatches instead. if label.startswith("address "): - return (f"{label} is not inside any defined function (of {total} defined). It may be in an " + return address_warning + (f"{label} is not inside any defined function (of {total} defined). It may be in an " f"unanalyzed region, data, or a different segment. Try re_disassemble_range(, " f"count=…) for a raw disassembly there regardless of function boundaries, or " f"re_decompile_at(, reanalyze=True) to force a deeper analysis pass.") @@ -614,7 +629,7 @@ def _format_decomp(out: dict, label: str, *, limit: int = _MAX) -> str: # (never the whole thing) with the true total, and point at the address-based path. shown = defined[:40] more = f" (+{total - len(shown)} more)" if total > len(shown) else "" - return (f"{label} not found among the {total} defined functions" + return address_warning + (f"{label} not found among the {total} defined functions" + (f": {', '.join(shown)}{more}" if shown else "") + " — if you have its address, try re_decompile_at()") addr = f" @ {focus['address']}" if focus.get("address") else "" @@ -652,7 +667,7 @@ def _format_decomp(out: dict, label: str, *, limit: int = _MAX) -> str: # read — and it was concatenated last, so a long function clipped away the actionable half # first. Reserve it like a paging hint. return _clip_with_hint( - f"// {name}{addr} (callees: {', '.join(_callee_names(focus.get('callees')))})" + f"{address_warning}// {name}{addr} (callees: {', '.join(_callee_names(focus.get('callees')))})" f"{engine_warn}{node_ref}\n{focus.get('pseudocode', '')}", hint=note.lstrip("\n"), limit=limit, obs_id=out.get("observation_id")) @@ -782,14 +797,20 @@ def _decomp(ctx: ToolContext, function: str | None, *, ftotal = len(fns) if focused: subj = function or address + payload = {"functions": fns, "functions_total": ftotal} + if isinstance(out.get("address_mapping"), dict): + payload["address_mapping"] = out["address_mapping"] obs, _cached = _record_obs( ctx, tool=req_tool, args=req_args, result_kind="function_list", - payload={"functions": fns, "functions_total": ftotal}, + payload=payload, summary=f"{subj!r} not found; {ftotal} functions available") else: + payload = {"functions": fns, "functions_total": ftotal} + if isinstance(out.get("address_mapping"), dict): + payload["address_mapping"] = out["address_mapping"] obs, _cached = _record_obs( ctx, tool=req_tool, args=req_args, result_kind="function_list", - payload={"functions": fns, "functions_total": ftotal}, + payload=payload, summary=f"{ftotal} functions" + (" (re-analyzed)" if reanalyze else "")) out["observation_id"] = obs.id if obs is not None else None return out @@ -1396,12 +1417,15 @@ def _list_functions(ctx: ToolContext, args: dict) -> str: # own row) under a DISTINCT kind from the raw `function_list` inventory _decomp wrote — else # search_symbols_project (which reads the newest `function_list` as a target's whole-program # name set) would see only this filtered page and under-report its function names. + page_payload = {"functions": page, "total": total, "grand_total": grand_total, + "offset": offset, "limit": limit} + if isinstance(out.get("address_mapping"), dict): + page_payload["address_mapping"] = out["address_mapping"] _obs, _ = _record_obs(ctx, tool="list_functions", args={k: v for k, v in (("pattern", pat), ("offset", offset), ("limit", limit)) if v}, result_kind="function_list_page", - payload={"functions": page, "total": total, "grand_total": grand_total, - "offset": offset, "limit": limit}, + payload=page_payload, summary=f"{total} functions{pat_note}; page {offset}-{next_offset}") header = f"functions{pat_note} ({total} total, showing {offset}-{next_offset})" @@ -1415,7 +1439,7 @@ def _list_functions(ctx: ToolContext, args: dict) -> str: # read as the whole program: name the true total and how many are beyond the returned set. tail += (f"\n…[note: {grand_total} functions defined; {withheld} are beyond the returned " f"inventory and not listed here — re_decompile_at reaches any function by address]") - prefix = f"{header}:\n" + prefix = _address_mapping_warning(out) + f"{header}:\n" return _clip_page(prefix, body, tail, _obs.id if _obs is not None else None) @@ -2448,7 +2472,7 @@ def fmt_group(group: dict) -> list[str]: "listens_on/connects_to edges:\n" + "\n".join(fmt_group(net))) text = "\n\n".join(parts) if parts else \ "no dangerous, format-string, or network sinks referenced in this target" - ctx.cache[key] = _clip(text) + ctx.cache[key] = _clip(_address_mapping_warning(out) + text) return ctx.cache[key] @@ -2573,8 +2597,11 @@ def _function_xrefs(ctx: ToolContext, function: str) -> str: _record_obs(ctx, tool="function_xrefs", args={"function": function}, result_kind="function_xrefs", payload=out, summary=f"{function}: {len(callers)} callers, {len(callees)} callees") - lines = [f"// {function}: callers (who calls it) and callees (what it calls){source_note}", + lines = [_address_mapping_warning(out).rstrip("\n"), + f"// {function}: callers (who calls it) and callees (what it calls){source_note}", "callers:"] + if not lines[0]: + lines.pop(0) lines += [f"- {c['caller']}" + (f" (@ {c['caller_addr']})" if c.get("caller_addr") else "") + (f" at {c['at']}" if c.get("at") else "") for c in callers] or [" (none)"] @@ -2660,17 +2687,19 @@ def _data_xrefs(ctx: ToolContext, address: str) -> str: if err: return err if out.get("error") or out.get("not_found"): - return _no_data_xrefs_msg(address, out, ctx) + return _address_mapping_warning(out) + _no_data_xrefs_msg(address, out, ctx) refs = out.get("data_refs") or [] _record_obs(ctx, tool="data_xrefs", args={"address": address}, result_kind="data_xrefs", payload=out, summary=f"{len(refs)} refs to {address}") if not refs: - ctx.cache[key] = (f"no references to {address} found" + ctx.cache[key] = (_address_mapping_warning(out) + f"no references to {address} found" + _pending_recovery_caveat(out, ctx)) return ctx.cache[key] more = out.get("total", len(refs)) - len(refs) - lines = [f"references to {address}:"] + lines = [_address_mapping_warning(out).rstrip("\n"), f"references to {address}:"] + if not lines[0]: + lines.pop(0) lines += [f"- {r['from_function']} at {r.get('at')} ({r.get('kind')})" for r in refs] if more > 0: lines.append(f" … and {more} more") @@ -2756,24 +2785,28 @@ def _call_graph_tool(ctx: ToolContext, function: str | None, depth) -> str: # TEXT), so record under args={} — it dedups to ONE Observation no matter how many rooted # views are requested, instead of re-storing the identical graph per root. from hexgraph.engine.re.ghidra import _call_graph_records + call_graph_payload = {"functions": _call_graph_records(edges)} + if isinstance(out.get("address_mapping"), dict): + call_graph_payload["address_mapping"] = out["address_mapping"] _record_obs(ctx, tool="call_graph", args={}, - result_kind="call_graph", payload={"functions": _call_graph_records(edges)}, + result_kind="call_graph", payload=call_graph_payload, summary=f"{len(edges)} call edges") if function: d = max(1, min(int(depth or 2), 6)) sub = _bfs_subgraph(edges, function, d) if not sub: - ctx.cache[key] = f"{function!r} not found in the call graph (or it calls nothing)" + ctx.cache[key] = (_address_mapping_warning(out) + + f"{function!r} not found in the call graph (or it calls nothing)") return ctx.cache[key] text = f"call graph from {function} (depth {d}, {len(sub)} edges){source_note}:\n" + \ "\n".join(f"- {a} → {b}" for a, b in sub) - ctx.cache[key] = _clip(text) + ctx.cache[key] = _clip(_address_mapping_warning(out) + text) return ctx.cache[key] shown = edges[:200] note = f"\n … and {len(edges) - len(shown)} more edges" if len(edges) > len(shown) else "" text = f"call graph ({len(edges)} edges){source_note}:\n" + \ "\n".join(f"- {p[0]} → {p[1]}" for p in shown if len(p) == 2) + note - ctx.cache[key] = _clip(text) + ctx.cache[key] = _clip(_address_mapping_warning(out) + text) return ctx.cache[key] @@ -3339,16 +3372,20 @@ def _search_code_scan(ctx: ToolContext, args: dict, *, bytes_pat, immediate) -> next_offset = offset + len(page) more = next_offset < total + scan_payload = {"mode": "scan", "bytes_pattern": bytes_pat, "immediate": immediate, + "hits": page, "total": total, "offset": offset, "limit": limit} + if isinstance(out.get("address_mapping"), dict): + scan_payload["address_mapping"] = out["address_mapping"] obs, _cached = _record_obs( ctx, tool="search_code", args={k: v for k, v in (("bytes_pattern", bytes_pat), ("immediate", immediate), ("offset", offset), ("limit", limit)) if v is not None and v != 0}, result_kind="search_code", - payload={"mode": "scan", "bytes_pattern": bytes_pat, "immediate": immediate, - "hits": page, "total": total, "offset": offset, "limit": limit}, + payload=scan_payload, summary=f"scan {subj}: {total} hit(s); page {offset}-{next_offset}") - header = f"search_code scan for {subj} ({total} hit(s), showing {offset}-{next_offset}):" + header = (_address_mapping_warning(out) + + f"search_code scan for {subj} ({total} hit(s), showing {offset}-{next_offset}):") body = "\n".join( f"- {h['addr']}" + (f" in {h['in_function']}" if h.get("in_function") else " (no function)") for h in page) or "(none)" diff --git a/src/hexgraph/agent/mcp_catalog.py b/src/hexgraph/agent/mcp_catalog.py index ff4e8502..e820ddf0 100644 --- a/src/hexgraph/agent/mcp_catalog.py +++ b/src/hexgraph/agent/mcp_catalog.py @@ -185,7 +185,7 @@ ("read", "re_hexdump", _t.hexdump, "Dump raw BYTES at a virtual ADDRESS as hex + ascii (bounded — default 256, max 4096) — inspect a DAT_ table, an embedded key, a struct, or a string constant's exact bytes. Reads the bytes in the sandbox via radare2 (`p8` at the vaddr, no whole-program analysis) and renders the hexdump host-side. Bytes in a .bss/zero-fill region read as 00 with a note; an unmapped address is reported, not faked. QUERY: records an Observation; adds no graph nodes. (For the INSTRUCTIONS at an address use re_disassemble_range; this is the raw-bytes view.)", {"type": "object", "properties": {"target_id": {"type": "string"}, "address": {"type": "string", "description": "hex virtual address, e.g. 0x4c1000"}, "length": {"type": "integer", "description": "bytes to dump (default 256, clamped 1-4096)"}}, "required": ["target_id", "address"]}), ("read", "re_function_info", _t.function_info, "Lightweight metadata for a function (by NAME or ADDRESS) WITHOUT a full decompile: address, size, prototype/signature + calling convention (when known — from the symbol table, a prior decompilation, or recon), and #callers / #callees (from the call graph). The cheap 'what is this function' triage before deciding to re_decompile_function it. Fields not yet recovered are marked unknown (decompile to recover). QUERY: records an Observation; adds no graph nodes.", {"type": "object", "properties": {"target_id": {"type": "string"}, "function": {"type": "string", "description": "function name (or pass address)"}, "address": {"type": "string", "description": "hex address in the function (or pass function)"}}, "required": ["target_id"]}), ("read", "re_search_code", _t.search_code, "Search the WHOLE binary's code (not just already-decompiled bodies, which re_search_decompiled covers). Two modes with VERY different costs. CHEAP — a BYTE/opcode pattern (`bytes_pattern`, hex pairs) or an IMMEDIATE constant (`immediate`) scanned across the mapped image: a FAST memory scan over the WARM Ghidra project (reuses the analysis, no whole-binary re-scan; radare2 raw-scan fallback when Ghidra's off), each hit mapped to its containing function. EXPENSIVE — a decompile-on-demand grep (`query` + `functions`): every named function with no already-recorded body costs a REAL DECOMPILE, tens of seconds each on a large target (a container spawn + JVM start + project open per function unless a Ghidra bridge is running), so 30 functions is ~10 minutes. Bodies already in the Observation store are reused for free, so grep functions you have already decompiled cheaply; before paying for the rest, prefer re_search_decompiled (already-decompiled bodies, no decompile) and re_xrefs (CALLERS of a symbol/sink — whole-program and indexed; this does NOT duplicate it). Both modes are paginated. Naming MORE than one page of functions (>50) with no offset/limit of your own runs the grep DETACHED over the whole set and returns a task handle — re-call the same search to poll it (once it finishes that same re-call hands you the completed result's search_code Observation, and never re-sweeps a set already swept) — because paging it synchronously would be several partial calls. Name a `limit` to page it yourself instead, in which case the grep stops after a wall-clock budget and reports the offset to resume from. The grep decompiles, so it needs a saved analysis for the ACTIVE persistent backend — Ghidra OR radare2 (run re_analyze first on a warm miss); the byte/immediate scan degrades to a radare2 raw scan instead. QUERY: records an Observation; adds no graph nodes.", {"type": "object", "properties": {"target_id": {"type": "string"}, "bytes_pattern": {"type": "string", "description": "hex byte pattern to scan for, e.g. 'deadbeef' or '48 8b'"}, "immediate": {"type": "string", "description": "an immediate/constant value to find (hex or decimal)"}, "functions": {"type": "array", "description": "bound the decompile-on-demand grep to these functions (with `query`). EACH one not already decompiled costs a full decompile — name only real candidates"}, "query": {"type": "string", "description": "substring to grep in the decompiled bodies of `functions`"}, "offset": {"type": "integer", "description": "page start: into the HITS for a byte/immediate scan, into the FUNCTIONS list for a grep"}, "limit": {"type": "integer", "description": "page size: hits for a scan (default 100, max 500); functions to decompile for a grep (default and max 50)"}, "max_chars": {"type": "integer", "description": _MAX_CHARS_DESC}}, "required": ["target_id"]}), - ("run", "re_script", _t.run_script, "ESCAPE-HATCH: run an AGENT-SUPPLIED PyGhidra Python-3 script against a target's already-built WARM Ghidra project and get its JSON output — full Ghidra-API power (data-flow slicing, BSim/FID, stack-frame analysis, custom P-Code walks) WITHOUT a pre-built tool per capability, for a query the other re_* verbs don't cover. GHIDRA-ONLY (radare2 unsupported — errors if headless Ghidra isn't the active backend). Runs inside the SAME hardened sandbox every probe uses (--network none, read-only rootfs, --cap-drop ALL, non-root) and opens the warm project READ-ONLY, so your script can inspect but NEVER mutate/corrupt it; the target binary is NEVER executed (Ghidra statically analyzes). PROJECT OWNERSHIP: before opening Ghidra, re_script reconciles one-shot HexGraph probes on the exact warm project. A probe whose launcher process is positively gone (for example after its MCP/client session was terminated) is stopped automatically; a live or uncertain owner is preserved and reported rather than disrupting another agent. A persistent bridge is also preserved and returns exact re_bridge_stop → retry → re_bridge_start remediation. A raced LockException is translated into the same ownership guidance. WARM-ONLY: if the target has no warm project yet it errors → run re_analyze(target) first (this never runs a cold analysis). CONTRACT (same as HexGraph's built-in postScripts): your script receives `out_path = getScriptArgs()[0]` and MUST write its JSON result to that path — HexGraph reads it back and returns it. Tiny example: `import json; fm = currentProgram.getFunctionManager(); n = len(list(fm.getFunctions(True))); f = open(getScriptArgs()[0], 'w'); f.write(json.dumps({'function_count': n})); f.close()`. program / currentProgram / flat / monitor / out_path / getScriptArgs() and the ghidra.* API are in scope — it's Python 3 (write JSON to out_path, or assign the `result` variable). Gated: features.ghidra.scripting — opt-in (hidden until enabled — it's an arbitrary-code-in-sandbox surface). Records a `script` Observation; adds NO graph nodes. Long output is truncated to max_chars (default 6000, clamped) with a marker; obs_get reads the full payload.", + ("run", "re_script", _t.run_script, "ESCAPE-HATCH: run an AGENT-SUPPLIED PyGhidra Python-3 script against a target's already-built WARM Ghidra project and get its JSON output — full Ghidra-API power (data-flow slicing, BSim/FID, stack-frame analysis, custom P-Code walks) WITHOUT a pre-built tool per capability, for a query the other re_* verbs don't cover. GHIDRA-ONLY (radare2 unsupported — errors if headless Ghidra isn't the active backend). Runs inside the SAME hardened sandbox every probe uses (--network none, read-only rootfs, --cap-drop ALL, non-root) and opens the warm project READ-ONLY, so your script can inspect but NEVER mutate/corrupt it; the target binary is NEVER executed (Ghidra statically analyzes). PROJECT OWNERSHIP: before opening Ghidra, re_script reconciles one-shot HexGraph probes on the exact warm project. A probe whose launcher process is positively gone (for example after its MCP/client session was terminated) is stopped automatically; a live or uncertain owner is preserved and reported rather than disrupting another agent. A persistent bridge is also preserved and returns exact re_bridge_stop → retry → re_bridge_start remediation. A raced LockException is translated into the same ownership guidance. ADDRESS SPACE: new PIE imports are normalized to the ELF preferred base. Existing warm projects are preserved even when they use Ghidra's legacy load bias; every result warns with their actual image base and load bias. The script namespace includes `address_mapping`, `file_offset_to_address(value)`, and `address_to_file_offset(value)` — use the first helper only for raw file offsets. WARM-ONLY: if the target has no warm project yet it errors → run re_analyze(target) first (this never runs a cold analysis). CONTRACT (same as HexGraph's built-in postScripts): your script receives `out_path = getScriptArgs()[0]` and MUST write its JSON result to that path — HexGraph reads it back and returns it. Tiny example: `import json; fm = currentProgram.getFunctionManager(); n = len(list(fm.getFunctions(True))); f = open(getScriptArgs()[0], 'w'); f.write(json.dumps({'function_count': n})); f.close()`. program / currentProgram / flat / monitor / out_path / getScriptArgs() / address_mapping / file_offset_to_address / address_to_file_offset and the ghidra.* API are in scope — it's Python 3 (write JSON to out_path, or assign the `result` variable). Gated: features.ghidra.scripting — opt-in (hidden until enabled — it's an arbitrary-code-in-sandbox surface). Records a `script` Observation; adds NO graph nodes. Long output is truncated to max_chars (default 6000, clamped) with a marker; obs_get reads the full payload.", {"type": "object", "properties": {"target_id": {"type": "string"}, "script": {"type": "string", "description": "PyGhidra/Jython source; writes its JSON result to getScriptArgs()[0]"}, "max_chars": {"type": "integer", "description": _MAX_CHARS_DESC}}, "required": ["target_id", "script"]}), ("read", "re_search_symbols_project", _t.search_symbols_project, "Search a symbol/function NAME pattern across ALL targets in a project -> which target(s) DEFINE or IMPORT it (the name analogue of re_yara_sweep). For each non-archived target it tests the pattern against its imports/exports/defined symbols (and function names when known), sourced from stored recon metadata + prior re_binutils_facts Observations — no per-target decompile, no probe. Returns {pattern, matches:[{target_id, name, kind}], scanned, targets_without_symbols}. Use to locate which binary in a firmware defines a given sink or a shared helper. `regex=true` for a regex; `kind` scopes to imports|exports|defined. A target with no stored symbols is listed in targets_without_symbols (re_binutils_facts it first) so a miss isn't read as authoritative. QUERY: reads the substrate; adds no graph nodes.", {"type": "object", "properties": {"project_id": {"type": "string"}, "pattern": {"type": "string", "description": "substring (or regex) to match against symbol/function names"}, "kind": {"type": "string", "enum": ["imports", "exports", "defined", "all"], "description": "scope (default all)"}, "regex": {"type": "boolean"}, "limit": {"type": "integer", "description": "max matches to return"}}, "required": ["project_id", "pattern"]}), diff --git a/src/hexgraph/agent/vr_skill.py b/src/hexgraph/agent/vr_skill.py index 0df276e2..4a524b97 100644 --- a/src/hexgraph/agent/vr_skill.py +++ b/src/hexgraph/agent/vr_skill.py @@ -418,7 +418,8 @@ to put names on `FUN_*` bodies, or an info-leak hunt for a routine that copies uninitialised stack into a reply. - **The contract.** The namespace exposes `program`/`currentProgram` (the analysed `Program`), - `flat` (a `FlatProgramAPI`), `monitor`, and `out_path`; `import ghidra.*` freely. Return + `flat` (a `FlatProgramAPI`), `monitor`, `out_path`, the reported `address_mapping`, and + `file_offset_to_address(value)` / `address_to_file_offset(value)`; `import ghidra.*` freely. Return results by writing JSON to `out_path` (or assigning a JSON-serialisable `result`) — that becomes the tool output (truncates to `max_chars`; recover the tail with `obs_get`). The script body is capped at 64 KiB and delivered off the argv; it records ONE `script` Observation, no graph nodes. @@ -428,6 +429,13 @@ rows = [{"name": v.getName(), "off": v.getStackOffset(), "size": v.getLength()} for v in f.getStackFrame().getStackVariables()] import json; open(out_path, "w").write(json.dumps(rows)) +- **Use the right address coordinate.** HexGraph normalizes new PIE imports to the ELF's preferred + base, so new Ghidra, radare2, and debugger virtual/module offsets agree. Existing warm projects + are preserved at their original Ghidra image base; each result warns with that base and load bias. + Raw file offsets remain + different whenever a `PT_LOAD` maps `p_offset` to `p_vaddr`; inspect `address_mapping` and pass + only those file coordinates through `file_offset_to_address(...)`. Inspect `address_mapping` + before combining addresses from a preserved legacy project with results from another backend. - **Warm-only, Ghidra-only, gated.** Run `re_analyze(target)` first; unavailable on a radare2-only project. **Gate: features.ghidra.scripting** — OFF by default, and the tool is HIDDEN from your list until the operator enables it. If you don't see `re_script`, scripting diff --git a/src/hexgraph/engine/observations.py b/src/hexgraph/engine/observations.py index 1fe44bd1..4479f08d 100644 --- a/src/hexgraph/engine/observations.py +++ b/src/hexgraph/engine/observations.py @@ -6,7 +6,9 @@ that aren't promoted into the graph yet — what both agent and user mine to decide what belongs there — and it gives "analyze once, reuse forever" for free: a repeat call with the same `(tool, args, content_hash, result_kind)` returns the existing -row flagged `cached` instead of re-running. +row flagged `cached` instead of re-running. Address-bearing Ghidra results additionally key on the +reported Program image base/load bias, because a preserved legacy warm project and a normalized new +import can validly describe the same bytes in different coordinates. Observations are NOT graph nodes; recording one creates ZERO nodes/edges. The link to the graph is bidirectional by reference only: an enriched node/edge/finding @@ -28,6 +30,28 @@ from hexgraph.engine import cas +_ADDRESS_MAPPING_ARG = "_ghidra_address_mapping" + + +def _args_with_address_mapping(args: dict[str, Any] | None, payload: Any) -> dict[str, Any] | None: + """Version address-bearing Ghidra results by the Program coordinate system. + + The same bytes and tool arguments can legitimately produce different addresses when an older + warm Ghidra project uses its historical PIE load bias while a new project uses the normalized + ELF base. Include the reported schema/base/bias in the dedup key so neither result can silently + reuse the other's Observation. Non-Ghidra payloads keep their long-standing key unchanged. + """ + mapping = payload.get("address_mapping") if isinstance(payload, dict) else None + if not isinstance(mapping, dict): + return args + fingerprint = ":".join(str(mapping.get(key, "")) for key in ( + "schema", "image_base", "preferred_image_base", "ghidra_load_bias", + )) + versioned = dict(args or {}) + versioned[_ADDRESS_MAPPING_ARG] = fingerprint + return versioned + + def _normalize_args(args: dict[str, Any] | None) -> dict[str, Any]: """Canonical form of the call args so dedup is order-insensitive. Drops None values so an omitted optional arg dedups with an explicit None.""" @@ -43,8 +67,8 @@ def _find_fresh( args: dict[str, Any] | None, content_hash: str | None, result_kind: str, ) -> Observation | None: """An existing OK Observation for the identical call against the identical bytes. - The dedup key is (tool, normalized args, content_hash, result_kind) — the same - bytes + the same call must yield the same answer (design §5.2).""" + The dedup key is (tool, normalized args, content_hash, result_kind); Ghidra mapping identity is + folded into normalized args by `record_observation` (design §5.2).""" key = _args_key(args) rows = ( session.query(Observation) @@ -83,7 +107,8 @@ def record_observation( """Record one tool call, or reuse a fresh identical one. Returns `(observation, cached)`. When an OK Observation already exists for the - same `(tool, normalized args, content_hash, result_kind)`, returns it with + same `(tool, normalized args, content_hash, result_kind)`, including any Ghidra mapping + fingerprint folded into the args, returns it with `cached=True` and stores nothing new ("analyze once, reuse forever", §5.2). Otherwise stores the full `payload` in CAS, sets `result_cas`/`size`, writes the row, and returns `cached=False`. Creates NO graph nodes/edges (curation gate). @@ -128,6 +153,10 @@ def record_observation( becomes durable when `session` commits — for callers that genuinely want the Observation to share the caller's transaction lifetime (e.g. a unit-of-work test, or a flow whose observation should roll back with it).""" + # A Ghidra Program's coordinate system is part of the deterministic call identity. This keeps + # preserved legacy warm projects and normalized new imports from sharing address-bearing rows. + args = _args_with_address_mapping(args, payload) + # Only OK results dedup — re-running after an error must be allowed to retry. The # lookup runs on the caller's session so it also sees this task's own just-flushed # rows; the durable checkpoint below is what makes a fresh row survive a later failure. diff --git a/src/hexgraph/engine/re/emulation.py b/src/hexgraph/engine/re/emulation.py index 93adff75..f32c5b32 100644 --- a/src/hexgraph/engine/re/emulation.py +++ b/src/hexgraph/engine/re/emulation.py @@ -156,4 +156,5 @@ def emulate_constant(session: Session, project: Any, target: Any, *, function: s return {"available": True, "function": function, "value": value, "value_hex": emu.get("value_hex"), "reached_ret": reached, "steps": emu.get("steps"), "width_bytes": emu.get("width_bytes"), - "observation_id": observation.id, "error": err} + "observation_id": observation.id, "error": err, + "address_mapping": out.get("address_mapping"), "warning": out.get("warning")} diff --git a/src/hexgraph/engine/re/ghidra.py b/src/hexgraph/engine/re/ghidra.py index ff7a5beb..1e6c9059 100644 --- a/src/hexgraph/engine/re/ghidra.py +++ b/src/hexgraph/engine/re/ghidra.py @@ -160,6 +160,10 @@ def enrich_target(session, project, target) -> dict: functions = list(data.get("functions") or []) calls = list(data.get("calls") or []) structs = list(data.get("structs") or []) + mapping = data.get("address_mapping") if isinstance(data.get("address_mapping"), dict) else None + + def _with_mapping(payload): + return {**payload, "address_mapping": mapping} if mapping is not None else payload def _record(tool, result_kind, payload, summary): # content_hash scopes the facts to the exact bytes (extract-at-write + passive @@ -171,13 +175,15 @@ def _record(tool, result_kind, payload, summary): # Function inventory + recovered prototypes/addresses → function_list facts. _record("enrich_recon", "function_list", - {"functions": [f if isinstance(f, dict) else {"name": f} for f in functions]}, + _with_mapping({"functions": [ + f if isinstance(f, dict) else {"name": f} for f in functions]}), f"{len(functions)} functions") # Call graph → `A calls B` relationship facts (edges self-wire among promoted fns). _record("enrich_recon", "call_graph", - {"functions": _call_graph_records(calls)}, f"{len(calls)} call edges") + _with_mapping({"functions": _call_graph_records(calls)}), f"{len(calls)} call edges") # Recovered structs → real-layout facts (the extractor drops built-ins). - _record("enrich_recon", "structs", {"structs": structs}, f"{len(structs)} structs") + _record("enrich_recon", "structs", _with_mapping({"structs": structs}), + f"{len(structs)} structs") return {"ok": True, "recorded": True, "functions": len(functions), "calls": len(calls), "structs": len(structs)} diff --git a/src/hexgraph/engine/re/taint.py b/src/hexgraph/engine/re/taint.py index 3f7a8a0a..84c2d514 100644 --- a/src/hexgraph/engine/re/taint.py +++ b/src/hexgraph/engine/re/taint.py @@ -103,7 +103,8 @@ def _run(a, **kw): return {"available": True, "flows": [], "analyzed": 0, "error": str(exc)} taint = out.get("taint") or {} return {"available": True, "flows": taint.get("flows", []), - "analyzed": taint.get("analyzed", 0), "error": out.get("error")} + "analyzed": taint.get("analyzed", 0), "error": out.get("error"), + "address_mapping": out.get("address_mapping"), "warning": out.get("warning")} def get_taint_analyzer() -> TaintAnalyzer: @@ -226,4 +227,5 @@ def analyze_taint(session: Session, project: Any, target: Any, *, return {"available": True, "flows": flows, "analyzed": result.get("analyzed", 0), "promoted": promoted, "observation_id": observation.id, "cached": cached, - "error": result.get("error")} + "error": result.get("error"), "address_mapping": result.get("address_mapping"), + "warning": result.get("warning")} diff --git a/src/hexgraph/sandbox/decompiler.py b/src/hexgraph/sandbox/decompiler.py index 521918f5..6a6b0689 100644 --- a/src/hexgraph/sandbox/decompiler.py +++ b/src/hexgraph/sandbox/decompiler.py @@ -44,9 +44,12 @@ def focus_only_payload(out: dict) -> dict: ftotal = (out or {}).get("functions_total") if not isinstance(ftotal, int) or ftotal < len(fns): ftotal = len(fns) - return {"functions": fns[:_FOCUS_PAYLOAD_FUNCTION_SAMPLE], - "functions_total": ftotal, - "focus": (out or {}).get("focus")} + payload = {"functions": fns[:_FOCUS_PAYLOAD_FUNCTION_SAMPLE], + "functions_total": ftotal, + "focus": (out or {}).get("focus")} + if isinstance((out or {}).get("address_mapping"), dict): + payload["address_mapping"] = out["address_mapping"] + return payload class Decompiler(ABC): diff --git a/src/hexgraph/sandbox/probes/ghidra_bridge_probe.py b/src/hexgraph/sandbox/probes/ghidra_bridge_probe.py index 2c1fbcd0..92c8cae9 100644 --- a/src/hexgraph/sandbox/probes/ghidra_bridge_probe.py +++ b/src/hexgraph/sandbox/probes/ghidra_bridge_probe.py @@ -35,7 +35,9 @@ def main() -> int: try: # WARM open REQUIRED (start_bridge gates on slot.exists()); cold_analyze=False so a missing # slot fails fast here rather than silently kicking off a huge cold analysis in the bridge. - with L.open_target(artifact, cold_analyze=False) as (program, flat, cached): + mapping_out = {} + with L.open_target( + artifact, cold_analyze=False, mapping_out=mapping_out) as (program, flat, cached): from ghidra.util.task import ConsoleTaskMonitor sys.stderr.write(f"ghidra_bridge_probe: project resident (cached={cached}); " @@ -45,7 +47,9 @@ def main() -> int: # fresh ConsoleTaskMonitor per request so a decompile timeout — which cancels the monitor # it's handed, permanently for a ConsoleTaskMonitor — can't poison later decompiles into # empty bodies. See pyghidra_lib._serve_one. - L.serve_bridge("0.0.0.0", PORT, program, flat, ConsoleTaskMonitor) # blocks forever + L.serve_bridge( + "0.0.0.0", PORT, program, flat, ConsoleTaskMonitor, + mapping_out.get("address_mapping")) # blocks forever except Exception as exc: # noqa: BLE001 sys.stderr.write(f"ghidra_bridge_probe: {exc}\n") return 4 diff --git a/src/hexgraph/sandbox/probes/ghidra_probe.py b/src/hexgraph/sandbox/probes/ghidra_probe.py index 09a98d9f..91e5ad04 100644 --- a/src/hexgraph/sandbox/probes/ghidra_probe.py +++ b/src/hexgraph/sandbox/probes/ghidra_probe.py @@ -215,20 +215,28 @@ def _run(m) -> dict: if mode == "recover_data_refs": # Its own short path: no core runs, so the warm slot is opened once — by the recovery itself # — and a detached recovery does not pay for a whole-program inventory it would discard. - return {"tool": "ghidra_probe", "mode": mode, "cached": True, - "fast_profile": L.fast_profile_applies(m["artifact"]), - "data_ref_recovery": _recover_data_refs(m["artifact"])} + recovery = _recover_data_refs(m["artifact"]) + mapping = recovery.pop("address_mapping", None) + result = {"tool": "ghidra_probe", "mode": mode, "cached": True, + "fast_profile": L.fast_profile_applies(m["artifact"]), + "data_ref_recovery": recovery, "address_mapping": mapping} + if isinstance(mapping, dict) and mapping.get("warning"): + result["warning"] = mapping["warning"] + return result # THE analysis chokepoint: ONLY the `analyze` mode (re_analyze) may build a cold analysis. Every # other mode is WARM-ONLY — main() already refused a cold miss with the re_analyze lead, so on the # warm path open_target never analyzes. script + search additionally open the program READ-ONLY # (immutable) so a query can't mutate the project. cold_analyze = mode == "analyze" read_only = mode in ("script", "search") + mapping_out = {} with L.open_target(m["artifact"], cold_analyze=cold_analyze, - read_only=read_only) as (program, flat, cached): + read_only=read_only, mapping_out=mapping_out) as (program, flat, cached): monitor = ConsoleTaskMonitor() if mode == "script": - result = L.script_core(program, flat, monitor, m["user_script"]) + result = L.script_core( + program, flat, monitor, m["user_script"], + address_mapping=mapping_out.get("address_mapping")) elif mode == "search": result = L.search_bytes_core(program, flat, monitor, bytes_pattern=m.get("search_bytes"), @@ -243,6 +251,9 @@ def _run(m) -> dict: result = L.decompile_core(program, flat, monitor, focus=m["focus"], rename=m["rename"]) result.setdefault("tool", "ghidra_probe") result["cached"] = cached + result["address_mapping"] = mapping_out.get("address_mapping") + if mapping_out.get("address_mapping", {}).get("warning"): + result.setdefault("warning", mapping_out["address_mapping"]["warning"]) # Whether this target's analysis runs under the fast profile is part of its RESULT, not a detail # an agent has to infer from empty xrefs: the profile disables the constant/scalar reference @@ -299,13 +310,17 @@ def _recover_data_refs(artifact) -> dict: through = marker.get("data_refs_recovered_through") total_added = total_scanned = slices = 0 last = {} + mapping = None try: while True: L.update_marker(data_ref_recovery_running=time.time()) # No Ghidra import here on purpose (this module keeps Ghidra lazy): # recover_data_refs_core constructs its own ConsoleTaskMonitor when passed None, so the # whole wrapper — including the skip and failure paths — stays stdlib-only. - with L.open_target(artifact, cold_analyze=False) as (program, _flat, _cached): + mapping_out = {} + with L.open_target( + artifact, cold_analyze=False, mapping_out=mapping_out) as (program, _flat, _cached): + mapping = mapping_out.get("address_mapping") last = L.recover_data_refs_core(program, None, start_after=through) slices += 1 total_added += last.get("added") or 0 @@ -345,12 +360,13 @@ def _recover_data_refs(artifact) -> dict: except Exception as exc: # noqa: BLE001 - recovery must never fail the analysis that preceded it with contextlib.suppress(Exception): L.update_marker(data_ref_recovery_running=None) - return {"error": str(exc), "added": total_added, "slices": slices} + return {"error": str(exc), "added": total_added, "slices": slices, + "address_mapping": mapping} return {"added": total_added, "scanned": total_scanned, "slices": slices, "seconds": round(time.time() - started, 1), "complete": not last.get("truncated", False), "through": through if last.get("truncated") else None, - "saved": last.get("saved", False)} + "saved": last.get("saved", False), "address_mapping": mapping} def main() -> int: @@ -393,6 +409,15 @@ def main() -> int: L.start() print(json.dumps(_run(m))) return 0 + except L.AddressMappingMismatch as exc: + # Cold imports are normalized before this check. A mismatch here means the newly imported + # program could not be put in the promised coordinate system; warm legacy bases are instead + # preserved and returned with an explicit warning by open_target. + tool = "ghidra_script" if mode == "script" else "ghidra_probe" + print(json.dumps({"error": str(exc), "analysis_invalid": True, "needs_analysis": True, + "address_mapping": exc.report, "functions": [], "focus": None, + "tool": tool})) + return 0 except Exception as exc: # noqa: BLE001 — always emit a structured payload, never nothing tool = "ghidra_script" if mode == "script" else "ghidra_probe" print(json.dumps({"error": f"ghidra probe failed: {exc}", "tb": traceback.format_exc(), diff --git a/src/hexgraph/sandbox/probes/pyghidra_lib.py b/src/hexgraph/sandbox/probes/pyghidra_lib.py index b2b09ec4..dea72692 100644 --- a/src/hexgraph/sandbox/probes/pyghidra_lib.py +++ b/src/hexgraph/sandbox/probes/pyghidra_lib.py @@ -19,6 +19,7 @@ import os import re import shutil +import struct import time SCRATCH = os.environ.get("TMPDIR", "/scratch") @@ -27,6 +28,36 @@ META_NAME = "meta.json" # the committed warm marker (matches ghidra_probe) _ADDR = re.compile(r"^0x[0-9a-fA-F]+$") +# Versioned contract for the ELF PT_LOAD <-> Ghidra address-space check recorded in meta.json and +# returned by every headless result. Bump when the validation semantics change so old markers are +# informational only; every warm open re-validates against the live Program before it is yielded. +ADDRESS_MAPPING_SCHEMA = 1 + + +class AddressMappingMismatch(RuntimeError): + """A newly imported Ghidra Program does not map the ELF at its declared virtual addresses. + + Warm projects are deliberately accepted at their existing base with a warning; this error is + reserved for a cold import that cannot satisfy the new coordinate contract. + """ + + def __init__(self, report): + self.report = report + details = [] + if report.get("unexpected_image_base"): + details.append( + f"image base {report.get('image_base')} instead of " + f"{report.get('preferred_image_base')}" + ) + missing = ", ".join(report.get("unmapped_addresses") or []) + if missing: + details.append(f"unmapped {missing}") + super().__init__( + "Ghidra project address mapping does not match the ELF PT_LOAD virtual-address map " + f"({'; '.join(details) or 'unknown mismatch'}). The new analysis cannot be committed; " + "check the ELF loader result before retrying." + ) + # Above this size a COLD import runs the "fast profile": disable the auto-analysis passes proven # pathological on a 100 MB+ monolith (see _slow) so recon still gets functions/call-graph/strings # but the O(n^2) / decompile-every-function passes don't grind for tens of minutes. Smaller binaries @@ -82,7 +113,7 @@ def _is_warm(proj_dir: str) -> bool: return os.path.isdir(proj_dir) and bool(os.listdir(proj_dir)) -def _commit_marker() -> None: +def _commit_marker(**fields) -> None: """Commit the warm marker atomically — the LAST step of a successful cold analyze. MERGES, like `update_marker`. A cold analyze normally runs after `_clear_partial` has removed @@ -91,7 +122,7 @@ def _commit_marker() -> None: completed or partially-completed pass and make the next run redo tens of minutes of work.""" marker = os.path.join(PROJECT_MOUNT, META_NAME) data = read_marker() - data.update({"program_name": PROJECT_NAME, "created_at": time.time()}) + data.update({"program_name": PROJECT_NAME, "created_at": time.time(), **fields}) tmp = marker + ".tmp" try: with open(tmp, "w") as fh: @@ -129,6 +160,292 @@ def update_marker(**fields) -> None: pass +def _hex(value): + """Stable JSON rendering for an address/offset, including a negative mapping delta.""" + value = int(value) + return ("-0x" + format(-value, "x")) if value < 0 else ("0x" + format(value, "x")) + + +def _elf_load_map(artifact): + """Parse the bounded ELF header + PT_LOAD table from `artifact` inside the sandbox. + + Returns None for a non-ELF (the cache still works, but no ELF coordinate contract applies). + Malformed/extended layouts return an `unverified` report instead of guessing. No target bytes + are parsed on the host, and only the fixed-size headers are read here. + """ + try: + file_size = os.path.getsize(artifact) + with open(artifact, "rb") as fh: + ident = fh.read(16) + if len(ident) != 16 or ident[:4] != b"\x7fELF": + return None + elf_class, elf_data = ident[4], ident[5] + if elf_class not in (1, 2) or elf_data not in (1, 2): + raise ValueError("unsupported ELF class or byte order") + endian = "<" if elf_data == 1 else ">" + hdr_fmt = endian + ("HHIIIIIHHHHHH" if elf_class == 1 else "HHIQQQIHHHHHH") + hdr_size = struct.calcsize(hdr_fmt) + raw = fh.read(hdr_size) + if len(raw) != hdr_size: + raise ValueError("truncated ELF header") + header = struct.unpack(hdr_fmt, raw) + entry = header[3] + phoff = header[4] + phentsize = header[8] + phnum = header[9] + # PN_XNUM stores the real count in section-header zero. It is uncommon and requires a + # second parser; fail explicitly instead of treating 65,535 hostile headers as real. + if phnum == 0xFFFF: + raise ValueError("extended ELF program-header counts are not supported") + if phnum > 4096: + raise ValueError("ELF program-header count exceeds validation limit") + ph_fmt = endian + ("IIIIIIII" if elf_class == 1 else "IIQQQQQQ") + ph_size = struct.calcsize(ph_fmt) + if phnum and (phentsize < ph_size or phoff + phentsize * phnum > file_size): + raise ValueError("ELF program-header table is out of bounds") + segments = [] + for i in range(phnum): + fh.seek(phoff + i * phentsize) + raw = fh.read(ph_size) + if len(raw) != ph_size: + raise ValueError("truncated ELF program header") + values = struct.unpack(ph_fmt, raw) + if elf_class == 1: + p_type, p_offset, p_vaddr, _paddr, p_filesz, p_memsz, p_flags, p_align = values + else: + p_type, p_flags, p_offset, p_vaddr, _paddr, p_filesz, p_memsz, p_align = values + if p_type != 1: # PT_LOAD + continue + if p_offset + p_filesz > file_size or p_filesz > p_memsz: + raise ValueError("ELF PT_LOAD range is out of bounds") + segments.append({ + "file_offset": p_offset, + "virtual_address": p_vaddr, + "file_size": p_filesz, + "memory_size": p_memsz, + "flags": p_flags, + "alignment": p_align, + }) + if not segments: + raise ValueError("ELF has no PT_LOAD segments") + return { + "elf_type": header[0], + "entry_point": entry, + "preferred_image_base": min(s["virtual_address"] for s in segments), + "segments": segments, + } + except (OSError, ValueError, struct.error) as exc: + return { + "schema": ADDRESS_MAPPING_SCHEMA, + "format": "elf", + "status": "unverified", + "reason": str(exc), + "coordinate_system": "elf_virtual_address", + } + + +def _address_offset(address): + """Convert a Ghidra Address to a non-negative Python integer.""" + try: + return int(address.getUnsignedOffset()) + except Exception: # noqa: BLE001 — older Ghidra Address implementations expose getOffset only + value = int(address.getOffset()) + if value >= 0: + return value + try: + bits = int(address.getAddressSpace().getSize()) + except Exception: # noqa: BLE001 + bits = 64 + return value + (1 << bits) + + +def _mapping_segment_json(segment, load_bias=0): + flags = int(segment["flags"]) + result = { + "file_offset": _hex(segment["file_offset"]), + "virtual_address": _hex(segment["virtual_address"]), + "file_size": _hex(segment["file_size"]), + "memory_size": _hex(segment["memory_size"]), + "flags": ("R" if flags & 4 else "-") + ("W" if flags & 2 else "-") + + ("X" if flags & 1 else "-"), + "file_to_virtual_delta": _hex(segment["virtual_address"] - segment["file_offset"]), + } + if load_bias: + result["ghidra_address"] = _hex(segment["virtual_address"] + load_bias) + return result + + +def validate_address_mapping(program, artifact, *, allow_existing_base=False): + """Validate a live Ghidra Program against the ELF's declared virtual-address ranges. + + Ghidra addresses are ELF *virtual addresses*, while debugger/module reports sometimes use raw + file offsets. Those differ whenever a PT_LOAD has `p_vaddr != p_offset`. Separately, Ghidra's + ELF loader gives PIEs a default non-zero image base (commonly +0x100000); HexGraph normalizes + that cold import back to the ELF's preferred base so every backend uses one coordinate system. + Cold imports require the canonical image base plus every non-empty PT_LOAD's start, file-backed + end, memory end, and ELF entry point at its declared virtual address. Warm projects may predate + that policy: with `allow_existing_base`, their existing load bias is validated and reported as + a warning rather than forcing an expensive rebuild. Unsupported/non-ELF formats are reported + honestly and not guessed at. + """ + parsed = _elf_load_map(artifact) + if parsed is None: + return { + "schema": ADDRESS_MAPPING_SCHEMA, + "format": "other", + "status": "not_applicable", + "coordinate_system": "ghidra_program_address", + "image_base": _hex(_address_offset(program.getImageBase())), + } + if parsed.get("status") == "unverified": + image_base = _hex(_address_offset(program.getImageBase())) + parsed["image_base"] = image_base + if allow_existing_base: + parsed["coordinate_system"] = "ghidra_program_address" + parsed["warning"] = ( + f"Warm Ghidra project uses image base {image_base}; its ELF address mapping " + f"could not be verified ({parsed.get('reason', 'unknown reason')})." + ) + return parsed + + image_base = _address_offset(program.getImageBase()) + preferred_base = parsed["preferred_image_base"] + load_bias = image_base - preferred_base + memory = program.getMemory() + space = program.getAddressFactory().getDefaultAddressSpace() + missing = [] + + def _mapped(value): + try: + return memory.getBlock(space.getAddress(int(value))) is not None + except Exception: # noqa: BLE001 — an out-of-space address is definitively not mapped + return False + + accepted_bias = load_bias if allow_existing_base else 0 + for segment in parsed["segments"]: + start = segment["virtual_address"] + accepted_bias + if not segment["memory_size"]: + continue + anchors = [start] + if segment["file_size"]: + anchors.append(start + segment["file_size"] - 1) + if segment["memory_size"]: + anchors.append(start + segment["memory_size"] - 1) + for anchor in dict.fromkeys(anchors): + if not _mapped(anchor): + missing.append(_hex(anchor)) + entry = parsed["entry_point"] + accepted_bias if parsed["entry_point"] else 0 + if entry and not _mapped(entry): + missing.append(_hex(entry)) + + report = { + "schema": ADDRESS_MAPPING_SCHEMA, + "format": "elf", + "status": "mismatch" if missing else "validated", + "coordinate_system": ("elf_virtual_address_plus_load_bias" + if accepted_bias else "elf_virtual_address"), + "image_base": _hex(image_base), + "preferred_image_base": _hex(preferred_base), + "ghidra_load_bias": _hex(load_bias), + "entry_point": _hex(parsed["entry_point"]), + "ghidra_entry_point": _hex(entry), + "load_segments": [_mapping_segment_json(s, accepted_bias) for s in parsed["segments"]], + } + if load_bias and not allow_existing_base: + report["status"] = "mismatch" + report["unexpected_image_base"] = True + if allow_existing_base and load_bias: + report["legacy_image_base"] = True + report["warning"] = ( + f"Warm Ghidra project uses image base {_hex(image_base)} " + f"(ELF preferred base {_hex(preferred_base)}; load bias {_hex(load_bias)}). " + "Ghidra addresses include this bias; new PIE imports use their ELF preferred base " + "(normally 0x0)." + ) + if missing: + report["status"] = "warning" if allow_existing_base else "mismatch" + detail = (" Existing project address ranges could not be fully verified." + if allow_existing_base else "") + base_note = ( + f"Warm Ghidra project uses image base {_hex(image_base)} " + f"(ELF preferred base {_hex(preferred_base)}; load bias {_hex(load_bias)})." + ) + report["warning"] = report.get("warning", base_note) + detail + if missing or (load_bias and not allow_existing_base): + report["unmapped_addresses"] = list(dict.fromkeys(missing))[:32] + if missing and allow_existing_base: + return report + if missing or (load_bias and not allow_existing_base): + raise AddressMappingMismatch(report) + return report + + +def normalize_program_image_base(program, artifact): + """Rebase a newly imported ELF Program to its preferred PT_LOAD base before analysis. + + Ghidra 12.x defaults PIE imports to 0x100000 even when the ELF's lowest PT_LOAD begins at zero. + That makes an exact address from the ELF, radare2, or a debugger resolve one megabyte away in + Ghidra. Cold imports are writable, so normalize once before auto-analysis and persist the shared + coordinate system. Warm projects are never mutated or rejected here; their existing image base + is reported with every result so expensive legacy analyses remain usable. + """ + parsed = _elf_load_map(artifact) + if (parsed is None or parsed.get("status") == "unverified" + or parsed.get("elf_type") != 3): # ET_DYN (PIE/shared object) + return + preferred = int(parsed["preferred_image_base"]) + current = _address_offset(program.getImageBase()) + if current == preferred: + return + space = program.getAddressFactory().getDefaultAddressSpace() + txid = program.startTransaction("hexgraph normalize ELF image base") + commit = False + try: + program.setImageBase(space.getAddress(preferred), True) + commit = True + finally: + program.endTransaction(txid, commit) + + +def _mapping_value(value): + """Normalize an int/hex string/Ghidra Address passed to a re_script mapping helper.""" + if hasattr(value, "getOffset"): + return _address_offset(value) + return int(str(value), 0) if isinstance(value, str) else int(value) + + +def mapping_file_offset_to_vaddr(mapping, value): + """Translate an ELF file offset into the reported Ghidra Program coordinate system.""" + offset = _mapping_value(value) + for segment in (mapping or {}).get("load_segments") or []: + start = int(segment["file_offset"], 0) + size = int(segment["file_size"], 0) + if start <= offset < start + size: + program_start = segment.get("ghidra_address", segment["virtual_address"]) + return int(program_start, 0) + offset - start + return None + + +def mapping_vaddr_to_file_offset(mapping, value): + """Translate a reported Ghidra Program address to an ELF file offset.""" + address = _mapping_value(value) + for segment in (mapping or {}).get("load_segments") or []: + start = int(segment.get("ghidra_address", segment["virtual_address"]), 0) + size = int(segment["file_size"], 0) + if start <= address < start + size: + return int(segment["file_offset"], 0) + address - start + return None + + +def _validate_warm_address_mapping(program, artifact, mapping_out=None): + """Publish a warm mapping without mutating or rejecting an expensive existing project.""" + address_mapping = validate_address_mapping(program, artifact, allow_existing_base=True) + update_marker(address_mapping=address_mapping) + if mapping_out is not None: + mapping_out["address_mapping"] = address_mapping + return address_mapping + + @contextlib.contextmanager def _read_only_program(project, prog_name): """Open the warm program IMMUTABLE via DomainFile.getReadOnlyDomainObject — the returned @@ -162,7 +479,7 @@ def _read_only_program(project, prog_name): @contextlib.contextmanager -def open_target(artifact, *, cold_analyze=True, read_only=False): +def open_target(artifact, *, cold_analyze=True, read_only=False, mapping_out=None): """Yield `(program, flat, cached)` for the target. WARM (a committed slot at PROJECT_MOUNT): open the resident project + program, NO re-analysis. COLD: import + analyze; persist into the slot (+ commit the marker) when the mount is present, else a throwaway /scratch project. @@ -170,7 +487,12 @@ def open_target(artifact, *, cold_analyze=True, read_only=False): `read_only` (re_script) opens the WARM program IMMUTABLE (getReadOnlyDomainObject) so an agent-supplied script can query but never write the persistent project; it also forces warm-only (no cold analysis for a read-only query) — a cold miss raises so the probe returns the re_analyze - lead. `cached` is True on the warm path. The Program is closed / the project released on exit.""" + lead. `cached` is True on the warm path. Before ANY warm Program is yielded, its address space + is checked against the artifact's ELF PT_LOAD virtual-address map. Legacy image bases are kept + intact and returned as warnings rather than invalidating a costly project. `mapping_out`, when + supplied, receives the JSON-safe validation report without changing + the long-standing `(program, flat, cached)` tuple. The Program is closed / project released on + exit.""" import pyghidra from ghidra.program.flatapi import FlatProgramAPI @@ -182,9 +504,11 @@ def open_target(artifact, *, cold_analyze=True, read_only=False): try: if read_only: with _read_only_program(project, prog_name) as program: + _validate_warm_address_mapping(program, artifact, mapping_out) yield program, FlatProgramAPI(program), True else: with pyghidra.program_context(project, "/" + prog_name) as program: + _validate_warm_address_mapping(program, artifact, mapping_out) yield program, FlatProgramAPI(program), True finally: with contextlib.suppress(Exception): @@ -216,10 +540,14 @@ def open_target(artifact, *, cold_analyze=True, read_only=False): program_name=prog_name, analyze=False, nested_project_location=False) as flat: program = flat.getCurrentProgram() + normalize_program_image_base(program, artifact) _analyze(program, artifact) + address_mapping = validate_address_mapping(program, artifact) + if mapping_out is not None: + mapping_out["address_mapping"] = address_mapping yield program, flat, False if persist: - _commit_marker() + _commit_marker(address_mapping=address_mapping) def _slow_analyzer(name: str) -> bool: @@ -1208,7 +1536,7 @@ def _sweep(names): # script can query the warm project but never mutate/corrupt it. Delivery is unchanged from the # Jython path (HEXGRAPH_USER_SCRIPT_B64 → the probe → here); only the runtime moved in-process. -def script_core(program, flat, monitor, user_script, *, out_path=None) -> dict: +def script_core(program, flat, monitor, user_script, *, out_path=None, address_mapping=None) -> dict: """Run the agent-supplied Python-3 `user_script` body against the RESIDENT (read-only) program and return its JSON result. The script's namespace exposes: @@ -1218,6 +1546,9 @@ def script_core(program, flat, monitor, user_script, *, out_path=None) -> dict: out_path — a scratch file path; write your JSON result there (the built-in-core convention) getScriptArgs() — a shim returning [out_path] (so a script written to the postScript `out_path = getScriptArgs()[0]` contract keeps working unchanged) + address_mapping — the reported ELF PT_LOAD / Ghidra coordinate report + file_offset_to_address(value) — translate an ELF file offset to a Ghidra Address + address_to_file_offset(value) — translate a Ghidra Address/vaddr to an ELF file offset result — OR: assign a JSON-serializable object to `result` instead of writing out_path The result is taken from an explicit `result` binding if the script set one, else parsed from @@ -1239,6 +1570,18 @@ def _get_script_args(): # Back-compat shim for scripts written to the Jython postScript contract (out_path = args[0]). return [out_path] + def _file_offset_to_address(value): + mapped = mapping_file_offset_to_vaddr(address_mapping, value) + if mapped is None: + raise ValueError(f"ELF file offset {_hex(_mapping_value(value))} is not file-backed") + return flat.toAddr(mapped) + + def _address_to_file_offset(value): + mapped = mapping_vaddr_to_file_offset(address_mapping, value) + if mapped is None: + raise ValueError(f"virtual address {_hex(_mapping_value(value))} is not file-backed") + return mapped + ns = { "__name__": "hexgraph_user_script", "__builtins__": __builtins__, @@ -1248,6 +1591,9 @@ def _get_script_args(): "monitor": monitor, "out_path": out_path, "getScriptArgs": _get_script_args, + "address_mapping": address_mapping, + "file_offset_to_address": _file_offset_to_address, + "address_to_file_offset": _address_to_file_offset, "result": None, } try: @@ -1427,7 +1773,7 @@ def bridge_dispatch(program, flat, monitor, req) -> dict: return {"error": f"bridge op {op} failed: {exc}", "tb": traceback.format_exc()} -def _serve_one(conn, program, flat, make_monitor) -> None: +def _serve_one(conn, program, flat, make_monitor, address_mapping=None) -> None: """Handle one connection: read one JSON request line, dispatch, write one JSON response line. Mints a FRESH TaskMonitor for THIS request via `make_monitor()`. A resident bridge serves many @@ -1448,11 +1794,15 @@ def _serve_one(conn, program, flat, make_monitor) -> None: resp = {"error": "bad request json"} else: resp = bridge_dispatch(program, flat, make_monitor(), req) + if address_mapping is not None and isinstance(resp, dict): + resp.setdefault("address_mapping", address_mapping) + if address_mapping.get("warning"): + resp.setdefault("warning", address_mapping["warning"]) fh.write((json.dumps(resp) + "\n").encode("utf-8")) fh.flush() -def serve_bridge(host, port, program, flat, make_monitor) -> None: +def serve_bridge(host, port, program, flat, make_monitor, address_mapping=None) -> None: """Block forever serving line-delimited JSON bridge requests over TCP against the RESIDENT (program, flat). Single-threaded — Ghidra program access is NOT concurrency-safe, so one connection + one request at a time (two host processes serialize). The caller binds AFTER the @@ -1472,7 +1822,7 @@ def serve_bridge(host, port, program, flat, make_monitor) -> None: except OSError: continue try: - _serve_one(conn, program, flat, make_monitor) + _serve_one(conn, program, flat, make_monitor, address_mapping) except Exception: # noqa: BLE001 — never let one connection kill the resident loop pass finally: diff --git a/tests/test_decompiler_fallback.py b/tests/test_decompiler_fallback.py index 95299a10..f466b962 100644 --- a/tests/test_decompiler_fallback.py +++ b/tests/test_decompiler_fallback.py @@ -91,6 +91,19 @@ def test_format_decomp_warns_on_fallback_engine(): assert "FALLBACK" not in _format_decomp(normal, "main") +def test_format_decomp_surfaces_preserved_warm_image_base_warning(): + out = { + "focus": {"name": "main", "address": "0x101000", "pseudocode": "return 0;"}, + "warning": "Warm Ghidra project uses image base 0x100000.", + "address_mapping": {"image_base": "0x100000", "ghidra_load_bias": "0x100000"}, + } + + text = _format_decomp(out, "main") + + assert text.startswith("// WARNING: Warm Ghidra project uses image base 0x100000.") + assert "// main @ 0x101000" in text + + def test_format_decomp_surfaces_promoted_node_id_for_mention(): """F11: _format_decomp renders the promoted node id in the header (truncation-safe) in the journal @-mention syntax, so a just-decompiled function is mention-able without a lookup.""" diff --git a/tests/test_ghidra_address_mapping.py b/tests/test_ghidra_address_mapping.py new file mode 100644 index 00000000..17e12f07 --- /dev/null +++ b/tests/test_ghidra_address_mapping.py @@ -0,0 +1,273 @@ +"""ELF PT_LOAD <-> Ghidra address-space validation and re_script translation helpers. + +No JVM/Docker: tiny fake Program/Memory objects exercise the same API calls the sandbox probe uses. +""" + +from __future__ import annotations + +import json +import os +import struct + +import pytest + +from hexgraph.sandbox.probes import pyghidra_lib as L + + +def _elf64(path, *, elf_type=3): + """Two-load ELF whose executable file offset -> vaddr delta is exactly +0x100000.""" + ident = b"\x7fELF" + bytes([2, 1, 1, 0]) + b"\0" * 8 + ehsize = 64 + phentsize = 56 + phnum = 2 + entry = 0x100100 + header = struct.pack( + " Date: Sat, 15 Aug 2026 10:57:16 -0400 Subject: [PATCH 2/2] fix: preserve Ghidra address provenance Addresses review comments r3789589252, r3789589270, and r3789589291 on PR #310. --- src/hexgraph/agent/agent_tools.py | 5 ++++- src/hexgraph/engine/re/ghidra_bridge.py | 7 +++++++ src/hexgraph/sandbox/decompiler.py | 10 ++++++++- src/hexgraph/sandbox/probes/pyghidra_lib.py | 21 ++++++++++++++++++- tests/test_decompiler_fallback.py | 15 ++++++++++++-- tests/test_ghidra_address_mapping.py | 23 +++++++++++++++++++++ tests/test_pyghidra_bridge.py | 7 ++++++- 7 files changed, 82 insertions(+), 6 deletions(-) diff --git a/src/hexgraph/agent/agent_tools.py b/src/hexgraph/agent/agent_tools.py index d1ed2d09..663f5aa1 100644 --- a/src/hexgraph/agent/agent_tools.py +++ b/src/hexgraph/agent/agent_tools.py @@ -605,7 +605,10 @@ def _format_decomp(out: dict, label: str, *, limit: int = _MAX) -> str: Truncates the body to `limit` chars with the ACTIONABLE marker (recovery knobs + sizes) rather than a bare `…[truncated]`, so a head-truncation can't silently hide a tail sink — the marker names both the max_chars re-call and the get_observation full-body path.""" - address_warning = _address_mapping_warning(out) + # A missing Ghidra focus can be replaced by radare2. That focus/address does not use the warm + # Ghidra project's legacy bias, so attaching its mapping warning to this focused result would + # describe the wrong coordinate system. The fallback-engine warning below remains explicit. + address_warning = "" if out.get("focus_fallback") else _address_mapping_warning(out) focus = out.get("focus") if not focus: defined = out.get("functions", []) or [] diff --git a/src/hexgraph/engine/re/ghidra_bridge.py b/src/hexgraph/engine/re/ghidra_bridge.py index 910e7812..c0888922 100644 --- a/src/hexgraph/engine/re/ghidra_bridge.py +++ b/src/hexgraph/engine/re/ghidra_bridge.py @@ -228,6 +228,13 @@ def decompile(self, program: str | None, function: str | None) -> dict: "functions_total": resp.get("functions_total"), "focus": focus, "calls": resp.get("calls") or [], "structs": resp.get("structs") or [], "tool": "ghidra_bridge"} + mapping = resp.get("address_mapping") + if isinstance(mapping, dict): + out["address_mapping"] = mapping + warning = resp.get("warning") or ( + mapping.get("warning") if isinstance(mapping, dict) else None) + if warning: + out["warning"] = warning if resp.get("error"): out["error"] = resp["error"] return out diff --git a/src/hexgraph/sandbox/decompiler.py b/src/hexgraph/sandbox/decompiler.py index 6a6b0689..498fe2fc 100644 --- a/src/hexgraph/sandbox/decompiler.py +++ b/src/hexgraph/sandbox/decompiler.py @@ -47,7 +47,15 @@ def focus_only_payload(out: dict) -> dict: payload = {"functions": fns[:_FOCUS_PAYLOAD_FUNCTION_SAMPLE], "functions_total": ftotal, "focus": (out or {}).get("focus")} - if isinstance((out or {}).get("address_mapping"), dict): + if (out or {}).get("focus_engine"): + payload["focus_engine"] = out["focus_engine"] + if (out or {}).get("focus_fallback"): + payload["focus_fallback"] = True + # A missed Ghidra focus can be replaced with a radare2 focus. Its address is in radare2's + # normalized virtual-address coordinate, not a preserved warm Ghidra project's biased space; + # never fingerprint that hybrid payload with the Ghidra mapping. + if (not (out or {}).get("focus_fallback") + and isinstance((out or {}).get("address_mapping"), dict)): payload["address_mapping"] = out["address_mapping"] return payload diff --git a/src/hexgraph/sandbox/probes/pyghidra_lib.py b/src/hexgraph/sandbox/probes/pyghidra_lib.py index dea72692..85590fbd 100644 --- a/src/hexgraph/sandbox/probes/pyghidra_lib.py +++ b/src/hexgraph/sandbox/probes/pyghidra_lib.py @@ -52,6 +52,8 @@ def __init__(self, report): missing = ", ".join(report.get("unmapped_addresses") or []) if missing: details.append(f"unmapped {missing}") + if report.get("reason"): + details.append(f"unverified ELF mapping: {report['reason']}") super().__init__( "Ghidra project address mapping does not match the ELF PT_LOAD virtual-address map " f"({'; '.join(details) or 'unknown mismatch'}). The new analysis cannot be committed; " @@ -226,6 +228,15 @@ def _elf_load_map(artifact): "flags": p_flags, "alignment": p_align, }) + if not segments and header[0] == 1: # ET_REL has no runtime PT_LOAD coordinate system + return { + "schema": ADDRESS_MAPPING_SCHEMA, + "format": "elf", + "elf_type": header[0], + "status": "not_applicable", + "coordinate_system": "ghidra_program_address", + "reason": "relocatable ELF has no PT_LOAD segments", + } if not segments: raise ValueError("ELF has no PT_LOAD segments") return { @@ -297,6 +308,9 @@ def validate_address_mapping(program, artifact, *, allow_existing_base=False): "coordinate_system": "ghidra_program_address", "image_base": _hex(_address_offset(program.getImageBase())), } + if parsed.get("status") == "not_applicable": + parsed["image_base"] = _hex(_address_offset(program.getImageBase())) + return parsed if parsed.get("status") == "unverified": image_base = _hex(_address_offset(program.getImageBase())) parsed["image_base"] = image_base @@ -306,7 +320,12 @@ def validate_address_mapping(program, artifact, *, allow_existing_base=False): f"Warm Ghidra project uses image base {image_base}; its ELF address mapping " f"could not be verified ({parsed.get('reason', 'unknown reason')})." ) - return parsed + return parsed + # A cold import has no durable analysis to preserve. Committing one whose ELF mapping this + # bounded parser could not verify would silently keep Ghidra's loader-selected base while + # promising the normalized coordinate contract. Warm projects take the warn-and-preserve + # path above; new projects fail closed and leave no committed marker. + raise AddressMappingMismatch(parsed) image_base = _address_offset(program.getImageBase()) preferred_base = parsed["preferred_image_base"] diff --git a/tests/test_decompiler_fallback.py b/tests/test_decompiler_fallback.py index f466b962..d3e7a4ac 100644 --- a/tests/test_decompiler_fallback.py +++ b/tests/test_decompiler_fallback.py @@ -55,7 +55,12 @@ def test_ghidra_falls_back_to_r2_for_explicit_focus_it_missed(monkeypatch): g = GhidraDecompiler.__new__(GhidraDecompiler) g.runner = object() # the fallback constructs its own R2Decompiler; never touched ghidra_out = {"functions": ["main", "helper"], "focus": None, - "calls": [["main", "helper"]], "structs": [{"name": "cfg_t"}]} + "calls": [["main", "helper"]], "structs": [{"name": "cfg_t"}], + "address_mapping": {"schema": 1, "image_base": "0x100000", + "preferred_image_base": "0x0", + "ghidra_load_bias": "0x100000", + "warning": "legacy warm mapping"}, + "warning": "legacy warm mapping"} monkeypatch.setattr(GhidraDecompiler, "_decompile_ghidra", lambda self, *a, **k: dict(ghidra_out)) r2_calls = [] @@ -75,6 +80,9 @@ def fake_r2_decompile(self, artifact, function=None, **kw): # it and the pseudocode isn't Ghidra-quality (r2dec can mis-resolve PLT/args / fabricate a call). assert out["focus_engine"] == "radare2" assert out["focus_fallback"] is True + payload = focus_only_payload(out) + assert payload["focus_engine"] == "radare2" and payload["focus_fallback"] is True + assert "address_mapping" not in payload def test_format_decomp_warns_on_fallback_engine(): @@ -82,9 +90,12 @@ def test_format_decomp_warns_on_fallback_engine(): can't silently read r2dec pseudocode as Ghidra-quality (the dogfood chased a fabricated call).""" fallback = {"focus": {"name": "cgi_handler", "address": "0x401200", "pseudocode": "void cgi_handler(){ system(x); }", "callees": []}, - "focus_engine": "radare2", "focus_fallback": True} + "focus_engine": "radare2", "focus_fallback": True, + "warning": "legacy Ghidra mapping", + "address_mapping": {"warning": "legacy Ghidra mapping"}} text = _format_decomp(fallback, "cgi_handler") assert "FALLBACK DECOMPILER" in text and "radare2" in text + assert "legacy Ghidra mapping" not in text assert text.index("FALLBACK") < text.index("system(x)") # warning precedes the body # a normal Ghidra focus carries NO warning normal = {"focus": {"name": "main", "pseudocode": "int main(){}", "callees": []}} diff --git a/tests/test_ghidra_address_mapping.py b/tests/test_ghidra_address_mapping.py index 17e12f07..6f2c671b 100644 --- a/tests/test_ghidra_address_mapping.py +++ b/tests/test_ghidra_address_mapping.py @@ -234,6 +234,18 @@ def test_non_elf_mapping_is_reported_without_guessing(tmp_path): } +def test_relocatable_elf_without_load_segments_is_not_rejected(tmp_path): + artifact = _elf64(tmp_path / "relocatable", elf_type=1) + data = bytearray(artifact.read_bytes()) + data[56:58] = b"\0\0" # ELF64 e_phnum + artifact.write_bytes(data) + + report = L.validate_address_mapping(_Program([], image_base=0), artifact) + + assert report["status"] == "not_applicable" + assert report["coordinate_system"] == "ghidra_program_address" + + def test_unparseable_warm_elf_reports_its_base_as_a_warning(tmp_path): artifact = tmp_path / "broken-elf" artifact.write_bytes(b"\x7fELF" + bytes([2, 1]) + b"\0" * 10) @@ -246,6 +258,17 @@ def test_unparseable_warm_elf_reports_its_base_as_a_warning(tmp_path): assert "image base 0x100000" in report["warning"] +def test_unparseable_cold_elf_is_rejected_before_commit(tmp_path): + artifact = tmp_path / "broken-elf" + artifact.write_bytes(b"\x7fELF" + bytes([2, 1]) + b"\0" * 10) + + with pytest.raises(L.AddressMappingMismatch) as raised: + L.validate_address_mapping(_Program([], image_base=0x100000), artifact) + + assert raised.value.report["status"] == "unverified" + assert "truncated ELF header" in str(raised.value) + + def test_probe_surfaces_cold_mapping_mismatch(tmp_path, monkeypatch, capsys): from hexgraph.sandbox.probes import ghidra_probe as G diff --git a/tests/test_pyghidra_bridge.py b/tests/test_pyghidra_bridge.py index 568cdb7e..1298e101 100644 --- a/tests/test_pyghidra_bridge.py +++ b/tests/test_pyghidra_bridge.py @@ -397,13 +397,18 @@ def serve(): def test_managed_ops_decompile_round_trip(): from hexgraph.engine.re.ghidra_bridge import _ManagedOps + mapping = {"status": "validated", "image_base": "0x100000", + "warning": "Warm Ghidra project uses image base 0x100000."} port, captured, t = _one_shot_server( - {"functions": ["f1", "f2"], "focus": {"name": "f1"}, "tool": "ghidra_bridge"}) + {"functions": ["f1", "f2"], "focus": {"name": "f1"}, "tool": "ghidra_bridge", + "address_mapping": mapping, "warning": mapping["warning"]}) out = _ManagedOps("127.0.0.1", port).decompile(None, "f1") t.join(timeout=5) assert captured["req"] == {"op": "decompile", "focus": "f1"} # client sent the right request assert out["functions"] == ["f1", "f2"] and out["focus"] == {"name": "f1"} assert out["tool"] == "ghidra_bridge" + assert out["address_mapping"] == mapping + assert out["warning"] == mapping["warning"] def test_managed_ops_error_response_reads_as_no_focus():