Commit 1ae4daa6 authored by Jörg Thalheim's avatar Jörg Thalheim
Browse files

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.
parent 5cbf1952
Loading
Loading
Loading
Loading
+47 −24
Original line number Diff line number Diff line
@@ -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([
        boot_json_str = run(
            [
                f"{BOOTSPEC_TOOLS}/bin/synthesize",
                "--version",
                "1",
                system_directory,
        "/dev/stdout"],
        universal_newlines=True)
                "/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([
    gen_list = run(
        [
            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')
            "/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)