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
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
|
"""Declarative probe specifications (the Frida-script-like layer).
A probe spec is JSON so it can be generated by another tool (or an AI
harness) and replayed unchanged:
{
"probes": [
{
"name": "sum-entry",
"symbol": "rpc_sum",
"when": "a1 > 0",
"on_enter": {
"log": ["a0", "a1", "a2"],
"dump": [{"reg": "a0", "len": 32, "label": "sum-arg"}],
"note": "sum called"
},
"on_leave": {"log": ["retval"]}
}
]
}
Conditions are evaluated over a register dictionary (``a0``..``a7``, ``pc``,
``ra``, or raw ``x10`` names). The evaluator is deliberately tiny: it only
accepts comparisons/boolean arithmetic over those names and integer
literals, never arbitrary Python.
"""
from __future__ import annotations
import json
import re
from .errors import GreenEyeError
_ALLOWED_CHARS = re.compile(r"^[\s0-9a-fA-FxX<>=!()+\-*/%&|^~]*$")
_NAME_RE = re.compile(r"\b[A-Za-z_][A-Za-z0-9_]*\b")
_ALLOWED_NAMES = re.compile(
r"^(retval|pc|ra|sp|gp|tp|fp|and|or|not|True|False|None"
r"|[a-z][0-9]?|x[0-9]{1,2}|[stga][0-9]{1,2})$"
)
def load_spec(path: str) -> dict:
with open(path, "r", encoding="utf-8") as fh:
spec = json.load(fh)
if "probes" not in spec or not isinstance(spec["probes"], list):
raise GreenEyeError("probe spec must be an object with a 'probes' list")
return spec
def compile_condition(expr: str | None):
if not expr:
return None
for name in _NAME_RE.findall(expr):
if not _ALLOWED_NAMES.match(name):
raise GreenEyeError(f"probe condition references disallowed name {name!r}")
if not _ALLOWED_CHARS.match(_NAME_RE.sub("0", expr)):
raise GreenEyeError(f"probe condition contains disallowed characters: {expr!r}")
code = compile(expr, "<probe-when>", "eval")
def check(regs: dict) -> bool:
scope = {
**{k: v for k, v in regs.items() if isinstance(v, int)},
"retval": regs.get("a0", 0),
}
try:
return bool(eval(code, {"__builtins__": {}}, scope)) # noqa: S307 - whitelisted above
except Exception:
return False
return check
def install(session, spec: dict) -> list:
"""Install every probe in the spec; returns the HookHandles."""
handles = []
for entry in spec["probes"]:
target = entry.get("symbol") or entry.get("addr")
if target is None:
raise GreenEyeError(f"probe {entry.get('name', '?')}: needs 'symbol' or 'addr'")
when = compile_condition(entry.get("when"))
name = entry.get("name", str(target))
on_enter_actions = entry.get("on_enter")
on_leave_actions = entry.get("on_leave")
def make_enter(name=name, when=when, actions=on_enter_actions):
def callback(hit):
if when is not None and not when(hit.regs):
return
_run_actions(hit, name, actions)
return callback
def make_leave(name=name, actions=on_leave_actions):
def callback(hit):
_run_actions(hit, name, actions, is_leave=True)
return callback
handle = session.hook(
target,
on_enter=make_enter() if on_enter_actions else None,
on_leave=make_leave() if on_leave_actions else None,
listener=f"probe:{name}",
)
session.note(f"probe installed: {name}", target=str(target))
handles.append(handle)
return handles
def _run_actions(hit, probe_name: str, actions: dict | None, is_leave: bool = False) -> None:
session = hit.session
if actions is None:
return
values = {}
for expr in actions.get("log", []):
values[expr] = _read_value(hit, expr)
if values:
session.recorder.event(
"probe_log",
probe=probe_name,
phase="leave" if is_leave else "enter",
addr=hit.addr,
symbol=hit.symbol.name if hit.symbol else None,
values=values,
)
for dump in actions.get("dump", []):
addr = _read_value(hit, dump.get("reg", "a0"))
if isinstance(addr, str):
continue
length = int(dump.get("len", 64))
label = dump.get("label", f"{probe_name}-dump")
try:
data = session.memory.read(addr & 0xFFFFFFFFFFFFFFFF, length)
except Exception as exc:
session.recorder.event("probe_dump_error", probe=probe_name, addr=addr, error=str(exc))
continue
path = session.recorder.blob(f"{label}-{addr:#x}.bin", data)
session.recorder.event(
"probe_dump",
probe=probe_name,
name=label,
addr=addr,
size=len(data),
path=path,
preview=data[:32].hex(),
)
for snapshot in actions.get("snapshot", []):
addr = _read_value(hit, snapshot.get("reg", "a0"))
if isinstance(addr, str):
continue
session.snapshot(addr & 0xFFFFFFFFFFFFFFFF, int(snapshot.get("len", 4096)), name=snapshot.get("label", probe_name))
if actions.get("note"):
session.note(actions["note"], probe=probe_name, addr=hit.addr)
def _read_value(hit, expr: str):
if expr == "retval":
return hit.retval if hit.retval is not None else hit.regs.get("a0", 0)
if expr in hit.regs:
return hit.regs[expr]
if expr.startswith("0x"):
return int(expr, 16)
try:
return int(expr, 0)
except ValueError:
return f"<unknown:{expr}>"
|
