diff --git a/CHANGELOG.md b/CHANGELOG.md index f12c178..cc1ad55 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -73,6 +73,35 @@ by deletion. Closing that gap needs a signed, versioned CRL snapshot/digest mechanism, which is not yet implemented. +- **[SECURITY][SDK]** `verify_attestation_chain` no longer treats the + `azure-cvm-sev-snp` platform label -- or a caller-supplied boolean -- as + proof that the manifest binding was verified. REPORT_DATA on Azure is + `sha256(runtime_data)`, never the manifest hash, so it cannot be checked + directly the way it is on bare-metal SEV-SNP; Azure's real binding is a + vTPM AK-signed quote over a PCR derived from the manifest hash, chained to + REPORT_DATA via the runtime data. An earlier revision of this fix took an + `azure_manifest_binding_verified` parameter and trusted whatever the + caller passed in -- which meant any caller (or future code) could get + `passed=True` by simply passing `True`, with no evidence ever checked. + That parameter is gone. `verify_attestation_chain` now establishes the + composite chain itself (PCR-in-quote, AK signature, AK identity in + runtime_data, runtime_data->REPORT_DATA binding -- see the new + `agent_manifest._azure_verify.verify_azure_manifest_binding`, also used + by `AzureCVMProvider.verify_manifest_in_report` so there is exactly one + implementation of this security property) directly from evidence carried + on the report (`quote_msg`, `quote_sig`, `ak_pub_pem`, `runtime_data_hex` + in `report.raw`, and the SNP report bytes). `report_data_matched` is a + definite `True`/`False` for Azure, the same as every other platform. A + `platform` value selects which verification procedure applies; it is + never itself evidence that the procedure ran, and neither is any + caller-supplied flag -- there is no longer one to supply. + +- **[SECURITY][SDK]** `verify_attestation_chain` platform dispatch is now an + explicit allow-list (`amd-sev-snp`, `azure-cvm-sev-snp`, `intel-tdx`, + `tpm`, `aws-nitro`) instead of a catch-all `else` that routed any + unrecognized platform label through the SNP verifier. Unsupported labels + now report `NOT_IMPLEMENTED` and cannot pass. Closes #363. + - **[SDK]** `_check_manifest_binding()` no longer masks a malformed `artifacts` or `artifacts.policy_bundle` as merely absent. Truthy non-objects (`"str"`, `[1]`, `True`) already correctly reported diff --git a/python/src/agent_manifest/_attestation.py b/python/src/agent_manifest/_attestation.py index 3815937..73637c9 100644 --- a/python/src/agent_manifest/_attestation.py +++ b/python/src/agent_manifest/_attestation.py @@ -16,8 +16,17 @@ manifest hash. Implemented below. NOTE: this applies to the *direct* SNP model where the guest controls ``REPORT_DATA``. On Azure confidential VMs the guest does not control ``REPORT_DATA`` (the paravisor binds the vTPM AK - there); manifest binding on Azure is via the vTPM quote produced by - ``AzureCVMProvider``, not this field. + there); manifest binding on Azure is instead a vTPM AK-signed quote over a + PCR derived from the manifest hash, chained to REPORT_DATA via the runtime + data. This function establishes that chain itself, by calling + :func:`._azure_verify.verify_azure_manifest_binding` against evidence + carried on the report (``quote_msg``, ``quote_sig``, ``ak_pub_pem``, + ``runtime_data_hex`` in ``report.raw``, and the SNP report bytes). A + ``platform`` value only selects which of the above applies; it is never + itself evidence that any of them ran, and neither is any boolean a caller + might supply -- this function does not accept one. The only way Azure's + binding step can report ``True`` is for this function to have verified + the cryptographic chain itself. :func:`verify_attestation_chain` **fails closed**: ``passed`` is ``True`` only when the hardware signature is ``VERIFIED``, the manifest-hash binding matches, @@ -29,10 +38,20 @@ from __future__ import annotations import hmac +import re from dataclasses import dataclass, field from enum import Enum from typing import Any, Optional +# `expected_manifest_hash` must be exactly the literal "sha256:" prefix +# followed by 64 hex characters -- see the call site below (the non-Azure +# REPORT_DATA binding step) for why a bare `split(":", 1)` is not +# sufficient: it would accept "md5:<64 hex>" or "foo:<64 hex>" just as +# readily as "sha256:<64 hex>", leaving the hash algorithm ambiguous and +# caller-controlled at the exact point this value is compared against the +# report's bound field. +_MANIFEST_HASH_RE = re.compile(r"sha256:[0-9a-fA-F]{64}") + class SignatureStatus(str, Enum): """Outcome of the hardware signature / quote-chain check.""" @@ -52,6 +71,15 @@ class ChainVerificationResult: accepted (or not requested), and the manifest-hash binding matched. Until the signature backends land (#204), ``passed`` is always ``False`` and ``reasons`` explains why. + + For ``"azure-cvm-sev-snp"`` reports, ``report_data_matched`` reflects the + outcome of :func:`._azure_verify.verify_azure_manifest_binding`, run + directly by :func:`verify_attestation_chain` against evidence on the + report -- never a caller-supplied flag. It is a definite ``True``/ + ``False`` here, the same as every other platform: ``True`` only when the + full composite chain (PCR-in-quote, AK signature, AK identity, runtime + data -> REPORT_DATA binding) verified; ``False`` for any missing, + malformed, or mismatched evidence, including simply having none at all. """ passed: bool @@ -204,6 +232,7 @@ def verify_attestation_chain( tpm_trusted_roots_pem: Optional[bytes] = None, expected_qualifying_data: Optional[bytes] = None, expected_pcr_digest: Optional[bytes] = None, + azure_expected_pcr_index: int = 16, ) -> ChainVerificationResult: """Verify a boot-time ``AttestationReport`` against expected values. @@ -224,27 +253,121 @@ def verify_attestation_chain( cert_chain_pem: The AMD KDS ``cert_chain`` blob (ASK then ARK, PEM). trusted_ark_der: Optional pinned AMD root (ARK) certificate. When given, the chain's ARK public key must match it. + azure_expected_pcr_index: For ``"azure-cvm-sev-snp"`` reports, the PCR + index the manifest hash was extended into -- i.e. the value the + caller's ``AzureCVMProvider`` was actually configured with (its + ``pcr_index``). Defaults to 16, ``AzureCVMProvider``'s own + default. This is verifier configuration the caller supplies, not + something inferred from a self-reported field on the report + (``report.raw`` is not signed by anything). Returns: A :class:`ChainVerificationResult`. ``passed`` is ``True`` only when the hardware signature is ``VERIFIED``, the manifest-hash binding matches, and the measurement is accepted (or no allow-list was requested). Without VCEK material the signature step is not performed and the result - cannot pass, because an unverified report proves nothing. + cannot pass, because an unverified report proves nothing. An + unrecognized ``report.platform`` value also cannot pass: the signature + step is reported as ``NOT_IMPLEMENTED`` rather than falling through to + a verifier for a different profile. For ``"azure-cvm-sev-snp"`` + reports, the manifest-hash binding step is established by this + function itself (:func:`._azure_verify.verify_azure_manifest_binding` + against evidence carried on the report) -- there is no caller-supplied + flag that can substitute for that check. A ``platform`` value only + selects which verification procedure applies; it is never itself + evidence that the procedure ran. """ reasons: list[str] = [] + platform = getattr(report, "platform", "") or "" # Step 3: manifest-hash binding (software-checkable). - expected_digest = expected_manifest_hash.split(":", 1)[-1].lower() - actual_hex = _report_data_hex(report) - if actual_hex is None: + # + # Does not apply on Azure via REPORT_DATA: the guest never controls that + # field there (the paravisor sets it to sha256(runtime_data) to bind the + # vTPM AK, not the manifest hash). Azure's real binding is a vTPM + # AK-signed quote over a PCR derived from the manifest hash, chained to + # REPORT_DATA via the runtime data. This step must never be set True from + # the platform label alone, and must never be set True from a + # caller-supplied flag either -- a `platform` value says which procedure + # applies, it is not evidence that the procedure ran, and neither is + # someone's say-so. The only way to establish it is to actually run the + # composite check, here, against evidence carried on the report itself. + azure_paravisor = platform == "azure-cvm-sev-snp" + if azure_paravisor: + from ._azure_verify import verify_azure_manifest_binding + + # report is typed Any here -- a caller-forged or malformed report + # object can have any truthy value on .raw (a str, int, list), not + # just the documented dict. `getattr(..., {}) or {}` only guards the + # falsy cases (missing attribute, None, {}, ""); a truthy non-dict + # still reaches `.get()` below and raises AttributeError, breaking + # this function's own fail-closed contract (see the measurement + # check further down, which already guards this the same way). + raw_azure_candidate = getattr(report, "raw", {}) or {} + raw_azure = raw_azure_candidate if isinstance(raw_azure_candidate, dict) else {} + report_data_matched = verify_azure_manifest_binding( + expected_manifest_hash=expected_manifest_hash, + expected_pcr_index=azure_expected_pcr_index, + quote_msg_b64=raw_azure.get("quote_msg"), + quote_sig_b64=raw_azure.get("quote_sig"), + ak_pub_pem=raw_azure.get("ak_pub_pem"), + runtime_data_hex=raw_azure.get("runtime_data_hex"), + # Only fall back to report.quote when the caller didn't supply + # snp_report_bytes at all (None) -- not merely "falsy". `or` + # here would treat an explicit `snp_report_bytes=b""` (or any + # other falsy-but-supplied value) the same as "not given", + # silently discarding the caller's supplied evidence and + # substituting self-reported report.quote instead. In a + # security-sensitive verifier, supplied (even if invalid) + # evidence must be rejected on its own terms -- + # verify_azure_manifest_binding's own type/emptiness guard + # already fails closed on b"" -- never silently replaced with a + # different source. + snp_report_bytes=( + snp_report_bytes + if snp_report_bytes is not None + else getattr(report, "quote", None) + ), + ) + if report_data_matched: + reasons.append( + "Azure manifest binding verified: PCR value is one extension " + "of the manifest hash, the AK-signed TPM quote covers that " + "PCR, the AK is the one runtime_data names, and runtime_data " + "is bound into this report's REPORT_DATA -- all checked " + "directly by verify_attestation_chain, not accepted from a " + "caller-supplied flag" + ) + else: + reasons.append( + "Azure manifest binding not established: the composite " + "check (PCR value in the AK-signed quote, AK signature, AK " + "identity in runtime_data, runtime_data->REPORT_DATA " + "binding) did not fully verify against the evidence on this " + "report (quote_msg/quote_sig/ak_pub_pem/runtime_data_hex in " + "report.raw and the SNP report bytes)" + ) + elif not isinstance(expected_manifest_hash, str) or not _MANIFEST_HASH_RE.fullmatch( + expected_manifest_hash + ): + # Fail closed on a malformed/wrong-algorithm expected_manifest_hash + # instead of silently treating whatever follows a colon as the + # SHA-256 digest (see _MANIFEST_HASH_RE above). report_data_matched = False - reasons.append("report has no 'report_data' field to check the manifest binding against") + reasons.append( + "expected_manifest_hash is not in the required 'sha256:<64 hex>' form" + ) else: - # The first 32 bytes (64 hex chars) of REPORT_DATA carry the digest. - report_data_matched = hmac.compare_digest(actual_hex[:64].lower(), expected_digest) - if not report_data_matched: - reasons.append("manifest hash does not match the report_data binding") + expected_digest = expected_manifest_hash[len("sha256:"):].lower() + actual_hex = _report_data_hex(report) + if actual_hex is None: + report_data_matched = False + reasons.append("report has no 'report_data' field to check the manifest binding against") + else: + # The first 32 bytes (64 hex chars) of REPORT_DATA carry the digest. + report_data_matched = hmac.compare_digest(actual_hex[:64].lower(), expected_digest) + if not report_data_matched: + reasons.append("manifest hash does not match the report_data binding") # Step 2: launch-measurement allow-list (software-checkable, optional). measurement_matched: Optional[bool] @@ -262,11 +385,13 @@ def verify_attestation_chain( reasons.append("launch measurement is not in the supplied allow-list") # Step 1: hardware signature / quote chain, dispatched by platform. - # AMD SEV-SNP verifies the report signature + VCEK<-ASK<-ARK chain (needs the - # VCEK material). Intel TDX verifies the self-contained DCAP quote + PCK chain - # to the pinned Intel SGX Root CA. Either way, without a verifiable signature - # the result cannot pass. - platform = getattr(report, "platform", "") or "" + # AMD SEV-SNP (bare-metal and Azure's paravisor variant, which carries a + # real SNP report too) verifies the report signature + VCEK<-ASK<-ARK + # chain (needs the VCEK material). Intel TDX verifies the self-contained + # DCAP quote + PCK chain to the pinned Intel SGX Root CA. TPM/AWS Nitro + # verify an AK-signed quote. Dispatch is an explicit allow-list, not a + # catch-all: an unrecognized platform label must fail closed rather than + # silently inherit a verifier meant for a different profile. if platform == "intel-tdx": signature = _verify_tdx_signature_step(report, reasons, trusted_tdx_root_pem) elif platform in ("tpm", "aws-nitro"): @@ -279,7 +404,7 @@ def verify_attestation_chain( expected_pcr_digest, reasons, ) - else: + elif platform in ("amd-sev-snp", "azure-cvm-sev-snp"): signature = _verify_snp_signature_step( report, snp_report_bytes, @@ -288,8 +413,11 @@ def verify_attestation_chain( trusted_ark_der, reasons, ) + else: + signature = SignatureStatus.NOT_IMPLEMENTED + reasons.append(f"platform {platform!r} is not a supported attestation profile") - passed = ( + passed = bool( signature == SignatureStatus.VERIFIED and report_data_matched and measurement_matched is not False diff --git a/python/src/agent_manifest/_azure_verify.py b/python/src/agent_manifest/_azure_verify.py new file mode 100644 index 0000000..2e03a2e --- /dev/null +++ b/python/src/agent_manifest/_azure_verify.py @@ -0,0 +1,366 @@ +"""Composite Azure vTPM/SNP manifest-binding verification. + +On Azure confidential VMs the guest does not control SNP's ``REPORT_DATA`` +(the Hyper-V paravisor binds the vTPM AK there instead), so the manifest hash +cannot be checked directly against ``REPORT_DATA`` the way it is on +bare-metal SEV-SNP. Azure's real binding is a four-link chain: + + 1. The manifest PCR value is exactly one extension of the manifest hash, + and that value is the PCR digest carried *inside* an AK-signed TPM quote + (not a free-text ``pcr_read`` string -- that is unauthenticated). + 2. The quote signature verifies under the claimed AK public key. + 3. That AK public key is the exact key the runtime data names as + ``HCLAkPub`` -- i.e. this AK, not some other key. + 4. The runtime data hashes into ``REPORT_DATA`` of the *signed* SNP report + bytes -- i.e. this runtime data (and therefore this AK) is bound into + this exact report. The report's own VCEK<-ASK<-ARK signature chain is a + separate, mandatory check performed by the caller; together the two + checks tie the AK all the way to silicon. + +:func:`verify_azure_manifest_binding` is the single place this chain is +established. Nothing else -- not a platform label, not a caller-supplied +boolean, not a self-reported flag on the report -- may substitute for +actually running it. Both :class:`AzureCVMProvider.verify_manifest_in_report` +and :func:`agent_manifest._attestation.verify_attestation_chain` call this +same function so there is exactly one implementation of the security +property, not two that can drift apart. +""" + +from __future__ import annotations + +import base64 +import binascii +import hashlib +import hmac +import json +import re +import struct +from typing import Any, Optional + + +# RFC 7515 base64url: unpadded, alphabet A-Za-z0-9-_ only. Python's +# ``base64.urlsafe_b64decode`` (and ``b64decode`` without ``validate=True``) +# silently *discards* any character outside the standard base64 alphabet +# before decoding, rather than rejecting it -- so a JWK member like +# ``"!!!!"`` decodes to the same bytes as ``""`` with +# no error. That lets attacker-corrupted-but-cryptographically-committed +# JWK material decode "successfully" to whatever bytes survive the silent +# filtering, defeating the "fail closed on malformed input" contract these +# helpers document. Reject anything outside the unpadded URL-safe alphabet +# up front, then decode strictly. +# +# `+` (one-or-more), not `*` (zero-or-more): this is only ever used to +# decode a JWK's ``n``/``e`` member, and an RSA public key's modulus and +# exponent are never legitimately empty. With `*`, ``fullmatch("")`` +# succeeds, so ``{"n": "", "e": ""}`` -- a malformed JWK -- would sail +# through the alphabet check and decode to ``b""``, reaching +# ``ak_public_numbers_from_runtime_data`` as ``("", 0)`` instead of the +# documented ``None`` fail-closed return. That empty key can never match a +# real AK's ``(n, e)``, so this does not by itself let anything +# authenticate -- but the validation contract is "reject malformed JWK +# input", and an empty string is malformed input, not a valid (if unusual) +# key. +_B64URL_ALPHABET_RE = re.compile(r"[A-Za-z0-9_-]+") + + +# `expected_manifest_hash` must be exactly this shape: the literal "sha256:" +# prefix followed by 64 hex characters -- nothing before, nothing after, and +# no other algorithm name accepted. See the call site in +# ``verify_azure_manifest_binding`` for why a bare ``split(":", 1)`` is not +# sufficient here. +_MANIFEST_HASH_RE = re.compile(r"sha256:[0-9a-fA-F]{64}") + + +def _strict_b64url_decode(value: str) -> bytes: + """Decode a JWK base64url member, rejecting anything but the exact, + unpadded, URL-safe alphabet. + + Raises ``binascii.Error`` (caught by callers, which fail closed) if + ``value`` contains characters outside ``A-Za-z0-9-_`` -- including + standard-base64 ``+``/``/`` (and their padding ``=``), any illegal + prefix/suffix such as stray punctuation, and (via ``fullmatch``, not a + ``$``-anchored pattern) a trailing newline, which Python's ``$`` would + otherwise let slip through one character before the end of the string -- + or if ``value`` is the empty string, since a JWK's ``n``/``e`` are never + legitimately empty and ``fullmatch`` on a zero-or-more-style pattern + would otherwise accept "". This is stricter than + ``base64.urlsafe_b64decode``, which silently drops out-of-alphabet + characters -- including embedded/trailing newlines -- instead of + raising. + """ + if not isinstance(value, str) or not _B64URL_ALPHABET_RE.fullmatch(value): + raise binascii.Error("invalid base64url alphabet") + padded = value + "=" * ((4 - len(value) % 4) % 4) + return base64.urlsafe_b64decode(padded) + + +def _find_hcl_ak_jwk(runtime_data: bytes) -> Optional[dict[str, Any]]: + """Return the ``HCLAkPub`` JWK dict from Azure HCL runtime-data JSON, or ``None``. + + Fails closed (returns ``None``, never raises) on any malformed input: + bad JSON, non-UTF-8 bytes, a top-level value that isn't an object, a + ``"keys"`` value that isn't a list (e.g. ``{"keys": 1}``, which would + otherwise blow up trying to iterate an int), or list entries that aren't + key-shaped objects. + """ + try: + parsed = json.loads(runtime_data) + except (json.JSONDecodeError, UnicodeDecodeError, TypeError): + return None + if not isinstance(parsed, dict): + return None + keys = parsed.get("keys", []) + if not isinstance(keys, list): + return None + return next((k for k in keys if isinstance(k, dict) and k.get("kid") == "HCLAkPub"), None) + + +def ak_modulus_hex_from_runtime_data(runtime_data: bytes) -> Optional[str]: + """Extract the ``HCLAkPub`` RSA modulus (hex) from Azure HCL runtime-data JSON. + + Returns ``None`` on any malformed input (bad JSON, missing key, bad + base64) rather than raising -- callers treat that as "cannot establish + the binding", i.e. fail closed. + + An RSA public key is the pair ``(n, e)``, not ``n`` alone: this helper is + kept for callers (e.g. matching a vTPM persistent handle by modulus) that + only need the modulus, but a security decision that means to authenticate + the *exact* key must use :func:`ak_public_numbers_from_runtime_data` + instead, which also checks the exponent. + + """ + ak = _find_hcl_ak_jwk(runtime_data) + if ak is None or "n" not in ak: + return None + try: + return _strict_b64url_decode(ak["n"]).hex() + except (binascii.Error, ValueError, TypeError): + return None + + +def ak_public_numbers_from_runtime_data(runtime_data: bytes) -> Optional[tuple[str, int]]: + """Extract the ``HCLAkPub`` RSA public numbers ``(n, e)`` from runtime data. + + Returns ``(modulus_hex, exponent_int)``, or ``None`` on any malformed + input (bad JSON, missing/malformed ``n`` or ``e``, bad base64) -- never + raises. Both values must be checked to authenticate the exact key: a + modulus match alone does not establish ``e`` also matches, and a + different ``e`` describes a different key even with the same ``n``. + """ + ak = _find_hcl_ak_jwk(runtime_data) + if ak is None or "n" not in ak or "e" not in ak: + return None + try: + n_hex = _strict_b64url_decode(ak["n"]).hex() + e_int = int.from_bytes(_strict_b64url_decode(ak["e"]), "big") + except (binascii.Error, ValueError, TypeError): + return None + return n_hex, e_int + + +def verify_azure_manifest_binding( + *, + expected_manifest_hash: str, + expected_pcr_index: int, + quote_msg_b64: Optional[str], + quote_sig_b64: Optional[str], + ak_pub_pem: Optional[str], + runtime_data_hex: Optional[str], + snp_report_bytes: Optional[bytes], +) -> bool: + """Authenticate the composite Azure manifest-binding chain end to end. + + Verifies, purely from the evidence given -- never from a platform label, + a caller-supplied boolean, or any self-reported flag: + + 1. the manifest PCR value (one extension of ``expected_manifest_hash``) + equals the PCR digest inside the AK-signed TPM quote, *and* that + digest is carried under a selection of exactly one PCR bank + (SHA-256) and exactly one PCR index -- ``expected_pcr_index`` -- not + merely a digest value that happens to match while the quote was + actually signed over a different bank or PCR; + 2. ``quote_sig`` is a valid AK signature over ``quote_msg`` under + ``ak_pub_pem``; + 3. ``ak_pub_pem`` is the exact key the runtime data names as + ``HCLAkPub`` -- both the modulus *and* the exponent, since an RSA + public key is the pair ``(n, e)`` and a modulus-only match lets a + different exponent describe a different key; + 4. the runtime data hashes into ``REPORT_DATA`` of the signed SNP + report bytes on ``snp_report_bytes``. + + ``expected_pcr_index`` is required and must come from verifier + configuration (e.g. the ``AzureCVMProvider`` the caller actually + configured), never inferred from a self-reported field like + ``report.raw["pcr_index"]`` -- that field is not signed by anything and + an attacker who controls the report controls it too. + + Returns ``True`` only when all four hold. Any missing, malformed, or + mismatched input returns ``False`` -- this function never raises. + """ + # Truthiness alone is not a type check: a truthy wrong-type value (an + # int, list, dict -- all trivially producible from forged JSON evidence + # or a caller passing the wrong variable) passes `and`-chained + # truthiness checks but then blows up as TypeError/AttributeError inside + # base64.b64decode / bytes.fromhex / struct.unpack / str.split below, + # which only catch format errors, not type errors. Require the exact + # expected type for every top-level argument -- including + # ``expected_manifest_hash`` and ``expected_pcr_index``, which are just + # as reachable from a misconfigured caller as the report-derived fields + # are from a forged report -- before touching any of them, so a + # wrong-type argument fails closed here instead of raising deeper in the + # chain. ``expected_pcr_index`` in particular must be ``int``: it is + # later put into a ``frozenset`` for a PCR-index comparison, and an + # unhashable value there (a list or dict) raises ``TypeError`` on the + # ``frozenset`` construction itself, not on anything this function + # already wraps in a ``try``. ``bool`` is deliberately excluded even + # though it is an ``int`` subclass in Python: a boolean PCR index is + # always a caller bug, not a legitimate value. + if ( + not isinstance(expected_manifest_hash, str) + or not isinstance(expected_pcr_index, int) + or isinstance(expected_pcr_index, bool) + or not isinstance(quote_msg_b64, str) + or not isinstance(quote_sig_b64, str) + or not isinstance(ak_pub_pem, str) + or not isinstance(runtime_data_hex, str) + or not isinstance(snp_report_bytes, (bytes, bytearray)) + or not expected_manifest_hash + or not quote_msg_b64 + or not quote_sig_b64 + or not ak_pub_pem + or not runtime_data_hex + or not snp_report_bytes + ): + return False + + # Imports are local to avoid a module-load-time cycle: _snp_verify and + # _tpm_verify are heavier modules not needed unless Azure binding is + # actually being checked. + from ._snp_verify import SnpVerificationError, parse_snp_report, verify_runtime_data_binding + from ._tpm_verify import ( + TPM_ALG_SHA256, + TPM_GENERATED_VALUE, + TPM_ST_ATTEST_QUOTE, + TpmVerificationError, + parse_tpm_quote, + verify_ak_signature, + ) + try: + quote_msg = base64.b64decode(quote_msg_b64, validate=True) + quote_sig = base64.b64decode(quote_sig_b64, validate=True) + runtime_data = bytes.fromhex(runtime_data_hex) + except (binascii.Error, ValueError, TypeError): + # TypeError is defense-in-depth: the isinstance guard above already + # rules out non-str inputs reaching this point, but keeping TypeError + # here too means this stays fail-closed even if that guard is ever + # loosened or bypassed by a future edit. + return False + + # 1a. Expected PCR value: one extension of the manifest hash into a PCR + # that started at zero. + # + # `expected_manifest_hash` must be exactly `"sha256:" + 64 hex chars` -- + # not merely have *some* prefix before the first `:`. A naive + # `split(":", 1)[-1]` accepts "md5:<64 hex>" or "foo:<64 hex>" just as + # readily as "sha256:<64 hex>", silently treating whatever hex follows + # any colon as the SHA-256 digest. That leaves the hash algorithm + # ambiguous and caller-controlled at the exact point this value gets + # baked into the manifest-to-PCR security binding. Require the literal + # "sha256:" prefix explicitly, not just "a colon happens to be present". + if not _MANIFEST_HASH_RE.fullmatch(expected_manifest_hash): + return False + digest = expected_manifest_hash[len("sha256:"):].lower() + try: + expected_pcr_value = hashlib.sha256(bytes(32) + bytes.fromhex(digest)).digest() + except (AttributeError, TypeError, ValueError): + # AttributeError/TypeError are defense-in-depth: the isinstance + # guard above already rules out a non-str expected_manifest_hash + # reaching this point, and the fullmatch above already guarantees + # valid hex, but keeping this try here too means this stays + # fail-closed even if either guard is ever loosened or bypassed by a + # future edit. + return False + # tpm2_quote's PCR digest, for a single sha256 bank with a single PCR + # selected, is sha256 of that PCR's raw (post-extend) value. + expected_quote_pcr_digest = hashlib.sha256(expected_pcr_value).digest() + + try: + quote = parse_tpm_quote(quote_msg) + except TpmVerificationError: + return False + if quote.magic != TPM_GENERATED_VALUE or quote.attest_type != TPM_ST_ATTEST_QUOTE: + return False + # The boot-time quote is produced with a fixed all-zero qualifying value; + # a fresh-nonce runtime quote (attest_runtime_state) must not be + # replayable here as boot proof. + if not hmac.compare_digest(quote.qualifying_data, bytes(16)): + return False + # 1b. That PCR digest is the one actually inside the AK-signed quote. + if not hmac.compare_digest(quote.pcr_digest, expected_quote_pcr_digest): + return False + + # 1c. The digest must be a digest *of the expected bank and PCR index*, + # not merely a byte string that matches while the signed selection names + # a different bank or PCR -- pcr_digest alone doesn't say which PCR(s) it + # was computed over. Require exactly one selection: exactly one bank + # (SHA-256) with exactly one PCR selected, and that PCR must be the one + # this verifier is configured to check. + if len(quote.pcr_selections) != 1: + return False + selection = quote.pcr_selections[0] + if selection.hash_alg != TPM_ALG_SHA256: + return False + try: + expected_indices = frozenset({expected_pcr_index}) + except TypeError: + # Defense-in-depth: the isinstance guard above already rules out an + # unhashable expected_pcr_index (list, dict) reaching this point -- + # frozenset() would otherwise raise TypeError constructing a set + # from an unhashable member -- but keeping this here too means this + # stays fail-closed even if that guard is ever loosened or bypassed + # by a future edit. + return False + if selection.indices() != expected_indices: + return False + + # 2. quote_sig is a valid AK signature over quote_msg under ak_pub_pem. + try: + from cryptography.hazmat.primitives.asymmetric import rsa + from cryptography.hazmat.primitives.serialization import load_pem_public_key + + ak_key = load_pem_public_key(ak_pub_pem.encode()) + except Exception: + return False + try: + if not verify_ak_signature(ak_key, quote.raw, quote_sig): + return False + except TpmVerificationError: + return False + + # 3. ak_pub_pem is the exact key runtime_data describes as HCLAkPub -- + # both the modulus and the exponent. An RSA public key is the pair + # (n, e); comparing n alone would let runtime_data claim a different + # exponent (a different key) while still "matching" on modulus. + if not isinstance(ak_key, rsa.RSAPublicKey): + return False + numbers = ak_key.public_numbers() + ak_modulus_hex = numbers.n.to_bytes((numbers.n.bit_length() + 7) // 8, "big").hex() + runtime_ak_numbers = ak_public_numbers_from_runtime_data(runtime_data) + if runtime_ak_numbers is None: + return False + runtime_ak_modulus_hex, runtime_ak_exponent = runtime_ak_numbers + if not hmac.compare_digest(ak_modulus_hex.lower(), runtime_ak_modulus_hex.lower()): + return False + if numbers.e != runtime_ak_exponent: + return False + + # 4. runtime_data is bound (via REPORT_DATA) into the signed SNP report + # bytes actually supplied. + try: + parsed_snp = parse_snp_report(snp_report_bytes) + except (SnpVerificationError, TypeError, struct.error): + return False + if not verify_runtime_data_binding(parsed_snp, runtime_data): + return False + + return True diff --git a/python/src/agent_manifest/_delegation.py b/python/src/agent_manifest/_delegation.py index 4070458..ff57827 100644 --- a/python/src/agent_manifest/_delegation.py +++ b/python/src/agent_manifest/_delegation.py @@ -388,10 +388,9 @@ def _verify_hops( manifest_id=manifest_id, ) - import base64 + from ._signing import _b64url_decode sig = hop["delegation_signature"] - pad = 4 - len(sig) % 4 - sig_bytes = base64.urlsafe_b64decode(sig + ("=" * pad if pad != 4 else "")) + sig_bytes = _b64url_decode(sig) verifier = Ed25519Verifier(pub_bytes) verifier._pub.verify(sig_bytes, pre) # raises InvalidSignature on failure @@ -634,8 +633,6 @@ def verify_hitl_approval( ValueError: If required fields are missing, malformed, or the approval has expired. """ - import base64 - import re # Establish the shapes this function reads before interpreting them, so a # malformed approval always produces the documented ValueError rather than @@ -712,24 +709,18 @@ def verify_hitl_approval( approver_id=approver_id, approval_method=approval.get("approval_method"), ) - # base64.b64decode(..., altchars=b"-_", validate=True) translates '-'/'_' - # to '+'/'/' *before* validating, so it also accepts the standard base64 - # alphabet mixed in as-is: swapping every '-' for '+' and '_' for '/' in - # an otherwise-valid signature decodes to the identical bytes and passes. - # Whitelist the URL-safe alphabet explicitly so no other representation - # of the same bytes is accepted. - if not re.fullmatch(r"[A-Za-z0-9_-]*", sig): - raise ValueError( - "HITL approval.approval_signature is not valid base64url: " - "contains characters outside the URL-safe alphabet" - ) + + # _b64url_decode() (shared with delegation-hop verification) already + # rejects the standard base64 alphabet and any non-URL-safe characters + # (CRYPTO-006), so reuse it here instead of re-implementing the same + # check inline. Wrapped to keep the field-qualified error message. + from ._signing import _b64url_decode + try: - pad = 4 - len(sig) % 4 - sig_bytes = base64.b64decode( - sig + ("=" * pad if pad != 4 else ""), altchars=b"-_", validate=True - ) + sig_bytes = _b64url_decode(sig) except ValueError as e: raise ValueError( f"HITL approval.approval_signature is not valid base64url: {e}" ) from e + Ed25519Verifier(approver_public_key)._pub.verify(sig_bytes, pre) diff --git a/python/src/agent_manifest/_hw_providers.py b/python/src/agent_manifest/_hw_providers.py index 253daee..3439f80 100644 --- a/python/src/agent_manifest/_hw_providers.py +++ b/python/src/agent_manifest/_hw_providers.py @@ -347,17 +347,14 @@ def _find_ak_handle(self, modulus_hex: str) -> str: ) def _ak_modulus_hex(self, runtime_data: bytes) -> str: - import base64 - import json + from ._azure_verify import ak_modulus_hex_from_runtime_data - keys = json.loads(runtime_data).get("keys", []) - ak = next((k for k in keys if k.get("kid") == "HCLAkPub"), None) - if ak is None: + modulus_hex = ak_modulus_hex_from_runtime_data(runtime_data) + if modulus_hex is None: raise AttestationUnavailableError( "runtime data does not carry the HCLAkPub attestation key." ) - n_b64 = ak["n"] + "=" * ((4 - len(ak["n"]) % 4) % 4) - return base64.urlsafe_b64decode(n_b64).hex() + return modulus_hex def extend_manifest_hash(self, manifest_json: dict[str, Any]) -> None: pre = self.manifest_pre_image(manifest_json) @@ -415,7 +412,11 @@ def _quote(self, nonce_hex: str) -> dict[str, str]: except OSError: pass blobs["snp_report"] = snp_raw.hex() - blobs["runtime_data"] = runtime.decode("utf-8", "replace") + # Exact bytes, not a decoded string: verify_manifest_in_report re-hashes + # this to recheck the REPORT_DATA binding, so a lossy decode/re-encode + # round trip (e.g. "replace" on non-UTF-8 bytes) must not be able to + # change what gets hashed. + blobs["runtime_data_hex"] = runtime.hex() blobs["measurement"] = rep.measurement.hex() blobs["report_data"] = rep.report_data.hex() return blobs @@ -436,37 +437,66 @@ def get_attestation_report(self) -> AttestationReport: raw={ "report_data": blobs["report_data"], "measurement": blobs["measurement"], - "runtime_data_binding_verified": True, + "runtime_data_hex": blobs["runtime_data_hex"], "ak_pub_pem": blobs["ak_pub_pem"], "pcr_index": self._pcr, "pcr_read": pcr_value, "quote_msg": blobs["quote_msg"], "quote_sig": blobs["quote_sig"], "quote_pcrs": blobs["quote_pcrs"], - "vcek_cert_chain_verified": False, + # Not "vcek_cert_chain_verified": True/False like SEVSNPProvider + # above -- this method never verifies the VCEK chain itself + # (verify_attestation_chain does that, given vcek_cert_der / + # cert_chain_pem by the caller). We deliberately don't invent + # a value for it here. }, ) def verify_manifest_in_report( self, report: AttestationReport, manifest_json: dict[str, Any] ) -> bool: - """Confirm the manifest PCR equals a single extension of the manifest hash. - - A resettable PCR starts at 0x00*32; after one extension its value is - ``sha256(0x00*32 || manifest_digest)``. Matching that proves the - manifest hash (and nothing else) was measured into the PCR. + """Authenticate the composite Azure manifest-binding chain end to end. + + ``pcr_read`` alone (the old check here) is a plain string on + ``report.raw`` that anyone constructing an ``AttestationReport`` can + set to whatever they like; it is not signed by anything and proves + nothing on its own. This method is the boundary that must actually + establish the chain -- see :func:`._azure_verify.verify_azure_manifest_binding` + for the four links it checks (PCR-in-quote, AK signature, AK identity + in runtime data, runtime-data->REPORT_DATA binding). That same + function is also called directly by + :func:`agent_manifest._attestation.verify_attestation_chain`, so + there is exactly one implementation of this security property -- + this method does not hand out a boolean for a caller to relay + elsewhere as if it were itself trustworthy evidence. + + Any missing/malformed/mismatched field fails closed (``False``), + never raises. """ - import hmac as _hmac - - digest = self.manifest_hash_value(manifest_json).split(":", 1)[-1] - expected = hashlib.sha256(bytes(32) + bytes.fromhex(digest)).hexdigest() - pcr_read = (report.raw or {}).get("pcr_read", "") - got = "" - for line in pcr_read.splitlines(): - s = line.strip() - if s.startswith(f"{self._pcr}:"): - got = s.split(":", 1)[1].strip().lower().removeprefix("0x") - return bool(got) and _hmac.compare_digest(got, expected) + from ._azure_verify import verify_azure_manifest_binding + + # report.raw is typed dict[str, Any], but this method receives an + # AttestationReport from a caller who can construct one with any + # value at all (dataclasses do not enforce field types at runtime) -- + # a truthy non-dict raw (a str, int, or list) is just as reachable as + # a forged report's individual fields. `raw or {}` only guards the + # falsy cases (None, {}, ""); a truthy non-dict still reaches + # `.get()` below and raises AttributeError, breaking this method's + # own fail-closed contract. Require raw to actually be a dict first. + raw = report.raw if isinstance(report.raw, dict) else {} + return verify_azure_manifest_binding( + expected_manifest_hash=self.manifest_hash_value(manifest_json), + # This provider's own configured PCR index -- verifier + # configuration -- not raw.get("pcr_index"), which is a + # self-reported field on the report that anyone constructing an + # AttestationReport can set to whatever they like. + expected_pcr_index=self._pcr, + quote_msg_b64=raw.get("quote_msg"), + quote_sig_b64=raw.get("quote_sig"), + ak_pub_pem=raw.get("ak_pub_pem"), + runtime_data_hex=raw.get("runtime_data_hex"), + snp_report_bytes=getattr(report, "quote", None), + ) def attest_runtime_state( self, diff --git a/python/src/agent_manifest/_providers.py b/python/src/agent_manifest/_providers.py index a367d9f..50daaf2 100644 --- a/python/src/agent_manifest/_providers.py +++ b/python/src/agent_manifest/_providers.py @@ -40,7 +40,7 @@ class AttestationUnavailableError(RuntimeError): class AttestationReport: """Portable attestation report returned by all providers.""" - platform: str # "tpm" | "sev-snp" | "tdx" | "opaque" + platform: str # "amd-sev-snp" | "azure-cvm-sev-snp" | "intel-tdx" | "tpm" | "aws-nitro" | "opaque" manifest_hash: str # "sha256:<64-hex>" — hash of the signed manifest pcr_values: dict[str, str] = field(default_factory=dict) # {"PCR15": "sha256:..."} quote: Optional[bytes] = None # raw platform quote/report blob diff --git a/python/src/agent_manifest/_revocation.py b/python/src/agent_manifest/_revocation.py index 769c4cd..aa3df6a 100644 --- a/python/src/agent_manifest/_revocation.py +++ b/python/src/agent_manifest/_revocation.py @@ -101,9 +101,8 @@ def verify_revocation_signature( cryptography.exceptions.InvalidSignature: If verification fails or revocation_signature is absent/null (CRL-001). """ - import base64 from cryptography.exceptions import InvalidSignature - from ._signing import Ed25519Verifier as _Ed25519Verifier, _key_id + from ._signing import Ed25519Verifier as _Ed25519Verifier, _b64url_decode, _key_id # CRL-001: null/empty signature must raise InvalidSignature, not ValueError if not record.revocation_signature: @@ -125,8 +124,14 @@ def verify_revocation_signature( } pre_image = canonicalize(pre_image_obj) - pad = 4 - len(sig) % 4 - sig_bytes = base64.urlsafe_b64decode(sig + ("=" * pad if pad != 4 else "")) + try: + sig_bytes = _b64url_decode(sig) + except ValueError as e: + # Preserve this function's documented contract (only + # InvalidSignature is raised) even though _b64url_decode raises + # ValueError for malformed base64url -- a malformed signature + # encoding is itself an invalid signature. + raise InvalidSignature(f"revocation_signature is not valid base64url: {e}") from e if len(sig_bytes) != 64: raise InvalidSignature( f"Ed25519 signature must be 64 bytes, got {len(sig_bytes)}" diff --git a/python/src/agent_manifest/_signing.py b/python/src/agent_manifest/_signing.py index d9234b5..abc9b15 100644 --- a/python/src/agent_manifest/_signing.py +++ b/python/src/agent_manifest/_signing.py @@ -169,7 +169,15 @@ def _b64url_encode(data: bytes) -> str: return base64.urlsafe_b64encode(data).rstrip(b"=").decode() -_B64URL_RE = re.compile(r"^[A-Za-z0-9\-_]*$") +# NOTE: intentionally NOT anchored with `$` -- Python's `$` matches at +# end-of-string OR just before a single trailing '\n', which would let +# "\n" slip past this alphabet check even though '\n' isn't +# in the URL-safe alphabet, and base64.urlsafe_b64decode() would then +# silently drop that '\n' and decode it identically to the un-suffixed +# value -- defeating the CRYPTO-006 guard below. fullmatch() (no anchors) +# does not have this hole. +_B64URL_RE = re.compile(r"[A-Za-z0-9\-_]*") + def _signed_at_now() -> str: @@ -181,7 +189,7 @@ def _signed_at_now() -> str: def _b64url_decode(s: str) -> bytes: # CRYPTO-006: reject standard base64 (+/) - only URL-safe chars allowed - if not _B64URL_RE.match(s): + if not isinstance(s, str) or not _B64URL_RE.fullmatch(s): raise ValueError( "Invalid base64url: contains non-URL-safe characters (use - and _ not + and /)" ) diff --git a/python/src/agent_manifest/_tpm_verify.py b/python/src/agent_manifest/_tpm_verify.py index 37e0a06..7f58297 100644 --- a/python/src/agent_manifest/_tpm_verify.py +++ b/python/src/agent_manifest/_tpm_verify.py @@ -63,6 +63,9 @@ _ALG_RSAPSS = 0x0016 _ALG_ECDSA = 0x0018 +# TPM2_ALG_ID value for the SHA-256 PCR bank. +TPM_ALG_SHA256 = 0x000B + class TpmVerificationError(Exception): """Raised when a TPM quote or its certificate chain fails verification.""" @@ -100,6 +103,28 @@ class TpmAttest: raw: bytes +@dataclass(frozen=True) +class PcrSelection: + """One ``TPMS_PCR_SELECTION`` entry: a bank (hash alg) and its selected PCRs. + + ``pcr_select`` is the raw ``pcrSelect`` bitmap bytes (TCG bit order: byte + ``i`` bit ``j`` selects PCR ``i*8+j``). Use :meth:`indices` rather than + reading the bitmap by hand. + """ + + hash_alg: int + pcr_select: bytes + + def indices(self) -> frozenset[int]: + """Return the set of PCR indices selected by ``pcr_select``.""" + selected = set() + for byte_index, byte in enumerate(self.pcr_select): + for bit in range(8): + if byte & (1 << bit): + selected.add(byte_index * 8 + bit) + return frozenset(selected) + + @dataclass(frozen=True) class TpmQuote: """The parsed subset of a TPM 2.0 quote (TPMS_ATTEST) that is appraised.""" @@ -107,6 +132,7 @@ class TpmQuote: magic: int attest_type: int qualifying_data: bytes # extraData: the verifier's nonce + pcr_selections: tuple[PcrSelection, ...] # TPML_PCR_SELECTION: bank(s) + PCR(s) selected pcr_digest: bytes # the platform measurement raw: bytes @@ -231,21 +257,36 @@ def parse_tpm_quote(attest: bytes) -> TpmQuote: ) attested = common.attested_raw pos = 0 - # TPML_PCR_SELECTION + # TPML_PCR_SELECTION. Each TPMS_PCR_SELECTION is hashAlg (2 bytes) + + # sizeofSelect (1 byte) + pcrSelect (sizeofSelect bytes). The bank and the + # actual PCR bitmap are captured here -- not merely skipped over -- because + # pcr_digest alone does not say *which* PCR(s) it is a digest of; a quote + # signed over a different PCR (or bank) can carry the same digest bytes + # and a caller checking only pcr_digest would be unable to tell. + if pos + 4 > len(attested): raise TpmVerificationError("TPM quote truncated reading PCR selection count") count = int.from_bytes(attested[pos:pos + 4], "big") pos += 4 + selections = [] for _ in range(count): if pos + 3 > len(attested): raise TpmVerificationError("TPM quote truncated reading a PCR selection") + hash_alg = int.from_bytes(attested[pos:pos + 2], "big") size_of_select = attested[pos + 2] - pos += 3 + size_of_select + pos += 3 + if pos + size_of_select > len(attested): + raise TpmVerificationError("TPM quote truncated reading a PCR select bitmap") + pcr_select = bytes(attested[pos:pos + size_of_select]) + pos += size_of_select + selections.append(PcrSelection(hash_alg=hash_alg, pcr_select=pcr_select)) + pcr_digest, _pos = _read_2b(attested, pos) return TpmQuote( magic=common.magic, attest_type=common.attest_type, qualifying_data=common.qualifying_data, + pcr_selections=tuple(selections), pcr_digest=pcr_digest, raw=common.raw, ) @@ -363,44 +404,37 @@ def _verify_ak_chain( return chain[0] -def verify_tpm_quote( - attest: bytes, +def verify_ak_signature( + ak_public_key: object, + attest_raw: bytes, signature: bytes | ParsedSignature, - ak_chain_pem: bytes, - *, - trusted_roots_pem: bytes, - expected_qualifying_data: bytes | None = None, - expected_pcr_digest: bytes | None = None, - verification_time: datetime | None = None, ) -> bool: - """Fully verify a TPM 2.0 quote offline (all four steps, fail-closed). - - Args: - attest: the raw ``TPMS_ATTEST`` blob the TPM signed. - signature: either the legacy bare AK signature (DER ECDSA or RSA - PKCS#1 v1.5 over SHA-256), a parsed :class:`ParsedSignature`, or a - marshalled ``TPMT_SIGNATURE``. Envelopes select RSASSA, RSAPSS, or - ECDSA and SHA-256, SHA-384, or SHA-512 from their algorithm ids. - Legacy bare input is recognized from the AK's signature shape: - modulus-sized bytes for RSA and DER for ECDSA. Other byte input - must be a well-formed envelope. - ak_chain_pem: the AK certificate chain (PEM, leaf first). - trusted_roots_pem: the caller's trusted vendor EK/AK roots (PEM). - expected_qualifying_data: if given, the quote's ``extraData`` (nonce) - must equal it. - expected_pcr_digest: if given, the quote's PCR digest must equal it. - verification_time: UTC-aware time used to check AK chain certificate - validity periods (default: current UTC time). Primarily useful - for deterministic tests. - - Returns: - ``True`` only when the structure, AK chain, AK signature, and any - supplied bindings all check out. Returns ``False`` on a well-formed but - invalid signature or a binding mismatch. Raises - :class:`TpmVerificationError` on a malformed quote / broken chain, a - malformed signature, an unsupported algorithm, or if ``cryptography`` - is unavailable. + """Verify a quote's AK signature over ``attest_raw``, given a trusted AK key. + + This is exactly step 3 of :func:`verify_tpm_quote` (the signature check), + factored out so a caller who establishes trust in the AK by some route + other than an X.509 chain can still reuse the same signature-verification + code instead of re-implementing ``TPMT_SIGNATURE``/legacy-signature framing + and algorithm selection. For example, an Azure vTPM AK is not chained to a + CA at all: it is trusted because its public key is embedded in runtime + data that is itself bound (via ``REPORT_DATA``) into a VCEK-signed SNP + report, verified elsewhere. ``ak_public_key`` must already be an + ``ec.EllipticCurvePublicKey`` or ``rsa.RSAPublicKey``; establishing that + the key is the *right* one is the caller's responsibility. + + Legacy bare input is recognized from the AK's signature shape: modulus- + sized bytes for RSA and DER for ECDSA (not the ``sigAlg`` prefix bytes, + which are untrusted input -- an attacker could otherwise flip a well- + formed envelope's leading bytes to look like a legacy bare signature and + move it into the wrong verification lane). Other byte input must be a + well-formed ``TPMT_SIGNATURE`` envelope. + + Returns ``True`` only when the signature verifies. Returns ``False`` on a + well-formed but invalid signature or key/algorithm mismatch. Raises + :class:`TpmVerificationError` on a malformed ``TPMT_SIGNATURE`` or an + unsupported algorithm. """ + try: from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives import hashes @@ -410,37 +444,16 @@ def verify_tpm_quote( "TPM quote verification requires the 'cryptography' package" ) from e - quote = parse_tpm_quote(attest) - - # Step 1: structural — this must be a TPM-generated quote. - if quote.magic != TPM_GENERATED_VALUE: - raise TpmVerificationError( - f"TPMS_ATTEST magic is not TPM_GENERATED (magic={quote.magic:#x})" - ) - if quote.attest_type != TPM_ST_ATTEST_QUOTE: - raise TpmVerificationError( - f"attestation is not a quote (type={quote.attest_type:#x})" - ) - - # Step 2: AK certificate chain up to a pinned trusted root. - ak = _verify_ak_chain(ak_chain_pem, trusted_roots_pem, verification_time=verification_time) - ak_key = ak.public_key() # type: ignore[attr-defined] - - # Step 3: AK signature over the TPMS_ATTEST blob. Use quote.raw rather than - # the argument: when the caller passes a size-prefixed TPM2B_ATTEST, the TPM - # signed the inner structure, so verifying over the outer bytes would fail a - # genuine quote. - signed = quote.raw parsed_signature: ParsedSignature | None = None if isinstance(signature, ParsedSignature): parsed_signature = signature - elif isinstance(ak_key, rsa.RSAPublicKey): + elif isinstance(ak_public_key, rsa.RSAPublicKey): # A bare RSA signature is exactly one modulus wide. Do not use the # untrusted sigAlg prefix as the discriminator: changing an envelope's # scheme must not silently move it into the legacy bare-signature lane. - if len(signature) != (ak_key.key_size + 7) // 8: + if len(signature) != (ak_public_key.key_size + 7) // 8: parsed_signature = parse_tpmt_signature(signature) - elif isinstance(ak_key, ec.EllipticCurvePublicKey): + elif isinstance(ak_public_key, ec.EllipticCurvePublicKey): # cryptography's legacy ECDSA input is a complete DER SEQUENCE. Testing # only its first 0x30 byte would let malformed sequence-prefixed input # fall through too, so require the actual signature representation. @@ -473,13 +486,13 @@ def verify_tpm_quote( ) try: - if isinstance(ak_key, ec.EllipticCurvePublicKey): + if isinstance(ak_public_key, ec.EllipticCurvePublicKey): if signature_algorithm not in (None, _ALG_ECDSA): raise TpmVerificationError( "TPMT_SIGNATURE algorithm does not match the EC attestation key" ) - ak_key.verify(bare_signature, signed, ec.ECDSA(digest)) - elif isinstance(ak_key, rsa.RSAPublicKey): + ak_public_key.verify(bare_signature, attest_raw, ec.ECDSA(digest)) + elif isinstance(ak_public_key, rsa.RSAPublicKey): signature_padding: padding.AsymmetricPadding if signature_algorithm in (None, _ALG_RSASSA): signature_padding = padding.PKCS1v15() @@ -491,11 +504,86 @@ def verify_tpm_quote( raise TpmVerificationError( "TPMT_SIGNATURE algorithm does not match the RSA attestation key" ) - ak_key.verify(bare_signature, signed, signature_padding, digest) + ak_public_key.verify(bare_signature, attest_raw, signature_padding, digest) else: raise TpmVerificationError("unsupported AK public-key type for TPM quote") except InvalidSignature: return False + return True + +def verify_tpm_quote( + attest: bytes, + signature: bytes | ParsedSignature, + ak_chain_pem: bytes, + *, + trusted_roots_pem: bytes, + expected_qualifying_data: bytes | None = None, + expected_pcr_digest: bytes | None = None, + verification_time: datetime | None = None, +) -> bool: + """Fully verify a TPM 2.0 quote offline (all four steps, fail-closed). + + Args: + attest: the raw ``TPMS_ATTEST`` blob the TPM signed. + signature: either the legacy bare AK signature (DER ECDSA or RSA + PKCS#1 v1.5 over SHA-256), a parsed :class:`ParsedSignature`, or a + marshalled ``TPMT_SIGNATURE``. Envelopes select RSASSA, RSAPSS, or + ECDSA and SHA-256, SHA-384, or SHA-512 from their algorithm ids. + Legacy bare input is recognized from the AK's signature shape: + modulus-sized bytes for RSA and DER for ECDSA. Other byte input + must be a well-formed envelope. + ak_chain_pem: the AK certificate chain (PEM, leaf first). + trusted_roots_pem: the caller's trusted vendor EK/AK roots (PEM). + expected_qualifying_data: if given, the quote's ``extraData`` (nonce) + must equal it. + expected_pcr_digest: if given, the quote's PCR digest must equal it. + verification_time: UTC-aware time used to check AK chain certificate + validity periods (default: current UTC time). Primarily useful + for deterministic tests. + + Returns: + ``True`` only when the structure, AK chain, AK signature, and any + supplied bindings all check out. Returns ``False`` on a well-formed but + invalid signature or a binding mismatch. Raises + :class:`TpmVerificationError` on a malformed quote / broken chain, a + malformed signature, an unsupported algorithm, or if ``cryptography`` + is unavailable. + """ + try: + import cryptography # noqa: F401 + except ImportError as e: # pragma: no cover + raise TpmVerificationError( + "TPM quote verification requires the 'cryptography' package" + ) from e + + quote = parse_tpm_quote(attest) + + # Step 1: structural — this must be a TPM-generated quote. + if quote.magic != TPM_GENERATED_VALUE: + raise TpmVerificationError( + f"TPMS_ATTEST magic is not TPM_GENERATED (magic={quote.magic:#x})" + ) + if quote.attest_type != TPM_ST_ATTEST_QUOTE: + raise TpmVerificationError( + f"attestation is not a quote (type={quote.attest_type:#x})" + ) + + # Step 2: AK certificate chain up to a pinned trusted root. + ak = _verify_ak_chain(ak_chain_pem, trusted_roots_pem, verification_time=verification_time) + ak_key = ak.public_key() # type: ignore[attr-defined] + + # Step 3: AK signature over the TPMS_ATTEST blob. Use quote.raw rather than + # the argument: when the caller passes a size-prefixed TPM2B_ATTEST, the TPM + # signed the inner structure, so verifying over the outer bytes would fail a + # genuine quote. + # Delegates to verify_ak_signature(), which contains this same key-shape + # classification logic (bare-signature-vs-TPMT_SIGNATURE, algorithm and + # hash selection). Kept as a single shared implementation rather than + # duplicated here, since _azure_verify.py also calls it directly to check + # an Azure vTPM AK-signed quote against an AK that is trusted via the SNP + # binding rather than an X.509 chain. + if not verify_ak_signature(ak_key, quote.raw, signature): + return False # Step 4: bindings (constant-time). if expected_qualifying_data is not None and not hmac.compare_digest( diff --git a/python/src/agent_manifest/_trace.py b/python/src/agent_manifest/_trace.py index 8dff715..b016648 100644 --- a/python/src/agent_manifest/_trace.py +++ b/python/src/agent_manifest/_trace.py @@ -112,6 +112,14 @@ # "", so neither has a shape this can enforce without # rejecting records the spec permits. `tool_id` is a reverse-domain identifier # with no stated grammar, so it is checked for emptiness only. +# NOTE: these patterns keep their `^`/`$` anchors (matching what the spec's +# JSON-schema-style grammar expects), but every call site below uses +# `.fullmatch()`, never `.match()`. `.match()` with a `$`-anchored pattern +# would let "\n" through, since Python's `$` matches at +# end-of-string OR just before a single trailing '\n'. `.fullmatch()` +# requires the pattern to consume the entire string and is not subject to +# that exception, even though the pattern text itself still has `$`. + _UUID_V7 = re.compile( r"^[0-9a-f]{8}-[0-9a-f]{4}-7[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$", re.IGNORECASE, @@ -468,22 +476,22 @@ def _envelope_format_failures(envelope: dict[str, Any]) -> list[str]: failures: list[str] = [] for field_name in ("trace_id", "agent_manifest_id"): - if not _UUID_V7.match(envelope[field_name]): + if not _UUID_V7.fullmatch(envelope[field_name]): failures.append(f"not_a_uuid_v7:{field_name}") # hitl_approval_id is " | null", so null is legal and a present # value must be well formed. approval_id = envelope.get("hitl_approval_id") if approval_id is not None and ( - not isinstance(approval_id, str) or not _UUID_V7.match(approval_id) + not isinstance(approval_id, str) or not _UUID_V7.fullmatch(approval_id) ): failures.append("not_a_uuid_v7:hitl_approval_id") - if not _SPIFFE.match(envelope["agent_id"]): + if not _SPIFFE.fullmatch(envelope["agent_id"]): failures.append("not_a_spiffe_uri:agent_id") for field_name in ("policy_hash", "catalog_hash"): - if not _SHA256.match(envelope[field_name]): + if not _SHA256.fullmatch(envelope[field_name]): failures.append(f"not_a_sha256_hash:{field_name}") if envelope["decision"] not in TRACE_DECISIONS: diff --git a/python/src/agent_manifest/_transparency.py b/python/src/agent_manifest/_transparency.py index f27ba20..2af2356 100644 --- a/python/src/agent_manifest/_transparency.py +++ b/python/src/agent_manifest/_transparency.py @@ -70,6 +70,9 @@ def publish_to_rekor( Raises: RuntimeError: If the Rekor API call fails. ImportError: If httpx is not installed. + ValueError: If public_key_b64url is not valid, strictly-encoded + base64url. + """ try: import httpx @@ -80,15 +83,14 @@ def publish_to_rekor( ) from ._canonicalize import canonicalize - from ._signing import SIGNED_FIELDS + from ._signing import SIGNED_FIELDS, _b64url_decode # Build the signed bytes (must match what was signed) subset = {k: manifest_dict[k] for k in SIGNED_FIELDS if k in manifest_dict} canonical_bytes = canonicalize(subset) # Decode public key from base64url to PEM for Rekor - pad = 4 - len(public_key_b64url) % 4 - pub_raw = base64.urlsafe_b64decode(public_key_b64url + ("=" * pad if pad != 4 else "")) + pub_raw = _b64url_decode(public_key_b64url) pub_pem = _raw_ed25519_to_pem(pub_raw) # Rekor hashedrekord entry format diff --git a/python/src/agent_manifest/_types.py b/python/src/agent_manifest/_types.py index 66cfdff..f97af2e 100644 --- a/python/src/agent_manifest/_types.py +++ b/python/src/agent_manifest/_types.py @@ -17,6 +17,15 @@ class ManifestId(str): The variant nibble (position 19) MUST be one of [89ab]. """ + # NOTE: keeps `^`/`$` anchors (needed for the JSON-schema `pattern` + # exported below to mean "matches exactly", per JSON Schema/ECMA 262 + # semantics for external consumers). Internal validation below uses + # `.fullmatch()`, never `.match()`, since `.match()` with a + # `$`-anchored pattern would let "\n" through -- Python's + # `$` matches at end-of-string OR just before one trailing '\n`. + # `.fullmatch()` requires the whole string and has no such exception. + + _PATTERN = re.compile( r"^[0-9a-f]{8}-[0-9a-f]{4}-7[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$", re.IGNORECASE, @@ -41,7 +50,7 @@ def __get_pydantic_json_schema__( def _validate(cls, v: Any) -> "ManifestId": if not isinstance(v, str): raise ValueError(f"ManifestId must be a string, got {type(v).__name__}") - if not cls._PATTERN.match(v): + if not cls._PATTERN.fullmatch(v): raise ValueError( f"'{v}' is not a valid UUID v7. " "Expected format: xxxxxxxx-xxxx-7xxx-[89ab]xxx-xxxxxxxxxxxx" @@ -57,6 +66,7 @@ class HashValue(str): shake256:<64 lowercase hex chars> (256-bit output, per RFC 8785 / FIPS 202) """ + # NOTE: same `.fullmatch()`-not-`.match()` reasoning as ManifestId above. _PATTERN = re.compile(r"^(sha256|shake256):[0-9a-f]{64}$") @classmethod @@ -78,7 +88,7 @@ def __get_pydantic_json_schema__( def _validate(cls, v: Any) -> "HashValue": if not isinstance(v, str): raise ValueError(f"HashValue must be a string, got {type(v).__name__}") - if not cls._PATTERN.match(v): + if not cls._PATTERN.fullmatch(v): prefix = v.split(":")[0] if ":" in v else v[:10] raise ValueError( f"Invalid hash value (prefix='{prefix}'). " diff --git a/python/tests/test_attestation_chain.py b/python/tests/test_attestation_chain.py index 91cd7c9..b46039a 100644 --- a/python/tests/test_attestation_chain.py +++ b/python/tests/test_attestation_chain.py @@ -6,6 +6,7 @@ signature and VCEK chain verify (synthetic self-consistent crypto). """ +import base64 import hashlib import pytest @@ -51,6 +52,49 @@ def test_missing_report_data_field(): assert result.report_data_matched is False +@ pytest.mark.parametrize( + "bad_hash", + [ + f"md5:{DIGEST}", # right length, wrong algorithm name + f"sha1:{DIGEST}", # right length, wrong algorithm name + f"foo:{DIGEST}", # nonsense algorithm name + "sha256:", # prefix only, no digest + "sha256:" + DIGEST[:-2], # 62 hex chars, one short + "sha256:" + DIGEST + "aa", # 66 hex chars, one too many + "sha256:" + "zz" * 32, # right length, invalid hex + DIGEST, # no prefix at all + ], +) +def test_report_data_binding_rejects_malformed_expected_manifest_hash_prefix(bad_hash): + """The non-Azure REPORT_DATA binding step has the same + ``expected_manifest_hash`` contract as the Azure composite check: it + must be exactly ``"sha256:" + 64 hex chars``. A verifier that merely + does ``split(":", 1)[-1]`` implicitly assumes "whatever follows a colon + is the SHA-256 digest", so ``"md5:<64 hex>"`` or a bare hex string with + no prefix at all would otherwise be treated as if it were a genuine + ``"sha256:<64 hex>"`` value. Even when the report's own report_data + happens to carry bytes that would match the trailing hex, a malformed + or wrong-algorithm ``expected_manifest_hash`` must fail closed rather + than accidentally "matching". + """ + # report_data carries the *real* digest so that, if the algorithm-prefix + # check were skipped, the naive split-based comparison would otherwise + # spuriously succeed -- this isolates the prefix-validation bug from an + # unrelated report_data mismatch. + report = _report(report_data_hex=DIGEST + "00" * 32) + result = verify_attestation_chain(report, expected_manifest_hash=bad_hash) + assert result.report_data_matched is False + + +def test_report_data_binding_accepts_well_formed_expected_manifest_hash(): + """Sanity check for the positive case: a genuine ``"sha256:" + 64 hex`` + value must still verify -- the stricter prefix check must not reject + well-formed input.""" + report = _report(report_data_hex=DIGEST + "00" * 32) + result = verify_attestation_chain(report, expected_manifest_hash=MANIFEST_HASH) + assert result.report_data_matched is True + + def test_measurement_allow_list_hit(): report = _report(report_data_hex=DIGEST + "00" * 32, measurement=MEASUREMENT) result = verify_attestation_chain( @@ -239,3 +283,631 @@ def test_full_chain_reads_snp_bytes_from_report_quote(): ) assert result.signature is SignatureStatus.VERIFIED assert result.passed is True + + +# --------------------------------------------------------------------------- +# Azure paravisor SNP + unsupported-platform dispatch. +# +# REPORT_DATA on Azure is sha256(runtime_data), never the manifest hash (the +# guest does not control it). verify_attestation_chain establishes Azure's +# real manifest binding itself (a vTPM AK-signed quote over a PCR derived +# from the manifest hash, chained to REPORT_DATA via the runtime data -- see +# agent_manifest._azure_verify.verify_azure_manifest_binding) directly from +# evidence carried on the report. There is no caller-supplied boolean +# anywhere in this API that can substitute for that check -- see #373: an +# earlier revision accepted an `azure_manifest_binding_verified` flag from +# the caller, which let anyone construct a report with no real evidence at +# all and still get passed=True by asserting the flag. Platform dispatch +# must also be an explicit allow-list, not a catch-all, so an unrecognized +# platform label can't silently borrow the SNP verifier. +# --------------------------------------------------------------------------- + + +def _azure_build_attest( + qualifying_data: bytes, + pcr_digest: bytes, + *, + hash_alg: int = 0x000B, + bitmap: bytes = b"\x00\x00\x01", + sizeof_select: int | None = None, +) -> bytes: + """Minimal structurally-valid TPMS_ATTEST (single PCR-bank selection).""" + + from agent_manifest._tpm_verify import TPM_GENERATED_VALUE, TPM_ST_ATTEST_QUOTE + + out = TPM_GENERATED_VALUE.to_bytes(4, "big") + out += TPM_ST_ATTEST_QUOTE.to_bytes(2, "big") + out += (0).to_bytes(2, "big") # qualifiedSigner: empty TPM2B_NAME + out += len(qualifying_data).to_bytes(2, "big") + qualifying_data + out += b"\x00" * 17 # clockInfo + out += b"\x00" * 8 # firmwareVersion + out += (1).to_bytes(4, "big") # TPML_PCR_SELECTION count + out += hash_alg.to_bytes(2, "big") + out += (sizeof_select if sizeof_select is not None else len(bitmap)).to_bytes(1, "big") + out += bitmap + out += len(pcr_digest).to_bytes(2, "big") + pcr_digest + return out + + +def _azure_tpmt_sign(ak_key, attest: bytes) -> bytes: + """RSASSA/SHA-256 TPMT_SIGNATURE over ``attest``.""" + from cryptography.hazmat.primitives import hashes + from cryptography.hazmat.primitives.asymmetric import padding + + sig = ak_key.sign(attest, padding.PKCS1v15(), hashes.SHA256()) + return (0x0014).to_bytes(2, "big") + (0x000B).to_bytes(2, "big") + len(sig).to_bytes(2, "big") + sig + + +def _azure_good_fixture(*, manifest_digest_hex: str = DIGEST): + """Build a fully self-consistent, cryptographically-valid Azure evidence set. + + Every field is genuinely tied together: the AK signs a quote over the + manifest PCR, the AK is embedded in runtime_data, and runtime_data is + bound (via REPORT_DATA) into the SNP report -- exactly the composite + chain verify_attestation_chain must authenticate for an Azure report to + pass. Mirrors the fixture in test_hw_providers.py (same crypto, reused + here because verify_attestation_chain must establish this chain itself + -- it is the object under test, not AzureCVMProvider). + """ + import base64 + import json + + from cryptography.hazmat.primitives.asymmetric import rsa + from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat + + expected_pcr_value = hashlib.sha256(bytes(32) + bytes.fromhex(manifest_digest_hex)).digest() + pcr_digest = hashlib.sha256(expected_pcr_value).digest() + + ak_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + ak_pub_pem = ak_key.public_key().public_bytes( + Encoding.PEM, PublicFormat.SubjectPublicKeyInfo + ).decode() + + quote_msg = _azure_build_attest(bytes(16), pcr_digest) + quote_sig = _azure_tpmt_sign(ak_key, quote_msg) + + numbers = ak_key.public_key().public_numbers() + modulus_bytes = numbers.n.to_bytes((numbers.n.bit_length() + 7) // 8, "big") + n_b64 = base64.urlsafe_b64encode(modulus_bytes).rstrip(b"=").decode() + runtime_data = json.dumps({"keys": [{"kid": "HCLAkPub", "n": n_b64, "e": "AQAB"}]}).encode() + + report_data = hashlib.sha256(runtime_data).digest() + bytes(32) + snp, vcek_der, chain = _synthetic_snp_with_chain(report_data.hex()[:64], MEASUREMENT) + + return { + "snp": snp, + "vcek_der": vcek_der, + "chain": chain, + "raw": { + "ak_pub_pem": ak_pub_pem, + "runtime_data_hex": runtime_data.hex(), + "quote_msg": base64.b64encode(quote_msg).decode(), + "quote_sig": base64.b64encode(quote_sig).decode(), + "measurement": MEASUREMENT, + }, + "ak_key": ak_key, + "pcr_digest": pcr_digest, + } + + +def _azure_report_from_fixture(fx, *, raw_overrides=None): + raw = dict(fx["raw"]) + if raw_overrides: + raw.update(raw_overrides) + return AttestationReport( + platform="azure-cvm-sev-snp", + manifest_hash=MANIFEST_HASH, + quote=fx["snp"], + raw=raw, + ) + + +def test_azure_report_full_composite_chain_passes(): + # The positive path: verify_attestation_chain establishes the entire + # Azure binding itself (PCR-in-quote, AK signature, AK identity, + # runtime_data->REPORT_DATA) purely from evidence on the report, plus + # the SNP signature/VCEK chain -- and only then can passed be True. + fx = _azure_good_fixture() + report = _azure_report_from_fixture(fx) + result = verify_attestation_chain( + report, + expected_manifest_hash=MANIFEST_HASH, + vcek_cert_der=fx["vcek_der"], + cert_chain_pem=fx["chain"], + ) + assert result.signature is SignatureStatus.VERIFIED + assert result.report_data_matched is True + assert result.passed is True + + +def test_azure_explicit_empty_snp_report_bytes_is_not_silently_replaced(): + # `snp_report_bytes or getattr(report, "quote", None)` would treat an + # explicit `snp_report_bytes=b""` the same as "not given" (both are + # falsy) and silently fall back to report.quote -- even though the + # caller *did* supply a value, just an invalid/empty one. That is wrong + # verifier semantics: supplied evidence must be judged on its own terms, + # never silently discarded and replaced with a different source. Build + # a report whose own report.quote is fully valid, matching SNP evidence + # (so a wrongful fallback would make this pass), but pass an explicit + # empty snp_report_bytes -- the composite check must still fail closed. + fx = _azure_good_fixture() + report = _azure_report_from_fixture(fx) + result = verify_attestation_chain( + report, + expected_manifest_hash=MANIFEST_HASH, + snp_report_bytes=b"", + vcek_cert_der=fx["vcek_der"], + cert_chain_pem=fx["chain"], + ) + assert result.report_data_matched is False + assert result.passed is False + + +def test_azure_omitted_snp_report_bytes_still_falls_back_to_report_quote(): + # Sanity check for the positive case: when the caller genuinely doesn't + # supply snp_report_bytes at all (the parameter default, None), falling + # back to report.quote is still the documented, correct behavior -- the + # fix must only change the falsy-but-supplied case, not remove the + # None fallback entirely. + fx = _azure_good_fixture() + report = _azure_report_from_fixture(fx) + result = verify_attestation_chain( + report, + expected_manifest_hash=MANIFEST_HASH, + vcek_cert_der=fx["vcek_der"], + cert_chain_pem=fx["chain"], + ) + assert result.report_data_matched is True + assert result.passed is True + + +def test_azure_report_with_valid_snp_signature_and_wrong_pcr_does_not_pass(): + # a correctly signed SNP report, but the PCR inside the AK-signed quote is wrong. + # A valid hardware signature must not be enough on its own. + fx = _azure_good_fixture() + wrong_pcr_digest = hashlib.sha256(bytes(32) + b"\x00" * 32).digest() + tampered_quote_msg = _azure_build_attest(bytes(16), wrong_pcr_digest) + tampered_sig = _azure_tpmt_sign(fx["ak_key"], tampered_quote_msg) + report = _azure_report_from_fixture( + fx, + raw_overrides={ + "quote_msg": base64.b64encode(tampered_quote_msg).decode(), + "quote_sig": base64.b64encode(tampered_sig).decode(), + }, + ) + result = verify_attestation_chain( + report, + expected_manifest_hash=MANIFEST_HASH, + vcek_cert_der=fx["vcek_der"], + cert_chain_pem=fx["chain"], + ) + assert result.signature is SignatureStatus.VERIFIED + assert result.report_data_matched is False + assert result.passed is False + + +def test_azure_report_wrong_pcr_selection_same_digest_bytes(): + # the signed selection bitmap changed from PCR16 (000001) to PCR17 (000002), + # pcr_digest bytes unchanged, re-signed with the runtime-data-bound AK. + # A verifier checking only pcr_digest returned signature=verified, report_data_matched=True, + # passed=True. Must fail now: the bank/PCR the quote was actually signed over must be + # checked, not just the digest value. + fx = _azure_good_fixture() + retargeted_quote_msg = _azure_build_attest( + bytes(16), fx["pcr_digest"], sizeof_select=3, bitmap=b"\x00\x00\x02" + ) + retargeted_sig = _azure_tpmt_sign(fx["ak_key"], retargeted_quote_msg) + report = _azure_report_from_fixture( + fx, + raw_overrides={ + "quote_msg": base64.b64encode(retargeted_quote_msg).decode(), + "quote_sig": base64.b64encode(retargeted_sig).decode(), + }, + ) + result = verify_attestation_chain( + report, + expected_manifest_hash=MANIFEST_HASH, + vcek_cert_der=fx["vcek_der"], + cert_chain_pem=fx["chain"], + azure_expected_pcr_index=16, + ) + assert result.signature is SignatureStatus.VERIFIED + assert result.report_data_matched is False + assert result.passed is False + + +def test_azure_report_ak_exponent_mismatch(): + # runtime-data JWK exponent changed to e=3 ("Aw") while the PEM + # AK key keeps e=65537, rebound into a freshly signed SNP report. + # A verifier comparing only the modulus returned + # passed=True. Must fail now: both n and e must match. + import json + + fx = _azure_good_fixture() + numbers = fx["ak_key"].public_key().public_numbers() + modulus_bytes = numbers.n.to_bytes((numbers.n.bit_length() + 7) // 8, "big") + n_b64 = base64.urlsafe_b64encode(modulus_bytes).rstrip(b"=").decode() + wrong_e_b64 = base64.urlsafe_b64encode((3).to_bytes(1, "big")).rstrip(b"=").decode() + runtime_data_wrong_e = json.dumps( + {"keys": [{"kid": "HCLAkPub", "n": n_b64, "e": wrong_e_b64}]} + ).encode() + + report_data = hashlib.sha256(runtime_data_wrong_e).digest() + bytes(32) + snp, vcek_der, chain = _synthetic_snp_with_chain(report_data.hex()[:64], MEASUREMENT) + report = _azure_report_from_fixture( + fx, raw_overrides={"runtime_data_hex": runtime_data_wrong_e.hex()} + ) + report.quote = snp + result = verify_attestation_chain( + report, + expected_manifest_hash=MANIFEST_HASH, + vcek_cert_der=vcek_der, + cert_chain_pem=chain, + ) + assert result.report_data_matched is False + assert result.passed is False + + +@pytest.mark.parametrize( + "corrupt", + [ + pytest.param(lambda n_b64: "!!!!" + n_b64, id="illegal-prefix"), + pytest.param(lambda n_b64: n_b64 + "!!!!", id="illegal-suffix"), + pytest.param(lambda n_b64: n_b64[:-1] + "+", id="standard-base64-plus-alias"), + pytest.param(lambda n_b64: n_b64[:-1] + "/", id="standard-base64-slash-alias"), + ], +) +def test_azure_report_malformed_jwk_modulus_encoding_fails_closed(corrupt): + # base64.urlsafe_b64decode() silently *discards* characters outside the + # base64 alphabet instead of raising -- so "!!!!" used to decode + # to the same bytes as "". Rebind the corrupted n back into a + # freshly, validly signed SNP report (report_data = sha256(runtime_data) + # still matches exactly) and re-sign the AK quote over the same PCR, so + # every other link in the chain is genuinely valid: only the JWK + # encoding is malformed. Must fail closed -- a correctly rebound and + # signed report must never pass with illegally-encoded JWK material. + import json + + fx = _azure_good_fixture() + numbers = fx["ak_key"].public_key().public_numbers() + modulus_bytes = numbers.n.to_bytes((numbers.n.bit_length() + 7) // 8, "big") + good_n_b64 = base64.urlsafe_b64encode(modulus_bytes).rstrip(b"=").decode() + corrupt_n_b64 = corrupt(good_n_b64) + + corrupt_runtime_data = json.dumps( + {"keys": [{"kid": "HCLAkPub", "n": corrupt_n_b64, "e": "AQAB"}]} + ).encode() + report_data = hashlib.sha256(corrupt_runtime_data).digest() + bytes(32) + snp, vcek_der, chain = _synthetic_snp_with_chain(report_data.hex()[:64], MEASUREMENT) + report = _azure_report_from_fixture( + fx, raw_overrides={"runtime_data_hex": corrupt_runtime_data.hex()} + ) + report.quote = snp + result = verify_attestation_chain( + report, + expected_manifest_hash=MANIFEST_HASH, + vcek_cert_der=vcek_der, + cert_chain_pem=chain, + ) + # The runtime data is still cryptographically committed into REPORT_DATA + # (report_data_matched would be True under the old permissive decoder), + # but the AK identity check must now reject the malformed JWK encoding. + assert result.passed is False + + +@pytest.mark.parametrize( + "corrupt", + [ + pytest.param(lambda e_b64: "!!!!" + e_b64, id="illegal-prefix"), + pytest.param(lambda e_b64: e_b64 + "!!!!", id="illegal-suffix"), + pytest.param(lambda e_b64: e_b64[:-1] + "+", id="standard-base64-plus-alias"), + pytest.param(lambda e_b64: e_b64[:-1] + "/", id="standard-base64-slash-alias"), + ], +) +def test_azure_report_malformed_jwk_exponent_encoding_fails_closed(corrupt): + # Twin of test_azure_report_malformed_jwk_modulus_encoding_fails_closed, + # but corrupting the JWK "e" member instead of "n". Both members go + # through the same _strict_b64url_decode() call inside + # ak_public_numbers_from_runtime_data(); the modulus test alone does not + # prove the exponent path is covered, since a future edit could special- + # case or bypass strict decoding for "e" specifically without this test + # noticing. n is kept well-formed here so only the exponent encoding is + # under test -- rebind REPORT_DATA to the new runtime data and re-derive + # a matching PCR/quote/SNP-report exactly as the modulus test does, so + # every other link in the chain is genuinely valid. + import json + + fx = _azure_good_fixture() + numbers = fx["ak_key"].public_key().public_numbers() + modulus_bytes = numbers.n.to_bytes((numbers.n.bit_length() + 7) // 8, "big") + good_n_b64 = base64.urlsafe_b64encode(modulus_bytes).rstrip(b"=").decode() + good_e_b64 = base64.urlsafe_b64encode( + numbers.e.to_bytes((numbers.e.bit_length() + 7) // 8, "big") + ).rstrip(b"=").decode() + corrupt_e_b64 = corrupt(good_e_b64) + + corrupt_runtime_data = json.dumps( + {"keys": [{"kid": "HCLAkPub", "n": good_n_b64, "e": corrupt_e_b64}]} + ).encode() + report_data = hashlib.sha256(corrupt_runtime_data).digest() + bytes(32) + snp, vcek_der, chain = _synthetic_snp_with_chain(report_data.hex()[:64], MEASUREMENT) + report = _azure_report_from_fixture( + fx, raw_overrides={"runtime_data_hex": corrupt_runtime_data.hex()} + ) + report.quote = snp + result = verify_attestation_chain( + report, + expected_manifest_hash=MANIFEST_HASH, + vcek_cert_der=vcek_der, + cert_chain_pem=chain, + ) + # REPORT_DATA still matches (it is a hash of the corrupted runtime_data + # itself, so corrupting the JWK content inside it does not break step 4); + # only the AK-identity check (step 3, which decodes "e") must reject. + assert result.passed is False + + +def test_ak_public_numbers_and_modulus_reject_illegal_alphabet_direct(): + # Direct unit coverage: the strict decoder must reject illegal + # prefixes/suffixes and standard-base64 +/ aliases for both n and e, + # in both the (n, e) helper and the modulus-only helper. + import json + + from agent_manifest._azure_verify import ( + ak_modulus_hex_from_runtime_data, + ak_public_numbers_from_runtime_data, + ) + + good_n = base64.urlsafe_b64encode(b"\x01\x00\x01\xff").rstrip(b"=").decode() + + for bad_n in ("!!!!" + good_n, good_n + "!!!!", good_n[:-1] + "+", good_n[:-1] + "/"): + runtime_data = json.dumps( + {"keys": [{"kid": "HCLAkPub", "n": bad_n, "e": "AQAB"}]} + ).encode() + assert ak_modulus_hex_from_runtime_data(runtime_data) is None + assert ak_public_numbers_from_runtime_data(runtime_data) is None + + # A bad exponent must also be rejected, even when n is well-formed. + for bad_e in ("!!!!AQAB", "AQAB!!!!", "AQA+", "AQA/"): + runtime_data = json.dumps( + {"keys": [{"kid": "HCLAkPub", "n": good_n, "e": bad_e}]} + ).encode() + assert ak_public_numbers_from_runtime_data(runtime_data) is None + + # Sanity: the well-formed value still decodes fine through both helpers. + good_runtime_data = json.dumps( + {"keys": [{"kid": "HCLAkPub", "n": good_n, "e": "AQAB"}]} + ).encode() + assert ak_modulus_hex_from_runtime_data(good_runtime_data) == "010001ff" + assert ak_public_numbers_from_runtime_data(good_runtime_data) == ("010001ff", 65537) + + +def test_ak_public_numbers_and_modulus_reject_empty_n_or_e(): + # The alphabet regex used to be `[A-Za-z0-9_-]*` (zero-or-more), so + # fullmatch("") succeeded and an empty n/e sailed through the alphabet + # check straight to base64 decoding as b"" -- turning a malformed JWK + # ({"n": "", "e": ""}) into ("", 0) instead of the documented + # fail-closed None. An RSA modulus/exponent is never legitimately + # empty, so both must be rejected the same way any other malformed + # alphabet input is. + import json + + from agent_manifest._azure_verify import ( + ak_modulus_hex_from_runtime_data, + ak_public_numbers_from_runtime_data, + ) + + good_n = base64.urlsafe_b64encode(b"\x01\x00\x01\xff").rstrip(b"=").decode() + + # Empty n, well-formed e. + runtime_data = json.dumps({"keys": [{"kid": "HCLAkPub", "n": "", "e": "AQAB"}]}).encode() + assert ak_modulus_hex_from_runtime_data(runtime_data) is None + assert ak_public_numbers_from_runtime_data(runtime_data) is None + + # Well-formed n, empty e. + runtime_data = json.dumps({"keys": [{"kid": "HCLAkPub", "n": good_n, "e": ""}]}).encode() + assert ak_public_numbers_from_runtime_data(runtime_data) is None + + # Both empty. + runtime_data = json.dumps({"keys": [{"kid": "HCLAkPub", "n": "", "e": ""}]}).encode() + assert ak_modulus_hex_from_runtime_data(runtime_data) is None + assert ak_public_numbers_from_runtime_data(runtime_data) is None + + +def test_strict_b64url_decode_rejects_trailing_newline_regex_dollar_hole(): + # Python's `$` regex anchor matches at end-of-string OR just before a + # single trailing '\n' -- so a naive `^[A-Za-z0-9_-]*$` alphabet check + # would let "AQAB\n" through even though '\n' isn't in the alphabet, + # and base64.urlsafe_b64decode then silently drops the '\n' and decodes + # it identically to "AQAB". Must use fullmatch (no '$'/'\Z' hole) so a + # trailing newline is rejected like any other illegal character. + import binascii + + from agent_manifest._azure_verify import _strict_b64url_decode + + assert _strict_b64url_decode("AQAB") == bytes.fromhex("010001") + with pytest.raises(binascii.Error): + _strict_b64url_decode("AQAB\n") + with pytest.raises(binascii.Error): + _strict_b64url_decode("AQAB\n\n") + + +def test_azure_report_malformed_runtime_data_keys_shape(): + # valid JSON `{"keys": 1}`, correctly hash-bound into the signed SNP report. + # The unpatched helper raised TypeError: 'int' object is not iterable, which + # leaked out of verify_attestation_chain despite the "never raises" contract. Must + # fail closed (return False, no exception) now. + import json + + fx = _azure_good_fixture() + malformed_runtime_data = json.dumps({"keys": 1}).encode() + report_data = hashlib.sha256(malformed_runtime_data).digest() + bytes(32) + snp, vcek_der, chain = _synthetic_snp_with_chain(report_data.hex()[:64], MEASUREMENT) + report = _azure_report_from_fixture( + fx, raw_overrides={"runtime_data_hex": malformed_runtime_data.hex()} + ) + report.quote = snp + result = verify_attestation_chain( + report, + expected_manifest_hash=MANIFEST_HASH, + vcek_cert_der=vcek_der, + cert_chain_pem=chain, + ) + assert result.report_data_matched is False + assert result.passed is False + + +def test_verify_azure_manifest_binding_requires_expected_pcr_index_kwarg(): + # expected_pcr_index must be a real, required parameter -- not something + # a caller can omit and have silently inferred from self-reported data. + import inspect + + from agent_manifest._azure_verify import verify_azure_manifest_binding + + sig = inspect.signature(verify_azure_manifest_binding) + assert "expected_pcr_index" in sig.parameters + assert sig.parameters["expected_pcr_index"].default is inspect.Parameter.empty + + +def test_ak_public_numbers_from_runtime_data_malformed_keys_shape_returns_none(): + # Direct unit coverage of the fail-closed helper fix: {"keys": 1} used to + # raise TypeError iterating an int; must now return None like every other + # malformed-input case. + import json + + from agent_manifest._azure_verify import ( + ak_modulus_hex_from_runtime_data, + ak_public_numbers_from_runtime_data, + ) + + malformed = json.dumps({"keys": 1}).encode() + assert ak_modulus_hex_from_runtime_data(malformed) is None + assert ak_public_numbers_from_runtime_data(malformed) is None + + +@pytest.mark.parametrize( + "runtime_json", + [ + {"keys": "not-a-list"}, + {"keys": [1, 2, 3]}, + {"keys": None}, + "not-a-dict-at-top-level", + 123, + [], + ], +) + + +def test_ak_public_numbers_from_runtime_data_fails_closed_on_malformed_shapes(runtime_json): + import json + + from agent_manifest._azure_verify import ak_public_numbers_from_runtime_data + + assert ak_public_numbers_from_runtime_data(json.dumps(runtime_json).encode()) is None + + +def test_azure_report_with_no_evidence_supplied_fails_closed(): + # Nothing supplied at all (the common/default case): must not be quietly + # assumed fine. + report = AttestationReport( + platform="azure-cvm-sev-snp", + manifest_hash=MANIFEST_HASH, + raw={"report_data": DIGEST + "00" * 32, "measurement": MEASUREMENT}, + ) + + result = verify_attestation_chain(report, expected_manifest_hash=MANIFEST_HASH) + assert result.report_data_matched is False + assert result.passed is False + assert any("not established" in r for r in result.reasons) + + +@pytest.mark.parametrize( + "bad_raw", + ["attacker-controlled-string", 12345, [1, 2, 3], b"attacker-controlled-bytes"], +) +def test_azure_report_non_dict_raw_fails_closed(bad_raw): + """``AttestationReport.raw`` is typed ``dict[str, Any]``, but a caller + constructing (or forging) a report is not stopped by that type hint at + runtime -- a truthy non-dict is just as reachable as any individual + malformed field on ``raw``. ``getattr(report, "raw", {}) or {}`` only + guards the falsy cases (missing attribute, ``None``, ``{}``, ``""``); a + truthy non-dict still reaches ``raw_azure.get(...)`` and raises + ``AttributeError``, breaking this function's documented fail-closed + contract. Must return False, never raise. + """ + report = AttestationReport( + platform="azure-cvm-sev-snp", + manifest_hash=MANIFEST_HASH, + raw=bad_raw, + ) + result = verify_attestation_chain(report, expected_manifest_hash=MANIFEST_HASH) + assert result.report_data_matched is False + assert result.passed is False + + +@pytest.mark.parametrize("arbitrary_hash", ["sha256:" + "11" * 32, "sha256:" + "22" * 32, "sha256:" + "ff" * 32]) +def test_azure_report_fails_closed_for_any_expected_manifest_hash(arbitrary_hash): + # The platform label and expected_manifest_hash alone must never combine + # into a pass, regardless of which hash is expected -- reproduces the + # that a signed report passed for three unrelated expected_manifest_hash + # values (11.., 22.., ff..). + report = AttestationReport( + platform="azure-cvm-sev-snp", + manifest_hash=MANIFEST_HASH, + raw={"report_data": DIGEST + "00" * 32, "measurement": MEASUREMENT}, + ) + result = verify_attestation_chain(report, expected_manifest_hash=arbitrary_hash) + assert result.passed is False + + +def test_azure_report_only_pcr_read_string_is_not_evidence(): + # a report that "looks Azure" (has a plausible-looking legacy + # pcr_read string, a self-reported runtime_data_binding_verified=True + # flag) but carries none of the actual quote/signature/AK material must not pass. + report = AttestationReport( + platform="azure-cvm-sev-snp", + manifest_hash=MANIFEST_HASH, + raw={ + "report_data": DIGEST + "00" * 32, + "measurement": MEASUREMENT, + "pcr_read": f" 16: 0x{'aa' * 32}", + "runtime_data_binding_verified": True, + "vcek_cert_chain_verified": True, + }, + ) + result = verify_attestation_chain(report, expected_manifest_hash=MANIFEST_HASH) + assert result.report_data_matched is False + assert result.passed is False + + +def test_verify_attestation_chain_no_longer_accepts_a_binding_boolean(): + # the parameter itself must be gone, not merely ignored, so no + # caller (old or new) can be under the impression that supplying it does + # anything. + import inspect + + sig = inspect.signature(verify_attestation_chain) + assert "azure_manifest_binding_verified" not in sig.parameters + + +@pytest.mark.parametrize("unsupported_platform", ["opaque", "", "quantum-tee-v9"]) +def test_unsupported_platform_label_does_not_borrow_the_snp_verifier(unsupported_platform): + # #363 regression matrix: same exact SNP evidence, only the platform + # label changes to an unrecognized/empty/future value. Dispatch must not + # fall through to SNP verification for any of them. + report = _report(report_data_hex=DIGEST + "00" * 32) + report.platform = unsupported_platform + result = verify_attestation_chain(report, expected_manifest_hash=MANIFEST_HASH) + assert result.signature is SignatureStatus.NOT_IMPLEMENTED + assert result.passed is False + assert any("not a supported attestation profile" in r for r in result.reasons) + + +def test_non_azure_snp_still_requires_report_data_to_match(): + # Confirms the fix is scoped to azure-cvm-sev-snp: direct-silicon SNP + # (amd-sev-snp) must still bind REPORT_DATA to the manifest hash directly. + wrong = hashlib.sha256(b"different").hexdigest() + report = _report(report_data_hex=wrong + "00" * 32) + result = verify_attestation_chain(report, expected_manifest_hash=MANIFEST_HASH) + assert result.report_data_matched is False + assert result.passed is False diff --git a/python/tests/test_hw_providers.py b/python/tests/test_hw_providers.py index f1284c4..da12de0 100644 --- a/python/tests/test_hw_providers.py +++ b/python/tests/test_hw_providers.py @@ -174,31 +174,773 @@ def raise_tpm(args): AzureCVMProvider() -def test_azure_verify_manifest_pcr_replay(monkeypatch): - import hashlib +import base64 # noqa: E402 +import hashlib # noqa: E402 +import json # noqa: E402 + +from cryptography.hazmat.primitives import hashes as _hashes # noqa: E402 +from cryptography.hazmat.primitives.asymmetric import padding as _padding # noqa: E402 +from cryptography.hazmat.primitives.asymmetric import rsa as _rsa # noqa: E402 +from cryptography.hazmat.primitives.serialization import ( # noqa: E402 + Encoding as _Encoding, + PublicFormat as _PublicFormat, +) + +from agent_manifest._tpm_verify import ( # noqa: E402 + TPM_GENERATED_VALUE, + TPM_ST_ATTEST_QUOTE, +) + + +def _azure_build_attest( + qualifying_data: bytes, pcr_digest: bytes, *, hash_alg: int = 0x000B, bitmap: bytes = b"\x00\x00\x01" +) -> bytes: + """Minimal structurally-valid TPMS_ATTEST (single PCR-bank selection).""" + out = TPM_GENERATED_VALUE.to_bytes(4, "big") + out += TPM_ST_ATTEST_QUOTE.to_bytes(2, "big") + out += (0).to_bytes(2, "big") # qualifiedSigner: empty TPM2B_NAME + out += len(qualifying_data).to_bytes(2, "big") + qualifying_data + out += b"\x00" * 17 # clockInfo + out += b"\x00" * 8 # firmwareVersion + out += (1).to_bytes(4, "big") # TPML_PCR_SELECTION count + out += hash_alg.to_bytes(2, "big") + out += len(bitmap).to_bytes(1, "big") # sizeofSelect + out += bitmap + out += len(pcr_digest).to_bytes(2, "big") + pcr_digest + return out + + +def _azure_tpmt_sign(ak_key, attest: bytes) -> bytes: + """RSASSA/SHA-256 TPMT_SIGNATURE over ``attest``.""" + sig = ak_key.sign(attest, _padding.PKCS1v15(), _hashes.SHA256()) + return (0x0014).to_bytes(2, "big") + (0x000B).to_bytes(2, "big") + len(sig).to_bytes(2, "big") + sig + + +def _azure_good_fixture(provider, manifest=SAMPLE_MANIFEST): + """Build a fully self-consistent, cryptographically-valid Azure evidence set. + + Returns a dict of the raw-report fields plus the SNP report bytes, all + genuinely tied together: the AK signs a quote over the manifest PCR, the + AK is embedded in runtime_data, and runtime_data is bound (via + REPORT_DATA) into the SNP report -- exactly the composite chain + ``verify_manifest_in_report`` must authenticate. + """ + digest = provider.manifest_hash_value(manifest).split(":", 1)[1] + expected_pcr_value = hashlib.sha256(bytes(32) + bytes.fromhex(digest)).digest() + pcr_digest = hashlib.sha256(expected_pcr_value).digest() + + ak_key = _rsa.generate_private_key(public_exponent=65537, key_size=2048) + ak_pub_pem = ak_key.public_key().public_bytes( + _Encoding.PEM, _PublicFormat.SubjectPublicKeyInfo + ).decode() + + quote_msg = _azure_build_attest(bytes(16), pcr_digest) + quote_sig = _azure_tpmt_sign(ak_key, quote_msg) + + numbers = ak_key.public_key().public_numbers() + modulus_bytes = numbers.n.to_bytes((numbers.n.bit_length() + 7) // 8, "big") + n_b64 = base64.urlsafe_b64encode(modulus_bytes).rstrip(b"=").decode() + runtime_data = json.dumps({"keys": [{"kid": "HCLAkPub", "n": n_b64, "e": "AQAB"}]}).encode() + + report_data = hashlib.sha256(runtime_data).digest() + bytes(32) + snp_report_bytes = _snp_report_with(report_data) + + return { + "digest": digest, + "manifest_hash": f"sha256:{digest}", + "snp_report_bytes": snp_report_bytes, + "raw": { + "ak_pub_pem": ak_pub_pem, + "runtime_data_hex": runtime_data.hex(), + "quote_msg": base64.b64encode(quote_msg).decode(), + "quote_sig": base64.b64encode(quote_sig).decode(), + }, + "ak_key": ak_key, + "pcr_digest": pcr_digest, + } + + +def _azure_report_from_fixture(fx, *, raw_overrides=None, quote_override=None): + raw = dict(fx["raw"]) + if raw_overrides: + raw.update(raw_overrides) + return AttestationReport( + platform="azure-cvm-sev-snp", + manifest_hash=fx["manifest_hash"], + quote=fx["snp_report_bytes"] if quote_override is None else quote_override, + raw=raw, + ) + + +def test_azure_verify_manifest_full_chain_passes(monkeypatch): + """Positive path: every link of the composite chain genuinely verifies.""" + import agent_manifest._hw_providers as hw + from agent_manifest._hw_providers import AzureCVMProvider + + + monkeypatch.setattr(hw, "_run_tpm", lambda args: b"ok") + provider = AzureCVMProvider(pcr_index=16) + fx = _azure_good_fixture(provider) + report = _azure_report_from_fixture(fx) + assert provider.verify_manifest_in_report(report, SAMPLE_MANIFEST) is True + + +def test_azure_verify_manifest_uses_provider_pcr_index_not_self_reported_raw(monkeypatch): + """A provider configured for PCR 17 must check PCR 17 -- and a self-reported + ``raw["pcr_index"]`` claiming a different value must not override that. + Exercises the fix for the gap: the expected index must come from verifier + configuration, not from the report itself. + """ + import agent_manifest._hw_providers as hw + from agent_manifest._hw_providers import AzureCVMProvider + + monkeypatch.setattr(hw, "_run_tpm", lambda args: b"ok") + provider = AzureCVMProvider(pcr_index=17) + # _azure_good_fixture always signs over the PCR16 bitmap regardless of + # provider._pcr, so build the PCR17-selection quote by hand here. + fx = _azure_good_fixture(provider) + quote_msg_17 = _azure_build_attest(bytes(16), fx["pcr_digest"], bitmap=b"\x00\x00\x02") + sig_17 = _azure_tpmt_sign(fx["ak_key"], quote_msg_17) + # Claim (falsely, and irrelevantly) that this report is about PCR 16. + report = _azure_report_from_fixture( + fx, + raw_overrides={ + "quote_msg": base64.b64encode(quote_msg_17).decode(), + "quote_sig": base64.b64encode(sig_17).decode(), + "pcr_index": 16, + }, + ) + # Still passes: the provider's own configured pcr_index (17) is what's + # actually checked, and it genuinely matches the signed selection. + assert provider.verify_manifest_in_report(report, SAMPLE_MANIFEST) is True + + other_provider = AzureCVMProvider(pcr_index=16) + # The same evidence checked against a *differently configured* provider + # (expecting PCR 16, but the quote was signed over PCR 17) must fail -- + # confirming the check is real and not vacuously true. + assert other_provider.verify_manifest_in_report(report, SAMPLE_MANIFEST) is False + +def test_azure_reject(monkeypatch): + """fabricated pcr_read, no AK evidence. + + A report whose ``raw`` carries a matching ``pcr_read`` string but omits + ``ak_pub_pem``, ``quote_msg``, ``quote_sig``, and any runtime-data binding + evidence must never verify -- ``pcr_read`` is not signed by anything. + """ import agent_manifest._hw_providers as hw from agent_manifest._hw_providers import AzureCVMProvider - monkeypatch.setattr(hw, "_run_tpm", lambda args: b"ok") # NV index "present" + monkeypatch.setattr(hw, "_run_tpm", lambda args: b"ok") provider = AzureCVMProvider(pcr_index=16) digest = provider.manifest_hash_value(SAMPLE_MANIFEST).split(":", 1)[1] - # A resettable PCR starts at 0; after one extend it is sha256(0x00*32 || digest). expected_pcr = hashlib.sha256(bytes(32) + bytes.fromhex(digest)).hexdigest() - good = AttestationReport( + report = AttestationReport( platform="azure-cvm-sev-snp", manifest_hash=f"sha256:{digest}", + quote=_snp_report_with(bytes(64)), raw={"pcr_read": f" 16: 0x{expected_pcr.upper()}", "pcr_index": 16}, ) - assert provider.verify_manifest_in_report(good, SAMPLE_MANIFEST) is True + assert provider.verify_manifest_in_report(report, SAMPLE_MANIFEST) is False + + +def test_azure_reject_missing_evidence_entirely(monkeypatch): + import agent_manifest._hw_providers as hw + from agent_manifest._hw_providers import AzureCVMProvider + + monkeypatch.setattr(hw, "_run_tpm", lambda args: b"ok") + provider = AzureCVMProvider(pcr_index=16) + report = AttestationReport(platform="azure-cvm-sev-snp", manifest_hash="sha256:" + "00" * 32, raw={}) + assert provider.verify_manifest_in_report(report, SAMPLE_MANIFEST) is False + + +def test_azure_reject_wrong_pcr_in_quote(monkeypatch): + """A correctly-signed quote over the WRONG PCR value must not pass.""" + import agent_manifest._hw_providers as hw + from agent_manifest._hw_providers import AzureCVMProvider + + monkeypatch.setattr(hw, "_run_tpm", lambda args: b"ok") + provider = AzureCVMProvider(pcr_index=16) + fx = _azure_good_fixture(provider) + wrong_digest = hashlib.sha256(bytes(32) + b"\x00" * 32).digest() + tampered_quote_msg = _azure_build_attest(bytes(16), wrong_digest) + tampered_sig = _azure_tpmt_sign(fx["ak_key"], tampered_quote_msg) + report = _azure_report_from_fixture( + fx, + raw_overrides={ + "quote_msg": base64.b64encode(tampered_quote_msg).decode(), + "quote_sig": base64.b64encode(tampered_sig).decode(), + }, + ) + assert provider.verify_manifest_in_report(report, SAMPLE_MANIFEST) is False + + +def test_azure_reject_wrong_pcr_selection_with_correct_digest_bytes(monkeypatch): + """signed selection bitmap changed from PCR16 to PCR17, + same pcr_digest bytes, re-signed with the runtime-data-bound AK. Checking + only the digest (and not which bank/PCR it was computed over) let this + pass before; it must not pass now. + """ + import agent_manifest._hw_providers as hw + from agent_manifest._hw_providers import AzureCVMProvider + + monkeypatch.setattr(hw, "_run_tpm", lambda args: b"ok") + provider = AzureCVMProvider(pcr_index=16) + fx = _azure_good_fixture(provider) + # PCR17 bitmap: bit 1 of byte 2 (byte*8+bit = 2*8+1 = 17). + retargeted_quote_msg = _azure_build_attest(bytes(16), fx["pcr_digest"], bitmap=b"\x00\x00\x02") + retargeted_sig = _azure_tpmt_sign(fx["ak_key"], retargeted_quote_msg) + report = _azure_report_from_fixture( + fx, + raw_overrides={ + "quote_msg": base64.b64encode(retargeted_quote_msg).decode(), + "quote_sig": base64.b64encode(retargeted_sig).decode(), + }, + ) + assert provider.verify_manifest_in_report(report, SAMPLE_MANIFEST) is False + + +def test_azure_reject_pcr_selection_wrong_bank(monkeypatch): + """Same PCR index (16) but a different bank (SHA-384 alg id) must not pass: + the verifier is configured to require the SHA-256 bank specifically. + """ + import agent_manifest._hw_providers as hw + from agent_manifest._hw_providers import AzureCVMProvider + + monkeypatch.setattr(hw, "_run_tpm", lambda args: b"ok") + provider = AzureCVMProvider(pcr_index=16) + fx = _azure_good_fixture(provider) + wrong_bank_quote_msg = _azure_build_attest(bytes(16), fx["pcr_digest"], hash_alg=0x000C) + wrong_bank_sig = _azure_tpmt_sign(fx["ak_key"], wrong_bank_quote_msg) + report = _azure_report_from_fixture( + fx, + raw_overrides={ + "quote_msg": base64.b64encode(wrong_bank_quote_msg).decode(), + "quote_sig": base64.b64encode(wrong_bank_sig).decode(), + }, + ) + assert provider.verify_manifest_in_report(report, SAMPLE_MANIFEST) is False + + +def test_azure_reject_ak_exponent_mismatch(monkeypatch): + """runtime-data JWK exponent changed to e=3 while the + PEM AK key keeps e=65537, rebound into a freshly-signed SNP report. + Comparing only the modulus let this pass before; both (n, e) must match. + """ + import agent_manifest._hw_providers as hw + from agent_manifest._hw_providers import AzureCVMProvider + + monkeypatch.setattr(hw, "_run_tpm", lambda args: b"ok") + provider = AzureCVMProvider(pcr_index=16) + fx = _azure_good_fixture(provider) + + numbers = fx["ak_key"].public_key().public_numbers() + modulus_bytes = numbers.n.to_bytes((numbers.n.bit_length() + 7) // 8, "big") + n_b64 = base64.urlsafe_b64encode(modulus_bytes).rstrip(b"=").decode() + # e=3 -> base64url("\x03") without padding, i.e. "Aw". + wrong_e_b64 = base64.urlsafe_b64encode((3).to_bytes(1, "big")).rstrip(b"=").decode() + runtime_data_wrong_e = json.dumps( + {"keys": [{"kid": "HCLAkPub", "n": n_b64, "e": wrong_e_b64}]} + ).encode() + + # Rebind REPORT_DATA to the new runtime data so binding step 4 alone + # would pass -- isolating that step 3 (exact key: n AND e) rejects. + snp_report_bytes = _snp_report_with(hashlib.sha256(runtime_data_wrong_e).digest() + bytes(32)) + report = _azure_report_from_fixture( + fx, + raw_overrides={"runtime_data_hex": runtime_data_wrong_e.hex()}, + quote_override=snp_report_bytes, + ) + assert provider.verify_manifest_in_report(report, SAMPLE_MANIFEST) is False + + +@pytest.mark.parametrize( + "corrupt", + [ + pytest.param(lambda b64: "!!!!" + b64, id="illegal-prefix"), + pytest.param(lambda b64: b64 + "!!!!", id="illegal-suffix"), + pytest.param(lambda b64: b64[:-1] + "+", id="standard-base64-plus-alias"), + pytest.param(lambda b64: b64[:-1] + "/", id="standard-base64-slash-alias"), + ], +) +def test_azure_reject_malformed_jwk_modulus_encoding(monkeypatch, corrupt): + """AzureCVMProvider.verify_manifest_in_report twin of + test_azure_report_malformed_jwk_modulus_encoding_fails_closed in + test_attestation_chain.py. base64.urlsafe_b64decode() silently discards + out-of-alphabet characters instead of raising, so an illegally-prefixed/ + suffixed or standard-base64 (+/) JWK "n" would decode to the same bytes + as the genuine value under a naive decoder. n is rebound back into a + freshly hash-bound runtime_data (REPORT_DATA = sha256(runtime_data)) so + every other link in the chain is genuinely valid -- only the JWK + encoding is malformed. Must fail closed, not pass. + """ + import agent_manifest._hw_providers as hw + from agent_manifest._hw_providers import AzureCVMProvider + + monkeypatch.setattr(hw, "_run_tpm", lambda args: b"ok") + provider = AzureCVMProvider(pcr_index=16) + fx = _azure_good_fixture(provider) + + numbers = fx["ak_key"].public_key().public_numbers() + modulus_bytes = numbers.n.to_bytes((numbers.n.bit_length() + 7) // 8, "big") + good_n_b64 = base64.urlsafe_b64encode(modulus_bytes).rstrip(b"=").decode() + corrupt_n_b64 = corrupt(good_n_b64) + + corrupt_runtime_data = json.dumps( + {"keys": [{"kid": "HCLAkPub", "n": corrupt_n_b64, "e": "AQAB"}]} + ).encode() + snp_report_bytes = _snp_report_with(hashlib.sha256(corrupt_runtime_data).digest() + bytes(32)) + report = _azure_report_from_fixture( + fx, + raw_overrides={"runtime_data_hex": corrupt_runtime_data.hex()}, + quote_override=snp_report_bytes, + ) + assert provider.verify_manifest_in_report(report, SAMPLE_MANIFEST) is False + + +@pytest.mark.parametrize( + "corrupt", + [ + pytest.param(lambda b64: "!!!!" + b64, id="illegal-prefix"), + pytest.param(lambda b64: b64 + "!!!!", id="illegal-suffix"), + pytest.param(lambda b64: b64[:-1] + "+", id="standard-base64-plus-alias"), + pytest.param(lambda b64: b64[:-1] + "/", id="standard-base64-slash-alias"), + ], +) +def test_azure_reject_malformed_jwk_exponent_encoding(monkeypatch, corrupt): + """Twin of test_azure_reject_malformed_jwk_modulus_encoding, corrupting + the JWK "e" member instead of "n". Both members go through the same + _strict_b64url_decode() call inside ak_public_numbers_from_runtime_data(); + the modulus test alone does not prove the exponent path is covered here + (AzureCVMProvider.verify_manifest_in_report), since a future edit could + special-case or bypass strict decoding for "e" specifically without this + test noticing. n is kept well-formed so only the exponent encoding is + under test. Must fail closed, not pass. + """ + import agent_manifest._hw_providers as hw + from agent_manifest._hw_providers import AzureCVMProvider + + monkeypatch.setattr(hw, "_run_tpm", lambda args: b"ok") + provider = AzureCVMProvider(pcr_index=16) + fx = _azure_good_fixture(provider) + + numbers = fx["ak_key"].public_key().public_numbers() + modulus_bytes = numbers.n.to_bytes((numbers.n.bit_length() + 7) // 8, "big") + good_n_b64 = base64.urlsafe_b64encode(modulus_bytes).rstrip(b"=").decode() + good_e_b64 = base64.urlsafe_b64encode( + numbers.e.to_bytes((numbers.e.bit_length() + 7) // 8, "big") + ).rstrip(b"=").decode() + corrupt_e_b64 = corrupt(good_e_b64) + + corrupt_runtime_data = json.dumps( + {"keys": [{"kid": "HCLAkPub", "n": good_n_b64, "e": corrupt_e_b64}]} + ).encode() + snp_report_bytes = _snp_report_with(hashlib.sha256(corrupt_runtime_data).digest() + bytes(32)) + report = _azure_report_from_fixture( + fx, + raw_overrides={"runtime_data_hex": corrupt_runtime_data.hex()}, + quote_override=snp_report_bytes, + ) + assert provider.verify_manifest_in_report(report, SAMPLE_MANIFEST) is False + + +def test_azure_reject_malformed_runtime_data_keys_shape(monkeypatch): + """valid JSON `{"keys": 1}`, correctly hash-bound into + the signed SNP report, must fail closed (return False) rather than raise. + """ + import agent_manifest._hw_providers as hw + from agent_manifest._hw_providers import AzureCVMProvider + + monkeypatch.setattr(hw, "_run_tpm", lambda args: b"ok") + provider = AzureCVMProvider(pcr_index=16) + fx = _azure_good_fixture(provider) + + malformed_runtime_data = json.dumps({"keys": 1}).encode() + snp_report_bytes = _snp_report_with(hashlib.sha256(malformed_runtime_data).digest() + bytes(32)) + report = _azure_report_from_fixture( + fx, + raw_overrides={"runtime_data_hex": malformed_runtime_data.hex()}, + quote_override=snp_report_bytes, + ) + # Must not raise TypeError -- must simply return False. + assert provider.verify_manifest_in_report(report, SAMPLE_MANIFEST) is False + + +@pytest.mark.parametrize( + "field,bad_value", + [ + ("quote_msg", 12345), + ("quote_msg", ["not", "a", "string"]), + ("quote_sig", 12345), + ("quote_sig", {"unexpected": "dict"}), + ("ak_pub_pem", 12345), + ("ak_pub_pem", b"already-bytes-not-str"), + ("runtime_data_hex", 12345), + ("runtime_data_hex", ["not", "hex"]), + ], +) +def test_azure_reject_wrong_type_raw_field(monkeypatch, field, bad_value): + """Attacker-forged evidence can put any JSON type in report.raw -- an int, + list, or dict is just as reachable as a malformed string. Each of these is + truthy, so a truthiness-only guard lets it through to + base64.b64decode()/bytes.fromhex(), which raise TypeError (not + binascii.Error/ValueError) for non-str/bytes input. The function's + documented contract is that malformed input returns False and never + raises; a wrong-type field must fail closed, not crash the caller. + """ + import agent_manifest._hw_providers as hw + from agent_manifest._hw_providers import AzureCVMProvider + + monkeypatch.setattr(hw, "_run_tpm", lambda args: b"ok") + provider = AzureCVMProvider(pcr_index=16) + fx = _azure_good_fixture(provider) + report = _azure_report_from_fixture(fx, raw_overrides={field: bad_value}) + # Must not raise TypeError -- must simply return False. + assert provider.verify_manifest_in_report(report, SAMPLE_MANIFEST) is False + + +@pytest.mark.parametrize( + "bad_value", + [12345, "not-bytes-a-string", ["not", "bytes"], {"unexpected": "dict"}], +) +def test_azure_reject_wrong_type_snp_report_bytes(monkeypatch, bad_value): + """snp_report_bytes is the raw SNP report attached to the report object + (``report.quote``), not a report.raw JSON field, but it is equally + attacker-controlled evidence. A wrong type reaches + ``parse_snp_report()``, whose ``len(report)`` / struct-unpack calls raise + TypeError for non-bytes-like input -- not the ``SnpVerificationError`` + the caller catches. Must fail closed, not raise. + """ + import agent_manifest._hw_providers as hw + from agent_manifest._hw_providers import AzureCVMProvider + + monkeypatch.setattr(hw, "_run_tpm", lambda args: b"ok") + provider = AzureCVMProvider(pcr_index=16) + fx = _azure_good_fixture(provider) + report = _azure_report_from_fixture(fx, quote_override=bad_value) + assert provider.verify_manifest_in_report(report, SAMPLE_MANIFEST) is False + + +@pytest.mark.parametrize( + "bad_value", + [12345, None, ["sha256:" + "aa" * 32], {"hash": "sha256:" + "aa" * 32}, b"sha256:" + b"aa" * 32], +) +def test_azure_reject_wrong_type_expected_manifest_hash(monkeypatch, bad_value): + """``expected_manifest_hash`` is a top-level argument of + ``verify_azure_manifest_binding`` just like the report-derived evidence + fields -- a misconfigured or programmatically-wrong-typed caller is just + as real a source of a wrong-type value here as a forged report is for + the other fields. ``expected_manifest_hash.split(":", 1)`` raises + ``AttributeError`` for a non-str value (``None``, ``int``, ``list``, + ``dict``) and ``TypeError`` for ``bytes`` (``bytes.split`` requires a + ``bytes`` separator, not the ``str`` ``":"`` literal used here) -- both + uncaught previously. Must fail closed, not raise. + """ + import agent_manifest._hw_providers as hw + from agent_manifest._azure_verify import verify_azure_manifest_binding + from agent_manifest._hw_providers import AzureCVMProvider + + monkeypatch.setattr(hw, "_run_tpm", lambda args: b"ok") + provider = AzureCVMProvider(pcr_index=16) + fx = _azure_good_fixture(provider) + result = verify_azure_manifest_binding( + expected_manifest_hash=bad_value, + expected_pcr_index=16, + quote_msg_b64=fx["raw"]["quote_msg"], + quote_sig_b64=fx["raw"]["quote_sig"], + ak_pub_pem=fx["raw"]["ak_pub_pem"], + runtime_data_hex=fx["raw"]["runtime_data_hex"], + snp_report_bytes=fx["snp_report_bytes"], + ) + assert result is False + + +@pytest.mark.parametrize( + "bad_hash", + [ + "md5:" + "aa" * 32, # right length, wrong algorithm name + "sha1:" + "aa" * 32, # right length, wrong algorithm name + "foo:" + "aa" * 32, # nonsense algorithm name + "sha256:", # prefix only, no digest + "sha256:" + "aa" * 31, # 62 hex chars, one short + "sha256:" + "aa" * 33, # 66 hex chars, one too many + "sha256:" + "zz" * 32, # right length, invalid hex + "aa" * 32, # no prefix at all + " sha256:" + "aa" * 32, # leading whitespace + "sha256:" + "aa" * 32 + " ", # trailing whitespace + "sha256:" + "AA" * 32 + "\n", # trailing newline after otherwise-valid hex + ], +) +def test_azure_reject_malformed_expected_manifest_hash_prefix(monkeypatch, bad_hash): + """``expected_manifest_hash`` must be exactly ``"sha256:" + 64 hex chars``. + + A verifier that merely does ``split(":", 1)[-1]`` implicitly assumes + "whatever follows a colon is the SHA-256 digest" -- so ``"md5:<64 + hex>"`` or ``"foo:<64 hex>"`` pass just as readily as a genuine + ``"sha256:<64 hex>"`` value, leaving the hash algorithm ambiguous and + caller-controlled at the exact point this value is baked into the + manifest-to-PCR security binding. Must fail closed for every + malformed/wrong-algorithm/wrong-length/non-hex/whitespace-padded shape, + even when a genuinely-matching PCR/quote/AK/runtime-data chain is + supplied for every other field. + """ + import agent_manifest._hw_providers as hw + from agent_manifest._azure_verify import verify_azure_manifest_binding + from agent_manifest._hw_providers import AzureCVMProvider + + monkeypatch.setattr(hw, "_run_tpm", lambda args: b"ok") + provider = AzureCVMProvider(pcr_index=16) + fx = _azure_good_fixture(provider) + result = verify_azure_manifest_binding( + expected_manifest_hash=bad_hash, + expected_pcr_index=16, + quote_msg_b64=fx["raw"]["quote_msg"], + quote_sig_b64=fx["raw"]["quote_sig"], + ak_pub_pem=fx["raw"]["ak_pub_pem"], + runtime_data_hex=fx["raw"]["runtime_data_hex"], + snp_report_bytes=fx["snp_report_bytes"], + ) + assert result is False - bad = AttestationReport( + +def test_azure_accept_well_formed_expected_manifest_hash(monkeypatch): + """Sanity check for the positive case: a genuine ``"sha256:" + 64 hex`` + value (matching the rest of the evidence chain) must still verify -- + the stricter prefix check must not reject well-formed input.""" + import agent_manifest._hw_providers as hw + from agent_manifest._azure_verify import verify_azure_manifest_binding + from agent_manifest._hw_providers import AzureCVMProvider + + monkeypatch.setattr(hw, "_run_tpm", lambda args: b"ok") + provider = AzureCVMProvider(pcr_index=16) + fx = _azure_good_fixture(provider) + result = verify_azure_manifest_binding( + expected_manifest_hash=fx["manifest_hash"], + expected_pcr_index=16, + quote_msg_b64=fx["raw"]["quote_msg"], + quote_sig_b64=fx["raw"]["quote_sig"], + ak_pub_pem=fx["raw"]["ak_pub_pem"], + runtime_data_hex=fx["raw"]["runtime_data_hex"], + snp_report_bytes=fx["snp_report_bytes"], + ) + assert result is True + + +@pytest.mark.parametrize( + "bad_value", + [[16], {16}, {"pcr": 16}, "16", 16.0, True], +) +def test_azure_reject_wrong_type_expected_pcr_index(monkeypatch, bad_value): + """``expected_pcr_index`` is later put into ``frozenset({expected_pcr_index})`` + for a PCR-selection comparison. An unhashable value (``list``, ``set``, + ``dict``) raises ``TypeError`` constructing that frozenset -- uncaught + previously, since it happens outside any ``try`` in the function. Other + wrong types (``str``, ``float``, ``bool``) are hashable so would not + crash the frozenset call, but are still not the ``int`` the function + contract requires and must not silently coerce into passing; explicit + type validation rejects all of them up front. Must fail closed, not + raise, for every case. + """ + import agent_manifest._hw_providers as hw + from agent_manifest._azure_verify import verify_azure_manifest_binding + from agent_manifest._hw_providers import AzureCVMProvider + + monkeypatch.setattr(hw, "_run_tpm", lambda args: b"ok") + provider = AzureCVMProvider(pcr_index=16) + fx = _azure_good_fixture(provider) + result = verify_azure_manifest_binding( + expected_manifest_hash=fx["manifest_hash"], + expected_pcr_index=bad_value, + quote_msg_b64=fx["raw"]["quote_msg"], + quote_sig_b64=fx["raw"]["quote_sig"], + ak_pub_pem=fx["raw"]["ak_pub_pem"], + runtime_data_hex=fx["raw"]["runtime_data_hex"], + snp_report_bytes=fx["snp_report_bytes"], + ) + assert result is False + + +@pytest.mark.parametrize( + "bad_raw", + ["attacker-controlled-string", 12345, [1, 2, 3], b"attacker-controlled-bytes"], +) +def test_azure_reject_non_dict_raw(monkeypatch, bad_raw): + """``AttestationReport.raw`` is typed ``dict[str, Any]``, but dataclasses do + not enforce field types at runtime -- a caller building (or forging) a + report can set ``raw`` to any truthy non-dict value just as easily as any + other field. ``raw = report.raw or {}`` only guards the falsy cases + (``None``, ``{}``, ``""``); a truthy non-dict still reaches + ``raw.get(...)`` and raises ``AttributeError``, breaking this method's + documented fail-closed contract. Must return False, never raise. + """ + import agent_manifest._hw_providers as hw + from agent_manifest._hw_providers import AzureCVMProvider + + monkeypatch.setattr(hw, "_run_tpm", lambda args: b"ok") + provider = AzureCVMProvider(pcr_index=16) + report = AttestationReport( platform="azure-cvm-sev-snp", - manifest_hash=f"sha256:{digest}", - raw={"pcr_read": " 16: 0x" + "00" * 32, "pcr_index": 16}, + manifest_hash="sha256:" + "aa" * 32, + quote=b"x", + raw=bad_raw, ) - assert provider.verify_manifest_in_report(bad, SAMPLE_MANIFEST) is False + assert provider.verify_manifest_in_report(report, SAMPLE_MANIFEST) is False + + +def test_azure_reject_tampered_signature(monkeypatch): + """A valid quote_msg with a corrupted quote_sig must not pass.""" + import agent_manifest._hw_providers as hw + from agent_manifest._hw_providers import AzureCVMProvider + + monkeypatch.setattr(hw, "_run_tpm", lambda args: b"ok") + provider = AzureCVMProvider(pcr_index=16) + fx = _azure_good_fixture(provider) + sig_bytes = bytearray(base64.b64decode(fx["raw"]["quote_sig"])) + sig_bytes[-1] ^= 0xFF + report = _azure_report_from_fixture( + fx, raw_overrides={"quote_sig": base64.b64encode(bytes(sig_bytes)).decode()} + ) + assert provider.verify_manifest_in_report(report, SAMPLE_MANIFEST) is False + + +def test_azure_reject_missing_ak_evidence(monkeypatch): + """Signature/quote present but ak_pub_pem omitted -- cannot pass.""" + import agent_manifest._hw_providers as hw + from agent_manifest._hw_providers import AzureCVMProvider + + monkeypatch.setattr(hw, "_run_tpm", lambda args: b"ok") + provider = AzureCVMProvider(pcr_index=16) + fx = _azure_good_fixture(provider) + report = _azure_report_from_fixture(fx, raw_overrides={"ak_pub_pem": None}) + assert provider.verify_manifest_in_report(report, SAMPLE_MANIFEST) is False + + +def test_azure_reject_ak_not_bound_in_runtime_data(monkeypatch): + """ak_pub_pem is a genuine key, but a DIFFERENT key is embedded in runtime_data.""" + import agent_manifest._hw_providers as hw + from agent_manifest._hw_providers import AzureCVMProvider + + monkeypatch.setattr(hw, "_run_tpm", lambda args: b"ok") + provider = AzureCVMProvider(pcr_index=16) + fx = _azure_good_fixture(provider) + + other_key = _rsa.generate_private_key(public_exponent=65537, key_size=2048) + other_numbers = other_key.public_key().public_numbers() + other_modulus = other_numbers.n.to_bytes((other_numbers.n.bit_length() + 7) // 8, "big") + other_n_b64 = base64.urlsafe_b64encode(other_modulus).rstrip(b"=").decode() + swapped_runtime_data = json.dumps( + {"keys": [{"kid": "HCLAkPub", "n": other_n_b64, "e": "AQAB"}]} + ).encode() + + # Keep REPORT_DATA consistent with the swapped runtime_data so binding + # step 4 alone would pass -- isolating that step 3 (AK identity) rejects. + snp_report_bytes = _snp_report_with(hashlib.sha256(swapped_runtime_data).digest() + bytes(32)) + report = _azure_report_from_fixture( + fx, + raw_overrides={"runtime_data_hex": swapped_runtime_data.hex()}, + quote_override=snp_report_bytes, + ) + assert provider.verify_manifest_in_report(report, SAMPLE_MANIFEST) is False + + +def test_azure_reject_runtime_data_not_bound_in_snp_report(monkeypatch): + """runtime_data/AK are internally consistent, but REPORT_DATA doesn't hash-bind to it.""" + import agent_manifest._hw_providers as hw + from agent_manifest._hw_providers import AzureCVMProvider + + monkeypatch.setattr(hw, "_run_tpm", lambda args: b"ok") + provider = AzureCVMProvider(pcr_index=16) + fx = _azure_good_fixture(provider) + wrong_snp_report = _snp_report_with(bytes(64)) # REPORT_DATA all zero + report = _azure_report_from_fixture(fx, quote_override=wrong_snp_report) + assert provider.verify_manifest_in_report(report, SAMPLE_MANIFEST) is False + + +def test_azure_reject_freshness_nonce_replayed_as_boot_quote(monkeypatch): + """A quote with a non-zero qualifying value (a runtime-attestation quote) is not boot proof.""" + import agent_manifest._hw_providers as hw + from agent_manifest._hw_providers import AzureCVMProvider + + monkeypatch.setattr(hw, "_run_tpm", lambda args: b"ok") + provider = AzureCVMProvider(pcr_index=16) + fx = _azure_good_fixture(provider) + quote_msg = _azure_build_attest(b"\x01" * 16, fx["pcr_digest"]) + quote_sig = _azure_tpmt_sign(fx["ak_key"], quote_msg) + report = _azure_report_from_fixture( + fx, + raw_overrides={ + "quote_msg": base64.b64encode(quote_msg).decode(), + "quote_sig": base64.b64encode(quote_sig).decode(), + }, + ) + assert provider.verify_manifest_in_report(report, SAMPLE_MANIFEST) is False + + +def test_azure_get_attestation_report_has_no_stale_verification_flags(monkeypatch): + """get_attestation_report() must never hand out a boolean that claims a + check ran when it didn't (#373 nitpick). Regression guard: an earlier + revision hardcoded raw["runtime_data_binding_verified"] = True + unconditionally, on every single report, whether or not anything about + that binding had actually been checked at that point (it hadn't -- + that's verify_attestation_chain's job, done later, from the quote/sig/AK + material also on this report). Nothing in the verification code path + ever reads that key (confirmed by grep across src/), so it was pure + dead-but-misleading metadata. It must not come back. + """ + import agent_manifest._hw_providers as hw + + provider = hw.AzureCVMProvider.__new__(hw.AzureCVMProvider) + provider._pcr = 16 + provider._manifest_hash = "sha256:" + "cd" * 32 + + fake_blobs = { + "quote_msg": "bXNn", + "quote_sig": "c2ln", + "quote_pcrs": "cGNycw==", + "ak_pub_pem": "-----BEGIN PUBLIC KEY-----\nfake\n-----END PUBLIC KEY-----\n", + "snp_report": (b"\x00" * 0x1A0).hex(), + "runtime_data_hex": b"{}".hex(), + "measurement": "ab" * 48, + "report_data": "cd" * 64, + } + monkeypatch.setattr(provider, "_quote", lambda nonce_hex: fake_blobs) + monkeypatch.setattr(hw, "_run_tpm", lambda args: b"16: 0x" + b"cd" * 32) + + report = provider.get_attestation_report() + + assert "runtime_data_binding_verified" not in report.raw + assert "vcek_cert_chain_verified" not in report.raw + # The fields verify_attestation_chain / verify_azure_manifest_binding + # actually consume must still be there. + for key in ("quote_msg", "quote_sig", "ak_pub_pem", "runtime_data_hex", "report_data", "measurement"): + assert key in report.raw + + +def test_azure_reject_caller_supplied_boolean_alone_is_not_evidence(): + """Freely constructing an AttestationReport and asserting truthy fields proves nothing. + + There is no boolean shortcut anywhere in the raw schema: constructing a + report with only descriptive/legacy fields set to favorable-looking + values (no actual quote/signature/AK material) must fail closed. + """ + + provider_pcr = 16 + report = AttestationReport( + platform="azure-cvm-sev-snp", + manifest_hash="sha256:" + "ab" * 32, + raw={ + "runtime_data_binding_verified": True, # self-reported, must not be trusted + "vcek_cert_chain_verified": True, + "pcr_index": provider_pcr, + }, + ) + import agent_manifest._hw_providers as hw + + az = hw.AzureCVMProvider.__new__(hw.AzureCVMProvider) + az._pcr = provider_pcr + az._manifest_hash = None + assert az.verify_manifest_in_report(report, SAMPLE_MANIFEST) is False # --------------------------------------------------------------------------- diff --git a/python/tests/test_signing.py b/python/tests/test_signing.py index 001b87e..336239a 100644 --- a/python/tests/test_signing.py +++ b/python/tests/test_signing.py @@ -205,6 +205,26 @@ def test_b64url_decode_accepts_valid(): assert result == b"Hello" +def test_b64url_decode_rejects_trailing_newline(): + # Python's `$` regex anchor matches at end-of-string OR just before a + # single trailing '\n'. A naive `^[...]*$` CRYPTO-006 check would let + # "AQAB\n" through even though '\n' isn't URL-safe-base64, and + # base64.urlsafe_b64decode() would then silently drop the '\n' and + # decode it to the exact same bytes as "AQAB" -- defeating the guard. + # (Chosen deliberately as a 4-char, unpadded base -- for other lengths + # the stale padding arithmetic happens to raise "Incorrect padding" for + # an unrelated reason, which would make this test pass against the old, + # vulnerable regex too and not actually catch a regression.) + assert _b64url_decode("AQAB") == bytes.fromhex("010001") + with pytest.raises(ValueError, match="non-URL-safe"): + _b64url_decode("AQAB\n") + + +def test_b64url_decode_rejects_non_str(): + with pytest.raises(ValueError, match="non-URL-safe"): + _b64url_decode(None) # type: ignore[arg-type] + + def test_ed25519_public_key_roundtrip_b64url(): kp = generate_ed25519() b64 = kp.public_b64url() diff --git a/python/tests/test_tpm_verify.py b/python/tests/test_tpm_verify.py index 25ad61d..316a505 100644 --- a/python/tests/test_tpm_verify.py +++ b/python/tests/test_tpm_verify.py @@ -175,6 +175,24 @@ def test_parse_extracts_fields(): assert q.pcr_digest == PCR +def test_parse_captures_pcr_selection_bank_and_indices(): + # The bank + PCR bitmap must actually be captured, not silently skipped: + # pcr_digest alone doesn't say which PCR(s) it is a digest of. + q = parse_tpm_quote(_build_attest(NONCE, PCR)) + assert len(q.pcr_selections) == 1 + selection = q.pcr_selections[0] + assert selection.hash_alg == 0x000B # TPM_ALG_SHA256 + assert selection.indices() == frozenset({16}) # bitmap 0x00 0x00 0x01 -> PCR16 + + +def test_pcr_selection_indices_multiple_bits(): + from agent_manifest._tpm_verify import PcrSelection + + # byte0 bits 0 and 7 -> PCR0 and PCR7; byte2 bit 1 -> PCR17. + sel = PcrSelection(hash_alg=0x000B, pcr_select=b"\x81\x00\x02") + assert sel.indices() == frozenset({0, 7, 17}) + + def test_parse_rejects_truncated(): with pytest.raises(TpmVerificationError): parse_tpm_quote(b"\xff") @@ -354,6 +372,27 @@ def test_verify_accepts_valid(kind): ) is True +def test_ak_signature_discriminator_uses_key_shape_not_sig_alg_prefix_bytes(): + # Regression for the merge-order gap the reviewer flagged: an earlier + # revision discriminated a legacy bare signature from a TPMT_SIGNATURE + # envelope by sniffing the leading two bytes against known sigAlg ids + # (0x0014/0x0016/0x0018). Those bytes are untrusted signature content, not + # a length-prefixed framing marker -- a bare RSA signature can coincide + # with one by chance. The discriminator must instead be the AK's own + # signature shape: exactly one-modulus-width bytes for RSA, or a + # decodable DER SEQUENCE for ECDSA. + from agent_manifest._tpm_verify import verify_ak_signature + + ak_key, chain, roots = _ak_chain("rsa") + attest = _build_attest(NONCE, PCR) + bare_sig = _sign(ak_key, attest) + # A bare RSA signature is exactly modulus-width; it is never mistaken for + # an envelope merely because its first two bytes happen to equal 0x0014. + assert len(bare_sig) == 2048 // 8 + ak_public_key = x509.load_pem_x509_certificates(chain)[0].public_key() + assert verify_ak_signature(ak_public_key, attest, bare_sig) is True + + def test_verify_rejects_tampered_attest(): ak_key, chain, roots = _ak_chain() attest = bytearray(_build_attest(NONCE, PCR)) diff --git a/python/tests/test_trace_verify.py b/python/tests/test_trace_verify.py index c79b568..d91636f 100644 --- a/python/tests/test_trace_verify.py +++ b/python/tests/test_trace_verify.py @@ -142,6 +142,34 @@ def test_valid_envelope_verifies_and_is_admissible(kp, trusted): assert result.trace_id == envelope["trace_id"] +@pytest.mark.parametrize( + "field,corrupt_value", + [ + ("policy_hash", "sha256:" + "a1" * 32 + "\n"), + ("catalog_hash", "sha256:" + "b2" * 32 + "\n"), + ("agent_id", "spiffe://example.org/agent/billing\n"), + ("trace_id", "0192f3a0-0000-7000-8000-00000000000a\n"), + ], +) +def test_trailing_newline_on_a_formatted_field_is_rejected(kp, trusted, field, corrupt_value): + # `_envelope_format_failures` validates trace_id/policy_hash/etc against + # `^...$`-anchored regexes via `.fullmatch()`. Historically, if any of + # those call sites had used `.match()` instead, Python's `$` (which + # matches at end-of-string OR just before a single trailing '\n') + # would have let "\n" through as if it were the + # unmodified value. Sign a real envelope with the corrupted, `\n`- + # suffixed value (so the signature is genuinely valid over the exact + # corrupted field) and confirm the format check still catches it. + envelope = _sign_envelope(_envelope(**{field: corrupt_value}), kp) + result = verify_trace_envelope(envelope, trusted_keys=trusted) + # `_envelope_format_failures` runs (and short-circuits) before signature + # verification, so a format-rejected envelope never reaches VERIFIED. + assert result.status is TraceStatus.MALFORMED + assert result.signature_verified is False + assert result.admissible is False + assert any(f.startswith(("not_a_", "illegal_")) for f in result.failures) + + @pytest.mark.parametrize( "field,value", [ diff --git a/python/tests/test_types.py b/python/tests/test_types.py new file mode 100644 index 0000000..400196a --- /dev/null +++ b/python/tests/test_types.py @@ -0,0 +1,64 @@ +"""Regression tests for the custom scalar validators in ``_types.py``. + +Covers a specific Python regex gotcha: ``$`` matches at end-of-string OR +just before a single trailing ``'\n'``. A validator built as +``re.compile(r"^...$").match(v)`` would therefore accept +``"\n"`` as if it were the unmodified, valid value. These +validators must use ``.fullmatch()`` (which has no such exception) so a +trailing newline is rejected like any other malformed suffix. +""" + +import pytest + +from agent_manifest._types import HashValue, ManifestId + +GOOD_MANIFEST_ID = "01890a5d-ac96-774b-bcce-b302099a8057" +GOOD_HASH = "sha256:" + "a" * 64 + + +def test_manifest_id_accepts_well_formed_value(): + assert ManifestId._validate(GOOD_MANIFEST_ID) == GOOD_MANIFEST_ID + + +def test_manifest_id_rejects_trailing_newline(): + with pytest.raises(ValueError, match="not a valid UUID v7"): + ManifestId._validate(GOOD_MANIFEST_ID + "\n") + + +def test_manifest_id_rejects_illegal_prefix_and_suffix(): + with pytest.raises(ValueError, match="not a valid UUID v7"): + ManifestId._validate("!!!!" + GOOD_MANIFEST_ID) + with pytest.raises(ValueError, match="not a valid UUID v7"): + ManifestId._validate(GOOD_MANIFEST_ID + "!!!!") + + +def test_hash_value_accepts_well_formed_value(): + assert HashValue._validate(GOOD_HASH) == GOOD_HASH + + +def test_hash_value_rejects_trailing_newline(): + with pytest.raises(ValueError, match="Invalid hash value"): + HashValue._validate(GOOD_HASH + "\n") + + +def test_hash_value_rejects_illegal_prefix_and_suffix(): + with pytest.raises(ValueError, match="Invalid hash value"): + HashValue._validate("!!!!" + GOOD_HASH) + with pytest.raises(ValueError, match="Invalid hash value"): + HashValue._validate(GOOD_HASH + "!!!!") + + +def test_manifest_id_json_schema_pattern_still_anchored(): + # The exported JSON-schema pattern is consumed by non-Python tools + # following JSON Schema/ECMA 262 semantics, where a `pattern` without + # `^`/`$` anchors means "contains a match" rather than "matches + # exactly". The anchors must stay in the exported string even though + # internal validation now uses `.fullmatch()` instead of relying on + # them. + assert ManifestId._PATTERN.pattern.startswith("^") + assert ManifestId._PATTERN.pattern.endswith("$") + + +def test_hash_value_json_schema_pattern_still_anchored(): + assert HashValue._PATTERN.pattern.startswith("^") + assert HashValue._PATTERN.pattern.endswith("$")