"""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()})"