diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/harness/log_decode.py')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/harness/log_decode.py | 660 |
1 files changed, 660 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/harness/log_decode.py b/debug/risc-v-gsp-rm-vm/harness/log_decode.py new file mode 100644 index 0000000..e1d351c --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/log_decode.py @@ -0,0 +1,660 @@ +#!/usr/bin/env python3 +"""Decode the two real GSP-RM log dump forms into one normalized record stream. + +The live GSP-RM writes its "server stderr" into an 8 KiB-word libos log ring. +Two dump forms exist in the campaign evidence and they are the inputs here: + +1. Raw ring dumps (debugfs ``gsp-phys-log-dump`` saved to evidence files). + A ``<LABEL>-raw: <n> bytes (...) VA=... dma=...`` header is followed by + hexdump rows ``00000000: 000002d2 00000000 ...`` (little-endian u64 words). + Each header's rows decode into one ``raw-hex`` record whose ``frames`` are + the u64 words and whose ``raw`` is the decoded byte buffer as lowercase + hex. A later ``<LABEL>-decoded:`` marker switches to the offline + liblogdecode text records ``rec 449 ts=0x.. meta=0x.. args=0x..`` which + become ``record`` entries (``args`` -> ``frames``); both the + ``args=[0x.., 0x..]`` and ``args=0x..,0x..`` spellings are accepted. + +2. Live dmesg / evidence lines, e.g.:: + + vgsp-nocat: rec seq=1 recType=5 errorCode=0x57f1344 subsystem=0x0 bugcheck=0x0 payload=1212 source="ASSERT" faultingEngine="" + vgsp-nocat: rec seq=1 assertCallStack count=1 [0x401000,0x402000] (offsets from load base 0x4000000) + vgsp-nocat: entry[0]: count=2 recType=5 errorCode=0x57f1344 firstSeq=1 lastSeq=2 source="ASSERT" + vgsp: WARNING: ... (NOCAT recType=5 errorCode=0x52a8b04) ... + rec 449 ts=0x18dbb29118ec6200 meta=0x0000000005bb9c48 args=0x1,0x2 + + ``key=value`` pairs fill ``rec_type`` / ``error_code`` / ``source_name`` / + ``seq``; ``assertCallStack [..]`` lines attach their bracketed frame list + to the immediately preceding record with the same ``seq``; anything that + matches none of the shapes is counted in ``unparsed`` (never dropped). + Lines starting with ``#``, ``<LABEL>-raw:``/``<LABEL>-decoded:`` headers and + blank lines are structure, not records. + +Report schema ``greeneye-log-decode/1``:: + + {"schema": "greeneye-log-decode/1", + "source": "<basename of the input>", + "count": <records after filters>, + "unparsed": <lines that matched no shape>, + "records": [{"kind": "raw-hex"|"record", "label"?: str, + "rec_type"?: int|str, "error_code"?: "0x...", + "source_name"?: str, "seq"?: int, + "frames": ["0x...", ...], "raw": str, + "resolved"?: [{"frame": "0x...", "symbol": str|None, + "offset": "0x..."|None}, ...]}], + "histogram": {"rec_type": {..}, "error_code": {..}, "source": {..}}} + +``resolved`` appears only when a symbol table is supplied and is aligned 1:1 +with ``frames``. ``--symbols`` accepts the project symbol-DB shape: a +top-level list, or an object with a ``symbols`` list, of entries +``{"name": str, "addr": int | "0x..."}`` (extra keys ignored). Each frame is +matched to the nearest symbol with ``addr <= frame`` and the offset +(``frame - addr``) is reported; an unreadable or malformed symbols file exits +non-zero with a clear message. + +Filtering (``decode`` subcommand): ``--kind`` selects raw-hex/record, +``--rec-type``/``--error-code`` accept decimal or ``0x`` hex, ``--source`` is +a case-insensitive substring on ``source_name`` and ``--limit`` caps the +records; ``count``/``histogram`` describe the filtered stream. ``--json +out.json`` writes the full report. + +Honesty: offline decoder only (L1 at best; no timing claims). Raw words are +exposed, not structurally ring-decoded; ``assertCallStack`` frames are offsets +from the load base named in the line and are symbolized as given. There is no +hardware, module, QEMU or guest interaction. + +Run:: + + python3 harness/log_decode.py decode <dump> [--symbols symbols.json] + python3 harness/log_decode.py demo +""" +from __future__ import annotations + +import argparse +import bisect +import json +import os +import re +import sys +import tempfile +from typing import Any + +SCHEMA = "greeneye-log-decode/1" +DEMO_FORMAT = "greeneye-log-decode-demo/1" + +#: How many ``resolved`` entries the human-readable output prints per record +#: (the JSON report always carries all of them). +RESOLVED_PRINT_LIMIT = 8 + + +class LogDecodeError(ValueError): + """An input or request the decoder cannot honour.""" + + +# --------------------------------------------------------------------------- +# Symbols: the project symbol-DB shape, nearest-below resolution +# --------------------------------------------------------------------------- + +def _parse_sym_addr(value: Any, where: str) -> int: + if isinstance(value, bool): + raise LogDecodeError(f"{where}: bad symbol addr {value!r}") + if isinstance(value, int): + return value + if isinstance(value, str): + try: + return int(value, 0) + except ValueError as exc: + raise LogDecodeError(f"{where}: bad symbol addr {value!r}") from exc + raise LogDecodeError(f"{where}: bad symbol addr {value!r}") + + +class SymbolTable: + """Nearest-below ``frame -> (name, offset)`` resolver. + + Holds the same entries the project symbol DBs carry; only ``addr`` and + ``name`` are used here. ``resolve`` returns ``None`` when no symbol has an + address at or below the frame. + """ + + def __init__(self, entries: list[tuple[int, str]]) -> None: + self._entries = sorted(entries, key=lambda item: item[0]) + self._addrs = [addr for addr, _ in self._entries] + + def __len__(self) -> int: + return len(self._entries) + + def resolve(self, addr: int) -> tuple[str, int] | None: + pos = bisect.bisect_right(self._addrs, addr) - 1 + if pos < 0: + return None + sym_addr, name = self._entries[pos] + return name, addr - sym_addr + + def annotate_frame(self, frame: str) -> dict: + try: + value = int(frame, 0) + except ValueError: + return {"frame": frame, "symbol": None, "offset": None} + hit = self.resolve(value) + if hit is None: + return {"frame": frame, "symbol": None, "offset": None} + name, offset = hit + return {"frame": frame, "symbol": name, "offset": f"{offset:#x}"} + + +def load_symbols(path: str) -> SymbolTable: + """Load the accepted symbol JSON; raise :class:`LogDecodeError` when bad.""" + if not os.path.isfile(path): + raise LogDecodeError(f"symbols file not found: {path}") + try: + with open(path, "r", encoding="utf-8") as fh: + payload = json.load(fh) + except OSError as exc: + raise LogDecodeError(f"cannot read symbols {path}: {exc}") from exc + except json.JSONDecodeError as exc: + raise LogDecodeError(f"symbols JSON {path} is not valid JSON: {exc}") from exc + if isinstance(payload, dict): + entries = payload.get("symbols") + if entries is None: + raise LogDecodeError( + f"symbols JSON {path}: object must have a 'symbols' list") + elif isinstance(payload, list): + entries = payload + else: + raise LogDecodeError( + f"symbols JSON {path}: expected a list or an object with 'symbols'") + if not isinstance(entries, list): + raise LogDecodeError(f"symbols JSON {path}: 'symbols' must be a list") + table: list[tuple[int, str]] = [] + for index, item in enumerate(entries): + where = f"symbols JSON {path}: entry[{index}]" + if not isinstance(item, dict): + raise LogDecodeError(f"{where} is not an object") + if "addr" not in item: + raise LogDecodeError(f"{where} has no 'addr'") + addr = _parse_sym_addr(item["addr"], where) + name = item.get("name") + if name is None: + name = f"sub_{addr:x}" + if not isinstance(name, str): + raise LogDecodeError(f"{where}: 'name' must be a string") + table.append((addr, name)) + return SymbolTable(table) + + +# --------------------------------------------------------------------------- +# Line shapes observed in the real evidence +# --------------------------------------------------------------------------- + +RAW_SECTION_RE = re.compile(r"^(?P<label>[A-Za-z0-9_.+-]+)-raw:\s+\d+\s+bytes\b") +DECODED_SECTION_RE = re.compile(r"^(?P<label>[A-Za-z0-9_.+-]+)-decoded:") +COMMENT_LABEL_RE = re.compile(r"^#\s*(?P<label>[A-Za-z][A-Za-z0-9_.+-]*)\s*$") +HEX_ROW_RE = re.compile( + r"^(?P<offset>[0-9a-fA-F]{6,16}):\s+" + r"(?P<words>[0-9a-fA-F]{1,16}(?:\s+[0-9a-fA-F]{1,16})*)\s*$") +DMSG_PREFIX_RE = re.compile( + r"^(?:\[[^\]]*\]\s*)?(?P<label>[A-Za-z0-9_.+-]+):\s+(?P<body>.*)$") +KV_RE = re.compile(r"(?P<key>[A-Za-z_][A-Za-z0-9_]*)=" + r"(?P<value>\"[^\"]*\"|[^\s,()]+)") +FRAME_LIST_RE = re.compile(r"\[([^\[\]]*)\]") +CONTINUATION_RE = re.compile(r"\brec\s+seq=(?P<seq>\d+)\s+assertCallStack\b") +DECODED_REC_RE = re.compile( + r"^rec\s+(?P<seq>\d+)\s+ts=(?P<ts>0x[0-9a-fA-F]+)\s+" + r"meta=(?P<meta>\S+)\s+args=(?P<args>.*)$") +REC_SEQ_RE = re.compile(r"\brec\s+seq=(?P<seq>\d+)") + + +def _parse_kv(body: str) -> dict[str, str]: + fields: dict[str, str] = {} + for match in KV_RE.finditer(body): + value = match.group("value") + if value.startswith('"') and value.endswith('"') and len(value) >= 2: + value = value[1:-1] + fields[match.group("key")] = value + return fields + + +def _int_or_text(value: str) -> int | str: + try: + return int(value, 0) + except ValueError: + return value + + +def _norm_hex(value: str) -> str: + try: + return f"{int(value, 0):#x}" + except ValueError: + return value + + +def _parse_frame_tokens(text: str) -> list[str]: + text = text.strip() + if text.startswith("[") and text.endswith("]"): + text = text[1:-1] + frames: list[str] = [] + for token in text.replace(",", " ").split(): + try: + frames.append(f"{int(token, 0):#x}") + except ValueError: + continue + return frames + + +def _frames_from_brackets(text: str) -> list[str]: + match = FRAME_LIST_RE.search(text) + if not match: + return [] + return _parse_frame_tokens(match.group(1)) + + +def _looks_like_record(body: str) -> bool: + if DECODED_REC_RE.match(body): + return True + if "recType=" in body or "errorCode=" in body: + return True + if REC_SEQ_RE.search(body): + return True + return False + + +def _record_from_body(body: str, label: str | None, raw: str | None = None) -> dict: + if raw is None: + raw = body + match = DECODED_REC_RE.match(body) + if match: + return { + "kind": "record", + "label": label, + "seq": int(match.group("seq")), + "frames": _parse_frame_tokens(match.group("args")), + "raw": raw, + } + record: dict[str, Any] = {"kind": "record", "label": label, + "frames": [], "raw": raw} + fields = _parse_kv(body) + if "recType" in fields: + record["rec_type"] = _int_or_text(fields["recType"]) + if "errorCode" in fields: + record["error_code"] = _norm_hex(fields["errorCode"]) + if "source" in fields: + record["source_name"] = fields["source"] + if "seq" in fields: + record["seq"] = int(fields["seq"], 0) + return record + + +def _raw_record(label: str | None, words: list[int]) -> dict: + data = b"".join(word.to_bytes(8, "little") for word in words) + return { + "kind": "raw-hex", + "label": label, + "source_name": label, + "frames": [f"{word:#x}" for word in words], + "raw": data.hex(), + } + + +# --------------------------------------------------------------------------- +# Decode +# --------------------------------------------------------------------------- + +def _histogram(records: list[dict]) -> dict[str, dict[str, int]]: + hist: dict[str, dict[str, int]] = {"rec_type": {}, "error_code": {}, + "source": {}} + for record in records: + if "rec_type" in record: + key = str(record["rec_type"]) + hist["rec_type"][key] = hist["rec_type"].get(key, 0) + 1 + if "error_code" in record: + key = str(record["error_code"]) + hist["error_code"][key] = hist["error_code"].get(key, 0) + 1 + if record.get("source_name"): + key = str(record["source_name"]) + hist["source"][key] = hist["source"].get(key, 0) + 1 + return {name: dict(sorted(values.items())) for name, values in hist.items()} + + +def decode_text(text: str, source: str = "<text>", + symbols: SymbolTable | None = None) -> dict: + """Decode one dump/evidence text into a ``greeneye-log-decode/1`` report.""" + records: list[dict] = [] + unparsed = 0 + raw_state: dict[str, Any] | None = None + current_label: str | None = None + + def flush_raw() -> None: + nonlocal raw_state + if raw_state is None: + return + records.append(_raw_record(raw_state["label"], raw_state["words"])) + raw_state = None + + for line in text.splitlines(): + stripped = line.strip() + if not stripped: + continue + if stripped.startswith("#"): + match = COMMENT_LABEL_RE.match(stripped) + if match: + current_label = match.group("label") + continue + match = RAW_SECTION_RE.match(stripped) + if match: + flush_raw() + raw_state = {"label": match.group("label"), "words": []} + current_label = match.group("label") + continue + match = DECODED_SECTION_RE.match(stripped) + if match: + flush_raw() + current_label = match.group("label") + continue + if raw_state is not None: + match = HEX_ROW_RE.match(stripped) + if match: + raw_state["words"].extend( + int(token, 16) for token in match.group("words").split()) + continue + flush_raw() + else: + match = HEX_ROW_RE.match(stripped) + if match: + raw_state = {"label": None, "words": []} + raw_state["words"].extend( + int(token, 16) for token in match.group("words").split()) + continue + + label = current_label + body = stripped + prefix = DMSG_PREFIX_RE.match(stripped) + if prefix: + label = prefix.group("label") + body = prefix.group("body").strip() + + continuation = CONTINUATION_RE.search(body) + if continuation: + seq = int(continuation.group("seq")) + frames = _frames_from_brackets(body) + last = records[-1] if records else None + if (last is not None and last.get("kind") == "record" + and last.get("seq") == seq and not last.get("frames")): + last["frames"] = frames + else: + records.append({"kind": "record", "label": label, "seq": seq, + "frames": frames, "raw": stripped}) + continue + + if _looks_like_record(body): + records.append(_record_from_body(body, label, stripped)) + continue + unparsed += 1 + + flush_raw() + if symbols is not None: + for record in records: + if record.get("frames"): + record["resolved"] = [symbols.annotate_frame(frame) + for frame in record["frames"]] + return { + "schema": SCHEMA, + "source": source, + "count": len(records), + "unparsed": unparsed, + "records": records, + "histogram": _histogram(records), + } + + +def decode_path(path: str, symbols_path: str | None = None) -> dict: + """Decode the file at ``path``; ``source`` in the report is its basename.""" + if not os.path.isfile(path): + raise LogDecodeError(f"input not found: {path}") + symbols = load_symbols(symbols_path) if symbols_path else None + try: + with open(path, "r", encoding="utf-8", errors="replace") as fh: + text = fh.read() + except OSError as exc: + raise LogDecodeError(f"cannot read input {path}: {exc}") from exc + return decode_text(text, source=os.path.basename(path), symbols=symbols) + + +# --------------------------------------------------------------------------- +# Filters +# --------------------------------------------------------------------------- + +def _want_int(value: Any, what: str) -> int: + if isinstance(value, bool): + raise LogDecodeError(f"bad {what} {value!r}") + if isinstance(value, int): + return value + try: + return int(str(value), 0) + except ValueError as exc: + raise LogDecodeError(f"bad {what} {value!r}") from exc + + +def _match_int(record_value: Any, wanted: int) -> bool: + if record_value is None: + return False + if isinstance(record_value, int): + return record_value == wanted + try: + return int(str(record_value), 0) == wanted + except ValueError: + return False + + +def apply_filters(report: dict, *, rec_type: Any = None, source: str | None = None, + error_code: Any = None, kind: str | None = None, + limit: int | None = None) -> dict: + """Return a copy of ``report`` with the requested record filters applied.""" + records = list(report["records"]) + if kind is not None: + if kind not in ("raw-hex", "record"): + raise LogDecodeError(f"bad kind {kind!r} (want raw-hex or record)") + records = [r for r in records if r["kind"] == kind] + if rec_type is not None: + wanted = _want_int(rec_type, "rec-type") + records = [r for r in records if _match_int(r.get("rec_type"), wanted)] + if error_code is not None: + wanted = _want_int(error_code, "error-code") + records = [r for r in records if _match_int(r.get("error_code"), wanted)] + if source is not None: + needle = source.lower() + records = [r for r in records + if needle in str(r.get("source_name") or "").lower()] + if limit is not None: + if limit < 0: + raise LogDecodeError(f"bad limit {limit!r} (must be >= 0)") + records = records[:limit] + filtered = dict(report) + filtered["records"] = records + filtered["count"] = len(records) + filtered["histogram"] = _histogram(records) + return filtered + + +# --------------------------------------------------------------------------- +# Demo: tiny crafted inputs for both real formats (tempdir only) +# --------------------------------------------------------------------------- + +_DEMO_RAW = """\ +LOGINIT-raw: 64 bytes (allocated 64) VA=0xffff0000 dma=0x1000 +00000000: 00000005 00000000 0000000a deadbeef +00000010: 01020304 05060708 090a0b0c 0d0e0f10 +""" + +_DEMO_RECORDS = """\ +[Tue Oct 6 12:40:52 2026] vgsp-nocat: rec seq=1 recType=5 errorCode=0x57f1344 subsystem=0x0 bugcheck=0x0 payload=1212 source="ASSERT" faultingEngine="" +[Tue Oct 6 12:40:52 2026] vgsp-nocat: rec seq=1 assertCallStack count=1 [0x401000,0x402000,0x0] (offsets from load base 0x4000000) +[Tue Oct 6 12:40:53 2026] vgsp-nocat: entry[0]: count=1 recType=5 errorCode=0x57f1344 firstSeq=1 lastSeq=1 source="ASSERT" +an unmatched filler line +rec 7 ts=0x18dbb29118ec6200 meta=0x0000000005bb9c48 args=0x401000,0x402000 +""" + +_DEMO_SYMBOLS = { + "symbols": [ + {"name": "gr_init", "addr": "0x401000"}, + {"name": "gr_done", "addr": int("0x403000", 0)}, + ] +} + + +def run_demo() -> dict: + """Decode crafted inputs for both formats; deterministic, tempdir only.""" + with tempfile.TemporaryDirectory(prefix="log-decode-demo-") as tmp: + raw_path = os.path.join(tmp, "raw.txt") + records_path = os.path.join(tmp, "records.txt") + symbols_path = os.path.join(tmp, "symbols.json") + with open(raw_path, "w", encoding="utf-8") as fh: + fh.write(_DEMO_RAW) + with open(records_path, "w", encoding="utf-8") as fh: + fh.write(_DEMO_RECORDS) + with open(symbols_path, "w", encoding="utf-8") as fh: + json.dump(_DEMO_SYMBOLS, fh, indent=2, sort_keys=True) + fh.write("\n") + raw_report = decode_path(raw_path) + records_report = decode_path(records_path, symbols_path) + + raw_records = raw_report["records"] + record_records = records_report["records"] + first = record_records[0] if record_records else {} + resolved = first.get("resolved") or [] + ok = ( + raw_report["count"] == 1 + and raw_report["unparsed"] == 0 + and len(raw_records) == 1 + and len(raw_records[0]["frames"]) == 8 + and raw_records[0]["frames"][3] == "0xdeadbeef" + and len(raw_records[0]["raw"]) == 2 * 8 * 8 + and records_report["count"] == 3 + and records_report["unparsed"] == 1 + and first.get("rec_type") == 5 + and first.get("error_code") == "0x57f1344" + and first.get("source_name") == "ASSERT" + and first.get("frames") == ["0x401000", "0x402000", "0x0"] + and len(resolved) == 3 + and resolved[0].get("symbol") == "gr_init" + and resolved[1].get("symbol") == "gr_init" + and resolved[1].get("offset") == "0x1000" + and resolved[2].get("symbol") is None + ) + return { + "format": DEMO_FORMAT, + "ok": ok, + "raw": raw_report, + "records": records_report, + } + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + +def _format_record(record: dict) -> list[str]: + bits = [f"{record['kind']:<8}"] + for key in ("label", "seq", "rec_type", "error_code", "source_name"): + if record.get(key) is not None: + bits.append(f"{key}={record[key]}") + bits.append(f"frames={len(record.get('frames') or [])}") + lines = [" ".join(bits)] + resolved = record.get("resolved") or [] + for item in resolved[:RESOLVED_PRINT_LIMIT]: + if item.get("symbol") is None: + lines.append(f" {item['frame']} <unknown>") + else: + lines.append( + f" {item['frame']} {item['symbol']}+{item['offset']}") + if len(resolved) > RESOLVED_PRINT_LIMIT: + lines.append(f" ... {len(resolved) - RESOLVED_PRINT_LIMIT} more") + return lines + + +def _write_json(report: dict, path: str) -> str: + target = os.path.abspath(path) + parent = os.path.dirname(target) + if parent: + os.makedirs(parent, exist_ok=True) + with open(target, "w", encoding="utf-8") as fh: + json.dump(report, fh, indent=2, sort_keys=True) + fh.write("\n") + return target + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="log_decode.py", + description="Decode GSP-RM log-ring and NOCAT record dumps (offline).", + ) + sub = parser.add_subparsers(dest="command") + decode = sub.add_parser( + "decode", help="decode a raw ring dump or dmesg/evidence log") + decode.add_argument("input", help="dump/log text file to decode") + decode.add_argument("--symbols", help="symbol DB JSON (list or {symbols:[..]})") + decode.add_argument("--rec-type", help="keep records with this recType") + decode.add_argument("--source", help="case-insensitive source substring") + decode.add_argument("--error-code", help="keep records with this errorCode") + decode.add_argument("--kind", choices=("raw-hex", "record"), + help="keep only this record kind") + decode.add_argument("--limit", type=int, help="max records to emit") + decode.add_argument("--json", help="write the full report to this file") + sub.add_parser("demo", help="decode crafted samples for both real formats") + return parser + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + try: + if args.command == "decode": + report = decode_path(args.input, args.symbols) + report = apply_filters( + report, + rec_type=args.rec_type, + source=args.source, + error_code=args.error_code, + kind=args.kind, + limit=args.limit, + ) + for record in report["records"]: + for line in _format_record(record): + print(line) + summary = { + "schema": report["schema"], + "source": report["source"], + "count": report["count"], + "unparsed": report["unparsed"], + "histogram": report["histogram"], + } + if args.json: + summary["report_json"] = _write_json(report, args.json) + print(json.dumps(summary, indent=2, sort_keys=True)) + return 0 + if args.command == "demo": + demo = run_demo() + raw = demo["raw"] + records = demo["records"] + print(f"{raw['source']}: records={raw['count']} " + f"unparsed={raw['unparsed']} " + f"frames={len(raw['records'][0]['frames'])} " + f"raw_bytes={len(raw['records'][0]['raw']) // 2}") + print(f"{records['source']}: records={records['count']} " + f"unparsed={records['unparsed']}") + print(f"histogram: {json.dumps(records['histogram'], sort_keys=True)}") + for item in (records["records"][0].get("resolved") or []): + if item.get("symbol") is None: + print(f" {item['frame']} <unknown>") + else: + print(f" {item['frame']} {item['symbol']}+{item['offset']}") + print(json.dumps({"format": demo["format"], "ok": demo["ok"]}, + sort_keys=True)) + return 0 if demo["ok"] else 3 + parser.print_help() + return 2 + except LogDecodeError as exc: + print(f"error: {exc}", file=sys.stderr) + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) |