3bee664484
All checks were successful
Deploy server-up (dev) / deploy (push) Successful in 17m1s
Ik sprong met tientallen (0.7.90 → 0.8.00 → … → 0.9.00 → 0.10.00) en liep daarmee de reeks uit: dit schema rolt over bij .90, dus na 0.9.90 hoort 1.0.00 te komen. "0.10.00" bestond niet, en schond ook het formaat v0.0.00 uit de projectafspraken. De vijftien uitgaven van deze sessie zijn hernummerd naar 0.7.81 t/m 0.7.95, aaneengesloten. In één doorloop met een tabel, want 0.8.81 wordt 0.7.90 en 0.7.90 wordt 0.7.81 — achtereenvolgende vervangingen zouden elkaar overschrijven. Meegenomen: verwijzingen in de documentatie, .env.example, install.sh en de tests. Verkorte vormen als "vanaf v0.10" zijn vervangen door het volledige nummer, want die waren niet automatisch te herleiden. De commit-onderwerpen in de geschiedenis dragen nog de oude nummers; VERSION en CHANGELOG zijn leidend. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Q9eqpADJSRs49SoGGr4NAy
414 lines
15 KiB
Python
414 lines
15 KiB
Python
"""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"")
|
|
|
|
|
|
# ── Rechten: wie is de eigenaar van wat er teruggezet wordt? ─────────────────
|
|
# Server Up draait vanaf v0.7.93 niet meer als root. Twee gevolgen die eerder
|
|
# geen rol speelden: appdata van een app met een eigen uid kan buiten bereik
|
|
# liggen, en `tarfile` met filter="data" geeft alles aan wie uitpakt.
|
|
|
|
def test_een_onleesbaar_bestand_laat_de_rest_van_de_backup_staan(env, monkeypatch):
|
|
"""Eén bestand waar we niet bij kunnen mocht niet de héle backup slopen —
|
|
juist dan heb je hem nodig."""
|
|
from core import backups, docker
|
|
monkeypatch.setattr(docker, "_run", lambda *a, **k: _mislukt())
|
|
_d, appdata = _stack_met_appdata(env)
|
|
(appdata / "umami" / "tweede.txt").write_bytes(b"deze wel")
|
|
|
|
# Rechten intrekken werkt niet als toets: root leest ook een bestand met
|
|
# modus 000, en de CI draait als root. De weigering komt daarom van het
|
|
# inpakken zelf — dat is precies waar hij vandaan zou komen.
|
|
echt = backups.tarfile.TarFile.add
|
|
|
|
def weiger_een(self, name, *a, **kw):
|
|
if str(name).endswith("bestand.txt"):
|
|
raise PermissionError("Permission denied")
|
|
return echt(self, name, *a, **kw)
|
|
|
|
monkeypatch.setattr(backups.tarfile.TarFile, "add", weiger_een)
|
|
meta = backups.create("umami")
|
|
|
|
assert meta["skipped"] == ["bestand.txt"], meta
|
|
with backups.tarfile.open(backups.backup_dir() / meta["file"]) as tar:
|
|
namen = tar.getnames()
|
|
assert "appdata/umami/tweede.txt" in namen, "de rest is ook weggevallen"
|
|
assert "appdata/umami/bestand.txt" not in namen
|
|
|
|
|
|
def test_wat_ontbreekt_staat_in_de_metadata(env, monkeypatch):
|
|
"""Een archief dat compleet lijkt maar het niet is, is erger dan een
|
|
archief dat zegt wat het mist."""
|
|
from core import backups, docker
|
|
monkeypatch.setattr(docker, "_run", lambda *a, **k: _mislukt())
|
|
_stack_met_appdata(env)
|
|
meta = backups.create("umami")
|
|
assert meta["skipped"] == [], "een geslaagde backup meldt niets ontbrekends"
|
|
assert "skipped" in backups.list_for("umami")[0]
|
|
|
|
|
|
def test_teruggezette_appdata_krijgt_de_puid_van_de_stack(env, monkeypatch):
|
|
"""filter="data" laat uid/gid vallen, dus zonder chown wordt alles van wie
|
|
uitpakt — en start de app niet meer."""
|
|
import json as _json
|
|
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_pad = d / ".serverup.json"
|
|
inhoud = _json.loads(meta_pad.read_text())
|
|
inhoud["values"].update({"puid": "1005", "pgid": "1006"})
|
|
meta_pad.write_text(_json.dumps(inhoud))
|
|
|
|
meta = backups.create("umami")
|
|
gevraagd = []
|
|
# `backups.os` ís de os-module, dus dit vangt ook wat tarfile zelf chownt —
|
|
# als root doet die dat namelijk wél. Daarom toetsen we op het exacte
|
|
# doelpad in plaats van op "alles wat langskwam".
|
|
monkeypatch.setattr(backups.os, "chown",
|
|
lambda p, u, g, **kw: gevraagd.append((str(p), u, g)))
|
|
ok, msg = backups.restore("umami", meta["file"])
|
|
|
|
assert ok, msg
|
|
doel = str(appdata / "umami")
|
|
assert (doel, 1005, 1006) in gevraagd, \
|
|
f"de teruggezette appdata kreeg niet de PUID van de stack: {gevraagd}"
|
|
|
|
|
|
def test_zonder_puid_valt_hij_terug_op_ons_eigen_account(env):
|
|
from core import backups
|
|
import os as _os
|
|
assert backups._verwachte_eigenaar({}) == (_os.getuid(), _os.getgid())
|
|
assert backups._verwachte_eigenaar({"puid": "1005", "pgid": "1006"}) == (1005, 1006)
|
|
# Een sjabloon met onzin erin mag geen TypeError geven.
|
|
assert backups._verwachte_eigenaar({"puid": "abc"}) == (_os.getuid(), _os.getgid())
|
|
|
|
|
|
def test_chown_die_niet_mag_stopt_het_terugzetten_niet(env, monkeypatch, tmp_path):
|
|
"""Draai je zonder de rechten om te chownen, dan is een waarschuwing beter
|
|
dan een mislukt herstel."""
|
|
from core import backups
|
|
monkeypatch.setattr(backups.os, "chown",
|
|
lambda *a, **k: (_ for _ in ()).throw(PermissionError("mag niet")))
|
|
map_ = tmp_path / "appdata-umami"
|
|
map_.mkdir()
|
|
(map_ / "a.txt").write_text("x")
|
|
regels = []
|
|
assert backups._zet_eigenaar(map_, 1005, 1006, log_fn=regels.append) is False
|
|
assert any("eigenaar" in r for r in regels), regels
|