server-up/tests/test_backups.py

"""Backups: maken, tonen, terugzetten, retentie — en de veiligheid daarvan."""
import io
import json
import tarfile
import time
from pathlib import Path

import pytest

from conftest import login


@pytest.fixture
def bk(env, monkeypatch):
    from core import backups, docker
    env["core"].patch({"BACKUP_DIR": str(env["tmp"] / "backups")})
    # Geen echte docker in de tests: stack stoppen/starten is een no-op.
    monkeypatch.setattr(docker, "compose_down", lambda *a, **k: 0)
    monkeypatch.setattr(docker, "compose_up", lambda *a, **k: 0)
    return backups


def _maak_stack(env, naam="vaultwarden", inhoud="services:\n  app:\n    image: nginx\n"):
    d = env["lib"] / naam
    d.mkdir(parents=True, exist_ok=True)
    (d / "docker-compose.yml").write_text(inhoud, encoding="utf-8")
    (d / ".env").write_text("PORT=8222\n", encoding="utf-8")
    return d


def test_maken_en_tonen(bk, env):
    _maak_stack(env)
    meta = bk.create("vaultwarden")
    assert meta["stack"] == "vaultwarden"
    assert meta["size"] > 0
    lijst = bk.list_for("vaultwarden")
    assert len(lijst) == 1
    assert lijst[0]["file"] == meta["file"]
    assert lijst[0]["size_human"]


def test_maken_van_onbekende_stack(bk):
    with pytest.raises(bk.BackupError):
        bk.create("bestaatniet")


def test_terugzetten_herstelt_gewijzigd_bestand(bk, env):
    d = _maak_stack(env)
    meta = bk.create("vaultwarden")
    (d / "docker-compose.yml").write_text("KAPOT", encoding="utf-8")

    ok, msg = bk.restore("vaultwarden", meta["file"])
    assert ok, msg
    assert "image: nginx" in (d / "docker-compose.yml").read_text()
    assert (d / ".env").read_text() == "PORT=8222\n"


def test_terugzetten_van_backup_van_andere_stack(bk, env):
    _maak_stack(env, "app-a")
    _maak_stack(env, "app-b")
    meta = bk.create("app-a")
    ok, msg = bk.restore("app-b", meta["file"])
    assert ok is False
    assert "hoort bij 'app-a'" in msg


@pytest.mark.parametrize("naam", [
    "../../etc/passwd.tar.gz", "/etc/shadow.tar.gz", "geen-tijdstempel.tar.gz",
    "vaultwarden_20260101_120000.txt", "", "..tar.gz",
])
def test_ongeldige_backupnaam(bk, naam):
    with pytest.raises(bk.BackupError):
        bk._archive_path(naam)


def test_terugzetten_weigert_archief_dat_buiten_de_map_schrijft(bk, env):
    """Een geprepareerd archief met '..' mag niets buiten de stackmap raken.

    Met een kaal `tar xzf` zou dit bestand op de host belanden; tarfile met
    filter='data' weigert de ingang.
    """
    _maak_stack(env)
    bd = Path(bk.backup_dir())
    bd.mkdir(parents=True, exist_ok=True)
    boos = bd / "vaultwarden_20260101_120000.tar.gz"
    with tarfile.open(boos, "w:gz") as tar:
        data = b"ontsnapt"
        info = tarfile.TarInfo("../../ontsnapt.txt")
        info.size = len(data)
        tar.addfile(info, io.BytesIO(data))

    ok, msg = bk.restore("vaultwarden", boos.name)
    assert ok is False
    assert not (env["tmp"] / "ontsnapt.txt").exists()
    assert not (env["lib"].parent / "ontsnapt.txt").exists()


def test_mislukte_restore_zet_de_oude_map_terug(bk, env):
    """Een kapot archief mag je niet zonder stack achterlaten."""
    d = _maak_stack(env)
    bd = Path(bk.backup_dir())
    bd.mkdir(parents=True, exist_ok=True)
    stuk = bd / "vaultwarden_20260101_120000.tar.gz"
    stuk.write_bytes(b"dit is geen gzip")

    ok, _ = bk.restore("vaultwarden", stuk.name)
    assert ok is False
    assert d.is_dir()
    assert "image: nginx" in (d / "docker-compose.yml").read_text()


def test_verwijderen(bk, env):
    _maak_stack(env)
    meta = bk.create("vaultwarden")
    ok, _ = bk.delete(meta["file"])
    assert ok
    assert bk.list_for("vaultwarden") == []


def test_retentie_houdt_er_hoogstens_keep(bk, env):
    _maak_stack(env)
    env["core"].patch({"BACKUP_KEEP": 3})
    metas = []
    for i in range(5):
        m = bk.create("vaultwarden")
        # Tijdstempels uit elkaar trekken; de naam heeft secondeprecisie.
        m["ts"] = time.time() - (5 - i)
        _herschrijf_ts(bk, m)
        metas.append(m)
    bk.prune("vaultwarden")
    assert len(bk.list_for("vaultwarden")) == 3


def test_retentie_bewaart_altijd_de_nieuwste(bk, env):
    """Ook als alles te oud is, blijft er één over."""
    _maak_stack(env)
    env["core"].patch({"BACKUP_KEEP": 0, "BACKUP_MAX_AGE_DAYS": 1})
    for i in range(3):
        m = bk.create("vaultwarden")
        m["ts"] = time.time() - 30 * 86400
        _herschrijf_ts(bk, m)
    bk.prune("vaultwarden")
    assert len(bk.list_for("vaultwarden")) == 1


def test_retentie_uit_doet_niets(bk, env):
    _maak_stack(env)
    env["core"].patch({"BACKUP_KEEP": 0, "BACKUP_MAX_AGE_DAYS": 0})
    for _ in range(3):
        bk.create("vaultwarden")
    assert bk.prune("vaultwarden") == []


def _herschrijf_ts(bk, meta):
    """Zet de tijdstempel in het metadata-bestand terug in de tijd."""
    p = Path(bk.backup_dir()) / meta["file"]
    mf = p.with_suffix("").with_suffix(".json")
    d = json.loads(mf.read_text())
    d["ts"] = meta["ts"]
    mf.write_text(json.dumps(d))


# ── API ──────────────────────────────────────────────────────────────────────

def test_endpoint_lijst(client, env, bk):
    login(client)
    _maak_stack(env)
    bk.create("vaultwarden")
    d = client.get("/api/backups").get_json()
    assert len(d["backups"]) == 1
    assert "BACKUP_KEEP" in d["settings"]


def test_endpoint_restore_weigert_ongeldige_naam(client, bk):
    csrf = login(client)
    r = client.post("/api/backups/vaultwarden/restore",
                    json={"file": "../../etc/passwd.tar.gz"},
                    headers={"X-CSRF-Token": csrf})
    assert r.status_code == 400


def test_endpoint_restore_zonder_bestand(client, bk):
    csrf = login(client)
    r = client.post("/api/backups/vaultwarden/restore", json={},
                    headers={"X-CSRF-Token": csrf})
    assert r.status_code == 400