import glob import os import re import shlex import shutil import subprocess import time from pathlib import Path from typing import Any, List, Optional, TYPE_CHECKING, Union, Dict, Callable from . import disk from .exceptions import DiskError, ServiceException, RequirementError, HardwareIncompatibilityError, SysCallError from .general import SysCommand from .hardware import SysInfo from .locale import LocaleConfiguration from .locale import verify_keyboard_layout, verify_x11_keyboard_layout from .luks import Luks2 from .mirrors import use_mirrors, MirrorConfiguration, add_custom_mirrors from .models.bootloader import Bootloader from .models.network_configuration import Nic from .models.users import User from .output import log, error, info, warn, debug from . import pacman from .pacman import Pacman from .plugins import plugins from .storage import storage if TYPE_CHECKING: _: Any # Any package that the Installer() is responsible for (optional and the default ones) __packages__ = ["base", "base-devel", "linux-firmware", "linux", "linux-lts", "linux-zen", "linux-hardened"] # Additional packages that are installed if the user is running the Live ISO with accessibility tools enabled __accessibility_packages__ = ["brltty", "espeakup", "alsa-utils"] def accessibility_tools_in_use() -> bool: return os.system('systemctl is-active --quiet espeakup.service') == 0 class Installer: def __init__( self, target: Path, disk_config: disk.DiskLayoutConfiguration, disk_encryption: Optional[disk.DiskEncryption] = None, base_packages: List[str] = [], kernels: Optional[List[str]] = None ): """ `Installer()` is the wrapper for most basic installation steps. It also wraps :py:func:`~archinstall.Installer.pacstrap` among other things. """ self.base_packages = base_packages or __packages__[:3] self.kernels = kernels or ['linux'] self._disk_config = disk_config self._disk_encryption = disk_encryption or disk.DiskEncryption(disk.EncryptionType.NoEncryption) self.target: Path = target self.init_time = time.strftime('%Y-%m-%d_%H-%M-%S') self.milliseconds = int(str(time.time()).split('.')[1]) self.helper_flags: Dict[str, Any] = {'base': False, 'bootloader': None} for kernel in self.kernels: self.base_packages.append(kernel) # If using accessibility tools in the live environment, append those to the packages list if accessibility_tools_in_use(): self.base_packages.extend(__accessibility_packages__) self.post_base_install: List[Callable] = [] # TODO: Figure out which one of these two we'll use.. But currently we're mixing them.. storage['session'] = self storage['installation_session'] = self self.modules: List[str] = [] self._binaries: List[str] = [] self._files: List[str] = [] # systemd, sd-vconsole and sd-encrypt will be replaced by udev, keymap and encrypt # if HSM is not used to encrypt the root volume. Check mkinitcpio() function for that override. self._hooks: List[str] = [ "base", "systemd", "autodetect", "keyboard", "sd-vconsole", "modconf", "block", "filesystems", "fsck" ] self._kernel_params: List[str] = [] self._fstab_entries: List[str] = [] self._zram_enabled = False self.pacman = Pacman(self.target, storage['arguments'].get('silent', False)) def __enter__(self) -> 'Installer': return self def __exit__(self, exc_type, exc_val, exc_tb): if exc_type is not None: error(exc_val) self.sync_log_to_install_medium() # We avoid printing /mnt/ because that might confuse people if they note it down # and then reboot, and a identical log file will be found in the ISO medium anyway. print(_("[!] A log file has been created here: {}").format(os.path.join(storage['LOG_PATH'], storage['LOG_FILE']))) print(_(" Please submit this issue (and file) to https://github.com/archlinux/archinstall/issues")) raise exc_val if not (missing_steps := self.post_install_check()): log('Installation completed without any errors. You may now reboot.', fg='green') self.sync_log_to_install_medium() return True else: warn('Some required steps were not successfully installed/configured before leaving the installer:') for step in missing_steps: warn(f' - {step}') warn(f"Detailed error logs can be found at: {storage['LOG_PATH']}") warn("Submit this zip file as an issue to https://github.com/archlinux/archinstall/issues") self.sync_log_to_install_medium() return False def _verify_service_stop(self): """ Certain services might be running that affects the system during installation. One such service is "reflector.service" which updates /etc/pacman.d/mirrorlist We need to wait for it before we continue since we opted in to use a custom mirror/region. """ info('Waiting for time sync (systemd-timesyncd.service) to complete.') while True: time_val = SysCommand('timedatectl show --property=NTPSynchronized --value').decode() if time_val and time_val.strip() == 'yes': break time.sleep(1) info('Waiting for automatic mirror selection (reflector) to complete.') while self._service_state('reflector') not in ('dead', 'failed', 'exited'): time.sleep(1) # info('Waiting for pacman-init.service to complete.') # while self._service_state('pacman-init') not in ('dead', 'failed', 'exited'): # time.sleep(1) info('Waiting for Arch Linux keyring sync (archlinux-keyring-wkd-sync) to complete.') # Wait for the timer to kick in while self._service_started('archlinux-keyring-wkd-sync.timer') is None: time.sleep(1) # Wait for the service to enter a finished state while self._service_state('archlinux-keyring-wkd-sync.service') not in ('dead', 'failed', 'exited'): time.sleep(1) def _verify_boot_part(self): """ Check that mounted /boot device has at minimum size for installation The reason this check is here is to catch pre-mounted device configuration and potentially configured one that has not gone through any previous checks (e.g. --silence mode) NOTE: this function should be run AFTER running the mount_ordered_layout function """ boot_mount = self.target / 'boot' lsblk_info = disk.get_lsblk_by_mountpoint(boot_mount) if len(lsblk_info) > 0: if lsblk_info[0].size < disk.Size(200, disk.Unit.MiB, disk.SectorSize.default()): raise DiskError( f'The boot partition mounted at {boot_mount} is not large enough to install a boot loader. ' f'Please resize it to at least 200MiB and re-run the installation.' ) def sanity_check(self): # self._verify_boot_part() self._verify_service_stop() def mount_ordered_layout(self): info('Mounting partitions in order') for mod in self._disk_config.device_modifications: # partitions have to mounted in the right order on btrfs the mountpoint will # be empty as the actual subvolumes are getting mounted instead so we'll use # '/' just for sorting sorted_part_mods = sorted(mod.partitions, key=lambda x: x.mountpoint or Path('/')) enc_partitions = [] if self._disk_encryption.encryption_type is not disk.EncryptionType.NoEncryption: enc_partitions = list(set(sorted_part_mods) & set(self._disk_encryption.partitions)) # attempt to decrypt all luks partitions luks_handlers = self._prepare_luks_partitions(enc_partitions) for part_mod in sorted_part_mods: if luks_handler := luks_handlers.get(part_mod): # mount encrypted partition self._mount_luks_partition(part_mod, luks_handler) else: # partition is not encrypted self._mount_partition(part_mod) def _prepare_luks_partitions(self, partitions: List[disk.PartitionModification]) -> Dict[disk.PartitionModification, Luks2]: return { part_mod: disk.device_handler.unlock_luks2_dev( part_mod.dev_path, part_mod.mapper_name, self._disk_encryption.encryption_password ) for part_mod in partitions if part_mod.mapper_name and part_mod.dev_path } def _mount_partition(self, part_mod: disk.PartitionModification): # it would be none if it's btrfs as the subvolumes will have the mountpoints defined if part_mod.mountpoint and part_mod.dev_path: target = self.target / part_mod.relative_mountpoint disk.device_handler.mount(part_mod.dev_path, target, options=part_mod.mount_options) if part_mod.fs_type == disk.FilesystemType.Btrfs and part_mod.dev_path: self._mount_btrfs_subvol(part_mod.dev_path, part_mod.btrfs_subvols) def _mount_luks_partition(self, part_mod: disk.PartitionModification, luks_handler: Luks2): # it would be none if it's btrfs as the subvolumes will have the mountpoints defined if part_mod.mountpoint and luks_handler.mapper_dev: target = self.target / part_mod.relative_mountpoint disk.device_handler.mount(luks_handler.mapper_dev, target, options=part_mod.mount_options) if part_mod.fs_type == disk.FilesystemType.Btrfs and luks_handler.mapper_dev: self._mount_btrfs_subvol(luks_handler.mapper_dev, part_mod.btrfs_subvols) def _mount_btrfs_subvol(self, dev_path: Path, subvolumes: List[disk.SubvolumeModification]): for subvol in subvolumes: mountpoint = self.target / subvol.relative_mountpoint mount_options = subvol.mount_options + [f'subvol={subvol.name}'] disk.device_handler.mount(dev_path, mountpoint, options=mount_options) def generate_key_files(self): for part_mod in self._disk_encryption.partitions: gen_enc_file = self._disk_encryption.should_generate_encryption_file(part_mod) luks_handler = Luks2( part_mod.safe_dev_path, mapper_name=part_mod.mapper_name, password=self._disk_encryption.encryption_password ) if gen_enc_file and not part_mod.is_root(): info(f'Creating key-file: {part_mod.dev_path}') luks_handler.create_keyfile(self.target) if part_mod.is_root() and not gen_enc_file: if self._disk_encryption.hsm_device: disk.Fido2.fido2_enroll( self._disk_encryption.hsm_device, part_mod, self._disk_encryption.encryption_password ) 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. if self.helper_flags.get('base-strapped', False) is True: if filename := storage.get('LOG_FILE', None): absolute_logfile = os.path.join(storage.get('LOG_PATH', './'), filename) if not os.path.isdir(f"{self.target}/{os.path.dirname(absolute_logfile)}"): os.makedirs(f"{self.target}/{os.path.dirname(absolute_logfile)}") shutil.copy2(absolute_logfile, f"{self.target}/{absolute_logfile}") return True def add_swapfile(self, size='4G', enable_resume=True, file='/swapfile'): if file[:1] != '/': file = f"/{file}" if len(file.strip()) <= 0 or file == '/': raise ValueError(f"The filename for the swap file has to be a valid path, not: {self.target}{file}") SysCommand(f'dd if=/dev/zero of={self.target}{file} bs={size} count=1') SysCommand(f'chmod 0600 {self.target}{file}') SysCommand(f'mkswap {self.target}{file}') self._fstab_entries.append(f'{file} none swap defaults 0 0') if enable_resume: resume_uuid = SysCommand(f'findmnt -no UUID -T {self.target}{file}').decode() resume_offset = SysCommand( f'/usr/bin/filefrag -v {self.target}{file}' ).decode().split('0:', 1)[1].split(":", 1)[1].split("..", 1)[0].strip() self._hooks.append('resume') self._kernel_params.append(f'resume=UUID={resume_uuid}') self._kernel_params.append(f'resume_offset={resume_offset}') def post_install_check(self, *args :str, **kwargs :str) -> List[str]: return [step for step, flag in self.helper_flags.items() if flag is False] def set_mirrors(self, mirror_config: MirrorConfiguration): for plugin in plugins.values(): if hasattr(plugin, 'on_mirrors'): if result := plugin.on_mirrors(mirror_config): mirror_config = result destination = f'{self.target}/etc/pacman.d/mirrorlist' if mirror_config.mirror_regions: use_mirrors(mirror_config.mirror_regions, destination) if mirror_config.custom_mirrors: add_custom_mirrors(mirror_config.custom_mirrors) def genfstab(self, flags :str = '-pU'): fstab_path = self.target / "etc" / "fstab" info(f"Updating {fstab_path}") try: gen_fstab = SysCommand(f'/usr/bin/genfstab {flags} {self.target}').decode() except SysCallError as err: raise RequirementError(f'Could not generate fstab, strapping in packages most likely failed (disk out of space?)\n Error: {err}') with open(fstab_path, 'a') as fp: fp.write(gen_fstab) if not fstab_path.is_file(): raise RequirementError(f'Could not create fstab file') for plugin in plugins.values(): if hasattr(plugin, 'on_genfstab'): if plugin.on_genfstab(self) is True: break with open(fstab_path, 'a') as fp: for entry in self._fstab_entries: fp.write(f'{entry}\n') for mod in self._disk_config.device_modifications: for part_mod in mod.partitions: if part_mod.fs_type != disk.FilesystemType.Btrfs: continue with fstab_path.open('r') as fp: fstab = fp.readlines() # Replace the {installation}/etc/fstab with entries # using the compress=zstd where the mountpoint has compression set. for index, line in enumerate(fstab): # So first we grab the mount options by using subvol=.*? as a locator. # And we also grab the mountpoint for the entry, for instance /var/log subvoldef = re.findall(',.*?subvol=.*?[\t ]', line) mountpoint = re.findall('[\t ]/.*?[\t ]', line) if not subvoldef or not mountpoint: continue for sub_vol in part_mod.btrfs_subvols: # We then locate the correct subvolume and check if it's compressed, # and skip entries where compression is already defined # We then sneak in the compress=zstd option if it doesn't already exist: if sub_vol.compress and str(sub_vol.mountpoint) == Path(mountpoint[0].strip()) and ',compress=zstd,' not in line: fstab[index] = line.replace(subvoldef[0], f',compress=zstd{subvoldef[0]}') break with fstab_path.open('w') as fp: fp.writelines(fstab) def set_hostname(self, hostname: str, *args :str, **kwargs :str) -> None: with open(f'{self.target}/etc/hostname', 'w') as fh: fh.write(hostname + '\n') def set_locale(self, locale_config: LocaleConfiguration) -> bool: modifier = '' lang = locale_config.sys_lang encoding = locale_config.sys_enc # This is a temporary patch to fix #1200 if '.' in locale_config.sys_lang: lang, potential_encoding = locale_config.sys_lang.split('.', 1) # Override encoding if encoding is set to the default parameter # and the "found" encoding differs. if locale_config.sys_enc == 'UTF-8' and locale_config.sys_enc != potential_encoding: encoding = potential_encoding # Make sure we extract the modifier, that way we can put it in if needed. if '@' in locale_config.sys_lang: lang, modifier = locale_config.sys_lang.split('@', 1) modifier = f"@{modifier}" # - End patch locale_gen = self.target / 'etc/locale.gen' locale_gen_lines = locale_gen.read_text().splitlines(True) # A locale entry in /etc/locale.gen may or may not contain the encoding # in the first column of the entry; check for both cases. entry_re = re.compile(rf'#{lang}(\.{encoding})?{modifier} {encoding}') for index, line in enumerate(locale_gen_lines): if entry_re.match(line): uncommented_line = line.removeprefix('#') locale_gen_lines[index] = uncommented_line locale_gen.write_text(''.join(locale_gen_lines)) lang_value = uncommented_line.split()[0] break else: error(f"Invalid locale: language '{locale_config.sys_lang}', encoding '{locale_config.sys_enc}'") return False try: SysCommand(f'/usr/bin/arch-chroot {self.target} locale-gen') except SysCallError as e: error(f'Failed to run locale-gen on target: {e}') return False (self.target / 'etc/locale.conf').write_text(f'LANG={lang_value}\n') return True def set_timezone(self, zone :str, *args :str, **kwargs :str) -> bool: if not zone: return True if not len(zone): return True # Redundant for plugin in plugins.values(): if hasattr(plugin, 'on_timezone'): if result := plugin.on_timezone(zone): zone = result if (Path("/usr") / "share" / "zoneinfo" / zone).exists(): (Path(self.target) / "etc" / "localtime").unlink(missing_ok=True) SysCommand(f'/usr/bin/arch-chroot {self.target} ln -s /usr/share/zoneinfo/{zone} /etc/localtime') return True else: warn(f'Time zone {zone} does not exist, continuing with system default') return False def activate_time_synchronization(self) -> None: info('Activating systemd-timesyncd for time synchronization using Arch Linux and ntp.org NTP servers') self.enable_service('systemd-timesyncd') def enable_espeakup(self) -> None: info('Enabling espeakup.service for speech synthesis (accessibility)') self.enable_service('espeakup') def enable_periodic_trim(self) -> None: info("Enabling periodic TRIM") # fstrim is owned by util-linux, a dependency of both base and systemd. self.enable_service("fstrim.timer") def enable_service(self, services: Union[str, List[str]]) -> None: if isinstance(services, str): services = [services] for service in services: info(f'Enabling service {service}') try: self.arch_chroot(f'systemctl enable {service}') except SysCallError as err: raise ServiceException(f"Unable to start service {service}: {err}") for plugin in plugins.values(): if hasattr(plugin, 'on_service'): plugin.on_service(service) def run_command(self, cmd :str, *args :str, **kwargs :str) -> SysCommand: return SysCommand(f'/usr/bin/arch-chroot {self.target} {cmd}') def arch_chroot(self, cmd :str, run_as :Optional[str] = None) -> SysCommand: if run_as: cmd = f"su - {run_as} -c {shlex.quote(cmd)}" return self.run_command(cmd) def drop_to_shell(self) -> None: subprocess.check_call(f"/usr/bin/arch-chroot {self.target}", shell=True) def configure_nic(self, nic: Nic): conf = nic.as_systemd_config() for plugin in plugins.values(): if hasattr(plugin, 'on_configure_nic'): conf = plugin.on_configure_nic( nic.iface, nic.dhcp, nic.ip, nic.gateway, nic.dns ) or conf with open(f"{self.target}/etc/systemd/network/10-{nic.iface}.network", "a") as netconf: netconf.write(str(conf)) def copy_iso_network_config(self, enable_services :bool = False) -> bool: # Copy (if any) iwd password and config files if os.path.isdir('/var/lib/iwd/'): if psk_files := glob.glob('/var/lib/iwd/*.psk'): if not os.path.isdir(f"{self.target}/var/lib/iwd"): os.makedirs(f"{self.target}/var/lib/iwd") if enable_services: # If we haven't installed the base yet (function called pre-maturely) if self.helper_flags.get('base', False) is False: self.base_packages.append('iwd') # This function will be called after minimal_installation() # as a hook for post-installs. This hook is only needed if # base is not installed yet. def post_install_enable_iwd_service(*args :str, **kwargs :str): self.enable_service('iwd') self.post_base_install.append(post_install_enable_iwd_service) # Otherwise, we can go ahead and add the required package # and enable it's service: else: self.pacman.strap('iwd') self.enable_service('iwd') for psk in psk_files: shutil.copy2(psk, f"{self.target}/var/lib/iwd/{os.path.basename(psk)}") # Copy (if any) systemd-networkd config files if netconfigurations := glob.glob('/etc/systemd/network/*'): if not os.path.isdir(f"{self.target}/etc/systemd/network/"): os.makedirs(f"{self.target}/etc/systemd/network/") for netconf_file in netconfigurations: shutil.copy2(netconf_file, f"{self.target}/etc/systemd/network/{os.path.basename(netconf_file)}") if enable_services: # If we haven't installed the base yet (function called pre-maturely) if self.helper_flags.get('base', False) is False: def post_install_enable_networkd_resolved(*args :str, **kwargs :str): self.enable_service(['systemd-networkd', 'systemd-resolved']) self.post_base_install.append(post_install_enable_networkd_resolved) # Otherwise, we can go ahead and enable the services else: self.enable_service(['systemd-networkd', 'systemd-resolved']) return True def mkinitcpio(self, flags: List[str]) -> bool: for plugin in plugins.values(): if hasattr(plugin, 'on_mkinitcpio'): # Allow plugins to override the usage of mkinitcpio altogether. if plugin.on_mkinitcpio(self): return True with open(f'{self.target}/etc/mkinitcpio.conf', 'w') as mkinit: mkinit.write(f"MODULES=({' '.join(self.modules)})\n") mkinit.write(f"BINARIES=({' '.join(self._binaries)})\n") mkinit.write(f"FILES=({' '.join(self._files)})\n") if not self._disk_encryption.hsm_device: # For now, if we don't use HSM we revert to the old # way of setting up encryption hooks for mkinitcpio. # This is purely for stability reasons, we're going away from this. # * systemd -> udev # * sd-vconsole -> keymap self._hooks = [hook.replace('systemd', 'udev').replace('sd-vconsole', 'keymap') for hook in self._hooks] mkinit.write(f"HOOKS=({' '.join(self._hooks)})\n") try: SysCommand(f'/usr/bin/arch-chroot {self.target} mkinitcpio {" ".join(flags)}', peek_output=True) return True except SysCallError as error: if error.worker: log(error.worker._trace_log.decode()) return False def _get_microcode(self) -> Optional[Path]: if not SysInfo.is_vm(): if vendor := SysInfo.cpu_vendor(): return vendor.get_ucode() return None def minimal_installation( self, testing: bool = False, multilib: bool = False, mkinitcpio: bool = True, hostname: str = 'archinstall', locale_config: LocaleConfiguration = LocaleConfiguration.default() ): _disable_fstrim = False for mod in self._disk_config.device_modifications: for part in mod.partitions: if part.fs_type is not None: if (pkg := part.fs_type.installation_pkg) is not None: self.base_packages.append(pkg) if (module := part.fs_type.installation_module) is not None: self.modules.append(module) if (binary := part.fs_type.installation_binary) is not None: self._binaries.append(binary) # https://github.com/archlinux/archinstall/issues/1837 if part.fs_type.fs_type_mount == 'btrfs': _disable_fstrim = True # There is not yet an fsck tool for NTFS. If it's being used for the root filesystem, the hook should be removed. if part.fs_type.fs_type_mount == 'ntfs3' and part.mountpoint == self.target: if 'fsck' in self._hooks: self._hooks.remove('fsck') if part in self._disk_encryption.partitions: if self._disk_encryption.hsm_device: # Required by mkinitcpio to add support for fido2-device options self.pacman.strap('libfido2') if 'sd-encrypt' not in self._hooks: self._hooks.insert(self._hooks.index('filesystems'), 'sd-encrypt') else: if 'encrypt' not in self._hooks: self._hooks.insert(self._hooks.index('filesystems'), 'encrypt') if not SysInfo.has_uefi(): self.base_packages.append('grub') if ucode := self._get_microcode(): (self.target / 'boot' / ucode).unlink(missing_ok=True) self.base_packages.append(ucode.stem) else: debug('Archinstall will not install any ucode.') # Determine whether to enable multilib/testing repositories before running pacstrap if testing flag is set. # This action takes place on the host system as pacstrap copies over package repository lists. pacman_conf = pacman.Config(self.target) if multilib: info("The multilib flag is set. This system will be installed with the multilib repository enabled.") pacman_conf.enable(pacman.Repo.Multilib) else: info("The multilib flag is not set. This system will be installed without multilib repositories enabled.") if testing: info("The testing flag is set. This system will be installed with testing repositories enabled.") pacman_conf.enable(pacman.Repo.Testing) if multilib: pacman_conf.enable(pacman.Repo.MultilibTesting) else: info("The testing flag is not set. This system will be installed without testing repositories enabled.") pacman_conf.apply() self.pacman.strap(self.base_packages) self.helper_flags['base-strapped'] = True pacman_conf.persist() # Periodic TRIM may improve the performance and longevity of SSDs whilst # having no adverse effect on other devices. Most distributions enable # periodic TRIM by default. # # https://github.com/archlinux/archinstall/issues/880 # https://github.com/archlinux/archinstall/issues/1837 # https://github.com/archlinux/archinstall/issues/1841 if not _disable_fstrim: self.enable_periodic_trim() # TODO: Support locale and timezone # os.remove(f'{self.target}/etc/localtime') # sys_command(f'/usr/bin/arch-chroot {self.target} ln -s /usr/share/zoneinfo/{localtime} /etc/localtime') # sys_command('/usr/bin/arch-chroot /mnt hwclock --hctosys --localtime') self.set_hostname(hostname) self.set_locale(locale_config) # TODO: Use python functions for this SysCommand(f'/usr/bin/arch-chroot {self.target} chmod 700 /root') if mkinitcpio and not self.mkinitcpio(['-P']): error(f"Error generating initramfs (continuing anyway)") self.helper_flags['base'] = True # Run registered post-install hooks for function in self.post_base_install: info(f"Running post-installation hook: {function}") function(self) for plugin in plugins.values(): if hasattr(plugin, 'on_install'): plugin.on_install(self) def setup_swap(self, kind :str = 'zram'): if kind == 'zram': info(f"Setting up swap on zram") self.pacman.strap('zram-generator') # We could use the default example below, but maybe not the best idea: https://github.com/archlinux/archinstall/pull/678#issuecomment-962124813 # zram_example_location = '/usr/share/doc/zram-generator/zram-generator.conf.example' # shutil.copy2(f"{self.target}{zram_example_location}", f"{self.target}/usr/lib/systemd/zram-generator.conf") with open(f"{self.target}/etc/systemd/zram-generator.conf", "w") as zram_conf: zram_conf.write("[zram0]\n") self.enable_service('systemd-zram-setup@zram0.service') self._zram_enabled = True else: raise ValueError(f"Archinstall currently only supports setting up swap on zram") def _get_efi_partition(self) -> Optional[disk.PartitionModification]: for layout in self._disk_config.device_modifications: if partition := layout.get_efi_partition(): return partition return None def _get_boot_partition(self) -> Optional[disk.PartitionModification]: for layout in self._disk_config.device_modifications: if boot := layout.get_boot_partition(): return boot return None def _get_root_partition(self) -> Optional[disk.PartitionModification]: for mod in self._disk_config.device_modifications: if root := mod.get_root_partition(): return root return None def _get_kernel_params( self, root_partition: disk.PartitionModification, id_root: bool = True, partuuid: bool = True ) -> List[str]: kernel_parameters = [] if root_partition in self._disk_encryption.partitions: # TODO: We need to detect if the encrypted device is a whole disk encryption, # or simply a partition encryption. Right now we assume it's a partition (and we always have) if self._disk_encryption and self._disk_encryption.hsm_device: debug(f'Root partition is an encrypted device, identifying by UUID: {root_partition.uuid}') # Note: UUID must be used, not PARTUUID for sd-encrypt to work kernel_parameters.append(f'rd.luks.name={root_partition.uuid}=root') # Note: tpm2-device and fido2-device don't play along very well: # https://github.com/archlinux/archinstall/pull/1196#issuecomment-1129715645 kernel_parameters.append('rd.luks.options=fido2-device=auto,password-echo=no') elif partuuid: debug(f'Root partition is an encrypted device, identifying by PARTUUID: {root_partition.partuuid}') kernel_parameters.append(f'cryptdevice=PARTUUID={root_partition.partuuid}:root') else: debug(f'Root partition is an encrypted device, identifying by UUID: {root_partition.uuid}') kernel_parameters.append(f'cryptdevice=UUID={root_partition.uuid}:root') if id_root: kernel_parameters.append('root=/dev/mapper/root') elif id_root: if partuuid: debug(f'Identifying root partition by PARTUUID: {root_partition.partuuid}') kernel_parameters.append(f'root=PARTUUID={root_partition.partuuid}') else: debug(f'Identifying root partition by UUID: {root_partition.uuid}') kernel_parameters.append(f'root=UUID={root_partition.uuid}') # Zswap should be disabled when using zram. # https://github.com/archlinux/archinstall/issues/881 if self._zram_enabled: kernel_parameters.append('zswap.enabled=0') if id_root: for sub_vol in root_partition.btrfs_subvols: if sub_vol.is_root(): kernel_parameters.append(f'rootflags=subvol={sub_vol.name}') break kernel_parameters.append('rw') kernel_parameters.append(f'rootfstype={root_partition.safe_fs_type.fs_type_mount}') kernel_parameters.extend(self._kernel_params) debug(f'kernel parameters: {" ".join(kernel_parameters)}') return kernel_parameters def _add_systemd_bootloader( self, boot_partition: disk.PartitionModification, root_partition: disk.PartitionModification, efi_partition: Optional[disk.PartitionModification], uki_enabled: bool = False ): self.pacman.strap('efibootmgr') if not SysInfo.has_uefi(): raise HardwareIncompatibilityError # TODO: Ideally we would want to check if another config # points towards the same disk and/or partition. # And in which case we should do some clean up. bootctl_options = [] if efi_partition and boot_partition != efi_partition: bootctl_options.append(f'--esp-path={efi_partition.mountpoint}') bootctl_options.append(f'--boot-path={boot_partition.mountpoint}') # Install the boot loader try: SysCommand(f"/usr/bin/arch-chroot {self.target} bootctl {' '.join(bootctl_options)} install") except SysCallError: # Fallback, try creating the boot loader without touching the EFI variables SysCommand(f"/usr/bin/arch-chroot {self.target} bootctl --no-variables {' '.join(bootctl_options)} install") # Ensure that the $BOOT/loader/ directory exists before we try to create files in it. # # As mentioned in https://github.com/archlinux/archinstall/pull/1859 - we store the # loader entries in $BOOT/loader/ rather than $ESP/loader/ # The current reasoning being that $BOOT works in both use cases as well # as being tied to the current installation. This may change. loader_dir = self.target / 'boot/loader' loader_dir.mkdir(parents=True, exist_ok=True) default_kernel = self.kernels[0] if uki_enabled: default_entry = f'arch-{default_kernel}.efi' else: entry_name = self.init_time + '_{kernel}{variant}.conf' default_entry = entry_name.format(kernel=default_kernel, variant='') default = f'default {default_entry}' # Modify or create a loader.conf loader_conf = loader_dir / 'loader.conf' try: loader_data = loader_conf.read_text().splitlines() except FileNotFoundError: loader_data = [ default, 'timeout 15' ] else: for index, line in enumerate(loader_data): if line.startswith('default'): loader_data[index] = default elif line.startswith('#timeout'): # We add in the default timeout to support dual-boot loader_data[index] = line.removeprefix('#') loader_conf.write_text('\n'.join(loader_data) + '\n') if uki_enabled: return # Ensure that the $BOOT/loader/entries/ directory exists before we try to create files in it entries_dir = loader_dir / 'entries' entries_dir.mkdir(parents=True, exist_ok=True) comments = ( '# Created by: archinstall', f'# Created on: {self.init_time}' ) microcode = [] if ucode := self._get_microcode(): microcode.append(f'initrd /{ucode}') else: debug('Archinstall will not add any ucode to systemd-boot config.') options = 'options ' + ' '.join(self._get_kernel_params(root_partition)) for kernel in self.kernels: for variant in ("", "-fallback"): # Setup the loader entry entry = [ *comments, f'title Arch Linux ({kernel}{variant})', f'linux /vmlinuz-{kernel}', *microcode, f'initrd /initramfs-{kernel}{variant}.img', options, ] name = entry_name.format(kernel=kernel, variant=variant) entry_conf = entries_dir / name entry_conf.write_text('\n'.join(entry) + '\n') self.helper_flags['bootloader'] = 'systemd' def _add_grub_bootloader( self, boot_partition: disk.PartitionModification, root_partition: disk.PartitionModification, efi_partition: Optional[disk.PartitionModification], uki_enabled: bool = False ): self.pacman.strap('grub') # no need? if not uki_enabled: grub_default = self.target / 'etc/default/grub' config = grub_default.read_text() kernel_parameters = ' '.join(self._get_kernel_params(root_partition, False, False)) config = re.sub(r'(GRUB_CMDLINE_LINUX=")("\n)', rf'\1{kernel_parameters}\2', config, 1) grub_default.write_text(config) info(f"GRUB boot partition: {boot_partition.dev_path}") if boot_partition == root_partition and root_partition.mountpoint: boot_dir = root_partition.mountpoint / 'boot' elif boot_partition.mountpoint: boot_dir = boot_partition.mountpoint else: raise ValueError('Could not detect boot directory') command = [ '/usr/bin/arch-chroot', str(self.target), 'grub-install', '--debug' ] if SysInfo.has_uefi(): if not efi_partition: raise ValueError('Could not detect efi partition') info(f"GRUB EFI partition: {efi_partition.dev_path}") self.pacman.strap('efibootmgr') # TODO: Do we need? Yes, but remove from minimal_installation() instead? boot_dir_arg = [] if boot_partition != efi_partition: boot_dir_arg.append(f'--boot-directory={boot_dir}') add_options = [ '--target=x86_64-efi', f'--efi-directory={efi_partition.mountpoint}', *boot_dir_arg, '--bootloader-id=GRUB', '--removable' ] command.extend(add_options) try: SysCommand(command, peek_output=True) except SysCallError: try: SysCommand(command, peek_output=True) except SysCallError as err: raise DiskError(f"Could not install GRUB to {self.target}{efi_partition.mountpoint}: {err}") else: info(f"GRUB boot partition: {boot_partition.dev_path}") parent_dev_path = disk.device_handler.get_parent_device_path(boot_partition.safe_dev_path) add_options = [ '--target=i386-pc', '--recheck', str(parent_dev_path) ] try: SysCommand(command + add_options, peek_output=True) except SysCallError as err: raise DiskError(f"Failed to install GRUB boot on {boot_partition.dev_path}: {err}") try: SysCommand( f'/usr/bin/arch-chroot {self.target} ' f'grub-mkconfig -o {boot_dir}/grub/grub.cfg' ) except SysCallError as err: raise DiskError(f"Could not configure GRUB: {err}") self.helper_flags['bootloader'] = "grub" def _add_limine_bootloader( self, boot_partition: disk.PartitionModification, efi_partition: Optional[disk.PartitionModification], root_partition: disk.PartitionModification ): self.pacman.strap('limine') info(f"Limine boot partition: {boot_partition.dev_path}") limine_path = self.target / 'usr' / 'share' / 'limine' hook_command = None if SysInfo.has_uefi(): if not efi_partition: raise ValueError('Could not detect efi partition') elif not efi_partition.mountpoint: raise ValueError('EFI partition is not mounted') info(f"Limine EFI partition: {efi_partition.dev_path}") try: efi_dir_path = self.target / efi_partition.mountpoint.relative_to('/') / 'EFI' / 'BOOT' efi_dir_path.mkdir(parents=True, exist_ok=True) for file in ('BOOTIA32.EFI', 'BOOTX64.EFI'): shutil.copy(limine_path / file, efi_dir_path) except Exception as err: raise DiskError(f'Failed to install Limine in {self.target}{efi_partition.mountpoint}: {err}') hook_command = f'/usr/bin/cp /usr/share/limine/BOOTIA32.EFI {efi_partition.mountpoint}/EFI/BOOT/' \ f' && /usr/bin/cp /usr/share/limine/BOOTX64.EFI {efi_partition.mountpoint}/EFI/BOOT/' else: parent_dev_path = disk.device_handler.get_parent_device_path(boot_partition.safe_dev_path) if unique_path := disk.device_handler.get_unique_path_for_device(parent_dev_path): parent_dev_path = unique_path try: # The `limine-bios.sys` file contains stage 3 code. shutil.copy(limine_path / 'limine-bios.sys', self.target / 'boot') # `limine bios-install` deploys the stage 1 and 2 to the disk. SysCommand(f'/usr/bin/arch-chroot {self.target} limine bios-install {parent_dev_path}', peek_output=True) except Exception as err: raise DiskError(f'Failed to install Limine on {parent_dev_path}: {err}') hook_command = f'/usr/bin/limine bios-install {parent_dev_path}' \ f' && /usr/bin/cp /usr/share/limine/limine-bios.sys /boot/' hook_contents = f'''[Trigger] Operation = Install Operation = Upgrade Type = Package Target = limine [Action] Description = Deploying Limine after upgrade... When = PostTransaction Exec = /bin/sh -c "{hook_command}" ''' hooks_dir = self.target / 'etc' / 'pacman.d' / 'hooks' hooks_dir.mkdir(parents=True, exist_ok=True) hook_path = hooks_dir / '99-limine.hook' hook_path.write_text(hook_contents) microcode = [] if ucode := self._get_microcode(): microcode = [f'MODULE_PATH=boot:///{ucode}'] kernel_params = ' '.join(self._get_kernel_params(root_partition)) config_contents = 'TIMEOUT=5\n' for kernel in self.kernels: for variant in ('', '-fallback'): entry = [ f'PROTOCOL=linux', f'KERNEL_PATH=boot:///vmlinuz-{kernel}', *microcode, f'MODULE_PATH=boot:///initramfs-{kernel}{variant}.img', f'CMDLINE={kernel_params}', ] config_contents += f'\n:Arch Linux ({kernel}{variant})\n' config_contents += '\n'.join([f' {it}' for it in entry]) + '\n' config_path = self.target / 'boot' / 'limine.cfg' config_path.write_text(config_contents) self.helper_flags['bootloader'] = "limine" def _add_efistub_bootloader( self, boot_partition: disk.PartitionModification, root_partition: disk.PartitionModification, uki_enabled: bool = False ): self.pacman.strap('efibootmgr') if not SysInfo.has_uefi(): raise HardwareIncompatibilityError # TODO: Ideally we would want to check if another config # points towards the same disk and/or partition. # And in which case we should do some clean up. if not uki_enabled: loader = '/vmlinuz-{kernel}' microcode = [] if ucode := self._get_microcode(): microcode.append(f'initrd=/{ucode}') else: debug('Archinstall will not add any ucode to firmware boot entry.') entries = ( *microcode, 'initrd=/initramfs-{kernel}.img', *self._get_kernel_params(root_partition) ) cmdline = [' '.join(entries)] else: loader = '/EFI/Linux/arch-{kernel}.efi' cmdline = [] parent_dev_path = disk.device_handler.get_parent_device_path(boot_partition.safe_dev_path) cmd_template = ( 'efibootmgr', '--create', '--disk', str(parent_dev_path), '--part', str(boot_partition.partn), '--label', 'Arch Linux ({kernel})', '--loader', loader, '--unicode', *cmdline, '--verbose' ) for kernel in self.kernels: # Setup the firmware entry cmd = [arg.format(kernel=kernel) for arg in cmd_template] SysCommand(cmd) self.helper_flags['bootloader'] = "efistub" def _config_uki( self, root_partition: disk.PartitionModification, efi_partition: Optional[disk.PartitionModification] ): if not efi_partition or not efi_partition.mountpoint: raise ValueError(f'Could not detect ESP at mountpoint {self.target}') # Set up kernel command line with open(self.target / 'etc/kernel/cmdline', 'w') as cmdline: kernel_parameters = self._get_kernel_params(root_partition) cmdline.write(' '.join(kernel_parameters) + '\n') ucode = self._get_microcode() esp = efi_partition.mountpoint diff_mountpoint = None if esp != Path('/efi'): diff_mountpoint = str(esp) image_re = re.compile('(.+_image="/([^"]+).+\n)') uki_re = re.compile('#((.+_uki=")/[^/]+(.+\n))') # Modify .preset files for kernel in self.kernels: preset = self.target / 'etc/mkinitcpio.d' / (kernel + '.preset') config = preset.read_text().splitlines(True) for index, line in enumerate(config): if not ucode and line.startswith('ALL_microcode='): config[index] = '#' + line # Avoid storing redundant image file elif m := image_re.match(line): image = self.target / m.group(2) image.unlink(missing_ok=True) config[index] = '#' + m.group(1) elif m := uki_re.match(line): if diff_mountpoint: config[index] = m.group(2) + diff_mountpoint + m.group(3) else: config[index] = m.group(1) elif line.startswith('#default_options='): config[index] = line.removeprefix('#') preset.write_text(''.join(config)) # Directory for the UKIs uki_dir = self.target / esp.relative_to(Path('/')) / 'EFI/Linux' uki_dir.mkdir(parents=True, exist_ok=True) # Build the UKIs if not self.mkinitcpio(['-P']): error(f"Error generating initramfs (continuing anyway)") def add_bootloader(self, bootloader: Bootloader, uki_enabled: bool = False): """ Adds a bootloader to the installation instance. Archinstall supports one of three types: * systemd-bootctl * grub * limine (beta) * efistub (beta) :param bootloader: Type of bootloader to be added """ for plugin in plugins.values(): if hasattr(plugin, 'on_add_bootloader'): # Allow plugins to override the boot-loader handling. # This allows for bot configuring and installing bootloaders. if plugin.on_add_bootloader(self): return True efi_partition = self._get_efi_partition() boot_partition = self._get_boot_partition() root_partition = self._get_root_partition() if boot_partition is None: raise ValueError(f'Could not detect boot at mountpoint {self.target}') if root_partition is None: raise ValueError(f'Could not detect root at mountpoint {self.target}') info(f'Adding bootloader {bootloader.value} to {boot_partition.dev_path}') if uki_enabled: self._config_uki(root_partition, efi_partition) match bootloader: case Bootloader.Systemd: self._add_systemd_bootloader(boot_partition, root_partition, efi_partition, uki_enabled) case Bootloader.Grub: self._add_grub_bootloader(boot_partition, root_partition, efi_partition, uki_enabled) case Bootloader.Efistub: self._add_efistub_bootloader(boot_partition, root_partition, uki_enabled) case Bootloader.Limine: self._add_limine_bootloader(boot_partition, efi_partition, root_partition) def add_additional_packages(self, packages: Union[str, List[str]]) -> bool: return self.pacman.strap(packages) def _enable_users(self, service: str, users: List[User]): for user in users: self.arch_chroot(f'systemctl enable --user {service}', run_as=user.username) def enable_sudo(self, entity: str, group :bool = False): info(f'Enabling sudo permissions for {entity}') sudoers_dir = f"{self.target}/etc/sudoers.d" # Creates directory if not exists if not (sudoers_path := Path(sudoers_dir)).exists(): sudoers_path.mkdir(parents=True) # Guarantees sudoer confs directory recommended perms os.chmod(sudoers_dir, 0o440) # Appends a reference to the sudoers file, because if we are here sudoers.d did not exist yet with open(f'{self.target}/etc/sudoers', 'a') as sudoers: sudoers.write('@includedir /etc/sudoers.d\n') # We count how many files are there already so we know which number to prefix the file with num_of_rules_already = len(os.listdir(sudoers_dir)) file_num_str = "{:02d}".format(num_of_rules_already) # We want 00_user1, 01_user2, etc # Guarantees that entity str does not contain invalid characters for a linux file name: # \ / : * ? " < > | safe_entity_file_name = re.sub(r'(\\|\/|:|\*|\?|"|<|>|\|)', '', entity) rule_file_name = f"{sudoers_dir}/{file_num_str}_{safe_entity_file_name}" with open(rule_file_name, 'a') as sudoers: sudoers.write(f'{"%" if group else ""}{entity} ALL=(ALL) ALL\n') # Guarantees sudoer conf file recommended perms os.chmod(Path(rule_file_name), 0o440) def create_users(self, users: Union[User, List[User]]): if not isinstance(users, list): users = [users] for user in users: self.user_create(user.username, user.password, user.groups, user.sudo) def user_create(self, user :str, password :Optional[str] = None, groups :Optional[List[str]] = None, sudo :bool = False) -> None: if groups is None: groups = [] # This plugin hook allows for the plugin to handle the creation of the user. # Password and Group management is still handled by user_create() handled_by_plugin = False for plugin in plugins.values(): if hasattr(plugin, 'on_user_create'): if result := plugin.on_user_create(self, user): handled_by_plugin = result if not handled_by_plugin: info(f'Creating user {user}') try: SysCommand(f'/usr/bin/arch-chroot {self.target} useradd -m -G wheel {user}') 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 if password: self.user_set_pw(user, password) if groups: for group in groups: SysCommand(f'/usr/bin/arch-chroot {self.target} gpasswd -a {user} {group}') if sudo and self.enable_sudo(user): self.helper_flags['user'] = True def user_set_pw(self, user :str, password :str) -> bool: info(f'Setting password for {user}') if user == 'root': # This means the root account isn't locked/disabled with * in /etc/passwd self.helper_flags['user'] = True combo = f'{user}:{password}' echo = shlex.join(['echo', combo]) sh = shlex.join(['sh', '-c', echo]) try: SysCommand(f"/usr/bin/arch-chroot {self.target} " + sh[:-1] + " | chpasswd'") return True except SysCallError: return False def user_set_shell(self, user :str, shell :str) -> bool: info(f'Setting shell for {user} to {shell}') try: SysCommand(f"/usr/bin/arch-chroot {self.target} sh -c \"chsh -s {shell} {user}\"") return True except SysCallError: return False def chown(self, owner :str, path :str, options :List[str] = []) -> bool: cleaned_path = path.replace('\'', '\\\'') try: SysCommand(f"/usr/bin/arch-chroot {self.target} sh -c 'chown {' '.join(options)} {owner} {cleaned_path}'") return True except SysCallError: return False def set_keyboard_language(self, language: str) -> bool: info(f"Setting keyboard language to {language}") if len(language.strip()): if not verify_keyboard_layout(language): error(f"Invalid keyboard language specified: {language}") return False # In accordance with https://github.com/archlinux/archinstall/issues/107#issuecomment-841701968 # Setting an empty keymap first, allows the subsequent call to set layout for both console and x11. from .boot import Boot with Boot(self) as session: os.system('/usr/bin/systemd-run --machine=archinstall --pty localectl set-keymap ""') try: session.SysCommand(["localectl", "set-keymap", language]) except SysCallError as err: raise ServiceException(f"Unable to set locale '{language}' for console: {err}") info(f"Keyboard language for this installation is now set to: {language}") else: info('Keyboard language was not changed from default (no language specified)') return True def set_x11_keyboard_language(self, language: str) -> bool: """ A fallback function to set x11 layout specifically and separately from console layout. This isn't strictly necessary since .set_keyboard_language() does this as well. """ info(f"Setting x11 keyboard language to {language}") if len(language.strip()): if not verify_x11_keyboard_layout(language): error(f"Invalid x11-keyboard language specified: {language}") return False from .boot import Boot with Boot(self) as session: session.SysCommand(["localectl", "set-x11-keymap", '""']) try: session.SysCommand(["localectl", "set-x11-keymap", language]) except SysCallError as err: raise ServiceException(f"Unable to set locale '{language}' for X11: {err}") else: info(f'X11-Keyboard language was not changed from default (no language specified)') return True def _service_started(self, service_name: str) -> Optional[str]: if os.path.splitext(service_name)[1] not in ('.service', '.target', '.timer'): service_name += '.service' # Just to be safe last_execution_time = SysCommand( f"systemctl show --property=ActiveEnterTimestamp --no-pager {service_name}", environment_vars={'SYSTEMD_COLORS': '0'} ).decode().lstrip('ActiveEnterTimestamp=') if not last_execution_time: return None return last_execution_time def _service_state(self, service_name: str) -> str: if os.path.splitext(service_name)[1] not in ('.service', '.target', '.timer'): service_name += '.service' # Just to be safe return SysCommand( f'systemctl show --no-pager -p SubState --value {service_name}', environment_vars={'SYSTEMD_COLORS': '0'} ).decode()