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


# ── De appdata en de databases ───────────────────────────────────────────────
# Een backup bevatte alleen de stackmap. LIBRARY_DIR en de appdata zijn buren,
# geen genestelde mappen, dus je zette een backup terug en hield een lege app
# over — met een groen vinkje erbij.

def _stack_met_appdata(env, naam="umami", inhoud=b"gegevens"):
    """Een geïnstalleerde stack met een appdata-map ernaast."""
    import json as _json
    d = env["lib"] / naam
    d.mkdir(parents=True, exist_ok=True)
    (d / "docker-compose.yml").write_text(
        "services:\n"
        f"  {naam}:\n"
        "    image: nginx\n"
        f"  {naam}-db:\n"
        "    image: postgres:16-alpine\n"
        f"    container_name: {naam}-db\n"
        "    environment:\n"
        "      - POSTGRES_USER=umami\n"
        "      - POSTGRES_DB=umami\n"
        "      - POSTGRES_PASSWORD=${DB_PASSWORD}\n")
    (d / ".env").write_text("DB_PASSWORD=geheim123\n")
    appdata = Path(str(env["tmp"])) / "appdata"
    (appdata / naam).mkdir(parents=True, exist_ok=True)
    (appdata / f"{naam}-db").mkdir(parents=True, exist_ok=True)
    (appdata / naam / "bestand.txt").write_bytes(inhoud)
    (d / ".serverup.json").write_text(_json.dumps({
        "source": naam,
        "values": {"appdata_dir": str(appdata), "service_name": naam},
    }))
    return d, appdata


def test_appdata_gaat_mee_in_het_archief(env, monkeypatch):
    import tarfile
    from core import backups, docker
    monkeypatch.setattr(docker, "_run", lambda *a, **k: _mislukt())
    _stack_met_appdata(env)

    meta = backups.create("umami")
    with tarfile.open(backups.backup_dir() / meta["file"]) as tar:
        namen = tar.getnames()
    assert "umami/docker-compose.yml" in namen
    assert "appdata/umami/bestand.txt" in namen
    assert "appdata/umami-db" in namen
    assert meta["appdata"] == ["umami", "umami-db"]


def test_appdata_kan_uit(env, monkeypatch):
    """Bij een mediabibliotheek loopt dit in de honderden gigabytes."""
    import tarfile
    from core import backups, docker
    monkeypatch.setattr(docker, "_run", lambda *a, **k: _mislukt())
    _stack_met_appdata(env)

    meta = backups.create("umami", appdata=False)
    with tarfile.open(backups.backup_dir() / meta["file"]) as tar:
        namen = tar.getnames()
    assert not [n for n in namen if n.startswith("appdata/")]
    assert meta["appdata"] == []


def test_terugzetten_herstelt_de_appdata(env, monkeypatch):
    from core import backups, docker
    monkeypatch.setattr(docker, "_run", lambda *a, **k: _mislukt())
    monkeypatch.setattr(docker, "compose_up", lambda *a, **k: 0)
    monkeypatch.setattr(docker, "compose_down", lambda *a, **k: 0)
    _d, appdata = _stack_met_appdata(env)

    meta = backups.create("umami")
    (appdata / "umami" / "bestand.txt").write_bytes(b"kapot")

    ok, msg = backups.restore("umami", meta["file"])
    assert ok, msg
    assert (appdata / "umami" / "bestand.txt").read_bytes() == b"gegevens"


def test_databasedienst_wordt_herkend_met_wachtwoord_uit_env(env, monkeypatch):
    """Het wachtwoord staat als ${DB_PASSWORD} in compose en met de waarde in
    .env — zonder die stap draait pg_dump met een lege sleutel."""
    from core import backups, docker
    monkeypatch.setattr(docker, "_run", lambda *a, **k: _mislukt())
    _stack_met_appdata(env)

    diensten = backups.databasediensten("umami")
    assert len(diensten) == 1
    d = diensten[0]
    assert d["container"] == "umami-db"
    assert d["soort"] == "postgres"
    assert d["gebruiker"] == "umami"
    assert d["wachtwoord"] == "geheim123"


def test_dump_die_mislukt_wordt_gemeld_en_niet_stil_overgeslagen(env, monkeypatch):
    """Een archief dat compleet lijkt terwijl de database ontbreekt is het
    ergste dat een backup kan doen."""
    from core import backups, docker
    monkeypatch.setattr(docker, "_run", lambda *a, **k: _mislukt(b"container niet gestart"))
    _stack_met_appdata(env)

    regels = []
    meta = backups.create("umami", log_fn=regels.append)
    assert meta["dumps"] == []
    assert any("Geen dump van umami-db" in r for r in regels), regels


def test_dump_belandt_in_het_archief(env, monkeypatch):
    import tarfile
    from core import backups, docker
    monkeypatch.setattr(docker, "_run",
                        lambda *a, **k: _gelukt(b"-- pg_dump\nCREATE TABLE x;\n"))
    _stack_met_appdata(env)

    meta = backups.create("umami")
    with tarfile.open(backups.backup_dir() / meta["file"]) as tar:
        namen = tar.getnames()
        inhoud = tar.extractfile("dumps/umami-db.sql").read()
    assert "dumps/umami-db.sql" in namen
    assert b"CREATE TABLE x;" in inhoud
    assert meta["dumps"] == ["umami-db.sql"]


def _mislukt(stderr=b"fout"):
    import subprocess
    return subprocess.CompletedProcess([], 1, b"", stderr)


def _gelukt(stdout=b""):
    import subprocess
    return subprocess.CompletedProcess([], 0, stdout, b"")