07d5effcec
All checks were successful
Deploy server-up (dev) / deploy (push) Successful in 6m57s
Een mediamap koppelen of een USB-stick doorgeven kon alleen door het compose-bestand met de hand te bewerken, en dat werd bij de eerstvolgende 'Instellingen wijzigen' overschreven. - Nieuwe stap 'Extra's' in het invulmenu: per onderdeel een map van de host (optioneel alleen lezen) of een apparaat. - De keuzes staan in .serverup.json en worden ná het renderen toegepast, dus ze overleven elke wijziging. - /api/devices vindt seriële poorten (/dev/serial/by-id, ttyUSB*, ttyACM*) en videokaarten, met een 'vast'-label voor by-id-paden. - Het hostpad wordt vooraf gecontroleerd via /api/paths/check. - Docker-socket, /etc, /proc, /sys en /boot worden geweigerd met uitleg, in compose_transform en nog eens in de API. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01H9Vcwu7owJZ3nj5dCf4VFT
385 lines
16 KiB
Python
385 lines
16 KiB
Python
"""Instellingen wijzigen na installatie, koppelen, en het audit-filter."""
|
|
import json
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
WORTEL = Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0, str(WORTEL / "server-up"))
|
|
|
|
from conftest import login
|
|
|
|
|
|
def _wacht_op_job(client, jid, tijdslimiet=10):
|
|
"""Installeren gebeurt in een achtergrondthread; wachten tot hij klaar is."""
|
|
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 _stack(env, naam="web", meta=None, compose=None):
|
|
d = env["lib"] / naam
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
(d / "docker-compose.yml").write_text(
|
|
compose or "services:\n web:\n image: nginx\n ports:\n - \"8080:80\"\n")
|
|
if meta is not None:
|
|
(d / ".serverup.json").write_text(json.dumps(meta), encoding="utf-8")
|
|
return d
|
|
|
|
|
|
# ── Herconfigureren ──────────────────────────────────────────────────────────
|
|
|
|
def test_config_zonder_opgeslagen_waarden(client, env):
|
|
"""Stacks van vóór deze versie hebben geen keuzes bewaard; dan moet de
|
|
interface uitleggen waarom de knop niet werkt in plaats van te falen."""
|
|
login(client)
|
|
_stack(env, "oud", {"source": "oud"})
|
|
d = client.get("/api/stacks/oud/config").get_json()
|
|
assert d["ok"] is False and d["editable"] is False
|
|
assert "compose bewerken" in d["msg"]
|
|
|
|
|
|
def test_config_van_onbekende_stack(client):
|
|
login(client)
|
|
assert client.get("/api/stacks/bestaatniet/config").status_code == 404
|
|
|
|
|
|
def test_stacklijst_meldt_of_wijzigen_kan(client, env):
|
|
login(client)
|
|
_stack(env, "nieuw", {"source": "x", "values": {"port": 80}})
|
|
_stack(env, "oud", {"source": "x"})
|
|
per_naam = {s["name"]: s for s in client.get("/api/stacks").get_json()}
|
|
assert per_naam["nieuw"]["editable"] is True
|
|
assert per_naam["oud"]["editable"] is False
|
|
|
|
|
|
def test_reconfigure_zonder_bronsjabloon(client, env):
|
|
csrf = login(client)
|
|
_stack(env, "web", {"source": "bestaatniet", "repo_id": "x", "values": {"a": 1}})
|
|
r = client.post("/api/stacks/web/reconfigure", json={"values": {"a": 2}},
|
|
headers={"X-CSRF-Token": csrf})
|
|
assert r.status_code == 400
|
|
assert "Bronsjabloon" in r.get_json()["msg"]
|
|
|
|
|
|
def test_reconfigure_van_onbekende_stack(client):
|
|
csrf = login(client)
|
|
r = client.post("/api/stacks/bestaatniet/reconfigure", json={"values": {}},
|
|
headers={"X-CSRF-Token": csrf})
|
|
assert r.status_code == 404
|
|
|
|
|
|
def test_installatie_bewaart_de_gemaakte_keuzes(client, env, monkeypatch):
|
|
"""Zonder opgeslagen waarden valt het formulier later niet te heropenen.
|
|
|
|
Was een controle op een letterlijke regel broncode; die brak zodra de regel
|
|
verplaatste terwijl het gedrag klopte.
|
|
"""
|
|
import json as _json
|
|
from core import git
|
|
csrf = login(client)
|
|
wortel = git.cache_dir("server-up") / "apps" / "keuzeapp"
|
|
(wortel / "files").mkdir(parents=True, exist_ok=True)
|
|
(wortel / "template.json").write_text(_json.dumps({
|
|
"kind": "compose", "metadata": {"name": "keuzeapp"},
|
|
"variables": [{"title": "Algemeen", "items": [
|
|
{"name": "service_name", "type": "str", "title": "Naam",
|
|
"default": "keuzeapp", "required": True},
|
|
{"name": "port", "type": "int", "title": "Poort", "default": 8099},
|
|
]}]}), encoding="utf-8")
|
|
(wortel / "files" / "compose.yaml").write_text(
|
|
'services:\n << service_name >>:\n image: nginx\n'
|
|
' ports:\n - "<< port >>:80"\n', encoding="utf-8")
|
|
|
|
r = client.post("/api/store/install",
|
|
json={"stack": "keuzeapp", "repo_id": "server-up",
|
|
"instance": "kz", "values": {"port": 9123}},
|
|
headers={"X-CSRF-Token": csrf})
|
|
assert r.status_code == 200, r.get_json()
|
|
_wacht_op_job(client, r.get_json()["job_id"])
|
|
|
|
meta = _json.loads((env["lib"] / "kz" / ".serverup.json").read_text())
|
|
assert meta["values"]["port"] == 9123
|
|
assert meta["source"] == "keuzeapp"
|
|
|
|
|
|
def _installeer_keuzeapp(client, csrf, poort=9123):
|
|
"""Zet een sjabloon in de git-cache en installeer het als stack 'kz'."""
|
|
import json as _json
|
|
from core import git
|
|
wortel = git.cache_dir("server-up") / "apps" / "keuzeapp"
|
|
(wortel / "files").mkdir(parents=True, exist_ok=True)
|
|
(wortel / "template.json").write_text(_json.dumps({
|
|
"kind": "compose", "metadata": {"name": "keuzeapp"},
|
|
"variables": [{"title": "Algemeen", "items": [
|
|
{"name": "service_name", "type": "str", "title": "Naam",
|
|
"default": "keuzeapp", "required": True},
|
|
{"name": "port", "type": "int", "title": "Poort", "default": 8099},
|
|
]}]}), encoding="utf-8")
|
|
(wortel / "files" / "compose.yaml").write_text(
|
|
'services:\n << service_name >>:\n image: nginx\n'
|
|
' ports:\n - "<< port >>:80"\n', encoding="utf-8")
|
|
|
|
r = client.post("/api/store/install",
|
|
json={"stack": "keuzeapp", "repo_id": "server-up",
|
|
"instance": "kz", "values": {"port": poort}},
|
|
headers={"X-CSRF-Token": csrf})
|
|
assert r.status_code == 200, r.get_json()
|
|
_wacht_op_job(client, r.get_json()["job_id"])
|
|
|
|
|
|
def test_geslaagde_wijziging_wordt_niet_teruggedraaid(client, env, monkeypatch):
|
|
"""De job draait in een eigen thread, waar `request` niet meer bestaat.
|
|
|
|
Het adres voor het audit-logboek werd daar alsnog uitgelezen: een
|
|
RuntimeError precies nádat de stack met de nieuwe instellingen al draaide.
|
|
De foutafhandeling zette vervolgens de backup terug, dus een geslaagde
|
|
wijziging werd alsnog ongedaan gemaakt.
|
|
"""
|
|
import app as app_module
|
|
csrf = login(client)
|
|
_installeer_keuzeapp(client, csrf)
|
|
|
|
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": "kz_test.tar.gz"})
|
|
monkeypatch.setattr(app_module.backups, "restore",
|
|
lambda *a, **k: pytest.fail("backup werd teruggezet"))
|
|
|
|
r = client.post("/api/stacks/kz/reconfigure",
|
|
json={"values": {"service_name": "keuzeapp", "port": 9124},
|
|
"connect_apps": False},
|
|
headers={"X-CSRF-Token": csrf})
|
|
assert r.status_code == 200, r.get_json()
|
|
d = _wacht_op_job(client, r.get_json()["job_id"])
|
|
|
|
tekst = "\n".join(r["text"] for r in d.get("lines", []))
|
|
assert "RuntimeError" not in tekst, tekst
|
|
assert d.get("status") == "done", tekst
|
|
compose = next((p for p in (env["lib"] / "kz").iterdir()
|
|
if p.name in ("compose.yaml", "compose.yml",
|
|
"docker-compose.yml", "docker-compose.yaml")))
|
|
assert '"9124:80"' in compose.read_text()
|
|
|
|
|
|
def test_wijziging_wordt_vastgelegd_met_het_adres_van_de_aanvrager(
|
|
client, env, monkeypatch):
|
|
"""Het IP-adres hoort in het audit-logboek te belanden, niet verloren te
|
|
gaan doordat we het pas in de achtergrondthread opvragen."""
|
|
import app as app_module
|
|
from core import audit
|
|
audit.init()
|
|
csrf = login(client)
|
|
_installeer_keuzeapp(client, csrf)
|
|
|
|
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": "kz_test.tar.gz"})
|
|
|
|
r = client.post("/api/stacks/kz/reconfigure",
|
|
json={"values": {"service_name": "keuzeapp", "port": 9125},
|
|
"connect_apps": False},
|
|
headers={"X-CSRF-Token": csrf})
|
|
_wacht_op_job(client, r.get_json()["job_id"])
|
|
|
|
regels = [x for x in audit.query(50, src="stacks")
|
|
if x["action"] == "reconfigure"]
|
|
assert regels and regels[0]["status"] == "ok"
|
|
assert regels[0]["ip"]
|
|
|
|
|
|
# ── Koppelen ─────────────────────────────────────────────────────────────────
|
|
|
|
def test_connect_van_onbekende_stack(client):
|
|
csrf = login(client)
|
|
r = client.post("/api/stacks/bestaatniet/connect", json={"connect": True},
|
|
headers={"X-CSRF-Token": csrf})
|
|
assert r.status_code == 404
|
|
|
|
|
|
def test_connect_start_een_job(client, env):
|
|
csrf = login(client)
|
|
_stack(env, "web", {"source": "x"})
|
|
d = client.post("/api/stacks/web/connect", json={"connect": True},
|
|
headers={"X-CSRF-Token": csrf}).get_json()
|
|
assert d["ok"] and d["job_id"]
|
|
|
|
|
|
def test_viewer_mag_niet_herconfigureren(client, env):
|
|
csrf = login(client)
|
|
_stack(env, "web", {"source": "x", "values": {}})
|
|
client.post("/api/auth/users",
|
|
json={"username": "kijker", "password": "eenlangwachtwoord",
|
|
"role": "viewer"}, headers={"X-CSRF-Token": csrf})
|
|
kijker = client.post("/api/auth/login",
|
|
json={"username": "kijker", "password": "eenlangwachtwoord"}
|
|
).get_json()["csrf_token"]
|
|
assert client.post("/api/stacks/web/reconfigure", json={"values": {}},
|
|
headers={"X-CSRF-Token": kijker}).status_code == 403
|
|
|
|
|
|
# ── Audit-filter ─────────────────────────────────────────────────────────────
|
|
|
|
@pytest.fixture
|
|
def gevuld(env):
|
|
from core import audit
|
|
audit.init()
|
|
audit.log("auth", "login", "ok", ref="ramon", ip="10.0.0.1")
|
|
audit.log("auth", "login", "error", ref="indringer", ip="10.9.9.9")
|
|
audit.log("stacks", "start", "ok", ref="jellyfin")
|
|
audit.log("backups", "create", "ok", ref="jellyfin")
|
|
return audit
|
|
|
|
|
|
def test_filter_op_bron(gevuld):
|
|
assert len(gevuld.query(100, src="auth")) == 2
|
|
assert gevuld.count(src="auth") == 2
|
|
|
|
|
|
def test_filter_op_actie_en_status(gevuld):
|
|
r = gevuld.query(100, src="auth", action="login", status="error")
|
|
assert len(r) == 1 and r[0]["ref"] == "indringer"
|
|
|
|
|
|
def test_zoeken_in_referentie_en_ip(gevuld):
|
|
assert len(gevuld.query(100, zoek="jellyfin")) == 2
|
|
assert len(gevuld.query(100, zoek="10.9.9.9")) == 1
|
|
|
|
|
|
def test_filter_op_periode(gevuld):
|
|
assert gevuld.count(sinds=time.time() - 60) == 4
|
|
assert gevuld.count(sinds=time.time() + 60) == 0
|
|
|
|
|
|
def test_doorbladeren(gevuld):
|
|
eerste = gevuld.query(2, 0)
|
|
tweede = gevuld.query(2, 2)
|
|
assert len(eerste) == 2 and len(tweede) == 2
|
|
assert {r["id"] for r in eerste}.isdisjoint({r["id"] for r in tweede})
|
|
|
|
|
|
def test_facetten_voor_de_keuzelijsten(gevuld):
|
|
f = gevuld.facets()
|
|
assert set(f["src"]) == {"auth", "stacks", "backups"}
|
|
assert "login" in f["action"]
|
|
assert set(f["status"]) == {"ok", "error"}
|
|
|
|
|
|
def test_audit_endpoint_filtert(client, env, gevuld):
|
|
login(client)
|
|
d = client.get("/api/audit?src=auth&status=error").get_json()
|
|
assert d["total"] == 1
|
|
assert d["entries"][0]["ref"] == "indringer"
|
|
assert "auth" in d["facets"]["src"]
|
|
assert d["grand_total"] >= d["total"]
|
|
|
|
|
|
def test_audit_endpoint_limiet_wordt_begrensd(client, env, gevuld):
|
|
login(client)
|
|
d = client.get("/api/audit?limit=99999").get_json()
|
|
assert d["limit"] <= 500
|
|
|
|
|
|
# ── Eigen mappen en apparaten ────────────────────────────────────────────────
|
|
|
|
def _wijzig(client, csrf, extras, poort=9200):
|
|
return client.post("/api/stacks/kz/reconfigure",
|
|
json={"values": {"service_name": "keuzeapp", "port": poort},
|
|
"connect_apps": False, "extras": extras},
|
|
headers={"X-CSRF-Token": csrf})
|
|
|
|
|
|
def test_datamap_en_apparaat_belanden_in_de_compose(client, env, monkeypatch):
|
|
"""Een mediamap of USB-stick toevoegen kon alleen via 'compose bewerken',
|
|
en dat werd bij de eerstvolgende wijziging overschreven."""
|
|
import yaml as _yaml
|
|
import app as app_module
|
|
csrf = login(client)
|
|
_installeer_keuzeapp(client, csrf)
|
|
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"})
|
|
|
|
r = _wijzig(client, csrf, {
|
|
"volumes": [{"service": "keuzeapp", "host": "/srv/media",
|
|
"container": "/media", "ro": True}],
|
|
"devices": [{"service": "keuzeapp", "host": "/dev/ttyUSB0"}]})
|
|
assert r.status_code == 200, r.get_json()
|
|
d = _wacht_op_job(client, r.get_json()["job_id"])
|
|
assert d.get("status") == "done", d.get("lines")
|
|
|
|
compose = next(p for p in (env["lib"] / "kz").iterdir()
|
|
if p.name.startswith(("compose", "docker-compose")))
|
|
svc = _yaml.safe_load(compose.read_text())["services"]["keuzeapp"]
|
|
assert "/srv/media:/media:ro" in svc["volumes"]
|
|
assert "/dev/ttyUSB0:/dev/ttyUSB0" in svc["devices"]
|
|
|
|
# En ze staan in de metadata, dus een volgende wijziging behoudt ze.
|
|
meta = json.loads((env["lib"] / "kz" / ".serverup.json").read_text())
|
|
assert meta["extras"]["volumes"][0]["host"] == "/srv/media"
|
|
|
|
|
|
def test_extras_overleven_een_volgende_wijziging(client, env, monkeypatch):
|
|
"""Het renderen begint elke keer opnieuw uit het sjabloon; zonder bewaarde
|
|
extra's zou je map er na één instellingswijziging weer af zijn."""
|
|
import yaml as _yaml
|
|
import app as app_module
|
|
csrf = login(client)
|
|
_installeer_keuzeapp(client, csrf)
|
|
for naam in ("validate_compose",):
|
|
monkeypatch.setattr(app_module.docker, naam, 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"})
|
|
|
|
_wacht_op_job(client, _wijzig(client, csrf, {
|
|
"volumes": [{"service": "keuzeapp", "host": "/srv/media",
|
|
"container": "/media"}]}).get_json()["job_id"])
|
|
# Nu een wijziging zonder extras mee te sturen: de bewaarde keuze telt.
|
|
r = client.post("/api/stacks/kz/reconfigure",
|
|
json={"values": {"service_name": "keuzeapp", "port": 9201},
|
|
"connect_apps": False},
|
|
headers={"X-CSRF-Token": csrf})
|
|
_wacht_op_job(client, r.get_json()["job_id"])
|
|
|
|
compose = next(p for p in (env["lib"] / "kz").iterdir()
|
|
if p.name.startswith(("compose", "docker-compose")))
|
|
svc = _yaml.safe_load(compose.read_text())["services"]["keuzeapp"]
|
|
assert "/srv/media:/media" in svc["volumes"]
|
|
|
|
|
|
def test_gevaarlijk_pad_wordt_geweigerd_met_uitleg(client, env):
|
|
"""Niet stil corrigeren: wie /etc intikt hoort te horen waarom dat niet gaat."""
|
|
csrf = login(client)
|
|
_installeer_keuzeapp(client, csrf)
|
|
r = _wijzig(client, csrf, {
|
|
"volumes": [{"service": "keuzeapp", "host": "/var/run/docker.sock",
|
|
"container": "/var/run/docker.sock"}]})
|
|
assert r.status_code == 400
|
|
assert "Docker-socket" in r.get_json()["msg"]
|
|
|
|
r = _wijzig(client, csrf, {
|
|
"devices": [{"service": "keuzeapp", "host": "/etc/shadow"}]})
|
|
assert r.status_code == 400
|
|
assert "/dev/" in r.get_json()["msg"]
|
|
|
|
|
|
def test_apparatenlijst_is_op_te_vragen(client, env):
|
|
login(client)
|
|
d = client.get("/api/devices").get_json()
|
|
assert "devices" in d and isinstance(d["devices"], list)
|
|
for dev in d["devices"]:
|
|
assert dev["path"].startswith("/dev/")
|