fix(verify/tdx,embodied_action): enforce hex format in verify_tdx_measurement and guard non-dict inputs in verify_embodied_action_evidence (#803)

src/cmcp_verify/embodied_action.py

@@ -14,7 +14,7 @@
 
 EMBODIED_ACTION_PROFILE = "cmcp.embodied_action_evidence.v0"
 
-_HASH_RE = re.compile(r"^sha(256|384):([0-9a-f]+)$")
+_HASH_RE = re.compile(r"^sha(256:[0-9a-f]{64}|384:[0-9a-f]{96})$")
 _ACTION_REF_FIELDS = ("agent_id", "action_type", "action_scope", "action_timestamp")
 _REQUIRED_PAYLOAD_FIELDS = (
     "profile",
@@ -96,30 +96,48 @@ def compute_action_ref(payload: dict[str, Any]) -> str:
 
 
 def _claim_policy_hash(claim_json: dict[str, Any]) -> str | None:
-    policy = claim_json.get("trace", {}).get("policy", {})
+    if not isinstance(claim_json, dict):
+        return None
+    trace = claim_json.get("trace")
+    if not isinstance(trace, dict):
+        return None
+    policy = trace.get("policy")
+    if not isinstance(policy, dict):
+        return None
     value = policy.get("bundle_hash")
     return value if isinstance(value, str) else None
 
 
 def _claim_catalog_hash(claim_json: dict[str, Any]) -> str | None:
-    catalog = claim_json.get("gateway", {}).get("catalog", {})
+    if not isinstance(claim_json, dict):
+        return None
+    gateway = claim_json.get("gateway")
+    if not isinstance(gateway, dict):
+        return None
+    catalog = gateway.get("catalog")
+    if not isinstance(catalog, dict):
+        return None
     value = catalog.get("hash")
     return value if isinstance(value, str) else None
 
 
 def _claim_agent_id(claim_json: dict[str, Any]) -> str | None:
-    identity = claim_json.get("gateway", {}).get("agent_identity")
+    if not isinstance(claim_json, dict):
+        return None
+    gateway = claim_json.get("gateway")
+    if not isinstance(gateway, dict):
+        return None
+    identity = gateway.get("agent_identity")
     if not isinstance(identity, dict):
         return None
     value = identity.get("agent_id")
     return value if isinstance(value, str) else None
 
 
 def _verify_hash_value(hash_value: str, payload: dict[str, Any]) -> bool:
-    match = _HASH_RE.match(hash_value)
-    if not match:
+    if not _HASH_RE.match(hash_value):
         return False
-    algorithm = f"sha{match.group(1)}"
+    algorithm = hash_value.split(":", 1)[0]
     return hmac.compare_digest(hash_value, hash_embodied_action_payload(payload, algorithm=algorithm))
 
 
@@ -157,6 +175,16 @@ def verify_embodied_action_evidence(
     failures: list[str] = []
     warnings: list[str] = []
 
+    if not isinstance(audit_entry, dict) or not isinstance(detached_payload, dict):
+        failures.append("audit_entry and detached_payload must be JSON objects")
+        return EmbodiedActionEvidenceResult(
+            verified=False,
+            receipt_state=ReceiptState.INVALID,
+            verified_fields=verified_fields,
+            failures=failures,
+            warnings=warnings,
+        )
+
     evidence = audit_entry.get("external_execution_evidence")
     if not isinstance(evidence, dict):
         failures.append("audit entry has no external_execution_evidence object")
@@ -231,26 +259,34 @@ def verify_embodied_action_evidence(
             verified_fields.append("governance_decision")
 
     if claim_json is not None:
-        policy_hash = _claim_policy_hash(claim_json)
-        if policy_hash is not None:
-            if detached_payload.get("policy_bundle_hash") != policy_hash:
+        if not isinstance(claim_json, dict):
+            failures.append("claim_json must be a JSON object when supplied")
+        else:
+            policy_hash = _claim_policy_hash(claim_json)
+            if policy_hash is None:
+                failures.append("claim_json.trace.policy.bundle_hash must be a string")
+            elif detached_payload.get("policy_bundle_hash") != policy_hash:
                 failures.append("policy_bundle_hash does not match TRACE Claim policy hash")
             else:
                 verified_fields.append("policy_bundle_hash")
 
-        catalog_hash = _claim_catalog_hash(claim_json)
-        if catalog_hash is not None:
-            if detached_payload.get("tool_catalog_hash") != catalog_hash:
+            catalog_hash = _claim_catalog_hash(claim_json)
+            if catalog_hash is None:
+                failures.append("claim_json.gateway.catalog.hash must be a string")
+            elif detached_payload.get("tool_catalog_hash") != catalog_hash:
                 failures.append("tool_catalog_hash does not match TRACE Claim catalog hash")
             else:
                 verified_fields.append("tool_catalog_hash")
 
-        agent_id = _claim_agent_id(claim_json)
-        if agent_id is not None:
-            if detached_payload.get("agent_id") != agent_id:
-                failures.append("agent_id does not match TRACE Claim agent identity")
-            else:
-                verified_fields.append("agent_id")
+            agent_id = _claim_agent_id(claim_json)
+            gateway = claim_json.get("gateway")
+            if isinstance(gateway, dict) and "agent_identity" in gateway and not agent_id:
+                failures.append("claim_json.gateway.agent_identity.agent_id must be a non-empty string")
+            elif agent_id is not None:
+                if detached_payload.get("agent_id") != agent_id:
+                    failures.append("agent_id does not match TRACE Claim agent identity")
+                else:
+                    verified_fields.append("agent_id")
 
     ros2 = detached_payload.get("ros2")
     if ros2 is not None:

src/cmcp_verify/tdx.py

@@ -3,6 +3,7 @@
 
 import hashlib
 import hmac
+import re
 import urllib.request
 from dataclasses import dataclass, field
 
@@ -19,6 +20,8 @@
 from cmcp_runtime.tee.tdreport import TdReport as _TdReport
 
 _TDREPORT_MIN_SIZE = TDREPORT_SIZE
+_MEASUREMENT_RE = re.compile(r"^sha384:[0-9a-f]{96}$")
+_HEX_RE = re.compile(r"^[0-9a-fA-F]+$")
 
 # Intel DCAP QE identity endpoint (used to confirm DCAP service reachability)
 _DCAP_QE_IDENTITY_URL = (
@@ -71,15 +74,7 @@ def verify_tdx_measurement(
     result = TDXVerificationResult(verified=True)
 
     # Step 1: Format check
-    if not measurement.startswith("sha384:"):
-        result.verified = False
-        result.failure_reason = "invalid_measurement_format"
-        result.unverified_fields.extend(["dcap_quote_signature", "tcb_status"])
-        result.details["dcap_chain"] = "requires_intel_dcap_service"
-        return result
-
-    hex_part = measurement[len("sha384:"):]
-    if len(hex_part) != 96:
+    if not isinstance(measurement, str) or not _MEASUREMENT_RE.fullmatch(measurement):
         result.verified = False
         result.failure_reason = "invalid_measurement_format"
         result.unverified_fields.extend(["dcap_quote_signature", "tcb_status"])
@@ -125,6 +120,19 @@ def verify_tdx_measurement(
             # non-zero expectation silently failed and was silently ignored.
             if report_data_hex is not None:
                 extracted_rd = bytes(tdreport.report_mac.report_data)
+                if (
+                    not isinstance(report_data_hex, str)
+                    or not report_data_hex
+                    or len(report_data_hex) % 2 != 0
+                    or not _HEX_RE.fullmatch(report_data_hex)
+                ):
+                    result.verified = False
+                    result.failure_reason = "report_data_mismatch"
+                    result.details["report_data_actual"] = extracted_rd[:16].hex()
+                    result.unverified_fields.extend(
+                        ["dcap_quote_signature", "tcb_status"]
+                    )
+                    return result
                 expected_rd = bytes.fromhex(report_data_hex[:REPORT_DATA_SIZE * 2])
                 if len(expected_rd) < REPORT_DATA_SIZE:
                     # Zero-pad: the guest writes a short nonce into a 64-byte

tests/unit/test_embodied_action_evidence.py

@@ -291,13 +291,21 @@ def test_agent_identity_mismatch_fails():
     assert any("agent_id" in failure for failure in result.failures)
 
 
-def test_non_object_agent_identity_is_ignored():
+@pytest.mark.parametrize("identity", [None, {}, {"agent_id": 42}, {"agent_id": ""}, "not-an-object"])
+def test_malformed_present_agent_identity_fails(identity):
     fixture = copy.deepcopy(_fixture())
-    fixture["trace_claim"]["gateway"]["agent_identity"] = "not-an-object"
-
+    fixture["trace_claim"]["gateway"]["agent_identity"] = identity
     result = _verify_fixture(fixture)
+    assert not result.verified
+    assert any("agent_identity.agent_id" in failure for failure in result.failures)
+
 
+def test_absent_agent_identity_keeps_optional_binding():
+    fixture = copy.deepcopy(_fixture())
+    fixture["trace_claim"]["gateway"].pop("agent_identity", None)
+    result = _verify_fixture(fixture)
     assert result.verified, result.failures
+    assert "agent_id" not in result.verified_fields
 
 
 def test_ros2_must_be_object_when_present():
@@ -440,6 +448,48 @@ def test_physical_completion_claim_fails_closed():
     assert any("physical completion" in failure for failure in result.failures)
 
 
+def test_non_dict_audit_entry_or_payload_and_short_hash_rejected():
+    fixture = copy.deepcopy(_fixture())
+    for bad in [None, 123, "str", []]:
+        res1 = verify_embodied_action_evidence(bad, fixture["detached_payload"])  # type: ignore[arg-type]
+        assert not res1.verified
+        assert res1.receipt_state == ReceiptState.INVALID
+        res2 = verify_embodied_action_evidence(fixture["audit_entry"], bad)  # type: ignore[arg-type]
+        assert not res2.verified
+        assert res2.receipt_state == ReceiptState.INVALID
+
+    # Explicit None claim_json is the no-Claim mode and succeeds
+    res_no_claim = verify_embodied_action_evidence(
+        fixture["audit_entry"],
+        fixture["detached_payload"],
+        None,
+    )
+    assert res_no_claim.verified, res_no_claim.failures
+    assert "policy_bundle_hash" not in res_no_claim.verified_fields
+
+    # A supplied but malformed claim_json fails closed without raising AttributeError
+    for bad_claim in (
+        {"trace": None, "gateway": None},
+        {"trace": {"policy": None}, "gateway": {"catalog": None}},
+        {},
+        "not-a-dict",
+    ):
+        res3 = verify_embodied_action_evidence(
+            fixture["audit_entry"],
+            fixture["detached_payload"],
+            bad_claim,  # type: ignore[arg-type]
+        )
+        assert not res3.verified
+        assert res3.failures
+
+    # Short sha256 digest (< 64 hex chars) is rejected by _HASH_RE
+    bad_hash_entry = copy.deepcopy(fixture["audit_entry"])
+    bad_hash_entry["external_execution_evidence"]["evidence_hash"] = "sha256:abcd"
+    res4 = verify_embodied_action_evidence(bad_hash_entry, fixture["detached_payload"])
+    assert not res4.verified
+    assert any("evidence_hash" in f for f in res4.failures)
+
+
 # -- canonical_json_bytes: RFC 8785/JCS compliance (agentrust-io/cmcp#588) ------
 
 

tests/unit/test_tdx_verify.py

@@ -217,3 +217,18 @@ def test_measurement_does_not_move_with_the_nonce(monkeypatch):
     window = slice(_OLD_MRTD_OFFSET, _OLD_MRTD_OFFSET + MRTD_SIZE)
     assert first_report[window] != second_report[window]
 
+
+def test_tdx_invalid_measurement_format_rejected_before_raw_evidence():
+    for bad_meas in (None, 123, "sha384:" + "Z" * 96, "sha384:" + "A" * 96, "sha384:abcd"):
+        res = verify_tdx_measurement(bad_meas, None)  # type: ignore[arg-type]
+        assert res.verified is False
+        assert res.failure_reason == "invalid_measurement_format"
+
+
+def test_tdx_malformed_report_data_hex_classified_as_report_data_mismatch(monkeypatch):
+    monkeypatch.setattr("cmcp_verify.tdx._check_dcap_reachable", lambda: False)
+    mrtd = _mrtd()
+    report = _make_tdreport(mrtd, bytes([0x11]) * REPORT_DATA_SIZE)
+    res = verify_tdx_measurement(_measurement_of(mrtd), report, report_data_hex="not-hex!")
+    assert res.verified is False
+    assert res.failure_reason == "report_data_mismatch"