made libary path detction more generic, added old_count to control how old files are cycled

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.
+* `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.
 * `paths` is a list of directores to create in the `build_dir`. They do not need a leading `/`.

ugrd/base/core.py

@@ -1,5 +1,5 @@
 __author__ = 'desultory'
-__version__ = '1.0.1'
+__version__ = '1.1.0'
 
 from pathlib import Path
 
@@ -82,15 +82,49 @@ def deploy_nodes(self):
             raise e
 
 
+def configure_library_paths(self):
+    """
+    Sets the export LD_LIBRARY_PATH variable to the library paths
+    """
+    library_paths = ":".join(self.config_dict['library_paths'])
+    self.logger.debug("Setting LD_LIBRARY_PATH to: %s" % library_paths)
+    return "export LD_LIBRARY_PATH=%s" % library_paths
+
+
+def _process_paths_multi(self, path):
+    """
+    Converts the input to a Path if it is not one
+    """
+    self.logger.log(5, "Processing path: %s" % path)
+    if not isinstance(path, Path):
+        path = Path(path)
+
+    self.logger.debug("Adding path: %s" % path)
+    self['paths'].append(path)
+
+
 def _process_binaries_multi(self, binary):
     """
     Processes binaries into the binaries list, adding dependencies along the way.
     """
     self.logger.debug("Processing binary: %s" % binary)
 
     dependencies = calculate_dependencies(self, binary)
-    self['dependencies'] = dependencies
-
+    # The first dependency will be the path of the binary itself, don't add this to the library paths
+    first_dep = True
+    for dependency in dependencies:
+        self['dependencies'] = dependency
+        if first_dep:
+            self.logger.debug("Skipping adding library path for first dependency: %s" % dependency)
+            first_dep = False
+            continue
+        if dependency.parent not in self['library_paths']:
+            self.logger.info("Adding library path: %s" % dependency.parent)
+            # Make it a string so NoDupFlatList can handle it
+            # It being derived from a path should ensure it's a proper path
+            self['library_paths'] = str(dependency.parent)
+
+    self.logger.debug("Adding binary: %s" % binary)
     self['binaries'].append(binary)
 
 
@@ -104,6 +138,7 @@ def _process_dependencies_multi(self, dependency):
     if not dependency.exists():
         raise FileNotFoundError("Dependency does not exist: %s" % dependency)
 
+    self.logger.debug("Adding dependency: %s" % dependency)
     self['dependencies'].append(dependency)
 
 
@@ -112,12 +147,13 @@ def _process_copies_multi(self, copy_name, copy_parameters):
     Processes a copy from the copies parameter
     Ensures the source and target are defined in the parameters.
     """
-    self.logger.debug("[%s] Processing copy: %s" % (copy_name, copy_parameters))
+    self.logger.log(5, "[%s] Processing copies: %s" % (copy_name, copy_parameters))
     if 'source' not in copy_parameters:
         raise ValueError("[%s] No source specified" % copy_name)
     if 'destination' not in copy_parameters:
         raise ValueError("[%s] No target specified" % copy_name)
 
+    self.logger.debug("[%s] Adding copies: %s" % (copy_name, copy_parameters))
     self['copies'][copy_name] = copy_parameters
 
 
@@ -126,12 +162,13 @@ def _process_symlinks_multi(self, symlink_name, symlink_parameters):
     Processes a symlink,
     Ensures the source and target are defined in the parameters.
     """
-    self.logger.debug("[%s] Processing symlink: %s" % (symlink_name, symlink_parameters))
+    self.logger.log(5, "[%s] Processing symlink: %s" % (symlink_name, symlink_parameters))
     if 'source' not in symlink_parameters:
         raise ValueError("[%s] No source specified" % symlink_name)
     if 'target' not in symlink_parameters:
         raise ValueError("[%s] No target specified" % symlink_name)
 
+    self.logger.debug("[%s] Adding symlink: %s -> %s" % (symlink_name, symlink_parameters['source'], symlink_parameters['target']))
     self['symlinks'][symlink_name] = symlink_parameters
 
 
@@ -152,6 +189,7 @@ def _process_nodes_multi(self, node_name, node_config):
         node_config['mode'] = 0o660
         self.logger.debug("[%s] No mode specified, assuming: %s" % (node_name, node_config['mode']))
 
+    self.logger.debug("[%s] Adding node: %s" % (node_name, node_config))
     self['nodes'][node_name] = node_config
 
 

ugrd/base/core.toml

@@ -2,6 +2,8 @@ file_owner = "portage"
 build_dir = "/tmp/initramfs"
 out_dir = "/tmp/initramfs_out"
 clean = true
+library_paths = [ "/lib64" ]
+old_count = 1
 
 [nodes.console]
 mode = 0o644
@@ -14,21 +16,27 @@ minor = 1
 		     "_process_copies_multi",
 		     "_process_symlinks_multi",
 		     "_process_file_owner",
-		     "_process_nodes_multi"
+		     "_process_nodes_multi",
+		     "_process_paths_multi",
 		   ]
 
 [imports.build_tasks]
 "ugrd.base.core" = [ "deploy_dependencies", "deploy_copies", "deploy_nodes", "deploy_symlinks" ]
 
+[imports.init_pre]
+"ugrd.base.core" = [ "configure_library_paths" ]
+
 [custom_parameters]
 binaries = "NoDupFlatList"  # Binaries which should be included in the intiramfs, dependencies resolved with lddtree
 dependencies = "NoDupFlatList"  # Add the dependencies property, used to define the dependencies of the initramfs
+library_paths = "NoDupFlatList"  # Add the library_paths property, used to define the library paths to add to LD_LIBRARY_PATH
 symlinks = "dict"  # Add the symlinks property, used to define the symlinks to be made in the initramfs
 copies = "dict"  # Add the copies property, used to define the files to be copied to the initramfs
 nodes = "dict"  # Add the nodes property, used to define the device nodes to be created
 paths = "NoDupFlatList"  # Paths to be copied to the initramfs
 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
 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

ugrd/base/cpio.py

@@ -38,6 +38,8 @@ def pack_cpio(self):
     self.logger.info("Creating CPIO file from packing list: %s" % packing_list)
 
     out_cpio = self.out_dir / self.config_dict['out_file']
+    if out_cpio.exists():
+        self._rotate_old(out_cpio)
 
     with open(out_cpio, 'wb') as cpio_file:
         cmd = run([str(cpio_path), packing_list], stdout=cpio_file)

ugrd/crypto/cryptsetup.py

@@ -1,6 +1,6 @@
 __author__ = 'desultory'
 
-__version__ = '0.8.1'
+__version__ = '0.8.2'
 
 
 CRYPTSETUP_PARAMETERS = ['key_type', 'partuuid', 'uuid', 'key_file', 'header_file', 'retries', 'key_command', 'reset_command', 'try_nokey']
@@ -56,13 +56,6 @@ def _process_cryptsetup_multi(self, mapped_name, config):
     self['cryptsetup'][mapped_name] = config
 
 
-def configure_library_dir(self):
-    """
-    exports the libtary path for cryptsetup
-    """
-    return 'export LD_LIBRARY_PATH=/lib64'
-
-
 def get_crypt_sources(self):
     """
     Goes through each cryptsetup device, sets $CRYPTSETUP_SOURCE_NAME to the source device
@@ -166,14 +159,3 @@ def crypt_init(self):
             out += ['fi']
     return out
 
-
-def find_libgcc(self):
-    """
-    Finds libgcc.so, adds a copies item for it.
-    """
-    ldconfig = self._run(['ldconfig', '-p']).stdout.decode().split("\n")
-    libgcc = [lib for lib in ldconfig if 'libgcc_s' in lib and 'libc6,x86-64' in lib][0]
-    source_path = libgcc.partition('=> ')[-1]
-    self.logger.debug("Source path for libgcc_s: %s" % source_path)
-
-    self.config_dict['copies']['libgcc_s'] = {'source': source_path, 'destination': '/lib64/'}

ugrd/crypto/cryptsetup.toml

@@ -10,18 +10,12 @@ cryptsetup_autoretry = true
 [imports.config_processing]
 "ugrd.crypto.cryptsetup" = [ "_process_cryptsetup_multi", "_process_cryptsetup_key_types_multi" ]
 
-[imports.init_pre]
-"ugrd.crypto.cryptsetup" = [ "configure_library_dir" ]
-
 [imports.init_early]
 "ugrd.crypto.cryptsetup" = [ "get_crypt_sources" ]
 
 [imports.init_main]
 "ugrd.crypto.cryptsetup" = [ "crypt_init" ]
 
-[imports.build_pre]
-"ugrd.crypto.cryptsetup" = [ "find_libgcc" ]
-
 [cryptsetup_key_types.keyfile]
 key_command = "cat {key_file} >"
 

ugrd/initramfs_dict.py

@@ -1,6 +1,6 @@
 
 __author__ = "desultory"
-__version__ = "0.8.2"
+__version__ = "0.8.3"
 
 from tomllib import load
 from pathlib import Path
@@ -43,6 +43,7 @@ def import_args(self, args: dict):
     def __setitem__(self, key, value):
         # If the type is registered, use the appropriate update function
         if expected_type := self.builtin_parameters.get(key, self['custom_parameters'].get(key)):
+            self.logger.log(5, "[%s] Expected type: %s" % (key, expected_type))
             if hasattr(self, f"_process_{key}"):
                 self.logger.debug("Using builtin setitem for: %s" % key)
                 getattr(self, f"_process_{key}")(value)
@@ -66,6 +67,8 @@ def __setitem__(self, key, value):
             else:
                 super().__setitem__(key, expected_type(value))
         else:  # Otherwise set it like a normal dict item
+            self.logger.debug("[%s] Unable to determine expected type, valid builtin types: %s" % (key, self.builtin_parameters.keys()))
+            self.logger.debug("[%s] Custom types: %s" % (key, self['custom_parameters'].keys()))
             if key.startswith('_'):
                 self.logger.warning("Setting unknown internal paramaters '%s' with value: %s" % (key, value))
             else:
@@ -97,6 +100,8 @@ def _process_custom_parameters(self, parameter_name, parameter_type):
             super().__setitem__(parameter_name, eval(parameter_type)())
         elif parameter_type == "bool":
             super().__setitem__(parameter_name, False)
+        elif parameter_type == "int":
+            super().__setitem__(parameter_name, 0)
 
     @handle_plural
     def _process_imports(self, import_type: str, import_value: dict):

ugrd/initramfs_generator.py

@@ -1,6 +1,6 @@
 
 __author__ = "desultory"
-__version__ = "0.8.0"
+__version__ = "0.8.1"
 
 from tomllib import load
 from pathlib import Path
@@ -42,7 +42,7 @@ def load_config(self):
 
         self.logger.debug("Loaded config: %s" % self.config_dict)
 
-        for parameter in ['build_dir', 'out_dir', 'clean']:
+        for parameter in ['build_dir', 'out_dir', 'clean', 'old_count']:
             dict_value = self.config_dict[parameter]
             if dict_value is not None:
                 setattr(self, parameter, dict_value)
@@ -192,8 +192,7 @@ def generate_structure(self):
             self._mkdir(self.build_dir)
 
         for subdir in set(self.config_dict['paths']):
-            subdir_path = Path(subdir)
-            subdir_relative_path = subdir_path.relative_to(subdir_path.anchor)
+            subdir_relative_path = subdir.relative_to(subdir.anchor)
             target_dir = self.build_dir / subdir_relative_path
 
             self._mkdir(target_dir)
@@ -266,6 +265,12 @@ def _write(self, file_name, contents, chmod_mask=0o644, in_build_dir=True):
             self.logger.debug("Parent directory for '%s' does not exist: %s" % (file_path.name, file_path))
             self._mkdir(file_path.parent)
 
+        if file_path.is_file():
+            self.logger.warning("File already exists: %s" % file_path)
+            if self.clean:
+                self.logger.info("Deleting file: %s" % file_path)
+                file_path.unlink()
+
         self.logger.debug("[%s] Writing contents:\n%s" % (file_path, pretty_print(contents)))
         with open(file_path, 'w') as file:
             file.writelines("\n".join(contents))
@@ -311,6 +316,51 @@ def _copy(self, source, dest=None, in_build_dir=True):
 
         self._chown(dest_path)
 
+    def _rotate_old(self, file_name: Path, sequence=0):
+        """
+        Copies a file to file_name.old then file_nane.old.n, where n is the next number in the sequence
+        """
+        # Nothing to do if the file doesn't exist
+        if not file_name.is_file():
+            self.logger.debug("File does not exist: %s" % file_name)
+            return
+
+        # 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)
+                file_name.unlink()
+                return
+            else:
+                # Fail if the cycle count is not set and clean is disabled
+                raise RuntimeError("Unable to cycle file, as cycle count is not set and clean is disabled: %s" % file_name)
+
+        self.logger.debug("[%d] Cycling file: %s" % (sequence, file_name))
+
+        # If the sequence is 0, we're cycling the file for the first time, just rename it to .old
+        suffix = '.old' if sequence == 0 else '.old.%d' % sequence
+        target_file = file_name.with_suffix(suffix)
+
+        self.logger.debug("[%d] Target file: %s" % (sequence, target_file))
+        # If the target file exists, cycle again
+        if target_file.is_file():
+            # First check if we've reached the cycle limit
+            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)
+                    target_file.unlink()
+                else:
+                    self.logger.debug("Cycle limit reached")
+                    return
+            else:
+                self.logger.debug("[%d] Target file exists, cycling again" % sequence)
+                self._rotate_old(target_file, sequence + 1)
+
+        # Finally, rename the file
+        self.logger.info("[%d] Cycling file: %s -> %s" % (sequence, file_name, target_file))
+        file_name.rename(target_file)
+
     def _get_build_path(self, path):
         """
         Returns the build path

ugrd/main.py

@@ -17,9 +17,9 @@ def main():
     # Add arguments for dracut compatibility
     argparser.add_argument('-c', '--config', action='store', help='Config file location')
     argparser.add_argument('--kver', action='store', help='Kernel version')
+    argparser.add_argument('--force', action='store_true', help='Enable build cleaning', default=True)
 
     # Add and ignore arguments for dracut compatibility
-    argparser.add_argument('--force', action='store_true', help='Not used, for dracut compatibility')
     argparser.add_argument('--kernel-image', action='store', help='Kernel image')
 
     # Add the argument for the output file
@@ -36,14 +36,10 @@ def main():
         logger.setLevel(20)
 
     kwargs = {'logger': logger}
-    if config := args.config:
-        kwargs['config'] = config
 
-    if config := args.kver:
-        kwargs['kernel_version'] = config
-
-    if config := args.output_file:
-        kwargs['out_file'] = config
+    for config, arg in {'clean': 'force', 'kernel_version': 'kver', 'config': 'config', 'out_file': 'output_file'}.items():
+        if arg := getattr(args, arg):
+            kwargs[config] = arg
 
     generator = InitramfsGenerator(**kwargs)
     try:

ugrd/zen_custom.py

@@ -1,7 +1,7 @@
 """
 A collection of classes and decorators
 """
-__version__ = '3.1.0'
+__version__ = '3.1.1'
 __author__ = 'desultory'
 
 import logging
@@ -108,7 +108,7 @@ def wrapper(self, *args):
             for key, value in focus_arg.items():
                 function(self, *(other_args + (key, value,)))
         else:
-            self.logger.debug("Arguments were not expanded: %s" % args)
+            self.logger.log(5, "Arguments were not expanded: %s" % args)
             return function(self, *args)
     return wrapper