server-up/tests/test_dbwachtwoord.py

"""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"