daynivo-kiosk-os/src/voxip_kiosk/agent.py

"""voxip-kiosk-agent — keeps the kiosk page where it should be.

Runs as user ``kiosk`` and steers Chromium through the DevTools Protocol on
127.0.0.1:

* shows the setup page (with setup code) until an admin password exists;
* leaves the local start page / Chromium error page for the kiosk URL as
  soon as that URL is reachable, and shows the offline page meanwhile;
* follows URL changes from the web interface without restarting Chromium;
* reloads periodically when ``refresh_interval`` is set — but never while
  the URL is unreachable, so a page that works offline keeps working;
* reloads a hung renderer, and closes the browser (systemd restarts it) if
  that does not help.
"""

from __future__ import annotations

import json
import logging
import signal
import socket
import ssl
import time
import urllib.error
import urllib.request
from pathlib import Path
from typing import Any, Callable
from urllib.parse import urlencode, urlsplit

from . import logutil, sysinfo
from .cdp import CDP, CDPError
from .config import admin_url, load_config
from .paths import Paths, load_paths
from .validation import mask_url

log = logging.getLogger("voxip-kiosk-agent")

TICK_SECONDS = 5
PROBE_MAX_AGE_LOADING = 10
PROBE_MAX_AGE_ONLINE = 60
HUNG_RELOAD_AFTER = 3
HUNG_RESTART_AFTER = 6
NO_NETWORK = "Geen netwerkverbinding (netwerkkabel los?)"
BLANK_URLS = {"", "about:blank", "chrome://newtab/", "chrome://new-tab-page/"}


def probe(url: str, timeout: float = 8.0) -> tuple[bool, str]:
    """Is ``url`` reachable over HTTPS with a valid certificate? → (ok, reason)."""
    req = urllib.request.Request(url, method="HEAD", headers={"User-Agent": "Daynivo-Kiosk-OS/agent"})
    try:
        with urllib.request.urlopen(req, timeout=timeout, context=ssl.create_default_context()):
            return True, ""
    except urllib.error.HTTPError as exc:
        if exc.code >= 500:
            return False, f"Serverfout {exc.code}"
        return True, ""
    except urllib.error.URLError as exc:
        return False, _reason(exc.reason)
    except (OSError, ValueError) as exc:
        return False, _reason(exc)


def _reason(exc: Any) -> str:
    if isinstance(exc, ssl.SSLCertVerificationError) or "CERTIFICATE_VERIFY_FAILED" in str(exc):
        return "Certificaat ongeldig"
    if isinstance(exc, ssl.SSLError):
        return "TLS-fout"
    if isinstance(exc, socket.gaierror):
        return "Naam niet gevonden (DNS)"
    if isinstance(exc, (socket.timeout, TimeoutError)):
        return "Time-out"
    if isinstance(exc, ConnectionRefusedError):
        return "Verbinding geweigerd"
    if isinstance(exc, OSError) and exc.errno in (101, 113):
        return "Geen netwerkverbinding"
    return "Niet bereikbaar"


class Agent:
    def __init__(self, paths: Paths, cdp: Any = None, probe_fn: Callable[[str], tuple[bool, str]] = probe,
                 clock: Callable[[], float] = time.monotonic, wall: Callable[[], float] = time.time,
                 hostname: Callable[[], str] = lambda: socket.gethostname().split(".")[0], addresses: Callable[[], dict] = sysinfo.ip_addresses):
        self.paths = paths
        self.cdp = cdp or CDP()
        self.probe_fn = probe_fn
        self.clock = clock
        self.wall = wall
        self.hostname = hostname
        self.addresses = addresses
        self.applied_url: str | None = None
        self.last_load: float | None = None
        self.hung = 0
        self.offline_since: float | None = None
        self.refresh_skipped_logged = False
        self._probe: tuple[str, float, bool, str] | None = None
        self.state: dict[str, Any] = {"last_error": None}
        self._written = ""
        self._setup_shown: str | None = None

    # -- helpers ------------------------------------------------------------

    def local_url(self, page: str, params: dict[str, Any]) -> str:
        return f"file://{self.paths.www_dir}/{page}.html?" + urlencode(params)

    def classify(self, url: str) -> str:
        prefix = f"file://{self.paths.www_dir}/"
        if url.startswith(prefix):
            return url[len(prefix):].split(".html", 1)[0]  # start / offline / setup
        if url.startswith("chrome-error://"):
            return "error"
        if url in BLANK_URLS:
            return "blank"
        return "remote"

    def reachable(self, url: str, max_age: float) -> tuple[bool, str]:
        now = self.clock()
        if self._probe and self._probe[0] == url and now - self._probe[1] < max_age:
            return self._probe[2], self._probe[3]
        ok, reason = self.probe_fn(url)
        self._probe = (url, now, ok, reason)
        self.state.update(target_reachable=ok, reason=reason, checked_at=int(self.wall()))
        return ok, reason

    def error(self, message: str) -> None:
        log.warning(message)
        self.state["last_error"] = {"at": int(self.wall()), "message": message}

    def has_network(self) -> bool:
        """False when no interface has an address, e.g. the cable is out and there is no wifi."""
        return bool(self.addresses())

    def show_offline(self, config: dict[str, Any], reason: str) -> None:
        if self.offline_since is None:
            self.offline_since = self.wall()
            self.state["offline_since"] = int(self.offline_since)
        self.error(f"Kiosk-URL niet bereikbaar ({reason}); offlinepagina getoond")
        self.navigate(self.offline_url(config, reason))

    def setup_required(self) -> bool:
        return self.paths.setup_code_file.exists() and not self.paths.admin_file.exists()

    def setup_url(self, config: dict[str, Any]) -> str:
        try:
            code = self.paths.setup_code_file.read_text(encoding="utf-8").strip()
        except OSError:
            code = ""
        ips = sorted({a.split("/")[0] for addrs in self.addresses().values() for a in addrs if ":" not in a})
        scheme = "https" if config["admin_https"] else "http"
        return self.local_url("setup", {"code": code, "host": self.hostname(), "ips": ",".join(ips),
                                        "port": config["admin_port"], "scheme": scheme, "name": config["device_name"]})

    def offline_url(self, config: dict[str, Any], reason: str) -> str:
        return self.local_url("offline", {"name": config["device_name"], "host": urlsplit(config["url"]).hostname or "",
                                          "url": mask_url(config["url"]), "admin": admin_url(config, self.hostname()),
                                          "reason": reason, "since": int(self.offline_since or self.wall())})

    def navigate(self, url: str) -> None:
        self.cdp.navigate(url)

    # -- main loop ----------------------------------------------------------

    def tick(self) -> None:
        config = load_config(self.paths.config_file)
        target = config["url"]
        self.state["updated"] = int(self.wall())
        try:
            page = self.cdp.main_page()
            if page is None:
                self.cdp.new_page(self.local_url("start", {"name": config["device_name"], "next": target}))
                return
            self._tick_page(config, target, page)
            self.state["browser"] = "ok"
        except CDPError as exc:
            self.state.update(browser="unreachable", page=None)
            self.hung = 0
            log.debug("DevTools niet bereikbaar: %s", exc)
        finally:
            self.write_state()

    def _tick_page(self, config: dict[str, Any], target: str, page: dict[str, Any]) -> None:
        url = page.get("url", "")
        kind = self.classify(url)
        self.state.update(page=kind, page_url=mask_url(url) if kind == "remote" else kind)

        if self.setup_required():
            self.state["setup_required"] = True
            want = self.setup_url(config)
            if kind != "setup" or want != self._setup_shown:
                self.navigate(want)
                self._setup_shown = want
            return
        self.state["setup_required"] = False
        self._setup_shown = None

        url_changed = self.applied_url is not None and target != self.applied_url
        if url_changed:
            log.info("Nieuwe kiosk-URL: %s", mask_url(target))
        self.applied_url = target

        if kind != "remote" or url_changed:
            ok, reason = self.reachable(target, 0 if url_changed else PROBE_MAX_AGE_LOADING)
            if not ok and not self.has_network():
                reason = NO_NETWORK
            if ok:
                if self.offline_since is not None:
                    log.info("Kiosk-URL weer bereikbaar na %.0f s", self.wall() - self.offline_since)
                self.navigate(target)
                self.last_load = self.clock()
                self.offline_since = None
                self.state["offline_since"] = None
            elif kind != "offline" or url_changed:
                self.show_offline(config, reason)
            self.hung = 0
            return

        # Without any network the page cannot recover by itself: say so on screen. When only
        # the server is unreachable the page stays, so Daynivo can show its offline cache.
        if not self.has_network():
            self.state.update(target_reachable=False, reason=NO_NETWORK, checked_at=int(self.wall()))
            self._probe = None
            self.hung = 0
            self.show_offline(config, NO_NETWORK)
            return

        if self.last_load is None:
            self.last_load = self.clock()

        if not self.cdp.responsive(page.get("webSocketDebuggerUrl")):
            self.hung += 1
            if self.hung == HUNG_RELOAD_AFTER:
                self.error("Pagina reageert niet; herladen")
                self.cdp.reload()
                self.last_load = self.clock()
            elif self.hung >= HUNG_RESTART_AFTER and config["browser_restart"]:
                self.error("Pagina reageert nog steeds niet; Chromium wordt herstart")
                self.hung = 0
                self.cdp.close_browser()
            return
        self.hung = 0

        ok, reason = self.reachable(target, PROBE_MAX_AGE_ONLINE)
        interval = config["refresh_interval"]
        if interval and self.clock() - self.last_load >= interval:
            if ok:
                log.info("Automatisch verversen (interval %s s)", interval)
                self.cdp.reload()
                self.last_load = self.clock()
                self.refresh_skipped_logged = False
            elif not self.refresh_skipped_logged:
                self.error(f"Verversen overgeslagen: kiosk-URL niet bereikbaar ({reason})")
                self.refresh_skipped_logged = True

    def write_state(self) -> None:
        state = dict(self.state, last_load_wall=int(self.wall() - (self.clock() - self.last_load)) if self.last_load else None)
        payload = json.dumps(state, sort_keys=True)
        if payload == self._written:
            return
        try:
            tmp = self.paths.agent_state_file.with_suffix(".tmp")
            tmp.write_text(payload, encoding="utf-8")
            tmp.replace(self.paths.agent_state_file)
            self._written = payload
        except OSError as exc:
            log.debug("Kon agentstatus niet schrijven: %s", exc)


def read_state(path: Path) -> dict[str, Any]:
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return {}
    return data if isinstance(data, dict) else {}


def main() -> int:
    logutil.setup()
    agent = Agent(load_paths())
    running = True

    def stop(*_: Any) -> None:
        nonlocal running
        running = False

    signal.signal(signal.SIGTERM, stop)
    log.info("Kiosk-agent gestart")
    while running:
        try:
            agent.tick()
        except Exception:  # never die on an unexpected page state; systemd would only restart us
            log.exception("Onverwachte fout in de kiosk-agent")
        for _ in range(TICK_SECONDS * 2):
            if not running:
                break
            time.sleep(0.5)
    return 0