#!/usr/bin/python3
"""ASU VPN tray applet.

Connecting happens in two phases. First openconnect-sso runs as you and opens a
browser window for the ASU SSO login; it prints back a host, a certificate
fingerprint and a short-lived session cookie. Then asuvpn-helper runs under
pkexec and feeds that cookie to openconnect, which needs root to set up the tun
device. Only the cookie crosses the privilege boundary, and the browser never
runs as root.

The helper stays attached to a pipe, so closing that pipe disconnects: no
second password prompt, and no orphaned root process if this applet dies.
"""

import argparse
import importlib.machinery
import importlib.util
import os
import re
import stat
import shlex
import shutil
import signal
import socket
import struct
import subprocess
import sys
import threading
import time
from pathlib import Path

# The GUI stack is optional at import time: `asuvpn status`, `log`, `disconnect`
# and `quit` are pure socket operations, and a missing typelib should not turn
# them into a traceback while a tunnel is up.
GUI_ERROR = None
try:
    import gi

    gi.require_version("Gtk", "3.0")
    gi.require_version("AyatanaAppIndicator3", "0.1")
    gi.require_version("Notify", "0.7")

    from gi.repository import AyatanaAppIndicator3 as AppIndicator
    from gi.repository import Gdk, GLib, Gtk, Notify
except (ImportError, ValueError) as exc:  # pragma: no cover - depends on the host
    GUI_ERROR = str(exc)


# Set before the contract is loaded, and this is the reason: loading it writes
# a __pycache__ into the directory beside us -- the one the helper runs out of
# as root -- created with the ambient umask and after install.sh has tightened
# everything else. Nothing here benefits from a bytecode cache; a few hundred
# microseconds once per run is not worth writing into that directory.
sys.dont_write_bytecode = True

def _contract():
    """Load the shared contract from beside this program, by explicit path.

    Not `import`: this is reached through a symlink in ~/.local/bin, so
    sys.path[0] is that directory and not the one its siblings live in. An
    explicit path built from the *resolved* location is the only form that
    works for both the symlink and a direct run.
    """
    here = os.path.dirname(os.path.realpath(__file__))
    path = os.path.join(here, "asuvpn_contract.py")
    if os.geteuid() == 0:
        for target in (here, path):
            info = os.stat(target)
            if info.st_mode & stat.S_IWOTH or (
                    info.st_mode & stat.S_IWGRP and info.st_gid != info.st_uid):
                raise SystemExit(f"refusing to load {target}: writable by others")
    loader = importlib.machinery.SourceFileLoader("asuvpn_contract", path)
    spec = importlib.util.spec_from_loader("asuvpn_contract", loader)
    module = importlib.util.module_from_spec(spec)
    loader.exec_module(module)
    return module


C = _contract()


def _xdg_dir(variable, fallback):
    """XDG base directory, matching GLib: a relative value is ignored."""
    return Path(C.xdg_dir(variable, fallback))


CONFIG_FILE = Path(C.config_path())
# Loaded once here for the one-shot CLI paths, and refreshed by the applet when
# the file changes, so most settings take effect without a reconnect. Problems
# in the file are reported by the applet's health tick, which re-reads it —
# the initial load's problems need no keeping.
SETTINGS = C.load_settings(CONFIG_FILE)[0]


def reload_settings():
    """Re-read the config. Returns the problems found, newest wins."""
    global SETTINGS  # one config per process, by design
    SETTINGS, problems = C.load_settings(CONFIG_FILE)
    return problems


APP_ID = "asuvpn-tray"
APP_NAME = "ASU VPN"
DEFAULT_SERVER = C.SETTINGS_BY_NAME["server"].default
HERE = Path(__file__).resolve().parent
HELPER = HERE / "asuvpn-helper"
LOG_FILE = _xdg_dir("XDG_CACHE_HOME", ".cache") / "asuvpn" / "session.log"
AUTOSTART_FILE = (
    _xdg_dir("XDG_CONFIG_HOME", ".config") / "autostart" / "asuvpn-tray.desktop"
)
MAX_LOG_LINES = 4000  # the in-memory tail; the file's cap is the log-max-kb setting
# How many times a tunnel that died on its own is rebuilt before the applet
# stops and waits to be asked. Spacing alone is not a bound: a network that is
# down stays down, and every attempt costs a browser window and a Duo push, so
# the third one is where unattended recovery stops being a kindness.
MAX_REBUILDS = 3
# What the badge appends while a rebuild is still owed. One spelling, because
# it is written in one place and taken off again in another, and two copies of
# a literal like that drift the moment either is reworded.
REBUILDING = "; rebuilding"
IFF_UP = 0x1

# Not 2: argparse already exits 2 for any bad command line, so a script could
# not tell "wrong server" from "you typo'd a flag".
EXIT_WRONG_SERVER = 4
EXIT_NOT_RUNNING = 3

# The eight states. Every situation has exactly one name: RECOVERING
# (openconnect is re-establishing its own session) and DEMOTED (established
# but not carrying traffic — the watchdog's verdict) used to hide inside
# CONNECTING behind a flag, and the flag's scattered readers were where the
# state bugs lived.
DISCONNECTED = "disconnected"
AUTHENTICATING = "authenticating"
CONNECTING = "connecting"
CONNECTED = "connected"
RECOVERING = "recovering"
DEMOTED = "demoted"
DISCONNECTING = "disconnecting"
FAILED = "failed"

ICONS = {
    DISCONNECTED: "network-vpn-disconnected-symbolic",
    AUTHENTICATING: "network-vpn-acquiring-symbolic",
    CONNECTING: "network-vpn-acquiring-symbolic",
    CONNECTED: "network-vpn-symbolic",
    RECOVERING: "network-vpn-acquiring-symbolic",
    DEMOTED: "network-vpn-acquiring-symbolic",
    DISCONNECTING: "network-vpn-acquiring-symbolic",
    FAILED: "network-error-symbolic",
}

BUSY_STATES = (AUTHENTICATING, CONNECTING, RECOVERING, DEMOTED, DISCONNECTING)

# The independent things that can find a tunnel unusable. Strikes are counted
# per source, never pooled: they fail for unrelated reasons and each has to
# clear on its own evidence (see _check_clear).
#
#   device  the kernel's view -- the device, its incarnation, its routes
#   probe   whether anything actually answers through the tunnel
#   dns     whether the resolver the VPN pushed is still configured on its link
#
# DEMOTED means "established but not usable", and the three are not usable in
# different ways, so each brings its own wording rather than sharing one that
# would be a lie for two of them. A tunnel whose DNS has been taken off the link
# is carrying traffic perfectly; what it cannot do is resolve the names that
# traffic is for.
CHECK_SOURCES = ("device", "probe", "dns")
DEMOTION_TEXT = {
    "device": "not carrying traffic",
    "probe": "not carrying traffic",
    "dns": "DNS not configured",
}


def fresh_strikes():
    """A clean strike count for every source.

    One function because this was written out three times and gained a source
    in none of them -- a new check would have raised KeyError on its first bad
    verdict, in a thread, at the moment it had something to report.
    """
    return dict.fromkeys(CHECK_SOURCES, 0)


# RECOVERING and DEMOTED keep the wording users already know; the detail line
# ("link lost, retrying" / "not carrying traffic — …") is what distinguishes
# them, exactly as before they were real states.
STATE_LABELS = {
    DISCONNECTED: "Disconnected",
    AUTHENTICATING: "Signing in…",
    CONNECTING: "Connecting…",
    CONNECTED: "Connected",
    RECOVERING: "Connecting…",
    DEMOTED: "Connecting…",
    DISCONNECTING: "Disconnecting…",
    FAILED: "Not connected",
}

# ---------------------------------------------------- the message vocabulary
#
# Everything that can happen is a message; the machine below is the only thing
# that turns messages into state. Watchers, workers, log readers and user
# verbs INJECT these — they never assign state themselves.
MSG_CONNECT = "connect"                    # user
MSG_DISCONNECT = "disconnect"              # user
MSG_RECONNECT = "reconnect"                # user; the ladder reaches the same
                                           # handler directly, keep_log=True
MSG_CANCEL = "cancel"                      # user
MSG_QUIT = "quit"                          # user
MSG_AUTH_OK = "auth-ok"                    # sign-in: host, cookie, fingerprint, attempt
MSG_AUTH_FAILED = "auth-failed"            # sign-in: reason, attempt
MSG_TUNNEL_UP = "tunnel-up"                # event/fallback: device, address, dns
MSG_LINK_LOST = "link-lost"                # event/fallback
MSG_DEVICE = "device-named"                # helper: device
MSG_FATAL = "fatal"                        # helper: sentence
MSG_WARNING = "warning"                    # helper: sentence
MSG_PROBLEM = "problem"                    # failure-pattern scan: sentence
MSG_CHECK = "check"                # watchdog: source, reason (None=ok), detail
MSG_HELPER_EXITED = "helper-exited"        # reader thread: status, generation
MSG_TEARDOWN_FINISHED = "teardown-finished"  # reconnect worker: old tunnel gone
MSG_TEARDOWN_TIMEOUT = "teardown-timeout"    # any teardown worker: during
MSG_REBUILD = "rebuild"                    # health tick: a dropped tunnel may
                                           # be worth rebuilding by itself

ANY = "*"

# The transition table: (state, message) -> handler. A pair not listed here
# is dropped, and the drop is logged — silence was how verdicts landing
# mid-teardown got to act. `asuvpn selftest` walks this table and drives the
# load-bearing rows with real message sequences.
TRANSITIONS = {
    (DISCONNECTED, MSG_CONNECT): "start_signin",
    (FAILED, MSG_CONNECT): "start_signin",
    (DEMOTED, MSG_CONNECT): "connect_means_reconnect",
    (DISCONNECTED, MSG_RECONNECT): "start_signin",
    (FAILED, MSG_RECONNECT): "start_signin",
    (CONNECTED, MSG_RECONNECT): "teardown_reconnect",
    (DEMOTED, MSG_RECONNECT): "teardown_reconnect",
    (DISCONNECTED, MSG_DISCONNECT): "already_disconnected",
    (FAILED, MSG_DISCONNECT): "already_disconnected",
    (AUTHENTICATING, MSG_DISCONNECT): "cancel_attempt",
    (CONNECTING, MSG_DISCONNECT): "cancel_attempt",
    (CONNECTED, MSG_DISCONNECT): "teardown_disconnect",
    (RECOVERING, MSG_DISCONNECT): "teardown_disconnect",
    (DEMOTED, MSG_DISCONNECT): "teardown_disconnect",
    (AUTHENTICATING, MSG_CANCEL): "cancel_attempt",
    (CONNECTING, MSG_CANCEL): "cancel_attempt",
    (RECOVERING, MSG_CANCEL): "cancel_attempt",
    (ANY, MSG_QUIT): "quit_everything",
    (AUTHENTICATING, MSG_AUTH_OK): "start_tunnel",
    (AUTHENTICATING, MSG_AUTH_FAILED): "signin_failed",
    (CONNECTING, MSG_TUNNEL_UP): "tunnel_came_up",
    (RECOVERING, MSG_TUNNEL_UP): "tunnel_came_up",
    (CONNECTED, MSG_TUNNEL_UP): "refresh_address",
    (DEMOTED, MSG_TUNNEL_UP): "adopt_only",
    (CONNECTED, MSG_LINK_LOST): "link_lost",
    (RECOVERING, MSG_LINK_LOST): "still_recovering",
    (CONNECTED, MSG_CHECK): "weigh_check",
    (RECOVERING, MSG_CHECK): "weigh_check",
    (DEMOTED, MSG_CHECK): "weigh_check",
    (ANY, MSG_DEVICE): "remember_device",
    (ANY, MSG_FATAL): "remember_fatal",
    (ANY, MSG_PROBLEM): "remember_problem",
    (ANY, MSG_WARNING): "helper_warning",
    (ANY, MSG_HELPER_EXITED): "helper_exited",
    (DISCONNECTING, MSG_TEARDOWN_FINISHED): "teardown_finished",
    (DISCONNECTING, MSG_TEARDOWN_TIMEOUT): "teardown_timed_out",
    # Only from FAILED, and only for a tunnel that had carried traffic: the
    # arming rule lives in the exit row, so this one cannot fire behind a
    # sign-in that never produced a working tunnel.
    (FAILED, MSG_REBUILD): "rebuild_dropped",
}

# Taken from the message catalogue of the installed openconnect, not from
# memory. openconnect renamed the success line to "Configured as …" in v8, and
# v9.12 contains no "Connected as" string at all — matching on that meant the
# applet could only reach CONNECTED when DTLS happened to negotiate. On a
# network that blocks UDP/443 or forces a proxy the tunnel works while the tray
# sits in "Connecting…" forever.
#
# "Configured as" is the authoritative line. "CSTP connected" covers the
# TLS-only case, and the DTLS line is kept because it arrives first when UDP is
# available. The "Connected as" forms are retained only for openconnect 7.x,
# which is why they are marked as no longer expected.
#
# Each entry is (literal, regex, still expected in the installed binary). The
# literal is what `asuvpn selftest` looks for in the binary's own message
# catalogue, so these are checked against the release in front of us rather
# than against anyone's memory of it.
CONNECTED_MESSAGES = (
    ("Configured as", r"Configured as (?P<addr>[0-9a-fA-F.:]+)", True),
    ("CSTP connected", r"CSTP connected", True),
    ("Established DTLS connection", r"Established DTLS connection", True),
    ("Connected %s as %s",
     r"Connected (?P<dev>[\w.-]+) as (?P<addr>[0-9a-fA-F.:]+)", False),
    ("Connected as", r"Connected as (?P<addr>[0-9a-fA-F.:]+)", False),
)
CONNECTED_PATTERNS = tuple(re.compile(r) for _, r, _ in CONNECTED_MESSAGES)
# openconnect-sso colours its logs (CSI sequences), and a hostile VPN banner
# could carry OSC sequences or bare control characters, which would reach a
# terminal through `asuvpn log`. Everything ESC-led, every C0 control except
# tab, and the C1 controls (which arrive as well-formed UTF-8, so decoding
# does not replace them) are removed where the log line is built.
ANSI_ESCAPE = re.compile(
    r"\x1b\[[0-9:;<=>?]*[ -/]*[@-~]"           # CSI, private forms included
    r"|\x1b\][^\x07\x1b]*(?:\x07|\x1b\\)?"     # OSC: titles, hyperlinks
    r"|[\x00-\x08\x0b-\x1f\x7f\u0080-\u009f]"    # bare C0, DEL, C1 controls
)

# Deliberately narrow. A bare /Reconnect/ matched thirty message templates in
# v9.12, two of which openconnect prints while the tunnel is perfectly healthy
# ("DTLS Rehandshake failed; reconnecting.", and the reconnect-after-drop
# banner on every CSTP connect) — so the badge flapped and fired a spurious
# "connected" notification each time. These three mean the link is actually
# down; a later "Configured as"/"CSTP connected" promotes it back.
RECONNECTING_MESSAGES = (
    ("Failed to reconnect to host", r"Failed to reconnect to host", True),
    ("Failed to reconnect to proxy", r"Failed to reconnect to proxy", True),
    ("Detected dead peer", r"Detected dead peer", True),
)
RECONNECTING_PATTERNS = tuple(re.compile(r) for _, r, _ in RECONNECTING_MESSAGES)

# Each verified present in the installed openconnect. "Failed to obtain WebVPN
# cookie" and "Login failed" were dead patterns carried over from an older
# release and have been dropped.
FAILURE_MESSAGES = (
    ("Cookie was rejected", r"Cookie was rejected", True),
    ("Failed to connect to", r"Failed to connect to", True),
    # Not the bare phrase. ASU's chain is not in the system trust store, so a
    # perfectly healthy connect always logs "Server certificate verify failed:
    # signer not found" and then succeeds on the --servercert pin instead --
    # which had us announce "openconnect reported a problem" on every single
    # session. A real pin mismatch still ends "verify failed: <something else>"
    # and is still caught.
    ("certificate verify failed",
     r"certificate verify failed(?!: signer not found)", True),
    ("Failed to open HTTPS connection", r"Failed to open HTTPS connection", True),
)
FAILURE_PATTERNS = tuple(
    re.compile(r, re.IGNORECASE) for _, r, _ in FAILURE_MESSAGES
)


def route_count(device):
    """(IPv4, IPv6) routes pointing at this device; None for an unreadable family.

    Counted per family, not summed: `ip route flush dev X` removes only IPv4,
    and on 2026-08-23 the six surviving IPv6 routes kept a summed count nonzero
    while every IPv4 path through the tunnel was gone — so the route check
    watched a broken tunnel and saw nothing.

    Read straight out of /proc rather than shelled out to `ip`: this runs on the
    GLib main loop every few seconds, and two file reads cost microseconds where
    a subprocess would stall the UI.
    """
    if not device:
        return 0, 0
    try:
        with open("/proc/net/route") as fh:
            next(fh, None)  # header
            v4 = sum(1 for line in fh
                     if (f := line.split()) and f[0] == device)
    except OSError:
        v4 = None
    try:
        with open("/proc/net/ipv6_route") as fh:
            v6 = sum(1 for line in fh
                     if (f := line.split()) and f[-1] == device)
    except OSError:
        v6 = None  # a kernel without IPv6; the v4 count still stands
    return v4, v6


def probe_tunnel(address, port, timeout):
    """Does anything answer through the tunnel? True, False, or None if unclear.

    This is the check that catches a black hole: a tunnel whose device is up and
    whose routes are installed, but which carries nothing. Every passive test
    passes in that state, and so does openconnect's own dead peer detection when
    the server never asked for it.

    A refusal counts as alive. Measured against a live ASU tunnel, one resolver
    completed the handshake in 29ms and another answered with a RST in 20ms --
    and the RST is just as good an answer, because it proves a packet crossed in
    each direction. Only silence means the tunnel is not carrying traffic, which
    is why nothing here cares whether the service is actually up.
    """
    if not address:
        return None, "no target"
    try:
        socket.create_connection((address, port), timeout=timeout).close()
        return True, f"{address} answered"
    except ConnectionRefusedError:
        return True, f"{address} refused the connection, which is still a reply"
    except TimeoutError:  # socket.timeout is this same class since 3.10
        return False, f"{address} did not answer in {timeout}s"
    except (OSError, ValueError) as exc:
        # No route, network down: the passive checks own that diagnosis and
        # give a better one, so this declines to add a second opinion.
        # ValueError covers getaddrinfo's UnicodeError on a name that cannot
        # be resolved at all -- equally inconclusive, and letting it raise
        # would kill the probe thread and wedge probing for good.
        return None, f"{address}: {exc.__class__.__name__}"


# The tray reads systemd-resolved; it never writes it. Configuring a link needs
# root, and asuvpn-notify already does it at the only moment it can be done
# correctly. What is left here is the question nothing else was asking: is the
# configuration it installed still there?
def parse_link_dns(text):
    """The resolver addresses in one `resolvectl dns <link>` answer.

    "Link 9 (asuvpn0): 192.0.2.53 2001:db8::1", or nothing after the colon when
    the link has none. Split on the first colon only, because an IPv6 address
    is full of them.

    A server configured for DNS-over-TLS prints as "addr#servername", so the
    suffix is dropped: the address is what was asked about, and comparing the
    decorated form would report a resolver missing that is sitting right there.
    Nothing this applet installs carries one -- it sets plain addresses -- but
    the link is not exclusively ours to configure.

    Separate from link_resolvers so it can be tested against real output
    without a resolvectl to run; the half that can be got wrong quietly is the
    parsing, not the subprocess.
    """
    _, sep, rest = text.partition(":")
    return [word.split("#", 1)[0] for word in rest.split()] if sep else []


def link_resolvers(device):
    """The resolvers systemd-resolved has on this link, or None if unanswerable.

    None is not a verdict. No resolvectl, resolved not running, the device
    already gone -- none of those is evidence that DNS is wrong, and the device
    checks give a far better diagnosis of the last one.

    Reading needs no privilege: this is a query over resolved's bus, and the
    per-link state files under /run/systemd/resolve/netif are 0700 to
    systemd-resolve, so the tool is the only way in from an ordinary account.
    """
    binary = C.resolvectl_path()
    if binary is None or not device:
        return None
    try:
        done = subprocess.run([binary, "dns", device], stdin=subprocess.DEVNULL,
                              stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
                              timeout=C.RESOLVECTL_TIMEOUT, check=False)
    except (OSError, ValueError, subprocess.SubprocessError):
        return None
    if done.returncode != 0:
        return None
    return parse_link_dns(done.stdout.decode("utf-8", "replace"))


def resolv_conf_lists(resolver):
    """Is this resolver named in /etc/resolv.conf?

    Only ever asked as a second opinion. asuvpn-notify hands DNS back to the
    stock vpnc-script whenever it cannot configure the link itself, and the
    stock script's fallback writes /etc/resolv.conf -- so a resolver that is not
    on the link may still be perfectly in force, installed the old way. Missing
    that would demote a working tunnel every twenty seconds and walk the
    recovery ladder up to an unattended sign-in, which is a Duo push every five
    minutes for a tunnel with nothing wrong with it.
    """
    try:
        with open("/etc/resolv.conf", encoding="utf-8", errors="replace") as fh:
            for line in fh:
                fields = line.split("#", 1)[0].split()
                if len(fields) >= 2 and fields[0] == "nameserver" \
                        and fields[1] == resolver:
                    return True
    except OSError:
        pass
    return False


def resolver_health(device, expected):
    """Why the tunnel's resolver is not in force anywhere, or None. Third check.

    This is the failure the other two cannot see, and the reason it needed its
    own source: the device is up, the routes are installed, packets cross the
    tunnel in both directions -- and internal names resolve to whatever the
    public internet says, because something took the resolver back off the link.
    systemd-resolved rewriting its own stub file does exactly that, and every
    other check passes throughout.

    The question is deliberately "is this resolver in force", not "is it on the
    link". Which mechanism installed it is asuvpn-notify's business and it has
    two; asserting the one it prefers would demote a tunnel using the other.

    Only the resolver is asserted, not the domains beside it. It is the whole of
    what makes the link resolve anything, it is the one value that reached here
    from the tunnel itself, and a check that tests less is a check that cannot
    be wrong about what it tests.
    """
    if not device or not expected:
        return None, "nothing to check"
    servers = link_resolvers(device)
    if servers is None:
        return None, "systemd-resolved did not answer"
    if expected in servers:
        return None, f"{expected} is still on {device}"
    if resolv_conf_lists(expected):
        return None, f"{expected} is in /etc/resolv.conf, not on {device}"
    return ("the resolver the VPN pushed is no longer in force",
            f"{device} has {' '.join(servers) or 'no resolvers'} and"
            f" /etc/resolv.conf does not name {expected} either")


def tunnel_health(device, ifindex, routes_seen=None):
    """Why the tunnel cannot be carrying traffic, or None. Facts, not packets.

    `routes_seen` says which route families this tunnel has been observed to
    have ({"routes4": bool, "routes6": bool}); a family the tunnel had, wholly
    gone, is a verdict even while the other survives — `ip route flush dev X`
    removes only IPv4, and the summed count this replaced looked healthy
    through exactly that break.

    Deliberately narrow, because three of the four checks that suggest
    themselves first are wrong here — each was tried against a live ASU tunnel
    and each would have raised a false alarm on a perfectly healthy link:

      * `operstate` is "unknown" for a tun device, never "up"
      * IFF_RUNNING is not set on a tun device either
      * there is no default route through it: ASU is a split tunnel, so the
        default route stays on the physical link and only the advertised
        prefixes are routed in (53 IPv4 + 6 IPv6 routes on a live session)
      * an idle tunnel moves no bytes at all, so flat counters mean nothing

    What survives is the set of things that are true of every working tunnel and
    false of a broken one: the device still exists, it is still the incarnation
    this session created, it is still administratively up, and the routes that
    make it useful are still installed. The last is the one openconnect's own
    dead-peer detection cannot see — it will happily keep a session alive while
    something else has wiped the routes out from under it.
    """
    if not device:
        return None, {}  # no event channel and no DEVICE line; nothing to watch
    if not C.INTERFACE_RE.match(device):
        return None, {"dev": device, "ignored": "not a usable device name"}
    facts = {"dev": device}
    current = C.interface_index(device)
    if current is None:
        return "the tunnel device is gone", facts
    facts["ifindex"] = current
    if ifindex is not None and current != ifindex:
        return "the tunnel device was replaced by a different one", facts
    try:
        with open(f"/sys/class/net/{device}/flags") as fh:
            flags = int(fh.read().strip(), 16)
        facts["flags"] = f"0x{flags:x}"
        if not flags & IFF_UP:
            return "the tunnel device is down", facts
    except (OSError, ValueError):
        pass  # unreadable flags are not evidence of a fault
    v4, v6 = route_count(device)
    facts["routes4"], facts["routes6"] = v4, v6
    # Every family we could read, and at least one we could: `v4 == 0 and
    # v6 == 0` looks like the same test and is not. A kernel built without
    # IPv6 has no /proc/net/ipv6_route, so v6 comes back None -- unreadable,
    # not zero -- and that comparison then stayed False through a table with
    # nothing in it at all. Unreadable is still never evidence of a fault, so
    # both families unreadable says nothing rather than everything.
    readable = [count for count in (v4, v6) if count is not None]
    if readable and not any(readable):
        return "no routes point at the tunnel any more", facts
    seen = routes_seen or {}
    if seen.get("routes4") and v4 == 0:
        return "the tunnel's IPv4 routes are gone", facts
    if seen.get("routes6") and v6 == 0:
        return "the tunnel's IPv6 routes are gone", facts
    return None, facts


def write_setting(name, value):
    """Change one setting in place, leaving every other line as the user left it.

    Rewritten rather than appended, and the whole file is regenerated only when
    it does not exist yet — a config file is the user's, and a program that
    reformats it on every toggle is not one anybody trusts.
    """
    setting = C.SETTINGS_BY_NAME[name]
    rendered = f"{name} = {setting.render(value)}"
    CONFIG_FILE.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
    os.chmod(CONFIG_FILE.parent, 0o700)  # a pre-existing directory keeps 0700 too
    try:
        lines = CONFIG_FILE.read_text(encoding="utf-8").splitlines()
    except FileNotFoundError:
        # Only a *missing* file earns a full regeneration. Any other read
        # failure propagates: quietly replacing an existing file the user
        # customised with rendered defaults would lose every other setting.
        lines = C.render_settings().splitlines()
    for index, line in enumerate(lines):
        if line.split("#", 1)[0].partition("=")[0].strip() == name:
            lines[index] = rendered
            break
    else:
        lines.append(rendered)
    # Written beside and renamed over, never truncated in place: the applet
    # re-reads this file on every health tick, and a read landing inside a
    # truncate-then-write window would see an empty file and run one whole
    # check on defaults. The scratch name carries the pid, because the CLI
    # and install.sh write the same file from other processes and a shared
    # scratch let one writer rename the other's half-written file into place.
    # Born 0600 like the config itself: write_text would create it with umask
    # permissions first and fix them a moment later.
    scratch = CONFIG_FILE.with_name(f"{CONFIG_FILE.name}.{os.getpid()}.tmp")
    fd = os.open(scratch, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as fh:
            fh.write("\n".join(lines) + "\n")
        os.replace(scratch, CONFIG_FILE)
    except OSError:
        scratch.unlink(missing_ok=True)  # nothing half-written survives
        raise
    reload_settings()


class Deadline:
    """Kill a child that outlasts `limit`, so a blocking read of it can end.

    A `read()` cannot be given a timeout, so the timeout has to act on the
    process instead: killing it closes the pipe, and the read returns. That
    is the same move Cancel makes, which is why `kill` is passed in rather
    than assumed.

    Module level and injectable for one reason: the failure this exists to
    prevent is a wait that never ends, and that is precisely the failure no
    test can observe by waiting for it. `asuvpn selftest` drives this against
    a real child that would outlive the run.
    """

    def __init__(self, proc, limit, kill):
        self.overdue = threading.Event()
        self._settled = threading.Event()
        self._proc, self._limit, self._kill = proc, limit, kill

    def __enter__(self):
        threading.Thread(target=self._watch, daemon=True).start()
        return self

    def _watch(self):
        if self._settled.wait(timeout=self._limit):
            return  # finished on its own; nothing to do
        self.overdue.set()  # set before the kill, so the waiter sees the cause
        self._kill(self._proc)

    def __exit__(self, *_exception):
        self._settled.set()  # releases the watcher; a no-op once it has fired
        return False


def parse_state_payload(payload):
    """Split a `[helper] STATE ...` payload into its state and its fields.

    Module level so `asuvpn selftest` can exercise it without a display: it is
    the one place a malformed or hostile event could reach the state machine.
    """
    words = payload.split()
    state = words[0] if words else ""
    fields = dict(part.split("=", 1) for part in words if "=" in part)
    return state, fields


def configured_server():
    return SETTINGS["server"] or None


def find_openconnect_sso():
    found = shutil.which("openconnect-sso")
    if found:
        return found
    fallback = Path.home() / ".local" / "bin" / "openconnect-sso"
    return str(fallback) if os.access(fallback, os.X_OK) else None


# openconnect-sso asks for a TOTP secret on the terminal whenever the keyring
# entry is empty, which is every run for a Duo-push setup like ASU's. Launched
# from a desktop icon there is no terminal, so getpass hits EOF and the whole
# sign-in dies. Answering with one blank line means "not required" and lets the
# browser handle the second factor.
#
# The same code path asks for the password first when the keyring has none. A
# blank answer there would be saved as the password and quietly break auto-fill,
# so probe the keyring first and refuse to guess instead.
CREDENTIALS_READY = "ready"
CREDENTIALS_NO_PASSWORD = "no-password"
CREDENTIALS_NONE = "no-credentials"
CREDENTIALS_UNKNOWN = "unknown"
CREDENTIALS_LOCKED = "locked"

_PROBE = """
import keyring
from openconnect_sso.config import APP_NAME, load

config = load()
username = getattr(config.credentials, "username", None) if config.credentials else None
if not username:
    print("no-credentials")
else:
    print("ready" if keyring.get_password(APP_NAME, username) else "no-password")
"""


def sso_interpreter(sso_path):
    """openconnect-sso is a console script; its shebang names its own venv python."""
    try:
        with open(sso_path, "rb") as fh:
            first_line = fh.readline().decode("utf-8", "replace").strip()
    except OSError:
        return None
    if not first_line.startswith("#!"):
        return None
    parts = first_line[2:].strip().split()
    return parts[0] if parts else None


def probe_credentials(sso_path):
    python = sso_interpreter(sso_path)
    if not python:
        return CREDENTIALS_UNKNOWN
    try:
        result = subprocess.run(
            [python, "-c", _PROBE],
            # Never the caller's stdin. Run with --foreground this inherits the
            # terminal, and anything in openconnect-sso's keyring path that
            # decided to prompt would silently eat the user's typing. The same
            # reasoning pinned stdin in the helper's run().
            stdin=subprocess.DEVNULL,
            capture_output=True,
            text=True,
            timeout=20,
        )
    except subprocess.TimeoutExpired:
        # Almost always a locked login keyring blocking on its unlock dialog.
        # Must fail closed: carrying on would feed our blank line to the
        # *password* prompt, and that blank would be saved as the password —
        # the exact outcome this probe exists to prevent.
        return CREDENTIALS_LOCKED
    except (OSError, subprocess.SubprocessError):
        return CREDENTIALS_UNKNOWN
    answer = result.stdout.strip().splitlines()
    if not answer:
        return CREDENTIALS_UNKNOWN
    return {
        "ready": CREDENTIALS_READY,
        "no-password": CREDENTIALS_NO_PASSWORD,
        "no-credentials": CREDENTIALS_NONE,
    }.get(answer[-1].strip(), CREDENTIALS_UNKNOWN)


class StateMachine:
    """The applet's decision core: messages in, state out, nowhere else.

    Separable from the GTK class on purpose — `asuvpn selftest` subclasses
    this with stubbed side effects and drives the very table that ships. The
    concrete class supplies the side effects: log, notify, _set_state,
    _send_helper, _act_start_signin, _kill_auth, _act_start_tunnel,
    autoreconnect_enabled, and the teardown worker threads.
    """

    # What the machine reads and writes. Declared so the boundary is written
    # down — and so the analysers hold every host to it.
    state: str
    detail: str
    server: str
    state_events: bool
    quitting: bool
    demoted_by: str | None
    strikes: dict
    nudge_spent: bool
    failure_notified: bool
    routes_seen: dict
    tunnel_device: str | None
    tunnel_ifindex: int | None
    tunnel_address: str
    tunnel_dns: str
    tunnel_ever_connected: bool
    tunnel_up_since: float | None
    last_failure: str | None
    dropped: bool
    rebuilds: int
    last_nudge: float
    last_autoreconnect: float
    # Values, not bare annotations, unlike their neighbours: pylint only
    # believes a member exists when something assigns it, and these three are
    # only ever read or bumped inside the machine. Their zeros are also truly
    # the state of a machine that has not run — every host re-initializes
    # them anyway.
    helper_generation = 0
    auth_generation = 0
    helper_spoke = False
    helper_proc: subprocess.Popen | None
    teardown_intent: str | None
    reconnect_keep_log: bool

    # The side effects the host must supply. Stubs, not annotations, so an
    # incomplete host fails loudly at the first call.
    def log(self, *_args, **_kwargs):
        raise NotImplementedError

    def notify(self, *_args, **_kwargs):
        raise NotImplementedError

    def _set_state(self, *_args, **_kwargs):
        raise NotImplementedError

    def _send_helper(self, *_args, **_kwargs):
        raise NotImplementedError

    def _act_start_signin(self, *_args, **_kwargs):
        raise NotImplementedError

    def _kill_auth(self, *_args, **_kwargs):
        raise NotImplementedError

    def _act_start_tunnel(self, *_args, **_kwargs):
        raise NotImplementedError

    def autoreconnect_enabled(self, *_args, **_kwargs):
        raise NotImplementedError

    def _disconnect_thread(self, *_args, **_kwargs):
        raise NotImplementedError

    def _reconnect_thread(self, *_args, **_kwargs):
        raise NotImplementedError

    def _cancel_tunnel_thread(self, *_args, **_kwargs):
        raise NotImplementedError

    def _quit_thread(self, *_args, **_kwargs):
        raise NotImplementedError

    # ------------------------------------------------------- state machine
    #
    # Messages in, state out — nowhere else. The user verbs, the sign-in
    # worker, the helper's reader, the watchdog and the teardown workers all
    # INJECT messages; the TRANSITIONS table at the top of the file decides
    # what each message means in the current state; the `_tr_*` handlers and
    # the `_act_*` actions they call are the only writers of `self.state`
    # and the incident bookkeeping.

    def dispatch(self, kind, **data):
        """The one place messages become state. Main loop only.

        A (state, message) pair the table does not list is dropped, and the
        drop is logged — silence is how verdicts landing mid-teardown once
        got to act on a tunnel the user was already closing.
        """
        name = TRANSITIONS.get((self.state, kind)) or TRANSITIONS.get((ANY, kind))
        if name is None:
            self.log(f"[tray] ignoring '{kind}' while {self.state}")
            return False
        getattr(self, f"_tr_{name}")(data)
        return False  # idle-callback friendly: never re-queues

    def post(self, kind, **data):
        """Inject a message from a worker thread; it runs on the main loop."""
        GLib.idle_add(lambda: bool(self.dispatch(kind, **data)))

    # The user verbs are message injections, nothing more.

    def connect(self):
        self.dispatch(MSG_CONNECT)

    def disconnect(self):
        self.dispatch(MSG_DISCONNECT)

    def reconnect(self):
        self.dispatch(MSG_RECONNECT)

    def cancel(self):
        self.dispatch(MSG_CANCEL)

    def quit(self):
        self.dispatch(MSG_QUIT)

    # -- rows: user verbs

    def _tr_start_signin(self, _data):
        # A human is at the keyboard. Whatever the applet still owed a dropped
        # tunnel is theirs to spend again, and nothing may keep retrying
        # behind them -- this row and the one below are the only ways out of
        # FAILED, which is the only state a rebuild is ever owed from.
        self._clear_drop()
        # Always a fresh log: this row serves the user verbs, and the
        # keep-the-log flavor travels only through the reconnect teardown
        # (_tr_teardown_reconnect), never through a connect from cold.
        self._act_start_signin(keep_log=False)

    def _tr_connect_means_reconnect(self, _data):
        # The menu offers Reconnect on a demoted tunnel, and the CLI verb
        # means the same intent — "get me connected" — which for a tunnel
        # that is up but carrying nothing is a teardown and a fresh sign-in.
        # Refusing silently read as "cannot connect" against exactly this
        # state on a live break.
        self._clear_drop()
        self.log("[tray] connect asked for while the tunnel is demoted;"
                 " reconnecting instead")
        self._tr_teardown_reconnect({"keep_log": False})

    def _tr_already_disconnected(self, _data):
        self._clear_drop()
        self._set_state(DISCONNECTED)

    def _tr_teardown_disconnect(self, _data):
        self._clear_drop()
        if self.helper_proc is None or self.helper_proc.poll() is not None:
            self._set_state(DISCONNECTED)
            return
        self.teardown_intent = "disconnect"
        self._set_state(DISCONNECTING)
        self.log("[tray] disconnecting")
        threading.Thread(target=self._disconnect_thread, daemon=True).start()

    def _tr_teardown_reconnect(self, data):
        keep = data.get("keep_log", False)
        if self.helper_proc is None or self.helper_proc.poll() is not None:
            # Nothing to tear down. The bump retires the dead helper's
            # still-queued exit message: the health timer can run before
            # queued idles, and without it that message would paint FAILED
            # over the sign-in this line starts. No state is published in
            # between — the sign-in sets the next one, so a `--wait` polling
            # this verb never samples a momentary DISCONNECTED.
            self.helper_generation += 1
            self._reset_tunnel_state()
            self._act_start_signin(keep_log=keep)
            return
        # Stay busy for the whole tear-down-then-connect sequence. Surfacing
        # a momentary "disconnected" would make `asuvpn reconnect --wait`
        # believe it had finished, and would flicker the panel icon.
        self.teardown_intent = "reconnect"
        self.reconnect_keep_log = keep
        self._set_state(DISCONNECTING, "reconnecting")
        self.log("[tray] reconnecting")
        threading.Thread(target=self._reconnect_thread, daemon=True).start()

    def _tr_cancel_attempt(self, _data):
        # Reachable mid-rebuild, which is the case that needs this: cancelling
        # the sign-in a rebuild started left the count spent, so a later drop
        # got fewer attempts than the next user was ever told about.
        self._clear_drop()
        self.auth_generation += 1  # invalidate any sign-in still in flight
        self.log("[tray] cancelled by user")
        self._kill_auth()
        if self.helper_proc is not None and self.helper_proc.poll() is None:
            # A tunnel may already be coming up, so report it as still
            # closing rather than claiming to be disconnected while
            # openconnect lives.
            self.teardown_intent = "cancel"
            self._set_state(DISCONNECTING)
            threading.Thread(target=self._cancel_tunnel_thread,
                             daemon=True).start()
            return
        self._set_state(DISCONNECTED)

    def _tr_quit_everything(self, _data):
        if self.quitting:
            return  # a second Quit must not start a second teardown
        self.quitting = True
        # Whichever branch we take, an in-flight sign-in has to die with us
        # or it outlives the tray as an orphan with a browser window.
        self.auth_generation += 1
        self._kill_auth()
        if self.helper_proc is not None and self.helper_proc.poll() is None:
            self.teardown_intent = "quit"
            self._set_state(DISCONNECTING)
            self.log("[tray] quitting, closing the tunnel first")
            threading.Thread(target=self._quit_thread, daemon=True).start()
        else:
            Gtk.main_quit()

    # -- rows: sign-in results

    def _tr_start_tunnel(self, data):
        if data["attempt"] != self.auth_generation:
            # Cancelled or superseded while the sign-in was finishing. Going
            # ahead would raise a root password prompt for a connection
            # nobody is waiting for any more.
            self.log("[tray] sign-in finished too late to be used; not connecting")
            return
        self._act_start_tunnel(data["host"], data["cookie"], data["fingerprint"])

    def _tr_signin_failed(self, data):
        if data["attempt"] != self.auth_generation:
            return
        message = data["reason"]
        self.log(f"[tray] {message}")
        # A failed *rebuild* attempt still owes the next one, and the badge's
        # promise has to say so between attempts — the rule is "; rebuilding"
        # while one is owed, and a sign-in that timed out unattended is
        # exactly the moment nobody is watching the log.
        self._set_state(FAILED, message + REBUILDING if self.dropped
                        else message)
        self.notify("Sign-in failed", message)

    # -- rows: what the tunnel reports

    def _tr_tunnel_came_up(self, data):
        self._adopt_tunnel(data["device"], data["address"], data["dns"])
        self._announce_connected(data["address"])

    def _tr_refresh_address(self, data):
        self._adopt_tunnel(data["device"], data["address"], data["dns"])
        address = data["address"]
        if address and self.detail != address:
            self._set_state(CONNECTED, address)

    def _tr_adopt_only(self, data):
        # A reconnect landed mid-demotion: adopt the tunnel — address,
        # device, resolver — but the promotion belongs to the source that
        # demoted. Taking openconnect's word here flapped the badge every
        # two minutes against a tunnel whose routes were gone.
        self._adopt_tunnel(data["device"], data["address"], data["dns"])

    def _tr_link_lost(self, _data):
        self._announce_link_lost()

    def _tr_still_recovering(self, _data):
        # openconnect retries emit attempt-reconnect on every attempt; one
        # announcement was made on the first, and a line per retry would spam
        # both the log and the ignore-trail during a long outage.
        return

    def _tr_remember_device(self, data):
        # Validated on the way in as well as on the way out: this is about
        # to be interpolated into a path under /sys, and "it was checked at
        # the other end" is not a property this side can verify.
        device = data["device"]
        self.tunnel_device = device if C.INTERFACE_RE.match(device) else None

    def _tr_remember_fatal(self, data):
        # The helper stopped before openconnect was started at all, so its
        # own sentence explains the exit status far better than the number.
        self.last_failure = data["sentence"]

    def _tr_remember_problem(self, data):
        self.last_failure = data["sentence"]

    def _tr_helper_warning(self, data):
        # Anything the helper marked WARNING: most seriously that the network
        # did not come back after teardown, but also degraded supervision —
        # the event channel dying, a vpnc-script it could not chain to. The
        # title says warning, not teardown, because not all of them are.
        self.notify("VPN warning", data["sentence"],
                    icon="dialog-warning-symbolic")

    # -- rows: the watchdog's verdicts

    def _tr_weigh_check(self, data):
        source, reason, detail = data["source"], data["reason"], data["detail"]
        if reason is None:
            self._check_clear(source, detail)
            return
        if self.state == RECOVERING:
            # openconnect owns its own recovery; the watchdog stands aside.
            return
        self._check_strike(source, reason, detail)

    def _check_clear(self, source, detail):
        """One source reports clear. Only the demoting source promotes.

        Sources are tracked separately on purpose: a probe failure leaves
        the device and its routes perfectly intact, so letting the device
        check promote the badge back would flap it every twenty seconds
        against a tunnel that is genuinely carrying nothing.
        """
        if self.strikes[source]:
            self.log(f"[tray] tunnel {source} check clear again ({detail})")
        self.strikes[source] = 0
        if self.state == DEMOTED and self.demoted_by == source:
            self.demoted_by = None
            self.nudge_spent = False
            self.failure_notified = False
            self._set_state(CONNECTED, self.tunnel_address)
            self.notify("VPN carrying traffic again", self.server)

    def _check_strike(self, source, reason, detail):
        strikes = max(1, SETTINGS["health-strikes"])
        self.strikes[source] += 1
        self.log(f"[tray] tunnel {source} check {self.strikes[source]}"
                 f"/{strikes}: {reason} ({detail})")
        if self.strikes[source] >= strikes:
            self.demoted_by = source
            self._act_ladder(reason)

    # -- rows: teardown outcomes

    def _tr_helper_exited(self, data):
        status, generation = data["status"], data["generation"]
        if generation != self.helper_generation:
            # An older tunnel finished dying after a new one took over.
            # Saying anything more would drop the handle on the live one.
            self.log(f"[tray] a previous tunnel closed (status {status})")
            return
        self.helper_proc = None
        # Read before the reset wipes it: a DEMOTED tunnel is no longer
        # CONNECTED but certainly was one, and "connection dropped" explains
        # its death far better than a bare exit status does.
        was_connected = (self.state in (CONNECTED, RECOVERING, DEMOTED)
                         or self.tunnel_ever_connected)
        # Read before the reset too, and for the same reason: how long this
        # session lasted is what tells a recovery from a flap.
        up_since = self.tunnel_up_since
        intent = self.teardown_intent
        self._reset_tunnel_state()
        if intent == "reconnect":
            # _tr_teardown_finished takes it from here; stay busy until then.
            self.log("[tray] tunnel closed, signing in again")
            return
        self.teardown_intent = None
        if intent == "cancel":
            self._set_state(DISCONNECTED)
            return
        if intent in ("disconnect", "quit") or self.state in (DISCONNECTING,
                                                              DISCONNECTED):
            self._set_state(DISCONNECTED)
            self.log("[tray] disconnected")
            self.notify("VPN disconnected", self.server)
            return
        if status is None:
            status = -1  # the reader thread died before it could reap
        # 126/127 are pkexec's own codes for "dismissed" and "not permitted",
        # but openconnect can return them too. Only read them that way if the
        # helper never got far enough to say anything.
        if status == 126 and not self.helper_spoke:
            message = "authorization cancelled"
        elif status == 127 and not self.helper_spoke:
            message = "not authorized to run the helper"
        elif self.last_failure:
            message = self.last_failure
        elif was_connected:
            message = "connection dropped"
        else:
            message = f"openconnect exited with status {status}"
        # A session that outlived the gap did not flap: whatever produced it
        # worked, so this death starts its own count. One that died sooner
        # keeps the count it inherited, and that is the whole reason
        # MAX_REBUILDS can bite -- every rebuilt tunnel reaches "connected",
        # so refunding the count there instead left the limit unreachable
        # against exactly the flap it exists for.
        if (up_since is not None
                and time.monotonic() - up_since
                >= SETTINGS["autoreconnect-min-gap"]):
            self.rebuilds = 0
        # The one place a rebuild is armed, and `was_connected` is the whole
        # test: a tunnel that carried traffic and then died is worth
        # rebuilding unasked, and a tunnel that never came up is not -- a
        # rejected cookie, a dismissed authorization or an unmatched
        # certificate pin will all do the same thing the second time, browser
        # window and Duo push included. Every rebuild has to re-earn this the
        # same way, which is what stops a retry becoming a loop.
        #
        # The setting is read here, unlike everywhere else, rather than only
        # where it is acted on: a box ticked hours after a tunnel died should
        # not open a browser window for it. Unticking it after the fact does
        # still stop the retries -- the handler re-checks -- because that is
        # the direction where acting on the newer answer means doing less.
        self.dropped = was_connected and self.autoreconnect_enabled()
        # The detail says what happens next, not only what happened: sitting
        # at a bare "Not connected" while three sign-ins are still owed reads
        # as a dead applet, both on the badge and through `asuvpn status`.
        self._set_state(FAILED, message + REBUILDING if self.dropped
                        else message)
        self.log(f"[tray] {message}")
        self.notify("VPN disconnected", message)
        if self.dropped:
            # Said here, once, rather than from the tick that acts on it:
            # that one runs every health-interval for as long as this lasts.
            self.log("[tray] rebuilding it shortly")
        elif was_connected:
            self.log("[tray] automatic sign-in is off; Connect when ready")

    def _tr_rebuild_dropped(self, _data):
        """Rebuild a tunnel that died on its own, while that is still sane.

        The escalation ladder cannot reach this case and never could: it
        escalates a tunnel that is *up* and useless, and this one is gone --
        no helper to nudge, no device to watch, no check to strike. So
        `autoreconnect` did nothing at all for the most ordinary break there
        is. openconnect gives up after its own reconnect timeout (300s by
        default), which a suspend or a WiFi outage clears easily, and the
        applet then sat at "Not connected" until somebody came back and
        clicked Connect -- with the setting that promises otherwise on.

        Driven by the health tick because FAILED has no events of its own and
        there is no other timer. Three things bound it, and each answers a
        different way this could become a nuisance:

          * `dropped`, armed only by an exit that followed real traffic, so a
            rejected cookie or a dismissed authorization is never retried;
          * `autoreconnect-min-gap`, shared with the ladder, so the two
            automatic sign-ins cannot add up to more than the user allowed;
          * `MAX_REBUILDS`, because a network that is down stays down and
            spacing alone would keep opening browser windows all afternoon.
        """
        if not self.dropped:
            # Nothing is owed. The health tick already tests this before it
            # dispatches, but the invariant belongs to the machine rather
            # than to whoever drives it -- and the give-up branch below is
            # once-per-drop only because this line makes it so.
            return
        if not self.autoreconnect_enabled():
            # Turned off since the drop -- by the menu or by `asuvpn
            # autoreconnect off`, which this row sees identically because the
            # config is re-read on every tick. Disarmed rather than merely
            # skipped: left armed, ticking the box again hours later opened a
            # browser window for a tunnel that had died before lunch, which
            # is the same surprise the arming rule above exists to prevent.
            # Nothing is said here; the exit row that armed it already did.
            self._disarm("")
            return
        if self.rebuilds >= MAX_REBUILDS:
            # Disarmed here, so this is said once and the tick stops asking.
            self._disarm("; gave up reconnecting")
            self.log(f"[tray] the tunnel did not come back in {MAX_REBUILDS}"
                     " attempts; waiting to be asked")
            self.notify("VPN still not connected",
                        "automatic sign-in gave up — Connect when ready",
                        icon="dialog-warning-symbolic")
            return
        now = time.monotonic()
        if now - self.last_autoreconnect < SETTINGS["autoreconnect-min-gap"]:
            # Deliberately silent: this row runs every health tick, and the
            # ladder's habit of logging what it is waiting for would put a
            # line in the log every twenty seconds for the whole gap.
            return
        self.rebuilds += 1
        self.last_autoreconnect = now
        self.log(f"[tray] rebuilding the dropped tunnel"
                 f" (attempt {self.rebuilds} of {MAX_REBUILDS})")
        # Announced for the same reason the ladder announces its sign-in: a
        # Duo push and a password dialog are about to arrive, and the user
        # should know what asked for them.
        self.notify("VPN signing in again", "the tunnel dropped")
        self._act_start_signin(keep_log=True)

    def _tr_teardown_finished(self, _data):
        keep = self.reconnect_keep_log
        self.reconnect_keep_log = False
        self.teardown_intent = None
        self.helper_proc = None
        # Straight into the sign-in, publishing no state of its own: the badge
        # goes disconnecting → authenticating. A momentary DISCONNECTED here
        # was readable by the IPC thread, and `asuvpn reconnect --wait`
        # sampling that instant reported the reconnect finished.
        self._act_start_signin(keep_log=keep)

    def _tr_teardown_timed_out(self, data):
        self.teardown_intent = None
        if data["during"] == "reconnect":
            # Carrying on would null out our only handle on a live root
            # tunnel and then start a second one beside it.
            self.reconnect_keep_log = False
            self._set_state(FAILED, "old tunnel would not close; not reconnecting")
            self.log("[tray] reconnect abandoned: the previous tunnel is still up")
            return
        self._set_state(FAILED, "helper did not exit — the tunnel may still be up")
        self.notify(
            "VPN warning",
            "The helper did not exit. Check 'asuvpn log'.",
            icon="dialog-warning-symbolic",
        )

    def _act_ladder(self, reason):
        """Escalate, cheapest first.

        The two recoveries cost wildly different amounts, so they are not
        interchangeable. Asking openconnect to re-establish reuses the session
        cookie: no sign-in, no Duo push, no polkit prompt, and the user need not
        even be at the keyboard. A tray-level reconnect throws the session away
        and starts again from the browser, which means a Duo approval and typing
        a password into a polkit dialog. So the free one is always tried first,
        and the expensive one only when it was asked for.
        """
        # Notified once per incident, not once per check: this is reached
        # repeatedly while the tunnel stays broken, and a desktop notification
        # every forty seconds is its own fault.
        first = self.state != DEMOTED
        self.log(f"[tray] the tunnel is not usable: {reason}")
        self.strikes = fresh_strikes()
        # demoted_by was set by the caller immediately before this; it is the
        # source whose verdict got us here, and different sources mean
        # different things by "not usable".
        headline = DEMOTION_TEXT.get(self.demoted_by, "not usable")
        self._set_state(DEMOTED, f"{headline} — {reason}")

        now = time.monotonic()
        # One nudge per incident. An incident no longer ends at the reconnect
        # event the nudge produces (see _adopt_tunnel), so "already nudged and
        # still demoted" is the evidence the nudge did not fix this — asking
        # again looped at the rate limit forever against a routes-lost break,
        # while the sign-in branch below was never reached. The timestamp still
        # guards incidents arriving back to back.
        if (not self.nudge_spent
                and now - self.last_nudge >= SETTINGS["nudge-min-gap"]
                and self._send_helper(C.CONTROL_RECONNECT)):
            self.nudge_spent = True
            self.last_nudge = now
            self.log("[tray] asked openconnect to re-establish"
                     " (same session, no sign-in needed)")
            if first:
                self.notify("VPN reconnecting", reason)
            return

        # The nudge was not sent. Said once per incident, at the moment that
        # becomes true — which may be cycles after the demotion, when a nudge
        # that looked promising did not stick.
        if not self.failure_notified:
            self.failure_notified = True
            self.notify(f"VPN {DEMOTION_TEXT.get(self.demoted_by, 'not usable')}",
                        reason, icon="dialog-warning-symbolic")

        # From here on, every path logs the decision it took and why. A log
        # that shows verdicts but no decisions reads as a hang: on 2026-08-23
        # the cycles after a spent nudge logged their strikes and then went
        # quiet, and what the watchdog was waiting for was anyone's guess.
        if self.nudge_spent:
            self.log("[tray] the free re-establish was already tried for this"
                     " incident and traffic did not return")
        elif now - self.last_nudge < SETTINGS["nudge-min-gap"]:
            # Held, not spent: the free option has not been tried for this
            # incident. Falling through here once bought with a Duo push what
            # waiting under two minutes would have gotten for free — so wait;
            # a later cycle lands past the gap and nudges.
            wait = int(SETTINGS["nudge-min-gap"] - (now - self.last_nudge))
            self.log(f"[tray] holding the free re-establish for another {wait}s"
                     " (nudge-min-gap)")
            return
        else:
            self.log("[tray] could not ask openconnect to re-establish;"
                     " the helper is not listening")

        # The free option is spent, or there is no one to ask: only now may
        # the expensive path be considered.
        if not self.autoreconnect_enabled():
            self.log("[tray] automatic sign-in is off; staying demoted until a"
                     " check clears — Reconnect or Disconnect when ready")
            return
        remaining = SETTINGS["autoreconnect-min-gap"] - (now - self.last_autoreconnect)
        if remaining > 0:
            self.log(f"[tray] automatic sign-in allowed again in"
                     f" {int(remaining)}s (autoreconnect-min-gap)")
            return
        self.last_autoreconnect = now
        self.log("[tray] signing in again to rebuild the tunnel")
        # Announced, because it is about to raise a Duo push and a password
        # dialog and the user should know what asked for them.
        self.notify("VPN signing in again",
                    "the tunnel could not be recovered without it")
        self._tr_teardown_reconnect({"keep_log": True})

    def _announce_connected(self, address):
        """The single place a tunnel becomes usable, whichever source said so.

        Both the script contract and the log-matching fallback land here, so the
        badge, the wording and the notification cannot drift apart between them
        — which they had, each building its own message.
        """
        first = not self.tunnel_ever_connected
        self.tunnel_ever_connected = True
        # There is a tunnel again, so nothing is owed. The *count* is
        # deliberately not cleared here: every rebuild reaches this line,
        # including the ones whose tunnel falls over seconds later, so
        # clearing it here made MAX_REBUILDS unreachable against exactly the
        # flap it exists for -- connect, die, connect, die, one Duo push per
        # gap forever. What earns the count back is a session that lasted,
        # judged on the way out in _tr_helper_exited.
        self.dropped = False
        self.tunnel_up_since = time.monotonic()
        # Reaching CONNECTED clears any earlier failure line; otherwise a blip
        # openconnect recovered from would be reported as the cause when the
        # tunnel eventually drops, hours later.
        self.last_failure = None
        self._set_state(CONNECTED, address)
        where = f"{self.server} — {address}".strip(" —")
        self.notify("VPN connected" if first else "VPN reconnected", where)

    def _announce_link_lost(self):
        """openconnect has lost the link and is re-establishing it itself.

        Worth saying out loud. Nothing else tells the user, a badge changing to
        a spinner is easy to miss, and the reassuring half — that this costs no
        sign-in and no password — is not something a spinner can convey.
        """
        self._set_state(RECOVERING, "link lost, retrying")
        self.notify("VPN connection lost",
                    f"{self.server} — reconnecting, no sign-in needed",
                    icon="dialog-warning-symbolic")

    def _apply_state_event(self, payload):
        """Translate a script-contract event into a message; decide nothing.

        The wire words are the contract's STATE_* constants — they were bare
        literals here once, restating what REASON_STATES already said. This
        used to act on state directly, carrying its own private copy of the
        teardown guard — the transition table owns every such decision now,
        so this watcher only reports what it saw.
        """
        state, fields = parse_state_payload(payload)
        if state not in (C.STATE_CONNECTED, C.STATE_CONNECTING,
                         C.STATE_DISCONNECTED):
            return
        self.state_events = True  # the log-matching fallback is now redundant
        if state == C.STATE_CONNECTED:
            self.dispatch(MSG_TUNNEL_UP,
                          device=fields.get("dev") or self.tunnel_device,
                          address=fields.get("addr", ""),
                          dns=fields.get("dns", ""))
        elif state == C.STATE_CONNECTING:
            self.dispatch(MSG_LINK_LOST)
        # A "disconnected" event is deliberately not a message: the exit path
        # already distinguishes a requested teardown from a dropped tunnel.

    def _disarm(self, ending):
        """Stop owing a rebuild, and stop the badge promising one.

        The detail keeps the failure that caused all this -- outside the log
        it is the only record of *why*, and it reaches the panel and
        `asuvpn status` alike -- so only the trailing promise is rewritten.
        Published rather than merely assigned, because two of the three ways
        a rebuild is called off change no state of their own, and a menu
        still offering to stop something that stopped is its own bug.
        """
        self.dropped = False
        self._set_state(FAILED, self.detail.removesuffix(REBUILDING) + ending)

    def _clear_drop(self):
        """Nothing is owed a dropped tunnel: a human took over.

        Called from the handlers a user verb is the *only* way to reach, so
        that acting on the applet always hands the decision back. The one
        user verb missing from that list is Reconnect: `_tr_teardown_reconnect`
        is shared with the escalation ladder, and clearing the count from
        there would let a demotion sign-in refund the drop budget every time
        round. Nothing is lost -- a session that outlives the gap refunds it
        anyway, judged on the way out.

        A tunnel merely *coming up* clears the flag but not the count; see
        _announce_connected, where getting that wrong made the limit
        unreachable against the flap it exists for.
        """
        self.dropped = False
        self.rebuilds = 0

    def _clear_incident(self):
        """A clean slate for the watchdog: no strikes, no demoter, no latch.

        One place, because this five-field reset already exists in two
        callers and a third divergent copy is how the tunnel-state reset
        went wrong once (see _reset_tunnel_state).
        """
        self.strikes = fresh_strikes()
        self.demoted_by = None
        self.nudge_spent = False
        self.failure_notified = False
        self.routes_seen = {"routes4": False, "routes6": False}

    def _adopt_tunnel(self, device, address, dns=""):
        """Record what the watchdog should be watching, once it is up."""
        if address:
            # An addr-less source (a fallback line like "CSTP connected"
            # names no address) must not blank one a real event delivered.
            self.tunnel_address = address
        # The probe target, taken from what this tunnel says to resolve
        # against. Nothing is hardcoded: a VPN that pushes no resolver simply
        # gets the device checks, which need no target at all.
        if dns:
            self.tunnel_dns = dns
        if self.state != DEMOTED:
            # A fresh adoption starts with a clean slate. One arriving *during*
            # a demotion is the reconnect a nudge produced, and openconnect
            # asserting "connected" is exactly what it asserted before the
            # watchdog found otherwise — so the slate stays dirty until the
            # demoting source itself passes (_check_clear does the promoting).
            # Clearing it here let a reconnect that fixed nothing restart the
            # incident every two minutes: badge flap, repeated notifications,
            # and an escalation that could never be reached.
            self._clear_incident()
        if not device or not C.INTERFACE_RE.match(device):
            return
        self.tunnel_device = device
        # Taken now, while openconnect has just created it.
        index = C.interface_index(device)
        if index is not None and index != self.tunnel_ifindex:
            # A different incarnation of the device. Which route families the
            # old one carried says nothing about this one, and a stale latch
            # here would hold a demotion forever against a working tunnel
            # whose new session simply has no routes of that family.
            self.routes_seen = {"routes4": False, "routes6": False}
        self.tunnel_ifindex = index

    def _reset_tunnel_state(self):
        """Forget everything the watchdog knows about a tunnel.

        One place, because this was open-coded in three and they had already
        diverged: the copy in the tunnel starter omitted tunnel_dns, so a new tunnel
        kept probing the *previous* one's resolver until an event replaced it —
        and if no event ever arrived, indefinitely, against an address that need
        not be reachable through the new tunnel at all.

        The rate-limit timestamps are deliberately not reset here. Every
        automatic recovery passes through this, so clearing them would clear the
        limits that stop recovery becoming a loop.
        """
        self.tunnel_device = None
        self.tunnel_ifindex = None
        self.tunnel_address = ""
        self.tunnel_dns = ""
        self.tunnel_ever_connected = False
        self.tunnel_up_since = None
        self._clear_incident()



class VpnTray(StateMachine):
    def __init__(self, server, extra_openconnect_args):
        self.server = server
        self.extra_args = extra_openconnect_args
        self.state = DISCONNECTED
        self.detail = ""
        self._status = (DISCONNECTED, "")
        self.auth_proc = None
        self.helper_proc = None
        # Set by the thread that reaps the helper. Teardown waits on this rather
        # than calling Popen.wait() a second time, which would contend for
        # CPython's waitpid lock and time out however fast the helper exits.
        self.helper_exited = None
        # Each tunnel gets a generation number. A helper that is still dying
        # must not be able to report exit over the top of its replacement, which
        # would drop the tray's only handle on a live root openconnect.
        self.helper_generation = 0
        # Sign-ins get one too: cancel, quit and every new connect bump it,
        # so a sign-in still in flight can tell nobody wants it any more and
        # cannot start a tunnel on top of a newer one.
        self.auth_generation = 0
        self.helper_spoke = False
        self.state_events = False
        self.last_failure = None
        # What the watchdog watches: the device from the helper's DEVICE line
        # or a state event, the ifindex read once when the tunnel comes up so a
        # device later replaced can be told from ours, the address so a
        # recovered tunnel regains its label, and the resolver to probe.
        self._reset_tunnel_state()
        self.probe_cycle = 0
        self.probe_in_flight = False
        self.dns_check_in_flight = False
        self._setting_problems = []
        self._health_interval = SETTINGS["health-interval"]
        # -inf, not 0: time.monotonic() counts from boot, so a zero here would
        # block the first nudge for the first nudge-min-gap seconds of uptime —
        # exactly when a tunnel brought up at login is most likely to need one.
        self.last_nudge = float("-inf")
        self.last_autoreconnect = float("-inf")
        # Whether the last tunnel died on its own, and how many times it has
        # been rebuilt for that death. Outside _reset_tunnel_state on purpose
        # -- see _clear_drop.
        self.dropped = False
        self.rebuilds = 0
        # Why the current teardown is happening ("disconnect", "cancel",
        # "reconnect", "quit"); the helper-exited rows read it.
        self.teardown_intent = None
        self.reconnect_keep_log = False
        self.quitting = False
        self._log_bytes = 0
        self._log_lock = threading.Lock()
        self._stdin_lock = threading.Lock()
        self._auth_lock = threading.Lock()
        self.log_lines = []
        self.log_window = None
        self.log_buffer = None
        self.log_view = None

        try:
            # 0700/0600: the log holds the address the VPN assigned, the routes
            # it installed and the DNS it used. Default umask made it 0644.
            LOG_FILE.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
            os.chmod(LOG_FILE.parent, 0o700)
            if LOG_FILE.exists():
                os.chmod(LOG_FILE, 0o600)
                # Count from the file's real size, not from zero: an applet
                # restarted over a large inherited log would otherwise let it
                # grow a whole extra cap before the first rotation.
                self._log_bytes = LOG_FILE.stat().st_size
        except OSError:
            pass  # an unwritable cache dir costs the log, not the applet

        # Which build wrote this log: the applet announces itself once, at
        # startup. (Connecting rotates the log, so on a long-lived applet
        # this line lives in the rotated generations rather than the
        # current file — `asuvpn --version` asks the running code directly.)
        self.log(f"[tray] {APP_NAME} {C.VERSION}"
                 f" (contract {C.CONTRACT_VERSION}) starting")

        Notify.init(APP_NAME)
        self.indicator = AppIndicator.Indicator.new(
            APP_ID, ICONS[DISCONNECTED], AppIndicator.IndicatorCategory.SYSTEM_SERVICES
        )
        self.indicator.set_status(AppIndicator.IndicatorStatus.ACTIVE)
        self.indicator.set_title(APP_NAME)
        self._build_menu()
        self._refresh()
        # On the main loop, not a thread: every check is a handful of file reads
        # and every outcome changes state, which may only happen here anyway.
        self._schedule_health_check()

    # ---------------------------------------------------------------- menu

    def _build_menu(self):
        self.menu = Gtk.Menu()

        self.status_item = Gtk.MenuItem(label="")
        self.status_item.set_sensitive(False)
        self.menu.append(self.status_item)

        # Kept as an attribute: while disconnecting there is no verb to offer,
        # and a separator that stays put then sits directly against the next
        # one -- two rules with nothing between them, which reads as a menu
        # that lost its items rather than one that has none to give.
        self.verb_separator = Gtk.SeparatorMenuItem()
        self.verb_separator.set_no_show_all(True)
        self.menu.append(self.verb_separator)

        self.connect_item = Gtk.MenuItem(label="Connect")
        self.connect_item.connect("activate", lambda _w: self.connect())
        self.menu.append(self.connect_item)

        self.disconnect_item = Gtk.MenuItem(label="Disconnect")
        self.disconnect_item.connect("activate", lambda _w: self.disconnect())
        self.menu.append(self.disconnect_item)

        self.reconnect_item = Gtk.MenuItem(label="Reconnect")
        self.reconnect_item.connect("activate", lambda _w: self.reconnect())
        self.menu.append(self.reconnect_item)

        self.cancel_item = Gtk.MenuItem(label="Cancel")
        self.cancel_item.connect("activate", lambda _w: self.cancel())
        self.menu.append(self.cancel_item)

        # These four swap according to state, so keep show_all() from reviving them.
        for item in (
            self.connect_item,
            self.disconnect_item,
            self.reconnect_item,
            self.cancel_item,
        ):
            item.set_no_show_all(True)

        self.menu.append(Gtk.SeparatorMenuItem())

        log_item = Gtk.MenuItem(label="Show log…")
        log_item.connect("activate", lambda _w: self.show_log_window())
        self.menu.append(log_item)

        # Named for what it does: puts the applet in the tray at login without
        # firing off a Duo push before anyone has asked for one. A local:
        # unlike autoreconnect_item, nothing re-reads it after this menu is
        # built — the toggled handler receives it as an argument.
        autostart_item = Gtk.CheckMenuItem(label="Start on login (applet only)")
        autostart_item.set_active(AUTOSTART_FILE.exists())
        autostart_item.connect("toggled", self._on_autostart_toggled)
        self.menu.append(autostart_item)

        # On by default, and it gates the two automatic sign-ins: the
        # ladder's last rung, which nothing reaches until the free
        # re-establish has been tried on a tunnel that is up and carrying
        # nothing, and the rebuild of a tunnel that died outright. Both spend
        # one autoreconnect-min-gap budget. Unticking it stops at the
        # demotion, or at "Not connected", and waits to be asked.
        self.autoreconnect_item = Gtk.CheckMenuItem(
            label="Reconnect automatically if traffic stops")
        self.autoreconnect_item.set_active(SETTINGS["autoreconnect"])
        self.autoreconnect_item.connect("toggled", self._on_autoreconnect_toggled)
        self.menu.append(self.autoreconnect_item)

        self.menu.append(Gtk.SeparatorMenuItem())

        quit_item = Gtk.MenuItem(label="Quit")
        quit_item.connect("activate", lambda _w: self.quit())
        self.menu.append(quit_item)

        # Which build answered — support's first question, so it is one
        # glance away rather than a terminal away. Insensitive, like the
        # status line; the same string `asuvpn --version` prints.
        version_item = Gtk.MenuItem(label=f"{APP_NAME} {C.VERSION}")
        version_item.set_sensitive(False)
        self.menu.append(version_item)

        self.menu.show_all()
        self.indicator.set_menu(self.menu)

    def status_line(self):
        """One tab-separated line, for the command line client.

        Read from a single tuple so a caller on the IPC thread cannot catch a
        new state paired with the previous detail.
        """
        state, detail = self._status
        return f"{state}\t{self.server}\t{detail}"

    def _refresh(self):
        text = STATE_LABELS[self.state]
        if self.detail:
            text = f"{text} — {self.detail}"
        self.status_item.set_label(f"{self.server}: {text}")
        self.indicator.set_icon_full(ICONS[self.state], text)
        # Hosts switch to the attention icon in ATTENTION status; without this the
        # panel would show a blank slot exactly when something went wrong.
        self.indicator.set_attention_icon_full(ICONS[self.state], text)
        self.indicator.set_status(
            AppIndicator.IndicatorStatus.ATTENTION
            if self.state == FAILED
            else AppIndicator.IndicatorStatus.ACTIVE
        )

        self.connect_item.set_visible(self.state in (DISCONNECTED, FAILED))
        # A DEMOTED tunnel is still a tunnel: it has a helper and a device,
        # and the two things a user wants at that moment are exactly
        # Disconnect and Reconnect — Cancel is the wrong word for tearing
        # down an established session.
        established = self.state in (CONNECTED, DEMOTED)
        # Offered in FAILED too, while a rebuild is still owed. The verb that
        # stops one already existed and already did the right thing --
        # (FAILED, disconnect) clears the budget and settles on disconnected,
        # which is how `asuvpn disconnect` has always stopped it -- the menu
        # simply never showed it, so the only way to call off a retry from
        # the panel was to untick a setting the user may well want kept.
        armed = self.state == FAILED and self.dropped
        self.disconnect_item.set_visible(established or armed)
        # Named for what it does there: "Disconnect" is the wrong word for a
        # tunnel that is already gone, and what is being stopped is the
        # sign-in that has not happened yet.
        self.disconnect_item.set_label("Stop reconnecting" if armed
                                       else "Disconnect")
        self.reconnect_item.set_visible(established)
        # Not while disconnecting: teardown must not be interrupted, and there
        # would be nothing left to cancel anyway.
        self.cancel_item.set_visible(
            self.state in (AUTHENTICATING, CONNECTING, RECOVERING))
        # Follows the verbs it introduces. DISCONNECTING offers none of them --
        # teardown must not be interrupted -- so the rule above them goes too.
        self.verb_separator.set_visible(any(
            item.get_visible() for item in (
                self.connect_item, self.disconnect_item,
                self.reconnect_item, self.cancel_item)))

    def _set_state(self, state, detail=""):
        self.state = state
        self.detail = detail
        self._status = (state, detail)  # published as one object, see status_line
        self._refresh()

    # ----------------------------------------------------------------- log

    def log(self, line):
        """Safe to call from any thread."""
        # openconnect-sso renders its structured logs with colour codes.
        line = f"{time.strftime('%H:%M:%S')}  {ANSI_ESCAPE.sub('', line).rstrip()}"
        with self._log_lock:
            self.log_lines.append(line)
            del self.log_lines[:-MAX_LOG_LINES]
            limit = SETTINGS["log-max-kb"]
            if limit and self._log_bytes > limit * 1024:
                # A session can stay up for weeks; roll the file over rather
                # than growing it forever. The overflow survives as .1.
                self._rotate_log_locked()
            try:
                # O_CREAT with 0600, because a plain append would create a
                # missing file with umask permissions — and the log holds the
                # assigned address, the routes and the DNS. An existing file
                # keeps its mode.
                fd = os.open(LOG_FILE,
                             os.O_WRONLY | os.O_APPEND | os.O_CREAT, 0o600)
                with os.fdopen(fd, "a") as fh:
                    fh.write(line + "\n")
                # Bytes, not characters: the cap and the rotation threshold
                # are file sizes, and this file is UTF-8 with multi-byte
                # dashes in the applet's own wording.
                self._log_bytes += len(line.encode("utf-8")) + 1
            except OSError:
                pass
        # The window is GTK, so it may only be touched from the main loop.
        GLib.idle_add(self._append_to_log_window, line)

    def _rotate_log_locked(self):
        """Rotate session.log → .1 → .2 …; the caller holds _log_lock.

        `log-keep` rotated files survive, the newest as .1. Files numbered
        beyond the current setting are pruned first, so lowering it takes
        effect at the next rotation; 0 keeps none, and rotation degenerates to
        the truncation it replaced.
        """
        keep = SETTINGS["log-keep"]
        try:
            for old in LOG_FILE.parent.glob(LOG_FILE.name + ".*"):
                suffix = old.name[len(LOG_FILE.name) + 1:]
                if suffix.isdigit() and int(suffix) > keep:
                    old.unlink(missing_ok=True)
            if keep > 0 and LOG_FILE.exists():
                for index in range(keep - 1, 0, -1):
                    older = LOG_FILE.with_name(f"{LOG_FILE.name}.{index}")
                    if older.exists():
                        older.replace(
                            LOG_FILE.with_name(f"{LOG_FILE.name}.{index + 1}"))
                LOG_FILE.replace(LOG_FILE.with_name(f"{LOG_FILE.name}.1"))
            # Born 0600 like every log file this applet creates: write_text
            # would create it with umask permissions first and fix them a
            # moment later, giving the one file two creation modes.
            os.close(os.open(LOG_FILE,
                             os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600))
            os.chmod(LOG_FILE, 0o600)  # an inherited pre-0600 file, truncated
        except OSError:
            pass  # an unwritable cache dir costs the log, not the applet
        # Re-measured rather than assumed zero: a failed rotation leaves the
        # old content in place, and a meter pretending otherwise would defer
        # the next attempt by a whole extra cap.
        try:
            self._log_bytes = LOG_FILE.stat().st_size
        except OSError:
            self._log_bytes = 0

    def _append_to_log_window(self, line):
        if self.log_buffer is not None:
            self.log_buffer.insert(self.log_buffer.get_end_iter(), line + "\n")
            self._scroll_log_to_end()
        return False

    def _scroll_log_to_end(self):
        if self.log_view is None or self.log_buffer is None:
            return
        end = self.log_buffer.get_end_iter()
        self.log_view.scroll_to_iter(end, 0.0, False, 0.0, 0.0)

    def show_log_window(self):
        if self.log_window is not None:
            self.log_window.present()
            return

        window = Gtk.Window(title=f"{APP_NAME} log")
        window.set_default_size(760, 460)
        window.set_icon_name(ICONS[CONNECTED])

        with self._log_lock:
            backlog = list(self.log_lines)
        self.log_buffer = Gtk.TextBuffer()
        self.log_buffer.set_text("\n".join(backlog) + ("\n" if backlog else ""))
        self.log_view = Gtk.TextView(buffer=self.log_buffer)
        self.log_view.set_editable(False)
        self.log_view.set_monospace(True)
        self.log_view.set_wrap_mode(Gtk.WrapMode.WORD_CHAR)

        scroller = Gtk.ScrolledWindow()
        scroller.set_policy(Gtk.PolicyType.AUTOMATIC, Gtk.PolicyType.AUTOMATIC)
        scroller.add(self.log_view)

        copy_button = Gtk.Button(label="Copy")
        copy_button.connect("clicked", self._on_copy_log)
        close_button = Gtk.Button(label="Close")
        close_button.connect("clicked", lambda _w: window.destroy())

        buttons = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=6)
        buttons.set_halign(Gtk.Align.END)
        buttons.set_margin_top(6)
        buttons.add(copy_button)
        buttons.add(close_button)

        box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=0)
        box.set_margin_top(10)
        box.set_margin_bottom(10)
        box.set_margin_start(10)
        box.set_margin_end(10)
        box.pack_start(scroller, True, True, 0)
        box.pack_start(buttons, False, False, 0)
        window.add(box)

        def on_destroy(_w):
            self.log_window = None
            self.log_buffer = None
            self.log_view = None

        window.connect("destroy", on_destroy)
        self.log_window = window
        window.show_all()
        self._scroll_log_to_end()

    def _on_copy_log(self, _widget):
        with self._log_lock:
            text = "\n".join(self.log_lines)
        clipboard = Gtk.Clipboard.get(Gdk.SELECTION_CLIPBOARD)
        clipboard.set_text(text, -1)

    def notify(self, summary, body="", icon=None):
        try:
            note = Notify.Notification.new(summary, body, icon or ICONS[self.state])
            note.show()
        except Exception:
            pass

    # ------------------------------------------------------------- actions

    def _act_start_signin(self, keep_log):
        """Sign in and bring the tunnel up.

        `keep_log` is the automatic-recovery flavor. Truncating on connect is
        right when a person asked for one — a fresh session, a fresh log. It
        is wrong when the applet reconnected by itself, because the lines
        explaining *why* are the only record of a breakage that was silent by
        definition.
        """
        # While a rebuild is owed, even a sign-in that cannot start keeps the
        # badge's trailing promise: the next attempt is still coming, and
        # dropping the suffix here read as the applet having given up.
        owed = REBUILDING if self.dropped else ""
        sso = find_openconnect_sso()
        if not sso:
            self._set_state(FAILED, "openconnect-sso not found" + owed)
            self.log("[tray] cannot find openconnect-sso on PATH or in ~/.local/bin")
            self.notify("Cannot start", "openconnect-sso is not installed.")
            return
        if not os.access(HELPER, os.X_OK):
            self._set_state(FAILED, "helper not executable" + owed)
            self.log(f"[tray] helper is missing or not executable: {HELPER}")
            return

        if not keep_log:
            with self._log_lock:
                # A fresh session starts a fresh log, but the previous
                # session's survives the turnover as session.log.1 — wiping it
                # here threw away the only record of whatever ended it.
                self._rotate_log_locked()
                self.log_lines.clear()
            if self.log_buffer is not None:
                self.log_buffer.set_text("")
        else:
            self.log("[tray] ---- reconnecting; the lines above say why ----")

        self.auth_generation += 1
        self._set_state(AUTHENTICATING, "waiting for ASU sign-in")
        self.log(f"[tray] authenticating to {self.server} via {sso}")
        threading.Thread(
            target=self._auth_thread, args=(sso, self.auth_generation), daemon=True
        ).start()

    def _reconnect_thread(self):
        if not self._teardown_helper():
            self.post(MSG_TEARDOWN_TIMEOUT, during="reconnect")
            return
        self.post(MSG_TEARDOWN_FINISHED)

    def _cancel_tunnel_thread(self):
        proc = self.helper_proc
        if proc is None:
            return
        # While the polkit dialog is still up, pkexec is an ordinary process of
        # ours and terminating it dismisses the dialog. Once it has become root
        # this fails with EPERM, and closing the control pipe is what stops it.
        try:
            proc.terminate()
        except (ProcessLookupError, PermissionError, OSError):
            pass
        if not self._teardown_helper():
            self.post(MSG_TEARDOWN_TIMEOUT, during="cancel")

    def _kill_auth(self):
        """Stop the sign-in subprocess, reading the handle exactly once."""
        with self._auth_lock:
            proc = self.auth_proc
            self.auth_proc = None
        if proc is not None and proc.poll() is None:
            self._kill_group(proc)

    def _quit_thread(self):
        # Same first move as cancel(): while the polkit dialog is still up,
        # pkexec is an ordinary process of ours and terminating it dismisses the
        # prompt. Without this, quitting mid-dialog left it orphaned on screen
        # and stalled the whole teardown budget waiting on a pipe nobody reads.
        # Unconditional, like cancel(). Testing self.state here was dead code:
        # quit() sets DISCONNECTING before starting this thread, so the test
        # never fired and a polkit dialog was left orphaned on screen while
        # teardown burned the full timeout on a pipe pkexec never reads.
        # Once pkexec has become root this fails with EPERM, which is fine.
        proc = self.helper_proc
        if proc is not None:
            try:
                proc.terminate()
            except (ProcessLookupError, PermissionError, OSError):
                pass
        # Quit regardless: refusing to exit because teardown stalled would leave
        # the user with an applet they cannot close either.
        if not self._teardown_helper():
            self.log("[tray] quitting anyway; the tunnel may still be up")
        GLib.idle_add(Gtk.main_quit)

    # ------------------------------------------------------- phase 1: auth

    def _auth_thread(self, sso, attempt):
        def superseded():
            # The generation is bumped by cancel, quit and every new connect,
            # so it alone says whether anyone still wants this sign-in.
            return attempt != self.auth_generation

        credentials = probe_credentials(sso)

        # The probe can take seconds, and until Popen below there is no process
        # for cancel() to kill. Checked before reporting anything, so a
        # superseded sign-in cannot overwrite a newer state with its verdict.
        if superseded():
            return

        blocked = {
            CREDENTIALS_NO_PASSWORD: (
                "no saved password — run: openconnect-sso --server "
                f"{self.server} --authenticate=shell"
            ),
            CREDENTIALS_LOCKED: "keyring did not answer — unlock it and try again",
        }.get(credentials)
        if blocked:
            self.post(MSG_AUTH_FAILED, reason=blocked, attempt=attempt)
            return
        if credentials == CREDENTIALS_NONE:
            self.log("[tray] no saved username; the browser will ask for everything")

        command = [
            sso,
            "--server",
            self.server,
            "--authenticate=shell",
        ]
        try:
            # Bound to a local and used only through it. Reading self.auth_proc
            # back on each line would let a second sign-in alias this one: two
            # threads waiting on one child, one closing the other's pipes, and a
            # live browser process nothing can kill.
            proc = subprocess.Popen(
                command,
                stdin=subprocess.PIPE,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True,
                errors="replace",
                start_new_session=True,
            )
        except OSError as exc:
            self.post(MSG_AUTH_FAILED,
                      reason=f"could not start openconnect-sso: {exc}",
                      attempt=attempt)
            return

        # Publish it only after a re-check, so a cancel that landed during the
        # spawn cannot leave an unkillable sign-in behind.
        with self._auth_lock:
            doomed = superseded()
            if not doomed:
                self.auth_proc = proc
        if doomed:
            self._kill_group(proc)
            # And reap it: nothing else ever wait()s on this generation, and
            # a kill without a wait keeps a zombie for the applet's lifetime.
            try:
                proc.wait(timeout=5)
            except subprocess.TimeoutExpired:
                pass
            return

        stderr_thread = threading.Thread(
            target=self._pump, args=(proc.stderr, "[sso] "), daemon=True
        )
        stderr_thread.start()

        # Blank answer to the "TOTP secret (leave blank if not required)" prompt.
        try:
            proc.stdin.write("\n")
            proc.stdin.flush()
            proc.stdin.close()
        except (OSError, ValueError):
            pass

        # A sign-in that has not finished by now is not going to, and it has
        # to end rather than merely be waited on. openconnect-sso opens a
        # browser window and blocks on a Duo approval; with nobody at the
        # keyboard -- which is exactly the situation an automatic rebuild runs
        # in -- it blocks for as long as the applet lives. The badge sits at
        # "Signing in…", the watchdog does not run in that state, and only
        # Cancel clears it.
        limit = SETTINGS["signin-timeout"]
        # Read stdout inside it rather than via communicate(), which would
        # fight the stderr pump thread for the same pipes.
        with Deadline(proc, limit, self._kill_group) as clock:
            try:
                stdout = proc.stdout.read()
            except (OSError, ValueError):
                stdout = ""
            returncode = proc.wait()
        stderr_thread.join(timeout=2)
        for stream in (proc.stdout, proc.stderr):
            try:
                stream.close()
            except (OSError, ValueError):
                pass
        with self._auth_lock:
            if self.auth_proc is proc:
                self.auth_proc = None

        if superseded():
            return
        if clock.overdue.is_set():
            # Named for what happened, not for the signal it died of: the
            # exit status of a process we killed says nothing useful.
            self.post(MSG_AUTH_FAILED,
                      reason=f"sign-in did not finish within {limit}s",
                      attempt=attempt)
            return
        if returncode != 0:
            self.post(MSG_AUTH_FAILED,
                      reason=f"sign-in failed (exit {returncode})",
                      attempt=attempt)
            return

        details = {}
        for line in stdout.splitlines():
            if "=" not in line:
                continue
            try:
                unquoted = shlex.split(line)
            except ValueError:
                continue
            if not unquoted:
                continue
            key, _, value = unquoted[0].partition("=")
            details[key.strip().upper()] = value

        if not {"HOST", "COOKIE", "FINGERPRINT"} <= details.keys():
            self.post(MSG_AUTH_FAILED, reason="sign-in returned no session",
                      attempt=attempt)
            return

        self.post(MSG_AUTH_OK, host=details["HOST"], cookie=details["COOKIE"],
                  fingerprint=details["FINGERPRINT"], attempt=attempt)

    # ----------------------------------------------------- phase 2: tunnel

    def _act_start_tunnel(self, host, cookie, fingerprint):
        if self.helper_proc is not None and self.helper_proc.poll() is None:
            # Overwriting helper_proc here would lose the only handle on a live
            # root openconnect, leaving it running with nothing able to stop it.
            # Say so in the UI: returning quietly strands us in "Signing in…".
            self._set_state(FAILED, "a tunnel is already running")
            self.log("[tray] a tunnel is already running; not starting a second")
            return
        self._set_state(CONNECTING, "authorizing")
        self.log(f"[tray] signed in, starting openconnect for {host}")
        command = [
            "pkexec",
            str(HELPER),
            "--host",
            host,
            "--fingerprint",
            fingerprint,
            "--ac-version",
            C.AC_VERSION,
        ]
        command += ["--dpd", str(SETTINGS["dpd"])]
        if SETTINGS["dns"]:
            # Presence is the switch. The value may well be empty -- that only
            # says no fallback domain was configured, and the helper derives
            # one from the server address in that case.
            command += ["--dns-domains", SETTINGS["dns-domains"]]
        if self.extra_args:
            command += ["--", *self.extra_args]

        self.helper_generation += 1
        generation = self.helper_generation
        exited = threading.Event()
        self.helper_spoke = False
        self.state_events = False
        self.last_failure = None
        self._reset_tunnel_state()
        try:
            proc = subprocess.Popen(
                command,
                stdin=subprocess.PIPE,
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,
                text=True,
                # Never strict: one invalid byte from a VPN banner would raise
                # out of the reader thread and strand the state machine.
                errors="replace",
                bufsize=1,
                # Its own session, so a Ctrl+C in the terminal reaches only us.
                # Teardown then has exactly one path: closing the control pipe.
                start_new_session=True,
            )
        except OSError as exc:
            # Nothing was started, so release anyone who might later wait on
            # this generation's exit rather than leaving them to time out.
            with self._stdin_lock:
                self.helper_proc = None
                self.helper_exited = exited
            exited.set()
            self._set_state(FAILED, f"could not run pkexec: {exc}")
            self.log(f"[tray] {exc}")
            return

        # Published as a pair, under the same lock _teardown_helper reads them
        # with. Assigning the event before the process meant a teardown running
        # between the two saw this generation's event beside the *previous*
        # generation's process — closing one pipe and then waiting out the full
        # timeout on an event the other would never set.
        with self._stdin_lock:
            self.helper_proc = proc
            self.helper_exited = exited

        try:
            proc.stdin.write(cookie + "\n")
            proc.stdin.flush()
        except OSError as exc:
            self.log(f"[tray] could not hand the session cookie to the helper: {exc}")

        threading.Thread(
            target=self._tunnel_thread,
            args=(proc, generation, exited),
            daemon=True,
        ).start()

    def _tunnel_thread(self, proc, generation, exited):
        # try/finally around the whole body: if this thread dies without setting
        # `exited`, every teardown burns the full timeout and the tray loses
        # track of a live root tunnel for good.
        returncode = None
        try:
            for raw in iter(proc.stdout.readline, ""):
                GLib.idle_add(self._on_tunnel_output, raw.rstrip("\n"), generation)
            returncode = proc.wait()
        except (OSError, ValueError) as exc:
            GLib.idle_add(self.log, f"[tray] lost the helper's output: {exc}")
            try:
                returncode = proc.wait()
            except OSError:
                returncode = -1
        finally:
            # Posted *before* the event is set, and the order matters. The
            # teardown worker wakes on `exited` and posts teardown-finished;
            # if that beat this message into the idle queue, the exit would be
            # weighed after the reconnect had already begun and a perfectly
            # healthy recovery would be reported as a failure. Setting the
            # event last closes the window.
            self.post(MSG_HELPER_EXITED, status=returncode, generation=generation)
            exited.set()

    # ------------------------------------------------------------- watchdog

    def autoreconnect_enabled(self):
        return SETTINGS["autoreconnect"]

    def _schedule_health_check(self):
        """Arm the next check at whatever the interval currently is.

        Rescheduled each time rather than installed once, so changing
        health-interval in the config takes effect without a restart — and
        setting it to 0 stops the watchdog entirely, which is the only honest
        way to offer an off switch for something that runs on a timer. The
        tick itself survives even then, at the schema's default cadence,
        because it is also the only thing that re-reads the config: without
        it, no setting change — including turning the watchdog back on —
        would ever be noticed again until a restart.
        """
        interval = SETTINGS["health-interval"]
        if interval <= 0:
            interval = C.SETTINGS_BY_NAME["health-interval"].default
        GLib.timeout_add_seconds(interval, self._health_check)

    def _health_check(self):
        """Ask the kernel, and occasionally the network, whether this still works.

        openconnect's dead peer detection covers the far end going away. It
        cannot see a break that is local — a resume, a WiFi change, another
        daemon rewriting the table — and when the server never asked for DPD it
        cannot see anything at all. Two independent sources cover that: what the
        kernel says about the device and its routes, which is free, and whether
        anything actually answers through the tunnel, which is not.
        """
        if self.quitting:
            return False
        # Picked up here so a config change applies to the next check rather
        # than the next connect. Problems are reported once, when they change.
        problems = reload_settings()
        if problems != self._setting_problems:
            self._setting_problems = problems
            for problem in problems:
                self.log(f"[config] {problem}")
        # The CLI writes the same file, so the checkbox has to follow it or the
        # menu shows the state from whenever it was built. Blocked while set, or
        # the toggled handler would write the value straight back and log a
        # change nobody made.
        item = self.autoreconnect_item
        if item.get_active() != SETTINGS["autoreconnect"]:
            item.handler_block_by_func(self._on_autoreconnect_toggled)
            item.set_active(SETTINGS["autoreconnect"])
            item.handler_unblock_by_func(self._on_autoreconnect_toggled)
        if SETTINGS["health-interval"] != self._health_interval:
            self._health_interval = SETTINGS["health-interval"]
            self._schedule_health_check()
            return False  # this timer is replaced by the one just armed
        # Before the watchdog's own off switch, and not gated by it: this is
        # not a check. `health-interval = 0` means "stop inspecting a live
        # tunnel", which is a different wish from "stay disconnected after one
        # drops" -- and the tick keeps running at the schema's cadence anyway,
        # precisely so something is still here to notice.
        if self.state == FAILED and self.dropped:
            self.dispatch(MSG_REBUILD)
            return True
        if SETTINGS["health-interval"] <= 0:
            return True  # watchdog off: keep reading the config, check nothing
        # The same three states that have MSG_CHECK rows in TRANSITIONS. The
        # table would drop a verdict sent from any other state anyway; this
        # gate only spares the log an "ignoring" line every interval.
        if (self.state not in (CONNECTED, RECOVERING, DEMOTED)
                or self.helper_proc is None):
            return True

        reason, facts = tunnel_health(self.tunnel_device, self.tunnel_ifindex,
                                      self.routes_seen)
        # Which families this tunnel has been observed to have. Latched from
        # the checks rather than captured at adoption, because the state event
        # arrives before vpnc-script has installed any routes at all.
        for family in ("routes4", "routes6"):
            if facts.get(family):
                self.routes_seen[family] = True
        # Latched here for the same reason, and against the same gap: in
        # log-matching fallback the tunnel-up line ("CSTP connected") can
        # arrive before openconnect has created the device, so the adoption
        # reads no index and "the tunnel device was replaced by a different
        # one" is then dead for the whole session. The first index this
        # tunnel is seen to have is the one to hold it to.
        if self.tunnel_ifindex is None and facts.get("ifindex") is not None:
            self.tunnel_ifindex = facts["ifindex"]
        # The verdict is a message like any other; the table weighs it
        # (RECOVERING stands aside — openconnect owns its own retry).
        self.dispatch(MSG_CHECK, source="device", reason=reason,
                      detail=self._facts(facts))
        if reason is not None:
            # No point probing a device that is already broken -- and that is
            # as true in RECOVERING as anywhere. It used to be excluded here,
            # so every third tick of an outage spent a thread and a five
            # second connect on a verdict the table then set aside unread,
            # openconnect owning its own recovery.
            return True

        # Every check, not every third like the probe: this asks the local
        # resolver daemon a question instead of sending a packet down the
        # tunnel, so there is nothing to ration. Gated on the setting because a
        # user who left DNS to vpnc-script has nothing here worth asserting,
        # and on tunnel_dns because a VPN that pushed no resolver never had one
        # installed to lose.
        if (SETTINGS["dns"] and self.tunnel_dns
                and not self.dns_check_in_flight):
            self._start_dns_check()

        self.probe_cycle += 1
        if (SETTINGS["probe"] and self._probe_target() and not self.probe_in_flight
                and self.probe_cycle % max(1, SETTINGS["probe-every"]) == 0):
            self._start_probe()
        return True

    def _start_dns_check(self):
        """Off the main loop, for the same reason the probe is.

        resolvectl answers in milliseconds, but it is a subprocess and a bus
        round trip, and a resolved that has wedged would otherwise freeze the
        applet for C.RESOLVECTL_TIMEOUT with the menu open.
        """
        self.dns_check_in_flight = True
        device, expected = self.tunnel_device, self.tunnel_dns
        generation = self.helper_generation

        def work():
            try:
                reason, detail = resolver_health(device, expected)
            except Exception as exc:
                # The reply has to arrive whatever happened: the in-flight flag
                # is cleared only in _dns_check_result, so a thread that died
                # here would disable this check for the applet's lifetime.
                reason, detail = None, f"dns check error: {exc.__class__.__name__}"
            GLib.idle_add(self._dns_check_result, reason, detail, generation)

        threading.Thread(target=work, daemon=True).start()

    def _dns_check_result(self, reason, detail, generation):
        self.dns_check_in_flight = False
        if self.quitting or generation != self.helper_generation:
            return False  # an answer about a tunnel that is no longer the live one
        # A verdict like any other. The table weighs it, and stands aside in
        # RECOVERING -- openconnect is rebuilding the link, and the resolver
        # goes back on it when `reconnect` runs asuvpn-notify again, which is
        # also precisely how a demotion from here gets repaired.
        self.dispatch(MSG_CHECK, source="dns", reason=reason, detail=detail)
        return False

    def _probe_target(self):
        """Where to probe: what the user configured, else what the VPN pushed."""
        return SETTINGS["probe-target"] or self.tunnel_dns

    def _start_probe(self):
        """Run the probe off the main loop; five seconds on it would freeze the UI."""
        self.probe_in_flight = True
        target, generation = self._probe_target(), self.helper_generation
        port, timeout = SETTINGS["probe-port"], SETTINGS["probe-timeout"]

        def work():
            try:
                alive, detail = probe_tunnel(target, port, timeout)
            except Exception as exc:
                # Whatever went wrong, the reply must arrive: probe_in_flight
                # is cleared only in _probe_result, and a dead thread here
                # would silently disable probing for the applet's lifetime.
                alive, detail = None, f"probe error: {exc.__class__.__name__}"
            GLib.idle_add(self._probe_result, alive, detail, generation)

        threading.Thread(target=work, daemon=True).start()

    def _probe_result(self, alive, detail, generation):
        self.probe_in_flight = False
        if self.quitting or generation != self.helper_generation:
            return False  # a reply for a tunnel that is no longer the live one
        if alive is None:
            return False  # unreachable or unconfigured; the device check owns it
        # A verdict, not an action: the table decides what it means — and in
        # DISCONNECTING it means nothing, which is exactly the guard a strike
        # needed the day it stomped a user's Disconnect.
        self.dispatch(MSG_CHECK, source="probe",
                      reason=None if alive else
                      "nothing answers through the tunnel",
                      detail=detail)
        return False

    @staticmethod
    def _facts(facts):
        """Formatted for the strike and recovery lines, as evidence.

        A quietly healthy check logs nothing — every twenty seconds forever
        would drown the log — so the facts appear on every failing check and
        on the recovery line, which is where the next silent break needs them.
        """
        return ", ".join(f"{k}={v}" for k, v in facts.items()) or "no detail"

    def _on_tunnel_output(self, line, generation):
        if generation != self.helper_generation:
            # Still log it. A superseded helper's last words are its teardown
            # diagnostics — exactly the "no default route" warning worth keeping.
            self.log(f"[old tunnel] {line}")
            return False
        message = C.decode_message(line)
        if message is not None:
            self.helper_spoke = True
            kind, payload = message
            self.log(line)
            if kind == C.KIND_DEVICE:
                self.dispatch(MSG_DEVICE, device=payload)
            elif kind == C.KIND_STATE:
                self._apply_state_event(payload)
            elif kind == C.KIND_FATAL:
                self.dispatch(MSG_FATAL, sentence=payload)
            elif kind == C.KIND_WARNING:
                self.dispatch(MSG_WARNING, sentence=payload)
            return False
        self.log(line)
        # The log patterns say which state we are in only as a fallback; they
        # never say why a connection failed, so the failure scan below stays
        # active even after events arrive. Scanning only in the fallback once
        # meant a post-connect failure was reported as a bare "exited with
        # status 1" while openconnect's own explanation sat in the log.
        if not self.state_events:
            if self.state == CONNECTED:
                # openconnect retries internally without exiting. Left
                # unhandled, the badge stays solid and `asuvpn status` keeps
                # exiting 0 while traffic goes nowhere.
                for pattern in RECONNECTING_PATTERNS:
                    if pattern.search(line):
                        self.dispatch(MSG_LINK_LOST)
                        return False
            # DEMOTED is included so a nudge-produced re-establish is adopted
            # (new address, new ifindex, fresh route-family latch) in fallback
            # mode too — the (DEMOTED, tunnel-up) row keeps the badge honest
            # either way. CONNECTED is deliberately not: mid-session DTLS and
            # rehandshake lines match these patterns on a healthy tunnel.
            if self.state in (CONNECTING, RECOVERING, DEMOTED):
                for pattern in CONNECTED_PATTERNS:
                    match = pattern.search(line)
                    if match:
                        address = (match.groupdict().get("addr") or "").strip()
                        # The same message the event path sends, so the table
                        # applies the same rules to both sources of truth.
                        self.dispatch(MSG_TUNNEL_UP,
                                      device=self.tunnel_device,
                                      address=address, dns="")
                        return False
        for pattern in FAILURE_PATTERNS:
            if pattern.search(line):
                # "[vpn] " is how the helper stops openconnect's output forging
                # helper messages; it is framing, not something to show anyone.
                # Scrubbed here, where the sentence is built: it becomes the
                # failure detail, which reaches a desktop notification and the
                # terminal via `asuvpn status` — paths the log scrubber never
                # sees, and this text is whatever the server chose to say.
                sentence = ANSI_ESCAPE.sub("", C.strip_relay(line)).strip()
                self.dispatch(MSG_PROBLEM, sentence=sentence)
                self.log(f"[tray] openconnect reported a problem: {line}")
                return False
        return False

    # ------------------------------------------------------------ teardown

    def _close_helper_stdin(self, proc):
        # Two teardown paths can arrive together (Disconnect then Quit), and a
        # write to a pipe the other one just closed raises ValueError, not
        # OSError — which would escape and strand the applet mid-teardown.
        with self._stdin_lock:
            try:
                if proc.stdin and not proc.stdin.closed:
                    proc.stdin.write(C.encode_control(C.CONTROL_QUIT))
                    proc.stdin.flush()
                    proc.stdin.close()
            except (OSError, ValueError):
                pass

    def _send_helper(self, verb):
        """Write one control verb, leaving the pipe open. False if it went nowhere."""
        with self._stdin_lock:
            proc = self.helper_proc
            if proc is None or proc.stdin is None or proc.stdin.closed:
                return False
            try:
                proc.stdin.write(C.encode_control(verb))
                proc.stdin.flush()
                return True
            except (OSError, ValueError):
                return False

    def _teardown_helper(self):
        """Close the control pipe and wait. False means it is still alive."""
        # Both handles read once, together: a tunnel starting between the two
        # reads would have us close one generation's pipe and wait on another's.
        with self._stdin_lock:
            proc = self.helper_proc
            exited = self.helper_exited
        if proc is None:
            return True
        self._close_helper_stdin(proc)
        if exited is not None and not exited.wait(timeout=SETTINGS["teardown-timeout"]):
            self.log("[tray] helper did not exit in time")
            return False
        return True

    def _disconnect_thread(self):
        if not self._teardown_helper():
            self.post(MSG_TEARDOWN_TIMEOUT, during="disconnect")

    @staticmethod
    def _kill_group(proc):
        try:
            os.killpg(os.getpgid(proc.pid), signal.SIGTERM)
        except (ProcessLookupError, PermissionError):
            try:
                proc.terminate()
            except ProcessLookupError:
                pass

    def _pump(self, stream, prefix):
        # ValueError, not OSError, is what a read from a stream closed under us
        # raises — and _auth_thread closes these once it stops waiting for the
        # pump to finish. Unhandled, it would kill this thread with a traceback.
        try:
            for raw in iter(stream.readline, ""):
                self.log(prefix + raw.rstrip("\n"))
        except (OSError, ValueError):
            pass

    # ----------------------------------------------------------- autostart

    def _on_autostart_toggled(self, item):
        try:
            self._write_autostart(item.get_active())
        except OSError as exc:
            self.log(f"[tray] could not update autostart: {exc}")
            item.handler_block_by_func(self._on_autostart_toggled)
            item.set_active(not item.get_active())
            item.handler_unblock_by_func(self._on_autostart_toggled)

    def _on_autoreconnect_toggled(self, item):
        try:
            write_setting("autoreconnect", item.get_active())
        except OSError as exc:
            self.log(f"[tray] could not change the setting: {exc}")
            item.handler_block_by_func(self._on_autoreconnect_toggled)
            item.set_active(not item.get_active())
            item.handler_unblock_by_func(self._on_autoreconnect_toggled)
            return
        self.log("[tray] automatic reconnection "
                 + ("enabled" if item.get_active() else "disabled"))

    def _write_autostart(self, enabled):
        if enabled:
            AUTOSTART_FILE.parent.mkdir(parents=True, exist_ok=True)
            AUTOSTART_FILE.write_text(
                "[Desktop Entry]\n"
                "Type=Application\n"
                f"Name={APP_NAME}\n"
                f'Exec="{Path(__file__).resolve()}" --server "{self.server}" tray\n'
                "Icon=asuvpn\n"
                "Terminal=false\n"
                "X-GNOME-Autostart-enabled=true\n"
            )
            self.log(f"[tray] autostart enabled ({AUTOSTART_FILE})")
        else:
            AUTOSTART_FILE.unlink(missing_ok=True)
            self.log("[tray] autostart disabled")


# ------------------------------------------------------- single instance IPC


def ipc_address():
    return "\0" + f"{APP_ID}-{os.getuid()}"


def peer_uid(conn):
    """The uid on the other end of a connected AF_UNIX socket, or None."""
    try:
        creds = conn.getsockopt(
            socket.SOL_SOCKET, socket.SO_PEERCRED, struct.calcsize("3I")
        )
        # uid_t is unsigned, and a short read raises struct.error, not OSError
        # — uncaught it would kill the whole IPC thread.
        _pid, uid, _gid = struct.unpack("3I", creds)
    except (OSError, struct.error):
        return None
    return uid


def send_to_running_instance(command, timeout=5.0):
    """Return the applet's reply, or None when no applet is running."""
    client = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    client.settimeout(timeout)
    try:
        client.connect(ipc_address())
    except OSError:
        return None
    # The name is an abstract socket, which any local user could have bound
    # first. The applet already refuses foreign clients via SO_PEERCRED; this
    # is the same check pointed the other way, so a squatter cannot feed this
    # command a fabricated state either. If a foreign process holds the name,
    # our applet is not running, and None is the honest answer.
    if peer_uid(client) != os.getuid():
        client.close()
        return None
    try:
        client.sendall(command.encode())
        client.shutdown(socket.SHUT_WR)
        # An empty read means the applet closed without answering — on its way
        # out, typically. Report that as "gone", not as an empty state.
        return client.recv(4096).decode("utf-8", "replace").strip() or None
    except OSError:
        return None
    finally:
        client.close()


def bind_ipc():
    """Claim this session's control socket, or None if another applet has it.

    Binding before building any UI is what makes the applet single-instance:
    two launches racing at login would otherwise both put an icon in the panel
    and both start a sign-in, and the loser would have no control channel.
    """
    server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    try:
        server.bind(ipc_address())
        server.listen(4)
    except OSError:
        server.close()
        return None
    return server


def serve_ipc(tray, server):
    actions = {
        "connect": tray.connect,
        "disconnect": tray.disconnect,
        "reconnect": tray.reconnect,
        "quit": tray.quit,
    }

    def loop():
        while True:
            try:
                conn, _ = server.accept()
            except OSError:
                return
            with conn:
                # An abstract AF_UNIX socket has no filesystem node and so no
                # permission bits: any local process can connect, and the uid in
                # the name enforces nothing. Without this check another user
                # could drop the tunnel, read the assigned VPN address, or —
                # worst — call `connect` to raise an admin password prompt on
                # this desktop whenever they liked.
                uid = peer_uid(conn)
                # Fail closed. Accepting when the credential cannot be read
                # would make an unreadable peer indistinguishable from us.
                if uid != os.getuid():
                    tray.log(f"[tray] refused a control connection from uid {uid}")
                    continue
                # A client that connects and never speaks would otherwise wedge
                # this single-threaded loop, and every later command with it.
                conn.settimeout(10)
                try:
                    command = conn.recv(256).decode("utf-8", "replace").strip()
                except OSError:
                    continue
                if command == "status":
                    reply = tray.status_line()
                elif command == "ping":
                    reply = "ok"
                elif command in actions:
                    # Run it on the main loop and wait for it to return, so the
                    # reply means "state has moved". Without this, a client that
                    # asks for status immediately afterwards still sees the old
                    # state and concludes nothing happened. Every action starts a
                    # worker and returns promptly, so this cannot stall.
                    finished = threading.Event()

                    # Both are bound as defaults, not captured: this closure
                    # outlives the loop iteration whenever the wait below times
                    # out, and a late-running invoke would otherwise signal a
                    # later command's event instead of its own.
                    def invoke(action=actions[command], done=finished):
                        try:
                            action()
                        finally:
                            done.set()
                        return False

                    GLib.idle_add(invoke)
                    # Shorter than the client's own 5s timeout, so a busy main
                    # loop produces an honest answer rather than the client
                    # timing out first and reporting "not running".
                    reply = "ok" if finished.wait(timeout=3) else "busy"
                else:
                    reply = f"unknown command: {command}"
                try:
                    conn.sendall((reply + "\n").encode())
                except OSError:
                    pass

    threading.Thread(target=loop, daemon=True).start()
    return server


# ------------------------------------------------------------- command line

COMMANDS = ("connect", "disconnect", "reconnect", "status", "log", "tray", "quit",
            "selftest", "autoreconnect")

COMMAND_HELP = """
commands:
  connect       sign in and bring the tunnel up (the default); a tunnel that
                is up but not carrying traffic is reconnected instead
  disconnect    close the tunnel, leaving the applet running; also calls
                off a rebuild the applet is about to attempt
  reconnect     sign in again and replace the current tunnel
  status        print the current state
  log           print this session's log, -f to follow it
  tray          run the applet in the foreground without connecting
  quit          close the tunnel and stop the applet
  selftest      check this installation against this machine
  autoreconnect show or set automatic reconnection: autoreconnect [on|off]

exit codes:
  0  the request was carried out; for status and --wait, connected
  1  status/--wait: not connected. disconnect: the tunnel did not close.
     quit: still shutting down. log: no log yet
  2  a bad command line (argparse's own code)
  3  status, or a --wait the applet vanished under: it is not running
  4  a different server was asked for than the running applet is using
"""


def run_selftest(argv):
    """Hand the whole command line to asuvpn-selftest.

    Its flags are its own; re-declaring them here would only let the two drift
    apart. Run with the same interpreter, so it inspects the python the applet
    actually uses rather than whatever is first on PATH.
    """
    selftest = HERE / "asuvpn-selftest"
    if not os.access(selftest, os.X_OK):
        print(f"{selftest} is missing or not executable", file=sys.stderr)
        return 1
    try:
        # -I mirrors the helper's isolation. Run privileged (sudo asuvpn
        # selftest), the script's user-owned directory must not be importable,
        # or a stdlib-named file dropped beside it executes as root.
        return subprocess.run(
            [sys.executable, "-I", str(selftest), *argv]).returncode
    except OSError as exc:
        print(f"could not run {selftest}: {exc}", file=sys.stderr)
        return 1


def run_autoreconnect(argv):
    if not argv:
        print(C.SETTINGS_BY_NAME["autoreconnect"].render(SETTINGS["autoreconnect"]))
        return 0
    if len(argv) > 1 or argv[0] not in ("on", "off"):
        print("usage: asuvpn autoreconnect [on|off]", file=sys.stderr)
        return 2
    try:
        write_setting("autoreconnect", argv[0] == "on")
    except OSError as exc:
        print(f"could not change the setting: {exc}", file=sys.stderr)
        return 1
    print(C.SETTINGS_BY_NAME["autoreconnect"].render(SETTINGS["autoreconnect"]))
    return 0


def show_log(follow):
    if not LOG_FILE.exists():
        print("no log yet — nothing has been connected in this session")
        return 1
    if send_to_running_instance("ping") is None:
        print(f"# {APP_NAME} is not running; this is the last session's log\n")
    try:
        fh = LOG_FILE.open()
        try:
            sys.stdout.write(fh.read())
            sys.stdout.flush()
            while follow:
                line = fh.readline()
                if line:
                    sys.stdout.write(line)
                    sys.stdout.flush()
                    continue
                try:
                    on_disk = os.stat(LOG_FILE)
                except OSError:
                    time.sleep(0.3)  # mid-rotation; the new file is imminent
                    continue
                # Connecting and the size cap both rotate the file, which
                # renames the inode this handle follows — switch to the new
                # one, or -f would silently tail yesterday's session.log.1.
                if os.fstat(fh.fileno()).st_ino != on_disk.st_ino:
                    fh.close()
                    fh = LOG_FILE.open()
                    continue
                if fh.tell() > on_disk.st_size:
                    fh.seek(0)  # truncated in place (log-keep = 0)
                time.sleep(0.3)
        finally:
            fh.close()
    except KeyboardInterrupt:
        pass
    except OSError as exc:
        print(f"cannot read {LOG_FILE}: {exc}", file=sys.stderr)
        return 1
    return 0


def wait_until_settled(timeout=300.0):
    """Poll the applet until it stops being busy. Signing in waits on a human."""
    deadline = time.monotonic() + timeout
    last = None
    misses = 0
    while time.monotonic() < deadline:
        reply = send_to_running_instance("status")
        if reply is None:
            # One unanswered poll is not proof the applet has gone; a busy main
            # loop can miss the socket timeout. Only give up after a few.
            misses += 1
            if misses >= 3:
                return None
        else:
            misses = 0
            last = reply
            if reply.split("\t", 1)[0] not in BUSY_STATES:
                return reply
        time.sleep(0.4)
    return last


def wait_until_gone():
    timeout = SETTINGS["teardown-timeout"] + 15
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        if send_to_running_instance("ping") is None:
            return True
        time.sleep(0.3)
    return False


def query_status(attempts=3):
    """Ask for the state, tolerating a momentarily busy main loop.

    One unanswered probe is not proof the applet has gone: a wedged idle queue
    can outlast the socket timeout, and reporting "not running" for a live
    tunnel is the worst answer this command can give.
    """
    for attempt in range(attempts):
        reply = send_to_running_instance("status")
        if reply is not None:
            return reply
        if attempt + 1 < attempts:
            time.sleep(0.5)
    return None


def print_status(reply):
    if reply is None:
        print(f"{APP_NAME}: not running")
        return EXIT_NOT_RUNNING
    state, _, rest = reply.partition("\t")
    server, _, detail = rest.partition("\t")
    line = f"{APP_NAME}: {STATE_LABELS.get(state, state)}"
    if server:
        line += f" ({server})"
    if detail:
        line += f" — {detail}"
    print(line)
    return 0 if state == CONNECTED else 1


def start_detached(child_args):
    """Relaunch ourselves in the background, then wait until the socket answers.

    A re-exec rather than a fork: GLib has already started worker threads by the
    time we get here, and forking away from them deadlocks the child.
    """
    command = [sys.executable, os.path.abspath(__file__), "--foreground", *child_args]
    try:
        subprocess.Popen(
            command,
            stdin=subprocess.DEVNULL,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            start_new_session=True,
        )
    except OSError as exc:
        print(f"could not start the applet: {exc}", file=sys.stderr)
        return 1

    for _ in range(150):
        if send_to_running_instance("ping") is not None:
            return 0
        time.sleep(0.1)
    print("the applet did not come up; try: asuvpn --foreground tray", file=sys.stderr)
    return 1


def main():
    parser = argparse.ArgumentParser(
        prog="asuvpn",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        description=f"{APP_NAME}: tray applet for openconnect-sso,"
        " with command line control.",
        epilog=COMMAND_HELP,
    )
    parser.add_argument(
        "command",
        nargs="?",
        default="connect",
        choices=COMMANDS,
        metavar="COMMAND",
        help="one of: " + ", ".join(COMMANDS),
    )
    # Left as None so we can tell "not given" from "same as the default", which
    # decides whether a mismatch with a running applet is worth complaining about.
    parser.add_argument(
        "--version",
        action="version",
        version=f"asuvpn {C.VERSION} (contract {C.CONTRACT_VERSION})",
    )
    parser.add_argument(
        "--server",
        default=None,
        metavar="HOST",
        help="VPN server to connect to (default: whatever install.sh was given)",
    )
    parser.add_argument(
        "-w",
        "--wait",
        action="store_true",
        help="with 'connect' or 'reconnect', wait for it to finish before returning",
    )
    parser.add_argument(
        "-f",
        "--follow",
        action="store_true",
        help="with 'log', keep following new lines",
    )
    parser.add_argument(
        "--foreground",
        action="store_true",
        help="do not fork into the background when starting the applet",
    )
    # Split at the first bare "--" by hand. argparse.REMAINDER would swallow
    # every later flag, so "asuvpn connect --wait" would silently pass --wait to
    # openconnect instead of to us.
    argv = sys.argv[1:]
    # Generating the config is the installer's job, but the schema lives in the
    # contract, so the program that can read it is the one that writes it. Kept
    # out of argparse deliberately: it is an installation step, not a command.
    if len(argv) == 2 and argv[0] == "--write-config":
        sys.stdout.write(C.render_settings({"server": argv[1]}))
        return 0
    # Intercepted before parsing so that --tier and --quiet reach the self-test
    # rather than being rejected here as unknown flags.
    if argv and argv[0] == "selftest":
        return run_selftest(argv[1:])
    # Intercepted for the same reason: it takes a value, and threading an
    # optional positional through argparse's `choices` would complicate every
    # other command to serve this one. The applet reads the file on each check,
    # so a change here takes effect on a running tray with no message passing.
    if argv and argv[0] == "autoreconnect":
        return run_autoreconnect(argv[1:])
    if "--" in argv:
        separator = argv.index("--")
        argv, passthrough = argv[:separator], argv[separator + 1 :]
    else:
        passthrough = []
    args = parser.parse_args(argv)

    command = args.command
    server = args.server or configured_server() or DEFAULT_SERVER

    if command == "selftest":  # reached via `asuvpn --server x selftest`
        return run_selftest([])
    if command == "autoreconnect":
        return run_autoreconnect([])
    if command == "log":
        return show_log(args.follow)
    if command == "status":
        return print_status(query_status())

    running = query_status()

    if command == "quit":
        if running is None:
            print(f"{APP_NAME} is not running")
            return 0
        send_to_running_instance("quit")
        if wait_until_gone():
            print(f"{APP_NAME} stopped")
            return 0
        print(f"{APP_NAME} is still shutting down", file=sys.stderr)
        return 1

    if command == "disconnect":
        if running is None:
            print(f"{APP_NAME} is not running")
            return 0
        send_to_running_instance("disconnect")
        reply = wait_until_settled(timeout=SETTINGS["teardown-timeout"] + 15)
        print_status(reply)
        # An action reports whether the action worked. Only `status` reports
        # connectedness, or `disconnect` would return 1 on success.
        return 0 if (reply or "").split("\t", 1)[0] == DISCONNECTED else 1

    if running is not None:
        running_state, _, rest = running.partition("\t")
        running_server = rest.partition("\t")[0]
        # Compare the *resolved* server, not just an explicit --server. Checking
        # only the flag meant that re-running install.sh with a new endpoint and
        # then typing a bare `asuvpn connect` silently drove the old applet at
        # the old server — the exact thing this guard claims to prevent.
        if running_server and server != running_server:
            print(
                f"{APP_NAME} is already running for {running_server}, "
                f"so it cannot connect to {server}.",
                file=sys.stderr,
            )
            print("Stop it first:  asuvpn quit", file=sys.stderr)
            return EXIT_WRONG_SERVER
        if command == "tray":
            print(f"{APP_NAME} is already running")
            return 0
        if passthrough:
            print(
                f"note: {APP_NAME} is already running, so these openconnect "
                "arguments are ignored. Use 'asuvpn quit' first.",
                file=sys.stderr,
            )
        if command == "connect" and running_state == CONNECTED:
            return print_status(running)
        send_to_running_instance(command)
        if args.wait:
            return print_status(wait_until_settled())
        print_status(send_to_running_instance("status"))
        return 0  # started; the outcome is not known yet without --wait

    # No applet yet, so become one. 'tray' stays in the foreground; the others
    # were asked to do something, so hand the terminal back once we are up.
    if command != "tray" and not args.foreground:
        child = ["--server", server, command]
        if passthrough:
            child += ["--", *passthrough]
        started = start_detached(child)
        if started != 0:
            return started
        if args.wait:
            return print_status(wait_until_settled())
        print_status(send_to_running_instance("status"))
        return 0  # started; the outcome is not known yet without --wait

    if GUI_ERROR is not None:
        print(f"cannot start the applet: {GUI_ERROR}", file=sys.stderr)
        print(
            "Install the GTK bindings (python3-gi, gir1.2-gtk-3.0, "
            "gir1.2-ayatanaappindicator3-0.1, gir1.2-notify-0.7) or re-run "
            "./bootstrap.sh.",
            file=sys.stderr,
        )
        return 1

    listener = bind_ipc()
    if listener is None:
        # Another applet claimed the socket between our probe above and now.
        print(f"{APP_NAME} is already running")
        return 0

    tray = VpnTray(server, passthrough)

    # Handlers first: a signal arriving after the sign-in has started but before
    # they were installed would kill the tray outright and orphan the browser.
    # SIGHUP matters too — closing the terminal would otherwise skip the
    # orderly quit and fall back to the helper's EOF path.
    for stop_signal in (signal.SIGINT, signal.SIGTERM, signal.SIGHUP):
        GLib.unix_signal_add(
            GLib.PRIORITY_DEFAULT, stop_signal, lambda: (tray.quit(), True)[1]
        )

    # Begin connecting before the socket starts answering. Otherwise a client
    # that has just been told we are up can sample the state before the queued
    # connect has run, see "disconnected", and conclude it already finished.
    if command != "tray":
        tray.connect()

    serve_ipc(tray, listener)
    Gtk.main()
    return 0


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