rothmanandClaude Opus 5.5
6733aacb6e
src/voxip_kiosk -> src/daynivo_kiosk, with every import, the launchers, install.sh, the Makefile and the manager's gunicorn app path. An upgrade replaces /usr/lib/voxip-kiosk/python as a whole, so the old package disappears; tested by upgrading an installation of the previous commit. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
329 lines
15 KiB
Python
329 lines
15 KiB
Python
import json
|
|
|
|
import pytest
|
|
|
|
from daynivo_kiosk.config import load_config
|
|
from daynivo_kiosk.web import auth
|
|
|
|
from .conftest import PASSWORD, csrf_from, token
|
|
|
|
|
|
# --- initial setup -------------------------------------------------------------
|
|
|
|
def test_everything_redirects_to_setup_until_configured(client, paths):
|
|
for path in ("/", "/kiosk", "/system", "/login", "/network", "/logs"):
|
|
resp = client.get(path)
|
|
assert resp.status_code == 302 and resp.headers["Location"].endswith("/setup"), path
|
|
assert paths.setup_code_file.exists()
|
|
assert oct(paths.setup_code_file.stat().st_mode & 0o777) == "0o640"
|
|
|
|
|
|
def test_setup_requires_code_and_strong_password(client, paths):
|
|
page = client.get("/setup")
|
|
tok = csrf_from(page)
|
|
code = paths.setup_code_file.read_text().strip()
|
|
|
|
resp = client.post("/setup", data={"csrf_token": tok, "setup_code": "AAAA-BBBB", "password": PASSWORD, "password_confirm": PASSWORD})
|
|
assert "setupcode klopt niet" in resp.get_data(as_text=True)
|
|
assert not paths.admin_file.exists()
|
|
|
|
resp = client.post("/setup", data={"csrf_token": tok, "setup_code": code, "password": "admin12345", "password_confirm": "admin12345"})
|
|
assert "te eenvoudig" in resp.get_data(as_text=True)
|
|
|
|
resp = client.post("/setup", data={"csrf_token": tok, "setup_code": code.lower().replace("-", ""), "password": PASSWORD, "password_confirm": PASSWORD})
|
|
assert resp.status_code == 302
|
|
stored = json.loads(paths.admin_file.read_text())
|
|
assert stored["password_hash"].startswith("$argon2id$")
|
|
assert PASSWORD not in paths.admin_file.read_text()
|
|
assert oct(paths.admin_file.stat().st_mode & 0o777) == "0o600"
|
|
assert not paths.setup_code_file.exists()
|
|
assert client.get("/").status_code == 200
|
|
assert client.get("/setup").status_code == 302
|
|
|
|
|
|
def test_setup_without_csrf_is_rejected(client, paths):
|
|
client.get("/setup")
|
|
code = paths.setup_code_file.read_text().strip()
|
|
resp = client.post("/setup", data={"setup_code": code, "password": PASSWORD, "password_confirm": PASSWORD})
|
|
assert resp.status_code == 303
|
|
assert not paths.admin_file.exists()
|
|
|
|
|
|
# --- login & session -------------------------------------------------------------
|
|
|
|
def test_login_logout(app, client):
|
|
app.extensions["voxip"].credentials.set_password(PASSWORD)
|
|
assert client.get("/").status_code == 302
|
|
tok = csrf_from(client.get("/login"))
|
|
assert "Onjuist" in client.post("/login", data={"csrf_token": tok, "password": "fout-wachtwoord"}).get_data(as_text=True)
|
|
assert client.post("/login", data={"csrf_token": tok, "password": PASSWORD}).status_code == 302
|
|
assert client.get("/").status_code == 200
|
|
client.post("/logout", data={"csrf_token": token(client)})
|
|
assert client.get("/").status_code == 302
|
|
|
|
|
|
def test_session_cookie_flags(app, client):
|
|
app.extensions["voxip"].credentials.set_password(PASSWORD)
|
|
tok = csrf_from(client.get("/login"))
|
|
resp = client.post("/login", data={"csrf_token": tok, "password": PASSWORD})
|
|
cookie = resp.headers["Set-Cookie"]
|
|
assert "HttpOnly" in cookie and "SameSite=Strict" in cookie
|
|
|
|
|
|
def test_login_throttling(app, client):
|
|
app.extensions["voxip"].credentials.set_password(PASSWORD)
|
|
tok = csrf_from(client.get("/login"))
|
|
for _ in range(5):
|
|
client.post("/login", data={"csrf_token": tok, "password": "verkeerd-wachtwoord"})
|
|
resp = client.post("/login", data={"csrf_token": tok, "password": PASSWORD})
|
|
assert "Te veel mislukte pogingen" in resp.get_data(as_text=True)
|
|
assert client.get("/").status_code == 302
|
|
|
|
|
|
def test_throttle_backoff_and_global_brake():
|
|
now = [0.0]
|
|
t = auth.LoginThrottle(clock=lambda: now[0], global_limit=8)
|
|
for _ in range(4):
|
|
t.failure("1.1.1.1")
|
|
assert t.retry_after("1.1.1.1") == 0
|
|
t.failure("1.1.1.1")
|
|
assert t.retry_after("1.1.1.1") == 30
|
|
t.failure("1.1.1.1")
|
|
assert t.retry_after("1.1.1.1") == 60
|
|
now[0] += 61
|
|
assert t.retry_after("1.1.1.1") == 0
|
|
t.success("1.1.1.1")
|
|
for i in range(8):
|
|
t.failure(f"10.0.0.{i}")
|
|
assert t.retry_after("192.168.0.99") > 0
|
|
|
|
|
|
def test_password_change_invalidates_other_sessions(app, logged_in):
|
|
other = app.test_client()
|
|
tok = csrf_from(other.get("/login"))
|
|
other.post("/login", data={"csrf_token": tok, "password": PASSWORD})
|
|
assert other.get("/").status_code == 200
|
|
|
|
new = "Nieuw-En-Lang-Wachtwoord-7"
|
|
resp = logged_in.post("/account/password", data={"csrf_token": token(logged_in), "current_password": PASSWORD,
|
|
"password": new, "password_confirm": new})
|
|
assert resp.status_code == 302
|
|
assert logged_in.get("/").status_code == 200
|
|
assert other.get("/").status_code == 302
|
|
|
|
|
|
def test_idle_session_expires(app, logged_in, monkeypatch):
|
|
from daynivo_kiosk.web import views
|
|
|
|
real = views.time.time
|
|
monkeypatch.setattr(views.time, "time", lambda: real() + views.IDLE_TIMEOUT + 5)
|
|
assert logged_in.get("/").status_code == 302
|
|
|
|
|
|
def test_password_policy():
|
|
assert auth.password_problem("kort") is not None
|
|
assert auth.password_problem("aaaaaaaaaaaa") is not None
|
|
assert auth.password_problem("Password123") is not None
|
|
assert auth.password_problem("lang-genoeg-en-uniek", "anders") == "De wachtwoorden komen niet overeen."
|
|
assert auth.password_problem("lang-genoeg-en-uniek", "lang-genoeg-en-uniek") is None
|
|
|
|
|
|
# --- network access & headers ------------------------------------------------------
|
|
|
|
@pytest.mark.parametrize("addr, allowed", [
|
|
("192.168.1.10", True), ("10.1.2.3", True), ("172.16.0.1", True), ("127.0.0.1", True), ("fe80::1", True),
|
|
("fd00::5", True), ("100.101.102.103", True), ("::ffff:192.168.1.10", True),
|
|
("8.8.8.8", False), ("2001:4860::8888", False), ("::ffff:8.8.8.8", False),
|
|
])
|
|
def test_public_clients_blocked(client, addr, allowed):
|
|
resp = client.get("/login", environ_base={"REMOTE_ADDR": addr})
|
|
assert (resp.status_code != 403) == allowed, addr
|
|
|
|
|
|
def test_same_subnet_global_ipv6_allowed():
|
|
from daynivo_kiosk.web.security import client_allowed
|
|
|
|
assert client_allowed("2a02:a210:1:2::99", False, ["2a02:a210:1:2::5/64"])
|
|
assert not client_allowed("2a02:a210:9:9::99", False, ["2a02:a210:1:2::5/64"])
|
|
assert client_allowed("8.8.8.8", True)
|
|
|
|
|
|
def test_security_headers(client):
|
|
resp = client.get("/setup")
|
|
assert "default-src 'self'" in resp.headers["Content-Security-Policy"]
|
|
assert resp.headers["X-Frame-Options"] == "DENY"
|
|
assert resp.headers["Cache-Control"] == "no-store"
|
|
|
|
|
|
def test_api_requires_login(app, client):
|
|
app.extensions["voxip"].credentials.set_password(PASSWORD)
|
|
resp = client.get("/api/status")
|
|
assert resp.status_code == 401 and resp.json["error"]
|
|
|
|
|
|
def test_post_without_csrf_rejected(logged_in, fake_helper):
|
|
resp = logged_in.post("/system/reboot")
|
|
assert resp.status_code == 303
|
|
assert fake_helper.calls == []
|
|
resp = logged_in.post("/system/reboot", data={"csrf_token": token(logged_in)}, headers={"Origin": "http://evil.example"})
|
|
assert fake_helper.calls == []
|
|
|
|
|
|
# --- pages ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.parametrize("path", ["/", "/kiosk", "/system", "/network", "/logs", "/logs?source=network&lines=50", "/account/password"])
|
|
def test_pages_render(logged_in, path):
|
|
resp = logged_in.get(path)
|
|
assert resp.status_code == 200, path
|
|
body = resp.get_data(as_text=True)
|
|
assert "Daynivo Kiosk Manager" in body and "9.9.9-test" in body
|
|
|
|
|
|
def test_dashboard_masks_token(logged_in, paths):
|
|
from daynivo_kiosk.config import save_config
|
|
|
|
save_config(paths.config_file, {"url": "https://portal.example.nl/display/hal?token=supergeheimtoken99"})
|
|
body = logged_in.get("/").get_data(as_text=True)
|
|
assert "supergeheimtoken" not in body and "token=…en99" in body
|
|
assert "supergeheimtoken" not in json.dumps(logged_in.get("/api/status").json)
|
|
|
|
|
|
def test_api_status_shape(logged_in):
|
|
data = logged_in.get("/api/status").json
|
|
for key in ("browser", "url", "uptime", "cpu", "temperature", "memory", "disk", "eth0", "wlan0", "network", "version"):
|
|
assert key in data
|
|
assert data["eth0"] == "192.168.10.20/24"
|
|
assert data["power"]["status"] == "warning"
|
|
|
|
|
|
# --- actions -------------------------------------------------------------------------
|
|
|
|
def test_kiosk_settings_valid_and_invalid(logged_in, paths):
|
|
tok = token(logged_in)
|
|
form = {"csrf_token": tok, "url": "https://nieuw.example.nl/scherm", "device_name": "Hal", "refresh_interval": "300",
|
|
"browser_restart": "1", "browser_profile": "ephemeral"}
|
|
logged_in.post("/kiosk/settings", data=form)
|
|
conf = load_config(paths.config_file)
|
|
assert conf["url"] == "https://nieuw.example.nl/scherm" and conf["refresh_interval"] == 300
|
|
assert conf["browser_profile"] == "ephemeral"
|
|
|
|
resp = logged_in.post("/kiosk/settings", data=dict(form, url="http://onveilig.example"), follow_redirects=True)
|
|
assert "Alleen HTTPS" in resp.get_data(as_text=True)
|
|
assert load_config(paths.config_file)["url"] == "https://nieuw.example.nl/scherm"
|
|
|
|
logged_in.post("/kiosk/settings", data={k: v for k, v in form.items() if k != "browser_restart"})
|
|
assert load_config(paths.config_file)["browser_restart"] is False
|
|
|
|
|
|
def test_reload_uses_devtools_not_restart(logged_in, fake_cdp, fake_helper):
|
|
logged_in.post("/kiosk/reload", data={"csrf_token": token(logged_in)})
|
|
assert fake_cdp.calls == [("reload", ())]
|
|
assert fake_helper.calls == []
|
|
|
|
|
|
def test_reload_failure_is_reported(logged_in, fake_cdp):
|
|
fake_cdp.fail = "DevTools niet bereikbaar"
|
|
resp = logged_in.post("/kiosk/reload", data={"csrf_token": token(logged_in)}, follow_redirects=True)
|
|
assert "Herladen mislukt" in resp.get_data(as_text=True)
|
|
|
|
|
|
@pytest.mark.parametrize("path, verb", [
|
|
("/kiosk/restart", "restart-browser"), ("/system/reboot", "reboot"), ("/system/poweroff", "poweroff"),
|
|
("/network/confirm", "network-confirm"), ("/network/rollback", "network-rollback"),
|
|
])
|
|
def test_helper_actions(logged_in, fake_helper, path, verb):
|
|
logged_in.post(path, data={"csrf_token": token(logged_in)})
|
|
assert fake_helper.calls[-1][0] == verb
|
|
|
|
|
|
def test_restart_and_reboot_are_separate(logged_in, fake_helper):
|
|
logged_in.post("/kiosk/restart", data={"csrf_token": token(logged_in)})
|
|
assert [c[0] for c in fake_helper.calls] == ["restart-browser"]
|
|
|
|
|
|
def test_hostname_passed_as_single_argument(logged_in, fake_helper):
|
|
logged_in.post("/system/hostname", data={"csrf_token": token(logged_in), "hostname": "Hal-BG"})
|
|
assert fake_helper.calls[-1] == ("set-hostname", ("hal-bg",), None)
|
|
|
|
|
|
def test_display_settings_saved_and_applied(logged_in, fake_helper, paths):
|
|
logged_in.post("/system/display", data={"csrf_token": token(logged_in), "rotation": "90", "resolution": "1280x720",
|
|
"zoom": "1,25", "screen_off_minutes": "10", "apply_now": "1"})
|
|
conf = load_config(paths.config_file)
|
|
assert conf["rotation"] == 90 and conf["resolution"] == "1280x720" and conf["zoom"] == 1.25
|
|
assert conf["screen_always_on"] is False
|
|
assert fake_helper.calls[-1][0] == "restart-browser"
|
|
|
|
|
|
def test_network_apply_validates_before_helper(logged_in, fake_helper):
|
|
tok = token(logged_in)
|
|
resp = logged_in.post("/network/apply", data={"csrf_token": tok, "interface": "eth0", "method": "static",
|
|
"address": "192.168.1.50", "gateway": "", "dns": ""}, follow_redirects=True)
|
|
assert "prefixlengte" in resp.get_data(as_text=True)
|
|
assert fake_helper.calls == []
|
|
logged_in.post("/network/apply", data={"csrf_token": tok, "interface": "wlan0", "method": "dhcp", "ssid": "Kantoor",
|
|
"psk": "geheim-wachtwoord", "dns": ""})
|
|
verb, args, payload = fake_helper.calls[-1]
|
|
assert verb == "network-apply" and args == ()
|
|
assert payload["wifi"] == {"ssid": "Kantoor", "psk": "geheim-wachtwoord", "hidden": False, "open": False}
|
|
|
|
|
|
def test_wifi_scan_json(logged_in, fake_helper):
|
|
fake_helper.responses["wifi-scan"] = {"networks": [{"ssid": "Kantoor", "signal": 70, "security": "WPA2", "in_use": True}]}
|
|
resp = logged_in.post("/api/network/scan", headers={"X-CSRF-Token": token(logged_in), "Accept": "application/json"})
|
|
assert resp.status_code == 200 and resp.json["networks"][0]["ssid"] == "Kantoor"
|
|
|
|
|
|
def test_helper_error_is_shown(logged_in, fake_helper):
|
|
fake_helper.fail = "Opdracht niet toegestaan"
|
|
resp = logged_in.post("/system/reboot", data={"csrf_token": token(logged_in)}, follow_redirects=True)
|
|
assert "Opdracht niet toegestaan" in resp.get_data(as_text=True)
|
|
|
|
|
|
def test_screenshot(logged_in, fake_cdp):
|
|
resp = logged_in.get("/kiosk/screenshot.jpg")
|
|
assert resp.status_code == 200 and resp.mimetype == "image/jpeg"
|
|
fake_cdp.fail = "weg"
|
|
assert logged_in.get("/kiosk/screenshot.jpg").status_code == 503
|
|
|
|
|
|
def test_audit_log_written(logged_in, paths):
|
|
logged_in.post("/kiosk/reload", data={"csrf_token": token(logged_in)})
|
|
events = [json.loads(line)["event"] for line in paths.audit_log.read_text().splitlines()]
|
|
assert "login" in events and "page_reload" in events
|
|
|
|
|
|
def test_admin_port_change_restarts_manager(logged_in, fake_helper, paths):
|
|
resp = logged_in.post("/system/admin", data={"csrf_token": token(logged_in), "admin_port": "8443"})
|
|
assert resp.status_code == 200 and ":8443/" in resp.get_data(as_text=True)
|
|
assert fake_helper.calls[-1][0] == "restart-manager"
|
|
assert load_config(paths.config_file)["admin_port"] == 8443
|
|
|
|
|
|
@pytest.mark.parametrize("path", ["/", "/kiosk", "/system", "/network", "/logs", "/account/password"])
|
|
def test_actions_menu_on_every_page(logged_in, path):
|
|
html = logged_in.get(path).get_data(as_text=True)
|
|
for action in ("/kiosk/reload", "/kiosk/restart", "/system/reboot", "/system/poweroff"):
|
|
assert f'action="{action}"' in html
|
|
|
|
|
|
def test_system_page_shows_update_state(logged_in, paths):
|
|
paths.update_state_file.write_text(json.dumps({"current": "1.1.0", "latest": "1.2.0", "available": True, "checked_at": 1_800_000_000}))
|
|
body = logged_in.get("/system").get_data(as_text=True)
|
|
assert "Software-updates" in body and "Bijwerken naar 1.2.0" in body
|
|
|
|
|
|
def test_auto_update_setting(logged_in, paths):
|
|
logged_in.post("/system/updates", data={"csrf_token": token(logged_in), "auto_update": "1"})
|
|
assert load_config(paths.config_file)["auto_update"] is True
|
|
logged_in.post("/system/updates", data={"csrf_token": token(logged_in)})
|
|
assert load_config(paths.config_file)["auto_update"] is False
|
|
|
|
|
|
def test_update_check_and_install_go_through_the_helper(logged_in, fake_helper):
|
|
fake_helper.responses["update-check"] = {"update": {"current": "1.1.0", "latest": "1.2.0", "available": True}}
|
|
resp = logged_in.post("/system/update/check", data={"csrf_token": token(logged_in)}, follow_redirects=True)
|
|
assert "Versie 1.2.0 is beschikbaar" in resp.get_data(as_text=True)
|
|
logged_in.post("/system/update/install", data={"csrf_token": token(logged_in)})
|
|
assert [c[0] for c in fake_helper.calls] == ["update-check", "update-install"]
|