"""SSH client helpers for remote filesystem and media inspection. All command execution goes through ``/bin/sh -c`` and all paths inserted into commands are shell-quoted by callers. This is important for two reasons: 1. The remote login shell may be fish/csh/etc.; internal commands are POSIX sh. 2. Media paths frequently contain spaces and punctuation. """ from __future__ import annotations import json import logging import posixpath import shlex from dataclasses import dataclass from io import StringIO from pathlib import Path from typing import Any import paramiko from media_library_viewer_api.services.known_hosts import has_known_host logger = logging.getLogger(__name__) @dataclass class CommandResult: """Plain result object returned by remote command execution.""" command: str exit_status: int stdout: str stderr: str class RemoteSSHClient: """SSH helper for read-only inspection plus explicit job execution.""" def __init__( self, host: str, 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 = 30, ): if not host or not username: raise ValueError("SSH host and username are required") self.host = host 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 self._client: paramiko.SSHClient | None = None def connect(self) -> paramiko.SSHClient: """Create or reuse the Paramiko connection. Host keys are trusted on first successful use when a managed known_hosts path is configured. Subsequent connections stay strict and reject host-key changes. """ if self._client: return self._client client = paramiko.SSHClient() client.load_system_host_keys() known_hosts_file = Path(self.known_hosts_path) if self.known_hosts_path else None trusted_before = bool(known_hosts_file and has_known_host(self.host, self.port, known_hosts_file)) if known_hosts_file and known_hosts_file.is_file(): client.load_host_keys(str(known_hosts_file)) client.set_missing_host_key_policy(paramiko.RejectPolicy() if trusted_before else paramiko.AutoAddPolicy()) connect_kwargs: dict[str, Any] = { "hostname": self.host, "port": self.port, "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 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 if known_hosts_file and not trusted_before: known_hosts_file.parent.mkdir(parents=True, exist_ok=True) client.save_host_keys(str(known_hosts_file)) 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() self._client = None def run(self, command: str, timeout: int | None = None) -> CommandResult: """Run a command through POSIX sh, independent of the user's login shell. Paramiko asks the SSH server to execute a command using the account's default shell. If that shell is fish/csh/etc., POSIX snippets containing `if ...; then`, pipes, redirects, or heredocs can fail. All internal app commands and job templates are written for POSIX shell, so explicitly dispatch through `/bin/sh -c`. """ client = self.connect() shell_command = f"/bin/sh -c {shlex.quote(command)}" logger.debug("SSH run host=%s timeout=%s command=%s", self.host, timeout or self.timeout, command) stdin, stdout, stderr = client.exec_command(shell_command, timeout=timeout or self.timeout) exit_status = stdout.channel.recv_exit_status() result = CommandResult( command=command, exit_status=exit_status, stdout=stdout.read().decode(errors="replace"), stderr=stderr.read().decode(errors="replace"), ) if result.exit_status == 0: logger.debug("SSH command ok host=%s exit_status=%s", self.host, result.exit_status) else: logger.warning( "SSH command failed host=%s exit_status=%s stderr=%s", self.host, result.exit_status, result.stderr.strip() or result.stdout.strip(), ) return result def list_dir(self, path: str) -> CommandResult: """List one remote directory as JSON. The command first verifies that ``path`` is a directory. Without that guard, running ``find`` on a file can look like an empty directory, which was a source of file-browser confusion. Output is NUL-delimited before Python serializes it, making spaces in filenames safe. """ quoted = shlex.quote(path) not_dir_message = shlex.quote(f"Not a directory: {path}") command = ( f"test -d {quoted} || " f"{{ echo {not_dir_message} >&2; exit 20; }}; " f"find {quoted} -maxdepth 1 -mindepth 1 -printf " "'%y\\t%s\\t%T@\\t%f\\0' | python3 -c " + shlex.quote( "import sys,json; data=sys.stdin.buffer.read().split(b'\\0'); " "rows=[]\n" "for row in data:\n" " if not row: continue\n" " t,s,m,n=row.decode('utf-8','replace').split('\\t',3)\n" " rows.append({'type':t,'size':int(s),'mtime':float(m),'name':n})\n" "print(json.dumps(rows))" ) ) result = self.run(command) logger.info("SSH list_dir path=%s exit_status=%s", path, result.exit_status) return result def stat_path(self, path: str) -> CommandResult: """Run stat for a remote file or directory path.""" quoted = shlex.quote(path) result = self.run(f"stat --printf='%F\\n%s bytes\\n%y\\n%n\\n' {quoted}") logger.info("SSH stat path=%s exit_status=%s", path, result.exit_status) return result def ffprobe_json(self, path: str) -> dict[str, Any]: """Run ffprobe and parse JSON output for a remote media file.""" quoted = shlex.quote(path) result = self.run( "ffprobe -v error -show_format -show_streams -print_format json " + quoted, timeout=60, ) logger.info("SSH ffprobe path=%s exit_status=%s", path, result.exit_status) if result.exit_status != 0: raise RuntimeError(result.stderr or result.stdout or "ffprobe failed") return json.loads(result.stdout) @staticmethod def join(parent: str, child: str) -> str: return posixpath.normpath(posixpath.join(parent, child))