diff options
Diffstat (limited to 'scripts/tpm_nvpcr.py')
| -rw-r--r-- | scripts/tpm_nvpcr.py | 133 |
1 files changed, 122 insertions, 11 deletions
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}") |