270 lines
8.3 KiB
Python
270 lines
8.3 KiB
Python
import json
|
|
import logging
|
|
import os
|
|
from dataclasses import dataclass
|
|
from enum import Enum
|
|
from pathlib import Path
|
|
from typing import TypedDict
|
|
|
|
from clan_lib.api import API
|
|
from clan_lib.cmd import Log, 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, nix_shell
|
|
from clan_lib.ssh.create import create_secret_key_nixos_anywhere
|
|
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
|
|
|
|
|
|
def get_machine_target_platform(machine: Machine) -> str | None:
|
|
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 = HardwareConfig.NIXOS_FACTER
|
|
password: str | None = None
|
|
debug: bool = False
|
|
|
|
|
|
@API.register
|
|
def run_machine_hardware_info_init(
|
|
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)
|
|
|
|
cmd = [
|
|
"nixos-anywhere",
|
|
"--flake",
|
|
f"{machine.flake}#{machine.name}",
|
|
"--phases",
|
|
"kexec",
|
|
"--generate-hardware-config",
|
|
str(opts.backend.value),
|
|
str(opts.backend.config_path(machine)),
|
|
]
|
|
|
|
environ = os.environ.copy()
|
|
if target_host.password:
|
|
cmd += [
|
|
"--env-password",
|
|
"--ssh-option",
|
|
"IdentitiesOnly=yes",
|
|
]
|
|
environ["SSHPASS"] = target_host.password
|
|
|
|
if target_host.private_key:
|
|
cmd += ["--ssh-option", f"IdentityFile={target_host.private_key}"]
|
|
|
|
if target_host.port:
|
|
cmd += ["--ssh-port", str(target_host.port)]
|
|
|
|
key_pair = create_secret_key_nixos_anywhere()
|
|
cmd += ["-i", str(key_pair.private)]
|
|
|
|
backup_file = None
|
|
if hw_file.exists():
|
|
backup_file = hw_file.with_suffix(".bak")
|
|
hw_file.replace(backup_file)
|
|
|
|
if opts.debug:
|
|
cmd += ["--debug"]
|
|
|
|
# REMOVEME when nixos-anywhere > 1.12.0
|
|
# In 1.12.0 and earlier, nixos-anywhere doesn't pass Nix options when attempting to get substituters
|
|
# which leads to the installation test failing with the error of not being able to substitute flake-parts
|
|
# see: https://github.com/nix-community/nixos-anywhere/pull/596
|
|
if "CLAN_TEST_STORE" in environ:
|
|
cmd += ["--no-use-machine-substituters"]
|
|
|
|
# Add nix options to nixos-anywhere
|
|
cmd.extend(opts.machine.flake.nix_options or [])
|
|
|
|
cmd += [target_host.target]
|
|
cmd = nix_shell(
|
|
["nixos-anywhere"],
|
|
cmd,
|
|
)
|
|
|
|
run(
|
|
cmd,
|
|
RunOpts(
|
|
log=Log.BOTH, prefix=machine.name, needs_user_terminal=True, env=environ
|
|
),
|
|
)
|
|
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.name}/{hw_file.name}: update hardware configuration",
|
|
)
|
|
try:
|
|
get_machine_target_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
|
|
|
|
|
|
@API.register
|
|
def run_machine_hardware_info_update(
|
|
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.host_connection() 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.name}/{hw_file.name}: update hardware configuration",
|
|
)
|
|
try:
|
|
get_machine_target_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 get_machine_hardware_config(machine: Machine) -> HardwareConfig:
|
|
"""Detect and return the full hardware configuration for the given machine.
|
|
|
|
Returns:
|
|
HardwareConfig: Structured hardware information, or None if unavailable.
|
|
|
|
"""
|
|
return HardwareConfig.detect_type(machine)
|
|
|
|
|
|
class MachineHardwareBrief(TypedDict):
|
|
hardware_config: HardwareConfig
|
|
platform: str | None
|
|
|
|
|
|
@API.register
|
|
def get_machine_hardware_summary(machine: Machine) -> MachineHardwareBrief:
|
|
"""Return a high-level summary of hardware config and platform type."""
|
|
return {
|
|
"hardware_config": get_machine_hardware_config(machine),
|
|
"platform": get_machine_target_platform(machine),
|
|
}
|