Add support for systemd-homed

Author: luca020400

archinstall/lib/disk/device_handler.py

@@ -271,12 +271,15 @@ def format(
 		options = []
 
 		match fs_type:
-			case FilesystemType.Btrfs | FilesystemType.F2fs | FilesystemType.Xfs:
+			case FilesystemType.Btrfs | FilesystemType.Xfs:
 				# Force overwrite
 				options.append('-f')
+			case  FilesystemType.F2fs:
+				# Force overwrite and enable encrypt
+				options.extend(('-f', '-O', 'encrypt'))
 			case FilesystemType.Ext2 | FilesystemType.Ext3 | FilesystemType.Ext4:
-				# Force create
-				options.append('-F')
+				# Force create and enable encrypt
+				options.extend(('-F', '-O', 'encrypt'))
 			case FilesystemType.Fat12 | FilesystemType.Fat16 | FilesystemType.Fat32:
 				mkfs_type = 'fat'
 				# Set FAT size

archinstall/lib/installer.py

@@ -36,7 +36,7 @@
 
 from .args import arch_config_handler
 from .exceptions import DiskError, HardwareIncompatibilityError, RequirementError, ServiceException, SysCallError
-from .general import SysCommand, run
+from .general import SysCommand, SysCommandWorker, run
 from .hardware import SysInfo
 from .locale.utils import verify_keyboard_layout, verify_x11_keyboard_layout
 from .luks import Luks2
@@ -1701,27 +1701,36 @@ def _create_user(self, user: User) -> None:
 		if not handled_by_plugin:
 			info(f'Creating user {user.username}')
 
-			cmd = f'arch-chroot {self.target} useradd -m'
+			if user.homed.use_homed:
+				self.enable_service('systemd-homed')
 
-			if user.sudo:
-				cmd += ' -G wheel'
+				identity = user.to_identity()
 
-			cmd += f' {user.username}'
+				cmd = f'arch-chroot {self.target} homectl create -P --identity=-'
+				run(cmd.split(" "), input_data=identity.encode())
+			else:
+				cmd = f'arch-chroot {self.target} useradd -m'
 
-			try:
-				SysCommand(cmd)
-			except SysCallError as err:
-				raise SystemError(f'Could not create user inside installation: {err}')
+				if user.sudo:
+					cmd += ' -G wheel'
+
+				cmd += f' {user.username}'
+
+				try:
+					SysCommand(cmd)
+				except SysCallError as err:
+					raise SystemError(f'Could not create user inside installation: {err}')
 
 		for plugin in plugins.values():
 			if hasattr(plugin, 'on_user_created'):
 				if result := plugin.on_user_created(self, user):
 					handled_by_plugin = result
 
-		self.set_user_password(user)
+		if not user.homed.use_homed:
+			self.set_user_password(user)
 
-		for group in user.groups:
-			SysCommand(f'arch-chroot {self.target} gpasswd -a {user.username} {group}')
+			for group in user.groups:
+				SysCommand(f'arch-chroot {self.target} gpasswd -a {user.username} {group}')
 
 		if user.sudo:
 			self.enable_sudo(user)

archinstall/lib/interactions/manage_users_conf.py

@@ -7,10 +7,10 @@
 from archinstall.tui.curses_menu import EditMenu, SelectMenu
 from archinstall.tui.menu_item import MenuItem, MenuItemGroup
 from archinstall.tui.result import ResultType
-from archinstall.tui.types import Alignment, Orientation
+from archinstall.tui.types import Alignment, FrameProperties, Orientation
 
 from ..menu.list_manager import ListManager
-from ..models.users import User
+from ..models.users import User, StorageMechanism, HomedConfiguration
 from ..utils.util import get_password
 
 
@@ -111,7 +111,55 @@ def _add_user(self) -> User | None:
 			case _:
 				raise ValueError('Unhandled result type')
 
-		return User(username, password, sudo)
+		homed_configuration = self._configure_homed()
+
+		return User(username, password, sudo, homed_configuration)
+
+
+	def _configure_homed(self) -> HomedConfiguration:
+		header = str(tr('Should the user use systemd-homed?\n'))
+
+		group = MenuItemGroup.yes_no()
+		group.focus_item = MenuItem.no()
+
+		result = SelectMenu[bool](
+			group,
+			header=header,
+			alignment=Alignment.CENTER,
+			columns=2,
+			orientation=Orientation.HORIZONTAL,
+			search_enabled=False,
+			allow_skip=False,
+		).run()
+
+		match result.type_:
+			case ResultType.Selection:
+				if result.item() != MenuItem.yes():
+					return HomedConfiguration(use_homed=False)
+			case _:
+				raise ValueError('Unhandled result type')
+
+		items = [
+			MenuItem('Directory', value=StorageMechanism.DIRECTORY),
+			MenuItem('LUKS', value=StorageMechanism.LUKS),
+			#MenuItem('fscrypt', value=StorageMechanism.FSCRYPT),
+		]
+
+		group = MenuItemGroup(items, sort_items=False)
+		result = SelectMenu[StorageMechanism](
+			group,
+			alignment=Alignment.CENTER,
+			frame=FrameProperties.min('Filesystem'),
+			allow_skip=False,
+		).run()
+
+		match result.type_:
+			case ResultType.Selection:
+				return HomedConfiguration(use_homed=True, storage_mechanism=result.item().value)
+			case _:
+				raise ValueError('Unhandled result type')
+
+		return HomedConfiguration(use_homed=False)
 
 
 def ask_for_additional_users(prompt: str = '', defined_users: list[User] = []) -> list[User]:

archinstall/lib/models/users.py

@@ -1,3 +1,5 @@
+import json
+
 from dataclasses import dataclass, field
 from enum import Enum
 from typing import NotRequired, TypedDict, override
@@ -100,12 +102,40 @@ def _check_password_strength(
 		return PasswordStrength.VERY_WEAK
 
 
+
+class StorageMechanism:
+	DIRECTORY = 'directory'
+	LUKS = 'luks'
+	FSCRYPT = 'fscrypt'
+
+
+class HomedConfiguration:
+	def __init__(
+		self,
+		use_homed: bool,
+		storage_mechanism: StorageMechanism | None = None,
+	):
+		self.use_homed = use_homed
+		self.storage_mechanism = storage_mechanism
+
+	@override
+	def __eq__(self, other: object) -> bool:
+		if not isinstance(other, HomedConfiguration):
+			return NotImplemented
+
+		return (
+			self.use_homed == other.use_homed
+			and self.storage_mechanism == other.storage_mechanism
+		)
+
+
 UserSerialization = TypedDict(
 	'UserSerialization',
 	{
 		'username': str,
 		'!password': NotRequired[str],
 		'sudo': bool,
+		'homed': HomedConfiguration,
 		'groups': list[str],
 		'enc_password': str | None,
 	},
@@ -158,18 +188,20 @@ class User:
 	username: str
 	password: Password
 	sudo: bool
+	homed: HomedConfiguration
 	groups: list[str] = field(default_factory=list)
 
 	@override
 	def __str__(self) -> str:
 		# safety overwrite to make sure password is not leaked
-		return f'User({self.username=}, {self.sudo=}, {self.groups=})'
+		return f'User({self.username=}, {self.sudo=}, {self.homed}, {self.groups=})'
 
-	def table_data(self) -> dict[str, str | bool | list[str]]:
+	def table_data(self) -> dict[str, str | str | bool | bool | list[str]]:
 		return {
 			'username': self.username,
 			'password': self.password.hidden(),
 			'sudo': self.sudo,
+			'homed': self.homed.use_homed,
 			'groups': self.groups,
 		}
 
@@ -178,9 +210,48 @@ def json(self) -> UserSerialization:
 			'username': self.username,
 			'enc_password': self.password.enc_password,
 			'sudo': self.sudo,
+			'homed': self.homed,
 			'groups': self.groups,
 		}
 
+	def to_identity(self) -> str:
+		if not self.homed.use_homed:
+			raise ValueError('Homed is not enabled for this user')
+
+		identity = {
+			'userName': self.username,
+			'disposition': 'regular',
+			'privileged': {
+				'hashedPassword': [
+					self.password.enc_password
+				],
+			},
+			'enforcePasswordPolicy': False,
+			'secret': {
+				'password': [
+					self.password.plaintext,
+				],
+			}
+		}
+
+		identity['memberOf'] = []
+
+		if self.sudo:
+			identity['memberOf'].append('wheel')
+
+		if self.groups:
+			identity['memberOf'].extend(self.groups)
+
+		match self.homed.storage_mechanism:
+			case StorageMechanism.DIRECTORY:
+				identity['storage'] = 'directory'
+			case StorageMechanism.LUKS:
+				identity['storage'] = 'luks'
+			case StorageMechanism.FSCRYPT:
+				identity['storage'] = 'fscrypt'
+
+		return json.dumps(identity)
+
 	@classmethod
 	def parse_arguments(
 		cls,
@@ -208,6 +279,7 @@ def parse_arguments(
 				username=username,
 				password=password,
 				sudo=entry.get('sudo', False) is True,
+				homed=entry.get('homed', HomedConfiguration(use_homed=False)),
 				groups=groups,
 			)