#!/usr/bin/env python3
"""Lavtomate desktop agent (#55).

Runs on the user's machine, connects to the Lavtomate server over an
authenticated WebSocket, and launches whitelisted (graphical) applications on
its behalf. Every launch is confirmed server-side; this agent is the final
defense-in-depth guard — it re-checks its OWN whitelist before executing
anything and never invokes a shell.

Config (environment, lowest effort first):
  LAVTOMATE_DESKTOP_SERVER   Lavtomate base URL (default http://localhost:8502)
  LAVTOMATE_DESKTOP_MACHINE  Machine id reported to the server (default: hostname)
  LAVTOMATE_API_KEY          If set, authenticate as a Bearer token (prod via
                            proxy-manager). Otherwise SPNEGO/Kerberos is used.
  LAVTOMATE_DESKTOP_WHITELIST
                            Path to a JSON whitelist file (see README). If
                            unset, ~/.config/lavtomate/desktop-whitelist.json
                            is used. Missing file → empty whitelist (launches
                            refused until you add entries).

Usage:
  python3 lavtomate_desktop_agent.py
  # or as a systemd user service — see lavtomate-desktop-agent.service
"""

from __future__ import annotations

import base64
import collections
import fnmatch
import json
import os
import re
import shlex
import shutil
import socket
import subprocess
import sys
import time
from pathlib import Path

import structlog

logger = structlog.get_logger()

# Re-read the whitelist at most this often, so edits apply without a restart.
WHITELIST_RELOAD_SECS = 10
PING_SECS = 20
# How often to report graphical-session liveness (active/idle) so the server can
# target the machine the user is actually sitting at. See compute_liveness().
LIVENESS_SECS = 30
RECONNECT_BACKOFF = [2, 5, 10, 20, 30]
# Session cookie set by lavtomate/proxy-manager on login — the web UI auth path.
SESSION_COOKIE_NAME = "lavtomate_session"

# Cap on a single delivered file (a .vv is ~3 KB; this is a safety bound).
MAX_FILE_BYTES = 256 * 1024
# Where files delivered with a `launch` (e.g. a SPICE .vv) are written. The
# server does NOT know this path (it depends on the agent user's runtime dir),
# so it sends `args` with the literal token `@FILE@` (or `@file:<name>@`) which
# the agent replaces with the real path after writing — no shell is ever used.
def file_dir() -> Path:
    override = os.environ.get("LAVTOMATE_DESKTOP_FILE_DIR")
    if override:
        return Path(override)
    base = os.environ.get("XDG_RUNTIME_DIR") or f"/tmp/lavtomate-{os.getuid()}"
    return Path(base) / "lavtomate-files"


# --------------------------------------------------------------------------- #
# Hardening (PATH-resolve, dangerous-binary denylist, rate-limit)
# --------------------------------------------------------------------------- #
# Hard denylist of binaries that are unsafe to launch unattended: any of them
# with an exec flag (terminal `-e`, shell/interpreter `-c`, ...) is full RCE as
# the user. The agent refuses them EVEN IF a (mis)configured whitelist lists them
# — the local whitelist must be "safe to run unattended". Bins are matched by the
# basename of the resolved path (so e.g. /usr/bin/xterm is caught too).
DANGEROUS_APPS = frozenset({
    # terminals — `-e`/`--execute` run an arbitrary command
    "xterm", "uxterm", "gnome-terminal", "konsole", "urxvt", "rxvt",
    "lxterminal", "mate-terminal", "xfce4-terminal", "terminator",
    "tilix", "guake", "alacritty", "kitty", "wezterm", "foot", "st",
    # shells / interpreters — `-c`/`-e` run arbitrary code
    "sh", "bash", "dash", "zsh", "fish", "csh", "tcsh",
    "python", "python2", "python3", "perl", "ruby", "node", "lua",
    "php", "tclsh", "wish",
})

# At most this many launches per sliding window (per agent process).
RATE_LIMIT_MAX = 10
RATE_LIMIT_WINDOW_SECS = 60.0


def _resolve_executable(app: str) -> str | None:
    """Resolve ``app`` to an absolute path via ``$PATH`` (``shutil.which``).

    Resolving here (instead of letting ``Popen`` do it at exec) lets us
    (a) check the denylist against the real binary, (b) enforce the resolved
    path when a whitelist entry pins an absolute path, and (c) hand ``Popen`` a
    fixed argv[0] so a PATH mutation between check and exec can't swap it.

    NOTE: for a whitelisted NAME (``firefox``) this is best-effort — a hijacked
    earlier PATH dir can still make ``which`` return a malicious binary. For full
    protection, pin absolute paths in the whitelist (they are then enforced).
    """
    return shutil.which(app)


def _is_dangerous(app: str) -> bool:
    """True if the resolved binary's basename is on the hard denylist."""
    resolved = _resolve_executable(app)
    name = os.path.basename(resolved) if resolved else app
    return name in DANGEROUS_APPS


# Sliding-window launch counters (one deque per agent process).
_launch_times: "collections.deque[float]" = collections.deque()


def _rate_limited() -> bool:
    """Sliding-window rate limit. Records the attempt only when not refused."""
    now = time.monotonic()
    cutoff = now - RATE_LIMIT_WINDOW_SECS
    while _launch_times and _launch_times[0] < cutoff:
        _launch_times.popleft()
    if len(_launch_times) >= RATE_LIMIT_MAX:
        return True
    _launch_times.append(now)
    return False


# --------------------------------------------------------------------------- #
# Auth
# --------------------------------------------------------------------------- #
def _spnego_header(hostname: str) -> str | None:
    """Build an `Authorization: Negotiate <token>` value for the HTTP service."""
    try:
        import gssapi
    except ImportError:
        logger.warning("gssapi not available — cannot do SPNEGO auth")
        return None
    try:
        name = gssapi.Name(f"HTTP@{hostname}", gssapi.NameType.hostbased_service)
        ctx = gssapi.SecurityContext(name=name, usage="initiate")
        token = ctx.step()
        if token:
            return "Negotiate " + base64.b64encode(token).decode("ascii")
    except Exception as e:
        logger.warning("SPNEGO token failed", hostname=hostname, error=str(e))
    return None


def have_kerberos_ticket() -> bool:
    """True if the default ccache holds a usable (unexpired) TGT.

    Gates the connection so the agent doesn't spam anonymous WS handshakes while
    there is no ticket yet (e.g. before the user runs ``kinit``) — the server
    would just reject them (HTTP 302). ``klist -s`` exits 0 iff the ccache has
    valid credentials; re-checked each cycle so a later ``kinit`` is picked up
    without a restart. If ``klist`` is absent we can't verify, so assume no
    ticket (wait rather than connect blindly).
    """
    try:
        r = subprocess.run(["klist", "-s"], capture_output=True, timeout=5)
        return r.returncode == 0
    except (FileNotFoundError, subprocess.TimeoutExpired):
        return False


def _parse_kv(text: str) -> dict:
    """Parse ``loginctl show-session`` KEY=VALUE output into a dict."""
    out: dict[str, str] = {}
    for line in text.splitlines():
        k, sep, v = line.partition("=")
        if sep:
            out[k.strip()] = v.strip()
    return out


def compute_liveness() -> dict:
    """Graphical-session liveness for THIS machine, for the agent's user.

    Returns ``{"active": bool | None, "idle_secs": float | None}``:

    * ``active`` True  — the user has an *attended* graphical session right now:
      a session with ``Class=user``, ``Type`` wayland/x11, ``Active=yes`` and
      ``State=active`` (i.e. they're at the keyboard, not just logged in).
      False — a graphical session exists but isn't active (locked / VT switched
      away), or there is none at all (headless box). None — couldn't tell
      (``loginctl`` absent/errored).
    * ``idle_secs`` — seconds idle on that session (0 when not idle), or None.

    Uses only ``loginctl`` (read via ``subprocess``) — no display bus / D-Bus /
    extra Python deps — so it works from a ``systemd --user`` unit that does not
    inherit the graphical session environment. The graphical ``Type`` filter is
    what keeps a headless box with an active SSH-tty from reporting active=True.
    """
    try:
        listing = subprocess.run(
            ["loginctl", "list-sessions", "--no-legend"],
            capture_output=True, text=True, timeout=5,
        )
        if listing.returncode != 0:
            return {"active": None, "idle_secs": None}
    except (FileNotFoundError, subprocess.TimeoutExpired):
        return {"active": None, "idle_secs": None}

    uid = str(os.getuid())
    now = time.time()
    for line in listing.stdout.splitlines():
        parts = line.split()
        # Columns: SESSION UID USER SEAT TTY ... — filter by our own UID.
        if len(parts) < 2 or parts[1] != uid:
            continue
        try:
            show = subprocess.run(
                ["loginctl", "show-session", parts[0]],
                capture_output=True, text=True, timeout=5,
            )
        except (FileNotFoundError, subprocess.TimeoutExpired):
            continue
        if show.returncode != 0:
            continue
        s = _parse_kv(show.stdout)
        if s.get("Class") != "user" or s.get("Type") not in ("wayland", "x11"):
            continue
        if s.get("Active") == "yes" and s.get("State") == "active":
            idle = 0.0
            if s.get("IdleHint") == "yes":
                try:
                    # IdleSinceHint is wall-clock microseconds since the epoch.
                    since = float(s.get("IdleSinceHint") or 0) / 1e6
                    if since > 0:
                        idle = max(0.0, now - since)
                except (TypeError, ValueError):
                    idle = 0.0
            return {"active": True, "idle_secs": idle}
    return {"active": False, "idle_secs": None}


def send_liveness(ws) -> None:
    """Compute liveness and send a ``liveness`` frame; never raises."""
    try:
        liv = compute_liveness()
        ws.send(json.dumps({"type": "liveness", **liv}))
        logger.info("liveness", active=liv["active"], idle_secs=liv["idle_secs"])
    except Exception as e:
        logger.warning("liveness report failed", error=str(e))


def auth_headers(server_url: str, api_key: str | None, machine_id: str) -> dict[str, str]:
    """Resolve the auth header(s) for the WS handshake.

    Preference order:
      1. an explicit API key (LAVTOMATE_API_KEY) → Bearer (for deployments that
         trust proxy headers / accept API keys on the WS);
      2. a session cookie self-provisioned from the Kerberos ticket — the same
         auth the web UI uses (login through /auth/kerberos), so it is accepted
         on the WS regardless of TRUST_PROXY;
      3. raw SPNEGO Negotiate — only where nginx ``auth_gss`` is applied directly
         (standalone deployments).
    """
    if api_key:
        return {"Authorization": f"Bearer {api_key}"}
    cookie = _kerberos_session_cookie(server_url)
    if cookie:
        return {"Cookie": f"{SESSION_COOKIE_NAME}={cookie}"}
    from urllib.parse import urlparse
    host = urlparse(server_url).hostname or "localhost"
    nego = _spnego_header(host)
    if nego:
        return {"Authorization": nego}
    logger.warning("No auth available — connecting anonymously (will likely be rejected)")
    return {}


def _kerberos_session_cookie(server_url: str) -> str | None:
    """Log in with the Kerberos ticket and return the session cookie value.

    Hits ``/auth/kerberos`` (nginx ``auth_gss`` SPNEGO) with the ticket via
    ``curl --negotiate``; proxy-manager validates it and sets the
    ``lavtomate_session`` cookie. This is exactly how the web UI authenticates,
    so lavtomate's WS handler accepts the cookie on the upgrade (no TRUST_PROXY
    required). Returns ``None`` if no ticket / curl missing / login failed.
    """
    import tempfile
    jar_fd, jar = tempfile.mkstemp(prefix="lav-agent-", suffix=".cookie")
    os.close(jar_fd)
    try:
        subprocess.run(
            ["curl", "-s", "--negotiate", "-u", ":",
             "-c", jar, "-o", "/dev/null", f"{server_url.rstrip('/')}/auth/kerberos"],
            capture_output=True, timeout=20,
        )
        try:
            lines = Path(jar).read_text(errors="replace").splitlines()
        except OSError:
            return None
        for line in lines:
            # Netscape jar: HttpOnly cookie lines start with "#HttpOnly_" —
            # don't confuse them with comment lines.
            if not line.strip() or (line.startswith("#") and not line.startswith("#HttpOnly_")):
                continue
            fields = line.split("\t")
            if len(fields) >= 7 and fields[5] == SESSION_COOKIE_NAME:
                logger.info("obtained session cookie via Kerberos ticket")
                return fields[6].strip('"')
    except (FileNotFoundError, subprocess.TimeoutExpired):
        return None
    finally:
        try:
            os.unlink(jar)
        except OSError:
            pass
    return None


# --------------------------------------------------------------------------- #
# Whitelist (defense in depth)
# --------------------------------------------------------------------------- #
def whitelist_path() -> Path:
    p = os.environ.get("LAVTOMATE_DESKTOP_WHITELIST")
    if p:
        return Path(p)
    return Path.home() / ".config" / "lavtomate" / "desktop-whitelist.json"


class Whitelist:
    """Local allowlist, reloaded from disk periodically."""

    def __init__(self) -> None:
        self._path = whitelist_path()
        self._mtime = 0.0
        self._entries: list[dict] = []
        self._reload(force=True)

    def _reload(self, force: bool = False) -> None:
        try:
            mtime = self._path.stat().st_mtime
        except FileNotFoundError:
            if force or self._entries:
                logger.info("whitelist file not found — empty whitelist", path=str(self._path))
                self._entries = []
                self._mtime = 0.0
            return
        if not force and mtime == self._mtime:
            return
        try:
            data = json.loads(self._path.read_text(encoding="utf-8"))
            entries = data if isinstance(data, list) else data.get("apps", [])
            assert isinstance(entries, list)
            self._entries = entries
            self._mtime = mtime
            logger.info("whitelist loaded", path=str(self._path), count=len(entries))
        except Exception as e:
            logger.warning("whitelist parse failed", error=str(e))

    def maybe_reload(self) -> None:
        if time.time() - getattr(self, "_last_check", 0) < WHITELIST_RELOAD_SECS:
            return
        self._last_check = time.time()
        self._reload()

    def check(self, app: str, args: str) -> tuple[bool, str]:
        """Return (allowed, reason)."""
        # #2: hard denylist — refuse terminals/shells/interpreters even when a
        # whitelist entry names them (any of them + an exec flag = full RCE).
        if _is_dangerous(app):
            return False, "refused: app is on the dangerous-binary denylist"
        self.maybe_reload()
        for e in self._entries:
            if e.get("app") != app:
                continue
            if e.get("enabled", True) is False:
                continue
            # #1: if the entry pins an absolute path, the resolved binary must
            # match it exactly — no $PATH hijack substitution at exec time.
            if os.path.isabs(app):
                resolved = _resolve_executable(app)
                if resolved is None or os.path.realpath(resolved) != os.path.realpath(app):
                    return False, "resolved path does not match the whitelisted absolute path"
            template = (e.get("arg_template") or "").strip()
            if not args:
                return True, "ok"
            if not template:
                return False, "arguments not allowed for this app"
            if e.get("use_regex"):
                try:
                    # #3: fullmatch (auto-anchored) — re.search would let a
                    # template like '^https://' accept 'https://evil foo'.
                    if re.fullmatch(template, args):
                        return True, "ok"
                except re.error:
                    pass
                return False, "arguments did not match regex template"
            if fnmatch.fnmatch(args, template):
                return True, "ok"
            return False, "arguments did not match glob template"
        return False, "app not in local whitelist"


# --------------------------------------------------------------------------- #
# Launch
# --------------------------------------------------------------------------- #
# A GUI app that exits within this window almost certainly failed to open
# (e.g. remote-viewer couldn't connect) — capture its exit code + stderr so the
# caller sees why instead of a silent "launched". Surviving past it = success.
LAUNCH_GRACE_SECS = 4.0


def _tail(path: str, lines: int = 12) -> str:
    try:
        data = Path(path).read_text(errors="replace").splitlines()
    except OSError:
        return ""
    return "\n".join(data[-lines:]).strip()


def launch(app: str, args: str) -> dict:
    """Launch the app detached. Returns the agent result envelope.

    No shell is ever involved — ``Popen([resolved, *args], start_new_session=True)``.
    ``app`` is resolved to an absolute path first (#1) so a PATH mutation can't
    swap the binary between the whitelist check and exec; the resolved path is
    logged. stderr/stdout go to a temp log; if the process exits within
    :data:`LAUNCH_GRACE_SECS`, we report the exit code and a log tail (startup
    failure, e.g. a SPICE connection error); if it survives, we treat it as a
    successful launch.
    """
    import tempfile
    # #6: per-agent sliding-window rate limit.
    if _rate_limited():
        logger.warning("launch refused: rate limit", app=app)
        return {"ok": False, "error": "rate limit: too many launches, retry shortly"}
    # #1: resolve to an absolute path and exec that (not a re-lookup via $PATH).
    resolved = _resolve_executable(app)
    if resolved is None:
        return {"ok": False, "error": f"executable not found in PATH: {app}"}
    argv = [resolved] + shlex.split(args)
    logger.info("launch resolved", app=app, resolved=resolved)
    log_fd, log_path = tempfile.mkstemp(prefix="lav-agent-", suffix=".log")
    os.close(log_fd)
    try:
        proc = subprocess.Popen(
            argv, start_new_session=True,
            stdout=open(log_path, "ab"),
            stderr=subprocess.STDOUT,
        )
    except FileNotFoundError:
        _rm(log_path)
        return {"ok": False, "error": f"executable not found: {app}"}
    except Exception as e:
        _rm(log_path)
        return {"ok": False, "error": f"launch failed: {e}"}

    try:
        proc.wait(timeout=LAUNCH_GRACE_SECS)
    except subprocess.TimeoutExpired:
        # Still running after the grace window → launched OK. Release the log
        # path (the app keeps writing to the unlinked inode via its inherited fd).
        _rm(log_path)
        return {"ok": True, "data": {"pid": proc.pid, "argv": argv}}

    # Exited within the window — capture why.
    rc = proc.returncode
    tail = _tail(log_path)
    _rm(log_path)
    logger.warning("launched app exited quickly", app=app, rc=rc)
    return {
        "ok": False,
        "data": {"pid": proc.pid, "argv": argv},
        "exit_code": rc,
        "log": tail,
        "error": f"{app} exited with code {rc}" + (f":\n{tail}" if tail else ""),
    }


def _rm(path: str) -> None:
    try:
        os.unlink(path)
    except OSError:
        pass


# --------------------------------------------------------------------------- #
# File delivery (e.g. a SPICE .vv) — no shell, sandboxed writes
# --------------------------------------------------------------------------- #
_FILE_NAME_RE = re.compile(r"[^A-Za-z0-9._-]+")


def _sanitize_file_name(name: str) -> str:
    """Reduce a server-supplied name to a safe single path component.

    Anything that is not [A-Za-z0-9._-] becomes '_'; leading dots are stripped
    so we never produce a dotfile or '..'. The result is a basename only — the
    sandbox directory is always prepended by the agent, never by the server.
    """
    name = _FILE_NAME_RE.sub("_", str(name)).strip("._")
    return name or "file"


def deliver_files(files: list[dict]) -> tuple[dict, str | None]:
    """Write delivered files into the sandbox.

    Each entry is ``{"name": <basename>, "content_b64": <base64 bytes>}``.
    Returns ``(name -> real_path, error_or_None)``. Writes are atomic and size-
    capped; the sandbox is created 0700.
    """
    if not files:
        return {}, None
    out: dict[str, str] = {}
    try:
        d = file_dir()
        d.mkdir(parents=True, exist_ok=True)
        os.chmod(d, 0o700)
    except OSError as e:
        return {}, f"cannot create file dir: {e}"
    for entry in files:
        if not isinstance(entry, dict):
            return out, "malformed file entry"
        name = _sanitize_file_name(entry.get("name", ""))
        raw = entry.get("content_b64") or ""
        try:
            data = base64.b64decode(raw, validate=False)
        except Exception as e:
            return out, f"bad base64 for {name}: {e}"
        if len(data) > MAX_FILE_BYTES:
            return out, f"file {name} exceeds {MAX_FILE_BYTES} bytes"
        target = d / name
        try:
            tmp = d / (".#" + name)
            tmp.write_bytes(data)
            os.replace(tmp, target)
        except OSError as e:
            return out, f"write {name} failed: {e}"
        out[name] = str(target)
    return out, None


def substitute_file_args(args: str, name_to_path: dict[str, str]) -> str:
    """Replace ``@file:<name>@`` and ``@FILE@`` tokens with real sandbox paths.

    If ``args`` is empty but files were delivered, default to the first file's
    path (the common single-file case, e.g. ``remote-viewer <vv>``).
    """
    if not name_to_path:
        return args
    if not args:
        return next(iter(name_to_path.values()))
    out = args
    for name, path in name_to_path.items():
        out = out.replace(f"@file:{name}@", path)
    if "@FILE@" in out and name_to_path:
        out = out.replace("@FILE@", next(iter(name_to_path.values())))
    return out


# --------------------------------------------------------------------------- #
# WS URL helpers
# --------------------------------------------------------------------------- #
def ws_url(server_url: str, machine_id: str) -> str:
    base = server_url.rstrip("/")
    if base.startswith("https://"):
        base = "wss://" + base[len("https://"):]
    elif base.startswith("http://"):
        base = "ws://" + base[len("http://"):]
    else:
        base = "ws://" + base
    from urllib.parse import quote
    return f"{base}/api/launch/ws?machine={quote(machine_id)}"


def _ssl_context():
    """SSL context for the WS connection.

    TLS verification is ON by default (the system trust store is used via
    ``ssl.create_default_context()`` — not certifi, so the Etersoft CA must be
    installed system-wide; the package Requires etersoft-ca-root >= 0.3). With
    that CA root the Basic Constraints are critical and OpenSSL 3 accepts the
    Lavtomate cert, so no override is needed. Set
    ``LAVTOMATE_DESKTOP_VERIFY_SSL=0`` only as a fallback on machines without
    the fixed CA root (the agent still authenticates at the app layer via
    Kerberos / API key).
    """
    import ssl
    ctx = ssl.create_default_context()
    if os.environ.get("LAVTOMATE_DESKTOP_VERIFY_SSL", "1") == "0":
        ctx.check_hostname = False
        ctx.verify_mode = ssl.CERT_NONE
    return ctx


# --------------------------------------------------------------------------- #
# Main loop
# --------------------------------------------------------------------------- #
def handle_message(payload: dict, wl: Whitelist) -> dict | None:
    """Process one server message; return a reply dict or None (no reply)."""
    mtype = payload.get("type")
    if mtype == "ping":
        return {"type": "pong"}
    if mtype == "launch":
        request_id = payload.get("request_id", "")
        app = (payload.get("app") or "").strip()
        args = (payload.get("args") or "").strip()
        if not app:
            return {"type": "result", "request_id": request_id, "ok": False,
                    "error": "no app specified"}
        # Delivered files (e.g. a SPICE .vv) are written to a sandbox FIRST,
        # then their real paths are substituted into args, so the whitelist
        # check below authorizes exactly what is executed. No shell is used.
        files = payload.get("files")
        name_to_path: dict[str, str] = {}
        if files:
            if not isinstance(files, list):
                return {"type": "result", "request_id": request_id, "ok": False,
                        "error": "files must be a list"}
            name_to_path, ferr = deliver_files(files)
            if ferr:
                logger.warning("file delivery failed", app=app, error=ferr)
                return {"type": "result", "request_id": request_id, "ok": False,
                        "error": f"file delivery failed: {ferr}"}
            args = substitute_file_args(args, name_to_path)
        allowed, reason = wl.check(app, args)
        if not allowed:
            logger.warning("launch refused", app=app, args=args, reason=reason)
            return {"type": "result", "request_id": request_id, "ok": False,
                    "error": f"refused locally: {reason}"}
        result = launch(app, args)
        logger.info("launch", app=app, args=args, ok=result.get("ok"),
                    pid=(result.get("data") or {}).get("pid"),
                    files=list(name_to_path) or None)
        return {"type": "result", "request_id": request_id, **result}
    return None


def run_once(server_url: str, machine_id: str, headers: dict, wl: Whitelist) -> None:
    """Connect, serve until the socket closes, then return."""
    from websockets.sync.client import connect
    uri = ws_url(server_url, machine_id)
    logger.info("connecting", uri=uri.replace(headers.get("Authorization", ""), "<auth>"))
    with connect(uri, additional_headers=headers, ping_interval=None,
                 open_timeout=15, close_timeout=5, ssl=_ssl_context()) as ws:
        logger.info("connected", machine=machine_id)
        last_ping = time.time()
        # Report liveness right away so the server knows our session state
        # without waiting for the first tick.
        send_liveness(ws)
        last_liveness = time.time()
        while True:
            try:
                raw = ws.recv(timeout=PING_SECS)
            except TimeoutError:
                raw = None
            now = time.time()
            # Tick liveness regardless of traffic (a steady stream of launches
            # would otherwise starve it — recv only times out when idle).
            if now - last_liveness >= LIVENESS_SECS:
                send_liveness(ws)
                last_liveness = now
            if raw is None:
                # idle — send an app-level ping to keep the server's liveness
                # probe happy and prove the connection is alive
                if now - last_ping >= PING_SECS:
                    ws.send(json.dumps({"type": "ping"}))
                    last_ping = now
                continue
            try:
                payload = json.loads(raw)
            except Exception:
                continue
            reply = handle_message(payload, wl)
            if reply is not None:
                ws.send(json.dumps(reply))


def main() -> int:
    structlog.configure(processors=[
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.dev.ConsoleRenderer(),
    ])
    server_url = os.environ.get("LAVTOMATE_DESKTOP_SERVER", "http://localhost:8502")
    machine_id = os.environ.get("LAVTOMATE_DESKTOP_MACHINE") or socket.gethostname()
    api_key = os.environ.get("LAVTOMATE_API_KEY") or None
    wl = Whitelist()

    attempt = 0
    while True:
        # Don't connect without credentials: with no API key and no Kerberos
        # ticket the server just rejects the WS (HTTP 302) — wait for a ticket
        # instead of hammering it. Re-checked each cycle, so a later `kinit` is
        # picked up without a restart. (API-key auth bypasses this — it never
        # needed Kerberos.)
        if not api_key and not have_kerberos_ticket():
            delay = RECONNECT_BACKOFF[min(attempt, len(RECONNECT_BACKOFF) - 1)]
            attempt += 1
            logger.warning(
                "no Kerberos ticket and no API key; not connecting — waiting "
                "(run kinit)", delay=delay)
            time.sleep(delay)
            continue
        headers = auth_headers(server_url, api_key, machine_id)
        try:
            run_once(server_url, machine_id, headers, wl)
            attempt = 0
        except KeyboardInterrupt:
            logger.info("shutting down")
            return 0
        except Exception as e:
            logger.warning("connection lost", error=str(e))
        delay = RECONNECT_BACKOFF[min(attempt, len(RECONNECT_BACKOFF) - 1)]
        attempt += 1
        logger.info("reconnecting", delay=delay)
        time.sleep(delay)


if __name__ == "__main__":
    sys.exit(main())
