From 01087e2da80863d28c6c8c6ee52b51ba0c0e7630 Mon Sep 17 00:00:00 2001 From: lassulus Date: Wed, 4 Oct 2023 15:32:04 +0200 Subject: [PATCH] clan_cli: refactor secrets code into Machine class --- pkgs/clan-cli/clan_cli/machines/install.py | 45 +++----- pkgs/clan-cli/clan_cli/machines/machines.py | 108 ++++++++++++++++++++ pkgs/clan-cli/clan_cli/machines/update.py | 58 ++++------- pkgs/clan-cli/clan_cli/secrets/generate.py | 39 ++----- pkgs/clan-cli/clan_cli/secrets/upload.py | 49 ++------- pkgs/clan-cli/clan_cli/ssh/__init__.py | 2 +- 6 files changed, 157 insertions(+), 144 deletions(-) create mode 100644 pkgs/clan-cli/clan_cli/machines/machines.py diff --git a/pkgs/clan-cli/clan_cli/machines/install.py b/pkgs/clan-cli/clan_cli/machines/install.py index 78531a3df..1ba1dcdae 100644 --- a/pkgs/clan-cli/clan_cli/machines/install.py +++ b/pkgs/clan-cli/clan_cli/machines/install.py @@ -1,30 +1,24 @@ import argparse -import json import subprocess from pathlib import Path from tempfile import TemporaryDirectory -from ..dirs import get_clan_flake_toplevel -from ..nix import nix_build, nix_config, nix_shell -from ..secrets.generate import run_generate_secrets -from ..secrets.upload import get_decrypted_secrets -from ..ssh import Host, parse_deployment_address +from ..machines.machines import Machine +from ..nix import nix_shell +from ..secrets.generate import generate_secrets -def install_nixos(h: Host, clan_dir: Path) -> None: +def install_nixos(machine: Machine) -> None: + h = machine.host target_host = f"{h.user or 'root'}@{h.host}" flake_attr = h.meta.get("flake_attr", "") - run_generate_secrets(h.meta["generateSecrets"], clan_dir) + generate_secrets(machine) with TemporaryDirectory() as tmpdir_: tmpdir = Path(tmpdir_) - get_decrypted_secrets( - h.meta["uploadSecrets"], - clan_dir, - target_directory=tmpdir / h.meta["secretsUploadDirectory"].lstrip("/"), - ) + machine.upload_secrets(tmpdir / machine.secrets_upload_directory) subprocess.run( nix_shell( @@ -32,7 +26,7 @@ def install_nixos(h: Host, clan_dir: Path) -> None: [ "nixos-anywhere", "-f", - f"{clan_dir}#{flake_attr}", + f"{machine.clan_dir}#{flake_attr}", "-t", "--no-reboot", "--extra-files", @@ -44,24 +38,11 @@ def install_nixos(h: Host, clan_dir: Path) -> None: ) -def install(args: argparse.Namespace) -> None: - clan_dir = get_clan_flake_toplevel() - config = nix_config() - system = config["system"] - json_file = subprocess.run( - nix_build( - [ - f'{clan_dir}#clanInternals.machines."{system}"."{args.machine}".config.system.clan.deployment.file' - ] - ), - stdout=subprocess.PIPE, - check=True, - text=True, - ).stdout.strip() - machine_json = json.loads(Path(json_file).read_text()) - host = parse_deployment_address(args.machine, args.target_host, machine_json) +def install_command(args: argparse.Namespace) -> None: + machine = Machine(args.machine) + machine.deployment_address = args.target_host - install_nixos(host, clan_dir) + install_nixos(machine) def register_install_parser(parser: argparse.ArgumentParser) -> None: @@ -76,4 +57,4 @@ def register_install_parser(parser: argparse.ArgumentParser) -> None: help="ssh address to install to in the form of user@host:2222", ) - parser.set_defaults(func=install) + parser.set_defaults(func=install_command) diff --git a/pkgs/clan-cli/clan_cli/machines/machines.py b/pkgs/clan-cli/clan_cli/machines/machines.py new file mode 100644 index 000000000..215b53ff7 --- /dev/null +++ b/pkgs/clan-cli/clan_cli/machines/machines.py @@ -0,0 +1,108 @@ +import json +import os +import subprocess +import sys +from pathlib import Path +from typing import Optional + +from ..dirs import get_clan_flake_toplevel +from ..nix import nix_build, nix_config, nix_eval +from ..ssh import Host, parse_deployment_address + + +def build_machine_data(machine_name: str, clan_dir: Path) -> dict: + config = nix_config() + system = config["system"] + + outpath = subprocess.run( + nix_build( + [ + f'path:{clan_dir}#clanInternals.machines."{system}"."{machine_name}".config.system.clan.deployment.file' + ] + ), + stdout=subprocess.PIPE, + check=True, + text=True, + ).stdout.strip() + return json.loads(Path(outpath).read_text()) + + +class Machine: + def __init__( + self, + name: str, + clan_dir: Optional[Path] = None, + machine_data: Optional[dict] = None, + ) -> None: + """ + Creates a Machine + @name: the name of the machine + @clan_dir: the directory of the clan, optional, if not set it will be determined from the current working directory + @machine_json: can be optionally used to skip evaluation of the machine, location of the json file with machine data + """ + self.name = name + if clan_dir is None: + self.clan_dir = get_clan_flake_toplevel() + else: + self.clan_dir = clan_dir + + if machine_data is None: + self.machine_data = build_machine_data(name, self.clan_dir) + else: + self.machine_data = machine_data + + self.deployment_address = self.machine_data["deploymentAddress"] + self.upload_secrets = self.machine_data["uploadSecrets"] + self.generate_secrets = self.machine_data["generateSecrets"] + self.secrets_upload_directory = self.machine_data["secretsUploadDirectory"] + + @property + def host(self) -> Host: + return parse_deployment_address( + self.name, self.deployment_address, meta={"machine": self} + ) + + def run_upload_secrets(self, secrets_dir: Path) -> None: + """ + Upload the secrets to the provided directory + @secrets_dir: the directory to store the secrets in + """ + env = os.environ.copy() + env["CLAN_DIR"] = str(self.clan_dir) + env["PYTHONPATH"] = str( + ":".join(sys.path) + ) # TODO do this in the clanCore module + env["SECRETS_DIR"] = str(secrets_dir) + subprocess.run( + [self.upload_secrets], + env=env, + check=True, + stdout=subprocess.PIPE, + text=True, + ) + + def eval_nix(self, attr: str) -> str: + """ + eval a nix attribute of the machine + @attr: the attribute to get + """ + output = subprocess.run( + nix_eval([f"path:{self.clan_dir}#{attr}"]), + stdout=subprocess.PIPE, + check=True, + text=True, + ).stdout.strip() + return output + + def build_nix(self, attr: str) -> Path: + """ + build a nix attribute of the machine + @attr: the attribute to get + """ + outpath = subprocess.run( + nix_build([f"path:{self.clan_dir}#{attr}"]), + stdout=subprocess.PIPE, + check=True, + text=True, + ).stdout.strip() + return Path(outpath) diff --git a/pkgs/clan-cli/clan_cli/machines/update.py b/pkgs/clan-cli/clan_cli/machines/update.py index d74ff0641..ff4e74994 100644 --- a/pkgs/clan-cli/clan_cli/machines/update.py +++ b/pkgs/clan-cli/clan_cli/machines/update.py @@ -3,12 +3,12 @@ import json import os import subprocess from pathlib import Path -from typing import Any from ..dirs import get_clan_flake_toplevel +from ..machines.machines import Machine from ..nix import nix_build, nix_command, nix_config -from ..secrets.generate import run_generate_secrets -from ..secrets.upload import run_upload_secrets +from ..secrets.generate import generate_secrets +from ..secrets.upload import upload_secrets from ..ssh import Host, HostGroup, HostKeyCheck, parse_deployment_address @@ -40,13 +40,8 @@ def deploy_nixos(hosts: HostGroup, clan_dir: Path) -> None: flake_attr = h.meta.get("flake_attr", "") - run_generate_secrets(h.meta["generateSecrets"], clan_dir) - run_upload_secrets( - h.meta["uploadSecrets"], - clan_dir, - target=target, - target_directory=h.meta["secretsUploadDirectory"], - ) + generate_secrets(h.meta["machine"]) + upload_secrets(h.meta["machine"]) target_host = h.meta.get("target_host") if target_host: @@ -81,49 +76,36 @@ def deploy_nixos(hosts: HostGroup, clan_dir: Path) -> None: hosts.run_function(deploy) -def build_json(targets: list[str]) -> list[dict[str, Any]]: - outpaths = subprocess.run( - nix_build(targets), +# function to speedup eval if we want to evauluate all machines +def get_all_machines(clan_dir: Path) -> HostGroup: + config = nix_config() + system = config["system"] + machines_json = subprocess.run( + nix_build([f'{clan_dir}#clanInternals.all-machines-json."{system}"']), stdout=subprocess.PIPE, check=True, text=True, ).stdout - parsed = [] - for outpath in outpaths.splitlines(): - parsed.append(json.loads(Path(outpath).read_text())) - return parsed - -def get_all_machines(clan_dir: Path) -> HostGroup: - config = nix_config() - system = config["system"] - what = f'{clan_dir}#clanInternals.all-machines-json."{system}"' - machines = build_json([what])[0] + machines = json.loads(Path(machines_json).read_text()) hosts = [] - for name, machine in machines.items(): + for name, machine_data in machines.items(): + # very hacky. would be better to do a MachinesGroup instead host = parse_deployment_address( - name, machine["deploymentAddress"], meta=machine + name, + machine_data["deploymentAddress"], + meta={"machine": Machine(name=name, machine_data=machine_data)}, ) hosts.append(host) return HostGroup(hosts) def get_selected_machines(machine_names: list[str], clan_dir: Path) -> HostGroup: - config = nix_config() - system = config["system"] - what = [] - for name in machine_names: - what.append( - f'{clan_dir}#clanInternals.machines."{system}"."{name}".config.system.clan.deployment.file' - ) - machines = build_json(what) hosts = [] - for i, machine in enumerate(machines): - host = parse_deployment_address( - machine_names[i], machine["deploymentAddress"], machine - ) - hosts.append(host) + for name in machine_names: + machine = Machine(name=name, clan_dir=clan_dir) + hosts.append(machine.host) return HostGroup(hosts) diff --git a/pkgs/clan-cli/clan_cli/secrets/generate.py b/pkgs/clan-cli/clan_cli/secrets/generate.py index 8433200f0..c83f5592c 100644 --- a/pkgs/clan-cli/clan_cli/secrets/generate.py +++ b/pkgs/clan-cli/clan_cli/secrets/generate.py @@ -1,45 +1,24 @@ import argparse import logging import os -import shlex import subprocess import sys -from pathlib import Path from clan_cli.errors import ClanError -from ..dirs import get_clan_flake_toplevel -from ..nix import nix_build, nix_config +from ..machines.machines import Machine log = logging.getLogger(__name__) -def build_generate_script(machine: str, clan_dir: Path) -> str: - config = nix_config() - system = config["system"] - - cmd = nix_build( - [ - f'path:{clan_dir}#clanInternals.machines."{system}"."{machine}".config.system.clan.generateSecrets' - ] - ) - proc = subprocess.run(cmd, stdout=subprocess.PIPE, text=True) - if proc.returncode != 0: - raise ClanError( - f"failed to generate secrets:\n{shlex.join(cmd)}\nexited with {proc.returncode}" - ) - - return proc.stdout.strip() - - -def run_generate_secrets(secret_generator_script: str, clan_dir: Path) -> None: +def generate_secrets(machine: Machine) -> None: env = os.environ.copy() - env["CLAN_DIR"] = str(clan_dir) + env["CLAN_DIR"] = str(machine.clan_dir) env["PYTHONPATH"] = ":".join(sys.path) # TODO do this in the clanCore module - print(f"generating secrets... {secret_generator_script}") + print(f"generating secrets... {machine.generate_secrets}") proc = subprocess.run( - [secret_generator_script], + [machine.generate_secrets], env=env, ) @@ -51,13 +30,9 @@ def run_generate_secrets(secret_generator_script: str, clan_dir: Path) -> None: print("successfully generated secrets") -def generate(machine: str) -> None: - clan_dir = get_clan_flake_toplevel() - run_generate_secrets(build_generate_script(machine, clan_dir), clan_dir) - - def generate_command(args: argparse.Namespace) -> None: - generate(args.machine) + machine = Machine(args.machine) + generate_secrets(machine) def register_generate_parser(parser: argparse.ArgumentParser) -> None: diff --git a/pkgs/clan-cli/clan_cli/secrets/upload.py b/pkgs/clan-cli/clan_cli/secrets/upload.py index b7c8e91be..53378daee 100644 --- a/pkgs/clan-cli/clan_cli/secrets/upload.py +++ b/pkgs/clan-cli/clan_cli/secrets/upload.py @@ -1,17 +1,14 @@ import argparse import json import logging -import os import shlex import subprocess -import sys from pathlib import Path from tempfile import TemporaryDirectory -from ..dirs import get_clan_flake_toplevel from ..errors import ClanError +from ..machines.machines import Machine from ..nix import nix_build, nix_config, nix_shell -from ..ssh import parse_deployment_address log = logging.getLogger(__name__) @@ -52,31 +49,13 @@ def get_deployment_info(machine: str, clan_dir: Path) -> dict: return json.load(open(proc.stdout.strip())) -def get_decrypted_secrets( - flake_attr: str, clan_dir: Path, target_directory: Path -) -> None: - env = os.environ.copy() - env["CLAN_DIR"] = str(clan_dir) - env["PYTHONPATH"] = ":".join(sys.path) # TODO do this in the clanCore module - print(f"uploading secrets... {flake_attr}") +def upload_secrets(machine: Machine) -> None: with TemporaryDirectory() as tempdir_: tempdir = Path(tempdir_) - env["SECRETS_DIR"] = str(tempdir) - proc = subprocess.run( - [flake_attr], - env=env, - check=True, - stdout=subprocess.PIPE, - text=True, - ) + machine.run_upload_secrets(tempdir) + host = machine.host - if proc.returncode != 0: - log.error("Stdout: %s", proc.stdout) - log.error("Stderr: %s", proc.stderr) - raise ClanError("failed to upload secrets") - - h = parse_deployment_address(flake_attr, target) - ssh_cmd = h.ssh_cmd() + ssh_cmd = host.ssh_cmd() subprocess.run( nix_shell( ["rsync"], @@ -87,28 +66,16 @@ def get_decrypted_secrets( "-az", "--delete", f"{str(tempdir)}/", - f"{h.user}@{h.host}:{target_directory}/", + f"{host.user}@{host.host}:{machine.secrets_upload_directory}/", ], ), check=True, ) -def upload_secrets(machine: str) -> None: - clan_dir = get_clan_flake_toplevel() - deployment_info = get_deployment_info(machine, clan_dir) - address = deployment_info.get("deploymentAddress", "") - secrets_upload_directory = deployment_info.get("secretsUploadDirectory", "") - run_upload_secrets( - build_upload_script(machine, clan_dir), - clan_dir, - address, - secrets_upload_directory, - ) - - def upload_command(args: argparse.Namespace) -> None: - upload_secrets(args.machine) + machine = Machine(args.machine) + upload_secrets(machine) def register_upload_parser(parser: argparse.ArgumentParser) -> None: diff --git a/pkgs/clan-cli/clan_cli/ssh/__init__.py b/pkgs/clan-cli/clan_cli/ssh/__init__.py index 729bbe329..2aba7e480 100644 --- a/pkgs/clan-cli/clan_cli/ssh/__init__.py +++ b/pkgs/clan-cli/clan_cli/ssh/__init__.py @@ -759,7 +759,7 @@ class HostGroup: def parse_deployment_address( - machine_name: str, host: str, meta: dict[str, str] = {} + machine_name: str, host: str, meta: dict[str, Any] = {} ) -> Host: parts = host.split("@") user: Optional[str] = None