fixes and improvements

This commit is contained in:
2026-05-07 20:07:54 +02:00
parent dd7b8f133c
commit 420dc34272
3 changed files with 63 additions and 19 deletions
@@ -20,8 +20,6 @@ from typing import Any
import paramiko
from media_library_viewer_api.services.known_hosts import ensure_known_host
logger = logging.getLogger(__name__)
@@ -66,26 +64,12 @@ class RemoteSSHClient:
def connect(self) -> paramiko.SSHClient:
"""Create or reuse the Paramiko connection.
Unknown host keys are recorded on first contact in the managed
known_hosts file when one is configured. After that, strict checking
remains in effect so host key changes are still rejected.
Host keys are expected to be managed ahead of time by startup synthesis
or explicit validation flows. Runtime connections only load the managed
known_hosts file and then let Paramiko enforce strict checking.
"""
if self._client:
return self._client
if self.known_hosts_path:
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():
+30
View File
@@ -0,0 +1,30 @@
from __future__ import annotations
from unittest.mock import MagicMock, patch
from media_library_viewer_api.clients.ssh import RemoteSSHClient
def test_connect_uses_existing_known_hosts_without_reprobing(tmp_path):
known_hosts_path = tmp_path / "known_hosts"
known_hosts_path.write_text("example.com ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAAFAKE\n")
ssh_client = MagicMock()
ssh_client.connect.return_value = None
with (
patch("media_library_viewer_api.clients.ssh.paramiko.SSHClient", return_value=ssh_client),
patch("media_library_viewer_api.clients.ssh.paramiko.RejectPolicy", return_value=object()),
):
client = RemoteSSHClient(
host="example.com",
username="alex",
known_hosts_path=str(known_hosts_path),
)
client.connect()
ssh_client.load_system_host_keys.assert_called_once_with()
ssh_client.load_host_keys.assert_called_once_with(str(known_hosts_path))
ssh_client.set_missing_host_key_policy.assert_called_once()
ssh_client.connect.assert_called_once()