diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/greeneye/rsp.py')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/greeneye/rsp.py | 583 |
1 files changed, 583 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/rsp.py b/debug/risc-v-gsp-rm-vm/greeneye/rsp.py new file mode 100644 index 0000000..0263c63 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/greeneye/rsp.py @@ -0,0 +1,583 @@ +"""Pure-Python GDB Remote Serial Protocol (RSP) client for QEMU's gdbstub. + +green-eye deliberately avoids GDB: the whole instrumentation stack is a Python +library, so sessions, probes and plugins are scriptable end to end. This +module implements the RSP subset QEMU needs: packet framing/acks, no-ack mode, +registers (with layout read from ``target.xml``), memory, software/hardware +breakpoints, watchpoints, continue/step/interrupt and monitor commands. + +Tested against QEMU 8.2 (``qemu-system-riscv64``). +""" +from __future__ import annotations + +import socket +import time +from dataclasses import dataclass, field as _field + +from .errors import RspProtocolError, RspTimeout + +# --------------------------------------------------------------------------- +# Packet framing +# --------------------------------------------------------------------------- + + +def checksum(data: bytes) -> int: + """RSP 8-bit checksum: sum of bytes mod 256.""" + return sum(data) & 0xFF + + +def encode_packet(data: bytes) -> bytes: + """Frame a payload as ``$<data>#<checksum>``.""" + return b"$" + data + b"#" + f"{checksum(data):02x}".encode("ascii") + + +def _hex_value(payload: str) -> int: + try: + return int(payload, 16) + except ValueError as exc: # pragma: no cover - defensive + raise RspProtocolError(f"malformed hex value: {payload!r}") from exc + + +# --------------------------------------------------------------------------- +# Stop replies +# --------------------------------------------------------------------------- + + +@dataclass +class StopReply: + """A parsed stop reply (``T``/``S``/``W``/``X`` packet).""" + + raw: str + signal: int | None = None + thread: str | None = None + regs: dict[int, int] = _field(default_factory=dict) + watch_addr: int | None = None + watch_kind: str | None = None + exit_status: int | None = None + + @property + def is_exit(self) -> bool: + return self.raw[:1] in ("W", "X") + + @property + def is_stop(self) -> bool: + return self.raw[:1] in ("T", "S") + + def __str__(self) -> str: # pragma: no cover - cosmetic + if self.is_exit: + return f"exit(status={self.exit_status})" + bits = [f"signal={self.signal}"] + if self.watch_addr is not None: + bits.append(f"{self.watch_kind}@{self.watch_addr:#x}") + if self.thread: + bits.append(f"thread={self.thread}") + return "stop(" + ", ".join(bits) + ")" + + +def parse_stop_reply(raw: str) -> StopReply: + """Parse a stop packet body (without the leading channel).""" + reply = StopReply(raw=raw) + if not raw: + return reply + head = raw[0] + if head in ("T", "S"): + if len(raw) >= 3: + try: + reply.signal = int(raw[1:3], 16) + except ValueError: + pass + rest = raw[3:] + for item in rest.split(";"): + if not item or ":" not in item: + continue + key, value = item.split(":", 1) + if key == "thread": + reply.thread = value + elif key in ("watch", "rwatch", "awatch"): + reply.watch_kind = key + reply.watch_addr = _hex_value(value) + else: + try: + reply.regs[int(key, 16)] = _hex_value(value) + except ValueError: + continue + elif head in ("W", "X"): + if len(raw) >= 3: + try: + reply.exit_status = int(raw[1:3], 16) + except ValueError: + pass + return reply + + +# --------------------------------------------------------------------------- +# Register layout +# --------------------------------------------------------------------------- + + +@dataclass +class Register: + number: int + name: str + bitsize: int + offset: int = 0 + + @property + def nbytes(self) -> int: + return (self.bitsize + 7) // 8 + + +#: Fallback layout for RV64 when target.xml is unavailable: x0..x31 + pc. +RV64_FALLBACK_REGS = [ + Register(n, name, 64, n * 8) + for n, name in enumerate( + ["x0", "ra", "sp", "gp", "tp", "t0", "t1", "t2"] + + [f"s{i}" for i in range(0, 2)] + + [f"a{i}" for i in range(0, 8)] + + [f"s{i}" for i in range(2, 12)] + + [f"t{i}" for i in range(3, 7)] + + ["pc"] + ) +] + + +# --------------------------------------------------------------------------- +# Client +# --------------------------------------------------------------------------- + + +class RspClient: + """A blocking RSP client over TCP.""" + + def __init__(self, host: str = "127.0.0.1", port: int = 4444, timeout: float = 10.0): + self.host = host + self.port = port + self.timeout = timeout + self.sock: socket.socket | None = None + self.no_ack = False + self.packet_size = 0x1000 + self.registers: list[Register] = [] + self._by_name: dict[str, Register] = {} + self._by_number: dict[int, Register] = {} + self.features: str = "" + + # -- lifecycle --------------------------------------------------------- + + def connect(self, retries: int = 1, retry_delay: float = 0.1) -> None: + last: Exception | None = None + for _ in range(max(1, retries)): + try: + self.sock = socket.create_connection((self.host, self.port), timeout=self.timeout) + self.sock.settimeout(self.timeout) + return + except OSError as exc: + last = exc + time.sleep(retry_delay) + raise RspTimeout(f"cannot connect to gdbstub {self.host}:{self.port}: {last}") + + def close(self) -> None: + if self.sock is not None: + try: + self.sock.close() + except OSError: + pass + self.sock = None + + @property + def connected(self) -> bool: + return self.sock is not None + + def __enter__(self) -> "RspClient": + if not self.connected: + self.connect() + return self + + def __exit__(self, *exc) -> None: + self.close() + + # -- low level I/O ----------------------------------------------------- + + def _send_raw(self, data: bytes) -> None: + if self.sock is None: + raise RspTimeout("not connected") + self.sock.sendall(data) + + def _read_exact(self, count: int, deadline: float) -> bytes: + if self.sock is None: + raise RspTimeout("not connected") + chunks = bytearray() + while len(chunks) < count: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise RspTimeout(f"timeout waiting for {count} bytes") + self.sock.settimeout(remaining) + try: + part = self.sock.recv(count - len(chunks)) + except socket.timeout as exc: + raise RspTimeout(f"timeout waiting for {count} bytes") from exc + if not part: + raise RspProtocolError("gdbstub closed the connection") + chunks += part + return bytes(chunks) + + def _send_packet(self, payload: bytes) -> None: + packet = encode_packet(payload) + self._send_raw(packet) + if self.no_ack: + return + deadline = time.monotonic() + self.timeout + while True: + ack = self._read_exact(1, deadline) + if ack == b"+": + return + if ack == b"-": + self._send_raw(packet) + continue + # Some stubs answer with a stop packet instead of an ack when a + # run-control command stops immediately; push it back for the + # packet reader by buffering is complex, so treat as protocol + # error only if it is not a start-of-packet marker. + if ack == b"$": + raise RspProtocolError("unexpected packet while waiting for ack") + + def _read_packet(self, timeout: float | None = None) -> str: + eff = self.timeout if timeout is None else timeout + deadline = time.monotonic() + eff + # Skip stale acks and interrupt notifications until a packet starts. + while True: + byte = self._read_exact(1, deadline) + if byte == b"$": + break + if byte == b"+": + continue + if byte == b"\x03": + return "S02" # async interrupt notification + body = bytearray() + while True: + byte = self._read_exact(1, deadline) + if byte == b"#": + break + body += byte + got = self._read_exact(2, deadline) + try: + expected = int(got.decode("ascii"), 16) + except ValueError as exc: + raise RspProtocolError(f"malformed checksum bytes {got!r}") from exc + if checksum(bytes(body)) != expected: + if not self.no_ack: + self._send_raw(b"-") + raise RspProtocolError(f"bad checksum for {bytes(body)!r}") + if not self.no_ack: + self._send_raw(b"+") + return bytes(body).decode("latin1") + + def command(self, payload: str | bytes, timeout: float | None = None) -> str: + """Send one packet and return the reply payload.""" + data = payload.encode("ascii") if isinstance(payload, str) else payload + self._send_packet(data) + return self._read_packet(timeout) + + # -- handshake --------------------------------------------------------- + + def sync(self) -> None: + """Negotiate features and register layout; consume the initial stop.""" + # QEMU acknowledges QStartNoAckMode with an EMPTY packet ($#00), then + # switches to no-ack mode once the client acks that reply. + try: + self._send_packet(b"QStartNoAckMode") + reply = self._read_packet() + if reply in ("", "OK"): + self.no_ack = True + except RspProtocolError: + pass # older stub: stay in ack mode + + self.packet_size = 0x1000 + try: + self.features = self.command("qSupported:multiprocess+;swbreak+;hwbreak+;qXfer:features:read+") + for token in self.features.split(";"): + if token.startswith("PacketSize="): + try: + self.packet_size = int(token.split("=", 1)[1], 16) + except ValueError: + pass + except RspProtocolError: + self.features = "" + + # Consume the initial stop reply (target is paused). + try: + self.command("?") + except RspProtocolError: + pass + # Select CPU 0 for register access in system mode. + for selection in ("Hg0", "Hc-1"): + try: + self.command(selection) + except RspProtocolError: + pass + self._load_registers() + + def _load_registers(self) -> None: + self.registers = [] + try: + xml = self._read_target_xml() + except RspProtocolError: + xml = "" + if xml: + self.registers = _parse_target_xml(xml) + if not self.registers: + self.registers = list(RV64_FALLBACK_REGS) + self._by_name = {r.name: r for r in self.registers} + self._by_number = {r.number: r for r in self.registers} + + def _read_target_xml(self, depth: int = 0) -> str: + """Fetch target.xml and recursively inline its xi:include files.""" + return self._collect_annex("target.xml", depth) + + def _collect_annex(self, annex: str, depth: int) -> str: + import re + + try: + xml = self._read_xfer("features", annex) + except RspProtocolError: + return "" + if depth >= 4: + return xml + parts = [xml] + for href in re.findall(r'xi:include[^>]*href="([^"]+)"', xml): + if href.startswith(("http:", "https:", "file:")): + continue + child = self._collect_annex(href, depth + 1) + if child: + parts.append(child) + return "\n".join(parts) + + def _read_xfer(self, kind: str, annex: str) -> str: + data = bytearray() + offset = 0 + while True: + reply = self.command(f"qXfer:{kind}:read:{annex}:{offset:x},fff") + if reply.startswith("E"): + raise RspProtocolError(f"qXfer {kind}:{annex} failed: {reply}") + if not reply or reply[0] not in ("l", "m"): + raise RspProtocolError(f"qXfer {kind}:{annex} malformed: {reply[:16]!r}") + chunk = reply[1:].encode("latin1") + data += chunk + offset += len(chunk) + if reply[0] == "l": + break + return data.decode("latin1", "replace") + + # -- registers --------------------------------------------------------- + + def read_registers(self) -> dict[str, int]: + payload = self.command("g") + raw = _unhex(payload) + out: dict[str, int] = {} + for reg in self.registers: + start = reg.offset + chunk = raw[start : start + reg.nbytes] + if len(chunk) < reg.nbytes: + break + out[reg.name] = int.from_bytes(chunk, "little", signed=False) + return out + + def read_reg(self, name_or_number: str | int) -> int: + reg = self._reg(name_or_number) + reply = self.command(f"p{reg.number:x}") + if reply.startswith("E"): + raise RspProtocolError(f"read register {reg.name} failed: {reply}") + return int.from_bytes(_unhex(reply), "little", signed=False) + + def write_reg(self, name_or_number: str | int, value: int) -> None: + reg = self._reg(name_or_number) + payload = value.to_bytes(reg.nbytes, "little", signed=False).hex() + reply = self.command(f"P{reg.number:x}={payload}") + if reply != "OK": + raise RspProtocolError(f"write register {reg.name} failed: {reply}") + + def _reg(self, name_or_number: str | int) -> Register: + if isinstance(name_or_number, int): + if name_or_number in self._by_number: + return self._by_number[name_or_number] + # allow ABI-numbered convenience: 0..31 map to x0..x31 + if name_or_number < len(self.registers): + return self.registers[name_or_number] + raise RspProtocolError(f"unknown register number {name_or_number}") + if name_or_number in self._by_name: + return self._by_name[name_or_number] + aliases = {"pc": "pc", "a0": "a0", "return": "ra"} + name_or_number = aliases.get(name_or_number, name_or_number) + if name_or_number in self._by_name: + return self._by_name[name_or_number] + raise RspProtocolError(f"unknown register {name_or_number!r}") + + @property + def pc(self) -> int: + return self.read_reg("pc") + + # -- memory ------------------------------------------------------------ + + def _mem_chunk(self) -> int: + # Each data byte costs two hex chars; leave room for the header. + return max(64, (self.packet_size - 128) // 2) + + def read_mem(self, addr: int, size: int, chunk: int = 0) -> bytes: + if chunk <= 0: + chunk = self._mem_chunk() + out = bytearray() + while len(out) < size: + want = min(chunk, size - len(out)) + here = addr + len(out) + reply = self.command(f"m{here:x},{want:x}") + if reply.startswith("E"): + raise RspProtocolError(f"memory read {here:#x}+{want:#x} failed: {reply}") + data = _unhex(reply) + if not data: + raise RspProtocolError(f"short memory read at {here:#x}") + out += data + return bytes(out) + + def write_mem(self, addr: int, data: bytes, chunk: int = 0) -> None: + if chunk <= 0: + chunk = self._mem_chunk() + view = memoryview(data) + offset = 0 + while offset < len(view): + part = bytes(view[offset : offset + chunk]) + here = addr + offset + reply = self.command(f"M{here:x},{len(part):x}:{part.hex()}") + if reply != "OK": + raise RspProtocolError(f"memory write {here:#x} failed: {reply}") + offset += len(part) + + def read_u32(self, addr: int) -> int: + return int.from_bytes(self.read_mem(addr, 4), "little") + + def read_u64(self, addr: int) -> int: + return int.from_bytes(self.read_mem(addr, 8), "little") + + def write_u32(self, addr: int, value: int) -> None: + self.write_mem(addr, value.to_bytes(4, "little")) + + def write_u64(self, addr: int, value: int) -> None: + self.write_mem(addr, value.to_bytes(8, "little")) + + def read_cstring(self, addr: int, max_len: int = 256) -> str: + raw = self.read_mem(addr, max_len) + return raw.split(b"\x00", 1)[0].decode("utf-8", "replace") + + # -- breakpoints / watchpoints ---------------------------------------- + + def set_sw_break(self, addr: int, kind: int = 4) -> None: + reply = self.command(f"Z0,{addr:x},{kind:x}") + if reply != "OK": + raise RspProtocolError(f"set software breakpoint {addr:#x} failed: {reply}") + + def clear_sw_break(self, addr: int, kind: int = 4) -> None: + reply = self.command(f"z0,{addr:x},{kind:x}") + if reply != "OK": + raise RspProtocolError(f"clear software breakpoint {addr:#x} failed: {reply}") + + def set_hw_break(self, addr: int, kind: int = 4) -> None: + reply = self.command(f"Z1,{addr:x},{kind:x}") + if reply != "OK": + raise RspProtocolError(f"set hardware breakpoint {addr:#x} failed: {reply}") + + def clear_hw_break(self, addr: int, kind: int = 4) -> None: + reply = self.command(f"z1,{addr:x},{kind:x}") + if reply != "OK": + raise RspProtocolError(f"clear hardware breakpoint {addr:#x} failed: {reply}") + + _WATCH_KINDS = {"w": 2, "r": 3, "a": 4} + + def set_watch(self, addr: int, size: int = 8, kind: str = "w") -> None: + code = self._WATCH_KINDS[kind] + reply = self.command(f"Z{code},{addr:x},{size:x}") + if reply != "OK": + raise RspProtocolError(f"set watchpoint {kind}@{addr:#x} failed: {reply}") + + def clear_watch(self, addr: int, size: int = 8, kind: str = "w") -> None: + code = self._WATCH_KINDS[kind] + reply = self.command(f"z{code},{addr:x},{size:x}") + if reply != "OK": + raise RspProtocolError(f"clear watchpoint {kind}@{addr:#x} failed: {reply}") + + # -- run control ------------------------------------------------------- + + def continue_(self, timeout: float | None = None) -> StopReply: + self._send_packet(b"c") + raw = self._read_packet(timeout) + return parse_stop_reply(raw) + + def step(self, timeout: float | None = None) -> StopReply: + self._send_packet(b"s") + raw = self._read_packet(timeout) + return parse_stop_reply(raw) + + def interrupt(self, timeout: float | None = None) -> StopReply: + self._send_raw(b"\x03") + raw = self._read_packet(timeout) + return parse_stop_reply(raw) + + # -- monitor ----------------------------------------------------------- + + def monitor(self, cmd: str) -> str: + payload = ("qRcmd," + cmd.encode("ascii").hex()).encode("ascii") + self._send_packet(payload) + out = bytearray() + while True: + reply = self._read_packet() + if reply == "OK": + break + if reply.startswith("E"): + raise RspProtocolError(f"monitor {cmd!r} failed: {reply}") + if reply.startswith("O"): + out += bytes.fromhex(reply[1:]) + else: # pragma: no cover - defensive + raise RspProtocolError(f"unexpected monitor reply: {reply[:32]!r}") + return out.decode("utf-8", "replace") + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _unhex(payload: str) -> bytes: + if len(payload) % 2: + payload = payload[:-1] # tolerate odd trailing nibble + try: + return bytes.fromhex(payload) + except ValueError as exc: + raise RspProtocolError(f"malformed hex payload: {payload[:32]!r}") from exc + + +_RE_REG = None + + +def _parse_target_xml(xml: str) -> list[Register]: + import re + + global _RE_REG + if _RE_REG is None: + _RE_REG = re.compile(r"<reg\b[^>]*?/?>", re.IGNORECASE) + attrs_re = re.compile(r'(\w+)\s*=\s*"([^"]*)"') + regs: list[Register] = [] + next_number = 0 + offset = 0 + for match in _RE_REG.finditer(xml): + attrs = {k: v for k, v in attrs_re.findall(match.group(0))} + name = attrs.get("name") + if not name: + continue + bitsize = int(attrs.get("bitsize", "64")) + number = int(attrs["regnum"], 0) if "regnum" in attrs else next_number + if "offset" in attrs: + try: + offset = int(attrs["offset"], 0) + except ValueError: + pass + regs.append(Register(number=number, name=name, bitsize=bitsize, offset=offset)) + offset += (bitsize + 7) // 8 + next_number = number + 1 + return regs |