Merge pull request 'clan-cli: add --json and --png flags to machine install' (#1320) from a-kenji-feat/clan-install-json into main

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

@@ -1,6 +1,8 @@
 import argparse
 import importlib
+import json
 import logging
+import os
 from dataclasses import dataclass
 from pathlib import Path
 from tempfile import TemporaryDirectory
@@ -9,12 +11,20 @@ from ..cmd import Log, run
 from ..facts.generate import generate_facts
 from ..machines.machines import Machine
 from ..nix import nix_shell
+from ..ssh.cli import is_ipv6, is_reachable, qrcode_scan
 
 log = logging.getLogger(__name__)
 
 
+class ClanError(Exception):
+    pass
+
+
 def install_nixos(
-    machine: Machine, kexec: str | None = None, debug: bool = False
+    machine: Machine,
+    kexec: str | None = None,
+    debug: bool = False,
+    password: str | None = None,
 ) -> None:
     secret_facts_module = importlib.import_module(machine.secret_facts_module)
     log.info(f"installing {machine.name}")
@@ -36,6 +46,9 @@ def install_nixos(
         upload_dir.mkdir(parents=True)
         secret_facts_store.upload(upload_dir)
 
+        if password:
+            os.environ["SSHPASS"] = password
+
         cmd = [
             "nixos-anywhere",
             "--flake",
@@ -44,6 +57,14 @@ def install_nixos(
             "--extra-files",
             str(tmpdir),
         ]
+
+        if password:
+            cmd += [
+                "--env-password",
+                "--ssh-option",
+                "IdentitiesOnly=yes",
+            ]
+
         if machine.target_host.port:
             cmd += ["--ssh-port", str(machine.target_host.port)]
         if kexec:
@@ -69,16 +90,38 @@ class InstallOptions:
     kexec: str | None
     confirm: bool
     debug: bool
+    json_ssh_deploy: dict[str, str] | None
 
 
 def install_command(args: argparse.Namespace) -> None:
+    json_ssh_deploy = None
+    if args.json:
+        json_file = Path(args.json)
+        if json_file.is_file():
+            json_ssh_deploy = json.loads(json_file.read_text())
+        else:
+            json_ssh_deploy = json.loads(args.json)
+    elif args.png:
+        json_ssh_deploy = json.loads(qrcode_scan(args.png))
+
+    if not json_ssh_deploy and not args.target_host:
+        raise ClanError("No target host provided, please provide a target host.")
+
+    if json_ssh_deploy:
+        target_host = f"root@{find_reachable_host_from_deploy_json(json_ssh_deploy)}"
+        password = json_ssh_deploy["pass"]
+    else:
+        target_host = args.target_host
+        password = None
+
     opts = InstallOptions(
         flake=args.flake,
         machine=args.machine,
-        target_host=args.target_host,
+        target_host=target_host,
         kexec=args.kexec,
         confirm=not args.yes,
         debug=args.debug,
+        json_ssh_deploy=json_ssh_deploy,
     )
     machine = Machine(opts.machine, flake=opts.flake)
     machine.target_host_address = opts.target_host
@@ -88,7 +131,27 @@ def install_command(args: argparse.Namespace) -> None:
         if ask != "y":
             return
 
-    install_nixos(machine, kexec=opts.kexec, debug=opts.debug)
+    install_nixos(machine, kexec=opts.kexec, debug=opts.debug, password=password)
+
+
+def find_reachable_host_from_deploy_json(deploy_json: dict[str, str]) -> str:
+    host = None
+    for addr in deploy_json["addrs"]:
+        if is_reachable(addr):
+            if is_ipv6(addr):
+                host = f"[{addr}]"
+            else:
+                host = addr
+            break
+    if not host:
+        raise ClanError(
+            f"""
+            Could not reach any of the host addresses provided in the json string.
+            Please doublecheck if they are reachable from your machine.
+            Try `ping [ADDR]` with one of the addresses: {deploy_json['addrs']}
+            """
+        )
+    return host
 
 
 def register_install_parser(parser: argparse.ArgumentParser) -> None:
@@ -117,6 +180,18 @@ def register_install_parser(parser: argparse.ArgumentParser) -> None:
     parser.add_argument(
         "target_host",
         type=str,
+        nargs="?",
         help="ssh address to install to in the form of user@host:2222",
     )
+    group = parser.add_mutually_exclusive_group(required=False)
+    group.add_argument(
+        "-j",
+        "--json",
+        help="specify the json file for ssh data (generated by starting the clan installer)",
+    )
+    group.add_argument(
+        "-P",
+        "--png",
+        help="specify the json file for ssh data as the qrcode image (generated by starting the clan installer)",
+    )
     parser.set_defaults(func=install_command)

pkgs/clan-cli/clan_cli/ssh/cli.py

@@ -1,4 +1,5 @@
 import argparse
+import ipaddress
 import json
 import logging
 import socket
@@ -96,6 +97,13 @@ def connect_ssh_from_json(ssh_data: dict[str, str]) -> None:
     ssh(host=ssh_data["tor"], password=ssh_data["pass"], torify=True)
 
 
+def is_ipv6(ip: str) -> bool:
+    try:
+        return isinstance(ipaddress.ip_address(ip), ipaddress.IPv6Address)
+    except ValueError:
+        return False
+
+
 def main(args: argparse.Namespace) -> None:
     if args.json:
         json_file = Path(args.json)