#!/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 https://lavtomate.office.etersoft.ru)
  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 getpass
import inspect
import json
import logging
import os
import re
import shlex
import shutil
import secrets
import socket
import stat
import subprocess
import sys
import time
from http.cookies import SimpleCookie
from pathlib import Path

from lavtomate_client import LoginRequired
from lavtomate_client import auth as lavtomate_auth
from lavtomate_client import transport as lavtomate_transport

logger = logging.getLogger("lavtomate.desktop_agent")

# Package version, put here by %install of lavtomate-mcp.spec; "dev" when run
# from the source tree. Reported to lavtomate with the liveness frame.
VERSION = "0.9.2-alt1"

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
# Caps on one launch's files, and how long delivered files stay (a .vv holds
# a console password; the app reads it at start)
MAX_FILES = 8
MAX_TOTAL_BYTES = 1024 * 1024
FILE_TTL_SECS = 3600
# 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",
    "ash", "ksh", "mksh", "pdksh", "pwsh",
    "python", "python2", "python3", "perl", "ruby", "node", "lua",
    "php", "tclsh", "wish", "julia", "Rscript", "expect", "gdb",
    "awk", "gawk", "mawk", "nawk",
    # run the command given in their arguments
    "env", "busybox", "toybox", "sudo", "doas", "pkexec", "su", "runuser",
    "nohup", "setsid", "nice", "ionice", "timeout", "xargs", "stdbuf", "script",
    "chroot", "systemd-run", "flatpak-spawn", "ssh",
})
# bash4, python3.12, tclsh8.6: the name without its version
_VERSION_SUFFIX = re.compile(r"[-.]?[0-9][0-9.]*$")

# 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).
    Relative PATH entries (``.``, ``bin``) are skipped and a relative path with a
    ``/`` is refused: the result never depends on the working directory.
    """
    if "/" in app and not os.path.isabs(app):
        return None
    path = os.pathsep.join(
        p for p in os.environ.get("PATH", os.defpath).split(os.pathsep) if os.path.isabs(p))
    found = shutil.which(app, path=path)
    return os.path.abspath(found) if found else None


_BLANKS = re.compile(r"[ \t]+")


def _blanks(text: str) -> str:
    """Runs of spaces and tabs as one space: they only separate arguments
    (the server's lavtomate.desktop.apps compares the same way)."""
    return _BLANKS.sub(" ", text).strip()


def split_args(args: str) -> tuple[list[str] | None, str | None]:
    """(argv, None) of an argument string, or (None, why it cannot be parsed)."""
    try:
        return shlex.split(args or ""), None
    except ValueError as e:
        return None, f"cannot parse arguments: {e}"


def _dangerous_name(path: str) -> bool:
    name = os.path.basename(path)
    return name in DANGEROUS_APPS or _VERSION_SUFFIX.sub("", name) in DANGEROUS_APPS


def _is_dangerous(app: str) -> bool:
    """True if the app, the binary it resolves to or the target of its links
    (``shell-link`` → /bin/bash → /bin/bash4) is on the hard denylist."""
    resolved = _resolve_executable(app)
    names = [app]
    if resolved:
        names += [resolved, os.path.realpath(resolved)]
    return any(_dangerous_name(n) for n in names)


def _untrusted(path: str) -> str | None:
    """Why the executable at ``path`` may have been replaced by someone else, or None.

    The file, a link to it and every directory above them must belong to root or
    to the agent's user and be writable by nobody else (a world-writable sticky
    directory such as /tmp is fine: others cannot replace entries there).
    Below a directory of ours that others cannot enter (0700) nothing is
    reachable by them, whatever the modes there (umask 002 and a shared group).
    """
    me = os.getuid()
    for start in (os.path.abspath(path), os.path.realpath(path)):
        parts = Path(start).parts
        for depth in range(1, len(parts) + 1):
            current = os.path.join(*parts[:depth])
            try:
                st = os.lstat(current)
            except OSError as e:
                return f"cannot check {current}: {e.strerror}"
            if st.st_uid not in (0, me):
                return f"{current} belongs to another user (uid {st.st_uid})"
            if stat.S_ISDIR(st.st_mode) and st.st_uid == me and not st.st_mode & 0o011:
                break
            if not stat.S_ISLNK(st.st_mode) and st.st_mode & 0o022 and not (
                    stat.S_ISDIR(st.st_mode) and st.st_mode & stat.S_ISVTX):
                return f"{current} is writable by other users"
    return None


# 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:
        return "Negotiate " + lavtomate_transport.spnego_token(f"https://{hostname}/")
    except LoginRequired as e:
        logger.warning('SPNEGO token failed hostname=%r error=%r', hostname, 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). Checked with python gssapi
    (lavtomate_client.auth.have_ticket); re-checked each cycle so a later
    ``kinit`` is picked up without a restart. Any error counts as no ticket
    (wait rather than connect blindly).
    """
    return lavtomate_auth.have_ticket()


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 compute_host_info() -> dict:
    """Local network identity for reaching this machine (e.g. over SSH).

    Returns ``{"hostname": str, "ips": [str], "user": str}`` — the host's
    non-loopback IPv4 addresses (plus any non-link-local IPv6), its hostname
    and the local account the agent runs as. The server surfaces these in
    ``launch_list_machines`` and Settings so the AI/user knows which address
    to ``ssh`` to. Pure socket stdlib — no subprocess, no extra deps —
    so it is safe to call from a ``systemd --user`` unit. Best-effort: ``ips``
    is empty if nothing could be determined.
    """
    try:
        hostname = socket.gethostname() or ""
    except Exception:
        hostname = ""

    ips: list[str] = []
    seen: set[str] = set()

    def _add(ip: str) -> None:
        # Skip loopback and IPv6 link-local (fe80::) — neither is dialable for SSH.
        if not ip or ip in seen or ip.startswith("127."):
            return
        if ":" in ip and ip.lower().startswith("fe80:"):
            return
        seen.add(ip)
        ips.append(ip)

    # Primary: the source IP the kernel would use for the default route. A UDP
    # "connect" sends no packets — it just makes the kernel pick the local
    # endpoint — so it yields the routable LAN address even behind NAT.
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        try:
            s.connect(("8.8.8.8", 80))
            _add(s.getsockname()[0])
        finally:
            s.close()
    except OSError:
        pass

    # Also resolve our own hostname — covers multi-interface hosts and IPv6
    # addresses the UDP trick can't reach.
    if hostname:
        try:
            for info in socket.getaddrinfo(hostname, None):
                _add(info[4][0])
        except (socket.gaierror, OSError):
            pass

    try:
        user = getpass.getuser()
    except Exception:
        user = ""

    return {"hostname": hostname, "ips": ips, "user": user}


def send_liveness(ws) -> None:
    """Send the periodic status frame; never raises.

    Carries graphical-session liveness (``active``/``idle_secs``), host
    network identity (``hostname``/``ips``/``user``) and the agent ``version``.
    They share a frame because all are sent on connect and every
    ``LIVENESS_SECS``; the host's address can
    change (DHCP, dock/undock), so reporting it on the same heartbeat keeps the
    server's view fresh rather than only-at-connect.
    """
    try:
        liv = compute_liveness()
        host = compute_host_info()
        ws.send(json.dumps({"type": "liveness", **liv, **host, "version": VERSION}))
        logger.info(
            'liveness active=%r idle_secs=%r ips=%r',
            liv["active"],
            liv["idle_secs"],
            host["ips"],
        )
    except Exception as e:
        logger.warning(
            'liveness report failed error=%r',
            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 a Negotiate token
    from python gssapi (lavtomate_client); proxy-manager validates it and sets
    the ``lavtomate_session`` cookie (redirects are not followed, the cookie
    comes with the 302). 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 / login failed.
    """
    try:
        _, _, headers = lavtomate_transport.negotiate_request(
            "GET", f"{server_url.rstrip('/')}/auth/kerberos", timeout=20, with_headers=True)
    except Exception as e:  # LoginRequired (no ticket), network errors
        logger.warning('Kerberos login failed error=%r', str(e))
        return None
    for value in (headers.get_all("Set-Cookie") or []) if headers else []:
        cookie = SimpleCookie()
        try:
            cookie.load(value)
        except Exception:
            continue
        if SESSION_COOKIE_NAME in cookie:
            logger.info("obtained session cookie via Kerberos ticket")
            return cookie[SESSION_COOKIE_NAME].value
    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"


def _parse_whitelist(raw: bytes) -> list[dict]:
    data = json.loads(raw.decode("utf-8"))
    entries = data if isinstance(data, list) else data.get("apps") if isinstance(data, dict) else None
    if not isinstance(entries, list) or not all(isinstance(e, dict) for e in entries):
        raise ValueError('expected a list of entries or {"apps": [...]}')
    return entries


class Whitelist:
    """Local allowlist, read from disk on every check.

    The file is small and launches are rare, so every check reads it: an edit
    applies at once, whatever its mtime. A file that cannot be read or parsed
    refuses every launch until it is fixed — the old policy is never kept,
    because whoever edited it expects the new one to apply.
    """

    def __init__(self) -> None:
        self._path = whitelist_path()
        self._raw: bytes | None = b""
        self._entries: list[dict] = []
        self._error: str | None = None
        self._reload()

    def _reload(self) -> None:
        try:
            raw = self._path.read_bytes()
        except FileNotFoundError:
            raw = None
        except OSError as e:
            raw = f"unreadable: {e}".encode()
            self._set(raw, [], f"cannot read the local whitelist {self._path}: {e.strerror}")
            return
        if raw == self._raw:
            return
        if raw is None:
            self._set(raw, [], None)
            logger.info('whitelist file not found — empty whitelist path=%r', str(self._path))
            return
        try:
            entries = _parse_whitelist(raw)
        except Exception as e:
            self._set(raw, [], f"the local whitelist {self._path} is invalid ({e}): "
                               "the new policy is not applied, every launch is refused until it is fixed")
            return
        self._set(raw, entries, None)
        logger.info('whitelist loaded path=%r count=%r', str(self._path), len(entries))

    def _set(self, raw, entries, error) -> None:
        if error and error != self._error:
            logger.warning('whitelist not applied, launches refused error=%r', error)
        self._raw, self._entries, self._error = raw, entries, error

    def maybe_reload(self) -> None:
        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()
        if self._error:
            return False, self._error
        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"
            resolved = _resolve_executable(app)
            if resolved is None:
                return False, f"executable not found in PATH: {app}"
            untrusted = _untrusted(resolved)
            if untrusted:
                return False, f"the executable may have been replaced: {untrusted}"
            # The arguments the app gets are those of one shlex.split; the
            # template is matched against the text with any run of spaces and
            # tabs between arguments as one space — which must not change any
            # argument ("a    b" in quotes is not "a b")
            argv, error = split_args(args)
            if error:
                return False, error
            args = _blanks(args or "")
            if split_args(args)[0] != argv:
                return False, ("an argument holds several spaces or tabs in a row: "
                               "the template cannot tell it from one space, refused")
            template = _blanks(e.get("arg_template") or "")
            if not argv:
                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


# --------------------------------------------------------------------------- #
# Hiding secrets (#19720 c17): arguments and the output of a launched app go
# to the journal and back to the server, so values that look like passwords,
# tokens or keys are replaced with "***" there. The app itself gets them as is.
# --------------------------------------------------------------------------- #
REDACTED = "***"
_SECRET_NAME = (r"[\w.-]*(?:password|passwd|passphrase|pwd|token|secret|credentials?"
                r"|cookie|api[-_]?key|private[-_]?key|access[-_]?key)[\w.-]*")
# --password=X, password: X, PGPASSWORD=X
_SECRET_ASSIGN = re.compile(
    rf"(?i)(\b{_SECRET_NAME}\s*[=:]\s*)(\"[^\"]*\"|'[^']*'|[^\s,;&]+)")
# --password X inside one text
_SECRET_FLAG_TEXT = re.compile(rf"(?i)((?:^|\s)--?{_SECRET_NAME}\s+)([^\s-]\S*)")
# --password as a separate argument: the next one is its value
_SECRET_OPTION = re.compile(rf"(?i)--?{_SECRET_NAME}")
# scheme://user:password@host
_URL_USERINFO = re.compile(r"(?i)(\b[a-z][a-z0-9+.-]*://[^\s/:@]*:)[^\s/@]+(@)")
# Authorization: Bearer X / Basic X
_AUTH_SCHEME = re.compile(r"(?i)(\b(?:bearer|basic)\s+)[A-Za-z0-9._~+/=-]{8,}")


def redact_text(text: str) -> str:
    """The text with the values of secret-looking names, URL passwords and
    Bearer/Basic credentials replaced with ``***``."""
    text = _SECRET_FLAG_TEXT.sub(rf"\g<1>{REDACTED}", text)
    text = _SECRET_ASSIGN.sub(rf"\g<1>{REDACTED}", text)
    text = _URL_USERINFO.sub(rf"\g<1>{REDACTED}\g<2>", text)
    return _AUTH_SCHEME.sub(rf"\g<1>{REDACTED}", text)


def redact_argv(argv: list[str]) -> list[str]:
    """argv with secrets hidden: ``--password X`` hides X, ``--token=X`` hides X."""
    out: list[str] = []
    hide_next = False
    for arg in argv:
        if hide_next:
            out.append(REDACTED)
            hide_next = False
        elif _SECRET_OPTION.fullmatch(arg):
            out.append(arg)
            hide_next = True
        else:
            out.append(redact_text(arg))
    return out


def _tail(path: str, lines: int = 12) -> str:
    """The last lines of the app's output, with secrets hidden."""
    try:
        data = Path(path).read_text(errors="replace").splitlines()
    except OSError:
        return ""
    return redact_text("\n".join(data[-lines:]).strip())


def _launch_log() -> str:
    """A new private (0600) file for the app's output: in the agent's own 0700
    directory (removed by its TTL and at logout), else in the temp directory."""
    import tempfile
    fd, d, error = _open_file_dir()
    if fd is not None:
        os.close(fd)
    log_fd, log_path = tempfile.mkstemp(prefix="launch-", suffix=".log",
                                        dir=None if error else d)
    os.close(log_fd)
    return log_path


def launch(app: str, args: list[str]) -> dict:
    """Launch the app detached with the argv the whitelist checked. 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.
    """
    # #6: per-agent sliding-window rate limit.
    if _rate_limited():
        logger.warning(
            'launch refused: rate limit app=%r',
            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, *args]
    logger.info(
        'launch resolved app=%r resolved=%r',
        app,
        resolved,
    )
    log_path = _launch_log()
    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": redact_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=%r rc=%r',
        app,
        rc,
    )
    return {
        "ok": False,
        "data": {"pid": proc.pid, "argv": redact_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:
    """Remove a file the agent made; a failure is logged (without the content)."""
    try:
        os.unlink(path)
    except FileNotFoundError:
        pass
    except OSError as e:
        logger.warning('cannot remove %s: %s', path, e.strerror or e)


# --------------------------------------------------------------------------- #
# 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 _check_dir_parent(parent: str) -> str | None:
    """Why files must not go under ``parent``, or None: it must have no links on
    its way, belong to root or to us and be writable by nobody else, unless it
    is a sticky directory like /tmp (others cannot replace our entries there)."""
    if os.path.realpath(parent) != parent:
        return f"{parent} goes through a symbolic link"
    try:
        st = os.stat(parent)
    except OSError as e:
        return f"cannot use {parent}: {e.strerror}"
    if st.st_uid not in (0, os.getuid()):
        return f"{parent} belongs to another user"
    if st.st_mode & 0o022 and not st.st_mode & stat.S_ISVTX:
        return f"{parent} is writable by other users"
    return None


def _open_file_dir() -> tuple[int | None, str, str | None]:
    """(directory fd, path, None) of the delivery directory, or (None, path, error).

    The directory is ours, mode 0700, not a link and not under one; files are
    then created relative to the fd, so a link put in place later is not followed.
    Delivered files older than FILE_TTL_SECS are removed.
    """
    d = os.path.abspath(file_dir())
    parent = os.path.dirname(d)
    if not os.path.lexists(parent):
        # /tmp/lavtomate-<uid> without XDG_RUNTIME_DIR
        error = _check_dir_parent(os.path.dirname(parent))
        if error:
            return None, d, error
        try:
            os.mkdir(parent, 0o700)
        except FileExistsError:
            pass
        except OSError as e:
            return None, d, f"cannot create {parent}: {e.strerror}"
    error = _check_dir_parent(parent)
    if error:
        return None, d, error
    try:
        os.mkdir(d, 0o700)
    except FileExistsError:
        pass
    except OSError as e:
        return None, d, f"cannot create file dir: {e}"
    try:
        fd = os.open(d, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC)
    except OSError as e:
        return None, d, f"cannot open file dir {d} (a symbolic link?): {e.strerror}"
    st = os.fstat(fd)
    if st.st_uid != os.getuid():
        os.close(fd)
        return None, d, f"file dir {d} belongs to another user"
    if stat.S_IMODE(st.st_mode) != 0o700:
        os.fchmod(fd, 0o700)
    cutoff = time.time() - FILE_TTL_SECS
    for entry in os.scandir(fd):
        try:
            if entry.is_file(follow_symlinks=False) and entry.stat(follow_symlinks=False).st_mtime < cutoff:
                os.unlink(entry.name, dir_fd=fd)
        except OSError:
            pass
    return fd, d, None


def _decode(entry, index: int) -> tuple[str, bytes | None, str | None]:
    """(name, bytes, None) of a file entry, or (name, None, error)."""
    if not isinstance(entry, dict):
        return "", None, f"file entry {index} is not an object"
    name = entry.get("name")
    raw = entry.get("content_b64")
    if not isinstance(name, str) or not name:
        return "", None, f"file entry {index} has no name"
    if not isinstance(raw, str):
        return name, None, f"file {name}: content_b64 must be a base64 string"
    # The size is known from the base64 length: refuse before decoding
    if len(raw) > 4 * ((MAX_FILE_BYTES + 2) // 3):
        return name, None, f"file {name} exceeds {MAX_FILE_BYTES} bytes"
    try:
        data = base64.b64decode(raw, validate=True)
    except ValueError as e:
        return name, None, f"bad base64 for {name}: {e}"
    if len(data) > MAX_FILE_BYTES:
        return name, None, f"file {name} exceeds {MAX_FILE_BYTES} bytes"
    return name, data, None


def plan_files(files: list) -> tuple[list[dict], dict[str, str], str | None]:
    """Check the delivered files and choose their paths, writing nothing yet.

    Returns ``(plan, name -> path, error_or_None)``. Every file gets a name of
    its own (``vm-101.<random>.vv``), so deliveries never overwrite each other;
    a file is reachable by its name as sent and as sanitized. The files are
    written by :func:`write_files` only once the launch is allowed.
    """
    if not files:
        return [], {}, None
    if not isinstance(files, list):
        return [], {}, "files must be a list"
    if len(files) > MAX_FILES:
        return [], {}, f"too many files: {len(files)} (at most {MAX_FILES})"
    fd, d, error = _open_file_dir()
    if error:
        return [], {}, error
    os.close(fd)
    plan: list[dict] = []
    names: dict[str, str] = {}
    total = 0
    for index, entry in enumerate(files):
        name, data, error = _decode(entry, index)
        if error:
            return [], {}, error
        total += len(data)
        if total > MAX_TOTAL_BYTES:
            return [], {}, f"files exceed {MAX_TOTAL_BYTES} bytes in total"
        safe = _sanitize_file_name(name)
        stem, ext = os.path.splitext(safe)
        unique = f"{stem}.{secrets.token_hex(4)}{ext}"
        path = os.path.join(d, unique)
        for key in dict.fromkeys((name, safe)):
            if key in names:
                return [], {}, f"two files are named {key}"
            names[key] = path
        plan.append({"name": unique, "data": data, "path": path})
    return plan, names, None


def write_files(plan: list[dict]) -> str | None:
    """Write planned files (0600, never through a link, never over a file);
    on any error remove what this call wrote and return why."""
    if not plan:
        return None
    fd, _, error = _open_file_dir()
    if error:
        return error
    written: list[str] = []
    try:
        for item in plan:
            try:
                out = os.open(item["name"], os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW
                              | os.O_CLOEXEC, 0o600, dir_fd=fd)
            except OSError as e:
                return f"write {item['name']} failed: {e}"
            written.append(item["name"])
            try:
                os.fchmod(out, 0o600)
                view = memoryview(item["data"])
                while view:
                    view = view[os.write(out, view):]
            except OSError as e:
                return f"write {item['name']} failed: {e}"
            finally:
                os.close(out)
        written = []
        return None
    finally:
        for name in written:
            try:
                os.unlink(name, dir_fd=fd)
            except OSError:
                pass
        os.close(fd)


def deliver_files(files: list) -> tuple[dict, str | None]:
    """Check and write delivered files at once: ``(name -> path, error_or_None)``."""
    plan, names, error = plan_files(files)
    if error:
        return {}, error
    error = write_files(plan)
    return ({}, error) if error else (names, None)


_FILE_TOKEN = re.compile(r"@FILE@|@file:([^@]*)@")


def substitute_file_args(args: str, name_to_path: dict[str, str]) -> tuple[str, str | None]:
    """Replace ``@file:<name>@`` and ``@FILE@`` with delivered paths, in one pass
    (a path holding a token is not substituted again) and shell-quoted, so a
    path with spaces, quotes or backslashes stays one argument.

    Returns ``(args, None)`` or ``(args, error)`` for a token with no file.
    Empty ``args`` with files means the first file's path (``remote-viewer <vv>``).
    """
    paths = list(dict.fromkeys(name_to_path.values()))
    if paths and not args:
        return shlex.quote(paths[0]), None
    missing: list[str] = []

    def path_of(match):
        name = match.group(1)
        path = paths[0] if name is None and paths else name_to_path.get(name or "")
        if path is None:
            missing.append(match.group(0))
            return match.group(0)
        return shlex.quote(path)

    out = _FILE_TOKEN.sub(path_of, args)
    if missing:
        return out, f"no delivered file for {missing[0]}"
    return out, None


# --------------------------------------------------------------------------- #
# 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).

    Never raises: an unexpected error is the launch's failure, not the end of
    the connection.
    """
    mtype = payload.get("type")
    if mtype == "ping":
        return {"type": "pong"}
    if mtype != "launch":
        return None
    request_id = payload.get("request_id", "")
    try:
        result = _launch_request(payload, wl)
    except Exception as e:
        logger.exception('launch request failed request_id=%r', request_id)
        result = {"ok": False, "error": f"agent error: {type(e).__name__}: {e}"}
    return {"type": "result", "request_id": request_id, **result}


def _launch_request(payload: dict, wl: Whitelist) -> dict:
    app = payload.get("app")
    args = payload.get("args") or ""
    if not isinstance(app, str) or not app.strip():
        return {"ok": False, "error": "no app specified"}
    if not isinstance(args, str):
        return {"ok": False, "error": "args must be a string"}
    app, args = app.strip(), args.strip()
    # Delivered files (e.g. a SPICE .vv) get their sandbox paths first; the
    # paths are substituted into args, so the whitelist check below authorizes
    # exactly what is executed, and the files are written only once the launch
    # is allowed. No shell is used.
    plan, name_to_path, error = plan_files(payload.get("files") or [])
    if not error and (name_to_path or "@" in args):
        args, error = substitute_file_args(args, name_to_path)
    if error:
        logger.warning('file delivery failed app=%r error=%r', app, error)
        return {"ok": False, "error": f"file delivery failed: {error}"}
    allowed, reason = wl.check(app, args)
    if not allowed:
        logger.warning('launch refused app=%r nargs=%d reason=%r',
                       app, len(split_args(args)[0] or []), reason)
        logger.debug('refused args=%r', redact_text(args))
        return {"ok": False, "error": f"refused locally: {reason}"}
    argv, error = split_args(args)
    if error:
        return {"ok": False, "error": f"refused locally: {error}"}
    error = write_files(plan)
    if error:
        logger.warning('file delivery failed app=%r error=%r', app, error)
        return {"ok": False, "error": f"file delivery failed: {error}"}
    result = launch(app, argv)
    logger.info(
        'launch app=%r nargs=%d ok=%r pid=%r files=%r',
        app,
        len(argv),
        result.get("ok"),
        (result.get("data") or {}).get("pid"),
        [item["path"] for item in plan] or None,
    )
    logger.debug('launch argv=%r', redact_argv(argv))
    if not result.get("ok"):
        for item in plan:
            _rm(item["path"])
    return result


def connect_kwargs(connect, ssl_ctx) -> dict:
    """Keyword arguments for ``websockets.sync.client.connect``.

    Keepalive is app-level (``{"type": "ping"}``), so the library's own pings
    are turned off. The sync client of websockets 12.0 (p10) has no keepalive
    and rejects ``ping_interval``, and takes the TLS context as ``ssl_context``
    (``ssl`` since 13.0).
    """
    params = inspect.signature(connect).parameters
    kwargs = {"open_timeout": 15, "close_timeout": 5}
    if "ping_interval" in params:
        kwargs["ping_interval"] = None
    if ssl_ctx is not None:
        kwargs["ssl" if "ssl" in params else "ssl_context"] = ssl_ctx
    return kwargs


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)
    # Credentials go in headers (Authorization or Cookie), never in the URI
    logger.info(
        'connecting uri=%r',
        uri,
    )
    ssl_ctx = _ssl_context() if uri.startswith("wss://") else None
    with connect(uri, additional_headers=headers,
                 **connect_kwargs(connect, ssl_ctx)) as ws:
        logger.info(
            'connected machine=%r',
            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:
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
        datefmt="%Y-%m-%dT%H:%M:%S%z",
    )
    server_url = os.environ.get("LAVTOMATE_DESKTOP_SERVER", "https://lavtomate.office.etersoft.ru")
    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=%r',
                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=%r',
                str(e),
            )
        delay = RECONNECT_BACKOFF[min(attempt, len(RECONNECT_BACKOFF) - 1)]
        attempt += 1
        logger.info(
            'reconnecting delay=%r',
            delay,
        )
        time.sleep(delay)


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