1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
|
"""Stdlib-only client for the idatui RPC socket. Drive the TUI from another pane.
# point at the socket the TUI was launched with (--rpc), or set IDATUI_RPC_SOCK
python -m idatui.rpcclient --sock /tmp/ida.sock state
python -m idatui.rpcclient goto target=main
python -m idatui.rpcclient keys g m a i n enter # 'keys' takes positionals
python -m idatui.rpcclient text "hello world" delay_ms=40
python -m idatui.rpcclient move dir=down n=5
python -m idatui.rpcclient screen # prints the raw screen text
Also usable as a library:
from idatui.rpcclient import RpcClient
with RpcClient(sock) as c:
c.call("goto", target="main")
No auth: whoever can r/w the socket drives the app.
"""
from __future__ import annotations
import json
import os
import socket
import sys
from typing import Any
class RpcClient:
#: Default read timeout (s). Bounds any single call so a slow/hung server
#: (e.g. Hex-Rays grinding on an undecompilable function) can't block the
#: CLI forever. Override via ctor or the IDATUI_RPC_TIMEOUT env var.
DEFAULT_TIMEOUT = 90.0
def __init__(self, sock_path: str, timeout: float | None = None):
self.path = sock_path
self._sock: socket.socket | None = None
self._buf = b""
self._id = 0
if timeout is None:
env = os.environ.get("IDATUI_RPC_TIMEOUT")
timeout = float(env) if env else self.DEFAULT_TIMEOUT
self.timeout = timeout
def __enter__(self) -> "RpcClient":
self.connect()
return self
def __exit__(self, *exc) -> None:
self.close()
def connect(self) -> None:
s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
s.connect(self.path)
if self.timeout and self.timeout > 0:
s.settimeout(self.timeout)
self._sock = s
def close(self) -> None:
if self._sock is not None:
self._sock.close()
self._sock = None
def call(self, method: str, **params: Any) -> Any:
assert self._sock is not None, "not connected"
self._id += 1
req = {"id": self._id, "method": method, "params": params}
self._sock.sendall(json.dumps(req).encode() + b"\n")
while b"\n" not in self._buf:
try:
chunk = self._sock.recv(65536)
except socket.timeout as e:
raise RpcError(
f"no response for {method!r} within {self.timeout}s "
f"(server busy or the op is hung, e.g. a failed decompile)"
) from e
if not chunk:
raise ConnectionError("server closed the connection")
self._buf += chunk
line, self._buf = self._buf.split(b"\n", 1)
resp = json.loads(line.decode())
if "error" in resp and resp["error"]:
raise RpcError(resp["error"].get("message", "rpc error"))
return resp.get("result")
class RpcError(Exception):
pass
def _coerce(v: str) -> Any:
"""Params are strings on the CLI; only booleans need coercing (the server
int()s numeric params itself, and addresses/names must stay strings)."""
low = v.lower()
if low == "true":
return True
if low == "false":
return False
if low == "null":
return None
return v
def main(argv: list[str]) -> int:
sock = os.environ.get("IDATUI_RPC_SOCK")
args = list(argv)
if args and args[0] == "--sock":
sock = args[1]
args = args[2:]
if not sock:
print("error: no socket (pass --sock PATH or set IDATUI_RPC_SOCK)",
file=sys.stderr)
return 2
if not args:
print("error: no method given", file=sys.stderr)
return 2
method, rest = args[0], args[1:]
params: dict[str, Any] = {}
if method == "keys":
params["keys"] = rest # every positional is a key name
elif method == "text" and rest and "=" not in rest[0]:
# first positional is the literal text; the rest may be key=value
params["text"] = rest[0]
for tok in rest[1:]:
k, _, val = tok.partition("=")
params[k] = _coerce(val)
else:
for tok in rest:
k, _, val = tok.partition("=")
if not _:
print(f"error: expected key=value, got {tok!r}", file=sys.stderr)
return 2
params[k] = _coerce(val)
try:
with RpcClient(sock) as c:
result = c.call(method, **params)
except (OSError, RpcError, ConnectionError) as e:
print(f"error: {e}", file=sys.stderr)
return 1
# 'screen' is meant to be read as text; everything else as JSON.
if method == "screen" and isinstance(result, dict) and "text" in result:
sys.stdout.write(result["text"])
else:
print(json.dumps(result, indent=2))
return 0
if __name__ == "__main__":
raise SystemExit(main(sys.argv[1:]))
|
