Added host-only mode to toggle host mount verifications

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

Author: desultory

readme.md

@@ -70,6 +70,7 @@ Modules write to a shared config dict that is accessible by other modules.
 * `build_dir` (/tmp/initramfs) Defines where the build will take place.
 * `out_dir` (/tmp/initramfs_out) Defines where packed files will be placed.
 * `clean` (true) forces the build dir to be cleaned on each run.
+* `hostonly` (true) adds additional checks to verify the initramfs will work on the build host.
 * `old_count` (1) Sets the number of old file to keep when running the `_rotate_old` function.
 * `file_owner` (portage) sets the owner for items pulled into the initramfs on the build system
 * `binaries` is a list used to define programs to be pulled into the initrams. `which` is used to find the path of added entries, and `lddtree` is used to resolve dependendies.

ugrd/base/core.toml

@@ -2,6 +2,7 @@ file_owner = "portage"
 build_dir = "/tmp/initramfs"
 out_dir = "/tmp/initramfs_out"
 clean = true
+hostonly = true
 library_paths = [ "/lib64" ]
 old_count = 1
 
@@ -38,6 +39,7 @@ build_dir = "Path"  # The directory where the initramfs is built
 out_dir = "Path"  # The directory where the initramfs is packed/output. If no packer is used, this is the final output directory.
 old_count = "int"  # The number of times to cycle old files before deleting
 clean = "bool"  # Add the clean property, used to define if the mounts should be cleaned up after boot
+hostonly = "bool"  # If true, the initramfs will be designed for the host creating it
 file_owner = "str"  # Add the file_owner property, used to define who should own the copied initramfs files
 _file_owner_uid = "int"  # Add the _file_owner_uid property, used to store the uid of the file owner
 _custom_init_file = "str"  # Add the _custom_init_file propety, used to set where the custom init file is located

ugrd/fs/mounts.py

@@ -1,17 +1,19 @@
 __author__ = 'desultory'
-__version__ = '1.1.1'
+__version__ = '1.1.2'
 
 from pathlib import Path
 
 
 MOUNT_PARAMETERS = ['destination', 'source', 'type', 'options', 'base_mount', 'skip_unmount', 'remake_mountpoint']
+SOURCE_TYPES = ['uuid', 'partuuid', 'label']
 
 
 def _process_mounts_multi(self, mount_name, mount_config):
     """
     Processes the passed mounts into fstab mount objects
     under 'mounts'
     """
+    # If the mount already exists, merge the options and update it
     if mount_name in self['mounts']:
         self.logger.info("Updating mount: %s" % mount_name)
         self.logger.debug("[%s] Updating mount with: %s" % (mount_name, mount_config))
@@ -21,14 +23,36 @@ def _process_mounts_multi(self, mount_name, mount_config):
             mount_config.pop('options')
         mount_config = dict(self['mounts'][mount_name], **mount_config)
 
-    for parameter in mount_config:
-        if parameter not in MOUNT_PARAMETERS:
+    # Validate the mount config
+    for parameter, value in mount_config.items():
+        if parameter == 'source' and isinstance(value, dict):
+            for source_type in SOURCE_TYPES:
+                if source_type in value:
+                    break
+            else:
+                self.logger.info("Valid source types: %s" % SOURCE_TYPES)
+                raise ValueError("Invalid source type in mount: %s" % value)
+        elif parameter not in MOUNT_PARAMETERS:
             raise ValueError("Invalid parameter in mount: %s" % parameter)
 
+    # Set defaults
     mount_config['destination'] = Path(mount_config.get('destination', mount_name))
     mount_config['base_mount'] = mount_config.get('base_mount', False)
     mount_config['options'] = set(mount_config.get('options', ''))
 
+    # Check if the mount exists on the host if it's not a base mount
+    if not mount_config['base_mount']:
+        # Only check the root mount after the source has been defined
+        if mount_name == 'root':
+            if 'source' in mount_config:
+                _validate_host_mount(self, mount_config, '/')
+        # The source must be defined for non-root mounts
+        elif 'source' not in mount_config:
+            raise ValueError("[%s] No source specified in mount: %s" % (mount_name, mount_config))
+        else:
+            _validate_host_mount(self, mount_config)
+
+    # Add imports based on the mount type
     if mount_type := mount_config.get('type'):
         if mount_type == 'vfat':
             self['_kmod_depend'] = 'vfat'
@@ -54,14 +78,11 @@ def _get_mount_source(self, mount, pad=False):
 
     out_str = ''
     if isinstance(source, dict):
-        if 'uuid' in source:
-            out_str = f"UUID={source['uuid']}"
-        elif 'partuuid' in source:
-            out_str = f"PARTUUID={source['partuuid']}"
-        elif 'label' in source:
-            out_str = f"LABEL={source['label']}"
-        else:
-            raise ValueError("Unable to process source entry: %s" % repr(source))
+        # Create the source string from the dict
+        for source_type in SOURCE_TYPES:
+            if source_type in source:
+                out_str = f"{source_type.upper()}={source[source_type]}"
+                break
     else:
         out_str = source
 
@@ -157,6 +178,11 @@ def _get_mounts_source(self, mount):
     """
     Returns the source device of a mountpoint on /proc/mounts
     """
+    # Make the mount a string and ensure it starts with a /
+    mount = str(mount)
+    if not mount.startswith('/') and not mount.startswith(' /'):
+        mount = '/' + mount
+
     self.logger.debug("Getting mount source for: %s" % mount)
     # Add space padding to the mount name
     mount = mount if mount.startswith(' ') else ' ' + mount
@@ -177,9 +203,16 @@ def _get_blkid_info(self, device):
     """
     Gets the blkid info for a device
     """
+    from subprocess import run
     self.logger.debug("Getting blkid info for: %s" % device)
 
-    mount_info = self._run(['blkid', str(device)]).stdout.decode().strip()
+    cmd = run(['blkid', str(device)], capture_output=True)
+    if cmd.returncode != 0:
+        self.logger.warning("Unable to find blkid info for: %s" % device)
+        return None
+
+    mount_info = cmd.stdout.decode().strip()
+
     if not mount_info:
         self.logger.warning("Unable to find blkid info for: %s" % device)
         return None
@@ -188,33 +221,50 @@ def _get_blkid_info(self, device):
     return mount_info
 
 
-def mount_root(self):
+def _validate_host_mount(self, mount, destination_path=None):
     """
-    Mounts the root partition.
-    Warns if the root partition isn't found on the current system.
+    Checks if a defined mount exists on the host
     """
-    root_source = self.config_dict['mounts']['root']['source']
-    host_root_dev = _get_mounts_source(self, '/')
+    if not self['hostonly']:
+        self.logger.debug("Skipping host mount check as hostonly is not set")
+        return
+
+    source = mount['source']
+    destination_path = mount['destination'] if destination_path is None else destination_path
 
-    # If the root device is a string, check that it's the same path as the host root mount
-    if isinstance(root_source, str):
-        if root_source != host_root_dev:
-            self.logger.warning("Root device mismatch. Expected: %s, Found: %s" % (root_source, host_root_dev))
-    elif isinstance(root_source, dict):
-        # If the root device is a dict, check that the uuid, partuuid, or label matches the host root mount
-        if blkid_info := _get_blkid_info(self, host_root_dev):
+    host_source_dev = _get_mounts_source(self, destination_path)
+    if not host_source_dev:
+        self.logger.error("Unable to find mount on host system: %s" % destination_path)
+        return
+    else:
+        self.logger.debug("Checking host volume: %s" % host_source_dev)
+
+    if isinstance(source, str):
+        if source != host_source_dev:
+            self.logger.warning("Host device mismatch. Expected: %s, Found: %s" % (source, host_source_dev))
+    elif isinstance(source, dict):
+        # If the source is a dict, check that the uuid, partuuid, or label matches the host mount
+        if blkid_info := _get_blkid_info(self, host_source_dev):
             # Unholy for-else, breaks if the uuid, partuuid, or label matches, otherwise warns
-            for key, value in root_source.items():
-                search_str = f"{key.upper()}=\"{value}\""
+            for key, value in source.items():
+                search_str = f'{key.upper()}="{value}"'
                 if value in blkid_info:
-                    self.logger.debug("Found root device match: %s" % search_str)
-                    break
+                    self.logger.debug("Fount host device match: %s" % search_str)
+                    return True
             else:
-                self.logger.warning("Configuration root device not found on host system. Expected: %s" % root_source)
-                self.logger.warning("Host system root device info: %s" % blkid_info)
+                self.logger.error("Mount device not found on host system. Expected: %s" % source)
+                self.logger.error("Host device info: %s" % blkid_info)
         else:
-            self.logger.warning("Unable to find blkid info for: %s" % root_source)
+            self.logger.warning("Unable to find blkid info for: %s" % host_source_dev)
+
+    raise ValueError("Unable to validate host mount: %s" % mount)
 
+
+def mount_root(self):
+    """
+    Mounts the root partition.
+    Warns if the root partition isn't found on the current system.
+    """
     root_path = self.config_dict['mounts']['root']['destination']
 
     return [f"mount {root_path} || (echo 'Failed to mount root partition' && bash)"]

ugrd/fs/mounts.toml

@@ -7,8 +7,6 @@ binaries = [
   "umount",
 ]
 
-paths = [ "mnt/root" ]
-
 mount_wait = false
 
 [imports.config_processing]