fix: pin enclave container image digest in attestation, fail closed

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

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()

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)