diff --git a/CHANGELOG.md b/CHANGELOG.md index 17822dba..0ffa4f73 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,26 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- **On SEV-SNP, TDX and Azure CVM the policy actually in force was committed to nothing (#552, follow-up to #432).** `gateway_measurement()` folds the installed code, the policy bundle and the effective configuration into one digest, and #432 extended it into a `TPM_NT_EXTEND` NV index and had the TPM certify it. That path is validated on real Azure Trusted Launch vTPM hardware, and it ran for the `tpm` provider only: `_measure_gateway` returned early for every other provider. + + The stated reason was that "SEV-SNP and TDX commit their own binding through the report's fields". That is true and it is not equivalent. Those fields carry the **launch** measurement, which is fixed at boot and does not move when the Cedar bundle reloads mid-session through `PolicyEvaluator._maybe_reload()`. So on exactly the platforms whose whole premise is hardware-rooted policy enforcement, nothing signed said which policy was running. + + No new commitment scheme was invented. `make_measurement_bound_nonce(tee_public_key, measurement_digest)` puts the already-validated digest into the second half of the attestation nonce, in the same 64-byte layout `make_audit_bound_nonce` already uses: `jwk_thumbprint(pubkey) (32) || measurement_digest (32)`. `gateway_measurement().digest` is a raw 32-byte SHA-256, so it drops in unreshaped and a verifier compares it against a digest it recomputes rather than against a hash of one. The `tpm` provider is deliberately not in the set: its NV index keeps an append-only history that `report_data` cannot. + + This replaces the 32 random salt bytes on those providers, and freshness survives the change for a reason worth stating rather than assuming: the gateway generates a new signing key on every start, so `report_data[:32]` still differs between two starts of byte-identical code, policy and config. + + `report_data` carries one value and no history, so a report built before a bundle reload still looks well-formed on its own, and nothing in it says whether it is current. The gateway now re-attests on **every** policy-bundle reload, including one that finds the bundle unchanged, wired from `PolicyEvaluator._maybe_reload` through a new `on_reload` hook to `refresh_measurement_binding`. Re-signing an unchanged digest is not redundant: the same digest signed now and that digest signed an hour ago are different assertions, and only the latest report reaches a verifier. The cost is bounded by `policy_reload_interval_seconds` rather than by request rate, because `PolicyStore.reload_if_stale` stamps its clock before the attempt, and it is zero in the default configuration where reloading is off. + + A refresh that cannot re-attest logs and keeps the previous report rather than refusing traffic: the stale binding no longer matches the recomputed digest, so a verifier rejects the claim, and failing closed here would trade a detectable weakness for an outage. That is the same trade AUDIT-006 already makes for a failed per-session attestation. + + Recompute-and-compare is what catches a gateway that did not refresh, so it is now a check rather than a note in the spec. `verify_trace_claim` gains an optional `expected_gateway_measurement` (raw 32 bytes, hex, or `sha256:`-prefixed) and a `MEASUREMENT_NOT_BOUND` failure reason. It is opt-in because the expected digest must be an out-of-band trust input like `ApprovedHashes`: a check that read it out of the claim would be asking the claim to vouch for itself. The digest is compared directly rather than re-hashed, which is the one way it differs from the AUDIT-006 check beside it. + + **Known limit, stated rather than implied.** The binding is on the gateway's **startup** report. A TRACE Claim for a session carries the per-session report when one was produced, and AUDIT-006 already commits the audit-chain root in the same `report_data[32:64]`. So on the normal session path a verifier sees the chain-root commitment, not the measurement, and step 7c applies to claims that fall back to the startup report. Carrying both in one 64-byte field is not possible as the layout stands, and #552 explicitly scopes the audit-chain-root binding out, so this is left for the issue that takes that on. + + **Behaviour change:** an unmeasurable gateway on `sev-snp`, `tdx` or `azure-cvm-sev-snp` is now fatal at startup in production, as it has been on `tpm` since #432, because extending the measurement to a platform extends the consequence of not having one. `CMCP_DEV_MODE=1` still downgrades it to a warning, which is what an editable install with no `RECORD` metadata needs. + + Not addressed here, per the issue: binding the audit-chain root, which is session activity rather than gateway identity and which AUDIT-006 already owns in the per-session report's `report_data[32:64]`, and validation on real SEV-SNP and TDX silicon. The contract lands with software-only proof: the round trip through `SoftwareOnlyProvider` composes the nonce, takes the report, and has an independent recompute match both halves, with a stale pre-reload measurement correctly rejected. + - **An approval record stopped verifying once its approvals expired (#533, follow-up to #531).** `verify_catalog_change` judged `approved_at` and `expires_at` against `time.time()`, so a record that was valid when the catalog was approved became permanently unverifiable, and the chain #517 exists to let an auditor replay could not be replayed. That made the record an authorization token with a lifetime rather than a provenance record. The interval is now what it says it is: an assertion about when the signature could have been produced. The caller passes `validity_instant`, a pinned checkpoint or transparency-receipt timestamp where it has one, and where it passes nothing each approval is judged at its own `approved_at`. Requiring a pin instead would mean an auditor cannot verify a record without also holding the pin, which is a worse default than the one it replaces. The `now` parameter is gone with the wall clock, and `time` is no longer imported. diff --git a/docs/spec/attestation.md b/docs/spec/attestation.md index ef47b788..e9aa3f05 100644 --- a/docs/spec/attestation.md +++ b/docs/spec/attestation.md @@ -281,7 +281,7 @@ nonce = JWK_thumbprint(tee_public_key) (32 bytes) || random_salt (32 bytes) ``` - `JWK_thumbprint(tee_public_key)`: the RFC 7638 JWK Thumbprint of the Ed25519 public key: SHA-256 over the canonical JSON of the required OKP members in lexicographic order (`crv`, `kty`, `x`). This is re-derivable by any verifier from `cnf.jwk.x`. -- `random_salt`: 32 random bytes generated once per enclave startup, so two enclave instances produce distinct nonces even with the same key (e.g. blue-green deploy). +- `random_salt`: 32 random bytes generated once per enclave startup, so two enclave instances produce distinct nonces even with the same key (e.g. blue-green deploy). On SEV-SNP, TDX and Azure CVM this half carries the gateway measurement instead; see §3.3.2. - The 64-byte value is passed as the `report_data` / `user_data` / `reportdata` / `qualifying_data` field when requesting the hardware attestation report. The field name varies by provider; the semantic is the same: a caller-supplied value included in the signed measurement. Verifier check (key binding, CRYPTO-001): @@ -294,6 +294,44 @@ assert actual_nonce[:32] == expected_fingerprint A TRACE Claim whose `cnf.jwk` public key was substituted after attestation fails this check, because the embedded `report_data` (hardware-signed) will not match the re-derived thumbprint. A claim produced by a different enclave instance carries a different key (and salt), so it fails too. +#### 3.3.2 Measurement binding (SEV-SNP, TDX, Azure CVM) + +On platforms whose hardware report carries only a **launch** measurement, the 32-byte salt is replaced by the gateway measurement digest: + +``` +nonce = JWK_thumbprint(tee_public_key) (32 bytes) || gateway_measurement.digest (32 bytes) +``` + +`gateway_measurement.digest` is the SHA-256 over the installed code, the policy bundle and the effective configuration defined for the TPM tier (see `docs/spec/tpm-security-model.md`). It is already a raw 32-byte SHA-256, so it occupies the second half unreshaped and a verifier compares it against a digest it recomputes, not against a hash of one. + +**Why the launch measurement is not sufficient.** `SNP_REPORT.measurement` and TDX's `MRTD` are fixed at boot. They do not move when the Cedar bundle reloads mid-session, so without this binding the policy actually in force is committed to nothing. The TPM tier solves the same problem with a `TPM_NT_EXTEND` NV index; these platforms have no such index. + +**Applies to** the `sev-snp`, `tdx` and `azure-cvm-sev-snp` providers. The `tpm` provider keeps the random salt of §3.3, because its measurement is committed by the NV index instead, which keeps an append-only history that `report_data` does not. + +**Freshness.** The salt is gone but freshness is not: the gateway generates a new signing key on every start, so `report_data[:32]` still differs between two starts of byte-identical code, policy and config. + +**Refreshed on every policy-bundle reload.** `report_data` holds one value and no history, so a report produced before a bundle reload still looks well-formed on its own, and nothing in it says whether it is current. The gateway re-attests on **every** reload, including a reload that finds the bundle unchanged: the same digest signed now and that digest signed an hour ago are different assertions, and only the latest report reaches a verifier. The hook is `PolicyEvaluator._maybe_reload` calling `refresh_measurement_binding`, fired whenever `PolicyStore.reload_if_stale` reports that the bundle was re-read. + +The cost is bounded by `policy_reload_interval_seconds`, not by request rate, because `reload_if_stale` stamps its clock before the attempt. It is zero in the default configuration, where reloading is off. + +A gateway that fails to refresh is caught verifier-side, not runtime-side: + +``` +expected = gateway_measurement_digest_supplied_by_the_verifier +actual_nonce = base64url_decode(trace.runtime.nonce) +assert actual_nonce[32:64] == expected +``` + +The digest is compared directly, not re-hashed: it is already a raw 32-byte SHA-256. This is the one way the check differs from the AUDIT-006 one below it, which commits `SHA-256(chain_root)`. + +`cmcp_verify.verify_trace_claim` implements this as step 7c, enabled by passing `expected_gateway_measurement` (raw 32 bytes, hex, or `sha256:`-prefixed hex). It is opt-in because the expected value has to be an out-of-band trust input, like `ApprovedHashes` or `trusted_ark_pem`: a check that read the expected digest out of the claim would be asking the claim to vouch for itself. A mismatch is `MEASUREMENT_NOT_BOUND` and is fatal in software-only mode too, since the digest is computed the same way there and a mismatch is a real disagreement about what is running. + +A re-attestation that fails is logged and the previous report is kept rather than the gateway refusing traffic; the stale binding fails the check above, so the weakness is detectable rather than silent. + +**Which report carries it.** This binding is on the gateway's **startup** report. A TRACE Claim for a session carries the per-session report instead when one was produced (AUDIT-006), and that report commits the audit-chain root in the same bytes. The startup report reaches a claim only where no per-session report was produced. Committing both gateway identity and session activity in one 64-byte field is not possible as the layout stands, and #552 scopes the audit-chain-root binding out. + +**Not the per-session report.** AUDIT-006 (§3.3.1 and below) puts `SHA-256(chain_root)` in `report_data[32:64]` of the *per-session* report. This section governs the *startup* report. The two are different reports and do not contend for the field. + **Session binding** is carried separately, by `gateway.session_id` inside the Ed25519-signed claim body: not by the nonce. The hardware report is generated once per enclave instance at startup, before any session exists, so it cannot bind a specific `session_id`. Because the signature covers `session_id`, a claim cannot be presented under a different session without breaking verification. See §3.3.1. #### 3.3.1 Session binding diff --git a/src/cmcp_runtime/cli.py b/src/cmcp_runtime/cli.py index ab1f1d87..e1f91f62 100644 --- a/src/cmcp_runtime/cli.py +++ b/src/cmcp_runtime/cli.py @@ -54,6 +54,7 @@ def build_server(ctx: RuntimeContext) -> MCPServer: from cmcp_runtime.mcp.server import MCPServer from cmcp_runtime.policy.evaluator import PolicyEvaluator from cmcp_runtime.session.manager import SessionManager + from cmcp_runtime.tee.report_binding import refresh_measurement_binding # Resolve provider string to canonical platform name for Cedar context. # Falls back to the raw provider string if not in the map (e.g. future providers). @@ -65,7 +66,16 @@ def build_server(ctx: RuntimeContext) -> MCPServer: # chain is backed by the durable SQLite store and TEE-anchored at creation. session_manager = SessionManager(ctx) session, audit_chain = session_manager.create_session() - policy_evaluator = PolicyEvaluator(bundle=ctx.policy_bundle, config=ctx.config) + # #552: a policy hot-reload changes what the gateway is running, and on SEV-SNP, + # TDX and Azure CVM that fact lives only in the current attestation report's + # report_data. Wire the reload to a re-attestation so the committed measurement + # is the live one; without this the binding is correct at startup and stale from + # the first reload onwards. A no-op on every other provider. + policy_evaluator = PolicyEvaluator( + bundle=ctx.policy_bundle, + config=ctx.config, + on_reload=lambda: refresh_measurement_binding(ctx), + ) proxy = CMCPProxy( catalog=ctx.catalog, policy_evaluator=policy_evaluator, diff --git a/src/cmcp_runtime/policy/evaluator.py b/src/cmcp_runtime/policy/evaluator.py index cb88354a..44417319 100644 --- a/src/cmcp_runtime/policy/evaluator.py +++ b/src/cmcp_runtime/policy/evaluator.py @@ -3,6 +3,7 @@ from __future__ import annotations import logging +from collections.abc import Callable from dataclasses import dataclass from typing import TYPE_CHECKING, Any @@ -50,8 +51,20 @@ class is instantiated. CedarBackend receives the already-loaded policy content so the measured hash covers exactly the bytes that will be evaluated. """ - def __init__(self, bundle: PolicyBundle | PolicyStore, config: Config) -> None: + def __init__( + self, + bundle: PolicyBundle | PolicyStore, + config: Config, + # Return value is ignored, so the hook is free to report what it did. + on_reload: Callable[[], object] | None = None, + ) -> None: self._mode = config.attestation.enforcement_mode + # #552: called after every policy-bundle reload so the platform can re-commit + # what is now in force. On SEV-SNP, TDX and Azure CVM the gateway measurement + # lives in the attestation report's report_data, which has no append-only + # history, so the report is only ever as current as the last time it was + # produced. None where nothing needs telling. + self._on_reload = on_reload # #479: same effective vocabulary SessionManager derives from this same # Config, so a session's max_sensitivity and this sensitivity_level_int # can never disagree about what a custom label ranks as. @@ -87,7 +100,7 @@ def __init__(self, bundle: PolicyBundle | PolicyStore, config: Config) -> None: def _maybe_reload(self) -> None: """Check for a stale bundle and rebuild the CedarBackend if the hash changed.""" - self._store.reload_if_stale() + reloaded = self._store.reload_if_stale() bundle = self._store.bundle if bundle.bundle_hash != self._current_hash: combined_policy = "\n\n".join( @@ -97,6 +110,47 @@ def _maybe_reload(self) -> None: self._annotations = parse_policy_annotations(combined_policy) self._current_hash = bundle.bundle_hash logger.info("PolicyEvaluator backend refreshed: new_hash=%s", self._current_hash) + # #552 asks for a refresh on **every** policy-bundle reload, not only on the + # ones that moved the hash, so this sits outside the branch above. + # ``reload_if_stale`` returns True exactly when the bundle was re-read from + # disk, which is what "a reload" means here: False when reloading is off, + # when the interval has not elapsed, and when the read failed. + if reloaded: + self._notify_reload() + + def _notify_reload(self) -> None: + """Re-commit what is running after a policy-bundle reload (#552). + + Fires on every reload, including one that found the bundle unchanged. The + TPM tier can afford to skip those because its NV index accumulates history; + ``report_data`` holds one value and no history, so what a verifier gets is + only ever the last report the gateway produced. Re-signing on each reload is + what keeps that report an assertion about now rather than about whenever the + policy last happened to change. + + This runs on the enforcement path, so the tool call that observes the reload + pays for the re-attestation. It is bounded by the reload interval, not by + request rate: ``reload_if_stale`` stamps its clock before the attempt, so the + cost is one TEE call per ``policy_reload_interval_seconds``, and none at all + in the default configuration where reloading is off. + + A failing callback is logged and swallowed on purpose. The callback re-binds + the gateway measurement into a hardware report; if the TEE cannot produce one + right now, refusing traffic would trade a *detectable* weakness for an outage. + Stale report_data no longer matches the measurement a verifier recomputes, so + the claim is rejected at verification instead. This is the same trade AUDIT-006 + makes in SessionManager.create_session for a failed per-session attestation. + """ + if self._on_reload is None: + return + try: + self._on_reload() + except Exception: # noqa: BLE001 - enforcement must not depend on the TEE + logger.warning( + "#552: post-reload attestation hook failed; report_data still commits " + "the previous policy bundle and verification will reject it", + exc_info=True, + ) def _advice_for_deny(self, policy_ids: tuple[str, ...]) -> dict[str, str]: """ diff --git a/src/cmcp_runtime/startup.py b/src/cmcp_runtime/startup.py index 5b3cdbd1..1389a880 100644 --- a/src/cmcp_runtime/startup.py +++ b/src/cmcp_runtime/startup.py @@ -43,6 +43,10 @@ gateway_measurement, ) from cmcp_runtime.tee.nras import AppraisalResult, try_appraise +from cmcp_runtime.tee.report_binding import ( + binds_measurement_into_report_data, + measurement_bound_nonce_for, +) from cmcp_runtime.tee.spiffe import SpiffeClientResult, fetch_svid logger = logging.getLogger(__name__) @@ -140,51 +144,111 @@ def _fatal(code: str, message: str, **fields: Any) -> None: logger.critical("%s", entry) -def _measure_gateway( - config: Config, tee_provider: TEEProvider, nonce: bytes -) -> tuple[GatewayMeasurement | None, ExtendResult | None, bytes | None]: - """Measure the gateway into the TPM NV index and have the TPM certify it (#432). - - Returns ``(measurement, extend, evidence)``. ``evidence`` is the signed - ``TPM2_NV_Certify`` pair, or None when the platform provisions no certified - attestation key to sign with: the extend still happens and is still a local - integrity control, but it is not remote-verifiable, so it is not presented as - evidence. All three are None when the platform has no TPM at all, which is not a - failure: SEV-SNP and TDX commit their own binding through the report's fields. - - A TPM platform that cannot be measured is fatal in production and a warning in - dev mode, matching how ``CMCP_POLICY_HASH`` is handled. The dev-mode escape - matters in practice because an editable install has no ``RECORD`` metadata, so - the code digest is genuinely uncomputable there rather than merely inconvenient. +def _degrade_measurement(config: Config, exc: MeasurementUnavailable) -> None: + """Abort on an unmeasurable gateway, or warn and continue in dev mode. + + Fatal in production and a warning in dev mode, matching how ``CMCP_POLICY_HASH`` + is handled. The dev-mode escape matters in practice because an editable install + has no ``RECORD`` metadata, so the code digest is genuinely uncomputable there + rather than merely inconvenient. """ - if tee_provider.provider_name() != "tpm": - logger.debug( - "Gateway measurement skipped: provider %s does not use an NV extend index", - tee_provider.provider_name(), + if config.dev_mode: + logger.warning( + "Gateway measurement unavailable (%s): %s. Continuing because " + "CMCP_DEV_MODE is set; the platform will not attest what code is running.", + exc, + exc.detail or "", ) - return None, None, None + return + _fatal( + "MEASUREMENT_UNAVAILABLE", + f"the gateway could not be measured: {exc}", + detail=exc.detail or "", + action="startup_aborted", + ) + sys.exit(1) - def _degrade(exc: MeasurementUnavailable) -> tuple[None, None, None]: - if config.dev_mode: - logger.warning( - "Gateway measurement unavailable (%s): %s. Continuing because " - "CMCP_DEV_MODE is set; the TPM will not attest what code is running.", - exc, - exc.detail or "", - ) - return None, None, None - _fatal( - "MEASUREMENT_UNAVAILABLE", - f"the gateway could not be measured into the TPM: {exc}", - detail=exc.detail or "", - action="startup_aborted", + +def _measure_gateway( + config: Config, tee_provider: TEEProvider +) -> GatewayMeasurement | None: + """Compute the gateway measurement on platforms that commit it (#432, #552). + + Two tiers commit it, by two different mechanisms. The ``tpm`` tier extends it + into a ``TPM_NT_EXTEND`` NV index and certifies it (#432). SEV-SNP, TDX and + Azure CVM have no such index, so they bind the same digest into the attestation + nonce and the hardware signs it as ``report_data`` (#552) -- their own report + fields carry only a *launch* measurement, which does not move when the policy + bundle reloads mid-session. + + Returns None where neither applies, which is not a failure. An unmeasurable + gateway on a platform that does commit it is fatal in production; see + :func:`_degrade_measurement`. + """ + provider_name = tee_provider.provider_name() + if provider_name != "tpm" and not binds_measurement_into_report_data(provider_name): + logger.debug( + "Gateway measurement skipped: provider %s commits no gateway measurement", + provider_name, ) - sys.exit(1) + return None try: - measurement = gateway_measurement(config) + return gateway_measurement(config) except MeasurementUnavailable as exc: - return _degrade(exc) + _degrade_measurement(config, exc) + return None + + +def _attestation_nonce( + key_fingerprint: bytes, + signing_key: SigningKey, + tee_provider: TEEProvider, + measurement: GatewayMeasurement | None, +) -> bytes: + """Choose the 64-byte nonce the hardware will sign into ``report_data``. + + Default (CRYPTO-001 + CRYPTO-002): ``jwk_thumbprint(key) || random_salt``. The + first 32 bytes let a verifier re-derive the fingerprint from ``cnf.jwk`` and + confirm it matches ``report_data[:32]``, binding the report to this keypair. The + salt makes two gateways sharing a keypair produce different nonces. + + On SEV-SNP, TDX and Azure CVM (#552) the salt is replaced by the gateway + measurement digest, so the code, policy and config actually running are what the + hardware signs. Freshness survives the change: the signing key is generated once + per start, so ``report_data[:32]`` still differs between two starts of identical + code, policy and config. + + The TPM tier keeps the salt. Its measurement is committed by the NV extend index + instead, which keeps the history that ``report_data`` does not. + """ + if measurement is not None and binds_measurement_into_report_data( + tee_provider.provider_name() + ): + return measurement_bound_nonce_for(signing_key.public_key_bytes, measurement) + return key_fingerprint + secrets.token_bytes(32) + + +def _extend_measurement( + config: Config, + tee_provider: TEEProvider, + measurement: GatewayMeasurement | None, + nonce: bytes, +) -> tuple[ExtendResult | None, bytes | None]: + """Extend the measurement into the TPM NV index and have the TPM certify it (#432). + + Returns ``(extend, evidence)``. ``evidence`` is the signed ``TPM2_NV_Certify`` + pair, or None when the platform provisions no certified attestation key to sign + with: the extend still happens and is still a local integrity control, but it is + not remote-verifiable, so it is not presented as evidence. Both are None on a + platform with no TPM, where #552's ``report_data`` binding does this job instead. + """ + if tee_provider.provider_name() != "tpm" or measurement is None: + return None, None + + def _degrade(exc: MeasurementUnavailable) -> tuple[None, None]: + _degrade_measurement(config, exc) + return None, None try: from tpm2_pytss.ESAPI import ESAPI @@ -219,7 +283,7 @@ def _degrade(exc: MeasurementUnavailable) -> tuple[None, None, None]: extend_result.index, evidence is not None, ) - return measurement, extend_result, evidence + return extend_result, evidence def _extend_and_certify( @@ -311,21 +375,25 @@ def run_startup(config_path: str) -> RuntimeContext: # (SHA-256 of the sorted JSON OKP key members) so verifiers can re-derive the fingerprint # from cnf.jwk and confirm it matches report_data[:32] -- binding the attestation report # to this specific keypair. - # The remaining 32 bytes are a random salt so two gateways with different random bytes - # produce different nonces even if they share the same keypair (blue-green deploy). + # What fills the remaining 32 bytes depends on the platform: a random salt, or the + # gateway measurement on the providers that have no NV index to hold it (#552). + # See _attestation_nonce. _x_b64 = base64.urlsafe_b64encode(signing_key.public_key_bytes).rstrip(b"=").decode() key_fingerprint = _jwk_thumbprint_sha256(_x_b64) - random_salt = secrets.token_bytes(32) - nonce = key_fingerprint + random_salt - - # Step 3b (#432): measure the gateway into the NV extend index BEFORE it serves - # traffic, and have the TPM certify the value either side of the extend so the - # measurement is signed evidence rather than a self-reported number. PCRs 0-7 - # cover firmware and the bootloader only, so without this the TPM enforced - # nothing about the gateway itself and a swapped policy bundle measured - # identically. - measurement, extend_result, measurement_evidence = _measure_gateway( - config, tee_provider, nonce + + # Step 3b (#432, #552): measure the gateway BEFORE it serves traffic. The + # measurement is computed first because on SEV-SNP, TDX and Azure CVM it goes + # into the nonce itself, so it has to exist before the nonce does. + measurement = _measure_gateway(config, tee_provider) + nonce = _attestation_nonce(key_fingerprint, signing_key, tee_provider, measurement) + + # On the TPM tier the measurement is committed by an NV extend index instead, + # certified either side of the extend so it is signed evidence rather than a + # self-reported number. PCRs 0-7 cover firmware and the bootloader only, so + # without this the TPM enforced nothing about the gateway itself and a swapped + # policy bundle measured identically. + extend_result, measurement_evidence = _extend_measurement( + config, tee_provider, measurement, nonce ) try: diff --git a/src/cmcp_runtime/tee/base.py b/src/cmcp_runtime/tee/base.py index 4dacb6ac..ac3107cb 100644 --- a/src/cmcp_runtime/tee/base.py +++ b/src/cmcp_runtime/tee/base.py @@ -132,6 +132,47 @@ def make_audit_bound_nonce(tee_public_key: bytes, chain_root_hex: str) -> bytes: return jwk_thumbprint(tee_public_key) + audit_root_commitment(chain_root_hex) +def make_measurement_bound_nonce( + tee_public_key: bytes, measurement_digest: bytes +) -> bytes: + """Compute the startup attestation nonce that commits the gateway measurement. + + Layout (64 bytes, #552), the same shape :func:`make_audit_bound_nonce` uses: + + jwk_thumbprint(pubkey) (32) || measurement_digest (32) + + The first 32 bytes keep the existing key binding intact (report_data[:32] is + the RFC 7638 thumbprint, re-derivable from cnf.jwk.x). The second 32 bytes are + :attr:`~cmcp_runtime.tee.measurement.GatewayMeasurement.digest` verbatim: it is + already a raw 32-byte SHA-256 over code, policy and config, so it drops into + the second half unreshaped and a verifier compares it against a digest it + recomputes rather than against a hash of one. + + **Why this exists.** SEV-SNP and TDX commit a *launch* measurement, which is + fixed at boot. It does not move when the Cedar bundle reloads mid-session via + ``PolicyEvaluator._maybe_reload()``, so on those platforms the policy actually + in force was committed to nothing. The TPM tier already solves this with an + ``TPM_NT_EXTEND`` NV index (#432); these platforms have no such index, so the + same validated digest goes into ``report_data`` instead. + + **Freshness still holds.** This replaces the random salt of :func:`make_nonce` + with a deterministic value, but the first half is the thumbprint of a signing + key generated once per gateway start, so two starts of identical code, policy + and config still produce different nonces. + + **No append-only history.** Unlike the NV index, ``report_data`` carries only + the current value, so a report built before a policy reload stays valid-looking + on its face. Re-attesting on every bundle reload is what makes the committed + measurement the live one; a verifier that recomputes and compares is what + catches a gateway that failed to. + """ + if len(measurement_digest) != 32: + raise ValueError( + f"measurement_digest must be 32 bytes, got {len(measurement_digest)}" + ) + return jwk_thumbprint(tee_public_key) + measurement_digest + + class SoftwareOnlyProvider(TEEProvider): """ Software-only attestation stub for CI and local development. diff --git a/src/cmcp_runtime/tee/report_binding.py b/src/cmcp_runtime/tee/report_binding.py new file mode 100644 index 00000000..9884c2de --- /dev/null +++ b/src/cmcp_runtime/tee/report_binding.py @@ -0,0 +1,178 @@ +"""Commit the gateway measurement into report_data where there is no NV index (#552). + +#432 answered "what to measure": :func:`~cmcp_runtime.tee.measurement.gateway_measurement` +folds the installed code, the policy bundle and the effective config into one +SHA-256, and the TPM tier extends it into a ``TPM_NT_EXTEND`` NV index that is +certified by the platform AK. That path is validated on real Azure Trusted Launch +vTPM hardware. + +It ran for the ``tpm`` provider only. The stated reason was that "SEV-SNP and TDX +commit their own binding through the report's fields", which is true and not +equivalent: **those fields carry the launch measurement, and a launch measurement +is boot-time.** It does not move when the Cedar bundle reloads mid-session through +``PolicyEvaluator._maybe_reload()``. So on SEV-SNP, TDX and Azure CVM the policy +actually in force was committed to nothing. + +## What this module does + +Nothing new is invented. The already-validated digest is wired into the one field +those platforms do sign over a caller-supplied value: ``report_data``. The nonce +layout is :func:`~cmcp_runtime.tee.base.make_measurement_bound_nonce`, which is the +same 64-byte shape AUDIT-006 already uses for the audit-chain root. + +## Why re-attesting on reload is not optional + +The NV index is append-only, so a verifier can appraise a *relation* between two +certified values and staleness shows up as a broken chain. ``report_data`` has no +history at all: it holds one value, and a report built before a bundle reload +looks perfectly well-formed on its own. The commitment is only as live as the last +report, so the report has to be replaced whenever the measurement moves. That is +what :func:`refresh_measurement_binding` is for, and a verifier that recomputes the +digest and compares is what catches a gateway that failed to do it. + +## Failure handling, and why it is not fail-closed + +A refresh that cannot re-attest logs and leaves the previous report in place; the +gateway keeps serving. Refusing traffic on a TEE hiccup would trade a *detectable* +weakness for an outage: a stale ``report_data`` no longer matches the recomputed +measurement, so the verifier rejects the claim. That mirrors how AUDIT-006 handles +a failed per-session attestation in :mod:`cmcp_runtime.session.manager`. + +The startup path is stricter, and deliberately so: see ``_gateway_measurement`` in +:mod:`cmcp_runtime.startup`, where an unmeasurable gateway on one of these +platforms is fatal in production for the same reason it is fatal on the TPM tier. + +Out of scope, per #552: binding the audit-chain root (that is session activity +rather than gateway identity, and AUDIT-006 already owns report_data[32:64] on the +per-session report), and validation on real SEV-SNP and TDX silicon. +""" + +from __future__ import annotations + +import logging +from typing import Any + +from cmcp_runtime.tee.base import AttestationReport, make_measurement_bound_nonce +from cmcp_runtime.tee.measurement import ( + GatewayMeasurement, + MeasurementUnavailable, + gateway_measurement, +) + +logger = logging.getLogger(__name__) + +# Providers whose hardware report commits a boot-time launch measurement and offers +# no append-only index of its own, so the gateway measurement goes into report_data. +# +# "tpm" is absent because it has the NV extend index, which is strictly better here: +# it keeps history, so staleness is visible as a broken chain rather than only as a +# mismatch. "opaque" is absent because its provider raises rather than producing a +# report, and "software-only" is absent because there is no hardware to commit to +# -- the round trip is still exercised there through SoftwareOnlyProvider in tests. +MEASUREMENT_BOUND_PROVIDERS: frozenset[str] = frozenset({ + "sev-snp", + "azure-cvm-sev-snp", + "tdx", +}) + + +def binds_measurement_into_report_data(provider_name: str) -> bool: + """True when this provider commits the gateway measurement via ``report_data``.""" + return provider_name in MEASUREMENT_BOUND_PROVIDERS + + +def measurement_bound_nonce_for( + signing_key_public_bytes: bytes, measurement: GatewayMeasurement +) -> bytes: + """Build the 64-byte nonce committing ``measurement`` alongside the key binding.""" + return make_measurement_bound_nonce(signing_key_public_bytes, measurement.digest) + + +def refresh_measurement_binding(ctx: Any) -> bool: + """Re-attest when the gateway measurement has moved, e.g. after a policy reload. + + ``ctx`` is a :class:`~cmcp_runtime.startup.RuntimeContext`; it is typed loosely + to keep this module free of an import cycle back through startup. + + Called on every policy-bundle reload, including reloads that found nothing + changed: see :meth:`~cmcp_runtime.policy.evaluator.PolicyEvaluator._notify_reload` + for why an unchanged measurement is still worth re-signing. + + Returns True only when a new report was obtained and installed. Every other + outcome -- a provider that does not use this binding, an unmeasurable gateway, + a failed attestation call -- returns False and leaves ``ctx`` exactly as it was. + + Both ``ctx.gateway_measurement`` and ``ctx.attestation_report`` are assigned only + after the new report is in hand, so a failure at any earlier step leaves neither + changed. They are two statements rather than one atomic swap, which is worth + stating exactly rather than overclaiming: a concurrent reader could observe the + new measurement beside the old report for one interpreter step. That is benign + today because nothing reads the pair together (``SessionManager`` reads only + ``attestation_report``), and it is recorded here so that stops being an accident + if something later does. + """ + provider = ctx.tee_provider + provider_name = provider.provider_name() + if not binds_measurement_into_report_data(provider_name): + return False + + try: + measurement = gateway_measurement(ctx.config) + except MeasurementUnavailable as exc: + logger.warning( + "#552: the gateway could not be re-measured after a policy-bundle " + "reload, so " + "report_data still commits the previous measurement and verification " + "will reject it: %s %s", + exc, + exc.detail or "", + ) + return False + + # No short-circuit on an unchanged digest. #552 asks for a refresh on every + # reload precisely because report_data carries no history: an identical digest + # re-signed now is a different assertion from the same digest signed an hour + # ago, and the second is the only thing skipping would leave a verifier. + current = getattr(ctx, "gateway_measurement", None) + + nonce = measurement_bound_nonce_for(ctx.signing_key.public_key_bytes, measurement) + try: + report = provider.get_attestation_report(nonce) + except Exception as exc: # noqa: BLE001 - any TEE fault leaves the old binding + logger.warning( + "#552: re-attestation failed after the gateway measurement changed, so " + "report_data still commits the previous measurement and verification " + "will reject it. provider=%s error=%s: %s", + provider_name, + type(exc).__name__, + exc, + ) + return False + + # Guard on the concrete type for the same reason AUDIT-006 does: a provider that + # returns something malformed must not displace a well-formed report. + if not isinstance(report, AttestationReport): + logger.warning( + "#552: the TEE provider returned a %s, not an AttestationReport - " + "keeping the previous report. provider=%s", + type(report).__name__, + provider_name, + ) + return False + + unchanged = current is not None and current.digest == measurement.digest + previous = current.digest_hex if current is not None else "none" + ctx.gateway_measurement = measurement + ctx.attestation_report = report + logger.info( + "#552: gateway measurement rebound into report_data: %s -> %s%s " + "(code=%s policy=%s config=%s) provider=%s", + previous, + measurement.digest_hex, + " (unchanged, re-signed)" if unchanged else "", + measurement.components["code"][:12], + measurement.components["policy"][:12], + measurement.components["config"][:12], + provider_name, + ) + return True diff --git a/src/cmcp_verify/verify.py b/src/cmcp_verify/verify.py index 0ecaad29..eb468980 100644 --- a/src/cmcp_verify/verify.py +++ b/src/cmcp_verify/verify.py @@ -149,6 +149,7 @@ class VerificationError(StrEnum): ATTESTATION_STALE = "ATTESTATION_STALE" CHAIN_BROKEN = "CHAIN_BROKEN" CHAIN_ROOT_NOT_BOUND = "CHAIN_ROOT_NOT_BOUND" + MEASUREMENT_NOT_BOUND = "MEASUREMENT_NOT_BOUND" CLAIM_MALFORMED = "CLAIM_MALFORMED" HARDWARE_ATTESTATION_FAILED = "HARDWARE_ATTESTATION_FAILED" AGENT_MANIFEST_MISMATCH = "AGENT_MANIFEST_MISMATCH" @@ -416,6 +417,105 @@ def _check_audit_chain_binding( return True, None +def _check_measurement_binding( + claim: dict[str, Any], + expected_measurement_digest: bytes, + *, + is_sw_only: bool, +) -> tuple[bool | None, str | None]: + """#552: verify report_data[32:64] commits the gateway measurement. + + On SEV-SNP, TDX and Azure CVM the gateway's startup attestation nonce is + + jwk_thumbprint(key) (32) || gateway_measurement.digest (32) + + so the hardware signs the digest over the installed code, the policy bundle and + the effective configuration. The digest is a raw 32-byte SHA-256 and goes in + unreshaped, so this compares it directly rather than re-hashing it, which is the + one way this check differs from :func:`_check_audit_chain_binding`. + + ``expected_measurement_digest`` must come from the verifier, not from the claim. + That is the whole point: the claim is what is being appraised. It is the same + out-of-band trust input as ``trusted_ark_pem`` or ``ApprovedHashes``, obtained + from the build that was approved to run. + + **This is the freshness check for a field with no history.** Unlike the TPM's + ``TPM_NT_EXTEND`` index, ``report_data`` holds one value and no chain, so a + report produced before a policy reload is internally well-formed and cannot be + told from a current one by inspection. The gateway re-attests on every + policy-bundle reload; recompute-and-compare here is what catches a gateway that + did not. A mismatch is FATAL: the code, policy or config now running is not the + one this hardware report committed. + + Returns: + (True, None) -- report_data[32:64] commits the expected measurement + (False, reason) -- mismatch or missing commitment; reject (fail closed) + (None, warning_msg) -- software-only / Level-0 mode; not hardware-backed + """ + if len(expected_measurement_digest) != 32: + return False, ( + f"expected measurement digest is {len(expected_measurement_digest)} bytes, " + "not 32; a caller supplied something that is not a raw SHA-256" + ) + + nonce_b64 = claim.get("trace", {}).get("runtime", {}).get("nonce", "") + if not nonce_b64: + if is_sw_only: + return None, "software-only mode -- measurement binding not applicable" + return False, ( + "trace.runtime.nonce is absent -- attestation report_data does not " + "commit the gateway measurement" + ) + + try: + padding = 4 - (len(nonce_b64) % 4) + padded = nonce_b64 + ("=" * padding if padding != 4 else "") + nonce_bytes = base64.urlsafe_b64decode(padded) + except Exception as exc: + return False, f"cannot decode trace.runtime.nonce: {exc}" + + if len(nonce_bytes) < 64: + if is_sw_only: + return None, ( + "software-only mode -- report_data does not carry a measurement " + "commitment in bytes [32:64]" + ) + return False, ( + f"trace.runtime.nonce is too short ({len(nonce_bytes)} bytes); " + "expected 64 bytes (key fingerprint || measurement digest)" + ) + + if not hmac.compare_digest(nonce_bytes[32:64], expected_measurement_digest): + # Fatal in software-only mode too: the digest is computed the same way there, + # so a mismatch is a real disagreement about what is running, not an absence + # of hardware. What software-only costs is provenance, not correctness. + return False, ( + "gateway measurement does not match report_data[32:64] -- the code, " + "policy bundle or configuration now running is not the one committed to " + "this attestation report. Either the gateway did not re-attest after a " + "policy reload, or it is not running what the verifier approved" + ) + + if is_sw_only: + logger.warning( + "#552: software-only (dev) mode -- measurement commitment matches but " + "provides no hardware provenance guarantee" + ) + return None, "software-only mode -- measurement binding not hardware-backed" + + return True, None + + +def _coerce_measurement_digest(value: str | bytes) -> bytes | None: + """Accept a raw 32-byte digest or its hex form, with or without ``sha256:``.""" + if isinstance(value, bytes): + return value + try: + return bytes.fromhex(value.removeprefix("sha256:")) + except ValueError: + return None + + def _validate_schema(claim: dict[str, Any]) -> tuple[bool, str | None]: """Validate claim structure using the RuntimeClaim Pydantic model.""" try: @@ -608,6 +708,7 @@ def verify_trace_claim( trusted_ark_pem: bytes | None = None, trusted_intel_root_pem: bytes | None = None, trusted_tpm_ca_pem: bytes | None = None, + expected_gateway_measurement: str | bytes | None = None, ) -> VerificationResult: """ Verify a TRACE Claim without trusting the operator. @@ -624,6 +725,10 @@ def verify_trace_claim( 6. Attestation freshness check 7. Audit chain consistency check 7b. AUDIT-006: audit-chain root binding -- report_data[32:64] commits SHA-256(chain_root) + 7c. #552: gateway measurement binding -- report_data[32:64] commits the gateway + measurement digest. Only runs when expected_gateway_measurement is supplied, + because the expected value has to come from the verifier rather than the + claim. See _check_measurement_binding for which report carries it. 8. Platform-specific attestation verification (dispatched per-platform) Returns VerificationResult with status and details. @@ -843,6 +948,32 @@ def verify_trace_claim( if root_binding_msg: details["audit_chain_binding"] = root_binding_msg + # Step 7c (#552): the gateway measurement binding. Opt-in, because the expected + # digest is an out-of-band trust input like ApprovedHashes: a check that read it + # out of the claim would be asking the claim to vouch for itself. + if expected_gateway_measurement is not None: + _expected_digest = _coerce_measurement_digest(expected_gateway_measurement) + if _expected_digest is None: + unverified.append("measurement_binding") + failure = failure or VerificationError.MEASUREMENT_NOT_BOUND + details["measurement_binding"] = ( + "expected_gateway_measurement is not valid hex or raw bytes" + ) + else: + m_binding, m_binding_msg = _check_measurement_binding( + claim_json, _expected_digest, is_sw_only=_is_sw_only + ) + if m_binding is True: + verified.append("measurement_binding") + elif m_binding is False: + unverified.append("measurement_binding") + failure = failure or VerificationError.MEASUREMENT_NOT_BOUND + details["measurement_binding"] = ( + m_binding_msg or "gateway measurement binding verification failed" + ) + elif m_binding_msg: + details["measurement_binding"] = m_binding_msg + # Step 8: Platform-specific attestation platform = _runtime.get("platform", "") diff --git a/tests/unit/test_measurement_report_binding.py b/tests/unit/test_measurement_report_binding.py new file mode 100644 index 00000000..c6bbf1be --- /dev/null +++ b/tests/unit/test_measurement_report_binding.py @@ -0,0 +1,915 @@ +"""Bind the gateway measurement into SEV-SNP / TDX / Azure-CVM report_data (#552). + +These are the properties #552 prototyped, kept as tests so the contract cannot +regress silently. They are software-only by design: SoftwareOnlyProvider echoes the +nonce into report_data, which is exactly the collector-side shape SEV-SNP and TDX +give, so the round trip proves the *contract*. Whether real silicon signs those 64 +bytes is a hardware question and #552 leaves it out of scope on purpose. + +The verifier stand-in below recomputes both halves from scratch, the way a relying +party would: the thumbprint from the public key, the digest from code, policy and +config. Nothing it checks is taken from the report itself. +""" + +from __future__ import annotations + +import hashlib +from collections.abc import Callable +from dataclasses import dataclass, field +from pathlib import Path + +import pytest + +from cmcp_runtime.config import Config +from cmcp_runtime.policy.bundle import PolicyBundle, PolicyManifest +from cmcp_runtime.policy.evaluator import PolicyEvaluator +from cmcp_runtime.tee.base import ( + SoftwareOnlyProvider, + jwk_thumbprint, + make_measurement_bound_nonce, +) +from cmcp_runtime.tee.measurement import GatewayMeasurement, gateway_measurement +from cmcp_runtime.tee.report_binding import ( + MEASUREMENT_BOUND_PROVIDERS, + binds_measurement_into_report_data, + measurement_bound_nonce_for, + refresh_measurement_binding, +) + +_KEY = bytes(range(32)) +_DIGEST = b"\xab" * 32 + + +# ── config and context stand-ins ────────────────────────────────────────────── + + +@dataclass +class _Nested: + mode: str = "enforcing" + validity_seconds: int = 86400 + + +@dataclass +class _Config: + policy_bundle_path: str = "policies/" + listen_addr: str = "0.0.0.0:8443" + dev_mode: bool = False + bearer_token: str | None = None + attestation: _Nested = field(default_factory=_Nested) + + +@dataclass +class _SigningKey: + public_key_bytes: bytes = _KEY + + +@dataclass +class _Ctx: + """The slice of RuntimeContext refresh_measurement_binding actually touches.""" + + config: _Config + tee_provider: object + signing_key: _SigningKey = field(default_factory=_SigningKey) + gateway_measurement: GatewayMeasurement | None = None + attestation_report: object | None = None + + +def _bundle(tmp_path: Path, content: str) -> str: + root = tmp_path / "policies" + root.mkdir(parents=True, exist_ok=True) + (root / "a.cedar").write_text(content) + return str(root) + + +@pytest.fixture +def stub_code(monkeypatch: pytest.MonkeyPatch) -> None: + """Pin code_digest: an editable test install has no RECORD to measure.""" + monkeypatch.setattr("cmcp_runtime.tee.measurement.code_digest", lambda: "c" * 64) + + +# ── nonce layout ────────────────────────────────────────────────────────────── + + +def test_nonce_is_thumbprint_then_measurement() -> None: + """The 64-byte layout #552 specifies, with the digest unreshaped.""" + nonce = make_measurement_bound_nonce(_KEY, _DIGEST) + + assert len(nonce) == 64 + assert nonce[:32] == jwk_thumbprint(_KEY) + assert nonce[32:64] == _DIGEST + + +def test_nonce_keeps_the_key_binding_intact() -> None: + """CRYPTO-001 still holds: report_data[:32] is re-derivable from cnf.jwk.x.""" + other = bytes(range(1, 33)) + assert make_measurement_bound_nonce(_KEY, _DIGEST)[:32] != ( + make_measurement_bound_nonce(other, _DIGEST)[:32] + ) + + +def test_nonce_rejects_a_wrong_sized_digest() -> None: + """A short digest would leave the second half partly zero, committing nothing.""" + with pytest.raises(ValueError, match="32 bytes"): + make_measurement_bound_nonce(_KEY, b"\xab" * 31) + + +def test_nonce_is_fresh_per_start_despite_being_deterministic() -> None: + """Replacing the random salt does not cost freshness: the key is per-start. + + make_nonce got its freshness from 32 random bytes. This layout has none, so the + property has to come from the first half instead, and it does: run_startup + generates a new signing key on every start, so two starts of byte-identical + code, policy and config still produce different report_data. + """ + start_a = make_measurement_bound_nonce(_KEY, _DIGEST) + start_b = make_measurement_bound_nonce(bytes(range(32, 64)), _DIGEST) + assert start_a != start_b + + +# ── which providers use it ──────────────────────────────────────────────────── + + +def test_tpm_does_not_use_report_data_binding() -> None: + """The TPM tier has the NV extend index, which keeps history report_data cannot.""" + assert not binds_measurement_into_report_data("tpm") + + +@pytest.mark.parametrize("provider", sorted(MEASUREMENT_BOUND_PROVIDERS)) +def test_launch_measurement_platforms_use_report_data_binding(provider: str) -> None: + assert binds_measurement_into_report_data(provider) + + +def test_the_three_platforms_from_the_issue_are_covered() -> None: + assert {"sev-snp", "azure-cvm-sev-snp", "tdx"} == MEASUREMENT_BOUND_PROVIDERS + + +# ── the measurement itself ──────────────────────────────────────────────────── + + +def test_measurement_is_deterministic_across_repeat_calls( + tmp_path: Path, stub_code: None +) -> None: + config = _Config(policy_bundle_path=_bundle(tmp_path, "permit(...);")) + assert gateway_measurement(config).digest == gateway_measurement(config).digest + + +def test_one_character_of_cedar_moves_only_the_policy_component( + tmp_path: Path, stub_code: None +) -> None: + """Component isolation is what makes a mismatch diagnosable rather than opaque.""" + path = _bundle(tmp_path, "permit(...);") + config = _Config(policy_bundle_path=path) + before = gateway_measurement(config) + + (Path(path) / "a.cedar").write_text("permit(...) ;") + after = gateway_measurement(config) + + assert after.components["policy"] != before.components["policy"] + assert after.components["code"] == before.components["code"] + assert after.components["config"] == before.components["config"] + assert after.digest != before.digest + + +# ── full round trip through a provider ──────────────────────────────────────── + + +def _verifier_recompute(public_key: bytes, config: _Config) -> bytes: + """What a relying party derives independently: neither half read off the report.""" + return jwk_thumbprint(public_key) + gateway_measurement(config).digest + + +def test_round_trip_matches_both_halves_of_report_data( + tmp_path: Path, stub_code: None +) -> None: + config = _Config(policy_bundle_path=_bundle(tmp_path, "permit(...);")) + measurement = gateway_measurement(config) + + nonce = measurement_bound_nonce_for(_KEY, measurement) + report = SoftwareOnlyProvider().get_attestation_report(nonce) + report_data = bytes.fromhex(report.report_data) + + expected = _verifier_recompute(_KEY, config) + assert report_data[:32] == expected[:32] + assert report_data[32:64] == expected[32:64] + + +def test_a_swapped_policy_bundle_no_longer_matches_the_report( + tmp_path: Path, stub_code: None +) -> None: + """The gap #552 closes: on these platforms this comparison used to pass anyway. + + The launch measurement is boot-time, so before this change nothing in the signed + report moved when the Cedar bundle did. + """ + path = _bundle(tmp_path, "permit(...);") + config = _Config(policy_bundle_path=path) + + nonce = measurement_bound_nonce_for(_KEY, gateway_measurement(config)) + report_data = bytes.fromhex( + SoftwareOnlyProvider().get_attestation_report(nonce).report_data + ) + + (Path(path) / "a.cedar").write_text("forbid(...);") + assert report_data[32:64] != _verifier_recompute(_KEY, config)[32:64] + + +# ── refresh on policy reload ────────────────────────────────────────────────── + + +class _CountingProvider(SoftwareOnlyProvider): + """SoftwareOnlyProvider that records the nonces it was asked to commit.""" + + def __init__(self, provider: str = "sev-snp") -> None: + self._provider = provider + self.nonces: list[bytes] = [] + + def provider_name(self) -> str: + return self._provider + + def get_attestation_report(self, nonce: bytes): # type: ignore[no-untyped-def] + self.nonces.append(nonce) + report = super().get_attestation_report(nonce) + report.provider = self._provider + return report + + +def test_reload_produces_a_report_reflecting_the_new_measurement( + tmp_path: Path, stub_code: None +) -> None: + path = _bundle(tmp_path, "permit(...);") + config = _Config(policy_bundle_path=path) + provider = _CountingProvider() + ctx = _Ctx(config=config, tee_provider=provider) + + assert refresh_measurement_binding(ctx) is True + first = bytes.fromhex(ctx.attestation_report.report_data) + + (Path(path) / "a.cedar").write_text("forbid(...);") + assert refresh_measurement_binding(ctx) is True + second = bytes.fromhex(ctx.attestation_report.report_data) + + assert second != first + assert second[32:64] == gateway_measurement(config).digest + assert len(provider.nonces) == 2 + + +def test_a_stale_pre_reload_report_is_rejected_on_recompute( + tmp_path: Path, stub_code: None +) -> None: + """The freshness property: report_data has no history, so staleness must show.""" + path = _bundle(tmp_path, "permit(...);") + config = _Config(policy_bundle_path=path) + ctx = _Ctx(config=config, tee_provider=_CountingProvider()) + + refresh_measurement_binding(ctx) + stale = bytes.fromhex(ctx.attestation_report.report_data) + + (Path(path) / "a.cedar").write_text("forbid(...);") + assert stale[32:64] != _verifier_recompute(_KEY, config)[32:64] + + +def test_every_reload_re_attests_even_when_the_measurement_is_unchanged( + tmp_path: Path, stub_code: None +) -> None: + """#552 asks for a refresh on every reload, not only on the ones that changed. + + report_data carries no history, so the same digest re-signed now and that digest + signed an hour ago are different assertions, and only the latest one reaches a + verifier. Skipping the unchanged case would leave the stale one standing. + """ + config = _Config(policy_bundle_path=_bundle(tmp_path, "permit(...);")) + provider = _CountingProvider() + ctx = _Ctx(config=config, tee_provider=provider) + + assert refresh_measurement_binding(ctx) is True + assert refresh_measurement_binding(ctx) is True + assert refresh_measurement_binding(ctx) is True + + assert len(provider.nonces) == 3 + assert provider.nonces[0] == provider.nonces[1] == provider.nonces[2] + assert bytes.fromhex(ctx.attestation_report.report_data)[32:64] == ( + gateway_measurement(config).digest + ) + + +def test_the_cost_is_bounded_by_the_reload_interval_not_the_request_rate( + tmp_path: Path, stub_code: None +) -> None: + """Evaluating tool calls between reloads must not each spend a TEE call. + + The bound comes from PolicyStore.reload_if_stale, which stamps its clock before + the attempt and returns False until the interval elapses, so the hook fires once + per interval however many calls arrive in it. + """ + from cmcp_runtime.policy.bundle import PolicyStore + + calls: list[int] = [] + store = PolicyStore( + bundle=_policy_bundle("permit(principal, action, resource);"), + bundle_path=str(tmp_path / "nonexistent"), + reload_interval_seconds=3600, + ) + evaluator = PolicyEvaluator(bundle=store, config=Config(), on_reload=lambda: calls.append(1)) + + for _ in range(5): + evaluator._maybe_reload() + + assert calls == [] # interval has not elapsed since the store was built + + +def test_refresh_is_a_no_op_on_the_tpm_provider(tmp_path: Path, stub_code: None) -> None: + """The TPM tier commits through the NV index; report_data is not its mechanism.""" + config = _Config(policy_bundle_path=_bundle(tmp_path, "permit(...);")) + provider = _CountingProvider(provider="tpm") + ctx = _Ctx(config=config, tee_provider=provider) + + assert refresh_measurement_binding(ctx) is False + assert provider.nonces == [] + assert ctx.attestation_report is None + + +def test_a_failed_re_attestation_leaves_the_previous_report_in_place( + tmp_path: Path, stub_code: None +) -> None: + """Availability over fail-closed, because the stale binding is detectable. + + The context must not end up advertising a measurement its report_data does not + commit, so the pair is replaced together or not at all. + """ + path = _bundle(tmp_path, "permit(...);") + config = _Config(policy_bundle_path=path) + provider = _CountingProvider() + ctx = _Ctx(config=config, tee_provider=provider) + refresh_measurement_binding(ctx) + good_report = ctx.attestation_report + good_measurement = ctx.gateway_measurement + + def _boom(nonce: bytes): # type: ignore[no-untyped-def] + raise RuntimeError("TEE unavailable") + + provider.get_attestation_report = _boom # type: ignore[method-assign] + (Path(path) / "a.cedar").write_text("forbid(...);") + + assert refresh_measurement_binding(ctx) is False + assert ctx.attestation_report is good_report + assert ctx.gateway_measurement is good_measurement + + +def test_an_unmeasurable_gateway_leaves_the_previous_report_in_place( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, stub_code: None +) -> None: + config = _Config(policy_bundle_path=_bundle(tmp_path, "permit(...);")) + ctx = _Ctx(config=config, tee_provider=_CountingProvider()) + refresh_measurement_binding(ctx) + good_report = ctx.attestation_report + + monkeypatch.setattr(config, "policy_bundle_path", str(tmp_path / "gone")) + assert refresh_measurement_binding(ctx) is False + assert ctx.attestation_report is good_report + + +# ── the evaluator hook ──────────────────────────────────────────────────────── + + +def _policy_bundle(content: str) -> PolicyBundle: + return PolicyBundle( + manifest=PolicyManifest( + version="1.0.0", + authored_at="2026-08-23T00:00:00Z", + author_identity="test", + commit_sha="abc", + ), + policy_files={"a.cedar": content}, + schema_content='{"cMCP": {}}', + bundle_hash="sha256:" + hashlib.sha256(content.encode()).hexdigest(), + ) + + +def _evaluator_with_hook( + hook: Callable[[], object], *, reloads: bool = True +) -> PolicyEvaluator: + """Build an evaluator whose store reports ``reloads`` from reload_if_stale.""" + evaluator = PolicyEvaluator( + bundle=_policy_bundle("permit(principal, action, resource);"), + config=Config(), + on_reload=hook, + ) + evaluator._store.reload_if_stale = lambda: reloads # type: ignore[method-assign] + return evaluator + + +def _swap_bundle(evaluator: PolicyEvaluator) -> None: + """Force the hash-changed branch without needing a real on-disk reload.""" + evaluator._store._bundle = _policy_bundle("forbid(principal, action, resource);") + + +def test_evaluator_calls_the_hook_on_every_reload() -> None: + """The literal reading of #552: every reload, not only the ones that changed. + + A PolicyStore whose reload_if_stale reports a reload fires the hook whether or + not the bundle hash moved, which is the whole point of the second bullet. + """ + calls: list[int] = [] + evaluator = _evaluator_with_hook(lambda: calls.append(1), reloads=True) + + evaluator._maybe_reload() + assert calls == [1] # unchanged bundle, still re-committed + + _swap_bundle(evaluator) + evaluator._maybe_reload() + assert calls == [1, 1] + + +def test_a_reload_that_did_not_happen_does_not_call_the_hook() -> None: + """No reload, no re-attestation: the hook tracks reloads, not evaluations.""" + calls: list[int] = [] + evaluator = _evaluator_with_hook(lambda: calls.append(1), reloads=False) + + evaluator._maybe_reload() + evaluator._maybe_reload() + assert calls == [] + + +def test_a_failed_reload_does_not_call_the_hook() -> None: + """reload_if_stale returns False when the read failed, and there is nothing new + to commit: the bundle in force is the one already bound.""" + calls: list[int] = [] + evaluator = _evaluator_with_hook(lambda: calls.append(1), reloads=False) + evaluator._store.reload_if_stale = lambda: False # type: ignore[method-assign] + + _swap_bundle(evaluator) + evaluator._maybe_reload() + assert calls == [] + + +def test_a_failing_hook_does_not_break_enforcement() -> None: + """A TEE fault must not take the gateway down; the verifier catches it instead.""" + + def _boom() -> None: + raise RuntimeError("TEE unavailable") + + evaluator = _evaluator_with_hook(_boom) + _swap_bundle(evaluator) + evaluator._maybe_reload() # must not raise + + +def test_no_hook_is_the_default() -> None: + """Every other caller of PolicyEvaluator keeps working untouched.""" + evaluator = PolicyEvaluator( + bundle=_policy_bundle("permit(principal, action, resource);"), + config=Config(), + ) + evaluator._store.reload_if_stale = lambda: True # type: ignore[method-assign] + _swap_bundle(evaluator) + evaluator._maybe_reload() # must not raise + + +# ── startup wiring ──────────────────────────────────────────────────────────── + + +def test_startup_binds_the_measurement_into_the_nonce_on_sev_snp( + tmp_path: Path, stub_code: None +) -> None: + from cmcp_runtime.startup import _attestation_nonce, _measure_gateway + + config = _Config(policy_bundle_path=_bundle(tmp_path, "permit(...);")) + provider = _CountingProvider() + + measurement = _measure_gateway(config, provider) + assert measurement is not None + + nonce = _attestation_nonce( + jwk_thumbprint(_KEY), _SigningKey(), provider, measurement + ) + assert nonce == jwk_thumbprint(_KEY) + gateway_measurement(config).digest + + +def test_startup_keeps_the_random_salt_on_the_tpm_tier( + tmp_path: Path, stub_code: None +) -> None: + """The TPM commits via the NV extend index, so its nonce is unchanged (#432).""" + from cmcp_runtime.startup import _attestation_nonce, _measure_gateway + + config = _Config(policy_bundle_path=_bundle(tmp_path, "permit(...);")) + provider = _CountingProvider(provider="tpm") + + measurement = _measure_gateway(config, provider) + assert measurement is not None # still measured, just committed elsewhere + + fingerprint = jwk_thumbprint(_KEY) + first = _attestation_nonce(fingerprint, _SigningKey(), provider, measurement) + second = _attestation_nonce(fingerprint, _SigningKey(), provider, measurement) + + assert first[:32] == fingerprint + assert first != second # the salt is random per call + assert first[32:64] != measurement.digest + + +def test_startup_skips_the_measurement_where_nothing_commits_it( + tmp_path: Path, stub_code: None +) -> None: + from cmcp_runtime.startup import _measure_gateway + + config = _Config(policy_bundle_path=_bundle(tmp_path, "permit(...);")) + assert _measure_gateway(config, SoftwareOnlyProvider()) is None + + +def test_an_unmeasurable_gateway_aborts_startup_on_sev_snp(tmp_path: Path) -> None: + """Fail-closed, for the same reason the TPM tier does: unmeasured is unattested. + + #552 extends the measurement to these platforms, so it extends the consequence + of not having one. Dev mode keeps the escape hatch an editable install needs. + """ + from cmcp_runtime.startup import _measure_gateway + + config = _Config(policy_bundle_path=str(tmp_path / "gone"), dev_mode=False) + with pytest.raises(SystemExit): + _measure_gateway(config, _CountingProvider()) + + +def test_dev_mode_downgrades_that_abort_to_a_warning(tmp_path: Path) -> None: + from cmcp_runtime.startup import _measure_gateway + + config = _Config(policy_bundle_path=str(tmp_path / "gone"), dev_mode=True) + assert _measure_gateway(config, _CountingProvider()) is None + + +# ── verifier-side recompute-and-compare ─────────────────────────────────────── +# +# The second bullet of #552 ends "Verifier-side recompute-and-compare has to do +# that job". report_data carries no history, so nothing in a single report says +# whether it is current. These are the checks that do that job. + + +def _claim_with_nonce(nonce: bytes) -> dict: + return {"trace": {"runtime": {"nonce": _b64url(nonce)}}} + + +def _b64url(data: bytes) -> str: + import base64 + + return base64.urlsafe_b64encode(data).rstrip(b"=").decode() + + +def test_verifier_accepts_a_report_committing_the_expected_measurement( + tmp_path: Path, stub_code: None +) -> None: + from cmcp_verify.verify import _check_measurement_binding + + config = _Config(policy_bundle_path=_bundle(tmp_path, "permit(...);")) + digest = gateway_measurement(config).digest + claim = _claim_with_nonce(make_measurement_bound_nonce(_KEY, digest)) + + assert _check_measurement_binding(claim, digest, is_sw_only=False) == (True, None) + + +def test_verifier_rejects_a_stale_report_after_a_policy_change( + tmp_path: Path, stub_code: None +) -> None: + """The freshness property, from the other side: recompute catches the staleness.""" + from cmcp_verify.verify import _check_measurement_binding + + path = _bundle(tmp_path, "permit(...);") + config = _Config(policy_bundle_path=path) + claim = _claim_with_nonce( + make_measurement_bound_nonce(_KEY, gateway_measurement(config).digest) + ) + + (Path(path) / "a.cedar").write_text("forbid(...);") + ok, reason = _check_measurement_binding( + claim, gateway_measurement(config).digest, is_sw_only=False + ) + + assert ok is False + assert "does not match report_data[32:64]" in (reason or "") + + +def test_verifier_compares_the_digest_without_re_hashing_it() -> None: + """AUDIT-006 commits SHA-256(chain_root); this commits the digest itself.""" + from cmcp_verify.verify import _check_measurement_binding + + claim = _claim_with_nonce(make_measurement_bound_nonce(_KEY, _DIGEST)) + + assert _check_measurement_binding(claim, _DIGEST, is_sw_only=False)[0] is True + rehashed = hashlib.sha256(_DIGEST).digest() + assert _check_measurement_binding(claim, rehashed, is_sw_only=False)[0] is False + + +def test_verifier_fails_closed_without_a_nonce() -> None: + from cmcp_verify.verify import _check_measurement_binding + + ok, reason = _check_measurement_binding({}, _DIGEST, is_sw_only=False) + assert ok is False + assert "does not" in (reason or "") + + +def test_verifier_rejects_an_expected_digest_that_is_not_a_sha256() -> None: + """A caller passing the wrong shape must not silently compare 31 bytes.""" + from cmcp_verify.verify import _check_measurement_binding + + claim = _claim_with_nonce(make_measurement_bound_nonce(_KEY, _DIGEST)) + ok, reason = _check_measurement_binding(claim, _DIGEST[:31], is_sw_only=False) + assert ok is False + assert "not 32" in (reason or "") + + +def test_a_mismatch_is_fatal_in_software_only_mode_too() -> None: + """Software-only costs provenance, not correctness: the digest is computed the + same way, so a mismatch is a real disagreement about what is running.""" + from cmcp_verify.verify import _check_measurement_binding + + claim = _claim_with_nonce(make_measurement_bound_nonce(_KEY, _DIGEST)) + assert _check_measurement_binding(claim, b"\xcd" * 32, is_sw_only=True)[0] is False + assert _check_measurement_binding(claim, _DIGEST, is_sw_only=True)[0] is None + + +def test_the_expected_digest_accepts_hex_and_the_sha256_prefix() -> None: + from cmcp_verify.verify import _coerce_measurement_digest + + assert _coerce_measurement_digest(_DIGEST) == _DIGEST + assert _coerce_measurement_digest(_DIGEST.hex()) == _DIGEST + assert _coerce_measurement_digest("sha256:" + _DIGEST.hex()) == _DIGEST + assert _coerce_measurement_digest("not-hex") is None + + +# ── the two seams the rest of the suite stubs or assumes ────────────────────── + + +def test_a_real_unchanged_bundle_reload_fires_the_hook(tmp_path: Path) -> None: + """Bullet 2 end to end, with nothing stubbed on the reload seam. + + Every other hook test replaces reload_if_stale to control what it reports. This + one uses a real PolicyStore over a real bundle on disk, lets the interval elapse, + and reloads a bundle that did not change. The hook must still fire: that is the + whole difference between "every reload" and "every reload that changed the hash". + """ + import json + + from cmcp_runtime.policy.bundle import PolicyStore, load_policy_bundle + + root = tmp_path / "bundle" + root.mkdir() + (root / "manifest.json").write_text( + json.dumps( + { + "version": "1.0.0", + "authored_at": "2026-08-23T00:00:00Z", + "author_identity": "test@example.com", + "commit_sha": "abc123", + } + ) + ) + (root / "allow.cedar").write_text("permit(principal, action, resource);") + (root / "schema.cedarschema").write_text('{"cMCP": {"entityTypes": {}, "actions": {}}}') + + bundle = load_policy_bundle(str(root)) + store = PolicyStore(bundle=bundle, bundle_path=str(root), reload_interval_seconds=1) + + calls: list[int] = [] + evaluator = PolicyEvaluator(bundle=store, config=Config(), on_reload=lambda: calls.append(1)) + + # Age the store past its interval rather than sleeping for it. + store._last_reload_at -= 3600 + evaluator._maybe_reload() + + assert calls == [1] + assert store.bundle.bundle_hash == bundle.bundle_hash # genuinely unchanged + + +def test_the_two_jwk_thumbprint_implementations_agree() -> None: + """startup and tee.base each derive the thumbprint; a divergence would split + report_data[:32] between the measurement-bound path and every other path. + + startup._jwk_thumbprint_sha256 builds the first 32 bytes for the salt nonce; + tee.base.jwk_thumbprint builds them for the measurement-bound one. Nothing else + in the tree forces them to agree, so this is the guard that does. + """ + import base64 + + from cmcp_runtime.startup import _jwk_thumbprint_sha256 + + x_b64 = base64.urlsafe_b64encode(_KEY).rstrip(b"=").decode() + assert _jwk_thumbprint_sha256(x_b64) == jwk_thumbprint(_KEY) + + +def test_all_three_providers_put_the_whole_nonce_into_report_data() -> None: + """The binding is only real if the 64 bytes reach the hardware-signed field. + + SEV-SNP and TDX write the nonce into REPORT_DATA / REPORTDATA directly. Azure + CVM cannot (the paravisor owns SNP REPORT_DATA) and commits sha256(nonce) into + the AK-signed quote instead, but still surfaces the nonce as report_data, which + is what cmcp_verify.azure_cvm re-derives against. All three therefore carry the + measurement in report_data[32:64] where the verifier check looks. + """ + import inspect + + from cmcp_runtime.tee.azure_cvm import AzureCVMProvider + from cmcp_runtime.tee.sev_snp import SEVSNPProvider + from cmcp_runtime.tee.tdx import TDXProvider + + providers = (SEVSNPProvider, TDXProvider, AzureCVMProvider) + + # Named explicitly rather than discovered, so this cannot quietly start + # inspecting the abstract base and passing for the wrong reason. + assert {p.__name__ for p in providers} == { + "SEVSNPProvider", + "TDXProvider", + "AzureCVMProvider", + } + for provider in providers: + source = inspect.getsource(provider.get_attestation_report) + assert "report_data=nonce.hex()" in source, provider.__name__ + + # And every one of them is in the set that gets the measurement-bound nonce. + assert {p().provider_name() for p in providers} == MEASUREMENT_BOUND_PROVIDERS + + +# ── fail-closed branches of the verifier check ──────────────────────────────── + + +def test_verifier_fails_closed_on_an_undecodable_nonce() -> None: + from cmcp_verify.verify import _check_measurement_binding + + claim = {"trace": {"runtime": {"nonce": "!!! not base64 !!!"}}} + ok, reason = _check_measurement_binding(claim, _DIGEST, is_sw_only=False) + assert ok is False + assert "cannot decode" in (reason or "") + + +def test_verifier_fails_closed_on_a_short_nonce() -> None: + """A 32-byte nonce has no second half, so there is no commitment to compare.""" + from cmcp_verify.verify import _check_measurement_binding + + claim = _claim_with_nonce(b"\x01" * 32) + ok, reason = _check_measurement_binding(claim, _DIGEST, is_sw_only=False) + assert ok is False + assert "too short" in (reason or "") + + +def test_a_short_nonce_is_only_advisory_in_software_only_mode() -> None: + from cmcp_verify.verify import _check_measurement_binding + + claim = _claim_with_nonce(b"\x01" * 32) + ok, reason = _check_measurement_binding(claim, _DIGEST, is_sw_only=True) + assert ok is None + assert "software-only" in (reason or "") + + +def test_a_missing_nonce_is_only_advisory_in_software_only_mode() -> None: + from cmcp_verify.verify import _check_measurement_binding + + ok, reason = _check_measurement_binding({}, _DIGEST, is_sw_only=True) + assert ok is None + assert "software-only" in (reason or "") + + +# ── the malformed-report guard ──────────────────────────────────────────────── + + +def test_a_provider_returning_a_non_report_does_not_displace_the_good_one( + tmp_path: Path, stub_code: None +) -> None: + """Same guard AUDIT-006 uses: a malformed return must not become the report.""" + config = _Config(policy_bundle_path=_bundle(tmp_path, "permit(...);")) + provider = _CountingProvider() + ctx = _Ctx(config=config, tee_provider=provider) + refresh_measurement_binding(ctx) + good_report = ctx.attestation_report + + provider.get_attestation_report = lambda nonce: {"not": "a report"} # type: ignore[method-assign] + + assert refresh_measurement_binding(ctx) is False + assert ctx.attestation_report is good_report + + +# ── the TPM extend path, which #552 restructured but does not change ────────── + + +def test_extend_degrades_when_tpm2_pytss_is_absent( + tmp_path: Path, stub_code: None +) -> None: + """tpm2-pytss is not installed in CI, so this is the branch that actually runs. + + Dev mode turns the missing library into a warning; production would abort, which + test_an_unmeasurable_gateway_aborts_startup_on_sev_snp covers for the other tier. + """ + from cmcp_runtime.startup import _extend_measurement, _measure_gateway + + config = _Config(policy_bundle_path=_bundle(tmp_path, "permit(...);"), dev_mode=True) + provider = _CountingProvider(provider="tpm") + measurement = _measure_gateway(config, provider) + + assert _extend_measurement(config, provider, measurement, b"\x00" * 64) == (None, None) + + +def test_extend_returns_the_result_and_evidence_on_success( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, stub_code: None +) -> None: + """The success path of the restructured _extend_measurement. + + The TPM itself is covered by test_gateway_measurement.py against a fake ESAPI + that enforces TPM_NT_EXTEND semantics. What is under test here is only the + orchestration: open the context, hand off, return what came back. + """ + import sys + import types + + from cmcp_runtime.startup import _extend_measurement, _measure_gateway + from cmcp_runtime.tee.measurement import ExtendResult + + class _Ctx_: + def __enter__(self): + return object() + + def __exit__(self, *exc): + return False + + esapi_mod = types.ModuleType("tpm2_pytss.ESAPI") + esapi_mod.ESAPI = _Ctx_ # type: ignore[attr-defined] + monkeypatch.setitem(sys.modules, "tpm2_pytss", types.ModuleType("tpm2_pytss")) + monkeypatch.setitem(sys.modules, "tpm2_pytss.ESAPI", esapi_mod) + + extend = ExtendResult(index=0x01500432, before=b"\x00" * 32, after=b"\x01" * 32, provisioned=False) + monkeypatch.setattr( + "cmcp_runtime.startup._extend_and_certify", + lambda ectx, measurement, nonce: (extend, b"evidence"), + ) + + config = _Config(policy_bundle_path=_bundle(tmp_path, "permit(...);")) + provider = _CountingProvider(provider="tpm") + measurement = _measure_gateway(config, provider) + + assert _extend_measurement(config, provider, measurement, b"\x00" * 64) == ( + extend, + b"evidence", + ) + + +def _stub_esapi(monkeypatch: pytest.MonkeyPatch, handoff) -> None: + """Install a tpm2_pytss whose ESAPI context hands off to ``handoff``.""" + import sys + import types + + class _EsapiCtx: + def __enter__(self): + return object() + + def __exit__(self, *exc): + return False + + esapi_mod = types.ModuleType("tpm2_pytss.ESAPI") + esapi_mod.ESAPI = _EsapiCtx # type: ignore[attr-defined] + monkeypatch.setitem(sys.modules, "tpm2_pytss", types.ModuleType("tpm2_pytss")) + monkeypatch.setitem(sys.modules, "tpm2_pytss.ESAPI", esapi_mod) + monkeypatch.setattr("cmcp_runtime.startup._extend_and_certify", handoff) + + +def test_extend_degrades_when_the_tpm_reports_the_measurement_unavailable( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, stub_code: None +) -> None: + from cmcp_runtime.startup import _extend_measurement, _measure_gateway + from cmcp_runtime.tee.measurement import MeasurementUnavailable + + def _raise(ectx, measurement, nonce): + raise MeasurementUnavailable("TPM2_NV_Extend failed", detail="index=0x1500432") + + _stub_esapi(monkeypatch, _raise) + config = _Config(policy_bundle_path=_bundle(tmp_path, "permit(...);"), dev_mode=True) + provider = _CountingProvider(provider="tpm") + measurement = _measure_gateway(config, provider) + + assert _extend_measurement(config, provider, measurement, b"\x00" * 64) == (None, None) + + +def test_any_tpm_fault_degrades_rather_than_escaping( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, stub_code: None +) -> None: + """A gateway must not proceed believing it was measured, nor crash on a TPM fault.""" + from cmcp_runtime.startup import _extend_measurement, _measure_gateway + + def _raise(ectx, measurement, nonce): + raise RuntimeError("tcti connection refused") + + _stub_esapi(monkeypatch, _raise) + config = _Config(policy_bundle_path=_bundle(tmp_path, "permit(...);"), dev_mode=True) + provider = _CountingProvider(provider="tpm") + measurement = _measure_gateway(config, provider) + + assert _extend_measurement(config, provider, measurement, b"\x00" * 64) == (None, None) + + +def test_a_tpm_fault_is_fatal_in_production( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, stub_code: None +) -> None: + from cmcp_runtime.startup import _extend_measurement, _measure_gateway + + def _raise(ectx, measurement, nonce): + raise RuntimeError("tcti connection refused") + + _stub_esapi(monkeypatch, _raise) + config = _Config(policy_bundle_path=_bundle(tmp_path, "permit(...);"), dev_mode=False) + provider = _CountingProvider(provider="tpm") + measurement = _measure_gateway(config, provider) + + with pytest.raises(SystemExit): + _extend_measurement(config, provider, measurement, b"\x00" * 64) diff --git a/tests/unit/test_verify.py b/tests/unit/test_verify.py index ed830d17..e444b253 100644 --- a/tests/unit/test_verify.py +++ b/tests/unit/test_verify.py @@ -60,18 +60,36 @@ def _make_nonce_for_key(key: SigningKey, chain_root_hex: str | None = None) -> s return (fingerprint + second_half).hex() +def _make_measurement_nonce_for_key(key: SigningKey, measurement_digest: bytes) -> str: + """report_data committing the gateway measurement (#552) instead of the chain root. + + Same first half as _make_nonce_for_key. The second half is the raw 32-byte + measurement digest, unreshaped: unlike AUDIT-006 this commits the digest itself + rather than a hash of it. + """ + x_b64 = base64.urlsafe_b64encode(key.public_key_bytes).rstrip(b"=").decode() + jwk_json = json.dumps( + {"crv": "Ed25519", "kty": "OKP", "x": x_b64}, + separators=(",", ":"), + sort_keys=True, + ).encode() + return (hashlib.sha256(jwk_json).digest() + measurement_digest).hex() + + def _make_signed_claim( policy_hash=POLICY_HASH, catalog_hash=CATALOG_HASH, provider="software-only", agent_identity: AgentIdentityInfo | None = None, + report_data: str | None = None, ): key = SigningKey() chain = AuditChain("test-session") measurement = "DEVELOPMENT_ONLY" if provider == "software-only" else "ab" * 32 # Bind both the key (report_data[:32]) and the chain root (report_data[32:64], # AUDIT-006) for hardware providers; software-only ignores report_data here. - report_data = ( + # A caller may override to bind something else, e.g. the #552 measurement. + report_data = report_data or ( _make_nonce_for_key(key, chain.chain_root) if provider != "software-only" else "00" * 32 @@ -835,3 +853,75 @@ def test_audit_chain_binding_software_only_not_applicable(): result = verify_trace_claim(claim_dict, _approved()) assert "audit_chain_binding" not in result.verified_fields assert "audit_chain_binding" not in result.unverified_fields + + +# ── #552: gateway measurement binding, step 7c ──────────────────────────────── +# +# _check_measurement_binding is unit-tested in test_measurement_report_binding.py. +# These cover the wiring into verify_trace_claim: that the parameter reaches the +# check, and that each outcome lands in the right result field. + +_MEASUREMENT = bytes(range(32)) +_APPROVED = ApprovedHashes(policy_bundle_hash=POLICY_HASH, tool_catalog_hash=CATALOG_HASH) + + +def _claim_binding_measurement(digest: bytes): + key = SigningKey() + return _make_signed_claim( + provider="sev-snp", + report_data=_make_measurement_nonce_for_key(key, digest), + ) + + +def test_measurement_binding_is_verified_when_it_matches(): + claim, _ = _claim_binding_measurement(_MEASUREMENT) + result = verify_trace_claim( + claim, _APPROVED, expected_gateway_measurement=_MEASUREMENT + ) + assert "measurement_binding" in result.verified_fields + + +def test_measurement_binding_mismatch_is_reported(): + claim, _ = _claim_binding_measurement(_MEASUREMENT) + result = verify_trace_claim( + claim, _APPROVED, expected_gateway_measurement=b"\xee" * 32 + ) + assert "measurement_binding" in result.unverified_fields + assert "does not match report_data[32:64]" in result.details["measurement_binding"] + + +def test_measurement_binding_accepts_the_digest_as_hex(): + claim, _ = _claim_binding_measurement(_MEASUREMENT) + result = verify_trace_claim( + claim, _APPROVED, expected_gateway_measurement="sha256:" + _MEASUREMENT.hex() + ) + assert "measurement_binding" in result.verified_fields + + +def test_measurement_binding_rejects_an_unparseable_expected_digest(): + """A caller passing junk must fail closed, not skip the check.""" + claim, _ = _claim_binding_measurement(_MEASUREMENT) + result = verify_trace_claim( + claim, _APPROVED, expected_gateway_measurement="not-a-digest" + ) + assert "measurement_binding" in result.unverified_fields + assert "not valid hex" in result.details["measurement_binding"] + + +def test_measurement_binding_is_advisory_in_software_only_mode(): + """A software-only claim carries no nonce at all, so there is nothing to bind.""" + claim, _ = _make_signed_claim(provider="software-only") + result = verify_trace_claim( + claim, _APPROVED, expected_gateway_measurement=_MEASUREMENT + ) + assert "measurement_binding" not in result.verified_fields + assert "software-only" in result.details["measurement_binding"] + + +def test_measurement_binding_is_skipped_when_no_expectation_is_supplied(): + """Opt-in: the expected digest is an out-of-band input the verifier must supply.""" + claim, _ = _claim_binding_measurement(_MEASUREMENT) + result = verify_trace_claim(claim, _APPROVED) + assert "measurement_binding" not in result.verified_fields + assert "measurement_binding" not in result.unverified_fields + assert "measurement_binding" not in result.details