feat(disk): fix menu rendering and complete backend data pipeline for custom LUKS2 cipher

This commit is contained in:
Alperen42v 2026-06-29 12:21:09 +03:00
parent 0f16801448
commit 0c5ec83b25
4 changed files with 157 additions and 7 deletions

View File

@ -17,6 +17,7 @@ from archinstall.lib.disk.utils import (
from archinstall.lib.exceptions import DiskError, SysCallError, UnknownFilesystemFormat
from archinstall.lib.log import debug, error, info, log
from archinstall.lib.models.device import (
DEFAULT_CIPHER,
DEFAULT_ITER_TIME,
BDevice,
BtrfsMountOption,
@ -280,6 +281,8 @@ class DeviceHandler:
enc_password: Password | None,
lock_after_create: bool = True,
iter_time: int = DEFAULT_ITER_TIME,
cipher: str = DEFAULT_CIPHER,
integrity: str | None = None,
) -> Luks2:
luks_handler = Luks2(
dev_path,
@ -287,7 +290,7 @@ class DeviceHandler:
password=enc_password,
)
key_file = luks_handler.encrypt(iter_time=iter_time)
key_file = luks_handler.encrypt(iter_time=iter_time, cipher=cipher, integrity=integrity)
udev_sync()
@ -318,7 +321,7 @@ class DeviceHandler:
password=enc_conf.encryption_password,
)
key_file = luks_handler.encrypt(iter_time=enc_conf.iter_time)
key_file = luks_handler.encrypt(iter_time=enc_conf.iter_time, cipher=enc_conf.cipher, integrity=enc_conf.integrity)
udev_sync()

View File

@ -7,6 +7,8 @@ from archinstall.lib.menu.helpers import Input, Selection, Table
from archinstall.lib.menu.menu_helper import MenuHelper
from archinstall.lib.menu.util import get_password
from archinstall.lib.models.device import (
AEAD_CIPHERS,
DEFAULT_CIPHER,
DEFAULT_ITER_TIME,
DeviceModification,
DiskEncryption,
@ -64,6 +66,14 @@ class DiskEncryptionMenu(AbstractSubMenu[DiskEncryption]):
preview_action=self._prev_password,
key='encryption_password',
),
MenuItem(
text=tr('Encryption cipher'),
action=self._select_cipher,
value=self._enc_config.cipher,
dependencies=[self._check_dep_enc_type],
preview_action=self._prev_cipher,
key='cipher',
),
MenuItem(
text=tr('Iteration time'),
action=select_iteration_time,
@ -103,6 +113,19 @@ class DiskEncryptionMenu(AbstractSubMenu[DiskEncryption]):
return await select_lvm_vols_to_encrypt(self._lvm_config, preset=preset)
return []
async def _select_cipher(self, preset: str | None) -> str | None:
cipher = await select_encryption_cipher(preset)
# AEAD ciphers (e.g. chacha20-random) require LUKS2 + --integrity;
# keep the underlying DiskEncryption.integrity in sync automatically
# so the user never has to set it manually or risk an invalid combo.
if cipher in AEAD_CIPHERS:
self._enc_config.integrity = AEAD_CIPHERS[cipher]
else:
self._enc_config.integrity = None
return cipher
def _check_dep_enc_type(self) -> bool:
enc_type: EncryptionType | None = self._item_group.find_by_key('encryption_type').value
if enc_type and enc_type != EncryptionType.NO_ENCRYPTION:
@ -129,6 +152,7 @@ class DiskEncryptionMenu(AbstractSubMenu[DiskEncryption]):
enc_type: EncryptionType | None = self._item_group.find_by_key('encryption_type').value
enc_password: Password | None = self._item_group.find_by_key('encryption_password').value
cipher: str | None = self._item_group.find_by_key('cipher').value
iter_time: int | None = self._item_group.find_by_key('iter_time').value
enc_partitions = self._item_group.find_by_key('partitions').value
enc_lvm_vols = self._item_group.find_by_key('lvm_volumes').value
@ -144,6 +168,9 @@ class DiskEncryptionMenu(AbstractSubMenu[DiskEncryption]):
enc_partitions = []
if enc_type != EncryptionType.NO_ENCRYPTION and enc_password and (enc_partitions or enc_lvm_vols):
cipher = cipher or DEFAULT_CIPHER
integrity = AEAD_CIPHERS.get(cipher)
return DiskEncryption(
encryption_password=enc_password,
encryption_type=enc_type,
@ -151,6 +178,8 @@ class DiskEncryptionMenu(AbstractSubMenu[DiskEncryption]):
lvm_volumes=enc_lvm_vols,
hsm_device=enc_config.hsm_device,
iter_time=iter_time or DEFAULT_ITER_TIME,
cipher=cipher,
integrity=integrity,
)
return None
@ -164,6 +193,9 @@ class DiskEncryptionMenu(AbstractSubMenu[DiskEncryption]):
if (enc_pwd := self._prev_password(item)) is not None:
output += f'\n{enc_pwd}'
if (cipher := self._prev_cipher(item)) is not None:
output += f'\n{cipher}'
if (iter_time := self._prev_iter_time(item)) is not None:
output += f'\n{iter_time}'
@ -196,6 +228,18 @@ class DiskEncryptionMenu(AbstractSubMenu[DiskEncryption]):
return None
def _prev_cipher(self, item: MenuItem) -> str | None:
cipher = self._item_group.find_by_key('cipher').value
if cipher:
output = f'{tr("Encryption cipher")}: {cipher}'
if cipher in AEAD_CIPHERS:
integrity = AEAD_CIPHERS[cipher]
output += f'\n{tr("Integrity")}: {integrity} ({tr("requires LUKS2")})'
return output
return None
def _prev_partitions(self, item: MenuItem) -> str | None:
if item.value:
output = tr('Partitions to be encrypted') + '\n'
@ -404,3 +448,39 @@ async def select_iteration_time(preset: int | None = None) -> int | None:
return int(result.get_value())
case ResultType.Reset:
return None
async def select_encryption_cipher(preset: str | None = None) -> str | None:
# Regular block-cipher modes: accepted directly by `cryptsetup --cipher`.
# AEAD modes (see AEAD_CIPHERS in models/device.py) need LUKS2 and a
# separate --integrity value, which DiskEncryption.__post_init__ and
# DiskEncryptionMenu._select_cipher fill in automatically.
options = [
'aes-xts-plain64',
'aes-cbc-essiv:sha256',
'serpent-xts-plain64',
'twofish-xts-plain64',
'chacha20-random', # AEAD, requires --integrity poly1305, LUKS2 only
]
if not preset:
preset = options[0]
items = [MenuItem(o, value=o) for o in options]
group = MenuItemGroup(items)
group.set_focus_by_value(preset)
result = await Selection[str](
group,
header=tr('Select encryption cipher'),
allow_skip=True,
allow_reset=True,
).show()
match result.type_:
case ResultType.Reset:
return None
case ResultType.Skip:
return preset
case ResultType.Selection:
return result.get_value()

View File

@ -3,12 +3,13 @@ from dataclasses import dataclass
from pathlib import Path
from subprocess import CalledProcessError
from types import TracebackType
from typing import cast
from archinstall.lib.command import SysCommand, SysCommandWorker, run
from archinstall.lib.disk.utils import get_lsblk_info, umount
from archinstall.lib.exceptions import DiskError, SysCallError
from archinstall.lib.log import debug, info
from archinstall.lib.models.device import DEFAULT_ITER_TIME
from archinstall.lib.models.device import AEAD_CIPHERS, DEFAULT_CIPHER, DEFAULT_ITER_TIME
from archinstall.lib.models.users import Password
from archinstall.lib.utils.util import generate_password
@ -73,11 +74,21 @@ class Luks2:
hash_type: str = 'sha512',
iter_time: int = DEFAULT_ITER_TIME,
key_file: Path | None = None,
cipher: str = DEFAULT_CIPHER,
integrity: str | None = None,
) -> Path | None:
debug(f'Luks2 encrypting: {self.luks_dev_path}')
key_file_arg, passphrase = self._get_passphrase_args(key_file)
is_aead = cipher in AEAD_CIPHERS
if is_aead and not integrity:
integrity = AEAD_CIPHERS[cipher]
if integrity and not is_aead:
raise DiskError(f'Integrity option "{integrity}" is only valid for AEAD ciphers, not "{cipher}"')
cmd = [
'cryptsetup',
'--batch-mode',
@ -86,10 +97,25 @@ class Luks2:
'luks2',
'--pbkdf',
'argon2id',
'--hash',
hash_type,
'--key-size',
str(key_size),
'--cipher',
cipher,
]
if is_aead:
# AEAD ciphers (e.g. chacha20-random) authenticate each sector via
# --integrity and do not take --hash or a fixed --key-size the same
# way length-preserving modes (aes-xts-plain64 etc.) do; key size is
# derived from the cipher/integrity combination by cryptsetup itself.
cmd += ['--integrity', cast(str, integrity)]
else:
cmd += [
'--hash',
hash_type,
'--key-size',
str(key_size),
]
cmd += [
'--iter-time',
str(iter_time),
*key_file_arg,

View File

@ -1462,12 +1462,26 @@ class EncryptionType(StrEnum):
return type_to_text[self]
DEFAULT_CIPHER = 'aes-xts-plain64'
# AEAD (Authenticated Encryption with Associated Data) ciphers require LUKS2
# and a separate --integrity parameter; they cannot be used like normal
# block-cipher + chainmode + ivmode strings (e.g. aes-xts-plain64).
# Mapping: cipher value used in --cipher -> required --integrity value
AEAD_CIPHERS: dict[str, str] = {
'chacha20-random': 'poly1305',
'aes-gcm-random': 'aead',
}
class _DiskEncryptionSerialization(TypedDict):
encryption_type: str
partitions: list[str]
lvm_volumes: list[str]
hsm_device: NotRequired[_Fido2DeviceSerialization]
iter_time: NotRequired[int]
cipher: NotRequired[str]
integrity: NotRequired[str]
@dataclass
@ -1478,6 +1492,8 @@ class DiskEncryption:
lvm_volumes: list[LvmVolume] = field(default_factory=list)
hsm_device: Fido2Device | None = None
iter_time: int = DEFAULT_ITER_TIME
cipher: str = DEFAULT_CIPHER
integrity: str | None = None
def __post_init__(self) -> None:
if self.encryption_type in [EncryptionType.LUKS, EncryptionType.LVM_ON_LUKS] and not self.partitions:
@ -1486,6 +1502,20 @@ class DiskEncryption:
if self.encryption_type == EncryptionType.LUKS_ON_LVM and not self.lvm_volumes:
raise ValueError('LuksOnLvm encryption require LMV volumes to be defined')
# AEAD ciphers (chacha20-random, aes-gcm-random) are only valid on LUKS2
# and require --integrity to be set; auto-fill it if missing so the
# cryptsetup call downstream doesn't end up with an invalid combination.
if self.cipher in AEAD_CIPHERS:
required_integrity = AEAD_CIPHERS[self.cipher]
if not self.integrity:
self.integrity = required_integrity
elif self.integrity != required_integrity:
raise ValueError(
f'Cipher {self.cipher!r} requires integrity {required_integrity!r}, got {self.integrity!r}'
)
elif self.integrity:
raise ValueError(f'Integrity option is only valid for AEAD ciphers, not {self.cipher!r}')
def should_generate_encryption_file(self, dev: PartitionModification | LvmVolume) -> bool:
if isinstance(dev, PartitionModification):
return dev in self.partitions and dev.mountpoint != Path('/')
@ -1505,6 +1535,12 @@ class DiskEncryption:
if self.iter_time != DEFAULT_ITER_TIME: # Only include if not default
obj['iter_time'] = self.iter_time
if self.cipher != DEFAULT_CIPHER: # Only include if not default
obj['cipher'] = self.cipher
if self.integrity:
obj['integrity'] = self.integrity
return obj
@staticmethod
@ -1549,11 +1585,16 @@ class DiskEncryption:
if vol.obj_id in disk_encryption.get('lvm_volumes', []):
volumes.append(vol)
cipher = disk_encryption.get('cipher', None) or DEFAULT_CIPHER
integrity = disk_encryption.get('integrity', None)
enc = cls(
EncryptionType(disk_encryption['encryption_type']),
password,
enc_partitions,
volumes,
cipher=cipher,
integrity=integrity,
)
if hsm := disk_encryption.get('hsm_device', None):