diff --git a/prototype/tdx_collateral.py b/prototype/tdx_collateral.py index 3afb748d..22b4b1ce 100644 --- a/prototype/tdx_collateral.py +++ b/prototype/tdx_collateral.py @@ -28,6 +28,7 @@ import base64 import hashlib import json +import math import re from dataclasses import dataclass, field from datetime import UTC, datetime @@ -182,11 +183,27 @@ def _unique(pairs: list[tuple[str, Any]]) -> dict[str, Any]: return value +def _no_constants(token: str) -> Any: + raise ValueError(f"non-finite JSON constant: {token}") + + +def _finite_float(token: str) -> float: + val = float(token) + if not math.isfinite(val): + raise ValueError(f"non-finite JSON float: {token}") + return val + + def _json(raw: bytes, name: str) -> Any: if type(raw) is not bytes or len(raw) > LIMIT: raise CollateralDenied(f"{name}_format") try: - return json.loads(raw, object_pairs_hook=_unique) + return json.loads( + raw, + object_pairs_hook=_unique, + parse_constant=_no_constants, + parse_float=_finite_float, + ) except (ValueError, RecursionError) as exc: if isinstance(exc, CollateralDenied): raise @@ -326,7 +343,9 @@ def _check_qcnl(raw: bytes, policy: CollateralPolicy) -> None: if config.get("use_secure_cert") is not True: raise CollateralDenied("qcnl_insecure") if ( - config.get("pck_cache_expire_hours") != 0 + type(config.get("pck_cache_expire_hours")) is not int + or config.get("pck_cache_expire_hours") != 0 + or type(config.get("verify_collateral_cache_expire_hours")) is not int or config.get("verify_collateral_cache_expire_hours") != 0 ): raise CollateralDenied("qcnl_cache") @@ -458,12 +477,17 @@ def _sequence(blob: bytes) -> list[tuple[int, bytes]]: def _oid(value: bytes) -> str: - if not value: + if not value or value[-1] & 0x80: raise CollateralDenied("pck_extension") - parts, number = [min(value[0] // 40, 2), value[0] - 40 * min(value[0] // 40, 2)], 0 + parts = [min(value[0] // 40, 2), value[0] - 40 * min(value[0] // 40, 2)] + number = 0 + in_subid = False for byte in value[1:]: + if not in_subid and byte == 0x80: + raise CollateralDenied("pck_extension") number = number << 7 | byte & 0x7F - if not byte & 0x80: + in_subid = bool(byte & 0x80) + if not in_subid: parts.append(number) number = 0 return ".".join(map(str, parts)) @@ -482,20 +506,27 @@ def _pck_extension(cert: x509.Certificate) -> dict[str, Any]: raise CollateralDenied("pck_extension") base = _SGX_EXTENSION.dotted_string values: dict[str, tuple[int, bytes]] = {} + seen_tcb = False for tag, item in _sequence(body): pair = _sequence(item) if tag == 0x30 else [] if len(pair) != 2 or pair[0][0] != 0x06: raise CollateralDenied("pck_extension") name = _oid(pair[0][1]) if name == base + ".2": - if pair[1][0] != 0x30: + if seen_tcb or pair[1][0] != 0x30: raise CollateralDenied("pck_extension") + seen_tcb = True for inner_tag, inner in _sequence(pair[1][1]): sub = _sequence(inner) if inner_tag == 0x30 else [] if len(sub) != 2 or sub[0][0] != 0x06: raise CollateralDenied("pck_extension") - values[_oid(sub[0][1])] = sub[1] + sub_name = _oid(sub[0][1]) + if sub_name in values: + raise CollateralDenied("pck_extension") + values[sub_name] = sub[1] else: + if name in values: + raise CollateralDenied("pck_extension") values[name] = pair[1] def integer(oid: str) -> int: @@ -601,7 +632,9 @@ def appraise_collateral( if ( not isinstance(snapshot, dict) or set(snapshot) != {"version", "tee_type", "fields"} + or type(snapshot["version"]) is not int or snapshot["version"] != 3 + or type(snapshot["tee_type"]) is not int or snapshot["tee_type"] != 0x81 or not isinstance(snapshot["fields"], dict) or set(snapshot["fields"]) != set(FIELDS) @@ -634,9 +667,19 @@ def appraise_collateral( tcb_info = _signed_json(fields["tcb_info"], "tcbInfo", tcb_chain[0], "tcb_info") qe_identity = _signed_json(fields["qe_identity"], "enclaveIdentity", qe_chain[0], "qe_identity") - if tcb_info.get("id") != "TDX" or tcb_info.get("version") != 3 or tcb_info.get("tcbType") != 0: + if ( + tcb_info.get("id") != "TDX" + or type(tcb_info.get("version")) is not int + or tcb_info.get("version") != 3 + or type(tcb_info.get("tcbType")) is not int + or tcb_info.get("tcbType") != 0 + ): raise CollateralDenied("tcb_info_kind") - if qe_identity.get("id") != "TD_QE" or qe_identity.get("version") != 2: + if ( + qe_identity.get("id") != "TD_QE" + or type(qe_identity.get("version")) is not int + or qe_identity.get("version") != 2 + ): raise CollateralDenied("qe_identity_kind") fresh_until = min( _window(tcb_info, at, "tcb_info"), @@ -688,6 +731,7 @@ def appraise_collateral( "qe_identity", ) or str(qe_identity.get("mrsigner", "")).lower() != qe[128:160].hex() + or type(qe_identity.get("isvprodid")) is not int or qe_identity.get("isvprodid") != int.from_bytes(qe[256:258], "little") ): raise CollateralDenied("qe_identity_mismatch") @@ -728,6 +772,12 @@ def appraise_collateral( status = level.get("tcbStatus") if not isinstance(status, str) or status not in _STATUS_TO_QV: raise CollateralDenied("tcb_status_unknown") + advisories = level.get("advisoryIDs") + if advisories is not None and ( + not isinstance(advisories, list) + or any(not isinstance(item, str) or not item for item in advisories) + ): + raise CollateralDenied("tcb_info_levels") module_status = None if tee_tcb_svn[1] > 0: @@ -829,7 +879,9 @@ def appraise( offline = appraise_collateral(packet.collateral, quote, trust_root, at=verified_at) number = policy.tcb_evaluation_data_number if ( - offline["tcb_evaluation_data_number"] != number + type(offline["tcb_evaluation_data_number"]) is not int + or offline["tcb_evaluation_data_number"] != number + or type(offline["qe_tcb_evaluation_data_number"]) is not int or offline["qe_tcb_evaluation_data_number"] != number ): raise CollateralDenied("tcb_evaluation_data_number") diff --git a/tests/test_tdx_collateral.py b/tests/test_tdx_collateral.py index 18cfc825..2f8e7e84 100644 --- a/tests/test_tdx_collateral.py +++ b/tests/test_tdx_collateral.py @@ -694,3 +694,252 @@ def test_pck_extension_rejects_non_minimal_and_negative_der_integers(): t._pck_extension, _build_pck_cert_with_sgx_ext(pcesvn_bytes=bad_int), ) + + +@pytest.mark.parametrize( + "raw", + [ + b'{"x": NaN}', + b'{"x": Infinity}', + b'{"x": -Infinity}', + b'{"x": 1e309}', + b'{"x": -1e309}', + ], +) +def test_json_rejects_nan_infinity_and_float_overflow(raw: bytes): + assert t._json(b'{"x": 1.5}', "record") == {"x": 1.5} + denied("record_format", t._json, raw, "record") + + +@pytest.mark.parametrize( + "bad_oid", + [ + b"\x2a\x86\x48\x86\xf8\x4d\x01\x0d\x01\x04\x81", + b"\x2a\x86\x48\x86\xf8\x4d\x01\x0d\x01\x80\x04", + ], +) +def test_oid_rejects_trailing_continuation_and_leading_zero_subidentifier( + bad_oid: bytes, +): + assert t._oid(b"\x2a\x86\x48\x86\xf8\x4d\x01\x0d\x01\x04") == "1.2.840.113741.1.13.1.4" + denied("pck_extension", t._oid, bad_oid) + + +@pytest.mark.parametrize("case", ["dup_tcb_block", "dup_outer_oid", "dup_inner_oid"]) +def test_pck_extension_rejects_duplicate_outer_and_inner_oids(case: str): + base = t._SGX_EXTENSION.dotted_string + tcb_entries = [ + _der_tlv(0x30, _der_oid(f"{base}.2.{i}") + _der_tlv(0x02, b"\x01")) + for i in range(1, 18) + ] + if case == "dup_inner_oid": + tcb_entries.append(_der_tlv(0x30, _der_oid(f"{base}.2.1") + _der_tlv(0x02, b"\x02"))) + tcb_seq = _der_tlv(0x30, b"".join(tcb_entries)) + outer = [ + _der_tlv(0x30, _der_oid(f"{base}.2") + tcb_seq), + _der_tlv(0x30, _der_oid(f"{base}.3") + _der_tlv(0x04, b"\x00\x00")), + _der_tlv(0x30, _der_oid(f"{base}.4") + _der_tlv(0x04, bytes.fromhex("00806f050000"))), + ] + if case == "dup_tcb_block": + outer.append(_der_tlv(0x30, _der_oid(f"{base}.2") + tcb_seq)) + elif case == "dup_outer_oid": + outer.append( + _der_tlv(0x30, _der_oid(f"{base}.4") + _der_tlv(0x04, bytes.fromhex("11806f050000"))) + ) + top_seq = _der_tlv(0x30, b"".join(outer)) + key = ec.generate_private_key(ec.SECP256R1()) + subject = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "Intel SGX PCK Certificate")]) + now = datetime(2026, 1, 1, tzinfo=UTC) + cert = ( + x509.CertificateBuilder() + .subject_name(subject) + .issuer_name(subject) + .public_key(key.public_key()) + .serial_number(1) + .not_valid_before(now) + .not_valid_after(datetime(2030, 1, 1, tzinfo=UTC)) + .add_extension(x509.UnrecognizedExtension(t._SGX_EXTENSION, top_seq), critical=False) + .sign(key, hashes.SHA256()) + ) + denied("pck_extension", t._pck_extension, cert) + + +@pytest.mark.parametrize( + ("field_name", "bad_val"), + [ + ("pck_cache_expire_hours", False), + ("pck_cache_expire_hours", 0.0), + ("verify_collateral_cache_expire_hours", False), + ("verify_collateral_cache_expire_hours", 0.0), + ], +) +def test_check_qcnl_rejects_boolean_and_float_cache_expire_hours( + field_name: str, bad_val: object +): + valid_qcnl = { + "tcb_update_type": "standard", + "pccs_url": t.INTEL_PCS + "sgx/certification/v4/", + "collateral_service": t.INTEL_PCS + "sgx/certification/v4/", + "use_secure_cert": True, + "pck_cache_expire_hours": 0, + "verify_collateral_cache_expire_hours": 0, + } + t._check_qcnl(json.dumps(valid_qcnl).encode(), STANDARD) + bad_qcnl = {**valid_qcnl, field_name: bad_val} + denied("qcnl_cache", t._check_qcnl, json.dumps(bad_qcnl).encode(), STANDARD) + + +@pytest.mark.parametrize( + ("mutation", "expected_reason"), + [ + ("snapshot_version_float", "collateral_structure"), + ("snapshot_tee_type_float", "collateral_structure"), + ("tcb_version_float", "tcb_info_kind"), + ("tcb_type_bool", "tcb_info_kind"), + ("qe_version_float", "qe_identity_kind"), + ("qe_isvprodid_bool", "qe_identity_mismatch"), + ("advisory_str", "tcb_info_levels"), + ("advisory_empty_str", "tcb_info_levels"), + ("advisory_int_item", "tcb_info_levels"), + ], +) +def test_appraise_collateral_rejects_non_int_fields_and_invalid_advisory_ids( + monkeypatch: pytest.MonkeyPatch, mutation: str, expected_reason: str +): + import sys + import types + + cert = _build_pck_cert_with_sgx_ext() + + class _DummyCrl: + next_update_utc = datetime(2026, 6, 1, tzinfo=UTC) + + def get_revoked_certificate_by_serial_number(self, _s): + return None + + tcb_doc = { + "id": "TDX", + "version": 3, + "tcbType": 0, + "issueDate": "2026-01-01T00:00:00Z", + "nextUpdate": "2026-06-01T00:00:00Z", + "fmspc": "00806F050000", + "pceId": "0000", + "tcbEvaluationDataNumber": 20, + "tdxModule": { + "mrsigner": "00" * 48, + "attributes": "00" * 8, + "attributesMask": "ff" * 8, + }, + "tcbLevels": [ + { + "tcb": { + "sgxtcbcomponents": [{"svn": 1} for _ in range(16)], + "pcesvn": 1, + "tdxtcbcomponents": [{"svn": 0} for _ in range(16)], + }, + "tcbDate": "2026-01-01T00:00:00Z", + "tcbStatus": "UpToDate", + "advisoryIDs": ["INTEL-SA-00001"], + } + ], + } + qe_doc = { + "id": "TD_QE", + "version": 2, + "issueDate": "2026-01-01T00:00:00Z", + "nextUpdate": "2026-06-01T00:00:00Z", + "miscselect": "00000000", + "miscselectMask": "ffffffff", + "attributes": "00" * 16, + "attributesMask": "ff" * 16, + "mrsigner": "00" * 32, + "isvprodid": 0, + "tcbEvaluationDataNumber": 20, + "tcbLevels": [{"tcb": {"isvsvn": 0}, "tcbStatus": "UpToDate"}], + } + if mutation == "tcb_version_float": + tcb_doc["version"] = 3.0 + elif mutation == "tcb_type_bool": + tcb_doc["tcbType"] = False + elif mutation == "qe_version_float": + qe_doc["version"] = 2.0 + elif mutation == "qe_isvprodid_bool": + qe_doc["isvprodid"] = False + elif mutation == "advisory_str": + tcb_doc["tcbLevels"][0]["advisoryIDs"] = "INTEL-SA-00001" + elif mutation == "advisory_empty_str": + tcb_doc["tcbLevels"][0]["advisoryIDs"] = [""] + elif mutation == "advisory_int_item": + tcb_doc["tcbLevels"][0]["advisoryIDs"] = [123] + + wcm_mod = types.ModuleType("wcm") + tdx_mod = types.ModuleType("wcm.tdx") + qv_mod = types.ModuleType("wcm._quote_verify") + + class _QuoteFormatError(Exception): + pass + + class _TrustStore: + def add_root(self, _r): + pass + + class _ParsedReport: + raw = bytes(584) + + class _ParsedQuote: + version = 4 + pck_leaf = cert + pck_intermediates = [cert] + qe_report = bytes(384) + report = _ParsedReport() + + qv_mod.QuoteFormatError = _QuoteFormatError # type: ignore[attr-defined] + qv_mod.TrustStore = _TrustStore # type: ignore[attr-defined] + tdx_mod.verify_tdx_quote = lambda *_a, **_k: types.SimpleNamespace(verified=True) # type: ignore[attr-defined] + tdx_mod.parse_tdx_quote = lambda _q: _ParsedQuote() # type: ignore[attr-defined] + monkeypatch.setitem(sys.modules, "wcm", wcm_mod) + monkeypatch.setitem(sys.modules, "wcm.tdx", tdx_mod) + monkeypatch.setitem(sys.modules, "wcm._quote_verify", qv_mod) + monkeypatch.setattr(t, "_chain", lambda *_a, **_k: [cert, cert]) + monkeypatch.setattr( + t, + "_cn", + lambda c: t._TCB_SIGNER if c is cert else t._PCK_CAS[0], + ) + monkeypatch.setattr(t, "_crl", lambda *_a, **_k: _DummyCrl()) + monkeypatch.setattr( + t, + "_signed_json", + lambda _raw, key, _s, _n: tcb_doc if key == "tcbInfo" else qe_doc, + ) + + def _chain_side_effect(_raw, _root, _when, name): + if name == "pck_crl": + return [types.SimpleNamespace(subject=cert.issuer, serial_number=2), cert] + return [cert, cert] + + monkeypatch.setattr(t, "_chain", _chain_side_effect) + monkeypatch.setattr( + t, + "_cn", + lambda c: t._TCB_SIGNER if c is cert else t._PCK_CAS[0], + ) + _ParsedQuote.pck_intermediates = [ + _chain_side_effect(b"", None, None, "pck_crl")[0] # type: ignore[list-item] + ] + + snapshot = { + "version": 3.0 if mutation == "snapshot_version_float" else 3, + "tee_type": 129.0 if mutation == "snapshot_tee_type_float" else 0x81, + "fields": {name: base64.b64encode(b"x\x00").decode() for name in t.FIELDS}, + } + denied( + expected_reason, + t.appraise_collateral, + json.dumps(snapshot).encode(), + b"quote", + cert, + at=1770000000, + ) +