diff --git a/NOTICE b/NOTICE new file mode 100644 index 000000000..393522537 --- /dev/null +++ b/NOTICE @@ -0,0 +1,8 @@ +solstone nvattest runtime notice + +The NVIDIA nvattest runtime is provided by NVIDIA/attestation-sdk under the +Apache License 2.0. + +solstone fetches nvattest setup artifacts into a local runtime cache when the +GPU attestation provider is installed. nvattest is not bundled in the solstone +wheel or source distribution. diff --git a/scripts/check_journal_io_mechanic.py b/scripts/check_journal_io_mechanic.py index a0e2d7019..a517c4f3a 100644 --- a/scripts/check_journal_io_mechanic.py +++ b/scripts/check_journal_io_mechanic.py @@ -99,6 +99,7 @@ EXCLUDED_FILES: frozenset[str] = frozenset( "solstone/think/providers/ced_install.py", "solstone/think/providers/rerank_install.py", "solstone/think/providers/rfdetr_install.py", + "solstone/think/providers/nvattest_install.py", "solstone/think/services/scout.py", "solstone/think/services/spl.py", "solstone/think/steward.py", diff --git a/solstone/think/providers/nvattest_install.py b/solstone/think/providers/nvattest_install.py new file mode 100644 index 000000000..1701a62d4 --- /dev/null +++ b/solstone/think/providers/nvattest_install.py @@ -0,0 +1,269 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""Install and locate NVIDIA nvattest runtime artifacts. + +This module performs no network access at import time. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import shutil +import stat +import tempfile +from dataclasses import dataclass +from pathlib import Path + +from solstone.think.providers.rfdetr_install import ( + RfdetrInstallError, +) +from solstone.think.providers.rfdetr_install import ( + _safe_extract_tarball as _rfdetr_safe_extract_tarball, +) +from solstone.think.utils import get_journal + +SPP_NVATTEST_DIR_ENV = "SPP_NVATTEST_DIR" +NVATTEST_VERSION = "1.2.2" +NVATTEST_ARCHIVE_NAME = "libnvat-linux-x86_64-1.2.2.1780962352-archive.tar.xz" +NVATTEST_ARCHIVE_URL = ( + "https://developer.download.nvidia.com/compute/nvat/redist/libnvat/" + f"linux-x86_64/{NVATTEST_ARCHIVE_NAME}" +) +NVATTEST_ARCHIVE_SHA256 = ( + "3f10da6fca794b7e3025c6645447947ec8bc45bcfde5b5b1d23241c7115630db" +) +SIDECAR_NAME = ".nvattest-install.json" + + +class NvattestInstallError(RuntimeError): + """nvattest artifact acquisition failure with a recovery reason code.""" + + def __init__(self, reason_code: str, message: str) -> None: + super().__init__(message) + self.reason_code = reason_code + + +@dataclass(frozen=True, slots=True) +class NvattestArchiveSpec: + version: str + url: str + archive_name: str + sha256: str + + +@dataclass(frozen=True, slots=True) +class NvattestInstallRecord: + version: str + archive_name: str + archive_sha256: str + + def to_json(self) -> str: + return ( + json.dumps( + { + "archive_name": self.archive_name, + "archive_sha256": self.archive_sha256, + "version": self.version, + }, + indent=2, + sort_keys=True, + ) + + "\n" + ) + + +NVATTEST_ARCHIVE_SPEC = NvattestArchiveSpec( + version=NVATTEST_VERSION, + url=NVATTEST_ARCHIVE_URL, + archive_name=NVATTEST_ARCHIVE_NAME, + sha256=NVATTEST_ARCHIVE_SHA256, +) + + +def cache_root(journal_path: str | Path | None = None) -> Path: + root = Path(journal_path) if journal_path is not None else Path(get_journal()) + return root / "cache" / "providers" / "nvattest" + + +def resolve_nvattest_dir( + explicit_override: str | Path | None = None, + *, + journal_path: str | Path | None = None, +) -> Path: + """Resolve the nvattest directory from override, env, then journal cache.""" + + if explicit_override is not None: + return Path(explicit_override).expanduser() + env_path = os.environ.get(SPP_NVATTEST_DIR_ENV) + if env_path: + return Path(env_path).expanduser() + return cache_root(journal_path) + + +def install_nvattest( + *, + force: bool = False, + spec: NvattestArchiveSpec = NVATTEST_ARCHIVE_SPEC, + journal_path: str | Path | None = None, +) -> Path: + """Download, verify, and install nvattest into the journal provider cache.""" + + root = cache_root(journal_path) + if not force and _installed(root): + return root + + archive = _archive_path(spec, journal_path) + extract_dir = root / ".extract" + _download_file(spec.url, archive, spec.sha256) + shutil.rmtree(extract_dir, ignore_errors=True) + try: + _safe_extract_nvattest_tarball(archive, extract_dir) + source = _find_extracted_root(extract_dir) + _install_extracted_tree(source, root) + _write_sidecar( + root / SIDECAR_NAME, + NvattestInstallRecord( + version=spec.version, + archive_name=spec.archive_name, + archive_sha256=spec.sha256, + ), + ) + return root + finally: + shutil.rmtree(extract_dir, ignore_errors=True) + archive.unlink(missing_ok=True) + + +def _installed(root: Path) -> bool: + return (root / "bin" / "nvattest").is_file() and (root / "lib").is_dir() + + +def _archive_path( + spec: NvattestArchiveSpec, + journal_path: str | Path | None = None, +) -> Path: + return cache_root(journal_path) / ".downloads" / spec.archive_name + + +def _sha256_file(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _verify_file(path: Path, expected_sha256: str) -> None: + if not path.is_file(): + raise NvattestInstallError("file_missing", f"nvattest asset missing: {path}") + actual_sha256 = _sha256_file(path) + if actual_sha256 != expected_sha256: + raise NvattestInstallError( + "sha256_mismatch", + ( + f"sha256 mismatch for {path.name}: " + f"expected {expected_sha256}, got {actual_sha256}" + ), + ) + + +def _tmp_path(dest: Path) -> Path: + return dest.with_name(f"{dest.name}.tmp") + + +def _download_file(url: str, dest: Path, expected_sha256: str) -> None: + import httpx + + dest.parent.mkdir(parents=True, exist_ok=True) + tmp = _tmp_path(dest) + dest.unlink(missing_ok=True) + tmp.unlink(missing_ok=True) + try: + with httpx.stream("GET", url, timeout=600.0, follow_redirects=True) as response: + response.raise_for_status() + with tmp.open("wb") as handle: + for chunk in response.iter_bytes(): + if chunk: + handle.write(chunk) + _verify_file(tmp, expected_sha256) + tmp.replace(dest) + except NvattestInstallError: + tmp.unlink(missing_ok=True) + raise + except Exception as exc: + tmp.unlink(missing_ok=True) + raise NvattestInstallError( + "download_failed", + f"failed to download nvattest archive: {exc}", + ) from exc + + +def _safe_extract_nvattest_tarball(tarball: Path, dest: Path) -> None: + try: + _rfdetr_safe_extract_tarball(tarball, dest) + except RfdetrInstallError as exc: + reason_code = getattr(exc, "reason_code", "archive_extract_failed") + if reason_code == "archive_path_traversal": + raise NvattestInstallError(reason_code, str(exc)) from exc + raise NvattestInstallError("archive_extract_failed", str(exc)) from exc + + +def _find_extracted_root(extract_dir: Path) -> Path: + if _installed(extract_dir): + return extract_dir + matches = [ + path + for path in extract_dir.rglob("nvattest") + if path.is_file() + and path.parent.name == "bin" + and (path.parent.parent / "lib").is_dir() + ] + if len(matches) != 1: + raise NvattestInstallError( + "archive_layout_invalid", + f"expected exactly one extracted nvattest binary, found {len(matches)}", + ) + return matches[0].parent.parent + + +def _install_extracted_tree(source: Path, root: Path) -> None: + binary = source / "bin" / "nvattest" + lib_dir = source / "lib" + if not binary.is_file() or not lib_dir.is_dir(): + raise NvattestInstallError( + "archive_layout_invalid", + "extracted archive must contain bin/nvattest and lib/", + ) + + root.mkdir(parents=True, exist_ok=True) + for name in ("bin", "lib", "include", "share"): + shutil.rmtree(root / name, ignore_errors=True) + for name in ("LICENSE",): + (root / name).unlink(missing_ok=True) + + for name in ("bin", "lib", "include", "share"): + src = source / name + if src.exists(): + shutil.move(str(src), str(root / name)) + license_src = source / "LICENSE" + if license_src.exists(): + shutil.move(str(license_src), str(root / "LICENSE")) + _chmod_executable(root / "bin" / "nvattest") + + +def _chmod_executable(path: Path) -> None: + mode = path.stat().st_mode + path.chmod(mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH) + + +def _write_sidecar(path: Path, record: NvattestInstallRecord) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + with tempfile.NamedTemporaryFile( + "w", dir=path.parent, delete=False, encoding="utf-8" + ) as handle: + handle.write(record.to_json()) + tmp_path = Path(handle.name) + tmp_path.replace(path) diff --git a/solstone/think/services/spp_attest/__init__.py b/solstone/think/services/spp_attest/__init__.py index 74ab1d682..c12f5fb5f 100644 --- a/solstone/think/services/spp_attest/__init__.py +++ b/solstone/think/services/spp_attest/__init__.py @@ -9,6 +9,10 @@ from solstone.think.services.spp_attest.binding import ( composite_binding_hash, ) from solstone.think.services.spp_attest.errors import VerificationError +from solstone.think.services.spp_attest.nvgpu.appraise import appraise_gpu_leg +from solstone.think.services.spp_attest.nvgpu.claims import GpuAppraisal +from solstone.think.services.spp_attest.nvgpu.errors import GpuAppraisalError +from solstone.think.services.spp_attest.nvgpu.evidence import to_nvattest_evidence from solstone.think.services.spp_attest.snp import ( AppraisalStep, CpuAppraisal, @@ -26,14 +30,18 @@ __all__ = [ "BINDING_DOMAIN", "AppraisalStep", "CpuAppraisal", + "GpuAppraisal", + "GpuAppraisalError", "GpuEnvelope", "Policy", "TpmQuoteVerifier", "VerificationError", "appraise_cpu_leg", + "appraise_gpu_leg", "check_envelope_nonce", "composite_binding_hash", "decode_gpu_envelope", "extract_spdm_nonce", + "to_nvattest_evidence", "verify_quote", ] diff --git a/solstone/think/services/spp_attest/nvgpu/__init__.py b/solstone/think/services/spp_attest/nvgpu/__init__.py new file mode 100644 index 000000000..6dc020a63 --- /dev/null +++ b/solstone/think/services/spp_attest/nvgpu/__init__.py @@ -0,0 +1,18 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""NVIDIA GPU-leg appraisal for SPP attestation.""" + +from __future__ import annotations + +from solstone.think.services.spp_attest.nvgpu.appraise import appraise_gpu_leg +from solstone.think.services.spp_attest.nvgpu.claims import GpuAppraisal +from solstone.think.services.spp_attest.nvgpu.errors import GpuAppraisalError +from solstone.think.services.spp_attest.nvgpu.evidence import to_nvattest_evidence + +__all__ = [ + "GpuAppraisal", + "GpuAppraisalError", + "appraise_gpu_leg", + "to_nvattest_evidence", +] diff --git a/solstone/think/services/spp_attest/nvgpu/appraise.py b/solstone/think/services/spp_attest/nvgpu/appraise.py new file mode 100644 index 000000000..507a5b725 --- /dev/null +++ b/solstone/think/services/spp_attest/nvgpu/appraise.py @@ -0,0 +1,136 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""Shell-out NVIDIA GPU-leg appraisal for SPP attestation.""" + +from __future__ import annotations + +import json +import subprocess +import tempfile +from pathlib import Path + +from solstone.think.services.spp_attest.nvgpu.binary import ( + build_nvattest_attest_command, +) +from solstone.think.services.spp_attest.nvgpu.claims import ( + GpuAppraisal, + NvattestAcceptance, + build_gpu_appraisal, + classify_nvattest_result, + parse_nvattest_stdout, +) +from solstone.think.services.spp_attest.nvgpu.errors import GpuAppraisalError +from solstone.think.services.spp_attest.nvgpu.evidence import to_nvattest_evidence +from solstone.think.services.spp_attest.snp import AppraisalStep +from solstone.think.services.spp_attest.tlv import GpuEnvelope + + +def appraise_gpu_leg( + envelope: GpuEnvelope, + owner_nonce: bytes, + *, + nvattest_dir: Path, + rim_store: str = "remote", + rim_dir: Path | None = None, +) -> GpuAppraisal: + """Appraise the NVIDIA GPU side of an SPP composite attestation. + + ``--rim-store remote`` performs network egress inside the subprocess: a fetch + of NVIDIA's signed RIM + OCSP reference data only (public golden + measurements + revocation), never journal content, and the verdict is still + computed locally. "Offline" here means no local GPU and no NRAS, not "no + network". ``rim_store="dir"`` + ``--rim-dir`` is the fully-offline-RIM path + (OCSP still needs network). + """ + + evidence_path: Path | None = None + try: + evidence = to_nvattest_evidence(envelope, owner_nonce) + with tempfile.NamedTemporaryFile( + "w", + dir=tempfile.gettempdir(), + delete=False, + encoding="utf-8", + prefix="solstone-nvattest-", + suffix=".json", + ) as handle: + evidence_path = Path(handle.name) + handle.write(json.dumps(evidence, sort_keys=True)) + handle.write("\n") + + command = build_nvattest_attest_command( + nvattest_dir=nvattest_dir, + evidence_file=evidence_path, + owner_nonce=owner_nonce, + rim_store=rim_store, + rim_dir=rim_dir, + ) + try: + completed = subprocess.run( + command.argv, + env=command.env, + capture_output=True, + text=True, + check=False, + ) + except OSError as exc: + raise GpuAppraisalError( + "nvattest_unavailable", + f"failed to execute nvattest: {exc}", + ) from exc + + try: + stdout_obj = parse_nvattest_stdout(completed.stdout) + except ValueError as exc: + raise GpuAppraisalError( + "gpu_appraisal_failed", + str(exc), + stderr=completed.stderr, + ) from exc + + decision = classify_nvattest_result( + completed.returncode, + stdout_obj, + owner_nonce=owner_nonce, + ) + if not isinstance(decision, NvattestAcceptance): + raise GpuAppraisalError( + decision.reason, + decision.detail, + stderr=completed.stderr, + ) + + steps = [ + _ok( + "nvattest", + "returncode=0 result_code=0 result_message=Ok", + ), + _ok( + "overall-eat", + "alg=none iss=NVAT-LOCAL-VERIFIER overall_att_result=True", + ), + _ok( + "gpu-claims", + "claims-version=3.0 report, driver-RIM, vbios-RIM checks passed", + ), + ] + try: + return build_gpu_appraisal( + claim=decision.claim, + envelope=envelope, + steps=steps, + ) + except ValueError as exc: + raise GpuAppraisalError( + "gpu_appraisal_failed", + str(exc), + stderr=completed.stderr, + ) from exc + finally: + if evidence_path is not None: + evidence_path.unlink(missing_ok=True) + + +def _ok(name: str, detail: str) -> AppraisalStep: + return AppraisalStep(name=name, status="ok", detail=detail) diff --git a/solstone/think/services/spp_attest/nvgpu/binary.py b/solstone/think/services/spp_attest/nvgpu/binary.py new file mode 100644 index 000000000..a26501ce4 --- /dev/null +++ b/solstone/think/services/spp_attest/nvgpu/binary.py @@ -0,0 +1,96 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""Pure nvattest binary path and command construction.""" + +from __future__ import annotations + +import os +from dataclasses import dataclass +from pathlib import Path + +from solstone.think.services.spp_attest.nvgpu.errors import GpuAppraisalError +from solstone.think.services.spp_attest.tlv import SPDM_NONCE_SIZE + + +@dataclass(frozen=True, slots=True) +class NvattestCommand: + argv: list[str] + env: dict[str, str] + + +def locate_nvattest(nvattest_dir: Path) -> tuple[Path, Path]: + """Return the nvattest binary and lib directory under an injected install dir.""" + + binary = nvattest_dir / "bin" / "nvattest" + lib_dir = nvattest_dir / "lib" + if not nvattest_dir.is_dir(): + raise GpuAppraisalError( + "nvattest_unavailable", + f"nvattest directory is missing: {nvattest_dir}", + ) + if not binary.is_file(): + raise GpuAppraisalError( + "nvattest_unavailable", + f"nvattest binary is missing: {binary}", + ) + if not lib_dir.is_dir(): + raise GpuAppraisalError( + "nvattest_unavailable", + f"nvattest lib directory is missing: {lib_dir}", + ) + return binary, lib_dir + + +def build_nvattest_attest_command( + *, + nvattest_dir: Path, + evidence_file: Path, + owner_nonce: bytes, + rim_store: str = "remote", + rim_dir: Path | None = None, +) -> NvattestCommand: + """Build the nvattest local-verifier attest command.""" + + if len(owner_nonce) != SPDM_NONCE_SIZE: + raise ValueError(f"owner_nonce is {len(owner_nonce)} bytes, expected 32") + if rim_store not in {"remote", "dir"}: + raise ValueError("rim_store must be 'remote' or 'dir'") + if rim_store == "dir" and rim_dir is None: + raise ValueError("rim_dir is required when rim_store == 'dir'") + if rim_store == "remote" and rim_dir is not None: + raise ValueError("rim_dir is only valid when rim_store == 'dir'") + + binary, lib_dir = locate_nvattest(nvattest_dir) + argv = [ + str(binary), + "--format", + "json", + "attest", + "--device", + "gpu", + "--gpu-evidence-source", + "file", + "--gpu-evidence-file", + str(evidence_file), + "--verifier", + "local", + "--rim-store", + rim_store, + ] + if rim_dir is not None: + argv.extend(["--rim-dir", str(rim_dir)]) + argv.extend(["--nonce", owner_nonce.hex()]) + return NvattestCommand( + argv=argv, env={**os.environ, "LD_LIBRARY_PATH": str(lib_dir)} + ) + + +def build_nvattest_version_command(*, nvattest_dir: Path) -> NvattestCommand: + """Build the nvattest version command.""" + + binary, lib_dir = locate_nvattest(nvattest_dir) + return NvattestCommand( + argv=[str(binary), "version"], + env={**os.environ, "LD_LIBRARY_PATH": str(lib_dir)}, + ) diff --git a/solstone/think/services/spp_attest/nvgpu/claims.py b/solstone/think/services/spp_attest/nvgpu/claims.py new file mode 100644 index 000000000..90b5d25a8 --- /dev/null +++ b/solstone/think/services/spp_attest/nvgpu/claims.py @@ -0,0 +1,294 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""Pure nvattest stdout parsing and claim appraisal.""" + +from __future__ import annotations + +import base64 +import binascii +import json +from dataclasses import dataclass +from typing import Any + +from solstone.think.services.spp_attest.nvgpu.errors import GpuAppraisalReason +from solstone.think.services.spp_attest.snp import AppraisalStep +from solstone.think.services.spp_attest.tlv import GpuEnvelope + + +@dataclass(frozen=True, slots=True) +class GpuAppraisal: + """GPU appraisal provenance. + + envelope_gpu_uuid is copied from our SPP GPU envelope field 6. nvattest never + consumes or verifies that UUID, so it is intentionally prefixed as envelope + metadata rather than an nvattest-verified claim. + """ + + steps: list[AppraisalStep] + driver_version: str + vbios_version: str + hwmodel: str + ueid: str + oemid: str + eat_nonce: str + claims_version: str + arch: str + envelope_gpu_uuid: str + + +@dataclass(frozen=True, slots=True) +class NvattestAcceptance: + stdout: dict[str, Any] + claim: dict[str, Any] + overall_payload: dict[str, Any] + + +@dataclass(frozen=True, slots=True) +class NvattestRejection: + reason: GpuAppraisalReason + detail: str + + +class _ClaimReject(Exception): + pass + + +_REPORT_TRUE_KEYS = ( + "x-nvidia-gpu-attestation-report-parsed", + "x-nvidia-gpu-attestation-report-signature-verified", + "x-nvidia-gpu-attestation-report-nonce-match", + "x-nvidia-gpu-attestation-report-cert-chain-fwid-match", + "x-nvidia-gpu-arch-check", +) +_DRIVER_RIM_TRUE_KEYS = ( + "x-nvidia-gpu-driver-rim-signature-verified", + "x-nvidia-gpu-driver-rim-version-match", + "x-nvidia-gpu-driver-rim-measurements-available", +) +_VBIOS_RIM_TRUE_KEYS = ( + "x-nvidia-gpu-vbios-rim-signature-verified", + "x-nvidia-gpu-vbios-rim-version-match", + "x-nvidia-gpu-vbios-rim-measurements-available", + "x-nvidia-gpu-vbios-index-no-conflict", +) +_CERT_CHAIN_KEYS = ( + "x-nvidia-gpu-attestation-report-cert-chain", + "x-nvidia-gpu-driver-rim-cert-chain", + "x-nvidia-gpu-vbios-rim-cert-chain", +) + + +def parse_nvattest_stdout(stdout: str) -> object: + """Parse nvattest stdout as JSON.""" + + if not stdout: + raise ValueError("nvattest stdout is empty") + try: + return json.loads(stdout) + except json.JSONDecodeError as exc: + raise ValueError(f"nvattest stdout did not parse: {exc}") from exc + + +def classify_nvattest_result( + returncode: int, + stdout_obj: object, + *, + owner_nonce: bytes, +) -> NvattestAcceptance | NvattestRejection: + """Return a fail-closed appraisal decision without consulting stderr.""" + + if not isinstance(stdout_obj, dict): + return NvattestRejection( + "gpu_appraisal_failed", "nvattest stdout is not an object" + ) + + try: + result_code = _required_stdout(stdout_obj, "result_code") + result_message = _required_stdout(stdout_obj, "result_message") + except _ClaimReject as exc: + return NvattestRejection("gpu_appraisal_failed", str(exc)) + result_code_is_green = ( + isinstance(result_code, int) + and not isinstance(result_code, bool) + and result_code == 0 + ) + result_message_is_green = isinstance(result_message, str) and result_message == "Ok" + if returncode != 0 or not result_code_is_green or not result_message_is_green: + return NvattestRejection( + _reason_for_result_code(result_code), + ( + "nvattest non-green result: " + f"returncode={returncode} result_code={result_code!r} " + f"result_message={result_message!r}" + ), + ) + + try: + claims = _required_stdout(stdout_obj, "claims") + except _ClaimReject as exc: + return NvattestRejection("gpu_appraisal_failed", str(exc)) + if not isinstance(claims, list) or len(claims) != 1: + return NvattestRejection( + "gpu_appraisal_failed", + "nvattest claims is not a list of exactly one object", + ) + claim = claims[0] + if not isinstance(claim, dict): + return NvattestRejection( + "gpu_appraisal_failed", "nvattest claim is not an object" + ) + + try: + overall_payload = _parse_overall_eat( + _required_stdout(stdout_obj, "detached_eat") + ) + _check_claim(claim, owner_nonce.hex()) + except _ClaimReject as exc: + return NvattestRejection("gpu_appraisal_failed", str(exc)) + return NvattestAcceptance( + stdout=stdout_obj, + claim=claim, + overall_payload=overall_payload, + ) + + +def build_gpu_appraisal( + *, + claim: dict[str, Any], + envelope: GpuEnvelope, + steps: list[AppraisalStep], +) -> GpuAppraisal: + """Build GPU provenance from accepted nvattest claims and envelope metadata.""" + + try: + return GpuAppraisal( + steps=steps, + driver_version=_claim_str(claim, "x-nvidia-gpu-driver-version"), + vbios_version=_claim_str(claim, "x-nvidia-gpu-vbios-version"), + hwmodel=_claim_str(claim, "hwmodel"), + ueid=_claim_str(claim, "ueid"), + oemid=_claim_str(claim, "oemid"), + eat_nonce=_claim_str(claim, "eat_nonce"), + claims_version=_claim_str(claim, "x-nvidia-gpu-claims-version"), + arch=envelope.field(7).decode("utf-8").upper(), + envelope_gpu_uuid=envelope.field(6).decode("utf-8"), + ) + except _ClaimReject as exc: + raise ValueError(str(exc)) from exc + + +def _reason_for_result_code(result_code: object) -> GpuAppraisalReason: + if result_code == 504: + return "gpu_nonce_mismatch" + return "gpu_appraisal_failed" + + +def _required_stdout(stdout: dict[str, Any], key: str) -> Any: + try: + return stdout[key] + except KeyError as exc: + raise _ClaimReject(f"nvattest stdout missing key {key!r}") from exc + + +def _required_claim(claim: dict[str, Any], key: str) -> Any: + try: + return claim[key] + except KeyError as exc: + raise _ClaimReject(f"nvattest claim missing key {key!r}") from exc + + +def _claim_str(claim: dict[str, Any], key: str) -> str: + value = _required_claim(claim, key) + if not isinstance(value, str): + raise _ClaimReject(f"nvattest claim {key!r} is not a string") + return value + + +def _require_equal(claim: dict[str, Any], key: str, expected: object) -> None: + value = _required_claim(claim, key) + if value != expected: + raise _ClaimReject(f"nvattest claim {key!r}={value!r}, expected {expected!r}") + + +def _require_is(claim: dict[str, Any], key: str, expected: object) -> None: + value = _required_claim(claim, key) + if value is not expected: + raise _ClaimReject(f"nvattest claim {key!r}={value!r}, expected {expected!r}") + + +def _check_claim(claim: dict[str, Any], owner_nonce_hex: str) -> None: + _require_equal(claim, "x-nvidia-gpu-claims-version", "3.0") + _require_equal(claim, "x-nvidia-device-type", "gpu") + _require_equal(claim, "measres", "success") + _require_is(claim, "secboot", True) + _require_equal(claim, "dbgstat", "disabled") + _require_equal(claim, "eat_nonce", owner_nonce_hex) + + for key in _REPORT_TRUE_KEYS: + _require_is(claim, key, True) + for key in _DRIVER_RIM_TRUE_KEYS: + _require_is(claim, key, True) + for key in _VBIOS_RIM_TRUE_KEYS: + _require_is(claim, key, True) + + # Deliberately not checking *-rim-fetched: legitimate --rim-store dir runs + # verify local RIM data while reporting those fetch markers as false. + _require_is(claim, "x-nvidia-mismatch-measurement-records", None) + for key in _CERT_CHAIN_KEYS: + _check_cert_chain(_required_claim(claim, key), key) + + +def _check_cert_chain(value: object, key: str) -> None: + if not isinstance(value, dict): + raise _ClaimReject(f"nvattest claim {key!r} is not an object") + _require_equal(value, "x-nvidia-cert-status", "valid") + _require_equal(value, "x-nvidia-cert-ocsp-status", "good") + _require_is(value, "x-nvidia-cert-ocsp-response-valid", True) + _require_is(value, "x-nvidia-cert-ocsp-nonce-matches", True) + _require_is(value, "x-nvidia-cert-revocation-reason", None) + + +def _parse_overall_eat(detached_eat: object) -> dict[str, Any]: + if not isinstance(detached_eat, list) or not detached_eat: + raise _ClaimReject("nvattest detached_eat does not contain an overall JWT") + overall = detached_eat[0] + if ( + not isinstance(overall, list) + or len(overall) != 2 + or overall[0] != "JWT" + or not isinstance(overall[1], str) + ): + raise _ClaimReject("nvattest detached_eat overall JWT has invalid shape") + + parts = overall[1].split(".") + if len(parts) != 3: + raise _ClaimReject("nvattest overall JWT does not have three segments") + header = _decode_jwt_segment(parts[0], "header") + payload = _decode_jwt_segment(parts[1], "payload") + try: + alg = header["alg"] + iss = payload["iss"] + overall_result = payload["x-nvidia-overall-att-result"] + except KeyError as exc: + raise _ClaimReject(f"nvattest overall JWT missing key {exc.args[0]!r}") from exc + if alg != "none": + raise _ClaimReject(f"nvattest overall JWT alg={alg!r}, expected 'none'") + if iss != "NVAT-LOCAL-VERIFIER": + raise _ClaimReject( + f"nvattest overall JWT iss={iss!r}, expected 'NVAT-LOCAL-VERIFIER'" + ) + if overall_result is not True: + raise _ClaimReject("nvattest overall attestation result is not true") + return payload + + +def _decode_jwt_segment(segment: str, label: str) -> dict[str, Any]: + try: + raw = base64.urlsafe_b64decode(segment + "=" * (-len(segment) % 4)) + decoded = json.loads(raw) + except (binascii.Error, json.JSONDecodeError, UnicodeDecodeError) as exc: + raise _ClaimReject(f"nvattest overall JWT {label} did not parse") from exc + if not isinstance(decoded, dict): + raise _ClaimReject(f"nvattest overall JWT {label} is not an object") + return decoded diff --git a/solstone/think/services/spp_attest/nvgpu/errors.py b/solstone/think/services/spp_attest/nvgpu/errors.py new file mode 100644 index 000000000..de0a599de --- /dev/null +++ b/solstone/think/services/spp_attest/nvgpu/errors.py @@ -0,0 +1,46 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""GPU appraisal error types.""" + +from __future__ import annotations + +from typing import Literal + +from solstone.think.services.spp_attest.errors import VerificationError + +GpuAppraisalReason = Literal[ + "nvattest_unavailable", + "gpu_nonce_mismatch", + "gpu_appraisal_failed", +] +STDERR_TAIL_CHARS = 4000 + + +class GpuAppraisalError(VerificationError): + """Raised when NVIDIA GPU appraisal fails.""" + + reason: GpuAppraisalReason + stderr: str + + def __init__( + self, + reason: GpuAppraisalReason, + message: str, + *, + stderr: str = "", + ) -> None: + self.reason = reason + self.stderr = _stderr_tail(stderr) + detail = f"{reason}: {message}" + if self.stderr: + detail = f"{detail}\nnvattest stderr tail:\n{self.stderr}" + super().__init__(detail) + + +def _stderr_tail(stderr: str) -> str: + if not stderr: + return "" + if len(stderr) <= STDERR_TAIL_CHARS: + return stderr + return "[truncated to last 4000 chars]\n" + stderr[-STDERR_TAIL_CHARS:] diff --git a/solstone/think/services/spp_attest/nvgpu/evidence.py b/solstone/think/services/spp_attest/nvgpu/evidence.py new file mode 100644 index 000000000..4222e4460 --- /dev/null +++ b/solstone/think/services/spp_attest/nvgpu/evidence.py @@ -0,0 +1,25 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""Transform SPP GPU envelopes into nvattest evidence JSON.""" + +from __future__ import annotations + +import base64 + +from solstone.think.services.spp_attest.tlv import GpuEnvelope + + +def to_nvattest_evidence( + envelope: GpuEnvelope, owner_nonce: bytes +) -> list[dict[str, str]]: + """Return the one-item nvattest serialized-evidence array.""" + + return [ + { + "arch": envelope.field(7).decode("utf-8").upper(), + "certificate": base64.b64encode(envelope.field(3)).decode("ascii"), + "evidence": base64.b64encode(envelope.field(2)).decode("ascii"), + "nonce": owner_nonce.hex(), + } + ] diff --git a/tests/fixtures/spp_attest/nvattest/negA.stderr b/tests/fixtures/spp_attest/nvattest/negA.stderr new file mode 100644 index 000000000..fc8d59c6d --- /dev/null +++ b/tests/fixtures/spp_attest/nvattest/negA.stderr @@ -0,0 +1,5 @@ +2026-07-11 12:02:02.694 [gpu/evidence.cpp:663 get_evidence] [error] Nonce from GPU evidence does not match the nonce used for attestation. +2026-07-11 12:02:02.694 [gpu/evidence.cpp:664 get_evidence] [error] Nonce from GPU evidence: a892c0c66c8a5abf562b00223099061ae2c789673515554ccee5ea276a377ef1 Nonce used for attestation: 0000000000000000000000000000000000000000000000000000000000000000 +2026-07-11 12:02:02.694 [gpu/evidence.cpp:665 get_evidence] [error] Does the nonce from serialized evidence JSON file match the nonce used for attestation? +2026-07-11 12:02:02.694 [attestation.cpp:318 attest_gpus] [error] Failed to collect GPU evidence +2026-07-11 12:02:02.694 [nvat.cpp:668 nvat_attest_device] [error] Error while performing device attestation diff --git a/tests/fixtures/spp_attest/nvattest/negA.stdout b/tests/fixtures/spp_attest/nvattest/negA.stdout new file mode 100644 index 000000000..18277e49a --- /dev/null +++ b/tests/fixtures/spp_attest/nvattest/negA.stdout @@ -0,0 +1,6 @@ +{ + "claims": {}, + "detached_eat": {}, + "result_code": 504, + "result_message": "GPU Evidence Nonce Mismatch" +} diff --git a/tests/fixtures/spp_attest/nvattest/negB.stderr b/tests/fixtures/spp_attest/nvattest/negB.stderr new file mode 100644 index 000000000..85a6ef0a8 --- /dev/null +++ b/tests/fixtures/spp_attest/nvattest/negB.stderr @@ -0,0 +1,5 @@ +2026-07-11 12:02:02.712 [gpu/evidence.cpp:663 get_evidence] [error] Nonce from GPU evidence does not match the nonce used for attestation. +2026-07-11 12:02:02.712 [gpu/evidence.cpp:664 get_evidence] [error] Nonce from GPU evidence: 0000000000000000000000000000000000000000000000000000000000000000 Nonce used for attestation: a892c0c66c8a5abf562b00223099061ae2c789673515554ccee5ea276a377ef1 +2026-07-11 12:02:02.712 [gpu/evidence.cpp:665 get_evidence] [error] Does the nonce from serialized evidence JSON file match the nonce used for attestation? +2026-07-11 12:02:02.712 [attestation.cpp:318 attest_gpus] [error] Failed to collect GPU evidence +2026-07-11 12:02:02.712 [nvat.cpp:668 nvat_attest_device] [error] Error while performing device attestation diff --git a/tests/fixtures/spp_attest/nvattest/negB.stdout b/tests/fixtures/spp_attest/nvattest/negB.stdout new file mode 100644 index 000000000..18277e49a --- /dev/null +++ b/tests/fixtures/spp_attest/nvattest/negB.stdout @@ -0,0 +1,6 @@ +{ + "claims": {}, + "detached_eat": {}, + "result_code": 504, + "result_message": "GPU Evidence Nonce Mismatch" +} diff --git a/tests/fixtures/spp_attest/nvattest/negC.stderr b/tests/fixtures/spp_attest/nvattest/negC.stderr new file mode 100644 index 000000000..90bcdf485 --- /dev/null +++ b/tests/fixtures/spp_attest/nvattest/negC.stderr @@ -0,0 +1,4 @@ +2026-07-11 12:02:35.866 [attestation.cpp:278 attest_device] [info] Attesting GPUs +2026-07-11 12:02:35.872 [gpu/evidence.cpp:255 generate_gpu_evidence_claims] [error] Nonce mismatch. nonce_from_ar: a892c0c66c8a5abf562b00223099061ae2c789673515554ccee5ea276a377ef1 m_nonce: 0000000000000000000000000000000000000000000000000000000000000000 +2026-07-11 12:02:35.872 [attestation.cpp:323 attest_gpus] [error] Failed to verify GPU evidence +2026-07-11 12:02:35.872 [nvat.cpp:668 nvat_attest_device] [error] Error while performing device attestation diff --git a/tests/fixtures/spp_attest/nvattest/negC.stdout b/tests/fixtures/spp_attest/nvattest/negC.stdout new file mode 100644 index 000000000..18277e49a --- /dev/null +++ b/tests/fixtures/spp_attest/nvattest/negC.stdout @@ -0,0 +1,6 @@ +{ + "claims": {}, + "detached_eat": {}, + "result_code": 504, + "result_message": "GPU Evidence Nonce Mismatch" +} diff --git a/tests/fixtures/spp_attest/nvattest/positive.stderr b/tests/fixtures/spp_attest/nvattest/positive.stderr new file mode 100644 index 000000000..f8bfd95c9 --- /dev/null +++ b/tests/fixtures/spp_attest/nvattest/positive.stderr @@ -0,0 +1 @@ +2026-07-11 12:01:01.740 [attestation.cpp:278 attest_device] [info] Attesting GPUs diff --git a/tests/fixtures/spp_attest/nvattest/positive.stdout b/tests/fixtures/spp_attest/nvattest/positive.stdout new file mode 100644 index 000000000..d60a83993 --- /dev/null +++ b/tests/fixtures/spp_attest/nvattest/positive.stdout @@ -0,0 +1,67 @@ +{ + "claims": [ + { + "dbgstat": "disabled", + "eat_nonce": "a892c0c66c8a5abf562b00223099061ae2c789673515554ccee5ea276a377ef1", + "hwmodel": "GH100 A01 GSP BROM", + "measres": "success", + "oemid": "5703", + "secboot": true, + "ueid": "630243178525731796206631439086965082795369051455", + "x-nvidia-device-type": "gpu", + "x-nvidia-gpu-arch-check": true, + "x-nvidia-gpu-attestation-report-cert-chain": { + "x-nvidia-cert-expiration-date": "9999-12-31T23:59:59Z", + "x-nvidia-cert-ocsp-nonce-matches": true, + "x-nvidia-cert-ocsp-response-valid": true, + "x-nvidia-cert-ocsp-status": "good", + "x-nvidia-cert-revocation-reason": null, + "x-nvidia-cert-status": "valid" + }, + "x-nvidia-gpu-attestation-report-cert-chain-fwid-match": true, + "x-nvidia-gpu-attestation-report-nonce-match": true, + "x-nvidia-gpu-attestation-report-parsed": true, + "x-nvidia-gpu-attestation-report-signature-verified": true, + "x-nvidia-gpu-claims-version": "3.0", + "x-nvidia-gpu-driver-rim-cert-chain": { + "x-nvidia-cert-expiration-date": "2028-04-23T05:42:57Z", + "x-nvidia-cert-ocsp-nonce-matches": true, + "x-nvidia-cert-ocsp-response-valid": true, + "x-nvidia-cert-ocsp-status": "good", + "x-nvidia-cert-revocation-reason": null, + "x-nvidia-cert-status": "valid" + }, + "x-nvidia-gpu-driver-rim-fetched": true, + "x-nvidia-gpu-driver-rim-measurements-available": true, + "x-nvidia-gpu-driver-rim-signature-verified": true, + "x-nvidia-gpu-driver-rim-version-match": true, + "x-nvidia-gpu-driver-version": "595.71.05", + "x-nvidia-gpu-vbios-index-no-conflict": true, + "x-nvidia-gpu-vbios-rim-cert-chain": { + "x-nvidia-cert-expiration-date": "2027-11-11T02:15:51Z", + "x-nvidia-cert-ocsp-nonce-matches": true, + "x-nvidia-cert-ocsp-response-valid": true, + "x-nvidia-cert-ocsp-status": "good", + "x-nvidia-cert-revocation-reason": null, + "x-nvidia-cert-status": "valid" + }, + "x-nvidia-gpu-vbios-rim-fetched": true, + "x-nvidia-gpu-vbios-rim-measurements-available": true, + "x-nvidia-gpu-vbios-rim-signature-verified": true, + "x-nvidia-gpu-vbios-rim-version-match": true, + "x-nvidia-gpu-vbios-version": "96.00.88.00.11", + "x-nvidia-mismatch-measurement-records": null + } + ], + "detached_eat": [ + [ + "JWT", + "eyJhbGciOiJub25lIn0.eyJlYXRfbm9uY2UiOiJhODkyYzBjNjZjOGE1YWJmNTYyYjAwMjIzMDk5MDYxYWUyYzc4OTY3MzUxNTU1NGNjZWU1ZWEyNzZhMzc3ZWYxIiwiZXhwIjoxNzgzNzk2NDYzLCJpYXQiOjE3ODM3OTI4NjMsImlzcyI6Ik5WQVQtTE9DQUwtVkVSSUZJRVIiLCJqdGkiOiIxNzQ2MzJhYzVkN2UyOGE4ZDk4MzMyMjZiM2QyMTI5YzE1ZTQ5MGYwZWFhYTI3NmU3MjYxZTA5ZjE2Mjg3ZTM1Iiwic3ViIjoiTlZJRElBLVBMQVRGT1JNLUFUVEVTVEFUSU9OIiwic3VibW9kcyI6eyJHUFUtMCI6WyJESUdFU1QiLFsiU0hBMjU2IiwiYTIxODkxOWQ1Y2JmNmI4MmYzYzExZWY5MDRhM2NjNWRlNGZlNDg2MDk0N2MwNzYyNDQxOWY5MzUyNzdiYjEyZSJdXX0sIngtbnZpZGlhLW92ZXJhbGwtYXR0LXJlc3VsdCI6dHJ1ZSwieC1udmlkaWEtdmVyIjoiMy4wIn0." + ], + { + "GPU-0": "eyJhbGciOiJub25lIn0.eyJkYmdzdGF0IjoiZGlzYWJsZWQiLCJlYXRfbm9uY2UiOiJhODkyYzBjNjZjOGE1YWJmNTYyYjAwMjIzMDk5MDYxYWUyYzc4OTY3MzUxNTU1NGNjZWU1ZWEyNzZhMzc3ZWYxIiwiZXhwIjoxNzgzNzk2NDYzLCJod21vZGVsIjoiR0gxMDAgQTAxIEdTUCBCUk9NIiwiaWF0IjoxNzgzNzkyODYzLCJpc3MiOiJOVkFULUxPQ0FMLVZFUklGSUVSIiwianRpIjoiMDQyMzUxMTRkNWRkNDg0OWJiNDZmYmU2YWFlZmYzYjM2NmFlZmM5NzY5YTE0MmU3YmVjMmZiOThjMDBkMzQxMyIsIm1lYXNyZXMiOiJzdWNjZXNzIiwib2VtaWQiOiI1NzAzIiwic2VjYm9vdCI6dHJ1ZSwidWVpZCI6IjYzMDI0MzE3ODUyNTczMTc5NjIwNjYzMTQzOTA4Njk2NTA4Mjc5NTM2OTA1MTQ1NSIsIngtbnZpZGlhLWRldmljZS10eXBlIjoiZ3B1IiwieC1udmlkaWEtZ3B1LWFyY2gtY2hlY2siOnRydWUsIngtbnZpZGlhLWdwdS1hdHRlc3RhdGlvbi1yZXBvcnQtY2VydC1jaGFpbiI6eyJ4LW52aWRpYS1jZXJ0LWV4cGlyYXRpb24tZGF0ZSI6Ijk5OTktMTItMzFUMjM6NTk6NTlaIiwieC1udmlkaWEtY2VydC1vY3NwLW5vbmNlLW1hdGNoZXMiOnRydWUsIngtbnZpZGlhLWNlcnQtb2NzcC1yZXNwb25zZS12YWxpZCI6dHJ1ZSwieC1udmlkaWEtY2VydC1vY3NwLXN0YXR1cyI6Imdvb2QiLCJ4LW52aWRpYS1jZXJ0LXJldm9jYXRpb24tcmVhc29uIjpudWxsLCJ4LW52aWRpYS1jZXJ0LXN0YXR1cyI6InZhbGlkIn0sIngtbnZpZGlhLWdwdS1hdHRlc3RhdGlvbi1yZXBvcnQtY2VydC1jaGFpbi1md2lkLW1hdGNoIjp0cnVlLCJ4LW52aWRpYS1ncHUtYXR0ZXN0YXRpb24tcmVwb3J0LW5vbmNlLW1hdGNoIjp0cnVlLCJ4LW52aWRpYS1ncHUtYXR0ZXN0YXRpb24tcmVwb3J0LXBhcnNlZCI6dHJ1ZSwieC1udmlkaWEtZ3B1LWF0dGVzdGF0aW9uLXJlcG9ydC1zaWduYXR1cmUtdmVyaWZpZWQiOnRydWUsIngtbnZpZGlhLWdwdS1jbGFpbXMtdmVyc2lvbiI6IjMuMCIsIngtbnZpZGlhLWdwdS1kcml2ZXItcmltLWNlcnQtY2hhaW4iOnsieC1udmlkaWEtY2VydC1leHBpcmF0aW9uLWRhdGUiOiIyMDI4LTA0LTIzVDA1OjQyOjU3WiIsIngtbnZpZGlhLWNlcnQtb2NzcC1ub25jZS1tYXRjaGVzIjp0cnVlLCJ4LW52aWRpYS1jZXJ0LW9jc3AtcmVzcG9uc2UtdmFsaWQiOnRydWUsIngtbnZpZGlhLWNlcnQtb2NzcC1zdGF0dXMiOiJnb29kIiwieC1udmlkaWEtY2VydC1yZXZvY2F0aW9uLXJlYXNvbiI6bnVsbCwieC1udmlkaWEtY2VydC1zdGF0dXMiOiJ2YWxpZCJ9LCJ4LW52aWRpYS1ncHUtZHJpdmVyLXJpbS1mZXRjaGVkIjp0cnVlLCJ4LW52aWRpYS1ncHUtZHJpdmVyLXJpbS1tZWFzdXJlbWVudHMtYXZhaWxhYmxlIjp0cnVlLCJ4LW52aWRpYS1ncHUtZHJpdmVyLXJpbS1zaWduYXR1cmUtdmVyaWZpZWQiOnRydWUsIngtbnZpZGlhLWdwdS1kcml2ZXItcmltLXZlcnNpb24tbWF0Y2giOnRydWUsIngtbnZpZGlhLWdwdS1kcml2ZXItdmVyc2lvbiI6IjU5NS43MS4wNSIsIngtbnZpZGlhLWdwdS12Ymlvcy1pbmRleC1uby1jb25mbGljdCI6dHJ1ZSwieC1udmlkaWEtZ3B1LXZiaW9zLXJpbS1jZXJ0LWNoYWluIjp7IngtbnZpZGlhLWNlcnQtZXhwaXJhdGlvbi1kYXRlIjoiMjAyNy0xMS0xMVQwMjoxNTo1MVoiLCJ4LW52aWRpYS1jZXJ0LW9jc3Atbm9uY2UtbWF0Y2hlcyI6dHJ1ZSwieC1udmlkaWEtY2VydC1vY3NwLXJlc3BvbnNlLXZhbGlkIjp0cnVlLCJ4LW52aWRpYS1jZXJ0LW9jc3Atc3RhdHVzIjoiZ29vZCIsIngtbnZpZGlhLWNlcnQtcmV2b2NhdGlvbi1yZWFzb24iOm51bGwsIngtbnZpZGlhLWNlcnQtc3RhdHVzIjoidmFsaWQifSwieC1udmlkaWEtZ3B1LXZiaW9zLXJpbS1mZXRjaGVkIjp0cnVlLCJ4LW52aWRpYS1ncHUtdmJpb3MtcmltLW1lYXN1cmVtZW50cy1hdmFpbGFibGUiOnRydWUsIngtbnZpZGlhLWdwdS12Ymlvcy1yaW0tc2lnbmF0dXJlLXZlcmlmaWVkIjp0cnVlLCJ4LW52aWRpYS1ncHUtdmJpb3MtcmltLXZlcnNpb24tbWF0Y2giOnRydWUsIngtbnZpZGlhLWdwdS12Ymlvcy12ZXJzaW9uIjoiOTYuMDAuODguMDAuMTEiLCJ4LW52aWRpYS1taXNtYXRjaC1tZWFzdXJlbWVudC1yZWNvcmRzIjpudWxsfQ." + } + ], + "result_code": 0, + "result_message": "Ok" +} diff --git a/tests/services/test_spp_attest_nvgpu.py b/tests/services/test_spp_attest_nvgpu.py new file mode 100644 index 000000000..6fb0113dc --- /dev/null +++ b/tests/services/test_spp_attest_nvgpu.py @@ -0,0 +1,432 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +from __future__ import annotations + +import base64 +import json +import subprocess +from copy import deepcopy +from pathlib import Path +from typing import Any + +import pytest + +import solstone.think.services.spp_attest.nvgpu.appraise as appraise_module +from solstone.think.services.spp_attest.nvgpu.binary import ( + build_nvattest_attest_command, +) +from solstone.think.services.spp_attest.nvgpu.errors import GpuAppraisalError +from solstone.think.services.spp_attest.nvgpu.evidence import to_nvattest_evidence +from solstone.think.services.spp_attest.tlv import decode_gpu_envelope + +FIXTURE_DIR = Path(__file__).resolve().parents[1] / "fixtures" / "spp_attest" +NVATTEST_DIR = FIXTURE_DIR / "nvattest" + + +def _envelope(): + return decode_gpu_envelope((FIXTURE_DIR / "gpu-envelope.tlv").read_bytes()) + + +def _owner_nonce() -> bytes: + return bytes.fromhex("".join((FIXTURE_DIR / "nonce.hex").read_text().split())) + + +def _stdout(name: str) -> str: + return (NVATTEST_DIR / f"{name}.stdout").read_text(encoding="utf-8") + + +def _stderr(name: str) -> str: + return (NVATTEST_DIR / f"{name}.stderr").read_text(encoding="utf-8") + + +def _positive_body() -> dict[str, Any]: + data = json.loads(_stdout("positive")) + assert isinstance(data, dict) + return data + + +def _fake_nvattest_dir(tmp_path: Path) -> Path: + root = tmp_path / "nvattest" + (root / "bin").mkdir(parents=True, exist_ok=True) + (root / "bin" / "nvattest").write_text("#!/bin/sh\n", encoding="utf-8") + (root / "lib").mkdir(exist_ok=True) + return root + + +def _run_appraisal_with_stdout( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + stdout: str, + *, + stderr: str = "", + returncode: int = 0, + observed: dict[str, Any] | None = None, + rim_store: str = "remote", + rim_dir: Path | None = None, +): + nvattest_dir = _fake_nvattest_dir(tmp_path) + + def fake_run(argv, **kwargs): + if observed is not None: + observed["argv"] = argv + observed["kwargs"] = kwargs + evidence_path = Path(argv[argv.index("--gpu-evidence-file") + 1]) + observed["evidence"] = json.loads(evidence_path.read_text(encoding="utf-8")) + return subprocess.CompletedProcess(argv, returncode, stdout, stderr) + + monkeypatch.setattr(appraise_module.subprocess, "run", fake_run) + return appraise_module.appraise_gpu_leg( + _envelope(), + _owner_nonce(), + nvattest_dir=nvattest_dir, + rim_store=rim_store, + rim_dir=rim_dir, + ) + + +def _body_stdout(body: dict[str, Any]) -> str: + return json.dumps(body, indent=2, sort_keys=True) + + +def _claim(body: dict[str, Any]) -> dict[str, Any]: + claims = body["claims"] + assert isinstance(claims, list) + claim = claims[0] + assert isinstance(claim, dict) + return claim + + +def _set_path(body: dict[str, Any], path: tuple[str, ...], value: Any) -> None: + node: dict[str, Any] = body + for key in path[:-1]: + node = node[key] + node[path[-1]] = value + + +def _delete_claim_key(body: dict[str, Any], key: str) -> None: + del _claim(body)[key] + + +def _decode_jwt_segment(segment: str) -> dict[str, Any]: + raw = base64.urlsafe_b64decode(segment + "=" * (-len(segment) % 4)) + data = json.loads(raw) + assert isinstance(data, dict) + return data + + +def _encode_jwt(header: dict[str, Any], payload: dict[str, Any]) -> str: + def encode(data: dict[str, Any]) -> str: + raw = json.dumps(data, separators=(",", ":"), sort_keys=True).encode("utf-8") + return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=") + + return f"{encode(header)}.{encode(payload)}." + + +def _mutate_overall_jwt(body: dict[str, Any], *, header=None, payload=None) -> None: + jwt = body["detached_eat"][0][1] + parts = jwt.split(".") + original_header = _decode_jwt_segment(parts[0]) + original_payload = _decode_jwt_segment(parts[1]) + if header: + original_header.update(header) + if payload: + original_payload.update(payload) + body["detached_eat"][0][1] = _encode_jwt(original_header, original_payload) + + +def test_to_nvattest_evidence_transforms_gpu_envelope() -> None: + envelope = _envelope() + evidence = to_nvattest_evidence(envelope, _owner_nonce()) + + assert isinstance(evidence, list) + assert len(evidence) == 1 + item = evidence[0] + assert item["arch"] == "HOPPER" + assert item["nonce"] == _owner_nonce().hex() + assert base64.b64decode(item["evidence"]) == envelope.field(2) + assert base64.b64decode(item["certificate"]) == envelope.field(3) + + +def test_appraise_gpu_leg_accepts_positive_capture( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + result = _run_appraisal_with_stdout( + monkeypatch, + tmp_path, + _stdout("positive"), + stderr=_stderr("positive"), + ) + + assert result.driver_version == "595.71.05" + assert result.vbios_version == "96.00.88.00.11" + assert result.hwmodel == "GH100 A01 GSP BROM" + assert result.envelope_gpu_uuid == "GPU-256cc88f-e93b-9396-b581-274543ea3235" + assert [step.name for step in result.steps] == [ + "nvattest", + "overall-eat", + "gpu-claims", + ] + + +@pytest.mark.parametrize( + ("name", "marker"), + [ + ("negA", "get_evidence"), + ("negB", "get_evidence"), + ("negC", "generate_gpu_evidence_claims"), + ], +) +def test_appraise_gpu_leg_classifies_negative_nonce_captures( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + name: str, + marker: str, +) -> None: + with pytest.raises(GpuAppraisalError) as exc_info: + _run_appraisal_with_stdout( + monkeypatch, + tmp_path, + _stdout(name), + stderr=_stderr(name), + ) + + assert exc_info.value.reason == "gpu_nonce_mismatch" + assert marker in exc_info.value.stderr + if name == "negC": + assert "nonce_from_ar" in exc_info.value.stderr + + +@pytest.mark.parametrize( + ("label", "mutate"), + [ + ( + "secboot false", + lambda body: _set_path(body, ("claims", 0, "secboot"), False), + ), + ( + "secboot string", + lambda body: _set_path(body, ("claims", 0, "secboot"), "true"), + ), + ("secboot int", lambda body: _set_path(body, ("claims", 0, "secboot"), 1)), + ( + "debug enabled", + lambda body: _set_path(body, ("claims", 0, "dbgstat"), "enabled"), + ), + ( + "driver rim signature", + lambda body: _set_path( + body, + ("claims", 0, "x-nvidia-gpu-driver-rim-signature-verified"), + False, + ), + ), + ( + "measres failure", + lambda body: _set_path(body, ("claims", 0, "measres"), "fail"), + ), + ( + "missing key", + lambda body: _delete_claim_key( + body, + "x-nvidia-gpu-attestation-report-parsed", + ), + ), + ( + "ocsp bad", + lambda body: _set_path( + body, + ( + "claims", + 0, + "x-nvidia-gpu-attestation-report-cert-chain", + "x-nvidia-cert-ocsp-status", + ), + "revoked", + ), + ), + ("claims dict", lambda body: _set_path(body, ("claims",), {})), + ("result code false", lambda body: _set_path(body, ("result_code",), False)), + ( + "claims version", + lambda body: _set_path( + body, + ("claims", 0, "x-nvidia-gpu-claims-version"), + "4.0", + ), + ), + ], +) +def test_appraise_gpu_leg_fail_closed_matrix( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + label: str, + mutate, +) -> None: + body = _positive_body() + mutate(body) + + with pytest.raises(GpuAppraisalError) as exc_info: + _run_appraisal_with_stdout(monkeypatch, tmp_path, _body_stdout(body)) + + assert label + assert exc_info.value.reason == "gpu_appraisal_failed" + + +@pytest.mark.parametrize("stdout", ["", "not json"]) +def test_appraise_gpu_leg_rejects_empty_or_garbage_stdout( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + stdout: str, +) -> None: + with pytest.raises(GpuAppraisalError) as exc_info: + _run_appraisal_with_stdout(monkeypatch, tmp_path, stdout) + + assert exc_info.value.reason == "gpu_appraisal_failed" + + +def test_appraise_gpu_leg_rejects_missing_nvattest_dir() -> None: + with pytest.raises(GpuAppraisalError) as exc_info: + appraise_module.appraise_gpu_leg( + _envelope(), + _owner_nonce(), + nvattest_dir=Path("/does/not/exist"), + ) + + assert exc_info.value.reason == "nvattest_unavailable" + + +@pytest.mark.parametrize( + "mutate", + [ + lambda body: _set_path(body, ("detached_eat",), {}), + lambda body: _set_path(body, ("detached_eat",), []), + lambda body: _mutate_overall_jwt(body, header={"alg": "HS256"}), + lambda body: _mutate_overall_jwt(body, payload={"iss": "OTHER"}), + ], +) +def test_overall_eat_is_veto_only( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + mutate, +) -> None: + body = _positive_body() + mutate(body) + + with pytest.raises(GpuAppraisalError): + _run_appraisal_with_stdout(monkeypatch, tmp_path, _body_stdout(body)) + + green_eat_flipped_claim = _positive_body() + _set_path( + green_eat_flipped_claim, + ("claims", 0, "x-nvidia-gpu-vbios-rim-version-match"), + False, + ) + with pytest.raises(GpuAppraisalError): + _run_appraisal_with_stdout( + monkeypatch, + tmp_path, + _body_stdout(green_eat_flipped_claim), + ) + + +def test_stderr_does_not_influence_acceptance( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + result = _run_appraisal_with_stdout( + monkeypatch, + tmp_path, + _stdout("positive"), + stderr=_stderr("negC"), + ) + + assert result.driver_version == "595.71.05" + + +def test_nonce_is_written_to_evidence_and_argv( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + observed: dict[str, Any] = {} + + _run_appraisal_with_stdout( + monkeypatch, + tmp_path, + _stdout("positive"), + observed=observed, + ) + + argv = observed["argv"] + evidence = observed["evidence"][0] + assert evidence["nonce"] == _owner_nonce().hex() + assert argv[argv.index("--nonce") + 1] == _owner_nonce().hex() + + +def test_nvattest_command_env_inherits_parent_and_sets_library_path( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + nvattest_dir = _fake_nvattest_dir(tmp_path) + monkeypatch.setenv("SPP_NVATTEST_PARENT_SENTINEL", "kept") + + command = build_nvattest_attest_command( + nvattest_dir=nvattest_dir, + evidence_file=tmp_path / "evidence.json", + owner_nonce=_owner_nonce(), + ) + + assert command.env["SPP_NVATTEST_PARENT_SENTINEL"] == "kept" + assert command.env["LD_LIBRARY_PATH"] == str(nvattest_dir / "lib") + + +def test_rim_store_dir_argv_shape( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + observed: dict[str, Any] = {} + rim_dir = tmp_path / "rim" + rim_dir.mkdir() + + _run_appraisal_with_stdout( + monkeypatch, + tmp_path, + _stdout("positive"), + observed=observed, + rim_store="dir", + rim_dir=rim_dir, + ) + + argv = observed["argv"] + assert argv[argv.index("--rim-store") + 1] == "dir" + assert argv[argv.index("--rim-dir") + 1] == str(rim_dir) + + with pytest.raises(ValueError, match="rim_dir is required"): + _run_appraisal_with_stdout( + monkeypatch, + tmp_path, + _stdout("positive"), + rim_store="dir", + ) + with pytest.raises(ValueError, match="only valid"): + _run_appraisal_with_stdout( + monkeypatch, + tmp_path, + _stdout("positive"), + rim_store="remote", + rim_dir=rim_dir, + ) + + +def test_driver_and_vbios_rim_fetched_are_not_predicate_inputs( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + body = deepcopy(_positive_body()) + _set_path(body, ("claims", 0, "x-nvidia-gpu-driver-rim-fetched"), False) + _set_path(body, ("claims", 0, "x-nvidia-gpu-vbios-rim-fetched"), False) + + result = _run_appraisal_with_stdout(monkeypatch, tmp_path, _body_stdout(body)) + + assert result.driver_version == "595.71.05" diff --git a/tests/services/test_spp_attest_purity.py b/tests/services/test_spp_attest_purity.py index a828a8c94..21ace85d6 100644 --- a/tests/services/test_spp_attest_purity.py +++ b/tests/services/test_spp_attest_purity.py @@ -4,8 +4,15 @@ from __future__ import annotations import ast +import subprocess from pathlib import Path +import pytest + +from solstone.think.services.spp_attest.nvgpu.appraise import appraise_gpu_leg +from solstone.think.services.spp_attest.nvgpu.errors import GpuAppraisalError +from solstone.think.services.spp_attest.tlv import decode_gpu_envelope + PACKAGE_DIR = ( Path(__file__).resolve().parents[2] / "solstone" @@ -13,7 +20,40 @@ PACKAGE_DIR = ( / "services" / "spp_attest" ) -BANNED_IMPORT_ROOTS = {"subprocess", "socket", "urllib", "requests", "shutil"} +FIXTURE_DIR = Path(__file__).resolve().parents[1] / "fixtures" / "spp_attest" +NVATTEST_FIXTURE_DIR = FIXTURE_DIR / "nvattest" +PURE_EXCLUDED = {"nvgpu/appraise.py"} +PURE_NON_VACUITY = { + "__init__.py", + "binding.py", + "errors.py", + "nvgpu/binary.py", + "nvgpu/claims.py", + "nvgpu/evidence.py", + "nvgpu/errors.py", + "nvgpu/__init__.py", + "snp.py", + "tlv.py", + "tpm_quote.py", +} +BANNED_IMPORT_ROOTS = { + "http", + "httpx", + "requests", + "shutil", + "socket", + "subprocess", + "tempfile", + "urllib", +} +APPRAISE_BANNED_IMPORT_ROOTS = { + "http", + "httpx", + "requests", + "shutil", + "socket", + "urllib", +} BANNED_WRITE_ATTRS = { "write_text", "write_bytes", @@ -29,9 +69,16 @@ BANNED_WRITE_NAMES = {"atomic_write", "atomic_replace"} WRITE_MODE_CHARS = frozenset({"w", "a", "x", "+"}) -def test_spp_attest_package_stays_pure_python_read_only() -> None: - files = sorted(PACKAGE_DIR.rglob("*.py")) +def test_spp_attest_package_stays_pure_python_read_only_except_nvgpu_appraise() -> None: + assert PURE_EXCLUDED == {"nvgpu/appraise.py"} + files = [ + path + for path in sorted(PACKAGE_DIR.rglob("*.py")) + if path.relative_to(PACKAGE_DIR).as_posix() not in PURE_EXCLUDED + ] assert files, f"no Python files found under {PACKAGE_DIR}" + scanned = {path.relative_to(PACKAGE_DIR).as_posix() for path in files} + assert PURE_NON_VACUITY <= scanned findings: list[str] = [] for path in files: @@ -42,6 +89,103 @@ def test_spp_attest_package_stays_pure_python_read_only() -> None: assert findings == [] +def test_nvgpu_appraise_impurity_is_narrow() -> None: + path = PACKAGE_DIR / "nvgpu" / "appraise.py" + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + imports = _import_roots(tree) + + assert "subprocess" in imports + assert "tempfile" in imports + + findings: list[str] = [] + for node in ast.walk(tree): + findings.extend( + _scan_node( + path, + node, + banned_import_roots=APPRAISE_BANNED_IMPORT_ROOTS, + banned_write_attrs=BANNED_WRITE_ATTRS - {"unlink"}, + ban_solstone_utils=True, + ) + ) + + assert findings == [] + + +def test_nvgpu_appraise_removes_temp_evidence_file_on_return_and_raise( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + nvattest_dir = tmp_path / "nvattest" + (nvattest_dir / "bin").mkdir(parents=True) + (nvattest_dir / "bin" / "nvattest").write_text("#!/bin/sh\n", encoding="utf-8") + (nvattest_dir / "lib").mkdir() + envelope = decode_gpu_envelope((FIXTURE_DIR / "gpu-envelope.tlv").read_bytes()) + owner_nonce = bytes.fromhex((FIXTURE_DIR / "nonce.hex").read_text().strip()) + observed: list[Path] = [] + + def positive_run(argv, **_kwargs): + evidence_path = Path(argv[argv.index("--gpu-evidence-file") + 1]) + assert evidence_path.is_file() + observed.append(evidence_path) + return subprocess.CompletedProcess( + argv, + 0, + (NVATTEST_FIXTURE_DIR / "positive.stdout").read_text(encoding="utf-8"), + (NVATTEST_FIXTURE_DIR / "positive.stderr").read_text(encoding="utf-8"), + ) + + monkeypatch.setattr( + "solstone.think.services.spp_attest.nvgpu.appraise.subprocess.run", + positive_run, + ) + appraise_gpu_leg(envelope, owner_nonce, nvattest_dir=nvattest_dir) + assert observed and not observed[-1].exists() + + def negative_run(argv, **_kwargs): + evidence_path = Path(argv[argv.index("--gpu-evidence-file") + 1]) + assert evidence_path.is_file() + observed.append(evidence_path) + return subprocess.CompletedProcess( + argv, + 0, + (NVATTEST_FIXTURE_DIR / "negC.stdout").read_text(encoding="utf-8"), + (NVATTEST_FIXTURE_DIR / "negC.stderr").read_text(encoding="utf-8"), + ) + + monkeypatch.setattr( + "solstone.think.services.spp_attest.nvgpu.appraise.subprocess.run", + negative_run, + ) + with pytest.raises(GpuAppraisalError): + appraise_gpu_leg(envelope, owner_nonce, nvattest_dir=nvattest_dir) + assert not observed[-1].exists() + + class FailingTempFile: + def __init__(self, path: Path) -> None: + self.path = path + self.name = str(path) + path.write_text("", encoding="utf-8") + + def __enter__(self): + return self + + def __exit__(self, *_exc_info) -> bool: + return False + + def write(self, _value: str) -> int: + raise OSError("disk full") + + failing_temp_file = FailingTempFile(tmp_path / "write-failure.json") + monkeypatch.setattr( + "solstone.think.services.spp_attest.nvgpu.appraise.tempfile.NamedTemporaryFile", + lambda *_args, **_kwargs: failing_temp_file, + ) + with pytest.raises(OSError, match="disk full"): + appraise_gpu_leg(envelope, owner_nonce, nvattest_dir=nvattest_dir) + assert not failing_temp_file.path.exists() + + def test_purity_scanner_bans_aliased_shutil_import() -> None: tree = ast.parse("import shutil as sh\nsh.which('tpm2_checkquote')\n") findings = [ @@ -53,30 +197,55 @@ def test_purity_scanner_bans_aliased_shutil_import() -> None: assert findings == ["snippet.py:1: banned import shutil"] -def _scan_node(path: Path, node: ast.AST) -> list[str]: +def _scan_node( + path: Path, + node: ast.AST, + *, + banned_import_roots: set[str] = BANNED_IMPORT_ROOTS, + banned_write_attrs: set[str] = BANNED_WRITE_ATTRS, + ban_solstone_utils: bool = False, +) -> list[str]: if isinstance(node, ast.Import): - return _scan_import(path, node) + return _scan_import(path, node, banned_import_roots) if isinstance(node, ast.ImportFrom): - return _scan_import_from(path, node) + return _scan_import_from( + path, + node, + banned_import_roots, + ban_solstone_utils, + ) if isinstance(node, ast.Call): - return _scan_call(path, node) + return _scan_call(path, node, banned_write_attrs) return [] -def _scan_import(path: Path, node: ast.Import) -> list[str]: +def _scan_import( + path: Path, + node: ast.Import, + banned_import_roots: set[str], +) -> list[str]: findings: list[str] = [] for alias in node.names: root = alias.name.split(".", maxsplit=1)[0] - if root in BANNED_IMPORT_ROOTS: + if root in banned_import_roots: findings.append(f"{path}:{node.lineno}: banned import {alias.name}") return findings -def _scan_import_from(path: Path, node: ast.ImportFrom) -> list[str]: +def _scan_import_from( + path: Path, + node: ast.ImportFrom, + banned_import_roots: set[str], + ban_solstone_utils: bool, +) -> list[str]: findings: list[str] = [] module = node.module or "" root = module.split(".", maxsplit=1)[0] - if root in BANNED_IMPORT_ROOTS: + if root in banned_import_roots: + findings.append(f"{path}:{node.lineno}: banned import from {module}") + if ban_solstone_utils and ( + module == "solstone.think.utils" or module.startswith("solstone.think.utils.") + ): findings.append(f"{path}:{node.lineno}: banned import from {module}") for alias in node.names: if alias.name in BANNED_WRITE_NAMES: @@ -84,14 +253,18 @@ def _scan_import_from(path: Path, node: ast.ImportFrom) -> list[str]: return findings -def _scan_call(path: Path, node: ast.Call) -> list[str]: +def _scan_call( + path: Path, + node: ast.Call, + banned_write_attrs: set[str], +) -> list[str]: func = node.func if isinstance(func, ast.Attribute): if _is_shutil_which(func): return [f"{path}:{node.lineno}: banned shutil.which call"] if _is_json_dump(func): return [f"{path}:{node.lineno}: banned json.dump call"] - if func.attr in BANNED_WRITE_ATTRS: + if func.attr in banned_write_attrs: return [f"{path}:{node.lineno}: banned write API {func.attr}"] if isinstance(func, ast.Name): if func.id in BANNED_WRITE_NAMES: @@ -117,6 +290,16 @@ def _is_json_dump(func: ast.Attribute) -> bool: ) +def _import_roots(tree: ast.AST) -> set[str]: + roots: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.Import): + roots.update(alias.name.split(".", maxsplit=1)[0] for alias in node.names) + elif isinstance(node, ast.ImportFrom) and node.module: + roots.add(node.module.split(".", maxsplit=1)[0]) + return roots + + def _open_uses_write_mode(node: ast.Call) -> bool: mode_node = None if len(node.args) >= 2: