fixes and improvements

backend/src/media_library_viewer_api/clients/ssh.py

@@ -14,6 +14,7 @@ import logging
 import posixpath
 import shlex
 from dataclasses import dataclass
+from io import StringIO
 from pathlib import Path
 from typing import Any
 
@@ -41,6 +42,8 @@ class RemoteSSHClient:
         username: str,
         port: int = 22,
         key_filename: str | None = None,
+        private_key: str | None = None,
+        private_key_passphrase: str | None = None,
         password: str | None = None,
         known_hosts_path: str | None = None,
         timeout: int = 20,
@@ -51,6 +54,8 @@ class RemoteSSHClient:
         self.username = username
         self.port = port
         self.key_filename = key_filename or None
+        self.private_key = private_key or None
+        self.private_key_passphrase = private_key_passphrase or None
         self.password = password or None
         self.known_hosts_path = known_hosts_path or None
         self.timeout = timeout
@@ -70,17 +75,32 @@ class RemoteSSHClient:
         if self.known_hosts_path and Path(self.known_hosts_path).is_file():
             client.load_host_keys(self.known_hosts_path)
         client.set_missing_host_key_policy(paramiko.RejectPolicy())
-        client.connect(
-            self.host,
-            port=self.port,
-            username=self.username,
-            key_filename=self.key_filename,
-            password=self.password,
-            timeout=self.timeout,
-        )
+        connect_kwargs: dict[str, Any] = {
+            "hostname": self.host,
+            "port": self.port,
+            "username": self.username,
+            "password": self.password,
+            "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)
         self._client = client
         return client
 
+    @staticmethod
+    def _load_private_key(private_key: str, passphrase: str | None = None) -> paramiko.PKey:
+        key_classes = [paramiko.Ed25519Key, paramiko.RSAKey, paramiko.ECDSAKey]
+        last_error: Exception | None = None
+        for key_class in key_classes:
+            try:
+                return key_class.from_private_key(StringIO(private_key), password=passphrase or None)
+            except Exception as exc:  # pragma: no cover - try multiple algorithms
+                last_error = exc
+        raise RuntimeError("Unable to load SSH private key") from last_error
+
     def close(self) -> None:
         if self._client:
             self._client.close()