fix(verify/nv_certify): enforce strict JSON, exact int version, validated base64, and lowercase hex in verify_gateway_measurement (#810)

src/cmcp_verify/nv_certify.py

@@ -55,6 +55,7 @@
 import hashlib
 import hmac
 import json
+import math
 from collections.abc import Callable
 from dataclasses import dataclass, field
 from typing import Any
@@ -277,15 +278,54 @@ def fail(reason: str, **extra: str) -> NvCertifyResult:
     verified.append("policy")
 
     # 1. Envelope.
+    def _no_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
+        out: dict[str, Any] = {}
+        for key, value in pairs:
+            if key in out:
+                raise ValueError(f"duplicate JSON key: {key!r}")
+            out[key] = value
+        return out
+
+    def _no_nonfinite(token: str) -> Any:
+        raise ValueError(f"non-finite JSON constant: {token}")
+
+    def _parse_float(token: str) -> float:
+        val = float(token)
+        if not math.isfinite(val):
+            raise ValueError(f"non-finite JSON number: {token}")
+        return val
+
+    def _decode_b64_str(value: Any) -> bytes:
+        if not isinstance(value, str) or not value:
+            raise TypeError("expected non-empty base64 string")
+        return base64.b64decode(value.encode("ascii"), validate=True)
+
     try:
-        payload = json.loads(envelope)
-        if int(payload.get("v", 0)) != ENVELOPE_VERSION:
-            return fail("unsupported_envelope_version", version=str(payload.get("v")))
-        pre_attest = base64.b64decode(payload["pre_attest"])
-        pre_sig = base64.b64decode(payload["pre_signature"])
-        post_attest = base64.b64decode(payload["post_attest"])
-        post_sig = base64.b64decode(payload["post_signature"])
-        claimed_digest = bytes.fromhex(payload["gateway_digest"])
+        if not isinstance(envelope, (bytes, bytearray, memoryview)):
+            raise TypeError("envelope must be bytes")
+        payload = json.loads(
+            bytes(envelope).decode("utf-8"),
+            object_pairs_hook=_no_duplicate_keys,
+            parse_constant=_no_nonfinite,
+            parse_float=_parse_float,
+        )
+        if not isinstance(payload, dict):
+            raise TypeError("envelope must be a JSON object")
+        version = payload.get("v")
+        if type(version) is not int or version != ENVELOPE_VERSION:
+            return fail("unsupported_envelope_version", version=str(version))
+        pre_attest = _decode_b64_str(payload["pre_attest"])
+        pre_sig = _decode_b64_str(payload["pre_signature"])
+        post_attest = _decode_b64_str(payload["post_attest"])
+        post_sig = _decode_b64_str(payload["post_signature"])
+        gw_hex = payload["gateway_digest"]
+        if (
+            not isinstance(gw_hex, str)
+            or len(gw_hex) != 64
+            or any(c not in "0123456789abcdef" for c in gw_hex)
+        ):
+            raise ValueError("gateway_digest must be 64 lowercase hex characters")
+        claimed_digest = bytes.fromhex(gw_hex)
     except Exception as exc:  # noqa: BLE001
         return fail("malformed_envelope", error=f"{type(exc).__name__}: {exc}")
     verified.append("envelope")

src/cmcp_verify/nv_policy.py

@@ -116,6 +116,10 @@ def measurement_nv_public_bytes(
     cMCP seeds a new extend index before its first certification, so every signed
     Name must include the TPM-managed ``TPMA_NV_WRITTEN`` state bit.
     """
+    if type(index) is not int:
+        raise TypeError("index must be an integer")
+    if not isinstance(written, bool):
+        raise TypeError("written must be a boolean")
     if not 0x01000000 <= index <= 0x01FFFFFF:
         raise ValueError(f"index is not a TPM NV handle: {index:#x}")
     attributes = MEASUREMENT_NV_WRITTEN_ATTRIBUTES if written else MEASUREMENT_NV_BASE_ATTRIBUTES

tests/unit/test_nv_certify.py

@@ -392,7 +392,8 @@ def test_non_sha256_length_claimed_gateway_digest_is_rejected() -> None:
     short_digest = bytes(31)
     result = verify(fx, make_envelope(fx, gateway_digest=short_digest))
     assert not result.verified
-    assert result.failure_reason == "gateway_digest_mismatch"
+    assert result.failure_reason == "malformed_envelope"
+    assert "envelope" not in result.verified_fields
 
 
 def test_replayed_pair_from_another_session_is_rejected() -> None:
@@ -607,3 +608,52 @@ def test_accumulated_index_still_verifies() -> None:
         accumulated = hashlib.sha256(accumulated + bytes([i]) * 32).digest()
     result = verify(fx, make_envelope(fx, pre_contents=accumulated))
     assert result.verified, result.failure_reason
+
+
+@pytest.mark.parametrize("bad_v", [True, False, 1.9, "1", None])
+def test_non_int_envelope_version_is_rejected(bad_v: object) -> None:
+    fx = Fixture()
+    env = json.loads(make_envelope(fx))
+    env["v"] = bad_v
+    result = verify(fx, json.dumps(env).encode())
+    assert not result.verified
+    assert result.failure_reason == "unsupported_envelope_version"
+
+
+@pytest.mark.parametrize(
+    "mutated_raw",
+    [
+        lambda raw: raw[:-1] + b',"v":1}',
+        lambda _raw: b'{"v":1,"x":NaN}',
+        lambda raw: raw[:-1] + b',"extra":1e1000}',
+        lambda raw: raw[:-1] + b',"extra":{"nested":[-1e1000]}}',
+        lambda _raw: b"[1, 2]",
+        lambda raw: json.dumps({**json.loads(raw), "pre_attest": "!!!not-base64!!!"}).encode(),
+        lambda raw: json.dumps(
+            {**json.loads(raw), "gateway_digest": GATEWAY_DIGEST.hex().upper()}
+        ).encode(),
+        lambda raw: json.dumps({**json.loads(raw), "gateway_digest": "a" * 62}).encode(),
+        lambda raw: json.dumps({**json.loads(raw), "gateway_digest": "a" * 66}).encode(),
+    ],
+)
+def test_envelope_rejects_duplicate_keys_nonfinite_non_base64_and_uppercase_hex(
+    mutated_raw,
+) -> None:
+    fx = Fixture()
+    raw = make_envelope(fx)
+    assert verify(fx, raw).verified
+    result = verify(fx, mutated_raw(raw))
+    assert not result.verified
+    assert result.failure_reason == "malformed_envelope"
+    assert "envelope" not in result.verified_fields
+
+
+@pytest.mark.parametrize(
+    ("bad_index", "bad_written"),
+    [(True, True), ("0x01500432", True), (0x01500432, "yes")],
+)
+def test_measurement_nv_public_bytes_validates_types(bad_index, bad_written) -> None:
+    from cmcp_verify.nv_policy import measurement_nv_public_bytes
+
+    with pytest.raises(TypeError):
+        measurement_nv_public_bytes(bad_index, written=bad_written)