reformat: ruff format + import sort, mechanically (see ruff.toml)

No behavior. Listed in .git-blame-ignore-revs (next commit).

1 files changed, 221 insertions, 93 deletions

diff --git a/idatui/pane.py b/idatui/pane.py
index 1eee847..67b462b 100644
--- a/idatui/pane.py
+++ b/idatui/pane.py
@@ -27,6 +27,7 @@ Requires: running inside tmux or zellij. Each pane leases a registered GUI or
 shared managed idalib database through IDA Nexus. Uses ~/ida-venv/bin/python for
 the TUI (needs textual) unless --python / IDATUI_PYTHON says otherwise.
 """
+
 from __future__ import annotations
 
 import argparse
@@ -42,7 +43,8 @@ from .rpcclient import RpcClient, RpcError
 
 REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 DEFAULT_PY = os.environ.get(
-    "IDATUI_PYTHON", os.path.expanduser("~/ida-venv/bin/python"))
+    "IDATUI_PYTHON", os.path.expanduser("~/ida-venv/bin/python")
+)
 
 
 def _sockdir() -> str:
@@ -106,26 +108,31 @@ def _mux_of_pane(pane: str) -> str:
 def _zellij_argv() -> list[str]:
     """Base zellij argv, pinned to our session when we know it (so it still works
     from a process that isn't itself attached)."""
-    session = (os.environ.get("IDATUI_ZELLIJ_SESSION")
-               or os.environ.get("ZELLIJ_SESSION_NAME"))
+    session = os.environ.get("IDATUI_ZELLIJ_SESSION") or os.environ.get(
+        "ZELLIJ_SESSION_NAME"
+    )
     return ["zellij", "-s", session] if session else ["zellij"]
 
 
 def _tmux(*args: str) -> str:
-    return subprocess.run(["tmux", *args], capture_output=True, text=True,
-                          check=True).stdout.strip()
+    return subprocess.run(
+        ["tmux", *args], capture_output=True, text=True, check=True
+    ).stdout.strip()
 
 
 def _zellij(*args: str) -> str:
-    return subprocess.run([*_zellij_argv(), *args], capture_output=True,
-                          text=True, check=True).stdout.strip()
+    return subprocess.run(
+        [*_zellij_argv(), *args], capture_output=True, text=True, check=True
+    ).stdout.strip()
 
 
 def _zellij_panes() -> list[dict[str, Any]]:
     try:
-        out = subprocess.run([*_zellij_argv(), "action", "list-panes",
-                              "--state", "--json"],
-                             capture_output=True, text=True)
+        out = subprocess.run(
+            [*_zellij_argv(), "action", "list-panes", "--state", "--json"],
+            capture_output=True,
+            text=True,
+        )
         rows = json.loads(out.stdout or "[]")
     except (OSError, ValueError):
         return []
@@ -148,8 +155,9 @@ def _pane_alive(pane: str, mux: str | None = None) -> bool:
             if str(row.get("id")) == want and bool(row.get("is_plugin")) is False:
                 return not row.get("exited", False)
         return False
-    out = subprocess.run(["tmux", "list-panes", "-a", "-F", "#{pane_id}"],
-                         capture_output=True, text=True)
+    out = subprocess.run(
+        ["tmux", "list-panes", "-a", "-F", "#{pane_id}"], capture_output=True, text=True
+    )
     return pane in out.stdout.split()
 
 
@@ -159,8 +167,9 @@ def _pane_exists(pane: str, mux: str | None = None) -> bool:
         return False
     if (mux or _mux_of_pane(pane)) == "zellij":
         want = pane.split("_", 1)[-1]
-        return any(str(r.get("id")) == want and not r.get("is_plugin")
-                   for r in _zellij_panes())
+        return any(
+            str(r.get("id")) == want and not r.get("is_plugin") for r in _zellij_panes()
+        )
     return _pane_alive(pane, "tmux")
 
 
@@ -169,24 +178,36 @@ def _pane_kill(pane: str, mux: str | None = None) -> None:
     if not pane:
         return
     if (mux or _mux_of_pane(pane)) == "zellij":
-        subprocess.run([*_zellij_argv(), "action", "close-pane",
-                        "--pane-id", pane], capture_output=True)
+        subprocess.run(
+            [*_zellij_argv(), "action", "close-pane", "--pane-id", pane],
+            capture_output=True,
+        )
     else:
         subprocess.run(["tmux", "kill-pane", "-t", pane], capture_output=True)
 
 
-def _pane_split(inner: list[str], *, mux: str, vertical: bool,
-                size: str | None, detached: bool) -> str:
+def _pane_split(
+    inner: list[str], *, mux: str, vertical: bool, size: str | None, detached: bool
+) -> str:
     """Open a pane running ``inner`` (argv) in REPO, and return its pane id."""
     if mux == "zellij":
         # zellij runs the argv directly (no shell) and takes the cwd as a flag,
         # so there's nothing to quote. --name labels the pane in the UI.
-        argv = [*_zellij_argv(), "action", "new-pane",
-                "--direction", "down" if vertical else "right",
-                "--cwd", REPO, "--name", "idatui"]
+        argv = [
+            *_zellij_argv(),
+            "action",
+            "new-pane",
+            "--direction",
+            "down" if vertical else "right",
+            "--cwd",
+            REPO,
+            "--name",
+            "idatui",
+        ]
         argv += ["--", *inner]
-        pane = subprocess.run(argv, capture_output=True, text=True,
-                              check=True).stdout.strip()
+        pane = subprocess.run(
+            argv, capture_output=True, text=True, check=True
+        ).stdout.strip()
         # zellij prints the new pane id ('terminal_3'); without it we could not
         # target this pane later, so treat a missing id as a hard failure.
         if not pane.startswith(("terminal_", "plugin_")):
@@ -196,13 +217,14 @@ def _pane_split(inner: list[str], *, mux: str, vertical: bool,
             # to the pane we were called from.
             origin = os.environ.get("ZELLIJ_PANE_ID")
             if origin:
-                subprocess.run([*_zellij_argv(), "action", "focus-pane-id",
-                                f"terminal_{origin}"], capture_output=True)
+                subprocess.run(
+                    [*_zellij_argv(), "action", "focus-pane-id", f"terminal_{origin}"],
+                    capture_output=True,
+                )
         return pane
 
     cmd = f"cd {REPO!r} && exec " + " ".join(_q(a) for a in inner)
-    split = ["split-window", "-v" if vertical else "-h",
-             "-P", "-F", "#{pane_id}"]
+    split = ["split-window", "-v" if vertical else "-h", "-P", "-F", "#{pane_id}"]
     if size:
         split += ["-l", str(size)]
     if detached:
@@ -224,9 +246,15 @@ def _pane_capture(pane: str, mux: str | None = None) -> str:
 
 # tmux key names -> zellij key names (zellij rejects e.g. "Escape", wants "Esc").
 _ZELLIJ_KEYS = {
-    "escape": "Esc", "bspace": "Backspace", "space": "Space",
-    "pageup": "PageUp", "pagedown": "PageDown", "ppage": "PageUp",
-    "npage": "PageDown", "ic": "Insert", "dc": "Delete",
+    "escape": "Esc",
+    "bspace": "Backspace",
+    "space": "Space",
+    "pageup": "PageUp",
+    "pagedown": "PageDown",
+    "ppage": "PageUp",
+    "npage": "PageDown",
+    "ic": "Insert",
+    "dc": "Delete",
 }
 
 
@@ -244,8 +272,17 @@ def _pane_keys(pane: str, keys: list[str], mux: str | None = None) -> None:
     """Inject real terminal keystrokes into the pane (the input-layer cross-check)."""
     mux = mux or _mux_of_pane(pane)
     if mux == "zellij":
-        subprocess.run([*_zellij_argv(), "action", "send-keys", "--pane-id", pane,
-                        *[_to_zellij_key(k) for k in keys]], check=True)
+        subprocess.run(
+            [
+                *_zellij_argv(),
+                "action",
+                "send-keys",
+                "--pane-id",
+                pane,
+                *[_to_zellij_key(k) for k in keys],
+            ],
+            check=True,
+        )
     else:
         subprocess.run(["tmux", "send-keys", "-t", pane, *keys], check=True)
 
@@ -256,8 +293,9 @@ def _pane_keys(pane: str, keys: list[str], mux: str | None = None) -> None:
 
 
 def _count_live_panes() -> int:
-    return sum(1 for r in _load_registry()
-               if _pane_alive(r.get("pane", ""), r.get("mux")))
+    return sum(
+        1 for r in _load_registry() if _pane_alive(r.get("pane", ""), r.get("mux"))
+    )
 
 
 def _reap_orphan_workers(force: bool = False) -> int:
@@ -272,20 +310,28 @@ def _reap_orphan_workers(force: bool = False) -> int:
 def spawn(args) -> int:
     mux = args.mux or _detect_mux()
     if mux.startswith("?"):
-        print(f"error: unknown multiplexer {mux[1:]!r} (want tmux or zellij)",
-              file=sys.stderr)
+        print(
+            f"error: unknown multiplexer {mux[1:]!r} (want tmux or zellij)",
+            file=sys.stderr,
+        )
         return 2
     if not mux:
-        print("error: not inside tmux or zellij (spawn creates a pane there). "
-              "Set $IDATUI_MUX=tmux|zellij to force a backend.", file=sys.stderr)
+        print(
+            "error: not inside tmux or zellij (spawn creates a pane there). "
+            "Set $IDATUI_MUX=tmux|zellij to force a backend.",
+            file=sys.stderr,
+        )
         return 2
     if not args.open and not getattr(args, "project", None):
         print("error: pass --open <binary> or --project <file>", file=sys.stderr)
         return 2
 
     sock = args.sock or os.path.join(_sockdir(), f"idatui-{secrets.token_hex(3)}.sock")
-    project = (os.path.abspath(os.path.expanduser(args.project))
-               if getattr(args, "project", None) else None)
+    project = (
+        os.path.abspath(os.path.expanduser(args.project))
+        if getattr(args, "project", None)
+        else None
+    )
     target = os.path.abspath(os.path.expanduser(args.open)) if args.open else None
     if target is not None and not os.path.exists(target):
         print(f"error: no such binary: {target}", file=sys.stderr)
@@ -317,17 +363,30 @@ def spawn(args) -> int:
         inner += ["--trace", os.path.abspath(os.path.expanduser(args.trace))]
 
     if args.size and mux == "zellij":
-        print("note: --size is tmux-only; zellij tiles the new pane evenly",
-              file=sys.stderr)
+        print(
+            "note: --size is tmux-only; zellij tiles the new pane evenly",
+            file=sys.stderr,
+        )
     try:
-        pane = _pane_split(inner, mux=mux, vertical=args.vertical,
-                           size=args.size, detached=args.detached)
+        pane = _pane_split(
+            inner,
+            mux=mux,
+            vertical=args.vertical,
+            size=args.size,
+            detached=args.detached,
+        )
     except (OSError, subprocess.CalledProcessError, RuntimeError) as e:
         print(f"error: could not create a {mux} pane: {e}", file=sys.stderr)
         return 2
 
-    row = {"sock": sock, "pane": pane, "mux": mux, "target": project or target,
-           "kind": "project" if project else "open", "started": time.time()}
+    row = {
+        "sock": sock,
+        "pane": pane,
+        "mux": mux,
+        "target": project or target,
+        "kind": "project" if project else "open",
+        "started": time.time(),
+    }
     reg = [r for r in _load_registry() if r.get("sock") != sock]
     reg.append(row)
     _save_registry(reg)
@@ -340,11 +399,17 @@ def spawn(args) -> int:
 
 def _q(s: str) -> str:
     import shlex
+
     return shlex.quote(s)
 
 
-def _wait_ready(sock: str, timeout: float, pane: str,
-                stuck_after: float = 45.0, mux: str | None = None) -> dict[str, Any]:
+def _wait_ready(
+    sock: str,
+    timeout: float,
+    pane: str,
+    stuck_after: float = 45.0,
+    mux: str | None = None,
+) -> dict[str, Any]:
     """Poll the socket + ping until the TUI reports ready (or timeout).
 
     Emits a one-time hint if IDA Nexus discovery/opening is still not ready after
@@ -367,10 +432,16 @@ def _wait_ready(sock: str, timeout: float, pane: str,
                 pass
         if not warned and (time.time() - start) > stuck_after:
             warned = True
-            why = ("RPC socket not created yet" if not os.path.exists(sock)
-                   else "TUI up but analysis not ready")
-            print(f"still waiting ({int(time.time() - start)}s): {why}. "
-                  f"Check IDA Nexus registrations and worker logs.", file=sys.stderr)
+            why = (
+                "RPC socket not created yet"
+                if not os.path.exists(sock)
+                else "TUI up but analysis not ready"
+            )
+            print(
+                f"still waiting ({int(time.time() - start)}s): {why}. "
+                f"Check IDA Nexus registrations and worker logs.",
+                file=sys.stderr,
+            )
         time.sleep(0.4)
     last = dict(last)
     last["ready"] = False
@@ -383,9 +454,12 @@ def _wait_ready(sock: str, timeout: float, pane: str,
 # --------------------------------------------------------------------------- #
 def stop(args) -> int:
     reg = _load_registry()
-    rows = [r for r in reg
-            if (args.sock and r.get("sock") == args.sock)
-            or (args.pane and r.get("pane") == args.pane)]
+    rows = [
+        r
+        for r in reg
+        if (args.sock and r.get("sock") == args.sock)
+        or (args.pane and r.get("pane") == args.pane)
+    ]
     if not rows and args.sock:  # allow stopping an untracked socket
         rows = [{"sock": args.sock, "pane": args.pane}]
     if not rows:
@@ -432,8 +506,10 @@ def stop(args) -> int:
         # Only ever reached on timeout: say so, because it means a save may have
         # been cut short rather than "clean teardown".
         out["force_killed"] = killed
-        out["warning"] = (f"pane(s) did not exit within {args.timeout}s and were "
-                          "killed; unsaved database changes may be lost")
+        out["warning"] = (
+            f"pane(s) did not exit within {args.timeout}s and were "
+            "killed; unsaved database changes may be lost"
+        )
     print(json.dumps(out))
     return 0
 
@@ -468,8 +544,16 @@ def list_panes(args) -> int:
 
 def reap(args) -> int:
     """Deprecated no-op; shared IDA Nexus workers are managed by leases."""
-    print(json.dumps({"reaped_workers": 0, "live_panes": _count_live_panes(),
-                      "forced": args.force, "deprecated": True}))
+    print(
+        json.dumps(
+            {
+                "reaped_workers": 0,
+                "live_panes": _count_live_panes(),
+                "forced": args.force,
+                "deprecated": True,
+            }
+        )
+    )
     return 0
 
 
@@ -520,56 +604,97 @@ def _resolve_pane(sock: str | None) -> str | None:
     else:
         print("error: several live panes, pass --pane or --sock:", file=sys.stderr)
         for r in live:
-            print(f"  {r.get('pane')}  {r.get('sock')}  {r.get('target')}",
-                  file=sys.stderr)
+            print(
+                f"  {r.get('pane')}  {r.get('sock')}  {r.get('target')}",
+                file=sys.stderr,
+            )
     return None
 
 
 def main(argv: list[str]) -> int:
     p = argparse.ArgumentParser(
         prog="idatui.pane",
-        description="spawn/manage idatui TUI panes in tmux or zellij")
+        description="spawn/manage idatui TUI panes in tmux or zellij",
+    )
     sub = p.add_subparsers(dest="cmd", required=True)
 
     sp = sub.add_parser("spawn", help="open a TUI pane and wait until ready")
-    sp.add_argument("--open", metavar="PATH",
-                    help="binary to open (its dir must be writable)")
-    sp.add_argument("--trace", metavar="FILE",
-                    help="Tenet execution trace to load alongside the binary")
-    sp.add_argument("--project", metavar="FILE",
-                    help="project file to open instead of a single binary; "
-                         "any --open paths are added to it (created if absent)")
-    sp.add_argument("--processor", metavar="NAME",
-                    help="IDA processor for a headerless blob: arm, armb, "
-                         "mipsb, metapc, … (passed to idatui.launch)")
-    sp.add_argument("--base", metavar="ADDR",
-                    help="load address for a headerless blob, e.g. 0x8000000 "
-                         "(16-byte aligned)")
-    sp.add_argument("--ida-args", metavar="STR", dest="ida_args",
-                    help="extra IDA command-line switches, passed through")
-    sp.add_argument("--sock", help="RPC socket path (default: auto in $XDG_RUNTIME_DIR)")
-    sp.add_argument("--python", default=DEFAULT_PY, help=f"python for the TUI ({DEFAULT_PY})")
-    sp.add_argument("--vertical", action="store_true", help="split vertically (stacked)")
-    sp.add_argument("--size", help="new pane size (tmux -l value, e.g. 60%% or 120; "
-                                   "ignored under zellij)")
+    sp.add_argument(
+        "--open", metavar="PATH", help="binary to open (its dir must be writable)"
+    )
+    sp.add_argument(
+        "--trace",
+        metavar="FILE",
+        help="Tenet execution trace to load alongside the binary",
+    )
+    sp.add_argument(
+        "--project",
+        metavar="FILE",
+        help="project file to open instead of a single binary; "
+        "any --open paths are added to it (created if absent)",
+    )
+    sp.add_argument(
+        "--processor",
+        metavar="NAME",
+        help="IDA processor for a headerless blob: arm, armb, "
+        "mipsb, metapc, … (passed to idatui.launch)",
+    )
+    sp.add_argument(
+        "--base",
+        metavar="ADDR",
+        help="load address for a headerless blob, e.g. 0x8000000 (16-byte aligned)",
+    )
+    sp.add_argument(
+        "--ida-args",
+        metavar="STR",
+        dest="ida_args",
+        help="extra IDA command-line switches, passed through",
+    )
+    sp.add_argument(
+        "--sock", help="RPC socket path (default: auto in $XDG_RUNTIME_DIR)"
+    )
+    sp.add_argument(
+        "--python", default=DEFAULT_PY, help=f"python for the TUI ({DEFAULT_PY})"
+    )
+    sp.add_argument(
+        "--vertical", action="store_true", help="split vertically (stacked)"
+    )
+    sp.add_argument(
+        "--size",
+        help="new pane size (tmux -l value, e.g. 60%% or 120; ignored under zellij)",
+    )
     sp.add_argument("--detached", action="store_true", help="don't focus the new pane")
-    sp.add_argument("--mux", choices=MUXES, default="",
-                    help="multiplexer to spawn in (default: autodetect from "
-                         "$ZELLIJ/$TMUX; $IDATUI_MUX overrides)")
-    sp.add_argument("--timeout", type=float, default=300.0,
-                    help="seconds to wait for readiness (fresh --open analysis is slow)")
+    sp.add_argument(
+        "--mux",
+        choices=MUXES,
+        default="",
+        help="multiplexer to spawn in (default: autodetect from "
+        "$ZELLIJ/$TMUX; $IDATUI_MUX overrides)",
+    )
+    sp.add_argument(
+        "--timeout",
+        type=float,
+        default=300.0,
+        help="seconds to wait for readiness (fresh --open analysis is slow)",
+    )
     sp.set_defaults(fn=spawn)
 
     st = sub.add_parser("stop", help="graceful quit + kill the pane")
     st.add_argument("--sock")
     st.add_argument("--pane")
-    st.add_argument("--timeout", type=float, default=600.0,
-                    help="seconds to wait for the pane to exit (it saves dirty "
-                         "databases on the way out) before force-killing it")
+    st.add_argument(
+        "--timeout",
+        type=float,
+        default=600.0,
+        help="seconds to wait for the pane to exit (it saves dirty "
+        "databases on the way out) before force-killing it",
+    )
     st.set_defaults(fn=stop)
 
     ls = sub.add_parser("list", help="list tracked panes")
-    ls.add_argument("--prune", action="store_true", help="drop dead panes (and their sockets)")
+    ls.add_argument(
+        "--prune", action="store_true", help="drop dead panes (and their sockets)"
+    )
     ls.set_defaults(fn=list_panes)
 
     rp = sub.add_parser("reap", help="deprecated no-op (IDA Nexus uses shared leases)")
@@ -582,8 +707,11 @@ def main(argv: list[str]) -> int:
     cp.add_argument("--mux", choices=MUXES, default="")
     cp.set_defaults(fn=capture)
 
-    kp = sub.add_parser("keys", help="inject real keystrokes into a pane "
-                                     "(tmux-style names, translated per mux)")
+    kp = sub.add_parser(
+        "keys",
+        help="inject real keystrokes into a pane "
+        "(tmux-style names, translated per mux)",
+    )
     kp.add_argument("keys", nargs="+", help="e.g. Escape, Enter, C-a, g m a i n")
     kp.add_argument("--pane")
     kp.add_argument("--sock", help="resolve the pane from this socket")