move integrity detection to header validation, add tests

Signed-off-by: Zen <[email protected]>

Author: desultory

src/ugrd/crypto/cryptsetup.py

@@ -8,7 +8,7 @@
 
 from ugrd.exceptions import AutodetectError, ValidationError
 from zenlib.util import colorize as c_
-from zenlib.util import contains, unset, pretty_print
+from zenlib.util import contains, unset
 
 _module_name = "ugrd.crypto.cryptsetup"
 
@@ -151,11 +151,6 @@ def _get_dm_slave_info(self, device_info: dict) -> (str, dict):
     # For integrity backed devices, get the slave's slave
     if self["_vblk_info"].get(slave_source, {}).get("uuid", "").startswith("CRYPT-SUBDEV"):
         slave_source = self["_vblk_info"][slave_source]["slaves"][0]
-        # Set the _dm-integrity flag on the cryptsetup configuration, if it exists
-        if device_info.get("name") in self["cryptsetup"]:
-            self["cryptsetup"][device_info["name"]]["_dm-integrity"] = True
-        else:
-            self.logger.warning(f"[{c_(device_info['name'], 'yellow')}] Unable to set _dm-integrity flag, cryptsetup configuration not found: {pretty_print(self['cryptsetup'])}")
     slave_name = self["_vblk_info"].get(slave_source, {}).get("name")
     search_paths = ["/dev/", "/dev/mapper/"]
 
@@ -238,6 +233,16 @@ def _detect_luks_header_sha(self, luks_info: dict) -> dict:
             self["kernel_modules"] = self._crypto_ciphers[digest["hash"]]["driver"]
 
 
+def _detect_luks_header_integrity(self, luks_info: dict) -> dict:
+    """ Reads the integrity algorithm from the LUKS header,
+    Enables the dm-integrity module, and returns the integrity type. """
+    for segment in luks_info.get("segments", {}).values():
+        if integrity_type := segment.get("integrity", {}).get("type"):
+            self["kernel_modules"] = "dm_integrity"
+            return integrity_type
+
+
+
 @contains("cryptsetup_header_validation", "Skipping cryptsetup header validation.", log_level=30)
 def _validate_cryptsetup_header(self, mapped_name: str) -> None:
     """Validates configured cryptsetup volumes against the LUKS header."""
@@ -274,6 +279,10 @@ def _validate_cryptsetup_header(self, mapped_name: str) -> None:
 
     _detect_luks_header_aes(self, luks_info)
     _detect_luks_header_sha(self, luks_info)
+    if integrity_type := _detect_luks_header_integrity(self, luks_info):
+        self.logger.info(f"[{c_(mapped_name, 'blue')}] Detected dm-integrity type: {c_(integrity_type, 'cyan')}")
+        self["cryptsetup"][mapped_name]["_dm-integrity"] = integrity_type
+
 
     if not self["argon2"]:  # if argon support was not detected, check if the header wants it
         for keyslot in luks_info.get("keyslots", {}).values():

src/ugrd/fs/mounts.py

@@ -1,5 +1,5 @@
 __author__ = "desultory"
-__version__ = "7.2.0"
+__version__ = "7.2.1"
 
 from pathlib import Path
 from re import search
@@ -562,7 +562,7 @@ def _autodetect_dm(self, mountpoint, device=None) -> None:
         self.logger.info(f"[{c_(dev_name, 'blue')}] Slave is a CRYPT-SUBDEV, using its slave instead: {c_(slave_source, 'cyan')}")
         # Add the kmod for it
         self.logger.info(f"[{c_(dev_name, 'blue')}] Adding kmod for CRYPT-SUBDEV: {c_('dm-crypt', 'magenta')}")
-        self["_kmod_auto"] = "dm_integrity"
+        self["kernel_modules"] = "dm_integrity"
 
     autodetect_mount_kmods(self, slave_source)
 

src/ugrd/fs/test_image.py

@@ -1,13 +1,14 @@
-__version__ = "2.0.0"
+__version__ = "2.1.0"
 
+from re import match
 from tempfile import TemporaryDirectory
 
 from zenlib.util import colorize as c_
 from zenlib.util import contains
 
-
 MIN_FS_SIZES = {"btrfs": 110, "f2fs": 50}
 
+
 @contains("test_flag", "A test flag must be set to create a test image", raise_exception=True)
 def init_banner(self):
     """Initialize the test image banner, set a random flag if not set."""
@@ -33,12 +34,13 @@ def _allocate_image(self, image_path, padding=0):
 
     with open(image_path, "wb") as f:
         total_size = (self.test_image_size + padding) * (2**20)  # Convert MB to bytes
-        self.logger.info(f"Allocating {self.test_image_size + padding}MB test image file: { c_(f.name, 'green')}")
-        self.logger.debug(f"[{f.name}] Total bytes: { c_(total_size, 'green')}")
+        self.logger.info(f"Allocating {self.test_image_size + padding}MB test image file: {c_(f.name, 'green')}")
+        self.logger.debug(f"[{f.name}] Total bytes: {c_(total_size, 'green')}")
         f.write(b"\0" * total_size)
 
+
 def _copy_fs_contents(self, image_path, build_dir):
-    """ Mount and copy the filesystem contents into the image,
+    """Mount and copy the filesystem contents into the image,
     for filesystems which cannot be created directly from a directory"""
     try:
         with TemporaryDirectory() as tmp_dir:
@@ -84,6 +86,14 @@ def make_test_luks_image(self, image_path):
     keyfile_path = _get_luks_keyfile(self)
     self.logger.info("Using LUKS keyfile: %s" % c_(keyfile_path, "green"))
     self.logger.info("Creating LUKS image: %s" % c_(image_path, "green"))
+    extra_args = []
+    if integrity_type := _get_luks_config(self).get("_dm-integrity"):
+        # If it's the type reported in the header like <type>(<algo>), turn it into <type>-<algo> for arg usage
+        if m := match(r"^(?P<type>\w+)\((?P<algo>[\w-]+)\)$", integrity_type):
+            integrity_type = f"{m['type']}-{m['algo']}"
+        self.logger.info(f"[{c_(image_path, 'green')}] LUKS integrity type: {c_(integrity_type, 'cyan')}")
+        extra_args.extend(["--integrity", integrity_type])
+
     self._run(
         [
             "cryptsetup",
@@ -94,6 +104,7 @@ def make_test_luks_image(self, image_path):
             "--batch-mode",
             "--key-file",
             keyfile_path,
+            *extra_args,
         ]
     )
     self.logger.info("Opening LUKS image: %s" % c_(image_path, "magenta"))

tests/test_cryptsetup.py

@@ -31,6 +31,16 @@ def test_cryptsetup_included_key(self):
         generator = InitramfsGenerator(logger=self.logger, config="tests/cryptsetup_included_key.toml")
         generator.build()
 
+    def test_cryptsetup_integrity(self):
+        """Tests LUKS based roots using a keyfile included in the initramfs with integrity protection"""
+        generator = InitramfsGenerator(
+            logger=self.logger,
+            config="tests/cryptsetup_included_key.toml",
+            kernel_modules=["dm-integrity"],  # Specify this because its usually auto-detected during header validation
+            cryptsetup={"root": {"_dm-integrity": "hmac(sha256)"}},  # Use the type like defined in the header, not the proper args to test processing
+        )
+        generator.build()
+
 
 if __name__ == "__main__":
     main()