Skip to content
Open
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
19 changes: 17 additions & 2 deletions src/ca2a_runtime/tee/tpm.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"


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


Expand Down Expand Up @@ -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
Expand Down
5 changes: 5 additions & 0 deletions src/ca2a_verify/tpm.py
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down
31 changes: 31 additions & 0 deletions tests/unit/test_tpm.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Loading