feat: prep SGX attestation

README.md

@@ -97,6 +97,31 @@ asyncio.run(main())
 - **Guaranteed zeroing**: Memory is zeroed after encryption completes
 - **Fallback mechanism**: Graceful degradation if SecureMemory module unavailable
 
+### SGX Enclave Attestation (operator-blind, opt-in)
+
+For the `high`/`maximum` tiers the server runs inside an Intel SGX enclave. The
+client can **verify a hardware attestation quote before sending any plaintext**,
+proving the server public key it encrypts to was generated inside a genuine
+enclave running the expected code — making the channel operator-blind. This is
+opt-in and additive; without it the client behaves exactly as before.
+
+```python
+from nomyo import SecureChatCompletion, AttestationPolicy, JwtQuoteVerifier
+
+client = SecureChatCompletion(
+    base_url="https://api.nomyo.ai", api_key="...",
+    attestation_policy=AttestationPolicy(
+        pinned_mrenclave={"<hex>"}, enforce=False,  # rollout: warn; True = hard-fail
+    ),
+    quote_verifier=JwtQuoteVerifier(verify_url="https://attest.example/verify",
+                                    jwks_url="https://attest.example/certs"),
+)
+```
+
+Install the verifier extra with `pip install nomyo[attestation]`. See
+[doc/attestation.md](doc/attestation.md) for policy, rollout/enforce, liveness,
+and how to self-host the verification service.
+
 ## 🔄 OpenAI Compatibility
 
 The `SecureChatCompletion` class provides **exact API compatibility** with OpenAI's `ChatCompletion.create()` method.

doc/attestation.md

@@ -0,0 +1,134 @@
+# SGX enclave attestation (operator-blind verification)
+
+For the `high` and `maximum` security tiers, the NOMYO server runs CPU inference
+inside an Intel SGX enclave. SGX **attestation** lets the client prove — *before
+sending any plaintext* — that the RSA public key it is about to encrypt to was
+generated inside a genuine enclave running the expected (measured) code. This
+upgrades the channel from "encrypted in transit, operator-trusted" to
+"operator-blind": only the enclave can decrypt, not the host operator.
+
+This is **opt-in** and **additive**. With no attestation configured the client
+behaves exactly as before.
+
+## How it works
+
+On each secure request for a tier that requires attestation, the client:
+
+1. Fetches the server public key (`GET /pki/public_key`).
+2. Fetches a hardware-signed DCAP quote (`GET /pki/attestation`).
+3. **Verifies the quote** (ECDSA signature + Intel PCK chain + TCB) via a
+   `QuoteVerifier` you provide (step 4).
+4. Checks policy: `tcb_status`, pinned `MRENCLAVE` (and optionally
+   `MRSIGNER` / `prod_id` / minimum `isv_svn`).
+5. Checks the **key binding**: `SHA-512(server_pubkey_der [|| challenge])` must
+   equal the `report_data` inside the verified quote (constant-time compare).
+6. Only if all checks pass does it encrypt the payload **to that exact attested
+   key** and send it. In `enforce` mode any failure raises and nothing is sent.
+
+## Quote verification (`QuoteVerifier`)
+
+Verifying a DCAP quote must not be hand-rolled. The client delegates step 4 to a
+`QuoteVerifier`. Three ways to supply one:
+
+### `JwtQuoteVerifier` (recommended)
+
+Talks to a remote verification service that returns a signed JWT of appraised
+claims — Intel Trust Authority, [Veraison](https://github.com/veraison), or a
+self-hosted wrapper around the Intel DCAP Quote Verification Library (QVL). The
+native/PCCS dependency lives on that one service, not on every client.
+
+Requires the optional dependency:
+
+```bash
+pip install nomyo[attestation]
+```
+
+```python
+from nomyo import SecureChatCompletion, AttestationPolicy, JwtQuoteVerifier
+
+verifier = JwtQuoteVerifier(
+    verify_url="https://attest.example.com/appraisal/v1/attest",
+    jwks_url="https://attest.example.com/certs",
+    issuer="https://attest.example.com",     # optional
+    audience="nomyo-client",                  # optional
+    # claim_map=...  # override if your service uses different claim names
+)
+
+policy = AttestationPolicy(
+    pinned_mrenclave={"<hex from gramine/build-enclave.sh>"},
+    allowed_tcb_statuses={"UpToDate"},
+    enforce=False,   # rollout: warn and proceed. Set True to hard-fail.
+)
+
+client = SecureChatCompletion(
+    base_url="https://api.nomyo.ai",
+    api_key="...",
+    attestation_policy=policy,
+    quote_verifier=verifier,
+)
+
+resp = await client.create(
+    model="...", messages=[...], security_tier="high",
+)
+```
+
+The default JWT claim names are in `DEFAULT_CLAIM_MAP`; override `claim_map` if
+your service differs (e.g. Veraison vs ITA).
+
+### `CallableQuoteVerifier` — bring your own
+
+```python
+from nomyo import CallableQuoteVerifier, VerifiedQuote
+
+async def my_verify(quote: bytes) -> VerifiedQuote:
+    # call your verifier (local QVL FFI, another service, ...)
+    return VerifiedQuote(mrenclave=..., mrsigner=..., isv_prod_id=...,
+                         isv_svn=..., tcb_status="UpToDate", report_data=...)
+
+client = SecureChatCompletion(..., attestation_policy=policy,
+                              quote_verifier=CallableQuoteVerifier(my_verify))
+```
+
+`my_verify` may be sync or async and must raise on any verification failure.
+
+## Policy (`AttestationPolicy`)
+
+| Field | Meaning | Default |
+|---|---|---|
+| `pinned_mrenclave` | Allowed enclave measurements (a **set** so you can roll builds) | `frozenset()` |
+| `pinned_mrsigner` | Allowed signer (optional) | `None` |
+| `pinned_prod_id` / `min_isv_svn` | Product id / minimum security version (optional) | `None` |
+| `allowed_tcb_statuses` | Accepted platform TCB levels | `{"UpToDate"}` |
+| `enforce` | Hard-fail vs warn-and-proceed | `False` |
+| `require_attestation_for_tier` | Tiers that must be operator-blind | `{"high","maximum"}` |
+| `liveness` | Use a fresh per-request challenge (see below) | `False` |
+
+Pin MRENCLAVE values as **configuration**, not constants — they change whenever
+the enclave build or its file paths change. Printed by
+`gramine/build-enclave.sh` at sign time. Do **not** silently allow `OutOfDate` or
+`Revoked` TCB statuses.
+
+## Rollout vs enforce (spec §6)
+
+- **`standard` tier:** attestation not expected — proceeds normally.
+- **`high`/`maximum`, `enforce=False` (rollout, default):** if attestation is
+  unavailable (server not yet under SGX) or fails, a loud warning is logged and
+  the request proceeds. Lets you deploy clients before SGX is live everywhere.
+- **`high`/`maximum`, `enforce=True` (target):** unavailable or failed
+  attestation ⇒ the request is refused with `AttestationError`, **no plaintext
+  fallback**.
+
+## Liveness (spec §7, optional)
+
+The default quote is bound to the key only and is cached server-side (proves
+binding, not recency). Set `liveness=True` to send a random 32-byte challenge per
+handshake; the server returns a fresh, uncached quote bound to
+`SHA-512(pubkey_der || challenge)`, proving the enclave is alive *now*. Liveness
+bypasses the client-side verification cache.
+
+## Self-hosting the verifier
+
+A self-hosted equivalent of Intel Trust Authority is a small service that runs
+the Intel DCAP QVL (`libsgx_dcap_quoteverify`) with a reachable PCCS, and exposes
+`POST /verify {quote} -> {token: <JWT>}` plus a JWKS endpoint. Veraison is an
+open-source option. Run that one service and point `JwtQuoteVerifier` at it.

nomyo/SecureCompletionClient.py

@@ -76,7 +76,7 @@ class SecureCompletionClient:
     - Response parsing
     """
 
-    def __init__(self, router_url: str = "https://api.nomyo.ai", allow_http: bool = False, secure_memory: bool = True, max_retries: int = 2):
+    def __init__(self, router_url: str = "https://api.nomyo.ai", allow_http: bool = False, secure_memory: bool = True, max_retries: int = 2, attestation_policy=None, quote_verifier=None):
         """
         Initialize the secure completion client.
 
@@ -87,6 +87,13 @@ class SecureCompletionClient:
             max_retries: Number of retries on retryable errors (429, 500, 502, 503, 504,
                          network errors). Uses exponential backoff. Default 2, matching
                          the OpenAI Python SDK default.
+            attestation_policy: Optional AttestationPolicy enabling SGX DCAP attestation
+                         verification before plaintext is sent. When None (default),
+                         attestation is disabled and behavior is unchanged.
+            quote_verifier: Optional QuoteVerifier used for the step-4 DCAP quote
+                         verification. Required to actually verify a quote; when None
+                         and a policy is set, attestation is treated as unverifiable
+                         (warn-and-proceed unless policy.enforce is True).
         """
         self.router_url = router_url.rstrip('/')
         self.private_key = None
@@ -96,6 +103,18 @@ class SecureCompletionClient:
         self._use_secure_memory = _SECURE_MEMORY_AVAILABLE and secure_memory
         self.max_retries = max_retries
 
+        # Optional SGX attestation: only construct the verifier when a policy is
+        # provided, so existing callers get byte-for-byte the previous behavior.
+        self._attestation = None
+        if attestation_policy is not None:
+            from .SgxAttestation import SgxAttestationVerifier
+            self._attestation = SgxAttestationVerifier(
+                router_url=self.router_url,
+                policy=attestation_policy,
+                verifier=quote_verifier,
+                allow_http=allow_http,
+            )
+
         # Validate HTTPS for security
         if not self.router_url.startswith("https://"):
             if allow_http:
@@ -337,11 +356,15 @@ class SecureCompletionClient:
         except Exception:
             raise ValueError("Failed to fetch server's public key")
 
-    async def _do_encrypt(self, payload_bytes: Union[bytes, bytearray], aes_key: Union[bytes, bytearray]) -> bytes:
+    async def _do_encrypt(self, payload_bytes: Union[bytes, bytearray], aes_key: Union[bytes, bytearray], server_public_key_pem: Optional[str] = None) -> bytes:
         """
         Core AES-256-GCM + RSA-OAEP encryption. Caller is responsible for
         memory protection of payload_bytes and aes_key before calling this.
         Accepts bytearray to avoid creating an unzeroed immutable bytes copy.
+
+        When ``server_public_key_pem`` is provided it is used directly (the
+        caller has already fetched — and possibly attested — that exact key);
+        otherwise the key is fetched here, preserving the previous behavior.
         """
         nonce = secrets.token_bytes(12)  # 96-bit nonce for GCM
         cipher = Cipher(
@@ -353,7 +376,8 @@ class SecureCompletionClient:
         ciphertext = encryptor.update(payload_bytes) + encryptor.finalize()
         tag = encryptor.tag
 
-        server_public_key_pem = await self.fetch_server_public_key()
+        if server_public_key_pem is None:
+            server_public_key_pem = await self.fetch_server_public_key()
         server_public_key = serialization.load_pem_public_key(
             server_public_key_pem.encode('utf-8'),
             backend=default_backend()
@@ -397,7 +421,7 @@ class SecureCompletionClient:
         logger.debug("Encrypted package size: %d bytes", len(package_json))
         return package_json
 
-    async def encrypt_payload(self, payload: Dict[str, Any]) -> bytes:
+    async def encrypt_payload(self, payload: Dict[str, Any], server_public_key_pem: Optional[str] = None) -> bytes:
         """
         Encrypt a payload using hybrid encryption (AES-256-GCM + RSA-OAEP).
 
@@ -406,6 +430,10 @@ class SecureCompletionClient:
 
         Args:
             payload: Dictionary containing the chat completion request
+            server_public_key_pem: Optional pre-fetched server public key PEM. When
+                provided the payload is encrypted to exactly this key (used after
+                SGX attestation so we encrypt to the attested key, with no
+                re-fetch); otherwise the key is fetched during encryption.
 
         Returns:
             Encrypted payload as bytes
@@ -441,11 +469,12 @@ class SecureCompletionClient:
                         with secure_bytearray(aes_key) as protected_aes_key:
                             return await self._do_encrypt(
                                 protected_payload.data,
-                                protected_aes_key.data
+                                protected_aes_key.data,
+                                server_public_key_pem
                             )
                 else:
                     logger.warning("Secure memory not available, using standard encryption")
-                    return await self._do_encrypt(payload_json, aes_key)
+                    return await self._do_encrypt(payload_json, aes_key, server_public_key_pem)
             finally:
                 del aes_key
 
@@ -641,8 +670,20 @@ class SecureCompletionClient:
                     f"Must be one of: {', '.join(valid_tiers)}"
                 )
 
-        # Step 1: Encrypt the payload
-        encrypted_payload = await self.encrypt_payload(payload)
+        # Step 0: SGX attestation handshake (when enabled). Fetch the server
+        # public key once, verify the enclave + key binding BEFORE building any
+        # plaintext, then encrypt to exactly that attested key. In enforce mode a
+        # failure raises here, so plaintext is never constructed or sent.
+        server_public_key_pem: Optional[str] = None
+        if self._attestation is not None:
+            server_public_key_pem = await self.fetch_server_public_key()
+            pubkey_der = self._spki_der_from_pem(server_public_key_pem)
+            await self._attestation.ensure_verified(
+                server_public_key_pem, pubkey_der, security_tier
+            )
+
+        # Step 1: Encrypt the payload (to the attested key when attestation ran)
+        encrypted_payload = await self.encrypt_payload(payload, server_public_key_pem)
 
         # Step 2: Prepare headers
         headers = {
@@ -798,6 +839,24 @@ class SecureCompletionClient:
                 logger.exception("Unexpected error in send_secure_request")
                 raise APIConnectionError("Request failed due to an unexpected error")
 
+    @staticmethod
+    def _spki_der_from_pem(pubkey_pem: str) -> bytes:
+        """
+        Return the X.509 SubjectPublicKeyInfo DER bytes for the server public key.
+
+        This is exactly the value the SGX report_data binding is computed over
+        (SHA-512 of these bytes); see doc/CLIENT_SGX_ATTESTATION.md §2.1/§2.3. We
+        re-serialize via the parsed key (rather than base64-decoding the PEM body
+        directly) so the bytes are canonical SPKI DER regardless of PEM wrapping.
+        """
+        key = serialization.load_pem_public_key(
+            pubkey_pem.encode('utf-8'), backend=default_backend()
+        )
+        return key.public_bytes(
+            encoding=serialization.Encoding.DER,
+            format=serialization.PublicFormat.SubjectPublicKeyInfo,
+        )
+
     def _validate_rsa_key(self, key, key_type: str = "private") -> None:
         """
         Validate that a key is a valid RSA key with appropriate size.

nomyo/SgxAttestation.py

@@ -0,0 +1,417 @@
+"""
+SGX DCAP attestation verification for the NOMYO secure-completion client.
+
+This implements the client side of ``doc/CLIENT_SGX_ATTESTATION.md`` (mirrored in
+the router repo): before sending any plaintext to a ``high``/``maximum`` tier
+endpoint, prove that the RSA public key the client is about to encrypt to was
+generated *inside* a genuine Intel SGX enclave running our measured code.
+
+The single most important rule (spec §1): verify the quote and the key binding
+**before** transmitting plaintext. A failed verification must block the request
+in ``enforce`` mode, with no plaintext fallback.
+
+Architecture
+------------
+Verifying a DCAP quote (ECDSA signature + PCK certificate chain + TCB/revocation)
+is intentionally *not* done here — it must not be hand-rolled. Instead this module
+defines a small :class:`QuoteVerifier` interface and performs everything around
+it (fetch, policy checks, the constant-time key binding, session caching). The
+``step 4`` quote verification is delegated to an injected verifier:
+
+* :class:`JwtQuoteVerifier` — opt-in, talks to a remote verification service
+  (Intel Trust Authority, Veraison, or a self-hosted QVL wrapper) that returns a
+  signed JWT of appraised claims. Requires the ``attestation`` extra (``pyjwt``).
+* :class:`CallableQuoteVerifier` — wrap any user-supplied callable.
+* Or any object implementing :class:`QuoteVerifier`.
+
+With no verifier configured the orchestrator treats the quote as unverifiable and
+applies the §6 rollout policy (warn-and-proceed unless ``enforce``).
+"""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import hmac
+import logging
+import secrets
+from dataclasses import dataclass, field
+from typing import Any, Awaitable, Callable, Dict, FrozenSet, Optional, Union
+
+import httpx
+
+from .SecureCompletionClient import SecurityError
+
+logger = logging.getLogger(__name__)
+
+# SGX report_data is a fixed 64-byte field; the binding is the full SHA-512 digest
+# of the server public key DER (optionally concatenated with a client challenge),
+# with no truncation or padding (spec §2.3).
+REPORT_DATA_LEN = 64
+
+
+class AttestationError(SecurityError):
+    """Raised when SGX attestation verification fails in ``enforce`` mode.
+
+    Subclasses :class:`SecurityError` so existing ``except SecurityError`` blocks
+    treat a failed attestation as the security failure it is.
+    """
+    pass
+
+
+@dataclass(frozen=True)
+class VerifiedQuote:
+    """Trusted result of verifying a DCAP quote.
+
+    These values come from *inside* the hardware-signed quote (as returned by a
+    :class:`QuoteVerifier`) — never from the untrusted ``/pki/attestation`` JSON.
+    """
+    mrenclave: str          # hex (lowercase)
+    mrsigner: str           # hex (lowercase)
+    isv_prod_id: int
+    isv_svn: int
+    tcb_status: str
+    report_data: bytes      # 64 bytes, the SGX report_data field
+
+
+@dataclass
+class AttestationPolicy:
+    """Client-side attestation policy (spec §5).
+
+    Ship the pinned values as configuration, not hard-coded constants: MRENCLAVE
+    changes whenever the enclave build or its file paths change, so pin a *set* to
+    allow rolling new builds.
+    """
+    pinned_mrenclave: FrozenSet[str] = field(default_factory=frozenset)
+    pinned_mrsigner: Optional[str] = None
+    pinned_prod_id: Optional[int] = None
+    min_isv_svn: Optional[int] = None
+    allowed_tcb_statuses: FrozenSet[str] = field(
+        default_factory=lambda: frozenset({"UpToDate"})
+    )
+    enforce: bool = False
+    require_attestation_for_tier: FrozenSet[str] = field(
+        default_factory=lambda: frozenset({"high", "maximum"})
+    )
+    liveness: bool = False
+
+    def __post_init__(self) -> None:
+        # Accept any iterable for the set-typed fields and normalise pins to
+        # lowercase hex so comparisons against verifier output are case-stable.
+        self.pinned_mrenclave = frozenset(
+            m.lower() for m in self.pinned_mrenclave
+        )
+        if self.pinned_mrsigner is not None:
+            self.pinned_mrsigner = self.pinned_mrsigner.lower()
+        self.allowed_tcb_statuses = frozenset(self.allowed_tcb_statuses)
+        self.require_attestation_for_tier = frozenset(
+            self.require_attestation_for_tier
+        )
+
+
+class QuoteVerifier:
+    """Interface for step-4 DCAP quote verification.
+
+    Implementations must verify the ECDSA signature, the PCK certificate chain to
+    the Intel root, and revocation/TCB, then return the trusted parsed values.
+    Raise on any verification failure.
+    """
+
+    async def verify(self, quote: bytes) -> VerifiedQuote:  # pragma: no cover - interface
+        raise NotImplementedError
+
+
+# A user callable may be sync or async and returns a VerifiedQuote.
+_VerifierCallable = Callable[[bytes], Union[VerifiedQuote, Awaitable[VerifiedQuote]]]
+
+
+class CallableQuoteVerifier(QuoteVerifier):
+    """Adapt a (sync or async) callable into a :class:`QuoteVerifier`."""
+
+    def __init__(self, func: _VerifierCallable) -> None:
+        self._func = func
+
+    async def verify(self, quote: bytes) -> VerifiedQuote:
+        import inspect
+        result = self._func(quote)
+        if inspect.isawaitable(result):
+            result = await result
+        if not isinstance(result, VerifiedQuote):
+            raise AttestationError(
+                "quote verifier callable must return a VerifiedQuote"
+            )
+        return result
+
+
+# Default mapping from JWT claim names to VerifiedQuote fields. Concrete services
+# differ (Intel Trust Authority vs Veraison vs a custom QVL wrapper), so this is
+# overridable per JwtQuoteVerifier instance.
+DEFAULT_CLAIM_MAP: Dict[str, str] = {
+    "mrenclave": "sgx_mrenclave",
+    "mrsigner": "sgx_mrsigner",
+    "isv_prod_id": "sgx_isvprodid",
+    "isv_svn": "sgx_isvsvn",
+    "tcb_status": "tcb_status",
+    "report_data": "sgx_report_data",
+}
+
+
+class JwtQuoteVerifier(QuoteVerifier):
+    """Verify a quote via a remote service that returns a signed JWT (spec §4A).
+
+    POSTs the base64 quote to ``verify_url``; the service replies with a JWT whose
+    signature is validated against ``jwks_url`` (and optional issuer/audience).
+    The validated claims are mapped to a :class:`VerifiedQuote`.
+
+    Requires the optional ``pyjwt`` dependency::
+
+        pip install nomyo[attestation]
+    """
+
+    def __init__(
+        self,
+        verify_url: str,
+        jwks_url: str,
+        *,
+        issuer: Optional[str] = None,
+        audience: Optional[str] = None,
+        claim_map: Optional[Dict[str, str]] = None,
+        algorithms: Optional[list] = None,
+        timeout: float = 30.0,
+        verify_ssl: bool = True,
+    ) -> None:
+        self.verify_url = verify_url
+        self.jwks_url = jwks_url
+        self.issuer = issuer
+        self.audience = audience
+        self.claim_map = dict(claim_map) if claim_map else dict(DEFAULT_CLAIM_MAP)
+        self.algorithms = algorithms or ["RS256", "ES256", "ES384"]
+        self.timeout = timeout
+        self.verify_ssl = verify_ssl
+
+    @staticmethod
+    def _require_jwt():
+        try:
+            import jwt  # PyJWT
+            return jwt
+        except ImportError as e:  # pragma: no cover - exercised via monkeypatch in tests
+            raise AttestationError(
+                "JwtQuoteVerifier requires the 'pyjwt' package. "
+                "Install the attestation extra: pip install nomyo[attestation]"
+            ) from e
+
+    async def verify(self, quote: bytes) -> VerifiedQuote:
+        jwt = self._require_jwt()
+
+        async with httpx.AsyncClient(timeout=self.timeout, verify=self.verify_ssl) as client:
+            resp = await client.post(
+                self.verify_url,
+                json={"quote": base64.b64encode(quote).decode("ascii")},
+            )
+        if resp.status_code != 200:
+            raise AttestationError(
+                f"quote verification service returned HTTP {resp.status_code}"
+            )
+        try:
+            token = resp.json()["token"]
+        except (ValueError, KeyError, TypeError) as e:
+            raise AttestationError("verifier response missing JWT 'token'") from e
+
+        # Validate the JWT signature against the service's published JWKS.
+        try:
+            jwk_client = jwt.PyJWKClient(self.jwks_url)
+            signing_key = jwk_client.get_signing_key_from_jwt(token)
+            options = {"verify_aud": self.audience is not None}
+            claims = jwt.decode(
+                token,
+                signing_key.key,
+                algorithms=self.algorithms,
+                issuer=self.issuer,
+                audience=self.audience,
+                options=options,
+            )
+        except Exception as e:
+            raise AttestationError(f"JWT validation failed: {e}") from e
+
+        return self._claims_to_verified_quote(claims)
+
+    def _claims_to_verified_quote(self, claims: Dict[str, Any]) -> VerifiedQuote:
+        cm = self.claim_map
+        try:
+            report_data_hex = claims[cm["report_data"]]
+            report_data = bytes.fromhex(report_data_hex)
+            return VerifiedQuote(
+                mrenclave=str(claims[cm["mrenclave"]]).lower(),
+                mrsigner=str(claims[cm["mrsigner"]]).lower(),
+                isv_prod_id=int(claims[cm["isv_prod_id"]]),
+                isv_svn=int(claims[cm["isv_svn"]]),
+                tcb_status=str(claims[cm["tcb_status"]]),
+                report_data=report_data,
+            )
+        except (KeyError, ValueError, TypeError) as e:
+            raise AttestationError(
+                f"verifier JWT missing/invalid expected claim: {e}"
+            ) from e
+
+
+class SgxAttestationVerifier:
+    """Orchestrates the §3 verification algorithm around a :class:`QuoteVerifier`.
+
+    One instance per :class:`SecureCompletionClient`. Caches the
+    ``(pubkey -> verified)`` result per server for the session to avoid
+    re-verifying on every call; re-verifies if the server pubkey changes.
+    Liveness mode bypasses the cache (freshness).
+    """
+
+    def __init__(
+        self,
+        router_url: str,
+        policy: AttestationPolicy,
+        verifier: Optional[QuoteVerifier] = None,
+        allow_http: bool = False,
+    ) -> None:
+        self.router_url = router_url.rstrip("/")
+        self.policy = policy
+        self.verifier = verifier
+        self.allow_http = allow_http
+        # Set of sha256(pubkey_der) hex that verified OK this session.
+        self._verified_keys: set = set()
+
+    async def ensure_verified(
+        self,
+        pubkey_pem: str,
+        pubkey_der: bytes,
+        security_tier: Optional[str],
+    ) -> Optional[VerifiedQuote]:
+        """Run the §3 algorithm for the given server public key + tier.
+
+        Returns the :class:`VerifiedQuote` on success, ``None`` when attestation
+        is not required for the tier or when it failed but ``enforce`` is off.
+        Raises :class:`AttestationError` when verification fails and
+        ``policy.enforce`` is True — in which case the caller MUST NOT send
+        plaintext.
+        """
+        tier = security_tier or "standard"
+        if tier not in self.policy.require_attestation_for_tier:
+            # e.g. standard tier: attestation not expected — proceed (spec §6).
+            return None
+
+        key_id = hashlib.sha256(pubkey_der).hexdigest()
+        if not self.policy.liveness and key_id in self._verified_keys:
+            logger.debug("SGX attestation: using cached verification for key %s", key_id[:16])
+            return None
+
+        if self.verifier is None:
+            return self._fail(
+                "no quote verifier configured (cannot perform step-4 DCAP "
+                "verification)"
+            )
+
+        challenge = secrets.token_bytes(32) if self.policy.liveness else b""
+
+        att = await self._fetch_attestation(challenge)
+        if att is None:
+            return self._fail("could not fetch attestation from server")
+        if not att.get("is_available"):
+            reason = att.get("reason", "unknown")
+            return self._fail(f"attestation unavailable: {reason}")
+
+        quote_b64 = att.get("quote_b64")
+        if not quote_b64:
+            return self._fail("attestation response missing quote_b64")
+        try:
+            quote = base64.b64decode(quote_b64)
+        except Exception:
+            return self._fail("attestation quote_b64 is not valid base64")
+
+        # Step 4: delegate the cryptographic quote verification.
+        try:
+            v = await self.verifier.verify(quote)
+        except AttestationError as e:
+            return self._fail(f"quote verification failed: {e}")
+        except Exception as e:
+            return self._fail(f"quote verification error: {type(e).__name__}: {e}")
+
+        # Step 5: policy checks.
+        policy_error = self._check_policy(v)
+        if policy_error is not None:
+            return self._fail(policy_error)
+
+        # Step 6: key-binding check (constant-time).
+        expected = hashlib.sha512(pubkey_der + challenge).digest()
+        if len(v.report_data) != REPORT_DATA_LEN or not hmac.compare_digest(
+            expected, v.report_data
+        ):
+            return self._fail(
+                "key binding mismatch: report_data does not match "
+                "SHA-512(server_pubkey_der[ || challenge]) — the attested key is "
+                "not the key being encrypted to"
+            )
+
+        # Step 7: success — cache (non-liveness only) and proceed.
+        if not self.policy.liveness:
+            self._verified_keys.add(key_id)
+        logger.info(
+            "SGX attestation verified: mrenclave=%s tcb=%s tier=%s",
+            v.mrenclave[:16], v.tcb_status, tier,
+        )
+        return v
+
+    def _check_policy(self, v: VerifiedQuote) -> Optional[str]:
+        """Return an error string if any policy check fails, else None."""
+        if v.tcb_status not in self.policy.allowed_tcb_statuses:
+            return (
+                f"TCB status '{v.tcb_status}' not in allowed "
+                f"{sorted(self.policy.allowed_tcb_statuses)}"
+            )
+        if self.policy.pinned_mrenclave:
+            if v.mrenclave.lower() not in self.policy.pinned_mrenclave:
+                return f"MRENCLAVE {v.mrenclave} not in pinned set"
+        else:
+            logger.warning(
+                "SGX attestation: no pinned_mrenclave configured — enclave "
+                "identity is NOT verified (configure AttestationPolicy.pinned_mrenclave)"
+            )
+        if self.policy.pinned_mrsigner is not None:
+            if v.mrsigner.lower() != self.policy.pinned_mrsigner:
+                return f"MRSIGNER {v.mrsigner} != pinned {self.policy.pinned_mrsigner}"
+        if self.policy.pinned_prod_id is not None:
+            if v.isv_prod_id != self.policy.pinned_prod_id:
+                return f"isv_prod_id {v.isv_prod_id} != pinned {self.policy.pinned_prod_id}"
+        if self.policy.min_isv_svn is not None:
+            if v.isv_svn < self.policy.min_isv_svn:
+                return f"isv_svn {v.isv_svn} < minimum {self.policy.min_isv_svn}"
+        return None
+
+    async def _fetch_attestation(self, challenge: bytes) -> Optional[Dict[str, Any]]:
+        """GET /pki/attestation; return parsed JSON or None on transport error."""
+        if not self.router_url.startswith("https://") and not self.allow_http:
+            raise AttestationError(
+                "Attestation must be fetched over HTTPS to prevent MITM attacks "
+                "(initialize with allow_http=True only for local development)."
+            )
+        url = f"{self.router_url}/pki/attestation"
+        params = {"challenge": challenge.hex()} if challenge else None
+        verify_ssl = self.router_url.startswith("https://")
+        try:
+            async with httpx.AsyncClient(timeout=900.0, verify=verify_ssl) as client:
+                resp = await client.get(url, params=params)
+            if resp.status_code != 200:
+                logger.warning(
+                    "SGX attestation: /pki/attestation returned HTTP %d", resp.status_code
+                )
+                return None
+            return resp.json()
+        except (httpx.HTTPError, ValueError) as e:
+            logger.warning("SGX attestation: failed to fetch attestation: %s", e)
+            return None
+
+    def _fail(self, reason: str) -> None:
+        """Apply the §6 rollout policy: hard-fail in enforce, else warn + proceed."""
+        if self.policy.enforce:
+            raise AttestationError(f"SGX attestation failed: {reason}")
+        logger.warning(
+            "SGX attestation: %s — proceeding because enforce=False (rollout mode)",
+            reason,
+        )
+        return None

nomyo/__init__.py

@@ -16,6 +16,15 @@ from .SecureCompletionClient import (
     ServerError,
     ServiceUnavailableError
 )
+from .SgxAttestation import (
+    AttestationError,
+    AttestationPolicy,
+    VerifiedQuote,
+    QuoteVerifier,
+    CallableQuoteVerifier,
+    JwtQuoteVerifier,
+    SgxAttestationVerifier,
+)
 
 __all__ = [
     'SecureChatCompletion',
@@ -28,6 +37,13 @@ __all__ = [
     'RateLimitError',
     'ServerError',
     'ServiceUnavailableError',
+    'AttestationError',
+    'AttestationPolicy',
+    'VerifiedQuote',
+    'QuoteVerifier',
+    'CallableQuoteVerifier',
+    'JwtQuoteVerifier',
+    'SgxAttestationVerifier',
 ]
 
 # Import secure memory module if available; extend __all__ only for what was imported
@@ -51,6 +67,6 @@ try:
 except ImportError:
     pass
 
-__version__ = "0.2.9"
+__version__ = "0.3.0"
 __author__ = "NOMYO AI"
 __license__ = "Apache-2.0"

nomyo/nomyo.py

@@ -52,7 +52,7 @@ class SecureChatCompletion:
         ```
     """
 
-    def __init__(self, base_url: str = "https://api.nomyo.ai", allow_http: bool = False, api_key: Optional[str] = None, secure_memory: bool = True, key_dir: Optional[str] = None, max_retries: int = 2):
+    def __init__(self, base_url: str = "https://api.nomyo.ai", allow_http: bool = False, api_key: Optional[str] = None, secure_memory: bool = True, key_dir: Optional[str] = None, max_retries: int = 2, attestation_policy=None, quote_verifier=None):
         """
         Initialize the secure chat completion client.
 
@@ -70,13 +70,22 @@ class SecureChatCompletion:
                      generated in memory for this session only.
             max_retries: Number of retries on retryable errors (429, 500, 502, 503, 504,
                         network errors). Uses exponential backoff. Default 2.
+            attestation_policy: Optional AttestationPolicy enabling SGX DCAP attestation
+                        verification before any plaintext is sent (see doc/attestation.md).
+                        When None (default) attestation is disabled.
+            quote_verifier: Optional QuoteVerifier performing the step-4 DCAP quote
+                        verification (e.g. JwtQuoteVerifier). Required to actually
+                        verify; without it a set policy treats attestation as
+                        unverifiable (warn-and-proceed unless policy.enforce).
         """
-        self.client = SecureCompletionClient(router_url=base_url, allow_http=allow_http, secure_memory=secure_memory, max_retries=max_retries)
+        self.client = SecureCompletionClient(router_url=base_url, allow_http=allow_http, secure_memory=secure_memory, max_retries=max_retries, attestation_policy=attestation_policy, quote_verifier=quote_verifier)
         self._keys_initialized = False
         self._keys_lock = asyncio.Lock()
         self.api_key = api_key
         self._key_dir = key_dir
         self._secure_memory_enabled = secure_memory
+        self._attestation_policy = attestation_policy
+        self._quote_verifier = quote_verifier
 
         if secure_memory and not _SECURE_MEMORY_AVAILABLE:
             import warnings
@@ -211,6 +220,8 @@ class SecureChatCompletion:
                 api_key=self.api_key,
                 secure_memory=self._secure_memory_enabled,
                 key_dir=self._key_dir,
+                attestation_policy=self._attestation_policy,
+                quote_verifier=self._quote_verifier,
             )
             instance = temp_client
         else:

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "nomyo"
-version = "0.2.9"
+version = "0.3.0"
 description = "OpenAI-compatible secure chat client with end-to-end encryption for NOMYO Inference Endpoints"
 authors = [
     {name = "NOMYO.AI", email = "ichi@nomyo.ai"},
@@ -40,6 +40,12 @@ dependencies = [
     "typing_extensions==4.15.0",
 ]
 
+[project.optional-dependencies]
+# SGX attestation via a remote JWT-returning verification service (Intel Trust
+# Authority / Veraison / self-hosted QVL wrapper). cryptography (core dep) covers
+# the JWKS RSA/EC signature validation; pyjwt provides JWT + JWKS handling.
+attestation = ["pyjwt>=2.8,<3"]
+
 [project.urls]
 Homepage = "https://www.nomyo.ai"
 Documentation = "https://bitfreedom.net/code/nomyo-ai/nomyo/wiki/NOMYO-Secure-Client-Documentation"