Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Handling encryption logic when only one non-root partition is encrypted (such as /home) #2575

Draft
wants to merge 4 commits into
base: master
Choose a base branch
from
Draft

Handling encryption logic when only one non-root partition is encrypted (such as /home) #2575

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
35c6311
Splitting out logic for creating keyfile and crypttab entries. Then m…
Torxed Jul 12, 2024
91abe1f
Fixing flake8 and some variable issues
Torxed Jul 12, 2024
7625c65
Renaming some more functions, and sorting out the LVM stuff, and chec…
Torxed Jul 12, 2024
4322f3a
Removed final debug output
Torxed Jul 15, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .flake8
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
[flake8]
count = True
# Several of the following could be autofixed or improved by running the code through psf/black
ignore = E123,E126,E128,E203,E231,E261,E302,E402,E722,F541,W191,W292,W293,W503,W504
ignore = E123,E126,E128,E203,E227,E231,E261,E302,E402,E722,F541,W191,W292,W293,W503,W504
max-complexity = 40
max-line-length = 236
show-source = True
Expand Down
15 changes: 13 additions & 2 deletions archinstall/lib/disk/device_model.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@

from ..exceptions import DiskError, SysCallError
from ..general import SysCommand
from ..output import debug, error
from ..output import debug, error, warn
from ..storage import storage

if TYPE_CHECKING:
Expand Down Expand Up @@ -1204,10 +1204,21 @@ def __post_init__(self):
raise ValueError('LuksOnLvm encryption require LMV volumes to be defined')

def should_generate_encryption_file(self, dev: PartitionModification | LvmVolume) -> bool:
if isinstance(dev, PartitionModification):
warn("Deprecation warning: should_generate_encryption_file() has been renamed to should_generate_encryption_keyfile() and should_generate_encryption_file will be removed.")
return self.should_generate_encryption_keyfile(dev)

def should_generate_encryption_keyfile(self, dev: PartitionModification | LvmVolume) -> bool:
# If all partitions being encrypted, are non-root file systems
# we will not generate key-files as their safety can not be guaranteed.
if all([part.mountpoint != Path('/') for part in self.partitions]):
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

just FYI there's a part.is_root() :)

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This check doesn't feel right here...on the caller side this is run for each partition now, but we can probably short circuit it

	def generate_keyfile_for_dev(self, dev: PartitionModification | LvmVolume) -> bool:
		# If all partitions being encrypted, are non-root file systems
		# we will not generate key-files as their safety can not be guaranteed.
		if isinstance(dev, PartitionModification):
			return dev in self.partitions and dev.mountpoint != Path('/')
		elif isinstance(dev, LvmVolume):
			return dev in self.lvm_volumes and dev.mountpoint != Path('/')

		return False

	def generate_keyfiles(self) -> bool:
		# If all partitions being encrypted, are non-root file systems
		# we will not generate key-files as their safety can not be guaranteed.
		if all([part.mountpoint != Path('/') for part in self.partitions]):
			return True
		return False

And on the caller

	def _setup_luks_unlock_partitions(self) -> None:
		if not self._disk_encryption.generate_keyfiles():
			return
		...

What do you think?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

just FYI there's a part.is_root() :)

Silly me just copied the code from two rows down, but now that you mention it I do remember seeing that function, I'll change!

return False
# If the device mountpoint is anything but root,
# we want to create a keyfile for unlocking
elif isinstance(dev, PartitionModification):
return dev in self.partitions and dev.mountpoint != Path('/')
elif isinstance(dev, LvmVolume):
return dev in self.lvm_volumes and dev.mountpoint != Path('/')

return False

def json(self) -> Dict[str, Any]:
Expand Down
40 changes: 31 additions & 9 deletions archinstall/lib/installer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -264,7 +264,7 @@ def _mount_lvm_layout(self, luks_handlers: Dict[Any, Luks2] = {}):
def _prepare_luks_partitions(
self,
partitions: List[disk.PartitionModification]
) -> Dict[disk.PartitionModification, Luks2]:
) -> Dict[disk.PartitionModification, Luks2] -> Dict[disk.PartitionModification, disk.device_handler.unlock_luks2_dev]:
return {
part_mod: disk.device_handler.unlock_luks2_dev(
part_mod.dev_path,
Expand Down Expand Up @@ -354,9 +354,13 @@ def _mount_btrfs_subvol(
disk.device_handler.mount(dev_path, mountpoint, options=mount_options)

def generate_key_files(self) -> None:
warn("Deprecation warning: generate_key_files() has been renamed to setup_luks_unlock() and generate_key_files will be removed.")
return self.setup_luks_unlock(self)

def setup_luks_unlock(self) -> None:
match self._disk_encryption.encryption_type:
case disk.EncryptionType.Luks:
self._generate_key_files_partitions()
self._setup_luks_unlock_partitions()
case disk.EncryptionType.LuksOnLvm:
self._generate_key_file_lvm_volumes()
case disk.EncryptionType.LvmOnLuks:
Expand All @@ -365,9 +369,13 @@ def generate_key_files(self) -> None:
# so we won't need any keyfile generation atm
pass

def _generate_key_files_partitions(self):
def _generate_key_files_partitions(self) -> None:
warn("Deprecation warning: _generate_key_files_partitions() has been renamed to _setup_luks_unlock_partitions() and _generate_key_files_partitions will be removed.")
return self._setup_luks_unlock_partitions(self)

def _setup_luks_unlock_partitions(self) -> None:
for part_mod in self._disk_encryption.partitions:
gen_enc_file = self._disk_encryption.should_generate_encryption_file(part_mod)
gen_enc_file = self._disk_encryption.should_generate_encryption_keyfile(part_mod)

luks_handler = Luks2(
part_mod.safe_dev_path,
Expand All @@ -377,7 +385,9 @@ def _generate_key_files_partitions(self):

if gen_enc_file and not part_mod.is_root():
debug(f'Creating key-file: {part_mod.dev_path}')
luks_handler.create_keyfile(self.target)
key_file = luks_handler.create_keyfile(self.target)
else:
key_file = None

if part_mod.is_root() and not gen_enc_file:
if self._disk_encryption.hsm_device:
Expand All @@ -387,19 +397,26 @@ def _generate_key_files_partitions(self):
self._disk_encryption.encryption_password
)

luks_handler.create_crypttab(
target_path=self.target,
key_file=key_file
)

def _generate_key_file_lvm_volumes(self):
for vol in self._disk_encryption.lvm_volumes:
gen_enc_file = self._disk_encryption.should_generate_encryption_file(vol)
gen_enc_file = self._disk_encryption.should_generate_encryption_keyfile(vol)

luks_handler = Luks2(
vol.safe_dev_path,
mapper_name=vol.mapper_name,
password=self._disk_encryption.encryption_password
)

if gen_enc_file and not vol.is_root():
info(f'Creating key-file: {vol.dev_path}')
luks_handler.create_keyfile(self.target)
if gen_enc_file and not part_mod.is_root():
debug(f'Creating key-file: {part_mod.dev_path}')
key_file = luks_handler.create_keyfile(self.target)
else:
key_file = None

if vol.is_root() and not gen_enc_file:
if self._disk_encryption.hsm_device:
Expand All @@ -409,6 +426,11 @@ def _generate_key_file_lvm_volumes(self):
self._disk_encryption.encryption_password
)

luks_handler.create_crypttab(
target_path=self.target,
key_file=key_file
)

def sync_log_to_install_medium(self) -> bool:
# Copy over the install log (if there is one) to the install medium if
# at least the base has been strapped in, otherwise we won't have a filesystem/structure to copy to.
Expand Down
47 changes: 38 additions & 9 deletions archinstall/lib/luks.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -191,7 +191,16 @@ def lock(self):

self._mapper_dev = None

def create_keyfile(self, target_path: Path, override: bool = False):
def create_crypttab(self, target_path: Path, key_file: Path|None = None, options: List[str]|None = None, override: bool = False):
crypttab_path = target_path / 'etc/crypttab'
self._crypttab(
crypttab_path,
key_file=key_file,
options=options or ["luks", "key-slot=1"],
override=override
)

def create_keyfile(self, target_path: Path, override: bool = False) -> Path:
"""
Routine to create keyfiles, so it can be moved elsewhere
"""
Expand All @@ -202,12 +211,11 @@ def create_keyfile(self, target_path: Path, override: bool = False):
# automatically load this key if we name the device to "xyzloop"
kf_path = Path(f'/etc/cryptsetup-keys.d/{self.mapper_name}.key')
key_file = target_path / kf_path.relative_to(kf_path.root)
crypttab_path = target_path / 'etc/crypttab'

if key_file.exists():
if not override:
info(f'Key file {key_file} already exists, keeping existing')
return
return kf_path
else:
info(f'Key file {key_file} already exists, overriding')

Expand All @@ -219,7 +227,8 @@ def create_keyfile(self, target_path: Path, override: bool = False):
key_file.chmod(0o400)

self._add_key(key_file)
self._crypttab(crypttab_path, kf_path, options=["luks", "key-slot=1"])

return kf_path

def _add_key(self, key_file: Path):
debug(f'Adding additional key-file {key_file}')
Expand All @@ -239,13 +248,33 @@ def _add_key(self, key_file: Path):
def _crypttab(
self,
crypttab_path: Path,
key_file: Path,
options: List[str]
options: List[str],
key_file: Path|None = None,
override: bool = False
) -> None:
debug(f'Adding crypttab entry for key {key_file}')

with open(crypttab_path, 'a') as crypttab:
with open(crypttab_path, 'r+') as crypttab:
existing_data = crypttab.readlines()
opt = ','.join(options)
uuid = self._get_luks_uuid()
row = f"{self.mapper_name} UUID={uuid} {key_file} {opt}\n"
crypttab.write(row)
new_row = f"{self.mapper_name} UUID={uuid} {key_file or 'none'} {opt}\n"

for index, existing_row in enumerate(existing_data):
if uuid in existing_row:
if override is False:
debug(f"Found {uuid} in {existing_row}")
return True
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is this return here right? If override is passed into the function should this terminate ?

Copy link
Member Author

@Torxed Torxed Jul 13, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's checking if we don't want to override - but it found that we already have added the data - so it's happy and returns "done" :)

else:
existing_data.pop(index)

# Make sure the last entry ends with \n
# before inserting our new entry at the end
if not existing_data[-1][-1:] == '\n':
existing_data.append('\n')

existing_data.append(new_row)

crypttab.seek(0)
crypttab.truncate()
crypttab.write(''.join(existing_data))
2 changes: 1 addition & 1 deletion archinstall/scripts/guided.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -128,7 +128,7 @@ def perform_installation(mountpoint: Path):
if disk_config.config_type != disk.DiskLayoutType.Pre_mount:
if disk_encryption and disk_encryption.encryption_type != disk.EncryptionType.NoEncryption:
# generate encryption key files for the mounted luks devices
installation.generate_key_files()
installation.setup_luks_unlock()

if mirror_config := archinstall.arguments.get('mirror_config', None):
installation.set_mirrors(mirror_config, on_target=False)
Expand Down
Loading