Make the root mount return a warning if that device is not detected

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

Author: desultory

readme.md

@@ -16,7 +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
 * Key entry over serial
-* Automatic CPIO generation
+* Automatic CPIO generation (gen_init_cpio)
 
 ## Usage
 
@@ -350,6 +350,8 @@ Modules can load other modules using the `modules` directive, be careful conside
 
 If a module depends on another module, it can be added to the `mod_depends` list in the module config. A `ValueError` will be thrown if the module is not present.
 
+`_module_name` can be set within a module for logging purposes, it is verified to be accurate when imported but optional.
+
 #### imports
 
 The most powerful part of a module is the `imports` directive.

ugrd/fs/mounts.py

@@ -173,32 +173,47 @@ def _get_mounts_source(self, mount):
     self.logger.warning("Unable to find mount source for: %s" % mount)
 
 
-def _get_lsblk_info(self, mount, output_fields="NAME,UUID,PARTUUID,LABEL"):
+def _get_blkid_info(self, device):
     """
-    Gets the lsblk info for a mountpoint
+    Gets the blkid info for a device
     """
-    from json import loads, JSONDecodeError
+    self.logger.debug("Getting blkid info for: %s" % device)
 
-    self.logger.debug("Getting lsblk info for: %s" % mount)
-
-    mount_info = self._run(['lsblk', '--json', '--output', output_fields, str(mount)])
-    try:
-        mount_info = loads(mount_info.stdout)
-    except JSONDecodeError:
-        self.logger.warning("Unable to parse lsblk info for: %s" % mount)
+    mount_info = self._run(['blkid', str(device)]).stdout.decode().strip()
+    if not mount_info:
+        self.logger.warning("Unable to find blkid info for: %s" % device)
         return None
 
-    self.logger.debug("Found lsblk info: %s" % mount_info)
+    self.logger.debug("Found blkid info: %s" % mount_info)
     return mount_info
 
+
 def mount_root(self):
     """
     Mounts the root partition.
     Warns if the root partition isn't found on the current system.
     """
-#    root_source = self.config_dict['mounts']['root']['source']
+    root_source = self.config_dict['mounts']['root']['source']
     host_root_dev = _get_mounts_source(self, '/')
-    lsblk_info = _get_lsblk_info(self, host_root_dev)
+
+    # 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):
+            # 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}\""
+                if value in blkid_info:
+                    self.logger.debug("Found root device match: %s" % search_str)
+                    break
+            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)
+        else:
+            self.logger.warning("Unable to find blkid info for: %s" % root_source)
 
     root_path = self.config_dict['mounts']['root']['destination']
 

ugrd/initramfs_generator.py

@@ -96,7 +96,7 @@ def _run_func(self, function, external=False, return_output=True):
                 self.logger.debug("[%s] Function returned string: %s" % (function.__name__, function_output))
                 return [function_output]
             else:
-                self.logger.debug("[%s] Function returned output: %s" % (function.__name__, function_output))
+                self.logger.debug("[%s] Function returned output: %s" % (function.__name__, pretty_print(function_output)))
                 return function_output
         else:
             self.logger.debug("[%s] Function returned no output" % function.__name__)
@@ -268,7 +268,7 @@ def _write(self, file_name, contents, chmod_mask=0o644, in_build_dir=True):
                 self.logger.info("Deleting file: %s" % file_path)
                 file_path.unlink()
 
-        self.logger.debug("[%s] Writing contents:\n%s" % (file_path, pretty_print(contents)))
+        self.logger.debug("[%s] Writing contents:\n%s" % (file_path, contents))
         with open(file_path, 'w') as file:
             file.writelines("\n".join(contents))
 
@@ -325,7 +325,7 @@ def _rotate_old(self, file_name: Path, sequence=0):
         # If the cycle count is not set, attempt to clean
         if not self.old_count:
             if self.clean:
-                self.logger.info("Deleting file: %s" % file_name)
+                self.logger.warning("Deleting file: %s" % file_name)
                 file_name.unlink()
                 return
             else:
@@ -345,7 +345,7 @@ def _rotate_old(self, file_name: Path, sequence=0):
             if sequence >= self.old_count:
                 # Clean the last file in the sequence if clean is enabled
                 if self.clean:
-                    self.logger.info("Deleting old file: %s" % target_file)
+                    self.logger.warning("Deleting old file: %s" % target_file)
                     target_file.unlink()
                 else:
                     self.logger.debug("Cycle limit reached")
@@ -392,7 +392,7 @@ def _run(self, args):
         """
         Runs a command, returns the object
         """
-        self.logger.debug("Running command: %s" % args)
+        self.logger.debug("Running command: %s" % ' '.join(args))
         cmd = run(args, capture_output=True)
         if cmd.returncode != 0:
             self.logger.error("Failed to run command: %s" % cmd.args)