diff --git a/pkgs/clan-cli/clan_cli/machines/hardware.py b/pkgs/clan-cli/clan_cli/machines/hardware.py index 86da4538b..8204316cf 100644 --- a/pkgs/clan-cli/clan_cli/machines/hardware.py +++ b/pkgs/clan-cli/clan_cli/machines/hardware.py @@ -1,17 +1,12 @@ import argparse -import json import logging -from dataclasses import dataclass -from enum import Enum -from pathlib import Path -from clan_lib.api import API -from clan_lib.cmd import RunOpts, run -from clan_lib.dirs import specific_machine_dir -from clan_lib.errors import ClanCmdError, ClanError -from clan_lib.git import commit_file +from clan_lib.machines.hardware import ( + HardwareConfig, + HardwareGenerateOptions, + generate_machine_hardware_info, +) from clan_lib.machines.machines import Machine -from clan_lib.nix import nix_config, nix_eval from clan_lib.ssh.remote import Remote from clan_cli.completions import add_dynamic_completer, complete_machines @@ -21,142 +16,6 @@ from .types import machine_name_type log = logging.getLogger(__name__) -class HardwareConfig(Enum): - NIXOS_FACTER = "nixos-facter" - NIXOS_GENERATE_CONFIG = "nixos-generate-config" - NONE = "none" - - def config_path(self, machine: Machine) -> Path: - machine_dir = specific_machine_dir(machine) - if self == HardwareConfig.NIXOS_FACTER: - return machine_dir / "facter.json" - return machine_dir / "hardware-configuration.nix" - - @classmethod - def detect_type(cls: type["HardwareConfig"], machine: Machine) -> "HardwareConfig": - hardware_config = HardwareConfig.NIXOS_GENERATE_CONFIG.config_path(machine) - - if hardware_config.exists() and "throw" not in hardware_config.read_text(): - return HardwareConfig.NIXOS_GENERATE_CONFIG - - if HardwareConfig.NIXOS_FACTER.config_path(machine).exists(): - return HardwareConfig.NIXOS_FACTER - - return HardwareConfig.NONE - - -@API.register -def show_machine_hardware_config(machine: Machine) -> HardwareConfig: - """ - Show hardware information for a machine returns None if none exist. - """ - return HardwareConfig.detect_type(machine) - - -@API.register -def show_machine_hardware_platform(machine: Machine) -> str | None: - """ - Show hardware information for a machine returns None if none exist. - """ - config = nix_config() - system = config["system"] - cmd = nix_eval( - [ - f"{machine.flake}#clanInternals.machines.{system}.{machine.name}", - "--apply", - "machine: { inherit (machine.pkgs) system; }", - "--json", - ] - ) - proc = run(cmd, RunOpts(prefix=machine.name)) - res = proc.stdout.strip() - - host_platform = json.loads(res) - return host_platform.get("system", None) - - -@dataclass -class HardwareGenerateOptions: - machine: Machine - backend: HardwareConfig - password: str | None = None - - -@API.register -def generate_machine_hardware_info( - opts: HardwareGenerateOptions, target_host: Remote -) -> HardwareConfig: - """ - Generate hardware information for a machine - and place the resulting *.nix file in the machine's directory. - """ - - machine = opts.machine - - hw_file = opts.backend.config_path(opts.machine) - hw_file.parent.mkdir(parents=True, exist_ok=True) - - if opts.backend == HardwareConfig.NIXOS_FACTER: - config_command = ["nixos-facter"] - else: - config_command = [ - "nixos-generate-config", - # Filesystems are managed by disko - "--no-filesystems", - "--show-hardware-config", - ] - - with target_host.ssh_control_master() as ssh, ssh.become_root() as sudo_ssh: - out = sudo_ssh.run(config_command, opts=RunOpts(check=False)) - if out.returncode != 0: - if "nixos-facter" in out.stderr and "not found" in out.stderr: - machine.error(str(out.stderr)) - msg = ( - "Please use our custom nixos install images from https://github.com/nix-community/nixos-images/releases/tag/nixos-unstable. " - "nixos-factor only works on nixos / clan systems currently." - ) - raise ClanError(msg) - - machine.error(str(out)) - msg = f"Failed to inspect {opts.machine}. Address: {target_host.target}" - raise ClanError(msg) - - backup_file = None - if hw_file.exists(): - backup_file = hw_file.with_suffix(".bak") - hw_file.replace(backup_file) - hw_file.write_text(out.stdout) - print(f"Successfully generated: {hw_file}") - - # try to evaluate the machine - # If it fails, the hardware-configuration.nix file is invalid - - commit_file( - hw_file, - opts.machine.flake.path, - f"machines/{opts.machine}/{hw_file.name}: update hardware configuration", - ) - try: - show_machine_hardware_platform(opts.machine) - if backup_file: - backup_file.unlink(missing_ok=True) - except ClanCmdError as e: - log.exception("Failed to evaluate hardware-configuration.nix") - # Restore the backup file - print(f"Restoring backup file {backup_file}") - if backup_file: - backup_file.replace(hw_file) - # TODO: Undo the commit - - msg = "Invalid hardware-configuration.nix file" - raise ClanError( - msg, - description=f"Configuration at '{hw_file}' is invalid. Please check the file and try again.", - ) from e - - return opts.backend - - def update_hardware_config_command(args: argparse.Namespace) -> None: host_key_check = args.host_key_check machine = Machine(flake=args.flake, name=args.machine) diff --git a/pkgs/clan-cli/clan_cli/machines/install.py b/pkgs/clan-cli/clan_cli/machines/install.py index 5943034e7..0aaba5c8f 100644 --- a/pkgs/clan-cli/clan_cli/machines/install.py +++ b/pkgs/clan-cli/clan_cli/machines/install.py @@ -1,17 +1,11 @@ import argparse import logging -import os import sys -from dataclasses import dataclass, field -from enum import Enum from pathlib import Path -from tempfile import TemporaryDirectory -from clan_lib.api import API -from clan_lib.cmd import Log, RunOpts, run from clan_lib.errors import ClanError +from clan_lib.machines.install import BuildOn, InstallOptions, install_machine from clan_lib.machines.machines import Machine -from clan_lib.nix import nix_shell from clan_lib.ssh.remote import Remote from clan_cli.completions import ( @@ -19,145 +13,12 @@ from clan_cli.completions import ( complete_machines, complete_target_host, ) -from clan_cli.facts.generate import generate_facts from clan_cli.machines.hardware import HardwareConfig from clan_cli.ssh.deploy_info import DeployInfo, find_reachable_host, ssh_command_parse -from clan_cli.vars.generate import generate_vars log = logging.getLogger(__name__) -class BuildOn(Enum): - AUTO = "auto" - LOCAL = "local" - REMOTE = "remote" - - -@dataclass -class InstallOptions: - machine: Machine - kexec: str | None = None - debug: bool = False - no_reboot: bool = False - phases: str | None = None - build_on: BuildOn | None = None - nix_options: list[str] = field(default_factory=list) - update_hardware_config: HardwareConfig = HardwareConfig.NONE - password: str | None = None - identity_file: Path | None = None - use_tor: bool = False - - -@API.register -def install_machine(opts: InstallOptions, target_host: Remote) -> None: - machine = opts.machine - - machine.debug(f"installing {machine.name}") - - generate_facts([machine]) - generate_vars([machine]) - - with ( - TemporaryDirectory(prefix="nixos-install-") as _base_directory, - ): - base_directory = Path(_base_directory).resolve() - activation_secrets = base_directory / "activation_secrets" - upload_dir = activation_secrets / machine.secrets_upload_directory.lstrip("/") - upload_dir.mkdir(parents=True) - machine.secret_facts_store.upload(upload_dir) - machine.secret_vars_store.populate_dir( - upload_dir, phases=["activation", "users", "services"] - ) - - partitioning_secrets = base_directory / "partitioning_secrets" - partitioning_secrets.mkdir(parents=True) - machine.secret_vars_store.populate_dir( - partitioning_secrets, phases=["partitioning"] - ) - - if opts.password: - os.environ["SSHPASS"] = opts.password - - cmd = [ - "nixos-anywhere", - "--flake", - f"{machine.flake}#{machine.name}", - "--extra-files", - str(activation_secrets), - ] - - for path in partitioning_secrets.rglob("*"): - if path.is_file(): - cmd.extend( - [ - "--disk-encryption-keys", - str( - "/run/partitioning-secrets" - / path.relative_to(partitioning_secrets) - ), - str(path), - ] - ) - - if opts.no_reboot: - cmd.append("--no-reboot") - - if opts.phases: - cmd += ["--phases", str(opts.phases)] - - if opts.update_hardware_config is not HardwareConfig.NONE: - cmd.extend( - [ - "--generate-hardware-config", - str(opts.update_hardware_config.value), - str(opts.update_hardware_config.config_path(machine)), - ] - ) - - if opts.password: - cmd += [ - "--env-password", - "--ssh-option", - "IdentitiesOnly=yes", - ] - - if opts.identity_file: - cmd += ["-i", str(opts.identity_file)] - - if opts.build_on: - cmd += ["--build-on", opts.build_on.value] - - if target_host.port: - cmd += ["--ssh-port", str(target_host.port)] - if opts.kexec: - cmd += ["--kexec", opts.kexec] - - if opts.debug: - cmd.append("--debug") - - # Add nix options to nixos-anywhere - cmd.extend(opts.nix_options) - - cmd.append(target_host.target) - if opts.use_tor: - # nix copy does not support tor socks proxy - # cmd.append("--ssh-option") - # cmd.append("ProxyCommand=nc -x 127.0.0.1:9050 -X 5 %h %p") - cmd = nix_shell( - [ - "nixos-anywhere", - "tor", - ], - ["torify", *cmd], - ) - else: - cmd = nix_shell( - ["nixos-anywhere"], - cmd, - ) - run(cmd, RunOpts(log=Log.BOTH, prefix=machine.name, needs_user_terminal=True)) - - def install_command(args: argparse.Namespace) -> None: try: # Only if the caller did not specify a target_host via args.target_host diff --git a/pkgs/clan-cli/clan_cli/ssh/results.py b/pkgs/clan-cli/clan_cli/ssh/results.py deleted file mode 100644 index 00b328d05..000000000 --- a/pkgs/clan-cli/clan_cli/ssh/results.py +++ /dev/null @@ -1,34 +0,0 @@ -from dataclasses import dataclass -from typing import Generic - -from clan_lib.errors import CmdOut -from clan_lib.ssh.remote import Remote - -from clan_cli.ssh import T - - -@dataclass -class HostResult(Generic[T]): - host: Remote - _result: T | Exception - - @property - def error(self) -> Exception | None: - """ - Returns an error if the command failed - """ - if isinstance(self._result, Exception): - return self._result - return None - - @property - def result(self) -> T: - """ - Unwrap the result - """ - if isinstance(self._result, Exception): - raise self._result - return self._result - - -Results = list[HostResult[CmdOut]] diff --git a/pkgs/clan-cli/clan_lib/api/disk.py b/pkgs/clan-cli/clan_lib/api/disk.py index cb7043dce..28a14befa 100644 --- a/pkgs/clan-cli/clan_lib/api/disk.py +++ b/pkgs/clan-cli/clan_lib/api/disk.py @@ -5,13 +5,12 @@ from dataclasses import dataclass from typing import Any, TypedDict from uuid import uuid4 -from clan_cli.machines.hardware import HardwareConfig, show_machine_hardware_config - from clan_lib.api import API from clan_lib.api.modules import Frontmatter, extract_frontmatter from clan_lib.dirs import TemplateType, clan_templates from clan_lib.errors import ClanError from clan_lib.git import commit_file +from clan_lib.machines.hardware import HardwareConfig, show_machine_hardware_config from clan_lib.machines.machines import Machine log = logging.getLogger(__name__) diff --git a/pkgs/clan-cli/clan_lib/machines/hardware.py b/pkgs/clan-cli/clan_lib/machines/hardware.py new file mode 100644 index 000000000..8324a7953 --- /dev/null +++ b/pkgs/clan-cli/clan_lib/machines/hardware.py @@ -0,0 +1,152 @@ +import json +import logging +from dataclasses import dataclass +from enum import Enum +from pathlib import Path + +from clan_lib.api import API +from clan_lib.cmd import RunOpts, run +from clan_lib.dirs import specific_machine_dir +from clan_lib.errors import ClanCmdError, ClanError +from clan_lib.git import commit_file +from clan_lib.machines.machines import Machine +from clan_lib.nix import nix_config, nix_eval +from clan_lib.ssh.remote import Remote + +log = logging.getLogger(__name__) + + +class HardwareConfig(Enum): + NIXOS_FACTER = "nixos-facter" + NIXOS_GENERATE_CONFIG = "nixos-generate-config" + NONE = "none" + + def config_path(self, machine: Machine) -> Path: + machine_dir = specific_machine_dir(machine) + if self == HardwareConfig.NIXOS_FACTER: + return machine_dir / "facter.json" + return machine_dir / "hardware-configuration.nix" + + @classmethod + def detect_type(cls: type["HardwareConfig"], machine: Machine) -> "HardwareConfig": + hardware_config = HardwareConfig.NIXOS_GENERATE_CONFIG.config_path(machine) + + if hardware_config.exists() and "throw" not in hardware_config.read_text(): + return HardwareConfig.NIXOS_GENERATE_CONFIG + + if HardwareConfig.NIXOS_FACTER.config_path(machine).exists(): + return HardwareConfig.NIXOS_FACTER + + return HardwareConfig.NONE + + +@API.register +def show_machine_hardware_config(machine: Machine) -> HardwareConfig: + """ + Show hardware information for a machine returns None if none exist. + """ + return HardwareConfig.detect_type(machine) + + +@API.register +def show_machine_hardware_platform(machine: Machine) -> str | None: + """ + Show hardware information for a machine returns None if none exist. + """ + config = nix_config() + system = config["system"] + cmd = nix_eval( + [ + f"{machine.flake}#clanInternals.machines.{system}.{machine.name}", + "--apply", + "machine: { inherit (machine.pkgs) system; }", + "--json", + ] + ) + proc = run(cmd, RunOpts(prefix=machine.name)) + res = proc.stdout.strip() + + host_platform = json.loads(res) + return host_platform.get("system", None) + + +@dataclass +class HardwareGenerateOptions: + machine: Machine + backend: HardwareConfig + password: str | None = None + + +@API.register +def generate_machine_hardware_info( + opts: HardwareGenerateOptions, target_host: Remote +) -> HardwareConfig: + """ + Generate hardware information for a machine + and place the resulting *.nix file in the machine's directory. + """ + + machine = opts.machine + + hw_file = opts.backend.config_path(opts.machine) + hw_file.parent.mkdir(parents=True, exist_ok=True) + + if opts.backend == HardwareConfig.NIXOS_FACTER: + config_command = ["nixos-facter"] + else: + config_command = [ + "nixos-generate-config", + # Filesystems are managed by disko + "--no-filesystems", + "--show-hardware-config", + ] + + with target_host.ssh_control_master() as ssh, ssh.become_root() as sudo_ssh: + out = sudo_ssh.run(config_command, opts=RunOpts(check=False)) + if out.returncode != 0: + if "nixos-facter" in out.stderr and "not found" in out.stderr: + machine.error(str(out.stderr)) + msg = ( + "Please use our custom nixos install images from https://github.com/nix-community/nixos-images/releases/tag/nixos-unstable. " + "nixos-factor only works on nixos / clan systems currently." + ) + raise ClanError(msg) + + machine.error(str(out)) + msg = f"Failed to inspect {opts.machine}. Address: {target_host.target}" + raise ClanError(msg) + + backup_file = None + if hw_file.exists(): + backup_file = hw_file.with_suffix(".bak") + hw_file.replace(backup_file) + hw_file.write_text(out.stdout) + print(f"Successfully generated: {hw_file}") + + # try to evaluate the machine + # If it fails, the hardware-configuration.nix file is invalid + + commit_file( + hw_file, + opts.machine.flake.path, + f"machines/{opts.machine}/{hw_file.name}: update hardware configuration", + ) + try: + show_machine_hardware_platform(opts.machine) + if backup_file: + backup_file.unlink(missing_ok=True) + except ClanCmdError as e: + log.exception("Failed to evaluate hardware-configuration.nix") + # Restore the backup file + print(f"Restoring backup file {backup_file}") + if backup_file: + backup_file.replace(hw_file) + # TODO: Undo the commit + + msg = "Invalid hardware-configuration.nix file" + raise ClanError( + msg, + description=f"Configuration at '{hw_file}' is invalid. Please check the file and try again.", + ) from e + + return opts.backend diff --git a/pkgs/clan-cli/clan_lib/machines/install.py b/pkgs/clan-cli/clan_lib/machines/install.py new file mode 100644 index 000000000..561fcd22d --- /dev/null +++ b/pkgs/clan-cli/clan_lib/machines/install.py @@ -0,0 +1,149 @@ +import logging +import os +from dataclasses import dataclass, field +from enum import Enum +from pathlib import Path +from tempfile import TemporaryDirectory + +from clan_cli.facts.generate import generate_facts +from clan_cli.machines.hardware import HardwareConfig +from clan_cli.vars.generate import generate_vars + +from clan_lib.api import API +from clan_lib.cmd import Log, RunOpts, run +from clan_lib.machines.machines import Machine +from clan_lib.nix import nix_shell +from clan_lib.ssh.remote import Remote + +log = logging.getLogger(__name__) + + +class BuildOn(Enum): + AUTO = "auto" + LOCAL = "local" + REMOTE = "remote" + + +@dataclass +class InstallOptions: + machine: Machine + kexec: str | None = None + debug: bool = False + no_reboot: bool = False + phases: str | None = None + build_on: BuildOn | None = None + nix_options: list[str] = field(default_factory=list) + update_hardware_config: HardwareConfig = HardwareConfig.NONE + password: str | None = None + identity_file: Path | None = None + use_tor: bool = False + + +@API.register +def install_machine(opts: InstallOptions, target_host: Remote) -> None: + machine = opts.machine + + machine.debug(f"installing {machine.name}") + + generate_facts([machine]) + generate_vars([machine]) + + with ( + TemporaryDirectory(prefix="nixos-install-") as _base_directory, + ): + base_directory = Path(_base_directory).resolve() + activation_secrets = base_directory / "activation_secrets" + upload_dir = activation_secrets / machine.secrets_upload_directory.lstrip("/") + upload_dir.mkdir(parents=True) + machine.secret_facts_store.upload(upload_dir) + machine.secret_vars_store.populate_dir( + upload_dir, phases=["activation", "users", "services"] + ) + + partitioning_secrets = base_directory / "partitioning_secrets" + partitioning_secrets.mkdir(parents=True) + machine.secret_vars_store.populate_dir( + partitioning_secrets, phases=["partitioning"] + ) + + if opts.password: + os.environ["SSHPASS"] = opts.password + + cmd = [ + "nixos-anywhere", + "--flake", + f"{machine.flake}#{machine.name}", + "--extra-files", + str(activation_secrets), + ] + + for path in partitioning_secrets.rglob("*"): + if path.is_file(): + cmd.extend( + [ + "--disk-encryption-keys", + str( + "/run/partitioning-secrets" + / path.relative_to(partitioning_secrets) + ), + str(path), + ] + ) + + if opts.no_reboot: + cmd.append("--no-reboot") + + if opts.phases: + cmd += ["--phases", str(opts.phases)] + + if opts.update_hardware_config is not HardwareConfig.NONE: + cmd.extend( + [ + "--generate-hardware-config", + str(opts.update_hardware_config.value), + str(opts.update_hardware_config.config_path(machine)), + ] + ) + + if opts.password: + cmd += [ + "--env-password", + "--ssh-option", + "IdentitiesOnly=yes", + ] + + if opts.identity_file: + cmd += ["-i", str(opts.identity_file)] + + if opts.build_on: + cmd += ["--build-on", opts.build_on.value] + + if target_host.port: + cmd += ["--ssh-port", str(target_host.port)] + if opts.kexec: + cmd += ["--kexec", opts.kexec] + + if opts.debug: + cmd.append("--debug") + + # Add nix options to nixos-anywhere + cmd.extend(opts.nix_options) + + cmd.append(target_host.target) + if opts.use_tor: + # nix copy does not support tor socks proxy + # cmd.append("--ssh-option") + # cmd.append("ProxyCommand=nc -x 127.0.0.1:9050 -X 5 %h %p") + cmd = nix_shell( + [ + "nixos-anywhere", + "tor", + ], + ["torify", *cmd], + ) + else: + cmd = nix_shell( + ["nixos-anywhere"], + cmd, + ) + run(cmd, RunOpts(log=Log.BOTH, prefix=machine.name, needs_user_terminal=True))