diff --git a/README.md b/README.md index 96bf7ed..001f461 100644 --- a/README.md +++ b/README.md @@ -1 +1,51 @@ # dotfiles + +## Installer ISO + +Build a bootable NixOS ISO with the installer preinstalled: + +```sh +nix build .#iso +``` + +Write `result/iso/nixos-zfs-installer.iso` to a USB stick (for example with `dd`) or boot it in a VM. The image is the minimal NixOS installation CD with ZFS enabled and `nixos-installer` on `PATH`, so once booted you can run: + +```sh +sudo nixos-installer +``` + +The ISO bundles the `.#installer-nixos` package, a variant of the binary that keeps its Nix store linkage instead of being patched for foreign distributions. + +## Installer binary + +Build the self-contained installer executable with: + +```sh +nix build .#installer +``` + +The flake package (defined in `python/installer/package.nix`) uses the Python builder in `python/installer/build.py`, which stages only the installer modules before running PyInstaller. You can also call it directly when `pyinstaller` and `patchelf` are on `PATH`: + +```sh +python -m python.installer.build --output ./nixos-installer +``` + +Copy `result/bin/nixos-installer` to the installer USB stick and run it as root from the NixOS live environment: + +```sh +sudo ./nixos-installer +``` + +Validate the live environment first with: + +```sh +./nixos-installer --check +``` + +Set `ENCRYPT_KEY` to enable LUKS during install: + +```sh +sudo env ENCRYPT_KEY='change-me' ./nixos-installer +``` + +The binary bundles the Python runtime and only the installer modules it imports. It still expects the NixOS installer environment to provide system install tools such as `parted`, `zfs`, `zpool`, `cryptsetup`, `nixos-generate-config`, and `nixos-install`. diff --git a/flake.nix b/flake.nix index 0d673cb..9058f6f 100644 --- a/flake.nix +++ b/flake.nix @@ -65,6 +65,35 @@ devShells = forEachSystem (pkgs: import ./shell.nix { inherit pkgs; }); formatter = forEachSystem (pkgs: pkgs.treefmt); + packages = forEachSystem ( + pkgs: + let + installer = pkgs.callPackage ./python/installer/package.nix { }; + installer-nixos = pkgs.callPackage ./python/installer/package.nix { patchElf = false; }; + in + { + inherit installer installer-nixos; + default = installer; + } + // lib.optionalAttrs (pkgs.stdenv.hostPlatform.system == "x86_64-linux") { + iso = self.nixosConfigurations.iso.config.system.build.isoImage; + } + ); + apps = forEachSystem ( + pkgs: + let + system = pkgs.stdenv.hostPlatform.system; + installer = { + type = "app"; + program = "${self.packages.${system}.installer}/bin/nixos-installer"; + meta.description = "One-file NixOS ZFS installer."; + }; + in + { + inherit installer; + default = installer; + } + ); nixosConfigurations = { bob = lib.nixosSystem { @@ -97,6 +126,12 @@ ]; specialArgs = { inherit inputs outputs; }; }; + iso = lib.nixosSystem { + modules = [ + ./systems/iso + ]; + specialArgs = { inherit inputs outputs; }; + }; }; }; } diff --git a/python/common.py b/python/common.py index 1e717dc..0fb07c2 100644 --- a/python/common.py +++ b/python/common.py @@ -3,28 +3,20 @@ from __future__ import annotations import logging -import sys from datetime import UTC, datetime from os import getenv from subprocess import PIPE, Popen from apprise import Apprise +from python.logging_config import configure_logger as _configure_logger + logger = logging.getLogger(__name__) def configure_logger(level: str = "INFO") -> None: - """Configure the logger. - - Args: - level (str, optional): The logging level. Defaults to "INFO". - """ - logging.basicConfig( - level=level, - datefmt="%Y-%m-%dT%H:%M:%S%z", - format="%(asctime)s %(levelname)s %(filename)s:%(lineno)d - %(message)s", - handlers=[logging.StreamHandler(sys.stdout)], - ) + """Configure the logger.""" + _configure_logger(level) def bash_wrapper(command: str) -> tuple[str, int]: diff --git a/python/installer/__main__.py b/python/installer/__main__.py index 2dc5c92..7c107e3 100644 --- a/python/installer/__main__.py +++ b/python/installer/__main__.py @@ -5,41 +5,37 @@ from __future__ import annotations import curses import logging import sys +from argparse import ArgumentParser from os import getenv from pathlib import Path from random import getrandbits -from subprocess import PIPE, Popen, run +from subprocess import run from time import sleep from typing import TYPE_CHECKING -from python.common import configure_logger from python.installer.tui import draw_menu +from python.logging_config import configure_logger +from python.process import require_commands, run_output if TYPE_CHECKING: from collections.abc import Sequence logger = logging.getLogger(__name__) - -def bash_wrapper(command: str) -> str: - """Execute a bash command and capture the output. - - Args: - command (str): The bash command to be executed. - - Returns: - Tuple[str, int]: A tuple containing the output of the command (stdout) as a string, - the error output (stderr) as a string (optional), and the return code as an integer. - """ - logger.debug(f"running {command=}") - # This is a acceptable risk - process = Popen(command.split(), stdout=PIPE, stderr=PIPE) - output, _ = process.communicate() - if process.returncode != 0: - error = f"Failed to run command {command=} return code {process.returncode=}" - raise RuntimeError(error) - - return output.decode() +REQUIRED_COMMANDS = ( + "blkdiscard", + "cryptsetup", + "find", + "lsblk", + "mkfs.vfat", + "mount", + "nixos-generate-config", + "nixos-install", + "parted", + "readlink", + "zfs", + "zpool", +) def partition_disk(disk: str, swap_size: int, reserve: int = 0) -> None: @@ -56,7 +52,7 @@ def partition_disk(disk: str, swap_size: int, reserve: int = 0) -> None: swap_size = max(swap_size, 1) reserve = max(reserve, 0) - bash_wrapper(f"blkdiscard -f {disk}") + run_output(("blkdiscard", "-f", disk)) if reserve > 0: msg = f"Creating swap partition on {disk=} with size {swap_size=}GiB and reserve {reserve=}GiB" @@ -72,14 +68,28 @@ def partition_disk(disk: str, swap_size: int, reserve: int = 0) -> None: logger.debug(f"{swap_partition=}") create_partitions = ( - f"parted --script --align=optimal {disk} -- " - "mklabel gpt " - "mkpart EFI 1MiB 4GiB " - f"mkpart root_pool 4GiB -{swap_start}GiB " - f"{swap_partition}" - "set 1 esp on" + "parted", + "--script", + "--align=optimal", + disk, + "--", + "mklabel", + "gpt", + "mkpart", + "EFI", + "1MiB", + "4GiB", + "mkpart", + "root_pool", + "4GiB", + f"-{swap_start}GiB", + *swap_partition.split(), + "set", + "1", + "esp", + "on", ) - bash_wrapper(create_partitions) + run_output(create_partitions) logger.info(f"{disk=} successfully partitioned") @@ -95,30 +105,43 @@ def create_zfs_pool(pool_disks: Sequence[str], mnt_dir: str) -> None: error = "disks must be a tuple of at least length 1" raise ValueError(error) - zpool_create = ( - "zpool create " - "-o ashift=12 " - "-o autotrim=on " - f"-R {mnt_dir} " - "-O acltype=posixacl " - "-O canmount=off " - "-O dnodesize=auto " - "-O normalization=formD " - "-O relatime=on " - "-O xattr=sa " - "-O mountpoint=legacy " - "-O compression=zstd " - "-O atime=off " - "root_pool " - ) + zpool_create = [ + "zpool", + "create", + "-o", + "ashift=12", + "-o", + "autotrim=on", + "-R", + mnt_dir, + "-O", + "acltype=posixacl", + "-O", + "canmount=off", + "-O", + "dnodesize=auto", + "-O", + "normalization=formD", + "-O", + "relatime=on", + "-O", + "xattr=sa", + "-O", + "mountpoint=legacy", + "-O", + "compression=zstd", + "-O", + "atime=off", + "root_pool", + ] if len(pool_disks) == 1: - zpool_create += pool_disks[0] + zpool_create.append(pool_disks[0]) else: - zpool_create += "mirror " - zpool_create += " ".join(pool_disks) + zpool_create.append("mirror") + zpool_create.extend(pool_disks) - bash_wrapper(zpool_create) - zpools = bash_wrapper("zpool list -o name") + run_output(zpool_create) + zpools = run_output(("zpool", "list", "-o", "name")) if "root_pool" not in zpools.splitlines(): logger.critical("Failed to create root_pool") sys.exit(1) @@ -126,11 +149,11 @@ def create_zfs_pool(pool_disks: Sequence[str], mnt_dir: str) -> None: def create_zfs_datasets() -> None: """Create ZFS datasets.""" - bash_wrapper("zfs create -o canmount=noauto -o reservation=10G root_pool/root") - bash_wrapper("zfs create root_pool/home") - bash_wrapper("zfs create root_pool/var -o reservation=1G") - bash_wrapper("zfs create -o compression=zstd-9 -o reservation=10G root_pool/nix") - datasets = bash_wrapper("zfs list -o name") + run_output(("zfs", "create", "-o", "canmount=noauto", "-o", "reservation=10G", "root_pool/root")) + run_output(("zfs", "create", "root_pool/home")) + run_output(("zfs", "create", "-o", "reservation=1G", "root_pool/var")) + run_output(("zfs", "create", "-o", "compression=zstd-9", "-o", "reservation=10G", "root_pool/nix")) + datasets = run_output(("zfs", "list", "-o", "name")) expected_datasets = { "root_pool/root", @@ -146,7 +169,7 @@ def create_zfs_datasets() -> None: def get_cpu_manufacturer() -> str: """Get the CPU manufacturer.""" - output = bash_wrapper("cat /proc/cpuinfo") + output = Path("/proc/cpuinfo").read_text() id_vendor = {"AuthenticAMD": "amd", "GenuineIntel": "intel"} @@ -160,7 +183,7 @@ def get_cpu_manufacturer() -> str: def get_boot_drive_id(disk: str) -> str: """Get the boot drive ID.""" - output = bash_wrapper(f"lsblk -o UUID {disk}-part1") + output = run_output(("lsblk", "-o", "UUID", f"{disk}-part1")) return output.splitlines()[1] @@ -220,21 +243,28 @@ def create_nix_hardware_file(mnt_dir: str, disks: Sequence[str], encrypt: str | def install_nixos(mnt_dir: str, disks: Sequence[str], encrypt: str | None) -> None: """Install NixOS.""" - bash_wrapper(f"mount -o X-mount.mkdir -t zfs root_pool/root {mnt_dir}") - bash_wrapper(f"mount -o X-mount.mkdir -t zfs root_pool/home {mnt_dir}/home") - bash_wrapper(f"mount -o X-mount.mkdir -t zfs root_pool/var {mnt_dir}/var") - bash_wrapper(f"mount -o X-mount.mkdir -t zfs root_pool/nix {mnt_dir}/nix") + run_output(("mount", "-o", "X-mount.mkdir", "-t", "zfs", "root_pool/root", mnt_dir)) + run_output(("mount", "-o", "X-mount.mkdir", "-t", "zfs", "root_pool/home", f"{mnt_dir}/home")) + run_output(("mount", "-o", "X-mount.mkdir", "-t", "zfs", "root_pool/var", f"{mnt_dir}/var")) + run_output(("mount", "-o", "X-mount.mkdir", "-t", "zfs", "root_pool/nix", f"{mnt_dir}/nix")) for disk in disks: - bash_wrapper(f"mkfs.vfat -n EFI {disk}-part1") + run_output(("mkfs.vfat", "-n", "EFI", f"{disk}-part1")) # set up mirroring afterwards if more than one disk - boot_partition = ( - f"mount -t vfat -o fmask=0077,dmask=0077,iocharset=iso8859-1,X-mount.mkdir {disks[0]}-part1 {mnt_dir}/boot" + run_output( + ( + "mount", + "-t", + "vfat", + "-o", + "fmask=0077,dmask=0077,iocharset=iso8859-1,X-mount.mkdir", + f"{disks[0]}-part1", + f"{mnt_dir}/boot", + ), ) - bash_wrapper(boot_partition) - bash_wrapper(f"nixos-generate-config --root {mnt_dir}") + run_output(("nixos-generate-config", "--root", mnt_dir)) create_nix_hardware_file(mnt_dir, disks, encrypt) @@ -249,18 +279,25 @@ def installer( ) -> None: """Main.""" logger.info("Starting installation") + require_commands(REQUIRED_COMMANDS) + disks = tuple(sorted(disks)) for disk in disks: partition_disk(disk, swap_size, reserve) - test = Popen(("printf", f"'{encrypt_key}'"), stdout=PIPE) if encrypt_key: sleep(1) - for command in ( - f"cryptsetup luksFormat --type luks2 {disk}-part2 -", - f"cryptsetup luksOpen {disk}-part2 luks-root-pool-{disk.split('/')[-1]}-part2 -", - ): - run(command, check=True, stdin=test.stdout) + key_input = encrypt_key.encode() + run( + ("cryptsetup", "luksFormat", "--type", "luks2", f"{disk}-part2", "-"), + input=key_input, + check=True, + ) + run( + ("cryptsetup", "luksOpen", f"{disk}-part2", f"luks-root-pool-{disk.split('/')[-1]}-part2", "-"), + input=key_input, + check=True, + ) # Fixed mount point for the new system; the installer runs as root on a fresh disk mnt_dir = "/tmp/nix_install" # noqa: S108 @@ -281,14 +318,32 @@ def installer( logger.info("Installation complete") -def main() -> None: +def main(argv: Sequence[str] | None = None) -> None: """Main.""" - configure_logger("DEBUG") + parser = ArgumentParser(description="Install this NixOS configuration onto a ZFS root pool.") + parser.add_argument( + "--check", + action="store_true", + help="validate that the live environment has the external installer commands and exit", + ) + parser.add_argument("--log-level", default=getenv("LOG_LEVEL", "DEBUG"), help="Python log level") + args = parser.parse_args(argv) + + configure_logger(args.log_level) + + if args.check: + require_commands(REQUIRED_COMMANDS) + logger.info("installer runtime dependencies are available") + return state = curses.wrapper(draw_menu) encrypt_key = getenv("ENCRYPT_KEY") + if not state.selected_device_ids: + logger.error("No disks selected; exiting without installing") + sys.exit(1) + logger.info("installing_nixos") logger.info(f"disks: {state.selected_device_ids}") logger.info(f"swap_size: {state.swap_size}") diff --git a/python/installer/build.py b/python/installer/build.py new file mode 100644 index 0000000..8b8bf4d --- /dev/null +++ b/python/installer/build.py @@ -0,0 +1,184 @@ +"""Build the one-file installer binary.""" + +from __future__ import annotations + +import logging +import os +import shutil +import stat +import subprocess +from argparse import ArgumentParser, Namespace +from pathlib import Path +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from collections.abc import Sequence + +logger = logging.getLogger(__name__) + +BINARY_NAME = "nixos-installer" +DEFAULT_INTERPRETER = "/lib64/ld-linux-x86-64.so.2" +INSTALLER_SOURCE_FILES = ( + Path("python/__init__.py"), + Path("python/logging_config.py"), + Path("python/process.py"), + Path("python/installer/__init__.py"), + Path("python/installer/__main__.py"), + Path("python/installer/tui.py"), +) + + +class InstallerBuildError(RuntimeError): + """Raised when the installer binary cannot be built.""" + + +class MissingSourceFileError(InstallerBuildError): + """Raised when a required source file is missing.""" + + def __init__(self, path: Path) -> None: + """Store the missing path.""" + super().__init__(f"Required installer source file is missing: {path}") + self.path = path + + +class MissingToolError(InstallerBuildError): + """Raised when a required build tool is missing.""" + + def __init__(self, tool: str) -> None: + """Store the missing tool name.""" + super().__init__(f"Required build tool is missing from PATH: {tool}") + self.tool = tool + + +def repo_root() -> Path: + """Return the repository root for direct script usage.""" + return Path(__file__).resolve().parents[2] + + +def require_tool(tool: str) -> str: + """Return the path to a tool or raise.""" + tool_path = shutil.which(tool) + if tool_path is None: + raise MissingToolError(tool) + return tool_path + + +def copy_installer_source(source_root: Path, destination: Path) -> None: + """Copy only the installer files into a minimal staging tree.""" + if destination.exists(): + shutil.rmtree(destination) + destination.mkdir(parents=True) + + for relative_path in INSTALLER_SOURCE_FILES: + source = source_root / relative_path + if not source.is_file(): + raise MissingSourceFileError(source) + + target = destination / relative_path + target.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(source, target) + + +def run_command(command: Sequence[str], *, env: dict[str, str] | None = None) -> None: + """Run a build command.""" + logger.info("running command=%s", command) + subprocess.run(command, check=True, env=env) + + +def pyinstaller_environment(staged_source: Path, build_root: Path) -> dict[str, str]: + """Return environment variables for PyInstaller.""" + env = os.environ.copy() + home = build_root / "home" + home.mkdir(parents=True, exist_ok=True) + + env["HOME"] = str(home) + if existing_pythonpath := env.get("PYTHONPATH"): + env["PYTHONPATH"] = f"{staged_source}{os.pathsep}{existing_pythonpath}" + else: + env["PYTHONPATH"] = str(staged_source) + + return env + + +def build_installer( + *, + source_root: Path, + build_root: Path, + output: Path, + interpreter: str, + patch_elf: bool, +) -> Path: + """Build the one-file installer binary.""" + pyinstaller = require_tool("pyinstaller") + if patch_elf: + patchelf = require_tool("patchelf") + + source_root = source_root.resolve() + build_root = build_root.resolve() + output = output.resolve() + + staged_source = build_root / "source" + dist_dir = build_root / "dist" + work_dir = build_root / "work" + spec_dir = build_root / "spec" + + copy_installer_source(source_root, staged_source) + output.parent.mkdir(parents=True, exist_ok=True) + + run_command( + ( + pyinstaller, + "--clean", + "--onefile", + "--name", + BINARY_NAME, + "--paths", + str(staged_source), + "--distpath", + str(dist_dir), + "--workpath", + str(work_dir), + "--specpath", + str(spec_dir), + str(staged_source / "python/installer/__main__.py"), + ), + env=pyinstaller_environment(staged_source, build_root), + ) + + built_binary = dist_dir / BINARY_NAME + shutil.copy2(built_binary, output) + output.chmod(output.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH) + + if patch_elf: + run_command((patchelf, "--set-interpreter", interpreter, "--remove-rpath", str(output))) + + return output + + +def parse_args(argv: Sequence[str] | None = None) -> Namespace: + """Parse command-line arguments.""" + parser = ArgumentParser(description="Build the one-file NixOS installer binary.") + parser.add_argument("--source-root", type=Path, default=repo_root(), help="repo or staged source root") + parser.add_argument("--build-root", type=Path, default=Path("build/nixos-installer"), help="temporary build root") + parser.add_argument("--output", type=Path, default=Path("dist/nixos-installer"), help="output binary path") + parser.add_argument("--interpreter", default=DEFAULT_INTERPRETER, help="ELF interpreter path for the USB binary") + parser.add_argument("--skip-patchelf", action="store_true", help="do not patch the final ELF binary") + parser.add_argument("--log-level", default="INFO", help="Python log level") + return parser.parse_args(argv) + + +def main(argv: Sequence[str] | None = None) -> None: + """Build the installer binary from the command line.""" + args = parse_args(argv) + logging.basicConfig(level=args.log_level, format="%(levelname)s %(message)s") + build_installer( + source_root=args.source_root, + build_root=args.build_root, + output=args.output, + interpreter=args.interpreter, + patch_elf=not args.skip_patchelf, + ) + + +if __name__ == "__main__": + main() diff --git a/python/installer/old_installer.py b/python/installer/old_installer.py deleted file mode 100644 index 8dd9214..0000000 --- a/python/installer/old_installer.py +++ /dev/null @@ -1,739 +0,0 @@ -"""Install NixOS on a ZFS pool.""" - -from __future__ import annotations - -import curses -import logging -import sys -from collections import defaultdict -from os import getenv -from pathlib import Path -from random import getrandbits -from subprocess import PIPE, Popen, run -from time import sleep -from typing import TYPE_CHECKING - -if TYPE_CHECKING: - from collections.abc import Sequence - -logger = logging.getLogger(__name__) -ESCAPE_KEY = 27 - - -def configure_logger(level: str = "INFO") -> None: - """Configure the logger. - - Args: - level (str, optional): The logging level. Defaults to "INFO". - """ - logging.basicConfig( - level=level, - datefmt="%Y-%m-%dT%H:%M:%S%z", - format="%(asctime)s %(levelname)s %(filename)s:%(lineno)d - %(message)s", - handlers=[logging.StreamHandler(sys.stdout)], - ) - - -def bash_wrapper(command: str) -> str: - """Execute a bash command and capture the output. - - Args: - command (str): The bash command to be executed. - - Returns: - Tuple[str, int]: A tuple containing the output of the command (stdout) as a string, - the error output (stderr) as a string (optional), and the return code as an integer. - """ - logger.debug(f"running {command=}") - # This is a acceptable risk - process = Popen(command.split(), stdout=PIPE, stderr=PIPE) - output, _ = process.communicate() - if process.returncode != 0: - error = f"Failed to run command {command=} return code {process.returncode=}" - raise RuntimeError(error) - - return output.decode() - - -def partition_disk(disk: str, swap_size: int, reserve: int = 0) -> None: - """Partition a disk. - - Args: - disk (str): The disk to partition. - swap_size (int): The size of the swap partition in GB. - minimum value is 1. - reserve (int, optional): The size of the reserve partition in GB. Defaults to 0. - minimum value is 0. - """ - logger.info(f"partitioning {disk=}") - swap_size = max(swap_size, 1) - reserve = max(reserve, 0) - - bash_wrapper(f"blkdiscard -f {disk}") - - if reserve > 0: - msg = f"Creating swap partition on {disk=} with size {swap_size=}GiB and reserve {reserve=}GiB" - logger.info(msg) - - swap_start = swap_size + reserve - swap_partition = f"mkpart swap -{swap_start}GiB -{reserve}GiB " - else: - logger.info(f"Creating swap partition on {disk=} with size {swap_size=}GiB") - swap_start = swap_size - swap_partition = f"mkpart swap -{swap_start}GiB 100% " - - logger.debug(f"{swap_partition=}") - - create_partitions = ( - f"parted --script --align=optimal {disk} -- " - "mklabel gpt " - "mkpart EFI 1MiB 4GiB " - f"mkpart root_pool 4GiB -{swap_start}GiB " - f"{swap_partition}" - "set 1 esp on" - ) - bash_wrapper(create_partitions) - - logger.info(f"{disk=} successfully partitioned") - - -def create_zfs_pool(pool_disks: Sequence[str], mnt_dir: str) -> None: - """Create a ZFS pool. - - Args: - pool_disks (Sequence[str]): A tuple of disks to use for the pool. - mnt_dir (str): The mount directory. - """ - if len(pool_disks) <= 0: - error = "disks must be a tuple of at least length 1" - raise ValueError(error) - - zpool_create = ( - "zpool create " - "-o ashift=12 " - "-o autotrim=on " - f"-R {mnt_dir} " - "-O acltype=posixacl " - "-O canmount=off " - "-O dnodesize=auto " - "-O normalization=formD " - "-O relatime=on " - "-O xattr=sa " - "-O mountpoint=legacy " - "-O compression=zstd " - "-O atime=off " - "root_pool " - ) - if len(pool_disks) == 1: - zpool_create += pool_disks[0] - else: - zpool_create += "mirror " - zpool_create += " ".join(pool_disks) - - bash_wrapper(zpool_create) - zpools = bash_wrapper("zpool list -o name") - if "root_pool" not in zpools.splitlines(): - logger.critical("Failed to create root_pool") - sys.exit(1) - - -def create_zfs_datasets() -> None: - """Create ZFS datasets.""" - bash_wrapper("zfs create -o canmount=noauto -o reservation=10G root_pool/root") - bash_wrapper("zfs create root_pool/home") - bash_wrapper("zfs create root_pool/var -o reservation=1G") - bash_wrapper("zfs create -o compression=zstd-9 -o reservation=10G root_pool/nix") - datasets = bash_wrapper("zfs list -o name") - - expected_datasets = { - "root_pool/root", - "root_pool/home", - "root_pool/var", - "root_pool/nix", - } - missing_datasets = expected_datasets.difference(datasets.splitlines()) - if missing_datasets: - logger.critical(f"Failed to create pools {missing_datasets}") - sys.exit(1) - - -def get_cpu_manufacturer() -> str: - """Get the CPU manufacturer.""" - output = bash_wrapper("cat /proc/cpuinfo") - - id_vendor = {"AuthenticAMD": "amd", "GenuineIntel": "intel"} - - for line in output.splitlines(): - if "vendor_id" in line: - return id_vendor[line.split(": ")[1].strip()] - error = "Failed to get CPU manufacturer" - raise RuntimeError(error) - - -def get_boot_drive_id(disk: str) -> str: - """Get the boot drive ID.""" - output = bash_wrapper(f"lsblk -o UUID {disk}-part1") - return output.splitlines()[1] - - -def create_nix_hardware_file(mnt_dir: str, disks: Sequence[str], *, encrypt: bool) -> None: - """Create a NixOS hardware file.""" - cpu_manufacturer = get_cpu_manufacturer() - - devices = "" - if encrypt: - disk = disks[0] - - devices = ( - f' luks.devices."luks-root-pool-{disk.split("/")[-1]}-part2"' - "= {\n" - f' device = "{disk}-part2";\n' - " bypassWorkqueues = true;\n" - " allowDiscards = true;\n" - " };\n" - ) - - host_id = format(getrandbits(32), "08x") - - nix_hardware = ( - "{ config, lib, modulesPath, ... }:\n" - "{\n" - ' imports = [ (modulesPath + "/installer/scan/not-detected.nix") ];\n\n' - " boot = {\n" - " initrd = {\n" - " availableKernelModules = [ \n" - ' "ahci"\n' - ' "ehci_pci"\n' - ' "nvme"\n' - ' "sd_mod"\n' - ' "usb_storage"\n' - ' "usbhid"\n' - ' "xhci_pci"\n' - " ];\n" - " kernelModules = [ ];\n" - f" {devices}" - " };\n" - f' kernelModules = [ "kvm-{cpu_manufacturer}" ];\n' - " extraModulePackages = [ ];\n" - " };\n\n" - " fileSystems = {\n" - ' "/" = lib.mkDefault {\n device = "root_pool/root";\n fsType = "zfs";\n };\n\n' - ' "/home" = {\n device = "root_pool/home";\n fsType = "zfs";\n };\n\n' - ' "/var" = {\n device = "root_pool/var";\n fsType = "zfs";\n };\n\n' - ' "/nix" = {\n device = "root_pool/nix";\n fsType = "zfs";\n };\n\n' - ' "/boot" = {\n' - f' device = "/dev/disk/by-uuid/{get_boot_drive_id(disks[0])}";\n' - ' fsType = "vfat";\n' - " options = [\n" - ' "fmask=0077"\n' - ' "dmask=0077"\n' - " ];\n" - " };\n" - " };\n\n" - " swapDevices = [ ];\n\n" - " networking.useDHCP = lib.mkDefault true;\n\n" - ' nixpkgs.hostPlatform = lib.mkDefault "x86_64-linux";\n' - f" hardware.cpu.{cpu_manufacturer}.updateMicrocode = lib.mkDefault " - "config.hardware.enableRedistributableFirmware;\n" - f' networking.hostId = "{host_id}";\n' - "}\n" - ) - - Path(f"{mnt_dir}/etc/nixos/hardware-configuration.nix").write_text(nix_hardware) - - -def install_nixos(mnt_dir: str, disks: Sequence[str], *, encrypt: bool) -> None: - """Install NixOS.""" - bash_wrapper(f"mount -o X-mount.mkdir -t zfs root_pool/root {mnt_dir}") - bash_wrapper(f"mount -o X-mount.mkdir -t zfs root_pool/home {mnt_dir}/home") - bash_wrapper(f"mount -o X-mount.mkdir -t zfs root_pool/var {mnt_dir}/var") - bash_wrapper(f"mount -o X-mount.mkdir -t zfs root_pool/nix {mnt_dir}/nix") - - for disk in disks: - bash_wrapper(f"mkfs.vfat -n EFI {disk}-part1") - - # set up mirroring afterwards if more than one disk - boot_partition = ( - f"mount -t vfat -o fmask=0077,dmask=0077,iocharset=iso8859-1,X-mount.mkdir {disks[0]}-part1 {mnt_dir}/boot" - ) - bash_wrapper(boot_partition) - - bash_wrapper(f"nixos-generate-config --root {mnt_dir}") - - create_nix_hardware_file(mnt_dir, disks, encrypt=encrypt) - - run(("nixos-install", "--root", mnt_dir), check=True) - - -def installer( - disks: set[str], - swap_size: int, - reserve: int, - encrypt_key: str | None, -) -> None: - """Main.""" - logger.info("Starting installation") - - for disk in disks: - partition_disk(disk, swap_size, reserve) - - if encrypt_key: - sleep(1) - key_input = encrypt_key.encode() - run( - ("cryptsetup", "luksFormat", "--type", "luks2", f"{disk}-part2", "-"), - input=key_input, - check=True, - ) - run( - ( - "cryptsetup", - "luksOpen", - f"{disk}-part2", - f"luks-root-pool-{disk.split('/')[-1]}-part2", - "-", - ), - input=key_input, - check=True, - ) - - # Fixed mount point for the new system; the installer runs as root on a fresh disk - mnt_dir = "/tmp/nix_install" # noqa: S108 - - Path(mnt_dir).mkdir(parents=True, exist_ok=True) - - if encrypt_key: - pool_disks = [f"/dev/mapper/luks-root-pool-{disk.split('/')[-1]}-part2" for disk in disks] - else: - pool_disks = [f"{disk}-part2" for disk in disks] - - create_zfs_pool(pool_disks, mnt_dir) - - create_zfs_datasets() - - install_nixos(mnt_dir, disks, encrypt=bool(encrypt_key)) - - logger.info("Installation complete") - - -class Cursor: - """Track cursor position and constrain movement to screen bounds.""" - - def __init__(self) -> None: - """Initialize cursor position and screen dimensions.""" - self.x_position = 0 - self.y_position = 0 - self.height = 0 - self.width = 0 - - def set_height(self, height: int) -> None: - """Set the maximum screen height.""" - self.height = height - - def set_width(self, width: int) -> None: - """Set the maximum screen width.""" - self.width = width - - def x_bounce_check(self, cursor: int) -> int: - """Clamp an x position to the screen width.""" - cursor = max(0, cursor) - return min(self.width - 1, cursor) - - def y_bounce_check(self, cursor: int) -> int: - """Clamp a y position to the screen height.""" - cursor = max(0, cursor) - return min(self.height - 1, cursor) - - def set_x(self, x: int) -> None: - """Set the cursor x position.""" - self.x_position = self.x_bounce_check(x) - - def set_y(self, y: int) -> None: - """Set the cursor y position.""" - self.y_position = self.y_bounce_check(y) - - def get_x(self) -> int: - """Get the cursor x position.""" - return self.x_position - - def get_y(self) -> int: - """Get the cursor y position.""" - return self.y_position - - def move_up(self) -> None: - """Move the cursor up one row.""" - self.set_y(self.y_position - 1) - - def move_down(self) -> None: - """Move the cursor down one row.""" - self.set_y(self.y_position + 1) - - def move_left(self) -> None: - """Move the cursor left one column.""" - self.set_x(self.x_position - 1) - - def move_right(self) -> None: - """Move the cursor right one column.""" - self.set_x(self.x_position + 1) - - def navigation(self, key: int) -> None: - """Move the cursor for a curses navigation key.""" - action = { - curses.KEY_DOWN: self.move_down, - curses.KEY_UP: self.move_up, - curses.KEY_RIGHT: self.move_right, - curses.KEY_LEFT: self.move_left, - } - - action.get(key, lambda: None)() - - -class State: - """State class to store the state of the program.""" - - def __init__(self) -> None: - """Initialize installer menu state.""" - self.key = 0 - self.cursor = Cursor() - - self.swap_size = 0 - self.show_swap_input = False - - self.reserve_size = 0 - self.show_reserve_input = False - - self.selected_device_ids = set() - - def get_selected_devices(self) -> tuple[str]: - """Get selected devices.""" - return tuple(self.selected_device_ids) - - -def get_device(raw_device: str) -> dict[str, str]: - """Parse an lsblk key-value device row.""" - raw_device_components = raw_device.split(" ") - return {thing.split("=")[0].lower(): thing.split("=")[1].strip('"') for thing in raw_device_components} - - -def get_devices() -> list[dict[str, str]]: - """Get a list of devices.""" - # --bytes - raw_devices = bash_wrapper("lsblk --paths --pairs").splitlines() - return [get_device(raw_device) for raw_device in raw_devices] - - -def get_device_id_mapping() -> dict[str, set[str]]: - """Get a list of device ids. - - Returns: - list[str]: the list of device ids - """ - device_ids = bash_wrapper("find /dev/disk/by-id -type l").splitlines() - - device_id_mapping: dict[str, set[str]] = defaultdict(set) - - for device_id in device_ids: - device = bash_wrapper(f"readlink -f {device_id}").strip() - device_id_mapping[device].add(device_id) - - return device_id_mapping - - -def calculate_device_menu_padding(devices: list[dict[str, str]], column: str, padding: int = 0) -> int: - """Calculate the width needed for a device menu column.""" - return max(len(device[column]) for device in devices) + padding - - -def draw_device_ids( - state: State, - row_number: int, - menu_start_x: int, - std_screen: curses.window, - menu_width: list[int], - device_ids: set[str], -) -> tuple[State, int]: - """Draw selectable device IDs for a device row.""" - for device_id in sorted(device_ids): - row_number = row_number + 1 - if row_number == state.cursor.get_y() and state.cursor.get_x() in menu_width: - std_screen.attron(curses.A_BOLD) - if state.key == ord(" "): - if device_id not in state.selected_device_ids: - state.selected_device_ids.add(device_id) - else: - state.selected_device_ids.remove(device_id) - - if device_id in state.selected_device_ids: - std_screen.attron(curses.color_pair(7)) - - std_screen.addstr(row_number, menu_start_x, f" {device_id}") - - std_screen.attroff(curses.color_pair(7)) - std_screen.attroff(curses.A_BOLD) - - return state, row_number - - -def draw_device_menu( - std_screen: curses.window, - devices: list[dict[str, str]], - device_id_mapping: dict[str, set[str]], - state: State, - menu_start_y: int = 0, - menu_start_x: int = 0, -) -> tuple[State, int]: - """Draw the device menu and handle user input. - - Args: - std_screen (curses.window): the curses window to draw on - devices (list[dict[str, str]]): the list of devices to draw - device_id_mapping (dict[str, set[str]]): the list of device ids to draw - state (State): the state object to update - menu_start_y (int, optional): the y position to start drawing the menu. Defaults to 0. - menu_start_x (int, optional): the x position to start drawing the menu. Defaults to 0. - - Returns: - State: the updated state object - """ - padding = 2 - - name_padding = calculate_device_menu_padding(devices, "name", padding) - size_padding = calculate_device_menu_padding(devices, "size", padding) - type_padding = calculate_device_menu_padding(devices, "type", padding) - mountpoints_padding = calculate_device_menu_padding(devices, "mountpoints", padding) - - device_header = ( - f"{'Name':{name_padding}}{'Size':{size_padding}}{'Type':{type_padding}}{'Mountpoints':{mountpoints_padding}}" - ) - - menu_width = range(menu_start_x, len(device_header) + menu_start_x) - - std_screen.addstr(menu_start_y, menu_start_x, device_header, curses.color_pair(5)) - devises_list_start = menu_start_y + 1 - - row_number = devises_list_start - - for device in devices: - row_number = row_number + 1 - device_name = device["name"] - device_row = ( - f"{device_name:{name_padding}}" - f"{device['size']:{size_padding}}" - f"{device['type']:{type_padding}}" - f"{device['mountpoints']:{mountpoints_padding}}" - ) - std_screen.addstr(row_number, menu_start_x, device_row) - - state, row_number = draw_device_ids( - state=state, - row_number=row_number, - menu_start_x=menu_start_x, - std_screen=std_screen, - menu_width=menu_width, - device_ids=device_id_mapping[device_name], - ) - - return state, row_number - - -def debug_menu(std_screen: curses.window, key: int) -> None: - """Draw debug information for the current curses screen.""" - height, width = std_screen.getmaxyx() - width_height = f"Width: {width}, Height: {height}" - std_screen.addstr(height - 4, 0, width_height, curses.color_pair(5)) - - key_pressed = f"Last key pressed: {key}"[: width - 1] - if key == 0: - key_pressed = "No key press detected..."[: width - 1] - std_screen.addstr(height - 3, 0, key_pressed) - - for i in range(8): - std_screen.addstr(height - 2, i * 3, f"{i}██", curses.color_pair(i)) - - -def status_bar( - std_screen: curses.window, - cursor: Cursor, - width: int, - height: int, -) -> None: - """Draw the footer status bar.""" - std_screen.attron(curses.A_REVERSE) - std_screen.attron(curses.color_pair(3)) - - status_bar = f"Press 'q' to exit | STATUS BAR | Pos: {cursor.get_x()}, {cursor.get_y()}" - std_screen.addstr(height - 1, 0, status_bar) - std_screen.addstr(height - 1, len(status_bar), " " * (width - len(status_bar) - 1)) - - std_screen.attroff(curses.color_pair(3)) - std_screen.attroff(curses.A_REVERSE) - - -def set_color() -> None: - """Initialize curses color pairs.""" - curses.start_color() - curses.use_default_colors() - for i in range(curses.COLORS): - curses.init_pair(i + 1, i, -1) - - -def get_text_input(std_screen: curses.window, prompt: str, y: int, x: int) -> str: - """Read text input from a curses screen.""" - curses.echo() - std_screen.addstr(y, x, prompt) - input_str = "" - while True: - key = std_screen.getch() - if key == ord("\n"): - break - if key == ESCAPE_KEY: - input_str = "" - break - if key in (curses.KEY_BACKSPACE, ord("\b"), 127): - input_str = input_str[:-1] - std_screen.addstr(y, x + len(prompt), input_str + " ") - else: - input_str += chr(key) - std_screen.refresh() - curses.noecho() - return input_str - - -def swap_size_input( - std_screen: curses.window, - state: State, - swap_offset: int, -) -> State: - """Handle swap size input.""" - swap_size_text = "Swap size (GB): " - std_screen.addstr(swap_offset, 0, f"{swap_size_text}{state.swap_size}") - if state.key == ord("\n") and state.cursor.get_y() == swap_offset: - state.show_swap_input = True - - if state.show_swap_input: - swap_size_str = get_text_input(std_screen, swap_size_text, swap_offset, 0) - try: - state.swap_size = int(swap_size_str) - state.show_swap_input = False - except ValueError: - std_screen.addstr(swap_offset, 0, "Invalid input. Press any key to continue.") - std_screen.getch() - state.show_swap_input = False - - return state - - -def reserve_size_input( - std_screen: curses.window, - state: State, - reserve_offset: int, -) -> State: - """Handle reserve size input.""" - reserve_size_text = "reserve size (GB): " - std_screen.addstr(reserve_offset, 0, f"{reserve_size_text}{state.reserve_size}") - if state.key == ord("\n") and state.cursor.get_y() == reserve_offset: - state.show_reserve_input = True - - if state.show_reserve_input: - reserve_size_str = get_text_input(std_screen, reserve_size_text, reserve_offset, 0) - try: - state.reserve_size = int(reserve_size_str) - state.show_reserve_input = False - except ValueError: - std_screen.addstr(reserve_offset, 0, "Invalid input. Press any key to continue.") - std_screen.getch() - state.show_reserve_input = False - - return state - - -def draw_menu(std_screen: curses.window) -> State: - """Draw the menu and handle user input. - - Args: - std_screen (curses.window): the curses window to draw on - - Returns: - State: the state object - """ - # Clear and refresh the screen for a blank canvas - std_screen.clear() - std_screen.refresh() - - set_color() - - state = State() - - devices = get_devices() - - device_id_mapping = get_device_id_mapping() - - # Loop where k is the last character pressed - while state.key != ord("q"): - std_screen.clear() - height, width = std_screen.getmaxyx() - - state.cursor.set_height(height) - state.cursor.set_width(width) - - state.cursor.navigation(state.key) - - state, device_menu_size = draw_device_menu( - std_screen=std_screen, - state=state, - devices=devices, - device_id_mapping=device_id_mapping, - ) - - swap_offset = device_menu_size + 2 - - swap_size_input( - std_screen=std_screen, - state=state, - swap_offset=swap_offset, - ) - reserve_size_input( - std_screen=std_screen, - state=state, - reserve_offset=swap_offset + 1, - ) - - status_bar(std_screen, state.cursor, width, height) - - debug_menu(std_screen, state.key) - - std_screen.move(state.cursor.get_y(), state.cursor.get_x()) - - std_screen.refresh() - - state.key = std_screen.getch() - - return state - - -def main() -> None: - """Run the installer menu and start installation.""" - configure_logger("DEBUG") - - state = curses.wrapper(draw_menu) - - encrypt_key = getenv("ENCRYPT_KEY") - - logger.info("installing_nixos") - logger.info(f"disks: {state.selected_device_ids}") - logger.info(f"swap_size: {state.swap_size}") - logger.info(f"reserve: {state.reserve_size}") - logger.info(f"encrypted: {bool(encrypt_key)}") - - sleep(3) - - installer( - disks=state.get_selected_devices(), - swap_size=state.swap_size, - reserve=state.reserve_size, - encrypt_key=encrypt_key, - ) - - -if __name__ == "__main__": - main() diff --git a/python/installer/package.nix b/python/installer/package.nix new file mode 100644 index 0000000..35bb9c1 --- /dev/null +++ b/python/installer/package.nix @@ -0,0 +1,49 @@ +{ + lib, + stdenv, + patchelf, + python314, + python314Packages, + patchElf ? true, +}: + +stdenv.mkDerivation { + pname = "nixos-installer"; + version = "0.1.0"; + src = ../../.; + + dontPatchELF = true; + dontStrip = true; + + nativeBuildInputs = [ + patchelf + python314 + python314Packages.pyinstaller + ]; + + buildPhase = '' + runHook preBuild + + export HOME="$TMPDIR" + python "$src/python/installer/build.py" \ + --source-root "$src" \ + --build-root "$TMPDIR/nixos-installer-build" \ + --output "$PWD/nixos-installer" ${lib.optionalString (!patchElf) "--skip-patchelf"} + + runHook postBuild + ''; + + installPhase = '' + runHook preInstall + + install -Dm755 nixos-installer $out/bin/nixos-installer + + runHook postInstall + ''; + + meta.description = + if patchElf then + "One-file NixOS ZFS installer patched to run on foreign Linux live environments." + else + "One-file NixOS ZFS installer linked against the Nix store, for the custom install ISO."; +} diff --git a/python/installer/tui.py b/python/installer/tui.py index 6224b18..d66efc9 100644 --- a/python/installer/tui.py +++ b/python/installer/tui.py @@ -5,32 +5,12 @@ from __future__ import annotations import curses import logging from collections import defaultdict -from subprocess import PIPE, Popen + +from python.process import run_output logger = logging.getLogger(__name__) -def bash_wrapper(command: str) -> str: - """Execute a bash command and capture the output. - - Args: - command (str): The bash command to be executed. - - Returns: - Tuple[str, int]: A tuple containing the output of the command (stdout) as a string, - the error output (stderr) as a string (optional), and the return code as an integer. - """ - logger.debug(f"running {command=}") - # This is a acceptable risk - process = Popen(command.split(), stdout=PIPE, stderr=PIPE) - output, _ = process.communicate() - if process.returncode != 0: - error = f"Failed to run command {command=} return code {process.returncode=}" - raise RuntimeError(error) - - return output.decode() - - class Cursor: """Cursor class.""" @@ -144,7 +124,7 @@ def get_device(raw_device: str) -> dict[str, str]: def get_devices() -> list[dict[str, str]]: """Get a list of devices.""" # --bytes - raw_devices = bash_wrapper("lsblk --paths --pairs").splitlines() + raw_devices = run_output(("lsblk", "--paths", "--pairs")).splitlines() return [get_device(raw_device) for raw_device in raw_devices] @@ -305,12 +285,12 @@ def get_device_id_mapping() -> dict[str, set[str]]: Returns: list[str]: the list of device ids """ - device_ids = bash_wrapper("find /dev/disk/by-id -type l").splitlines() + device_ids = run_output(("find", "/dev/disk/by-id", "-type", "l")).splitlines() device_id_mapping: dict[str, set[str]] = defaultdict(set) for device_id in device_ids: - device = bash_wrapper(f"readlink -f {device_id}").strip() + device = run_output(("readlink", "-f", device_id)).strip() device_id_mapping[device].add(device_id) return device_id_mapping diff --git a/python/logging_config.py b/python/logging_config.py new file mode 100644 index 0000000..40b9f9b --- /dev/null +++ b/python/logging_config.py @@ -0,0 +1,16 @@ +"""Logging helpers shared by command-line tools.""" + +from __future__ import annotations + +import logging +import sys + + +def configure_logger(level: str = "INFO") -> None: + """Configure process-wide logging.""" + logging.basicConfig( + level=level, + datefmt="%Y-%m-%dT%H:%M:%S%z", + format="%(asctime)s %(levelname)s %(filename)s:%(lineno)d - %(message)s", + handlers=[logging.StreamHandler(sys.stdout)], + ) diff --git a/python/process.py b/python/process.py new file mode 100644 index 0000000..341961e --- /dev/null +++ b/python/process.py @@ -0,0 +1,52 @@ +"""Small subprocess helpers.""" + +from __future__ import annotations + +import logging +import shutil +from subprocess import run +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from collections.abc import Iterable, Sequence + +logger = logging.getLogger(__name__) + + +class CommandError(RuntimeError): + """Raised when an external command fails.""" + + def __init__(self, command: Sequence[str], returncode: int, stderr: str) -> None: + """Store command failure details.""" + command_text = " ".join(command) + super().__init__(f"Failed to run command {command_text!r}: exit {returncode}\n{stderr}") + self.command = command + self.returncode = returncode + self.stderr = stderr + + +class MissingCommandsError(RuntimeError): + """Raised when required external commands are not available.""" + + def __init__(self, commands: Sequence[str]) -> None: + """Store missing command details.""" + missing = ", ".join(commands) + super().__init__(f"Missing required installer commands: {missing}") + self.commands = commands + + +def require_commands(commands: Iterable[str]) -> None: + """Raise when one or more executables are missing from PATH.""" + missing_commands = sorted({command for command in commands if shutil.which(command) is None}) + if missing_commands: + raise MissingCommandsError(missing_commands) + + +def run_output(command: Sequence[str]) -> str: + """Run a command and return stdout.""" + logger.debug("running command=%s", command) + result = run(command, capture_output=True, text=True, check=False) + if result.returncode != 0: + raise CommandError(command, result.returncode, result.stderr) + + return result.stdout diff --git a/systems/iso/default.nix b/systems/iso/default.nix new file mode 100644 index 0000000..b8399aa --- /dev/null +++ b/systems/iso/default.nix @@ -0,0 +1,34 @@ +{ + outputs, + lib, + pkgs, + modulesPath, + ... +}: +{ + imports = [ + "${modulesPath}/installer/cd-dvd/installation-cd-minimal.nix" + ]; + + nixpkgs.hostPlatform = "x86_64-linux"; + + image.baseName = lib.mkForce "nixos-zfs-installer"; + + # Keep the live kernel and ZFS in sync with the deployed systems so pools + # created by the installer import cleanly on first boot. + boot = { + kernelPackages = pkgs.linuxPackages_6_18; + zfs.package = pkgs.zfs_2_4; + supportedFilesystems.zfs = true; + }; + + networking.hostName = "installer"; + + environment.systemPackages = [ + outputs.packages.${pkgs.stdenv.hostPlatform.system}.installer-nixos + ]; + + services.getty.helpLine = '' + Run "sudo nixos-installer" to install NixOS onto a ZFS root pool. + ''; +}