From 50a9530007c5eac41db6747d97b094cc8cad7a51 Mon Sep 17 00:00:00 2001 From: Imran Siddique Date: Fri, 31 Jul 2026 19:19:20 -0700 Subject: [PATCH] feat(tee): implement TPM attest() and carry verifiable evidence (#73) TpmProvider.detect() returned True on any host with a TPM device node while attest() raised unconditionally, so on every Azure Trusted Launch VM and most modern client hardware the provider was selected and then failed, with an error claiming no TPM was present when one was. AttestationReport also had no field for a quote, a signature, or a chain, so even a working collector could not supply anything a relying party could verify. attest() now produces a real quote, ported from cmcp's hardware-validated path: the platform attestation key at persistent handle 0x81000003 with its certificate chunk-read from NV 0x01C101D0 (a single read of a 1596-byte cert fails with TPM_RC_VALUE, since TPM2_NV_Read is bounded by TPM2_PT_NV_BUFFER_MAX), AIA chain assembly so verification stays offline, and a transient restricted signing key where no certified platform key exists. detect() is True only where attest() can actually run. The quote commits both the offered channel key and the nonce, so public_key and nonce become signed facts rather than assertions and sealing to "a key from a verified report" is actually hardware-rooted. Fields are length-prefixed: with a delimiter, a value containing it shifts the split without changing the digest, and nonce is an arbitrary caller-supplied string. Verification delegates to agent_manifest.verify_tpm_quote rather than being a third copy of one verifier (cmcp#447). cA2A keeps only TPMT_SIGNATURE, which agent-manifest does not model. Trust anchors stay caller-supplied per #440, with the hardware-validated Azure root as an opt-in constant. Two deliberate departures from cmcp: no SHA-1 PCR fallback and no unsigned-PCR-read tier. cmcp downgrades to software-only in each case; cA2A raises, because a report labelled sha256: that measured SHA-1 banks is a mislabel waiting to happen and a tpm report that can never verify is worse than an honest error. Not demonstrated: collect-then-verify in one process on hardware. Installing agent-manifest on the Azure test VM conflicts with the distribution's tpm2_pytss, so the two halves could not run together there. Recorded in LIMITATIONS.md. Refs #73 Co-Authored-By: Claude Opus 5 (1M context) --- CHANGELOG.md | 13 + LIMITATIONS.md | 4 +- docs/spec/attestation.md | 45 +- docs/spec/component-model.md | 2 +- docs/spec/failure-modes.md | 4 +- docs/tutorials/integrating-with-a2a.md | 4 +- docs/tutorials/reproducing-the-claims.md | 2 +- pyproject.toml | 5 + src/ca2a_runtime/tee/__init__.py | 7 +- src/ca2a_runtime/tee/base.py | 30 +- src/ca2a_runtime/tee/tpm.py | 459 +++++++++++++++++++- src/ca2a_verify/tpm.py | 256 +++++++++-- src/ca2a_verify/tpm_roots.py | 76 ++++ tests/unit/test_tpm.py | 12 +- tests/unit/test_tpm_attest.py | 519 +++++++++++++++++++++++ 15 files changed, 1373 insertions(+), 65 deletions(-) create mode 100644 src/ca2a_verify/tpm_roots.py create mode 100644 tests/unit/test_tpm_attest.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 5cf984f..0296b0f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- **`TpmProvider.attest` now produces a real TPM quote.** Previously `detect()` returned True on any host with a TPM device node while `attest()` raised unconditionally, so on every Azure Trusted Launch VM and most modern client hardware the provider was selected and then failed, with an error that claimed no TPM was present when one was (#73). The collector is ported from cmcp's hardware-validated path: it prefers the platform attestation key at persistent handle `0x81000003` with its certificate chunk-read from NV `0x01C101D0` (a single read of a 1596-byte certificate fails with `TPM_RC_VALUE`, because `TPM2_NV_Read` is bounded by `TPM2_PT_NV_BUFFER_MAX`), assembles the chain by walking each certificate's AIA extension so verification stays offline later, and falls back to a transient restricted signing key where no certified platform key exists. `detect()` now returns True only where `attest()` can actually run, and `AttestationUnsupported` names the piece that is actually missing. +- **`AttestationReport` can carry evidence.** Four optional fields (`raw_evidence`, `quote_signature`, `attestation_key_pem`, `attestation_key_chain_pem`), named to match cmcp's model so evidence is portable between the runtimes. Without them a report was `platform`, `measurement`, `public_key` and `nonce` with nothing signed behind it, so a relying party could not verify anything and the hardware tier could not supply verifiable evidence by construction. Absent on `software-only`, which has no evidence. +- **The quote commits the offered key, not just the nonce.** `extraData` carries `sha256("ca2a-tpm-v1|" || len32(public_key) || public_key || len32(nonce) || nonce)`, and `ca2a_verify.tpm.verify_tpm_report` re-derives it from the report's own fields and requires equality. That is what promotes `public_key` and `nonce` from assertion to signed fact, so sealing to "a key from a verified report" is actually rooted in hardware; committing the nonce alone would sign for freshness only. Fields are length-prefixed rather than delimiter-joined because a delimiter lets a value containing it shift the split without changing the digest, and `nonce` is an arbitrary caller-supplied string. The returned measurement is read out of the signed quote, and a report whose `measurement` disagrees with it is rejected. +- **`ca2a_verify.tpm.tpm_verifier(roots)`** returns a `Verifier` for `verify_offer`, so a TPM peer reaches `assurance="hardware"`. `ca2a_verify.tpm_roots.AZURE_VTPM_ROOT_2023_PEM` carries the one root validated on hardware as an opt-in constant; nothing is trusted implicitly, and supplying no root is refused rather than treated as trust-anything. + +### Changed + +- **TPM quote cryptography now delegates to `agent_manifest.verify_tpm_quote`** instead of being cA2A's own third copy of one verifier (cmcp#447). cA2A keeps only what agent-manifest does not model: `TPMT_SIGNATURE`, the envelope `tpm2_quote -s` and tpm2-pytss `signature.marshal()` emit, unwrapped to the bare signature agent-manifest takes. `verify_tpm_quote` keeps its signature and behavior, including the magic and attest-type checks, which agent-manifest enforces too. +- **No SHA-1 PCR fallback and no unsigned-PCR-read tier**, both deliberate departures from cmcp's collector. cmcp downgrades to `software-only` in each case; cA2A raises. A report labelled `sha256:` that measured SHA-1 banks is a mislabel waiting to happen, and a `tpm` report that can never verify is worse than an honest error. The collector also cross-checks its own PCR read against the quote's `pcrDigest`, so a PCR selection mismatch is caught before evidence ships. +- Docs corrected where they stated the opposite of the code: `detect()` returning False for every hardware provider was asserted in the attestation spec, the component model, failure modes, and two tutorials. + ### Changed - **BREAKING: TRACE records now carry the v0.2 profile** `tag:agentrust-io.com,2026:trace-v0.2`. Pins move to `agentrust-trace>=0.5` and `agentrust-trace-tests>=0.4,<0.5`, which have to move together: the conformance suite cut over rather than dual-accepting, so 0.4.0 of the suite fails a v0.1 record and 0.3.x fails a v0.2 one. The v0.1 URI named `agentrust.io`, a domain this project never controlled, which RFC 4151 does not permit for a tag URI (agentrust-io/trace-spec#107). Nothing else about the record format changed. diff --git a/LIMITATIONS.md b/LIMITATIONS.md index c85b103..8b181bb 100644 --- a/LIMITATIONS.md +++ b/LIMITATIONS.md @@ -12,7 +12,9 @@ cA2A is a pre-release profile in active design. This document states plainly wha - **Hardware-attested live binding.** The `verifier` seam in `ca2a_runtime.attestation` has now been driven off a real SEV-SNP quote on an Azure confidential VM: `verify_offer` returned `assurance="hardware"`, a payload was sealed to a channel key a hardware-verified measurement vouches for, and both a measurement mismatch and a stale nonce were rejected. See [docs/hardware-validation.md](docs/hardware-validation.md). Two gaps remain. First, the reference server/client still run in **software mode** by default (`assurance="none"`); the hardware path is a validated capability, not the default configuration, and there is no CLI listener (`ca2a start`) since serving is via `ca2a_runtime.transport.server.serve`. Second, attestation on that run was one-directional: a follow-on cross-operator run (an Azure SEV-SNP peer calling a GCP Intel TDX peer, recorded in the same document) had the caller appraise the callee's real TDX quote before sealing, but the callee did not appraise the caller in return. Mutual simultaneous attestation is the remaining step, and both peers were driven by one operator's harness. - **Sealed peer channel (hardware property).** The channel is implemented: a payload is sealed to the peer's attested X25519 key (X25519 ECDH, HKDF-SHA256, ChaCha20-Poly1305), and only the holder of the peer's private key can open it. On a live call the handshake now gates the seal on a channel key the caller has appraised, but in software mode that appraisal is `assurance="none"`. Until the seal is bound to a hardware-verified measurement (above), do not assume a payload is confined to a specific attested measurement. Adapter-decoded `sealed_payload` bytes are opaque ciphertext only. -- **Real hardware attestation.** The **SEV-SNP and Intel TDX verifiers now appraise genuine hardware evidence end to end**: a real Azure CVM SEV-SNP report (VCEK chain to the AMD ARK-Milan root, ECDSA-P384 report signature, measurement binding) and a real GCP C3 DCAP v4 TDX quote (PCK chain to the Intel SGX Root CA, QE binding, quote signature, MRTD binding), both fail-closed and both rejecting a tampered copy. Runs are recorded in [docs/hardware-validation.md](docs/hardware-validation.md). The **TPM 2.0 verifier** (AK chain to a caller-supplied vendor root, AK signature, magic/type, qualifying-data and PCR-digest binding) is implemented but synthetic-vector validated only. Quote *generation* still requires the respective hardware for all three. This validates the verifier, not a running attested peer: until the `verifier` seam in `ca2a_runtime.attestation` is driven off a live quote on a confidential VM, cA2A must not be described as attested across trust domains. +- **Real hardware attestation.** The **SEV-SNP and Intel TDX verifiers now appraise genuine hardware evidence end to end**: a real Azure CVM SEV-SNP report (VCEK chain to the AMD ARK-Milan root, ECDSA-P384 report signature, measurement binding) and a real GCP C3 DCAP v4 TDX quote (PCK chain to the Intel SGX Root CA, QE binding, quote signature, MRTD binding), both fail-closed and both rejecting a tampered copy. Runs are recorded in [docs/hardware-validation.md](docs/hardware-validation.md). The **TPM 2.0 verifier** (AK chain to a caller-supplied vendor root, AK signature, magic/type, and the key-and-nonce binding) is implemented, and its cryptography is delegated to `agent_manifest.verify_tpm_quote` rather than being a third copy. Quote *generation* still requires the respective hardware for SEV-SNP and TDX. This validates those verifiers, not a running attested peer: until the `verifier` seam in `ca2a_runtime.attestation` is driven off a live quote on a confidential VM, cA2A must not be described as attested across trust domains. +- **TPM collect-then-verify in one process.** `TpmProvider.attest` now produces a real quote (platform attestation key at a persistent handle with its certificate chunk-read from NV, AIA chain assembly, transient-key fallback), and `ca2a_verify.tpm.verify_tpm_report` appraises it. Both halves are exercised against synthetic self-consistent vectors, and a genuine Azure Trusted Launch vTPM quote parses and verifies under `CA2A_TPM_FIXTURE_DIR`. What has **not** been demonstrated is one process collecting and then verifying on hardware: on the Azure test VM, installing `agent-manifest` conflicts with the distribution's `tpm2_pytss` (older `cryptography`), and a venv fixes the install but then breaks the `tpm2_pytss` import, so the two could not run together there. Treat the TPM tier as validated in halves, not end to end. +- **TPM evidence proves key provenance only where a vendor root exists.** A quote signed by the transient fallback key is a verifiable signature but carries no certificate chain, so it proves nothing about *where* the key lives, and `verify_tpm_report` rejects it. Provenance therefore requires a platform-provisioned certified attestation key. Azure Trusted Launch provides one; a GCP Shielded VM does not (probed 2026-07-31: no EK certificate, no persistent handles, and `get-shielded-identity` returns a bare `ekPub` with no certificate). Client firmware TPM vendor roots are not yet published in a pinnable form. Separately, the TCG event log is 0 bytes on both Azure and GCP, so PCR values cannot be attributed to specific boot events on either cloud. ## Out of scope diff --git a/docs/spec/attestation.md b/docs/spec/attestation.md index 069d29f..6cd41ff 100644 --- a/docs/spec/attestation.md +++ b/docs/spec/attestation.md @@ -6,10 +6,19 @@ Before a peer is trusted with a delegated task, it proves it is running attested A provider implements `BaseProvider`: -- `detect()` returns whether the provider is available on the current host. +- `detect()` returns whether the provider is available on the current host. Available means `attest` can actually produce evidence here, not merely that the hardware exists: a provider that returns True and then raises would be selected and then fail. - `attest(public_key, nonce)` returns an `AttestationReport` binding `public_key` to the host's hardware measurement under `nonce`. -An `AttestationReport` carries `platform`, `measurement`, the bound `public_key`, and the `nonce`. +An `AttestationReport` carries `platform`, `measurement`, the bound `public_key`, and the `nonce`. Those four fields are what a report *claims*; on their own they are an assertion, since any peer can populate them with any values. Four further fields carry the evidence that makes them checkable, and they are named to match cmcp's report model so evidence is portable between the two runtimes: + +| Field | Contents | +|---|---| +| `raw_evidence` | the raw blob the hardware signed (for TPM, the bare `TPMS_ATTEST`) | +| `quote_signature` | the signature over `raw_evidence` (for TPM, a marshalled `TPMT_SIGNATURE`) | +| `attestation_key_pem` | the key that produced that signature | +| `attestation_key_chain_pem` | the leaf-first certificate chain for that key | + +All four are absent on `software-only`, which has no evidence by construction. A report claiming a hardware platform with no evidence cannot be verified, so it fails closed rather than being trusted. ## Providers @@ -18,7 +27,7 @@ An `AttestationReport` carries `platform`, `measurement`, the bound `public_key` | `software-only` | none | Available; for development and CI. Reports `platform: software-only`, never a hardware platform string. | | `sev-snp` | AMD SEV-SNP | Verifier implemented (see below). Report generation requires a real SEV-SNP guest. | | `tdx` | Intel TDX | Verifier implemented (see below). Quote generation requires a real TDX guest. | -| `tpm` | TPM 2.0 / vTPM | Verifier implemented (see below). Quote generation requires a real TPM. | +| `tpm` | TPM 2.0 / vTPM | Verifier and collector both implemented (see below). `attest` produces a real quote on a Linux host with a TPM and tpm2-pytss. | | `opaque` | OPAQUE Confidential Runtime | Tier 3, explicit opt-in, not auto-selected | ## SEV-SNP verification @@ -41,13 +50,37 @@ What is validated. The chain-verification path accepts the genuine self-signed I ## TPM verification -`ca2a_verify.tpm.verify_tpm_quote` appraises a TPM 2.0 quote (`TPMS_ATTEST`) offline: the AK certificate chain is verified to a trusted root, the AK signature over the attest blob is verified (ECDSA-SHA256 or RSA PKCS#1 v1.5), the structure is confirmed to be a TPM-generated quote (magic and type), and the qualifying data (the verifier's nonce) and the PCR digest (the platform measurement) are checked against expected values. +`ca2a_verify.tpm.verify_tpm_report` appraises a peer's TPM report offline: the AK certificate chain is verified to a trusted root, the AK signature over the attest blob is verified (ECDSA-SHA256 or RSA PKCS#1 v1.5), the structure is confirmed to be a TPM-generated quote (magic and type), and the key-and-nonce binding below is checked. `verify_tpm_quote` is the lower-level form taking an attest blob and a bare signature directly. + +The cryptography is not implemented in cA2A. Steps 1, 2 and 4 delegate to `agent_manifest.verify_tpm_quote`, the canonical hardware-validated implementation cA2A already depends on; three divergent copies of one TPM verifier is the problem being retired (cmcp#447). What cA2A keeps is the piece agent-manifest does not model: `TPMT_SIGNATURE`, the envelope `tpm2_quote -s` and tpm2-pytss `signature.marshal()` actually emit, which is unwrapped to the bare signature agent-manifest takes. + +### The signed binding + +A TPM quote commits caller-chosen bytes in its `extraData` (qualifying data) field. cA2A commits **both** the offered channel public key and the nonce: + +``` +extraData = sha256("ca2a-tpm-v1|" || len32(public_key) || public_key || len32(nonce) || nonce) +``` + +Committing the nonce alone would sign for freshness only, leaving `public_key` an unsigned assertion, and sealing a payload "to a key from a verified report" would not actually be rooted in hardware. The verifier re-derives this value from the report's own fields and requires equality, which is what promotes `public_key` and `nonce` from claim to signed fact. A report whose key was substituted after the quote was taken is rejected. + +Two encoding details are load-bearing. The value is hashed to 32 bytes rather than carried raw because `TPM2B_DATA` is capped below 64 bytes on some platforms (Azure returns `TPM_RC_SIZE`). Each field is length-prefixed rather than delimiter-joined because with a delimiter a value containing it shifts the split without changing the digest, so `("a|b", "c")` and `("a", "b|c")` would commit identical bytes and a peer could bind a key other than the one it appears to offer. `nonce` is an arbitrary caller-supplied string, so that is reachable rather than theoretical. + +The measurement is `sha256:` followed by the quote's own `pcrDigest`, over PCRs 0-7 in the SHA-256 bank. The collector separately reads those PCRs and requires its digest to equal the quoted one, which catches a PCR selection mismatch before evidence ships. The verifier returns the measurement read out of the signed quote rather than the report's `measurement` field, and rejects a report where the two disagree. + +### Trust anchors + +TPM attestation keys chain to per-vendor roots, not to one published root the way SEV-SNP and TDX do. cA2A does not decide which vendors a deployment trusts: the verifier takes caller-supplied roots and consults nothing implicitly. `ca2a_verify.tpm_roots.AZURE_VTPM_ROOT_2023_PEM` is the one root validated against hardware, available so a deployment on that platform need not re-derive it, but trusting it stays an explicit import. Supplying no root at all is refused, because a chain validated against no anchor would accept any self-consistent chain. + +### Two deliberate differences from cmcp's collector + +cmcp falls back to the SHA-1 PCR bank and downgrades the report to `software-only`; cA2A requires the SHA-256 bank and raises instead, because a report labelled `sha256:` that measured SHA-1 banks is a mislabel waiting to happen. cmcp can also emit a report whose only evidence is an unsigned PCR read, marked software-only; in cA2A, failing to produce a signed quote raises, because the platform string on a cA2A report is the provider's identity and a `tpm` report that can never verify is worse than an honest error. -What is validated. Unlike SEV-SNP and TDX, TPM attestation keys chain to per-vendor EK roots, so there is no single published root to validate against; the caller supplies the vendor roots it trusts, and the verifier is exercised against synthetic self-consistent vectors. Producing a quote (`TpmProvider.attest`) fails closed off a real TPM. +What is validated. The collector's checks, the TPM interaction shapes, and report verification end to end are exercised against synthetic self-consistent vectors in `tests/unit/test_tpm_attest.py`. A quote from a genuine Azure Trusted Launch vTPM parses and verifies under `CA2A_TPM_FIXTURE_DIR`. What has **not** been demonstrated is a full collect-then-verify pass in one process on hardware: on the Azure test VM, installing `agent-manifest` conflicts with the distribution's `tpm2_pytss`, so the collector and the verifier could not be run together there. See [LIMITATIONS.md](../../LIMITATIONS.md). ## Fail closed -Providers without a backend `detect()` to False, so they are never selected automatically, and verification fails closed when evidence is absent or invalid. This is deliberate: cA2A must not be described as attested across trust domains until a backend verifies a quote against a golden measurement on real hardware. See [LIMITATIONS.md](../../LIMITATIONS.md). +A provider `detect()`s to True only where `attest` works on that host, and verification fails closed when evidence is absent or invalid. `sev-snp` and `tdx` have verifiers but no collector yet, so their `attest` raises `AttestationUnsupported`; `software-only` returns False from `detect` so a no-guarantee posture is always an explicit choice. See [LIMITATIONS.md](../../LIMITATIONS.md). ## Why this is the critical path diff --git a/docs/spec/component-model.md b/docs/spec/component-model.md index 0494dd4..ca696db 100644 --- a/docs/spec/component-model.md +++ b/docs/spec/component-model.md @@ -22,7 +22,7 @@ The cA2A runtime is a set of small, composable modules under `src/`. Each maps t ### tee -`ca2a_runtime.tee.base` defines the provider interface and evidence model. `AttestationReport` is a frozen dataclass binding a `public_key` to a `measurement` under a `nonce` on a named `platform`. `BaseProvider` is an ABC with `detect()` and `attest(public_key, nonce)`. Real hardware providers (TPM, SEV-SNP, TDX, OPAQUE) are Tier 3 and not implemented; their `detect()` returns False so they are never auto-selected, and verification fails closed. See [attestation](attestation.md). +`ca2a_runtime.tee.base` defines the provider interface and evidence model. `AttestationReport` is a frozen dataclass binding a `public_key` to a `measurement` under a `nonce` on a named `platform`, plus four optional evidence fields (`raw_evidence`, `quote_signature`, `attestation_key_pem`, `attestation_key_chain_pem`) that make those claims checkable. `BaseProvider` is an ABC with `detect()` and `attest(public_key, nonce)`, and the two must agree: `detect()` is True only where `attest()` works. TPM has a collector; SEV-SNP, TDX and OPAQUE have verifiers but no collector yet, so their `attest()` raises and verification fails closed. See [attestation](attestation.md). ### config diff --git a/docs/spec/failure-modes.md b/docs/spec/failure-modes.md index 38d9110..63acd64 100644 --- a/docs/spec/failure-modes.md +++ b/docs/spec/failure-modes.md @@ -104,7 +104,9 @@ Peer attestation proves a peer runs measured code before a task is trusted to it - `ATTESTATION_UNSUPPORTED` (`AttestationUnsupported`): no attestation backend is available for the requested platform. - `ATTESTATION_FAILED` (`AttestationFailed`): a backend ran but the measurement or quote did not verify. -Real hardware providers (`tpm`, `sev-snp`, `tdx`, `opaque`) return `False` from `detect()` in this release, so they are never auto-selected, and verification against an absent backend fails closed rather than assuming a peer is trustworthy. The `software-only` provider is for development and CI and never reports a hardware platform string. Until at least one real hardware backend verifies a quote, cA2A must not be described as attested across a trust boundary. This is Tier 3 on the [roadmap](../../ROADMAP.md) and a shared critical path with cmcp. +A provider returns `True` from `detect()` only where `attest()` can actually produce evidence on that host, so a provider is never selected and then found broken. `tpm` has a collector: on a Linux host with a TPM and tpm2-pytss it produces a real quote, and `AttestationUnsupported` names the actual missing piece rather than claiming no TPM is present. `sev-snp`, `tdx` and `opaque` have verifiers but no collector yet, so their `attest()` raises. The `software-only` provider returns `False` from `detect()`, so a no-guarantee posture is always an explicit choice, and it never reports a hardware platform string. + +Verification fails closed on absent or invalid evidence. A report claiming a hardware platform with no `raw_evidence` or `quote_signature`, or with no certificate chain for the key that signed it, is rejected rather than trusted. The remaining gap is a full collect-then-verify pass in one process on hardware, which is Tier 3 on the [roadmap](../../ROADMAP.md) and a shared critical path with cmcp. ## Missing or failed attestation is denial diff --git a/docs/tutorials/integrating-with-a2a.md b/docs/tutorials/integrating-with-a2a.md index a5eff79..94c5ba0 100644 --- a/docs/tutorials/integrating-with-a2a.md +++ b/docs/tutorials/integrating-with-a2a.md @@ -137,7 +137,7 @@ ca2a validate-config --config ca2a.yaml # ok: provider=auto enforcement=enforcing ``` -`Ca2aConfig` accepts `provider` from `auto`, `tpm`, `sev-snp`, `tdx`, `opaque`, `software-only`, and `enforcement_mode` from `enforcing`, `advisory`, `silent`. Hardware providers `detect()` to `False` until their backend lands, so `auto` never selects one silently. `max_delegation_depth` is passed straight through to `verify_chain`. The config is parsed and validated now; the runtime that consumes it on a live inbound call is Tier 2. +`Ca2aConfig` accepts `provider` from `auto`, `tpm`, `sev-snp`, `tdx`, `opaque`, `software-only`, and `enforcement_mode` from `enforcing`, `advisory`, `silent`. A provider `detect()`s to `True` only where it can actually attest, so `auto` never selects one that would then fail; `software-only` is never auto-selected at all. `max_delegation_depth` is passed straight through to `verify_chain`. The config is parsed and validated now; the runtime that consumes it on a live inbound call is Tier 2. ## What awaits Tier 2 and Tier 3 @@ -146,7 +146,7 @@ Do not present any of the following as usable. They are design today, and the co | Capability | Step | Status | What happens today | |---|---|---|---| | Runtime peer-call enforcement | reads chain off the wire, gates the inbound call | Tier 2, not built | No live path accepts a credential on an inbound A2A task | -| Peer attestation check | 2 | Tier 3, not built | `detect()` is `False` for all hardware providers; verification fails closed (`ATTESTATION_UNSUPPORTED` / `ATTESTATION_FAILED`) | +| Peer attestation check | 2 | Tier 3, partly built | `tpm` collects and verifies a real quote; `sev-snp` / `tdx` / `opaque` verify but cannot collect. No live inbound path runs the check yet, and verification fails closed (`ATTESTATION_UNSUPPORTED` / `ATTESTATION_FAILED`) | | Cedar scope intersection | 3 | Tier 2, not built | Runtime does not consult a policy; `policy_bundle_path` only reserves the surface | | Sealed payload channel | 4 | Tier 2, fails closed | `SealedChannel.seal` / `open` raise `SEALED_CHANNEL_ERROR` rather than send plaintext | | Live provenance emission | 5b | Tier 2, not built | The runtime does not emit and link records automatically on the inbound path | diff --git a/docs/tutorials/reproducing-the-claims.md b/docs/tutorials/reproducing-the-claims.md index fe36d2f..4ba016e 100644 --- a/docs/tutorials/reproducing-the-claims.md +++ b/docs/tutorials/reproducing-the-claims.md @@ -139,7 +139,7 @@ Exit code 0. See the [Sealed Channel](../spec/sealed-channel.md) page. python experiments/claim6-cross-operator-attestation/run.py ``` -This SKIPs. Real hardware attestation backends (SEV-SNP VCEK chain, Intel TDX quote via QVL/PCS, TPM AK cert plus checkquote) are Tier 3 and not implemented. Every `BaseProvider.detect()` returns `False`, so no provider can produce a quote and no counterparty can verify one. The script probes for a provider, finds none, prints a software-only illustration of the `AttestationReport` shape clearly marked as carrying no assurance, then SKIPs. +This SKIPs. The SEV-SNP and TDX collectors are Tier 3 and not implemented, so on the SEV-SNP path this experiment exercises no provider that can produce a quote. (`TpmProvider.attest` does now produce a real quote on a Linux host with a TPM, but this experiment is written against SEV-SNP.) The script probes for a provider, finds none, prints a software-only illustration of the `AttestationReport` shape clearly marked as carrying no assurance, then SKIPs. ```text KEY RESULT: SKIP: cross-operator attestation is gated on Tier 3 (real diff --git a/pyproject.toml b/pyproject.toml index 930f030..47b0609 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -38,6 +38,11 @@ dependencies = [ ] [project.optional-dependencies] +# TPM quote collection. Only the collector needs these bindings; verifying a +# peer's quote does not, so this stays out of the base dependencies. +tpm = [ + "tpm2-pytss>=2.2; sys_platform == 'linux'", +] dev = [ "pytest>=8.0", "pytest-asyncio>=0.23", diff --git a/src/ca2a_runtime/tee/__init__.py b/src/ca2a_runtime/tee/__init__.py index d033672..8a8f7e8 100644 --- a/src/ca2a_runtime/tee/__init__.py +++ b/src/ca2a_runtime/tee/__init__.py @@ -1,8 +1,11 @@ """TEE provider abstraction for peer attestation. cA2A reuses the pluggable provider model from cmcp: a provider produces an -attestation report that binds a public key to a hardware measurement. Real -hardware backends are Tier 3 (see ROADMAP.md) and fail closed until implemented. +attestation report that binds a public key to a hardware measurement. + +:class:`~ca2a_runtime.tee.tpm.TpmProvider` collects a real quote. The SEV-SNP and +TDX collectors are Tier 3 (see ROADMAP.md) and fail closed until implemented, +though their verifiers appraise genuine hardware evidence. """ from ca2a_runtime.tee.base import AttestationReport, BaseProvider diff --git a/src/ca2a_runtime/tee/base.py b/src/ca2a_runtime/tee/base.py index 2db36e8..24dfcc2 100644 --- a/src/ca2a_runtime/tee/base.py +++ b/src/ca2a_runtime/tee/base.py @@ -8,20 +8,42 @@ @dataclass(frozen=True) class AttestationReport: - """An attestation report binding a public key to a hardware measurement.""" + """An attestation report binding a public key to a hardware measurement. + + The first four fields are what the report *claims*. On their own they are an + assertion: any peer can populate them with any values. The evidence fields + are what makes them checkable, so a relying party can reach + ``assurance="hardware"`` (see :mod:`ca2a_runtime.attestation`). + + Evidence field names match cmcp's report model so evidence is portable + between the two runtimes rather than needing a translation layer. + """ platform: str measurement: str public_key: str # raw hex of the key bound to this measurement nonce: str + # Signed evidence. All absent on the software-only provider, which has none + # by construction. A report claiming a hardware platform with no evidence + # cannot verify, so it fails closed rather than being trusted. + raw_evidence: bytes | None = None # the raw blob the hardware signed + quote_signature: bytes | None = None # the signature over raw_evidence + attestation_key_pem: bytes | None = None # the key that produced it + attestation_key_chain_pem: bytes | None = None # leaf-first chain for that key class BaseProvider(ABC): """Normalized interface every TEE provider implements. - Real hardware providers (TPM, SEV-SNP, TDX, OPAQUE) are Tier 3 and are not - implemented in this release; ``detect`` returns False for them so they are - never selected automatically. + ``detect`` and ``attest`` must agree: a provider returns True from ``detect`` + only where ``attest`` can actually produce evidence on that host. Returning + True and then raising is the one combination to avoid, because the provider + gets selected and then fails. + + :class:`~ca2a_runtime.tee.tpm.TpmProvider` implements ``attest``. SEV-SNP and + TDX do not yet: their collectors are Tier 3 (see ROADMAP.md), so they raise + :class:`~ca2a_runtime.errors.AttestationUnsupported` while their verifiers are + exercised against report vectors. """ platform: str = "base" diff --git a/src/ca2a_runtime/tee/tpm.py b/src/ca2a_runtime/tee/tpm.py index 44f5ec2..af833b3 100644 --- a/src/ca2a_runtime/tee/tpm.py +++ b/src/ca2a_runtime/tee/tpm.py @@ -7,19 +7,44 @@ :mod:`ca2a_verify.tpm`. Unlike AMD SEV-SNP and Intel TDX, TPM attestation keys chain to per-vendor EK -roots rather than one published root, so there is no single real root to -validate against. The verifier is exercised against synthetic self-consistent -vectors, and a production deployment supplies its own trusted vendor roots. -Producing a quote requires a real TPM, so :meth:`TpmProvider.attest` fails closed. +roots rather than one published root, so there is no single real root to validate +against. The verifier therefore takes the roots the caller trusts; +:mod:`ca2a_verify.tpm_roots` ships the one root cA2A has validated on hardware as +an opt-in constant rather than a default. + +:meth:`TpmProvider.attest` produces a real quote. What it binds is the point: +``extraData`` commits :func:`tpm_qualifying_data`, a digest over *both* the +channel public key and the caller's nonce. Without that, a report's +``public_key`` field would stay an unsigned assertion and sealing a payload to +"a key from a verified report" would not actually be rooted in hardware. + +Two deliberate differences from cmcp's collector, which this is otherwise ported +from: + +- **No SHA-1 fallback.** cmcp falls back to the SHA-1 PCR bank and downgrades the + report to software-only. cA2A requires the SHA-256 bank and raises instead. A + report labelled ``sha256:`` that measured SHA-1 banks is a mislabel waiting to + happen, and cA2A has no legacy deployments to keep working. +- **No unsigned-PCR-read tier.** cmcp can emit a report whose only evidence is an + unsigned PCR read, marked software-only. Here, failing to produce a signed + quote raises: the platform string on a cA2A report is the provider's identity + and a ``tpm`` report that can never verify is worse than an honest error. """ from __future__ import annotations +import hashlib +import logging +import sys from dataclasses import dataclass +from pathlib import Path +from typing import Any from ca2a_runtime.errors import AttestationFailed, AttestationUnsupported from ca2a_runtime.tee.base import AttestationReport, BaseProvider +logger = logging.getLogger(__name__) + TPM_GENERATED_VALUE = 0xFF544347 TPM_ST_ATTEST_QUOTE = 0x8018 CLOCK_INFO_LEN = 17 @@ -27,6 +52,57 @@ TPM_DEVICES = ("/dev/tpmrm0", "/dev/tpm0") +# The PCR selection quoted, and the bank the measurement is labelled with. +QUOTE_PCR_SELECTION = "sha256:0,1,2,3,4,5,6,7" +_PCR_COUNT = 8 + +# Domain-separated binding committed into the quote's extraData. Versioned +# because it is wire format: a peer and its verifier must derive it identically. +# See docs/spec/attestation.md. +_QUALIFYING_DATA_PREFIX = b"ca2a-tpm-v1|" + +# Platforms that provision an attestation key expose it at a persistent handle +# with a certificate in NV. Azure Trusted Launch uses these two; both are +# TCG-conventional ranges rather than Azure inventions, so probing is harmless +# elsewhere and simply finds nothing. +_PLATFORM_AK_HANDLE = 0x81000003 +_PLATFORM_AK_CERT_NV_INDEX = 0x01C101D0 + +# TPM2_NV_Read is bounded by TPM2_PT_NV_BUFFER_MAX, so a certificate larger than +# that must be read in chunks. +_NV_READ_CHUNK_BYTES = 512 +_TPM2_CAP_TPM_PROPERTIES = 0x00000006 +_TPM2_PT_NV_BUFFER_MAX = 0x0000012B + +# Walking the certificate AIA extension at collection time is what lets a relying +# party verify offline later: the chain travels with the evidence. +_AIA_FETCH_TIMEOUT_SECONDS = 5 +_AIA_MAX_DEPTH = 4 +_AIA_CA_ISSUERS_OID = "1.3.6.1.5.5.7.48.2" + + +def tpm_qualifying_data(public_key: str, nonce: str) -> bytes: + """Return the 32 bytes a cA2A TPM quote commits in ``extraData``. + + Binds the channel public key and the nonce together, so one signature covers + freshness *and* which key is being offered. A caller that seals to the key in + a verified report is then sealing to a key the TPM signed for. + + Hashed rather than concatenated raw because ``TPM2B_DATA`` is capped below 64 + bytes on some platforms (Azure returns ``TPM_RC_SIZE``); 32 bytes always fits. + + Each field is length-prefixed rather than separated by a delimiter. With a + delimiter, a value containing it moves the split without changing the digest: + ``("a|b", "c")`` and ``("a", "b|c")`` would commit identical bytes, so a peer + could bind a key other than the one it appears to offer. ``nonce`` is an + arbitrary caller-supplied string, so that is reachable rather than theoretical. + """ + parts = [] + for field in (public_key.encode(), nonce.encode()): + parts.append(len(field).to_bytes(4, "big")) + parts.append(field) + return hashlib.sha256(_QUALIFYING_DATA_PREFIX + b"".join(parts)).digest() + def _read_u16(buf: bytes, pos: int) -> tuple[int, int]: if pos + 2 > len(buf): @@ -80,19 +156,380 @@ def parse(cls, blob: bytes) -> TpmQuote: ) +def _tpm2_pytss_available() -> bool: + """True when the tpm2-pytss bindings can be imported.""" + try: + import tpm2_pytss # noqa: F401 + except Exception as exc: # noqa: BLE001 - a broken install must read as absent + logger.debug("tpm2-pytss unavailable: %s", exc) + return False + return True + + class TpmProvider(BaseProvider): - """TPM 2.0 provider. Quote generation requires a real TPM.""" + """TPM 2.0 provider. Produces a real quote over PCRs 0-7 in the SHA-256 bank. + + ``detect`` reports True only where ``attest`` can actually run: Linux, a TPM + device node, and the tpm2-pytss bindings present. Reporting True on a host + where attestation then fails is the failure mode this provider used to have. + """ platform = "tpm" @classmethod def detect(cls) -> bool: - import os - - return any(os.path.exists(dev) for dev in TPM_DEVICES) + if sys.platform != "linux": + return False + if not any(Path(dev).exists() for dev in TPM_DEVICES): + return False + return _tpm2_pytss_available() def attest(self, public_key: str, nonce: str) -> AttestationReport: - raise AttestationUnsupported( - "TPM quote generation requires a real TPM", - detail="no TPM device present; run on a host with a TPM 2.0 / vTPM", + """Quote PCRs 0-7, committing ``public_key`` and ``nonce`` in extraData. + + Raises :class:`AttestationUnsupported` when this host cannot produce a + quote at all, and :class:`AttestationFailed` when it can but the attempt + did not yield verifiable evidence. + """ + self._require_host() + return self._collect(public_key, nonce) + + # ── host preconditions ──────────────────────────────────────────────────── + + @classmethod + def _require_host(cls) -> None: + """Fail with the actual reason, never a generic "no TPM present".""" + if sys.platform != "linux": + raise AttestationUnsupported( + "TPM quote generation is only implemented on Linux", + detail=f"running on {sys.platform}; the TPM resource manager is a Linux device", + ) + present = [dev for dev in TPM_DEVICES if Path(dev).exists()] + if not present: + raise AttestationUnsupported( + "TPM quote generation requires a TPM device", + detail=( + f"none of {', '.join(TPM_DEVICES)} exist; run on a host with a " + "TPM 2.0 or vTPM" + ), + ) + if not _tpm2_pytss_available(): + raise AttestationUnsupported( + "TPM quote generation requires the tpm2-pytss bindings", + detail=( + f"a TPM is present at {present[0]} but tpm2-pytss is not importable; " + "install the 'tpm' extra" + ), + ) + + # ── collection ──────────────────────────────────────────────────────────── + + def _collect(self, public_key: str, nonce: str) -> AttestationReport: + from tpm2_pytss.ESAPI import ESAPI + from tpm2_pytss.types import TPM2B_DATA, TPML_PCR_SELECTION + + qualifying_data = tpm_qualifying_data(public_key, nonce) + + with ESAPI() as ectx: + pcr_digest_local = self._read_pcrs(ectx, TPML_PCR_SELECTION) + try: + ak_handle, ak_public, chain_pem, transient = self._attestation_key(ectx) + except Exception as exc: # noqa: BLE001 - keep the error contract + raise AttestationFailed( + "no attestation key was available to sign a quote", + detail=f"{type(exc).__name__}: {exc}", + ) from exc + try: + quoted, signature = ectx.quote( + ak_handle, + QUOTE_PCR_SELECTION, + TPM2B_DATA(qualifying_data), + ) + attest_blob = bytes(quoted.attestationData) + signature_blob = bytes(signature.marshal()) + ak_pem = bytes(ak_public.to_pem()) + except Exception as exc: # noqa: BLE001 - any TPM fault means no evidence + raise AttestationFailed( + "TPM2_Quote did not produce a signed quote", + detail=f"{type(exc).__name__}: {exc}", + ) from exc + finally: + if transient: + try: + ectx.flush_context(ak_handle) + except Exception as exc: # noqa: BLE001 + logger.debug("flushing the transient attestation key failed: %s", exc) + + quote = TpmQuote.parse(attest_blob) + self._check_quote(quote, qualifying_data, pcr_digest_local) + + if transient: + logger.warning( + "Quoted with a transient attestation key: this host provisions no " + "certified platform key, so the report carries no chain and a verifier " + "using vendor roots will reject it." + ) + + return AttestationReport( + platform=self.platform, + # The digest the TPM signed, not a locally recomputed one. + measurement="sha256:" + quote.pcr_digest.hex(), + public_key=public_key, + nonce=nonce, + raw_evidence=attest_blob, + quote_signature=signature_blob, + attestation_key_pem=ak_pem, + attestation_key_chain_pem=chain_pem, + ) + + @staticmethod + def _read_pcrs(ectx: Any, selection_cls: Any) -> bytes: + """Return sha256 over the concatenated SHA-256 PCRs 0-7. + + ``pcr_read`` returns a ``TPML_DIGEST``, so its ``digests`` are iterated + once. Treating it as a list of banks yields ``bytes`` of a structure and + silently corrupts the measurement. + """ + try: + _, _, digests = ectx.pcr_read(selection_cls.parse(QUOTE_PCR_SELECTION)) + raw = [bytes(digest) for digest in digests.digests] + except Exception as exc: # noqa: BLE001 + raise AttestationUnsupported( + "the SHA-256 PCR bank could not be read", + detail=( + f"{type(exc).__name__}: {exc}. cA2A does not fall back to SHA-1: a " + "report labelled sha256 must measure the SHA-256 bank." + ), + ) from exc + if len(raw) < _PCR_COUNT: + raise AttestationFailed( + "the TPM returned fewer PCRs than were selected", + detail=f"got {len(raw)}, expected {_PCR_COUNT}", + ) + return hashlib.sha256(b"".join(raw[:_PCR_COUNT])).digest() + + @staticmethod + def _check_quote(quote: TpmQuote, qualifying_data: bytes, pcr_digest_local: bytes) -> None: + """Check the TPM signed what this collector believes it asked for. + + The PCR cross-check catches a selection mismatch at collection time: the + quote's ``pcrDigest`` is the TPM's own digest over the selected PCRs, so it + must equal the digest computed from the separate PCR read. If it does not, + the measurement about to be shipped describes different PCRs than were + measured, and the peer would be attesting to the wrong thing. + """ + if quote.magic != TPM_GENERATED_VALUE: + raise AttestationFailed( + "the quote is not TPM-generated", detail=f"magic={quote.magic:#x}" + ) + if quote.attest_type != TPM_ST_ATTEST_QUOTE: + raise AttestationFailed( + "the attestation is not a quote", detail=f"type={quote.attest_type:#x}" + ) + if quote.qualifying_data != qualifying_data: + raise AttestationFailed( + "the quote does not commit the requested key and nonce binding", + detail="extraData does not match the derived qualifying data", + ) + if quote.pcr_digest != pcr_digest_local: + raise AttestationFailed( + "the quoted PCR digest does not match the PCRs that were read", + detail=( + f"quote={quote.pcr_digest.hex()} read={pcr_digest_local.hex()}; the " + "measurement would describe a different PCR selection" + ), + ) + + # ── attestation key ─────────────────────────────────────────────────────── + + def _attestation_key(self, ectx: Any) -> tuple[Any, Any, bytes | None, bool]: + """Return ``(handle, public, chain_pem, transient)`` for the signing key. + + Prefers the platform attestation key: one the platform provisioned and + certified, at a persistent handle with its certificate in NV. Only that + key's signature says anything about *where* the key lives, because the + certificate chains to a vendor root. + + Falls back to a transient restricted signing key. That still yields a + verifiable signature but no provenance, so it is returned with no chain and + cannot be mistaken for the stronger tier. + """ + cert_der = self._read_nv(ectx, _PLATFORM_AK_CERT_NV_INDEX) + if cert_der is not None: + try: + handle = ectx.tr_from_tpmpublic(_PLATFORM_AK_HANDLE) + public, _, _ = ectx.read_public(handle) + chain = self._chain_from_leaf(cert_der) + if chain and self._certifies(chain, public): + logger.info( + "Using the platform attestation key at %#x with its certificate " + "from NV %#x.", + _PLATFORM_AK_HANDLE, + _PLATFORM_AK_CERT_NV_INDEX, + ) + return handle, public, chain, False + logger.warning( + "The certificate at NV %#x does not certify the key at %#x; falling " + "back to a transient attestation key.", + _PLATFORM_AK_CERT_NV_INDEX, + _PLATFORM_AK_HANDLE, + ) + except Exception as exc: # noqa: BLE001 + logger.warning( + "The platform attestation key is unusable (%s); falling back to a " + "transient key.", + exc, + ) + + handle, public = self._create_attestation_key(ectx) + return handle, public, None, True + + @staticmethod + def _read_nv(ectx: Any, index: int) -> bytes | None: + """Read an NV index in full, or None when it is not defined or readable. + + Chunked because ``TPM2_NV_Read`` is bounded by ``TPM2_PT_NV_BUFFER_MAX``. + Requesting the whole object at once fails with ``TPM_RC_VALUE`` on the size + parameter, which is why a 1596-byte certificate cannot be fetched in one + call even though the index is plainly readable. + """ + try: + handle = ectx.tr_from_tpmpublic(index) + public, _ = ectx.nv_read_public(handle) + total = int(public.nvPublic.dataSize) + except Exception as exc: # noqa: BLE001 + logger.debug("NV index %#x is not defined: %s", index, exc) + return None + + chunk = _NV_READ_CHUNK_BYTES + try: + caps = ectx.get_capability(_TPM2_CAP_TPM_PROPERTIES, _TPM2_PT_NV_BUFFER_MAX, 1) + for prop in caps[1].data.tpmProperties: + if int(prop.property) == _TPM2_PT_NV_BUFFER_MAX: + chunk = min(chunk, int(prop.value)) + break + except Exception as exc: # noqa: BLE001 + logger.debug("NV buffer max unavailable, using %d: %s", chunk, exc) + + out = bytearray() + try: + while len(out) < total: + want = min(chunk, total - len(out)) + out += bytes(ectx.nv_read(handle, want, len(out))) + except Exception as exc: # noqa: BLE001 + logger.debug("NV index %#x read failed at offset %d: %s", index, len(out), exc) + return None + return bytes(out) + + @staticmethod + def _certifies(chain_pem: bytes, tpm_public: Any) -> bool: + """True when the chain's leaf carries the TPM key's public key.""" + from cryptography import x509 + from cryptography.hazmat.primitives import serialization + + try: + leaf = x509.load_pem_x509_certificates(chain_pem)[0] + except Exception as exc: # noqa: BLE001 + logger.debug("could not load the leaf certificate: %s", exc) + return False + return leaf.public_key().public_bytes( + serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo + ) == bytes(tpm_public.to_pem()) + + @staticmethod + def _chain_from_leaf(leaf_der: bytes) -> bytes: + """Build a leaf-first PEM chain by following each certificate's AIA extension. + + Assembled at collection time on purpose: shipping the chain with the + evidence is what keeps verification offline later. A self-signed + certificate or a missing AIA ends the walk, and whatever was gathered is + returned, so a partial chain still travels rather than being discarded. + """ + import urllib.request + + from cryptography import x509 + from cryptography.hazmat.primitives.serialization import Encoding, pkcs7 + + def load_any(data: bytes) -> list[x509.Certificate]: + for loader in (x509.load_der_x509_certificate, x509.load_pem_x509_certificate): + try: + return [loader(data)] + except Exception as exc: # noqa: BLE001 + logger.debug("not an X.509 certificate via %s: %s", loader.__name__, exc) + for bundle_loader in ( + pkcs7.load_der_pkcs7_certificates, + pkcs7.load_pem_pkcs7_certificates, + ): + try: + found = list(bundle_loader(data)) + except Exception as exc: # noqa: BLE001 + logger.debug("not a PKCS#7 bundle via %s: %s", bundle_loader.__name__, exc) + continue + if found: + return found + return [] + + chain = load_any(leaf_der) + if not chain: + logger.warning( + "The attestation key certificate parsed as neither X.509 nor PKCS#7; no " + "chain will be shipped." + ) + return b"" + + while len(chain) < _AIA_MAX_DEPTH: + current = chain[-1] + if current.subject == current.issuer: + break + try: + aia = current.extensions.get_extension_for_class( + x509.AuthorityInformationAccess + ).value + except x509.ExtensionNotFound: + break + issuer = None + for access in aia: + if access.access_method.dotted_string != _AIA_CA_ISSUERS_OID: + continue + url = access.access_location.value + if not isinstance(url, str) or not url.startswith(("http://", "https://")): + continue + try: + # http(s) only, checked immediately above. + with urllib.request.urlopen( # noqa: S310 # nosec B310 + url, timeout=_AIA_FETCH_TIMEOUT_SECONDS + ) as response: + found = load_any(response.read()) + except Exception as exc: # noqa: BLE001 + logger.debug("AIA fetch failed for %s: %s", url, exc) + continue + if found: + issuer = found[0] + break + if issuer is None: + break + chain.append(issuer) + + return b"".join(c.public_bytes(Encoding.PEM) for c in chain) + + @staticmethod + def _create_attestation_key(ectx: Any) -> tuple[Any, Any]: + """Create a transient restricted signing key and return ``(handle, public)``. + + Restricted signing is what makes the key usable for TPM2_Quote: the TPM will + only sign TPM-generated structures with it, so a quote cannot be forged by + asking the key to sign arbitrary bytes. + """ + from tpm2_pytss.constants import ESYS_TR + from tpm2_pytss.types import TPM2B_PUBLIC, TPM2B_SENSITIVE_CREATE + + template = TPM2B_PUBLIC.parse( + "rsa2048:rsassa:null", + objectAttributes=( + "restricted|sign|fixedtpm|fixedparent|sensitivedataorigin|userwithauth|noda" + ), + ) + handle, public, _, _, _ = ectx.create_primary( + TPM2B_SENSITIVE_CREATE(), template, ESYS_TR.OWNER ) + return handle, public diff --git a/src/ca2a_verify/tpm.py b/src/ca2a_verify/tpm.py index ddc0264..e37a64c 100644 --- a/src/ca2a_verify/tpm.py +++ b/src/ca2a_verify/tpm.py @@ -2,42 +2,147 @@ Appraisal is fail-closed: -1. The AK certificate chain is verified up to a trusted (vendor-supplied) root. +1. The AK certificate chain is verified up to a trusted (caller-supplied) root. 2. The AK signature over the ``TPMS_ATTEST`` blob is verified (ECDSA or RSA). 3. The structure is confirmed to be a TPM-generated quote (magic and type). -4. The qualifying data (the verifier's nonce) and the PCR digest (the platform - measurement) are checked against expected values. +4. The qualifying data (the key-and-nonce binding) and the PCR digest (the + platform measurement) are checked against expected values. -There is no single published TPM root; the caller supplies the vendor roots it -trusts. TPM AK signature schemes vary; this verifier supports ECDSA (over -SHA-256) and RSA PKCS#1 v1.5 (SHA-256) attestation keys. +The cryptography is **not** implemented here. Steps 1, 2 and 4 are delegated to +``agent_manifest.verify_tpm_quote``, which cA2A already depends on and which is +hardware-validated. Three divergent copies of one TPM verifier is the problem +being retired; see cmcp#447. + +What does live here is the piece agent-manifest does not model: ``TPMT_SIGNATURE``, +the wire format ``tpm2_quote -s`` and tpm2-pytss ``signature.marshal()`` actually +emit. agent-manifest takes a bare signature, so the envelope is unwrapped here. + +There is no single published TPM root, so the caller supplies the vendor roots it +trusts. :mod:`ca2a_verify.tpm_roots` carries the one root validated on hardware as +an opt-in constant. Verifying against no root at all is refused. """ from __future__ import annotations +import struct +from dataclasses import dataclass + from cryptography import x509 -from cryptography.exceptions import InvalidSignature -from cryptography.hazmat.primitives.asymmetric import ec, padding, rsa -from cryptography.hazmat.primitives.hashes import SHA256 +from cryptography.hazmat.primitives.serialization import Encoding +from ca2a_runtime.attestation import Verifier from ca2a_runtime.errors import AttestationFailed -from ca2a_runtime.tee.tpm import TPM_GENERATED_VALUE, TPM_ST_ATTEST_QUOTE, TpmQuote -from ca2a_verify.sev_snp import verify_cert_chain +from ca2a_runtime.tee.base import AttestationReport +from ca2a_runtime.tee.tpm import TpmQuote, tpm_qualifying_data + +__all__ = [ + "ParsedSignature", + "parse_tpmt_signature", + "tpm_verifier", + "verify_tpm_quote", + "verify_tpm_report", +] + +# TPM2_ALG_ID values for the signing schemes a quote can use. +_ALG_RSASSA = 0x0014 +_ALG_RSAPSS = 0x0016 +_ALG_ECDSA = 0x0018 + -__all__ = ["verify_tpm_quote"] +@dataclass(frozen=True) +class ParsedSignature: + """A parsed ``TPMT_SIGNATURE``: the algorithm ids and the bare signature.""" + sig_alg: int + hash_alg: int + signature: bytes -def _verify_ak_signature(ak: x509.Certificate, signature: bytes, message: bytes) -> None: - key = ak.public_key() + +def parse_tpmt_signature(blob: bytes) -> ParsedSignature: + """Unwrap a ``TPMT_SIGNATURE`` into a bare signature. + + Layout: ``sigAlg`` (2), ``hashAlg`` (2), then the algorithm-specific body. For + RSA that is a size-prefixed ``TPM2B_PUBLIC_KEY_RSA``. For ECDSA it is two + size-prefixed integers, R then S, which are re-encoded as a DER sequence + because that is what ``cryptography`` verifies against. + + Raises :class:`AttestationFailed` on anything malformed. + """ + if len(blob) < 6: + raise AttestationFailed("TPMT_SIGNATURE too short") try: - if isinstance(key, ec.EllipticCurvePublicKey): - key.verify(signature, message, ec.ECDSA(SHA256())) - elif isinstance(key, rsa.RSAPublicKey): - key.verify(signature, message, padding.PKCS1v15(), SHA256()) - else: - raise AttestationFailed("unsupported AK public-key type for TPM quote") - except InvalidSignature as exc: - raise AttestationFailed("TPM quote signature failed to verify") from exc + sig_alg, hash_alg = struct.unpack_from(">HH", blob, 0) + offset = 4 + + if sig_alg in (_ALG_RSASSA, _ALG_RSAPSS): + (size,) = struct.unpack_from(">H", blob, offset) + offset += 2 + if len(blob) < offset + size: + raise AttestationFailed("TPMT_SIGNATURE truncated inside the RSA signature") + return ParsedSignature(sig_alg, hash_alg, blob[offset : offset + size]) + + if sig_alg == _ALG_ECDSA: + from cryptography.hazmat.primitives.asymmetric.utils import encode_dss_signature + + parts: list[bytes] = [] + for _ in range(2): + (size,) = struct.unpack_from(">H", blob, offset) + offset += 2 + if len(blob) < offset + size: + raise AttestationFailed( + "TPMT_SIGNATURE truncated inside the ECDSA signature" + ) + parts.append(blob[offset : offset + size]) + offset += size + return ParsedSignature( + sig_alg, + hash_alg, + encode_dss_signature( + int.from_bytes(parts[0], "big"), int.from_bytes(parts[1], "big") + ), + ) + except struct.error as exc: + raise AttestationFailed("TPMT_SIGNATURE is malformed", detail=str(exc)) from exc + + raise AttestationFailed( + "unsupported TPM signature algorithm", detail=f"sigAlg={sig_alg:#06x}" + ) + + +def _delegate( + attest: bytes, + signature: bytes, + ak_chain_pem: bytes, + trusted_roots_pem: bytes, + expected_qualifying_data: bytes | None, + expected_pcr_digest: bytes | None, +) -> None: + """Run agent-manifest's verifier, translating its outcome into cA2A errors.""" + try: + from agent_manifest import verify_tpm_quote as _verify + except ImportError as exc: # pragma: no cover - a declared dependency + raise AttestationFailed( + "agent-manifest is required to verify a TPM quote", detail=str(exc) + ) from exc + + try: + ok = _verify( + attest, + signature, + ak_chain_pem, + trusted_roots_pem=trusted_roots_pem, + expected_qualifying_data=expected_qualifying_data, + expected_pcr_digest=expected_pcr_digest, + ) + except Exception as exc: # noqa: BLE001 - raises on a malformed quote or broken chain + raise AttestationFailed( + "TPM quote verification failed", detail=f"{type(exc).__name__}: {exc}" + ) from exc + if not ok: + raise AttestationFailed( + "TPM quote verification failed", + detail="the AK signature or an expected binding did not match", + ) def verify_tpm_quote( @@ -49,24 +154,107 @@ def verify_tpm_quote( expected_pcr_digest: bytes | None = None, expected_qualifying_data: bytes | None = None, ) -> TpmQuote: - """Appraise a TPM 2.0 quote offline. Raises AttestationFailed on any failure.""" + """Appraise a TPM 2.0 quote offline. Raises AttestationFailed on any failure. + + ``signature`` is the bare AK signature. For a marshalled ``TPMT_SIGNATURE`` + (what real tooling emits), unwrap it with :func:`parse_tpmt_signature` first, + or use :func:`verify_tpm_report`, which does that for you. + """ + if not ak_chain: + raise AttestationFailed("no AK certificate chain was supplied") + if not trusted_roots: + raise AttestationFailed( + "no trusted root was supplied", + detail="a chain validated against no anchor would accept any chain", + ) + quote = TpmQuote.parse(attest) + _delegate( + attest, + signature, + b"".join(c.public_bytes(Encoding.PEM) for c in ak_chain), + b"".join(c.public_bytes(Encoding.PEM) for c in trusted_roots), + expected_qualifying_data, + expected_pcr_digest, + ) + return quote + + +def verify_tpm_report( + report: AttestationReport, expected_nonce: str, *, trusted_roots_pem: bytes +) -> str: + """Verify a peer's TPM report and return the measurement it proves. - if quote.magic != TPM_GENERATED_VALUE: + This is the function that turns a report's ``public_key`` and ``nonce`` from + assertions into signed facts. The expected qualifying data is re-derived from + the report's own fields with :func:`~ca2a_runtime.tee.tpm.tpm_qualifying_data` + and required to equal what the TPM committed in ``extraData``, so a report + whose key or nonce was edited after the quote was taken is rejected. + + The returned measurement is read out of the signed quote, not copied from the + report's ``measurement`` field, so what the caller acts on is the value the TPM + signed. A mismatch between the two is itself a failure. + """ + if report.nonce != expected_nonce: + raise AttestationFailed( + "the report nonce does not match the expected nonce", + detail="stale or replayed attestation report", + ) + if report.raw_evidence is None or report.quote_signature is None: raise AttestationFailed( - "TPMS_ATTEST magic is not TPM_GENERATED", detail=f"magic={quote.magic:#x}" + "the report carries no TPM quote to verify", + detail=( + "a report claiming the 'tpm' platform must ship raw_evidence and " + "quote_signature; without them there is nothing to appraise" + ), ) - if quote.attest_type != TPM_ST_ATTEST_QUOTE: + if not report.attestation_key_chain_pem: raise AttestationFailed( - "attestation is not a quote", detail=f"type={quote.attest_type:#x}" + "the report carries no attestation key certificate chain", + detail=( + "the quote may be signed by a transient key, which proves a signature " + "but not that the key lives in a TPM" + ), + ) + if not trusted_roots_pem: + raise AttestationFailed( + "no trusted root was supplied", + detail="a chain validated against no anchor would accept any chain", ) - verify_cert_chain(ak_chain, trusted_roots) - _verify_ak_signature(ak_chain[0], signature, attest) + expected_qualifying_data = tpm_qualifying_data(report.public_key, report.nonce) + parsed = parse_tpmt_signature(report.quote_signature) - if expected_qualifying_data is not None and quote.qualifying_data != expected_qualifying_data: - raise AttestationFailed("qualifying data (nonce) does not match the expected value") - if expected_pcr_digest is not None and quote.pcr_digest != expected_pcr_digest: - raise AttestationFailed("PCR digest does not match the expected measurement") + quote = TpmQuote.parse(report.raw_evidence) + _delegate( + report.raw_evidence, + parsed.signature, + report.attestation_key_chain_pem, + trusted_roots_pem, + expected_qualifying_data, + None, + ) - return quote + measurement = "sha256:" + quote.pcr_digest.hex() + if report.measurement != measurement: + raise AttestationFailed( + "the reported measurement is not the one the TPM signed", + detail=f"report={report.measurement} quote={measurement}", + ) + return measurement + + +def tpm_verifier(trusted_roots_pem: bytes) -> Verifier: + """Return a :data:`~ca2a_runtime.attestation.Verifier` bound to these roots. + + Lets :func:`~ca2a_runtime.attestation.verify_offer` reach + ``assurance="hardware"`` on a TPM peer: + + peer = verify_offer(offer, expected_nonce=n, + verifier=tpm_verifier(AZURE_VTPM_ROOT_2023_PEM)) + """ + + def _verifier(report: AttestationReport, expected_nonce: str) -> str: + return verify_tpm_report(report, expected_nonce, trusted_roots_pem=trusted_roots_pem) + + return _verifier diff --git a/src/ca2a_verify/tpm_roots.py b/src/ca2a_verify/tpm_roots.py new file mode 100644 index 0000000..b1538bc --- /dev/null +++ b/src/ca2a_verify/tpm_roots.py @@ -0,0 +1,76 @@ +"""Vendor trust anchors for TPM attestation key chains, available but never default. + +TPM attestation keys chain to per-vendor roots, not to one published root the way +AMD SEV-SNP and Intel TDX do. cA2A therefore does not decide which vendors a +deployment trusts: :func:`ca2a_verify.tpm.verify_tpm_report` takes the roots the +caller supplies, and nothing here is consulted implicitly. + +What this module provides is the one root cA2A has actually validated against +hardware, so a deployment on that platform does not have to re-derive it. Trusting +it stays an explicit import: + + from ca2a_verify.tpm_roots import AZURE_VTPM_ROOT_2023_PEM + +A chain that validates to an unpinned root is not evidence, because an attacker +who can present any self-consistent chain would pass. So passing no roots at all +is refused rather than treated as "trust anything". +""" + +from __future__ import annotations + +# Azure Trusted Launch / confidential VM virtual TPM. +# +# Chain observed on a Standard_D2s_v5 in eastus, 2026-07-31, with the platform +# attestation key certificate read from vTPM NV index 0x01C101D0: +# +# CN=.TrustedVM.Azure.windows.net +# CN=Azure Cloud Virtual TPM CA - 11 +# CN=Azure Cloud Virtual TPM CA 2025 +# CN=Azure Virtual TPM Root Certificate Authority 2023 <- this cert +# +# The intermediates are fetchable over the certificate AIA extension, so only the +# self-signed root is pinned here. This is the same anchor cmcp pins in +# cmcp_verify/tpm_roots.py; the two were validated against the same chain. +AZURE_VTPM_ROOT_2023_PEM = b"""\ +-----BEGIN CERTIFICATE----- +MIIFsDCCA5igAwIBAgIQUfQx2iySCIpOKeDZKd5KpzANBgkqhkiG9w0BAQwFADBp +MQswCQYDVQQGEwJVUzEeMBwGA1UEChMVTWljcm9zb2Z0IENvcnBvcmF0aW9uMTow +OAYDVQQDEzFBenVyZSBWaXJ0dWFsIFRQTSBSb290IENlcnRpZmljYXRlIEF1dGhv +cml0eSAyMDIzMB4XDTIzMDYwMTE4MDg1M1oXDTQ4MDYwMTE4MTU0MVowaTELMAkG +A1UEBhMCVVMxHjAcBgNVBAoTFU1pY3Jvc29mdCBDb3Jwb3JhdGlvbjE6MDgGA1UE +AxMxQXp1cmUgVmlydHVhbCBUUE0gUm9vdCBDZXJ0aWZpY2F0ZSBBdXRob3JpdHkg +MjAyMzCCAiIwDQYJKoZIhvcNAQEBBQADggIPADCCAgoCggIBALoMMwvdRJ7+bW00 +adKE1VemNqJS+268Ure8QcfZXVOsVO22+PL9WRoPnWo0r5dVoomYGbobh4HC72s9 +sGY6BGRe+Ui2LMwuWnirBtOjaJ34r1ZieNMcVNJT/dXW5HN/HLlm/gSKlWzqCEx6 +gFFAQTvyYl/5jYI4Oe05zJ7ojgjK/6ZHXpFysXnyUITJ9qgjn546IJh/G5OMC3mD +fFU7A/GAi+LYaOHSzXj69Lk1vCftNq9DcQHtB7otO0VxFkRLaULcfu/AYHM7FC/S +q6cJb9Au8K/IUhw/5lJSXZawLJwHpcEYzETm2blad0VHsACaLNucZL5wBi8GEusQ +9Wo8W1p1rUCMp89pufxa3Ar9sYZvWeJlvKggWcQVUlhvvIZEnT+fteEvwTdoajl5 +qSvZbDPGCPjb91rSznoiLq8XqgQBBFjnEiTL+ViaZmyZPYUsBvBY3lKXB1l2hgga +hfBIag4j0wcgqlL82SL7pAdGjq0Fou6SKgHnkkrV5CNxUBBVMNCwUoj5mvEjd5mF +7XPgfM98qNABb2Aqtfl+VuCkU/G1XvFoTqS9AkwbLTGFMS9+jCEU2rw6wnKuGv1T +x9iuSdNvsXt8stx4fkVeJvnFpJeAIwBZVgKRSTa3w3099k0mW8qGiMnwCI5SfdZ2 +SJyD4uEmszsnieE6wAWd1tLLg1jvAgMBAAGjVDBSMA4GA1UdDwEB/wQEAwIBhjAP +BgNVHRMBAf8EBTADAQH/MB0GA1UdDgQWBBRL/iZalMH2M8ODSCbd8+WwZLKqlTAQ +BgkrBgEEAYI3FQEEAwIBADANBgkqhkiG9w0BAQwFAAOCAgEALgNAyg8I0ANNO/8I +2BhpTOsbywN2YSmShAmig5h4sCtaJSM1dRXwA+keY6PCXQEt/PRAQAiHNcOF5zbu +OU1Bw/Z5Z7k9okt04eu8CsS2Bpc+POg9js6lBtmigM5LWJCH1goMD0kJYpzkaCzx +1TdD3yjo0xSxgGhabk5Iu1soD3OxhUyIFcxaluhwkiVINt3Jhy7G7VJTlEwkk21A +oOrQxUsJH0f2GXjYShS1r9qLPzLf7ykcOm62jHGmLZVZujBzLIdNk1bljP9VuGW+ +cISBwzkNeEMMFufcL2xh6s/oiUnXicFWvG7E6ioPnayYXrHy3Rh68XLnhfpzeCzv +bz/I4yMV38qGo/cAY2OJpXUuuD/ZbI5rT+lRBEkDW1kxHP8cpwkRwGopV8+gX2KS +UucIIN4l8/rrNDEX8T0b5U+BUqiO7Z5YnxCya/H0ZIwmQnTlLRTU2fW+OGG+xyIr +jMi/0l6/yWPUkIAkNtvS/yO7USRVLPbtGVk3Qre6HcqacCXzEjINcJhGEVg83Y8n +M+Y+a9J0lUnHytMSFZE85h88OseRS2QwqjozUo2j1DowmhSSUv9Na5Ae22ycciBk +EZSq8a4rSlwqthaELNpeoTLUk6iVoUkK/iLvaMvrkdj9yJY1O/gvlfN2aiNTST/2 +bd+PA4RBToG9rXn6vNkUWdbLibU= +-----END CERTIFICATE----- +""" + +# Platform key to the root validated on it. Deliberately not consulted by the +# verifier: a deployment names the platform it trusts. +KNOWN_VENDOR_ROOTS: dict[str, bytes] = { + "azure-vtpm": AZURE_VTPM_ROOT_2023_PEM, +} + +__all__ = ["AZURE_VTPM_ROOT_2023_PEM", "KNOWN_VENDOR_ROOTS"] diff --git a/tests/unit/test_tpm.py b/tests/unit/test_tpm.py index da248f5..2a70dae 100644 --- a/tests/unit/test_tpm.py +++ b/tests/unit/test_tpm.py @@ -115,8 +115,16 @@ def test_short_quote_rejected() -> None: TpmQuote.parse(b"\x00\x00") -def test_provider_detect_and_attest() -> None: - assert TpmProvider.detect() is False +def test_provider_fails_closed_off_tpm_hardware() -> None: + """Off a TPM host, detect is False and attest raises rather than inventing evidence. + + This asserts the pair *agrees*. It deliberately does not assert ``detect() is + False`` unconditionally: on a host that does have a TPM and the bindings, True + is now the correct answer because ``attest`` works there. The contract and both + branches are covered in test_tpm_attest.py. + """ + if TpmProvider.detect(): + pytest.skip("this host has a TPM; the hardware path is covered separately") with pytest.raises(AttestationUnsupported): TpmProvider().attest("deadbeef", "nonce") diff --git a/tests/unit/test_tpm_attest.py b/tests/unit/test_tpm_attest.py new file mode 100644 index 0000000..5063bd6 --- /dev/null +++ b/tests/unit/test_tpm_attest.py @@ -0,0 +1,519 @@ +"""Tests for the TPM collector and for verifying a peer's TPM report. + +Two halves: + +- the collector's checks and its TPM interaction shape, driven with a fake ESAPI + context so they run without a TPM. The shapes asserted here are the ones that + were wrong on real hardware in cmcp: ``pcr_read`` returns a ``TPML_DIGEST`` whose + digests are iterated once, and ``TPM2_NV_Read`` must be chunked. +- report verification end to end over a synthetic quote and AK chain, including + that the key-and-nonce binding is what actually gates it. + +Synthetic vectors, not hardware. What was validated on hardware is recorded in +docs/testing/hardware-validation.md. +""" + +from __future__ import annotations + +import hashlib +import struct +from datetime import UTC, datetime, timedelta +from types import SimpleNamespace + +import pytest +from cryptography import x509 +from cryptography.hazmat.primitives.asymmetric import ec, padding, rsa +from cryptography.hazmat.primitives.asymmetric.utils import decode_dss_signature +from cryptography.hazmat.primitives.hashes import SHA256, SHA384 +from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat +from cryptography.x509.oid import NameOID + +from ca2a_runtime.attestation import ChannelOffer, verify_offer +from ca2a_runtime.errors import AttestationFailed, AttestationUnsupported +from ca2a_runtime.tee.base import AttestationReport +from ca2a_runtime.tee.tpm import ( + TPM_GENERATED_VALUE, + TPM_ST_ATTEST_QUOTE, + TpmProvider, + TpmQuote, + tpm_qualifying_data, +) +from ca2a_verify.tpm import ( + parse_tpmt_signature, + tpm_verifier, + verify_tpm_report, +) +from tests.unit.conftest import make_ec_cert +from tests.unit.test_tpm import build_attest + +_ALG_RSASSA = 0x0014 +_ALG_ECDSA = 0x0018 +_ALG_SHA256 = 0x000B + +PUBLIC_KEY = "aa" * 32 +NONCE = "deadbeef" + + +# ── qualifying data: the signed binding ─────────────────────────────────────── + + +def test_qualifying_data_is_32_bytes() -> None: + assert len(tpm_qualifying_data(PUBLIC_KEY, NONCE)) == 32 + + +def test_qualifying_data_changes_with_either_input() -> None: + base = tpm_qualifying_data(PUBLIC_KEY, NONCE) + assert tpm_qualifying_data("bb" * 32, NONCE) != base + assert tpm_qualifying_data(PUBLIC_KEY, "cafe") != base + + +@pytest.mark.parametrize( + ("a", "b"), + [ + (("ab", "cd"), ("abc", "d")), + # A delimiter-joined encoding collides here: both sides join to "a|b|c". + # nonce is an arbitrary caller-supplied string, so this is reachable. + (("a|b", "c"), ("a", "b|c")), + (("", "ab"), ("ab", "")), + ], +) +def test_qualifying_data_is_unambiguous(a: tuple[str, str], b: tuple[str, str]) -> None: + """No two distinct (key, nonce) pairs may commit the same bytes. + + Otherwise a peer shifts the split and binds a key other than the one it + appears to be offering. + """ + assert tpm_qualifying_data(*a) != tpm_qualifying_data(*b) + + +def test_qualifying_data_is_domain_separated() -> None: + """A bare sha256 of the same payload must not collide with the binding.""" + naive = hashlib.sha256(PUBLIC_KEY.encode() + b"|" + NONCE.encode()).digest() + assert tpm_qualifying_data(PUBLIC_KEY, NONCE) != naive + + +# ── TPMT_SIGNATURE, the format agent-manifest does not model ────────────────── + + +def _tpmt_ecdsa(signature_der: bytes) -> bytes: + r, s = decode_dss_signature(signature_der) + r_b = r.to_bytes(32, "big") + s_b = s.to_bytes(32, "big") + return ( + struct.pack(">HH", _ALG_ECDSA, _ALG_SHA256) + + struct.pack(">H", len(r_b)) + + r_b + + struct.pack(">H", len(s_b)) + + s_b + ) + + +def _tpmt_rsassa(signature: bytes) -> bytes: + return ( + struct.pack(">HH", _ALG_RSASSA, _ALG_SHA256) + + struct.pack(">H", len(signature)) + + signature + ) + + +def test_parse_tpmt_signature_round_trips_ecdsa() -> None: + key = ec.generate_private_key(ec.SECP256R1()) + der = key.sign(b"message", ec.ECDSA(SHA256())) + parsed = parse_tpmt_signature(_tpmt_ecdsa(der)) + assert parsed.sig_alg == _ALG_ECDSA + # Re-encoded as DER, so cryptography verifies it directly. + key.public_key().verify(parsed.signature, b"message", ec.ECDSA(SHA256())) + + +def test_parse_tpmt_signature_round_trips_rsassa() -> None: + key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + sig = key.sign(b"message", padding.PKCS1v15(), SHA256()) + parsed = parse_tpmt_signature(_tpmt_rsassa(sig)) + assert parsed.sig_alg == _ALG_RSASSA + assert parsed.signature == sig + + +def test_parse_tpmt_signature_rejects_truncated() -> None: + key = ec.generate_private_key(ec.SECP256R1()) + blob = _tpmt_ecdsa(key.sign(b"m", ec.ECDSA(SHA256()))) + with pytest.raises(AttestationFailed): + parse_tpmt_signature(blob[:-4]) + + +def test_parse_tpmt_signature_rejects_unsupported_algorithm() -> None: + with pytest.raises(AttestationFailed, match="unsupported"): + parse_tpmt_signature(struct.pack(">HH", 0x0011, _ALG_SHA256) + b"\x00" * 8) + + +def test_parse_tpmt_signature_rejects_short_blob() -> None: + with pytest.raises(AttestationFailed): + parse_tpmt_signature(b"\x00\x14") + + +# ── synthetic AK chains and reports ─────────────────────────────────────────── + + +def _ec_ak_chain() -> tuple[ec.EllipticCurvePrivateKey, bytes, bytes]: + """Return (ak_key, chain_pem_leaf_first, root_pem).""" + root_key = ec.generate_private_key(ec.SECP256R1()) + root = make_ec_cert("vendor-root", "vendor-root", root_key, root_key) + ak_key = ec.generate_private_key(ec.SECP256R1()) + ak = make_ec_cert("AK", "vendor-root", ak_key, root_key) + return ak_key, ak.public_bytes(Encoding.PEM) + root.public_bytes(Encoding.PEM), ( + root.public_bytes(Encoding.PEM) + ) + + +def _rsa_ak_chain() -> tuple[rsa.RSAPrivateKey, bytes, bytes]: + """An RSA AK under an EC root: the shape Azure's vTPM actually presents.""" + root_key = ec.generate_private_key(ec.SECP256R1()) + root = make_ec_cert("vendor-root", "vendor-root", root_key, root_key) + ak_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + now = datetime.now(UTC) + ak = ( + x509.CertificateBuilder() + .subject_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "AK")])) + .issuer_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "vendor-root")])) + .public_key(ak_key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now - timedelta(days=1)) + .not_valid_after(now + timedelta(days=3650)) + .sign(root_key, SHA384()) + ) + return ak_key, ak.public_bytes(Encoding.PEM) + root.public_bytes(Encoding.PEM), ( + root.public_bytes(Encoding.PEM) + ) + + +def _report( + *, + public_key: str = PUBLIC_KEY, + nonce: str = NONCE, + pcr_digest: bytes = b"\x11" * 32, + qualifying_data: bytes | None = None, + measurement: str | None = None, + rsa_ak: bool = False, +) -> tuple[AttestationReport, bytes]: + """Build a signed synthetic TPM report. Returns (report, root_pem).""" + if qualifying_data is None: + qualifying_data = tpm_qualifying_data(public_key, nonce) + attest = build_attest(qualifying_data=qualifying_data, pcr_digest=pcr_digest) + + if rsa_ak: + rsa_key, chain_pem, root_pem = _rsa_ak_chain() + sig_blob = _tpmt_rsassa(rsa_key.sign(attest, padding.PKCS1v15(), SHA256())) + ak_pem = rsa_key.public_key().public_bytes( + Encoding.PEM, PublicFormat.SubjectPublicKeyInfo + ) + else: + ec_key, chain_pem, root_pem = _ec_ak_chain() + sig_blob = _tpmt_ecdsa(ec_key.sign(attest, ec.ECDSA(SHA256()))) + ak_pem = ec_key.public_key().public_bytes( + Encoding.PEM, PublicFormat.SubjectPublicKeyInfo + ) + + report = AttestationReport( + platform="tpm", + measurement=measurement or ("sha256:" + pcr_digest.hex()), + public_key=public_key, + nonce=nonce, + raw_evidence=attest, + quote_signature=sig_blob, + attestation_key_pem=ak_pem, + attestation_key_chain_pem=chain_pem, + ) + return report, root_pem + + +# ── report verification ─────────────────────────────────────────────────────── + + +def test_valid_report_verifies_and_returns_the_signed_measurement() -> None: + report, root_pem = _report() + measurement = verify_tpm_report(report, NONCE, trusted_roots_pem=root_pem) + assert measurement == "sha256:" + ("11" * 32) + + +def test_valid_rsa_report_verifies() -> None: + """RSASSA is what the Azure vTPM signs with, so it must verify too.""" + report, root_pem = _report(rsa_ak=True) + assert verify_tpm_report(report, NONCE, trusted_roots_pem=root_pem) + + +def test_substituted_public_key_is_rejected() -> None: + """The whole point of the binding: editing the offered key breaks the quote. + + Before the key was committed into extraData, this field was an unsigned + assertion and a peer could swap it after quoting. + """ + report, root_pem = _report() + swapped = AttestationReport(**{**report.__dict__, "public_key": "bb" * 32}) + with pytest.raises(AttestationFailed): + verify_tpm_report(swapped, NONCE, trusted_roots_pem=root_pem) + + +def test_substituted_nonce_is_rejected() -> None: + report, root_pem = _report() + swapped = AttestationReport(**{**report.__dict__, "nonce": "0bad"}) + with pytest.raises(AttestationFailed): + verify_tpm_report(swapped, "0bad", trusted_roots_pem=root_pem) + + +def test_stale_nonce_is_rejected() -> None: + report, root_pem = _report() + with pytest.raises(AttestationFailed, match="nonce"): + verify_tpm_report(report, "a-different-nonce", trusted_roots_pem=root_pem) + + +def test_measurement_disagreeing_with_the_quote_is_rejected() -> None: + """A report may not claim a measurement the quote does not carry.""" + report, root_pem = _report(measurement="sha256:" + ("99" * 32)) + with pytest.raises(AttestationFailed, match="not the one the TPM signed"): + verify_tpm_report(report, NONCE, trusted_roots_pem=root_pem) + + +def test_report_without_evidence_is_rejected() -> None: + report, root_pem = _report() + bare = AttestationReport( + platform="tpm", + measurement=report.measurement, + public_key=report.public_key, + nonce=report.nonce, + ) + with pytest.raises(AttestationFailed, match="no TPM quote"): + verify_tpm_report(bare, NONCE, trusted_roots_pem=root_pem) + + +def test_report_without_a_chain_is_rejected() -> None: + """A transient-key quote verifies as a signature but proves no provenance.""" + report, root_pem = _report() + chainless = AttestationReport(**{**report.__dict__, "attestation_key_chain_pem": None}) + with pytest.raises(AttestationFailed, match="chain"): + verify_tpm_report(chainless, NONCE, trusted_roots_pem=root_pem) + + +def test_no_trusted_root_is_refused() -> None: + """Verifying against no anchor would accept any self-consistent chain.""" + report, _ = _report() + with pytest.raises(AttestationFailed, match="no trusted root"): + verify_tpm_report(report, NONCE, trusted_roots_pem=b"") + + +def test_untrusted_root_is_rejected() -> None: + report, _ = _report() + stranger_key = ec.generate_private_key(ec.SECP256R1()) + stranger = make_ec_cert("stranger", "stranger", stranger_key, stranger_key) + with pytest.raises(AttestationFailed): + verify_tpm_report( + report, NONCE, trusted_roots_pem=stranger.public_bytes(Encoding.PEM) + ) + + +def test_tampered_attest_is_rejected() -> None: + report, root_pem = _report() + tampered = bytearray(report.raw_evidence or b"") + tampered[-1] ^= 0xFF + bad = AttestationReport(**{**report.__dict__, "raw_evidence": bytes(tampered)}) + with pytest.raises(AttestationFailed): + verify_tpm_report(bad, NONCE, trusted_roots_pem=root_pem) + + +# ── the handshake reaches hardware assurance ────────────────────────────────── + + +def test_tpm_verifier_reaches_hardware_assurance_in_verify_offer() -> None: + """End to end: a TPM offer gets assurance="hardware", not "none".""" + report, root_pem = _report() + offer = ChannelOffer(channel_public_key=PUBLIC_KEY, report=report) + peer = verify_offer(offer, expected_nonce=NONCE, verifier=tpm_verifier(root_pem)) + assert peer.assurance == "hardware" + assert peer.public_key == PUBLIC_KEY + assert peer.measurement == "sha256:" + ("11" * 32) + + +def test_tpm_offer_without_a_verifier_still_fails_closed() -> None: + report, _ = _report() + offer = ChannelOffer(channel_public_key=PUBLIC_KEY, report=report) + with pytest.raises(AttestationFailed, match="requires a hardware verifier"): + verify_offer(offer, expected_nonce=NONCE) + + +# ── collector checks ────────────────────────────────────────────────────────── + + +def _quote(qualifying_data: bytes, pcr_digest: bytes, **kw: int) -> TpmQuote: + return TpmQuote.parse( + build_attest(qualifying_data=qualifying_data, pcr_digest=pcr_digest, **kw) + ) + + +def test_check_quote_accepts_a_consistent_quote() -> None: + qd = tpm_qualifying_data(PUBLIC_KEY, NONCE) + digest = b"\x22" * 32 + TpmProvider._check_quote(_quote(qd, digest), qd, digest) + + +def test_check_quote_rejects_a_pcr_selection_mismatch() -> None: + """The cross-check that stops a report describing PCRs that were not measured.""" + qd = tpm_qualifying_data(PUBLIC_KEY, NONCE) + with pytest.raises(AttestationFailed, match="does not match the PCRs that were read"): + TpmProvider._check_quote(_quote(qd, b"\x22" * 32), qd, b"\x33" * 32) + + +def test_check_quote_rejects_a_binding_mismatch() -> None: + digest = b"\x22" * 32 + with pytest.raises(AttestationFailed, match="key and nonce binding"): + TpmProvider._check_quote(_quote(b"other", digest), b"expected", digest) + + +def test_check_quote_rejects_non_tpm_magic() -> None: + qd, digest = b"qd", b"\x22" * 32 + with pytest.raises(AttestationFailed, match="not TPM-generated"): + TpmProvider._check_quote(_quote(qd, digest, magic=0x00000000), qd, digest) + + +def test_check_quote_rejects_a_non_quote_attestation() -> None: + qd, digest = b"qd", b"\x22" * 32 + with pytest.raises(AttestationFailed, match="not a quote"): + TpmProvider._check_quote(_quote(qd, digest, attest_type=0x8017), qd, digest) + assert TPM_GENERATED_VALUE and TPM_ST_ATTEST_QUOTE # constants stay exported + + +# ── the TPM interaction shapes that were wrong on hardware ──────────────────── + + +class _FakeSelection: + @staticmethod + def parse(_spec: str) -> str: + return "selection" + + +def test_read_pcrs_iterates_the_digest_list_once() -> None: + """``pcr_read`` returns a TPML_DIGEST: its digests are the PCR values. + + Treating them as banks and descending a second level yields bytes of a + structure and silently corrupts the measurement, which is the bug this asserts + against. + """ + pcrs = [bytes([i]) * 32 for i in range(8)] + ectx = SimpleNamespace( + pcr_read=lambda _sel: (None, None, SimpleNamespace(digests=pcrs)) + ) + assert TpmProvider._read_pcrs(ectx, _FakeSelection) == hashlib.sha256( + b"".join(pcrs) + ).digest() + + +def test_read_pcrs_refuses_a_short_read() -> None: + ectx = SimpleNamespace( + pcr_read=lambda _sel: (None, None, SimpleNamespace(digests=[b"\x01" * 32] * 3)) + ) + with pytest.raises(AttestationFailed, match="fewer PCRs"): + TpmProvider._read_pcrs(ectx, _FakeSelection) + + +def test_read_pcrs_does_not_fall_back_to_sha1() -> None: + """A report labelled sha256 must measure the SHA-256 bank, or fail.""" + + def _boom(_sel: str) -> None: + raise RuntimeError("no sha256 bank") + + ectx = SimpleNamespace(pcr_read=_boom) + with pytest.raises(AttestationUnsupported, match="SHA-256 PCR bank"): + TpmProvider._read_pcrs(ectx, _FakeSelection) + + +class _FakeNvCtx: + """An ESAPI stand-in that enforces the real TPM2_NV_Read size bound.""" + + def __init__(self, data: bytes, buffer_max: int) -> None: + self.data = data + self.buffer_max = buffer_max + self.reads: list[tuple[int, int]] = [] + + def tr_from_tpmpublic(self, index: int) -> str: + return f"handle-{index:#x}" + + def nv_read_public(self, _handle: str) -> tuple[SimpleNamespace, None]: + return SimpleNamespace(nvPublic=SimpleNamespace(dataSize=len(self.data))), None + + def get_capability(self, _cap: int, prop: int, _count: int) -> tuple[object, object]: + props = SimpleNamespace( + data=SimpleNamespace( + tpmProperties=[SimpleNamespace(property=prop, value=self.buffer_max)] + ) + ) + return (None, props) + + def nv_read(self, _handle: str, size: int, offset: int) -> bytes: + if size > self.buffer_max: + raise RuntimeError("TPM_RC_VALUE: size exceeds TPM2_PT_NV_BUFFER_MAX") + self.reads.append((size, offset)) + return self.data[offset : offset + size] + + +def test_read_nv_chunks_a_certificate_larger_than_the_buffer_max() -> None: + """A 1596-byte AK certificate cannot be fetched in one TPM2_NV_Read. + + Azure reports TPM2_PT_NV_BUFFER_MAX of 1024, so a single full-size read fails + with TPM_RC_VALUE even though the index is plainly readable. + """ + cert = bytes(range(256)) * 6 + b"tail" # 1540 bytes + ectx = _FakeNvCtx(cert, buffer_max=1024) + assert TpmProvider._read_nv(ectx, 0x01C101D0) == cert + assert len(ectx.reads) > 1 + assert all(size <= 1024 for size, _ in ectx.reads) + + +def test_read_nv_returns_none_for_an_undefined_index() -> None: + class _Undefined: + def tr_from_tpmpublic(self, _index: int) -> str: + raise RuntimeError("TPM_RC_HANDLE") + + assert TpmProvider._read_nv(_Undefined(), 0x01C00002) is None + + +# ── detect and attest agree ─────────────────────────────────────────────────── + + +def test_detect_is_false_without_a_tpm(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("ca2a_runtime.tee.tpm.sys.platform", "linux") + monkeypatch.setattr("ca2a_runtime.tee.tpm.Path.exists", lambda _self: False) + assert TpmProvider.detect() is False + + +def test_detect_is_false_without_the_bindings(monkeypatch: pytest.MonkeyPatch) -> None: + """A TPM with no tpm2-pytss cannot attest, so detect must not claim it can.""" + monkeypatch.setattr("ca2a_runtime.tee.tpm.sys.platform", "linux") + monkeypatch.setattr("ca2a_runtime.tee.tpm.Path.exists", lambda _self: True) + monkeypatch.setattr("ca2a_runtime.tee.tpm._tpm2_pytss_available", lambda: False) + assert TpmProvider.detect() is False + + +def test_detect_is_true_when_attest_can_run(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("ca2a_runtime.tee.tpm.sys.platform", "linux") + monkeypatch.setattr("ca2a_runtime.tee.tpm.Path.exists", lambda _self: True) + monkeypatch.setattr("ca2a_runtime.tee.tpm._tpm2_pytss_available", lambda: True) + assert TpmProvider.detect() is True + + +def test_attest_does_not_claim_a_tpm_is_absent_when_one_is_present( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The misleading error from issue #73: "no TPM present" on a host with a TPM.""" + monkeypatch.setattr("ca2a_runtime.tee.tpm.sys.platform", "linux") + monkeypatch.setattr("ca2a_runtime.tee.tpm.Path.exists", lambda _self: True) + monkeypatch.setattr("ca2a_runtime.tee.tpm._tpm2_pytss_available", lambda: False) + with pytest.raises(AttestationUnsupported) as excinfo: + TpmProvider().attest(PUBLIC_KEY, NONCE) + assert "tpm2-pytss" in str(excinfo.value) + assert "is present" in (excinfo.value.detail or "") + + +def test_attest_reports_a_missing_device_as_a_missing_device( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr("ca2a_runtime.tee.tpm.sys.platform", "linux") + monkeypatch.setattr("ca2a_runtime.tee.tpm.Path.exists", lambda _self: False) + with pytest.raises(AttestationUnsupported, match="requires a TPM device"): + TpmProvider().attest(PUBLIC_KEY, NONCE)