clan-cli: add machines install

pkgs/clan-cli/clan_cli/machines/__init__.py

@@ -3,6 +3,7 @@ import argparse
 
 from .create import register_create_parser
 from .delete import register_delete_parser
+from .install import register_install_parser
 from .list import register_list_parser
 from .update import register_update_parser
 
@@ -27,3 +28,6 @@ def register_parser(parser: argparse.ArgumentParser) -> None:
 
     list_parser = subparser.add_parser("list", help="List machines")
     register_list_parser(list_parser)
+
+    install_parser = subparser.add_parser("install", help="Install a machine")
+    register_install_parser(install_parser)

pkgs/clan-cli/clan_cli/machines/install.py

@@ -0,0 +1,79 @@
+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
+
+
+def install_nixos(h: Host, clan_dir: Path) -> None:
+    target_host = f"{h.user or 'root'}@{h.host}"
+
+    flake_attr = h.meta.get("flake_attr", "")
+
+    run_generate_secrets(h.meta["generateSecrets"], clan_dir)
+
+    with TemporaryDirectory() as tmpdir_:
+        tmpdir = Path(tmpdir_)
+        get_decrypted_secrets(
+            h.meta["uploadSecrets"],
+            clan_dir,
+            target_directory=tmpdir / h.meta["secretsUploadDirectory"].lstrip("/"),
+        )
+
+        subprocess.run(
+            nix_shell(
+                ["nixos-anywhere"],
+                [
+                    "nixos-anywhere",
+                    "-f",
+                    f"{clan_dir}#{flake_attr}",
+                    "-t",
+                    "--no-reboot",
+                    "--extra-files",
+                    str(tmpdir),
+                    target_host,
+                ],
+            ),
+            check=True,
+        )
+
+
+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)
+
+    install_nixos(host, clan_dir)
+
+
+def register_install_parser(parser: argparse.ArgumentParser) -> None:
+    parser.add_argument(
+        "machine",
+        type=str,
+        help="machine to install",
+    )
+    parser.add_argument(
+        "target_host",
+        type=str,
+        help="ssh address to install to in the form of user@host:2222",
+    )
+
+    parser.set_defaults(func=install)

pkgs/clan-cli/clan_cli/secrets/upload.py

@@ -52,8 +52,8 @@ def get_deployment_info(machine: str, clan_dir: Path) -> dict:
     return json.load(open(proc.stdout.strip()))
 
 
-def run_upload_secrets(
+def get_decrypted_secrets(
-    flake_attr: str, clan_dir: Path, target: str, target_directory: str
+    flake_attr: str, clan_dir: Path, target_directory: Path
 ) -> None:
     env = os.environ.copy()
     env["CLAN_DIR"] = str(clan_dir)