add colored text

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

src/ugrd/base/base.py

@@ -1,10 +1,10 @@
 __author__ = "desultory"
-__version__ = "5.2.0"
+__version__ = "5.2.1"
 
 from importlib.metadata import version
 from pathlib import Path
 
-from zenlib.util import contains, unset
+from zenlib.util import contains, unset, colorize
 
 
 @contains("hostonly")
@@ -40,7 +40,7 @@ def autodetect_init(self) -> None:
     from shutil import which
 
     if init := which("init"):
-        self.logger.info("Detected init at: %s", init)
+        self.logger.info("Detected init at: %s", colorize(init, "cyan", bright=True))
         self["init_target"] = init
     else:
         raise FileNotFoundError("init_target is not specified and could not be detected.")

src/ugrd/base/core.py

@@ -1,18 +1,18 @@
 __author__ = "desultory"
-__version__ = "3.11.2"
+__version__ = "3.11.3"
 
 from pathlib import Path
 from typing import Union
 
-from zenlib.util import contains
+from zenlib.util import contains, colorize
 from zenlib.types import NoDupFlatList
 
 def detect_tmpdir(self) -> None:
     """Reads TMPDIR from the environment, sets it as the temporary directory."""
     from os import environ
 
     if tmpdir := environ.get("TMPDIR"):
-        self.logger.info("Detected TMPDIR: %s" % tmpdir)
+        self.logger.info("Detected TMPDIR: %s" % (colorize(tmpdir, "cyan")))
         self["tmpdir"] = Path(tmpdir)
 
 
@@ -24,7 +24,7 @@ def clean_build_dir(self) -> None:
     build_dir = self._get_build_path("/")
 
     if build_dir.is_dir():
-        self.logger.warning("Cleaning build directory: %s" % build_dir)
+        self.logger.warning("Cleaning build directory: %s" % colorize(build_dir, "yellow"))
         rmtree(build_dir)
     else:
         self.logger.info("Build directory does not exist, skipping cleaning: %s" % build_dir)
@@ -54,7 +54,7 @@ def calculate_dependencies(self, binary: str) -> list[Path]:
     dependencies = run(["lddtree", "-l", str(binary_path)], capture_output=True)
 
     if dependencies.returncode != 0:
-        self.logger.warning("Unable to calculate dependencies for: %s" % binary)
+        self.logger.warning("Unable to calculate dependencies for: %s" % colorize(binary, "red", bold=True, bright=True))
         raise RuntimeError("Unable to resolve dependencies, error: %s" % dependencies.stderr.decode("utf-8"))
 
     dependency_paths = []
@@ -198,7 +198,7 @@ def _process_out_file(self, out_file: str) -> None:
     out_file = str(out_file)
     if out_file == "./" or out_file == ".":
         current_dir = Path(".").resolve()
-        self.logger.info("Setting out_dir to current directory: %s" % current_dir)
+        self.logger.info("Setting out_dir to current directory: %s" % colorize(current_dir, "cyan", bold=True))
         self["out_dir"] = current_dir
         return
 
@@ -215,7 +215,7 @@ def _process_out_file(self, out_file: str) -> None:
 
     if str(out_file.parent) != ".":  # If the parent isn't the curent dir, set the out_dir to the parent
         self["out_dir"] = out_file.parent
-        self.logger.info("Resolved out_dir to: %s" % self["out_dir"])
+        self.logger.info("Resolved out_dir to: %s" % colorize(self["out_dir"], "green"))
         out_file = out_file.name
 
     self.data["out_file"] = out_file
@@ -402,7 +402,7 @@ def _process_masks_multi(self, runlevel: str, function: str) -> None:
     if runlevel not in self["masks"]:
         self.logger.debug("Creating new mask: %s" % runlevel)
         self["masks"][runlevel] = NoDupFlatList(logger=self.logger)
-    self.logger.info("[%s] Adding mask: %s" % (runlevel, function))
+    self.logger.info("[%s] Adding mask: %s" % (runlevel, colorize(function, "red")))
     self["masks"][runlevel] = function
 
 

src/ugrd/crypto/cryptsetup.py

@@ -1,9 +1,9 @@
 __author__ = "desultory"
-__version__ = "3.7.3"
+__version__ = "3.8.0"
 
 from pathlib import Path
 
-from zenlib.util import contains
+from zenlib.util import colorize, contains
 
 _module_name = "ugrd.crypto.cryptsetup"
 
@@ -61,7 +61,7 @@ def _validate_crypysetup_key(self, key_parameters: dict) -> None:
     if key_parameters.get("include_key"):
         return self.logger.info("Skipping key validation for included key.")
     elif key_parameters.get("validate_key") is False:
-        return self.logger.info("Skipping key validation for: %s" % key_parameters["key_file"])
+        return self.logger.info("Skipping key validation for: %s" % colorize(key_parameters["key_file"], "yellow"))
 
     key_path = Path(key_parameters["key_file"])
 
@@ -92,7 +92,9 @@ def _validate_cryptsetup_config(self, mapped_name: str) -> None:
         if not Path(
             config["header_file"]
         ).exists():  # Make sure the header file exists, it may not be present at build time
-            self.logger.warning("[%s] Header file not found: %s" % (mapped_name, config["header_file"]))
+            self.logger.warning(
+                "[%s] Header file not found: %s" % (mapped_name, colorize(config["header_file"], "yellow"))
+            )
     elif not any([config.get("partuuid"), config.get("uuid"), config.get("path")]):
         if not self["autodetect_root_luks"]:
             raise ValueError(
@@ -107,7 +109,7 @@ def _process_cryptsetup_multi(self, mapped_name: str, config: dict) -> None:
     """Processes the cryptsetup configuration"""
     for parameter in config:
         if parameter not in CRYPTSETUP_PARAMETERS:
-            self.logger.error("[%s] Unknown parameter: %s" % (mapped_name, parameter))
+            self.logger.error("[%s] Unknown parameter: %s" % (mapped_name, colorize(parameter, "red", bold=True)))
 
     config = _merge_cryptsetup(self, mapped_name, config)  # Merge the config with the existing configuration
     self.logger.debug("[%s] Processing cryptsetup configuration: %s" % (mapped_name, config))
@@ -183,9 +185,9 @@ def _read_cryptsetup_header(self, mapped_name: str, slave_device: str = None) ->
         return luks_info
     except RuntimeError as e:
         if not self["cryptsetup"][mapped_name].get("header_file"):
-            self.logger.error("Unable to read LUKS header: %s" % e)
+            self.logger.error("Unable to read LUKS header: %s" % colorize(e, "red", bold=True, bright=True))
         else:
-            self.logger.warning("Cannot read detached LUKS header for validation: %s" % e)
+            self.logger.warning("Cannot read detached LUKS header for validation: %s" % colorize(e, "yellow"))
     return {}
 
 
@@ -199,7 +201,7 @@ def _detect_luks_aes_module(self, luks_cipher_name: str) -> None:
     if crypto_config["module"] == "kernel":
         self.logger.debug("Cipher kernel modules are builtin: %s" % crypto_name)
     else:
-        self.logger.info("[%s] Adding kernel module for LUKS cipher: %s" % (crypto_name, crypto_config["module"]))
+        self.logger.info("[%s] Adding kernel module for LUKS cipher: %s" % (crypto_name, colorize(crypto_config["module"], "cyan")))
         self["_kmod_auto"] = crypto_config["module"]
 
 
@@ -230,7 +232,7 @@ def _validate_cryptsetup_header(self, mapped_name: str) -> None:
     """Validates configured cryptsetup volumes against the LUKS header."""
     cryptsetup_info = self["cryptsetup"][mapped_name]
     if cryptsetup_info.get("validate_header") is False:
-        return self.logger.warning("Skipping cryptsetup header validation for: %s" % mapped_name)
+        return self.logger.warning("Skipping cryptsetup header validation for: %s" % colorize(mapped_name, "yellow"))
 
     luks_info = _read_cryptsetup_header(self, mapped_name)
     if not luks_info:
@@ -250,7 +252,7 @@ def _validate_cryptsetup_header(self, mapped_name: str) -> None:
             if keyslot.get("kdf", {}).get("type") == "argon2id":
                 raise FileNotFoundError("[%s] Missing cryptsetup dependency: libargon2.so" % mapped_name)
 
-    if 'header_file' in cryptsetup_info:
+    if "header_file" in cryptsetup_info:
         self["check_included_or_mounted"] = cryptsetup_info["header_file"]  # Add the header file to the check list
 
 
@@ -266,7 +268,7 @@ def _validate_cryptsetup_device(self, mapped_name) -> None:
 
     cryptsetup_info = self["cryptsetup"][mapped_name]  # Get the cryptsetup information
     if cryptsetup_info.get("validate") is False:
-        return self.logger.warning("Skipping cryptsetup device validation: %s" % mapped_name)
+        return self.logger.warning("Skipping cryptsetup device validation: %s" % colorize(mapped_name, "yellow"))
 
     slave_device, blkid_info = _get_dm_slave_info(self, dm_info)  # Get the blkid information
 

src/ugrd/crypto/smartcard.py

@@ -1,7 +1,7 @@
 __author__ = "desultory"
-__version__ = "1.2.0"
+__version__ = "1.2.1"
 
-from zenlib.util import contains
+from zenlib.util import contains, colorize
 from pathlib import Path
 
 
@@ -11,7 +11,7 @@ def _process_sc_public_key(self, key: str) -> None:
     if not key_path.exists():
         raise FileNotFoundError(f"Smartcard public key file not found: {key}")
     self.data["sc_public_key"] = key_path
-    self.logger.info("Using smartcard public key file: %s", key_path)
+    self.logger.info("Using smartcard public key file: %s", colorize(key_path, "green"))
     self["dependencies"] = key_path