clan_cli: refactor secrets code into Machine class

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:

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: