fix: re_search_code byte/immediate scan no longer runs a whole-binary analysis (2547s timeout) (#269)

* fix: re_search_code byte/immediate scan no longer runs a whole-binary analysis (2547s timeout)

The byte/immediate scan shelled to xrefs_probe.py, which runs a whole-binary radare2 `aaa` before
EVERY mode — but a raw memory scan doesn't need analysis (the `aaa` existed only to map a hit to its
containing function), so on a ~480MB target it hit the ~2547s size-scaled timeout. Fix, tiered like
the re_xrefs routing: a warm Ghidra Memory.findBytes scan (pyghidra_lib.search_bytes_core; served
headless via `ghidra_probe --search` WARM-ONLY + a resident-bridge `search` op) routed through
sandbox.decompiler.ghidra_op_backend — reuses the warm project / a live bridge, each hit mapped to its
function; the r2 xrefs_probe fallback now SKIPS `aaa` for search mode so it's fast too. Also fixes the
read-only warm open (_read_only_program, shared with re_script), latently broken under jpype (a boxed
JInt DEFAULT_VERSION + a Python object() that can't bind java.lang.Object).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix: address PR #269 review nits — guard found.add() overflow, reject negative immediates

Wrap search_bytes_core's found.add(1) so an AddressOverflowException at the very top of the address
space stops the scan cleanly instead of raising out and dropping already-found hits. And _search_patterns
now rejects negative immediates to match the r2 fallback's _IMM regex (which has no leading '-'), so the
same `immediate` resolves identically on both backends — a two's-complement value is searchable via
bytes_pattern.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: branover

src/hexgraph/agent/agent_tools.py

@@ -2385,26 +2385,58 @@ def _search_code_grep(ctx: ToolContext, args: dict, *, query: str, functions) ->
     return _clip("\n".join(lines))
 
 
-def _search_code_scan(ctx: ToolContext, args: dict, *, bytes_pat, immediate) -> str:
-    """The byte/immediate scan: run the r2 `--mode search` probe (`/xj`//`/vj`) over the mapped
-    image and map each hit to its containing function, PAGINATED (total + next offset reported,
-    the no-silent-caps discipline). QUERY: records an Observation; adds no graph nodes."""
+def _ghidra_search(ctx: ToolContext, *, bytes_pat, immediate) -> dict | None:
+    """The WARM Ghidra memory scan (`GhidraDecompiler`/bridge `search_bytes`): `Memory.findBytes`
+    over the resident/warm image, each hit mapped to its containing function — a fast scan that
+    reuses the warm analysis, NOT the r2 whole-binary `aaa` sweep. Returns the probe dict on a
+    COMPLETED run, or None when Ghidra couldn't answer (not the active backend / no warm project /
+    a probe fault) so the caller falls back to r2. Best-effort: never raises into the caller."""
+    from hexgraph.sandbox.decompiler import ghidra_op_backend
+
+    try:
+        out = ghidra_op_backend(ctx.target).search_bytes(
+            ctx.target.path, bytes_pattern=bytes_pat,
+            immediate=(str(immediate) if immediate is not None else None), project=ctx.project)
+    except Exception:  # noqa: BLE001 — a warm-path failure DEGRADES to r2, never aborts the tool
+        return None
+    if not isinstance(out, dict) or out.get("error"):
+        return None
+    return out
+
+
+def _r2_search(ctx: ToolContext, *, bytes_pat, immediate):
+    """The radare2 raw-scan fallback (`xrefs_probe --mode search`, `/xj`//`/vj`) for when Ghidra
+    can't answer. The probe no longer runs a whole-binary `aaa` for a search, so it's fast (function
+    mapping best-effort from the symbol table). Returns the probe dict, or a string error message."""
     from hexgraph.sandbox.executor import get_executor
-    from hexgraph.sandbox.runner import docker_available
 
-    if not docker_available():
-        return "search_code byte/immediate scan unavailable (Docker/sandbox not running)"
-    extra = ["--mode", "search"]
-    if bytes_pat:
-        extra += ["--bytes", str(bytes_pat)]
-        subj = f"bytes {bytes_pat!r}"
-    else:
-        extra += ["--imm", str(immediate)]
-        subj = f"immediate {immediate!r}"
+    extra = ["--mode", "search"] + (["--bytes", str(bytes_pat)] if bytes_pat
+                                    else ["--imm", str(immediate)])
     try:
-        out = get_executor().run_json_probe("xrefs_probe.py", ctx.target.path, extra_args=extra)
+        return get_executor().run_json_probe("xrefs_probe.py", ctx.target.path, extra_args=extra)
     except Exception as exc:  # noqa: BLE001
         return f"search_code scan failed: {exc}"
+
+
+def _search_code_scan(ctx: ToolContext, args: dict, *, bytes_pat, immediate) -> str:
+    """The byte/immediate scan, PAGINATED (total + next offset reported, the no-silent-caps
+    discipline). Prefers the WARM Ghidra loaded-memory scan (the byte-scan analog of re_xrefs
+    consulting the warm reference index — a fast memory scan over the resident image, NOT the r2
+    whole-binary `aaa` sweep that times out on a large target); falls back to the r2 raw scan (also
+    fast now — no `aaa` for a search) when Ghidra can't answer. QUERY: records an Observation."""
+    from hexgraph.sandbox.runner import docker_available
+
+    if not docker_available():
+        return "search_code byte/immediate scan unavailable (Docker/sandbox not running)"
+    subj = f"bytes {bytes_pat!r}" if bytes_pat else f"immediate {immediate!r}"
+
+    # Warm Ghidra first (reuses the warm project / a live bridge); r2 raw scan otherwise.
+    out = _ghidra_search(ctx, bytes_pat=bytes_pat, immediate=immediate) \
+        if _ghidra_xrefs_active() else None
+    if out is None:
+        out = _r2_search(ctx, bytes_pat=bytes_pat, immediate=immediate)
+    if isinstance(out, str):  # a string error from the r2 fallback
+        return out
     if isinstance(out, dict) and out.get("error"):
         return f"search_code: {out['error']}"
 

src/hexgraph/agent/mcp_catalog.py

@@ -182,7 +182,7 @@
     ("read", "re_resolve", _t.resolve_address, "Triage a hex ADDRESS (a crash address, a pointer, a DAT_ label) WITHOUT a full decompile: returns {nearest_symbol + offset, section, containing_function (name+bounds when the symbol table knows it)}. Assembled server-side from the symbol table + section table (re_binutils_facts / pyelftools over the on-disk ELF) — cheap orientation before you spend a re_decompile_at. On a stripped binary it still resolves the section + nearest dynamic symbol (a FUN_ name needs a decompile). QUERY: records an Observation; adds no graph nodes.", {"type": "object", "properties": {"target_id": {"type": "string"}, "address": {"type": "string", "description": "hex address, e.g. 0x401200"}}, "required": ["target_id", "address"]}),
     ("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. Maps the vaddr to a file offset via the ELF program headers/sections and reads the on-disk artifact server-side (no decompile, no Docker). 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): a decompile-on-demand grep (`query`) over a BOUNDED candidate set you name in `functions` (so you control the cost — an unbounded whole-binary decompile is intentionally NOT offered), OR a BYTE/opcode pattern (`bytes_pattern`, hex pairs) / an IMMEDIATE constant (`immediate`) scanned across the mapped image (radare2 `/xj`//`/vj`, mapped to the containing function). To find CALLERS of a symbol/sink use re_xrefs (whole-program, indexed) — this does NOT duplicate it. Paginated. 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`)"}, "query": {"type": "string", "description": "substring to grep in the decompiled bodies of `functions`"}, "offset": {"type": "integer"}, "limit": {"type": "integer"}}, "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): a decompile-on-demand grep (`query`) over a BOUNDED candidate set you name in `functions` (so you control the cost — an unbounded whole-binary decompile is intentionally NOT offered), OR a BYTE/opcode pattern (`bytes_pattern`, hex pairs) / 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. To find CALLERS of a symbol/sink use re_xrefs (whole-program, indexed) — this does NOT duplicate it. Paginated. 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`)"}, "query": {"type": "string", "description": "substring to grep in the decompiled bodies of `functions`"}, "offset": {"type": "integer"}, "limit": {"type": "integer"}}, "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). 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: opt-in via features.ghidra.scripting (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"]}),

src/hexgraph/engine/re/ghidra_bridge.py

@@ -206,6 +206,9 @@ def run_emulate(self, function: str | None) -> dict:
     def rename_function(self, address: str, new_name: str) -> dict:
         return self._rpc({"op": "rename", "address": address, "new_name": new_name})
 
+    def search_bytes(self, bytes_pattern, immediate) -> dict:
+        return self._rpc({"op": "search", "bytes_pattern": bytes_pattern, "immediate": immediate})
+
 
 def connect_managed(host: str, port: int) -> BridgeOps:
     """Ops for HexGraph's managed resident bridge (custom JSON RPC). Unlike `connect_ops`, needs NO
@@ -265,6 +268,10 @@ def run_emulate(self, artifact: str, function: str, *, project=None) -> dict:
     def rename_function(self, artifact: str, *, address: str, new_name: str, project=None) -> dict:
         return self._managed().rename_function(address, new_name)
 
+    def search_bytes(self, artifact: str, *, bytes_pattern: str | None = None,
+                     immediate: str | None = None, project=None) -> dict:
+        return self._managed().search_bytes(bytes_pattern, immediate)
+
 
 def list_open_programs(ops: BridgeOps | None = None) -> list[dict]:
     return (ops or connect_ops()).list_programs()

src/hexgraph/sandbox/decompiler.py

@@ -317,6 +317,30 @@ def xrefs(self, artifact: str, *, mode: str, subject: str | None = None, project
                 return self.runner.run_json_probe("ghidra_probe.py", artifact, extra_args=args)
             return self._run_locked(slot, artifact, args)
 
+    def search_bytes(self, artifact: str, *, bytes_pattern: str | None = None,
+                     immediate: str | None = None, project=None) -> dict:
+        """Scan the warm project's LOADED MEMORY for a BYTE pattern / IMMEDIATE value (`--search`),
+        each hit mapped to its containing function. WARM-ONLY on purpose: a byte scan must NEVER
+        trigger a cold whole-binary analysis — that cold import/`aaa` IS the 2547s timeout the r2
+        xrefs_probe hits. Without a committed warm slot this returns an `error` so the caller degrades
+        to the r2 raw-scan path (fast, no analysis) instead of cold-importing here. Held under the
+        slot lock (read-only open, so the warm project is never mutated); emits the same JSON shape as
+        the r2 xrefs_probe `search` mode."""
+        slot = self._resolve_slot(artifact, project)
+        if slot is None or not slot.exists():
+            return {"tool": "ghidra_search", "mode": "search",
+                    "error": "no warm Ghidra analysis for this target (search is warm-only)"}
+        args = ["--search"]
+        if bytes_pattern:
+            args += ["--sbytes", str(bytes_pattern)]
+        elif immediate is not None:
+            args += ["--simm", str(immediate)]
+        with slot.lock() as locked:
+            if not locked:
+                return {"tool": "ghidra_search", "mode": "search",
+                        "error": "ghidra project busy (locked); try again"}
+            return self._run_locked(slot, artifact, args)
+
     def _run_locked(self, slot, artifact: str, args, *, force_cold: bool = False):
         """Run the probe with the slot held exclusively. Decides cold vs warm on the AUTHORITATIVE
         committed marker (`slot.exists()`); cleans a partially-written slot before a cold run; and

src/hexgraph/sandbox/probes/ghidra_probe.py

@@ -132,15 +132,30 @@ def _check() -> int:
     return 0
 
 
+def _flag_value(argv, flag):
+    """The value immediately following `flag` in argv, or None (for `--sbytes <hex>` / `--simm <v>`)."""
+    if flag in argv:
+        i = argv.index(flag)
+        if i + 1 < len(argv):
+            return argv[i + 1]
+    return None
+
+
 def _parse(argv):
     """Parse the probe argv into a mode dict. Mirrors the Jython arg grammar exactly."""
     artifact = argv[1]
     rest = argv[2:]
     m = {"artifact": artifact, "mode": "decompile", "focus": None,
-         "rename": None, "xrefs_mode": "sinks", "xrefs_subject": "", "user_script": None}
+         "rename": None, "xrefs_mode": "sinks", "xrefs_subject": "", "user_script": None,
+         "search_bytes": None, "search_imm": None}
     if "--script" in rest:
         m["mode"] = "script"  # run the agent-supplied script over the WARM program, read-only
         return m
+    if "--search" in rest:
+        m["mode"] = "search"  # byte/immediate memory scan over the WARM program, read-only
+        m["search_bytes"] = _flag_value(rest, "--sbytes")
+        m["search_imm"] = _flag_value(rest, "--simm")
+        return m
     if "--analyze" in rest:
         m["mode"] = "analyze"  # cold whole-binary analysis, no focus
         return m
@@ -177,12 +192,19 @@ def _run(m) -> dict:
     from ghidra.util.task import ConsoleTaskMonitor
 
     mode = m["mode"]
-    read_only = mode == "script"
+    # script + search are READ-ONLY, WARM-ONLY queries: they open the warm program immutable and
+    # never trigger a cold analysis (a byte scan that cold-analyzes a large target IS the timeout
+    # bug; the host falls back to the r2 raw scan on a warm miss instead).
+    read_only = mode in ("script", "search")
     with L.open_target(m["artifact"], cold_analyze=not read_only,
                        read_only=read_only) as (program, flat, cached):
         monitor = ConsoleTaskMonitor()
         if mode == "script":
             result = L.script_core(program, flat, monitor, m["user_script"])
+        elif mode == "search":
+            result = L.search_bytes_core(program, flat, monitor,
+                                         bytes_pattern=m.get("search_bytes"),
+                                         immediate=m.get("search_imm"))
         elif mode == "taint":
             result = L.taint_core(program, flat, monitor)
         elif mode == "emulate":