yadunut/clan-core
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Merge pull request 'clan-cli: Move hardware.py to clan_lib/machines part2' (#4066) from Qubasa/clan-core:move_to_clan_lib2 into main

Browse Source 
Reviewed-on: https://git.clan.lol/clan/clan-core/pulls/4066
This commit is contained in:
Luis Hebendanz
2025-06-24 09:19:41 +00:00
parent 74abda09e7 168e1d8c96
commit feb47ffe33
6 changed files with 308 additions and 322 deletions
Download Patch File Download Diff File Expand all files Collapse all files

151
pkgs/clan-cli/clan_cli/machines/hardware.py
View File

@@ -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)

141
pkgs/clan-cli/clan_cli/machines/install.py
View File

@@ -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

34
pkgs/clan-cli/clan_cli/ssh/results.py
View File

@@ -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]]

3
pkgs/clan-cli/clan_lib/api/disk.py
View File

@@ -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__)

152
pkgs/clan-cli/clan_lib/machines/hardware.py Normal file
View File

@@ -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

149
pkgs/clan-cli/clan_lib/machines/install.py Normal file
View File

@@ -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))