diff --git a/solstone/think/services/spp_attest/ratls/__init__.py b/solstone/think/services/spp_attest/ratls/__init__.py new file mode 100644 index 000000000..1ba2ea7bd --- /dev/null +++ b/solstone/think/services/spp_attest/ratls/__init__.py @@ -0,0 +1,46 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +from __future__ import annotations + +from solstone.think.services.spp_attest.ratls.contract import ( + CERTIFICATE_BINDING_DOMAIN, + COMPOSITE_EVIDENCE_OID, + EXPORTER_BYTES, + EXPORTER_LABEL, + EXPORTER_PROOF_MEDIA_TYPE, + EXPORTER_PROOF_PATH, + OWNER_NONCE_BYTES, + PREFACE_MAGIC, + CompositeEvidence, + ExporterProof, + certificate_binding, + exporter_binding, + exporter_context, +) +from solstone.think.services.spp_attest.ratls.verify import ( + RatlsVerificationError, + VerifiedCertificateEvidence, + verify_certificate_evidence, + verify_exporter_proof, +) + +__all__ = [ + "CERTIFICATE_BINDING_DOMAIN", + "COMPOSITE_EVIDENCE_OID", + "EXPORTER_BYTES", + "EXPORTER_LABEL", + "EXPORTER_PROOF_MEDIA_TYPE", + "EXPORTER_PROOF_PATH", + "OWNER_NONCE_BYTES", + "PREFACE_MAGIC", + "CompositeEvidence", + "ExporterProof", + "RatlsVerificationError", + "VerifiedCertificateEvidence", + "certificate_binding", + "exporter_binding", + "exporter_context", + "verify_certificate_evidence", + "verify_exporter_proof", +] diff --git a/solstone/think/services/spp_attest/ratls/channel.py b/solstone/think/services/spp_attest/ratls/channel.py new file mode 100644 index 000000000..f0ffdebba --- /dev/null +++ b/solstone/think/services/spp_attest/ratls/channel.py @@ -0,0 +1,220 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""Socket-backed RA-TLS channel establishment for SPP confidential transport.""" + +from __future__ import annotations + +import socket +from collections.abc import Callable +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path + +from cryptography.hazmat.primitives import serialization +from OpenSSL import SSL + +from solstone.think.services.spp_attest.composite import CompositeVerdict +from solstone.think.services.spp_attest.ratls.contract import ( + EXPORTER_BYTES, + EXPORTER_LABEL, + EXPORTER_PROOF_MEDIA_TYPE, + EXPORTER_PROOF_PATH, + PREFACE_MAGIC, + exporter_context, +) +from solstone.think.services.spp_attest.ratls.verify import ( + RatlsVerificationError, + VerifiedCertificateEvidence, + verify_certificate_evidence, + verify_exporter_proof, +) +from solstone.think.services.spp_attest.snp import Policy + +MAX_PROOF_RESPONSE_HEADERS = 16 * 1024 +MAX_PROOF_RESPONSE_BYTES = 8 * 1024 * 1024 + + +class RatlsChannelError(RuntimeError): + def __init__(self, reason_code: str) -> None: + self.reason_code = reason_code + super().__init__(f"confidential attestation rejected ({reason_code})") + + +@dataclass(frozen=True, slots=True) +class RatlsEndpoint: + host: str + port: int + server_name: bytes = b"spp-engine" + + +@dataclass(slots=True) +class AttestedChannel: + raw_socket: socket.socket + tls: SSL.Connection + verified: VerifiedCertificateEvidence + last_used_monotonic: float + + @property + def verdict(self) -> CompositeVerdict: + return self.verified.verdict + + def close(self) -> None: + try: + self.tls.shutdown() + except Exception: + pass + try: + self.tls.close() + except Exception: + pass + try: + self.raw_socket.close() + except Exception: + pass + + +def _tls_context() -> SSL.Context: + context = SSL.Context(SSL.TLS_CLIENT_METHOD) + context.set_min_proto_version(SSL.TLS1_3_VERSION) + context.set_max_proto_version(SSL.TLS1_3_VERSION) + context.set_verify(SSL.VERIFY_NONE, lambda *_args: True) + return context + + +def _recv_proof_response(connection: SSL.Connection) -> bytes: + data = bytearray() + marker = b"\r\n\r\n" + while marker not in data: + if len(data) >= MAX_PROOF_RESPONSE_HEADERS: + raise RatlsChannelError("proof_http_failed") + chunk = connection.recv(min(4096, MAX_PROOF_RESPONSE_HEADERS - len(data))) + if not chunk: + raise RatlsChannelError("proof_http_failed") + data.extend(chunk) + + head, body = bytes(data).split(marker, 1) + lines = head.split(b"\r\n") + if not lines or lines[0] != b"HTTP/1.1 200 OK": + raise RatlsChannelError("proof_http_failed") + content_length: int | None = None + for line in lines[1:]: + name, separator, value = line.partition(b":") + if not separator: + raise RatlsChannelError("proof_http_failed") + lowered = name.strip().lower() + if ( + lowered == b"content-type" + and value.strip() != EXPORTER_PROOF_MEDIA_TYPE.encode("ascii") + ): + raise RatlsChannelError("proof_http_failed") + if lowered == b"content-length": + try: + content_length = int(value.strip()) + except ValueError as exc: + raise RatlsChannelError("proof_http_failed") from exc + if content_length is None or content_length > MAX_PROOF_RESPONSE_BYTES: + raise RatlsChannelError("proof_http_failed") + while len(body) < content_length: + chunk = connection.recv(min(65536, content_length - len(body))) + if not chunk: + raise RatlsChannelError("proof_http_failed") + body += chunk + return body[:content_length] + + +def establish_attested_channel( + endpoint: RatlsEndpoint, + *, + owner_nonce: bytes, + nvattest_dir: Path, + now: datetime, + roots_dir: Path | None = None, + policy: Policy | None = None, + quote_verifier: Callable[..., None] | None = None, + composite_verifier: Callable[..., CompositeVerdict], + socket_timeout_s: float = 30.0, + monotonic_now: Callable[[], float], +) -> AttestedChannel: + raw: socket.socket | None = None + connection: SSL.Connection | None = None + try: + raw = socket.create_connection( + (endpoint.host, endpoint.port), timeout=socket_timeout_s + ) + raw.sendall(PREFACE_MAGIC + owner_nonce) + connection = SSL.Connection(_tls_context(), raw) + connection.setblocking(1) + connection.set_connect_state() + connection.set_tlsext_host_name(endpoint.server_name) + connection.do_handshake() + + peer = connection.get_peer_certificate() + if peer is None: + raise RatlsChannelError("tls_handshake_failed") + certificate_der = peer.to_cryptography().public_bytes( + serialization.Encoding.DER + ) + verified = verify_certificate_evidence( + certificate_der=certificate_der, + owner_nonce=owner_nonce, + now=now, + nvattest_dir=nvattest_dir, + roots_dir=roots_dir, + policy=policy, + quote_verifier=quote_verifier, + composite_verifier=composite_verifier, + ) + + tls_exporter = connection.export_keying_material( + EXPORTER_LABEL, + EXPORTER_BYTES, + exporter_context(owner_nonce, verified.tls_spki_der), + ) + request = ( + f"GET {EXPORTER_PROOF_PATH} HTTP/1.1\r\n" + "Host: spp-engine\r\n" + "Content-Length: 0\r\n\r\n" + ).encode("ascii") + connection.sendall(request) + proof_der = _recv_proof_response(connection) + verify_exporter_proof( + proof_der=proof_der, + evidence=verified.evidence, + tls_exporter=tls_exporter, + owner_nonce=owner_nonce, + ) + raw.settimeout(None) + return AttestedChannel( + raw_socket=raw, + tls=connection, + verified=verified, + last_used_monotonic=monotonic_now(), + ) + except RatlsVerificationError: + if connection is not None: + try: + connection.close() + except Exception: + pass + elif raw is not None: + raw.close() + raise + except RatlsChannelError: + if connection is not None: + try: + connection.close() + except Exception: + pass + elif raw is not None: + raw.close() + raise + except (OSError, SSL.Error) as exc: + if connection is not None: + try: + connection.close() + except Exception: + pass + elif raw is not None: + raw.close() + raise RatlsChannelError("gateway_unreachable") from exc diff --git a/solstone/think/services/spp_attest/ratls/contract.py b/solstone/think/services/spp_attest/ratls/contract.py new file mode 100644 index 000000000..c44e86a57 --- /dev/null +++ b/solstone/think/services/spp_attest/ratls/contract.py @@ -0,0 +1,257 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc +"""SPP RA-TLS v1 wire-contract constants and minimal DER codec. + +This module is the code source of truth for the engine-side contract. The +checked-in ``ratls-contract.json`` artifact is generated from these constants +with ``python3 ratls_contract.py generate``. Consumers must read that artifact +or port this codec; they must not re-type identifiers from prose. +""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass +from typing import Final + +PREFACE_MAGIC: Final = b"SPPRAT1\x00" +OWNER_NONCE_BYTES: Final = 32 +EXPORTER_BYTES: Final = 32 +# UUIDv5 dd12d0b5-d2b9-567d-8182-5e68403c1712 split into uint32 arcs. +# cryptography's OID parser caps each arc at uint32, so the standard single +# 128-bit 2.25 UUID integer cannot be represented by our shared parser. +COMPOSITE_EVIDENCE_OID: Final = "2.25.3708997813.3535365757.2172800616.1077671698" +EXPORTER_LABEL: Final = b"EXPERIMENTAL-sol-spp-engine-attestation-v1" +EXPORTER_CONTEXT_DOMAIN: Final = b"sol-spp-ratls-exporter-context-v1" +CERTIFICATE_BINDING_DOMAIN: Final = b"sol-spp-ratls-certificate-bind-v1" +EXPORTER_BINDING_DOMAIN: Final = b"sol-spp-ratls-exporter-bind-v1" +EXPORTER_PROOF_PATH: Final = "/._sol/spp/exporter-proof" +COMPOSITE_MEDIA_TYPE: Final = "application/vnd.sol.spp-composite-evidence-v1+der" +EXPORTER_PROOF_MEDIA_TYPE: Final = "application/vnd.sol.spp-exporter-proof-v1+der" +PROTOCOL_VERSION: Final = 1 + + +COMPOSITE_FIELDS: Final = ( + "version", + "owner_nonce", + "tls_spki_der", + "amd_report", + "hcl_report", + "ak_public_key_pem", + "quote_message", + "quote_signature", + "quote_pcrs", + "amd_ark_pem", + "amd_ask_pem", + "amd_vcek_pem", + "gpu_envelope", +) +EXPORTER_PROOF_FIELDS: Final = ( + "version", + "owner_nonce", + "tls_spki_der", + "tls_exporter", + "quote_message", + "quote_signature", + "quote_pcrs", +) + + +def certificate_binding( + owner_nonce: bytes, spki_der: bytes, gpu_envelope: bytes +) -> bytes: + return hashlib.sha256( + CERTIFICATE_BINDING_DOMAIN + + owner_nonce + + hashlib.sha256(spki_der).digest() + + hashlib.sha256(gpu_envelope).digest() + ).digest() + + +def exporter_context(owner_nonce: bytes, spki_der: bytes) -> bytes: + return hashlib.sha256( + EXPORTER_CONTEXT_DOMAIN + owner_nonce + hashlib.sha256(spki_der).digest() + ).digest() + + +def exporter_binding( + owner_nonce: bytes, spki_der: bytes, tls_exporter: bytes, gpu_envelope: bytes +) -> bytes: + return hashlib.sha256( + EXPORTER_BINDING_DOMAIN + + owner_nonce + + hashlib.sha256(spki_der).digest() + + tls_exporter + + hashlib.sha256(gpu_envelope).digest() + ).digest() + + +def _der_length(length: int) -> bytes: + if length < 0: + raise ValueError("DER length must be non-negative") + if length < 128: + return bytes([length]) + encoded = length.to_bytes((length.bit_length() + 7) // 8, "big") + return bytes([0x80 | len(encoded)]) + encoded + + +def _der_tlv(tag: int, value: bytes) -> bytes: + return bytes([tag]) + _der_length(len(value)) + value + + +def _der_integer(value: int) -> bytes: + if value < 0: + raise ValueError("only non-negative DER integers are supported") + encoded = value.to_bytes(max(1, (value.bit_length() + 7) // 8), "big") + if encoded[0] & 0x80: + encoded = b"\x00" + encoded + return _der_tlv(0x02, encoded) + + +def _der_octets(value: bytes) -> bytes: + return _der_tlv(0x04, value) + + +def _der_sequence(parts: list[bytes]) -> bytes: + return _der_tlv(0x30, b"".join(parts)) + + +def encode_sequence(version: int, fields: list[bytes]) -> bytes: + """Encode a fixed-order DER sequence containing one integer and octets.""" + + return _der_sequence( + [_der_integer(version), *(_der_octets(item) for item in fields)] + ) + + +def _read_length(data: bytes, offset: int) -> tuple[int, int]: + if offset >= len(data): + raise ValueError("truncated DER length") + first = data[offset] + offset += 1 + if first < 128: + return first, offset + count = first & 0x7F + if count == 0 or count > 4 or offset + count > len(data): + raise ValueError("invalid DER length") + encoded = data[offset : offset + count] + if encoded[0] == 0: + raise ValueError("non-minimal DER length") + length = int.from_bytes(encoded, "big") + if length < 128: + raise ValueError("non-minimal DER length") + return length, offset + count + + +def _read_tlv(data: bytes, offset: int, expected_tag: int) -> tuple[bytes, int]: + if offset >= len(data) or data[offset] != expected_tag: + raise ValueError(f"expected DER tag 0x{expected_tag:02x}") + length, value_offset = _read_length(data, offset + 1) + end = value_offset + length + if end > len(data): + raise ValueError("truncated DER value") + return data[value_offset:end], end + + +def decode_sequence(data: bytes, octet_count: int) -> tuple[int, list[bytes]]: + """Strictly decode the fixed sequence used by both SPP v1 envelopes.""" + + body, end = _read_tlv(data, 0, 0x30) + if end != len(data): + raise ValueError("trailing bytes after DER sequence") + integer, offset = _read_tlv(body, 0, 0x02) + if not integer or (len(integer) > 1 and integer[0] == 0 and integer[1] < 0x80): + raise ValueError("non-minimal DER integer") + version = int.from_bytes(integer, "big") + fields: list[bytes] = [] + for _ in range(octet_count): + value, offset = _read_tlv(body, offset, 0x04) + fields.append(value) + if offset != len(body): + raise ValueError("unexpected field in DER sequence") + return version, fields + + +@dataclass(frozen=True) +class CompositeEvidence: + owner_nonce: bytes + tls_spki_der: bytes + amd_report: bytes + hcl_report: bytes + ak_public_key_pem: bytes + quote_message: bytes + quote_signature: bytes + quote_pcrs: bytes + amd_ark_pem: bytes + amd_ask_pem: bytes + amd_vcek_pem: bytes + gpu_envelope: bytes + + def to_der(self) -> bytes: + return encode_sequence(PROTOCOL_VERSION, list(self.__dict__.values())) + + @classmethod + def from_der(cls, data: bytes) -> "CompositeEvidence": + version, fields = decode_sequence(data, len(COMPOSITE_FIELDS) - 1) + if version != PROTOCOL_VERSION: + raise ValueError(f"unsupported composite evidence version {version}") + return cls(*fields) + + +@dataclass(frozen=True) +class ExporterProof: + owner_nonce: bytes + tls_spki_der: bytes + tls_exporter: bytes + quote_message: bytes + quote_signature: bytes + quote_pcrs: bytes + + def to_der(self) -> bytes: + return encode_sequence(PROTOCOL_VERSION, list(self.__dict__.values())) + + @classmethod + def from_der(cls, data: bytes) -> "ExporterProof": + version, fields = decode_sequence(data, len(EXPORTER_PROOF_FIELDS) - 1) + if version != PROTOCOL_VERSION: + raise ValueError(f"unsupported exporter proof version {version}") + return cls(*fields) + + +def contract_artifact() -> dict[str, object]: + return { + "protocol_version": PROTOCOL_VERSION, + "preface": { + "magic_ascii_nul": "SPPRAT1", + "owner_nonce_bytes": OWNER_NONCE_BYTES, + }, + "x509_extension": { + "oid": COMPOSITE_EVIDENCE_OID, + "critical": True, + "encoding": "DER", + "fields": list(COMPOSITE_FIELDS), + "media_type": COMPOSITE_MEDIA_TYPE, + }, + "exporter": { + "label": EXPORTER_LABEL.decode("ascii"), + "context_domain": EXPORTER_CONTEXT_DOMAIN.decode("ascii"), + "length": EXPORTER_BYTES, + "proof_path": EXPORTER_PROOF_PATH, + "proof_encoding": "DER", + "proof_fields": list(EXPORTER_PROOF_FIELDS), + "proof_media_type": EXPORTER_PROOF_MEDIA_TYPE, + }, + "binding": { + "certificate_domain": CERTIFICATE_BINDING_DOMAIN.decode("ascii"), + "certificate_formula": "SHA256(domain || nonce || SHA256(tls_spki_der) || SHA256(SPPGPU1_TLV))", + "exporter_domain": EXPORTER_BINDING_DOMAIN.decode("ascii"), + "exporter_formula": "SHA256(domain || nonce || SHA256(tls_spki_der) || tls_exporter || SHA256(SPPGPU1_TLV))", + "exporter_context_formula": "SHA256(context_domain || nonce || SHA256(tls_spki_der))", + }, + "ingress_gate": "No credential or inference bytes are admitted until the certificate evidence and exporter proof both verify.", + } + + +def render_contract_artifact() -> str: + return json.dumps(contract_artifact(), indent=2, sort_keys=True) + "\n" diff --git a/solstone/think/services/spp_attest/ratls/ratls-contract.json b/solstone/think/services/spp_attest/ratls/ratls-contract.json new file mode 100644 index 000000000..279a81bd8 --- /dev/null +++ b/solstone/think/services/spp_attest/ratls/ratls-contract.json @@ -0,0 +1,53 @@ +{ + "binding": { + "certificate_domain": "sol-spp-ratls-certificate-bind-v1", + "certificate_formula": "SHA256(domain || nonce || SHA256(tls_spki_der) || SHA256(SPPGPU1_TLV))", + "exporter_context_formula": "SHA256(context_domain || nonce || SHA256(tls_spki_der))", + "exporter_domain": "sol-spp-ratls-exporter-bind-v1", + "exporter_formula": "SHA256(domain || nonce || SHA256(tls_spki_der) || tls_exporter || SHA256(SPPGPU1_TLV))" + }, + "exporter": { + "context_domain": "sol-spp-ratls-exporter-context-v1", + "label": "EXPERIMENTAL-sol-spp-engine-attestation-v1", + "length": 32, + "proof_encoding": "DER", + "proof_fields": [ + "version", + "owner_nonce", + "tls_spki_der", + "tls_exporter", + "quote_message", + "quote_signature", + "quote_pcrs" + ], + "proof_media_type": "application/vnd.sol.spp-exporter-proof-v1+der", + "proof_path": "/._sol/spp/exporter-proof" + }, + "ingress_gate": "No credential or inference bytes are admitted until the certificate evidence and exporter proof both verify.", + "preface": { + "magic_ascii_nul": "SPPRAT1", + "owner_nonce_bytes": 32 + }, + "protocol_version": 1, + "x509_extension": { + "critical": true, + "encoding": "DER", + "fields": [ + "version", + "owner_nonce", + "tls_spki_der", + "amd_report", + "hcl_report", + "ak_public_key_pem", + "quote_message", + "quote_signature", + "quote_pcrs", + "amd_ark_pem", + "amd_ask_pem", + "amd_vcek_pem", + "gpu_envelope" + ], + "media_type": "application/vnd.sol.spp-composite-evidence-v1+der", + "oid": "2.25.3708997813.3535365757.2172800616.1077671698" + } +} diff --git a/solstone/think/services/spp_attest/ratls/verify.py b/solstone/think/services/spp_attest/ratls/verify.py new file mode 100644 index 000000000..9a0cc1ce2 --- /dev/null +++ b/solstone/think/services/spp_attest/ratls/verify.py @@ -0,0 +1,174 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""Pure RA-TLS evidence verification for SPP confidential transport.""" + +from __future__ import annotations + +import hashlib +from collections.abc import Callable +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path + +from cryptography import x509 +from cryptography.hazmat.primitives import serialization +from cryptography.x509.oid import ObjectIdentifier + +from solstone.think.models import AttestationFailedError +from solstone.think.services.spp_attest.composite import ( + CompositeVerdict, + verify_composite, +) +from solstone.think.services.spp_attest.errors import VerificationError +from solstone.think.services.spp_attest.ratls.contract import ( + CERTIFICATE_BINDING_DOMAIN, + COMPOSITE_EVIDENCE_OID, + CompositeEvidence, + ExporterProof, + exporter_binding, +) +from solstone.think.services.spp_attest.snp import CpuBundle, Policy +from solstone.think.services.spp_attest.tpm_quote import verify_quote + + +class RatlsVerificationError(RuntimeError): + def __init__(self, reason_code: str) -> None: + self.reason_code = reason_code + super().__init__(f"confidential attestation rejected ({reason_code})") + + +@dataclass(frozen=True, slots=True) +class VerifiedCertificateEvidence: + evidence: CompositeEvidence + verdict: CompositeVerdict + tls_spki_der: bytes + + +def _cpu_bundle_from_evidence(evidence: CompositeEvidence) -> CpuBundle: + return CpuBundle( + hcl_report=evidence.hcl_report, + standalone_report=evidence.amd_report, + cert_pems=( + evidence.amd_ark_pem, + evidence.amd_ask_pem, + evidence.amd_vcek_pem, + ), + ak_public_key_pem=evidence.ak_public_key_pem, + nonce=evidence.owner_nonce, + quote_message=evidence.quote_message, + quote_signature=evidence.quote_signature, + quote_pcrs=evidence.quote_pcrs, + ) + + +def verify_certificate_evidence( + *, + certificate_der: bytes, + owner_nonce: bytes, + now: datetime, + nvattest_dir: Path, + roots_dir: Path | None = None, + policy: Policy | None = None, + quote_verifier: Callable[..., None] | None = None, + composite_verifier: Callable[..., CompositeVerdict] = verify_composite, +) -> VerifiedCertificateEvidence: + try: + certificate = x509.load_der_x509_certificate(certificate_der) + except ValueError as exc: + raise RatlsVerificationError("certificate_invalid") from exc + + try: + extension = certificate.extensions.get_extension_for_oid( + ObjectIdentifier(COMPOSITE_EVIDENCE_OID) + ) + except x509.ExtensionNotFound as exc: + raise RatlsVerificationError("certificate_extension_missing") from exc + if not extension.critical: + raise RatlsVerificationError("certificate_extension_not_critical") + if not isinstance(extension.value, x509.UnrecognizedExtension): + raise RatlsVerificationError("certificate_extension_invalid") + + try: + evidence = CompositeEvidence.from_der(extension.value.value) + except ValueError as exc: + raise RatlsVerificationError("certificate_evidence_invalid") from exc + if evidence.owner_nonce != owner_nonce: + raise RatlsVerificationError("nonce_mismatch") + + tls_spki_der = certificate.public_key().public_bytes( + serialization.Encoding.DER, + serialization.PublicFormat.SubjectPublicKeyInfo, + ) + if evidence.tls_spki_der != tls_spki_der: + raise RatlsVerificationError("spki_mismatch") + + try: + verdict = composite_verifier( + _cpu_bundle_from_evidence(evidence), + envelope_tlv=evidence.gpu_envelope, + channel_binding=hashlib.sha256(tls_spki_der).digest(), + owner_nonce=owner_nonce, + now=now, + nvattest_dir=nvattest_dir, + binding_domain=CERTIFICATE_BINDING_DOMAIN, + roots_dir=roots_dir, + policy=policy, + quote_verifier=quote_verifier, + ) + except AttestationFailedError as exc: + reason = getattr(exc, "detail", "") + if "nonce_mismatch" in reason: + code = "nonce_mismatch" + elif "cpu_verification_failed" in reason: + code = "cpu_verification_failed" + elif "gpu_nonce_mismatch" in reason: + code = "gpu_nonce_mismatch" + elif "nvattest_unavailable" in reason: + code = "nvattest_unavailable" + elif "gpu_appraisal_failed" in reason: + code = "gpu_appraisal_failed" + else: + code = "composite_appraisal_failed" + raise RatlsVerificationError(code) from exc + + return VerifiedCertificateEvidence( + evidence=evidence, + verdict=verdict, + tls_spki_der=tls_spki_der, + ) + + +def verify_exporter_proof( + *, + proof_der: bytes, + evidence: CompositeEvidence, + tls_exporter: bytes, + owner_nonce: bytes, +) -> None: + try: + proof = ExporterProof.from_der(proof_der) + except ValueError as exc: + raise RatlsVerificationError("exporter_proof_invalid") from exc + if proof.owner_nonce != owner_nonce: + raise RatlsVerificationError("nonce_mismatch") + if proof.tls_spki_der != evidence.tls_spki_der: + raise RatlsVerificationError("spki_mismatch") + if proof.tls_exporter != tls_exporter: + raise RatlsVerificationError("exporter_mismatch") + + try: + verify_quote( + ak_pub_pem=evidence.ak_public_key_pem, + quote_msg=proof.quote_message, + quote_sig=proof.quote_signature, + quote_pcrs=proof.quote_pcrs, + expected_binding=exporter_binding( + owner_nonce, + evidence.tls_spki_der, + tls_exporter, + evidence.gpu_envelope, + ), + ) + except VerificationError as exc: + raise RatlsVerificationError("exporter_quote_failed") from exc diff --git a/tests/services/test_spp_attest_purity.py b/tests/services/test_spp_attest_purity.py index 07cbee41b..1cb315da6 100644 --- a/tests/services/test_spp_attest_purity.py +++ b/tests/services/test_spp_attest_purity.py @@ -22,7 +22,7 @@ PACKAGE_DIR = ( ) FIXTURE_DIR = Path(__file__).resolve().parents[1] / "fixtures" / "spp_attest" NVATTEST_FIXTURE_DIR = FIXTURE_DIR / "nvattest" -PURE_EXCLUDED = {"nvgpu/appraise.py"} +PURE_EXCLUDED = {"nvgpu/appraise.py", "ratls/channel.py"} PURE_NON_VACUITY = { "__init__.py", "binding.py", @@ -34,6 +34,9 @@ PURE_NON_VACUITY = { "nvgpu/evidence.py", "nvgpu/errors.py", "nvgpu/__init__.py", + "ratls/__init__.py", + "ratls/contract.py", + "ratls/verify.py", "snp.py", "tlv.py", "tpm_quote.py", @@ -84,7 +87,7 @@ WRITE_MODE_CHARS = frozenset({"w", "a", "x", "+"}) def test_spp_attest_package_stays_pure_python_read_only_except_nvgpu_appraise() -> None: - assert PURE_EXCLUDED == {"nvgpu/appraise.py"} + assert PURE_EXCLUDED == {"nvgpu/appraise.py", "ratls/channel.py"} files = [ path for path in sorted(PACKAGE_DIR.rglob("*.py")) @@ -129,6 +132,31 @@ def test_nvgpu_appraise_impurity_is_narrow() -> None: assert findings == [] +def test_ratls_channel_impurity_is_narrow() -> None: + path = PACKAGE_DIR / "ratls" / "channel.py" + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + imports = _import_roots(tree) + + assert "socket" in imports + assert not ( + {"httpx", "requests", "subprocess", "tempfile", "shutil", "urllib"} & imports + ) + + findings: list[str] = [] + os_names = _imported_module_names(tree, "os") + for node in ast.walk(tree): + findings.extend( + _scan_node( + path, + node, + banned_import_roots=BANNED_IMPORT_ROOTS - {"socket"}, + os_names=os_names, + ) + ) + + assert findings == [] + + def test_nvgpu_appraise_removes_temp_evidence_file_on_return_and_raise( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, diff --git a/tests/services/test_spp_attest_ratls_contract.py b/tests/services/test_spp_attest_ratls_contract.py new file mode 100644 index 000000000..813fa2889 --- /dev/null +++ b/tests/services/test_spp_attest_ratls_contract.py @@ -0,0 +1,62 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from solstone.think.services.spp_attest.ratls.contract import ( + COMPOSITE_FIELDS, + PROTOCOL_VERSION, + CompositeEvidence, + decode_sequence, + encode_sequence, + render_contract_artifact, +) + +CONTRACT_PATH = ( + Path(__file__).resolve().parents[2] + / "solstone" + / "think" + / "services" + / "spp_attest" + / "ratls" + / "ratls-contract.json" +) +COMPOSITE_OCTET_COUNT = len(COMPOSITE_FIELDS) - 1 + + +def test_ratls_contract_artifact_matches_generated_constants() -> None: + assert CONTRACT_PATH.read_text(encoding="utf-8") == render_contract_artifact() + + +def test_composite_evidence_der_round_trips() -> None: + fields = [f"field-{index}".encode() for index in range(COMPOSITE_OCTET_COUNT)] + evidence = CompositeEvidence(*fields) + + assert CompositeEvidence.from_der(evidence.to_der()) == evidence + + +@pytest.mark.parametrize( + "payload", + [ + encode_sequence(PROTOCOL_VERSION, [b"x"] * COMPOSITE_OCTET_COUNT) + b"\x00", + b"\x31\x00", + b"\x30\x81\x01\x05", + b"\x30\x04\x02\x02\x00\x01", + encode_sequence(PROTOCOL_VERSION, [b"x"]), + encode_sequence(2, [b"x"] * COMPOSITE_OCTET_COUNT), + ], +) +def test_composite_evidence_strict_der_rejects_invalid_encodings( + payload: bytes, +) -> None: + with pytest.raises(ValueError): + CompositeEvidence.from_der(payload) + + +def test_decode_sequence_rejects_wrong_field_count() -> None: + with pytest.raises(ValueError): + decode_sequence(encode_sequence(PROTOCOL_VERSION, [b"x"]), 2) diff --git a/tests/services/test_spp_attest_ratls_verify.py b/tests/services/test_spp_attest_ratls_verify.py new file mode 100644 index 000000000..ad06c539b --- /dev/null +++ b/tests/services/test_spp_attest_ratls_verify.py @@ -0,0 +1,323 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any + +import pytest +from cryptography import x509 +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import ec +from cryptography.x509.oid import NameOID, ObjectIdentifier + +from solstone.think.models import AttestationFailedError +from solstone.think.services.spp_attest.composite import CompositeVerdict +from solstone.think.services.spp_attest.errors import VerificationError +from solstone.think.services.spp_attest.nvgpu.claims import GpuAppraisal +from solstone.think.services.spp_attest.ratls import verify as ratls_verify +from solstone.think.services.spp_attest.ratls.contract import ( + CERTIFICATE_BINDING_DOMAIN, + COMPOSITE_EVIDENCE_OID, + CompositeEvidence, + ExporterProof, + exporter_binding, +) +from solstone.think.services.spp_attest.ratls.verify import ( + RatlsVerificationError, + verify_certificate_evidence, + verify_exporter_proof, +) +from solstone.think.services.spp_attest.snp import AppraisalStep, CpuAppraisal + +NOW = datetime(2026, 7, 12, tzinfo=timezone.utc) + + +def _cpu_appraisal() -> CpuAppraisal: + return CpuAppraisal( + steps=[AppraisalStep("cpu", "ok", "fixture")], + hcla_version=1, + report_version=3, + cpuid={"family": 25, "model": 1, "step": 1}, + tcb={}, + pcr_sha256="00", + host_data="11", + measurement="22", + chip_id="33", + ) + + +def _gpu_appraisal() -> GpuAppraisal: + return GpuAppraisal( + steps=[AppraisalStep("gpu", "ok", "fixture")], + driver_version="570.0", + vbios_version="1.0", + hwmodel="B200", + ueid="ueid", + oemid="oem", + eat_nonce="nonce", + claims_version="1", + arch="blackwell", + envelope_gpu_uuid="gpu-uuid", + ) + + +def _verdict(now: datetime = NOW) -> CompositeVerdict: + return CompositeVerdict( + verified=True, + legs=("cpu", "gpu"), + substrate="fixture", + checked_at=now, + cpu_provenance=_cpu_appraisal(), + gpu_provenance=_gpu_appraisal(), + ) + + +def _key_and_spki() -> tuple[ec.EllipticCurvePrivateKey, bytes]: + key = ec.generate_private_key(ec.SECP256R1()) + spki = key.public_key().public_bytes( + serialization.Encoding.DER, + serialization.PublicFormat.SubjectPublicKeyInfo, + ) + return key, spki + + +def _evidence( + owner_nonce: bytes, spki: bytes, *, ak: bytes = b"ak-pem" +) -> CompositeEvidence: + return CompositeEvidence( + owner_nonce=owner_nonce, + tls_spki_der=spki, + amd_report=b"amd-report", + hcl_report=b"hcl-report", + ak_public_key_pem=ak, + quote_message=b"p1-message", + quote_signature=b"p1-signature", + quote_pcrs=b"p1-pcrs", + amd_ark_pem=b"ark", + amd_ask_pem=b"ask", + amd_vcek_pem=b"vcek", + gpu_envelope=b"gpu-envelope", + ) + + +def _certificate_der( + key: ec.EllipticCurvePrivateKey, + evidence: CompositeEvidence, + *, + critical: bool = True, +) -> bytes: + subject = issuer = x509.Name( + [x509.NameAttribute(NameOID.COMMON_NAME, "spp-engine-test")] + ) + cert = ( + x509.CertificateBuilder() + .subject_name(subject) + .issuer_name(issuer) + .public_key(key.public_key()) + .serial_number(1001) + .not_valid_before(NOW - timedelta(days=1)) + .not_valid_after(NOW + timedelta(days=1)) + .add_extension( + x509.UnrecognizedExtension( + ObjectIdentifier(COMPOSITE_EVIDENCE_OID), + evidence.to_der(), + ), + critical=critical, + ) + .sign(key, hashes.SHA256()) + ) + return cert.public_bytes(serialization.Encoding.DER) + + +def test_verify_certificate_evidence_passes_bytes_bundle_to_composite_verifier( + tmp_path: Path, +) -> None: + nonce = b"n" * 32 + key, spki = _key_and_spki() + evidence = _evidence(nonce, spki) + seen: dict[str, Any] = {} + + def composite_verifier(bundle, **kwargs): + seen["bundle"] = bundle + seen["kwargs"] = kwargs + return _verdict(kwargs["now"]) + + verified = verify_certificate_evidence( + certificate_der=_certificate_der(key, evidence), + owner_nonce=nonce, + now=NOW, + nvattest_dir=tmp_path, + quote_verifier="quote-verifier", + composite_verifier=composite_verifier, + ) + + assert verified.evidence == evidence + assert verified.tls_spki_der == spki + assert seen["bundle"].hcl_report == evidence.hcl_report + assert seen["bundle"].standalone_report == evidence.amd_report + assert seen["bundle"].cert_pems == ( + evidence.amd_ark_pem, + evidence.amd_ask_pem, + evidence.amd_vcek_pem, + ) + assert seen["bundle"].ak_public_key_pem == evidence.ak_public_key_pem + assert seen["bundle"].nonce == nonce + assert seen["kwargs"]["binding_domain"] == CERTIFICATE_BINDING_DOMAIN + assert seen["kwargs"]["owner_nonce"] == nonce + assert seen["kwargs"]["quote_verifier"] == "quote-verifier" + + +def test_verify_certificate_evidence_rejects_noncritical_extension( + tmp_path: Path, +) -> None: + nonce = b"n" * 32 + key, spki = _key_and_spki() + evidence = _evidence(nonce, spki) + + with pytest.raises(RatlsVerificationError) as exc_info: + verify_certificate_evidence( + certificate_der=_certificate_der(key, evidence, critical=False), + owner_nonce=nonce, + now=NOW, + nvattest_dir=tmp_path, + ) + + assert exc_info.value.reason_code == "certificate_extension_not_critical" + + +def test_verify_certificate_evidence_rejects_nonce_mismatch(tmp_path: Path) -> None: + key, spki = _key_and_spki() + evidence = _evidence(b"n" * 32, spki) + + with pytest.raises(RatlsVerificationError) as exc_info: + verify_certificate_evidence( + certificate_der=_certificate_der(key, evidence), + owner_nonce=b"m" * 32, + now=NOW, + nvattest_dir=tmp_path, + ) + + assert exc_info.value.reason_code == "nonce_mismatch" + + +def test_verify_certificate_evidence_rejects_spki_mismatch(tmp_path: Path) -> None: + nonce = b"n" * 32 + cert_key, _cert_spki = _key_and_spki() + _foreign_key, foreign_spki = _key_and_spki() + evidence = _evidence(nonce, foreign_spki) + + with pytest.raises(RatlsVerificationError) as exc_info: + verify_certificate_evidence( + certificate_der=_certificate_der(cert_key, evidence), + owner_nonce=nonce, + now=NOW, + nvattest_dir=tmp_path, + ) + + assert exc_info.value.reason_code == "spki_mismatch" + + +def test_verify_certificate_evidence_maps_composite_failure(tmp_path: Path) -> None: + nonce = b"n" * 32 + key, spki = _key_and_spki() + evidence = _evidence(nonce, spki) + + def composite_verifier(_bundle, **_kwargs): + raise AttestationFailedError( + "the CPU leg rejected evidence (cpu_verification_failed)" + ) + + with pytest.raises(RatlsVerificationError) as exc_info: + verify_certificate_evidence( + certificate_der=_certificate_der(key, evidence), + owner_nonce=nonce, + now=NOW, + nvattest_dir=tmp_path, + composite_verifier=composite_verifier, + ) + + assert exc_info.value.reason_code == "cpu_verification_failed" + + +def test_verify_exporter_proof_binds_quote_to_exporter(monkeypatch) -> None: + nonce = b"n" * 32 + _key, spki = _key_and_spki() + tls_exporter = b"e" * 32 + evidence = _evidence(nonce, spki) + proof = ExporterProof( + owner_nonce=nonce, + tls_spki_der=spki, + tls_exporter=tls_exporter, + quote_message=b"p2-message", + quote_signature=b"p2-signature", + quote_pcrs=b"p2-pcrs", + ) + seen: dict[str, Any] = {} + + def verify_quote(**kwargs): + seen.update(kwargs) + + monkeypatch.setattr(ratls_verify, "verify_quote", verify_quote) + + verify_exporter_proof( + proof_der=proof.to_der(), + evidence=evidence, + tls_exporter=tls_exporter, + owner_nonce=nonce, + ) + + assert seen == { + "ak_pub_pem": evidence.ak_public_key_pem, + "quote_msg": proof.quote_message, + "quote_sig": proof.quote_signature, + "quote_pcrs": proof.quote_pcrs, + "expected_binding": exporter_binding( + nonce, + spki, + tls_exporter, + evidence.gpu_envelope, + ), + } + + +def test_verify_exporter_proof_rejects_exporter_mismatch() -> None: + nonce = b"n" * 32 + _key, spki = _key_and_spki() + evidence = _evidence(nonce, spki) + proof = ExporterProof(nonce, spki, b"foreign", b"msg", b"sig", b"pcrs") + + with pytest.raises(RatlsVerificationError) as exc_info: + verify_exporter_proof( + proof_der=proof.to_der(), + evidence=evidence, + tls_exporter=b"mine", + owner_nonce=nonce, + ) + + assert exc_info.value.reason_code == "exporter_mismatch" + + +def test_verify_exporter_proof_rejects_quote_under_wrong_ak(monkeypatch) -> None: + nonce = b"n" * 32 + _key, spki = _key_and_spki() + tls_exporter = b"e" * 32 + evidence = _evidence(nonce, spki, ak=b"ak-a") + proof = ExporterProof(nonce, spki, tls_exporter, b"msg", b"sig", b"pcrs") + + def verify_quote(**_kwargs): + raise VerificationError("bad quote signature") + + monkeypatch.setattr(ratls_verify, "verify_quote", verify_quote) + + with pytest.raises(RatlsVerificationError) as exc_info: + verify_exporter_proof( + proof_der=proof.to_der(), + evidence=evidence, + tls_exporter=tls_exporter, + owner_nonce=nonce, + ) + + assert exc_info.value.reason_code == "exporter_quote_failed"