diff --git a/CHANGELOG.md b/CHANGELOG.md index c53c446..516b5d1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,8 +7,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- **A refusal is now evidence.** `enforce_peer_call` previously raised on an over-scoped call and emitted nothing, so a denial left no artifact and an auditor saw a gap where a hop should be. It now builds a linked denial record (`provenance.denial_record_for`) carrying the requested capability, the effective scope it fell outside, and the reason, and attaches it to `ScopeNotPermitted.record`. The call still fails closed; the record is evidence of the refusal, not a way to continue. `verify_dag` treats a denial as terminal and rejects any chain that continues past a refused hop, `cross_check_chain` matches only hop records positionally against the chain while still requiring a denial to name a credential that is in it, and `ca2a verify-dag` reports `outcome: denied` with the requested capability and effective scope. Denial fields are omitted from an allow record's hashed body, so existing allow-record hashes are unchanged. +- **`examples/rejection-with-proof/`**: the front-door demo. An agent asks for authority nobody delegated to it, is refused, and the refusal verifies offline through the shipped CLI against the committed chain and DAG. The callee's local policy deliberately permits the capability, so the refusal turns on delegation rather than on the callee not supporting it. The example runs in CI (`tests/unit/test_example_rejection_with_proof.py`) so the README cannot promise something the code stopped doing. It makes no attestation claim and says so. + ### Changed +- Certificate-chain verification now delegates to agent-manifest's shared generic verifier (`agent-manifest>=0.5`) instead of ca2a's own copy. `ca2a_verify.verify_cert_chain` (used by the SEV-SNP VCEK, TDX PCK, and TPM AK chains alike) is now a thin wrapper over `agent_manifest.verify_cert_chain`, re-raising `CertChainError` as `AttestationFailed` to preserve ca2a's contract. This removes the org's last duplicated cert-chain verifier; ca2a keeps its own report/quote parsers, per-format signature checks, appraisal shapes, trust-root pinning, and `verify_offer` semantics. Behavior unchanged (all 201 tests pass unchanged). - Bumped `agentrust-trace` to `>=0.4`. The previous `<0.4` cap worked around the published 0.3.0 lacking the TRACE A2A `delegation` block that cA2A records carry; 0.4.0 ships that block, so the cap is removed. cA2A still owns the diff --git a/LIMITATIONS.md b/LIMITATIONS.md index 342d892..b412069 100644 --- a/LIMITATIONS.md +++ b/LIMITATIONS.md @@ -12,7 +12,7 @@ cA2A is a pre-release profile in active design. This document states plainly wha - **Hardware-attested live binding.** The live transport and handshake run today in **software mode only** (the reference server/client above, `assurance="none"`). The remaining gap is hardware: driving the `verifier` seam in `ca2a_runtime.attestation` off a real SEV-SNP, TDX, or TPM quote (wrapping `ca2a_verify`) so the peer's channel key is bound to a hardware-verified enclave measurement, and relying on the enclave to hold the channel private key. There is also no CLI listener (`ca2a start`); serving is via `ca2a_runtime.transport.server.serve`. A software-mode live call is Tier 2 progress, not a claim that cA2A is attested across trust domains. - **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 verifier is implemented**: report parsing, VCEK certificate chain verification, ECDSA-P384 report-signature verification, and measurement/report-data binding, all fail-closed. The chain path is validated against the genuine AMD Milan root chain; the report-signature path is validated with synthetic vectors, since a real report plus VCEK pair needs SEV-SNP hardware. Report generation (`SevSnpProvider.attest`) still requires a real SEV-SNP guest. The **Intel TDX verifier** (DCAP Quote v4: PCK chain to the genuine Intel SGX Root CA, QE report, attestation-key binding, quote signature, MRTD binding) and the **TPM 2.0 verifier** (AK chain to a caller-supplied vendor root, AK signature, magic/type, qualifying-data and PCR-digest binding) are also implemented and synthetic-vector validated; quote generation needs the respective hardware. Until a backend verifies a real quote end to end against a golden measurement on hardware, cA2A must not be described as fully 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, 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. ## Out of scope diff --git a/README.md b/README.md index c826145..8c30ffb 100644 --- a/README.md +++ b/README.md @@ -71,6 +71,38 @@ pip install --pre ca2a-runtime ca2a verify-chain --chain ./examples/minimal/chain.json ``` +### See it refuse a call, and prove why + +An agent asks for authority nobody delegated to it, is refused, and hands over a +record a third party checks offline without trusting the operator that produced it: + +```bash +python examples/rejection-with-proof/demo.py +``` + +``` +ALLOW tool:search effective scope ['tool:search'] +DENY tool:purchase capability 'tool:purchase' is not in the effective scope + requested tool:purchase + effective ['tool:search'] +``` + +```bash +ca2a verify-dag --dag examples/rejection-with-proof/dag.json \ + --chain examples/rejection-with-proof/chain.json +``` + +```json +{"verified": true, "records": 4, "outcome": "denied", + "requested_capability": "tool:purchase", "effective_scope": ["tool:search"], + "cross_checked": true} +``` + +The callee's own policy permits `tool:purchase`. It is refused anyway, because +the delegated scope does not carry it. See +[examples/rejection-with-proof/](examples/rejection-with-proof/) for what this +does and does not claim; it makes no attestation claim. + See [docs/quickstart.md](docs/quickstart.md) for the full walkthrough. --- diff --git a/ROADMAP.md b/ROADMAP.md index 78ff695..42d0dc5 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -30,11 +30,11 @@ Already implemented and tested elsewhere; cA2A depends on it rather than reimple Real hardware attestation verification (SEV-SNP VCEK chain, Intel TDX quote via QVL/PCS, TPM AK cert + checkquote). This is a dependency for any cross-operator trust claim, single-agent or multi-agent, and is shared with cmcp. At least one real hardware backend must land before cA2A is marketed as attested across trust domains, so the demo matches the claim. -- **SEV-SNP verifier: landed.** Report parsing, VCEK chain verification (validated against the real AMD Milan root), ECDSA-P384 report-signature verification, and measurement/report-data binding, all fail-closed. Report generation still requires a real SEV-SNP guest. See `ca2a_verify.sev_snp` and [docs/spec/attestation.md](docs/spec/attestation.md). -- **TDX verifier: landed.** DCAP Quote v4 parsing, PCK chain to the genuine Intel SGX Root CA, QE report signature, attestation-key binding, quote signature, and MRTD binding, all fail-closed. Quote generation requires a real TDX guest. See `ca2a_verify.tdx`. +- **SEV-SNP verifier: landed and validated on real evidence.** Report parsing, VCEK chain verification, ECDSA-P384 report-signature verification, and measurement/report-data binding, all fail-closed, run against a genuine Azure CVM report (see [docs/hardware-validation.md](docs/hardware-validation.md)). Report generation still requires a real SEV-SNP guest. See `ca2a_verify.sev_snp` and [docs/spec/attestation.md](docs/spec/attestation.md). +- **TDX verifier: landed and validated on real evidence.** DCAP Quote v4 parsing (including the nested type-6 QE certification data), PCK chain to the genuine Intel SGX Root CA, QE report signature, attestation-key binding, quote signature, and MRTD binding, all fail-closed, run against a genuine GCP C3 quote. Quote generation requires a real TDX guest. See `ca2a_verify.tdx`. - **TPM 2.0 verifier: landed.** TPMS_ATTEST parsing, AK chain to a caller-supplied vendor root, AK signature (ECDSA or RSA), magic/type checks, and qualifying-data/PCR-digest binding, all fail-closed. Quote generation requires a real TPM. See `ca2a_verify.tpm`. - **Cross-operator attestation (C6): validated in software.** A two-operator harness (SEV-SNP verifier + measurement pinning + sealed channel) shows independent keys, mutual attestation, confidential cross-operator delegation, and binary-swap detection. All six claims (C1-C6) are now validated experiments. -- **Pending:** end-to-end validation of the SEV-SNP, TDX, and TPM signature paths against real hardware quotes on a confidential VM. The transport that parses A2A messages into a `PeerRequest` has **landed** (`ca2a_runtime.transport.a2a`), running in software mode; the hardware seam is the `verifier` callable in `ca2a_runtime.attestation`. +- **Pending:** the TPM signature path against a real quote, and a live attested peer (the `verifier` seam driven off a hardware quote on a running confidential VM). SEV-SNP and TDX appraisal of real evidence is done. The transport that parses A2A messages into a `PeerRequest` has **landed** (`ca2a_runtime.transport.a2a`), running in software mode; the hardware seam is the `verifier` callable in `ca2a_runtime.attestation`. ## v1.0: Stable profile diff --git a/docs/hardware-validation.md b/docs/hardware-validation.md new file mode 100644 index 0000000..ac6b26d --- /dev/null +++ b/docs/hardware-validation.md @@ -0,0 +1,82 @@ +# Hardware validation + +What `ca2a_verify` has been run against real confidential-computing hardware, +what it has not, and how to reproduce each run. [LIMITATIONS.md](../LIMITATIONS.md) +and [ROADMAP.md](../ROADMAP.md) link here rather than restating it. + +The rule: no document describes cA2A as attested for a platform until a genuine +quote from that platform has been verified end to end by the committed verifier, +and the run is recorded below. + +## Current state + +| Platform | Parsing | Certificate chain | Report signature | Verified against real hardware evidence | +|---|---|---|---|---| +| AMD SEV-SNP (Azure CVM, vTPM-rooted) | Yes | Yes, to the real AMD ARK-Milan root | Yes | **Yes**, 2026-07-27, capture of 2026-07-20 | +| Intel TDX (GCP C3, non-paravisor) | Yes | Yes, to the real Intel SGX Root CA | Yes | **Yes**, 2026-07-27, capture of 2026-07-21 | +| TPM 2.0 | Yes | Yes, to a caller-supplied vendor root | Yes | No. Synthetic vectors only | + +## What these runs do and do not establish + +They establish that the appraisal path in `ca2a_verify` accepts genuine evidence +from that silicon and fails closed on a tampered copy. They do **not** establish +that cA2A runs inside a TEE: quote *generation* still requires the hardware, and +the `verifier` seam in `ca2a_runtime.attestation` is not yet driven off a live +quote on a running peer. A live cA2A call is still software mode +(`assurance="none"`). + +So the claim that is now true is "the cA2A verifier appraises real SEV-SNP and +TDX evidence." The claim that is still not true is "cA2A is attested across trust +domains." The second needs a peer whose channel key is bound to a hardware +measurement on a live confidential VM, which stays on the critical path in +[ROADMAP.md](../ROADMAP.md). + +Verification is also bounded to a remote or rogue-admin adversary, not to one +with physical access: [TEE.fail](https://tee.fail) extracts attestation keys from +fully-patched SEV-SNP and TDX with a sub-$1000 DDR5 interposer. + +## SEV-SNP, Azure confidential VM + +Evidence: an SNP report and its VCEK plus the AMD ASK/ARK chain, captured from an +Azure DCasv5 CVM (family 0x19 / model 0x01, Milan) via the vTPM NV index +`0x01400001`. Azure SEV-SNP is paravisor-mediated, so `REPORT_DATA` binds the +vTPM attestation key rather than a cA2A-supplied value. + +``` +CA2A_SNP_FIXTURE_DIR= pytest tests/unit/test_sev_snp.py +``` + +The directory holds `snp_report.bin`, `vcek.der` and `cert_chain.pem`. Not +committed: the report's 64-byte `CHIP_ID` is a per-CPU hardware identifier, and +zeroing it invalidates the signature, so a redacted vector cannot exercise the +signature path. + +## Intel TDX, GCP C3 confidential VM + +Evidence: a DCAP v4 ECDSA quote from a GCP C3 CVM (non-paravisor TDX, kernel +6.17, configfs-TSM `tdx_guest` provider). + +``` +CA2A_TDX_QUOTE= pytest tests/unit/test_tdx.py +``` + +Not committed: the PCK certificate identifies the CPU. + +This run found the parser defect fixed alongside this page. Genuine DCAP v4 +quotes nest the Quoting Enclave material under a type-6 +`QE_REPORT_CERTIFICATION_DATA` header; `TdxQuote.parse` read the QE report six +bytes early and threw on every real quote. The synthetic fixture emitted the same +flat layout, so the tests validated the defect. Failure was closed, so this was a +false negative rather than an unsound accept, but the TDX path had never worked +against real evidence. Synthetic self-consistency is not validation. + +## Not yet validated + +- **TPM 2.0**: implemented and synthetic-vector validated; needs a real AK-signed + quote plus a vendor AK certificate chain. +- **Live attested peer binding**: the handshake gating the sealed channel on a + verified channel key runs in software mode. Driving it off a real quote on a + confidential VM is the remaining hardware property, and the precondition for + describing cA2A as attested across trust domains. +- **Cross-operator run on real hardware**: the two-operator harness + (`examples/cross-operator-delegation`) is validated in software. diff --git a/examples/rejection-with-proof/README.md b/examples/rejection-with-proof/README.md new file mode 100644 index 0000000..3beb060 --- /dev/null +++ b/examples/rejection-with-proof/README.md @@ -0,0 +1,89 @@ +# Rejection with proof + +An agent asks for authority nobody delegated to it. The call is refused, and the +refusal is handed over as a linked provenance record that a third party verifies +offline, from the committed files, without trusting the operator that produced +them. + +``` +python examples/rejection-with-proof/demo.py +``` + +``` +Delegation chain, narrowing at every hop + depth 0 orchestrator scope=['task:read', 'task:write', 'tool:purchase', 'tool:search'] + depth 1 researcher scope=['task:read', 'tool:purchase', 'tool:search'] + depth 2 retriever scope=['tool:search'] + +ALLOW tool:search effective scope ['tool:search'] +DENY tool:purchase capability 'tool:purchase' is not in the effective scope + requested tool:purchase + effective ['tool:search'] + why the leaf grant carries ['tool:search']; the callee's own + policy would have permitted it, but nobody delegated it +``` + +Then check it the way an auditor would, with no access to the runtime that +produced it: + +``` +ca2a verify-chain --chain examples/rejection-with-proof/chain.json +ca2a verify-dag --dag examples/rejection-with-proof/dag.json \ + --chain examples/rejection-with-proof/chain.json +``` + +```json +{"verified": true, "records": 4, "leaf_scope": ["tool:search"], + "outcome": "denied", "requested_capability": "tool:purchase", + "effective_scope": ["tool:search"], + "denial_reason": "capability 'tool:purchase' is not in the effective scope", + "cross_checked": true} +``` + +## Why the callee's policy is permissive here + +The callee's local policy deliberately **allows** `tool:purchase`. The call is +still refused, because the retriever's delegated scope does not carry it. That is +the distinction cA2A exists to enforce: authorization is the intersection of what +the callee permits and what was actually delegated, and the second half is what +transport-layer auth cannot express. If the demo used a callee that simply did +not support purchasing, the refusal would prove nothing. + +## What makes the refusal checkable + +1. **The chain is signed and attenuating.** Each hop's scope is a provable + subset of its parent's, so `sorted(leaf.scope)` is not the callee's word for + it: it is derivable from signatures the callee did not produce. +2. **The denial record states the gap.** It carries the capability requested and + the effective scope it fell outside, so a verifier reproduces the decision + rather than believing the outcome. +3. **The denial links into the same DAG.** It is hash-chained to the preceding + hop, so it cannot be silently dropped from an audit trail. Reparenting it + breaks the link, which the demo shows at the end. +4. **A denial is terminal.** `verify_dag` rejects any chain that continues past a + refused hop, so a trail cannot claim work proceeded after the denial. + +## What this example does not claim + +Everything here is real and hardware-independent: chain verification, scope +attenuation, the scope-intersect-policy decision, and DAG verification need no +TEE. This example makes **no attestation claim**. It does not seal a payload, +does not attest a peer, and proves nothing about where the code ran. It proves +what was authorized and what was refused. + +For the attestation side, see +[`../cross-operator-delegation/`](../cross-operator-delegation/), which is +explicit that its attestation step uses synthetic vectors, and +[`../../docs/hardware-validation.md`](../../docs/hardware-validation.md) for what +has been verified against real silicon. + +## Files + +| File | What it is | +|---|---| +| `demo.py` | The runnable story. Regenerates the two JSON files. | +| `chain.json` | The signed delegation chain, root to leaf. | +| `dag.json` | The provenance DAG: three hop records plus the terminal denial. | + +Keys are generated per run, so re-running the demo rewrites both files with new +signatures. The committed copies are a sample, not a fixture. diff --git a/examples/rejection-with-proof/chain.json b/examples/rejection-with-proof/chain.json new file mode 100644 index 0000000..9884480 --- /dev/null +++ b/examples/rejection-with-proof/chain.json @@ -0,0 +1,42 @@ +{ + "chain": [ + { + "credential_id": "cred-0-orchestrator", + "issuer": "667e9aa02b1e19e4ef9a24627e10b97e099532dff71a1065a343a6037a8e3c3a", + "subject": "2108c06f9eb264876fc88c0c79c0986177a87222a776ce1b0de7c24483084ea3", + "scope": [ + "task:read", + "task:write", + "tool:purchase", + "tool:search" + ], + "depth": 0, + "parent_id": null, + "signature": "bf10c51cda37517c28f20f66f9f216eade1fd49c766003f2343a7bf5a8b0d423b704f936dd55851a3debb2afb982dd8ba685dd18c171c3bb6a025442063cb20f" + }, + { + "credential_id": "cred-1-researcher", + "issuer": "2108c06f9eb264876fc88c0c79c0986177a87222a776ce1b0de7c24483084ea3", + "subject": "de5c05ea1d7946c8f2c6ffde727cc6ed0da7b906ff2e1219f5916a3cc372f46b", + "scope": [ + "task:read", + "tool:purchase", + "tool:search" + ], + "depth": 1, + "parent_id": "cred-0-orchestrator", + "signature": "1702778d81160bc46f1044b1b628e28b010d9d45c0767c1d1ce50b9f4ebfb3fe954455805ed5e9fdf7d72cc8ec18bb83cd23966ba88d95de21fa94b8b4026f06" + }, + { + "credential_id": "cred-2-retriever", + "issuer": "de5c05ea1d7946c8f2c6ffde727cc6ed0da7b906ff2e1219f5916a3cc372f46b", + "subject": "6eb31885c7ac3d745186ed36a2e77b754325a1ceba6f81b76994a36866d68f52", + "scope": [ + "tool:search" + ], + "depth": 2, + "parent_id": "cred-1-researcher", + "signature": "08999679fff8d3d0940fb94c085ed549511aa026bf9b702da91e8244387db3d86c50004e547950009cf5ce7bf7625bf67893583011097622bed1782de844cf0d" + } + ] +} diff --git a/examples/rejection-with-proof/dag.json b/examples/rejection-with-proof/dag.json new file mode 100644 index 0000000..1d3397c --- /dev/null +++ b/examples/rejection-with-proof/dag.json @@ -0,0 +1,51 @@ +{ + "records": [ + { + "record_id": "rec-0-orchestrator", + "credential_id": "cred-0-orchestrator", + "subject": "2108c06f9eb264876fc88c0c79c0986177a87222a776ce1b0de7c24483084ea3", + "scope": [ + "task:read", + "task:write", + "tool:purchase", + "tool:search" + ], + "parent_record_hash": null + }, + { + "record_id": "rec-1-researcher", + "credential_id": "cred-1-researcher", + "subject": "de5c05ea1d7946c8f2c6ffde727cc6ed0da7b906ff2e1219f5916a3cc372f46b", + "scope": [ + "task:read", + "tool:purchase", + "tool:search" + ], + "parent_record_hash": "2bb2e3f8b86b5874219bf2a5818bf576863daeb947b2d3461ea6eb6abd8622e2" + }, + { + "record_id": "rec-2-retriever", + "credential_id": "cred-2-retriever", + "subject": "6eb31885c7ac3d745186ed36a2e77b754325a1ceba6f81b76994a36866d68f52", + "scope": [ + "tool:search" + ], + "parent_record_hash": "35a4fad0f11b40ed72e47239c3d844ddd262bc1d4fd439913cc5bd0d916b57ec" + }, + { + "record_id": "rec-denied-purchase", + "credential_id": "cred-2-retriever", + "subject": "6eb31885c7ac3d745186ed36a2e77b754325a1ceba6f81b76994a36866d68f52", + "scope": [ + "tool:search" + ], + "parent_record_hash": "56d85126ea94f17f88fe4d0b1df35c300b2f9f79a2dfbcae814be097616da7ab", + "decision": "deny", + "requested_capability": "tool:purchase", + "effective_scope": [ + "tool:search" + ], + "denial_reason": "capability 'tool:purchase' is not in the effective scope" + } + ] +} diff --git a/examples/rejection-with-proof/demo.py b/examples/rejection-with-proof/demo.py new file mode 100644 index 0000000..7c9a476 --- /dev/null +++ b/examples/rejection-with-proof/demo.py @@ -0,0 +1,209 @@ +#!/usr/bin/env python3 +"""Rejection with proof: an agent exceeds its delegated authority and is refused. + +Story: an orchestrator (A) delegates a narrowed scope to a researcher (B), which +delegates a narrower scope still to a retriever (C). C then asks for a capability +nobody ever delegated to it. The call is refused, and the refusal is not a log +line or a stack trace: it is a linked provenance record that an auditor verifies +offline, from the committed artifacts, without trusting the operator that +produced them. + +Two things make the refusal checkable by a third party: + + 1. the delegation chain is signed and attenuating, so the effective scope at + each hop is a provable subset of its parent's; and + 2. the denial record states what was requested and what the effective scope + actually was, and links into the same hash-chained DAG as the allowed hops, + so it cannot be dropped without breaking the chain. + +Run (from the repo root): + + python examples/rejection-with-proof/demo.py + +Then re-verify what it wrote, as an auditor would, with no access to the runtime: + + ca2a verify-chain --chain examples/rejection-with-proof/chain.json + ca2a verify-dag --dag examples/rejection-with-proof/dag.json \ + --chain examples/rejection-with-proof/chain.json + +HONEST LABELING (see LIMITATIONS.md): everything here is real and +hardware-independent. Chain verification, scope attenuation, the scope-intersect- +policy decision and DAG verification need no TEE. This example makes no +attestation claim: it does not seal a payload, does not attest a peer, and proves +nothing about where the code ran. It proves what was authorized and what was +refused. +""" +# ruff: noqa: T201 +from __future__ import annotations + +import json +import os +import subprocess +import sys +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parent.parent.parent +sys.path.insert(0, str(REPO_ROOT / "src")) + +from ca2a_runtime.delegation import DelegationCredential, new_keypair # noqa: E402 +from ca2a_runtime.errors import ScopeNotPermitted # noqa: E402 +from ca2a_runtime.peer import enforce_peer_call # noqa: E402 +from ca2a_runtime.policy import LocalPolicy # noqa: E402 +from ca2a_runtime.provenance import DelegationRecord, record_for, verify_dag # noqa: E402 + +HERE = Path(__file__).resolve().parent + +# A delegates to B delegates to C, narrowing at every hop. +SCOPES = [ + frozenset({"task:read", "task:write", "tool:search", "tool:purchase"}), + frozenset({"task:read", "tool:search", "tool:purchase"}), + frozenset({"tool:search"}), +] +NAMES = ["orchestrator", "researcher", "retriever"] + +# What the retriever actually asks the callee for. It was never delegated this. +REQUESTED = "tool:purchase" + +# The callee's own policy. Deliberately permissive about purchase: the point is +# that a locally allowed capability is still refused when it was not delegated, +# so the refusal cannot be dismissed as "the callee just did not support it". +CALLEE_POLICY = LocalPolicy.of(["task:read", "tool:search", "tool:purchase"]) + + +def build_chain(scopes: list[frozenset[str]]) -> list[DelegationCredential]: + """A correctly signed, narrowing root-to-leaf chain (one hop per scope).""" + chain: list[DelegationCredential] = [] + priv, pub = new_keypair() + parent_id: str | None = None + for depth, scope in enumerate(scopes): + next_priv, next_pub = new_keypair() + cred = DelegationCredential( + credential_id=f"cred-{depth}-{NAMES[depth]}", + issuer=pub, + subject=next_pub, + scope=scope, + depth=depth, + parent_id=parent_id, + ).sign(priv) + chain.append(cred) + parent_id = cred.credential_id + priv, pub = next_priv, next_pub + return chain + + +def dump(obj: object, name: str) -> Path: + path = HERE / name + path.write_text(json.dumps(obj, indent=2) + "\n", encoding="utf-8") + return path + + +def main() -> int: + chain = build_chain(SCOPES) + + print("Delegation chain, narrowing at every hop") + for cred, name in zip(chain, NAMES, strict=True): + print(f" depth {cred.depth} {name:<13} scope={sorted(cred.scope)}") + print() + + # Each delegation hop emits a record as it is taken, so the DAG mirrors the + # chain. record_for is what enforce_peer_call emits on an allowed call. + records: list[DelegationRecord] = [] + parent_hash: str | None = None + for depth, cred in enumerate(chain): + rec = record_for(cred, record_id=f"rec-{depth}-{NAMES[depth]}", + parent_record_hash=parent_hash) + records.append(rec) + parent_hash = rec.record_hash() + + # The retriever asks for something it was delegated. Allowed. + granted = enforce_peer_call( + chain, + "tool:search", + policy=CALLEE_POLICY, + record_id="rec-check", + parent_record_hash=parent_hash, + ) + print(f"ALLOW tool:search effective scope {sorted(granted.effective_scope)}") + + # Then it asks for a capability its grant never carried. + try: + enforce_peer_call( + chain, + REQUESTED, + policy=CALLEE_POLICY, + record_id="rec-denied-purchase", + parent_record_hash=parent_hash, + ) + except ScopeNotPermitted as exc: + denial = exc.record + records.append(denial) + print(f"DENY {REQUESTED} {exc}") + print(f" requested {denial.requested_capability}") + print(f" effective {sorted(denial.effective_scope or frozenset())}") + print(" why the leaf grant carries " + f"{sorted(chain[-1].scope)}; the callee's own policy would have " + "permitted it, but nobody delegated it") + else: # pragma: no cover - the demo is meaningless if this path is taken + print("ERROR: the over-scoped call was NOT refused") + return 1 + print() + + chain_path = dump({"chain": [{**c.body(), "signature": c.signature} for c in chain]}, + "chain.json") + dag_path = dump({"records": [r.body() for r in records]}, "dag.json") + + # An auditor's view: verify the DAG from the records alone. + verify_dag(records) + print(f"DAG verifies offline: {len(records)} records, " + f"leaf documents a {records[-1].decision}") + print(f" wrote {chain_path.relative_to(REPO_ROOT)}") + print(f" wrote {dag_path.relative_to(REPO_ROOT)}") + print() + + # And the same check through the shipped CLI, which is what a third party + # actually runs. No runtime, no operator, just the committed files. + print("Re-verified through the CLI, as a third party would:") + for argv in ( + ["verify-chain", "--chain", str(chain_path)], + ["verify-dag", "--dag", str(dag_path), "--chain", str(chain_path)], + ): + env = {**os.environ, "PYTHONPATH": str(REPO_ROOT / "src")} + proc = subprocess.run( # noqa: S603 + [sys.executable, "-m", "ca2a_runtime.cli", *argv], + capture_output=True, + text=True, + cwd=REPO_ROOT, + env=env, + check=False, + ) + print(f" $ ca2a {' '.join(argv[:1])} ...") + print(f" {proc.stdout.strip() or proc.stderr.strip()}") + if proc.returncode != 0: + return proc.returncode + + print() + print("Tamper check: reparenting the denial breaks the DAG.") + forged = DelegationRecord( + record_id=records[-1].record_id, + credential_id=records[-1].credential_id, + subject=records[-1].subject, + scope=records[-1].scope, + parent_record_hash="0" * 64, + decision=records[-1].decision, + requested_capability=records[-1].requested_capability, + effective_scope=records[-1].effective_scope, + denial_reason=records[-1].denial_reason, + ) + try: + verify_dag([records[0], forged]) + except Exception as exc: # noqa: BLE001 - any link failure is the point + print(f" rejected: {exc}") + else: # pragma: no cover + print(" ERROR: a reparented denial record was accepted") + return 1 + + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/pyproject.toml b/pyproject.toml index eca3b67..7c1afeb 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -31,6 +31,9 @@ dependencies = [ # (see ROADMAP Tier 0). Pin a compatible minor; cross-repo skew is a known # risk documented in LIMITATIONS.md. "agentrust-trace>=0.4", + # Shared hardware-attestation verification (generic cert-chain verifier, + # SNP/TDX/TPM primitives) consumed via PyPI instead of duplicated per repo. + "agent-manifest>=0.5", "rfc8785>=0.1", ] diff --git a/src/ca2a_runtime/cli.py b/src/ca2a_runtime/cli.py index c7eee53..df9f07e 100644 --- a/src/ca2a_runtime/cli.py +++ b/src/ca2a_runtime/cli.py @@ -77,6 +77,22 @@ def _load_records(path: str) -> list[DelegationRecord]: if item.get("parent_record_hash") is None else str(item["parent_record_hash"]) ), + decision=str(item.get("decision", "allow")), + requested_capability=( + None + if item.get("requested_capability") is None + else str(item["requested_capability"]) + ), + effective_scope=( + None + if item.get("effective_scope") is None + else frozenset(str(s) for s in item["effective_scope"]) + ), + denial_reason=( + None + if item.get("denial_reason") is None + else str(item["denial_reason"]) + ), ) ) except (KeyError, TypeError, ValueError) as exc: @@ -96,11 +112,19 @@ def _cmd_verify_dag(args: argparse.Namespace) -> int: except CA2AError as exc: print(json.dumps({"verified": False, "code": exc.code, "error": str(exc)})) return 1 - out = { + leaf = records[-1] + out: dict[str, Any] = { "verified": True, "records": len(records), - "leaf_scope": sorted(records[-1].scope), + "leaf_scope": sorted(leaf.scope), } + if leaf.denied: + # The DAG verifies AND it documents a refusal. Both are true, and a + # verifier that only printed "verified" would bury the interesting part. + out["outcome"] = "denied" + out["requested_capability"] = leaf.requested_capability + out["effective_scope"] = sorted(leaf.effective_scope or frozenset()) + out["denial_reason"] = leaf.denial_reason if args.chain: out["cross_checked"] = cross_checked print(json.dumps(out)) diff --git a/src/ca2a_runtime/errors.py b/src/ca2a_runtime/errors.py index 80a351f..b13cd86 100644 --- a/src/ca2a_runtime/errors.py +++ b/src/ca2a_runtime/errors.py @@ -77,11 +77,22 @@ class ProvenanceLinkBroken(CA2AError): class ScopeNotPermitted(CA2AError): """A requested capability is not in the effective scope (the delegated - scope intersected with the callee's local policy).""" + scope intersected with the callee's local policy). + + Carries the signed-in-place provenance record for the refusal on ``record`` + when the caller supplied enough context to build one. The call still fails + closed; the record is evidence of the refusal, not a way to continue. + """ code = "SCOPE_NOT_PERMITTED" http_status = 403 + def __init__( + self, message: str, *, detail: str | None = None, record: object | None = None + ) -> None: + super().__init__(message, detail=detail) + self.record = record + class TransportError(CA2AError): """cA2A A2A-extension metadata was present but malformed or incomplete. diff --git a/src/ca2a_runtime/peer.py b/src/ca2a_runtime/peer.py index addda54..a6c232b 100644 --- a/src/ca2a_runtime/peer.py +++ b/src/ca2a_runtime/peer.py @@ -26,7 +26,7 @@ from ca2a_runtime.delegation.credential import DelegationCredential, verify_chain from ca2a_runtime.errors import ScopeNotPermitted, SealedChannelError from ca2a_runtime.policy import Policy -from ca2a_runtime.provenance import DelegationRecord, record_for +from ca2a_runtime.provenance import DelegationRecord, denial_record_for, record_for def effective_scope( @@ -66,9 +66,21 @@ def enforce_peer_call( """ effective = effective_scope(chain, policy, max_depth=max_depth) if requested_capability not in effective: + reason = f"capability {requested_capability!r} is not in the effective scope" raise ScopeNotPermitted( - f"capability {requested_capability!r} is not in the effective scope", + reason, detail=f"effective={sorted(effective)}", + # The refusal is evidence too: emit a linked denial record so an + # auditor walking the DAG sees why the call stopped here, rather + # than a gap where a hop should be. + record=denial_record_for( + chain[-1], + record_id=record_id, + parent_record_hash=parent_record_hash, + requested_capability=requested_capability, + effective_scope=effective, + reason=reason, + ), ) record = record_for(chain[-1], record_id=record_id, parent_record_hash=parent_record_hash) return PeerDecision( diff --git a/src/ca2a_runtime/provenance.py b/src/ca2a_runtime/provenance.py index eac0082..1d3cf27 100644 --- a/src/ca2a_runtime/provenance.py +++ b/src/ca2a_runtime/provenance.py @@ -21,23 +21,49 @@ @dataclass(frozen=True) class DelegationRecord: - """A provenance record for a single delegation hop.""" + """A provenance record for a single delegation hop. + + A record is emitted whether the hop was allowed or denied. A denial record + carries what was asked for, what the effective scope actually was, and why + the call was refused, so a refusal is evidence rather than an absence of + evidence. The denial fields are omitted from the hashed body on an allow + record, so allow-record hashes are unchanged from records emitted before + denial records existed. + """ record_id: str credential_id: str subject: str scope: frozenset[str] parent_record_hash: str | None = None + decision: str = "allow" + requested_capability: str | None = None + effective_scope: frozenset[str] | None = None + denial_reason: str | None = None def body(self) -> dict[str, Any]: """The hashed portion of the record.""" - return { + body: dict[str, Any] = { "record_id": self.record_id, "credential_id": self.credential_id, "subject": self.subject, "scope": sorted(self.scope), "parent_record_hash": self.parent_record_hash, } + if self.decision != "allow": + body["decision"] = self.decision + if self.requested_capability is not None: + body["requested_capability"] = self.requested_capability + if self.effective_scope is not None: + body["effective_scope"] = sorted(self.effective_scope) + if self.denial_reason is not None: + body["denial_reason"] = self.denial_reason + return body + + @property + def denied(self) -> bool: + """True when this record documents a refused call.""" + return self.decision == "deny" def record_hash(self) -> str: """SHA-256 over the canonical body. Any field change changes this value.""" @@ -59,6 +85,35 @@ def record_for( ) +def denial_record_for( + credential: DelegationCredential, + record_id: str, + parent_record_hash: str | None, + *, + requested_capability: str, + effective_scope: frozenset[str], + reason: str, +) -> DelegationRecord: + """Build the provenance record a hop emits when it refuses a call. + + The record links into the same DAG as allow records, so an auditor walking + the chain sees the refusal in place rather than a gap. It states the + capability that was requested and the effective scope it fell outside of, + which is what makes the refusal checkable offline against the signed chain. + """ + return DelegationRecord( + record_id=record_id, + credential_id=credential.credential_id, + subject=credential.subject, + scope=credential.scope, + parent_record_hash=parent_record_hash, + decision="deny", + requested_capability=requested_capability, + effective_scope=effective_scope, + denial_reason=reason, + ) + + def verify_dag(records: list[DelegationRecord]) -> list[DelegationRecord]: """Verify a root-to-leaf provenance chain and return it in order. @@ -68,7 +123,9 @@ def verify_dag(records: list[DelegationRecord]) -> list[DelegationRecord]: - every later record's parent_record_hash must equal the recomputed hash of the immediately preceding record (so a tampered or reparented record is detected because its hash no longer matches the stored link); - - no record_id may repeat. + - no record_id may repeat; + - a denial record is terminal: nothing may be chained onto a refused hop, + so a chain claiming work continued past a denial is rejected. """ if not records: raise ProvenanceLinkBroken("empty provenance chain") @@ -84,6 +141,12 @@ def verify_dag(records: list[DelegationRecord]) -> list[DelegationRecord]: if prev is None: if rec.parent_record_hash is not None: raise ProvenanceLinkBroken("root record must not reference a parent") + elif prev.denied: + raise ProvenanceLinkBroken( + f"record {i} is chained onto a denied hop", + detail=f"record {prev.record_id} refused " + f"{prev.requested_capability!r}; no hop may follow it", + ) else: expected = prev.record_hash() if rec.parent_record_hash != expected: @@ -101,15 +164,37 @@ def cross_check_chain( ) -> None: """Confirm a verified provenance chain lines up with a delegation chain. - Ties provenance to authority: record ``i`` must reference credential ``i`` - and carry the same subject. Raises ProvenanceLinkBroken on any mismatch. + Ties provenance to authority: hop record ``i`` must reference credential + ``i`` and carry the same subject. Raises ProvenanceLinkBroken on any + mismatch. + + A denial record documents a refused call, not a delegation hop, so it is not + matched positionally against the chain. It must still name a credential that + is actually in the chain, otherwise a refusal could be attributed to + authority that was never granted. """ - if len(records) != len(chain): + hops = [r for r in records if not r.denied] + if len(hops) != len(chain): raise ProvenanceLinkBroken( - f"provenance length {len(records)} does not match chain length {len(chain)}" + f"provenance length {len(hops)} does not match chain length {len(chain)}" ) - for i, (rec, cred) in enumerate(zip(records, chain, strict=True)): + for i, (rec, cred) in enumerate(zip(hops, chain, strict=True)): if rec.credential_id != cred.credential_id: raise ProvenanceLinkBroken(f"record {i} credential_id does not match the chain") if rec.subject != cred.subject: raise ProvenanceLinkBroken(f"record {i} subject does not match the chain") + + by_id = {c.credential_id: c for c in chain} + for rec in records: + if not rec.denied: + continue + under = by_id.get(rec.credential_id) + if under is None: + raise ProvenanceLinkBroken( + f"denial record {rec.record_id} references credential " + f"{rec.credential_id!r}, which is not in the chain" + ) + if rec.subject != under.subject: + raise ProvenanceLinkBroken( + f"denial record {rec.record_id} subject does not match the chain" + ) diff --git a/src/ca2a_runtime/tee/tdx.py b/src/ca2a_runtime/tee/tdx.py index e90a839..607fed4 100644 --- a/src/ca2a_runtime/tee/tdx.py +++ b/src/ca2a_runtime/tee/tdx.py @@ -7,10 +7,11 @@ :mod:`ca2a_verify.tdx`. Producing a quote requires a real TDX guest, so :meth:`TdxProvider.attest` fails -closed off hardware. Byte offsets follow the Intel DCAP Quote v4 layout; the -verifier is exercised against synthetic self-consistent vectors plus the real -Intel SGX Root CA in the test suite. End-to-end validation against a real -hardware quote requires a TDX guest and remains open. +closed off hardware. Byte offsets follow the Intel DCAP Quote v4 layout, including +the nested type-6 QE certification data that wraps the QE report and the PCK +chain. The verifier is exercised against synthetic self-consistent vectors plus +the real Intel SGX Root CA in the test suite, and against a genuine GCP C3 quote +when ``CA2A_TDX_QUOTE`` points at one; see ``docs/hardware-validation.md``. """ from __future__ import annotations @@ -38,7 +39,10 @@ QE_REPORT_DATA_OFFSET = 320 # within the QE SGX report TEE_TYPE_TDX = 0x81 +# Intel DCAP certification-data types, each preceded by a uint16 type + uint32 size. CERT_TYPE_PCK_CHAIN = 5 +CERT_TYPE_QE_REPORT = 6 +CERT_DATA_HEADER_LEN = 6 TDX_GUEST_DEVICE = "/dev/tdx_guest" @@ -79,17 +83,34 @@ def parse(cls, blob: bytes) -> TdxQuote: pos += QUOTE_SIG_LEN att_key = blob[pos : pos + ATT_KEY_LEN] pos += ATT_KEY_LEN - qe_report = blob[pos : pos + QE_REPORT_LEN] - pos += QE_REPORT_LEN - qe_report_sig = blob[pos : pos + QUOTE_SIG_LEN] - pos += QUOTE_SIG_LEN - (qe_auth_len,) = struct.unpack_from(" None: """Verify a leaf-to-root certificate chain against a set of trusted roots. - ``chain`` is ordered leaf first (VCEK), root last (ARK). Each certificate - must be directly issued by the next, and the final certificate must match a - trusted root by fingerprint. Raises AttestationFailed on any failure. + ``chain`` is ordered leaf first (VCEK), root last (ARK). Delegates to + agent-manifest's shared, algorithm-agnostic cert-chain verifier (one + implementation across the org, consumed via PyPI) and re-raises its failure + as AttestationFailed to preserve ca2a's error contract. Used for the SEV-SNP + VCEK chain, the TDX PCK chain, and the TPM AK chain alike. Raises + AttestationFailed on any failure. """ - if not chain: - raise AttestationFailed("empty certificate chain") - - for i in range(len(chain) - 1): - child, issuer = chain[i], chain[i + 1] - try: - child.verify_directly_issued_by(issuer) - except (ValueError, TypeError, InvalidSignature) as exc: - raise AttestationFailed( - f"certificate at position {i} is not validly issued by the next", - detail=str(exc), - ) from exc - - root = chain[-1] - trusted = {c.fingerprint(SHA256()) for c in trusted_roots} - if root.fingerprint(SHA256()) not in trusted: + from agent_manifest import CertChainError + from agent_manifest import verify_cert_chain as _shared_verify_cert_chain + + try: + _shared_verify_cert_chain(chain, trusted_roots) + except CertChainError as exc: raise AttestationFailed( - "chain root is not a trusted AMD root", - detail=root.subject.rfc4514_string(), - ) + "certificate chain verification failed", detail=str(exc) + ) from exc def verify_sev_snp_report( diff --git a/tests/unit/test_example_rejection_with_proof.py b/tests/unit/test_example_rejection_with_proof.py new file mode 100644 index 0000000..2239a51 --- /dev/null +++ b/tests/unit/test_example_rejection_with_proof.py @@ -0,0 +1,72 @@ +"""The rejection-with-proof example must keep working, and keep refusing. + +The example is the repo's front door: an agent exceeds its delegated authority, +is refused, and the refusal verifies offline. A silent break here would leave a +README promising something the code no longer does, so the demo runs in CI. +""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +from pathlib import Path + +import pytest + +REPO_ROOT = Path(__file__).resolve().parents[2] +DEMO = REPO_ROOT / "examples" / "rejection-with-proof" / "demo.py" + + +@pytest.fixture(scope="module") +def demo_run(tmp_path_factory) -> subprocess.CompletedProcess[str]: + env = {**os.environ, "PYTHONPATH": str(REPO_ROOT / "src")} + return subprocess.run( # noqa: S603 + [sys.executable, str(DEMO)], + capture_output=True, + text=True, + cwd=REPO_ROOT, + env=env, + check=False, + ) + + +def test_demo_exits_clean(demo_run) -> None: + assert demo_run.returncode == 0, demo_run.stderr + + +def test_demo_refuses_the_over_scoped_call(demo_run) -> None: + out = demo_run.stdout + assert "DENY tool:purchase" in out + assert "ALLOW tool:search" in out + # the failure mode that would make the demo a lie + assert "the over-scoped call was NOT refused" not in out + + +def test_demo_denial_verifies_through_the_cli(demo_run) -> None: + """The CLI line the README quotes must actually say denied and verified.""" + line = next( + ln for ln in demo_run.stdout.splitlines() if '"outcome": "denied"' in ln + ) + payload = json.loads(line.strip()) + assert payload["verified"] is True + assert payload["cross_checked"] is True + assert payload["requested_capability"] == "tool:purchase" + assert payload["effective_scope"] == ["tool:search"] + + +def test_demo_rejects_a_reparented_denial(demo_run) -> None: + assert "rejected:" in demo_run.stdout + assert "a reparented denial record was accepted" not in demo_run.stdout + + +def test_committed_artifacts_are_shaped_as_the_readme_says() -> None: + here = DEMO.parent + chain = json.loads((here / "chain.json").read_text(encoding="utf-8"))["chain"] + records = json.loads((here / "dag.json").read_text(encoding="utf-8"))["records"] + assert len(chain) == 3 + assert [len(c["scope"]) for c in chain] == [4, 3, 1] # narrowing at every hop + assert records[-1]["decision"] == "deny" + assert records[-1]["requested_capability"] == "tool:purchase" + assert all("decision" not in r for r in records[:-1]) # allow records stay bare diff --git a/tests/unit/test_peer.py b/tests/unit/test_peer.py index 7b39382..77a67b3 100644 --- a/tests/unit/test_peer.py +++ b/tests/unit/test_peer.py @@ -71,3 +71,63 @@ def test_enforce_record_links_to_parent() -> None: parent_record_hash="abc123") assert decision.record.parent_record_hash == "abc123" assert decision.record.credential_id == _chain()[-1].credential_id + + +def test_denial_carries_a_linked_provenance_record() -> None: + """A refusal is evidence: it emits a record, it does not just raise.""" + policy = LocalPolicy.of(["read"]) + with pytest.raises(ScopeNotPermitted) as exc: + enforce_peer_call(_chain(), "write", policy=policy, record_id="rec-deny") + + record = exc.value.record + assert record is not None + assert record.denied + assert record.decision == "deny" + assert record.record_id == "rec-deny" + assert record.requested_capability == "write" + assert record.effective_scope == frozenset({"read"}) + assert "write" in record.denial_reason + + +def test_denial_record_links_to_its_parent_hop() -> None: + chain = _chain() + allowed = enforce_peer_call(chain, "read", policy=LocalPolicy.of(["read"]), record_id="rec-0") + with pytest.raises(ScopeNotPermitted) as exc: + enforce_peer_call( + chain, + "write", + policy=LocalPolicy.of(["read"]), + record_id="rec-1", + parent_record_hash=allowed.record.record_hash(), + ) + denial = exc.value.record + # the refusal sits in the DAG where the next hop would have been + assert verify_dag([allowed.record, denial]) == [allowed.record, denial] + + +def test_nothing_may_be_chained_onto_a_denied_hop() -> None: + from ca2a_runtime.errors import ProvenanceLinkBroken + + chain = _chain() + with pytest.raises(ScopeNotPermitted) as exc: + enforce_peer_call(chain, "write", policy=LocalPolicy.of(["read"]), record_id="rec-deny") + denial = exc.value.record + forged = enforce_peer_call( + chain, + "read", + policy=LocalPolicy.of(["read"]), + record_id="rec-after", + parent_record_hash=denial.record_hash(), + ).record + with pytest.raises(ProvenanceLinkBroken, match="chained onto a denied hop"): + verify_dag([denial, forged]) + + +def test_allow_record_hash_is_unchanged_by_the_denial_fields() -> None: + """Denial fields are omitted from an allow record's body, so hashes are stable.""" + decision = enforce_peer_call( + _chain(), "read", policy=LocalPolicy.of(["read"]), record_id="rec-0" + ) + assert "decision" not in decision.record.body() + assert "requested_capability" not in decision.record.body() + assert "effective_scope" not in decision.record.body() diff --git a/tests/unit/test_sev_snp.py b/tests/unit/test_sev_snp.py index a1fe604..0bf3ee5 100644 --- a/tests/unit/test_sev_snp.py +++ b/tests/unit/test_sev_snp.py @@ -8,6 +8,7 @@ from __future__ import annotations +import os import struct from datetime import UTC, datetime, timedelta from pathlib import Path @@ -157,3 +158,32 @@ def test_provider_detect_and_attest() -> None: assert SevSnpProvider.detect() is False # no /dev/sev-guest in this environment with pytest.raises(AttestationUnsupported): SevSnpProvider().attest("deadbeef", "nonce") + + +@pytest.mark.skipif( + not os.environ.get("CA2A_SNP_FIXTURE_DIR"), + reason="set CA2A_SNP_FIXTURE_DIR to a dir with snp_report.bin + vcek.der + " + "cert_chain.pem (a real SEV-SNP capture) to run the hardware test", +) +def test_real_snp_report_verifies_to_the_amd_root() -> None: + """Full-chain appraisal of a genuine SEV-SNP report. + + The capture is not committed: the report's 64-byte CHIP_ID is a per-CPU + hardware identifier. See docs/hardware-validation.md. + """ + d = Path(os.environ["CA2A_SNP_FIXTURE_DIR"]) + report = (d / "snp_report.bin").read_bytes() + vcek = x509.load_der_x509_certificate((d / "vcek.der").read_bytes()) + rest = x509.load_pem_x509_certificates((d / "cert_chain.pem").read_bytes()) + ark = rest[-1] + assert ark.subject == ark.issuer, "last cert in the chain must be the self-signed ARK" + + parsed = verify_sev_snp_report(report, [vcek, *rest], trusted_roots=[ark]) + assert len(parsed.measurement) == 48 + assert parsed.measurement != bytes(48) + + # and it fails closed on a flipped bit in the signed body + tampered = bytearray(report) + tampered[0x50] ^= 0x01 + with pytest.raises(AttestationFailed): + verify_sev_snp_report(bytes(tampered), [vcek, *rest], trusted_roots=[ark]) diff --git a/tests/unit/test_tdx.py b/tests/unit/test_tdx.py index 9a8841d..ffc5af7 100644 --- a/tests/unit/test_tdx.py +++ b/tests/unit/test_tdx.py @@ -9,6 +9,7 @@ from __future__ import annotations import hashlib +import os import struct from pathlib import Path @@ -66,9 +67,14 @@ def build_quote(mrtd: bytes, report_data: bytes, *, root_key, root_name="test-in qe_report_sig = _sig_rs(pck_key, bytes(qe_report)) pem = pck.public_bytes(Encoding.PEM) + root.public_bytes(Encoding.PEM) - sig = (quote_sig + att_raw + bytes(qe_report) + qe_report_sig - + struct.pack(" None: assert TdxProvider.detect() is False with pytest.raises(AttestationUnsupported): TdxProvider().attest("deadbeef", "nonce") + + +def test_flat_signature_layout_rejected(quote_and_root) -> None: + """A quote without the type-6 QE wrapper is malformed, not silently misread. + + The flat shape is what the parser used to assume; genuine DCAP v4 quotes nest + the QE material, so anything claiming the flat layout must fail closed. + """ + off = 48 + 584 + 4 + 128 # header + TD report + sig_len + quote_sig + att_key + flat = bytearray(quote_and_root["quote"]) + flat[off:off + 2] = struct.pack(" None: + """Full-chain appraisal of a genuine TDX quote (see docs/hardware-validation.md).""" + from ca2a_verify.tdx import verify_tdx_quote + + intel_root = x509.load_pem_x509_certificates(FIXTURE.read_bytes())[0] + quote_bytes = Path(os.environ["CA2A_TDX_QUOTE"]).read_bytes() + quote = verify_tdx_quote(quote_bytes, trusted_roots=[intel_root]) + assert quote.version == 4 + assert quote.tee_type == 0x81 + assert len(quote.measurement) == 48 + assert quote.measurement != bytes(48)