From e2fd3a4d345aafaaf972a10648020e13f816cb0f Mon Sep 17 00:00:00 2001 From: Raghuvar Karthik Durga <177662415+drk-ram@users.noreply.github.com> Date: Sat, 31 Oct 2026 10:07:19 -0500 Subject: [PATCH] fix(tpm): reject trailing quote bytes, bound AIA HTTP fetches, and verify report.platform in verify_tpm_report (#282) Signed-off-by: Raghuvar Karthik Durga <177662415+drk-ram@users.noreply.github.com> --- src/ca2a_runtime/tee/tpm.py | 19 +++++++++++++++++-- src/ca2a_verify/tpm.py | 5 +++++ tests/unit/test_tpm.py | 31 +++++++++++++++++++++++++++++++ 3 files changed, 53 insertions(+), 2 deletions(-) diff --git a/src/ca2a_runtime/tee/tpm.py b/src/ca2a_runtime/tee/tpm.py index 2d6f8f3a..b1ee68d0 100644 --- a/src/ca2a_runtime/tee/tpm.py +++ b/src/ca2a_runtime/tee/tpm.py @@ -79,6 +79,7 @@ # party verify offline later: the chain travels with the evidence. _AIA_FETCH_TIMEOUT_SECONDS = 5 _AIA_MAX_DEPTH = 4 +_AIA_MAX_CERT_BYTES = 64 * 1024 _AIA_CA_ISSUERS_OID = "1.3.6.1.5.5.7.48.2" @@ -123,6 +124,9 @@ class TpmQuote: @classmethod def parse(cls, blob: bytes) -> TpmQuote: + if not isinstance(blob, (bytes, bytearray, memoryview)): + raise AttestationFailed("TPM quote must be bytes", detail=f"got {type(blob).__name__}") + blob = bytes(blob) if len(blob) < 6: raise AttestationFailed("TPM quote too short") magic = int.from_bytes(blob[0:4], "big") @@ -141,12 +145,17 @@ def parse(cls, blob: bytes) -> TpmQuote: size_of_select = blob[pos + 2] pos += 3 + size_of_select pcr_digest, pos = _read_2b(blob, pos) + if pos != len(blob): + raise AttestationFailed( + "TPM quote carries trailing bytes after PCR digest", + detail=f"parsed {pos} bytes, got {len(blob)} bytes", + ) return cls( magic=magic, attest_type=attest_type, qualifying_data=qualifying_data, pcr_digest=pcr_digest, - raw=bytes(blob), + raw=blob, ) @@ -495,7 +504,13 @@ def load_any(data: bytes) -> list[x509.Certificate]: with urllib.request.urlopen( # noqa: S310 # nosec B310 url, timeout=_AIA_FETCH_TIMEOUT_SECONDS ) as response: - found = load_any(response.read()) + raw = response.read(_AIA_MAX_CERT_BYTES + 1) + if len(raw) > _AIA_MAX_CERT_BYTES: + logger.debug( + "AIA response exceeded %d bytes for %s", _AIA_MAX_CERT_BYTES, url + ) + continue + found = load_any(raw) except Exception as exc: # noqa: BLE001 logger.debug("AIA fetch failed for %s: %s", url, exc) continue diff --git a/src/ca2a_verify/tpm.py b/src/ca2a_verify/tpm.py index 54840dc0..0457fd16 100644 --- a/src/ca2a_verify/tpm.py +++ b/src/ca2a_verify/tpm.py @@ -197,6 +197,11 @@ def verify_tpm_report( report's ``measurement`` field, so what the caller acts on is the value the TPM signed. A mismatch between the two is itself a failure. """ + if report.platform != "tpm": + raise AttestationFailed( + "expected a 'tpm' attestation report", + detail=f"got platform={report.platform!r}", + ) if report.nonce != expected_nonce: raise AttestationFailed( "the report nonce does not match the expected nonce", diff --git a/tests/unit/test_tpm.py b/tests/unit/test_tpm.py index bf893d8f..239ac791 100644 --- a/tests/unit/test_tpm.py +++ b/tests/unit/test_tpm.py @@ -188,3 +188,34 @@ def test_real_tpm_quote_parses_and_verifies() -> None: tampered[100] ^= 0x01 with pytest.raises(InvalidSignature): ak_pub.verify(signature, bytes(tampered), padding.PKCS1v15(), hashes.SHA256()) + + +@pytest.mark.parametrize( + "bad_blob", + [ + build_attest(qualifying_data=b"n", pcr_digest=b"\x11" * 32) + b"\x00", + build_attest(qualifying_data=b"n", pcr_digest=b"\x11" * 32) + b"trailing", + "not-bytes", # type: ignore[list-item] + ], +) +def test_tpm_quote_parse_rejects_trailing_bytes_and_non_bytes(bad_blob: bytes) -> None: + with pytest.raises(AttestationFailed): + TpmQuote.parse(bad_blob) + + +@pytest.mark.parametrize("wrong_platform", ["tdx", "sev-snp", "software-only", ""]) +def test_verify_tpm_report_rejects_non_tpm_platform(wrong_platform: str) -> None: + from ca2a_runtime.tee.base import AttestationReport + from ca2a_verify.tpm import verify_tpm_report + + report = AttestationReport( + platform=wrong_platform, + measurement="sha256:" + "11" * 32, + public_key="aa" * 32, + nonce="fresh", + raw_evidence=build_attest(qualifying_data=b"n", pcr_digest=b"\x11" * 32), + quote_signature=b"\x00" * 64, + attestation_key_chain_pem=b"-----BEGIN CERTIFICATE-----\n", + ) + with pytest.raises(AttestationFailed, match="expected a 'tpm' attestation report"): + verify_tpm_report(report, "fresh", trusted_roots_pem=b"-----BEGIN CERTIFICATE-----\n")