pyrox.dev
1#! @python3@/bin/python3 -B
2import argparse
3import ctypes
4import datetime
5import errno
6import glob
7import os
8import os.path
9import re
10import shutil
11import subprocess
12import sys
13import warnings
14import json
15from typing import NamedTuple, Dict, List
16from dataclasses import dataclass
17
18# These values will be replaced with actual values during the package build
19EFI_SYS_MOUNT_POINT = "@efiSysMountPoint@"
20BOOT_MOUNT_POINT = "@bootMountPoint@"
21LOADER_CONF = f"{EFI_SYS_MOUNT_POINT}/loader/loader.conf" # Always stored on the ESP
22NIXOS_DIR = "@nixosDir@"
23TIMEOUT = "@timeout@"
24EDITOR = "@editor@" == "1"
25CONSOLE_MODE = "@consoleMode@"
26BOOTSPEC_TOOLS = "@bootspecTools@"
27DISTRO_NAME = "@distroName@"
28NIX = "@nix@"
29SYSTEMD = "@systemd@"
30CONFIGURATION_LIMIT = int("@configurationLimit@")
31CAN_TOUCH_EFI_VARIABLES = "@canTouchEfiVariables@"
32GRACEFUL = "@graceful@"
33COPY_EXTRA_FILES = "@copyExtraFiles@"
34CHECK_MOUNTPOINTS = "@checkMountpoints@"
35
36@dataclass
37class BootSpec:
38 init: str
39 initrd: str
40 kernel: str
41 kernelParams: List[str]
42 label: str
43 system: str
44 toplevel: str
45 specialisations: Dict[str, "BootSpec"]
46 sortKey: str
47 initrdSecrets: str | None = None
48
49
50libc = ctypes.CDLL("libc.so.6")
51
52class SystemIdentifier(NamedTuple):
53 profile: str | None
54 generation: int
55 specialisation: str | None
56
57
58def copy_if_not_exists(source: str, dest: str) -> None:
59 if not os.path.exists(dest):
60 shutil.copyfile(source, dest)
61
62
63def generation_dir(profile: str | None, generation: int) -> str:
64 if profile:
65 return "/nix/var/nix/profiles/system-profiles/%s-%d-link" % (profile, generation)
66 else:
67 return "/nix/var/nix/profiles/system-%d-link" % (generation)
68
69def system_dir(profile: str | None, generation: int, specialisation: str | None) -> str:
70 d = generation_dir(profile, generation)
71 if specialisation:
72 return os.path.join(d, "specialisation", specialisation)
73 else:
74 return d
75
76BOOT_ENTRY = """title {title}
77sort-key {sort_key}
78version Generation {generation} {description}
79linux {kernel}
80initrd {initrd}
81options {kernel_params}
82"""
83
84def generation_conf_filename(profile: str | None, generation: int, specialisation: str | None) -> str:
85 pieces = [
86 "nixos",
87 profile or None,
88 "generation",
89 str(generation),
90 f"specialisation-{specialisation}" if specialisation else None,
91 ]
92 return "-".join(p for p in pieces if p) + ".conf"
93
94
95def write_loader_conf(profile: str | None, generation: int, specialisation: str | None) -> None:
96 with open(f"{LOADER_CONF}.tmp", 'w') as f:
97 if TIMEOUT != "":
98 f.write(f"timeout {TIMEOUT}\n")
99 f.write("default %s\n" % generation_conf_filename(profile, generation, specialisation))
100 if not EDITOR:
101 f.write("editor 0\n")
102 f.write(f"console-mode {CONSOLE_MODE}\n")
103 f.flush()
104 os.fsync(f.fileno())
105 os.rename(f"{LOADER_CONF}.tmp", LOADER_CONF)
106
107
108def get_bootspec(profile: str | None, generation: int) -> BootSpec:
109 system_directory = system_dir(profile, generation, None)
110 boot_json_path = os.path.realpath("%s/%s" % (system_directory, "boot.json"))
111 if os.path.isfile(boot_json_path):
112 boot_json_f = open(boot_json_path, 'r')
113 bootspec_json = json.load(boot_json_f)
114 else:
115 boot_json_str = subprocess.check_output([
116 f"{BOOTSPEC_TOOLS}/bin/synthesize",
117 "--version",
118 "1",
119 system_directory,
120 "/dev/stdout"],
121 universal_newlines=True)
122 bootspec_json = json.loads(boot_json_str)
123 return bootspec_from_json(bootspec_json)
124
125def bootspec_from_json(bootspec_json: Dict) -> BootSpec:
126 specialisations = bootspec_json['org.nixos.specialisation.v1']
127 specialisations = {k: bootspec_from_json(v) for k, v in specialisations.items()}
128 systemdBootExtension = bootspec_json.get('org.nixos.systemd-boot', {})
129 sortKey = systemdBootExtension.get('sortKey', 'nixos')
130 return BootSpec(
131 **bootspec_json['org.nixos.bootspec.v1'],
132 specialisations=specialisations,
133 sortKey=sortKey
134 )
135
136
137def copy_from_file(file: str, dry_run: bool = False) -> str:
138 store_file_path = os.path.realpath(file)
139 suffix = os.path.basename(store_file_path)
140 store_dir = os.path.basename(os.path.dirname(store_file_path))
141 efi_file_path = f"{NIXOS_DIR}/{store_dir}-{suffix}.efi"
142 if not dry_run:
143 copy_if_not_exists(store_file_path, f"{BOOT_MOUNT_POINT}{efi_file_path}")
144 return efi_file_path
145
146def write_entry(profile: str | None, generation: int, specialisation: str | None,
147 machine_id: str, bootspec: BootSpec, current: bool) -> None:
148 if specialisation:
149 bootspec = bootspec.specialisations[specialisation]
150 kernel = copy_from_file(bootspec.kernel)
151 initrd = copy_from_file(bootspec.initrd)
152
153 title = "{name}{profile}{specialisation}".format(
154 name=DISTRO_NAME,
155 profile=" [" + profile + "]" if profile else "",
156 specialisation=" (%s)" % specialisation if specialisation else "")
157
158 try:
159 if bootspec.initrdSecrets is not None:
160 subprocess.check_call([bootspec.initrdSecrets, f"{BOOT_MOUNT_POINT}%s" % (initrd)])
161 except subprocess.CalledProcessError:
162 if current:
163 print("failed to create initrd secrets!", file=sys.stderr)
164 sys.exit(1)
165 else:
166 print("warning: failed to create initrd secrets "
167 f'for "{title} - Configuration {generation}", an older generation', file=sys.stderr)
168 print("note: this is normal after having removed "
169 "or renamed a file in `boot.initrd.secrets`", file=sys.stderr)
170 entry_file = f"{BOOT_MOUNT_POINT}/loader/entries/%s" % (
171 generation_conf_filename(profile, generation, specialisation))
172 tmp_path = "%s.tmp" % (entry_file)
173 kernel_params = "init=%s " % bootspec.init
174
175 kernel_params = kernel_params + " ".join(bootspec.kernelParams)
176 build_time = int(os.path.getctime(system_dir(profile, generation, specialisation)))
177 build_date = datetime.datetime.fromtimestamp(build_time).strftime('%F')
178
179 with open(tmp_path, 'w') as f:
180 f.write(BOOT_ENTRY.format(title=title,
181 sort_key=bootspec.sortKey,
182 generation=generation,
183 kernel=kernel,
184 initrd=initrd,
185 kernel_params=kernel_params,
186 description=f"{bootspec.label}, built on {build_date}"))
187 if machine_id is not None:
188 f.write("machine-id %s\n" % machine_id)
189 f.flush()
190 os.fsync(f.fileno())
191 os.rename(tmp_path, entry_file)
192
193
194def get_generations(profile: str | None = None) -> list[SystemIdentifier]:
195 gen_list = subprocess.check_output([
196 f"{NIX}/bin/nix-env",
197 "--list-generations",
198 "-p",
199 "/nix/var/nix/profiles/%s" % ("system-profiles/" + profile if profile else "system"),
200 "--option", "build-users-group", ""],
201 universal_newlines=True)
202 gen_lines = gen_list.split('\n')
203 gen_lines.pop()
204
205 configurationLimit = CONFIGURATION_LIMIT
206 configurations = [
207 SystemIdentifier(
208 profile=profile,
209 generation=int(line.split()[0]),
210 specialisation=None
211 )
212 for line in gen_lines
213 ]
214 return configurations[-configurationLimit:]
215
216
217def remove_old_entries(gens: list[SystemIdentifier]) -> None:
218 rex_profile = re.compile(r"^" + re.escape(BOOT_MOUNT_POINT) + "/loader/entries/nixos-(.*)-generation-.*\.conf$")
219 rex_generation = re.compile(r"^" + re.escape(BOOT_MOUNT_POINT) + "/loader/entries/nixos.*-generation-([0-9]+)(-specialisation-.*)?\.conf$")
220 known_paths = []
221 for gen in gens:
222 bootspec = get_bootspec(gen.profile, gen.generation)
223 known_paths.append(copy_from_file(bootspec.kernel, True))
224 known_paths.append(copy_from_file(bootspec.initrd, True))
225 for path in glob.iglob(f"{BOOT_MOUNT_POINT}/loader/entries/nixos*-generation-[1-9]*.conf"):
226 if rex_profile.match(path):
227 prof = rex_profile.sub(r"\1", path)
228 else:
229 prof = None
230 try:
231 gen_number = int(rex_generation.sub(r"\1", path))
232 except ValueError:
233 continue
234 if not (prof, gen_number, None) in gens:
235 os.unlink(path)
236 for path in glob.iglob(f"{BOOT_MOUNT_POINT}/{NIXOS_DIR}/*"):
237 if not path in known_paths and not os.path.isdir(path):
238 os.unlink(path)
239
240
241def cleanup_esp() -> None:
242 for path in glob.iglob(f"{EFI_SYS_MOUNT_POINT}/loader/entries/nixos*"):
243 os.unlink(path)
244 if os.path.isdir(f"{EFI_SYS_MOUNT_POINT}/{NIXOS_DIR}"):
245 shutil.rmtree(f"{EFI_SYS_MOUNT_POINT}/{NIXOS_DIR}")
246
247
248def get_profiles() -> list[str]:
249 if os.path.isdir("/nix/var/nix/profiles/system-profiles/"):
250 return [x
251 for x in os.listdir("/nix/var/nix/profiles/system-profiles/")
252 if not x.endswith("-link")]
253 else:
254 return []
255
256def install_bootloader(args: argparse.Namespace) -> None:
257 try:
258 with open("/etc/machine-id") as machine_file:
259 machine_id = machine_file.readlines()[0]
260 except IOError as e:
261 if e.errno != errno.ENOENT:
262 raise
263 # Since systemd version 232 a machine ID is required and it might not
264 # be there on newly installed systems, so let's generate one so that
265 # bootctl can find it and we can also pass it to write_entry() later.
266 cmd = [f"{SYSTEMD}/bin/systemd-machine-id-setup", "--print"]
267 machine_id = subprocess.run(
268 cmd, text=True, check=True, stdout=subprocess.PIPE
269 ).stdout.rstrip()
270
271 if os.getenv("NIXOS_INSTALL_GRUB") == "1":
272 warnings.warn("NIXOS_INSTALL_GRUB env var deprecated, use NIXOS_INSTALL_BOOTLOADER", DeprecationWarning)
273 os.environ["NIXOS_INSTALL_BOOTLOADER"] = "1"
274
275 # flags to pass to bootctl install/update
276 bootctl_flags = []
277
278 if BOOT_MOUNT_POINT != EFI_SYS_MOUNT_POINT:
279 bootctl_flags.append(f"--boot-path={BOOT_MOUNT_POINT}")
280
281 if CAN_TOUCH_EFI_VARIABLES != "1":
282 bootctl_flags.append("--no-variables")
283
284 if GRACEFUL == "1":
285 bootctl_flags.append("--graceful")
286
287 if os.getenv("NIXOS_INSTALL_BOOTLOADER") == "1":
288 # bootctl uses fopen() with modes "wxe" and fails if the file exists.
289 if os.path.exists(LOADER_CONF):
290 os.unlink(LOADER_CONF)
291
292 subprocess.check_call([f"{SYSTEMD}/bin/bootctl", f"--esp-path={EFI_SYS_MOUNT_POINT}"] + bootctl_flags + ["install"])
293 else:
294 # Update bootloader to latest if needed
295 available_out = subprocess.check_output([f"{SYSTEMD}/bin/bootctl", "--version"], universal_newlines=True).split()[2]
296 installed_out = subprocess.check_output([f"{SYSTEMD}/bin/bootctl", f"--esp-path={EFI_SYS_MOUNT_POINT}", "status"], universal_newlines=True)
297
298 # See status_binaries() in systemd bootctl.c for code which generates this
299 installed_match = re.search(r"^\W+File:.*/EFI/(?:BOOT|systemd)/.*\.efi \(systemd-boot ([\d.]+[^)]*)\)$",
300 installed_out, re.IGNORECASE | re.MULTILINE)
301
302 available_match = re.search(r"^\((.*)\)$", available_out)
303
304 if installed_match is None:
305 raise Exception("could not find any previously installed systemd-boot")
306
307 if available_match is None:
308 raise Exception("could not determine systemd-boot version")
309
310 installed_version = installed_match.group(1)
311 available_version = available_match.group(1)
312
313 if installed_version < available_version:
314 print("updating systemd-boot from %s to %s" % (installed_version, available_version))
315 subprocess.check_call([f"{SYSTEMD}/bin/bootctl", f"--esp-path={EFI_SYS_MOUNT_POINT}"] + bootctl_flags + ["update"])
316
317 os.makedirs(f"{BOOT_MOUNT_POINT}/{NIXOS_DIR}", exist_ok=True)
318 os.makedirs(f"{BOOT_MOUNT_POINT}/loader/entries", exist_ok=True)
319
320 gens = get_generations()
321 for profile in get_profiles():
322 gens += get_generations(profile)
323
324 remove_old_entries(gens)
325
326 for gen in gens:
327 try:
328 bootspec = get_bootspec(gen.profile, gen.generation)
329 is_default = os.path.dirname(bootspec.init) == args.default_config
330 write_entry(*gen, machine_id, bootspec, current=is_default)
331 for specialisation in bootspec.specialisations.keys():
332 write_entry(gen.profile, gen.generation, specialisation, machine_id, bootspec, current=is_default)
333 if is_default:
334 write_loader_conf(*gen)
335 except OSError as e:
336 # See https://github.com/NixOS/nixpkgs/issues/114552
337 if e.errno == errno.EINVAL:
338 profile = f"profile '{gen.profile}'" if gen.profile else "default profile"
339 print("ignoring {} in the list of boot entries because of the following error:\n{}".format(profile, e), file=sys.stderr)
340 else:
341 raise e
342
343 if BOOT_MOUNT_POINT != EFI_SYS_MOUNT_POINT:
344 # Cleanup any entries in ESP if xbootldrMountPoint is set.
345 # If the user later unsets xbootldrMountPoint, entries in XBOOTLDR will not be cleaned up
346 # automatically, as we don't have information about the mount point anymore.
347 cleanup_esp()
348
349 for root, _, files in os.walk(f"{BOOT_MOUNT_POINT}/{NIXOS_DIR}/.extra-files", topdown=False):
350 relative_root = root.removeprefix(f"{BOOT_MOUNT_POINT}/{NIXOS_DIR}/.extra-files").removeprefix("/")
351 actual_root = os.path.join(f"{BOOT_MOUNT_POINT}", relative_root)
352
353 for file in files:
354 actual_file = os.path.join(actual_root, file)
355
356 if os.path.exists(actual_file):
357 os.unlink(actual_file)
358 os.unlink(os.path.join(root, file))
359
360 if not len(os.listdir(actual_root)):
361 os.rmdir(actual_root)
362 os.rmdir(root)
363
364 os.makedirs(f"{BOOT_MOUNT_POINT}/{NIXOS_DIR}/.extra-files", exist_ok=True)
365
366 subprocess.check_call(COPY_EXTRA_FILES)
367
368
369def main() -> None:
370 parser = argparse.ArgumentParser(description=f"Update {DISTRO_NAME}-related systemd-boot files")
371 parser.add_argument('default_config', metavar='DEFAULT-CONFIG', help=f"The default {DISTRO_NAME} config to boot")
372 args = parser.parse_args()
373
374 subprocess.check_call(CHECK_MOUNTPOINTS)
375
376 try:
377 install_bootloader(args)
378 finally:
379 # Since fat32 provides little recovery facilities after a crash,
380 # it can leave the system in an unbootable state, when a crash/outage
381 # happens shortly after an update. To decrease the likelihood of this
382 # event sync the efi filesystem after each update.
383 rc = libc.syncfs(os.open(f"{BOOT_MOUNT_POINT}", os.O_RDONLY))
384 if rc != 0:
385 print(f"could not sync {BOOT_MOUNT_POINT}: {os.strerror(rc)}", file=sys.stderr)
386
387 if BOOT_MOUNT_POINT != EFI_SYS_MOUNT_POINT:
388 rc = libc.syncfs(os.open(EFI_SYS_MOUNT_POINT, os.O_RDONLY))
389 if rc != 0:
390 print(f"could not sync {EFI_SYS_MOUNT_POINT}: {os.strerror(rc)}", file=sys.stderr)
391
392
393if __name__ == '__main__':
394 main()