diff --git a/declib/api/decompiler_client.py b/declib/api/decompiler_client.py index 69e3488..ae1a76f 100644 --- a/declib/api/decompiler_client.py +++ b/declib/api/decompiler_client.py @@ -595,6 +595,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 disassemble_range(self, start: int, end: int, **kwargs) -> Optional[str]: """Disassemble an arbitrary address span, function or not.""" return self._send_request({ diff --git a/declib/api/decompiler_interface.py b/declib/api/decompiler_interface.py index 34fc757..6895d16 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 0b24f9e..1765e64 100644 --- a/declib/cli/decompiler_cli.py +++ b/declib/cli/decompiler_cli.py @@ -797,10 +797,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, ) @@ -2145,6 +2155,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: @@ -3097,6 +3177,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 f3813ad..5f21a54 100644 --- a/tests/test_decompiler_cli.py +++ b/tests/test_decompiler_cli.py @@ -2250,6 +2250,76 @@ 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"]) class TestServerStartupDiagnostics(unittest.TestCase): """Startup/teardown diagnostics. No backend required."""