rothmanandClaude Opus 5.5
eb0242c182
Services (daynivo-kiosk-*, daynivo-network-guard, daynivo-boot-import), programs in /usr/lib/daynivo-kiosk, folders /etc, /var/lib, /var/log and /var/cache daynivo-kiosk*, the system user daynivo, DAYNIVO_* settings, the plymouth theme, network profiles daynivo-eth0/wlan0, the image stage and its first user daynivoadmin. install.sh moves a kiosk installed up to 1.2.x over once: old services stopped and removed, folders moved (configuration, password, browser profile, logs) with symlinks under the old names, the voxip user renamed with the same uid, network profiles renamed unless a change awaits confirmation, and old leftovers removed. Tested by upgrading a configured 1.2.0 installation in a container, and running the install a second time. Old names keep working where people type or place them: voxip-config.json, voxip-authorized_keys, voxip-network-reset, voxip-kiosk-ctl, and VOXIP_DEFAULT_URL / VOXIP_DEVICE_NAME for image builds. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
604 lines
22 KiB
Python
604 lines
22 KiB
Python
"""Routes of the Daynivo Kiosk Manager."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import socket
|
||
import time
|
||
from typing import Any
|
||
|
||
from flask import (Blueprint, Response, abort, current_app, flash, g, jsonify, redirect, render_template, request,
|
||
session, url_for)
|
||
|
||
from .. import netconfig, netinfo, services, sysinfo
|
||
from ..agent import read_state
|
||
from ..updater import read_state as read_update_state
|
||
from ..cdp import CDPError
|
||
from ..config import ConfigError, load_config, save_config
|
||
from ..helper_client import HelperError
|
||
from ..validation import ValidationError, mask_url
|
||
from . import security
|
||
from .auth import password_problem
|
||
|
||
bp = Blueprint("ui", __name__)
|
||
audit_log = logging.getLogger("daynivo-kiosk-audit")
|
||
|
||
IDLE_TIMEOUT = 3600
|
||
PUBLIC_ENDPOINTS = {"ui.login", "ui.setup", "static"}
|
||
|
||
|
||
def svc():
|
||
return current_app.extensions["daynivo"]
|
||
|
||
|
||
def cfg() -> dict[str, Any]:
|
||
if "cfg" not in g:
|
||
g.cfg = load_config(svc().paths.config_file)
|
||
return g.cfg
|
||
|
||
|
||
def client_ip() -> str:
|
||
return request.remote_addr or "?"
|
||
|
||
|
||
def audit(event: str, **details: Any) -> None:
|
||
record = {"time": int(time.time()), "event": event, "ip": client_ip(), **details}
|
||
audit_log.info(json.dumps(record, ensure_ascii=False))
|
||
try:
|
||
with open(svc().paths.audit_log, "a", encoding="utf-8") as fh:
|
||
fh.write(json.dumps(record, ensure_ascii=False) + "\n")
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
def wants_json() -> bool:
|
||
return request.path.startswith("/api/") or request.headers.get("Accept", "").startswith("application/json")
|
||
|
||
|
||
_local_nets_cache: tuple[float, list[str]] = (0.0, [])
|
||
|
||
|
||
def local_networks() -> list[str]:
|
||
global _local_nets_cache
|
||
now = time.monotonic()
|
||
if now - _local_nets_cache[0] > 60:
|
||
nets = [a for addrs in sysinfo.ip_addresses().values() for a in addrs]
|
||
_local_nets_cache = (now, nets)
|
||
return _local_nets_cache[1]
|
||
|
||
|
||
# -- request pipeline ----------------------------------------------------------
|
||
|
||
@bp.before_app_request
|
||
def guard():
|
||
if not security.client_allowed(request.remote_addr, cfg()["admin_allow_public"], local_networks()):
|
||
abort(403)
|
||
if request.endpoint == "static":
|
||
return None
|
||
creds = svc().credentials
|
||
if not creds.is_configured():
|
||
if request.endpoint != "ui.setup":
|
||
return redirect(url_for("ui.setup"))
|
||
elif request.endpoint == "ui.setup":
|
||
return redirect(url_for("ui.dashboard"))
|
||
|
||
if session.get("auth"):
|
||
expired = time.time() - session.get("seen", 0) > IDLE_TIMEOUT
|
||
if expired or session.get("epoch") != creds.session_epoch():
|
||
session.clear()
|
||
flash("Je sessie is verlopen. Log opnieuw in.", "info")
|
||
if request.endpoint not in PUBLIC_ENDPOINTS and not session.get("auth"):
|
||
if wants_json():
|
||
return jsonify(error="Niet ingelogd"), 401
|
||
return redirect(url_for("ui.login"))
|
||
if session.get("auth"):
|
||
session["seen"] = int(time.time())
|
||
|
||
if request.method == "POST" and not security.csrf_valid(request):
|
||
audit("csrf_rejected", path=request.path)
|
||
if wants_json():
|
||
return jsonify(error="Beveiligingstoken ongeldig. Herlaad de pagina."), 400
|
||
flash("Beveiligingstoken ongeldig of verlopen. Probeer het opnieuw.", "error")
|
||
return redirect(request.path if request.endpoint in PUBLIC_ENDPOINTS else url_for("ui.dashboard")), 303
|
||
return None
|
||
|
||
|
||
@bp.after_app_request
|
||
def headers(response: Response):
|
||
return security.apply_headers(response, secure=current_app.config["SESSION_COOKIE_SECURE"])
|
||
|
||
|
||
@bp.app_context_processor
|
||
def template_globals():
|
||
return {
|
||
"csrf_token": security.csrf_token,
|
||
"version": svc().version,
|
||
"device_name": cfg()["device_name"],
|
||
"hostname": socket.gethostname().split(".")[0],
|
||
"logged_in": bool(session.get("auth")),
|
||
"admin_port": cfg()["admin_port"],
|
||
"admin_scheme": "https" if cfg()["admin_https"] else "http",
|
||
}
|
||
|
||
|
||
@bp.app_errorhandler(403)
|
||
def forbidden(_):
|
||
return render_template("error.html", title="Geen toegang",
|
||
message="De beheerinterface is alleen bereikbaar vanaf het lokale netwerk of via VPN."), 403
|
||
|
||
|
||
@bp.app_errorhandler(404)
|
||
def not_found(_):
|
||
return render_template("error.html", title="Niet gevonden", message="Deze pagina bestaat niet."), 404
|
||
|
||
|
||
@bp.app_errorhandler(413)
|
||
def too_large(_):
|
||
return render_template("error.html", title="Te groot", message="Het verzoek is te groot."), 413
|
||
|
||
|
||
# -- setup & login --------------------------------------------------------------
|
||
|
||
@bp.route("/setup", methods=["GET", "POST"])
|
||
def setup():
|
||
creds = svc().credentials
|
||
error = None
|
||
if request.method == "POST":
|
||
ip = client_ip()
|
||
wait = svc().throttle.retry_after(ip)
|
||
password = request.form.get("password", "")
|
||
if wait:
|
||
error = f"Te veel pogingen. Probeer het over {wait} seconden opnieuw."
|
||
elif not creds.check_setup_code(request.form.get("setup_code", "")):
|
||
svc().throttle.failure(ip)
|
||
audit("setup_code_rejected")
|
||
error = "De setupcode klopt niet. Je vindt de code op het scherm van de kiosk."
|
||
else:
|
||
error = password_problem(password, request.form.get("password_confirm", ""))
|
||
if error is None:
|
||
epoch = creds.set_password(password)
|
||
svc().throttle.success(ip)
|
||
session.clear()
|
||
session.update(auth=True, epoch=epoch, seen=int(time.time()))
|
||
session.permanent = True
|
||
audit("setup_completed")
|
||
flash("Het beheerderswachtwoord is ingesteld. Welkom!", "success")
|
||
return redirect(url_for("ui.dashboard"))
|
||
return render_template("setup.html", error=error)
|
||
|
||
|
||
@bp.route("/login", methods=["GET", "POST"])
|
||
def login():
|
||
if session.get("auth"):
|
||
return redirect(url_for("ui.dashboard"))
|
||
error = None
|
||
if request.method == "POST":
|
||
ip = client_ip()
|
||
wait = svc().throttle.retry_after(ip)
|
||
if wait:
|
||
error = f"Te veel mislukte pogingen. Probeer het over {wait} seconden opnieuw."
|
||
elif svc().credentials.verify(request.form.get("password", "")):
|
||
svc().throttle.success(ip)
|
||
session.clear()
|
||
session.update(auth=True, epoch=svc().credentials.session_epoch(), seen=int(time.time()))
|
||
session.permanent = True
|
||
audit("login")
|
||
return redirect(url_for("ui.dashboard"))
|
||
else:
|
||
svc().throttle.failure(ip)
|
||
audit("login_failed")
|
||
error = "Onjuist wachtwoord."
|
||
return render_template("login.html", error=error)
|
||
|
||
|
||
@bp.post("/logout")
|
||
def logout():
|
||
audit("logout")
|
||
session.clear()
|
||
flash("Je bent uitgelogd.", "info")
|
||
return redirect(url_for("ui.login"))
|
||
|
||
|
||
@bp.route("/account/password", methods=["GET", "POST"])
|
||
def change_password():
|
||
if request.method == "POST":
|
||
creds = svc().credentials
|
||
if not creds.verify(request.form.get("current_password", "")):
|
||
audit("password_change_failed")
|
||
flash("Het huidige wachtwoord klopt niet.", "error")
|
||
else:
|
||
password = request.form.get("password", "")
|
||
problem = password_problem(password, request.form.get("password_confirm", ""))
|
||
if problem:
|
||
flash(problem, "error")
|
||
else:
|
||
epoch = creds.set_password(password)
|
||
session["epoch"] = epoch
|
||
audit("password_changed")
|
||
flash("Wachtwoord gewijzigd. Andere sessies zijn uitgelogd.", "success")
|
||
return redirect(url_for("ui.dashboard"))
|
||
return render_template("password.html", active="account")
|
||
|
||
|
||
# -- dashboard -------------------------------------------------------------------
|
||
|
||
def browser_state(units: dict[str, Any], agent: dict[str, Any]) -> tuple[str, str]:
|
||
unit = units.get("browser", {})
|
||
if unit.get("status") != "ok":
|
||
return unit.get("status", "unknown"), unit.get("label", "Onbekend")
|
||
if agent.get("browser") == "unreachable":
|
||
return "starting", "Bezig met starten"
|
||
page = agent.get("page")
|
||
if page == "offline":
|
||
return "warning", "Actief — offlinepagina"
|
||
if page == "setup":
|
||
return "warning", "Actief — wacht op initiële configuratie"
|
||
if page in ("start", "blank", "error"):
|
||
return "starting", "Pagina wordt geladen"
|
||
return "ok", "Actief"
|
||
|
||
|
||
def collect_status() -> dict[str, Any]:
|
||
conf = cfg()
|
||
paths = svc().paths
|
||
agent = read_state(paths.agent_state_file)
|
||
units = services.summary()
|
||
b_status, b_label = browser_state(units, agent)
|
||
mem = sysinfo.memory()
|
||
disk = sysinfo.disk("/")
|
||
temp = sysinfo.temperature_c()
|
||
load = sysinfo.load_average()
|
||
cpu = sysinfo.cpu_percent()
|
||
throttle = sysinfo.throttled()
|
||
reachable = agent.get("target_reachable")
|
||
if reachable is None:
|
||
net_status, net_label = "unknown", "Nog niet gecontroleerd"
|
||
elif reachable:
|
||
net_status, net_label = "ok", "Online — kiosk-URL bereikbaar"
|
||
else:
|
||
net_status, net_label = "error", f"Offline — {agent.get('reason') or 'niet bereikbaar'}"
|
||
addresses = sysinfo.ip_addresses()
|
||
return {
|
||
"browser": {"status": b_status, "label": b_label, "restarts": units.get("browser", {}).get("restarts", "0")},
|
||
"agent": units.get("agent", {}),
|
||
"url": mask_url(conf["url"]),
|
||
"refresh_interval": conf["refresh_interval"],
|
||
"uptime": sysinfo.format_duration(sysinfo.uptime_seconds()),
|
||
"cpu": f"{cpu:.0f}%".replace(".", ",") if cpu is not None else "–",
|
||
"cpu_percent": cpu or 0,
|
||
"load": " / ".join(f"{x:.2f}".replace(".", ",") for x in load) if load else "–",
|
||
"temperature": f"{temp:.1f} °C".replace(".", ",") if temp is not None else "–",
|
||
"temperature_status": "error" if temp and temp >= 80 else "warning" if temp and temp >= 70 else "ok",
|
||
"memory": f"{sysinfo.format_bytes(mem['used'])} van {sysinfo.format_bytes(mem['total'])}" if mem else "–",
|
||
"memory_percent": mem["percent"] if mem else 0,
|
||
"disk": f"{sysinfo.format_bytes(disk['used'])} van {sysinfo.format_bytes(disk['total'])}" if disk else "–",
|
||
"disk_percent": disk["percent"] if disk else 0,
|
||
"eth0": ", ".join(addresses.get("eth0", [])) or "Niet verbonden",
|
||
"wlan0": ", ".join(addresses.get("wlan0", [])) or "Niet verbonden",
|
||
"network": {"status": net_status, "label": net_label},
|
||
"power": {"status": "error" if throttle and throttle["now"] else "warning" if throttle and throttle["ever"] else "ok",
|
||
"label": ", ".join(throttle["flags"]) if throttle and throttle["flags"] else "In orde"} if throttle else None,
|
||
"last_error": agent.get("last_error"),
|
||
"version": svc().version,
|
||
"os": sysinfo.os_release(),
|
||
"model": sysinfo.model(),
|
||
}
|
||
|
||
|
||
@bp.get("/")
|
||
def dashboard():
|
||
return render_template("dashboard.html", active="dashboard", s=collect_status())
|
||
|
||
|
||
@bp.get("/api/status")
|
||
def api_status():
|
||
return jsonify(collect_status())
|
||
|
||
|
||
# -- kiosk -----------------------------------------------------------------------
|
||
|
||
@bp.get("/kiosk")
|
||
def kiosk():
|
||
agent = read_state(svc().paths.agent_state_file)
|
||
return render_template("kiosk.html", active="kiosk", conf=cfg(), agent=agent,
|
||
errors=services.recent_errors(25), masked_url=mask_url(cfg()["url"]),
|
||
s={"browser": dict(zip(("status", "label"), browser_state(services.summary(), agent)))})
|
||
|
||
|
||
def _form_bool(name: str) -> bool:
|
||
return request.form.get(name) in ("1", "on", "true", "yes")
|
||
|
||
|
||
def _save(updates: dict[str, Any]) -> dict[str, Any] | None:
|
||
try:
|
||
return save_config(svc().paths.config_file, updates)
|
||
except ConfigError as exc:
|
||
for message in exc.errors.values():
|
||
flash(message, "error")
|
||
except OSError as exc:
|
||
flash(f"Opslaan mislukt: {exc}", "error")
|
||
return None
|
||
|
||
|
||
@bp.post("/kiosk/settings")
|
||
def kiosk_settings():
|
||
updates = {
|
||
"url": request.form.get("url", ""),
|
||
"device_name": request.form.get("device_name", ""),
|
||
"refresh_interval": request.form.get("refresh_interval", "0"),
|
||
"browser_restart": _form_bool("browser_restart"),
|
||
"browser_profile": request.form.get("browser_profile", "persistent"),
|
||
}
|
||
old = cfg()
|
||
new = _save(updates)
|
||
if new is not None:
|
||
changed = sorted(k for k in updates if old.get(k) != new.get(k))
|
||
audit("kiosk_settings", changed=changed, url=mask_url(new["url"]))
|
||
message = "Instellingen opgeslagen."
|
||
if "url" in changed:
|
||
message += " De kiosk opent de nieuwe URL binnen enkele seconden."
|
||
if "browser_profile" in changed:
|
||
message += " Het profieltype geldt na het herstarten van Chromium."
|
||
flash(message, "success")
|
||
return redirect(url_for("ui.kiosk"))
|
||
|
||
|
||
def _action_response(ok: bool, message: str, back: str):
|
||
if wants_json():
|
||
return jsonify(ok=ok, message=message), (200 if ok else 500)
|
||
flash(message, "success" if ok else "error")
|
||
return redirect(back)
|
||
|
||
|
||
@bp.post("/kiosk/reload")
|
||
def kiosk_reload():
|
||
try:
|
||
svc().cdp.reload()
|
||
audit("page_reload")
|
||
return _action_response(True, "De pagina wordt herladen.", request.referrer or url_for("ui.dashboard"))
|
||
except CDPError as exc:
|
||
current_app.logger.warning("Herladen mislukt: %s", exc)
|
||
return _action_response(
|
||
False,
|
||
"Herladen mislukt: Chromium reageert niet. Als hij net (her)start, probeer het over een halve minuut opnieuw.",
|
||
request.referrer or url_for("ui.dashboard"),
|
||
)
|
||
|
||
|
||
def _helper_action(event: str, verb: str, *args: str, back: str):
|
||
try:
|
||
result = svc().helper.call(verb, *args)
|
||
audit(event, **({"args": list(args)} if args else {}))
|
||
return _action_response(True, result.get("message", "Uitgevoerd."), back)
|
||
except HelperError as exc:
|
||
audit(event + "_failed", error=str(exc))
|
||
return _action_response(False, str(exc), back)
|
||
|
||
|
||
@bp.post("/kiosk/restart")
|
||
def kiosk_restart():
|
||
return _helper_action("browser_restart", "restart-browser", back=request.referrer or url_for("ui.dashboard"))
|
||
|
||
|
||
@bp.get("/kiosk/screenshot.jpg")
|
||
def kiosk_screenshot():
|
||
try:
|
||
data = svc().cdp.screenshot(quality=70)
|
||
except CDPError as exc:
|
||
return Response(f"Screenshot niet beschikbaar: {exc}", status=503, mimetype="text/plain")
|
||
return Response(data, mimetype="image/jpeg")
|
||
|
||
|
||
# -- system ----------------------------------------------------------------------
|
||
|
||
def display_info() -> dict[str, Any]:
|
||
try:
|
||
return json.loads(svc().paths.display_info_file.read_text(encoding="utf-8"))
|
||
except (OSError, ValueError):
|
||
return {}
|
||
|
||
|
||
@bp.get("/system")
|
||
def system():
|
||
info = display_info()
|
||
outputs = info.get("outputs") or []
|
||
modes = outputs[0]["modes"] if outputs else []
|
||
return render_template("system.html", active="system", conf=cfg(), modes=modes,
|
||
output=outputs[0] if outputs else None, s=collect_status(), update=update_view())
|
||
|
||
|
||
def _when(timestamp: Any) -> str | None:
|
||
if not isinstance(timestamp, (int, float)):
|
||
return None
|
||
return time.strftime("%d-%m-%Y %H:%M", time.localtime(timestamp))
|
||
|
||
|
||
def update_view() -> dict[str, Any]:
|
||
"""The updater's last result (update.json, written by root) in display form."""
|
||
state = read_update_state(svc().paths)
|
||
return {
|
||
**state,
|
||
"checked": _when(state.get("checked_at")),
|
||
"finished": _when(state.get("finished_at")),
|
||
# A run that started more than two hours ago and never finished is reported as failed.
|
||
"running": state.get("status") == "updating" and time.time() - state.get("started_at", 0) < 7200,
|
||
}
|
||
|
||
|
||
@bp.post("/system/display")
|
||
def system_display():
|
||
updates = {
|
||
"rotation": request.form.get("rotation", "0"),
|
||
"resolution": request.form.get("resolution", "auto"),
|
||
"zoom": request.form.get("zoom", "1").replace(",", "."),
|
||
"screen_always_on": _form_bool("screen_always_on"),
|
||
"screen_off_minutes": request.form.get("screen_off_minutes", "10"),
|
||
}
|
||
if _save(updates) is None:
|
||
return redirect(url_for("ui.system"))
|
||
audit("display_settings", **{k: v for k, v in updates.items()})
|
||
if _form_bool("apply_now"):
|
||
return _helper_action("browser_restart", "restart-browser", back=url_for("ui.system"))
|
||
flash("Scherminstellingen opgeslagen. Ze worden actief bij de volgende herstart van Chromium.", "success")
|
||
return redirect(url_for("ui.system"))
|
||
|
||
|
||
@bp.post("/system/hostname")
|
||
def system_hostname():
|
||
return _helper_action("hostname", "set-hostname", request.form.get("hostname", "").strip().lower(), back=url_for("ui.system"))
|
||
|
||
|
||
@bp.post("/system/reboot")
|
||
def system_reboot():
|
||
return _helper_action("reboot", "reboot", back=url_for("ui.dashboard"))
|
||
|
||
|
||
@bp.post("/system/poweroff")
|
||
def system_poweroff():
|
||
return _helper_action("poweroff", "poweroff", back=url_for("ui.dashboard"))
|
||
|
||
|
||
@bp.post("/system/admin")
|
||
def system_admin():
|
||
updates = {
|
||
"admin_port": request.form.get("admin_port", "8080"),
|
||
"admin_https": _form_bool("admin_https"),
|
||
"admin_allow_public": _form_bool("admin_allow_public"),
|
||
}
|
||
old = cfg()
|
||
new = _save(updates)
|
||
if new is None:
|
||
return redirect(url_for("ui.system"))
|
||
audit("admin_settings", **updates)
|
||
if any(old[k] != new[k] for k in ("admin_port", "admin_https")):
|
||
scheme = "https" if new["admin_https"] else "http"
|
||
target = f"{scheme}://{request.host.rsplit(':', 1)[0]}:{new['admin_port']}/"
|
||
try:
|
||
svc().helper.call("restart-manager")
|
||
except HelperError as exc:
|
||
flash(f"Opgeslagen, maar herstarten van de beheerinterface mislukte: {exc}", "error")
|
||
return redirect(url_for("ui.system"))
|
||
return render_template("moved.html", target=target)
|
||
flash("Instellingen opgeslagen.", "success")
|
||
return redirect(url_for("ui.system"))
|
||
|
||
|
||
@bp.post("/system/updates")
|
||
def system_updates():
|
||
auto = _form_bool("auto_update")
|
||
if _save({"auto_update": auto}) is not None:
|
||
audit("update_settings", auto_update=auto)
|
||
flash("Automatisch bijwerken staat " + ("aan." if auto else "uit."), "success")
|
||
return redirect(url_for("ui.system") + "#updates")
|
||
|
||
|
||
@bp.post("/system/update/check")
|
||
def system_update_check():
|
||
back = url_for("ui.system") + "#updates"
|
||
try:
|
||
state = svc().helper.call("update-check", timeout=60).get("update", {})
|
||
except HelperError as exc:
|
||
return _action_response(False, str(exc), back)
|
||
if state.get("check_error"):
|
||
return _action_response(False, f"Controleren mislukt: {state['check_error']}", back)
|
||
if state.get("available"):
|
||
return _action_response(True, f"Versie {state.get('latest')} is beschikbaar.", back)
|
||
return _action_response(True, f"Je hebt de nieuwste versie ({state.get('current')}).", back)
|
||
|
||
|
||
@bp.post("/system/update/install")
|
||
def system_update_install():
|
||
return _helper_action("update_install", "update-install", back=url_for("ui.system") + "#updates")
|
||
|
||
|
||
# -- logs ------------------------------------------------------------------------
|
||
|
||
@bp.get("/logs")
|
||
def logs():
|
||
source = request.args.get("source", "kiosk")
|
||
if source not in services.LOG_SOURCES:
|
||
source = "kiosk"
|
||
try:
|
||
lines = int(request.args.get("lines", 100))
|
||
except ValueError:
|
||
lines = 100
|
||
priority = request.args.get("priority") or None
|
||
text = services.journal(source, lines, priority)
|
||
return render_template("logs.html", active="logs", text=text, source=source, lines=lines, priority=priority,
|
||
sources=services.LOG_SOURCES, line_options=services.LOG_LINES)
|
||
|
||
|
||
# -- network -----------------------------------------------------------------------
|
||
|
||
def _pending() -> dict[str, Any] | None:
|
||
pending = netinfo.pending_change(svc().paths.network_pending_file)
|
||
if pending:
|
||
pending = {k: pending.get(k) for k in ("interface", "method", "address", "created", "deadline")}
|
||
pending["remaining"] = max(0, int(pending.get("deadline") or 0) - int(time.time()))
|
||
return pending
|
||
|
||
|
||
@bp.get("/network")
|
||
def network():
|
||
return render_template("network.html", active="network", net=netinfo.overview(), pending=_pending(),
|
||
rollback_seconds=netconfig.ROLLBACK_SECONDS)
|
||
|
||
|
||
@bp.get("/api/network/pending")
|
||
def api_network_pending():
|
||
return jsonify(pending=_pending())
|
||
|
||
|
||
@bp.post("/network/apply")
|
||
def network_apply():
|
||
interface = request.form.get("interface", "")
|
||
payload: dict[str, Any] = {
|
||
"interface": interface,
|
||
"method": request.form.get("method", "dhcp"),
|
||
"address": request.form.get("address", ""),
|
||
"gateway": request.form.get("gateway", ""),
|
||
"dns": request.form.get("dns", ""),
|
||
}
|
||
if interface == "wlan0":
|
||
payload["wifi"] = {
|
||
"ssid": request.form.get("ssid", ""),
|
||
"psk": request.form.get("psk", ""),
|
||
"hidden": _form_bool("hidden"),
|
||
"open": _form_bool("open"),
|
||
}
|
||
try:
|
||
netconfig.validate_request(payload) # early feedback; the helper validates again
|
||
svc().helper.call("network-apply", payload=payload, timeout=60)
|
||
except (ValidationError, HelperError) as exc:
|
||
flash(str(exc), "error")
|
||
return redirect(url_for("ui.network"))
|
||
audit("network_apply", interface=interface, method=payload["method"], address=payload["address"])
|
||
flash(f"Netwerkwijziging toegepast. Bevestig binnen {netconfig.ROLLBACK_SECONDS // 60} minuten, anders worden de "
|
||
"vorige instellingen automatisch teruggezet.", "warning")
|
||
return redirect(url_for("ui.network"))
|
||
|
||
|
||
@bp.post("/network/confirm")
|
||
def network_confirm():
|
||
return _helper_action("network_confirm", "network-confirm", back=url_for("ui.network"))
|
||
|
||
|
||
@bp.post("/network/rollback")
|
||
def network_rollback():
|
||
return _helper_action("network_rollback", "network-rollback", back=url_for("ui.network"))
|
||
|
||
|
||
@bp.post("/network/country")
|
||
def network_country():
|
||
return _helper_action("wifi_country", "set-wifi-country", request.form.get("country", "").strip().upper(), back=url_for("ui.network"))
|
||
|
||
|
||
@bp.post("/api/network/scan")
|
||
def network_scan():
|
||
try:
|
||
result = svc().helper.call("wifi-scan", timeout=60)
|
||
except HelperError as exc:
|
||
return jsonify(error=str(exc)), 500
|
||
return jsonify(networks=result.get("networks", []))
|