backup-tool/tests/integration/test_ssh_backup_restore.py

"""Opt-in live forced-SFTP chroot coverage; all SSH keys are generated under tmp_path."""

from __future__ import annotations

import os
import socket
import subprocess
import time
from pathlib import Path

import httpx
import pytest
from backup_tool.db.engine import create_engine
from backup_tool.db.models import Backup, Execution, Repository
from backup_tool.snapshot import verify_published_snapshot
from backup_tool.worker import Worker
from sqlalchemy import select

ROOT = Path(__file__).resolve().parents[2]
PASSWORD = "correct-horse-battery-staple"


def _enabled() -> bool:
    return os.environ.get("BACKUP_TOOL_SSH_INTEGRATION") == "1"


def _port() -> int:
    with socket.socket() as listener:
        listener.bind(("127.0.0.1", 0))
        address = listener.getsockname()
    if not isinstance(address, tuple) or not isinstance(address[1], int):
        raise RuntimeError("could not allocate SSH fixture port")
    return address[1]


def _wait(port: int) -> None:
    deadline = time.monotonic() + 60
    while time.monotonic() < deadline:
        try:
            with socket.create_connection(("127.0.0.1", port), timeout=1):
                return
        except OSError:
            time.sleep(0.25)
    raise AssertionError("SSHD fixture did not become reachable")


@pytest.fixture
def sshd_fixture(tmp_path: Path):
    fixture = tmp_path / "fixture"
    host = fixture / "host"
    source = fixture / "source"
    host.mkdir(parents=True)
    source.mkdir()
    fixture.chmod(0o755)
    host.chmod(0o755)
    source.chmod(0o755)
    private = fixture / "client"
    for target in (host / "ssh_host_ed25519_key", private):
        subprocess.run(
            ["ssh-keygen", "-q", "-t", "ed25519", "-N", "", "-f", str(target)],
            check=True,
        )
    (host / "ssh_host_ed25519_key").chmod(0o644)
    (fixture / "authorized_keys").write_text(private.with_suffix(".pub").read_text())
    # Public keys are copied into a backup-owned 0600 tmpfs file at startup.
    (fixture / "authorized_keys").chmod(0o644)
    port = _port()
    environment = os.environ | {
        "SSH_FIXTURE_DIR": str(fixture),
        "SSH_FIXTURE_PORT": str(port),
        "COMPOSE_PROJECT_NAME": f"backup-tool-ssh-{os.getpid()}-{port}",
    }
    command = ["docker", "compose", "-f", "tests/compose.ssh.yaml"]
    try:
        subprocess.run([*command, "up", "--build", "-d"], cwd=ROOT, env=environment, check=True)
        _wait(port)
        host_key = " ".join(host.joinpath("ssh_host_ed25519_key.pub").read_text().split()[:2])
        yield {
            "port": port,
            "key": private,
            "host_key": host_key,
            "source": source,
            "env": environment,
        }
    finally:
        subprocess.run(
            [*command, "down", "--volumes", "--remove-orphans"],
            cwd=ROOT,
            env=environment,
            check=False,
        )


async def _login(client: httpx.AsyncClient) -> dict[str, str]:
    response = await client.post("/api/v2/setup", json={"username": "admin", "password": PASSWORD})
    assert response.status_code == 201
    return {"X-CSRF-Token": client.cookies["backup_tool_csrf"]}


@pytest.mark.skipif(not _enabled(), reason="set BACKUP_TOOL_SSH_INTEGRATION=1")
@pytest.mark.asyncio
async def test_forced_sftp_chroot_probe_backup_and_restore(app_client, sshd_fixture) -> None:
    client, settings = app_client
    source = sshd_fixture["source"]
    (source / "nested").mkdir()
    (source / "nested" / "hello.txt").write_text("hello ssh\n")
    headers = await _login(client)
    secret = await client.post(
        "/api/v2/admin/secrets",
        json={"purpose": "ssh_private_key", "value": sshd_fixture["key"].read_text()},
        headers=headers,
    )
    assert secret.status_code == 201
    remote = await client.post(
        "/api/v2/sources",
        json={
            "name": "ssh",
            "kind": "ssh",
            "private_key_secret_id": secret.json()["id"],
            "public_config": {
                "hostname": "127.0.0.1",
                "port": sshd_fixture["port"],
                "username": "backup",
                "host_key": sshd_fixture["host_key"],
                "root": "/",
            },
        },
        headers=headers,
    )
    assert remote.status_code == 201, remote.text
    assert (
        await client.post(f"/api/v2/sources/{remote.json()['id']}/probe", headers=headers)
    ).json() == {"entry_count": 1}
    repository = await client.post(
        "/api/v2/repositories",
        json={"name": "repo", "relative_path": "ssh"},
        headers=headers,
    )
    job = await client.post(
        "/api/v2/jobs",
        json={
            "name": "ssh-job",
            "source_id": remote.json()["id"],
            "repository_id": repository.json()["id"],
            "requested_mode": "full",
            "exclusions": [],
            "retention": {},
            "enabled": True,
            "allow_empty": False,
        },
        headers=headers,
    )
    execution = await client.post(f"/api/v2/jobs/{job.json()['id']}/executions", headers=headers)
    worker = Worker(settings, owner="ssh-live")
    try:
        assert await worker.run_once()
    finally:
        await worker.engine.dispose()
    engine = create_engine(settings)
    try:
        from sqlalchemy.ext.asyncio import async_sessionmaker

        async with async_sessionmaker(engine, expire_on_commit=False)() as db:
            stored = await db.get(Execution, execution.json()["id"])
            backup = await db.scalar(
                select(Backup).where(Backup.execution_id == execution.json()["id"])
            )
            stored_repository = await db.get(Repository, repository.json()["id"])
    finally:
        await engine.dispose()
    assert stored is not None and stored.state == "committed", (
        stored.operator_message if stored else None
    )
    assert backup is not None and stored_repository is not None
    manifest = verify_published_snapshot(
        Path(stored_repository.root),
        Path(stored_repository.root) / "manifests" / f"{backup.manifest_id}.json",
        stored_repository.signing_public_key,
    )
    assert any(entry["path"] == "data/nested/hello.txt" for entry in manifest["entries"])
    destination = settings.restore_roots[0] / "ssh-restored"
    restore = await client.post(
        f"/api/v2/backups/{backup.id}/restores",
        json={
            "destination": str(destination),
            "selection": ["data/nested"],
            "overwrite_policy": "fail",
        },
        headers=headers,
    )
    assert restore.status_code == 202
    restore_worker = Worker(settings, owner="ssh-restore")
    try:
        assert await restore_worker.run_once()
    finally:
        await restore_worker.engine.dispose()
    assert (destination / "data" / "nested" / "hello.txt").read_text() == "hello ssh\n"