Added a HSM menu entry (#1196)

* Added a HSM menu entry, but also a safety check to make sure a FIDO device is connected

* flake8 complaints

* Adding FIDO lookup using cryptenroll listing

* Added systemd-cryptenroll --fido2-device=list

* Removed old _select_hsm call

* Fixed flake8 complaints

* Added support for locking and unlocking with a HSM

* Removed hardcoded paths in favor of PR merge

* Removed hardcoded paths in favor of PR merge

* Fixed mypy complaint

* Flake8 issue

* Added sd-encrypt for HSM and revert back to encrypt when HSM is not used (stability reason)

* Added /etc/vconsole.conf and tweaked fido2_enroll() to use the proper paths

* Spelling error

* Using UUID instead of PARTUUID when using HSM. I can't figure out how to get sd-encrypt to use PARTUUID instead. Added a Partition().part_uuid function. Actually renamed .uuid to .part_uuid and created a .uuid instead.

* Adding missing package libfido2 and removed tpm2-device=auto as it overrides everything and forces password prompt to be used over FIDO2, no matter the order of the options.

* Added some notes to clarify some choices.

* Had to move libfido2 package install to later in the chain, as there's not even a base during mounting :P

Author: Torxed

.flake8

@@ -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
+ignore = E123,E126,E128,E203,E231,E261,E302,E402,E722,F541,W191,W292,W293,W503
 max-complexity = 40
 max-line-length = 236
 show-source = True

archinstall/__init__.py

@@ -45,6 +45,11 @@
 from .lib.translation import Translation, DeferredTranslation
 from .lib.plugins import plugins, load_plugin # This initiates the plugin loading ceremony
 from .lib.configuration import *
+from .lib.udev import udevadm_info
+from .lib.hsm import (
+	get_fido2_devices,
+	fido2_enroll
+)
 parser = ArgumentParser()
 
 __version__ = "2.4.2"

archinstall/lib/configuration.py

@@ -1,12 +1,23 @@
 import json
 import logging
-from pathlib import Path
+import pathlib
 from typing import Optional, Dict
 
 from .storage import storage
 from .general import JSON, UNSAFE_JSON
 from .output import log
-
+from .exceptions import RequirementError
+from .hsm import get_fido2_devices
+
+def configuration_sanity_check():
+	if storage['arguments'].get('HSM'):
+		if not get_fido2_devices():
+			raise RequirementError(
+				f"In order to use HSM to pair with the disk encryption,"
+				+ f" one needs to be accessible through /dev/hidraw* and support"
+				+ f" the FIDO2 protocol. You can check this by running"
+				+ f" 'systemd-cryptenroll --fido2-device=list'."
+			)
 
 class ConfigurationOutput:
 	def __init__(self, config: Dict):
@@ -21,7 +32,7 @@ def __init__(self, config: Dict):
 		self._user_credentials = {}
 		self._disk_layout = None
 		self._user_config = {}
-		self._default_save_path = Path(storage.get('LOG_PATH', '.'))
+		self._default_save_path = pathlib.Path(storage.get('LOG_PATH', '.'))
 		self._user_config_file = 'user_configuration.json'
 		self._user_creds_file = "user_credentials.json"
 		self._disk_layout_file = "user_disk_layout.json"
@@ -84,7 +95,7 @@ def show(self):
 
 		print()
 
-	def _is_valid_path(self, dest_path :Path) -> bool:
+	def _is_valid_path(self, dest_path :pathlib.Path) -> bool:
 		if (not dest_path.exists()) or not (dest_path.is_dir()):
 			log(
 				'Destination directory {} does not exist or is not a directory,\n Configuration files can not be saved'.format(dest_path.resolve()),
@@ -93,26 +104,26 @@ def _is_valid_path(self, dest_path :Path) -> bool:
 			return False
 		return True
 
-	def save_user_config(self, dest_path :Path = None):
+	def save_user_config(self, dest_path :pathlib.Path = None):
 		if self._is_valid_path(dest_path):
 			with open(dest_path / self._user_config_file, 'w') as config_file:
 				config_file.write(self.user_config_to_json())
 
-	def save_user_creds(self, dest_path :Path = None):
+	def save_user_creds(self, dest_path :pathlib.Path = None):
 		if self._is_valid_path(dest_path):
 			if user_creds := self.user_credentials_to_json():
 				target = dest_path / self._user_creds_file
 				with open(target, 'w') as config_file:
 					config_file.write(user_creds)
 
-	def save_disk_layout(self, dest_path :Path = None):
+	def save_disk_layout(self, dest_path :pathlib.Path = None):
 		if self._is_valid_path(dest_path):
 			if disk_layout := self.disk_layout_to_json():
 				target = dest_path / self._disk_layout_file
 				with target.open('w') as config_file:
 					config_file.write(disk_layout)
 
-	def save(self, dest_path :Path = None):
+	def save(self, dest_path :pathlib.Path = None):
 		if not dest_path:
 			dest_path = self._default_save_path
 

archinstall/lib/disk/blockdevice.py

@@ -275,7 +275,7 @@ def get_partition(self, uuid :str) -> Partition:
 		count = 0
 		while count < 5:
 			for partition_uuid, partition in self.partitions.items():
-				if partition.uuid.lower() == uuid.lower():
+				if partition.part_uuid.lower() == uuid.lower():
 					return partition
 			else:
 				log(f"uuid {uuid} not found. Waiting for {count +1} time",level=logging.DEBUG)

archinstall/lib/disk/filesystem.py

@@ -150,7 +150,7 @@ def load_layout(self, layout :Dict[str, Any]) -> None:
 
 			if partition.get('boot', False):
 				log(f"Marking partition {partition['device_instance']} as bootable.")
-				self.set(self.partuuid_to_index(partition['device_instance'].uuid), 'boot on')
+				self.set(self.partuuid_to_index(partition['device_instance'].part_uuid), 'boot on')
 
 			prev_partition = partition
 
@@ -193,7 +193,7 @@ def use_entire_disk(self, root_filesystem_type :str = 'ext4') -> Partition:
 	def add_partition(self, partition_type :str, start :str, end :str, partition_format :Optional[str] = None) -> Partition:
 		log(f'Adding partition to {self.blockdevice}, {start}->{end}', level=logging.INFO)
 
-		previous_partition_uuids = {partition.uuid for partition in self.blockdevice.partitions.values()}
+		previous_partition_uuids = {partition.part_uuid for partition in self.blockdevice.partitions.values()}
 
 		if self.mode == MBR:
 			if len(self.blockdevice.partitions) > 3:
@@ -210,7 +210,7 @@ def add_partition(self, partition_type :str, start :str, end :str, partition_for
 			count = 0
 			while count < 10:
 				new_uuid = None
-				new_uuid_set = (previous_partition_uuids ^ {partition.uuid for partition in self.blockdevice.partitions.values()})
+				new_uuid_set = (previous_partition_uuids ^ {partition.part_uuid for partition in self.blockdevice.partitions.values()})
 
 				if len(new_uuid_set) > 0:
 					new_uuid = new_uuid_set.pop()
@@ -236,7 +236,7 @@ def add_partition(self, partition_type :str, start :str, end :str, partition_for
 		# TODO: This should never be able to happen
 		log(f"Could not find the new PARTUUID after adding the partition.", level=logging.ERROR, fg="red")
 		log(f"Previous partitions: {previous_partition_uuids}", level=logging.ERROR, fg="red")
-		log(f"New partitions: {(previous_partition_uuids ^ {partition.uuid for partition in self.blockdevice.partitions.values()})}", level=logging.ERROR, fg="red")
+		log(f"New partitions: {(previous_partition_uuids ^ {partition.part_uuid for partition in self.blockdevice.partitions.values()})}", level=logging.ERROR, fg="red")
 		raise DiskError(f"Could not add partition using: {parted_string}")
 
 	def set_name(self, partition: int, name: str) -> bool:

archinstall/lib/disk/partition.py

@@ -184,7 +184,7 @@ def partition_type(self) -> Optional[str]:
 			return device['pttype']
 
 	@property
-	def uuid(self) -> Optional[str]:
+	def part_uuid(self) -> Optional[str]:
 		"""
 		Returns the PARTUUID as returned by lsblk.
 		This is more reliable than relying on /dev/disk/by-partuuid as
@@ -197,6 +197,26 @@ def uuid(self) -> Optional[str]:
 
 			time.sleep(max(0.1, storage['DISK_TIMEOUTS'] * i))
 
+			partuuid = self._safe_part_uuid
+			if partuuid:
+				return partuuid
+
+		raise DiskError(f"Could not get PARTUUID for {self.path} using 'blkid -s PARTUUID -o value {self.path}'")
+
+	@property
+	def uuid(self) -> Optional[str]:
+		"""
+		Returns the UUID as returned by lsblk for the **partition**.
+		This is more reliable than relying on /dev/disk/by-uuid as
+		it doesn't seam to be able to detect md raid partitions.
+		For bind mounts all the subvolumes share the same uuid
+		"""
+		for i in range(storage['DISK_RETRY_ATTEMPTS']):
+			if not self.partprobe():
+				raise DiskError(f"Could not perform partprobe on {self.device_path}")
+
+			time.sleep(max(0.1, storage['DISK_TIMEOUTS'] * i))
+
 			partuuid = self._safe_uuid
 			if partuuid:
 				return partuuid
@@ -216,6 +236,28 @@ def _safe_uuid(self) -> Optional[str]:
 
 			log(f"Could not reliably refresh PARTUUID of partition {self.device_path} due to partprobe error.", level=logging.DEBUG)
 
+		try:
+			return SysCommand(f'blkid -s UUID -o value {self.device_path}').decode('UTF-8').strip()
+		except SysCallError as error:
+			if self.block_device.info.get('TYPE') == 'iso9660':
+				# Parent device is a Optical Disk (.iso dd'ed onto a device for instance)
+				return None
+
+			log(f"Could not get PARTUUID of partition using 'blkid -s UUID -o value {self.device_path}': {error}")
+
+	@property
+	def _safe_part_uuid(self) -> Optional[str]:
+		"""
+		A near copy of self.uuid but without any delays.
+		This function should only be used where uuid is not crucial.
+		For instance when you want to get a __repr__ of the class.
+		"""
+		if not self.partprobe():
+			if self.block_device.info.get('TYPE') == 'iso9660':
+				return None
+
+			log(f"Could not reliably refresh PARTUUID of partition {self.device_path} due to partprobe error.", level=logging.DEBUG)
+
 		try:
 			return SysCommand(f'blkid -s PARTUUID -o value {self.device_path}').decode('UTF-8').strip()
 		except SysCallError as error:

archinstall/lib/general.py

@@ -135,6 +135,8 @@ def _encode(obj :Any) -> Any:
 			return obj.isoformat()
 		elif isinstance(obj, (list, set, tuple)):
 			return [json.loads(json.dumps(item, cls=JSON)) for item in obj]
+		elif isinstance(obj, (pathlib.Path)):
+			return str(obj)
 		else:
 			return obj
 

archinstall/lib/hsm/__init__.py

@@ -0,0 +1,4 @@
+from .fido import (
+	get_fido2_devices,
+	fido2_enroll
+)

archinstall/lib/hsm/fido.py

@@ -0,0 +1,47 @@
+import typing
+import pathlib
+from ..general import SysCommand, SysCommandWorker, clear_vt100_escape_codes
+from ..disk.partition import Partition
+
+def get_fido2_devices() -> typing.Dict[str, typing.Dict[str, str]]:
+	"""
+	Uses systemd-cryptenroll to list the FIDO2 devices
+	connected that supports FIDO2.
+	Some devices might show up in udevadm as FIDO2 compliant
+	when they are in fact not.
+
+	The drawback of systemd-cryptenroll is that it uses human readable format.
+	That means we get this weird table like structure that is of no use.
+
+	So we'll look for `MANUFACTURER` and `PRODUCT`, we take their index
+	and we split each line based on those positions.
+	"""
+	worker = clear_vt100_escape_codes(SysCommand(f"systemd-cryptenroll --fido2-device=list").decode('UTF-8'))
+
+	MANUFACTURER_POS = 0
+	PRODUCT_POS = 0
+	devices = {}
+	for line in worker.split('\r\n'):
+		if '/dev' not in line:
+			MANUFACTURER_POS = line.find('MANUFACTURER')
+			PRODUCT_POS = line.find('PRODUCT')
+			continue
+
+		path = line[:MANUFACTURER_POS].rstrip()
+		manufacturer = line[MANUFACTURER_POS:PRODUCT_POS].rstrip()
+		product = line[PRODUCT_POS:]
+
+		devices[path] = {
+			'manufacturer' : manufacturer,
+			'product' : product
+		}
+
+	return devices
+	
+def fido2_enroll(hsm_device_path :pathlib.Path, partition :Partition, password :str) -> bool:
+	worker = SysCommandWorker(f"systemd-cryptenroll --fido2-device={hsm_device_path} {partition.real_device}", peak_output=True)
+	pw_inputted = False
+	while worker.is_alive():
+		if pw_inputted is False and bytes(f"please enter current passphrase for disk {partition.real_device}", 'UTF-8') in worker._trace_log.lower():
+			worker.write(bytes(password, 'UTF-8'))
+			pw_inputted = True