From dd04f55c8e1bf65f42908ef5c79e927bcad3666a Mon Sep 17 00:00:00 2001 From: Jer Miller Date: Fri, 24 Jul 2026 08:49:37 -0600 Subject: [PATCH] feat(spp): enforce production PCR pin at confidential transport chokepoint Add the production PCR pin registry and wire production_policy() into _establish_channel_locked, the single production establish chokepoint for confidential transport. Preserve the new pcr_pin_mismatch reason across SNP appraisal, composite verification, RA-TLS verification, transport recorded state, and brain_cli mapping via PcrPinMismatchError caught before the generic CPU-leg handler. Promote check_pcr_fingerprint as the shared digest-and-compare helper, remove _check_pcr_policy, and apply the same check to phase-2 exporter proof after quote signature verification. Policy() remains record-mode by default; no wire contract, fixture, loopback script, or engine behavior changes. --- solstone/think/brain_cli.py | 1 + .../think/services/spp_attest/composite.py | 10 +- solstone/think/services/spp_attest/errors.py | 4 + solstone/think/services/spp_attest/pins.py | 23 +++ .../services/spp_attest/ratls/channel.py | 1 + .../think/services/spp_attest/ratls/verify.py | 19 +- solstone/think/services/spp_attest/snp.py | 16 +- solstone/think/services/spp_transport.py | 2 + tests/services/test_spp_attest_composite.py | 44 +++++ tests/services/test_spp_attest_pins.py | 31 ++++ tests/services/test_spp_attest_purity.py | 1 + .../services/test_spp_attest_ratls_channel.py | 166 ++++++++++++++++++ .../services/test_spp_attest_ratls_verify.py | 48 ++++- tests/services/test_spp_attest_snp.py | 13 ++ tests/services/test_spp_transport.py | 147 +++++++++++++++- tests/test_brain_cli.py | 1 + 16 files changed, 516 insertions(+), 11 deletions(-) create mode 100644 solstone/think/services/spp_attest/pins.py create mode 100644 tests/services/test_spp_attest_pins.py create mode 100644 tests/services/test_spp_attest_ratls_channel.py diff --git a/solstone/think/brain_cli.py b/solstone/think/brain_cli.py index 6ace31cc6..90b0dc801 100644 --- a/solstone/think/brain_cli.py +++ b/solstone/think/brain_cli.py @@ -88,6 +88,7 @@ _SPP_ATTESTATION_FAILURE_REASON_TO_BRAIN_REASON = { "certificate_extension_invalid": "attestation_rejected", "certificate_evidence_invalid": "attestation_rejected", "nonce_mismatch": "attestation_rejected", + "pcr_pin_mismatch": "attestation_rejected", "spki_mismatch": "attestation_rejected", "cpu_verification_failed": "attestation_rejected", "gpu_nonce_mismatch": "attestation_rejected", diff --git a/solstone/think/services/spp_attest/composite.py b/solstone/think/services/spp_attest/composite.py index b992dbe6b..ce4ea10ce 100644 --- a/solstone/think/services/spp_attest/composite.py +++ b/solstone/think/services/spp_attest/composite.py @@ -14,7 +14,10 @@ from typing import NoReturn from solstone.think.models import AttestationFailedError from solstone.think.services.spp_attest.binding import BINDING_DOMAIN -from solstone.think.services.spp_attest.errors import VerificationError +from solstone.think.services.spp_attest.errors import ( + PcrPinMismatchError, + 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 @@ -78,6 +81,11 @@ def verify_composite( policy=policy, quote_verifier=quote_verifier, ) + except PcrPinMismatchError: + _raise_attestation_failed( + "the CPU leg rejected the evidence (pcr_pin_mismatch)", + "confidential attestation CPU PCR pin mismatch", + ) except VerificationError: _raise_attestation_failed( "the CPU leg rejected the evidence (cpu_verification_failed)", diff --git a/solstone/think/services/spp_attest/errors.py b/solstone/think/services/spp_attest/errors.py index 7c9bb2192..2c07f595c 100644 --- a/solstone/think/services/spp_attest/errors.py +++ b/solstone/think/services/spp_attest/errors.py @@ -6,3 +6,7 @@ from __future__ import annotations class VerificationError(RuntimeError): """Raised when SPP attestation evidence fails appraisal.""" + + +class PcrPinMismatchError(VerificationError): + """Raised when a TPM PCR fingerprint is outside the pinned policy.""" diff --git a/solstone/think/services/spp_attest/pins.py b/solstone/think/services/spp_attest/pins.py new file mode 100644 index 000000000..43198a796 --- /dev/null +++ b/solstone/think/services/spp_attest/pins.py @@ -0,0 +1,23 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""Production PCR pin policy for SPP composite attestation.""" + +from __future__ import annotations + +from solstone.think.services.spp_attest.snp import Policy + +# substrate: spp-engine-01 (processing.solstone.app:9443, Azure Standard_NCC40ads_H100_v5) +# pcr_sha256 pin: b162f46105c80d3e45028e37cc649404c9d65297ad1cda8f953208582060b0e3 +# Provenance: captured live from the production substrate, 2026-07-24, +# operator decision record; observed identical across two fresh RA-TLS sessions +# via the journal-side CPU-leg appraisal. +PRODUCTION_PCR_SHA256_PINS: frozenset[str] = frozenset( + { + "b162f46105c80d3e45028e37cc649404c9d65297ad1cda8f953208582060b0e3", + } +) + + +def production_policy() -> Policy: + return Policy(pcr_mode="pin", pcr_pins=set(PRODUCTION_PCR_SHA256_PINS)) diff --git a/solstone/think/services/spp_attest/ratls/channel.py b/solstone/think/services/spp_attest/ratls/channel.py index 8ee5da159..ead3b05e0 100644 --- a/solstone/think/services/spp_attest/ratls/channel.py +++ b/solstone/think/services/spp_attest/ratls/channel.py @@ -185,6 +185,7 @@ def establish_attested_channel( evidence=verified.evidence, tls_exporter=tls_exporter, owner_nonce=owner_nonce, + policy=policy, ) raw.settimeout(None) return AttestedChannel( diff --git a/solstone/think/services/spp_attest/ratls/verify.py b/solstone/think/services/spp_attest/ratls/verify.py index a6dc2be31..6a96bf7d6 100644 --- a/solstone/think/services/spp_attest/ratls/verify.py +++ b/solstone/think/services/spp_attest/ratls/verify.py @@ -20,7 +20,10 @@ 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.errors import ( + PcrPinMismatchError, + VerificationError, +) from solstone.think.services.spp_attest.ratls.contract import ( CERTIFICATE_BINDING_DOMAIN, COMPOSITE_EVIDENCE_OID, @@ -28,7 +31,11 @@ from solstone.think.services.spp_attest.ratls.contract import ( ExporterProof, exporter_binding, ) -from solstone.think.services.spp_attest.snp import CpuBundle, Policy +from solstone.think.services.spp_attest.snp import ( + CpuBundle, + Policy, + check_pcr_fingerprint, +) from solstone.think.services.spp_attest.tpm_quote import verify_quote @@ -120,6 +127,8 @@ def verify_certificate_evidence( reason = getattr(exc, "detail", "") if "nonce_mismatch" in reason: code = "nonce_mismatch" + elif "pcr_pin_mismatch" in reason: + code = "pcr_pin_mismatch" elif "cpu_verification_failed" in reason: code = "cpu_verification_failed" elif "gpu_nonce_mismatch" in reason: @@ -147,6 +156,7 @@ def verify_exporter_proof( evidence: CompositeEvidence, tls_exporter: bytes, owner_nonce: bytes, + policy: Policy | None = None, ) -> None: try: proof = ExporterProof.from_der(proof_der) @@ -174,3 +184,8 @@ def verify_exporter_proof( ) except VerificationError: raise RatlsVerificationError("exporter_quote_failed") + + try: + check_pcr_fingerprint(proof.quote_pcrs, policy or Policy()) + except PcrPinMismatchError: + raise RatlsVerificationError("pcr_pin_mismatch") diff --git a/solstone/think/services/spp_attest/snp.py b/solstone/think/services/spp_attest/snp.py index 29f6bbb7f..ef4a1ad30 100644 --- a/solstone/think/services/spp_attest/snp.py +++ b/solstone/think/services/spp_attest/snp.py @@ -27,7 +27,10 @@ from solstone.think.services.spp_attest.binding import ( check_envelope_nonce, composite_binding_hash, ) -from solstone.think.services.spp_attest.errors import VerificationError +from solstone.think.services.spp_attest.errors import ( + PcrPinMismatchError, + VerificationError, +) from solstone.think.services.spp_attest.tlv import decode_gpu_envelope from solstone.think.services.spp_attest.tpm_quote import verify_quote @@ -349,8 +352,7 @@ def appraise_cpu_leg( ) ) - pcr_sha256 = hashlib.sha256(bundle.quote_pcrs).hexdigest() - _check_pcr_policy(pcr_sha256, policy) + pcr_sha256 = check_pcr_fingerprint(bundle.quote_pcrs, policy) if policy.pcr_mode == "record": steps.append(_ok("pcr-policy", f"record-then-pin v1 fingerprint={pcr_sha256}")) else: @@ -638,14 +640,16 @@ def _verify_ak_binding(runtime: dict[str, Any], ak_public_key_pem: bytes) -> Non ) -def _check_pcr_policy(pcr_sha256: str, policy: Policy) -> None: +def check_pcr_fingerprint(quote_pcrs: bytes, policy: Policy) -> str: + pcr_sha256 = hashlib.sha256(quote_pcrs).hexdigest() if policy.pcr_mode == "record": - return + return pcr_sha256 if policy.pcr_mode != "pin": raise VerificationError(f"unknown PCR policy mode {policy.pcr_mode!r}") pins = {pin.lower() for pin in policy.pcr_pins} if pcr_sha256.lower() not in pins: - raise VerificationError(f"PCR fingerprint {pcr_sha256} not in pinned policy") + raise PcrPinMismatchError(f"PCR fingerprint {pcr_sha256} not in pinned policy") + return pcr_sha256 def _is_ca(cert: x509.Certificate) -> bool: diff --git a/solstone/think/services/spp_transport.py b/solstone/think/services/spp_transport.py index 5afa4af5c..b9de5e90c 100644 --- a/solstone/think/services/spp_transport.py +++ b/solstone/think/services/spp_transport.py @@ -27,6 +27,7 @@ from solstone.think.providers.nvattest_install import ( from solstone.think.services import spp from solstone.think.services.spp_attest.cadence import AttestationSession from solstone.think.services.spp_attest.composite import verify_composite +from solstone.think.services.spp_attest.pins import production_policy from solstone.think.services.spp_attest.ratls.channel import ( AttestedChannel, RatlsChannelError, @@ -221,6 +222,7 @@ def _establish_channel_locked( if nvattest_dir is not None else resolve_nvattest_dir(block.get("nvattest_dir")), now=now, + policy=production_policy(), composite_verifier=verify_composite, monotonic_now=time.monotonic, epoch=_EPOCH, diff --git a/tests/services/test_spp_attest_composite.py b/tests/services/test_spp_attest_composite.py index be023cb64..3ec59c71e 100644 --- a/tests/services/test_spp_attest_composite.py +++ b/tests/services/test_spp_attest_composite.py @@ -3,6 +3,7 @@ from __future__ import annotations +import json import logging import shutil import subprocess @@ -19,6 +20,7 @@ from solstone.think.services.spp_attest.nvgpu.errors import GpuAppraisalError from solstone.think.services.spp_attest.snp import ( AppraisalStep, CpuBundle, + Policy, load_cpu_bundle, ) from solstone.think.services.spp_attest.tlv import GpuEnvelope @@ -41,6 +43,11 @@ def _channel_binding() -> bytes: return (FIXTURE_DIR / "guest_x25519.pub.der").read_bytes() +def _fixture_pcr_sha256() -> str: + data = json.loads((FIXTURE_DIR / "cpu-appraisal.json").read_text(encoding="utf-8")) + return data["pcr_sha256"] + + def _copy_bundle(tmp_path: Path) -> Path: bundle = tmp_path / "bundle" shutil.copytree(FIXTURE_DIR, bundle) @@ -277,6 +284,43 @@ def test_verify_composite_rejects_mutated_channel_binding_without_leak( _assert_owner_message_safe(exc_info.value) +def test_verify_composite_enforces_pcr_pin_policy_from_real_cpu_fixture_without_leak( + tmp_path: Path, +) -> None: + matching_pin = _fixture_pcr_sha256() + positive = verify_composite( + _cpu_bundle(), + envelope_tlv=_envelope_tlv(), + channel_binding=_channel_binding(), + owner_nonce=_owner_nonce(), + now=NOW, + nvattest_dir=tmp_path / "unused", + policy=Policy(pcr_mode="pin", pcr_pins={matching_pin}), + gpu_appraiser=_safe_gpu_appraiser, + ) + assert positive.verified is True + + wrong_pin = "00" * 32 + with pytest.raises(AttestationFailedError) as exc_info: + verify_composite( + _cpu_bundle(), + envelope_tlv=_envelope_tlv(), + channel_binding=_channel_binding(), + owner_nonce=_owner_nonce(), + now=NOW, + nvattest_dir=tmp_path / "unused", + policy=Policy(pcr_mode="pin", pcr_pins={wrong_pin}), + gpu_appraiser=_safe_gpu_appraiser, + ) + + assert exc_info.value.detail == ( + "the CPU leg rejected the evidence (pcr_pin_mismatch)" + ) + assert matching_pin not in exc_info.value.detail + assert wrong_pin not in exc_info.value.detail + _assert_owner_message_safe(exc_info.value) + + def test_verify_composite_maps_gpu_nonce_mismatch_without_leak( tmp_path: Path, caplog: pytest.LogCaptureFixture, diff --git a/tests/services/test_spp_attest_pins.py b/tests/services/test_spp_attest_pins.py new file mode 100644 index 000000000..0b3cc511f --- /dev/null +++ b/tests/services/test_spp_attest_pins.py @@ -0,0 +1,31 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +from __future__ import annotations + +from solstone.think.services.spp_attest.pins import ( + PRODUCTION_PCR_SHA256_PINS, + production_policy, +) + + +def test_production_pcr_pin_registry_literals() -> None: + assert PRODUCTION_PCR_SHA256_PINS == frozenset( + {"b162f46105c80d3e45028e37cc649404c9d65297ad1cda8f953208582060b0e3"} + ) + + first = production_policy() + second = production_policy() + + assert first.pcr_mode == "pin" + assert first.pcr_pins == set(PRODUCTION_PCR_SHA256_PINS) + assert second.pcr_pins == set(PRODUCTION_PCR_SHA256_PINS) + assert first.pcr_pins is not second.pcr_pins + assert first.pcr_pins is not PRODUCTION_PCR_SHA256_PINS + assert second.pcr_pins is not PRODUCTION_PCR_SHA256_PINS + + first.pcr_pins.add("00" * 32) + assert first.pcr_pins != second.pcr_pins + assert PRODUCTION_PCR_SHA256_PINS == frozenset( + {"b162f46105c80d3e45028e37cc649404c9d65297ad1cda8f953208582060b0e3"} + ) diff --git a/tests/services/test_spp_attest_purity.py b/tests/services/test_spp_attest_purity.py index 28595bd9c..758107371 100644 --- a/tests/services/test_spp_attest_purity.py +++ b/tests/services/test_spp_attest_purity.py @@ -34,6 +34,7 @@ PURE_NON_VACUITY = { "nvgpu/evidence.py", "nvgpu/errors.py", "nvgpu/__init__.py", + "pins.py", "ratls/__init__.py", "ratls/contract.py", "ratls/verify.py", diff --git a/tests/services/test_spp_attest_ratls_channel.py b/tests/services/test_spp_attest_ratls_channel.py new file mode 100644 index 000000000..e96a797ab --- /dev/null +++ b/tests/services/test_spp_attest_ratls_channel.py @@ -0,0 +1,166 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +from __future__ import annotations + +from datetime import datetime, timezone +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from solstone.think.services.spp_attest.ratls import channel +from solstone.think.services.spp_attest.ratls.contract import ( + EXPORTER_BYTES, + EXPORTER_PROOF_MEDIA_TYPE, + EXPORTER_PROOF_PATH, + PREFACE_MAGIC, +) +from solstone.think.services.spp_attest.snp import Policy + + +def test_establish_attested_channel_passes_policy_to_exporter_proof( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + owner_nonce = b"n" * 32 + proof_der = b"proof-der" + tls_exporter = b"e" * EXPORTER_BYTES + tls_spki_der = b"tls-spki" + evidence = object() + verdict = object() + policy = Policy(pcr_mode="pin", pcr_pins={"abc"}) + captured_certificate: dict[str, object] = {} + captured_exporter: dict[str, object] = {} + connections: list[object] = [] + + class FakeRawSocket: + def __init__(self) -> None: + self.sent: list[bytes] = [] + self.timeout = 30.0 + self.closed = False + + def sendall(self, data: bytes) -> None: + self.sent.append(data) + + def settimeout(self, timeout: float | None) -> None: + self.timeout = timeout + + def close(self) -> None: + self.closed = True + + class FakeCertificate: + def public_bytes(self, _encoding) -> bytes: + return b"certificate-der" + + class FakePeer: + def to_cryptography(self) -> FakeCertificate: + return FakeCertificate() + + class FakeConnection: + def __init__(self, context, raw) -> None: + self.context = context + self.raw = raw + self.sent: list[bytes] = [] + self.recv_chunks = [ + ( + b"HTTP/1.1 200 OK\r\n" + + f"Content-Type: {EXPORTER_PROOF_MEDIA_TYPE}\r\n".encode("ascii") + + f"Content-Length: {len(proof_der)}\r\n\r\n".encode("ascii") + + proof_der + ) + ] + self.closed = False + connections.append(self) + + def setblocking(self, _flag: int) -> None: + pass + + def set_connect_state(self) -> None: + pass + + def set_tlsext_host_name(self, _name: bytes) -> None: + pass + + def do_handshake(self) -> None: + pass + + def get_peer_certificate(self) -> FakePeer: + return FakePeer() + + def export_keying_material( + self, + _label: bytes, + _size: int, + _context: bytes, + ) -> bytes: + return tls_exporter + + def sendall(self, data: bytes) -> None: + self.sent.append(data) + + def recv(self, _size: int) -> bytes: + return self.recv_chunks.pop(0) if self.recv_chunks else b"" + + def close(self) -> None: + self.closed = True + + raw = FakeRawSocket() + monkeypatch.setattr( + channel.socket, + "create_connection", + lambda _address, timeout: raw, + ) + monkeypatch.setattr(channel, "_tls_context", lambda: object()) + monkeypatch.setattr(channel.SSL, "Connection", FakeConnection) + + def fake_verify_certificate_evidence(**kwargs): + captured_certificate.update(kwargs) + return SimpleNamespace( + tls_spki_der=tls_spki_der, + evidence=evidence, + verdict=verdict, + ) + + def fake_verify_exporter_proof(**kwargs): + captured_exporter.update(kwargs) + + monkeypatch.setattr( + channel, + "verify_certificate_evidence", + fake_verify_certificate_evidence, + ) + monkeypatch.setattr(channel, "verify_exporter_proof", fake_verify_exporter_proof) + + established = channel.establish_attested_channel( + channel.RatlsEndpoint("spp.example.test", 9443), + owner_nonce=owner_nonce, + nvattest_dir=tmp_path, + now=datetime(2026, 7, 24, tzinfo=timezone.utc), + policy=policy, + composite_verifier=lambda *_args, **_kwargs: verdict, + monotonic_now=lambda: 123.0, + epoch=7, + ) + + assert raw.sent == [PREFACE_MAGIC + owner_nonce] + assert raw.timeout is None + assert connections + assert connections[0].sent == [ + ( + f"GET {EXPORTER_PROOF_PATH} HTTP/1.1\r\n" + "Host: spp-engine\r\n" + "Content-Length: 0\r\n\r\n" + ).encode("ascii") + ] + assert captured_certificate["policy"] is policy + assert captured_exporter == { + "proof_der": proof_der, + "evidence": evidence, + "tls_exporter": tls_exporter, + "owner_nonce": owner_nonce, + "policy": policy, + } + assert established.verdict is verdict + assert established.epoch == 7 + assert established.last_used_monotonic == 123.0 diff --git a/tests/services/test_spp_attest_ratls_verify.py b/tests/services/test_spp_attest_ratls_verify.py index 1532187cc..198d12e3d 100644 --- a/tests/services/test_spp_attest_ratls_verify.py +++ b/tests/services/test_spp_attest_ratls_verify.py @@ -3,6 +3,7 @@ from __future__ import annotations +import hashlib from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any @@ -30,7 +31,7 @@ from solstone.think.services.spp_attest.ratls.verify import ( verify_certificate_evidence, verify_exporter_proof, ) -from solstone.think.services.spp_attest.snp import AppraisalStep, CpuAppraisal +from solstone.think.services.spp_attest.snp import AppraisalStep, CpuAppraisal, Policy NOW = datetime(2026, 7, 12, tzinfo=timezone.utc) @@ -227,6 +228,10 @@ def test_verify_certificate_evidence_rejects_spki_mismatch(tmp_path: Path) -> No "the CPU leg rejected evidence (cpu_verification_failed)", "cpu_verification_failed", ), + ( + "the CPU leg rejected the evidence (pcr_pin_mismatch)", + "pcr_pin_mismatch", + ), ( "the GPU leg rejected evidence (nvattest_integrity_failed)", "nvattest_integrity_failed", @@ -315,6 +320,47 @@ def test_verify_exporter_proof_rejects_exporter_mismatch() -> None: assert exc_info.value.reason_code == "exporter_mismatch" +def test_verify_exporter_proof_rejects_pcr_pin_mismatch_after_valid_quote( + monkeypatch, +) -> None: + nonce = b"n" * 32 + _key, spki = _key_and_spki() + tls_exporter = b"e" * 32 + evidence = _evidence(nonce, spki) + proof = ExporterProof( + nonce, + spki, + tls_exporter, + b"p2-message", + b"p2-signature", + b"p2-pcrs", + ) + seen: dict[str, Any] = {} + + def verify_quote(**kwargs): + seen.update(kwargs) + + monkeypatch.setattr(ratls_verify, "verify_quote", verify_quote) + policy = Policy( + pcr_mode="pin", + pcr_pins={hashlib.sha256(evidence.quote_pcrs).hexdigest()}, + ) + + with pytest.raises(RatlsVerificationError) as exc_info: + verify_exporter_proof( + proof_der=proof.to_der(), + evidence=evidence, + tls_exporter=tls_exporter, + owner_nonce=nonce, + policy=policy, + ) + + assert evidence.quote_pcrs == b"p1-pcrs" + assert proof.quote_pcrs == b"p2-pcrs" + assert seen["quote_pcrs"] == proof.quote_pcrs + assert exc_info.value.reason_code == "pcr_pin_mismatch" + + def test_verify_exporter_proof_rejects_quote_under_wrong_ak(monkeypatch) -> None: nonce = b"n" * 32 _key, spki = _key_and_spki() diff --git a/tests/services/test_spp_attest_snp.py b/tests/services/test_spp_attest_snp.py index 05862d7a6..6815609f7 100644 --- a/tests/services/test_spp_attest_snp.py +++ b/tests/services/test_spp_attest_snp.py @@ -4,6 +4,7 @@ from __future__ import annotations import datetime as dt +import hashlib import json import shutil from pathlib import Path @@ -20,6 +21,7 @@ from solstone.think.services.spp_attest import ( load_cpu_bundle, ) from solstone.think.services.spp_attest import snp as snp_module +from solstone.think.services.spp_attest.snp import check_pcr_fingerprint FIXTURE_DIR = Path(__file__).resolve().parents[1] / "fixtures" / "spp_attest" PCR_SHA256_HEX = "b162f46105c80d3e45028e37cc649404c9d65297ad1cda8f953208582060b0e3" @@ -265,6 +267,17 @@ def test_appraise_cpu_leg_rejects_non_matching_pcr_pin() -> None: _appraise(policy=policy) +def test_check_pcr_fingerprint_accepts_rotation_overlap() -> None: + old_pcrs = b"old-pcrs" + new_pcrs = b"new-pcrs" + old_pin = hashlib.sha256(old_pcrs).hexdigest() + new_pin = hashlib.sha256(new_pcrs).hexdigest() + policy = Policy(pcr_mode="pin", pcr_pins={old_pin, new_pin}) + + assert check_pcr_fingerprint(old_pcrs, policy) == old_pin + assert check_pcr_fingerprint(new_pcrs, policy) == new_pin + + def test_appraise_cpu_leg_rejects_tlv_splice_before_appraisal_steps() -> None: envelope_tlv = _mutate_tlv_field_one_nonce( (FIXTURE_DIR / "gpu-envelope.tlv").read_bytes() diff --git a/tests/services/test_spp_transport.py b/tests/services/test_spp_transport.py index 8f122b83f..462dd0b1e 100644 --- a/tests/services/test_spp_transport.py +++ b/tests/services/test_spp_transport.py @@ -12,14 +12,29 @@ from types import SimpleNamespace from unittest.mock import Mock 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.journal_io.locking import hold_lock from solstone.think.models import AttestationFailedError from solstone.think.providers import nvattest_install from solstone.think.services import spp, spp_transport from solstone.think.services.spp_attest.cadence import AttestationSession +from solstone.think.services.spp_attest.composite import verify_composite from solstone.think.services.spp_attest.ratls.channel import RatlsChannelError -from solstone.think.services.spp_attest.ratls.verify import RatlsVerificationError +from solstone.think.services.spp_attest.ratls.contract import ( + COMPOSITE_EVIDENCE_OID, + CompositeEvidence, +) +from solstone.think.services.spp_attest.ratls.verify import ( + RatlsVerificationError, + verify_certificate_evidence, +) +from solstone.think.services.spp_attest.snp import Policy + +SPP_FIXTURE_DIR = Path(__file__).resolve().parents[1] / "fixtures" / "spp_attest" class _FakeChannel: @@ -115,6 +130,64 @@ def _patch_listener(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(spp_transport, "_start_listener_locked", fake_start_listener) +def _fixture_owner_nonce() -> bytes: + return bytes.fromhex("".join((SPP_FIXTURE_DIR / "nonce.hex").read_text().split())) + + +def _fixture_composite_evidence(owner_nonce: bytes, spki: bytes) -> CompositeEvidence: + certs_dir = SPP_FIXTURE_DIR / "certs" + return CompositeEvidence( + owner_nonce=owner_nonce, + tls_spki_der=spki, + amd_report=(SPP_FIXTURE_DIR / "report.bin").read_bytes(), + hcl_report=(SPP_FIXTURE_DIR / "hcl_report.bin").read_bytes(), + ak_public_key_pem=(SPP_FIXTURE_DIR / "akpub.pem").read_bytes(), + quote_message=(SPP_FIXTURE_DIR / "quote.msg").read_bytes(), + quote_signature=(SPP_FIXTURE_DIR / "quote.sig").read_bytes(), + quote_pcrs=(SPP_FIXTURE_DIR / "quote.pcrs").read_bytes(), + amd_ark_pem=(certs_dir / "ark.pem").read_bytes(), + amd_ask_pem=(certs_dir / "ask.pem").read_bytes(), + amd_vcek_pem=(certs_dir / "vcek.pem").read_bytes(), + gpu_envelope=(SPP_FIXTURE_DIR / "gpu-envelope.tlv").read_bytes(), + ) + + +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 _certificate_der( + key: ec.EllipticCurvePrivateKey, + evidence: CompositeEvidence, +) -> 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(datetime.now(timezone.utc) - timedelta(days=1)) + .not_valid_after(datetime.now(timezone.utc) + timedelta(days=1)) + .add_extension( + x509.UnrecognizedExtension( + ObjectIdentifier(COMPOSITE_EVIDENCE_OID), + evidence.to_der(), + ), + critical=True, + ) + .sign(key, hashes.SHA256()) + ) + return cert.public_bytes(serialization.Encoding.DER) + + def _stale_session(verdict: object) -> AttestationSession: old = datetime.now(timezone.utc) - timedelta(hours=2) return AttestationSession( @@ -125,6 +198,34 @@ def _stale_session(verdict: object) -> AttestationSession: ) +def test_establish_channel_locked_passes_production_pcr_policy( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + block = _write_confidential_config(tmp_path, monkeypatch) + captured: dict[str, object] = {} + + def fake_establish(_endpoint, **kwargs): + captured.update(kwargs) + return _FakeChannel(object(), epoch=kwargs["epoch"]) + + monkeypatch.setattr(spp_transport, "establish_attested_channel", fake_establish) + + spp_transport._establish_channel_locked( + block, + datetime.now(timezone.utc), + nvattest_dir=tmp_path / "nvattest", + ) + + policy = captured["policy"] + assert isinstance(policy, Policy) + assert policy.pcr_mode == "pin" + assert ( + "b162f46105c80d3e45028e37cc649404c9d65297ad1cda8f953208582060b0e3" + in policy.pcr_pins + ) + + def test_verify_confidential_attestation_reuses_then_rotates_stale_session_inline( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, @@ -489,6 +590,7 @@ RATLS_VERIFICATION_REASON_CODES = ( "certificate_extension_invalid", "certificate_evidence_invalid", "nonce_mismatch", + "pcr_pin_mismatch", "spki_mismatch", "cpu_verification_failed", "gpu_nonce_mismatch", @@ -556,6 +658,49 @@ def test_attestation_failure_buckets_real_reason_codes_at_transport_catch_site( assert failure.reason_code == reason_code +def test_pcr_pin_mismatch_reason_records_from_real_composite_origin( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + block = _write_confidential_config(tmp_path, monkeypatch) + observed: dict[str, str] = {} + monkeypatch.setattr( + spp_transport.secrets, + "token_bytes", + lambda _size: _fixture_owner_nonce(), + ) + + def fake_establish(_endpoint, **kwargs): + key, spki = _key_and_spki() + evidence = _fixture_composite_evidence(kwargs["owner_nonce"], spki) + bad_policy = Policy(pcr_mode="pin", pcr_pins={"00" * 32}) + try: + verify_certificate_evidence( + certificate_der=_certificate_der(key, evidence), + owner_nonce=kwargs["owner_nonce"], + now=kwargs["now"], + nvattest_dir=kwargs["nvattest_dir"], + policy=bad_policy, + quote_verifier=lambda **_kwargs: None, + composite_verifier=verify_composite, + ) + except RatlsVerificationError as exc: + observed["reason_code"] = exc.reason_code + raise + raise AssertionError("fixture unexpectedly passed a non-matching PCR pin") + + monkeypatch.setattr(spp_transport, "establish_attested_channel", fake_establish) + + with pytest.raises(AttestationFailedError): + spp_transport.verify_confidential_attestation(block) + + assert observed["reason_code"] == "pcr_pin_mismatch" + failure = spp.get_attestation_state().failure + assert failure is not None + assert failure.kind == "failed" + assert failure.reason_code == "pcr_pin_mismatch" + + @pytest.mark.parametrize( ("status", "kind", "reason_code"), [ diff --git a/tests/test_brain_cli.py b/tests/test_brain_cli.py index 77e53dedb..2e17ead64 100644 --- a/tests/test_brain_cli.py +++ b/tests/test_brain_cli.py @@ -424,6 +424,7 @@ def _write_unhealthy_record(journal: Path) -> None: ("nvattest_integrity_failed", "nvattest_integrity_failed", "failed"), ("gateway_unreachable", "attestation_not_verified", "blocked"), ("attestation_failed", "attestation_rejected", "failed"), + ("pcr_pin_mismatch", "attestation_rejected", "failed"), ], ) def test_spp_prerequisite_maps_failure_reason_code( -- 2.51.2