stackpilot/backend/tests/test_destinations.py

"""Backup destinations: errors that say what happened, and names that stay names.

An SFTP host that refused the connection was a bare 500 (paramiko connects in
the Transport constructor, outside every handler), so the "Test" button showed
nothing useful. And a backup file name from the API went straight into
``posixpath.join`` — ``../`` let a delete reach outside the backup folder.
"""
from __future__ import annotations

import socket

import pytest


@pytest.mark.parametrize("name", ["../etc/passwd", "a/b", "..", "", "x\\y", "a\0b"])
def test_backup_names_must_be_bare_file_names(name):
    from services import backup_destination_service as dest

    with pytest.raises(dest.DestinationError):
        dest._check_file_name(name)


def test_a_real_backup_name_passes():
    from services import backup_destination_service as dest

    dest._check_file_name("backup-immich-full-20260101-030000.tar.gz")


def _free_port() -> int:
    with socket.socket() as s:
        s.bind(("127.0.0.1", 0))
        return s.getsockname()[1]


def _destination(kind: str, config: dict):
    from models.backup_destination import BackupDestination
    from services import backup_destination_service as dest

    return BackupDestination(id=99, name="t", type=kind, config=dest.dump_config(config))


def test_an_unreachable_sftp_host_is_a_destination_error():
    from services import backup_destination_service as dest

    d = _destination(
        "sftp", {"host": "127.0.0.1", "port": _free_port(), "username": "u", "password": "p"}
    )
    with pytest.raises(dest.DestinationError, match="SFTP connection"):
        dest.test(d)


def test_sftp_delete_refuses_a_path_before_connecting():
    from services import backup_destination_service as dest

    d = _destination("sftp", {"host": "198.51.100.1", "username": "u"})
    with pytest.raises(dest.DestinationError, match="Invalid backup file name"):
        dest.delete(d, "../../important")


def test_a_malformed_s3_endpoint_is_a_destination_error():
    from services import backup_destination_service as dest

    d = _destination("s3", {"bucket": "b", "endpoint_url": "not a url"})
    with pytest.raises(dest.DestinationError, match="Invalid S3 settings"):
        dest.test(d)


def test_the_test_endpoint_reports_instead_of_failing(as_admin, db):
    from sqlmodel import Session

    from database import engine
    from models.backup_destination import BackupDestination
    from services import backup_destination_service as dest

    with Session(engine) as session:
        row = BackupDestination(
            name="refusing",
            type="sftp",
            config=dest.dump_config(
                {"host": "127.0.0.1", "port": _free_port(), "username": "u", "password": "p"}
            ),
        )
        session.add(row)
        session.commit()
        session.refresh(row)
        dest_id = row.id
    try:
        response = as_admin.post(f"/api/backups/destinations/{dest_id}/test")
        assert response.status_code == 200
        assert response.json()["ok"] is False
        assert "SFTP connection" in response.json()["error"]
    finally:
        with Session(engine) as session:
            session.delete(session.get(BackupDestination, dest_id))
            session.commit()


def test_a_destination_in_use_by_a_schedule_cannot_be_deleted(as_admin, db):
    """Deleted from under it, the schedule failed on every run from then on."""
    from sqlmodel import Session

    from database import engine
    from models.backup_destination import BackupDestination
    from models.backup_schedule import BackupSchedule
    from services import backup_destination_service as dest

    with Session(engine) as session:
        row = BackupDestination(name="nas", type="sftp", config=dest.dump_config({"host": "x"}))
        session.add(row)
        session.commit()
        session.refresh(row)
        schedule = BackupSchedule(destination_id=row.id, all_stacks=True)
        session.add(schedule)
        session.commit()
        session.refresh(schedule)
        dest_id, schedule_id = row.id, schedule.id
    try:
        response = as_admin.delete(f"/api/backups/destinations/{dest_id}")
        assert response.status_code == 409
        assert "schedule" in response.json()["detail"]
    finally:
        with Session(engine) as session:
            session.delete(session.get(BackupSchedule, schedule_id))
            session.commit()
    assert as_admin.delete(f"/api/backups/destinations/{dest_id}").status_code == 200


# --------------------------------------------------------------------------- #
# SFTP host keys: a real SSH handshake against an in-process paramiko server.
# --------------------------------------------------------------------------- #


class _SshServer:
    """One-connection SSH server with an SFTP subsystem and a known host key."""

    def __init__(self, host_key):
        import threading

        import paramiko

        self.host_key = host_key
        self.passwords: list[str] = []
        outer = self

        class Iface(paramiko.ServerInterface):
            def get_allowed_auths(self, username):
                return "password"

            def check_auth_password(self, username, password):
                outer.passwords.append(password)
                return paramiko.AUTH_SUCCESSFUL

            def check_channel_request(self, kind, chanid):
                return paramiko.OPEN_SUCCEEDED

        self._iface = Iface
        self.sock = socket.socket()
        self.sock.bind(("127.0.0.1", 0))
        self.sock.listen(1)
        self.port = self.sock.getsockname()[1]
        threading.Thread(target=self._run, daemon=True).start()

    def _run(self):
        import paramiko

        conn, _ = self.sock.accept()
        transport = paramiko.Transport(conn)
        transport.add_server_key(self.host_key)
        transport.set_subsystem_handler("sftp", paramiko.SFTPServer, paramiko.SFTPServerInterface)
        try:
            transport.start_server(server=self._iface())
            # Kept: a channel that is garbage collected closes itself, and the
            # client saw "Channel closed" in the middle of the SFTP handshake.
            self.channel = transport.accept(10)
            while transport.is_active():
                transport.join(0.1)
        except Exception:  # noqa: BLE001 - the client hung up on purpose
            pass
        finally:
            self.sock.close()

    def cfg(self, **extra):
        return {"host": "127.0.0.1", "port": self.port, "username": "u", "password": "pw", **extra}


@pytest.fixture
def host_key():
    import paramiko

    return paramiko.ECDSAKey.generate()


def _connect(cfg, pin=None):
    from services import backup_destination_service as dest

    sftp, transport = dest._sftp_connect(cfg, pin)
    sftp.close()
    transport.close()


def test_the_first_host_key_is_pinned(host_key):
    server = _SshServer(host_key)
    pinned = []
    _connect(server.cfg(), pinned.append)
    assert pinned == [f"{host_key.get_name()} {host_key.get_base64()}"]


@pytest.mark.parametrize("form", ["line", "known_hosts", "fingerprint"])
def test_a_pinned_key_lets_its_server_in(host_key, form):
    from services import backup_destination_service as dest

    line = f"{host_key.get_name()} {host_key.get_base64()}"
    stored = {
        "line": line,
        "known_hosts": f"[nas]:2222 {line}",
        "fingerprint": dest.host_key_fingerprint(host_key),
    }[form]
    server = _SshServer(host_key)
    pinned = []
    _connect(server.cfg(host_key=stored), pinned.append)
    assert pinned == []  # nothing new to pin
    assert server.passwords == ["pw"]


def test_another_key_is_refused_before_the_password_is_sent(host_key):
    import paramiko

    from services import backup_destination_service as dest

    impostor = _SshServer(host_key)
    real = paramiko.ECDSAKey.generate()
    with pytest.raises(dest.DestinationError, match="does not match"):
        _connect(impostor.cfg(host_key=f"{real.get_name()} {real.get_base64()}"))
    assert impostor.passwords == []


def test_the_fingerprint_of_a_stored_key_is_the_one_ssh_keygen_prints(host_key):
    from services import backup_destination_service as dest

    line = f"{host_key.get_name()} {host_key.get_base64()}"
    assert dest.fingerprint_of(line) == dest.host_key_fingerprint(host_key)
    assert dest.fingerprint_of("SHA256:abc") == "SHA256:abc"


def test_moving_a_destination_to_another_host_forgets_its_key(as_admin, db):
    from sqlmodel import Session

    from database import engine
    from models.backup_destination import BackupDestination
    from services import backup_destination_service as dest

    with Session(engine) as session:
        row = BackupDestination(
            name="moving",
            type="sftp",
            config=dest.dump_config({"host": "old", "username": "u", "host_key": "ssh-ed25519 AAAA"}),
        )
        session.add(row)
        session.commit()
        session.refresh(row)
        dest_id = row.id
    try:
        # The form sends every field back, the shown host key included.
        same = {"host": "old", "username": "u", "host_key": "ssh-ed25519 AAAA", "path": "/b"}
        as_admin.put(f"/api/backups/destinations/{dest_id}", json={"config": same})
        moved = {**same, "host": "new"}
        assert as_admin.put(
            f"/api/backups/destinations/{dest_id}", json={"config": moved}
        ).status_code == 200
        with Session(engine) as session:
            cfg = dest.parse_config(session.get(BackupDestination, dest_id))
        assert cfg["host"] == "new" and "host_key" not in cfg
    finally:
        with Session(engine) as session:
            session.delete(session.get(BackupDestination, dest_id))
            session.commit()