We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Logic error in `encrypt()` loop
The loop in `Luks2.encrypt()` contains a logic error.

https://github.com/archlinux/archinstall/blob/c9755b35e7364d208d836ec051d365354be6dd63/archinstall/lib/luks.py#L93

If `DISK_RETRY_ATTEMPTS` is `0` then an initial attempt will not occur and no code in the loop will be executed.

The maximum number of retries that can occur is erroneously one less than indicated by `DISK_RETRY_ATTEMPTS`; the initial attempt is when `retry_attempt` is `0` and all others after it are retries.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `archinstall/lib/luks.py`
Content:
```
1 from __future__ import annotations
2 
3 import shlex
4 import time
5 from dataclasses import dataclass
6 from pathlib import Path
7 from typing import Optional, List
8 
9 from . import disk
10 from .general import SysCommand, generate_password, SysCommandWorker
11 from .output import info, debug
12 from .exceptions import SysCallError, DiskError
13 from .storage import storage
14 
15 
16 @dataclass
17 class Luks2:
18 	luks_dev_path: Path
19 	mapper_name: Optional[str] = None
20 	password: Optional[str] = None
21 	key_file: Optional[Path] = None
22 	auto_unmount: bool = False
23 
24 	# will be set internally after unlocking the device
25 	_mapper_dev: Optional[Path] = None
26 
27 	@property
28 	def mapper_dev(self) -> Optional[Path]:
29 		if self.mapper_name:
30 			return Path(f'/dev/mapper/{self.mapper_name}')
31 		return None
32 
33 	def __post_init__(self):
34 		if self.luks_dev_path is None:
35 			raise ValueError('Partition must have a path set')
36 
37 	def __enter__(self):
38 		self.unlock(self.key_file)
39 
40 	def __exit__(self, *args: str, **kwargs: str):
41 		if self.auto_unmount:
42 			self.lock()
43 
44 	def _default_key_file(self) -> Path:
45 		return Path(f'/tmp/{self.luks_dev_path.name}.disk_pw')
46 
47 	def _password_bytes(self) -> bytes:
48 		if not self.password:
49 			raise ValueError('Password for luks2 device was not specified')
50 
51 		if isinstance(self.password, bytes):
52 			return self.password
53 		else:
54 			return bytes(self.password, 'UTF-8')
55 
56 	def encrypt(
57 		self,
58 		key_size: int = 512,
59 		hash_type: str = 'sha512',
60 		iter_time: int = 10000,
61 		key_file: Optional[Path] = None
62 	) -> Path:
63 		info(f'Luks2 encrypting: {self.luks_dev_path}')
64 
65 		byte_password = self._password_bytes()
66 
67 		if not key_file:
68 			if self.key_file:
69 				key_file = self.key_file
70 			else:
71 				key_file = self._default_key_file()
72 
73 				with open(key_file, 'wb') as fh:
74 					fh.write(byte_password)
75 
76 		cryptsetup_args = shlex.join([
77 			'/usr/bin/cryptsetup',
78 			'--batch-mode',
79 			'--verbose',
80 			'--type', 'luks2',
81 			'--pbkdf', 'argon2id',
82 			'--hash', hash_type,
83 			'--key-size', str(key_size),
84 			'--iter-time', str(iter_time),
85 			'--key-file', str(key_file),
86 			'--use-urandom',
87 			'luksFormat', str(self.luks_dev_path),
88 		])
89 
90 		# Retry formatting the volume because archinstall can some times be too quick
91 		# which generates a "Device /dev/sdX does not exist or access denied." between
92 		# setting up partitions and us trying to encrypt it.
93 		for retry_attempt in range(storage['DISK_RETRY_ATTEMPTS']):
94 			try:
95 				SysCommand(cryptsetup_args)
96 				break
97 			except SysCallError as err:
98 				time.sleep(storage['DISK_TIMEOUTS'])
99 
100 				if retry_attempt != storage['DISK_RETRY_ATTEMPTS'] - 1:
101 					continue
102 
103 				if err.exit_code == 1:
104 					info(f'luks2 partition currently in use: {self.luks_dev_path}')
105 					info('Attempting to unmount, crypt-close and trying encryption again')
106 
107 					self.lock()
108 					# Then try again to set up the crypt-device
109 					SysCommand(cryptsetup_args)
110 				else:
111 					raise DiskError(f'Could not encrypt volume "{self.luks_dev_path}": {err}')
112 
113 		return key_file
114 
115 	def _get_luks_uuid(self) -> str:
116 		command = f'/usr/bin/cryptsetup luksUUID {self.luks_dev_path}'
117 
118 		try:
119 			return SysCommand(command).decode()
120 		except SysCallError as err:
121 			info(f'Unable to get UUID for Luks device: {self.luks_dev_path}')
122 			raise err
123 
124 	def is_unlocked(self) -> bool:
125 		return self.mapper_name is not None and Path(f'/dev/mapper/{self.mapper_name}').exists()
126 
127 	def unlock(self, key_file: Optional[Path] = None):
128 		"""
129 		Unlocks the luks device, an optional key file location for unlocking can be specified,
130 		otherwise a default location for the key file will be used.
131 
132 		:param key_file: An alternative key file
133 		:type key_file: Path
134 		"""
135 		debug(f'Unlocking luks2 device: {self.luks_dev_path}')
136 
137 		if not self.mapper_name:
138 			raise ValueError('mapper name missing')
139 
140 		byte_password = self._password_bytes()
141 
142 		if not key_file:
143 			if self.key_file:
144 				key_file = self.key_file
145 			else:
146 				key_file = self._default_key_file()
147 
148 				with open(key_file, 'wb') as fh:
149 					fh.write(byte_password)
150 
151 		wait_timer = time.time()
152 		while Path(self.luks_dev_path).exists() is False and time.time() - wait_timer < 10:
153 			time.sleep(0.025)
154 
155 		SysCommand(f'/usr/bin/cryptsetup open {self.luks_dev_path} {self.mapper_name} --key-file {key_file} --type luks2')
156 
157 		if not self.mapper_dev or not self.mapper_dev.is_symlink():
158 			raise DiskError(f'Failed to open luks2 device: {self.luks_dev_path}')
159 
160 	def lock(self):
161 		disk.device_handler.umount(self.luks_dev_path)
162 
163 		# Get crypt-information about the device by doing a reverse lookup starting with the partition path
164 		# For instance: /dev/sda
165 		disk.device_handler.partprobe(self.luks_dev_path)
166 		lsblk_info = disk.get_lsblk_info(self.luks_dev_path)
167 
168 		# For each child (sub-partition/sub-device)
169 		for child in lsblk_info.children:
170 			# Unmount the child location
171 			for mountpoint in child.mountpoints:
172 				debug(f'Unmounting {mountpoint}')
173 				disk.device_handler.umount(mountpoint, recursive=True)
174 
175 			# And close it if possible.
176 			debug(f"Closing crypt device {child.name}")
177 			SysCommand(f"cryptsetup close {child.name}")
178 
179 		self._mapper_dev = None
180 
181 	def create_keyfile(self, target_path: Path, override: bool = False):
182 		"""
183 		Routine to create keyfiles, so it can be moved elsewhere
184 		"""
185 		if self.mapper_name is None:
186 			raise ValueError('Mapper name must be provided')
187 
188 		# Once we store the key as ../xyzloop.key systemd-cryptsetup can
189 		# automatically load this key if we name the device to "xyzloop"
190 		kf_path = Path(f'/etc/cryptsetup-keys.d/{self.mapper_name}.key')
191 		key_file = target_path / kf_path.relative_to(kf_path.root)
192 		crypttab_path = target_path / 'etc/crypttab'
193 
194 		if key_file.exists():
195 			if not override:
196 				info(f'Key file {key_file} already exists, keeping existing')
197 				return
198 			else:
199 				info(f'Key file {key_file} already exists, overriding')
200 
201 		key_file.parent.mkdir(parents=True, exist_ok=True)
202 
203 		with open(key_file, "w") as keyfile:
204 			keyfile.write(generate_password(length=512))
205 
206 		key_file.chmod(0o400)
207 
208 		self._add_key(key_file)
209 		self._crypttab(crypttab_path, kf_path, options=["luks", "key-slot=1"])
210 
211 	def _add_key(self, key_file: Path):
212 		info(f'Adding additional key-file {key_file}')
213 
214 		command = f'/usr/bin/cryptsetup -q -v luksAddKey {self.luks_dev_path} {key_file}'
215 		worker = SysCommandWorker(command, environment_vars={'LC_ALL': 'C'})
216 		pw_injected = False
217 
218 		while worker.is_alive():
219 			if b'Enter any existing passphrase' in worker and pw_injected is False:
220 				worker.write(self._password_bytes())
221 				pw_injected = True
222 
223 		if worker.exit_code != 0:
224 			raise DiskError(f'Could not add encryption key {key_file} to {self.luks_dev_path}: {worker.decode()}')
225 
226 	def _crypttab(
227 		self,
228 		crypttab_path: Path,
229 		key_file: Path,
230 		options: List[str]
231 	) -> None:
232 		info(f'Adding crypttab entry for key {key_file}')
233 
234 		with open(crypttab_path, 'a') as crypttab:
235 			opt = ','.join(options)
236 			uuid = self._get_luks_uuid()
237 			row = f"{self.mapper_name} UUID={uuid} {key_file} {opt}\n"
238 			crypttab.write(row)
239
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/archinstall/lib/luks.py b/archinstall/lib/luks.py
--- a/archinstall/lib/luks.py
+++ b/archinstall/lib/luks.py
@@ -90,14 +90,14 @@
 		# Retry formatting the volume because archinstall can some times be too quick
 		# which generates a "Device /dev/sdX does not exist or access denied." between
 		# setting up partitions and us trying to encrypt it.
-		for retry_attempt in range(storage['DISK_RETRY_ATTEMPTS']):
+		for retry_attempt in range(storage['DISK_RETRY_ATTEMPTS'] + 1):
 			try:
 				SysCommand(cryptsetup_args)
 				break
 			except SysCallError as err:
 				time.sleep(storage['DISK_TIMEOUTS'])
 
-				if retry_attempt != storage['DISK_RETRY_ATTEMPTS'] - 1:
+				if retry_attempt != storage['DISK_RETRY_ATTEMPTS']:
 					continue
 
 				if err.exit_code == 1:

{"golden_diff": "diff --git a/archinstall/lib/luks.py b/archinstall/lib/luks.py\n--- a/archinstall/lib/luks.py\n+++ b/archinstall/lib/luks.py\n@@ -90,14 +90,14 @@\n \t\t# Retry formatting the volume because archinstall can some times be too quick\n \t\t# which generates a \"Device /dev/sdX does not exist or access denied.\" between\n \t\t# setting up partitions and us trying to encrypt it.\n-\t\tfor retry_attempt in range(storage['DISK_RETRY_ATTEMPTS']):\n+\t\tfor retry_attempt in range(storage['DISK_RETRY_ATTEMPTS'] + 1):\n \t\t\ttry:\n \t\t\t\tSysCommand(cryptsetup_args)\n \t\t\t\tbreak\n \t\t\texcept SysCallError as err:\n \t\t\t\ttime.sleep(storage['DISK_TIMEOUTS'])\n \n-\t\t\t\tif retry_attempt != storage['DISK_RETRY_ATTEMPTS'] - 1:\n+\t\t\t\tif retry_attempt != storage['DISK_RETRY_ATTEMPTS']:\n \t\t\t\t\tcontinue\n \n \t\t\t\tif err.exit_code == 1:\n", "issue": "Logic error in `encrypt()` loop\nThe loop in `Luks2.encrypt()` contains a logic error.\r\n\r\nhttps://github.com/archlinux/archinstall/blob/c9755b35e7364d208d836ec051d365354be6dd63/archinstall/lib/luks.py#L93\r\n\r\nIf `DISK_RETRY_ATTEMPTS` is `0` then an initial attempt will not occur and no code in the loop will be executed.\r\n\r\nThe maximum number of retries that can occur is erroneously one less than indicated by `DISK_RETRY_ATTEMPTS`; the initial attempt is when `retry_attempt` is `0` and all others after it are retries.\n", "before_files": [{"content": "from __future__ import annotations\n\nimport shlex\nimport time\nfrom dataclasses import dataclass\nfrom pathlib import Path\nfrom typing import Optional, List\n\nfrom . import disk\nfrom .general import SysCommand, generate_password, SysCommandWorker\nfrom .output import info, debug\nfrom .exceptions import SysCallError, DiskError\nfrom .storage import storage\n\n\n@dataclass\nclass Luks2:\n\tluks_dev_path: Path\n\tmapper_name: Optional[str] = None\n\tpassword: Optional[str] = None\n\tkey_file: Optional[Path] = None\n\tauto_unmount: bool = False\n\n\t# will be set internally after unlocking the device\n\t_mapper_dev: Optional[Path] = None\n\n\t@property\n\tdef mapper_dev(self) -> Optional[Path]:\n\t\tif self.mapper_name:\n\t\t\treturn Path(f'/dev/mapper/{self.mapper_name}')\n\t\treturn None\n\n\tdef __post_init__(self):\n\t\tif self.luks_dev_path is None:\n\t\t\traise ValueError('Partition must have a path set')\n\n\tdef __enter__(self):\n\t\tself.unlock(self.key_file)\n\n\tdef __exit__(self, *args: str, **kwargs: str):\n\t\tif self.auto_unmount:\n\t\t\tself.lock()\n\n\tdef _default_key_file(self) -> Path:\n\t\treturn Path(f'/tmp/{self.luks_dev_path.name}.disk_pw')\n\n\tdef _password_bytes(self) -> bytes:\n\t\tif not self.password:\n\t\t\traise ValueError('Password for luks2 device was not specified')\n\n\t\tif isinstance(self.password, bytes):\n\t\t\treturn self.password\n\t\telse:\n\t\t\treturn bytes(self.password, 'UTF-8')\n\n\tdef encrypt(\n\t\tself,\n\t\tkey_size: int = 512,\n\t\thash_type: str = 'sha512',\n\t\titer_time: int = 10000,\n\t\tkey_file: Optional[Path] = None\n\t) -> Path:\n\t\tinfo(f'Luks2 encrypting: {self.luks_dev_path}')\n\n\t\tbyte_password = self._password_bytes()\n\n\t\tif not key_file:\n\t\t\tif self.key_file:\n\t\t\t\tkey_file = self.key_file\n\t\t\telse:\n\t\t\t\tkey_file = self._default_key_file()\n\n\t\t\t\twith open(key_file, 'wb') as fh:\n\t\t\t\t\tfh.write(byte_password)\n\n\t\tcryptsetup_args = shlex.join([\n\t\t\t'/usr/bin/cryptsetup',\n\t\t\t'--batch-mode',\n\t\t\t'--verbose',\n\t\t\t'--type', 'luks2',\n\t\t\t'--pbkdf', 'argon2id',\n\t\t\t'--hash', hash_type,\n\t\t\t'--key-size', str(key_size),\n\t\t\t'--iter-time', str(iter_time),\n\t\t\t'--key-file', str(key_file),\n\t\t\t'--use-urandom',\n\t\t\t'luksFormat', str(self.luks_dev_path),\n\t\t])\n\n\t\t# Retry formatting the volume because archinstall can some times be too quick\n\t\t# which generates a \"Device /dev/sdX does not exist or access denied.\" between\n\t\t# setting up partitions and us trying to encrypt it.\n\t\tfor retry_attempt in range(storage['DISK_RETRY_ATTEMPTS']):\n\t\t\ttry:\n\t\t\t\tSysCommand(cryptsetup_args)\n\t\t\t\tbreak\n\t\t\texcept SysCallError as err:\n\t\t\t\ttime.sleep(storage['DISK_TIMEOUTS'])\n\n\t\t\t\tif retry_attempt != storage['DISK_RETRY_ATTEMPTS'] - 1:\n\t\t\t\t\tcontinue\n\n\t\t\t\tif err.exit_code == 1:\n\t\t\t\t\tinfo(f'luks2 partition currently in use: {self.luks_dev_path}')\n\t\t\t\t\tinfo('Attempting to unmount, crypt-close and trying encryption again')\n\n\t\t\t\t\tself.lock()\n\t\t\t\t\t# Then try again to set up the crypt-device\n\t\t\t\t\tSysCommand(cryptsetup_args)\n\t\t\t\telse:\n\t\t\t\t\traise DiskError(f'Could not encrypt volume \"{self.luks_dev_path}\": {err}')\n\n\t\treturn key_file\n\n\tdef _get_luks_uuid(self) -> str:\n\t\tcommand = f'/usr/bin/cryptsetup luksUUID {self.luks_dev_path}'\n\n\t\ttry:\n\t\t\treturn SysCommand(command).decode()\n\t\texcept SysCallError as err:\n\t\t\tinfo(f'Unable to get UUID for Luks device: {self.luks_dev_path}')\n\t\t\traise err\n\n\tdef is_unlocked(self) -> bool:\n\t\treturn self.mapper_name is not None and Path(f'/dev/mapper/{self.mapper_name}').exists()\n\n\tdef unlock(self, key_file: Optional[Path] = None):\n\t\t\"\"\"\n\t\tUnlocks the luks device, an optional key file location for unlocking can be specified,\n\t\totherwise a default location for the key file will be used.\n\n\t\t:param key_file: An alternative key file\n\t\t:type key_file: Path\n\t\t\"\"\"\n\t\tdebug(f'Unlocking luks2 device: {self.luks_dev_path}')\n\n\t\tif not self.mapper_name:\n\t\t\traise ValueError('mapper name missing')\n\n\t\tbyte_password = self._password_bytes()\n\n\t\tif not key_file:\n\t\t\tif self.key_file:\n\t\t\t\tkey_file = self.key_file\n\t\t\telse:\n\t\t\t\tkey_file = self._default_key_file()\n\n\t\t\t\twith open(key_file, 'wb') as fh:\n\t\t\t\t\tfh.write(byte_password)\n\n\t\twait_timer = time.time()\n\t\twhile Path(self.luks_dev_path).exists() is False and time.time() - wait_timer < 10:\n\t\t\ttime.sleep(0.025)\n\n\t\tSysCommand(f'/usr/bin/cryptsetup open {self.luks_dev_path} {self.mapper_name} --key-file {key_file} --type luks2')\n\n\t\tif not self.mapper_dev or not self.mapper_dev.is_symlink():\n\t\t\traise DiskError(f'Failed to open luks2 device: {self.luks_dev_path}')\n\n\tdef lock(self):\n\t\tdisk.device_handler.umount(self.luks_dev_path)\n\n\t\t# Get crypt-information about the device by doing a reverse lookup starting with the partition path\n\t\t# For instance: /dev/sda\n\t\tdisk.device_handler.partprobe(self.luks_dev_path)\n\t\tlsblk_info = disk.get_lsblk_info(self.luks_dev_path)\n\n\t\t# For each child (sub-partition/sub-device)\n\t\tfor child in lsblk_info.children:\n\t\t\t# Unmount the child location\n\t\t\tfor mountpoint in child.mountpoints:\n\t\t\t\tdebug(f'Unmounting {mountpoint}')\n\t\t\t\tdisk.device_handler.umount(mountpoint, recursive=True)\n\n\t\t\t# And close it if possible.\n\t\t\tdebug(f\"Closing crypt device {child.name}\")\n\t\t\tSysCommand(f\"cryptsetup close {child.name}\")\n\n\t\tself._mapper_dev = None\n\n\tdef create_keyfile(self, target_path: Path, override: bool = False):\n\t\t\"\"\"\n\t\tRoutine to create keyfiles, so it can be moved elsewhere\n\t\t\"\"\"\n\t\tif self.mapper_name is None:\n\t\t\traise ValueError('Mapper name must be provided')\n\n\t\t# Once we store the key as ../xyzloop.key systemd-cryptsetup can\n\t\t# automatically load this key if we name the device to \"xyzloop\"\n\t\tkf_path = Path(f'/etc/cryptsetup-keys.d/{self.mapper_name}.key')\n\t\tkey_file = target_path / kf_path.relative_to(kf_path.root)\n\t\tcrypttab_path = target_path / 'etc/crypttab'\n\n\t\tif key_file.exists():\n\t\t\tif not override:\n\t\t\t\tinfo(f'Key file {key_file} already exists, keeping existing')\n\t\t\t\treturn\n\t\t\telse:\n\t\t\t\tinfo(f'Key file {key_file} already exists, overriding')\n\n\t\tkey_file.parent.mkdir(parents=True, exist_ok=True)\n\n\t\twith open(key_file, \"w\") as keyfile:\n\t\t\tkeyfile.write(generate_password(length=512))\n\n\t\tkey_file.chmod(0o400)\n\n\t\tself._add_key(key_file)\n\t\tself._crypttab(crypttab_path, kf_path, options=[\"luks\", \"key-slot=1\"])\n\n\tdef _add_key(self, key_file: Path):\n\t\tinfo(f'Adding additional key-file {key_file}')\n\n\t\tcommand = f'/usr/bin/cryptsetup -q -v luksAddKey {self.luks_dev_path} {key_file}'\n\t\tworker = SysCommandWorker(command, environment_vars={'LC_ALL': 'C'})\n\t\tpw_injected = False\n\n\t\twhile worker.is_alive():\n\t\t\tif b'Enter any existing passphrase' in worker and pw_injected is False:\n\t\t\t\tworker.write(self._password_bytes())\n\t\t\t\tpw_injected = True\n\n\t\tif worker.exit_code != 0:\n\t\t\traise DiskError(f'Could not add encryption key {key_file} to {self.luks_dev_path}: {worker.decode()}')\n\n\tdef _crypttab(\n\t\tself,\n\t\tcrypttab_path: Path,\n\t\tkey_file: Path,\n\t\toptions: List[str]\n\t) -> None:\n\t\tinfo(f'Adding crypttab entry for key {key_file}')\n\n\t\twith open(crypttab_path, 'a') as crypttab:\n\t\t\topt = ','.join(options)\n\t\t\tuuid = self._get_luks_uuid()\n\t\t\trow = f\"{self.mapper_name} UUID={uuid} {key_file} {opt}\\n\"\n\t\t\tcrypttab.write(row)\n", "path": "archinstall/lib/luks.py"}], "after_files": [{"content": "from __future__ import annotations\n\nimport shlex\nimport time\nfrom dataclasses import dataclass\nfrom pathlib import Path\nfrom typing import Optional, List\n\nfrom . import disk\nfrom .general import SysCommand, generate_password, SysCommandWorker\nfrom .output import info, debug\nfrom .exceptions import SysCallError, DiskError\nfrom .storage import storage\n\n\n@dataclass\nclass Luks2:\n\tluks_dev_path: Path\n\tmapper_name: Optional[str] = None\n\tpassword: Optional[str] = None\n\tkey_file: Optional[Path] = None\n\tauto_unmount: bool = False\n\n\t# will be set internally after unlocking the device\n\t_mapper_dev: Optional[Path] = None\n\n\t@property\n\tdef mapper_dev(self) -> Optional[Path]:\n\t\tif self.mapper_name:\n\t\t\treturn Path(f'/dev/mapper/{self.mapper_name}')\n\t\treturn None\n\n\tdef __post_init__(self):\n\t\tif self.luks_dev_path is None:\n\t\t\traise ValueError('Partition must have a path set')\n\n\tdef __enter__(self):\n\t\tself.unlock(self.key_file)\n\n\tdef __exit__(self, *args: str, **kwargs: str):\n\t\tif self.auto_unmount:\n\t\t\tself.lock()\n\n\tdef _default_key_file(self) -> Path:\n\t\treturn Path(f'/tmp/{self.luks_dev_path.name}.disk_pw')\n\n\tdef _password_bytes(self) -> bytes:\n\t\tif not self.password:\n\t\t\traise ValueError('Password for luks2 device was not specified')\n\n\t\tif isinstance(self.password, bytes):\n\t\t\treturn self.password\n\t\telse:\n\t\t\treturn bytes(self.password, 'UTF-8')\n\n\tdef encrypt(\n\t\tself,\n\t\tkey_size: int = 512,\n\t\thash_type: str = 'sha512',\n\t\titer_time: int = 10000,\n\t\tkey_file: Optional[Path] = None\n\t) -> Path:\n\t\tinfo(f'Luks2 encrypting: {self.luks_dev_path}')\n\n\t\tbyte_password = self._password_bytes()\n\n\t\tif not key_file:\n\t\t\tif self.key_file:\n\t\t\t\tkey_file = self.key_file\n\t\t\telse:\n\t\t\t\tkey_file = self._default_key_file()\n\n\t\t\t\twith open(key_file, 'wb') as fh:\n\t\t\t\t\tfh.write(byte_password)\n\n\t\tcryptsetup_args = shlex.join([\n\t\t\t'/usr/bin/cryptsetup',\n\t\t\t'--batch-mode',\n\t\t\t'--verbose',\n\t\t\t'--type', 'luks2',\n\t\t\t'--pbkdf', 'argon2id',\n\t\t\t'--hash', hash_type,\n\t\t\t'--key-size', str(key_size),\n\t\t\t'--iter-time', str(iter_time),\n\t\t\t'--key-file', str(key_file),\n\t\t\t'--use-urandom',\n\t\t\t'luksFormat', str(self.luks_dev_path),\n\t\t])\n\n\t\t# Retry formatting the volume because archinstall can some times be too quick\n\t\t# which generates a \"Device /dev/sdX does not exist or access denied.\" between\n\t\t# setting up partitions and us trying to encrypt it.\n\t\tfor retry_attempt in range(storage['DISK_RETRY_ATTEMPTS'] + 1):\n\t\t\ttry:\n\t\t\t\tSysCommand(cryptsetup_args)\n\t\t\t\tbreak\n\t\t\texcept SysCallError as err:\n\t\t\t\ttime.sleep(storage['DISK_TIMEOUTS'])\n\n\t\t\t\tif retry_attempt != storage['DISK_RETRY_ATTEMPTS']:\n\t\t\t\t\tcontinue\n\n\t\t\t\tif err.exit_code == 1:\n\t\t\t\t\tinfo(f'luks2 partition currently in use: {self.luks_dev_path}')\n\t\t\t\t\tinfo('Attempting to unmount, crypt-close and trying encryption again')\n\n\t\t\t\t\tself.lock()\n\t\t\t\t\t# Then try again to set up the crypt-device\n\t\t\t\t\tSysCommand(cryptsetup_args)\n\t\t\t\telse:\n\t\t\t\t\traise DiskError(f'Could not encrypt volume \"{self.luks_dev_path}\": {err}')\n\n\t\treturn key_file\n\n\tdef _get_luks_uuid(self) -> str:\n\t\tcommand = f'/usr/bin/cryptsetup luksUUID {self.luks_dev_path}'\n\n\t\ttry:\n\t\t\treturn SysCommand(command).decode()\n\t\texcept SysCallError as err:\n\t\t\tinfo(f'Unable to get UUID for Luks device: {self.luks_dev_path}')\n\t\t\traise err\n\n\tdef is_unlocked(self) -> bool:\n\t\treturn self.mapper_name is not None and Path(f'/dev/mapper/{self.mapper_name}').exists()\n\n\tdef unlock(self, key_file: Optional[Path] = None):\n\t\t\"\"\"\n\t\tUnlocks the luks device, an optional key file location for unlocking can be specified,\n\t\totherwise a default location for the key file will be used.\n\n\t\t:param key_file: An alternative key file\n\t\t:type key_file: Path\n\t\t\"\"\"\n\t\tdebug(f'Unlocking luks2 device: {self.luks_dev_path}')\n\n\t\tif not self.mapper_name:\n\t\t\traise ValueError('mapper name missing')\n\n\t\tbyte_password = self._password_bytes()\n\n\t\tif not key_file:\n\t\t\tif self.key_file:\n\t\t\t\tkey_file = self.key_file\n\t\t\telse:\n\t\t\t\tkey_file = self._default_key_file()\n\n\t\t\t\twith open(key_file, 'wb') as fh:\n\t\t\t\t\tfh.write(byte_password)\n\n\t\twait_timer = time.time()\n\t\twhile Path(self.luks_dev_path).exists() is False and time.time() - wait_timer < 10:\n\t\t\ttime.sleep(0.025)\n\n\t\tSysCommand(f'/usr/bin/cryptsetup open {self.luks_dev_path} {self.mapper_name} --key-file {key_file} --type luks2')\n\n\t\tif not self.mapper_dev or not self.mapper_dev.is_symlink():\n\t\t\traise DiskError(f'Failed to open luks2 device: {self.luks_dev_path}')\n\n\tdef lock(self):\n\t\tdisk.device_handler.umount(self.luks_dev_path)\n\n\t\t# Get crypt-information about the device by doing a reverse lookup starting with the partition path\n\t\t# For instance: /dev/sda\n\t\tdisk.device_handler.partprobe(self.luks_dev_path)\n\t\tlsblk_info = disk.get_lsblk_info(self.luks_dev_path)\n\n\t\t# For each child (sub-partition/sub-device)\n\t\tfor child in lsblk_info.children:\n\t\t\t# Unmount the child location\n\t\t\tfor mountpoint in child.mountpoints:\n\t\t\t\tdebug(f'Unmounting {mountpoint}')\n\t\t\t\tdisk.device_handler.umount(mountpoint, recursive=True)\n\n\t\t\t# And close it if possible.\n\t\t\tdebug(f\"Closing crypt device {child.name}\")\n\t\t\tSysCommand(f\"cryptsetup close {child.name}\")\n\n\t\tself._mapper_dev = None\n\n\tdef create_keyfile(self, target_path: Path, override: bool = False):\n\t\t\"\"\"\n\t\tRoutine to create keyfiles, so it can be moved elsewhere\n\t\t\"\"\"\n\t\tif self.mapper_name is None:\n\t\t\traise ValueError('Mapper name must be provided')\n\n\t\t# Once we store the key as ../xyzloop.key systemd-cryptsetup can\n\t\t# automatically load this key if we name the device to \"xyzloop\"\n\t\tkf_path = Path(f'/etc/cryptsetup-keys.d/{self.mapper_name}.key')\n\t\tkey_file = target_path / kf_path.relative_to(kf_path.root)\n\t\tcrypttab_path = target_path / 'etc/crypttab'\n\n\t\tif key_file.exists():\n\t\t\tif not override:\n\t\t\t\tinfo(f'Key file {key_file} already exists, keeping existing')\n\t\t\t\treturn\n\t\t\telse:\n\t\t\t\tinfo(f'Key file {key_file} already exists, overriding')\n\n\t\tkey_file.parent.mkdir(parents=True, exist_ok=True)\n\n\t\twith open(key_file, \"w\") as keyfile:\n\t\t\tkeyfile.write(generate_password(length=512))\n\n\t\tkey_file.chmod(0o400)\n\n\t\tself._add_key(key_file)\n\t\tself._crypttab(crypttab_path, kf_path, options=[\"luks\", \"key-slot=1\"])\n\n\tdef _add_key(self, key_file: Path):\n\t\tinfo(f'Adding additional key-file {key_file}')\n\n\t\tcommand = f'/usr/bin/cryptsetup -q -v luksAddKey {self.luks_dev_path} {key_file}'\n\t\tworker = SysCommandWorker(command, environment_vars={'LC_ALL': 'C'})\n\t\tpw_injected = False\n\n\t\twhile worker.is_alive():\n\t\t\tif b'Enter any existing passphrase' in worker and pw_injected is False:\n\t\t\t\tworker.write(self._password_bytes())\n\t\t\t\tpw_injected = True\n\n\t\tif worker.exit_code != 0:\n\t\t\traise DiskError(f'Could not add encryption key {key_file} to {self.luks_dev_path}: {worker.decode()}')\n\n\tdef _crypttab(\n\t\tself,\n\t\tcrypttab_path: Path,\n\t\tkey_file: Path,\n\t\toptions: List[str]\n\t) -> None:\n\t\tinfo(f'Adding crypttab entry for key {key_file}')\n\n\t\twith open(crypttab_path, 'a') as crypttab:\n\t\t\topt = ','.join(options)\n\t\t\tuuid = self._get_luks_uuid()\n\t\t\trow = f\"{self.mapper_name} UUID={uuid} {key_file} {opt}\\n\"\n\t\t\tcrypttab.write(row)\n", "path": "archinstall/lib/luks.py"}]}