diff options
| -rw-r--r-- | docs/RPC.md | 1 | ||||
| -rw-r--r-- | idatui/diag.py | 121 | ||||
| -rw-r--r-- | idatui/domain.py | 6 | ||||
| -rw-r--r-- | idatui/edit_ctl.py | 21 | ||||
| -rw-r--r-- | idatui/rpc.py | 11 | ||||
| -rw-r--r-- | idatui/trace_ctl.py | 7 | ||||
| -rw-r--r-- | tests/test_diag.py | 174 | ||||
| -rw-r--r-- | tests/test_scenarios.py | 9 |
8 files changed, 343 insertions, 7 deletions
diff --git a/docs/RPC.md b/docs/RPC.md index 3ea7f33..e433551 100644 --- a/docs/RPC.md +++ b/docs/RPC.md @@ -71,6 +71,7 @@ stalls. `target` is a name, `0xADDR`, or omitted (= current function). | `xrefs_to` | `target`, `limit?=200` | `[{frm,to,type,fn_addr,fn_name}]` — who references it. | | `xrefs_from` | `target`, `limit?=200` | what it references. For a **function** (name/entry ea): whole-body callees + string/data refs from the decompiler, `[{to,name,string,is_func,type}]`. For an explicit **0xADDR**: address-scoped `[{frm,to,type,fn_addr,fn_name}]`. | | `resolve` | `name` | `{ea}` (or `{ea:null}`). | +| `diag` | `n?=10`, `clear?` | `{recent:[{when,what,error,where,thread}], log}` — errors the app swallowed rather than crashing on. The answer to "the verb said success and the pane shows nothing": inside a full-screen TUI a traceback has nowhere to go, so it comes out here. Set `$IDATUI_LOG=/tmp/x.log` when spawning for the same entries plus tracebacks, on disk. | ### Semantic verbs High-level ops type through the **real prompts** with a per-char delay diff --git a/idatui/diag.py b/idatui/diag.py new file mode 100644 index 0000000..b45a72d --- /dev/null +++ b/idatui/diag.py @@ -0,0 +1,121 @@ +"""Where swallowed errors go. + +A TUI must not die because one background load failed, so this codebase catches +broadly -- around fifty ``except Exception`` sites, two dozen of which resolve +to ``pass``. That is the right policy and it has one bad consequence: with +forty-odd ``@work(thread=True)`` workers, a failure in a background load leaves +no trace at all. The view just stays empty, and there is nothing to read +afterwards because the app owns the screen. + +``kittygfx`` already solved this for itself with ``$IDATUI_KITTY_LOG``. This is +the same idea for everything else: + +* ``$IDATUI_LOG=/tmp/x.log`` writes every swallowed error to a file. Unset (the + default) it costs an ``os.environ`` lookup and nothing else. +* The last few are kept in memory regardless, so ``drive diag`` can ask a live + app "what went wrong recently?" -- which is the question you actually have + when a driver reports success and the pane shows nothing. + +Use it where an exception would otherwise vanish:: + + with swallow("decomp_map(%#x)" % ea): + self._apply_split_map(ea, self.program.decomp_map(ea)) + +NOT for expected control flow. ``query_one`` raising because a modal owns the +screen is normal and happens constantly; wrapping that would bury the real +entries in noise. The test for whether it belongs here is "would I want to see +this after the fact?". +""" +from __future__ import annotations + +import contextlib +import os +import threading +import time +import traceback +from collections import deque + +#: Bounded on purpose: this is a debugging aid inside a long-running TUI, not an +#: audit log. Old entries are worth less than the memory. +_MAX = 50 +_ring: deque[dict] = deque(maxlen=_MAX) +_lock = threading.Lock() + + +def _logfile() -> str | None: + """Read the env var per call, not once at import. + + The pilot suite and the RPC tests set it after importing the app, and a + cached value would silently disable the thing being tested. + """ + return os.environ.get("IDATUI_LOG") or None + + +def log(msg: str) -> None: + """Append a line to ``$IDATUI_LOG``. No-op when it isn't set.""" + path = _logfile() + if not path: + return + try: + with open(path, "a", encoding="utf-8") as fh: + fh.write(f"{time.strftime('%H:%M:%S')} {msg}\n") + except OSError: + pass # a broken log path must never break the app + + +def note(what: str, exc: BaseException) -> None: + """Record a swallowed exception: in the ring always, in the log if enabled.""" + entry = { + "when": time.time(), + "what": what, + "error": f"{type(exc).__name__}: {exc}", + "where": _origin(exc), + "thread": threading.current_thread().name, + } + with _lock: + _ring.append(entry) + log(f"[swallowed] {what}: {entry['error']} ({entry['where']})") + if _logfile(): + log("".join(traceback.format_exception( + type(exc), exc, exc.__traceback__)).rstrip()) + + +def _origin(exc: BaseException) -> str: + """file:line where it was actually raised (the deepest frame we have).""" + tb = exc.__traceback__ + last = None + while tb is not None: + last = tb + tb = tb.tb_next + if last is None: + return "?" + f = last.tb_frame + return f"{os.path.basename(f.f_code.co_filename)}:{last.tb_lineno}" + + +@contextlib.contextmanager +def swallow(what: str, *, reraise: tuple = ()): + """Run a block, record anything it raises, and carry on. + + ``reraise`` lets a caller keep the exceptions it genuinely handles -- most + usefully ``IDAConnectionError``, which the app turns into a reconnect and + must not have eaten here. + """ + try: + yield + except reraise: + raise + except Exception as e: # noqa: BLE001 -- the whole point + note(what, e) + + +def recent(n: int = 10) -> list[dict]: + """The last ``n`` swallowed errors, newest last.""" + with _lock: + items = list(_ring) + return items[-n:] if n > 0 else items + + +def clear() -> None: + with _lock: + _ring.clear() diff --git a/idatui/domain.py b/idatui/domain.py index 222f7bb..b496638 100644 --- a/idatui/domain.py +++ b/idatui/domain.py @@ -30,6 +30,7 @@ from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass, field, replace from typing import Callable, TYPE_CHECKING +from . import diag from .errors import IDAToolError if TYPE_CHECKING: # type hint only @@ -1605,7 +1606,10 @@ class Program: try: with urllib.request.urlopen(url, timeout=timeout) as r: return json.loads(r.read().decode("utf-8", "replace")) - except Exception: # noqa: BLE001 -- fall back to the truncated preview + except Exception as e: # noqa: BLE001 -- fall back to the truncated preview + # The user gets CLIPPED pseudocode with no indication that a fetch + # failed rather than the function genuinely being that short. + diag.note(f"decompile: full-body fetch {url}", e) return None def strings(self, min_len: int = 4, refresh: bool = False) -> list[StrLit]: diff --git a/idatui/edit_ctl.py b/idatui/edit_ctl.py index 7803a7a..54f4d89 100644 --- a/idatui/edit_ctl.py +++ b/idatui/edit_ctl.py @@ -26,6 +26,7 @@ from typing import TYPE_CHECKING from textual.widgets import DataTable +from . import diag from .errors import IDAToolError if TYPE_CHECKING: # pragma: no cover @@ -210,7 +211,11 @@ class EditController: resolved: int | None = None try: resolved = prog.resolve(old) - except Exception: # noqa: BLE001 + except Exception as e: # noqa: BLE001 + # Not cosmetic: an unresolved name is renamed as DATA instead of as + # a function, so a lookup that failed for a transport reason quietly + # applies the wrong kind of edit. + diag.note(f"rename: resolve({old!r})", e) resolved = None if resolved is not None: fn = prog.function_of(resolved) @@ -291,7 +296,11 @@ class EditController: # it already did. try: fn = app.program.function_of(addr) - except Exception: # noqa: BLE001 + except Exception as e: # noqa: BLE001 + # If this throws we don't learn the address IS a function start, so + # the index keeps the old name and every readback says the rename + # never happened. + diag.note(f"name: function_of({addr:#x})", e) fn = None is_func_start = fn is not None and fn.addr == addr lm = app.program.listing(addr) @@ -403,7 +412,10 @@ class EditController: if kind is None and app._looks_like_symbol(word): try: tgt = app.program.resolve(word) - except Exception: # noqa: BLE001 + except Exception as e: # noqa: BLE001 + # Falls through to case (3), which retypes the enclosing + # function -- a different edit from the one asked for. + diag.note(f"retype: resolve({word!r})", e) tgt = None if tgt is not None: tft = app.program.func_types(tgt) @@ -495,6 +507,7 @@ class EditController: try: app.program.make_data(ea, type_decl) except Exception as e: # noqa: BLE001 + diag.note(f"make_data({ea:#x}, {type_decl!r})", e) app.call_from_thread(app._status, f"make data: {e}") return app.program.bump_items() @@ -555,6 +568,7 @@ class EditController: app.call_from_thread(app._status, f"format: {e.message}", True) return except Exception as e: # noqa: BLE001 -- surface transport failures too + diag.note(f"op_format({where}, {ea:#x})", e) app.call_from_thread(app._status, f"format: {e}", True) return text = " ".join((r.get("text") or "").split()) @@ -690,6 +704,7 @@ class EditController: anchor.refresh_functions = True app.program.undefine(ea) except Exception as e: # noqa: BLE001 -- surface soft/hard tool errors + diag.note(f"edit_item({kind}, {ea:#x})", e) app.call_from_thread(app._status, f"{kind}: {e}") return # Structure changed everywhere: drop all item/function/decomp caches. diff --git a/idatui/rpc.py b/idatui/rpc.py index b4ca80e..698e4cf 100644 --- a/idatui/rpc.py +++ b/idatui/rpc.py @@ -28,6 +28,7 @@ from typing import Any from rich.console import Console from ._sync import drain, settle +from . import diag from .app import DecompView, GraphView, HexView, ListingView, ViewMode PROTO_VERSION = 1 @@ -889,6 +890,16 @@ class RpcServer: return functions(app, params.get("filter"), int(params.get("limit", 50))) # -- projects ------------------------------------------------------ # + if method == "diag": + # What has been swallowed lately. A driver that got "success" and a + # pane showing nothing has no other way to ask; inside a + # full-screen TUI there is nowhere for a traceback to go. + if params.get("clear"): + diag.clear() + return {"cleared": True} + return {"recent": diag.recent(int(params.get("n", 10))), + "log": os.environ.get("IDATUI_LOG") or None} + if method == "trace": tc = app.trace_ctl if tc.trace is None: diff --git a/idatui/trace_ctl.py b/idatui/trace_ctl.py index 46d4b17..8072801 100644 --- a/idatui/trace_ctl.py +++ b/idatui/trace_ctl.py @@ -21,6 +21,8 @@ import bisect import os from typing import TYPE_CHECKING +from . import diag + if TYPE_CHECKING: # pragma: no cover from .app import IdaTui @@ -264,7 +266,10 @@ class TraceController: # or twice, once for each of two parallel maps — would be felt. try: app._apply_split_map(ea, app.program.decomp_map(ea)) - except Exception: # noqa: BLE001 + except Exception as e: # noqa: BLE001 + # The pseudocode simply stops being painted with the trail, with + # nothing on screen to say why. + diag.note(f"trail: decomp_map({ea:#x})", e) self.trail_map, self.trail_map_ea = [], ea self.trail_line_of, self.trail_eas = {}, [] self.trail_span = None diff --git a/tests/test_diag.py b/tests/test_diag.py new file mode 100644 index 0000000..e3b966a --- /dev/null +++ b/tests/test_diag.py @@ -0,0 +1,174 @@ +#!/usr/bin/env python3 +"""idatui.diag — the channel swallowed errors go down. + +Pure: no IDA, no worker, no Textual. +""" +from __future__ import annotations + +import os +import sys +import tempfile +import threading + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +#: pure: a ring buffer and a log file. +#: Read by tests/run.py (--fast skips every NEEDS_IDA file). +NEEDS_IDA = False + +from idatui import diag # noqa: E402 + +PASS = FAIL = 0 + + +def check(name, ok, detail=""): + global PASS, FAIL + if ok: + PASS += 1 + print(f" ok {name}") + else: + FAIL += 1 + print(f" FAIL {name} {detail}") + + +def t_swallow_keeps_going(): + diag.clear() + ran = [] + with diag.swallow("a thing"): + raise ValueError("nope") + ran.append("after") + check("swallow() does not propagate", ran == ["after"]) + r = diag.recent() + check("the error is recorded", len(r) == 1, str(r)) + check("with what was being attempted", r[0]["what"] == "a thing", str(r[0])) + check("and the exception type and message", + r[0]["error"] == "ValueError: nope", r[0]["error"]) + check("and where it was actually raised", + r[0]["where"].startswith("test_diag.py:"), r[0]["where"]) + + +def t_reraise(): + """The app turns IDAConnectionError into a reconnect; swallow() must not eat + the exceptions its caller genuinely handles.""" + diag.clear() + + class Wanted(Exception): + pass + + try: + with diag.swallow("keeps its own", reraise=(Wanted,)): + raise Wanted("mine") + check("reraise lets the listed type through", False, "not raised") + except Wanted: + check("reraise lets the listed type through", True) + check("and a reraised error is not recorded twice", + diag.recent() == [], str(diag.recent())) + with diag.swallow("still swallows others", reraise=(Wanted,)): + raise ValueError("other") + check("other types are still swallowed", len(diag.recent()) == 1) + + +def t_ring_is_bounded(): + diag.clear() + for i in range(diag._MAX + 25): + diag.note(f"item {i}", RuntimeError(str(i))) + r = diag.recent(1000) + check("the ring is bounded", len(r) == diag._MAX, f"{len(r)}") + check("it keeps the NEWEST entries", + r[-1]["what"] == f"item {diag._MAX + 24}", r[-1]["what"]) + check("recent(n) returns the last n, newest last", + [e["what"] for e in diag.recent(3)] + == [f"item {diag._MAX + 22}", f"item {diag._MAX + 23}", + f"item {diag._MAX + 24}"], str(diag.recent(3))) + + +def t_log_file(): + diag.clear() + with tempfile.TemporaryDirectory() as d: + path = os.path.join(d, "x.log") + os.environ["IDATUI_LOG"] = path + try: + with diag.swallow("logged thing"): + raise KeyError("missing") + finally: + os.environ.pop("IDATUI_LOG", None) + body = open(path, encoding="utf-8").read() + check("the log records what was attempted", "logged thing" in body, body[:200]) + check("and the error", "KeyError" in body, body[:200]) + check("and a traceback, which the ring doesn't carry", + "Traceback" in body and "t_log_file" in body, body[:300]) + + +def t_log_is_off_by_default(): + diag.clear() + os.environ.pop("IDATUI_LOG", None) + with diag.swallow("unlogged"): + raise ValueError("x") + check("without $IDATUI_LOG nothing is written, but the ring still has it", + len(diag.recent()) == 1) + + +def t_broken_log_path_is_harmless(): + """A bad log path must never be the thing that breaks the app.""" + diag.clear() + os.environ["IDATUI_LOG"] = "/nonexistent-dir-xyz/deep/x.log" + try: + with diag.swallow("still fine"): + raise ValueError("boom") + check("an unwritable log path doesn't raise", True) + check("and the error is still recorded in the ring", + len(diag.recent()) == 1) + finally: + os.environ.pop("IDATUI_LOG", None) + + +def t_env_read_per_call(): + """The pilot and the RPC tests set $IDATUI_LOG after importing the app, so a + value cached at import would silently disable the thing under test.""" + diag.clear() + with tempfile.TemporaryDirectory() as d: + path = os.path.join(d, "late.log") + os.environ["IDATUI_LOG"] = path # set AFTER import + try: + diag.log("hello") + finally: + os.environ.pop("IDATUI_LOG", None) + check("a log path set after import is honoured", + os.path.exists(path) and "hello" in open(path).read()) + + +def t_thread_safe(): + diag.clear() + def go(n): + for i in range(40): + diag.note(f"t{n}-{i}", RuntimeError("x")) + ts = [threading.Thread(target=go, args=(n,)) for n in range(6)] + for t in ts: + t.start() + for t in ts: + t.join(10) + r = diag.recent(1000) + check("concurrent notes don't corrupt the ring", + len(r) == diag._MAX and all("what" in e for e in r), f"{len(r)}") + check("the recording thread is captured", + all(e["thread"] for e in r)) + + +def main() -> int: + for fn in (t_swallow_keeps_going, t_reraise, t_ring_is_bounded, t_log_file, + t_log_is_off_by_default, t_broken_log_path_is_harmless, + t_env_read_per_call, t_thread_safe): + print(f"\n{fn.__name__}") + try: + fn() + except Exception as e: # noqa: BLE001 + import traceback + check(f"{fn.__name__} did not crash", False, f"{type(e).__name__}: {e}") + traceback.print_exc() + diag.clear() + print(f"\n{PASS} passed, {FAIL} failed") + return 1 if FAIL else 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_scenarios.py b/tests/test_scenarios.py index 9934462..a36ea80 100644 --- a/tests/test_scenarios.py +++ b/tests/test_scenarios.py @@ -1362,11 +1362,16 @@ async def s_follow_xrefs(c: Ctx): orig_name = app._cur.name depth = len(app._nav) await c.press("enter") - await c.wait(lambda: len(app._nav) > depth, 25) + # Wait for exactly what the check asserts. Waiting only on the nav depth let + # the check run while _cur was still the function we jumped FROM -- the + # follow pushes the source entry before it opens the target -- so this + # failed about one run in ten with cur == orig, at full speed, looking like + # a code regression. + await c.wait(lambda: len(app._nav) > depth and app._cur.ea != orig, 25) c.check("Enter follows the call into another function", app._cur.ea != orig and len(app._nav) > depth, f"cur={app._cur.ea:#x}") await c.press("escape") - await c.pause(0.1) + await c.wait(lambda: app._cur.ea == orig, 15) c.check("Esc returns from the follow", app._cur.ea == orig, f"cur={app._cur.ea:#x}") dis.cursor = call_idx dis.refresh() |