manage/backend/clients/ssh.py

"""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 posixpath
import shlex
from dataclasses import dataclass
from typing import Any

import paramiko


@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,
        password: str | None = None,
        timeout: int = 20,
    ):
        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.password = password or None
        self.timeout = timeout
        self._client: paramiko.SSHClient | None = None

    def connect(self) -> paramiko.SSHClient:
        """Create or reuse the Paramiko connection.

        Unknown host keys are rejected. Users should connect once manually with
        ssh so the server is present in known_hosts.
        """
        if self._client:
            return self._client
        client = paramiko.SSHClient()
        client.load_system_host_keys()
        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,
        )
        self._client = client
        return client

    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)}"
        stdin, stdout, stderr = client.exec_command(shell_command, timeout=timeout or self.timeout)
        exit_status = stdout.channel.recv_exit_status()
        return CommandResult(
            command=command,
            exit_status=exit_status,
            stdout=stdout.read().decode(errors="replace"),
            stderr=stderr.read().decode(errors="replace"),
        )

    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.
        """
        # JSON-ish output: type, size, mtime epoch, filename. Handles spaces/newlines reasonably via NUL boundaries.
        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))"
            )
        )
        return self.run(command)

    def stat_path(self, path: str) -> CommandResult:
        """Run stat for a remote file or directory path."""
        quoted = shlex.quote(path)
        return self.run(f"stat --printf='%F\\n%s bytes\\n%y\\n%n\\n' {quoted}")

    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,
        )
        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))