From 5ab7fbc1478c4b702e2190130c68deb376ca9099 Mon Sep 17 00:00:00 2001
From: Raghuvar Karthik Durga <177662415+drk-ram@users.noreply.github.com>
Date: Thu, 29 Oct 2026 12:13:50 -0500
Subject: [PATCH] prototype(azure_execution_binding): bind HCLAkPub exponent e
 and key size, reject duplicate JSON keys, and harden _secure_boot_state and
 authenticode_sha256

Closes #613

Signed-off-by: Raghuvar Karthik Durga <177662415+drk-ram@users.noreply.github.com>
---
 prototype/azure_execution_binding.py          | 116 ++--
 ...n_appraisal_and_azure_binding_hardening.py | 528 ++++++++++++++++++
 2 files changed, 611 insertions(+), 33 deletions(-)
 create mode 100644 tests/test_execution_appraisal_and_azure_binding_hardening.py

diff --git a/prototype/azure_execution_binding.py b/prototype/azure_execution_binding.py
index 70c129f1..1dfaa7d4 100644
--- a/prototype/azure_execution_binding.py
+++ b/prototype/azure_execution_binding.py
@@ -43,6 +43,32 @@ class ExecutionDenied(ValueError):
     pass
 
 
+def _pairs(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
+    result: dict[str, Any] = {}
+    for key, value in pairs:
+        if key in result:
+            raise ExecutionDenied("duplicate_json_key")
+        result[key] = value
+    return result
+
+
+def _b64url_uint(value: Any, label: str) -> int:
+    if (
+        not isinstance(value, str)
+        or not re.fullmatch(r"[A-Za-z0-9_-]+", value)
+        or len(value) % 4 == 1
+    ):
+        raise ValueError(f"HCLAkPub {label}")
+    raw = base64.b64decode(value + "=" * (-len(value) % 4), altchars=b"-_", validate=True)
+    if (
+        not raw
+        or raw[0] == 0
+        or base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii") != value
+    ):
+        raise ValueError(f"HCLAkPub {label} encoding")
+    return int.from_bytes(raw, "big")
+
+
 def _digest_set(values: frozenset[str], name: str) -> None:
     if not isinstance(values, frozenset) or any(
         not isinstance(v, str) or not _HEX64.fullmatch(v) for v in values
@@ -221,10 +247,17 @@ def _secure_boot_state(events: list[UefiEvent]) -> bool:
     for event in events:
         if event.pcr != 7 or event.event_type != EV_EFI_VARIABLE_DRIVER_CONFIG:
             continue
+        if not isinstance(event.data, bytes) or len(event.data) < 32:
+            raise ExecutionDenied("secure_boot_state")
         if hashlib.sha256(event.data).digest() != event.sha256:
             raise ExecutionDenied("uefi_variable_digest")
         name_len, data_len = struct.unpack_from("<QQ", event.data, 16)
-        name = event.data[32 : 32 + 2 * name_len].decode("utf-16-le")
+        if 32 + 2 * name_len + data_len != len(event.data):
+            raise ExecutionDenied("secure_boot_state")
+        try:
+            name = event.data[32 : 32 + 2 * name_len].decode("utf-16-le")
+        except UnicodeDecodeError as exc:
+            raise ExecutionDenied("secure_boot_state") from exc
         if name == "SecureBoot":
             found.append(event.data[32 + 2 * name_len : 32 + 2 * name_len + data_len])
     if len(found) != 1 or found[0] not in (b"\x00", b"\x01"):
@@ -466,24 +499,22 @@ def authenticate_ak(
     if parsed.report_data != hashlib.sha256(runtime_json).digest() + bytes(32):
         raise ExecutionDenied("hcl_runtime_binding")
     try:
-        doc = json.loads(runtime_json)
+        doc = json.loads(runtime_json, object_pairs_hook=_pairs)
         keys = [k for k in doc["keys"] if isinstance(k, dict) and k.get("kid") == "HCLAkPub"]
-        if len(keys) != 1 or _JWK_PRIVATE_PARAMS.intersection(keys[0]):
-            raise ValueError("HCLAkPub")
-        n = keys[0]["n"]
-        if not isinstance(n, str) or not re.fullmatch(r"[A-Za-z0-9_-]+", n):
-            raise ValueError("HCLAkPub n")
-        n_bytes = base64.b64decode(n + "=" * (-len(n) % 4), altchars=b"-_", validate=True)
         if (
-            not n_bytes
-            or n_bytes[0] == 0
-            or base64.urlsafe_b64encode(n_bytes).rstrip(b"=").decode("ascii") != n
+            len(keys) != 1
+            or keys[0].get("kty") != "RSA"
+            or _JWK_PRIVATE_PARAMS.intersection(keys[0])
         ):
-            raise ValueError("HCLAkPub n encoding")
-        modulus = int.from_bytes(n_bytes, "big")
+            raise ValueError("HCLAkPub")
+        modulus = _b64url_uint(keys[0]["n"], "n")
+        exponent = _b64url_uint(keys[0]["e"], "e")
+        if modulus.bit_length() < 2048 or exponent != 65537:
+            raise ValueError("HCLAkPub key size or exponent")
     except (KeyError, ValueError, TypeError, AttributeError) as exc:
         raise ExecutionDenied("hcl_ak") from exc
-    if modulus != ak.public_numbers().n:
+    numbers = ak.public_numbers()
+    if (modulus, exponent) != (numbers.n, numbers.e):
         raise ExecutionDenied("ak_not_hcl_bound")
     return ak
 
@@ -505,7 +536,7 @@ def _runtime_claims(packet: dict[str, Any]) -> dict[str, Any]:
     runtime = _b64(packet.get("hcl_b64"), "hcl")[32 + 1184 :]
     json_len = int.from_bytes(runtime[16:20], "little")
     try:
-        claims = json.loads(runtime[20 : 20 + json_len])
+        claims = json.loads(runtime[20 : 20 + json_len], object_pairs_hook=_pairs)
         vm_id = claims["vm-configuration"]["vmUniqueId"]
         secure_boot = claims["vm-configuration"]["secure-boot"]
     except (KeyError, ValueError, TypeError) as exc:
@@ -580,33 +611,52 @@ def execution_component(
 
 def authenticode_sha256(image: bytes) -> str:
     """PE Authenticode SHA-256, the digest firmware logs for boot applications."""
-    pe = struct.unpack_from("<I", image, 0x3C)[0]
-    if image[pe : pe + 4] != b"PE\x00\x00":
+    if not isinstance(image, bytes) or len(image) < 0x40:
         raise ValueError("not a PE image")
-    sections = struct.unpack_from("<H", image, pe + 6)[0]
-    optional_size = struct.unpack_from("<H", image, pe + 20)[0]
-    opt = pe + 24
-    magic = struct.unpack_from("<H", image, opt)[0]
-    datadir = opt + (112 if magic == 0x20B else 96)
-    checksum = opt + 64
-    cert_entry = datadir + 8 * 4
-    headers_size = struct.unpack_from("<I", image, opt + 60)[0]
-    cert_offset, cert_size = struct.unpack_from("<II", image, cert_entry)
+    try:
+        pe = struct.unpack_from("<I", image, 0x3C)[0]
+        if pe + 24 > len(image) or image[pe : pe + 4] != b"PE\x00\x00":
+            raise ValueError("not a PE image")
+        sections = struct.unpack_from("<H", image, pe + 6)[0]
+        optional_size = struct.unpack_from("<H", image, pe + 20)[0]
+        opt = pe + 24
+        magic = struct.unpack_from("<H", image, opt)[0]
+        if magic not in (0x10B, 0x20B):
+            raise ValueError("malformed PE image")
+        datadir = opt + (112 if magic == 0x20B else 96)
+        checksum = opt + 64
+        cert_entry = datadir + 8 * 4
+        headers_size = struct.unpack_from("<I", image, opt + 60)[0]
+        cert_offset, cert_size = struct.unpack_from("<II", image, cert_entry)
+        table = opt + optional_size
+        if (
+            cert_entry + 8 > table
+            or table + 40 * sections > headers_size
+            or headers_size > len(image)
+        ):
+            raise ValueError("malformed PE image")
+        if (cert_offset == 0) != (cert_size == 0) or (
+            cert_offset and (cert_offset < headers_size or cert_offset + cert_size != len(image))
+        ):
+            raise ValueError("malformed PE image")
+        raw = []
+        for i in range(sections):
+            size, pointer = struct.unpack_from("<II", image, table + 40 * i + 16)
+            if size:
+                raw.append((pointer, size))
+    except struct.error as exc:
+        raise ValueError("malformed PE image") from exc
+    end = len(image) - (cert_size if cert_offset else 0)
     digest = hashlib.sha256()
     digest.update(image[:checksum])
     digest.update(image[checksum + 4 : cert_entry])
     digest.update(image[cert_entry + 8 : headers_size])
-    table = opt + optional_size
-    raw = []
-    for i in range(sections):
-        size, pointer = struct.unpack_from("<II", image, table + 40 * i + 16)
-        if size:
-            raw.append((pointer, size))
     hashed = headers_size
     for pointer, size in sorted(raw):
+        if pointer < hashed or pointer + size > end:
+            raise ValueError("malformed PE image")
         digest.update(image[pointer : pointer + size])
         hashed = max(hashed, pointer + size)
-    end = len(image) - (cert_size if cert_offset else 0)
     if end > hashed:
         digest.update(image[hashed:end])
     return digest.hexdigest()
diff --git a/tests/test_execution_appraisal_and_azure_binding_hardening.py b/tests/test_execution_appraisal_and_azure_binding_hardening.py
new file mode 100644
index 00000000..0320ac2f
--- /dev/null
+++ b/tests/test_execution_appraisal_and_azure_binding_hardening.py
@@ -0,0 +1,528 @@
+"""Regression and boundary tests for Azure SEV-SNP/vTPM execution binding (#613)."""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import json
+import struct
+import sys
+from dataclasses import dataclass
+from pathlib import Path
+from types import ModuleType
+from typing import Any, cast
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric import rsa
+from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat
+
+from prototype import azure_execution_binding as x
+
+FIXTURES = Path(__file__).parent / "fixtures" / "azure-execution-20260929"
+AK = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+
+
+def _b64(data: bytes) -> str:
+    return base64.b64encode(data).decode("ascii")
+
+
+def _b64url(data: bytes) -> str:
+    return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii")
+
+
+def _install_wcm_stub(monkeypatch: pytest.MonkeyPatch) -> None:
+    @dataclass
+    class _ParsedReport:
+        policy: int
+        vmpl: int
+        report_data: bytes
+
+    wcm_mod = ModuleType("wcm")
+    certs_mod = ModuleType("wcm._certificates")
+    quote_mod = ModuleType("wcm._quote_verify")
+    snp_mod = ModuleType("wcm.snp")
+
+    cast(Any, certs_mod).load_pem_certificate = lambda pem, **kw: pem
+    cast(Any, quote_mod).verify_cert_chain = lambda vcek, inter, trust, now: []
+    cast(Any, snp_mod).extract_snp_report_from_hcl = lambda hcl: hcl[32 : 32 + 1184]
+    cast(Any, snp_mod).verify_snp_report_signature = lambda report, vcek: True
+    cast(Any, snp_mod).parse_snp_report = lambda report: _ParsedReport(
+        policy=int.from_bytes(report[8:16], "little"),
+        vmpl=report[0x30],
+        report_data=report[0x50:0x90],
+    )
+    monkeypatch.setitem(sys.modules, "wcm", wcm_mod)
+    monkeypatch.setitem(sys.modules, "wcm._certificates", certs_mod)
+    monkeypatch.setitem(sys.modules, "wcm._quote_verify", quote_mod)
+    monkeypatch.setitem(sys.modules, "wcm.snp", snp_mod)
+
+
+def _hcl_blob(runtime_json: bytes) -> bytes:
+    report = bytearray(1184)
+    report[0x50:0x90] = hashlib.sha256(runtime_json).digest() + bytes(32)
+    return (
+        bytes(32)
+        + bytes(report)
+        + bytes(16)
+        + len(runtime_json).to_bytes(4, "little")
+        + runtime_json
+    )
+
+
+def _ak_pem(public_key: rsa.RSAPublicKey | None = None) -> str:
+    key = public_key or AK.public_key()
+    return key.public_bytes(Encoding.PEM, PublicFormat.SubjectPublicKeyInfo).decode("ascii")
+
+
+def _hcl_packet(
+    runtime_json: bytes | None = None,
+    *,
+    jwk: dict[str, Any] | None = None,
+    ak_pem: str | None = None,
+) -> dict[str, Any]:
+    n_bytes = AK.public_key().public_numbers().n.to_bytes(256, "big")
+    key_obj: dict[str, Any] = (
+        jwk
+        if jwk is not None
+        else {
+            "kid": "HCLAkPub",
+            "kty": "RSA",
+            "e": "AQAB",
+            "n": _b64url(n_bytes),
+        }
+    )
+    raw_json = (
+        runtime_json
+        if runtime_json is not None
+        else json.dumps(
+            {
+                "keys": [key_obj],
+                "vm-configuration": {
+                    "vmUniqueId": "ca2e2e97-d879-49b7-ab25-f10ac9bd8667",
+                    "secure-boot": False,
+                },
+            }
+        ).encode("utf-8")
+    )
+    return {
+        "hcl_b64": _b64(_hcl_blob(raw_json)),
+        "vcek_pem": "-----BEGIN CERTIFICATE-----\n-----END CERTIFICATE-----\n",
+        "intermediates_pem": [],
+        "ak_pem": ak_pem if ak_pem is not None else _ak_pem(),
+    }
+
+
+def _real_b1() -> tuple[dict[str, Any], x.ExecutionPolicy, str]:
+    packet = json.loads((FIXTURES / "b1-approved-packet.json").read_text(encoding="utf-8"))
+    challenge = (FIXTURES / "b1-approved.challenge").read_text(encoding="utf-8").strip()
+    raw_policy = json.loads((FIXTURES / "execution-policy.json").read_text(encoding="utf-8"))
+    policy = x.ExecutionPolicy(
+        required_boot_application=raw_policy["required_boot_application"],
+        approved_boot_applications=frozenset(raw_policy["approved_boot_applications"]),
+        required_application=raw_policy["required_application"],
+        approved_executables=frozenset(raw_policy["approved_executables"]),
+        denied_executables=frozenset(raw_policy["denied_executables"]),
+        secure_boot=raw_policy["secure_boot"],
+    )
+    return packet, policy, challenge
+
+
+def _build_pe_image(
+    *,
+    magic: int = 0x20B,
+    section_data: bytes = b"\x90" * 64,
+    cert_bytes: bytes = b"",
+) -> tuple[bytes, str]:
+    pe = 0x40
+    opt = pe + 24
+    optional_size = 240 if magic == 0x20B else 224
+    datadir = opt + (112 if magic == 0x20B else 96)
+    checksum = opt + 64
+    cert_entry = datadir + 8 * 4
+    table = opt + optional_size
+    headers_size = 512
+    section_offset = headers_size
+    section_size = len(section_data)
+    cert_offset = section_offset + section_size if cert_bytes else 0
+    cert_size = len(cert_bytes)
+    total_size = section_offset + section_size + cert_size
+
+    buf = bytearray(total_size)
+    buf[0:2] = b"MZ"
+    struct.pack_into("<I", buf, 0x3C, pe)
+    buf[pe : pe + 4] = b"PE\x00\x00"
+    struct.pack_into("<H", buf, pe + 6, 1)  # NumberOfSections = 1
+    struct.pack_into("<H", buf, pe + 20, optional_size)
+    struct.pack_into("<H", buf, opt, magic)
+    struct.pack_into("<I", buf, opt + 60, headers_size)
+    struct.pack_into("<I", buf, checksum, 0xDEADBEEF)
+    struct.pack_into("<II", buf, cert_entry, cert_offset, cert_size)
+    buf[table : table + 6] = b".text\x00"
+    struct.pack_into("<II", buf, table + 16, section_size, section_offset)
+    buf[section_offset : section_offset + section_size] = section_data
+    if cert_bytes:
+        buf[cert_offset : cert_offset + cert_size] = cert_bytes
+
+    expected = hashlib.sha256()
+    expected.update(buf[:checksum])
+    expected.update(buf[checksum + 4 : cert_entry])
+    expected.update(buf[cert_entry + 8 : headers_size])
+    expected.update(section_data)
+    return bytes(buf), expected.hexdigest()
+
+
+def test_authenticate_ak_rejects_duplicate_json_keys(monkeypatch: pytest.MonkeyPatch) -> None:
+    _install_wcm_stub(monkeypatch)
+    b1_packet, b1_policy, b1_challenge = _real_b1()
+    appraised = x.appraise(
+        b1_packet, object(), b1_policy, expected_challenge=b1_challenge
+    )
+    assert appraised["observed_application"] == "sha256:" + b1_policy.required_application
+
+    n_b64 = _b64url(AK.public_key().public_numbers().n.to_bytes(256, "big"))
+    other_n_b64 = _b64url(
+        rsa.generate_private_key(public_exponent=65537, key_size=2048)
+        .public_key()
+        .public_numbers()
+        .n.to_bytes(256, "big")
+    )
+    dup_top_level = (
+        b'{"keys":[{"kid":"other","kty":"RSA","e":"AQAB","n":"'
+        + other_n_b64.encode("ascii")
+        + b'"}],"keys":[{"kid":"HCLAkPub","kty":"RSA","e":"AQAB","n":"'
+        + n_b64.encode("ascii")
+        + b'"}]}'
+    )
+    with pytest.raises(x.ExecutionDenied, match="^hcl_ak$"):
+        x.authenticate_ak(_hcl_packet(dup_top_level), object())
+
+    dup_jwk_key = (
+        b'{"keys":[{"kid":"HCLAkPub","kty":"RSA","e":"AQAB","n":"'
+        + other_n_b64.encode("ascii")
+        + b'","n":"'
+        + n_b64.encode("ascii")
+        + b'"}]}'
+    )
+    with pytest.raises(x.ExecutionDenied, match="^hcl_ak$"):
+        x.authenticate_ak(_hcl_packet(dup_jwk_key), object())
+
+
+def test_runtime_claims_rejects_duplicate_json_keys() -> None:
+    b1_packet, _, _ = _real_b1()
+    assert x._runtime_claims(b1_packet) == {
+        "vm_id": "ca2e2e97-d879-49b7-ab25-f10ac9bd8667",
+        "secure_boot": False,
+    }
+
+    dup_vm_id = (
+        b'{"vm-configuration":{"vmUniqueId":"not-a-uuid",'
+        b'"vmUniqueId":"ca2e2e97-d879-49b7-ab25-f10ac9bd8667","secure-boot":false}}'
+    )
+    with pytest.raises(x.ExecutionDenied, match="^hcl_vm_configuration$"):
+        x._runtime_claims(_hcl_packet(dup_vm_id))
+
+    dup_secboot = (
+        b'{"vm-configuration":{"vmUniqueId":"ca2e2e97-d879-49b7-ab25-f10ac9bd8667",'
+        b'"secure-boot":true,"secure-boot":false}}'
+    )
+    with pytest.raises(x.ExecutionDenied, match="^hcl_vm_configuration$"):
+        x._runtime_claims(_hcl_packet(dup_secboot))
+
+
+def test_authenticate_ak_rejects_non_rsa_kty(monkeypatch: pytest.MonkeyPatch) -> None:
+    _install_wcm_stub(monkeypatch)
+    b1_packet, _, _ = _real_b1()
+    assert x.authenticate_ak(b1_packet, object()).public_numbers().e == 65537
+
+    valid_n = _b64url(AK.public_key().public_numbers().n.to_bytes(256, "big"))
+    for bad_jwk in (
+        {"kid": "HCLAkPub", "e": "AQAB", "n": valid_n},
+        {"kid": "HCLAkPub", "kty": "EC", "e": "AQAB", "n": valid_n},
+        {"kid": "HCLAkPub", "kty": "OKP", "e": "AQAB", "n": valid_n},
+        {"kid": "HCLAkPub", "kty": "oct", "e": "AQAB", "n": valid_n},
+        {"kid": "HCLAkPub", "kty": "rsa", "e": "AQAB", "n": valid_n},
+    ):
+        with pytest.raises(x.ExecutionDenied, match="^hcl_ak$"):
+            x.authenticate_ak(_hcl_packet(jwk=bad_jwk), object())
+
+
+def test_authenticate_ak_rejects_invalid_or_noncanonical_exponent_encoding(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    _install_wcm_stub(monkeypatch)
+    assert x.authenticate_ak(_hcl_packet(), object()).public_numbers().e == 65537
+
+    valid_n = _b64url(AK.public_key().public_numbers().n.to_bytes(256, "big"))
+    for bad_jwk in (
+        {"kid": "HCLAkPub", "kty": "RSA", "n": valid_n},
+        {"kid": "HCLAkPub", "kty": "RSA", "e": 65537, "n": valid_n},
+        {"kid": "HCLAkPub", "kty": "RSA", "e": None, "n": valid_n},
+        {"kid": "HCLAkPub", "kty": "RSA", "e": "", "n": valid_n},
+        {"kid": "HCLAkPub", "kty": "RSA", "e": "A", "n": valid_n},
+        {"kid": "HCLAkPub", "kty": "RSA", "e": "AQAB=", "n": valid_n},
+        {"kid": "HCLAkPub", "kty": "RSA", "e": "AQAB$", "n": valid_n},
+        {"kid": "HCLAkPub", "kty": "RSA", "e": _b64url(b"\x00\x01\x00\x01"), "n": valid_n},
+        {"kid": "HCLAkPub", "kty": "RSA", "e": "AQB", "n": valid_n},
+        {"kid": "HCLAkPub", "kty": "RSA", "e": "AQAB", "n": "A" * 341},
+    ):
+        with pytest.raises(x.ExecutionDenied, match="^hcl_ak$"):
+            x.authenticate_ak(_hcl_packet(jwk=bad_jwk), object())
+
+
+def test_authenticate_ak_rejects_sub_2048_bit_modulus(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    _install_wcm_stub(monkeypatch)
+    assert x.authenticate_ak(_hcl_packet(), object()).public_numbers().n.bit_length() == 2048
+
+    for small_int, byte_len in (((1 << 1023) | 1, 128), ((1 << 2047) - 1, 256)):
+        small_n = _b64url(small_int.to_bytes(byte_len, "big"))
+        small_key = rsa.RSAPublicNumbers(65537, small_int).public_key()
+        bad_jwk = {"kid": "HCLAkPub", "kty": "RSA", "e": "AQAB", "n": small_n}
+        with pytest.raises(x.ExecutionDenied, match="^hcl_ak$"):
+            x.authenticate_ak(_hcl_packet(jwk=bad_jwk, ak_pem=_ak_pem(small_key)), object())
+
+
+def test_authenticate_ak_rejects_non_65537_exponent(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    _install_wcm_stub(monkeypatch)
+    assert x.authenticate_ak(_hcl_packet(), object()).public_numbers().e == 65537
+
+    valid_n = _b64url(AK.public_key().public_numbers().n.to_bytes(256, "big"))
+    for bad_e in (
+        _b64url((3).to_bytes(1, "big")),
+        _b64url((17).to_bytes(1, "big")),
+        _b64url((65539).to_bytes(3, "big")),
+    ):
+        bad_jwk = {"kid": "HCLAkPub", "kty": "RSA", "e": bad_e, "n": valid_n}
+        with pytest.raises(x.ExecutionDenied, match="^hcl_ak$"):
+            x.authenticate_ak(_hcl_packet(jwk=bad_jwk), object())
+
+
+def test_authenticate_ak_rejects_mismatched_ak_pem_exponent(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    _install_wcm_stub(monkeypatch)
+    good_ak = x.authenticate_ak(_hcl_packet(), object())
+    assert (good_ak.public_numbers().n, good_ak.public_numbers().e) == (
+        AK.public_key().public_numbers().n,
+        65537,
+    )
+
+    mismatched_e_key = rsa.RSAPublicNumbers(3, AK.public_key().public_numbers().n).public_key()
+    with pytest.raises(x.ExecutionDenied, match="^ak_not_hcl_bound$"):
+        x.authenticate_ak(_hcl_packet(ak_pem=_ak_pem(mismatched_e_key)), object())
+
+
+def test_secure_boot_state_rejects_non_bytes_and_short_event_data() -> None:
+    b1_packet, _, _ = _real_b1()
+    b1_events = x.parse_uefi_log(base64.b64decode(b1_packet["uefi_event_log_b64"]))
+    assert x._secure_boot_state(b1_events) is False
+
+    short_data = bytes(20)
+    short_ev = x.UefiEvent(
+        pcr=7,
+        event_type=x.EV_EFI_VARIABLE_DRIVER_CONFIG,
+        sha256=hashlib.sha256(short_data).digest(),
+        data=short_data,
+    )
+    with pytest.raises(x.ExecutionDenied, match="^secure_boot_state$"):
+        x._secure_boot_state([short_ev])
+
+    non_bytes_ev = x.UefiEvent(
+        pcr=7,
+        event_type=x.EV_EFI_VARIABLE_DRIVER_CONFIG,
+        sha256=bytes(32),
+        data=cast(Any, None),
+    )
+    with pytest.raises(x.ExecutionDenied, match="^secure_boot_state$"):
+        x._secure_boot_state([non_bytes_ev])
+
+
+def test_secure_boot_state_rejects_mismatched_variable_payload_length() -> None:
+    name_utf16 = "SecureBoot".encode("utf-16-le")
+    valid_data = bytes(16) + struct.pack("<QQ", len("SecureBoot"), 1) + name_utf16 + b"\x01"
+    valid_ev = x.UefiEvent(
+        pcr=7,
+        event_type=x.EV_EFI_VARIABLE_DRIVER_CONFIG,
+        sha256=hashlib.sha256(valid_data).digest(),
+        data=valid_data,
+    )
+    assert x._secure_boot_state([valid_ev]) is True
+
+    truncated_data = bytes(16) + struct.pack("<QQ", len("SecureBoot"), 9999) + name_utf16 + b"\x01"
+    truncated_name = bytes(16) + struct.pack("<QQ", 10, 1) + name_utf16[:6] + b"\x01"
+    trailing_bytes = (
+        bytes(16) + struct.pack("<QQ", len("SecureBoot"), 1) + name_utf16 + b"\x01\x00"
+    )
+    for bad_payload in (truncated_data, truncated_name, trailing_bytes):
+        ev = x.UefiEvent(
+            pcr=7,
+            event_type=x.EV_EFI_VARIABLE_DRIVER_CONFIG,
+            sha256=hashlib.sha256(bad_payload).digest(),
+            data=bad_payload,
+        )
+        with pytest.raises(x.ExecutionDenied, match="^secure_boot_state$"):
+            x._secure_boot_state([ev])
+
+
+def test_secure_boot_state_rejects_invalid_utf16_variable_name() -> None:
+    name_utf16 = "SecureBoot".encode("utf-16-le")
+    valid_data = bytes(16) + struct.pack("<QQ", len("SecureBoot"), 1) + name_utf16 + b"\x01"
+    valid_ev = x.UefiEvent(
+        pcr=7,
+        event_type=x.EV_EFI_VARIABLE_DRIVER_CONFIG,
+        sha256=hashlib.sha256(valid_data).digest(),
+        data=valid_data,
+    )
+    assert x._secure_boot_state([valid_ev]) is True
+
+    surrogate_name = bytes(16) + struct.pack("<QQ", 1, 1) + b"\x00\xd8" + b"\x01"
+    ev = x.UefiEvent(
+        pcr=7,
+        event_type=x.EV_EFI_VARIABLE_DRIVER_CONFIG,
+        sha256=hashlib.sha256(surrogate_name).digest(),
+        data=surrogate_name,
+    )
+    with pytest.raises(x.ExecutionDenied, match="^secure_boot_state$"):
+        x._secure_boot_state([ev])
+
+
+def test_authenticode_sha256_rejects_non_bytes_and_truncated_pe_headers() -> None:
+    pe64, expected64 = _build_pe_image(magic=0x20B, cert_bytes=b"CERT" * 8)
+    assert x.authenticode_sha256(pe64) == expected64
+
+    pe32, expected32 = _build_pe_image(magic=0x10B, cert_bytes=b"")
+    assert x.authenticode_sha256(pe32) == expected32
+
+    for non_pe in (
+        cast(Any, None),
+        cast(Any, "MZ"),
+        cast(Any, bytearray(pe64)),
+        b"MZ" + bytes(20),
+        b"MZ" + bytes(0x3A) + struct.pack("<I", 0x200) + bytes(64),
+        b"MZ" + bytes(0x3A) + struct.pack("<I", 0x40) + b"PE\x00\x00" + bytes(16),
+    ):
+        with pytest.raises(ValueError, match="^not a PE image$"):
+            x.authenticode_sha256(non_pe)
+
+    truncated_optional = b"MZ" + bytes(0x3A) + struct.pack("<I", 0x40) + b"PE\x00\x00" + bytes(20)
+    with pytest.raises(ValueError, match="^malformed PE image$"):
+        x.authenticode_sha256(truncated_optional)
+
+
+def test_authenticode_sha256_rejects_unknown_optional_header_magic() -> None:
+    pe64, expected64 = _build_pe_image(magic=0x20B)
+    assert x.authenticode_sha256(pe64) == expected64
+
+    bad_magic = bytearray(pe64)
+    struct.pack_into("<H", bad_magic, 0x40 + 24, 0x999)
+    with pytest.raises(ValueError, match="^malformed PE image$"):
+        x.authenticode_sha256(bytes(bad_magic))
+
+
+def test_authenticode_sha256_rejects_out_of_bounds_header_and_section_table_sizes() -> None:
+    pe64, expected64 = _build_pe_image(magic=0x20B)
+    assert x.authenticode_sha256(pe64) == expected64
+
+    small_opt = bytearray(pe64)
+    struct.pack_into("<H", small_opt, 0x40 + 20, 32)
+
+    oversized_headers = bytearray(pe64)
+    struct.pack_into("<I", oversized_headers, 0x40 + 24 + 60, len(pe64) + 64)
+
+    too_many_sections = bytearray(pe64)
+    struct.pack_into("<H", too_many_sections, 0x40 + 6, 20)
+
+    for malformed in (bytes(small_opt), bytes(oversized_headers), bytes(too_many_sections)):
+        with pytest.raises(ValueError, match="^malformed PE image$"):
+            x.authenticode_sha256(malformed)
+
+
+def test_authenticode_sha256_rejects_invalid_certificate_table_bounds() -> None:
+    pe64, expected64 = _build_pe_image(magic=0x20B, cert_bytes=b"CERT" * 8)
+    assert x.authenticode_sha256(pe64) == expected64
+
+    cert_entry = 0x40 + 24 + 112 + 32
+
+    oob_cert = bytearray(pe64)
+    struct.pack_into("<II", oob_cert, cert_entry, 512, 4096)
+
+    cert_in_headers = bytearray(pe64[:512])
+    struct.pack_into("<H", cert_in_headers, 0x40 + 6, 0)
+    struct.pack_into("<II", cert_in_headers, cert_entry, 256, 256)
+
+    zero_offset_nonzero_cert = bytearray(pe64)
+    struct.pack_into("<II", zero_offset_nonzero_cert, cert_entry, 0, 32)
+
+    for malformed in (bytes(oob_cert), bytes(cert_in_headers), bytes(zero_offset_nonzero_cert)):
+        with pytest.raises(ValueError, match="^malformed PE image$"):
+            x.authenticode_sha256(malformed)
+
+
+def test_authenticode_sha256_rejects_overlapping_or_out_of_bounds_sections() -> None:
+    pe64, expected64 = _build_pe_image(magic=0x20B, cert_bytes=b"CERT" * 8)
+    assert x.authenticode_sha256(pe64) == expected64
+
+    table = 0x40 + 24 + 240
+
+    oob_section = bytearray(pe64)
+    struct.pack_into("<II", oob_section, table + 16, 4096, 512)
+
+    header_overlapping_section = bytearray(pe64)
+    struct.pack_into("<II", header_overlapping_section, table + 16, 64, 256)
+
+    for malformed in (bytes(oob_section), bytes(header_overlapping_section)):
+        with pytest.raises(ValueError, match="^malformed PE image$"):
+            x.authenticode_sha256(malformed)
+
+
+@pytest.mark.parametrize(
+    "case",
+    [
+        "table_past_headers",
+        "headers_past_eof",
+        "cert_in_headers",
+        "cert_past_eof",
+        "cert_before_eof",
+        "nonzero_offset_zero_size",
+        "section_in_section",
+        "section_in_cert",
+    ],
+)
+def test_authenticode_sha256_rejects_each_bounds_clause_alone(case: str) -> None:
+    plain, expected_plain = _build_pe_image(magic=0x20B)
+    signed, expected_signed = _build_pe_image(magic=0x20B, cert_bytes=b"CERT" * 8)
+    assert x.authenticode_sha256(plain) == expected_plain
+    assert x.authenticode_sha256(signed) == expected_signed
+
+    opt, table = 0x40 + 24, 0x40 + 24 + 240
+    image = bytearray(
+        signed
+        if case in ("cert_past_eof", "cert_before_eof", "section_in_cert")
+        else plain
+    )
+    if case == "table_past_headers":
+        struct.pack_into("<H", image, 0x40 + 6, 5)
+    elif case == "headers_past_eof":
+        struct.pack_into("<H", image, 0x40 + 6, 0)
+        struct.pack_into("<I", image, opt + 60, len(plain) + 64)
+    elif case == "cert_in_headers":
+        del image[512:]
+        struct.pack_into("<H", image, 0x40 + 6, 0)
+        struct.pack_into("<II", image, opt + 144, 256, 256)
+    elif case == "cert_past_eof":
+        struct.pack_into("<II", image, opt + 144, len(signed) - 8, 32)
+    elif case == "cert_before_eof":
+        struct.pack_into("<II", image, opt + 144, len(signed) - 32, 16)
+    elif case == "nonzero_offset_zero_size":
+        struct.pack_into("<II", image, opt + 144, len(plain), 0)
+    elif case == "section_in_section":
+        struct.pack_into("<H", image, 0x40 + 6, 2)
+        struct.pack_into("<II", image, table + 56, 32, 544)
+    else:
+        struct.pack_into("<II", image, table + 16, 80, 512)
+    with pytest.raises(ValueError, match="^malformed PE image$"):
+        x.authenticode_sha256(bytes(image))
+