diff --git a/.github/workflows/codeql.yml b/.github/workflows/codeql.yml index 8743fe32..ecb8832f 100644 --- a/.github/workflows/codeql.yml +++ b/.github/workflows/codeql.yml @@ -38,6 +38,5 @@ jobs: - name: Perform CodeQL Analysis uses: github/codeql-action/analyze@v4.37.7 - continue-on-error: true with: category: /language:python diff --git a/tests/unit/test_azure_cvm_verify.py b/tests/unit/test_azure_cvm_verify.py index 7703273c..ce863c1a 100644 --- a/tests/unit/test_azure_cvm_verify.py +++ b/tests/unit/test_azure_cvm_verify.py @@ -6,6 +6,7 @@ real attestation; it proves the verifier accepts a well-formed chain and fails closed on tampering. A real-hardware fixture test is env-gated at the bottom. """ + from __future__ import annotations import base64 @@ -37,7 +38,7 @@ def _name(cn: str) -> x509.Name: return x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, cn)]) -def _cert(subject, issuer_name, subject_pub, issuer_key): +def _cert(subject, issuer_name, subject_pub, issuer_key, *, is_ca: bool = False): now = datetime.now(UTC) return ( x509.CertificateBuilder() @@ -47,6 +48,7 @@ def _cert(subject, issuer_name, subject_pub, issuer_key): .serial_number(x509.random_serial_number()) .not_valid_before(now - timedelta(days=1)) .not_valid_after(now + timedelta(days=3650)) + .add_extension(x509.BasicConstraints(ca=is_ca, path_length=None), critical=True) .sign(issuer_key, hashes.SHA384()) ) @@ -55,8 +57,8 @@ def _synthetic_chain(): ark_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) ask_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) vcek_key = ec.generate_private_key(ec.SECP384R1()) - ark = _cert(_name("ARK"), _name("ARK"), ark_key.public_key(), ark_key) - ask = _cert(_name("ASK"), _name("ARK"), ask_key.public_key(), ark_key) + ark = _cert(_name("ARK"), _name("ARK"), ark_key.public_key(), ark_key, is_ca=True) + ask = _cert(_name("ASK"), _name("ARK"), ask_key.public_key(), ark_key, is_ca=True) vcek = _cert(_name("VCEK"), _name("ASK"), vcek_key.public_key(), ask_key) chain_pem = ( vcek.public_bytes(Encoding.PEM) @@ -67,17 +69,23 @@ def _synthetic_chain(): def _b64u_int(n: int) -> str: - return base64.urlsafe_b64encode(n.to_bytes((n.bit_length() + 7) // 8 or 1, "big")).rstrip(b"=").decode() + return ( + base64.urlsafe_b64encode(n.to_bytes((n.bit_length() + 7) // 8 or 1, "big")) + .rstrip(b"=") + .decode() + ) def _runtime_data(ak_pub: rsa.RSAPublicKey) -> bytes: pn = ak_pub.public_numbers() - return json.dumps({ - "keys": [{"kid": "HCLAkPub", "kty": "RSA", "e": _b64u_int(pn.e), "n": _b64u_int(pn.n)}] - }).encode() + return json.dumps( + {"keys": [{"kid": "HCLAkPub", "kty": "RSA", "e": _b64u_int(pn.e), "n": _b64u_int(pn.n)}]} + ).encode() -def _signed_snp(vcek_key, runtime: bytes, measurement_bytes: bytes = b"\x11" * 48) -> tuple[bytes, str]: +def _signed_snp( + vcek_key, runtime: bytes, measurement_bytes: bytes = b"\x11" * 48 +) -> tuple[bytes, str]: buf = bytearray(_REPORT_SIZE) buf[0x00:0x04] = (2).to_bytes(4, "little") buf[_SIG_ALGO_OFFSET : _SIG_ALGO_OFFSET + 4] = (1).to_bytes(4, "little") @@ -98,7 +106,8 @@ def _tpm2b_attest(extra_data: bytes) -> bytes: struct.pack(">I", 0xFF544347) # magic + struct.pack(">H", 0x8018) # type: TPM_ST_ATTEST_QUOTE + struct.pack(">H", 0) # qualifiedSigner (empty TPM2B) - + struct.pack(">H", len(extra_data)) + extra_data # extraData + + struct.pack(">H", len(extra_data)) + + extra_data # extraData + b"\x00" * 40 # clockInfo + firmwareVersion + attested (unparsed tail) ) return struct.pack(">H", len(body)) + body @@ -114,19 +123,23 @@ def _build_evidence(nonce: bytes, *, include_chain: bool = True, ak_key=None, qu ak_key = ak_key or rsa.generate_private_key(public_exponent=65537, key_size=2048) runtime = _runtime_data(ak_key.public_key()) snp, measurement = _signed_snp(vcek_key, runtime) - quote_msg = _tpm2b_attest(quote_extra if quote_extra is not None else hashlib.sha256(nonce).digest()) + quote_msg = _tpm2b_attest( + quote_extra if quote_extra is not None else hashlib.sha256(nonce).digest() + ) quote_sig = _tpmt_signature(ak_key, quote_msg) - envelope = json.dumps({ - "v": 1, - "snp_report": base64.b64encode(snp).decode(), - "runtime_data": base64.b64encode(runtime).decode(), - "quote_msg": base64.b64encode(quote_msg).decode(), - "quote_sig": base64.b64encode(quote_sig).decode(), - "ak_pub_pem": ak_key.public_key().public_bytes( - Encoding.PEM, PublicFormat.SubjectPublicKeyInfo - ).decode(), - "vcek_chain_pem": base64.b64encode(chain_pem if include_chain else b"").decode(), - }).encode() + envelope = json.dumps( + { + "v": 1, + "snp_report": base64.b64encode(snp).decode(), + "runtime_data": base64.b64encode(runtime).decode(), + "quote_msg": base64.b64encode(quote_msg).decode(), + "quote_sig": base64.b64encode(quote_sig).decode(), + "ak_pub_pem": ak_key.public_key() + .public_bytes(Encoding.PEM, PublicFormat.SubjectPublicKeyInfo) + .decode(), + "vcek_chain_pem": base64.b64encode(chain_pem if include_chain else b"").decode(), + } + ).encode() return envelope, measurement, ark_pem @@ -135,8 +148,14 @@ def test_happy_path_verifies() -> None: envelope, measurement, ark_pem = _build_evidence(nonce) res = verify_azure_cvm_measurement(measurement, envelope, nonce.hex(), ark_pem) assert res.verified is True, res.failure_reason - for f in ("measurement", "runtime_data_binding", "ak_binding", "quote_nonce_binding", - "vcek_cert_chain", "report_signature"): + for f in ( + "measurement", + "runtime_data_binding", + "ak_binding", + "quote_nonce_binding", + "vcek_cert_chain", + "report_signature", + ): assert f in res.verified_fields, f @@ -155,7 +174,9 @@ def test_wrong_ak_rejected() -> None: envelope, measurement, ark_pem = _build_evidence(nonce) env = json.loads(envelope) other = rsa.generate_private_key(public_exponent=65537, key_size=2048) - env["ak_pub_pem"] = other.public_key().public_bytes(Encoding.PEM, PublicFormat.SubjectPublicKeyInfo).decode() + env["ak_pub_pem"] = ( + other.public_key().public_bytes(Encoding.PEM, PublicFormat.SubjectPublicKeyInfo).decode() + ) res = verify_azure_cvm_measurement(measurement, json.dumps(env).encode(), nonce.hex(), ark_pem) assert res.verified is False assert res.failure_reason == "ak_mismatch" @@ -218,6 +239,7 @@ def fake_tpm(args): elif args[0] == "tpm2_readpublic": out = dict(zip(args, args[1:], strict=False)) from cryptography.hazmat.primitives.serialization import PublicFormat + __import__("pathlib").Path(out["-o"]).write_bytes( ak_key.public_key().public_bytes(Encoding.PEM, PublicFormat.SubjectPublicKeyInfo) ) @@ -230,7 +252,9 @@ def fake_tpm(args): assert report.report_data == nonce.hex() assert report.measurement == measurement - res = verify_azure_cvm_measurement(report.measurement, report.raw_evidence, nonce.hex(), ark_pem) + res = verify_azure_cvm_measurement( + report.measurement, report.raw_evidence, nonce.hex(), ark_pem + ) assert res.verified is True, res.failure_reason diff --git a/tests/unit/test_snp_signature_verify.py b/tests/unit/test_snp_signature_verify.py index 83b68e94..611f03db 100644 --- a/tests/unit/test_snp_signature_verify.py +++ b/tests/unit/test_snp_signature_verify.py @@ -8,6 +8,7 @@ A test against a genuine Azure SEV-SNP report and the real AMD KDS VCEK chain is marked skipped below and unblocks when that hardware fixture lands. """ + from __future__ import annotations import hashlib @@ -38,7 +39,7 @@ def _name(cn: str) -> x509.Name: return x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, cn)]) -def _cert(subject, issuer_name, subject_pub, issuer_key): +def _cert(subject, issuer_name, subject_pub, issuer_key, *, is_ca: bool = False): now = datetime.now(UTC) return ( x509.CertificateBuilder() @@ -48,6 +49,7 @@ def _cert(subject, issuer_name, subject_pub, issuer_key): .serial_number(x509.random_serial_number()) .not_valid_before(now - timedelta(days=1)) .not_valid_after(now + timedelta(days=3650)) + .add_extension(x509.BasicConstraints(ca=is_ca, path_length=None), critical=True) .sign(issuer_key, hashes.SHA384()) ) @@ -57,8 +59,10 @@ def _synthetic_chain(): ark_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) ask_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) vcek_key = ec.generate_private_key(ec.SECP384R1()) - ark = _cert(_name("ARK"), _name("ARK"), ark_key.public_key(), ark_key) # self-signed - ask = _cert(_name("ASK"), _name("ARK"), ask_key.public_key(), ark_key) + ark = _cert( + _name("ARK"), _name("ARK"), ark_key.public_key(), ark_key, is_ca=True + ) # self-signed + ask = _cert(_name("ASK"), _name("ARK"), ask_key.public_key(), ark_key, is_ca=True) vcek = _cert(_name("VCEK"), _name("ASK"), vcek_key.public_key(), ask_key) chain_pem = ( vcek.public_bytes(Encoding.PEM) @@ -85,7 +89,9 @@ def _signed_report(vcek_key, *, measurement_bytes: bytes, report_data: bytes) -> def test_valid_chain_and_signature_verifies() -> None: chain_pem, ark_pem, vcek_key = _synthetic_chain() - report, measurement = _signed_report(vcek_key, measurement_bytes=b"\x11" * 48, report_data=b"\x00" * 64) + report, measurement = _signed_report( + vcek_key, measurement_bytes=b"\x11" * 48, report_data=b"\x00" * 64 + ) res = verify_sev_snp_measurement( measurement=measurement, raw_evidence=report, @@ -99,7 +105,9 @@ def test_valid_chain_and_signature_verifies() -> None: def test_tampered_report_fails_closed() -> None: chain_pem, ark_pem, vcek_key = _synthetic_chain() - report, measurement = _signed_report(vcek_key, measurement_bytes=b"\x22" * 48, report_data=b"\x00" * 64) + report, measurement = _signed_report( + vcek_key, measurement_bytes=b"\x22" * 48, report_data=b"\x00" * 64 + ) tampered = bytearray(report) tampered[0x10] ^= 0xFF # flip a byte inside the signed region res = verify_sev_snp_measurement( @@ -116,7 +124,9 @@ def test_tampered_report_fails_closed() -> None: def test_wrong_pinned_ark_fails_closed() -> None: chain_pem, _good_ark, vcek_key = _synthetic_chain() _other_chain, other_ark_pem, _ = _synthetic_chain() # a different, untrusted ARK - report, measurement = _signed_report(vcek_key, measurement_bytes=b"\x33" * 48, report_data=b"\x00" * 64) + report, measurement = _signed_report( + vcek_key, measurement_bytes=b"\x33" * 48, report_data=b"\x00" * 64 + ) res = verify_sev_snp_measurement( measurement=measurement, raw_evidence=report, @@ -131,7 +141,9 @@ def test_missing_chain_stays_unverified_not_passed() -> None: # Backward compatible: with no chain supplied, the cert chain is reported as # unverified rather than silently trusted. _chain, _ark, vcek_key = _synthetic_chain() - report, measurement = _signed_report(vcek_key, measurement_bytes=b"\x44" * 48, report_data=b"\x00" * 64) + report, measurement = _signed_report( + vcek_key, measurement_bytes=b"\x44" * 48, report_data=b"\x00" * 64 + ) res = verify_sev_snp_measurement(measurement=measurement, raw_evidence=report) assert "vcek_cert_chain" in res.unverified_fields diff --git a/tests/unit/test_tdx_quote_verify.py b/tests/unit/test_tdx_quote_verify.py index d5394c75..96ed4871 100644 --- a/tests/unit/test_tdx_quote_verify.py +++ b/tests/unit/test_tdx_quote_verify.py @@ -13,6 +13,7 @@ Set ``CMCP_TDX_FIXTURE_DIR`` to a directory holding a real ``tdx_quote.bin`` to run the full-chain hardware tests at the bottom against the pinned Intel SGX Root CA. """ + from __future__ import annotations import hashlib @@ -43,7 +44,7 @@ def _name(cn: str) -> x509.Name: return x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, cn)]) -def _cert(subject: str, issuer: str, sub_pub, iss_priv) -> x509.Certificate: +def _cert(subject: str, issuer: str, sub_pub, iss_priv, *, is_ca: bool = False) -> x509.Certificate: now = datetime.now(UTC) return ( x509.CertificateBuilder() @@ -53,6 +54,7 @@ def _cert(subject: str, issuer: str, sub_pub, iss_priv) -> x509.Certificate: .serial_number(x509.random_serial_number()) .not_valid_before(now - timedelta(days=1)) .not_valid_after(now + timedelta(days=3650)) + .add_extension(x509.BasicConstraints(ca=is_ca, path_length=None), critical=True) .sign(iss_priv, hashes.SHA256()) ) @@ -71,8 +73,8 @@ def _pck_chain(): root_k = ec.generate_private_key(ec.SECP256R1()) inter_k = ec.generate_private_key(ec.SECP256R1()) leaf_k = ec.generate_private_key(ec.SECP256R1()) - root = _cert("Intel Root", "Intel Root", root_k.public_key(), root_k) # self-signed - inter = _cert("Intel PCK Intermediate", "Intel Root", inter_k.public_key(), root_k) + root = _cert("Intel Root", "Intel Root", root_k.public_key(), root_k, is_ca=True) # self-signed + inter = _cert("Intel PCK Intermediate", "Intel Root", inter_k.public_key(), root_k, is_ca=True) leaf = _cert("Intel PCK Leaf", "Intel PCK Intermediate", leaf_k.public_key(), inter_k) chain_pem = leaf.public_bytes(Encoding.PEM) + inter.public_bytes(Encoding.PEM) return chain_pem, root.public_bytes(Encoding.PEM), leaf_k, root_k @@ -86,14 +88,14 @@ def _build_quote(*, report_data: bytes = _RD, qe_auth: bytes = b"") -> tuple[byt header = bytearray(_QUOTE_HEADER_LEN) header[2:4] = (2).to_bytes(2, "little") # att_key_type = ECDSA-P256 body = bytearray(_TD_REPORT_BODY_LEN) - body[_TD_BODY_REPORT_DATA_OFF:_TD_BODY_REPORT_DATA_OFF + 64] = report_data + body[_TD_BODY_REPORT_DATA_OFF : _TD_BODY_REPORT_DATA_OFF + 64] = report_data signed_region = bytes(header) + bytes(body) quote_sig = _raw_sig(att_k, signed_region) chain_pem, root_pem, leaf_k, _root_k = _pck_chain() qe_report = bytearray(384) bind = hashlib.sha256(att_pub_raw + qe_auth).digest() - qe_report[_QE_REPORT_DATA_OFF:_QE_REPORT_DATA_OFF + 32] = bind + qe_report[_QE_REPORT_DATA_OFF : _QE_REPORT_DATA_OFF + 32] = bind qe_report_sig = _raw_sig(leaf_k, bytes(qe_report)) # Intel DCAP v4 nests the QE material: the bytes after the attestation key are @@ -104,13 +106,13 @@ def _build_quote(*, report_data: bytes = _RD, qe_auth: bytes = b"") -> tuple[byt cert_data += bytes(qe_report) cert_data += qe_report_sig cert_data += len(qe_auth).to_bytes(2, "little") + qe_auth - cert_data += (5).to_bytes(2, "little") # cert_data_type (PCK chain) + cert_data += (5).to_bytes(2, "little") # cert_data_type (PCK chain) cert_data += len(chain_pem).to_bytes(4, "little") + chain_pem sig = bytearray() sig += quote_sig sig += att_pub_raw - sig += (6).to_bytes(2, "little") # cert_data_type (QE report) + sig += (6).to_bytes(2, "little") # cert_data_type (QE report) sig += len(cert_data).to_bytes(4, "little") + bytes(cert_data) quote = signed_region + len(sig).to_bytes(4, "little") + bytes(sig) @@ -160,7 +162,7 @@ def test_parse_rejects_flat_signature_layout() -> None: quote, _ = _build_quote() off = _QUOTE_HEADER_LEN + _TD_REPORT_BODY_LEN + 4 + 128 flat = bytearray(quote) - flat[off:off + 2] = (5).to_bytes(2, "little") # PCK chain where the QE report belongs + flat[off : off + 2] = (5).to_bytes(2, "little") # PCK chain where the QE report belongs with pytest.raises(ValueError, match="certification data type"): parse_td_quote(bytes(flat)) @@ -169,11 +171,11 @@ def test_parse_rejects_flat_signature_layout() -> None: # Every one of these lengths is attacker-controlled: Python slicing clamps an # overstated length instead of raising, so the parser must reject the mismatch # rather than verify a silently shorter buffer than the producer declared. -_SIG_LEN_OFF = _QUOTE_HEADER_LEN + _TD_REPORT_BODY_LEN # uint32 signature-data size +_SIG_LEN_OFF = _QUOTE_HEADER_LEN + _TD_REPORT_BODY_LEN # uint32 signature-data size _SIG_OFF = _SIG_LEN_OFF + 4 -_QE_CERT_SIZE_OFF = _SIG_OFF + 130 # uint32, after type-6 header +_QE_CERT_SIZE_OFF = _SIG_OFF + 130 # uint32, after type-6 header _CERT_DATA_OFF = _SIG_OFF + 134 -_QE_AUTH_LEN_OFF = _CERT_DATA_OFF + 384 + 64 # uint16 QE auth data size +_QE_AUTH_LEN_OFF = _CERT_DATA_OFF + 384 + 64 # uint16 QE auth data size def test_parse_matches_shared_agent_manifest_parser() -> None: @@ -185,7 +187,7 @@ def test_parse_matches_shared_agent_manifest_parser() -> None: shared = parse_tdx_quote_signature(quote) assert pq.signed_region == shared.signed_body - assert pq.signed_region == quote[:_QUOTE_HEADER_LEN + _TD_REPORT_BODY_LEN] + assert pq.signed_region == quote[: _QUOTE_HEADER_LEN + _TD_REPORT_BODY_LEN] assert pq.quote_sig == shared.quote_signature assert pq.att_pubkey_raw == shared.attestation_key assert pq.qe_report == shared.qe_report @@ -201,8 +203,8 @@ def test_parse_rejects_oversized_declared_signature_size() -> None: """An overstated signature-data length must fail closed, not be clamped.""" quote, _root = _build_quote() tampered = bytearray(quote) - declared = int.from_bytes(tampered[_SIG_LEN_OFF:_SIG_LEN_OFF + 4], "little") - tampered[_SIG_LEN_OFF:_SIG_LEN_OFF + 4] = (declared + 64).to_bytes(4, "little") + declared = int.from_bytes(tampered[_SIG_LEN_OFF : _SIG_LEN_OFF + 4], "little") + tampered[_SIG_LEN_OFF : _SIG_LEN_OFF + 4] = (declared + 64).to_bytes(4, "little") with pytest.raises(ValueError, match="signature data"): parse_td_quote(bytes(tampered)) @@ -211,8 +213,8 @@ def test_parse_rejects_oversized_qe_certification_size() -> None: """An overstated type-6 QE certification-data length must fail closed.""" quote, _root = _build_quote() tampered = bytearray(quote) - declared = int.from_bytes(tampered[_QE_CERT_SIZE_OFF:_QE_CERT_SIZE_OFF + 4], "little") - tampered[_QE_CERT_SIZE_OFF:_QE_CERT_SIZE_OFF + 4] = (declared + 64).to_bytes(4, "little") + declared = int.from_bytes(tampered[_QE_CERT_SIZE_OFF : _QE_CERT_SIZE_OFF + 4], "little") + tampered[_QE_CERT_SIZE_OFF : _QE_CERT_SIZE_OFF + 4] = (declared + 64).to_bytes(4, "little") with pytest.raises(ValueError, match="QE certification data"): parse_td_quote(bytes(tampered)) @@ -221,7 +223,7 @@ def test_parse_rejects_oversized_qe_auth_size() -> None: """A QE auth length running past the certification data must fail closed.""" quote, _root = _build_quote() tampered = bytearray(quote) - tampered[_QE_AUTH_LEN_OFF:_QE_AUTH_LEN_OFF + 2] = (0xFFFF).to_bytes(2, "little") + tampered[_QE_AUTH_LEN_OFF : _QE_AUTH_LEN_OFF + 2] = (0xFFFF).to_bytes(2, "little") with pytest.raises(ValueError, match="QE auth data"): parse_td_quote(bytes(tampered))