Merge pull request #167 from nix-community/sd-stub-tpm2

feat: minimal poc for TPM measurements à la sd-stub

nix/tests/lanzaboote.nix

@@ -3,10 +3,90 @@
 }:
 
 let
-  inherit (pkgs) lib;
+  inherit (pkgs) lib system;
+
+  mkSecureBootTest = { name, machine ? { }, useSecureBoot ? true, useTPM2 ? false, readEfiVariables ? false, testScript }:
+    let
+      tpmSocketPath = "/tmp/swtpm-sock";
+      tpmDeviceModels = {
+        x86_64-linux = "tpm-tis";
+        aarch64-linux = "tpm-tis-device";
+      };
+      # Should go to nixpkgs.
+      efiVariablesHelpers = ''
+        import struct
+
+        SD_LOADER_GUID = "4a67b082-0a4c-41cf-b6c7-440b29bb8c4f"
+        def read_raw_variable(var: str) -> bytes:
+            attr_var = machine.succeed(f"cat /sys/firmware/efi/efivars/{var}-{SD_LOADER_GUID}").encode('raw_unicode_escape')
+            _ = attr_var[:4] # First 4 bytes are attributes according to https://www.kernel.org/doc/html/latest/filesystems/efivarfs.html
+            value = attr_var[4:]
+            return value
+        def read_string_variable(var: str, encoding='utf-16-le') -> str:
+            return read_raw_variable(var).decode(encoding).rstrip('\x00')
+        # By default, it will read a 4 byte value, read `struct` docs to change the format.
+        def assert_variable_uint(var: str, expected: int, format: str = 'I'):
+            with subtest(f"Is `{var}` set to {expected} (uint)"):
+              value, = struct.unpack(f'<{format}', read_raw_variable(var))
+              assert value == expected, f"Unexpected variable value in `{var}`, expected: `{expected}`, actual: `{value}`"
+        def assert_variable_string(var: str, expected: str, encoding='utf-16-le'):
+            with subtest(f"Is `{var}` correctly set"):
+                value = read_string_variable(var, encoding)
+                assert value == expected, f"Unexpected variable value in `{var}`, expected: `{expected.encode(encoding)!r}`, actual: `{value.encode(encoding)!r}`"
+        def assert_variable_string_contains(var: str, expected_substring: str):
+            with subtest(f"Do `{var}` contain expected substrings"):
+                value = read_string_variable(var).strip()
+                assert expected_substring in value, f"Did not find expected substring in `{var}`, expected substring: `{expected_substring}`, actual value: `{value}`"
+      '';
+      tpm2Initialization = ''
+        import subprocess
+        from tempfile import TemporaryDirectory
+
+        # From systemd-initrd-luks-tpm2.nix
+        class Tpm:
+            def __init__(self):
+                self.state_dir = TemporaryDirectory()
+                self.start()
+
+            def start(self):
+                self.proc = subprocess.Popen(["${pkgs.swtpm}/bin/swtpm",
+                    "socket",
+                    "--tpmstate", f"dir={self.state_dir.name}",
+                    "--ctrl", "type=unixio,path=${tpmSocketPath}",
+                    "--tpm2",
+                    ])
+
+                # Check whether starting swtpm failed
+                try:
+                    exit_code = self.proc.wait(timeout=0.2)
+                    if exit_code is not None and exit_code != 0:
+                        raise Exception("failed to start swtpm")
+                except subprocess.TimeoutExpired:
+                    pass
+
+            """Check whether the swtpm process exited due to an error"""
+            def check(self):
+                exit_code = self.proc.poll()
+                if exit_code is not None and exit_code != 0:
+                  raise Exception("swtpm process died")
+
+        tpm = Tpm()
+
+        @polling_condition
+        def swtpm_running():
+          tpm.check()
+      '';
+    in
+    pkgs.nixosTest {
+      inherit name;
+
+      testScript = ''
+        ${lib.optionalString useTPM2 tpm2Initialization}
+        ${lib.optionalString readEfiVariables efiVariablesHelpers}
+        ${testScript}
+      '';
+
 
-  mkSecureBootTest = { name, machine ? { }, useSecureBoot ? true, testScript }: pkgs.nixosTest {
-    inherit name testScript;
       nodes.machine = { lib, ... }: {
         imports = [
           lanzabooteModule
@@ -17,9 +97,23 @@ let
           useBootLoader = true;
           useEFIBoot = true;
 
+          efi.OVMF = pkgs.OVMF.override {
+            secureBoot = useSecureBoot;
+            tpmSupport = useTPM2; # This is needed otherwise OVMF won't initialize the TPM2 protocol.
+          };
+
+
+          qemu.options = lib.mkIf useTPM2 [
+            "-chardev socket,id=chrtpm,path=${tpmSocketPath}"
+            "-tpmdev emulator,id=tpm_dev_0,chardev=chrtpm"
+            "-device ${tpmDeviceModels.${system}},tpmdev=tpm_dev_0"
+          ];
+
           inherit useSecureBoot;
         };
 
+        boot.initrd.availableKernelModules = lib.mkIf useTPM2 [ "tpm_tis" ];
+
         boot.loader.efi = {
           canTouchEfiVariables = true;
         };
@@ -271,9 +365,8 @@ in
   export-efi-variables = mkSecureBootTest {
     name = "lanzaboote-exports-efi-variables";
     machine.environment.systemPackages = [ pkgs.efibootmgr ];
+    readEfiVariables = true;
     testScript = ''
-      import struct
-
       # We will choose to boot directly on the stub.
       # To perform this trick, we will boot first with systemd-boot.
       # Then, we will add a new boot entry in EFI with higher priority
@@ -301,7 +394,6 @@ in
           "test -e /sys/firmware/efi/efivars/LoaderEntrySelected-4a67b082-0a4c-41cf-b6c7-440b29bb8c4f && false || true"
       )
 
-      SD_LOADER_GUID = "4a67b082-0a4c-41cf-b6c7-440b29bb8c4f"
       expected_variables = ["LoaderDevicePartUUID",
         "LoaderImageIdentifier",
         "LoaderFirmwareInfo",
@@ -310,20 +402,6 @@ in
         "StubFeatures"
       ]
 
-      def read_raw_variable(var: str) -> bytes:
-          attr_var = machine.succeed(f"cat /sys/firmware/efi/efivars/{var}-{SD_LOADER_GUID}").encode('raw_unicode_escape')
-          return attr_var[4:] # First 4 bytes are attributes according to https://www.kernel.org/doc/html/latest/filesystems/efivarfs.html
-      def read_string_variable(var: str, encoding='utf-16-le') -> str:
-          return read_raw_variable(var).decode(encoding).rstrip('\x00')
-      def assert_variable_string(var: str, expected: str, encoding='utf-16-le'):
-          with subtest(f"Is `{var}` correctly set"):
-              value = read_string_variable(var, encoding)
-              assert value == expected, f"Unexpected variable value in `{var}`, expected: `{expected.encode(encoding)!r}`, actual: `{value.encode(encoding)!r}`"
-      def assert_variable_string_contains(var: str, expected_substring: str):
-          with subtest(f"Do `{var}` contain expected substrings"):
-              value = read_string_variable(var).strip()
-              assert expected_substring in value, f"Did not find expected substring in `{var}`, expected substring: `{expected_substring}`, actual value: `{value}`"
-
       # Debug all systemd loader specification GUID EFI variables loaded by the current environment.
       print(machine.succeed(f"ls /sys/firmware/efi/efivars/*-{SD_LOADER_GUID}"))
       with subtest("Check if supported variables are exported"):
@@ -344,4 +422,28 @@ in
           assert struct.unpack('<Q', read_raw_variable("StubFeatures")) != 0
     '';
   };
+
+  tpm2-export-efi-variables = mkSecureBootTest {
+    name = "lanzaboote-tpm2-exports-efi-variables";
+    useTPM2 = true;
+    readEfiVariables = true;
+    testScript = ''
+      machine.start()
+
+      # TODO: the other variables are not yet supported.
+      expected_variables = [
+        "StubPcrKernelImage"
+      ]
+
+      # Debug all systemd loader specification GUID EFI variables loaded by the current environment.
+      print(machine.succeed(f"ls /sys/firmware/efi/efivars/*-{SD_LOADER_GUID}"))
+      with subtest("Check if supported variables are exported"):
+          for expected_var in expected_variables:
+            machine.succeed(f"test -e /sys/firmware/efi/efivars/{expected_var}-{SD_LOADER_GUID}")
+
+      # "Static" parts of the UKI is measured in PCR11
+      assert_variable_uint("StubPcrKernelImage", 11)
+    '';
+  };
+
 }

rust/stub/Cargo.lock

@@ -92,6 +92,7 @@ dependencies = [
  "bitflags 2.2.1",
  "goblin",
  "log",
+ "sha1_smol",
  "sha2",
  "uefi",
  "uefi-services",
@@ -176,6 +177,12 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "sha1_smol"
+version = "1.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ae1a47186c03a32177042e55dbc5fd5aee900b8e0069a8d70fba96a9375cd012"
+
 [[package]]
 name = "sha2"
 version = "0.10.6"

rust/stub/Cargo.toml

@@ -17,6 +17,8 @@ log = { version = "0.4.17", default-features = false, features = [ "max_level_in
 
 # Use software implementation because the UEFI target seems to need it.
 sha2 = { version = "0.10.6", default-features = false, features = ["force-soft"] }
+# SHA1 for TPM TCG interface version 1.
+sha1_smol = "1.0.0"
 
 [profile.release]
 opt-level = "s"

rust/stub/src/main.rs

@@ -6,16 +6,21 @@ extern crate alloc;
 
 mod efivars;
 mod linux_loader;
+mod measure;
 mod pe_loader;
 mod pe_section;
+mod tpm;
 mod uefi_helpers;
+mod unified_sections;
 
 use alloc::vec::Vec;
 use efivars::{export_efi_variables, get_loader_features, EfiLoaderFeatures};
-use log::{debug, info, warn};
+use log::{info, warn};
+use measure::measure_image;
 use pe_loader::Image;
 use pe_section::{pe_section, pe_section_as_string};
 use sha2::{Digest, Sha256};
+use tpm::tpm_available;
 use uefi::{
     prelude::*,
     proto::{
@@ -238,10 +243,26 @@ fn main(handle: Handle, mut system_table: SystemTable<Boot>) -> Status {
         warn!("Hash mismatch for initrd!");
     }
 
+    if tpm_available(system_table.boot_services()) {
+        info!("TPM available, will proceed to measurements.");
+        unsafe {
+            // Iterate over unified sections and measure them
+            // For now, ignore failures during measurements.
+            // TODO: in the future, devise a threat model where this can fail
+            // and ensure this hard-fail correctly.
+            let _ = measure_image(
+                &system_table,
+                booted_image_file(system_table.boot_services()).unwrap(),
+            );
+            // TODO: Measure kernel parameters
+            // TODO: Measure sysexts
+        }
+    }
+
     if let Ok(features) = get_loader_features(system_table.runtime_services()) {
         if !features.contains(EfiLoaderFeatures::RandomSeed) {
             // FIXME: process random seed then on the disk.
-            debug!("Random seed is available, but lanzaboote does not support it yet.");
+            info!("Random seed is available, but lanzaboote does not support it yet.");
         }
     }
     export_efi_variables(&system_table).expect("Failed to export stub EFI variables");

rust/stub/src/measure.rs

@@ -0,0 +1,65 @@
+use log::info;
+use uefi::{
+    cstr16,
+    proto::tcg::PcrIndex,
+    table::{runtime::VariableAttributes, Boot, SystemTable},
+};
+
+use crate::{
+    efivars::BOOT_LOADER_VENDOR_UUID, pe_section::pe_section_data, tpm::tpm_log_event_ascii,
+    uefi_helpers::PeInMemory, unified_sections::UnifiedSection,
+};
+
+const TPM_PCR_INDEX_KERNEL_IMAGE: PcrIndex = PcrIndex(11);
+
+pub unsafe fn measure_image(
+    system_table: &SystemTable<Boot>,
+    image: PeInMemory,
+) -> uefi::Result<u32> {
+    let runtime_services = system_table.runtime_services();
+    let boot_services = system_table.boot_services();
+
+    // SAFETY: We get a slice that represents our currently running
+    // image and then parse the PE data structures from it. This is
+    // safe, because we don't touch any data in the data sections that
+    // might conceivably change while we look at the slice.
+    // (data sections := all unified sections that can be measured.)
+    let pe_binary = unsafe { image.as_slice() };
+    let pe = goblin::pe::PE::parse(pe_binary).map_err(|_err| uefi::Status::LOAD_ERROR)?;
+
+    let mut measurements = 0;
+    for section in pe.sections {
+        let section_name = section.name().map_err(|_err| uefi::Status::UNSUPPORTED)?;
+        if let Ok(unified_section) = UnifiedSection::try_from(section_name) {
+            // UNSTABLE: && in the previous if is an unstable feature
+            // https://github.com/rust-lang/rust/issues/53667
+            if unified_section.should_be_measured() {
+                // Here, perform the TPM log event in ASCII.
+                if let Some(data) = pe_section_data(pe_binary, &section) {
+                    info!("Measuring section `{}`...", section_name);
+                    if tpm_log_event_ascii(
+                        boot_services,
+                        TPM_PCR_INDEX_KERNEL_IMAGE,
+                        data,
+                        section_name,
+                    )? {
+                        measurements += 1;
+                    }
+                }
+            }
+        }
+    }
+
+    if measurements > 0 {
+        // If we did some measurements, expose a variable encoding the PCR where
+        // we have done the measurements.
+        runtime_services.set_variable(
+            cstr16!("StubPcrKernelImage"),
+            &BOOT_LOADER_VENDOR_UUID,
+            VariableAttributes::BOOTSERVICE_ACCESS | VariableAttributes::RUNTIME_ACCESS,
+            &TPM_PCR_INDEX_KERNEL_IMAGE.0.to_le_bytes(),
+        )?;
+    }
+
+    Ok(measurements)
+}

rust/stub/src/pe_section.rs

@@ -5,8 +5,21 @@
 #![allow(clippy::bind_instead_of_map)]
 
 use alloc::{borrow::ToOwned, string::String};
+use goblin::pe::section_table::SectionTable;
 
-/// Extracts the data of a section of a loaded PE file.
+/// Extracts the data of a section in a loaded PE file
+/// based on the section table.
+pub fn pe_section_data<'a>(pe_data: &'a [u8], section: &SectionTable) -> Option<&'a [u8]> {
+    let section_start: usize = section.virtual_address.try_into().ok()?;
+
+    assert!(section.virtual_size <= section.size_of_raw_data);
+    let section_end: usize = section_start + usize::try_from(section.virtual_size).ok()?;
+
+    Some(&pe_data[section_start..section_end])
+}
+
+/// Extracts the data of a section of a loaded PE file
+/// based on the section name.
 pub fn pe_section<'a>(pe_data: &'a [u8], section_name: &str) -> Option<&'a [u8]> {
     let pe_binary = goblin::pe::PE::parse(pe_data).ok()?;
 
@@ -14,14 +27,7 @@ pub fn pe_section<'a>(pe_data: &'a [u8], section_name: &str) -> Option<&'a [u8]>
         .sections
         .iter()
         .find(|s| s.name().map(|n| n == section_name).unwrap_or(false))
-        .and_then(|s| {
+        .and_then(|s| pe_section_data(pe_data, s))
-            let section_start: usize = s.virtual_address.try_into().ok()?;
-
-            assert!(s.virtual_size <= s.size_of_raw_data);
-            let section_end: usize = section_start + usize::try_from(s.virtual_size).ok()?;
-
-            Some(&pe_data[section_start..section_end])
-        })
 }
 
 /// Extracts the data of a section of a loaded PE image and returns it as a string.

rust/stub/src/tpm.rs

@@ -0,0 +1,106 @@
+use alloc::vec;
+use core::mem::{self, MaybeUninit};
+use log::warn;
+use uefi::{
+    prelude::BootServices,
+    proto::tcg::{
+        v1::{self, Sha1Digest},
+        v2, EventType, PcrIndex,
+    },
+    table::boot::ScopedProtocol,
+};
+
+fn open_capable_tpm2(boot_services: &BootServices) -> uefi::Result<ScopedProtocol<v2::Tcg>> {
+    let tpm_handle = boot_services.get_handle_for_protocol::<v2::Tcg>()?;
+    let mut tpm_protocol = boot_services.open_protocol_exclusive::<v2::Tcg>(tpm_handle)?;
+
+    let capabilities = tpm_protocol.get_capability()?;
+
+    /*
+     * Here's systemd-stub perform a cast to EFI_TCG_BOOT_SERVICE_CAPABILITY
+     * indicating there could be some quirks to workaround.
+     * It should probably go to uefi-rs?
+    if capabilities.structure_version.major == 1 && capabilities.structure_version.minor == 0 {
+
+    }*/
+
+    if !capabilities.tpm_present() {
+        warn!("Capability `TPM present` is not there for the existing TPM TCGv2 protocol");
+        return Err(uefi::Status::UNSUPPORTED.into());
+    }
+
+    Ok(tpm_protocol)
+}
+
+fn open_capable_tpm1(boot_services: &BootServices) -> uefi::Result<ScopedProtocol<v1::Tcg>> {
+    let tpm_handle = boot_services.get_handle_for_protocol::<v1::Tcg>()?;
+    let mut tpm_protocol = boot_services.open_protocol_exclusive::<v1::Tcg>(tpm_handle)?;
+
+    let status_check = tpm_protocol.status_check()?;
+
+    if status_check.protocol_capability.tpm_deactivated()
+        || !status_check.protocol_capability.tpm_present()
+    {
+        warn!("Capability `TPM present` is not there or `TPM deactivated` is there for the existing TPM TCGv1 protocol");
+        return Err(uefi::Status::UNSUPPORTED.into());
+    }
+
+    Ok(tpm_protocol)
+}
+
+pub fn tpm_available(boot_services: &BootServices) -> bool {
+    open_capable_tpm2(boot_services).is_ok() || open_capable_tpm1(boot_services).is_ok()
+}
+
+/// Log an event in the TPM with `buffer` as data.
+/// Returns a boolean whether the measurement has been done or not in case of success.
+pub fn tpm_log_event_ascii(
+    boot_services: &BootServices,
+    pcr_index: PcrIndex,
+    buffer: &[u8],
+    description: &str,
+) -> uefi::Result<bool> {
+    if pcr_index.0 == u32::MAX {
+        return Ok(false);
+    }
+    if let Ok(mut tpm2) = open_capable_tpm2(boot_services) {
+        let required_size = mem::size_of::<u32>()
+            // EventHeader is private…
+            + mem::size_of::<u32>() + mem::size_of::<u16>() + mem::size_of::<PcrIndex>() + mem::size_of::<EventType>()
+            + description.len();
+
+        let mut event_buffer = vec![MaybeUninit::<u8>::uninit(); required_size];
+        let event = v2::PcrEventInputs::new_in_buffer(
+            event_buffer.as_mut_slice(),
+            pcr_index,
+            EventType::IPL,
+            description.as_bytes(),
+        )?;
+        // FIXME: what do we want as flags here?
+        tpm2.hash_log_extend_event(Default::default(), buffer, event)?;
+    } else if let Ok(mut tpm1) = open_capable_tpm1(boot_services) {
+        let required_size = mem::size_of::<PcrIndex>()
+            + mem::size_of::<EventType>()
+            + mem::size_of::<Sha1Digest>()
+            + mem::size_of::<u32>()
+            + description.len();
+
+        let mut event_buffer = vec![MaybeUninit::<u8>::uninit(); required_size];
+
+        // Compute sha1 of the event data
+        let mut m = sha1_smol::Sha1::new();
+        m.update(description.as_bytes());
+
+        let event = v1::PcrEvent::new_in_buffer(
+            event_buffer.as_mut_slice(),
+            pcr_index,
+            EventType::IPL,
+            m.digest().bytes(),
+            description.as_bytes(),
+        )?;
+
+        tpm1.hash_log_extend_event(event, Some(buffer))?;
+    }
+
+    Ok(true)
+}

rust/stub/src/unified_sections.rs

@@ -0,0 +1,40 @@
+/// List of PE sections that have a special meaning with respect to
+/// UKI specification.
+/// This is the canonical order in which they are measured into TPM
+/// PCR 11.
+/// !!! DO NOT REORDER !!!
+#[repr(u8)]
+pub enum UnifiedSection {
+    Linux = 0,
+    OsRel = 1,
+    CmdLine = 2,
+    Initrd = 3,
+    Splash = 4,
+    Dtb = 5,
+    PcrSig = 6,
+    PcrPkey = 7,
+}
+
+impl TryFrom<&str> for UnifiedSection {
+    type Error = uefi::Error;
+    fn try_from(value: &str) -> Result<Self, Self::Error> {
+        Ok(match value {
+            ".linux" => Self::Linux,
+            ".osrel" => Self::OsRel,
+            ".cmdline" => Self::CmdLine,
+            ".initrd" => Self::Initrd,
+            ".splash" => Self::Splash,
+            ".dtb" => Self::Dtb,
+            ".pcrsig" => Self::PcrSig,
+            ".pcrpkey" => Self::PcrPkey,
+            _ => return Err(uefi::Status::INVALID_PARAMETER.into()),
+        })
+    }
+}
+
+impl UnifiedSection {
+    /// Whether this section should be measured into TPM.
+    pub fn should_be_measured(&self) -> bool {
+        !matches!(self, UnifiedSection::PcrSig)
+    }
+}