From 12651dcd93151cd8ac8a5206b19b6c7016407b27 Mon Sep 17 00:00:00 2001 From: Quentin BUISSON-DEBON Date: Sat, 26 Sep 2026 17:04:22 +0200 Subject: [PATCH] CMP 170HX 23-card checker: PCIe cap-mod checks, VBIOS backup, cmpunlocker verify, optional Vulkan VRAM test --- .gitignore | 5 + README.md | 0 check_cards.py | 339 ++++++++++++++++++++++++++++ cxlib.py | 590 +++++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 934 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100755 check_cards.py create mode 100644 cxlib.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..24de2a4 --- /dev/null +++ b/.gitignore @@ -0,0 +1,5 @@ +__pycache__/ +*.pyc +harness/ +cards/ +progress.json diff --git a/README.md b/README.md new file mode 100644 index 0000000..e69de29 diff --git a/check_cards.py b/check_cards.py new file mode 100755 index 0000000..2d2256f --- /dev/null +++ b/check_cards.py @@ -0,0 +1,339 @@ +#!/usr/bin/env python3 +"""CMP 170HX batch card checker - guided one-card-at-a-time workflow (23 cards). + +Per card: detect GPU -> auto serial -> VBIOS backup -> PCIe link/AER/config checks +-> retrain + rescan tests -> dmesg scrape -> cmpunlocker verify.sh (read-only) +-> optional 5-min Vulkan VRAM stress (last step, ask yes/no) -> cards//report.json. + +Scoring (lenient): FAIL only on no-device/link-down; WARN on correctable AER/Xid/ +width< x16/verify issues. +""" +import argparse +import json +import sys +import time +from pathlib import Path + +import cxlib as X +from cxlib import (C, ask, detect_gpus, err, info, now_iso, ok, sanitize, step, warn) + +TOTAL_DEFAULT = X.DEFAULT_TOTAL + + +# ---------------------------------------------------------------- scoring + +def score(rep): + fails, warns = [], [] + if not rep.get("bdf"): + fails.append("no CMP 170HX device detected on the bus") + link = rep.get("link", {}) + if rep.get("bdf"): + if link.get("lspci_rc") not in (0, None): + fails.append("lspci could not read the device") + if link.get("lnksta_width") in ("x0", "x1", None) or not link.get("lnksta_speed"): + fails.append(f"link down or untrained: {link.get('lnksta_raw') or 'no LnkSta'}") + width = link.get("lnksta_width") + if width and width != "x16": + warns.append(f"link width {width} (capacitor mod expects x16 - mod likely " + f"missing/incomplete on this card)") + if link.get("lnksta_gen") == 1: + warns.append("link negotiated Gen1 only (cmpunlocker Gen2 patch/service not active?)") + if rep.get("rescan", {}).get("ok") is False: + fails.append("device lost after bus rescan") + if rep.get("retrain", {}).get("ok") is False: + fails.append("link did not survive retrain") + aer = rep.get("aer_after", {}) + aer_before = rep.get("aer_before", {}) + if (aer.get("ce_active") or aer.get("parent_ce_active") + or (aer.get("aer_correctable_sum", 0) > aer_before.get("aer_correctable_sum", 0))): + warns.append("correctable AER errors present/incremented") + if aer.get("ue_active"): + fails.append(f"uncorrectable AER errors active: {aer['ue_active']}") + dm = rep.get("dmesg", {}) + if dm.get("errors"): + warns.append(f"{len(dm['errors'])} kernel error line(s) related to this card (see dmesg.log)") + vu = rep.get("vulkan_test", {}) + if vu.get("requested") and vu.get("ran"): + if vu.get("tool_errors"): + warns.append("errors reported during Vulkan VRAM stress") + if dm.get("errors") and vu.get("ran"): + warns.append("kernel errors around VRAM stress window") + if not rep.get("bios", {}).get("ok"): + warns.append("VBIOS backup failed/incomplete") + ver = rep.get("cmpunlocker_verify", {}) + if ver.get("ran") and ver.get("status") not in ("unlocked-ok",): + warns.append(f"cmpunlocker verify status: {ver.get('status')}") + + result = "FAIL" if fails else ("WARN" if warns else "PASS") + return result, fails, warns + + +# ---------------------------------------------------------------- one card + +def find_gpu(timeout=45): + """Wait for a CMP 170HX to appear, offering rescan retries.""" + deadline = time.time() + timeout + while time.time() < deadline: + bdfs = detect_gpus() + if len(bdfs) == 1: + return bdfs[0] + if len(bdfs) > 1: + warn(f"multiple CMP 170HX devices present: {', '.join(bdfs)}") + return bdfs[0] + time.sleep(2) + warn("no CMP 170HX detected yet") + if ask("trigger PCI bus rescan and retry? (y/N)", "n").lower().startswith("y"): + if X.rescan_bus(): + ok("bus rescanned") + else: + err("rescan not permitted - reseat the card and press Enter") + input() + return find_gpu(timeout=30) + return None + + +def process_card(index, total, args): + header = f"Card {index}/{total}" + step(header) + bdf = find_gpu() + if not bdf: + err("no device - skipping this card") + return None + devid = "" + for line in X.run(["lspci", "-D", "-s", bdf, "-nn"])["out"].splitlines(): + if "[10de:" in line: + devid = line[line.index("[10de:") + 1:line.index("[10de:") + 10].rstrip("]") + ok(f"device found at {bdf} ({devid})") + + nf = X.nvflash_bin(args) + serial, serial_src = X.get_serial(nf, args) + folder = X.CARDS_DIR / sanitize(serial) + folder.mkdir(parents=True, exist_ok=True) + info(f"folder: {folder}") + + rep = { + "card_index": index, "total": total, "serial": serial, "serial_source": serial_src, + "started_at": now_iso(), "bdf": bdf, "device_id": devid, "host": X.run(["uname", "-n"])["out"].strip(), + "kernel": X.run(["uname", "-r"])["out"].strip(), + } + + step(f"[{index}/{total}] VBIOS backup") + rep["bios"] = X.save_bios(nf, folder / "bios.rom", args) + (ok if rep["bios"]["ok"] else warn)(f"bios: {rep['bios']['method'] or 'FAILED'} " + f"{rep['bios']['size']} bytes - {rep['bios']['note']}") + + step(f"[{index}/{total}] PCIe link + config") + link, lspci_txt = X.check_link(bdf) + rep["link"] = link + (folder / "lspci.txt").write_text(lspci_txt) + (folder / "config_space.hex").write_text(X.dump_config(bdf)) + info(f"LnkSta: {link.get('lnksta_raw') or 'n/a'} (setpci gen={link.get('setpci_current_gen')})") + rep["aer_before"] = X.aer_snapshot(bdf) + + step(f"[{index}/{total}] retrain + rescan tests") + rep["retrain"] = X.retrain_test(bdf) + info(f"retrain: ok={rep['retrain']['ok']} - {rep['retrain']['note']}") + rep["rescan"] = X.rescan_test(bdf) + info(f"rescan: ok={rep['rescan']['ok']} - {rep['rescan']['note']}") + rep["aer_after"] = X.aer_snapshot(bdf) + link2, lspci2 = X.check_link(bdf) + rep["link_after"] = link2 + if lspci2: + (folder / "lspci_after.txt").write_text(lspci2) + + step(f"[{index}/{total}] kernel log scrape") + dm = X.dmesg_capture(bdf) + rep["dmesg"] = dm + (folder / "dmesg.log").write_text("\n".join(dm.get("related", []))) + info(f"{dm.get('note')} (source: {dm.get('source')})") + if dm.get("errors"): + for e in dm["errors"][-5:]: + warn(e.strip()[:160]) + + step(f"[{index}/{total}] nvidia-smi + cmpunlocker verify") + rep["nvidia_smi"] = X.nvidia_smi_info() + rep["cmpunlocker_verify"] = X.run_verify_unlocker() + v = rep["cmpunlocker_verify"] + (folder / "verify_unlocker.log").write_text( + "\n".join(v.get("tail", [])) or (v.get("note") or "not run")) + info(f"verify: {'ran rc=' + str(v.get('rc')) + ' -> ' + str(v.get('status')) if v.get('ran') else 'skipped - ' + str(v.get('note'))}") + + step(f"[{index}/{total}] optional Vulkan VRAM test (last step)") + if args.vulkan == "never": + rep["vulkan_test"] = {"requested": False, "ran": False, "note": "disabled via --vulkan never"} + elif args.vulkan == "always": + rep["vulkan_test"] = X.run_vulkan_test() + else: + if ask("Run 5-min Vulkan VRAM stress test now? (y/N)", "n").lower().startswith("y"): + rep["vulkan_test"] = X.run_vulkan_test() + else: + rep["vulkan_test"] = {"requested": False, "ran": False, "note": "operator declined"} + info("vulkan test skipped") + + result, fails, warns = score(rep) + rep.update(finished_at=now_iso(), result=result, failures=fails, warnings=warns) + (folder / "report.json").write_text(json.dumps(rep, indent=2)) + + print() + if result == "PASS": + ok(f"{serial}: PASS") + elif result == "WARN": + warn(f"{serial}: WARN") + else: + err(f"{serial}: FAIL") + for f in fails: + err(" FAIL: " + f) + for w in warns: + warn(" WARN: " + w) + info(f"report: {folder / 'report.json'}") + return rep + + +# ---------------------------------------------------------------- progress + +def load_progress(): + try: + return json.loads(X.PROGRESS_FILE.read_text()) + except Exception: + return {"done": []} + + +def save_progress(prog): + X.PROGRESS_FILE.write_text(json.dumps(prog, indent=2)) + + +def scoreboard(results): + step("Scoreboard") + counts = {"PASS": 0, "WARN": 0, "FAIL": 0} + for r in results: + if not r: + continue + counts[r["result"]] = counts.get(r["result"], 0) + 1 + color = {"PASS": C.GREEN, "WARN": C.YELLOW, "FAIL": C.RED}[r["result"]] + print(f" {color}{r['result']:<4}{C.RESET} #{r['card_index']:>2} " + f"{r['serial']:<24} {r.get('bdf') or '-':<12} " + f"{r.get('link', {}).get('lnksta_width') or '?':<4} " + f"{r.get('link', {}).get('lnksta_speed') or '?'}") + skipped = [r for r in results if r is None] + print(f"\n PASS={counts.get('PASS', 0)} WARN={counts.get('WARN', 0)} " + f"FAIL={counts.get('FAIL', 0)} skipped={len(skipped)} " + f"total_seen={len(results)}/{TOTAL_DEFAULT}") + + +# ---------------------------------------------------------------- demo + +def demo(args): + """Validate report pipeline without hardware: writes cards/DEMO-*.""" + step("Demo mode (no hardware)") + args.vulkan = "never" + args.serial = "DEMO-SERIAL-001" + args.skip_bios = True + real_find = globals()["find_gpu"] + globals()["find_gpu"] = lambda timeout=0: "0000:0a:00.0" + + class FakeRun: + pass + + orig_detect, orig_check = X.detect_gpus, X.check_link + orig_aer, orig_dm, orig_cfg = X.aer_snapshot, X.dmesg_capture, X.dump_config + orig_retrain, orig_rescan = X.retrain_test, X.rescan_test + orig_smi, orig_verify = X.nvidia_smi_info, X.run_verify_unlocker + X.detect_gpus = lambda: ["0000:0a:00.0"] + X.check_link = lambda b: ({"lnksta_speed": "5GT/s", "lnksta_width": "x4", "lnksta_gen": 2, + "lnkcap_speed": "8GT/s", "lnkcap_width": "x16", + "lnksta_raw": "LnkSta: Speed 5GT/s, Width x4, TrErr-", + "setpci_current_gen": 2, "lspci_rc": 0}, "demo lspci output\n") + X.aer_snapshot = lambda b: {"aer_sysfs_correctable": None, "aer_sysfs_fatal": None, + "aer_correctable_sum": 0, "ce_active": [], "ue_active": [], + "parent_bdf": None, "parent_ce_active": []} + X.dmesg_capture = lambda b: {"source": "demo", "related": ["demo line"], "errors": [], + "note": "demo"} + X.dump_config = lambda b: "demo config space dump\n" + X.retrain_test = lambda b: {"attempted": True, "ok": True, "note": "x4 -> x4", + "width_before": "x4", "width_after": "x4"} + X.rescan_test = lambda b: {"attempted": True, "ok": True, "note": "re-enumerated"} + X.nvidia_smi_info = lambda: {"available": False} + X.run_verify_unlocker = lambda: {"ran": False, "note": "demo"} + try: + rep = process_card(1, TOTAL_DEFAULT, args) + finally: + globals()["find_gpu"] = real_find + X.detect_gpus, X.check_link = orig_detect, orig_check + X.aer_snapshot, X.dmesg_capture, X.dump_config = orig_aer, orig_dm, orig_cfg + X.retrain_test, X.rescan_test = orig_retrain, orig_rescan + X.nvidia_smi_info, X.run_verify_unlocker = orig_smi, orig_verify + return rep + + +# ---------------------------------------------------------------- main + +def main(): + global TOTAL_DEFAULT + ap = argparse.ArgumentParser(description="CMP 170HX guided 23-card checker") + ap.add_argument("--preflight", action="store_true", help="only run tool check/install/clone") + ap.add_argument("--demo", action="store_true", help="run pipeline with fake hardware (self-test)") + ap.add_argument("--cards", type=int, default=TOTAL_DEFAULT, help="total cards (default 23)") + ap.add_argument("--start", type=int, default=1, help="first card index") + ap.add_argument("--nvflash", help="path to nvflash binary") + ap.add_argument("--serial", help="override serial for the next card (single-card run)") + ap.add_argument("--skip-bios", action="store_true", help="skip VBIOS backup") + ap.add_argument("--vulkan", choices=("ask", "always", "never"), default="ask", + help="5-min Vulkan VRAM test policy (default ask, runs as last step)") + ap.add_argument("--no-install", action="store_true", help="do not auto-install missing packages") + ap.add_argument("--no-clone", action="store_true", help="do not clone harness repos") + ap.add_argument("--restart", action="store_true", help="ignore previous progress.json") + ap.add_argument("--no-pause", action="store_true", help="do not wait for card insertion") + args = ap.parse_args() + + TOTAL_DEFAULT = args.cards + + pf = X.preflight(auto_install=not args.no_install, do_clone=not args.no_clone) + if args.preflight: + return 0 + if args.demo: + demo(args) + return 0 + + X.CARDS_DIR.mkdir(exist_ok=True) + prog = {"done": []} if args.restart else load_progress() + + results = [] + for i in range(args.start, args.cards + 1): + if str(i) in prog.get("done", []): + info(f"card {i} already done (progress.json) - skipping") + results.append(None) + continue + if not args.no_pause: + ans = ask(f"Insert card {i}/{args.cards}, then press Enter " + f"(s=skip this card, q=quit)", "") + if ans.lower() == "q": + break + if ans.lower() == "s": + results.append(None) + prog.setdefault("done", []).append(str(i)) + prog.setdefault("skipped", []).append(str(i)) + save_progress(prog) + continue + rep = process_card(i, args.cards, args) + results.append(rep) + if rep: + prog.setdefault("done", []).append(str(i)) + save_progress(prog) + else: + if not ask("retry this card? (y/N)", "n").lower().startswith("y"): + break + + if results: + scoreboard(results) + done = len([r for r in results if r]) + info(f"{done} card(s) processed. Output in {X.CARDS_DIR}") + return 0 + + +if __name__ == "__main__": + try: + sys.exit(main()) + except KeyboardInterrupt: + print() + warn("interrupted - progress saved in progress.json") + sys.exit(130) diff --git a/cxlib.py b/cxlib.py new file mode 100644 index 0000000..82e04ef --- /dev/null +++ b/cxlib.py @@ -0,0 +1,590 @@ +#!/usr/bin/env python3 +"""CMP 170HX checker - shared library (preflight, detection, PCIe checks, BIOS, verify, Vulkan).""" +import datetime +import hashlib +import os +import re +import shutil +import subprocess +import sys +import time +from pathlib import Path + +SCRIPT_DIR = Path(__file__).resolve().parent +HARNESS_DIR = SCRIPT_DIR / "harness" +CARDS_DIR = SCRIPT_DIR / "cards" +PROGRESS_FILE = SCRIPT_DIR / "progress.json" + +REPOS = { + "170th-Street": "https://github.com/amoghmunikote/170th-Street.git", + "cmpunlocker": "https://github.com/amoghmunikote/cmpunlocker.git", +} +VERIFY_SH = HARNESS_DIR / "cmpunlocker" / "verify.sh" +GPU_IDS = ("20c2", "2082") +DEFAULT_TOTAL = 23 +VULKAN_TEST_SECONDS = 300 + +SPEED_TO_GEN = {"2.5GT/s": 1, "5GT/s": 2, "8GT/s": 3, "16GT/s": 4, "32GT/s": 5, "64GT/s": 6} + +USE_COLOR = sys.stdout.isatty() + + +class C: + RESET = "\033[0m" if USE_COLOR else "" + BOLD = "\033[1m" if USE_COLOR else "" + DIM = "\033[2m" if USE_COLOR else "" + GREEN = "\033[32m" if USE_COLOR else "" + YELLOW = "\033[33m" if USE_COLOR else "" + RED = "\033[31m" if USE_COLOR else "" + CYAN = "\033[36m" if USE_COLOR else "" + + +def info(msg): + print(f"{C.CYAN}[*]{C.RESET} {msg}", flush=True) + + +def ok(msg): + print(f"{C.GREEN}[+]{C.RESET} {msg}", flush=True) + + +def warn(msg): + print(f"{C.YELLOW}[!]{C.RESET} {msg}", flush=True) + + +def err(msg): + print(f"{C.RED}[-]{C.RESET} {msg}", flush=True) + + +def step(msg): + print(f"\n{C.BOLD}== {msg} =={C.RESET}", flush=True) + + +def ask(prompt, default=""): + suffix = f" [{default}]" if default else "" + try: + val = input(f"{C.BOLD}?{C.RESET} {prompt}{suffix}: ").strip() + except EOFError: + return default + return val if val else default + + +def now_iso(): + return datetime.datetime.now().isoformat(timespec="seconds") + + +def sanitize(name): + return re.sub(r"[^A-Za-z0-9._-]+", "_", name).strip("_") or "UNKNOWN" + + +# ---------------------------------------------------------------- subprocess + +def run(cmd, timeout=60, env=None): + try: + p = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout, env=env) + return {"rc": p.returncode, "out": p.stdout, "err": p.stderr, "cmd": cmd} + except FileNotFoundError: + return {"rc": 127, "out": "", "err": f"not found: {cmd[0]}", "cmd": cmd} + except subprocess.TimeoutExpired as e: + out = e.stdout.decode(errors="replace") if isinstance(e.stdout, bytes) else (e.stdout or "") + return {"rc": 124, "out": out, "err": "timeout", "cmd": cmd} + + +def as_root(cmd, timeout=60): + if os.geteuid() == 0: + return run(cmd, timeout=timeout) + if shutil.which("sudo"): + r = run(["sudo", "-n"] + cmd, timeout=timeout) + if r["rc"] != 0 and "password is required" in (r["err"] + r["out"]).lower(): + r["err"] = "sudo requires a password (sudo -n failed)" + return r + return {"rc": 1, "out": "", "err": "no root: neither uid 0 nor sudo available", "cmd": cmd} + + +def which(name): + return shutil.which(name) + + +def have_root(): + return os.geteuid() == 0 or bool(shutil.which("sudo")) + + +# ---------------------------------------------------------------- preflight + +PM_CANDIDATES = [ + ("apt-get", ["apt-get", "install", "-y"]), + ("dnf", ["dnf", "install", "-y"]), + ("yum", ["yum", "install", "-y"]), + ("pacman", ["pacman", "-S", "--noconfirm"]), + ("zypper", ["zypper", "--non-interactive", "install"]), + ("xbps-install", ["xbps-install", "-y"]), +] + + +def detect_pm(): + for name, cmd in PM_CANDIDATES: + if which(name): + return cmd + return None + + +def install_packages(packages): + pm = detect_pm() + if not pm: + return False, "no supported package manager found (apt/dnf/pacman/zypper/xbps)" + installed, log = [], [] + for pkg in packages: + r = as_root(pm + [pkg], timeout=300) + log.append(f"$ {' '.join(pm + [pkg])}\nrc={r['rc']}\n{r['out'][-1500:]}{r['err'][-800:]}") + if r["rc"] == 0: + installed.append(pkg) + else: + warn(f"could not install '{pkg}' via {pm[0]} (rc={r['rc']})") + return bool(installed), "\n".join(log) + + +def ensure_repos(): + for name, url in REPOS.items(): + dest = HARNESS_DIR / name + if (dest / ".git").exists(): + continue + info(f"cloning {name} ...") + r = run(["git", "clone", "--depth", "1", url, str(dest)], timeout=300) + if r["rc"] != 0: + err(f"git clone failed for {name}: {r['err'].strip()}") + return False + ok(f"cloned {name}") + return True + + +TOOL_CHECKS = { + "git": ("critical", "clone harness repos"), + "lspci": ("critical", "PCIe checks (package: pciutils)"), + "setpci": ("critical", "link generation probe (package: pciutils)"), + "nvidia-smi": ("critical-if-card-present", "GPU state / cmpunlocker verify"), + "nvflash": ("critical", "VBIOS backup + serial (not packaged: manual install)"), + "dmesg": ("optional", "kernel log scrape"), + "journalctl": ("optional", "kernel log scrape fallback"), + "vulkaninfo": ("optional", "vulkan presence (package: vulkan-tools)"), + "vkmark": ("optional", "5-min VRAM stress (package: vkmark)"), + "vkcube": ("optional", "VRAM stress fallback (package: vulkan-tools)"), +} +PKG_FOR = {"lspci": "pciutils", "setpci": "pciutils", "vulkaninfo": "vulkan-tools", + "vkcube": "vulkan-tools", "vkmark": "vkmark", "git": "git"} + + +def preflight(auto_install=True, do_clone=True): + step("Preflight") + report = {"tools": {}, "root": have_root()} + critical_missing, optional_missing = [], [] + + for tool, (level, why) in TOOL_CHECKS.items(): + path = which(tool) + report["tools"][tool] = {"found": bool(path), "path": path, "level": level, "why": why} + if not path: + (critical_missing if level.startswith("critical") else optional_missing).append(tool) + + if auto_install: + pkgs = sorted({PKG_FOR[t] for t in critical_missing + optional_missing if t in PKG_FOR}) + if pkgs: + if detect_pm(): + info(f"attempting to install: {' '.join(pkgs)}") + _, log = install_packages(pkgs) + report["install_log"] = log + else: + report["install_log"] = "no package manager - manual install required" + warn(f"no package manager found - install manually: {' '.join(pkgs)}") + for tool in list(critical_missing) + list(optional_missing): + if which(tool): + if tool in critical_missing: + critical_missing.remove(tool) + if tool in optional_missing: + optional_missing.remove(tool) + report["tools"][tool].update(found=True, path=which(tool)) + + if do_clone: + ensure_repos() + report["repos_ok"] = all((HARNESS_DIR / n / ".git").exists() for n in REPOS) + + report["verify_sh_exists"] = VERIFY_SH.exists() + + for t, st in report["tools"].items(): + if st["found"]: + mark = f"{C.GREEN}OK{C.RESET}" + elif st["level"].startswith("critical"): + mark = f"{C.RED}MISSING{C.RESET}" + else: + mark = f"{C.YELLOW}missing (optional){C.RESET}" + print(f" {t:<14} {mark} {C.DIM}{st['why']}{C.RESET}") + + if "nvflash" in critical_missing: + warn("nvflash missing - VBIOS backup will fall back to sysfs ROM, serial to manual entry") + print(" Download Linux x64 nvflash from https://www.nvidia.com/en-us/drivers/nvflash/") + print(" and pass --nvflash /path/to/nvflash, or place it on PATH.") + if "nvidia-smi" in critical_missing: + warn("nvidia-smi missing - cmpunlocker verify will be skipped (install nvidia-open driver)") + if not report["root"]: + warn("no sudo and not root: privileged steps (rescan, retrain, rom, dmesg) will be skipped") + + ok("preflight complete") + return report + + +# ---------------------------------------------------------------- GPU detect + +def detect_gpus(): + bdfs = [] + for dev_id in GPU_IDS: + r = run(["lspci", "-D", "-d", f"10de:{dev_id}"]) + for line in r["out"].splitlines(): + parts = line.split() + if not parts: + continue + bdf = parts[0] + if re.match(r"^[0-9a-fA-F]{4}:[0-9a-fA-F]{2}:[0-9a-fA-F]{2}\.[0-9a-f]$", bdf): + bdfs.append(bdf.lower()) + return sorted(set(bdfs)) + + +def rescan_bus(): + r = as_root(["sh", "-c", "echo 1 > /sys/bus/pci/rescan"], timeout=15) + if r["rc"] == 0: + time.sleep(2) + return r["rc"] == 0 + + +def sysfs(bdf): + return Path("/sys/bus/pci/devices") / bdf + + +def read_file(path, binary=False): + try: + with open(path, "rb" if binary else "r", errors="replace") as f: + return f.read() + except (PermissionError, FileNotFoundError, OSError): + return None + + +# ---------------------------------------------------------------- PCIe checks + +def parse_link(text): + out = {"lnksta_speed": None, "lnksta_width": None, "lnksta_gen": None, + "lnkcap_speed": None, "lnkcap_width": None, "lnksta_raw": None} + for line in text.splitlines(): + s = line.strip() + if s.startswith("LnkSta:") and not s.startswith("LnkSta2"): + out["lnksta_raw"] = s + m = re.search(r"Speed\s+([^,\s]+)", s) + if m: + out["lnksta_speed"] = m.group(1) + out["lnksta_gen"] = SPEED_TO_GEN.get(m.group(1)) + m = re.search(r"Width\s+(x\d+)", s) + if m: + out["lnksta_width"] = m.group(1) + elif s.startswith("LnkCap:"): + m = re.search(r"Speed\s+([^,\s]+)", s) + if m: + out["lnkcap_speed"] = m.group(1) + m = re.search(r"Width\s+(x\d+)", s) + if m: + out["lnkcap_width"] = m.group(1) + return out + + +def parse_aer_flags(text): + res = {"CESta": [], "UESta": []} + for line in text.splitlines(): + s = line.strip() + for key in ("CESta", "UESta"): + if s.startswith(key + ":"): + body = s.split(":", 1)[1] + res[key].extend(re.findall(r"([A-Za-z0-9_]+)\+", body)) + return res + + +def parent_bdf(bdf): + path = Path(os.path.realpath(sysfs(bdf))) + while True: + path = path.parent + name = path.name + if re.match(r"^[0-9a-fA-F]{4}:[0-9a-fA-F]{2}:[0-9a-fA-F]{2}\.[0-9a-f]$", name): + return name.lower() + if name.startswith("pci") or path.parent == path: + return None + + +def aer_snapshot(bdf): + snap = {"aer_sysfs_correctable": None, "aer_sysfs_fatal": None, + "aer_correctable_sum": 0, "ce_active": [], "ue_active": [], + "parent_bdf": None, "parent_ce_active": []} + base = sysfs(bdf) + data = read_file(base / "aer_dev_correctable") + if data is not None: + snap["aer_sysfs_correctable"] = data.strip() + snap["aer_correctable_sum"] = sum(int(x) for x in re.findall(r"\d+", data)) + fatal = read_file(base / "aer_dev_fatal") + if fatal is not None: + snap["aer_sysfs_fatal"] = fatal.strip() + r = run(["lspci", "-vvv", "-D", "-s", bdf], timeout=30) + if r["rc"] == 0: + aer = parse_aer_flags(r["out"]) + snap["ce_active"] = aer["CESta"] + snap["ue_active"] = aer["UESta"] + parent = parent_bdf(bdf) + if parent: + snap["parent_bdf"] = parent + rp = run(["lspci", "-vvv", "-D", "-s", parent], timeout=30) + if rp["rc"] == 0: + snap["parent_ce_active"] = parse_aer_flags(rp["out"])["CESta"] + return snap + + +def link_gen_setpci(bdf): + r = as_root(["setpci", "-s", bdf, "CAP_EXP+12.w"], timeout=15) + val = r["out"].strip() + if re.match(r"^[0-9a-fA-F]{4}$", val): + return int(val, 16) & 0x0F + return None + + +def check_link(bdf): + r = run(["lspci", "-vvv", "-D", "-s", bdf], timeout=30) + link = parse_link(r["out"]) if r["rc"] == 0 else {} + link["setpci_current_gen"] = link_gen_setpci(bdf) + link["lspci_rc"] = r["rc"] + return link, r["out"] + + +def dump_config(bdf): + r = run(["lspci", "-nnvvxxxx", "-D", "-s", bdf], timeout=30) + return r["out"] if r["rc"] == 0 else f"lspci failed rc={r['rc']}: {r['err']}" + + +def retrain_test(bdf): + result = {"attempted": False, "ok": None, "note": "", "width_before": None, "width_after": None} + link_before, _ = check_link(bdf) + result["width_before"] = link_before.get("lnksta_width") + node = sysfs(bdf) / "link" / "retrain" + if not node.exists(): + result["note"] = "sysfs link/retrain not available on this kernel" + return result + result["attempted"] = True + r = as_root(["sh", "-c", f"echo 1 > {node}"], timeout=15) + if r["rc"] != 0: + result["note"] = f"retrain write failed: {r['err'].strip()}" + return result + time.sleep(3) + if bdf not in detect_gpus(): + result.update(ok=False, note="device vanished after retrain") + return result + link_after, _ = check_link(bdf) + result["width_after"] = link_after.get("lnksta_width") + result["ok"] = result["width_after"] == result["width_before"] and result["width_after"] is not None + result["note"] = f"{result['width_before']} -> {result['width_after']}" + return result + + +def rescan_test(bdf): + result = {"attempted": True, "ok": None, "note": ""} + if not rescan_bus(): + result["note"] = "bus rescan not permitted (no root)" + return result + time.sleep(1) + present = bdf in detect_gpus() + result["ok"] = present + result["note"] = "re-enumerated at same BDF" if present else "device missing after rescan" + return result + + +def dmesg_capture(bdf): + text, source = "", None + r = as_root(["dmesg", "--ctime"], timeout=20) + if r["rc"] == 0 and r["out"]: + text, source = r["out"], "dmesg" + else: + r2 = run(["journalctl", "-k", "-b", "--no-pager", "-o", "short-precise"], timeout=25) + if r2["rc"] == 0 and r2["out"]: + text, source = r2["out"], "journalctl" + if not text: + return {"source": None, "related": [], "errors": [], "note": "kernel log unavailable (no root?)"} + short = bdf[5:] + related, errors = [], [] + for line in text.splitlines(): + low = line.lower() + if (bdf in line or short in line or "10de" in low or "nvrm" in low + or "xid" in low or "aer:" in low or "pcieport" in low): + related.append(line) + if any(k in low for k in ("xid", "error", "aer:", "fail", "corrupted", "fallen off")): + errors.append(line) + return {"source": source, "related": related[-200:], "errors": errors[-50:], + "note": f"{len(related)} related lines"} + + +# ---------------------------------------------------------------- BIOS / serial + +def nvflash_bin(args): + return args.nvflash if getattr(args, "nvflash", None) else which("nvflash") + + +def nvflash_list(nf): + for flags in (["--list"], ["-l"]): + r = as_root([nf] + flags, timeout=40) + if r["rc"] == 0 and r["out"].strip(): + return r["out"] + return None + + +def get_serial(nf, args): + if getattr(args, "serial", None): + return args.serial, "cli-override" + if nf: + listing = nvflash_list(nf) + if listing: + m = re.search(r"UUID:\s*(GPU-[0-9a-fA-F\-]+)", listing, re.I) + if m: + info(f"serial auto-read from nvflash UUID: {m.group(1)}") + return m.group(1), "nvflash-uuid" + m = re.search(r"Serial(?:Number)?\s*[:=]\s*(\S+)", listing, re.I) + if m: + info(f"serial auto-read from nvflash: {m.group(1)}") + return m.group(1), "nvflash-list" + q = as_root([nf, "--query"], timeout=40) + m = re.search(r"Serial(?:Number)?\s*[:=]\s*(\S+)", q["out"], re.I) + if m: + info(f"serial auto-read from nvflash --query: {m.group(1)}") + return m.group(1), "nvflash-query" + warn("nvflash present but exposes no serial/UUID for this card") + else: + warn("nvflash unavailable - cannot auto-read serial") + serial = "" + while not serial: + serial = ask("Enter serial number from the card label (folder name)") + if not serial: + warn("serial is required for the folder name") + return serial, "manual-label" + + +def save_bios(nf, out_path, args): + result = {"method": None, "path": None, "sha256": None, "size": 0, "ok": False, "note": ""} + if getattr(args, "skip_bios", False): + result["note"] = "bios backup skipped (--skip-bios)" + return result + if nf: + r = as_root([nf, "--save", str(out_path)], timeout=90) + if r["rc"] == 0 and out_path.exists() and out_path.stat().st_size > 0: + data = out_path.read_bytes() + if data.strip(b"\x00") == b"": + result["note"] = "nvflash wrote an all-zero image" + else: + result.update(method="nvflash", path=str(out_path), size=len(data), + sha256=hashlib.sha256(data).hexdigest(), ok=True) + return result + else: + result["note"] = f"nvflash --save failed rc={r['rc']}: {(r['err'] or r['out'])[-200:]}" + else: + result["note"] = "nvflash unavailable" + + for bdf in detect_gpus(): + rom = sysfs(bdf) / "rom" + if as_root(["sh", "-c", f"echo 1 > {rom}"], timeout=15)["rc"] != 0: + result["note"] += " | sysfs rom enable failed" + continue + data = read_file(rom, binary=True) + as_root(["sh", "-c", f"echo 0 > {rom}"], timeout=15) + if data and len(data) > 512 and data.strip(b"\x00") != b"": + out_path.write_bytes(data) + result.update(method="sysfs-rom", path=str(out_path), size=len(data), + sha256=hashlib.sha256(data).hexdigest(), ok=True) + result["note"] = (result["note"] + " | saved via sysfs rom").strip(" |") + return result + result["note"] += " | sysfs rom empty/all-zero" + return result + + +# ---------------------------------------------------------------- nvidia / verify + +def nvidia_smi_info(): + if not which("nvidia-smi"): + return {"available": False} + q = run(["nvidia-smi", "--query-gpu=name,uuid,memory.total,pcie.link.gen.current," + "pcie.link.gen.max,pcie.link.width.current,temperature.gpu,driver_version", + "--format=csv,noheader"], timeout=30) + if q["rc"] != 0: + return {"available": False, "error": (q["err"] or q["out"])[-300:]} + rows = [[f.strip() for f in line.split(",")] for line in q["out"].strip().splitlines()] + return {"available": True, "gpus": rows} + + +def run_verify_unlocker(): + if not VERIFY_SH.exists(): + return {"ran": False, "note": "harness/cmpunlocker not cloned"} + if not which("nvidia-smi"): + return {"ran": False, "note": "nvidia-smi missing - unlock verify skipped"} + r = as_root(["bash", str(VERIFY_SH)], timeout=180) + lines = [l for l in (r["out"] + r["err"]).splitlines() if l.strip()] + status = "unknown" + if r["rc"] == 0 and any("unlocked memory" in l.lower() for l in lines): + status = "unlocked-ok" + elif "STOCK" in r["out"]: + status = "stock-memory" + elif "MISSING" in r["out"]: + status = "gpu-missing-in-smi" + elif r["rc"] != 0: + status = f"verify-failed-rc{r['rc']}" + return {"ran": True, "rc": r["rc"], "status": status, "tail": lines[-30:]} + + +# ---------------------------------------------------------------- Vulkan test + +def vulkan_tool(): + for t in ("vkmark", "vkcube", "vulkaninfo"): + if which(t): + return t + return None + + +def run_vulkan_test(seconds=VULKAN_TEST_SECONDS): + result = {"tool": None, "ran": False, "requested": True, "duration_s": seconds, + "rc": None, "note": "", "log_tail": []} + tool = vulkan_tool() + if not tool: + result["note"] = "no vulkan tool available (install vkmark or vulkan-tools)" + warn(result["note"]) + return result + result["tool"] = tool + info(f"running {tool} for {seconds}s VRAM stress - press Ctrl+C to abort early") + output = [] + deadline = time.time() + seconds + try: + if tool == "vkmark": + while time.time() < deadline: + r = run(["vkmark", "--benchmark"], timeout=60) + output.append(r["out"]) + if r["rc"] not in (0, 124): + break + elif tool == "vkcube": + p = subprocess.Popen(["vkcube"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True) + try: + p.wait(timeout=seconds) + except subprocess.TimeoutExpired: + p.terminate() + try: + p.wait(timeout=10) + except subprocess.TimeoutExpired: + p.kill() + output.append(p.communicate()[0] or "") + result["rc"] = p.returncode + else: + r = run(["vulkaninfo"], timeout=seconds) + output.append(r["out"] + r["err"]) + result["rc"] = r["rc"] + except KeyboardInterrupt: + result["note"] = "aborted by user" + warn("vulkan test aborted") + text = "\n".join(output) + result["ran"] = True + result["log_tail"] = [l for l in text.splitlines() if l.strip()][-40:] + bad = [l for l in text.splitlines() + if re.search(r"(error|fail|out of memory|oom|corrupt)", l, re.I)] + result["tool_errors"] = bad[-20:] + ok(f"{tool} stress finished ({seconds}s)") + return result