Add missing frontend and backend files

backend/src/media_library_viewer_api/services/known_hosts.py

@@ -0,0 +1,85 @@
+"""Managed known_hosts synthesis for SSH clients.
+
+Instead of mounting a host-side ``known_hosts`` file, the backend can discover
+and persist host keys for configured SSH machines inside its own cache volume.
+This keeps strict host-key checking enabled without exposing a whole SSH
+configuration directory into the container.
+"""
+
+from __future__ import annotations
+
+import logging
+import socket
+from pathlib import Path
+from typing import Any
+
+import paramiko
+
+logger = logging.getLogger(__name__)
+
+
+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:
+    sock = socket.create_connection((host, int(port or 22)), timeout=timeout)
+    transport = paramiko.Transport(sock)
+    try:
+        transport.start_client(timeout=timeout)
+        key = transport.get_remote_server_key()
+        if key is None:
+            raise RuntimeError(f"Unable to read SSH host key for {host}:{port}")
+        return key
+    finally:
+        transport.close()
+        sock.close()
+
+
+def ensure_known_host(host: str, port: int, known_hosts_path: Path, *, strict: bool = True) -> bool:
+    """Ensure a host key entry exists for the given host/port.
+
+    Returns ``True`` when the file was changed. If ``strict`` is enabled and the
+    existing key differs, a ``RuntimeError`` is raised instead of silently
+    overwriting the entry.
+    """
+    if not host:
+        return False
+    known_hosts_path.parent.mkdir(parents=True, exist_ok=True)
+    host_key = _fetch_server_key(host, port)
+    host_alias = _host_alias(host, port)
+    host_keys = paramiko.HostKeys()
+    if known_hosts_path.exists():
+        host_keys.load(str(known_hosts_path))
+
+    existing = host_keys.lookup(host_alias)
+    key_type = host_key.get_name()
+    if existing and key_type in existing:
+        if existing[key_type].get_base64() == host_key.get_base64():
+            return False
+        if strict:
+            raise RuntimeError(f"SSH host key mismatch for {host_alias}")
+
+    host_keys.add(host_alias, key_type, host_key)
+    host_keys.save(str(known_hosts_path))
+    logger.info("Recorded SSH host key host=%s port=%s file=%s", host, port, known_hosts_path)
+    return True
+
+
+def ensure_known_hosts_for_machines(
+    machines: list[dict[str, Any]],
+    known_hosts_path: Path,
+    *,
+    strict: bool = True,
+) -> int:
+    changed = 0
+    for machine in machines:
+        if str(machine.get("mode") or "").lower() != "ssh":
+            continue
+        host = str(machine.get("host") or "").strip()
+        port = int(machine.get("port") or 22)
+        if not host:
+            continue
+        if ensure_known_host(host, port, known_hosts_path, strict=strict):
+            changed += 1
+    return changed