Require signed PCR 11 policy for TPM root unlock

3 files changed, 299 insertions, 11 deletions

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)