clean luks-autodetect

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

Author: desultory

readme.md

@@ -16,6 +16,7 @@ The original goal of this project was to create an initramfs suitable for decryp
   - Allows for late insertion of a smartcard
   - Can fail back to plain password entry
 * Auto-detection and validation of the root mount using `/proc/mounts`
+* Auto-detection and validation of LUKS root mounts.
 * Auto-detection and validation of the btrfs subvolume used for the root mount, if present.
 * Dynamic BTRFS subvolume selection at boot time using `subvol_selector`.
 * Auto-detection of kernel modules using `lspci` and `lsmod`
@@ -297,6 +298,7 @@ Similarly `ugrd.kmod.novideo` `nonetwork`, and `nosound` exist to ignore video,
 ### Filesystem modules
 
 `autodetect_root` (true) Set the root mount parameter based on the current root label or uuid.
+`autodetect_root_luks` (true) Attempt to automatically configure LUKS mounts for the root device.
 
 #### ugrd.fs.mounts
 

src/ugrd/crypto/cryptsetup.py

@@ -1,5 +1,5 @@
 __author__ = 'desultory'
-__version__ = '1.4.3'
+__version__ = '1.4.4'
 
 from zenlib.util import check_dict
 
@@ -106,7 +106,7 @@ def get_crypt_sources(self) -> list[str]:
         if parameters.get('path') and not self['validate']:
             self.logger.warning("Using device paths is unreliable and can result in boot failures.")
             out += [f"export CRYPTSETUP_SOURCE_{name}={parameters.get('path')}"]
-        elif not parameters.get('partuuid') and not parameters.get('uuid'):
+        elif not parameters.get('partuuid') and not parameters.get('uuid') and parameters.get('path'):
             raise ValueError("Validation must be disabled to use device paths with the cryptsetup module.")
         else:
             try:

src/ugrd/fs/mounts.py

@@ -193,61 +193,67 @@ def _get_dm_devices(self, f_major=None, f_minor=None) -> dict:
     return dm_devices
 
 
-@check_dict('autodetect_root', value=True, log_level=10, message="Skipping root autodetection, autodetect_root is not set.")
-@check_dict({'mounts': {'root': 'source'}}, unset=True, log_level=30, message="Skipping root autodetection, root source is already set.")
-def autodetect_root(self) -> None:
-    """ Sets self['mounts']['root']['source'] based on the host mount. """
-    root_mount_info = _get_blkid_info(self, _get_mounts_source_device(self, '/'))
-    self.logger.debug("Detected root mount info: %s" % root_mount_info)
-
+@check_dict('autodetect_root_luks', value=True, log_level=10, message="Skipping LUKS autodetection, autodetect_root_luks is not set.")
+def _autodetect_root_luks(self, root_mount_info: dict) -> None:
     # Check if the mount is under /dev/mapper or starts with /dev/dm-
-    if root_mount_info['name'].startswith('/dev/mapper') or root_mount_info['name'].startswith('/dev/dm-'):
-        mount_loc = Path(root_mount_info['name']).resolve()
-        self.logger.debug("Detected a device mapper mount: %s" % mount_loc)
-
-        major, minor = mount_loc.stat().st_rdev >> 8, mount_loc.stat().st_rdev & 0xFF
-        self.logger.debug("[%s] Major: %s, Minor: %s" % (mount_loc, major, minor))
-        dm_info = _get_dm_devices(self, major, minor)
-
-        if len(dm_info) > 1:  # there should only be one device mapper device associated with the mount
-            self.logger.error("Device mapper devices: %s" % dm_info)
-            raise RuntimeError("Multiple device mapper devices found for: %s" % mount_loc)
-
-        mapped_name, dm_info = dm_info.popitem()
-
-        if mount_loc.name != dm_info['name'] and mount_loc.name != mapped_name:
-            raise ValueError("Device mapper device name mismatch: %s != %s" % (mount_loc.name, dm_info['name']))
+    if not root_mount_info['name'].startswith('/dev/mapper') and not root_mount_info['name'].startswith('/dev/dm-'):
+        self.logger.debug("Root mount is not a device mapper mount: %s" % root_mount_info['name'])
+        return
+    mount_loc = Path(root_mount_info['name']).resolve()
+    self.logger.debug("Detected a device mapper mount: %s" % mount_loc)
+
+    major, minor = mount_loc.stat().st_rdev >> 8, mount_loc.stat().st_rdev & 0xFF
+    self.logger.debug("[%s] Major: %s, Minor: %s" % (mount_loc, major, minor))
+    dm_info = _get_dm_devices(self, major, minor)
+
+    if len(dm_info) > 1:  # there should only be one device mapper device associated with the mount
+        self.logger.error("Device mapper devices: %s" % dm_info)
+        raise RuntimeError("Multiple device mapper devices found for: %s" % mount_loc)
+
+    mapped_name, dm_info = dm_info.popitem()
+
+    if mount_loc.name != dm_info['name'] and mount_loc.name != mapped_name:
+        raise ValueError("Device mapper device name mismatch: %s != %s" % (mount_loc.name, dm_info['name']))
+
+    if len(dm_info['holders']) > 0:
+        self.logger.error("Device mapper holders: %s" % dm_info['holders'])
+        raise RuntimeError("LUKS volumes should not have holders, potential LVM volume: %s" % mount_loc.name)
+
+    if len(dm_info['slaves']) == 0:
+        raise RuntimeError("No slaves found for device mapper device, unknown type: %s" % mount_loc.name)
+    elif len(dm_info['slaves']) > 1:
+        self.logger.error("Device mapper slaves: %s" % dm_info['slaves'])
+        raise RuntimeError("Multiple slaves found for device mapper device, unknown type: %s" % mount_loc.name)
+
+    luks_mount = _get_blkid_info(self, Path('/dev/' + dm_info['slaves'][0]))
+    if luks_mount.get('type') != 'crypto_LUKS':
+        if not luks_mount.get('uuid'):
+            self.logger.error("[%s] Unknown device mapper slave type: %s" % (dm_info['slaves'][0], luks_mount.get('type')))
+        else:
+            raise RuntimeError("[%s] Unknown device mapper slave type: %s" % (dm_info['slaves'][0], luks_mount.get('type')))
 
-        if len(dm_info['holders']) > 0:
-            self.logger.error("Device mapper holders: %s" % dm_info['holders'])
-            raise RuntimeError("LUKS volumes should not have holders, potential LVM volume: %s" % mount_loc.name)
+    if 'ugrd.crypto.cryptsetup' not in self['modules']:
+        self.logger.info("Autodetected LUKS mount, enabling the cryptsetup module: %s" % luks_mount['name'])
+        self['modules'] = 'ugrd.crypto.cryptsetup'
 
-        if len(dm_info['slaves']) == 0:
-            raise RuntimeError("No slaves found for device mapper device, unknown type: %s" % mount_loc.name)
-        elif len(dm_info['slaves']) > 1:
-            self.logger.error("Device mapper slaves: %s" % dm_info['slaves'])
-            raise RuntimeError("Multiple slaves found for device mapper device, unknown type: %s" % mount_loc.name)
+    if uuid := luks_mount.get('uuid'):
+        self.logger.info("[%s] Detected LUKS volume uuid: %s" % (mount_loc.name, uuid))
+        self['cryptsetup'] = {dm_info['name']: {'uuid': uuid}}
+    elif partuuid := luks_mount.get('partuuid'):
+        self.logger.info("[%s] Detected LUKS volume partuuid: %s" % (mount_loc.name, partuuid))
+        self['cryptsetup'] = {dm_info['name']: {'partuuid': partuuid}}
 
-        luks_mount = _get_blkid_info(self, Path('/dev/' + dm_info['slaves'][0]))
-        if luks_mount.get('type') != 'crypto_LUKS':
-            if not luks_mount.get('uuid'):
-                self.logger.error("[%s] Unknown device mapper slave type: %s" % (dm_info['slaves'][0], luks_mount.get('type')))
-            else:
-                raise RuntimeError("[%s] Unknown device mapper slave type: %s" % (dm_info['slaves'][0], luks_mount.get('type')))
+    self.logger.info("[%s] Configuring cryptsetup for LUKS mount (%s) on: %s\n%s" %
+                     (mount_loc.name, dm_info['name'], luks_mount['name'], pretty_print(self['cryptsetup'])))
 
-        if 'ugrd.crypto.cryptsetup' not in self['modules']:
-            self.logger.info("Autodetected LUKS mount, enabling the cryptsetup module: %s" % luks_mount['name'])
-            self['modules'] = 'ugrd.crypto.cryptsetup'
 
-        if uuid := luks_mount.get('uuid'):
-            self.logger.info("[%s] Detected LUKS volume uuid: %s" % (mount_loc.name, uuid))
-            self['cryptsetup'] = {dm_info['name']: {'uuid': uuid}}
-        elif partuuid := luks_mount.get('partuuid'):
-            self.logger.info("[%s] Detected LUKS volume partuuid: %s" % (mount_loc.name, partuuid))
-            self['cryptsetup'] = {dm_info['name']: {'partuuid': partuuid}}
-
-        self.logger.info("[%s] Configuring cryptsetup for LUKS mount (%s) on: %s\n%s" %
-                         (mount_loc.name, dm_info['name'], luks_mount['name'], pretty_print(self['cryptsetup'])))
+@check_dict('autodetect_root', value=True, log_level=10, message="Skipping root autodetection, autodetect_root is not set.")
+@check_dict({'mounts': {'root': 'source'}}, unset=True, log_level=30, message="Skipping root autodetection, root source is already set.")
+def autodetect_root(self) -> None:
+    """ Sets self['mounts']['root']['source'] based on the host mount. """
+    root_mount_info = _get_blkid_info(self, _get_mounts_source_device(self, '/'))
+    self.logger.debug("Detected root mount info: %s" % root_mount_info)
+    _autodetect_root_luks(self, root_mount_info)
 
     mount_info = {'root': {'type': 'auto', 'base_mount': False}}
 

src/ugrd/fs/mounts.toml

@@ -8,6 +8,7 @@ binaries = [
 
 mount_wait = false
 autodetect_root = true
+autodetect_root_luks = true
 
 [imports.config_processing]
 "ugrd.fs.mounts" = [ "_process_mounts_multi", "_process_mount_timeout" ]
@@ -42,6 +43,7 @@ mount_wait = "bool"  # Add the mount_wait property, if defined, user input will
 mount_timeout = "float"  # Add the mount_timeout property, used to define the timeout for mount_wait
 mount_cmd = "str"  # The mount command called by mount_root, can be overridden
 autodetect_root = "bool"  # Add the autodetect_root property, if defined, the root mount will be autodetected
+autodetect_root_luks = "bool"  # Whether or not to try to autodetect LUKS partitions
 
 
 # Define the base of the root mount

src/ugrd/main.py

@@ -23,6 +23,8 @@ def main():
                  {'flags': ['--no-firmware'], 'action': 'store_false', 'help': 'Exclude firmware files.', 'dest': 'kmod_pull_firmware'},
                  {'flags': ['--autodetect-root'], 'action': 'store_true', 'help': 'Autodetect the root partition.'},
                  {'flags': ['--no-autodetect-root'], 'action': 'store_false', 'help': 'Do not autodetect the root partition.', 'dest': 'autodetect_root'},
+                 {'flags': ['--autodetect-root-luks'], 'action': 'store_true', 'help': 'Autodetect LUKS volumes under the root partition.'},
+                 {'flags': ['--no-autodetect-root-luks'], 'action': 'store_false', 'help': 'Do not autodetect root LUKS volumes.', 'dest': 'autodetect_root'},
                  {'flags': ['--print-config'], 'action': 'store_true', 'help': 'Print the final config dict.'},
                  {'flags': ['out_file'], 'action': 'store', 'help': 'Output file location', 'nargs': '?'}]