solpbc.org / solstone
personal memory agent
0
Fork 0

Configure Feed

Select the types of activity you want to include in your feed.

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.

author
Jer Miller date (Jul 24, 2026, 8:49 AM -0600) commit dd04f55c parent b4e7411b
+516 -11
+1
solstone/think/brain_cli.py
··· 88 88 "certificate_extension_invalid": "attestation_rejected", 89 89 "certificate_evidence_invalid": "attestation_rejected", 90 90 "nonce_mismatch": "attestation_rejected", 91 + "pcr_pin_mismatch": "attestation_rejected", 91 92 "spki_mismatch": "attestation_rejected", 92 93 "cpu_verification_failed": "attestation_rejected", 93 94 "gpu_nonce_mismatch": "attestation_rejected",
+9 -1
solstone/think/services/spp_attest/composite.py
··· 14 14 15 15 from solstone.think.models import AttestationFailedError 16 16 from solstone.think.services.spp_attest.binding import BINDING_DOMAIN 17 - from solstone.think.services.spp_attest.errors import VerificationError 17 + from solstone.think.services.spp_attest.errors import ( 18 + PcrPinMismatchError, 19 + VerificationError, 20 + ) 18 21 from solstone.think.services.spp_attest.nvgpu.appraise import appraise_gpu_leg 19 22 from solstone.think.services.spp_attest.nvgpu.claims import GpuAppraisal 20 23 from solstone.think.services.spp_attest.nvgpu.errors import GpuAppraisalError ··· 77 80 roots_dir=roots_dir, 78 81 policy=policy, 79 82 quote_verifier=quote_verifier, 83 + ) 84 + except PcrPinMismatchError: 85 + _raise_attestation_failed( 86 + "the CPU leg rejected the evidence (pcr_pin_mismatch)", 87 + "confidential attestation CPU PCR pin mismatch", 80 88 ) 81 89 except VerificationError: 82 90 _raise_attestation_failed(
+4
solstone/think/services/spp_attest/errors.py
··· 6 6 7 7 class VerificationError(RuntimeError): 8 8 """Raised when SPP attestation evidence fails appraisal.""" 9 + 10 + 11 + class PcrPinMismatchError(VerificationError): 12 + """Raised when a TPM PCR fingerprint is outside the pinned policy."""
+23
solstone/think/services/spp_attest/pins.py
··· 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 + # Copyright (c) 2026 sol pbc 3 + 4 + """Production PCR pin policy for SPP composite attestation.""" 5 + 6 + from __future__ import annotations 7 + 8 + from solstone.think.services.spp_attest.snp import Policy 9 + 10 + # substrate: spp-engine-01 (processing.solstone.app:9443, Azure Standard_NCC40ads_H100_v5) 11 + # pcr_sha256 pin: b162f46105c80d3e45028e37cc649404c9d65297ad1cda8f953208582060b0e3 12 + # Provenance: captured live from the production substrate, 2026-07-24, 13 + # operator decision record; observed identical across two fresh RA-TLS sessions 14 + # via the journal-side CPU-leg appraisal. 15 + PRODUCTION_PCR_SHA256_PINS: frozenset[str] = frozenset( 16 + { 17 + "b162f46105c80d3e45028e37cc649404c9d65297ad1cda8f953208582060b0e3", 18 + } 19 + ) 20 + 21 + 22 + def production_policy() -> Policy: 23 + return Policy(pcr_mode="pin", pcr_pins=set(PRODUCTION_PCR_SHA256_PINS))
+1
solstone/think/services/spp_attest/ratls/channel.py
··· 185 185 evidence=verified.evidence, 186 186 tls_exporter=tls_exporter, 187 187 owner_nonce=owner_nonce, 188 + policy=policy, 188 189 ) 189 190 raw.settimeout(None) 190 191 return AttestedChannel(
+17 -2
solstone/think/services/spp_attest/ratls/verify.py
··· 20 20 CompositeVerdict, 21 21 verify_composite, 22 22 ) 23 - from solstone.think.services.spp_attest.errors import VerificationError 23 + from solstone.think.services.spp_attest.errors import ( 24 + PcrPinMismatchError, 25 + VerificationError, 26 + ) 24 27 from solstone.think.services.spp_attest.ratls.contract import ( 25 28 CERTIFICATE_BINDING_DOMAIN, 26 29 COMPOSITE_EVIDENCE_OID, ··· 28 31 ExporterProof, 29 32 exporter_binding, 30 33 ) 31 - from solstone.think.services.spp_attest.snp import CpuBundle, Policy 34 + from solstone.think.services.spp_attest.snp import ( 35 + CpuBundle, 36 + Policy, 37 + check_pcr_fingerprint, 38 + ) 32 39 from solstone.think.services.spp_attest.tpm_quote import verify_quote 33 40 34 41 ··· 120 127 reason = getattr(exc, "detail", "") 121 128 if "nonce_mismatch" in reason: 122 129 code = "nonce_mismatch" 130 + elif "pcr_pin_mismatch" in reason: 131 + code = "pcr_pin_mismatch" 123 132 elif "cpu_verification_failed" in reason: 124 133 code = "cpu_verification_failed" 125 134 elif "gpu_nonce_mismatch" in reason: ··· 147 156 evidence: CompositeEvidence, 148 157 tls_exporter: bytes, 149 158 owner_nonce: bytes, 159 + policy: Policy | None = None, 150 160 ) -> None: 151 161 try: 152 162 proof = ExporterProof.from_der(proof_der) ··· 174 184 ) 175 185 except VerificationError: 176 186 raise RatlsVerificationError("exporter_quote_failed") 187 + 188 + try: 189 + check_pcr_fingerprint(proof.quote_pcrs, policy or Policy()) 190 + except PcrPinMismatchError: 191 + raise RatlsVerificationError("pcr_pin_mismatch")
+10 -6
solstone/think/services/spp_attest/snp.py
··· 27 27 check_envelope_nonce, 28 28 composite_binding_hash, 29 29 ) 30 - from solstone.think.services.spp_attest.errors import VerificationError 30 + from solstone.think.services.spp_attest.errors import ( 31 + PcrPinMismatchError, 32 + VerificationError, 33 + ) 31 34 from solstone.think.services.spp_attest.tlv import decode_gpu_envelope 32 35 from solstone.think.services.spp_attest.tpm_quote import verify_quote 33 36 ··· 349 352 ) 350 353 ) 351 354 352 - pcr_sha256 = hashlib.sha256(bundle.quote_pcrs).hexdigest() 353 - _check_pcr_policy(pcr_sha256, policy) 355 + pcr_sha256 = check_pcr_fingerprint(bundle.quote_pcrs, policy) 354 356 if policy.pcr_mode == "record": 355 357 steps.append(_ok("pcr-policy", f"record-then-pin v1 fingerprint={pcr_sha256}")) 356 358 else: ··· 638 640 ) 639 641 640 642 641 - def _check_pcr_policy(pcr_sha256: str, policy: Policy) -> None: 643 + def check_pcr_fingerprint(quote_pcrs: bytes, policy: Policy) -> str: 644 + pcr_sha256 = hashlib.sha256(quote_pcrs).hexdigest() 642 645 if policy.pcr_mode == "record": 643 - return 646 + return pcr_sha256 644 647 if policy.pcr_mode != "pin": 645 648 raise VerificationError(f"unknown PCR policy mode {policy.pcr_mode!r}") 646 649 pins = {pin.lower() for pin in policy.pcr_pins} 647 650 if pcr_sha256.lower() not in pins: 648 - raise VerificationError(f"PCR fingerprint {pcr_sha256} not in pinned policy") 651 + raise PcrPinMismatchError(f"PCR fingerprint {pcr_sha256} not in pinned policy") 652 + return pcr_sha256 649 653 650 654 651 655 def _is_ca(cert: x509.Certificate) -> bool:
+2
solstone/think/services/spp_transport.py
··· 27 27 from solstone.think.services import spp 28 28 from solstone.think.services.spp_attest.cadence import AttestationSession 29 29 from solstone.think.services.spp_attest.composite import verify_composite 30 + from solstone.think.services.spp_attest.pins import production_policy 30 31 from solstone.think.services.spp_attest.ratls.channel import ( 31 32 AttestedChannel, 32 33 RatlsChannelError, ··· 221 222 if nvattest_dir is not None 222 223 else resolve_nvattest_dir(block.get("nvattest_dir")), 223 224 now=now, 225 + policy=production_policy(), 224 226 composite_verifier=verify_composite, 225 227 monotonic_now=time.monotonic, 226 228 epoch=_EPOCH,
+44
tests/services/test_spp_attest_composite.py
··· 3 3 4 4 from __future__ import annotations 5 5 6 + import json 6 7 import logging 7 8 import shutil 8 9 import subprocess ··· 19 20 from solstone.think.services.spp_attest.snp import ( 20 21 AppraisalStep, 21 22 CpuBundle, 23 + Policy, 22 24 load_cpu_bundle, 23 25 ) 24 26 from solstone.think.services.spp_attest.tlv import GpuEnvelope ··· 39 41 40 42 def _channel_binding() -> bytes: 41 43 return (FIXTURE_DIR / "guest_x25519.pub.der").read_bytes() 44 + 45 + 46 + def _fixture_pcr_sha256() -> str: 47 + data = json.loads((FIXTURE_DIR / "cpu-appraisal.json").read_text(encoding="utf-8")) 48 + return data["pcr_sha256"] 42 49 43 50 44 51 def _copy_bundle(tmp_path: Path) -> Path: ··· 274 281 assert exc_info.value.detail == ( 275 282 "the CPU leg rejected the evidence (cpu_verification_failed)" 276 283 ) 284 + _assert_owner_message_safe(exc_info.value) 285 + 286 + 287 + def test_verify_composite_enforces_pcr_pin_policy_from_real_cpu_fixture_without_leak( 288 + tmp_path: Path, 289 + ) -> None: 290 + matching_pin = _fixture_pcr_sha256() 291 + positive = verify_composite( 292 + _cpu_bundle(), 293 + envelope_tlv=_envelope_tlv(), 294 + channel_binding=_channel_binding(), 295 + owner_nonce=_owner_nonce(), 296 + now=NOW, 297 + nvattest_dir=tmp_path / "unused", 298 + policy=Policy(pcr_mode="pin", pcr_pins={matching_pin}), 299 + gpu_appraiser=_safe_gpu_appraiser, 300 + ) 301 + assert positive.verified is True 302 + 303 + wrong_pin = "00" * 32 304 + with pytest.raises(AttestationFailedError) as exc_info: 305 + verify_composite( 306 + _cpu_bundle(), 307 + envelope_tlv=_envelope_tlv(), 308 + channel_binding=_channel_binding(), 309 + owner_nonce=_owner_nonce(), 310 + now=NOW, 311 + nvattest_dir=tmp_path / "unused", 312 + policy=Policy(pcr_mode="pin", pcr_pins={wrong_pin}), 313 + gpu_appraiser=_safe_gpu_appraiser, 314 + ) 315 + 316 + assert exc_info.value.detail == ( 317 + "the CPU leg rejected the evidence (pcr_pin_mismatch)" 318 + ) 319 + assert matching_pin not in exc_info.value.detail 320 + assert wrong_pin not in exc_info.value.detail 277 321 _assert_owner_message_safe(exc_info.value) 278 322 279 323
+31
tests/services/test_spp_attest_pins.py
··· 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 + # Copyright (c) 2026 sol pbc 3 + 4 + from __future__ import annotations 5 + 6 + from solstone.think.services.spp_attest.pins import ( 7 + PRODUCTION_PCR_SHA256_PINS, 8 + production_policy, 9 + ) 10 + 11 + 12 + def test_production_pcr_pin_registry_literals() -> None: 13 + assert PRODUCTION_PCR_SHA256_PINS == frozenset( 14 + {"b162f46105c80d3e45028e37cc649404c9d65297ad1cda8f953208582060b0e3"} 15 + ) 16 + 17 + first = production_policy() 18 + second = production_policy() 19 + 20 + assert first.pcr_mode == "pin" 21 + assert first.pcr_pins == set(PRODUCTION_PCR_SHA256_PINS) 22 + assert second.pcr_pins == set(PRODUCTION_PCR_SHA256_PINS) 23 + assert first.pcr_pins is not second.pcr_pins 24 + assert first.pcr_pins is not PRODUCTION_PCR_SHA256_PINS 25 + assert second.pcr_pins is not PRODUCTION_PCR_SHA256_PINS 26 + 27 + first.pcr_pins.add("00" * 32) 28 + assert first.pcr_pins != second.pcr_pins 29 + assert PRODUCTION_PCR_SHA256_PINS == frozenset( 30 + {"b162f46105c80d3e45028e37cc649404c9d65297ad1cda8f953208582060b0e3"} 31 + )
+1
tests/services/test_spp_attest_purity.py
··· 34 34 "nvgpu/evidence.py", 35 35 "nvgpu/errors.py", 36 36 "nvgpu/__init__.py", 37 + "pins.py", 37 38 "ratls/__init__.py", 38 39 "ratls/contract.py", 39 40 "ratls/verify.py",
+166
tests/services/test_spp_attest_ratls_channel.py
··· 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 + # Copyright (c) 2026 sol pbc 3 + 4 + from __future__ import annotations 5 + 6 + from datetime import datetime, timezone 7 + from pathlib import Path 8 + from types import SimpleNamespace 9 + 10 + import pytest 11 + 12 + from solstone.think.services.spp_attest.ratls import channel 13 + from solstone.think.services.spp_attest.ratls.contract import ( 14 + EXPORTER_BYTES, 15 + EXPORTER_PROOF_MEDIA_TYPE, 16 + EXPORTER_PROOF_PATH, 17 + PREFACE_MAGIC, 18 + ) 19 + from solstone.think.services.spp_attest.snp import Policy 20 + 21 + 22 + def test_establish_attested_channel_passes_policy_to_exporter_proof( 23 + tmp_path: Path, 24 + monkeypatch: pytest.MonkeyPatch, 25 + ) -> None: 26 + owner_nonce = b"n" * 32 27 + proof_der = b"proof-der" 28 + tls_exporter = b"e" * EXPORTER_BYTES 29 + tls_spki_der = b"tls-spki" 30 + evidence = object() 31 + verdict = object() 32 + policy = Policy(pcr_mode="pin", pcr_pins={"abc"}) 33 + captured_certificate: dict[str, object] = {} 34 + captured_exporter: dict[str, object] = {} 35 + connections: list[object] = [] 36 + 37 + class FakeRawSocket: 38 + def __init__(self) -> None: 39 + self.sent: list[bytes] = [] 40 + self.timeout = 30.0 41 + self.closed = False 42 + 43 + def sendall(self, data: bytes) -> None: 44 + self.sent.append(data) 45 + 46 + def settimeout(self, timeout: float | None) -> None: 47 + self.timeout = timeout 48 + 49 + def close(self) -> None: 50 + self.closed = True 51 + 52 + class FakeCertificate: 53 + def public_bytes(self, _encoding) -> bytes: 54 + return b"certificate-der" 55 + 56 + class FakePeer: 57 + def to_cryptography(self) -> FakeCertificate: 58 + return FakeCertificate() 59 + 60 + class FakeConnection: 61 + def __init__(self, context, raw) -> None: 62 + self.context = context 63 + self.raw = raw 64 + self.sent: list[bytes] = [] 65 + self.recv_chunks = [ 66 + ( 67 + b"HTTP/1.1 200 OK\r\n" 68 + + f"Content-Type: {EXPORTER_PROOF_MEDIA_TYPE}\r\n".encode("ascii") 69 + + f"Content-Length: {len(proof_der)}\r\n\r\n".encode("ascii") 70 + + proof_der 71 + ) 72 + ] 73 + self.closed = False 74 + connections.append(self) 75 + 76 + def setblocking(self, _flag: int) -> None: 77 + pass 78 + 79 + def set_connect_state(self) -> None: 80 + pass 81 + 82 + def set_tlsext_host_name(self, _name: bytes) -> None: 83 + pass 84 + 85 + def do_handshake(self) -> None: 86 + pass 87 + 88 + def get_peer_certificate(self) -> FakePeer: 89 + return FakePeer() 90 + 91 + def export_keying_material( 92 + self, 93 + _label: bytes, 94 + _size: int, 95 + _context: bytes, 96 + ) -> bytes: 97 + return tls_exporter 98 + 99 + def sendall(self, data: bytes) -> None: 100 + self.sent.append(data) 101 + 102 + def recv(self, _size: int) -> bytes: 103 + return self.recv_chunks.pop(0) if self.recv_chunks else b"" 104 + 105 + def close(self) -> None: 106 + self.closed = True 107 + 108 + raw = FakeRawSocket() 109 + monkeypatch.setattr( 110 + channel.socket, 111 + "create_connection", 112 + lambda _address, timeout: raw, 113 + ) 114 + monkeypatch.setattr(channel, "_tls_context", lambda: object()) 115 + monkeypatch.setattr(channel.SSL, "Connection", FakeConnection) 116 + 117 + def fake_verify_certificate_evidence(**kwargs): 118 + captured_certificate.update(kwargs) 119 + return SimpleNamespace( 120 + tls_spki_der=tls_spki_der, 121 + evidence=evidence, 122 + verdict=verdict, 123 + ) 124 + 125 + def fake_verify_exporter_proof(**kwargs): 126 + captured_exporter.update(kwargs) 127 + 128 + monkeypatch.setattr( 129 + channel, 130 + "verify_certificate_evidence", 131 + fake_verify_certificate_evidence, 132 + ) 133 + monkeypatch.setattr(channel, "verify_exporter_proof", fake_verify_exporter_proof) 134 + 135 + established = channel.establish_attested_channel( 136 + channel.RatlsEndpoint("spp.example.test", 9443), 137 + owner_nonce=owner_nonce, 138 + nvattest_dir=tmp_path, 139 + now=datetime(2026, 7, 24, tzinfo=timezone.utc), 140 + policy=policy, 141 + composite_verifier=lambda *_args, **_kwargs: verdict, 142 + monotonic_now=lambda: 123.0, 143 + epoch=7, 144 + ) 145 + 146 + assert raw.sent == [PREFACE_MAGIC + owner_nonce] 147 + assert raw.timeout is None 148 + assert connections 149 + assert connections[0].sent == [ 150 + ( 151 + f"GET {EXPORTER_PROOF_PATH} HTTP/1.1\r\n" 152 + "Host: spp-engine\r\n" 153 + "Content-Length: 0\r\n\r\n" 154 + ).encode("ascii") 155 + ] 156 + assert captured_certificate["policy"] is policy 157 + assert captured_exporter == { 158 + "proof_der": proof_der, 159 + "evidence": evidence, 160 + "tls_exporter": tls_exporter, 161 + "owner_nonce": owner_nonce, 162 + "policy": policy, 163 + } 164 + assert established.verdict is verdict 165 + assert established.epoch == 7 166 + assert established.last_used_monotonic == 123.0
+47 -1
tests/services/test_spp_attest_ratls_verify.py
··· 3 3 4 4 from __future__ import annotations 5 5 6 + import hashlib 6 7 from datetime import datetime, timedelta, timezone 7 8 from pathlib import Path 8 9 from typing import Any ··· 30 31 verify_certificate_evidence, 31 32 verify_exporter_proof, 32 33 ) 33 - from solstone.think.services.spp_attest.snp import AppraisalStep, CpuAppraisal 34 + from solstone.think.services.spp_attest.snp import AppraisalStep, CpuAppraisal, Policy 34 35 35 36 NOW = datetime(2026, 7, 12, tzinfo=timezone.utc) 36 37 ··· 228 229 "cpu_verification_failed", 229 230 ), 230 231 ( 232 + "the CPU leg rejected the evidence (pcr_pin_mismatch)", 233 + "pcr_pin_mismatch", 234 + ), 235 + ( 231 236 "the GPU leg rejected evidence (nvattest_integrity_failed)", 232 237 "nvattest_integrity_failed", 233 238 ), ··· 313 318 ) 314 319 315 320 assert exc_info.value.reason_code == "exporter_mismatch" 321 + 322 + 323 + def test_verify_exporter_proof_rejects_pcr_pin_mismatch_after_valid_quote( 324 + monkeypatch, 325 + ) -> None: 326 + nonce = b"n" * 32 327 + _key, spki = _key_and_spki() 328 + tls_exporter = b"e" * 32 329 + evidence = _evidence(nonce, spki) 330 + proof = ExporterProof( 331 + nonce, 332 + spki, 333 + tls_exporter, 334 + b"p2-message", 335 + b"p2-signature", 336 + b"p2-pcrs", 337 + ) 338 + seen: dict[str, Any] = {} 339 + 340 + def verify_quote(**kwargs): 341 + seen.update(kwargs) 342 + 343 + monkeypatch.setattr(ratls_verify, "verify_quote", verify_quote) 344 + policy = Policy( 345 + pcr_mode="pin", 346 + pcr_pins={hashlib.sha256(evidence.quote_pcrs).hexdigest()}, 347 + ) 348 + 349 + with pytest.raises(RatlsVerificationError) as exc_info: 350 + verify_exporter_proof( 351 + proof_der=proof.to_der(), 352 + evidence=evidence, 353 + tls_exporter=tls_exporter, 354 + owner_nonce=nonce, 355 + policy=policy, 356 + ) 357 + 358 + assert evidence.quote_pcrs == b"p1-pcrs" 359 + assert proof.quote_pcrs == b"p2-pcrs" 360 + assert seen["quote_pcrs"] == proof.quote_pcrs 361 + assert exc_info.value.reason_code == "pcr_pin_mismatch" 316 362 317 363 318 364 def test_verify_exporter_proof_rejects_quote_under_wrong_ak(monkeypatch) -> None:
+13
tests/services/test_spp_attest_snp.py
··· 4 4 from __future__ import annotations 5 5 6 6 import datetime as dt 7 + import hashlib 7 8 import json 8 9 import shutil 9 10 from pathlib import Path ··· 20 21 load_cpu_bundle, 21 22 ) 22 23 from solstone.think.services.spp_attest import snp as snp_module 24 + from solstone.think.services.spp_attest.snp import check_pcr_fingerprint 23 25 24 26 FIXTURE_DIR = Path(__file__).resolve().parents[1] / "fixtures" / "spp_attest" 25 27 PCR_SHA256_HEX = "b162f46105c80d3e45028e37cc649404c9d65297ad1cda8f953208582060b0e3" ··· 263 265 264 266 with pytest.raises(VerificationError, match="not in pinned policy"): 265 267 _appraise(policy=policy) 268 + 269 + 270 + def test_check_pcr_fingerprint_accepts_rotation_overlap() -> None: 271 + old_pcrs = b"old-pcrs" 272 + new_pcrs = b"new-pcrs" 273 + old_pin = hashlib.sha256(old_pcrs).hexdigest() 274 + new_pin = hashlib.sha256(new_pcrs).hexdigest() 275 + policy = Policy(pcr_mode="pin", pcr_pins={old_pin, new_pin}) 276 + 277 + assert check_pcr_fingerprint(old_pcrs, policy) == old_pin 278 + assert check_pcr_fingerprint(new_pcrs, policy) == new_pin 266 279 267 280 268 281 def test_appraise_cpu_leg_rejects_tlv_splice_before_appraisal_steps() -> None:
+146 -1
tests/services/test_spp_transport.py
··· 12 12 from unittest.mock import Mock 13 13 14 14 import pytest 15 + from cryptography import x509 16 + from cryptography.hazmat.primitives import hashes, serialization 17 + from cryptography.hazmat.primitives.asymmetric import ec 18 + from cryptography.x509.oid import NameOID, ObjectIdentifier 15 19 16 20 from solstone.think.journal_io.locking import hold_lock 17 21 from solstone.think.models import AttestationFailedError 18 22 from solstone.think.providers import nvattest_install 19 23 from solstone.think.services import spp, spp_transport 20 24 from solstone.think.services.spp_attest.cadence import AttestationSession 25 + from solstone.think.services.spp_attest.composite import verify_composite 21 26 from solstone.think.services.spp_attest.ratls.channel import RatlsChannelError 22 - from solstone.think.services.spp_attest.ratls.verify import RatlsVerificationError 27 + from solstone.think.services.spp_attest.ratls.contract import ( 28 + COMPOSITE_EVIDENCE_OID, 29 + CompositeEvidence, 30 + ) 31 + from solstone.think.services.spp_attest.ratls.verify import ( 32 + RatlsVerificationError, 33 + verify_certificate_evidence, 34 + ) 35 + from solstone.think.services.spp_attest.snp import Policy 36 + 37 + SPP_FIXTURE_DIR = Path(__file__).resolve().parents[1] / "fixtures" / "spp_attest" 23 38 24 39 25 40 class _FakeChannel: ··· 115 130 monkeypatch.setattr(spp_transport, "_start_listener_locked", fake_start_listener) 116 131 117 132 133 + def _fixture_owner_nonce() -> bytes: 134 + return bytes.fromhex("".join((SPP_FIXTURE_DIR / "nonce.hex").read_text().split())) 135 + 136 + 137 + def _fixture_composite_evidence(owner_nonce: bytes, spki: bytes) -> CompositeEvidence: 138 + certs_dir = SPP_FIXTURE_DIR / "certs" 139 + return CompositeEvidence( 140 + owner_nonce=owner_nonce, 141 + tls_spki_der=spki, 142 + amd_report=(SPP_FIXTURE_DIR / "report.bin").read_bytes(), 143 + hcl_report=(SPP_FIXTURE_DIR / "hcl_report.bin").read_bytes(), 144 + ak_public_key_pem=(SPP_FIXTURE_DIR / "akpub.pem").read_bytes(), 145 + quote_message=(SPP_FIXTURE_DIR / "quote.msg").read_bytes(), 146 + quote_signature=(SPP_FIXTURE_DIR / "quote.sig").read_bytes(), 147 + quote_pcrs=(SPP_FIXTURE_DIR / "quote.pcrs").read_bytes(), 148 + amd_ark_pem=(certs_dir / "ark.pem").read_bytes(), 149 + amd_ask_pem=(certs_dir / "ask.pem").read_bytes(), 150 + amd_vcek_pem=(certs_dir / "vcek.pem").read_bytes(), 151 + gpu_envelope=(SPP_FIXTURE_DIR / "gpu-envelope.tlv").read_bytes(), 152 + ) 153 + 154 + 155 + def _key_and_spki() -> tuple[ec.EllipticCurvePrivateKey, bytes]: 156 + key = ec.generate_private_key(ec.SECP256R1()) 157 + spki = key.public_key().public_bytes( 158 + serialization.Encoding.DER, 159 + serialization.PublicFormat.SubjectPublicKeyInfo, 160 + ) 161 + return key, spki 162 + 163 + 164 + def _certificate_der( 165 + key: ec.EllipticCurvePrivateKey, 166 + evidence: CompositeEvidence, 167 + ) -> bytes: 168 + subject = issuer = x509.Name( 169 + [x509.NameAttribute(NameOID.COMMON_NAME, "spp-engine-test")] 170 + ) 171 + cert = ( 172 + x509.CertificateBuilder() 173 + .subject_name(subject) 174 + .issuer_name(issuer) 175 + .public_key(key.public_key()) 176 + .serial_number(1001) 177 + .not_valid_before(datetime.now(timezone.utc) - timedelta(days=1)) 178 + .not_valid_after(datetime.now(timezone.utc) + timedelta(days=1)) 179 + .add_extension( 180 + x509.UnrecognizedExtension( 181 + ObjectIdentifier(COMPOSITE_EVIDENCE_OID), 182 + evidence.to_der(), 183 + ), 184 + critical=True, 185 + ) 186 + .sign(key, hashes.SHA256()) 187 + ) 188 + return cert.public_bytes(serialization.Encoding.DER) 189 + 190 + 118 191 def _stale_session(verdict: object) -> AttestationSession: 119 192 old = datetime.now(timezone.utc) - timedelta(hours=2) 120 193 return AttestationSession( ··· 122 195 started_at=old, 123 196 tpm_heartbeat_at=old, 124 197 gpu_reattest_at=old, 198 + ) 199 + 200 + 201 + def test_establish_channel_locked_passes_production_pcr_policy( 202 + tmp_path: Path, 203 + monkeypatch: pytest.MonkeyPatch, 204 + ) -> None: 205 + block = _write_confidential_config(tmp_path, monkeypatch) 206 + captured: dict[str, object] = {} 207 + 208 + def fake_establish(_endpoint, **kwargs): 209 + captured.update(kwargs) 210 + return _FakeChannel(object(), epoch=kwargs["epoch"]) 211 + 212 + monkeypatch.setattr(spp_transport, "establish_attested_channel", fake_establish) 213 + 214 + spp_transport._establish_channel_locked( 215 + block, 216 + datetime.now(timezone.utc), 217 + nvattest_dir=tmp_path / "nvattest", 218 + ) 219 + 220 + policy = captured["policy"] 221 + assert isinstance(policy, Policy) 222 + assert policy.pcr_mode == "pin" 223 + assert ( 224 + "b162f46105c80d3e45028e37cc649404c9d65297ad1cda8f953208582060b0e3" 225 + in policy.pcr_pins 125 226 ) 126 227 127 228 ··· 489 590 "certificate_extension_invalid", 490 591 "certificate_evidence_invalid", 491 592 "nonce_mismatch", 593 + "pcr_pin_mismatch", 492 594 "spki_mismatch", 493 595 "cpu_verification_failed", 494 596 "gpu_nonce_mismatch", ··· 554 656 assert failure is not None 555 657 assert failure.kind == kind 556 658 assert failure.reason_code == reason_code 659 + 660 + 661 + def test_pcr_pin_mismatch_reason_records_from_real_composite_origin( 662 + tmp_path: Path, 663 + monkeypatch: pytest.MonkeyPatch, 664 + ) -> None: 665 + block = _write_confidential_config(tmp_path, monkeypatch) 666 + observed: dict[str, str] = {} 667 + monkeypatch.setattr( 668 + spp_transport.secrets, 669 + "token_bytes", 670 + lambda _size: _fixture_owner_nonce(), 671 + ) 672 + 673 + def fake_establish(_endpoint, **kwargs): 674 + key, spki = _key_and_spki() 675 + evidence = _fixture_composite_evidence(kwargs["owner_nonce"], spki) 676 + bad_policy = Policy(pcr_mode="pin", pcr_pins={"00" * 32}) 677 + try: 678 + verify_certificate_evidence( 679 + certificate_der=_certificate_der(key, evidence), 680 + owner_nonce=kwargs["owner_nonce"], 681 + now=kwargs["now"], 682 + nvattest_dir=kwargs["nvattest_dir"], 683 + policy=bad_policy, 684 + quote_verifier=lambda **_kwargs: None, 685 + composite_verifier=verify_composite, 686 + ) 687 + except RatlsVerificationError as exc: 688 + observed["reason_code"] = exc.reason_code 689 + raise 690 + raise AssertionError("fixture unexpectedly passed a non-matching PCR pin") 691 + 692 + monkeypatch.setattr(spp_transport, "establish_attested_channel", fake_establish) 693 + 694 + with pytest.raises(AttestationFailedError): 695 + spp_transport.verify_confidential_attestation(block) 696 + 697 + assert observed["reason_code"] == "pcr_pin_mismatch" 698 + failure = spp.get_attestation_state().failure 699 + assert failure is not None 700 + assert failure.kind == "failed" 701 + assert failure.reason_code == "pcr_pin_mismatch" 557 702 558 703 559 704 @pytest.mark.parametrize(
+1
tests/test_brain_cli.py
··· 424 424 ("nvattest_integrity_failed", "nvattest_integrity_failed", "failed"), 425 425 ("gateway_unreachable", "attestation_not_verified", "blocked"), 426 426 ("attestation_failed", "attestation_rejected", "failed"), 427 + ("pcr_pin_mismatch", "attestation_rejected", "failed"), 427 428 ], 428 429 ) 429 430 def test_spp_prerequisite_maps_failure_reason_code(