diff --git a/src/cmcp_gateway/startup.py b/src/cmcp_gateway/startup.py index 92fe250a..e4cd96e8 100644 --- a/src/cmcp_gateway/startup.py +++ b/src/cmcp_gateway/startup.py @@ -1,4 +1,4 @@ -"""Gateway startup sequence with fail-closed validation — implements issue #66.""" +"""Gateway startup sequence with fail-closed validation — implements issue #66.""" from __future__ import annotations @@ -97,13 +97,16 @@ def run_startup(config_path: str) -> GatewayContext: signing_key = SigningKey() logger.info("Signing key generated: %s...", signing_key.public_key_hex[:16]) - # CRYPTO-002: nonce must be session-unique. Use SHA-256(public_key || random_session_id) - # so two gateways with different random bytes produce different nonces even if they - # share the same keypair (e.g. during blue-green deploy). + # CRYPTO-001 + CRYPTO-002: the first 32 bytes of the nonce are SHA-256(public_key_bytes) + # so verifiers can re-derive the fingerprint from the public key in cnf.jwk and confirm + # it matches report_data[:32] -- binding the attestation report to this specific keypair. + # The remaining 32 bytes are a random salt so two gateways with different random bytes + # produce different nonces even if they share the same keypair (blue-green deploy). import hashlib import secrets - session_id = secrets.token_bytes(32) - nonce = hashlib.sha256(signing_key.public_key_bytes + session_id).digest() + key_fingerprint = hashlib.sha256(signing_key.public_key_bytes).digest() + random_salt = secrets.token_bytes(32) + nonce = key_fingerprint + random_salt try: attestation_report = tee_provider.get_attestation_report(nonce) except Exception as exc: diff --git a/src/cmcp_verify/verify.py b/src/cmcp_verify/verify.py index 6b798b8a..c2aca23e 100644 --- a/src/cmcp_verify/verify.py +++ b/src/cmcp_verify/verify.py @@ -1,5 +1,5 @@ """ -TRACE Claim verification — implements issue #59. +TRACE Claim verification -- implements issue #59. Verifies a cMCP TRACE Claim without trusting the gateway operator. Provider-specific attestation verification (TPM, SEV-SNP) is dispatched @@ -9,7 +9,9 @@ from __future__ import annotations import base64 +import hashlib import json +import logging from dataclasses import dataclass, field from datetime import UTC, datetime from enum import StrEnum @@ -21,6 +23,8 @@ from cmcp_gateway.audit.trace_claim import GatewayClaim +logger = logging.getLogger(__name__) + _SW_ONLY_FIRMWARE = "software-only-dev-mode" _KNOWN_PLATFORMS = { @@ -119,6 +123,82 @@ def _verify_signature(claim: dict[str, Any]) -> tuple[bool, str | None]: return False, "Ed25519 signature verification failed" +def _verify_key_binding( + claim: dict[str, Any], + *, + is_sw_only: bool, +) -> tuple[bool | None, str | None]: + """ + CRYPTO-001: verify that cnf.jwk public key fingerprint matches report_data[:32]. + + The gateway embeds SHA-256(public_key_bytes) as the first 32 bytes of the nonce + it submits to the TEE when requesting the attestation report. The TEE hardware + commits that nonce into the signed report_data field. The nonce is stored as + trace.runtime.nonce (base64url of the full 64-byte value). + + Verifiers re-derive SHA-256(cnf.jwk.x public key bytes) and compare it against + nonce[:32]. A mismatch means the public key was substituted after attestation; + the claim must be rejected with PUBLIC_KEY_NOT_BOUND. + + Returns: + (True, None) -- fingerprint matches; binding verified + (False, reason) -- mismatch or missing data; binding rejected + (None, warning_msg) -- software-only / Level-0 mode; binding not applicable + """ + if is_sw_only: + logger.warning( + "CRYPTO-001: software-only (dev) mode -- TEE key binding cannot be verified; " + "this claim provides no hardware provenance guarantee" + ) + return None, "software-only mode -- TEE key binding not applicable" + + # Extract the public key bytes from cnf.jwk.x + x_b64 = claim.get("trace", {}).get("cnf", {}).get("jwk", {}).get("x", "") + if not x_b64: + return False, "trace.cnf.jwk.x is missing -- cannot verify key binding" + + try: + padding = 4 - (len(x_b64) % 4) + padded = x_b64 + ("=" * padding if padding != 4 else "") + pub_key_bytes = base64.urlsafe_b64decode(padded) + except Exception as exc: + return False, f"cannot decode trace.cnf.jwk.x: {exc}" + + # Compute SHA-256(public_key_bytes) -- the expected fingerprint + expected_fingerprint = hashlib.sha256(pub_key_bytes).digest() + + # Extract the nonce from trace.runtime.nonce (base64url, first 32 bytes = fingerprint) + nonce_b64 = claim.get("trace", {}).get("runtime", {}).get("nonce", "") + if not nonce_b64: + return False, ( + "trace.runtime.nonce is absent -- attestation report_data does not " + "bind this public key to TEE hardware" + ) + + try: + padding = 4 - (len(nonce_b64) % 4) + padded = nonce_b64 + ("=" * padding if padding != 4 else "") + nonce_bytes = base64.urlsafe_b64decode(padded) + except Exception as exc: + return False, f"cannot decode trace.runtime.nonce: {exc}" + + if len(nonce_bytes) < 32: + return False, ( + f"trace.runtime.nonce is too short ({len(nonce_bytes)} bytes); " + "expected at least 32 bytes for key fingerprint" + ) + + actual_fingerprint = nonce_bytes[:32] + if actual_fingerprint != expected_fingerprint: + return False, ( + "cnf.jwk public key fingerprint does not match report_data[:32] -- " + "the public key was not bound to this TEE attestation report; " + "possible key substitution attack" + ) + + return True, None + + def _check_attestation_freshness( claim: dict[str, Any], max_age_seconds: int, @@ -169,6 +249,8 @@ def verify_trace_claim( Steps: 1. Pydantic schema validation (GatewayClaim) 2. Ed25519 signature verification over canonical claim body + 2b. CRYPTO-001: TEE key binding -- verify cnf.jwk fingerprint matches report_data[:32] + 2c. Optional out-of-band trusted_public_key_hex cross-check 3. trace.policy.bundle_hash check against approved.policy_bundle_hash 4. gateway.catalog.hash check against approved.tool_catalog_hash 5. Attestation freshness check @@ -215,29 +297,43 @@ def verify_trace_claim( failure = VerificationError.SIGNATURE_INVALID details["signature_error"] = sig_err or "invalid signature" - # Step 2b: Public key binding — verify JWK x matches an externally-trusted key. - # Without this, a malicious gateway can sign with any key and embed it in the claim. + # Step 2b: CRYPTO-001 -- TEE key binding via report_data fingerprint. + # The nonce submitted to the TEE at attestation time encodes SHA-256(public_key_bytes) + # in its first 32 bytes. Hardware commits this nonce into the signed report_data field. + # Verifiers re-derive the fingerprint from cnf.jwk.x and compare to nonce[:32]. + # An attacker who substitutes their own keypair cannot forge the TEE-signed nonce, + # so verification fails even when the Ed25519 signature is self-consistent. _runtime = claim_json.get("trace", {}).get("runtime", {}) _is_sw_only = ( _runtime.get("platform") == "tpm2" and _runtime.get("firmware_version") == _SW_ONLY_FIRMWARE ) + + binding_result, binding_msg = _verify_key_binding(claim_json, is_sw_only=_is_sw_only) + if binding_result is True: + verified.append("public_key_binding") + elif binding_result is False: + unverified.append("public_key_binding") + # Key binding failure is a higher-priority security signal than a signature failure: + # a substituted key means the signing key itself cannot be trusted. + failure = VerificationError.PUBLIC_KEY_NOT_BOUND + details["public_key_binding"] = binding_msg or "TEE key binding verification failed" + # binding_result is None: software-only mode -- skip (no penalty, no credit) + + # Step 2c: Optional out-of-band trusted_public_key_hex cross-check. + # Callers may supply an externally-pinned public key hex to add an additional + # cross-check independent of the in-claim nonce. Recorded as "trusted_public_key" + # so consumers can distinguish the two mechanisms. _x_b64 = claim_json.get("trace", {}).get("cnf", {}).get("jwk", {}).get("x", "") if trusted_public_key_hex: actual_hex = _jwk_x_to_hex(_x_b64) if _x_b64 else None normalized = trusted_public_key_hex.lower().removeprefix("0x") if actual_hex == normalized: - verified.append("public_key_binding") + verified.append("trusted_public_key") else: - unverified.append("public_key_binding") + unverified.append("trusted_public_key") failure = failure or VerificationError.PUBLIC_KEY_NOT_BOUND - details["public_key_binding"] = "trace.cnf.jwk.x does not match trusted_public_key_hex" - elif not _is_sw_only: - unverified.append("public_key_binding") - failure = failure or VerificationError.PUBLIC_KEY_NOT_BOUND - details["public_key_binding"] = ( - "no trusted_public_key_hex provided — TEE key binding cannot be verified" - ) + details["trusted_public_key"] = "trace.cnf.jwk.x does not match trusted_public_key_hex" # Step 3: Policy bundle hash claimed_policy = claim_json.get("trace", {}).get("policy", {}).get("bundle_hash", "") @@ -395,4 +491,4 @@ def verify_trace_claim( attestation_age_seconds=age, is_attestation_fresh=is_fresh, details=details, - ) + ) \ No newline at end of file diff --git a/tests/unit/test_verify.py b/tests/unit/test_verify.py index d002545e..2e828a6d 100644 --- a/tests/unit/test_verify.py +++ b/tests/unit/test_verify.py @@ -2,6 +2,10 @@ from __future__ import annotations +import base64 +import hashlib +import json +import secrets from datetime import UTC, datetime, timedelta from cmcp_gateway.audit.chain import AuditChain @@ -26,10 +30,23 @@ CATALOG_HASH = "sha256:" + "b" * 64 +def _make_nonce_for_key(key: SigningKey) -> str: + """Build a report_data hex string matching the CRYPTO-001 format. + + First 32 bytes: SHA-256(public_key_bytes) -- verifiable key fingerprint. + Next 32 bytes: random salt -- session uniqueness (CRYPTO-002). + """ + fingerprint = hashlib.sha256(key.public_key_bytes).digest() + salt = secrets.token_bytes(32) + return (fingerprint + salt).hex() + + def _make_signed_claim(policy_hash=POLICY_HASH, catalog_hash=CATALOG_HASH, provider="software-only"): key = SigningKey() chain = AuditChain("test-session") measurement = "DEVELOPMENT_ONLY" if provider == "software-only" else "ab" * 32 + # Use proper CRYPTO-001 report_data for hardware providers; software-only ignores it. + report_data = _make_nonce_for_key(key) if provider != "software-only" else "00" * 32 claim = generate_trace_claim( session_id="test-session", @@ -37,7 +54,7 @@ def _make_signed_claim(policy_hash=POLICY_HASH, catalog_hash=CATALOG_HASH, provi attestation_report=AttestationReportInfo( provider=provider, measurement=measurement, - report_data="00" * 32, + report_data=report_data, attestation_generated_at=datetime.now(tz=UTC).isoformat(), attestation_validity_seconds=86400, ), @@ -71,7 +88,7 @@ def _approved(): return ApprovedHashes(policy_bundle_hash=POLICY_HASH, tool_catalog_hash=CATALOG_HASH) -# ── Signature verification ──────────────────────────────────────────────────── +# -- Signature verification --------------------------------------------------- def test_valid_signature_is_verified(): @@ -96,14 +113,14 @@ def test_empty_signature_fails(): def test_tampered_claim_body_fails_signature(): - """TRACE-002 — signature fails if claim body is modified after signing.""" + """TRACE-002 -- signature fails if claim body is modified after signing.""" claim_dict, _ = _make_signed_claim() claim_dict["gateway"]["session_id"] = "tampered-session" result = verify_trace_claim(claim_dict, _approved()) assert result.failure_reason == VerificationError.SIGNATURE_INVALID -# ── Hash checks ─────────────────────────────────────────────────────────────── +# -- Hash checks -------------------------------------------------------------- def test_matching_policy_hash_is_verified(): @@ -136,7 +153,7 @@ def test_mismatched_catalog_hash_fails(): assert "tool_catalog.hash" in result.unverified_fields -# ── Attestation freshness ───────────────────────────────────────────────────── +# -- Attestation freshness ---------------------------------------------------- def test_fresh_attestation_is_verified(): @@ -153,7 +170,7 @@ def test_stale_attestation_fails(): assert result.is_attestation_fresh is False -# ── Audit chain ─────────────────────────────────────────────────────────────── +# -- Audit chain -------------------------------------------------------------- def test_valid_audit_chain_is_verified(): @@ -169,7 +186,7 @@ def test_missing_audit_chain_root_fails(): assert "audit_chain" in result.unverified_fields -# ── Status ──────────────────────────────────────────────────────────────────── +# -- Status ------------------------------------------------------------------- def test_software_only_provider_is_partially_verified(): @@ -190,11 +207,11 @@ def test_all_software_only_verified_fields_are_present(): assert "audit_chain" in result.verified_fields -# ── TEE-001: known hardware platform without verifier ───────────────────────── +# -- TEE-001: known hardware platform without verifier ------------------------ def test_known_hardware_platform_without_verifier_is_partially_verified(): - """TEE-001 — amd-sev-snp with no verifier impl must be PARTIALLY_VERIFIED not VERIFIED.""" + """TEE-001 -- amd-sev-snp must be PARTIALLY_VERIFIED not VERIFIED.""" claim_dict, key = _make_signed_claim(provider="sev-snp") result = verify_trace_claim( claim_dict, _approved(), trusted_public_key_hex=key.public_key_hex @@ -204,39 +221,223 @@ def test_known_hardware_platform_without_verifier_is_partially_verified(): assert "hardware_attestation" in result.unverified_fields -# ── CRYPTO-001: public key binding ──────────────────────────────────────────── +# -- CRYPTO-001: TEE key binding via report_data fingerprint ------------------ + + +def test_tee_key_binding_happy_path(): + """CRYPTO-001 -- valid key with correct fingerprint in nonce passes binding check.""" + key = SigningKey() + chain = AuditChain("test-session") + fingerprint = hashlib.sha256(key.public_key_bytes).digest() + salt = secrets.token_bytes(32) + report_data = (fingerprint + salt).hex() + + claim = generate_trace_claim( + session_id="test-session", + signing_key=key, + attestation_report=AttestationReportInfo( + provider="sev-snp", + measurement="ab" * 32, + report_data=report_data, + attestation_generated_at=datetime.now(tz=UTC).isoformat(), + attestation_validity_seconds=86400, + ), + policy_bundle=PolicyBundleInfo( + hash=POLICY_HASH, + enforcement_mode="enforcing", + policy_version="1.0.0", + ), + tool_catalog=ToolCatalogInfo(hash=CATALOG_HASH), + call_summary=CallSummary( + tool_calls_total=0, + tool_calls_allowed=0, + tool_calls_denied=0, + tool_calls_faulted=0, + tools_invoked=[], + session_max_sensitivity="public", + call_graph_summary=CallGraphSummary( + compliance_domains_touched=[], + cross_boundary_events=[], + ), + ), + audit_chain_root=chain.chain_root, + audit_chain_tip=chain.chain_tip, + audit_chain_length=chain.length, + do_sign=True, + ) + claim_dict = _to_dict(claim) + result = verify_trace_claim(claim_dict, _approved()) + assert "public_key_binding" in result.verified_fields, ( + f"Expected public_key_binding in verified; " + f"verified={result.verified_fields}, " + f"unverified={result.unverified_fields}, details={result.details}" + ) + assert "public_key_binding" not in result.unverified_fields + + +def test_tee_key_binding_attack_path_mismatched_fingerprint(): + """CRYPTO-001 -- attacker generates a fresh keypair and signs a claim. + + The attacker embeds their own public key in cnf.jwk. The nonce in + trace.runtime was committed by the gateway using the *gateway* key + (SHA-256(gateway_key)), not the attacker key. Verification must reject + the claim with PUBLIC_KEY_NOT_BOUND even though the Ed25519 signature + over the claim body is self-consistent. + """ + gateway_key = SigningKey() + attacker_key = SigningKey() + + chain = AuditChain("test-session") + gateway_fingerprint = hashlib.sha256(gateway_key.public_key_bytes).digest() + salt = secrets.token_bytes(32) + report_data = (gateway_fingerprint + salt).hex() + + # Build a valid claim signed by the gateway key. + claim = generate_trace_claim( + session_id="test-session", + signing_key=gateway_key, + attestation_report=AttestationReportInfo( + provider="sev-snp", + measurement="ab" * 32, + report_data=report_data, + attestation_generated_at=datetime.now(tz=UTC).isoformat(), + attestation_validity_seconds=86400, + ), + policy_bundle=PolicyBundleInfo( + hash=POLICY_HASH, + enforcement_mode="enforcing", + policy_version="1.0.0", + ), + tool_catalog=ToolCatalogInfo(hash=CATALOG_HASH), + call_summary=CallSummary( + tool_calls_total=0, + tool_calls_allowed=0, + tool_calls_denied=0, + tool_calls_faulted=0, + tools_invoked=[], + session_max_sensitivity="public", + call_graph_summary=CallGraphSummary( + compliance_domains_touched=[], + cross_boundary_events=[], + ), + ), + audit_chain_root=chain.chain_root, + audit_chain_tip=chain.chain_tip, + audit_chain_length=chain.length, + do_sign=False, + ) + claim_dict = _to_dict(claim) + + # Attacker replaces cnf.jwk with their own public key. + attacker_x = base64.urlsafe_b64encode(attacker_key.public_key_bytes).rstrip(b"=").decode() + claim_dict["trace"]["cnf"]["jwk"]["x"] = attacker_x + claim_dict["trace"]["cnf"]["jwk"]["kid"] = f"cmcp-{attacker_key.public_key_hex[:8]}" + + # Attacker re-signs the body so Ed25519 verification passes. + body = {k: v for k, v in claim_dict.items() if k != "signature"} + body_bytes = json.dumps(body, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode() + raw_sig = attacker_key.sign(body_bytes) + claim_dict["signature"] = base64.urlsafe_b64encode(raw_sig).rstrip(b"=").decode() + + result = verify_trace_claim(claim_dict, _approved()) + + # Ed25519 signature must pass (self-consistent with attacker key). + assert "signature" in result.verified_fields, ( + "Expected attacker-re-signed claim to pass Ed25519 check" + ) + # TEE key binding must fail: nonce encodes gateway_key fingerprint, not attacker key. + assert "public_key_binding" in result.unverified_fields, ( + f"Expected public_key_binding in unverified; " + f"verified={result.verified_fields}, details={result.details}" + ) + assert result.failure_reason == VerificationError.PUBLIC_KEY_NOT_BOUND + + +def test_tee_key_binding_absent_nonce_fails(): + """CRYPTO-001 -- hardware claim with no nonce in runtime is rejected.""" + claim_dict, _ = _make_signed_claim(provider="sev-snp") + claim_dict["trace"]["runtime"].pop("nonce", None) + result = verify_trace_claim(claim_dict, _approved()) + assert "public_key_binding" in result.unverified_fields + assert result.failure_reason == VerificationError.PUBLIC_KEY_NOT_BOUND + + +def test_tee_key_binding_software_only_exempt(): + """CRYPTO-001 -- software-only provider is exempt from TEE key binding check.""" + claim_dict, _ = _make_signed_claim(provider="software-only") + result = verify_trace_claim(claim_dict, _approved()) + assert "public_key_binding" not in result.unverified_fields + assert "public_key_binding" not in result.verified_fields + + +# -- CRYPTO-001: trusted_public_key_hex out-of-band cross-check (legacy) ------ def test_matching_trusted_public_key_is_verified(): - """CRYPTO-001 — trusted_public_key_hex matching JWK adds public_key_binding to verified.""" - claim_dict, key = _make_signed_claim() + """trusted_public_key_hex matching JWK adds trusted_public_key to verified.""" + claim_dict, key = _make_signed_claim(provider="sev-snp") result = verify_trace_claim( claim_dict, _approved(), trusted_public_key_hex=key.public_key_hex ) - assert "public_key_binding" in result.verified_fields - assert "public_key_binding" not in result.unverified_fields + assert "trusted_public_key" in result.verified_fields + assert "trusted_public_key" not in result.unverified_fields def test_mismatched_trusted_public_key_fails(): - """CRYPTO-001 — wrong trusted key → PUBLIC_KEY_NOT_BOUND.""" - claim_dict, _ = _make_signed_claim() + """Wrong trusted_public_key_hex -> PUBLIC_KEY_NOT_BOUND.""" + claim_dict, _ = _make_signed_claim(provider="sev-snp") result = verify_trace_claim( claim_dict, _approved(), trusted_public_key_hex="00" * 32 ) - assert "public_key_binding" in result.unverified_fields + assert "trusted_public_key" in result.unverified_fields assert result.failure_reason == VerificationError.PUBLIC_KEY_NOT_BOUND def test_no_trusted_key_for_hardware_platform_fails(): - """CRYPTO-001 — hardware platform without trusted_public_key_hex → PUBLIC_KEY_NOT_BOUND.""" - claim_dict, _ = _make_signed_claim(provider="sev-snp") + """CRYPTO-001 -- hardware platform without nonce fingerprint -> PUBLIC_KEY_NOT_BOUND.""" + key = SigningKey() + chain = AuditChain("test-session") + claim = generate_trace_claim( + session_id="test-session", + signing_key=key, + attestation_report=AttestationReportInfo( + provider="sev-snp", + measurement="ab" * 32, + report_data="00" * 64, + attestation_generated_at=datetime.now(tz=UTC).isoformat(), + attestation_validity_seconds=86400, + ), + policy_bundle=PolicyBundleInfo( + hash=POLICY_HASH, + enforcement_mode="enforcing", + policy_version="1.0.0", + ), + tool_catalog=ToolCatalogInfo(hash=CATALOG_HASH), + call_summary=CallSummary( + tool_calls_total=0, + tool_calls_allowed=0, + tool_calls_denied=0, + tool_calls_faulted=0, + tools_invoked=[], + session_max_sensitivity="public", + call_graph_summary=CallGraphSummary( + compliance_domains_touched=[], + cross_boundary_events=[], + ), + ), + audit_chain_root=chain.chain_root, + audit_chain_tip=chain.chain_tip, + audit_chain_length=chain.length, + do_sign=True, + ) + claim_dict = _to_dict(claim) result = verify_trace_claim(claim_dict, _approved()) assert "public_key_binding" in result.unverified_fields assert result.failure_reason == VerificationError.PUBLIC_KEY_NOT_BOUND def test_no_trusted_key_for_software_only_is_not_penalized(): - """CRYPTO-001 — software-only is exempt from the trusted key binding requirement.""" + """CRYPTO-001 -- software-only is exempt from the TEE key binding requirement.""" claim_dict, _ = _make_signed_claim() result = verify_trace_claim(claim_dict, _approved()) assert "public_key_binding" not in result.unverified_fields