From 6739045b173cc1be1f46bdf4b6167cbf9a883089 Mon Sep 17 00:00:00 2001 From: sommerfeld Date: Wed, 30 Sep 2026 08:54:59 +0100 Subject: Require signed PCR 11 policy for TPM root unlock --- scripts/tpm_negative_uki.py | 69 +++++++++++++++++++++++ scripts/tpm_nvpcr.py | 133 ++++++++++++++++++++++++++++++++++++++++---- scripts/tpm_unlock.py | 108 +++++++++++++++++++++++++++++++++++ 3 files changed, 299 insertions(+), 11 deletions(-) create mode 100644 scripts/tpm_negative_uki.py create mode 100644 scripts/tpm_unlock.py (limited to 'scripts') diff --git a/scripts/tpm_negative_uki.py b/scripts/tpm_negative_uki.py new file mode 100644 index 0000000..b9ce8b2 --- /dev/null +++ b/scripts/tpm_negative_uki.py @@ -0,0 +1,69 @@ +"""Prepare a UKI that retains NvPCR approval but lacks root-unlock approval.""" + +import json +import os +import subprocess +import sys +from pathlib import Path + +import pefile + +from scripts.tpm_nvpcr import PUBLIC_KEY, read_pe_sections, verify_pcr_signatures + + +def remove_root_approval(source: Path, destination: Path, public_key: bytes) -> None: + if destination.exists(): + raise ValueError("Refusing to replace an existing test image") + original = read_pe_sections(source) + verify_pcr_signatures(original, public_key) + with destination.open("xb") as output: + output.write(source.read_bytes()) + try: + subprocess.run( + ["sbattach", "--remove", str(destination)], check=True, capture_output=True + ) + pe = pefile.PE(str(destination), fast_load=True) + sections = [s for s in pe.sections if s.Name.rstrip(b"\0") == b".pcrsig"] + if len(sections) != 1: + raise ValueError("Expected one .pcrsig section") + section = sections[0] + length = min(section.Misc_VirtualSize, section.SizeOfRawData) + offset = section.PointerToRawData + image = bytearray(destination.read_bytes()) + old = bytes(image[offset : offset + length]) + if old.count(b"luks-root") != 1: + raise ValueError("Expected exactly one luks-root approval") + new = old.replace(b"luks-root", b"deny-root") + policies = json.loads(new.decode().rstrip("\0")) + if any(item.get("ref") == "luks-root" for item in policies.get("sha256", [])): + raise ValueError("Root approval remains in test image") + image[offset : offset + length] = new + destination.write_bytes(image) + changed = read_pe_sections(destination) + if any( + original.get(name) != changed.get(name) + for name in original + if name != ".pcrsig" + ): + raise ValueError("Test image changed a measured section") + verify_pcr_signatures(changed, public_key, refs=("initrd",)) + except Exception: + destination.unlink() + raise + + +def main() -> None: + if len(sys.argv) != 3: + raise ValueError("Usage: tpm_negative_uki.py SOURCE DESTINATION") + if os.geteuid() != 0 or Path("/etc/hostname").read_text().strip() != "halley2": + raise ValueError("Run as root on Halley2") + remove_root_approval(Path(sys.argv[1]), Path(sys.argv[2]), PUBLIC_KEY.read_bytes()) + print(f"prepared: {sys.argv[2]} (sign with Secure Boot key before testing)") + + +if __name__ == "__main__": + try: + main() + except (OSError, TypeError, ValueError, subprocess.CalledProcessError) as exc: + print(f"Negative UKI preparation failed: {exc}", file=sys.stderr) + sys.exit(1) diff --git a/scripts/tpm_nvpcr.py b/scripts/tpm_nvpcr.py index 4f110fa..ddf9fd2 100644 --- a/scripts/tpm_nvpcr.py +++ b/scripts/tpm_nvpcr.py @@ -1,15 +1,23 @@ """Read-only checks for Halley2's signed NvPCR UKIs.""" +import base64 import hashlib import json import os import subprocess import sys +import tempfile from pathlib import Path +import pefile +from cryptography.exceptions import InvalidSignature +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import padding, rsa + BOOT = Path("/boot/EFI/Linux") PUBLIC_KEY = Path("/etc/systemd/tpm2-pcr-public-key.pem") SB_CERT = Path("/var/lib/sbctl/keys/db/db.pem") +ROOT_UUID = "81520bbc-1e7a-45e6-9465-cfc2e8b18945" IMAGES = ( "arch-linux-hardened.efi", "arch-linux-hardened-fallback.efi", @@ -27,6 +35,23 @@ INITRD_UNITS = ( "systemd-pcrextend.socket", "systemd-pcrextend@.service", ) +POLICY_REFS = ("initrd", "luks-root") +MEASURED_SECTIONS = frozenset( + ( + ".linux", + ".osrel", + ".cmdline", + ".initrd", + ".ucode", + ".splash", + ".dtb", + ".uname", + ".sbat", + ".pcrpkey", + ".profile", + ".hwids", + ) +) def run(*command: str) -> str: @@ -50,23 +75,24 @@ def check_sections(sections: dict, public_key: bytes) -> None: if not isinstance(signature, dict) or not isinstance(signature.get("text"), str): raise TypeError("UKI has no readable PCR signatures") policies = json.loads(signature["text"]) - if not isinstance(policies, dict) or not any( - isinstance(item, dict) - and item.get("pcrs") == [11] - and item.get("ref") == "initrd" - and isinstance(item.get("sig"), str) - and bool(item["sig"]) - for item in policies.get("sha256", []) - ): - raise ValueError("UKI has no signed initrd PCR 11 policy") + for ref in POLICY_REFS: + if not isinstance(policies, dict) or not any( + isinstance(item, dict) + and item.get("pcrs") == [11] + and item.get("ref") == ref + and isinstance(item.get("sig"), str) + and bool(item["sig"]) + for item in policies.get("sha256", []) + ): + raise ValueError(f"UKI has no signed {ref} PCR 11 policy") cmdline = sections.get(".cmdline") if not isinstance(cmdline, dict) or not isinstance(cmdline.get("text"), str): raise TypeError("UKI has no kernel command line") args = cmdline["text"].split() roots = [arg.split("=", 2)[1] for arg in args if arg.startswith("rd.luks.name=")] - if len(roots) != 1: - raise ValueError("UKI must specify exactly one LUKS root device") + if roots != [ROOT_UUID]: + raise ValueError("UKI has the wrong LUKS root UUID") options = [ arg.removeprefix("rd.luks.options=") for arg in args @@ -74,6 +100,8 @@ def check_sections(sections: dict, public_key: bytes) -> None: ] if len(options) != 1 or not options[0].startswith(roots[0] + "="): raise ValueError("Root measurement options have the wrong LUKS UUID") + if "root=/dev/mapper/root" not in args: + raise ValueError("UKI does not boot the expected root mapping") selected = options[0].split("=", 1)[1].split(",") if not all(option in selected for option in MEASUREMENT_OPTIONS): raise ValueError("Root measurement or TPM unlock option is missing") @@ -86,6 +114,88 @@ def check_initrd_listing(listing: str) -> None: raise ValueError(f"UKI initrd lacks {unit}") +def read_pe_sections(path: Path) -> dict[str, bytes]: + pe = pefile.PE(str(path), fast_load=True) + sections = {} + for section in pe.sections: + name = section.Name.rstrip(b"\0").decode("ascii") + if name in sections: + raise ValueError(f"Duplicate UKI section: {name}") + size = min(section.Misc_VirtualSize, section.SizeOfRawData) + sections[name] = section.get_data(length=size) + if ".dtbauto" in sections: + raise ValueError( + "Automatic device tree selection is not supported by this checker" + ) + return sections + + +def verify_pcr_signatures( + sections: dict[str, bytes], public_key: bytes, refs: tuple[str, ...] = POLICY_REFS +) -> None: + if sections.get(".pcrpkey") != public_key: + raise ValueError("UKI PCR public key does not match the installed key") + try: + signed = json.loads(sections[".pcrsig"].decode().rstrip("\0")) + key = serialization.load_pem_public_key(public_key) + except (KeyError, ValueError, TypeError) as exc: + raise ValueError("Invalid UKI PCR signature or public key") from exc + if not isinstance(key, rsa.RSAPublicKey): + raise TypeError("PCR public key must be RSA") + + measured = { + name: data for name, data in sections.items() if name in MEASURED_SECTIONS + } + if ".linux" not in measured or ".initrd" not in measured: + raise ValueError("UKI lacks a kernel or initrd") + + with tempfile.TemporaryDirectory() as directory: + options = [] + for name, data in measured.items(): + section_path = Path(directory) / name.removeprefix(".") + section_path.write_bytes(data) + options.append(f"--{name.removeprefix('.')}={section_path}") + for ref in refs: + expected = json.loads( + run( + "/usr/lib/systemd/systemd-measure", + "policy-digest", + "--json=short", + "--bank=sha256", + "--phase=enter-initrd", + f"--policyref={ref}", + f"--public-key={PUBLIC_KEY}", + *options, + ) + )["sha256"] + if len(expected) != 1: + raise ValueError(f"Unexpected PCR policy count for {ref}") + matching = [ + item + for item in signed.get("sha256", []) + if item.get("pcrs") == [11] and item.get("ref") == ref + ] + if len(matching) != 1: + raise ValueError(f"Expected one {ref} PCR 11 signature") + actual = matching[0] + policy = expected[0] + if any( + actual.get(field) != policy.get(field) + for field in ("pcrs", "pkfp", "ref", "pol") + ): + raise ValueError(f"{ref} PCR 11 signature does not match UKI contents") + try: + signature = base64.b64decode(actual["sig"], validate=True) + key.verify( + signature, + bytes.fromhex(policy["tbs"]), + padding.PKCS1v15(), + hashes.SHA256(), + ) + except (KeyError, ValueError, InvalidSignature) as exc: + raise ValueError(f"Invalid {ref} PCR 11 signature") from exc + + def check_image(path: Path, public_key: bytes) -> None: sections = json.loads( run( @@ -101,6 +211,7 @@ def check_image(path: Path, public_key: bytes) -> None: if not isinstance(sections, dict) or "_profiles" in sections: raise ValueError(f"Unsupported multi-profile UKI: {path}") check_sections(sections, public_key) + verify_pcr_signatures(read_pe_sections(path), public_key) check_initrd_listing(run("lsinitcpio", "--list", str(path))) run("sbverify", "--cert", str(SB_CERT), str(path)) print(f"verified: {path}") diff --git a/scripts/tpm_unlock.py b/scripts/tpm_unlock.py new file mode 100644 index 0000000..0082d2d --- /dev/null +++ b/scripts/tpm_unlock.py @@ -0,0 +1,108 @@ +"""Audit the LUKS token used for automatic root unlock on Halley2.""" + +import base64 +import binascii +import json +import os +import subprocess +import sys +from pathlib import Path + +from cryptography.hazmat.primitives import serialization + +from scripts.tpm_nvpcr import PUBLIC_KEY + +DEVICE = "/dev/nvme0n1p2" + + +def decoded(value: object, field: str) -> bytes: + if not isinstance(value, str): + raise TypeError(f"Missing {field}") + try: + result = base64.b64decode(value, validate=True) + except (ValueError, binascii.Error) as exc: + raise ValueError(f"Invalid {field}") from exc + if not result: + raise ValueError(f"Empty {field}") + return result + + +def check_header( + header: dict, public_key: bytes, *, trial: bool = False +) -> tuple[str, str]: + tokens = header.get("tokens", {}) + slots = header.get("keyslots", {}) + if not isinstance(tokens, dict) or len(tokens) != 1: + raise ValueError("Expected exactly one automatic-unlock token") + if not isinstance(slots, dict) or "0" not in slots: + raise ValueError("Passphrase keyslot 0 is missing") + token_id, token = next(iter(tokens.items())) + if not isinstance(token, dict) or token.get("type") != "systemd-tpm2": + raise ValueError("Expected one systemd TPM2 token") + assigned = token.get("keyslots") + if not isinstance(assigned, list) or len(assigned) != 1: + raise ValueError("TPM token must name exactly one keyslot") + slot = assigned[0] + if slot == "0" or slot not in slots: + raise ValueError("TPM token has no separate live keyslot") + if not trial and set(slots) != {"0", slot}: + raise ValueError("Unexpected extra keyslot after migration") + + if token.get("tpm2-pcrs") != [] or token.get("tpm2-pcr-bank") != "sha256": + raise ValueError("Unexpected direct PCR binding or PCR bank") + if token.get("tpm2_pcrlock") is not True: + raise ValueError("TPM token does not require pcrlock") + decoded(token.get("tpm2_pcrlock_nv"), "pcrlock NV reference") + if ( + token.get("tpm2_pubkey_pcrs") != [11] + or token.get("tpm2_pubkey_ref") != "luks-root" + ): + raise ValueError( + "TPM token does not require the signed luks-root PCR 11 policy" + ) + embedded = serialization.load_pem_public_key( + decoded(token.get("tpm2_pubkey"), "PCR public key") + ) + installed = serialization.load_pem_public_key(public_key) + encoding = serialization.Encoding.DER + fmt = serialization.PublicFormat.SubjectPublicKeyInfo + if embedded.public_bytes(encoding, fmt) != installed.public_bytes(encoding, fmt): + raise ValueError("TPM token has the wrong PCR public key") + blobs = token.get("tpm2-blob") + hashes = token.get("tpm2-policy-hash") + if not isinstance(blobs, list) or len(blobs) != 2: + raise ValueError("Combined TPM policy must have two sealed blobs") + if not isinstance(hashes, list) or len(hashes) != 2: + raise ValueError("Combined TPM policy must have two policy hashes") + for blob in blobs: + decoded(blob, "sealed blob") + for value in hashes: + if not isinstance(value, str) or len(bytes.fromhex(value)) != 32: + raise ValueError("Invalid TPM policy hash") + return token_id, slot + + +def main() -> None: + if len(sys.argv) > 2 or (len(sys.argv) == 2 and sys.argv[1] != "--trial"): + raise ValueError("Usage: tpm_unlock.py [--trial]") + if os.geteuid() != 0 or Path("/etc/hostname").read_text().strip() != "halley2": + raise ValueError("Run as root on Halley2") + header = json.loads( + subprocess.check_output( + ["cryptsetup", "luksDump", "--dump-json-metadata", DEVICE], text=True + ) + ) + token_id, slot = check_header( + header, PUBLIC_KEY.read_bytes(), trial="--trial" in sys.argv + ) + print( + f"verified: token {token_id}, keyslot {slot}, pcrlock and signed PCR 11 metadata" + ) + + +if __name__ == "__main__": + try: + main() + except (OSError, TypeError, ValueError, subprocess.CalledProcessError) as exc: + print(f"TPM unlock check failed: {exc}", file=sys.stderr) + sys.exit(1) -- cgit v1.3.1