diff --git a/packages/syft-enclave/src/syft_enclaves/attestation.py b/packages/syft-enclave/src/syft_enclaves/attestation.py index 238b50cddd4..665ef596a83 100644 --- a/packages/syft-enclave/src/syft_enclaves/attestation.py +++ b/packages/syft-enclave/src/syft_enclaves/attestation.py @@ -9,9 +9,11 @@ from __future__ import annotations from dataclasses import dataclass, field +from typing import Optional from google.auth.transport import requests as google_requests from google.oauth2 import id_token +from pydantic import BaseModel from syft_client.version import SYFT_CLIENT_VERSION @@ -21,11 +23,34 @@ "signer@confidentialspace-sign.iam.gserviceaccount.com" ) -# The verifier expects the enclave to run the same syft-client version as -# this client. Tracks releases automatically — no manual bump needed when -# syft-client's version changes. -EXPECTED_SYFT_VERSION = SYFT_CLIENT_VERSION -EXPECTED_IMAGE_DIGEST = "" # TODO: set after enclave image is published +# Google mints Confidential Space attestation tokens with a short lifetime +# (~30 minutes), but the enclave only writes its token to SYFT_version.json once +# at boot and does not yet refresh it. So after ~30 minutes every peer would +# reject the enclave. +# +# TODO: remove this once the enclave periodically refreshes its attestation +# token in SYFT_version.json — then the real (short) expiry can be honoured. +JWT_EXPIRY_GRACE_SECONDS = 30 * 24 * 60 * 60 # ~1 month + + +class AppraisalPolicy(BaseModel): + """Reference values the verifier appraises attestation evidence against. + + In RATS terms this is the *appraisal policy*: the + set of trusted reference values the enclave's evidence is compared to. + + The image digest is intentionally not shipped as a constant — the data + owner supplies the digest they independently confirmed. Left unset + (``None``), the image-digest check is skipped and the image is not pinned. + """ + + model_config = {"frozen": True} + + # None → image-digest check skipped (no image pinned). Set a "sha256:..." + # digest to pin, and require, a specific enclave image. + expected_image_digest: Optional[str] = None + # By default, the enclave must run the same version of syft-client as the verifier. + expected_syft_version: Optional[str] = SYFT_CLIENT_VERSION class AttestationError(Exception): @@ -70,7 +95,11 @@ def print_checklist(self) -> None: print(f"{icon} {check.label:<20s} — {check.detail}") -def verify_attestation_token(token: str, verbose: bool = True) -> AttestationResult: +def verify_attestation_token( + token: str, + policy: AppraisalPolicy | None = None, + verbose: bool = True, +) -> AttestationResult: """Verify an attestation JWT and return the result checklist. Runs every check before raising — so a failure in one (e.g. ``dbgstat``) @@ -83,13 +112,28 @@ def verify_attestation_token(token: str, verbose: bool = True) -> AttestationRes to inspect. ``passed=None`` ("skipped") does not count as a failure. + + Args: + token: the attestation JWT to verify. + policy: reference values to appraise the evidence against. Defaults to + the shipped ``AppraisalPolicy()`` (module-level pinned digest and + version). Pass a custom policy to appraise against your own + independently-verified image digest. + verbose: print the check progress and final checklist. """ + policy = policy or AppraisalPolicy() + expected_image_digest = policy.expected_image_digest + expected_syft_version = policy.expected_syft_version + result = AttestationResult() if verbose: print("🔒 Verifying enclave attestation...") - # 1. JWT signature + expiry — fail-fast (no claims → no point continuing) + # 1. JWT signature + expiry — fail-fast (no claims → no point continuing). + # clock_skew_in_seconds widens the accepted expiry window (token valid until + # exp + grace) as a stopgap for the enclave not yet refreshing its token — + # see JWT_EXPIRY_GRACE_SECONDS. if verbose: print(" ⏳ JWT signature ...") try: @@ -99,6 +143,7 @@ def verify_attestation_token(token: str, verbose: bool = True) -> AttestationRes request, audience=ATTESTATION_AUDIENCE, certs_url=CONFIDENTIAL_COMPUTING_CERTS_URL, + clock_skew_in_seconds=JWT_EXPIRY_GRACE_SECONDS, ) result.add( "jwt_signature", @@ -159,7 +204,7 @@ def verify_attestation_token(token: str, verbose: bool = True) -> AttestationRes eat_nonce = [eat_nonce] actual_version_nonce = eat_nonce[0] if eat_nonce else None # Must match the format produced by syft_enclaves.tee_token.build_eat_nonce. - expected_version_nonce = f"syft-client-{EXPECTED_SYFT_VERSION}" + expected_version_nonce = f"syft-client-{expected_syft_version}" if not actual_version_nonce: result.add( "version_match", @@ -172,7 +217,7 @@ def verify_attestation_token(token: str, verbose: bool = True) -> AttestationRes "version_match", "Version match", True, - f"enclave runs expected syft-client {EXPECTED_SYFT_VERSION}", + f"enclave runs expected syft-client {expected_syft_version}", ) else: result.add( @@ -182,21 +227,29 @@ def verify_attestation_token(token: str, verbose: bool = True) -> AttestationRes f"version mismatch (enclave={actual_version_nonce!r}, expected={expected_version_nonce!r})", ) - # 5. Image digest + # 5. Image digest. The expected digest is supplied by the data owner via the + # AppraisalPolicy . When none is supplied the check is + # SKIPPED (passed=None), not failed. if verbose: print(" ⏳ Image digest ...") container = claims.get("submods", {}).get("container", {}) image_digest = container.get("image_digest") - if not EXPECTED_IMAGE_DIGEST: + if not expected_image_digest: result.add( "image_digest", "Image digest", None, - f"digest {image_digest[:20]}... (expected digest not configured, skipped)" - if image_digest - else "no digest in token (expected digest not configured, skipped)", + "no expected image digest supplied — pass one via AppraisalPolicy " + "(attest_peer(..., expected_image_digest=...)) to pin the image (skipped)", + ) + elif not image_digest: + result.add( + "image_digest", + "Image digest", + False, + "no image digest in token — cannot verify enclave is running the released image", ) - elif image_digest == EXPECTED_IMAGE_DIGEST: + elif image_digest == expected_image_digest: result.add( "image_digest", "Image digest", @@ -208,7 +261,7 @@ def verify_attestation_token(token: str, verbose: bool = True) -> AttestationRes "image_digest", "Image digest", False, - f"digest mismatch (got {image_digest or 'none'}, expected {EXPECTED_IMAGE_DIGEST[:20]}...)", + f"digest mismatch (got {image_digest}, expected {expected_image_digest})", ) # Finalize — print full checklist, then raise once if anything failed diff --git a/packages/syft-enclave/src/syft_enclaves/client.py b/packages/syft-enclave/src/syft_enclaves/client.py index 7437ff460ff..6bdac8af9af 100644 --- a/packages/syft-enclave/src/syft_enclaves/client.py +++ b/packages/syft-enclave/src/syft_enclaves/client.py @@ -16,6 +16,10 @@ PartyApprovalStatus, enclave_approval_file_name, ) +from syft_enclaves.attestation import ( + AppraisalPolicy, + verify_attestation_token, +) from syft_perms.syftperm_context import SyftPermContext from syft_enclaves.enclave_job_client import EnclaveJobClient @@ -74,11 +78,30 @@ def approve_peer_request( def reject_peer_request(self, email_or_peer: str | Peer): self._manager.reject_peer_request(email_or_peer) - def attest_peer(self, peer_email: str): + def attest_peer( + self, + peer_email: str, + expected_image_digest: str | None = None, + policy: "AppraisalPolicy | None" = None, + ): """Verify an enclave peer's attestation by re-reading SYFT_version.json from Drive. Returns None (with an info print) when no token is available; - raises AttestationError only when verification of an existing token fails.""" - from syft_enclaves.attestation import verify_attestation_token + raises AttestationError only when verification of an existing token fails. + + Args: + peer_email: the enclave peer to attest. + expected_image_digest: a "sha256:..." container image digest you + trust — . When set, the attestation + is appraised against it. + policy: a full ``AppraisalPolicy`` for finer control (image digest + *and* syft-client version). Mutually exclusive with + ``expected_image_digest``. + """ + + if expected_image_digest is not None and policy is not None: + raise ValueError("Pass either expected_image_digest or policy, not both.") + if expected_image_digest is not None: + policy = AppraisalPolicy(expected_image_digest=expected_image_digest) version_info = ( self._manager.peer_manager.connection_router.read_peer_version_file( @@ -96,7 +119,7 @@ def attest_peer(self, peer_email: str): "(not running in a Confidential Space); skipping attestation." ) return None - return verify_attestation_token(version_info.attestation_token) + return verify_attestation_token(version_info.attestation_token, policy=policy) def sync(self): self._manager.sync() diff --git a/packages/syft-enclave/tests/test_attestation.py b/packages/syft-enclave/tests/test_attestation.py index 54d0c2b9039..1682b28e0a8 100644 --- a/packages/syft-enclave/tests/test_attestation.py +++ b/packages/syft-enclave/tests/test_attestation.py @@ -4,15 +4,23 @@ import pytest +from syft_client.version import SYFT_CLIENT_VERSION + from syft_enclaves.attestation import ( - EXPECTED_SYFT_VERSION, + JWT_EXPIRY_GRACE_SECONDS, + AppraisalPolicy, AttestationError, AttestationResult, verify_attestation_token, ) FAKE_IMAGE_DIGEST = "sha256:abc123" -EXPECTED_VERSION_NONCE = f"syft-client-{EXPECTED_SYFT_VERSION}" +EXPECTED_VERSION_NONCE = f"syft-client-{SYFT_CLIENT_VERSION}" + +# The image digest is not shipped as a constant — it's supplied per-call via an +# AppraisalPolicy. A policy pinning the fake token's digest is used by the +# tests that need the image-digest check to pass. +DEFAULT_TEST_POLICY = AppraisalPolicy(expected_image_digest=FAKE_IMAGE_DIGEST) def _valid_claims(**overrides): @@ -34,7 +42,12 @@ def _valid_claims(**overrides): @pytest.fixture def mock_verify(): - """Patch google id_token.verify_token to return valid claims.""" + """Patch google id_token.verify_token to return valid claims. + + Tests use ``_verify`` (or pass ``DEFAULT_TEST_POLICY``) so the fake token's + digest matches the policy and the image_digest check passes. Tests + targeting image_digest pass their own policy. + """ with ( patch("syft_enclaves.attestation.id_token.verify_token") as mock_vt, patch("syft_enclaves.attestation.google_requests.Request"), @@ -45,22 +58,26 @@ def mock_verify(): class TestVerifyAttestationToken: def test_all_checks_pass(self, mock_verify): - # Configure EXPECTED_IMAGE_DIGEST so the image_digest check resolves - # to True (not None/skipped) and all five checks pass. - with patch( - "syft_enclaves.attestation.EXPECTED_IMAGE_DIGEST", - FAKE_IMAGE_DIGEST, - ): - result = verify_attestation_token("fake-token", verbose=False) - assert result.all_passed() - assert len(result.checks) == 5 - assert all(c.passed for c in result.checks) + result = verify_attestation_token( + "fake-token", policy=DEFAULT_TEST_POLICY, verbose=False + ) + assert result.all_passed() + assert len(result.checks) == 5 + assert all(c.passed for c in result.checks) def test_jwt_signature_failure(self, mock_verify): mock_verify.side_effect = ValueError("bad signature") with pytest.raises(AttestationError, match="JWT signature"): verify_attestation_token("fake-token", verbose=False) + def test_jwt_expiry_grace_passed_through(self, mock_verify): + """The enclave doesn't yet refresh its token, so the verifier accepts an + expired token for a grace window (~1 month) via clock_skew_in_seconds.""" + verify_attestation_token("fake-token", verbose=False) + _, kwargs = mock_verify.call_args + assert kwargs["clock_skew_in_seconds"] == JWT_EXPIRY_GRACE_SECONDS + assert JWT_EXPIRY_GRACE_SECONDS == 30 * 24 * 60 * 60 + def test_secure_boot_disabled(self, mock_verify): mock_verify.return_value = _valid_claims(secboot=False) with pytest.raises(AttestationError, match="secure_boot"): @@ -86,7 +103,7 @@ def test_version_mismatch(self, mock_verify): def test_version_unprefixed_rejected(self, mock_verify): """A bare version (pre-fix sender) must be rejected, not accepted.""" - mock_verify.return_value = _valid_claims(eat_nonce=[EXPECTED_SYFT_VERSION]) + mock_verify.return_value = _valid_claims(eat_nonce=[SYFT_CLIENT_VERSION]) with pytest.raises(AttestationError, match="version_match"): verify_attestation_token("fake-token", verbose=False) @@ -106,31 +123,38 @@ def test_version_as_string(self, mock_verify): assert version_check.passed is True def test_image_digest_mismatch(self, mock_verify): - with patch( - "syft_enclaves.attestation.EXPECTED_IMAGE_DIGEST", - "sha256:expected", - ): - mock_verify.return_value = _valid_claims() - with pytest.raises(AttestationError, match="image_digest"): - verify_attestation_token("fake-token", verbose=False) - - def test_image_digest_skipped_when_not_configured(self, mock_verify): - """When EXPECTED_IMAGE_DIGEST is empty, image check is skipped (passed=None).""" - result = verify_attestation_token("fake-token", verbose=False) + policy = AppraisalPolicy(expected_image_digest="sha256:expected") + with pytest.raises(AttestationError, match="image_digest"): + verify_attestation_token("fake-token", policy=policy, verbose=False) + + def test_image_digest_skipped_when_not_supplied(self, mock_verify): + """No expected digest supplied → the image-digest check is skipped + (passed=None), not failed. The default policy pins no image.""" + result = verify_attestation_token( + "fake-token", policy=AppraisalPolicy(), verbose=False + ) image_check = next(c for c in result.checks if c.name == "image_digest") - # Skipped checks use passed=None — distinguishes "not run" from "failed". assert image_check.passed is None - assert "skipped" in image_check.detail + assert "no expected image digest supplied" in image_check.detail + + def test_image_digest_fails_when_supplied_but_missing_from_token(self, mock_verify): + """A pinned policy plus a token with no image_digest claim must be + rejected — the verifier can't confirm which image is running.""" + claims = _valid_claims() + del claims["submods"]["container"]["image_digest"] + mock_verify.return_value = claims + with pytest.raises(AttestationError, match="image_digest"): + verify_attestation_token( + "fake-token", policy=DEFAULT_TEST_POLICY, verbose=False + ) def test_image_digest_matches(self, mock_verify): - with patch( - "syft_enclaves.attestation.EXPECTED_IMAGE_DIGEST", - FAKE_IMAGE_DIGEST, - ): - result = verify_attestation_token("fake-token", verbose=False) - image_check = next(c for c in result.checks if c.name == "image_digest") - assert image_check.passed - assert "matches" in image_check.detail + result = verify_attestation_token( + "fake-token", policy=DEFAULT_TEST_POLICY, verbose=False + ) + image_check = next(c for c in result.checks if c.name == "image_digest") + assert image_check.passed + assert "matches" in image_check.detail def test_error_carries_result(self, mock_verify): mock_verify.return_value = _valid_claims(secboot=False)