Clan-app: connection check & show remote devices

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

@@ -3,12 +3,13 @@ import json
 import logging
 from dataclasses import dataclass
 from pathlib import Path
+from typing import Literal
 
 from clan_cli.api import API
 from clan_cli.cmd import run_no_stdout
-from clan_cli.errors import ClanError
+from clan_cli.errors import ClanCmdError, ClanError
 from clan_cli.inventory import Machine, load_inventory_eval
-from clan_cli.nix import nix_eval
+from clan_cli.nix import nix_eval, nix_shell
 
 log = logging.getLogger(__name__)
 
@@ -66,6 +67,52 @@ def list_nixos_machines(flake_url: str | Path) -> list[str]:
         raise ClanError(f"Error decoding machines from flake: {e}")
 
 
+@dataclass
+class ConnectionOptions:
+    keyfile: str | None = None
+    timeout: int = 2
+
+
+@API.register
+def check_machine_online(
+    flake_url: str | Path, machine_name: str, opts: ConnectionOptions | None
+) -> Literal["Online", "Offline"]:
+    machine = load_inventory_eval(flake_url).machines.get(machine_name)
+    if not machine:
+        raise ClanError(f"Machine {machine_name} not found in inventory")
+
+    hostname = machine.deploy.targetHost
+
+    if not hostname:
+        raise ClanError(f"Machine {machine_name} does not specify a targetHost")
+
+    timeout = opts.timeout if opts and opts.timeout else 2
+
+    cmd = nix_shell(
+        ["nixpkgs#util-linux", *(["nixpkgs#openssh"] if hostname else [])],
+        [
+            "ssh",
+            *(["-i", f"{opts.keyfile}"] if opts and opts.keyfile else []),
+            # Disable strict host key checking
+            "-o StrictHostKeyChecking=no",
+            # Disable known hosts file
+            "-o UserKnownHostsFile=/dev/null",
+            f"-o ConnectTimeout={timeout}",
+            f"{hostname}",
+            "true",
+            "&> /dev/null",
+        ],
+    )
+    try:
+        proc = run_no_stdout(cmd)
+        if proc.returncode != 0:
+            return "Offline"
+
+        return "Online"
+    except ClanCmdError:
+        return "Offline"
+
+
 def list_command(args: argparse.Namespace) -> None:
     flake_path = args.flake.path
     for name in list_nixos_machines(flake_path):