Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 37 additions & 11 deletions src/cmcp_verify/verify.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand Down
34 changes: 32 additions & 2 deletions tests/unit/test_evidence_envelope_all_platforms.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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(
Expand All @@ -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:
Expand Down
13 changes: 9 additions & 4 deletions tests/unit/test_tpm_claim_signature_verify.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand Down Expand Up @@ -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"
Expand Down Expand Up @@ -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.

Expand All @@ -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,
)


Expand Down Expand Up @@ -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
Expand Down
24 changes: 16 additions & 8 deletions tests/unit/test_tpm_verify.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
PolicyBundleInfo,
ToolCatalogInfo,
_to_dict,
canonical_json,
generate_trace_claim,
)
from cmcp_verify.tpm import verify_tpm_measurement
Expand Down Expand Up @@ -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")
Expand All @@ -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,
Expand All @@ -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)
Expand All @@ -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


Expand All @@ -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:
Expand Down
87 changes: 87 additions & 0 deletions tests/unit/test_verify_malformed_claim_shapes.py
Original file line number Diff line number Diff line change
@@ -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
Loading