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 MirrorConfiguration
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/<log path> 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 remove_mod(self, mod: str):
		if mod in self._modules:
			self._modules.remove(mod)

	def append_mod(self, mod: str):
		if mod not in self._modules:
			self._modules.append(mod)

	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.
		"""

		if not storage['arguments'].get('skip_ntp', False):
			info(_('Waiting for time sync (timedatectl show) to complete.'))

			_started_wait = time.time()
			_notified = False
			while True:
				if not _notified and time.time() - _started_wait > 5:
					_notified = True
					warn(
						_("Time synchronization not completing, while you wait - check the docs for workarounds: https://archinstall.readthedocs.io/"))

				time_val = SysCommand('timedatectl show --property=NTPSynchronized --value').decode()
				if time_val and time_val.strip() == 'yes':
					break
				time.sleep(1)
		else:
			info(
				_('Skipping waiting for automatic time sync (this can cause issues if time is out of sync during installation)'))

		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, on_target: bool = False):
		"""
		Set the mirror configuration for the installation.

		:param mirror_config: The mirror configuration to use.
		:type mirror_config: MirrorConfiguration

		:on_target: Whether to set the mirrors on the target system or the live system.
		:param on_target: bool
		"""
		debug('Setting mirrors')

		for plugin in plugins.values():
			if hasattr(plugin, 'on_mirrors'):
				if result := plugin.on_mirrors(mirror_config):
					mirror_config = result

		if on_target:
			local_pacman_conf = Path(f'{self.target}/etc/pacman.conf')
			local_mirrorlist_conf = Path(f'{self.target}/etc/pacman.d/mirrorlist')
		else:
			local_pacman_conf = Path('/etc/pacman.conf')
			local_mirrorlist_conf = Path('/etc/pacman.d/mirrorlist')

		mirrorlist_config = mirror_config.mirrorlist_config()
		pacman_config = mirror_config.pacman_config()

		if pacman_config:
			debug(f'Pacman config: {pacman_config}')

			with local_pacman_conf.open('a') as fp:
				fp.write(pacman_config)

		if mirrorlist_config:
			debug(f'Mirrorlist: {mirrorlist_config}')

			with local_mirrorlist_conf.open('a') as fp:
				fp.write(mirrorlist_config)

	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)
		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('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]
	):
		self.pacman.strap('grub')  # no need?

		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()

		diff_mountpoint = None

		if efi_partition.mountpoint != Path('/efi'):
			diff_mountpoint = str(efi_partition.mountpoint)

		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 / efi_partition.relative_mountpoint / 'EFI/Linux'
		uki_dir.mkdir(parents=True, exist_ok=True)

		# Build the UKIs
		if not self.mkinitcpio(['-P']):
			error('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)
			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()