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
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
|
"""QEMU target lifecycle for green-eye.
Owns the ``qemu-system-riscv64`` process, the gdbstub endpoint, the serial
capture and the QEMU trace log. Both firmware paths are supported:
* ``kernel=`` fast path - QEMU loads an ELF linked for RAM and sets the PC;
* ``segments=`` probe path - each PT_LOAD segment is placed with a loader
device (raw), so a firmware linked for GPU physical addresses can be
relocated into RAM and observed.
"""
from __future__ import annotations
import os
import shutil
import signal
import socket
import subprocess
import tempfile
import time
from .errors import TargetError
DEFAULT_QEMU = "qemu-system-riscv64"
def find_qemu() -> str:
env = os.environ.get("GREENEYE_QEMU")
if env:
return env
found = shutil.which(DEFAULT_QEMU)
if not found:
raise TargetError(
f"{DEFAULT_QEMU} not found. Install it (Ubuntu: qemu-system-misc) "
"or set GREENEYE_QEMU to the binary."
)
return found
def free_port() -> int:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
sock.bind(("127.0.0.1", 0))
return sock.getsockname()[1]
class QemuTarget:
def __init__(
self,
*,
machine: str = "virt",
memory_mb: int = 256,
gdb_port: int = 0,
bios: str = "none",
extra_args: list[str] | None = None,
qemu: str | None = None,
kernel: str | None = None,
segments: list[tuple[int, bytes]] | None = None,
dlog: str | None = None,
uart_log: str | None = None,
qemu_log: str | None = None,
workdir: str | None = None,
keep_alive: bool = False,
):
self.qemu = qemu or find_qemu()
self.machine = machine
self.memory_mb = memory_mb
self.host = "127.0.0.1"
self.port = gdb_port or free_port()
self.bios = bios
self.extra_args = list(extra_args or [])
self.kernel = os.path.abspath(kernel) if kernel else None
self.segments = list(segments or [])
self.dlog = dlog
self.workdir = workdir or tempfile.mkdtemp(prefix="greeneye-qemu-")
os.makedirs(self.workdir, exist_ok=True)
self.uart_log = uart_log or os.path.join(self.workdir, "uart.log")
self.qemu_log = qemu_log or os.path.join(self.workdir, "qemu.log")
self.keep_alive = keep_alive
self.process: subprocess.Popen | None = None
self._segment_files: list[str] = []
self._cmd: list[str] = []
# -- lifecycle ---------------------------------------------------------
def _build_command(self, paused: bool) -> list[str]:
cmd = [
self.qemu,
"-machine",
self.machine,
"-m",
f"{self.memory_mb}M",
"-display",
"none",
"-monitor",
"none",
"-serial",
f"file:{self.uart_log}",
"-gdb",
f"tcp:{self.host}:{self.port}",
]
if self.bios:
cmd += ["-bios", self.bios]
if paused:
cmd += ["-S"]
if self.kernel:
cmd += ["-kernel", self.kernel]
for index, (addr, blob) in enumerate(self.segments):
seg_path = os.path.join(self.workdir, f"segment-{index:02d}-{addr:#x}.bin")
with open(seg_path, "wb") as fh:
fh.write(blob)
self._segment_files.append(seg_path)
cmd += ["-device", f"loader,file={seg_path},addr={addr:#x},force-raw=on"]
if self.dlog:
cmd += ["-d", self.dlog, "-D", self.qemu_log]
cmd += self.extra_args
return cmd
def start(self, paused: bool = True, timeout: float = 15.0) -> "QemuTarget":
self._cmd = self._build_command(paused)
try:
self.process = subprocess.Popen(
self._cmd,
stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE,
cwd=self.workdir,
start_new_session=True,
)
except OSError as exc:
raise TargetError(f"cannot start QEMU: {exc}") from exc
try:
self.wait_for_port(timeout=timeout)
except TargetError:
self.kill()
raise
return self
def wait_for_port(self, timeout: float = 15.0) -> None:
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
if self.process is not None and self.process.poll() is not None:
stderr = b""
if self.process.stderr is not None:
try:
stderr = self.process.stderr.read() or b""
except OSError:
pass
raise TargetError(
f"QEMU exited early rc={self.process.returncode}: "
f"{stderr.decode('utf-8', 'replace')[-2000:]}"
)
try:
with socket.create_connection((self.host, self.port), timeout=0.25):
return
except OSError:
time.sleep(0.05)
raise TargetError(f"gdbstub port {self.port} not ready after {timeout}s")
def alive(self) -> bool:
return self.process is not None and self.process.poll() is None
def kill(self, timeout: float = 5.0) -> None:
if self.process is None:
return
if self.process.poll() is None:
try:
os.killpg(os.getpgid(self.process.pid), signal.SIGTERM)
except (ProcessLookupError, PermissionError):
try:
self.process.terminate()
except ProcessLookupError:
pass
try:
self.process.wait(timeout=timeout)
except subprocess.TimeoutExpired:
try:
os.killpg(os.getpgid(self.process.pid), signal.SIGKILL)
except (ProcessLookupError, PermissionError):
try:
self.process.kill()
except ProcessLookupError:
pass
self.process.wait(timeout=timeout)
if self.process.stderr is not None:
try:
self.process.stderr.close()
except OSError:
pass
self.process = None
# -- observation -------------------------------------------------------
def read_uart(self) -> str:
try:
with open(self.uart_log, "rb") as fh:
return fh.read().decode("utf-8", "replace")
except FileNotFoundError:
return ""
def wait_for_uart(self, marker: str, timeout: float = 5.0, interval: float = 0.05) -> bool:
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
if marker in self.read_uart():
return True
time.sleep(interval)
return False
def read_qemu_log(self) -> str:
try:
with open(self.qemu_log, "rb") as fh:
return fh.read().decode("utf-8", "replace")
except FileNotFoundError:
return ""
@property
def command_line(self) -> list[str]:
return list(self._cmd)
def shell_command(self) -> str:
import shlex
return " ".join(shlex.quote(part) for part in self._cmd)
def __repr__(self) -> str: # pragma: no cover - cosmetic
return f"QemuTarget(port={self.port}, alive={self.alive()})"
|
