diff --git a/src/ugrd/crypto/cryptsetup.py b/src/ugrd/crypto/cryptsetup.py index 5b58a13e..ce7f21ba 100644 --- a/src/ugrd/crypto/cryptsetup.py +++ b/src/ugrd/crypto/cryptsetup.py @@ -6,7 +6,7 @@ from re import search from textwrap import dedent -from ugrd import InitramfsProtocol, InitramfsConfig +from ugrd import InitramfsConfig, InitramfsProtocol from ugrd.exceptions import AutodetectError, ValidationError from zenlib.util import colorize as c_ from zenlib.util import contains, unset @@ -27,6 +27,7 @@ "try_nokey", "include_key", "include_header", + "validate", "validate_key", "validate_header", "_dm-integrity", # Internal parameter for when dm-integrity was detected. Mostly used for test automation @@ -63,18 +64,18 @@ def _process_cryptsetup_key_types_multi(self: InitramfsConfig, key_type: str, co @contains("validate", "Skipping cryptsetup keyfile validation.", log_level=30) -def _validate_cryptsetup_key(self: InitramfsProtocol, key_parameters: dict[str, str]) -> None: +def _validate_cryptsetup_key(self: InitramfsProtocol, config: dict[str, str]) -> None: """Validates the cryptsetup keyfile config key_parameters is actually all config but this function only makes use of bits related to keyfiles """ - if key_parameters.get("include_key"): + if config.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(f"Skipping key validation for: {c_(key_parameters['key_file'], 'yellow')}") + elif config.get("validate_key") is False or config.get("validate") is False: + return self.logger.info(f"Skipping key validation for: {c_(config['key_file'], 'yellow')}") - key_path = Path(key_parameters["key_file"]) + key_path = Path(config["key_file"]) - if self["cryptsetup_keyfile_validation"] and key_parameters.get("validate_key") is not False: + if self["cryptsetup_keyfile_validation"] and config.get("validate_key") is not False: if not key_path.exists(): # Do a preliminary check to see if the key file exists raise FileNotFoundError("Key file not found: %s" % key_path) self["check_included_or_mounted"] = key_path # Add a proper check for later @@ -85,7 +86,8 @@ def _validate_cryptsetup_config(self: InitramfsProtocol, mapped_name: str) -> No try: config = self["cryptsetup"][mapped_name] except KeyError: - raise KeyError("No cryptsetup configuration found for: %s" % mapped_name) + raise KeyError(f"No cryptsetup configuration found for: {c_(mapped_name, 'red')}") + self.logger.log(5, "[%s] Validating cryptsetup configuration: %s" % (mapped_name, config)) for parameter in config: if parameter not in CRYPTSETUP_PARAMETERS: @@ -97,15 +99,17 @@ def _validate_cryptsetup_config(self: InitramfsProtocol, mapped_name: str) -> No "A partuuid or device path must be specified when using detached headers: %s" % mapped_name ) if config.get("uuid"): - raise ValueError("A UUID cannot be used with a detached header: %s" % mapped_name) + raise ValueError(f"A UUID cannot be used with a detached header: {c_(mapped_name, 'red')}") 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, c_(config["header_file"], "yellow"))) + self.logger.warning( + f"[{c_(mapped_name, 'blue')}] Header file not found: {c_(config['header_file'], 'yellow')}" + ) elif not any([config.get("partuuid"), config.get("uuid"), config.get("path")]): if not self["autodetect_luks"]: raise ValidationError( - "A device uuid, partuuid, or path must be specified for cryptsetup mount: %s" % mapped_name + f"A device uuid, partuuid, or path must be specified for cryptsetup mount: {c_(mapped_name, 'red')}" ) if config.get("key_file"): @@ -168,7 +172,7 @@ def _get_dm_slave_info(self: InitramfsProtocol, device_info: dict) -> tuple[str, raise AutodetectError("No slave device information found for: %s" % device_info) -def _read_cryptsetup_header(self: InitramfsProtocol, mapped_name: str, slave_device: str | None= None) -> dict: +def _read_cryptsetup_header(self: InitramfsProtocol, mapped_name: str, slave_device: str | None = None) -> dict: """Reads LUKS header information from a device or header file into a dict""" header_file = self["cryptsetup"][mapped_name].get("header_file") if not header_file: @@ -303,14 +307,14 @@ def _validate_cryptsetup_device(self: InitramfsProtocol, mapped_name) -> None: """Validates a cryptsetup device against the device mapper information, blkid information, and cryptsetup information. Uses `cryptsetup luksDump` to check that the device is a LUKS device.""" + cryptsetup_info = self["cryptsetup"][mapped_name] # Get the cryptsetup information + if cryptsetup_info.get("validate") is False: + return self.logger.warning(f"Skipping cryptsetup device validation: {c_(mapped_name, 'yellow')}") + dm_info = _get_dm_info(self, mapped_name) if not dm_info["uuid"].startswith("CRYPT-LUKS"): # Ensure the device is a crypt device - raise ValueError("Device is not a crypt device: %s" % dm_info) - - 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" % c_(mapped_name, "yellow")) + raise ValueError(f"Device is not a crypt device: {c_(dm_info, 'red')}") slave_device, blkid_info = _get_dm_slave_info(self, dm_info) # Get the blkid information diff --git a/tests/test_cryptsetup.py b/tests/test_cryptsetup.py index e482e395..86ab5027 100644 --- a/tests/test_cryptsetup.py +++ b/tests/test_cryptsetup.py @@ -31,6 +31,12 @@ def test_cryptsetup_included_key(self): generator = InitramfsGenerator(logger=self.logger, config="tests/cryptsetup_included_key.toml") generator.build() + def test_cryptsetup_bad_key_no_verify(self): + """Tests LUKS based roots using an invalid keyfile but verification is disabled""" + generator = InitramfsGenerator(logger=self.logger, config="tests/cryptsetup_included_key.toml", cryptsetup={"root": {"key_file": "asdfasdfasdfasdfasdfasdfasdf", "validate": False, "include_key": False}}) + with self.assertRaises(RuntimeError): + generator.build() + def test_cryptsetup_integrity(self): """Tests LUKS based roots using a keyfile included in the initramfs with integrity protection""" generator = InitramfsGenerator(