feat(installer): add one-file installer build and custom install ISO
Replace old_installer.py with a curses TUI installer packaged as a one-file PyInstaller binary (python/installer/build.py, wrapped by python/installer/package.nix). The default .#installer is patched to run on foreign Linux live media; the new .#installer-nixos variant keeps its Nix store interpreter so it runs on NixOS. Add systems/iso, a minimal NixOS install CD with kernel 6.18 and ZFS 2.4 matching the deployed systems and the installer on PATH; build it with nix build .#iso. Shared logging and subprocess helpers move out of common.py into python/logging_config.py and python/process.py.
This commit is contained in:
+4
-12
@@ -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]:
|
||||
|
||||
+130
-75
@@ -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}")
|
||||
|
||||
@@ -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()
|
||||
@@ -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()
|
||||
@@ -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.";
|
||||
}
|
||||
+5
-25
@@ -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
|
||||
|
||||
@@ -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)],
|
||||
)
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user