ci: fail closed when CodeQL analysis fails

.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

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
 
 

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
 

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))