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
|
"""Session recording: events.jsonl + snapshots + summary.
A recorded session is the durable artifact risc-v-gsp-rm-debugger produces: a complete,
symbol-annotatable map of what the firmware did (hook hits, watch hits,
memory writes, snapshots, coverage steps) that can be replayed offline,
diffed between runs (bare-metal vs VFIO-mdev, old GSP vs new GSP) and
annotated back into Ghidra.
"""
from __future__ import annotations
import json
import os
import time
import zlib
FORMAT = "greeneye-session/1"
class NullRecorder:
dir: str | None = None
def event(self, event_type: str, **fields) -> None:
pass
def snapshot(self, label: str, addr: int, data: bytes) -> None:
pass
def blob(self, name: str, data: bytes) -> None:
pass
def close(self) -> None:
pass
class Recorder:
def __init__(self, root: str, *, name: str = "session", meta: dict | None = None):
self.dir = root
os.makedirs(self.dir, exist_ok=True)
self.name = name
self.meta = dict(meta or {})
self.meta.update({"format": FORMAT, "name": name, "created": time.time(), "greeneye": _version()})
self._events_path = os.path.join(self.dir, "events.jsonl")
self._fh = open(self._events_path, "w", encoding="utf-8", buffering=1)
self._t0 = time.monotonic()
self._counts: dict[str, int] = {}
self._snapshots: list[dict] = []
self._closed = False
@classmethod
def new(cls, *, base: str | None = None, name: str = "session", meta: dict | None = None) -> "Recorder":
base = base or os.environ.get("GREENEYE_DUMPS") or os.path.join(os.getcwd(), "dumps")
stamp = time.strftime("%Y%m%d-%H%M%S")
return cls(os.path.join(base, f"{stamp}-{name}"), name=name, meta=meta)
# -- writing -----------------------------------------------------------
def event(self, event_type: str, **fields) -> None:
if self._closed:
return
record = {"t": round(time.monotonic() - self._t0, 6), "type": event_type}
for key, value in fields.items():
record[key] = _jsonable(value)
self._fh.write(json.dumps(record, sort_keys=True) + "\n")
self._counts[event_type] = self._counts.get(event_type, 0) + 1
def snapshot(self, label: str, addr: int, data: bytes) -> str:
rel = os.path.join("snapshots", f"{label}-{addr:#x}-{len(data)}B.bin")
path = os.path.join(self.dir, rel)
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "wb") as fh:
fh.write(zlib.compress(data))
info = {"label": label, "addr": addr, "size": len(data), "path": rel, "sha256": _sha(data)}
self._snapshots.append(info)
self.event("snapshot_saved", **info)
return rel
def blob(self, name: str, data: bytes) -> str:
rel = os.path.join("blobs", name)
path = os.path.join(self.dir, rel)
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "wb") as fh:
fh.write(data)
self.event("blob_saved", path=rel, size=len(data))
return rel
# -- closing -----------------------------------------------------------
def close(self, write_summary: bool = True) -> None:
if self._closed:
return
self.event("session_end")
self._fh.close()
self._closed = True
self.meta["counts"] = self._counts
self.meta["snapshots"] = self._snapshots
with open(os.path.join(self.dir, "session.json"), "w", encoding="utf-8") as fh:
json.dump(self.meta, fh, indent=1, sort_keys=True)
if write_summary:
try:
from .analyze import render_summary
report = render_summary(self.dir)
with open(os.path.join(self.dir, "summary.md"), "w", encoding="utf-8") as fh:
fh.write(report)
except Exception as exc: # pragma: no cover - never break close
with open(os.path.join(self.dir, "summary-error.txt"), "w", encoding="utf-8") as fh:
fh.write(f"summary failed: {exc}\n")
def __enter__(self) -> "Recorder":
return self
def __exit__(self, *exc) -> None:
self.close()
def load_events(session_dir: str) -> list[dict]:
path = session_dir
if os.path.isdir(session_dir):
path = os.path.join(session_dir, "events.jsonl")
events: list[dict] = []
with open(path, "r", encoding="utf-8") as fh:
for line in fh:
line = line.strip()
if line:
events.append(json.loads(line))
return events
def load_meta(session_dir: str) -> dict:
with open(os.path.join(session_dir, "session.json"), "r", encoding="utf-8") as fh:
return json.load(fh)
def _jsonable(value):
if isinstance(value, bytes):
return value.hex()
if isinstance(value, (set, tuple)):
return list(value)
if isinstance(value, dict):
return {str(k): _jsonable(v) for k, v in value.items()}
return value
def _sha(data: bytes) -> str:
import hashlib
return hashlib.sha256(data).hexdigest()
def _version() -> str:
try:
from . import __version__
return __version__
except Exception: # pragma: no cover
return "unknown"
|
