From 6ffad28da75560dfba5a0c242839403f871e9c53 Mon Sep 17 00:00:00 2001 From: "Altru.dev" Date: Sun, 6 Sep 2026 16:30:13 -0700 Subject: [PATCH] fix(verify): preserve structural claim failure Closes #592. Reconstruct #596 on current main with the strict schema prerequisite preserved. Move legacy TPM fixtures to the signed gateway evidence envelope and pin schema refusal for unsupported platform aliases. Supersedes the earlier draft implementation and its overbroad compatibility workaround. AI-assisted implementation and verification; Altru.dev remains responsible for the contribution. Signed-off-by: Altru.dev --- src/cmcp_verify/verify.py | 48 +++++++--- .../test_evidence_envelope_all_platforms.py | 34 +++++++- tests/unit/test_tpm_claim_signature_verify.py | 13 ++- tests/unit/test_tpm_verify.py | 24 +++-- .../test_verify_malformed_claim_shapes.py | 87 +++++++++++++++++++ 5 files changed, 181 insertions(+), 25 deletions(-) create mode 100644 tests/unit/test_verify_malformed_claim_shapes.py diff --git a/src/cmcp_verify/verify.py b/src/cmcp_verify/verify.py index 6f5af0f4..29dd987b 100644 --- a/src/cmcp_verify/verify.py +++ b/src/cmcp_verify/verify.py @@ -531,13 +531,29 @@ def _coerce_measurement_digest(value: str | bytes) -> bytes | None: return None -def _validate_schema(claim: dict[str, Any]) -> tuple[bool, str | None]: +def _validation_error_path(exc: ValidationError) -> str: + """Return the most specific non-missing schema location for diagnostics.""" + errors = exc.errors() + for error in errors: + if error["type"] == "missing": + continue + loc = error["loc"] + if loc: + return ".".join(str(part) for part in loc) + for error in errors: + loc = error["loc"] + if loc: + return ".".join(str(part) for part in loc) + return "claim" + + +def _validate_schema(claim: dict[str, Any]) -> tuple[bool, str | None, str | None]: """Validate claim structure using the RuntimeClaim Pydantic model.""" try: RuntimeClaim.model_validate(claim) - return True, None + return True, None, None except ValidationError as exc: - return False, str(exc) + return False, str(exc), _validation_error_path(exc) @dataclass @@ -834,14 +850,24 @@ def verify_trace_claim( failure: VerificationError | None = None details: dict[str, str] = {} - # Step 1: Schema validation - schema_ok, schema_err = _validate_schema(claim_json) - if schema_ok: - verified.append("schema") - else: - unverified.append("schema") - failure = VerificationError.CLAIM_MALFORMED - details["schema_error"] = schema_err or "schema validation failed" + # Step 1: Schema establishment. Structural malformation wins and stops + # interpretation: without a valid shape the verifier has not established the + # bytes or fields to which a signature/key-binding verdict would refer. + schema_ok, schema_err, malformed_field = _validate_schema(claim_json) + if not schema_ok: + return VerificationResult( + status=VerificationStatus.UNVERIFIED, + verified_fields=[], + unverified_fields=["schema"], + failure_reason=VerificationError.CLAIM_MALFORMED, + attestation_age_seconds=-1, + is_attestation_fresh=False, + details={ + "schema_error": schema_err or "schema validation failed", + "malformed_field": malformed_field or "claim", + }, + ) + verified.append("schema") # Step 2: Signature sig_ok, sig_err = _verify_signature(claim_json) diff --git a/tests/unit/test_evidence_envelope_all_platforms.py b/tests/unit/test_evidence_envelope_all_platforms.py index ef4d7e9c..20e59746 100644 --- a/tests/unit/test_evidence_envelope_all_platforms.py +++ b/tests/unit/test_evidence_envelope_all_platforms.py @@ -43,7 +43,12 @@ generate_trace_claim, ) from cmcp_runtime.tee.base import jwk_thumbprint -from cmcp_verify.verify import ApprovedHashes, verify_trace_claim +from cmcp_verify.verify import ( + ApprovedHashes, + VerificationError, + VerificationStatus, + verify_trace_claim, +) POLICY_HASH = "sha256:" + "a" * 64 CATALOG_HASH = "sha256:" + "b" * 64 @@ -141,7 +146,6 @@ class _Result: ("azure-cvm-sev-snp", None, "cmcp_verify.azure_cvm", "verify_azure_cvm_measurement"), ("sev-snp", None, "cmcp_verify.sev_snp", "verify_sev_snp_measurement"), ("tdx", None, "cmcp_verify.tdx", "verify_tdx_measurement"), - ("tdx", "opaque", "cmcp_verify.opaque", "verify_opaque_measurement"), ], ) def test_platform_branch_reads_evidence_from_the_envelope( @@ -166,6 +170,32 @@ def test_platform_branch_reads_evidence_from_the_envelope( assert spy.kwargs["raw_evidence"] == EVIDENCE +@pytest.mark.parametrize("platform", ["sev-snp", "tdx", "opaque", "opaque-managed"]) +def test_unsupported_platform_cannot_bypass_schema( + monkeypatch: pytest.MonkeyPatch, platform: str, +) -> None: + """Legacy aliases cannot make unsupported platform claims reach crypto. + + TRACE 0.10.0 permits amd-sev-snp and intel-tdx, but not these aliases. + The old best-effort schema path could dispatch them despite the rejection. + No alias-to-platform promotion is authorized by this fixture repair. + """ + claim = _claim("tdx", platform_override=platform) + + def unexpected_crypto(**kwargs: object) -> None: + pytest.fail("schema-rejected platform reached cryptographic interpretation") + + monkeypatch.setattr("cmcp_verify.verify._verify_signature", unexpected_crypto) + monkeypatch.setattr("cmcp_verify.verify._verify_key_binding", unexpected_crypto) + result = verify_trace_claim(claim, _approved()) + + assert result.status is VerificationStatus.UNVERIFIED + assert result.failure_reason is VerificationError.CLAIM_MALFORMED + assert result.verified_fields == [] + assert result.unverified_fields == ["schema"] + assert result.details["malformed_field"] == "trace.runtime.platform" + + def test_sev_snp_reads_the_cert_chain_from_the_envelope( monkeypatch: pytest.MonkeyPatch, ) -> None: diff --git a/tests/unit/test_tpm_claim_signature_verify.py b/tests/unit/test_tpm_claim_signature_verify.py index 52a29b02..663d5eda 100644 --- a/tests/unit/test_tpm_claim_signature_verify.py +++ b/tests/unit/test_tpm_claim_signature_verify.py @@ -185,6 +185,7 @@ def _make_tpm2_claim( raw_evidence: bytes | None = None, quote_signature: bytes | None = None, cert_chain: bytes | None = None, + ek_cert_chain: bytes | None = None, ) -> dict: """A fully valid tpm2 claim: key-bound, chain-root-bound, correctly signed. @@ -215,6 +216,7 @@ def _make_tpm2_claim( raw_evidence=_b64(raw_evidence) if raw_evidence is not None else None, quote_signature=_b64(quote_signature) if quote_signature is not None else None, cert_chain=_b64(cert_chain) if cert_chain is not None else None, + ek_cert_chain=_b64(ek_cert_chain) if ek_cert_chain is not None else None, ), policy_bundle=PolicyBundleInfo( hash=POLICY_HASH, enforcement_mode="enforcing", policy_version="1.0.0" @@ -257,6 +259,7 @@ def _claim_for( *, tamper_attest: bool = False, omit: str | None = None, + ek_cert_chain: bytes | None = None, ) -> dict: """Build a signed claim carrying evidence through the producer path. @@ -278,6 +281,7 @@ def _claim_for( raw_evidence=attest, quote_signature=None if omit == "quote_signature" else signature, cert_chain=None if omit == "cert_chain" else chain_pem, + ek_cert_chain=ek_cert_chain, ) @@ -421,10 +425,11 @@ def test_a_separately_supplied_ek_chain_is_verified() -> None: assert established == ["ak_cert_chain", "ek_cert_chain"] # And end to end through the claim path. - claim = _claim_for(_pem(ak_cert) + _pem(ca_cert), ak_key) - claim["trace"]["runtime"]["ek_cert_chain"] = base64.b64encode( - _pem(ek_cert) + _pem(ca_cert) - ).decode() + claim = _claim_for( + _pem(ak_cert) + _pem(ca_cert), + ak_key, + ek_cert_chain=_pem(ek_cert) + _pem(ca_cert), + ) result = verify_trace_claim(claim, _approved(), trusted_tpm_ca_pem=_pem(ca_cert)) assert "ek_cert_chain" in result.verified_fields assert "ek_cert_chain" not in result.unverified_fields diff --git a/tests/unit/test_tpm_verify.py b/tests/unit/test_tpm_verify.py index 6f3753bd..e59d50e9 100644 --- a/tests/unit/test_tpm_verify.py +++ b/tests/unit/test_tpm_verify.py @@ -16,6 +16,7 @@ PolicyBundleInfo, ToolCatalogInfo, _to_dict, + canonical_json, generate_trace_claim, ) from cmcp_verify.tpm import verify_tpm_measurement @@ -208,9 +209,8 @@ def _make_tpm2_claim( ) -> dict: """Build a signed claim with tpm2 platform. - firmware_version and raw_evidence are injected directly into the serialized dict - after signing, since AttestationReportInfo does not carry those fields and - verify_trace_claim reads them from the raw dict. + Build through the producer path. Evidence belongs in the cmcp-owned gateway + envelope, not in the schema-closed TRACE runtime object. """ key = key or SigningKey() chain = AuditChain("tpm-session") @@ -227,6 +227,7 @@ def _make_tpm2_claim( report_data="00" * 32, attestation_generated_at=datetime.now(tz=UTC).isoformat(), attestation_validity_seconds=86400, + raw_evidence=raw_evidence_b64, ), policy_bundle=PolicyBundleInfo( hash=POLICY_HASH, @@ -249,7 +250,7 @@ def _make_tpm2_claim( audit_chain_root=chain.chain_root, audit_chain_tip=chain.chain_tip, audit_chain_length=chain.length, - do_sign=True, + do_sign=False, ) claim_dict = _to_dict(claim) @@ -258,9 +259,12 @@ def _make_tpm2_claim( claim_dict["trace"]["runtime"]["firmware_version"] = firmware_version if measurement != gen_measurement: claim_dict["trace"]["runtime"]["measurement"] = measurement - if raw_evidence_b64 is not None: - claim_dict["trace"]["runtime"]["raw_evidence"] = raw_evidence_b64 + # Sign last: every assertion and evidence field must be covered by the + # envelope signature, including the deliberately invalid test measurement. + claim_dict["signature"] = ( + base64.urlsafe_b64encode(key.sign(canonical_json(claim_dict))).rstrip(b"=").decode() + ) return claim_dict @@ -279,8 +283,12 @@ def test_tpm2_valid_measurement_triggers_tpm_path() -> None: def test_tpm2_invalid_measurement_format_fails() -> None: claim_dict = _make_tpm2_claim(measurement="bad-measurement") result = verify_trace_claim(claim_dict, _approved()) - assert "tpm_failure" in result.details - assert result.details["tpm_failure"] == "invalid_measurement_format" + # The declared measurement is invalid at the schema boundary; the TPM + # parser must not interpret a record whose structure was not established. + assert result.failure_reason == "CLAIM_MALFORMED" + assert result.verified_fields == [] + assert result.unverified_fields == ["schema"] + assert result.details["malformed_field"] == "trace.runtime.measurement" def test_software_only_stays_in_sw_path() -> None: diff --git a/tests/unit/test_verify_malformed_claim_shapes.py b/tests/unit/test_verify_malformed_claim_shapes.py new file mode 100644 index 00000000..f5f1cc44 --- /dev/null +++ b/tests/unit/test_verify_malformed_claim_shapes.py @@ -0,0 +1,87 @@ +"""Regression matrix for malformed TRACE Claim structure (#592). + +These vectors pin the maintainer-ruling structural boundary: malformed external +claim structure is classified as CLAIM_MALFORMED before signature or key-binding +interpretation, and the result identifies which intermediate failed to parse. +""" + +from __future__ import annotations + +import pytest + +import cmcp_verify.verify as verify_module +from cmcp_verify.verify import ( + ApprovedHashes, + VerificationError, + VerificationResult, + VerificationStatus, + verify_trace_claim, +) + +_APPROVED = ApprovedHashes( + policy_bundle_hash="sha256:" + "a" * 64, + tool_catalog_hash="sha256:" + "b" * 64, +) + +_CASES = [ + ("trace-string", {"trace": "bad"}, "trace"), + ("trace-list", {"trace": []}, "trace"), + ("trace-null", {"trace": None}, "trace"), + ("cnf-string", {"trace": {"cnf": "bad"}}, "trace.cnf"), + ("cnf-list", {"trace": {"cnf": []}}, "trace.cnf"), + ("cnf-bool", {"trace": {"cnf": True}}, "trace.cnf"), + ("jwk-string", {"trace": {"cnf": {"jwk": "bad"}}}, "trace.cnf.jwk"), + ("jwk-list", {"trace": {"cnf": {"jwk": []}}}, "trace.cnf.jwk"), + ("jwk-x-integer", {"trace": {"cnf": {"jwk": {"x": 1}}}}, "trace.cnf.jwk.x"), + ("jwk-x-list", {"trace": {"cnf": {"jwk": {"x": []}}}}, "trace.cnf.jwk.x"), + ("gateway-string", {"trace": {}, "gateway": "bad"}, "gateway"), + ("gateway-list", {"trace": {}, "gateway": []}, "gateway"), + ( + "audit-chain-string", + {"trace": {}, "gateway": {"audit_chain": "bad"}}, + "gateway.audit_chain", + ), + ( + "attestation-evidence-string", + { + "trace": {"runtime": {"platform": "tpm2"}}, + "gateway": {"attestation_evidence": "bad"}, + }, + "gateway.attestation_evidence", + ), +] + + +@pytest.mark.parametrize( + ("label", "claim", "malformed_field"), + _CASES, + ids=[case[0] for case in _CASES], +) +def test_malformed_claim_shape_is_structural_failure( + label: str, + claim: dict[str, object], + malformed_field: str, +) -> None: + result = verify_trace_claim(claim, _APPROVED) + + assert isinstance(result, VerificationResult), label + assert result.status is VerificationStatus.UNVERIFIED, label + assert result.failure_reason is VerificationError.CLAIM_MALFORMED, label + assert result.verified_fields == [], label + assert result.unverified_fields == ["schema"], label + assert result.attestation_age_seconds == -1, label + assert result.is_attestation_fresh is False, label + assert result.details.get("malformed_field") == malformed_field, label + assert "schema_error" in result.details, label + + +def test_malformed_claim_stops_before_crypto(monkeypatch: pytest.MonkeyPatch) -> None: + def unexpected_crypto(*args: object, **kwargs: object) -> tuple[bool, str | None]: + pytest.fail("malformed claim reached cryptographic interpretation") + + monkeypatch.setattr(verify_module, "_verify_signature", unexpected_crypto) + monkeypatch.setattr(verify_module, "_verify_key_binding", unexpected_crypto) + + result = verify_module.verify_trace_claim({"trace": "bad"}, _APPROVED) + + assert result.failure_reason is VerificationError.CLAIM_MALFORMED