Merge pull request 'cleanup_install' (#4373) from Qubasa/clan-core:cleanup_install into main

Reviewed-on: https://git.clan.lol/clan/clan-core/pulls/4373

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

@@ -1,6 +1,7 @@
 import argparse
 import logging
 from pathlib import Path
+from typing import get_args
 
 from clan_lib.flake import require_flake
 from clan_lib.machines.hardware import (
@@ -10,6 +11,7 @@ from clan_lib.machines.hardware import (
 )
 from clan_lib.machines.machines import Machine
 from clan_lib.machines.suggestions import validate_machine_names
+from clan_lib.ssh.host_key import HostKeyCheck
 from clan_lib.ssh.remote import Remote
 
 from clan_cli.completions import add_dynamic_completer, complete_machines
@@ -59,7 +61,7 @@ def register_update_hardware_config(parser: argparse.ArgumentParser) -> None:
     )
     parser.add_argument(
         "--host-key-check",
-        choices=["strict", "ask", "tofu", "none"],
+        choices=list(get_args(HostKeyCheck)),
         default="ask",
         help="Host key (.ssh/known_hosts) check mode.",
     )

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

@@ -2,11 +2,13 @@ import argparse
 import logging
 import sys
 from pathlib import Path
+from typing import get_args
 
 from clan_lib.errors import ClanError
 from clan_lib.flake import require_flake
 from clan_lib.machines.install import BuildOn, InstallOptions, run_machine_install
 from clan_lib.machines.machines import Machine
+from clan_lib.ssh.host_key import HostKeyCheck
 from clan_lib.ssh.remote import Remote
 
 from clan_cli.completions import (
@@ -65,6 +67,15 @@ def install_command(args: argparse.Namespace) -> None:
             if ask != "y":
                 return None
 
+        if args.identity_file:
+            target_host = target_host.override(private_key=args.identity_file)
+
+        if password:
+            target_host = target_host.override(password=password)
+
+        if use_tor:
+            target_host = target_host.override(tor_socks=True)
+
         return run_machine_install(
             InstallOptions(
                 machine=machine,
@@ -72,11 +83,8 @@ def install_command(args: argparse.Namespace) -> None:
                 phases=args.phases,
                 debug=args.debug,
                 no_reboot=args.no_reboot,
-                build_on=BuildOn(args.build_on) if args.build_on is not None else None,
+                build_on=args.build_on if args.build_on is not None else None,
                 update_hardware_config=HardwareConfig(args.update_hardware_config),
-                password=password,
-                identity_file=args.identity_file,
-                use_tor=use_tor,
             ),
             target_host=target_host,
         )
@@ -99,13 +107,14 @@ def register_install_parser(parser: argparse.ArgumentParser) -> None:
     )
     parser.add_argument(
         "--host-key-check",
-        choices=["strict", "ask", "tofu", "none"],
+        choices=list(get_args(HostKeyCheck)),
         default="ask",
         help="Host key (.ssh/known_hosts) check mode.",
     )
+
     parser.add_argument(
         "--build-on",
-        choices=[x.value for x in BuildOn],
+        choices=list(get_args(BuildOn)),
         default=None,
         help="where to build the NixOS configuration",
     )

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

@@ -1,6 +1,7 @@
 import argparse
 import logging
 import sys
+from typing import get_args
 
 from clan_lib.async_run import AsyncContext, AsyncOpts, AsyncRuntime
 from clan_lib.errors import ClanError
@@ -12,6 +13,7 @@ from clan_lib.machines.machines import Machine
 from clan_lib.machines.suggestions import validate_machine_names
 from clan_lib.machines.update import run_machine_update
 from clan_lib.nix import nix_config
+from clan_lib.ssh.host_key import HostKeyCheck
 from clan_lib.ssh.remote import Remote
 
 from clan_cli.completions import (
@@ -174,7 +176,7 @@ def register_update_parser(parser: argparse.ArgumentParser) -> None:
 
     parser.add_argument(
         "--host-key-check",
-        choices=["strict", "ask", "tofu", "none"],
+        choices=list(get_args(HostKeyCheck)),
         default="ask",
         help="Host key (.ssh/known_hosts) check mode.",
     )

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

@@ -4,7 +4,7 @@ import logging
 import textwrap
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any
+from typing import Any, get_args
 
 from clan_lib.cmd import run
 from clan_lib.errors import ClanError
@@ -220,7 +220,7 @@ def register_parser(parser: argparse.ArgumentParser) -> None:
 
     parser.add_argument(
         "--host-key-check",
-        choices=["strict", "ask", "tofu", "none"],
+        choices=list(get_args(HostKeyCheck)),
         default="tofu",
         help="Host key (.ssh/known_hosts) check mode.",
     )

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

@@ -1,9 +1,9 @@
 import logging
 import os
 from dataclasses import dataclass
-from enum import Enum
 from pathlib import Path
 from tempfile import TemporaryDirectory
+from typing import Literal
 
 from clan_cli.facts.generate import generate_facts
 from clan_cli.machines.hardware import HardwareConfig
@@ -18,10 +18,7 @@ from clan_lib.ssh.remote import Remote
 log = logging.getLogger(__name__)
 
 
-class BuildOn(Enum):
-    AUTO = "auto"
-    LOCAL = "local"
-    REMOTE = "remote"
+BuildOn = Literal["auto", "local", "remote"]
 
 
 @dataclass
@@ -33,9 +30,6 @@ class InstallOptions:
     phases: str | None = None
     build_on: BuildOn | None = None
     update_hardware_config: HardwareConfig = HardwareConfig.NONE
-    password: str | None = None
-    identity_file: Path | None = None
-    use_tor: bool = False
 
 
 @API.register
@@ -75,8 +69,8 @@ def run_machine_install(opts: InstallOptions, target_host: Remote) -> None:
             machine.name, partitioning_secrets, phases=["partitioning"]
         )
 
-        if opts.password:
-            os.environ["SSHPASS"] = opts.password
+        if target_host.password:
+            os.environ["SSHPASS"] = target_host.password
 
         cmd = [
             "nixos-anywhere",
@@ -114,18 +108,18 @@ def run_machine_install(opts: InstallOptions, target_host: Remote) -> None:
                 ]
             )
 
-        if opts.password:
+        if target_host.password:
             cmd += [
                 "--env-password",
                 "--ssh-option",
                 "IdentitiesOnly=yes",
             ]
 
-        if opts.identity_file:
-            cmd += ["-i", str(opts.identity_file)]
+        if target_host.private_key:
+            cmd += ["-i", str(target_host.private_key)]
 
         if opts.build_on:
-            cmd += ["--build-on", opts.build_on.value]
+            cmd += ["--build-on", opts.build_on]
 
         if target_host.port:
             cmd += ["--ssh-port", str(target_host.port)]
@@ -139,7 +133,7 @@ def run_machine_install(opts: InstallOptions, target_host: Remote) -> None:
         cmd.extend(opts.machine.flake.nix_options or [])
 
         cmd.append(target_host.target)
-        if opts.use_tor:
+        if target_host.tor_socks:
             # 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")

pkgs/clan-cli/clan_lib/ssh/host_key.py

@@ -7,7 +7,7 @@ from clan_lib.errors import ClanError
 HostKeyCheck = Literal[
     "strict",  # Strictly check ssh host keys, prompt for unknown ones
     "ask",  # Ask for confirmation on first use
-    "tofu",  # Trust on ssh keys on first use
+    "accept-new",  # Trust on ssh keys on first use
     "none",  # Do not check ssh host keys
 ]
 
@@ -21,7 +21,7 @@ def hostkey_to_ssh_opts(host_key_check: HostKeyCheck) -> list[str]:
             return ["-o", "StrictHostKeyChecking=yes"]
         case "ask":
             return []
-        case "tofu":
+        case "accept-new" | "tofu":
             return ["-o", "StrictHostKeyChecking=accept-new"]
         case "none":
             return [

pkgs/clan-cli/clan_lib/ssh/remote.py

@@ -531,7 +531,7 @@ def check_machine_ssh_reachable(
         f"Checking SSH reachability for {remote.target} on port {remote.port or 22}",
     )
 
-    address_family = socket.AF_INET6 if ":" in remote.address else socket.AF_INET
+    address_family = socket.AF_INET6 if remote.is_ipv6() else socket.AF_INET
     for _ in range(opts.retries):
         with socket.socket(address_family, socket.SOCK_STREAM) as sock:
             sock.settimeout(opts.timeout)