"""Een gewijzigd databasewachtwoord moet ook in de database terechtkomen. `POSTGRES_PASSWORD` en verwanten worden door de officiële images alleen bij een lege datamap gebruikt. Wie dat veld in de instellingen wijzigde, kreeg een app met een nieuw wachtwoord en een database met het oude — en dus een stack die niet meer kon inloggen, met een foutmelding die nergens naar het wachtwoord verwees. """ import json import sys import time from pathlib import Path import pytest import yaml WORTEL = Path(__file__).resolve().parent.parent sys.path.insert(0, str(WORTEL / "server-up")) from conftest import login COMPOSE = """\ services: web: image: nginx environment: - DB_PASS=${DB_PASSWORD} web-db: image: postgres:16-alpine container_name: web-db environment: - POSTGRES_USER=app - POSTGRES_DB=app - POSTGRES_PASSWORD=${DB_PASSWORD} """ def _diensten(env_waarden): from core import backups return backups.diensten_uit_compose(COMPOSE, env_waarden) # ── Herkennen ──────────────────────────────────────────────────────────────── def test_gewijzigd_wachtwoord_wordt_herkend(env): from core import dbwachtwoord voor = _diensten({"DB_PASSWORD": "oud"}) na = _diensten({"DB_PASSWORD": "nieuw"}) veranderd = dbwachtwoord.wijzigingen(voor, na) assert len(veranderd) == 1 assert veranderd[0]["service"] == "web-db" assert veranderd[0]["oud_wachtwoord"] == "oud" assert veranderd[0]["wachtwoord"] == "nieuw" def test_gelijk_wachtwoord_levert_geen_werk_op(env): from core import dbwachtwoord zelfde = _diensten({"DB_PASSWORD": "geheim"}) assert dbwachtwoord.wijzigingen(zelfde, zelfde) == [] def test_nieuwe_database_telt_niet_mee(env): """Een dienst die er nog niet was heeft geen oud wachtwoord om mee in te loggen; die krijgt zijn wachtwoord gewoon bij de eerste start.""" from core import dbwachtwoord assert dbwachtwoord.wijzigingen([], _diensten({"DB_PASSWORD": "x"})) == [] # ── De commando's ──────────────────────────────────────────────────────────── def _cmd(soort, oud="oud", nieuw="nieuw", gebruiker="app"): from core import dbwachtwoord dienst = {"soort": soort, "gebruiker": gebruiker, "database": "app", "container": f"web-{soort}", "wachtwoord": nieuw, "oud_wachtwoord": oud} return dbwachtwoord.commando(dienst, nieuw) def test_postgres_zet_het_wachtwoord_om(): cmd, omgeving = _cmd("postgres", oud="Ou6-oudgeheim", nieuw="Zg3-nieuwgeheim") assert "ALTER USER app WITH PASSWORD" in omgeving["SU_SQL"] assert omgeving["PGPASSWORD"] == "Ou6-oudgeheim" assert "| psql -v ON_ERROR_STOP=1 -U app -d app" in cmd[-1] # Geen van beide wachtwoorden staat in het commando zelf: argumenten zijn # op de host zichtbaar in de procestabel. assert "Zg3-nieuwgeheim" not in cmd[-1] assert "Ou6-oudgeheim" not in cmd[-1] def test_postgres_laat_een_backslash_met_rust(): """Postgres neemt een backslash letterlijk; verdubbelen zou het wachtwoord veranderen en de app alsnog buitensluiten.""" _, omgeving = _cmd("postgres", nieuw="pas\\woord") assert omgeving["SU_SQL"] == "ALTER USER app WITH PASSWORD 'pas\\woord';" _, mysql_omgeving = _cmd("mysql", nieuw="pas\\woord") assert "pas\\\\woord" in mysql_omgeving["SU_SQL"] def test_mysql_en_mariadb(): for soort, client in (("mysql", "mysql"), ("mariadb", "mariadb")): cmd, omgeving = _cmd(soort, oud="Ou6-oudgeheim", nieuw="Zg3-nieuwgeheim") assert f"| {client} -u app" in cmd[-1] assert "ALTER USER USER() IDENTIFIED BY" in omgeving["SU_SQL"] assert omgeving["SU_OUD"] == "Ou6-oudgeheim" # Via stdin, dus geen wachtwoord in de procestabel van de host. assert "Zg3-nieuwgeheim" not in cmd[-1] assert "Ou6-oudgeheim" not in cmd[-1] def test_mongo_werkt_met_mongo_en_mongosh(): cmd, omgeving = _cmd("mongo", gebruiker="root", oud="Ou6-oud", nieuw="Zg3-nieuw") assert "updateUser" in omgeving["SU_JS"] assert "mongosh" in cmd[-1] and "bin=mongo" in cmd[-1] assert "Zg3-nieuw" not in cmd[-1] and "Ou6-oud" not in cmd[-1] def test_aanhalingstekens_breken_het_statement_niet(): """Een wachtwoord met een quote erin mag geen SQL of JavaScript openbreken.""" _, omgeving = _cmd("mysql", nieuw="pas'woord\\raar") assert "pas''woord" in omgeving["SU_SQL"] assert "\\\\raar" in omgeving["SU_SQL"] _, mongo_omgeving = _cmd("mongo", nieuw='pas"woord\\raar') assert '\\"woord' in mongo_omgeving["SU_JS"] def test_onbekend_soort_geeft_een_duidelijke_fout(): from core import dbwachtwoord with pytest.raises(dbwachtwoord.WachtwoordFout): dbwachtwoord.commando({"soort": "sqlite", "gebruiker": "", "oud_wachtwoord": "x", "container": "c"}, "y") # ── Omzetten in de container ───────────────────────────────────────────────── def test_gestopte_database_wordt_gemeld(env, monkeypatch): from core import dbwachtwoord, docker monkeypatch.setattr(docker, "container_draait", lambda n: False) ok, msg = dbwachtwoord.omzetten( {"soort": "postgres", "gebruiker": "app", "database": "app", "container": "web-db", "oud_wachtwoord": "oud"}, "nieuw") assert ok is False assert "draait niet" in msg def test_wachtwoord_lekt_niet_in_de_foutmelding(env, monkeypatch): """Een joblog is zichtbaar voor iedereen die de app mag bedienen.""" from core import dbwachtwoord, docker class Mislukt: returncode = 1 stdout = "" stderr = 'FATAL: password authentication failed voor "supergeheim123"' monkeypatch.setattr(docker, "container_draait", lambda n: True) monkeypatch.setattr(docker, "_run", lambda *a, **k: Mislukt()) ok, msg = dbwachtwoord.omzetten( {"soort": "postgres", "gebruiker": "app", "database": "app", "container": "web-db", "oud_wachtwoord": "supergeheim123"}, "nieuw") assert ok is False assert "supergeheim123" not in msg assert "***" in msg # ── De hele weg door de wijziging heen ─────────────────────────────────────── def _sjabloon(client, csrf): from core import git wortel = git.cache_dir("server-up") / "apps" / "dbapp" (wortel / "files").mkdir(parents=True, exist_ok=True) (wortel / "template.json").write_text(json.dumps({ "kind": "compose", "metadata": {"name": "dbapp"}, "variables": [{"title": "Algemeen", "items": [ {"name": "service_name", "type": "str", "title": "Naam", "default": "dbapp", "required": True}, {"name": "db_password", "type": "str", "title": "Wachtwoord", "default": "", "required": True, "secret": True, "generate": "auto"}, ]}]}), encoding="utf-8") (wortel / "files" / "compose.yaml").write_text( 'services:\n' ' << service_name >>:\n image: nginx\n' ' << service_name >>-db:\n' ' image: postgres:16-alpine\n' ' container_name: << service_name >>-db\n' ' environment:\n' ' - POSTGRES_USER=app\n' ' - POSTGRES_DB=app\n' ' - POSTGRES_PASSWORD=<< db_password >>\n', encoding="utf-8") r = client.post("/api/store/install", json={"stack": "dbapp", "repo_id": "server-up", "instance": "dbapp", "connect_apps": False, "values": {"service_name": "dbapp", "db_password": "eerstewachtwoord"}}, headers={"X-CSRF-Token": csrf}) assert r.status_code == 200, r.get_json() _wacht(client, r.get_json()["job_id"]) def _wacht(client, jid, tijdslimiet=10): einde = time.time() + tijdslimiet while time.time() < einde: d = client.get(f"/api/jobs/{jid}").get_json() or {} if d.get("status") in ("done", "error"): return d time.sleep(0.05) return {} def _stub_docker(monkeypatch, app_module, omzetten_lukt=True): monkeypatch.setattr(app_module.docker, "validate_compose", lambda *a, **k: (True, "")) monkeypatch.setattr(app_module.docker, "compose_down", lambda *a, **k: 0) monkeypatch.setattr(app_module.docker, "compose_up", lambda *a, **k: 0) monkeypatch.setattr(app_module.backups, "create", lambda *a, **k: {"file": "x.tar.gz"}) gedaan = [] def _omzetten(dienst, nieuw, log_fn=None): gedaan.append((dienst["container"], dienst["oud_wachtwoord"], nieuw)) if omzetten_lukt: if log_fn: log_fn("Wachtwoord omgezet") return True, "" return False, "Wachtwoord van web-db niet gewijzigd: verbinding geweigerd" monkeypatch.setattr(app_module.dbwachtwoord, "omzetten", _omzetten) return gedaan def test_wijziging_zet_het_wachtwoord_ook_in_de_database(client, env, monkeypatch): import app as app_module csrf = login(client) _sjabloon(client, csrf) gedaan = _stub_docker(monkeypatch, app_module) r = client.post("/api/stacks/dbapp/reconfigure", json={"values": {"service_name": "dbapp", "db_password": "tweedewachtwoord"}, "connect_apps": False}, headers={"X-CSRF-Token": csrf}) assert r.status_code == 200, r.get_json() d = _wacht(client, r.get_json()["job_id"]) assert d.get("status") == "done", d.get("lines") assert gedaan == [("dbapp-db", "eerstewachtwoord", "tweedewachtwoord")] def test_zonder_wachtwoordwijziging_gebeurt_er_niets_met_de_database( client, env, monkeypatch): import app as app_module csrf = login(client) _sjabloon(client, csrf) gedaan = _stub_docker(monkeypatch, app_module) r = client.post("/api/stacks/dbapp/reconfigure", json={"values": {"service_name": "dbapp", "db_password": "eerstewachtwoord"}, "connect_apps": False}, headers={"X-CSRF-Token": csrf}) _wacht(client, r.get_json()["job_id"]) assert gedaan == [] def test_mislukte_omzetting_breekt_de_wijziging_af(client, env, monkeypatch): """Een instelling die niet doorgaat is beter dan een app die niet meer bij zijn gegevens kan.""" import app as app_module csrf = login(client) _sjabloon(client, csrf) _stub_docker(monkeypatch, app_module, omzetten_lukt=False) hersteld = [] monkeypatch.setattr(app_module.backups, "restore", lambda *a, **k: hersteld.append(a) or (True, "ok")) r = client.post("/api/stacks/dbapp/reconfigure", json={"values": {"service_name": "dbapp", "db_password": "derdewachtwoord"}, "connect_apps": False}, headers={"X-CSRF-Token": csrf}) d = _wacht(client, r.get_json()["job_id"]) assert d.get("status") == "error" tekst = "\n".join(x["text"] for x in d.get("lines", [])) assert "niet meer bij zijn database" in tekst assert hersteld, "de backup hoort teruggezet te worden"