From c900da6e2b4d290147a4aeed08c8f028a21e0702 Mon Sep 17 00:00:00 2001 From: Mohammed Zoheb Shaik Date: Mon, 24 Aug 2026 22:40:58 +0400 Subject: [PATCH 1/2] fix(tee): bind the gateway measurement into SEV-SNP and TDX report_data gateway_measurement() folds installed code, the policy bundle and the effective config 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 premise is hardware-rooted policy enforcement, nothing signed said which policy was running. No new commitment scheme is invented. make_measurement_bound_nonce() 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 outside the set: its NV index keeps an append-only history that report_data cannot. This replaces the 32 random salt bytes on those providers. Freshness survives 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. Refreshed on every policy-bundle reload, not only on the reloads that moved the hash. report_data holds one value and no history, so nothing in a report says whether it is current: the same digest signed now and that digest signed an hour ago are different assertions, and only the latest report reaches a verifier. PolicyEvaluator._maybe_reload calls the new on_reload hook whenever PolicyStore.reload_if_stale reports the bundle was re-read, and the hook runs refresh_measurement_binding. The cost is bounded by policy_reload_interval_seconds rather than by request rate, because reload_if_stale stamps its clock before the attempt, and it is zero in the default configuration where reloading is off. Recompute-and-compare is what catches a gateway that did not refresh, so it is 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 has to 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, and a mismatch is fatal in software-only mode too since the digest is computed the same way there. A refresh that cannot re-attest logs and keeps the previous report rather than refusing traffic. Failing closed there would trade a detectable weakness for an outage: stale report_data no longer matches the recomputed digest, so a verifier rejects the claim. That is the same trade AUDIT-006 already makes for a failed per-session attestation. Neither ctx.gateway_measurement nor ctx.attestation_report is touched until the new report is in hand, so a failure at any earlier step leaves both unchanged. They are two assignments rather than one atomic swap, which the docstring now states exactly instead of overclaiming: a concurrent reader could see the new measurement beside the old report for one interpreter step. Benign today because nothing reads the pair together, and recorded so it stops being an accident if something later does. 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. 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 where 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 the new check 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 scopes the audit-chain-root binding out, so this is left for the issue that takes that on. Also out of scope per the issue: validation on real SEV-SNP and TDX silicon. Tests: 40, software-only by design. SoftwareOnlyProvider echoes the nonce into report_data, which is the collector-side shape SEV-SNP and TDX give, so the round trip proves the contract and leaves the silicon question where the issue leaves it. They cover the five properties prototyped on the issue: a deterministic measurement across repeat calls, a one-character Cedar change moving the policy component and only the policy component, a full round trip whose independent recompute matches both halves of report_data, a mid-session reload producing a report that reflects the new measurement, and a stale pre-reload measurement correctly rejected on recompute. Verified by mutation, twelve guards altered one at a time. Each is killed: measuring only on tpm fails 2 tests; never using the measurement-bound nonce fails 1; dropping the post-reload hook call fails 2; refreshing only when the hash changed, which is the non-literal reading of the second bullet, fails 2; firing the hook on evaluations that did not reload fails 3; letting a failing hook escape fails 1; skipping re-attestation on an unchanged digest fails 1; installing the measurement before the report is in hand fails 1; skipping the nonce length check fails 1; re-hashing the digest verifier-side fails 3; accepting an expected digest of any length fails 1; and treating a software-only mismatch as not applicable fails 1. Three checks were added after auditing the change against the issue text rather than against itself. The reload hook was only ever exercised with reload_if_stale stubbed, so a real PolicyStore over a real on-disk bundle now reloads an unchanged bundle and asserts the hook still fires, which is the whole of the second bullet with nothing mocked on the seam. startup._jwk_thumbprint_sha256 and tee.base.jwk_thumbprint each build report_data[:32] and nothing forced them to agree, so a test does. And all three providers are asserted to place the whole 64-byte nonce in report_data, since the binding is only real if it reaches the field the hardware signs. Closes #552. Signed-off-by: Mohammed Zoheb Shaik --- CHANGELOG.md | 20 + docs/spec/attestation.md | 40 +- src/cmcp_runtime/cli.py | 12 +- src/cmcp_runtime/policy/evaluator.py | 58 +- src/cmcp_runtime/startup.py | 172 +++-- src/cmcp_runtime/tee/base.py | 41 + src/cmcp_runtime/tee/report_binding.py | 178 +++++ src/cmcp_verify/verify.py | 131 ++++ tests/unit/test_measurement_report_binding.py | 724 ++++++++++++++++++ 9 files changed, 1320 insertions(+), 56 deletions(-) create mode 100644 src/cmcp_runtime/tee/report_binding.py create mode 100644 tests/unit/test_measurement_report_binding.py 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..30963cc4 --- /dev/null +++ b/tests/unit/test_measurement_report_binding.py @@ -0,0 +1,724 @@ +"""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 From a2a8f942d0a1d827f6999c600c8ed3fcd1da6854 Mon Sep 17 00:00:00 2001 From: Mohammed Zoheb Shaik Date: Mon, 24 Aug 2026 23:23:11 +0400 Subject: [PATCH 2/2] test: cover the paths #552 added but did not exercise Codecov reported 80% patch coverage on #563, 26 added lines with no test. The gap was not evenly spread and one part of it mattered. The whole of step 7c in verify_trace_claim was untested. _check_measurement_binding had unit tests, but nothing ever passed expected_gateway_measurement through the public entry point, so the wiring could have been broken in either direction and the suite would have stayed green. Six tests now cover it: a match landing in verified_fields, a mismatch landing in unverified_fields with the reason, hex and sha256:-prefixed digests, an unparseable expectation failing closed rather than skipping the check, software-only staying advisory, and no expectation supplied leaving the field absent entirely. _make_signed_claim grows an optional report_data so a caller can bind something other than the audit-chain root. The rest were branches rather than wiring. The verifier's fail-closed paths for an undecodable nonce and a nonce too short to hold a commitment, and the software-only variants of both, which return advisory rather than fatal. The guard that stops a provider returning a non-AttestationReport from displacing a good report, which is the same guard AUDIT-006 uses. And the TPM fault paths in _extend_measurement: tpm2-pytss absent, a MeasurementUnavailable from the extend, and any other TPM fault, degrading in dev mode and aborting in production. Those TPM lines are not new behaviour. They read as added because #552 split _measure_gateway into _measure_gateway and _extend_measurement, and the diff attributes the moved body to this branch. Covering them is still right: they were untested before the split too. Every line #563 adds is now covered. Patch coverage measured locally by intersecting the branch diff with coverage's missing-line report, across all six changed source files: 0 uncovered added lines, down from 26. Suite: 50 tests in test_measurement_report_binding.py plus 6 in test_verify.py. Full unit run 1258 passed, with the same 9 pre-existing agent_manifest SDK failures as before this branch. Ruff and mypy clean. The twelve mutations from #552 are all still killed. Signed-off-by: Mohammed Zoheb Shaik --- tests/unit/test_measurement_report_binding.py | 191 ++++++++++++++++++ tests/unit/test_verify.py | 92 ++++++++- 2 files changed, 282 insertions(+), 1 deletion(-) diff --git a/tests/unit/test_measurement_report_binding.py b/tests/unit/test_measurement_report_binding.py index 30963cc4..c6bbf1be 100644 --- a/tests/unit/test_measurement_report_binding.py +++ b/tests/unit/test_measurement_report_binding.py @@ -722,3 +722,194 @@ def test_all_three_providers_put_the_whole_nonce_into_report_data() -> None: # 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