Merge pull request 'update_hardware_config: use host.run rather than adhoc ssh command' (#3486) from control-master into main

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

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

@@ -7,14 +7,14 @@ from pathlib import Path
 
 from clan_lib.api import API
 
-from clan_cli.cmd import RunOpts, run, run_no_stdout
+from clan_cli.cmd import RunOpts, run_no_stdout
 from clan_cli.completions import add_dynamic_completer, complete_machines
 from clan_cli.dirs import specific_machine_dir
 from clan_cli.errors import ClanCmdError, ClanError
 from clan_cli.flake import Flake
 from clan_cli.git import commit_file
 from clan_cli.machines.machines import Machine
-from clan_cli.nix import nix_config, nix_eval, nix_shell
+from clan_cli.nix import nix_config, nix_eval
 
 from .types import machine_name_type
 
@@ -119,6 +119,10 @@ def generate_machine_hardware_info(opts: HardwareGenerateOptions) -> HardwareCon
     """
 
     machine = Machine(opts.machine, flake=opts.flake)
+
+    if opts.keyfile is not None:
+        machine.private_key = Path(opts.keyfile)
+
     if opts.target_host is not None:
         machine.override_target_host = opts.target_host
 
@@ -136,41 +140,19 @@ def generate_machine_hardware_info(opts: HardwareGenerateOptions) -> HardwareCon
         ]
 
     host = machine.target_host
-
+    host.ssh_options["StrictHostKeyChecking"] = "accept-new"
-    # HACK: to make non-root user work
+    host.ssh_options["UserKnownHostsFile"] = "/dev/null"
-    if host.user != "root":
-        config_command.insert(0, "sudo")
-
-    deps = ["openssh"]
     if opts.password:
-        deps += ["sshpass"]
+        host.password = opts.password
 
-    cmd = nix_shell(
+    out = host.run(config_command, become_root=True, opts=RunOpts(check=False))
-        deps,
-        [
-            *(["sshpass", "-p", opts.password] if opts.password else []),
-            "ssh",
-            *(["-i", f"{opts.keyfile}"] if opts.keyfile else []),
-            # Disable known hosts file
-            "-o",
-            "UserKnownHostsFile=/dev/null",
-            # Disable strict host key checking. The GUI user cannot type "yes" into the ssh terminal.
-            "-o",
-            "StrictHostKeyChecking=accept-new",
-            *(
-                ["-p", str(machine.target_host.port)]
-                if machine.target_host.port
-                else []
-            ),
-            host.target,
-            *config_command,
-        ],
-    )
-    out = run(cmd, RunOpts(needs_user_terminal=True, prefix=machine.name, check=False))
     if out.returncode != 0:
         if "nixos-facter" in out.stderr and "not found" in out.stderr:
             machine.error(str(out.stderr))
-            msg = "Please use our custom nixos install images. nixos-factor only works on nixos / clan systems currently."
+            msg = (
+                "Please use our custom nixos install images from https://github.com/nix-community/nixos-images/releases/tag/nixos-unstable. "
+                "nixos-factor only works on nixos / clan systems currently."
+            )
             raise ClanError(msg)
 
         machine.error(str(out))

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

@@ -87,7 +87,7 @@ def ssh_shell_from_deploy(
     deploy_info: DeployInfo, runtime: AsyncRuntime, host_key_check: HostKeyCheck
 ) -> None:
     if host := find_reachable_host(deploy_info, host_key_check):
-        host.connect_ssh_shell(password=deploy_info.pwd)
+        host.interactive_ssh()
     else:
         log.info("Could not reach host via clearnet 'addrs'")
         log.info(f"Trying to reach host via tor '{deploy_info.tor}'")
@@ -96,8 +96,7 @@ def ssh_shell_from_deploy(
             msg = "No tor address provided, please provide a tor address."
             raise ClanError(msg)
         if ssh_tor_reachable(TorTarget(onion=deploy_info.tor, port=22)):
-            host = Host(host=deploy_info.tor)
+            host = Host(host=deploy_info.tor, password=deploy_info.pwd, tor_socks=True)
-            host.connect_ssh_shell(password=deploy_info.pwd, tor_socks=True)
         else:
             msg = "Could not reach host via tor either."
             raise ClanError(msg)

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

@@ -29,12 +29,14 @@ class Host:
     user: str | None = None
     port: int | None = None
     private_key: Path | None = None
+    password: str | None = None
     forward_agent: bool = False
     command_prefix: str | None = None
     host_key_check: HostKeyCheck = HostKeyCheck.ASK
     meta: dict[str, Any] = field(default_factory=dict)
     verbose_ssh: bool = False
     ssh_options: dict[str, str] = field(default_factory=dict)
+    tor_socks: bool = False
 
     def __post_init__(self) -> None:
         if not self.command_prefix:
@@ -201,18 +203,16 @@ class Host:
     def ssh_cmd(
         self,
         verbose_ssh: bool = False,
-        tor_socks: bool = False,
         tty: bool = False,
-        password: str | None = None,
     ) -> list[str]:
         packages = []
         password_args = []
-        if password:
+        if self.password:
             packages.append("sshpass")
             password_args = [
                 "sshpass",
                 "-p",
-                password,
+                self.password,
             ]
 
         ssh_opts = self.ssh_cmd_opts
@@ -221,7 +221,7 @@ class Host:
         if tty:
             ssh_opts.extend(["-t"])
 
-        if tor_socks:
+        if self.tor_socks:
             packages.append("netcat")
             ssh_opts.append("-o")
             ssh_opts.append("ProxyCommand=nc -x 127.0.0.1:9050 -X 5 %h %p")
@@ -235,12 +235,8 @@ class Host:
 
         return nix_shell(packages, cmd)
 
-    def connect_ssh_shell(
+    def interactive_ssh(self) -> None:
-        self, *, password: str | None = None, tor_socks: bool = False
+        subprocess.run(self.ssh_cmd())
-    ) -> None:
-        cmd = self.ssh_cmd(tor_socks=tor_socks, password=password)
-
-        subprocess.run(cmd)
 
 
 def is_ssh_reachable(host: Host) -> bool: