systemd-boot-builder: use safe run wrapper

we have a zoo of ways to call subprocesses.
Let's just replace this with one function that has reasonable defaults.
i.e. I catched instances where commands where run in a shell.

1 files changed, 47 insertions, 24 deletions

diff --git a/nixos/modules/system/boot/loader/systemd-boot/systemd-boot-builder.py b/nixos/modules/system/boot/loader/systemd-boot/systemd-boot-builder.py
index cbcc38618cc4..724fb090c4b5 100644
--- a/nixos/modules/system/boot/loader/systemd-boot/systemd-boot-builder.py
+++ b/nixos/modules/system/boot/loader/systemd-boot/systemd-boot-builder.py
@@ -49,6 +49,11 @@ class BootSpec:
 
 libc = ctypes.CDLL("libc.so.6")
 
+FILE = None | int
+
+def run(cmd: List[str], stdout: FILE = None) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(cmd, check=True, text=True, stdout=stdout)
+
 class SystemIdentifier(NamedTuple):
     profile: str | None
     generation: int
@@ -112,13 +117,16 @@ def get_bootspec(profile: str | None, generation: int) -> BootSpec:
         boot_json_f = open(boot_json_path, 'r')
         bootspec_json = json.load(boot_json_f)
     else:
-        boot_json_str = subprocess.check_output([
-        f"{BOOTSPEC_TOOLS}/bin/synthesize",
-        "--version",
-        "1",
-        system_directory,
-        "/dev/stdout"],
-        universal_newlines=True)
+        boot_json_str = run(
+            [
+                f"{BOOTSPEC_TOOLS}/bin/synthesize",
+                "--version",
+                "1",
+                system_directory,
+                "/dev/stdout",
+            ],
+            stdout=subprocess.PIPE,
+        ).stdout
         bootspec_json = json.loads(boot_json_str)
     return bootspec_from_json(bootspec_json)
 
@@ -157,7 +165,7 @@ def write_entry(profile: str | None, generation: int, specialisation: str | None
 
     try:
         if bootspec.initrdSecrets is not None:
-            subprocess.check_call([bootspec.initrdSecrets, f"{BOOT_MOUNT_POINT}%s" % (initrd)])
+            run([bootspec.initrdSecrets, f"{BOOT_MOUNT_POINT}%s" % (initrd)])
     except subprocess.CalledProcessError:
         if current:
             print("failed to create initrd secrets!", file=sys.stderr)
@@ -192,13 +200,17 @@ def write_entry(profile: str | None, generation: int, specialisation: str | None
 
 
 def get_generations(profile: str | None = None) -> list[SystemIdentifier]:
-    gen_list = subprocess.check_output([
-        f"{NIX}/bin/nix-env",
-        "--list-generations",
-        "-p",
-        "/nix/var/nix/profiles/%s" % ("system-profiles/" + profile if profile else "system")],
-        universal_newlines=True)
-    gen_lines = gen_list.split('\n')
+    gen_list = run(
+        [
+            f"{NIX}/bin/nix-env",
+            "--list-generations",
+            "-p",
+            "/nix/var/nix/profiles/%s"
+            % ("system-profiles/" + profile if profile else "system"),
+        ],
+        stdout=subprocess.PIPE,
+    ).stdout
+    gen_lines = gen_list.split("\n")
     gen_lines.pop()
 
     configurationLimit = CONFIGURATION_LIMIT
@@ -263,9 +275,7 @@ def install_bootloader(args: argparse.Namespace) -> None:
         # be there on newly installed systems, so let's generate one so that
         # bootctl can find it and we can also pass it to write_entry() later.
         cmd = [f"{SYSTEMD}/bin/systemd-machine-id-setup", "--print"]
-        machine_id = subprocess.run(
-          cmd, text=True, check=True, stdout=subprocess.PIPE
-        ).stdout.rstrip()
+        machine_id = run(cmd, stdout=subprocess.PIPE).stdout.rstrip()
 
     if os.getenv("NIXOS_INSTALL_GRUB") == "1":
         warnings.warn("NIXOS_INSTALL_GRUB env var deprecated, use NIXOS_INSTALL_BOOTLOADER", DeprecationWarning)
@@ -288,11 +298,20 @@ def install_bootloader(args: argparse.Namespace) -> None:
         if os.path.exists(LOADER_CONF):
             os.unlink(LOADER_CONF)
 
-        subprocess.check_call([f"{SYSTEMD}/bin/bootctl", f"--esp-path={EFI_SYS_MOUNT_POINT}"] + bootctl_flags + ["install"])
+        run(
+            [f"{SYSTEMD}/bin/bootctl", f"--esp-path={EFI_SYS_MOUNT_POINT}"]
+            + bootctl_flags
+            + ["install"]
+        )
     else:
         # Update bootloader to latest if needed
-        available_out = subprocess.check_output([f"{SYSTEMD}/bin/bootctl", "--version"], universal_newlines=True).split()[2]
-        installed_out = subprocess.check_output([f"{SYSTEMD}/bin/bootctl", f"--esp-path={EFI_SYS_MOUNT_POINT}", "status"], universal_newlines=True)
+        available_out = run(
+            [f"{SYSTEMD}/bin/bootctl", "--version"], stdout=subprocess.PIPE
+        ).stdout.split()[2]
+        installed_out = run(
+            [f"{SYSTEMD}/bin/bootctl", f"--esp-path={EFI_SYS_MOUNT_POINT}", "status"],
+            stdout=subprocess.PIPE,
+        ).stdout
 
         # See status_binaries() in systemd bootctl.c for code which generates this
         installed_match = re.search(r"^\W+File:.*/EFI/(?:BOOT|systemd)/.*\.efi \(systemd-boot ([\d.]+[^)]*)\)$",
@@ -311,7 +330,11 @@ def install_bootloader(args: argparse.Namespace) -> None:
 
         if installed_version < available_version:
             print("updating systemd-boot from %s to %s" % (installed_version, available_version))
-            subprocess.check_call([f"{SYSTEMD}/bin/bootctl", f"--esp-path={EFI_SYS_MOUNT_POINT}"] + bootctl_flags + ["update"])
+            run(
+                [f"{SYSTEMD}/bin/bootctl", f"--esp-path={EFI_SYS_MOUNT_POINT}"]
+                + bootctl_flags
+                + ["update"]
+            )
 
     os.makedirs(f"{BOOT_MOUNT_POINT}/{NIXOS_DIR}", exist_ok=True)
     os.makedirs(f"{BOOT_MOUNT_POINT}/loader/entries", exist_ok=True)
@@ -362,7 +385,7 @@ def install_bootloader(args: argparse.Namespace) -> None:
 
     os.makedirs(f"{BOOT_MOUNT_POINT}/{NIXOS_DIR}/.extra-files", exist_ok=True)
 
-    subprocess.check_call(COPY_EXTRA_FILES)
+    run([COPY_EXTRA_FILES])
 
 
 def main() -> None:
@@ -370,7 +393,7 @@ def main() -> None:
     parser.add_argument('default_config', metavar='DEFAULT-CONFIG', help=f"The default {DISTRO_NAME} config to boot")
     args = parser.parse_args()
 
-    subprocess.check_call(CHECK_MOUNTPOINTS)
+    run([CHECK_MOUNTPOINTS])
 
     try:
         install_bootloader(args)