Add SSH connection validation for settings

backend/src/media_library_viewer_api/clients/ssh.py

@@ -48,7 +48,7 @@ class RemoteSSHClient:
         private_key_passphrase: str | None = None,
         password: str | None = None,
         known_hosts_path: str | None = None,
-        timeout: int = 20,
+        timeout: int = 30,
     ):
         if not host or not username:
             raise ValueError("SSH host and username are required")
@@ -73,7 +73,19 @@ class RemoteSSHClient:
         if self._client:
             return self._client
         if self.known_hosts_path:
-            ensure_known_host(self.host, self.port, Path(self.known_hosts_path), strict=True)
+            try:
+                ensure_known_host(self.host, self.port, Path(self.known_hosts_path), strict=True)
+            except Exception as exc:
+                message = str(exc).lower()
+                if "protocol banner" in message:
+                    raise RuntimeError(
+                        f"SSH banner not received from {self.host}:{self.port}. "
+                        "The host key could not be recorded because the backend could not talk to SSH."
+                    ) from exc
+                raise RuntimeError(
+                    f"SSH host key lookup failed for {self.host}:{self.port}. "
+                    "Confirm the host and port are correct and that SSH is reachable."
+                ) from exc
         client = paramiko.SSHClient()
         client.load_system_host_keys()
         if self.known_hosts_path and Path(self.known_hosts_path).is_file():
@@ -85,12 +97,28 @@ class RemoteSSHClient:
             "username": self.username,
             "password": self.password,
             "timeout": self.timeout,
+            "banner_timeout": self.timeout,
+            "auth_timeout": self.timeout,
         }
         if self.private_key:
             connect_kwargs["pkey"] = self._load_private_key(self.private_key, self.private_key_passphrase)
         else:
             connect_kwargs["key_filename"] = self.key_filename
-        client.connect(**connect_kwargs)
+        try:
+            client.connect(**connect_kwargs)
+        except Exception as exc:
+            message = str(exc).lower()
+            if "protocol banner" in message:
+                raise RuntimeError(
+                    f"SSH banner not received from {self.host}:{self.port}. "
+                    "Confirm the host, port, and firewall; the backend could not complete the SSH handshake."
+                ) from exc
+            if "no authentication methods available" in message or "authentication failed" in message:
+                raise RuntimeError(
+                    f"SSH authentication failed for {self.host}:{self.port}. "
+                    "Check the selected key, passphrase, username, or password."
+                ) from exc
+            raise
         self._client = client
         return client
 

backend/src/media_library_viewer_api/routers/settings.py

@@ -9,8 +9,11 @@ import paramiko
 from fastapi import APIRouter, Depends, HTTPException, status
 from pydantic import BaseModel, Field
 
+from media_library_viewer_api.clients.ssh import RemoteSSHClient
+from media_library_viewer_api.config import get_settings
 from media_library_viewer_api.dependencies import get_monitoring_poller, get_settings_store
 from media_library_viewer_api.services.db_maintenance import remove_sqlite_database
+from media_library_viewer_api.services.known_hosts import ensure_known_host
 from media_library_viewer_api.services.media_index import MediaIndex
 from media_library_viewer_api.services.settings_store import SettingsStore
 
@@ -49,6 +52,106 @@ def get_machines(store: SettingsStore = Depends(get_settings_store)) -> list[dic
     return store.list_machines()
 
 
+def _resolve_ssh_client(
+    machine: MonitoringMachineInput,
+    store: SettingsStore,
+) -> tuple[RemoteSSHClient, str, int]:
+    host = machine.host.strip()
+    username = machine.username.strip()
+    port = int(machine.port or 22)
+    if not host or not username:
+        raise HTTPException(status_code=400, detail="SSH machine is missing host or username")
+
+    private_key = machine.ssh_private_key
+    passphrase = machine.ssh_private_key_passphrase
+    if machine.ssh_key_id:
+        ssh_key = store.get_ssh_key(machine.ssh_key_id)
+        if ssh_key:
+            private_key = str(ssh_key.get("private_key") or private_key)
+            passphrase = str(ssh_key.get("passphrase") or passphrase)
+
+    key_filename = ""
+    if machine.key_directory and machine.key_name:
+        key_filename = f"{machine.key_directory}/{machine.key_name}"
+
+    settings = get_settings()
+    client = RemoteSSHClient(
+        host=host,
+        username=username,
+        port=port,
+        key_filename=key_filename or None,
+        private_key=private_key or None,
+        private_key_passphrase=passphrase or None,
+        password=machine.password or None,
+        known_hosts_path=str(settings.ssh_known_hosts_file),
+    )
+    return client, host, port
+
+
+@router.post("/machines/test-ssh")
+def test_machine_ssh(
+    machine: MonitoringMachineInput,
+    store: SettingsStore = Depends(get_settings_store),
+) -> dict[str, Any]:
+    if str(machine.mode or "").strip().lower() != "ssh":
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="SSH validation only applies to SSH machines")
+
+    client, host, port = _resolve_ssh_client(machine, store)
+    settings = get_settings()
+    try:
+        known_hosts_updated = ensure_known_host(host, port, settings.ssh_known_hosts_file, strict=True)
+    except RuntimeError as exc:
+        message = str(exc)
+        lowered = message.lower()
+        if "protocol banner" in lowered:
+            raise HTTPException(
+                status_code=status.HTTP_502_BAD_GATEWAY,
+                detail=(
+                    f"SSH banner not received from {host}:{port}; the backend could not record the host key. "
+                    "Confirm the SSH service, host, and port are reachable."
+                ),
+            ) from exc
+        raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=message) from exc
+
+    try:
+        client.connect()
+    except Exception as exc:
+        message = str(exc)
+        lowered = message.lower()
+        if "protocol banner" in lowered:
+            raise HTTPException(
+                status_code=status.HTTP_502_BAD_GATEWAY,
+                detail=(
+                    f"SSH banner not received from {host}:{port}; the backend recorded the host key, "
+                    "but SSH auth could not be validated. Confirm the SSH service is running."
+                ),
+            ) from exc
+        if "no authentication methods available" in lowered or "authentication failed" in lowered:
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail=(
+                    f"SSH banner received from {host}:{port}, but authentication failed. "
+                    "Check the selected SSH key, passphrase, username, or password."
+                ),
+            ) from exc
+        raise HTTPException(
+            status_code=status.HTTP_502_BAD_GATEWAY,
+            detail=f"SSH validation failed for {host}:{port}: {message}",
+        ) from exc
+    finally:
+        client.close()
+
+    return {
+        "status": "ok",
+        "message": (
+            f"SSH connection succeeded for {host}:{port}; host key {'was recorded' if known_hosts_updated else 'was already trusted'} and authentication worked."
+        ),
+        "host": host,
+        "port": port,
+        "known_hosts_updated": known_hosts_updated,
+    }
+
+
 @router.post("/machines", status_code=status.HTTP_201_CREATED)
 def post_machine(
     machine: MonitoringMachineInput,

backend/src/media_library_viewer_api/services/known_hosts.py

@@ -22,7 +22,7 @@ def _host_alias(host: str, port: int) -> str:
     return host if int(port or 22) == 22 else f"[{host}]:{int(port or 22)}"
 
 
-def _fetch_server_key(host: str, port: int, timeout: int = 10) -> paramiko.PKey:
+def _fetch_server_key(host: str, port: int, timeout: int = 30) -> paramiko.PKey:
     sock = socket.create_connection((host, int(port or 22)), timeout=timeout)
     transport = paramiko.Transport(sock)
     try:
@@ -31,6 +31,11 @@ def _fetch_server_key(host: str, port: int, timeout: int = 10) -> paramiko.PKey:
         if key is None:
             raise RuntimeError(f"Unable to read SSH host key for {host}:{port}")
         return key
+    except Exception as exc:
+        raise RuntimeError(
+            f"Unable to read SSH protocol banner from {host}:{port}. "
+            "Confirm the host is running an SSH server on that port and is reachable from the backend."
+        ) from exc
     finally:
         transport.close()
         sock.close()