From d714b9641218da8bb8cc654d4ab93d1c83a1fe93 Mon Sep 17 00:00:00 2001 From: Eric Gustafson Date: Sun, 26 Jul 2026 17:01:07 +0000 Subject: [PATCH] Add `annotate` for bulk comments/renames, and say what to do about --raw Two remaining gaps from a CTF where DecLib was the sole decompiler interface. **Bulk annotation.** Annotation is produced in bulk -- you read a binary, build a mapping of address to label, then want to write it back. Applying that one `comment set` at a time is N round-trips, so people drove the Python client directly instead: one challenge passed `renames={0x5440: 'detect_branch_trampoline', ...}` straight to DecompilerClient, and another skipped DecLib entirely, opening the IDB with idapro and calling ida_funcs.set_func_cmt in a loop. decompiler annotate --file notes.json echo '{"0x401000": "parses the header"}' | decompiler annotate Accepts a list of {"addr", "comment"?, "name"?} records, or the shorthand {addr: comment} object -- the shape people already build by hand. Backed by `apply_annotations` on DecompilerInterface, which like decompile_many defaults to an in-process loop and so collapses N round-trips into one for every backend. Renames are verified by read-back rather than assumed: a backend can accept the write and quietly not apply it, and a silently miscounted rename is worse than a reported failure. Per-item failures are returned to the caller (address, field, reason) instead of only reaching the server log, and one bad record no longer discards the batch. Failure text is never blank. An exception raised without arguments has an empty str(), which surfaced as "failed 0xdeadbe00 (name): " with no reason at all -- the same class of bug as the propagate-server-error-text branch. It now reads "KeyError (no message)". **`--raw` inside `batch`.** The rejection is correct -- batch results are structured, and raw text on stdout would corrupt them -- but "not allowed" left the caller guessing, and we logged 7 hits on it. The message now names the replacement: the text is already in results[].result.text. Tests: bulk apply keeps going past a bad record, missing/empty records are reported rather than raised, a blank exception cannot produce a blank reason, and the --raw message names its replacement. Co-Authored-By: Claude Opus 5 (1M context) --- declib/api/decompiler_client.py | 7 +++ declib/api/decompiler_interface.py | 74 ++++++++++++++++++++++++ declib/cli/decompiler_cli.py | 91 +++++++++++++++++++++++++++++- tests/test_decompiler_cli.py | 70 +++++++++++++++++++++++ 4 files changed, 241 insertions(+), 1 deletion(-) diff --git a/declib/api/decompiler_client.py b/declib/api/decompiler_client.py index ef1e6f6..a458e31 100644 --- a/declib/api/decompiler_client.py +++ b/declib/api/decompiler_client.py @@ -585,6 +585,13 @@ def decompile_many(self, addrs: List[int], **kwargs) -> Dict[int, Optional[str]] "args": [list(addrs)], "kwargs": kwargs, }) + def apply_annotations(self, items: List[Dict]) -> Dict[str, int]: + """Apply many comments/renames in one round-trip.""" + return self._send_request({ + "type": "method_call", "method_name": "apply_annotations", + "args": [list(items)], + }) + def read_memory(self, addr: int, size: int) -> Optional[bytes]: """Read raw bytes from the loaded program.""" return self._send_request({"type": "method_call", "method_name": "read_memory", "args": [addr, size]}) diff --git a/declib/api/decompiler_interface.py b/declib/api/decompiler_interface.py index 935bcfa..9a92132 100644 --- a/declib/api/decompiler_interface.py +++ b/declib/api/decompiler_interface.py @@ -409,6 +409,80 @@ def decompile_many(self, addrs: List[int], **kwargs) -> Dict[int, Optional[str]] dec = None out[addr] = dec.text if dec is not None else None return out + def apply_annotations(self, items: List[Dict]) -> Dict[str, int]: + """Apply many comments and/or function renames in one call. + + Each item is ``{"addr": int, "comment": str?, "name": str?}``. Both + fields are optional; an item may carry either or both. + + Like :meth:`decompile_many`, the default loops in-process — which + through DecompilerServer collapses N round-trips into one. Annotation + is naturally produced in bulk (a dict of address to label, built while + reading a binary), and applying it one CLI call at a time is what + pushes people back to writing a script against the raw backend API. + + A failed item is counted, not raised, so one bad address does not + discard the rest of the batch. + + @param items: annotation records, as above. + @return: {"comments": n, "names": n, "failed": n} + """ + applied = {"comments": 0, "names": 0, "failed": 0, "errors": []} + + def _fail(addr, what, reason): + # str(exc) is empty for exceptions raised without arguments, which + # would surface as "failed 0x...: " with no reason at all. Name the + # type in that case rather than reporting a blank. + text = str(reason) + if not text.strip() and isinstance(reason, BaseException): + text = f"{type(reason).__name__} (no message)" + applied["failed"] += 1 + applied["errors"].append({ + "addr": addr, "field": what, "error": text, + }) + + for item in items: + addr = item.get("addr") + if addr is None: + _fail(None, "addr", "record has no address") + continue + + comment = item.get("comment") + name = item.get("name") + if comment is None and name is None: + _fail(addr, "record", "neither 'comment' nor 'name' given") + continue + + if comment is not None: + try: + self.comments[addr] = Comment(addr=addr, comment=comment) + applied["comments"] += 1 + except Exception as e: # noqa: BLE001 + self.warning(f"apply_annotations: comment at {hex(addr)} failed: {e}") + _fail(addr, "comment", e) + + if name is not None: + try: + func = self.functions[addr] + if func is None: + raise ValueError("no function starts at this address") + func.name = name + if func.header is not None: + func.header.name = name + self.functions[addr] = func + # Read back rather than assuming: a backend can accept the + # write and quietly not apply it, and a silently-miscounted + # rename is worse than a reported failure. + check = self.functions[addr] + if check is not None and getattr(check, "name", None) != name: + raise ValueError( + f"backend kept the name {getattr(check, 'name', None)!r}" + ) + applied["names"] += 1 + except Exception as e: # noqa: BLE001 + self.warning(f"apply_annotations: rename at {hex(addr)} failed: {e}") + _fail(addr, "name", e) + return applied def xrefs_to(self, artifact: Artifact, decompile=False, only_code=False) -> List[Artifact]: """ diff --git a/declib/cli/decompiler_cli.py b/declib/cli/decompiler_cli.py index fb4c05b..2398ee9 100644 --- a/declib/cli/decompiler_cli.py +++ b/declib/cli/decompiler_cli.py @@ -700,10 +700,20 @@ def _execute_batch_operation(operation: Dict, batch_args) -> Dict: ) for option in ("--raw", "--help", "-h"): if _batch_option_present(argv, option): + # Rejecting --raw is correct: batch results are structured, and raw + # text printed straight to stdout would corrupt them. But saying + # only "not allowed" leaves the caller guessing, so name the + # replacement -- the text they wanted is already in the result. + if option == "--raw": + error = ("--raw is not allowed inside a structured batch: batch " + "results are always JSON. Drop it — each entry's text " + "is in results[].result.text.") + else: + error = f"{option} is not allowed inside a structured batch." return _batch_result( operation, exit_code=EXIT_USER_ERROR, - error=f"{option} is not allowed inside a structured batch.", + error=error, duration_ms=0, ) @@ -1831,6 +1841,76 @@ def cmd_export(args) -> int: print(f" {len(failed)} function(s) did not decompile " f"(listed in export.json)") return EXIT_OK +def cmd_annotate(args) -> int: + """Apply a batch of comments and renames in one call. + + Annotation is produced in bulk -- you read a binary, build up a mapping of + address to label, and want to write it back. Doing that one `comment set` + at a time is N round-trips, which is why people reach for the raw backend + API instead. + + Input is JSON on stdin (or --file): either a list of records, or an object + mapping address to a comment. Records look like: + + {"addr": "0x401000", "comment": "parses the header", "name": "parse_hdr"} + """ + raw = Path(args.file).read_text() if args.file else sys.stdin.read() + if not raw.strip(): + raise SystemExit("no annotation input (give JSON on stdin or --file)") + try: + parsed = json.loads(raw) + except json.JSONDecodeError as e: + raise SystemExit(f"input is not valid JSON: {e}") + + # Accept the shorthand {"0x401000": "a comment", ...} as well as records, + # because that is the shape people already build by hand. + records: List[Dict] = [] + if isinstance(parsed, dict): + for key, value in parsed.items(): + if isinstance(value, dict): + records.append({"addr": key, **value}) + else: + records.append({"addr": key, "comment": value}) + elif isinstance(parsed, list): + records = list(parsed) + else: + raise SystemExit("expected a JSON list of records or an object keyed by address") + + with _with_client(args) as client: + items: List[Dict] = [] + for rec in records: + if not isinstance(rec, dict) or "addr" not in rec: + raise SystemExit(f"every record needs an 'addr': {rec!r}") + addr_value, _ = _parse_target(str(rec["addr"])) + if addr_value is None: + raise SystemExit(f"invalid address {rec['addr']!r}") + item = {"addr": _to_lifted_addr(client, addr_value)} + if rec.get("comment") is not None: + item["comment"] = str(rec["comment"]) + if rec.get("name") is not None: + item["name"] = str(rec["name"]) + if len(item) == 1: + raise SystemExit( + f"record for {rec['addr']} has neither 'comment' nor 'name'" + ) + items.append(item) + + if not items: + raise SystemExit("nothing to apply") + + result = client.apply_annotations(items) or {} + result["requested"] = len(items) + if args.json: + _emit(args, result) + else: + print(f"applied {result.get('comments', 0)} comment(s) and " + f"{result.get('names', 0)} rename(s) from {len(items)} record(s)") + for err in result.get("errors") or []: + where = err.get("addr") + where = _format_addr_hex(where) if isinstance(where, int) else where + print(f" failed {where} ({err.get('field')}): {err.get('error')}", + file=sys.stderr) + return EXIT_OK if not result.get("failed") else EXIT_USER_ERROR def cmd_get_callers(args) -> int: @@ -2778,6 +2858,15 @@ def build_parser() -> argparse.ArgumentParser: _add_server_filter_args(p_export) _add_output_args(p_export) p_export.set_defaults(func=cmd_export) + p_annotate = sub.add_parser( + "annotate", + help="Apply many comments/renames at once from JSON (stdin or --file).", + ) + p_annotate.add_argument("--file", + help="Read JSON from this path instead of stdin.") + _add_server_filter_args(p_annotate) + _add_output_args(p_annotate) + p_annotate.set_defaults(func=cmd_annotate) p_lf = sub.add_parser("list_functions", help="List functions in the binary.") p_lf.add_argument("--filter", dest="filter", help="Regex to filter function names.") diff --git a/tests/test_decompiler_cli.py b/tests/test_decompiler_cli.py index bd163aa..8f47aad 100644 --- a/tests/test_decompiler_cli.py +++ b/tests/test_decompiler_cli.py @@ -2104,3 +2104,73 @@ def test_safe_filename_is_filesystem_safe_and_unique(self): self.assertEqual(_safe_filename("", 0x55), "00000055.c") # Two same-named functions at different addresses cannot collide. self.assertNotEqual(_safe_filename("f", 1), _safe_filename("f", 2)) +class TestAnnotateBatch(unittest.TestCase): + """Bulk annotation plumbing that needs no backend.""" + + def _apply(self, items): + """Run apply_annotations against a stand-in interface.""" + from declib.api.decompiler_interface import DecompilerInterface + + class _Fake(DecompilerInterface): + def __init__(self): + self.comments = {} + self.functions = {} + self._warnings = [] + + def warning(self, msg): + self._warnings.append(msg) + + fake = _Fake() + return fake, DecompilerInterface.apply_annotations(fake, items) + + def test_missing_address_is_reported_not_raised(self): + _, res = self._apply([{"comment": "orphan"}]) + self.assertEqual(res["failed"], 1) + self.assertEqual(res["errors"][0]["field"], "addr") + + def test_record_with_neither_field_is_reported(self): + _, res = self._apply([{"addr": 0x1000}]) + self.assertEqual(res["failed"], 1) + self.assertIn("neither", res["errors"][0]["error"]) + + def test_one_bad_record_does_not_discard_the_batch(self): + fake, res = self._apply([ + {"addr": 0x1000, "comment": "kept"}, + {"addr": 0x2000}, # invalid + {"addr": 0x3000, "comment": "also kept"}, + ]) + self.assertEqual(res["comments"], 2) + self.assertEqual(res["failed"], 1) + + def test_failure_reason_is_never_blank(self): + """A bare exception must not surface as 'failed 0x...: '.""" + from declib.api.decompiler_interface import DecompilerInterface + + class _Exploding(DecompilerInterface): + def __init__(self): + self.comments = self + self.functions = {} + + def __setitem__(self, key, value): + raise KeyError() # empty str(), the blank-error case + + def warning(self, msg): + pass + + fake = _Exploding() + res = DecompilerInterface.apply_annotations(fake, [{"addr": 0x1000, "comment": "x"}]) + self.assertEqual(res["failed"], 1) + text = res["errors"][0]["error"] + self.assertTrue(text.strip(), "error text was blank") + self.assertIn("KeyError", text) + + +class TestBatchRawMessage(unittest.TestCase): + def test_raw_rejection_names_the_replacement(self): + """Rejecting --raw is right; leaving the caller guessing is not.""" + from declib.cli.decompiler_cli import _execute_batch_operation + + result = _execute_batch_operation( + {"id": "x", "argv": ["decompile", "0x1000", "--raw"]}, None + ) + self.assertIn("results[].result.text", result["error"])