Merge pull request #123 from desultory/dev

Detect aes kmods from the LUKS header

Author: desultory

src/ugrd/crypto/cryptsetup.py

@@ -1,5 +1,5 @@
 __author__ = "desultory"
-__version__ = "3.5.0"
+__version__ = "3.6.1"
 
 from pathlib import Path
 
@@ -195,14 +195,31 @@ def _read_cryptsetup_header(self, mapped_name: str, slave_device: str = None) ->
             self.logger.warning("Cannot read detached LUKS header for validation: %s" % e)
     return {}
 
-def _check_luks_header_aes(self, luks_info: dict) -> dict:
-    """Checks for aes requirements in the LUKS header"""
+def _detect_luks_aes_module(self, luks_cipher_name: str) -> None:
+    """Using the cipher name from the LUKS header, detects the corresponding kernel module."""
+    if luks_cipher_name.startswith("aes"):
+        self["_kmod_auto"] = "aes"  # Try to enable the aesni module for any aes type
+
+    aes_type = luks_cipher_name.split("-")[1]  # Get the cipher type from the name
+    self["_kmod_auto"] = aes_type  # Add the aes type to the kernel modules
+
+    crypto_name = f"{aes_type}(aes)"  # Format the name like the /proc/crypto entry
+    crypto_config = self["_crypto_ciphers"][crypto_name]
+    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["_kmod_auto"] = crypto_config["module"]
+
+def _detect_luks_header_aes(self, luks_info: dict) -> dict:
+    """Checks the cipher type in the LUKS header, reads /proc/crypto to find the
+    corresponding driver. If it's not builtin, adds the module to the kernel modules."""
     for keyslot in luks_info.get("keyslots", {}).values():
-        if keyslot.get("area", {}).get("encryption") == "aes-xts-plain64":
-            return True
+        if keyslot.get("area", {}).get("encryption").startswith("aes"):
+            _detect_luks_aes_module(self, keyslot["area"]["encryption"])
     for segment in luks_info.get("segments", {}).values():
-        if segment.get("encryption") == "aes-xts-plain64":
-            return True
+        if segment.get("encryption").startswith("aes"):
+            _detect_luks_aes_module(self, segment["encryption"])
 
 def _detect_luks_header_sha(self, luks_info: dict) -> dict:
     """Reads the hash algorithm from the LUKS header,
@@ -229,10 +246,7 @@ def _validate_cryptsetup_header(self, mapped_name: str) -> None:
         if luks_info.get("uuid") != uuid:
             raise ValueError("[%s] LUKS UUID mismatch, found '%s', expected: %s" % (mapped_name, luks_info["uuid"], uuid))
 
-    if _check_luks_header_aes(self, luks_info):
-        self.logger.debug("[%s] LUKS uses aes-xts-plain64" % mapped_name)
-        self["kernel_modules"] = self._crypto_ciphers["xts(aes)"]["driver"]  # Placeholder, this driver is wrong!
-
+    _detect_luks_header_aes(self, luks_info)
     _detect_luks_header_sha(self, luks_info)
 
     if not self["argon2"]:  # if argon support was not detected, check if the header wants it
@@ -302,6 +316,8 @@ def get_value(line):
                 continue  # Skip lines until a name is found
             elif line.startswith("driver"):
                 self._crypto_ciphers[current_name]["driver"] = get_value(line)
+            elif line.startswith("module"):
+                self._crypto_ciphers[current_name]["module"] = get_value(line)
 
 
 @contains("validate", "Skipping cryptsetup configuration validation.", log_level=30)