#!/usr/bin/python3 -I
"""Checks the ASU VPN applet against the machine it is installed on.

Ordinary unit tests would not have caught a single one of the bugs that hurt
this project. Matching on "Connected as", which openconnect stopped saying in
v8; hardcoding the vpnc-script path, which differs between distributions;
comparing whole argv elements when getopt bundles short options — every one was
a wrong assumption about the *environment*, and a self-contained test would have
passed happily while the code was dead.

So the checks come in three tiers, and the middle one is the point:

  logic        this project's own rules, driven in-process
  environment  our assumptions put to the installed binaries themselves
  wiring       asuvpn-notify actually executed, end to end

Nothing here needs privileges or talks to the VPN. The real openconnect is
run only to describe itself — `--version` for its script path, `--help` for
its retry default — and the network is touched only by two probe exercises
that cannot leave anything behind: a loopback connect and a SYN to RFC 5737
documentation space, which must never answer. Run it after installing, or
whenever something behaves oddly:

    asuvpn selftest
    asuvpn selftest --tier environment
    asuvpn selftest --quiet
"""

import sys

# Before anything imports the siblings. Loading them would otherwise drop a
# __pycache__ into the installed directory -- the one the helper runs out of as
# root -- created with the ambient umask and after install.sh has already
# tightened everything else. A check has no business writing to the tree it is
# checking.
sys.dont_write_bytecode = True

import argparse  # noqa: E402
import importlib.machinery  # noqa: E402
import importlib.util  # noqa: E402
import os  # noqa: E402
import re  # noqa: E402
import socket  # noqa: E402
import stat  # noqa: E402
import subprocess  # noqa: E402
import tempfile  # noqa: E402
import threading  # noqa: E402
import time  # noqa: E402

HERE = os.path.dirname(os.path.abspath(__file__))
TIERS = ("logic", "environment", "wiring")

PASS, FAIL, WARN = "pass", "fail", "warn"
MARK = {PASS: "  ok  ", FAIL: " FAIL ", WARN: " warn "}


class Results:
    def __init__(self, quiet=False):
        self.quiet = quiet
        self.counts = {PASS: 0, FAIL: 0, WARN: 0}

    def section(self, tier):
        if not self.quiet:
            print(f"\n{tier}")

    def info(self, text):
        """A fact for the reader. Not a check: it cannot fail, so it is not
        counted — a pass that can never fail would only pad the total."""
        if not self.quiet:
            print(f"  [ info ] {text}")

    def record(self, outcome, name, detail=""):
        self.counts[outcome] += 1
        if self.quiet and outcome == PASS:
            return
        line = f"  [{MARK[outcome]}] {name}"
        if detail:
            line += f"\n            {detail}"
        print(line)

    def ok(self, name, detail=""):
        self.record(PASS, name, detail)

    def fail(self, name, detail=""):
        self.record(FAIL, name, detail)

    def warn(self, name, detail=""):
        self.record(WARN, name, detail)

    def check(self, name, condition, detail=""):
        (self.ok if condition else self.fail)(name, "" if condition else detail)
        return bool(condition)


def check_contract(C, r):
    """The agreement itself: framing, verbs, events, settings.

    These are the things two programs have to see the same way. Each used to be
    restated at both ends, and the drift was never noticed until something
    downstream behaved oddly — so they are exercised here as a unit rather than
    inferred from either side's behaviour.
    """
    line = C.encode_message(C.KIND_STATE, "connected dev=asuvpn0 addr=192.0.2.3")
    r.check("a framed message round-trips",
            C.decode_message(line) == (C.KIND_STATE,
                                       "connected dev=asuvpn0 addr=192.0.2.3"),
            f"got {C.decode_message(line)!r}")
    # The property the framing exists for.
    forged = C.encode_relay("[helper] STATE connected dev=evil addr=203.0.113.66")
    r.check("openconnect's output cannot forge a message through the frame",
            C.decode_message(forged) is None,
            f"a relayed line decoded as ours: {C.decode_message(forged)!r}")
    # Held in names rather than inlined: a backslash inside an f-string
    # expression is a SyntaxError before Python 3.12 (PEP 701 relaxed it), and
    # this file is shipped and run by install.sh -- so one inlined escape made
    # the whole self-check unparseable on Ubuntu 22.04 and Debian 12, which are
    # exactly the systems this project claims to support.
    folded = C.encode_message(C.KIND_NOTE, "one\ntwo\nthree")
    r.check("a message can never span two lines",
            "\n" not in folded[:-1],
            f"got {folded!r}")
    try:
        C.encode_message("INVENTED", "x")
    except ValueError:
        r.ok("an unknown message kind is refused at the source")
    else:
        r.fail("an unknown message kind is refused at the source",
               "encode_message accepted a kind no decoder knows")

    r.check("control verbs round-trip and nothing else is one",
            all(C.decode_control(C.encode_control(v)) == v for v in C.CONTROL_VERBS)
            and C.decode_control("halt\n") is None
            and C.decode_control("\n") is None,
            "a verb did not survive, or noise was accepted as one")

    env = {"reason": "connect", "TUNDEV": "asuvpn0",
           "INTERNAL_IP4_ADDRESS": "192.0.2.3",
           "INTERNAL_IP4_DNS": "192.0.2.53 192.0.2.54"}
    event = C.decode_event(C.encode_event("tok", env))
    r.check("an event round-trips every field the watchdog needs",
            event and event["token"] == "tok" and event["reason"] == "connect"
            and event["TUNDEV"] == "asuvpn0"
            and event["INTERNAL_IP4_DNS"].split()[0] == "192.0.2.53",
            f"got {event!r}")
    hostile = C.encode_event("tok", {"reason": "connect", "TUNDEV": "a\tb\nc"})
    r.check("separators in a value cannot add a field or end the datagram",
            C.decode_event(hostile)["TUNDEV"] == "a b c",
            f"got {C.decode_event(hostile)!r}")
    # Held in a name because the detail has to report the value that was
    # actually asserted, and a backslash cannot appear inside an f-string
    # expression before Python 3.12 — which this still supports.
    short = C.decode_event(b"only\ttwo")
    r.check("a short datagram is refused rather than indexed into",
            short is None,
            f"got {short!r} — the fields are read by position, so a truncated"
            " datagram must be rejected whole rather than indexed into")

    values, problems = C.parse_settings(
        "server = x.edu\nprobe = off\ndpd = 0\nnope = 1\nprobe-every = -3\n"
        "probe-port = 70000\nprobe-port = 0\nteardown-timeout = 10\n")
    r.check("settings parse by type, and every default survives a bad file",
            values["server"] == "x.edu" and values["probe"] is False
            and values["dpd"] == 0
            and values["probe-every"] == C.SETTINGS_BY_NAME["probe-every"].default
            and values["probe-port"] == C.SETTINGS_BY_NAME["probe-port"].default,
            f"got {values!r}")
    # probe-port 0 is among the refused lines above: it is not a connectable
    # port, so accepting it would make every probe silently inconclusive —
    # the watchdog's probe half blind while `probe = on` claims coverage.
    r.check("unknown keys and out-of-range values are reported, not raised",
            len(problems) == 5, f"got {problems!r}")
    # The schema's own floor: a teardown wait shorter than the helper's
    # 15+10+5 signal escalation reports every clean disconnect as a failure,
    # so the setting's documentation forbids it and the parser enforces that.
    r.check("a teardown-timeout below the signal escalation is refused",
            values["teardown-timeout"]
            == C.SETTINGS_BY_NAME["teardown-timeout"].default
            and any("teardown-timeout" in p for p in problems),
            f"got {values['teardown-timeout']!r}, problems={problems!r}")
    rendered, round_trip_problems = C.parse_settings(C.render_settings())
    r.check("the generated config file parses back to the defaults, cleanly",
            rendered == C.defaults() and not round_trip_problems,
            f"problems={round_trip_problems!r} — render_settings and"
            " parse_settings disagree, so the file install.sh writes does"
            " not mean what the schema says")


def load(name, filename):
    """Import a sibling program. They have no .py extension, hence the loader.

    With the same refusal the programs' own loaders apply: run privileged,
    nothing is executed out of a directory a second principal can write.
    Every load goes through here — the contract and the sibling sources
    alike — so a tampered file is refused before it runs, not after. This
    was the one loader in the project that skipped the check.
    """
    path = os.path.join(HERE, filename)
    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(name, path)
    spec = importlib.util.spec_from_loader(name, loader)
    module = importlib.util.module_from_spec(spec)
    loader.exec_module(module)
    return module


# --------------------------------------------------------------------- logic


def check_option_blocklist(helper, r):
    """Options that would take openconnect out of the helper's supervision."""
    refuse = [
        "--background", "-b", "-bv", "-vb", "--syslog", "-l", "-lv",
        "--pid-file", "--pid-file=/tmp/x", "--cookieonly", "--authenticate",
    ]
    allow = ["-v", "-i", "--interface", "lo", "--script", "-s", "-4", "--os=linux-64"]
    missed = [o for o in refuse if helper.unsupported_option(o) is None]
    r.check(
        "supervision-breaking options are refused, bundles included",
        not missed,
        f"accepted: {', '.join(missed)} — getopt reads -bv as -b -v, so a bundle"
        " can smuggle in --background and openconnect then daemonises",
    )
    wrong = [o for o in allow if helper.unsupported_option(o) is not None]
    r.check(
        "ordinary options are still allowed through",
        not wrong,
        f"wrongly refused: {', '.join(wrong)}",
    )


def check_interface_names(C, r):
    """Anything failing this reaches `ip link delete` running as root."""
    good = ["asuvpn0", "tun0", "a", "x" * 15, "eth0.100", "my-vpn_1"]
    # The trailing newline is the regex trap: $ matches before it, \Z does
    # not, and "asuvpn0\n" once passed as a device name because of exactly that.
    bad = ["", "../../etc/passwd", "x" * 16, "-lo", "a/b", "a b", ".hidden",
           "asuvpn0\n"]
    rejected = [n for n in good if not C.INTERFACE_RE.match(n)]
    accepted = [n for n in bad if C.INTERFACE_RE.match(n)]
    r.check("usable interface names are accepted", not rejected,
            f"wrongly rejected: {rejected}")
    r.check("unusable interface names are rejected", not accepted,
            f"wrongly accepted: {accepted} — these reach 'ip link delete' as root")


def check_option_parsing(helper, r):
    """getopt gives openconnect the *last* occurrence; so must we."""
    argv = ["--interface", "docker0", "-i", "asuvpn3", "--script=/a/b"]
    r.check(
        "the last --interface wins, as it does for openconnect",
        helper.requested_interface(argv) == "asuvpn3",
        f"got {helper.requested_interface(argv)!r}; taking the first would aim"
        " teardown at a device openconnect never touched",
    )
    r.check("--script=VALUE is recognised",
            helper.requested_script(argv) == "/a/b",
            f"got {helper.requested_script(argv)!r}")
    dpd = helper.requested_dpd(["--force-dpd", "10", "--force-dpd=45"])
    r.check("the last --force-dpd wins, like every other option reader",
            dpd == "45",
            f"got {dpd!r}; a reader with different semantics from its"
            " siblings is a trap for whoever starts using the value")
    stripped = list(helper.strip_script_option(["-s", "/a/b", "--foo", "--script=/c"]))
    r.check("the caller's --script is removed before ours is added",
            stripped == ["--foo"], f"got {stripped}")


def check_reason_contract(C, r):
    """The five documented vpnc-script reasons, and pre-init's absence."""
    expected = {"connect": "connected", "reconnect": "connected",
                "attempt-reconnect": "connecting", "disconnect": "disconnected"}
    r.check("every documented reason maps to a state",
            expected == C.REASON_STATES,
            f"got {C.REASON_STATES}")
    r.check("pre-init produces no state (the tunnel is not configured yet)",
            C.REASON_STATES.get("pre-init") is None,
            f"got {C.REASON_STATES.get('pre-init')!r} — openconnect runs the"
            " script once before anything is configured; mapping that to a"
            " state would badge a tunnel that does not exist yet")


def check_permission_rules(C, r):
    """A shared group is a second principal; a user-private group is not."""
    with tempfile.TemporaryDirectory() as base:
        d = os.path.join(base, "app")
        os.mkdir(d)
        os.chown(d, -1, os.getgid())
        modes = [
            (0o755, None, "0755"),
            (0o777, "world-writable", "0777"),
            (0o757, "world-writable", "0757"),
        ]
        if os.getuid() == os.getgid():
            modes.insert(1, (0o775, None,
                             "0775 with a user-private group (umask 002"
                             " checkout)"))
        else:
            # The rule under test exempts gid == uid; on an account whose
            # primary group is shared, this fixture cannot stage a
            # user-private group, and failing would be crying wolf.
            r.warn("0775 with a user-private group is allowed",
                   "this account's primary gid differs from its uid, so the"
                   " user-private-group case cannot be staged here")
        for mode, expected, label in modes:
            os.chmod(d, mode)
            got = C.unsafe_write_access(d)
            r.check(f"{label} is {'allowed' if expected is None else 'refused'}",
                    got == expected, f"got {got!r}, expected {expected!r}")
        others = [g for g in os.getgroups() if g != os.getgid()]
        if not others:
            r.warn("a genuinely shared group is refused",
                   "no secondary group on this account to test with")
        else:
            # A deliberately permissive fixture: this mode is exactly what the
            # check under test is supposed to object to.
            os.chmod(d, 0o775)  # nosec B103  # noqa: S103
            try:
                os.chown(d, -1, others[0])
            except OSError as exc:
                r.warn("a genuinely shared group is refused", str(exc))
            else:
                got = C.unsafe_write_access(d)
                r.check("a genuinely shared group is refused",
                        got == f"writable by group {others[0]}", f"got {got!r}")
            os.chown(d, -1, os.getgid())
        # Restored to a sane mode before the directory is removed.
        os.chmod(d, 0o755)  # nosec B103  # noqa: S103


def check_tunnel_health(tray, r):
    """The watchdog's verdicts, against devices that are certainly present.

    `lo` stands in for a tun device deliberately: it too reports
    operstate=unknown and carries routes, so it exercises the exact conditions
    that made the obvious checks wrong. Three of the four things one would reach
    for first — operstate == "up", IFF_RUNNING, a default route through the
    device — are false on a healthy tunnel, and a watchdog built on them would
    cry wolf on every session.
    """
    try:
        with open("/sys/class/net/lo/ifindex") as fh:
            lo_index = int(fh.read().strip())
    except (OSError, ValueError):
        r.warn("a healthy device is not reported as broken", "no lo to test with")
        return
    reason, facts = tray.tunnel_health("lo", lo_index)
    r.check("a healthy device is not reported as broken", reason is None,
            f"lo was called broken: {reason!r} ({facts}) — a watchdog that"
            " false-alarms on a working link is worse than none")
    reason, _ = tray.tunnel_health("asuvpn99", None)
    r.check("a missing device is reported",
            reason is not None and "gone" in reason, f"got {reason!r}")
    reason, _ = tray.tunnel_health("lo", lo_index + 100000)
    r.check("a device that is not the one we created is reported",
            reason is not None and "replaced" in reason, f"got {reason!r}")
    reason, _ = tray.tunnel_health(None, None)
    r.check("no device to watch is not an error", reason is None,
            f"got {reason!r}")
    # The name arrives over a pipe and is interpolated into a path under /sys.
    # The helper validates it on the way out; this is the other end doing the
    # same rather than taking that on trust.
    hostile = ["../../etc/passwd", "a/b", "", "x" * 16, ".hidden", "-lo",
               "asuvpn0 ", "asuvpn0\nevil", "asuvpn0\n"]
    # "used as a path" means it reached the /sys lookup: facts carry the device
    # and no note saying it was set aside. An empty name returns before either.
    leaked = []
    for name in hostile:
        facts = tray.tunnel_health(name, None)[1]
        if "dev" in facts and "ignored" not in facts:
            leaked.append(name)
    r.check("a device name that could not name a device is never used as a path",
            not leaked, f"these were used unvalidated: {leaked}")
    fake = tray.route_count("asuvpn-nope")
    r.check("routes are counted for a real device and not for a fake one",
            tray.route_count("lo") != (0, 0) and fake[0] == 0 and not fake[1],
            f"lo={tray.route_count('lo')} fake={fake}")
    # The rule, staged deterministically: a family this tunnel was seen to
    # have, wholly gone, is a verdict even while the other family survives.
    # `ip route flush dev X` removes only IPv4, and a summed count sat green
    # through exactly that break on a live tunnel. Staged by standing in for
    # route_count, because what lo happens to carry differs per host — an
    # earlier version read the live counts and was vacuous wherever both
    # families happened to be populated.
    seen_all = {"routes4": True, "routes6": True}
    original_count = tray.route_count
    try:
        tray.route_count = lambda _dev: (0, 6)
        gone4, _ = tray.tunnel_health("lo", lo_index, seen_all)
        tray.route_count = lambda _dev: (3, 0)
        gone6, _ = tray.tunnel_health("lo", lo_index, seen_all)
        tray.route_count = lambda _dev: (3, 6)
        healthy, _ = tray.tunnel_health("lo", lo_index, seen_all)
        # A kernel built without IPv6 has no /proc/net/ipv6_route, so the v6
        # count is unreadable rather than zero. An empty table must still be
        # a verdict there -- `v4 == 0 and v6 == 0` reads like this test and
        # stayed False through exactly that machine -- while unreadable on
        # its own must stay silent, because it is not evidence of a fault.
        tray.route_count = lambda _dev: (0, None)
        no_v6_empty, _ = tray.tunnel_health("lo", lo_index)
        tray.route_count = lambda _dev: (3, None)
        no_v6_fine, _ = tray.tunnel_health("lo", lo_index)
        tray.route_count = lambda _dev: (None, None)
        unreadable, _ = tray.tunnel_health("lo", lo_index)
    finally:
        tray.route_count = original_count
    r.check("an empty table is a verdict even where IPv6 cannot be read",
            no_v6_empty is not None and "no routes" in no_v6_empty
            and no_v6_fine is None and unreadable is None,
            f"empty={no_v6_empty!r} populated={no_v6_fine!r}"
            f" unreadable={unreadable!r} — on a kernel without IPv6 this is"
            " the only check standing between a wiped table and a green badge")
    r.check("a route family the tunnel had, wholly gone, is a verdict",
            gone4 is not None and "IPv4" in gone4
            and gone6 is not None and "IPv6" in gone6
            and healthy is None,
            f"v4-gone={gone4!r} v6-gone={gone6!r} both-present={healthy!r} —"
            " the per-family rule does not hold")


def check_notifications(tray, r):
    """What the user is actually told when the tunnel comes and goes.

    Exercised through a stand-in `self`, so no display is needed. The point is
    that a drop is announced at all: the badge changing to a spinner is easy to
    miss, and the reassuring half — that openconnect recovers this one on its
    own, with no sign-in — is not something a spinner can convey.
    """

    class Stub:
        server = "vpn.example.edu"
        state = tray.CONNECTING
        detail = ""
        tunnel_ever_connected = False
        last_failure = "stale"
        # A tunnel that comes up settles whatever the applet still owed a
        # dropped one, so the real method is borrowed rather than stubbed:
        # a no-op here would let that half of _announce_connected rot.
        _clear_drop = tray.VpnTray._clear_drop
        dropped = True
        rebuilds = 2
        tunnel_up_since = None

        def __init__(self):
            self.notes = []

        def notify(self, summary, body="", icon=None):
            self.notes.append(summary)

        def _set_state(self, state, detail=""):
            self.state, self.detail = state, detail

        def _refresh(self):
            pass

    stub = Stub()
    tray.VpnTray._announce_connected(stub, "192.0.2.3")
    first = stub.notes[-1]
    r.check("a first connect is announced as a connection",
            first == "VPN connected" and stub.state == tray.CONNECTED,
            f"got {first!r}, state {stub.state!r}")
    r.check("reaching connected clears any earlier failure line",
            stub.last_failure is None, f"got {stub.last_failure!r}")
    r.check("reaching connected settles the debt but not the rebuild count",
            not stub.dropped and stub.rebuilds == 2
            and stub.tunnel_up_since is not None,
            f"dropped={stub.dropped} rebuilds={stub.rebuilds}"
            f" up_since={stub.tunnel_up_since} — every rebuild reaches this"
            " line, so clearing the count here makes MAX_REBUILDS unreachable"
            " against exactly the flap it exists for")

    tray.VpnTray._announce_link_lost(stub)
    r.check("a dropped tunnel is announced, not just badged",
            stub.notes[-1] == "VPN connection lost"
            and stub.state == tray.RECOVERING,
            f"got {stub.notes[-1]!r}, state {stub.state!r} — without this the"
            " user is never told the tunnel went away")

    tray.VpnTray._announce_connected(stub, "192.0.2.19")
    r.check("coming back is announced as a reconnection, not a fresh connect",
            stub.notes[-1] == "VPN reconnected",
            f"got {stub.notes[-1]!r} — the wording is how the user tells a"
            " recovery from a connection they asked for")


def machine_stub(tray):
    """A StateMachine host with the full declared surface, side effects recorded.

    One builder, because three checks each carried a near-identical
    twenty-line stub and fixture copies drift exactly like production copies
    do. Checks set state, autoreconnect and the bookkeeping per case, on the
    instance. The base class comes out of a path-loaded module, which the
    type checker cannot see into — the same limitation as the loaders'
    ModuleSpec residue, suppressed here at its one occurrence.
    """

    class Stub(tray.StateMachine):  # type: ignore[misc, name-defined]
        server = "vpn.example.edu"

        def __init__(self):
            self.state = tray.CONNECTING
            self.detail = ""
            self.state_events = False
            self.quitting = False
            self.demoted_by = None
            self.strikes = tray.fresh_strikes()
            self.nudge_spent = self.failure_notified = False
            self.routes_seen = {"routes4": False, "routes6": False}
            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.last_failure = None
            self.dropped = False
            self.rebuilds = 0
            self.last_nudge = float("-inf")
            self.last_autoreconnect = float("-inf")
            self.helper_generation = 0
            self.auth_generation = 0
            self.helper_spoke = True
            self.helper_proc = None
            self.teardown_intent = None
            self.reconnect_keep_log = False
            self.autoreconnect = False
            self.killed = 0
            self.notes, self.sent, self.logged = [], [], []
            self.signins, self.published, self.started = [], [], []

        def notify(self, summary, body="", icon=None):
            self.notes.append(summary)

        def log(self, line):
            self.logged.append(line)

        def _set_state(self, state, detail=""):
            self.state, self.detail = state, detail
            self.published.append(state)

        def _send_helper(self, verb):
            self.sent.append(verb)
            return True

        def _act_start_signin(self, keep_log):
            self.signins.append(keep_log)

        def _act_start_tunnel(self, host, cookie, fingerprint):
            self.started.append(host)

        def _kill_auth(self):
            self.killed += 1

        def autoreconnect_enabled(self):
            return self.autoreconnect

    return Stub


def check_demotion_rules(tray, r):
    """A demotion outlives openconnect's word, and a nudge is one per incident.

    Learned live on 2026-08-23, from `ip route flush dev asuvpn0`: the nudge's
    reconnect event promoted the badge and reset the incident even though the
    probe had never passed again — so the badge flapped every two minutes, the
    notification repeated, and because the demote cadence equalled the nudge
    rate limit, the sign-in branch was unreachable even when opted in.

    Driven the way the applet itself now works: messages injected into the
    real transition table, on a stand-in `self` that stubs only the side
    effects (log, notify, the badge, the worker threads). The machine under
    test is the machine that ships.
    """
    rows_missing = [(state, kind)
                    for (state, kind), name in tray.TRANSITIONS.items()
                    if not callable(getattr(tray.VpnTray, "_tr_" + name, None))]
    r.check("every transition row names a real handler", not rows_missing,
            f"rows without handlers: {rows_missing}")
    # And the reverse: a handler no row names is dead code the analysers
    # cannot see, because rows reach handlers by computed name — deleting a
    # row while its handler survives would otherwise pass everything.
    named = set(tray.TRANSITIONS.values())
    orphans = sorted(name for name in dir(tray.VpnTray)
                     if name.startswith("_tr_") and name[4:] not in named)
    r.check("no handler is orphaned by the table", not orphans,
            f"handlers no row names: {orphans}")

    Stub = machine_stub(tray)
    stub = Stub()
    strikes = max(1, tray.SETTINGS["health-strikes"])
    stub._apply_state_event("connected dev=asuvpn0 addr=192.0.2.1 dns=192.0.2.53")
    for _ in range(strikes):
        stub.dispatch(tray.MSG_CHECK, source="probe",
                      reason="nothing answers", detail="no reply")
    r.check("a broken tunnel is demoted and nudged exactly once",
            stub.state == tray.DEMOTED and stub.sent == ["reconnect"]
            and stub.notes[-1] == "VPN reconnecting",
            f"state={stub.state!r} sent={stub.sent} notes={stub.notes}")

    stub._apply_state_event("connected dev=asuvpn0 addr=192.0.2.2 dns=192.0.2.53")
    r.check("a reconnect event mid-demotion adopts the tunnel but not the badge",
            stub.state == tray.DEMOTED
            and stub.tunnel_address == "192.0.2.2"
            and "VPN reconnected" not in stub.notes,
            f"state={stub.state!r} addr={stub.tunnel_address!r}"
            f" notes={stub.notes}")

    # Wind the clock past the rate limit before the next round: the incident
    # flag, not the timestamp, must be what refuses a second nudge — otherwise
    # a reset flag hides behind the gap and this check proves nothing. A third
    # round follows so the warning's once-per-incident guard is what holds the
    # count at one, not the fixture stopping early.
    stub.last_nudge -= max(1, tray.SETTINGS["nudge-min-gap"]) + 1
    for _ in range(2 * strikes):
        stub.dispatch(tray.MSG_CHECK, source="probe",
                      reason="nothing answers", detail="no reply")
    r.check("an incident takes one nudge, then says the free option is spent",
            stub.sent == ["reconnect"]
            and stub.notes.count("VPN not carrying traffic") == 1,
            f"sent={stub.sent} notes={stub.notes}")
    # The decision trail: a cycle that declines to act has to say what it
    # declined and what it is waiting for, or the log reads as a hang — which
    # is exactly how the 2026-08-23 live break read.
    r.check("a cycle that declines to act logs its decision and its reason",
            any("already tried" in line for line in stub.logged)
            and any("sign-in is off" in line for line in stub.logged),
            f"logged={stub.logged}")

    stub.dispatch(tray.MSG_CHECK, source="device", reason=None,
                  detail="the device checks still pass")
    r.check("a source that did not demote cannot promote",
            stub.state == tray.DEMOTED,
            f"state={stub.state!r} — the device check cleared a probe"
            " demotion, which flaps the badge against a tunnel that is"
            " genuinely carrying nothing")

    # A verdict landing after teardown began must change nothing: this once
    # stomped a user's Disconnect back into the escalation ladder. The table
    # simply has no row for it.
    stub.state = tray.DISCONNECTING
    stub.dispatch(tray.MSG_CHECK, source="probe",
                  reason="nothing answers", detail="no reply")
    r.check("a probe verdict arriving mid-teardown changes nothing",
            stub.state == tray.DISCONNECTING and stub.strikes["probe"] == 0,
            f"state={stub.state!r} strikes={stub.strikes}")

    # openconnect re-establishing its own session outranks the watchdog: a
    # bad check landing in RECOVERING is weighed and deliberately set aside.
    # Counting it would strike, demote and nudge a tunnel openconnect is
    # already in the middle of fixing.
    stub.state = tray.RECOVERING
    stub.dispatch(tray.MSG_CHECK, source="probe",
                  reason="nothing answers", detail="no reply")
    r.check("a verdict during openconnect's own recovery is set aside",
            stub.state == tray.RECOVERING and stub.strikes["probe"] == 0,
            f"state={stub.state!r} strikes={stub.strikes}")
    stub.state = tray.DEMOTED

    # A blip, a clear, another blip: the clear must zero the count, or two
    # unrelated hiccups hours apart add up to a false demotion — and a
    # spurious nudge with it. Pinned at two strikes; the user's file may
    # legally say one, which would demote on the first blip.
    original_strikes = tray.SETTINGS["health-strikes"]
    tray.SETTINGS["health-strikes"] = 2
    try:
        blip = Stub()
        blip._apply_state_event("connected dev=asuvpn0 addr=192.0.2.9 dns=")
        blip.dispatch(tray.MSG_CHECK, source="probe",
                      reason="nothing answers", detail="no reply")
        blip.dispatch(tray.MSG_CHECK, source="probe", reason=None,
                      detail="answered")
        blip.dispatch(tray.MSG_CHECK, source="probe",
                      reason="nothing answers", detail="no reply")
    finally:
        tray.SETTINGS["health-strikes"] = original_strikes
    r.check("a clear resets the strike count; blips do not accumulate",
            blip.state == tray.CONNECTED and blip.strikes["probe"] == 1
            and blip.sent == [],
            f"state={blip.state!r} strikes={blip.strikes} sent={blip.sent}")

    stub.dispatch(tray.MSG_CHECK, source="probe", reason=None, detail="answered")
    r.check("the demoting source is what promotes, and it resets the incident",
            stub.state == tray.CONNECTED
            and stub.notes[-1] == "VPN carrying traffic again"
            and not stub.nudge_spent and not stub.failure_notified,
            f"state={stub.state!r} notes={stub.notes}")

    # A nudge that is merely held (rate-limited) is not spent: escalating past
    # it would buy with a Duo push what waiting under the gap gets for free.
    original_gap = tray.SETTINGS["nudge-min-gap"]
    tray.SETTINGS["nudge-min-gap"] = 120
    try:
        held = Stub()
        held.autoreconnect = True
        held._apply_state_event(
            "connected dev=asuvpn0 addr=192.0.2.1 dns=192.0.2.53")
        held.last_nudge = time.monotonic() - 1
        for _ in range(strikes):
            held.dispatch(tray.MSG_CHECK, source="probe",
                          reason="nothing answers", detail="no reply")
    finally:
        tray.SETTINGS["nudge-min-gap"] = original_gap
    r.check("a held nudge is waited out, never escalated past",
            held.signins == [] and held.sent == []
            and any("holding the free re-establish" in line
                    for line in held.logged),
            f"signins={held.signins} sent={held.sent} logged={held.logged}")

    # A reconnect that lands on a *different* device incarnation must not
    # inherit the old one's route-family observations: a stale IPv6 latch
    # held a demotion forever against a working v4-only session.
    if tray.C.interface_index("lo") is None:
        r.warn("a replaced device forgets the old one's route families",
               "no lo to stage the adoption with")
    else:
        fresh = Stub()
        fresh.state = tray.DEMOTED
        fresh.routes_seen = {"routes4": True, "routes6": True}
        fresh.tunnel_ifindex = -1  # provably not lo's real index
        fresh._adopt_tunnel("lo", "192.0.2.9", "")
        r.check("a replaced device forgets the old one's route families",
                fresh.routes_seen == {"routes4": False, "routes6": False}
                and fresh.tunnel_ifindex not in (None, -1),
                f"routes_seen={fresh.routes_seen}"
                f" ifindex={fresh.tunnel_ifindex}")

    # The verb a user reaches for. A demoted tunnel is established, so the
    # menu offers Reconnect — and `connect` has to mean the same intent there,
    # not a silent refusal that reads as "cannot connect". A plain CONNECTED
    # still declines, as ever.
    demoted = Stub()
    demoted.state = tray.DEMOTED
    demoted.dispatch(tray.MSG_CONNECT)
    connected = Stub()
    connected.state = tray.CONNECTED
    connected.dispatch(tray.MSG_CONNECT)
    r.check("connect on a demoted tunnel means reconnect, and only there",
            demoted.signins == [False]
            and any("reconnecting instead" in line for line in demoted.logged)
            and connected.signins == [] and connected.state == tray.CONNECTED,
            f"demoted={demoted.signins} connected={connected.signins}"
            f" state={connected.state!r}")

    # The log-matching fallback must feed the same table. A nudge-produced
    # re-establish that arrives only as a log line (no event channel) still
    # lands on the (DEMOTED, tunnel-up) row: adopt the tunnel, never promote
    # the badge. The event path proved this above; this is the other source.
    fallback = Stub()
    fallback.state = tray.DEMOTED
    fallback.tunnel_address = "192.0.2.1"
    fallback.helper_generation = 1
    tray.VpnTray._on_tunnel_output(fallback, "Configured as 192.0.2.7", 1)
    r.check("a fallback tunnel-up mid-demotion adopts but does not promote",
            fallback.state == tray.DEMOTED
            and fallback.tunnel_address == "192.0.2.7",
            f"state={fallback.state!r} addr={fallback.tunnel_address!r}")
    # Some connected patterns carry no address ("CSTP connected"); adopting
    # from one of those must not blank the address a real source delivered.
    tray.VpnTray._on_tunnel_output(fallback, "CSTP connected. DPD 30", 1)
    r.check("an addr-less fallback line does not blank a known address",
            fallback.tunnel_address == "192.0.2.7",
            f"addr={fallback.tunnel_address!r}")

    # A failure sentence lifted from openconnect's output becomes the detail
    # the user sees outside the log — in a notification and on the terminal
    # via `asuvpn status` — so it is scrubbed where it is built. The log's
    # own scrubber never sees those paths.
    # With events latched, deliberately: the failure scan must keep running
    # after the fallback state-matching retires, or a post-connect failure is
    # reported as a bare exit status while openconnect's own sentence rots
    # in the log — the exact regression the scan's comment records.
    fallback.state_events = True
    hostile = "[vpn] Failed to connect to host \x1b]0;evil\x07" + chr(0x9B) + "x"
    tray.VpnTray._on_tunnel_output(fallback, hostile, 1)
    kept = fallback.last_failure or ""
    r.check("a failure sentence is scrubbed before it becomes the detail",
            "Failed to connect" in kept
            and not any(ord(ch) < 0x20 or 0x7f <= ord(ch) <= 0x9f
                        for ch in kept),
            f"last_failure={kept!r}")


def check_teardown_rows(tray, r):
    """The teardown outcomes the sandbox cannot reach quickly.

    A helper that outlives teardown-timeout takes the whole signal ladder to
    stage for real, so the two timeout rows are driven here, on the shipping
    table. The reconnect handoff is pinned too: it must publish no state of
    its own, because a momentary DISCONNECTED between teardown and sign-in
    was readable by the IPC thread, and `asuvpn reconnect --wait` sampling
    that instant reported the reconnect finished.
    """

    Stub = machine_stub(tray)

    def tearing_down():
        # Mid-teardown with the reconnect intent: the default the timeout and
        # handoff rows are weighed against. The non-None intent is what makes
        # the handoff's `teardown_intent is None` assert below non-vacuous.
        stub = Stub()
        stub.state = tray.DISCONNECTING
        stub.teardown_intent = "reconnect"
        return stub

    timed = tearing_down()
    timed.reconnect_keep_log = True
    timed.dispatch(tray.MSG_TEARDOWN_TIMEOUT, during="reconnect")
    r.check("a reconnect whose old tunnel will not close is abandoned",
            timed.state == tray.FAILED and timed.signins == []
            and timed.reconnect_keep_log is False
            and any("abandoned" in line for line in timed.logged),
            f"state={timed.state!r} signins={timed.signins}"
            f" logged={timed.logged} — carrying on would start a second root"
            " tunnel beside one that is still up")

    stuck = tearing_down()
    stuck.teardown_intent = "disconnect"
    stuck.dispatch(tray.MSG_TEARDOWN_TIMEOUT, during="disconnect")
    r.check("a helper that will not exit is reported, with a warning raised",
            stuck.state == tray.FAILED and stuck.notes == ["VPN warning"],
            f"state={stuck.state!r} notes={stuck.notes}")

    handoff = tearing_down()
    handoff.reconnect_keep_log = True
    handoff.dispatch(tray.MSG_TEARDOWN_FINISHED)
    r.check("the teardown-to-sign-in handoff publishes no state of its own",
            handoff.signins == [True] and handoff.published == []
            and handoff.teardown_intent is None and handoff.helper_proc is None,
            f"signins={handoff.signins} published={handoff.published} — a"
            " momentary state here is what `reconnect --wait` mistakes for"
            " completion")

    dead = Stub()
    dead.state = tray.CONNECTED
    dead.teardown_intent = None
    before = dead.helper_generation
    dead.dispatch(tray.MSG_RECONNECT)
    r.check("a reconnect with no live helper signs in without a state detour",
            dead.signins == [False] and dead.published == []
            and dead.helper_generation == before + 1,
            f"signins={dead.signins} published={dead.published}"
            f" generation={dead.helper_generation} (was {before}) — the bump"
            " retires the dead helper's still-queued exit message")


def check_signin_deadline(tray, r):
    """A sign-in always ends, against a child that would otherwise outlive us.

    The gap this closes: `openconnect-sso` opens a browser window and blocks
    on a Duo approval, and `_auth_thread` waited on it with no bound at all.
    From the menu that is fine -- somebody is looking at the window. From an
    automatic rebuild it is not: with nobody at the keyboard the applet sat
    in AUTHENTICATING for as long as it lived, a state the watchdog does not
    run in and only Cancel leaves.

    Driven against a real process, because the failure is a wait that never
    returns and no test can observe that by waiting for it. `sleep 300` would
    outlive the whole suite, so if the deadline does not fire this check
    hangs rather than passing quietly -- which is the honest failure mode.
    """
    killed = []

    def kill(proc):
        killed.append(proc.pid)
        tray.VpnTray._kill_group(proc)

    proc = subprocess.Popen(["sleep", "300"], stdout=subprocess.PIPE,
                            start_new_session=True)
    began = time.monotonic()
    try:
        with tray.Deadline(proc, 1, kill) as clock:
            proc.stdout.read()   # ends only because the deadline closes the pipe
            returncode = proc.wait()
    finally:
        if proc.poll() is None:  # a broken check must not orphan the child
            tray.VpnTray._kill_group(proc)
            proc.wait()
    elapsed = time.monotonic() - began
    r.check("a sign-in that never finishes is ended, not waited on",
            clock.overdue.is_set() and killed == [proc.pid]
            and returncode != 0 and elapsed < 30,
            f"overdue={clock.overdue.is_set()} killed={killed}"
            f" rc={returncode} elapsed={elapsed:.1f}s")

    # And the other half: a child that finishes inside its limit must not be
    # killed, or every ordinary sign-in would be reported as timed out.
    quick = subprocess.Popen(["true"], stdout=subprocess.PIPE,
                             start_new_session=True)
    with tray.Deadline(quick, 30, kill) as fast:
        quick.stdout.read()
        quick.wait()
    r.check("a sign-in that finishes in time is left alone",
            not fast.overdue.is_set() and killed == [proc.pid],
            f"overdue={fast.overdue.is_set()} killed={killed}")

    # The setting cannot be turned off, because the guarantee above would
    # then have a hole exactly where an unattended rebuild runs.
    setting = tray.C.SETTINGS_BY_NAME["signin-timeout"]
    refused = []
    for value in ("0", "-1", "5", "99999"):
        try:
            setting.parse(value)
        except ValueError:
            refused.append(value)
    r.check("signin-timeout has no off, and no nonsense either",
            refused == ["0", "-1", "5", "99999"] and setting.default == 300,
            f"refused={refused} default={setting.default}")


def check_rebuild_rules(tray, r):
    """A tunnel that died on its own is rebuilt, and only while that is sane.

    The gap this closes: `autoreconnect` used to gate one rung of the
    demotion ladder and nothing else, so it did nothing at all for the most
    ordinary break there is -- openconnect giving up after its own reconnect
    timeout, which any suspend or WiFi outage longer than five minutes
    produces. The applet landed on FAILED and stayed there, with the setting
    that promises otherwise switched on.

    Three bounds are what make retrying safe rather than rude, and each is
    driven here on the shipping table: only a tunnel that carried traffic is
    ever rebuilt, the attempts share the ladder's rate limit, and they are
    counted so a network that is simply down cannot spend the afternoon
    opening browser windows.
    """
    # Pinned, not read live: `autoreconnect-min-gap = 0` is a legal line in
    # the user's file, and these fixtures assume a real gap — with zero, the
    # held case acts at once and the flap case refunds itself every lap, so
    # the suite would fail against designed behaviour.
    original_gap = tray.SETTINGS["autoreconnect-min-gap"]
    tray.SETTINGS["autoreconnect-min-gap"] = 300
    try:

        Stub = machine_stub(tray)

        def dropped_tunnel(**kwargs):
            """A tunnel that was carrying traffic and whose helper then died."""
            stub = Stub()
            stub.state = tray.CONNECTED
            stub.tunnel_ever_connected = True
            stub.autoreconnect = True
            for name, value in kwargs.items():
                setattr(stub, name, value)
            stub.dispatch(tray.MSG_HELPER_EXITED, status=1,
                          generation=stub.helper_generation)
            return stub

        stub = dropped_tunnel()
        r.check("a dropped tunnel arms a rebuild and says so",
                stub.state == tray.FAILED and stub.dropped
                and "rebuilding" in stub.detail
                and any("rebuilding it shortly" in line for line in stub.logged),
                f"state={stub.state!r} dropped={stub.dropped}"
                f" detail={stub.detail!r} logged={stub.logged} — a bare"
                " \"Not connected\" while three sign-ins are still owed reads as"
                " a dead applet, on the badge and through `asuvpn status`")

        stub.dispatch(tray.MSG_REBUILD)
        r.check("the rebuild signs in again, keeping the log that explains why",
                stub.signins == [True] and stub.rebuilds == 1
                and stub.notes[-1] == "VPN signing in again",
                f"signins={stub.signins} rebuilds={stub.rebuilds}"
                f" notes={stub.notes} — a rebuild that truncated the log would"
                " leave a fresh one saying nothing happened")

        # A failed attempt still owes the next one, and the badge must keep
        # saying so between attempts — an unattended sign-in timing out is
        # exactly the moment nobody is reading the log.
        stub.state = tray.AUTHENTICATING  # where the real sign-in would sit
        stub.dispatch(tray.MSG_AUTH_FAILED,
                      reason="sign-in did not finish within 300s",
                      attempt=stub.auth_generation)
        r.check("a failed rebuild attempt keeps the badge's promise",
                stub.state == tray.FAILED and stub.dropped
                and stub.detail.endswith("; rebuilding")
                and "did not finish" in stub.detail,
                f"detail={stub.detail!r} dropped={stub.dropped} — between"
                " attempts the badge must still say a rebuild is owed")

        # The rate limit is the ladder's, not a second one beside it: a rebuild
        # landing inside the gap must wait, or the two automatic sign-ins add up
        # to more than the user allowed.
        held = dropped_tunnel(last_autoreconnect=time.monotonic())
        held.dispatch(tray.MSG_REBUILD)
        r.check("a rebuild inside autoreconnect-min-gap waits its turn",
                held.signins == [] and held.rebuilds == 0,
                f"signins={held.signins} rebuilds={held.rebuilds}")

        # The bound that matters most: a network that is down stays down, and
        # spacing alone would keep opening browser windows all afternoon.
        capped = dropped_tunnel()
        for _ in range(tray.MAX_REBUILDS + 2):
            capped.last_autoreconnect = float("-inf")  # never the limiter here
            capped.dispatch(tray.MSG_REBUILD)
        r.check("rebuilding gives up after MAX_REBUILDS and says so once",
                len(capped.signins) == tray.MAX_REBUILDS
                and not capped.dropped
                and capped.detail == "connection dropped; gave up reconnecting"
                and capped.notes.count("VPN still not connected") == 1,
                f"signins={capped.signins} dropped={capped.dropped}"
                f" detail={capped.detail!r} notes={capped.notes} — the detail has"
                " to stop promising a rebuild that is not coming, while keeping"
                " the failure that caused it: outside the log that is the only"
                " record of why, and it reaches `asuvpn status` too")

        # A cancel lands mid-rebuild, which is the case this covers: leaving the
        # count spent gave a later drop fewer attempts than anyone was told about.
        cancelled = dropped_tunnel(rebuilds=2)
        cancelled.state = tray.AUTHENTICATING
        cancelled.dispatch(tray.MSG_CANCEL)
        r.check("cancelling a rebuild hands the whole budget back",
                cancelled.rebuilds == 0 and not cancelled.dropped
                and cancelled.killed == 1
                and cancelled.state == tray.DISCONNECTED,
                f"rebuilds={cancelled.rebuilds} dropped={cancelled.dropped}"
                f" killed={cancelled.killed} state={cancelled.state!r}")

        # The arming rule, which is what keeps a retry from becoming a loop: a
        # rejected cookie or a dismissed authorization will do exactly the same
        # thing the second time, browser window and Duo push included.
        never = Stub()
        never.state = tray.CONNECTING
        never.autoreconnect = True  # the arming rule, not the setting, must refuse
        never.dispatch(tray.MSG_HELPER_EXITED, status=1, generation=0)
        r.check("a tunnel that never came up is not rebuilt",
                never.state == tray.FAILED and not never.dropped
                and never.signins == [],
                f"dropped={never.dropped} signins={never.signins}")

        off = dropped_tunnel(autoreconnect=False)
        off.dispatch(tray.MSG_REBUILD)
        r.check("autoreconnect off leaves a dropped tunnel alone, and says so",
                off.signins == [] and off.rebuilds == 0 and not off.dropped
                and any("Connect when ready" in line for line in off.logged),
                f"signins={off.signins} dropped={off.dropped}"
                f" logged={off.logged}")

        # Read at the drop, not at the moment of acting: a box ticked long after
        # a tunnel died must not open a browser window for it. The opposite
        # direction still applies immediately, because stopping is the kind one.
        stale = dropped_tunnel(autoreconnect=False)
        stale.autoreconnect = True
        stale.dispatch(tray.MSG_REBUILD)
        turned_off = dropped_tunnel()
        turned_off.autoreconnect = False
        turned_off.dispatch(tray.MSG_REBUILD)
        # And unticking disarms rather than pausing: left armed, ticking the box
        # again later opened a browser window for a tunnel long since dead.
        turned_off.autoreconnect = True
        turned_off.dispatch(tray.MSG_REBUILD)
        r.check("the setting is read when the tunnel dies, and again before acting",
                stale.signins == [] and turned_off.signins == []
                and not turned_off.dropped
                and turned_off.detail == "connection dropped",
                f"ticked-after={stale.signins} unticked-after={turned_off.signins}"
                f" still-armed={turned_off.dropped}"
                f" detail={turned_off.detail!r}")

        # Called off from the panel. `asuvpn disconnect` has always cleared a
        # pending rebuild -- (FAILED, disconnect) is the row -- but the menu did
        # not offer the verb in that state, so the only way to stop one from the
        # panel was to untick a setting the user may want kept. The row is what
        # the menu item now reaches; this is that row.
        called_off = dropped_tunnel(rebuilds=1)
        called_off.dispatch(tray.MSG_DISCONNECT)
        r.check("a pending rebuild can be called off without changing the setting",
                not called_off.dropped and called_off.rebuilds == 0
                and called_off.state == tray.DISCONNECTED
                and called_off.autoreconnect,
                f"dropped={called_off.dropped} rebuilds={called_off.rebuilds}"
                f" state={called_off.state!r} setting={called_off.autoreconnect}")

        # A human taking over releases the budget: nothing may keep retrying
        # behind them, and their own attempt must not inherit a spent count.
        taken = dropped_tunnel()
        taken.rebuilds = tray.MAX_REBUILDS
        taken.dispatch(tray.MSG_CONNECT)
        r.check("a user connect clears what the applet still owed the drop",
                not taken.dropped and taken.rebuilds == 0
                and taken.signins == [False],
                f"dropped={taken.dropped} rebuilds={taken.rebuilds}"
                f" signins={taken.signins}")


        # A tunnel coming up settles what was owed, but does not refund the
        # count -- see the notifications tier for why that line is load-bearing.
        back = dropped_tunnel()
        back.rebuilds = 2
        back.state = tray.CONNECTING
        back.dispatch(tray.MSG_TUNNEL_UP, device="", address="192.0.2.5", dns="")
        r.check("a tunnel that comes back stops owing, and starts proving itself",
                not back.dropped and back.rebuilds == 2
                and back.state == tray.CONNECTED,
                f"dropped={back.dropped} rebuilds={back.rebuilds}"
                f" state={back.state!r}")

        # What refunds the count is a session that lasted. This is the whole
        # difference between "the rebuild worked" and "it came up and fell over".
        gap = tray.SETTINGS["autoreconnect-min-gap"]
        held_up = dropped_tunnel(rebuilds=2, tunnel_up_since=time.monotonic() - gap - 1)
        r.check("a session that outlived the gap starts the count over",
                held_up.rebuilds == 0,
                f"rebuilds={held_up.rebuilds} — a recovery that held is not a flap")

        # The case the limit exists for, driven end to end on the table: a
        # tunnel that comes up and falls over must not be able to buy an
        # unbounded run of Duo pushes by touching "connected" each time round.
        flap = Stub()
        flap.state = tray.CONNECTING
        flap.tunnel_ever_connected = True
        flap.autoreconnect = True
        for _ in range(tray.MAX_REBUILDS + 3):
            flap.last_autoreconnect = float("-inf")  # never the limiter here
            flap.dispatch(tray.MSG_TUNNEL_UP, device="", address="192.0.2.5", dns="")
            flap.dispatch(tray.MSG_HELPER_EXITED, status=1,
                          generation=flap.helper_generation)
            flap.dispatch(tray.MSG_REBUILD)
            flap.state = tray.CONNECTING  # the sign-in the stub does not run
        r.check("a flapping tunnel cannot outrun the rebuild limit",
                len(flap.signins) == tray.MAX_REBUILDS,
                f"signins={len(flap.signins)} of a permitted {tray.MAX_REBUILDS}"
                " — each round reaches 'connected', which is exactly how an"
                " earlier reset let this run forever")


    finally:
        tray.SETTINGS["autoreconnect-min-gap"] = original_gap


def check_signin_races(tray, r):
    """A sign-in that nobody wants any more must change nothing.

    Cancel, quit and every new connect bump `auth_generation`; a sign-in
    still in flight carries the generation it started with. Without the
    stale-attempt guards, a cancelled sign-in finishing late would start a
    root tunnel nobody asked for — a browser window and a password prompt
    landing minutes after the user pressed Cancel.
    """
    Stub = machine_stub(tray)
    late = Stub()
    late.state = tray.AUTHENTICATING
    late.auth_generation = 2
    late.dispatch(tray.MSG_AUTH_OK, host="h", cookie="c", fingerprint="f",
                  attempt=1)
    stale_ignored = late.started == [] and late.state == tray.AUTHENTICATING
    late.dispatch(tray.MSG_AUTH_OK, host="h", cookie="c", fingerprint="f",
                  attempt=2)
    r.check("a superseded sign-in cannot start a tunnel; the current one can",
            stale_ignored and late.started == ["h"],
            f"stale_ignored={stale_ignored} started={late.started} — the"
            " generation is the only thing standing between a cancelled"
            " sign-in and a root tunnel nobody wants")

    failed = Stub()
    failed.state = tray.AUTHENTICATING
    failed.auth_generation = 2
    failed.dispatch(tray.MSG_AUTH_FAILED, reason="too late to matter",
                    attempt=1)
    r.check("a superseded failure report changes nothing either",
            failed.state == tray.AUTHENTICATING and failed.notes == [],
            f"state={failed.state!r} notes={failed.notes}")


def check_start_refusals_keep_promise(tray, r):
    """Even a sign-in that cannot start keeps the badge's rebuild promise.

    A rebuild whose openconnect-sso has vanished fails before anything runs;
    the next attempt is still owed, and a badge that drops "; rebuilding"
    there reads as the applet having given up while it has not.
    """

    class Holder:
        dropped = True

        def __init__(self):
            self.published = []

        def _set_state(self, state, detail=""):
            self.published.append((state, detail))

        def log(self, line):
            pass

        def notify(self, summary, body="", icon=None):
            pass

    original = tray.find_openconnect_sso
    tray.find_openconnect_sso = lambda: None
    try:
        holder = Holder()
        tray.VpnTray._act_start_signin(holder, keep_log=True)
    finally:
        tray.find_openconnect_sso = original
    r.check("a rebuild's failed start keeps the badge's promise",
            holder.published
            and holder.published[-1][1].endswith("; rebuilding"),
            f"published={holder.published}")


def check_log_rotation(tray, r):
    """The session log is capped and rotated, and log-keep bounds the pile.

    Asked for after a live session: the log used to be wiped on every connect
    (destroying the record of whatever ended the last session) and rewritten
    in place at a hardcoded size. Now it rotates — connect and the size cap
    both shift session.log → .1 → .2 … — and `log-keep` bounds how many
    survive, 0 meaning none.
    """
    original_log = tray.LOG_FILE
    original_keep = tray.SETTINGS["log-keep"]
    holder = type("Holder", (), {"_log_bytes": 999})()
    with tempfile.TemporaryDirectory() as base:
        tray.LOG_FILE = type(original_log)(base) / "session.log"
        try:
            tray.SETTINGS["log-keep"] = 2
            problems = []
            for generation in ("one", "two", "three"):
                tray.LOG_FILE.write_text(generation)
                tray.VpnTray._rotate_log_locked(holder)
            def generation_text(name):
                try:
                    return (tray.LOG_FILE.parent / name).read_text()
                except OSError:
                    return "<missing>"

            gen1 = generation_text("session.log.1")
            gen2 = generation_text("session.log.2")
            mode = os.stat(tray.LOG_FILE).st_mode & 0o777
            if gen1 != "three" or gen2 != "two":
                problems.append(f"expected three/two, got {gen1!r}/{gen2!r}")
            if (tray.LOG_FILE.parent / "session.log.3").exists():
                problems.append("a fourth generation survived log-keep = 2")
            if tray.LOG_FILE.read_text() != "":
                problems.append("the fresh log is not empty")
            if mode != 0o600:
                problems.append(f"fresh log mode is {oct(mode)}, not 0600")
            if holder._log_bytes != 0:
                problems.append("the byte counter was not reset")
            r.check("the log rotates, keeps log-keep files, and starts fresh",
                    not problems, "; ".join(problems))
            tray.SETTINGS["log-keep"] = 0
            tray.LOG_FILE.write_text("gone")
            tray.VpnTray._rotate_log_locked(holder)
            leftovers = sorted(p.name for p in
                               tray.LOG_FILE.parent.glob("session.log.*"))
            r.check("log-keep 0 truncates and keeps no rotated files",
                    tray.LOG_FILE.read_text() == "" and not leftovers,
                    f"left behind: {leftovers}")
        finally:
            tray.LOG_FILE = original_log
            tray.SETTINGS["log-keep"] = original_keep


def check_config_editing(tray, r):
    """write_setting's contract: change one line, keep the user's file.

    `asuvpn autoreconnect on|off` and the menu checkbox both land here. A
    missing file earns the full documented render; an existing file keeps
    every other line as the user left it; and an unreadable file raises
    instead of being quietly replaced — a silent fall-back to defaults here
    would let one toggle discard every customization, server included.
    """
    original_config = tray.CONFIG_FILE
    problems = []
    try:
        with tempfile.TemporaryDirectory() as base:
            tray.CONFIG_FILE = type(original_config)(base) / "cfg" / "asuvpn.conf"
            parent = tray.CONFIG_FILE.parent
            tray.write_setting("autoreconnect", True)
            text = tray.CONFIG_FILE.read_text()
            if "autoreconnect = on" not in text.splitlines():
                problems.append(f"the toggle did not land: {text!r}")
            if not text.startswith("# ASU VPN settings."):
                problems.append("a missing file was not regenerated in full")
            mode = os.stat(tray.CONFIG_FILE).st_mode & 0o777
            if mode != 0o600:
                problems.append(f"config mode is {oct(mode)}, not 0600")
            dmode = os.stat(parent).st_mode & 0o777
            if dmode != 0o700:
                problems.append(f"config directory is {oct(dmode)}, not 0700")
            r.check("a missing config is regenerated in full, born private",
                    not problems, "; ".join(problems))

            problems = []
            tray.CONFIG_FILE.write_text(
                "server = my.example.edu  # mine\ndpd = 45\n")
            tray.write_setting("autoreconnect", False)
            lines = tray.CONFIG_FILE.read_text().splitlines()
            if lines[:2] != ["server = my.example.edu  # mine", "dpd = 45"]:
                problems.append(f"the user's lines were not preserved: {lines}")
            if "autoreconnect = off" not in lines:
                problems.append(f"the toggle did not land: {lines}")
            if os.geteuid() != 0:  # root reads through a 0000 mode
                os.chmod(tray.CONFIG_FILE, 0)
                try:
                    tray.write_setting("autoreconnect", True)
                except OSError:
                    pass
                else:
                    problems.append("an unreadable config was quietly replaced"
                                    " with defaults")
                finally:
                    os.chmod(tray.CONFIG_FILE, 0o600)
                after = tray.CONFIG_FILE.read_text().splitlines()
                if after != lines:
                    problems.append("the refused edit still changed the file:"
                                    f" {after}")
            leftovers = sorted(p.name for p in parent.glob("*.tmp"))
            if leftovers:
                problems.append(f"scratch files left behind: {leftovers}")
            r.check("one setting changes; the rest of the file is the user's",
                    not problems, "; ".join(problems))
    finally:
        tray.CONFIG_FILE = original_config
        tray.reload_settings()


def check_liveness_probe(tray, r):
    """The probe's verdicts, against endpoints every machine has.

    The rule worth pinning down is that a *refusal* counts as alive. Measured
    against a live ASU tunnel, one pushed resolver completed the handshake in
    29ms and another answered with a RST in 20ms -- and the RST is just as good,
    because it proves a packet crossed in each direction. Treating it as a
    failure would declare a perfectly good tunnel dead whenever the service
    behind the probe happened to be down.
    """
    # A port the kernel just handed out and took back: provably nothing
    # listens there, so the connect is answered with a RST. Port 9 was used
    # before, but a host running a discard service would have turned this
    # into a successful connect — the same True for the wrong reason.
    lease = socket.socket()
    lease.bind(("127.0.0.1", 0))
    port = lease.getsockname()[1]
    lease.close()
    alive, detail = tray.probe_tunnel("127.0.0.1", port, 3)
    r.check("a refused connection counts as a live tunnel", alive is True,
            f"got alive={alive!r} ({detail}) — a RST is a reply, and treating"
            " it as death would condemn a working tunnel")
    alive, detail = tray.probe_tunnel("", 53, 3)
    # A name getaddrinfo cannot even encode raises UnicodeError, which is not
    # an OSError; unhandled it killed the probe thread, and probe_in_flight
    # then wedged True and disabled probing for the applet's lifetime.
    bad_alive, bad_detail = tray.probe_tunnel("x" * 300, 53, 3)
    r.check("an unusable probe target is inconclusive, not a failure or a crash",
            alive is None and bad_alive is None,
            f"empty: alive={alive!r} ({detail});"
            f" unresolvable: alive={bad_alive!r} ({bad_detail})")
    # RFC 5737 documentation space: it either black-holes or is unreachable.
    # Either way the one answer it must never give is "alive".
    alive, detail = tray.probe_tunnel("198.51.100.1", 53, 2)
    r.check("an address that cannot answer is never reported as alive",
            alive is not True, f"got alive={alive!r} ({detail})")


def check_state_payload(tray, r):
    """The one place a malformed or hostile event meets the state machine."""
    cases = [
        ("connected dev=asuvpn0 addr=192.0.2.3", "connected", "asuvpn0", "192.0.2.3"),
        ("connecting dev=asuvpn0 addr=", "connecting", "asuvpn0", ""),
        ("disconnected dev=tun0 addr=2001:db8::ff", "disconnected", "tun0",
         "2001:db8::ff"),
    ]
    bad = []
    for payload, state, dev, addr in cases:
        got_state, fields = tray.parse_state_payload(payload)
        if (got_state, fields.get("dev"), fields.get("addr")) != (state, dev, addr):
            bad.append(payload)
    r.check("state payloads parse into state, device and address", not bad,
            f"misparsed: {bad}")
    for payload in ("", "   ", "bogus", "= = =", "connected"):
        try:
            tray.parse_state_payload(payload)
        except Exception as exc:  # any escape here would kill the reader thread
            r.fail("malformed payloads do not raise", f"{payload!r}: {exc}")
            return
    r.ok("malformed payloads do not raise")


def check_log_scrubbing(tray, r):
    """Nothing a server says can drive the reader's terminal.

    openconnect's output — a VPN banner included — ends up in the session log,
    and `asuvpn log` prints that to a terminal. Colour codes are stripped for
    tidiness, but the rule is really about control: an OSC sequence can retitle
    the window or plant a fake hyperlink, and stray control characters can
    redraw a line into saying something it does not say.
    """
    hostile = [
        "\x1b[31mred\x1b[0m text",
        "\x1b[?25lhidden cursor",
        "\x1b]0;you have been hacked\x07banner",
        "\x1b]8;;https://evil.example\x1b\\link\x1b]8;;\x1b\\",
        "bell\x07 and \x08backspace and \x7fdelete",
        # C1 controls built with chr(): decoded from well-formed UTF-8
        # they survive errors="replace", so only the scrubber stops them.
        "C1 via UTF-8: " + chr(0x9B) + "hostile and " + chr(0x9D) + "more",
        "lone escape \x1b at the end",
    ]

    def residue(text):
        return [ch for ch in text
                if (ord(ch) < 0x20 and ch != "\t") or 0x7f <= ord(ch) <= 0x9f]

    leaked = []
    for line in hostile:
        cleaned = tray.ANSI_ESCAPE.sub("", line)
        if residue(cleaned):
            leaked.append(f"{line!r} -> {cleaned!r}")
    kept = tray.ANSI_ESCAPE.sub("", "keeps\ttabs and-hyphens intact")
    if leaked:
        r.fail("hostile control sequences never reach the log", "; ".join(leaked))
    elif kept != "keeps\ttabs and-hyphens intact":
        r.fail("hostile control sequences never reach the log",
               f"plain text was mangled by the scrubber: {kept!r}")
    else:
        r.ok("hostile control sequences never reach the log")


def check_secret_redaction(tray, r):
    """No session token reaches the log, whoever wrote the line.

    The cookie is kept out of argv and out of the log by construction: it
    travels on a pipe into the tray and on stdin to the helper. But that holds
    only for the paths this project wrote, and the log carries lines this
    project did not write -- openconnect's own output, relayed verbatim.
    `openconnect --authenticate` prints COOKIE= on stdout, and the
    external-browser sign-in carries a token back through a redirect URL, so
    the last gate before bytes reach the file is where this has to be enforced.

    The useful half of a line has to survive, or the redaction gets turned off
    the first time somebody needs to debug a connection.
    """
    # Every value here is deliberately a word rather than anything shaped like
    # a real token. A fixture that looks like a credential trips secret
    # scanners, invites a reader to wonder whether it ever was one, and is
    # exactly the habit that put somebody's resolver in this repository once
    # already. The regex does not care what the value looks like.
    leaky = [
        "COOKIE='NOT-A-REAL-COOKIE-example-value'",
        "HOST='192.0.2.1' COOKIE=NOT-A-REAL-COOKIE FINGERPRINT='469bb424'",
        "GET /?acSamlv2Token=NOT-A-REAL-TOKEN.example.value HTTP/1.1",
        "Set-Cookie: webvpn=NOT-A-REAL-SESSION; path=/; secure",
        "sso-token: NOT-A-REAL-SSO-TOKEN",
    ]
    secrets = ("NOT-A-REAL-COOKIE", "NOT-A-REAL-TOKEN",
               "NOT-A-REAL-SESSION", "NOT-A-REAL-SSO-TOKEN")
    survived = [line for line in leaky
                for secret in secrets
                if secret in tray.redact_secrets(line)]
    r.check("no session token survives into the log",
            not survived,
            f"these still carry their secret after redaction: {survived}")

    kept = tray.redact_secrets(
        "HOST='192.0.2.1' COOKIE=NOT-A-REAL-COOKIE FINGERPRINT='469bb'")
    r.check("redaction keeps the half of the line worth reading",
            "192.0.2.1" in kept and "469bb" in kept and "COOKIE" in kept,
            f"got {kept!r} — the key stays visible so the reader knows"
            " something was removed, and the address and fingerprint are not"
            " secrets; a redactor that eats them gets switched off")

    ordinary = "[vpn] Connected to 192.0.2.4:443"
    r.check("an ordinary line passes through untouched",
            tray.redact_secrets(ordinary) == ordinary,
            f"got {tray.redact_secrets(ordinary)!r}")


def check_log_write_path(tray, r):
    """The scrubber must be wired where the log is written, not merely exist.

    The regex check above proves the pattern; this proves the binding. A
    hostile line pushed through the real VpnTray.log must land clean in the
    file, and a file the logger creates must be born 0600 — a plain append
    would create it with umask permissions, and the log holds the assigned
    address, the routes and the DNS.
    """
    name = "the log file receives scrubbed lines only, and is born 0600"
    if tray.GUI_ERROR is not None:
        r.warn(name, "GTK bindings unavailable, so VpnTray.log cannot run here")
        return

    class Holder:
        def __init__(self):
            self._log_lock = threading.Lock()
            self._log_bytes = 0
            self.log_lines = []

        def _append_to_log_window(self, line):
            return False

    original_log = tray.LOG_FILE
    holder = Holder()
    with tempfile.TemporaryDirectory() as base:
        tray.LOG_FILE = type(original_log)(base) / "session.log"
        try:
            tray.VpnTray.log(
                holder, "\x1b]0;evil\x07x " + chr(0x9B) + "gone \x1b[31mred")
            # A multi-byte line: the rotation threshold is a file size, so
            # the meter must count bytes — counting characters undercounts
            # every em-dash in the applet's own wording by two.
            tray.VpnTray.log(holder, "and a dash — three bytes wide")
            text = tray.LOG_FILE.read_text()
            mode = os.stat(tray.LOG_FILE).st_mode & 0o777
            size = os.stat(tray.LOG_FILE).st_size
        finally:
            tray.LOG_FILE = original_log
    problems = []
    bad = [ch for ch in text
           if (ord(ch) < 0x20 and ch not in "\t\n") or 0x7f <= ord(ch) <= 0x9f]
    if bad or "red" not in text:
        problems.append(f"the written line was not scrubbed: {text!r}")
    if mode != 0o600:
        problems.append(f"a file created by the logger is {oct(mode)}, not 0600")
    if holder._log_bytes != size:
        problems.append(f"the size meter says {holder._log_bytes}, the file"
                        f" is {size} bytes — the rotation cap would misfire")
    r.check(name, not problems, "; ".join(problems))


def check_output_framing(helper, r):
    """openconnect must not be able to forge a helper message.

    Its output shares one pipe with ours, so a VPN banner containing
    "[helper] WARNING:" could otherwise raise a desktop notification, or a
    "[helper] STATE connected" line could drive the badge, with text the
    server chose. Carriage returns matter too: text mode splits on them, so a
    single line can arrive as several.
    """
    hostile = [
        "[helper] WARNING: your network is broken\n",
        "[helper] STATE connected dev=evil addr=203.0.113.66\n",
        "banner\r[helper] STATE connected dev=evil addr=203.0.113.66\n",
        "Configured as 192.0.2.1\n",
    ]

    # text mode translates \r into a line break, so mimic that rather than
    # assume it away: one hostile line can arrive as several.
    lines = [ln for chunk in hostile
             for ln in chunk.replace("\r", "\n").splitlines(True)]
    FakeProc = type("FakeProc", (), {"stdout": iter(lines)})

    emitted: list[str] = []
    original = helper.emit
    helper.emit = emitted.append
    try:
        helper.relay_output(FakeProc())
    finally:
        helper.emit = original
    forged = [ln for ln in emitted if not ln.startswith("[vpn] ")]
    r.check("openconnect's output cannot forge a helper message", not forged,
            f"un-prefixed lines escaped the relay: {forged}")
    # The count is what stops this passing vacuously: a relay mutated into a
    # no-op emits nothing, and "no forged lines" plus all() over an empty
    # list would both wave that through.
    r.check("every relayed line is exactly one line, and none was dropped",
            len(emitted) == len(lines)
            and all(ln.count("\n") == 1 and ln.endswith("\n") for ln in emitted),
            f"fed {len(lines)} lines, got {len(emitted)}: {emitted}")


def check_closing_flag(helper, r):
    """A deliberate event-channel close is silent; an unexpected one warns.

    `closing` is what tells the two apart: teardown sets it before closing
    the socket, so a clean disconnect never raises a spurious "the state
    event channel stopped" warning — while the same failure mid-tunnel must
    warn, because state precision just silently degraded to log matching.
    Driven synchronously on an already-closed socket: recvfrom raises at
    once, which is the same except-branch the live close reaches.
    """
    emitted: list[str] = []
    original = helper.emit
    try:
        helper.emit = emitted.append
        helper.closing.set()
        gone = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
        gone.close()
        helper.serve_events(gone, "tok", "asuvpn0", {})

        def warnings():
            decoded = (helper.C.decode_message(line) for line in emitted)
            return [m for m in decoded if m and m[0] == helper.C.KIND_WARNING]

        silent = not warnings()
        emitted.clear()
        helper.closing.clear()
        gone = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
        gone.close()
        helper.serve_events(gone, "tok", "asuvpn0", {})
        # The KIND, not the wording: WARNING is what raises a notification,
        # and the sentence itself is the helper's to reword.
        warned = bool(warnings())
    finally:
        helper.closing.clear()
        helper.emit = original
    r.check("a deliberate event-channel close is silent; a surprise one warns",
            silent and warned,
            f"deliberate-close silent={silent}, surprise-close warned={warned}"
            " — one way every clean disconnect cries wolf, the other way a"
            " dead channel is never reported")


def check_teardown_ownership(helper, r):
    """Only the device this session created is ever deleted.

    verify_teardown runs as root and can call `ip link delete`; the ifindex
    captured when openconnect created the device is what proves the name
    still means *our* tunnel. Two helpers racing can both pick asuvpn0 — the
    loser must not delete the winner's live tunnel. No sandbox scenario can
    reach this (the namespace cannot create devices), so the decision is
    driven here with the module's own seams stood in for.
    """

    class Result:
        returncode = 0
        stdout = "default via 198.51.100.1 dev wlan0\n"

    deleted, logged = [], []
    originals = {name: getattr(helper, name) for name in
                 ("find_ip", "tunnel_interfaces", "interface_exists",
                  "run", "log")}
    # ifindex now lives in the contract, and each program loads its own copy of
    # that module, so patching this one reaches the helper and nothing else.
    original_index = helper.C.interface_index

    def fake_run(command, timeout=5):
        deleted.append(list(command))
        return Result()

    helper.find_ip = lambda: "/fake/ip"
    helper.tunnel_interfaces = set  # called with no args; an empty set back
    helper.interface_exists = lambda name: True
    helper.run = fake_run
    helper.log = logged.append
    try:
        helper.C.interface_index = lambda name: 7
        helper.verify_teardown("asuvpn0", set(), False, 9)  # not our incarnation
        foreign_deletes = [c for c in deleted if "delete" in c]
        left_alone = any("leaving it alone" in line for line in logged)
        deleted.clear()
        logged.clear()
        helper.verify_teardown("asuvpn0", set(), False, 7)  # ours: index matches
        ours_deleted = ["/fake/ip", "link", "delete", "asuvpn0"] in deleted
        # And the after-check: an empty routing table is the one outcome that
        # must never pass silently — it is the user's only signal that their
        # network did not come back. The WARNING prefix is what the helper's
        # log() turns into the kind that raises a notification.
        logged.clear()
        Result.stdout = ""
        helper.interface_exists = lambda name: False
        helper.verify_teardown("asuvpn0", set(), False, 7)
        broken_said = any(line.startswith("WARNING") for line in logged)
        Result.stdout = "default via 198.51.100.1 dev wlan0\n"
    finally:
        for name, value in originals.items():
            setattr(helper, name, value)
        helper.C.interface_index = original_index
    r.check("only the device this session created is ever deleted",
            not foreign_deletes and left_alone and ours_deleted,
            f"foreign deletes={foreign_deletes} left_alone={left_alone}"
            f" ours_deleted={ours_deleted} — a name match without the ifindex"
            " match must never reach `ip link delete` as root")
    r.check("a teardown that left no default route says so, as a warning",
            broken_said,
            f"logged={logged} — silence here reads exactly like a clean"
            " teardown while the machine may have no working network")


# --------------------------------------------------------------- environment


def read_catalogue(binary):
    """The message strings of openconnect, wrapper binary and library both."""
    blobs = []
    for path in filter(None, [binary]):
        try:
            with open(path, "rb") as fh:
                blobs.append(fh.read())
        except OSError:
            pass
    libs = []
    try:
        out = subprocess.run(["ldd", binary], stdin=subprocess.DEVNULL,
                             capture_output=True, text=True, timeout=10)
        libs = re.findall(r"(/\S*libopenconnect\S*)", out.stdout or "")
    except (OSError, subprocess.SubprocessError):
        pass
    for lib in libs:
        try:
            with open(lib, "rb") as fh:
                blobs.append(fh.read())
        except OSError:
            pass
    return b"".join(blobs), libs


def check_openconnect(helper, r):
    """openconnect is installed and is the binary the helper would run.

    The first thing that has to be true for any of this to work, and the one
    the rest of the environment tier builds on -- every check below reads this
    binary's own strings, its --help and its --version, so a wrong answer here
    makes all of them meaningless rather than merely absent.
    """
    binary = helper.find_openconnect()
    if not binary:
        r.fail("openconnect is installed",
               "not at /usr/sbin, /usr/bin or on PATH; nothing can connect")
        return None
    r.ok("openconnect is installed", binary)
    return binary


def check_gives_up(helper, r, binary):
    """openconnect stops retrying by itself, which is what arms the rebuild.

    The whole `(failed, rebuild)` path rests on this one external fact: when
    the link stays down, openconnect eventually gives up and exits, the
    helper follows it, and the tray gets an exit to act on. If it retried
    forever instead, the applet would sit in `recovering` and nothing would
    ever rebuild anything -- and the sandbox could not tell us, because the
    stand-in there gives up because we told it to.

    That is exactly the shape of the bug this suite exists for: the old fake
    printed `Connected as`, which the real binary had stopped saying, and
    every test passed against a dead applet. So the number the README quotes
    is read back out of the binary in front of us rather than remembered.
    """
    if not binary:
        return
    # --help, not --version: this asks the binary to describe its own
    # retry behaviour, which is the assumption under test. run() returns a
    # CompletedProcess or None; openconnect prints --help to stdout and
    # exits 0, but the stream is not worth being fussy about here.
    result = helper.run([binary, "--help"], timeout=10)
    text = (result.stdout or "") + (result.stderr or "") if result else ""
    match = re.search(r"--reconnect-timeout=SECONDS[^\n]*?"
                      r"default is (\d+) seconds", text)
    if not match:
        r.warn("openconnect gives up retrying rather than trying forever",
               "this build does not describe --reconnect-timeout in --help,"
               " so its default could not be read. If it retries forever, a"
               " dropped tunnel stays in Connecting… and is never rebuilt.")
        return
    seconds = int(match.group(1))
    r.ok("openconnect gives up retrying rather than trying forever",
         f"--reconnect-timeout defaults to {seconds}s, after which it exits"
         " and the tray rebuilds")
    # Only the documentation quotes a figure, so only the documentation can
    # be wrong about it. Not a failure: a distribution is free to ship a
    # different default, and the applet works whatever it is.
    if seconds != 300:
        r.warn("the documented five-minute figure matches this build",
               f"this openconnect gives up after {seconds}s, not 300;"
               " README and DESIGN both say five minutes")


def check_default_script(helper, C, binary, r):
    """The path we chain to, taken from openconnect rather than guessed.

    Passing --script replaces openconnect's own default. Get this wrong and the
    tunnel comes up with no routes and no DNS, restores nothing on the way out,
    and the tray reports "Connected" throughout.
    """
    if not binary:
        return
    reported = helper.vpnc_script_from_binary(binary)
    script = reported or C.FALLBACK_VPNC_SCRIPT
    derived = reported is not None
    if not os.access(script, os.X_OK):
        r.fail("openconnect's default vpnc-script is executable",
               f"{script} is missing or not executable. The helper will leave"
               " openconnect's own default in place and fall back to reading"
               " its output, so state will be less precise but routing safe.")
        return
    r.ok("openconnect's default vpnc-script is executable",
         f"{script}" + ("" if derived else "  (assumed, not reported by the binary)"))
    if not derived:
        r.warn("the default script path came from openconnect itself",
               "the binary did not report one, so the built-in fallback was"
               " used; correct on Debian and Ubuntu, a guess elsewhere")
    try:
        with open(script, encoding="utf-8", errors="replace") as fh:
            text = fh.read()
    except OSError as exc:
        r.warn("the default vpnc-script handles every reason we send", str(exc))
        return
    # Positive signal only. The stock script (and vpnc's original) dispatches
    # with `case "$reason" in ... connect) ...`, but nothing obliges a custom
    # one to: `if [ "$reason" = connect ]` is just as valid. Reporting a
    # *failure* on something we cannot actually determine would teach people to
    # ignore this suite, so an unrecognised script is a warning that says so.
    missing = [reason for reason in C.REASON_STATES
               if not re.search(rf"^\s*{re.escape(reason)}\)", text, re.M)]
    if not missing:
        r.ok("the default vpnc-script handles every reason we send")
    elif len(missing) == len(C.REASON_STATES):
        r.warn("the default vpnc-script handles every reason we send",
               f"{script} does not dispatch on $reason in the usual"
               " `case` form, so this could not be confirmed. Fine for a"
               " custom script; worth a look if you did not write it.")
    else:
        r.fail("the default vpnc-script handles every reason we send",
               f"it has case branches, but none for: {', '.join(missing)}."
               " Those transitions will not configure or restore anything.")


def check_message_catalogue(tray, binary, r):
    """Our log patterns, checked against the binary's own strings.

    This is the check that would have caught matching on "Connected as". The
    patterns are only a fallback now — state comes from the script contract —
    so a miss is a warning, not a failure. But a set with *nothing* left in it
    is dead code pretending to be a safety net.
    """
    if not binary:
        return
    catalogue, libs = read_catalogue(binary)
    if len(catalogue) < 1024:
        r.warn("log patterns still match the installed openconnect",
               "could not read the binary's message catalogue")
        return
    for label, table in (("connect", tray.CONNECTED_MESSAGES),
                         ("reconnect", tray.RECONNECTING_MESSAGES),
                         ("failure", tray.FAILURE_MESSAGES)):
        required = [(frag, needed) for frag, _, needed in table]
        present = [frag for frag, _ in required if frag.encode() in catalogue]
        gone = [frag for frag, needed in required
                if needed and frag.encode() not in catalogue]
        if not present:
            r.fail(f"the {label} fallback patterns still match this openconnect",
                   "not one fragment appears in the binary; these patterns are"
                   " dead and the fallback would never fire")
        elif gone:
            r.warn(f"the {label} fallback patterns still match this openconnect",
                   f"no longer present: {', '.join(gone)}")
        else:
            r.ok(f"the {label} fallback patterns still match this openconnect",
                 f"{len(present)}/{len(required)} fragments present"
                 + (f" in {os.path.basename(libs[0])}" if libs else ""))


def check_standin_catalogue(binary, r):
    """The sandbox stand-in may only say things openconnect can say.

    A stand-in is only as good as the strings it imitates: an earlier fake
    printed "Connected as", which the real binary had stopped saying, and
    every test passed while the applet was dead against the real thing. So
    every literal the scenario sandbox's fake openconnect prints must either
    start with "[stand-in] " — declared harness telemetry, which no real line
    begins with — or begin with a prefix the installed binary's own catalogue
    contains. Wording stays copied, never remembered.

    An installed copy has no sandbox beside it and nothing to drift, so the
    check passes vacuously there; it bites in a checkout.
    """
    name = "the sandbox stand-in only speaks lines from the installed catalogue"
    fake = os.path.join(HERE, "tests", "sandbox", "bin", "openconnect")
    if not os.path.exists(fake):
        if os.path.isdir(os.path.join(HERE, "tests")):
            # A checkout with tests/ but no stand-in at the expected path is
            # a moved or renamed fake — the honesty check silently passing
            # forever is exactly how a drifting stand-in gets loose.
            r.warn(name, "tests/ exists but tests/sandbox/bin/openconnect"
                   " does not; the honesty check has nothing to hold")
        else:
            r.ok(name, "no sandbox beside this copy; installed copies have none")
        return
    if not binary:
        r.warn(name, "openconnect itself is missing, so there is no catalogue"
               " to hold the stand-in against")
        return
    try:
        with open(fake, encoding="utf-8") as fh:
            source = fh.read()
    except OSError as exc:
        r.fail(name, f"cannot read {fake}: {exc}")
        return
    if re.search(r"\b(?:say|print)\(f?'", source):
        r.fail(name, "the stand-in prints through single quotes, which this"
               " check cannot read; keep its output double-quoted")
        return
    literals = re.findall(r'(?:\bsay|\bprint)\(f?"([^"]*)"', source)
    if not literals:
        r.fail(name, "no printed literals found in the stand-in; this check no"
               " longer understands how the fake speaks")
        return
    catalogue, _ = read_catalogue(binary)
    if len(catalogue) < 1024:
        r.warn(name, "could not read the binary's message catalogue")
        return
    # A message's literal prefix is everything before its first placeholder;
    # ten characters keeps incidental fragments from vouching for a line.
    prefixes = {m.split(b"%", 1)[0]
                for m in re.findall(rb"[\x20-\x7e]{10,}", catalogue)}
    prefixes = {p for p in prefixes if len(p) >= 10}
    telemetry = [t for t in literals if t.startswith("[stand-in] ")]
    imitated = [t for t in literals if not t.startswith("[stand-in] ")]
    invented = [t for t in imitated
                if not any(t.encode().startswith(p) for p in prefixes)]
    if invented:
        r.fail(name, "invented wording the installed binary cannot say: "
               + "; ".join(repr(t) for t in invented))
    else:
        r.ok(name, f"{len(imitated)} imitated lines anchored to the catalogue,"
             f" {len(telemetry)} marked [stand-in]")


def check_script_variables(binary, r):
    """The vpnc-script variables the DNS handover reads, in the installed binary.

    The same reasoning as the message catalogue next door, and the same failure
    it exists to prevent: an assumption about openconnect that quietly stops
    holding. The `CISCO_` prefix is historical -- inherited from vpnc, long
    before openconnect spoke anything but Cisco -- and openconnect normalises
    every protocol it speaks into these same names rather than inventing one
    set per gateway type. If a future release ever stops doing that, split DNS
    silently narrows to the derived fallback domain and nobody finds out; this
    is what says so instead.
    """
    catalogue, libs = read_catalogue(binary)
    if not catalogue:
        r.warn("the vpnc-script variables the DNS handover reads",
               "no binary to read; skipped")
        return
    wanted = ("CISCO_DEF_DOMAIN", "CISCO_SPLIT_DNS", "INTERNAL_IP4_DNS")
    missing = [name for name in wanted if name.encode() not in catalogue]
    r.check("openconnect still names the domains a split tunnel serves",
            not missing,
            f"missing {missing} from {libs or binary} — without these the"
            " gateway's own split-DNS list cannot be read, and scoping falls"
            " back to the domain derived from the server address")
    # Not an assertion, because openconnect merging both families into
    # INTERNAL_IP4_DNS is correct and expected; it is recorded because the
    # variable's name says IPv4 and the values are not all IPv4.
    if b"INTERNAL_IP6_DNS" not in catalogue:
        r.info("this openconnect sets no INTERNAL_IP6_DNS: every resolver,"
               " both families, arrives in INTERNAL_IP4_DNS — which is why"
               " they are validated as addresses, not as IPv4 addresses")


# The unauthenticated handshake every AnyConnect client makes before any login:
# "here is what I can do, what do you want?". No credentials are sent and none
# can be -- there is nothing in this exchange but a capability list.
SSO_PROBE_TIMEOUT = 10
SSO_PROBE_REQUEST = (
    '<?xml version="1.0" encoding="UTF-8"?>\n'
    '<config-auth client="vpn" type="init" aggregate-auth-version="2">'
    "<version who=\"vpn\">{version}</version><device-id>linux-64</device-id>"
    "<capabilities><auth-method>single-sign-on-v2</auth-method></capabilities>"
    "<group-access>https://{server}/</group-access></config-auth>"
)


def check_sso_method(C, r):
    """The gateway still offers the sign-in method this applet implements.

    openconnect-sso drives an embedded browser and reads the session cookie out
    of it, which is the `single-sign-on-v2` method. It is not the only one a
    Cisco gateway can offer: `single-sign-on-external-browser` hands the login
    to your normal browser instead, and a gateway configured for that and
    nothing else would leave every sign-in here failing with an unhelpful
    error, on a day nobody changed this code.

    So ask. The exchange is the same unauthenticated capability handshake any
    client makes first, it sends no credentials, and it answers in one line the
    question that otherwise costs an afternoon: what does this server actually
    accept? Measured against ASU's gateway, external-browser is refused outright
    (`error 108`) and v2 is issued -- so this is the method that has to keep
    working, and this check is what notices if that stops being true.

    Skips rather than fails when the server cannot be reached. A self-check that
    goes red on a train is one people learn to ignore.
    """
    import http.client

    settings, _ = C.load_settings(C.config_path())
    server = settings.get("server") or C.SETTINGS_BY_NAME["server"].default
    # The endpoint comes out of a config file the user owns and is about to be
    # dialled. Held to the same rule install.sh applies to --server: a hostname
    # and nothing else, so nothing here can carry a path or a credential.
    if not re.fullmatch(r"[A-Za-z0-9]([A-Za-z0-9.-]*[A-Za-z0-9])?", server) \
            or ".." in server:
        r.fail("the gateway still offers the sign-in method this applet uses",
               f"{server!r} in the config is not a plain hostname; refusing to"
               " turn it into a request")
        return

    # HTTPSConnection rather than urlopen: it can only speak https, so there is
    # no scheme to get wrong and no URL for a config value to smuggle one into.
    # The alternative was a linter suppression on urlopen, which would have
    # been a promise instead of a guarantee.
    body = SSO_PROBE_REQUEST.format(version=C.AC_VERSION, server=server)
    connection = None
    try:
        connection = http.client.HTTPSConnection(server,
                                                 timeout=SSO_PROBE_TIMEOUT)
        connection.request(
            "POST", "/", body=body.encode(),
            headers={"User-Agent": f"AnyConnect Linux_64 {C.AC_VERSION}",
                     "X-Transcend-Version": "1", "X-Aggregate-Auth": "1",
                     "X-Support-HTTP-Auth": "true",
                     "Content-Type": "application/xml; charset=utf-8"})
        page = connection.getresponse().read(65536).decode("utf-8", "replace")
    except (OSError, http.client.HTTPException, ValueError) as exc:
        r.warn(f"{server} still offers the sign-in method this applet uses",
               f"could not ask it ({exc.__class__.__name__}); offline, behind a"
               " captive portal, or the endpoint moved. Not a verdict.")
        return
    finally:
        if connection is not None:
            connection.close()

    if "sso-v2-login" in page:
        r.ok("the gateway still offers the sign-in method this applet uses",
             f"{server} issued an sso-v2 login URL")
        return
    if "single-sign-on-external-browser" in page and "error id" not in page:
        r.fail("the gateway still offers the sign-in method this applet uses",
               f"{server} wants single-sign-on-external-browser, which"
               " openconnect-sso cannot drive. Sign-in will fail until this"
               " applet learns openconnect's --external-browser flow.")
        return
    if "<form" in page and "password" in page:
        r.fail("the gateway still offers the sign-in method this applet uses",
               f"{server} answered with a plain username/password form instead"
               " of SSO. Either the endpoint changed or the group did; the"
               " browser sign-in this applet performs will not be used.")
        return
    r.fail("the gateway still offers the sign-in method this applet uses",
           f"{server} answered with neither an sso-v2 login URL nor a"
           f" recognisable form: {page[:160]!r}")


def check_desktop_stack(tray, r):
    """The GTK, AppIndicator and libnotify bindings the applet imports.

    These come from apt, not from pip, so a working Python says nothing about
    them -- and their absence is not a crash but a tray that never appears,
    which reads as "the applet is broken" rather than "a package is missing".
    Named individually because the fix is a different package for each.
    """
    if tray.GUI_ERROR is None:
        r.ok("GTK, AppIndicator and libnotify bindings are importable")
    else:
        r.fail("GTK, AppIndicator and libnotify bindings are importable",
               f"{tray.GUI_ERROR}\n            the CLI still works; the tray"
               " icon will not. Re-run ./bootstrap.sh")
    if any(os.access(os.path.join(d, "pkexec"), os.X_OK)
           for d in os.environ.get("PATH", "").split(os.pathsep) if d):
        r.ok("pkexec is available")
    else:
        r.fail("pkexec is available",
               "without it nothing can be elevated; install policykit-1")
    sso = tray.find_openconnect_sso()
    if sso:
        r.ok("openconnect-sso is installed", sso)
        check_sso_environment(tray, sso, r)
    else:
        r.fail("openconnect-sso is installed",
               "not on PATH or in ~/.local/bin; sign-in cannot run")


def check_sso_environment(tray, sso, r):
    """The setuptools pin, checked by its effect rather than by its command.

    openconnect-sso imports pkg_resources -- in its sign-in browser process --
    and current setuptools no longer ships it, so bootstrap.sh pins
    setuptools<71 inside its venv. That pin silently did
    nothing for a long time: `pipx inject` skips a package that is already
    present *without looking at the version*, prints a notice that was being
    redirected to /dev/null, and returns 0 — so every run reported success while
    the venv kept a setuptools that breaks sign-in at import time. It needs
    --force.

    A pin is only observable in its effect, so that is what is checked here.
    """
    name = "openconnect-sso can import pkg_resources"
    python = tray.sso_interpreter(sso)
    if not python or not os.access(python, os.X_OK):
        r.warn(name, f"could not work out which python runs {sso}")
        return
    probe = ("import pkg_resources, importlib.metadata as m\n"
             "print(m.version('setuptools'))\n")
    try:
        out = subprocess.run([python, "-c", probe], stdin=subprocess.DEVNULL,
                             capture_output=True, text=True, timeout=60)
    except (OSError, subprocess.SubprocessError) as exc:
        r.warn(name, str(exc))
        return
    if out.returncode == 0:
        r.ok(name, f"setuptools {out.stdout.strip()} in its own venv")
    else:
        r.fail(name,
               "sign-in dies at import. The setuptools pin did not take:\n"
               "            pipx inject openconnect-sso 'setuptools<71' --force")


def check_install_scripts(C, r):
    """The shell installers' hand-copied server default, held to the schema.

    bash cannot load this schema, so install.sh and bootstrap.sh each state
    the default once, with a comment promising to stay in sync by hand. This
    is that promise, enforced. Installed copies have no scripts beside them
    and nothing to drift, so the check passes vacuously there; it bites in a
    checkout, which is where the default would be edited.
    """
    name = "the shell installers' server default matches the schema"
    default = C.SETTINGS_BY_NAME["server"].default
    scripts = [os.path.join(HERE, s) for s in ("install.sh", "bootstrap.sh")]
    present = [s for s in scripts if os.path.exists(s)]
    if not present:
        r.ok(name, "no installer scripts beside this copy; nothing to drift")
        return
    wrong = []
    for script in present:
        try:
            with open(script, encoding="utf-8") as fh:
                text = fh.read()
        except OSError as exc:
            r.warn(name, f"cannot read {script}: {exc}")
            return
        if f'SERVER="{default}"' not in text:
            wrong.append(os.path.basename(script))
    r.check(name, not wrong,
            f"{', '.join(wrong)}: no SERVER=\"{default}\" line — the by-hand"
            " sync with the schema's default has drifted")


def check_no_bytecode(r):
    """Nothing may write into the directory the helper runs out of as root.

    This has regressed twice — once when the self-test began importing its
    siblings, and again when every program started loading the shared contract.
    Both times a __pycache__ appeared beside the helper, created with the
    ambient umask and after install.sh had tightened everything else. Cheap to
    check, evidently not obvious enough to remember.
    """
    stray = [name for name in os.listdir(HERE) if name == "__pycache__"]
    r.check("loading the contract writes nothing beside the helper", not stray,
            f"found {stray} in {HERE} — created with the ambient umask, in the"
            " one directory that is executed as root")


def check_installed_permissions(C, r):
    """The helper's own refusal criteria, applied now instead of at first connect."""
    problems = []
    # asuvpn_contract.py is on this list because the helper *executes* it as
    # root. It was missed when the contract was introduced: the check listed the
    # programs it knew about, and a new file that runs with the same privilege
    # is exactly the kind of thing such a list stops covering.
    for path in (HERE, os.path.join(HERE, "asuvpn-helper"),
                 os.path.join(HERE, "asuvpn-notify"),
                 os.path.join(HERE, "asuvpn_contract.py")):
        reason = C.unsafe_write_access(path)
        if reason:
            problems.append(f"{path} is {reason}")
    r.check("nothing the helper runs as root is writable by anyone else",
            not problems,
            "; ".join(problems) + "\n            the helper refuses to start"
            " like this. Fix with: chmod go-w " + HERE)
    for path in (os.path.join(HERE, "asuvpn-helper"),
                 os.path.join(HERE, "asuvpn-notify")):
        if not os.access(path, os.X_OK):
            r.fail("the helper and the notify script are executable",
                   f"{path} is not executable")
            return
    r.ok("the helper and the notify script are executable")


# -------------------------------------------------------------------- wiring


def check_split_dns(C, r):
    """The rules that decide which names go down the tunnel.

    These feed a command line run as root, so the interesting cases are the
    hostile ones: a domain that would parse as an option, one carrying shell
    punctuation, one that is really a whole flag. Every one of them has to be
    dropped rather than escaped -- and a domain so broad it would capture the
    whole internet has to be refused even when it is well formed.
    """
    r.check("comma and space separated domain lists both parse",
            C.split_domains("a.example.com,b.example.com", "c.example.com") ==
            ["a.example.com", "b.example.com", "c.example.com"],
            f"got {C.split_domains('a.example.com,b.example.com', 'c.example.com')}")
    r.check("domains are deduped and case folded, order kept",
            C.split_domains("EXAMPLE.com example.COM b.example.com") ==
            ["example.com", "b.example.com"],
            f"got {C.split_domains('EXAMPLE.com example.COM b.example.com')}"
            " — the first search domain is tried first, so order is meaning")
    hostile = C.split_domains("-x.example.com", "a;b.example.com", "$(id).example.com",
                              "a|b.example.com", "--dns=x", "..", "example..com",
                              "a`id`.example.com", "*.example.com", "a/b.example.com")
    r.check("anything that cannot be a domain is dropped, not escaped",
            hostile == [],
            f"got {hostile} — these become argv for a program run as root")
    r.check("a single label is not a domain",
            C.split_domains("localhost", "asu") == [],
            f"got {C.split_domains('localhost', 'asu')}")
    r.check("the domain list is bounded",
            len(C.split_domains(" ".join(f"h{i}.example.com" for i in range(200))))
            == C.MAX_DNS_DOMAINS,
            "a gateway must not be able to make the command line unbounded")
    hostile = C.split_resolvers("--listen", "192.0.2.53; id", "not-an-ip",
                                "192.0.2.999", "")
    r.check("a resolver that is not an address is dropped, not escaped",
            hostile == [],
            f"got {hostile} — these become argv for a program run as root,"
            " and a leading dash would be read as an option")
    resolvers = C.split_resolvers("192.0.2.53 192.0.2.53 192.0.2.54",
                                  "2001:db8::1")
    r.check("real resolvers survive, both families, order and dedup kept",
            resolvers == ["192.0.2.53", "192.0.2.54", "2001:db8::1"],
            f"got {resolvers} — the first is asked first, and is the probe"
            " target too")
    canonical = C.split_resolvers("2001:0db8::1 2001:db8:0:0:0:0:0:1",
                                  "192.0.2.53")
    r.check("resolvers come back canonical, so both ends can compare them",
            canonical == ["2001:db8::1", "192.0.2.53"],
            f"got {canonical} — resolvectl echoes systemd's spelling of an"
            " IPv6 address, so a gateway's spelling would have the tray"
            " reporting a resolver missing that is sitting right there;"
            " the two spellings must also dedup to one")
    r.check("a gateway's own name yields the domain it serves",
            C.parent_domain("vpn.example.com") == "example.com",
            f"got {C.parent_domain('vpn.example.com')!r}")
    r.check("a two-label name yields nothing rather than a public suffix",
            C.parent_domain("example.com") == "" and C.parent_domain("host") == "",
            f"got {C.parent_domain('example.com')!r} and {C.parent_domain('host')!r}"
            " — routing '.edu' to the tunnel would send half the internet"
            " down it")


def check_resolver_health(tray, r):
    """The third check: is the resolver the VPN pushed still on its link?

    The failure it exists for passes every other check -- device up, routes
    installed, packets crossing -- so the verdicts have to be exact. In
    particular "resolved did not answer" is not a verdict at all: a machine
    without systemd-resolved never had this configured, and demoting a healthy
    tunnel over an unanswerable question is worse than not asking it.
    """
    original = tray.link_resolvers
    try:
        tray.link_resolvers = lambda device: ["192.0.2.53"]
        reason, _ = tray.resolver_health("asuvpn0", "192.0.2.53")
        r.check("a link still holding the pushed resolver is healthy",
                reason is None, f"got {reason!r}")

        tray.link_resolvers = lambda device: []
        reason, detail = tray.resolver_health("asuvpn0", "192.0.2.53")
        r.check("a link the resolver was taken off is a verdict",
                reason is not None and "no resolvers" in detail,
                f"got {reason!r} / {detail!r} — this is the exact state"
                " systemd-resolved rewriting its stub file leaves behind")

        tray.link_resolvers = lambda device: ["198.51.100.1"]
        reason, _ = tray.resolver_health("asuvpn0", "192.0.2.53")
        r.check("a link holding somebody else's resolver is a verdict",
                reason is not None, f"got {reason!r}")

        # The parsing half of the same question. link_resolvers feeds every
        # case above, and the half that can be got wrong quietly is how one
        # line of resolvectl output is read -- not the subprocess around it.
        parsed = tray.parse_link_dns(
            "Link 9 (asuvpn0): 192.0.2.53#dns.example.com 2001:db8::1\n")
        r.check("a DNS-over-TLS server is recognised by its address",
                parsed == ["192.0.2.53", "2001:db8::1"],
                f"got {parsed} — resolvectl decorates a DoT server with the"
                " name it validates against, and the #suffix must not make a"
                " resolver that is sitting right there look absent")
        empty = tray.parse_link_dns("Link 9 (asuvpn0):\n")
        r.check("a link with no resolvers parses as none, not as unanswerable",
                empty == [],
                f"got {empty} — [] is the verdict-bearing case; None would"
                " mean the question could not be asked")

        tray.link_resolvers = lambda device: None
        reason, _ = tray.resolver_health("asuvpn0", "192.0.2.53")
        r.check("an unanswerable question is not a verdict",
                reason is None, f"got {reason!r}")

        original_conf = tray.resolv_conf_lists
        try:
            tray.link_resolvers = lambda device: []
            tray.resolv_conf_lists = lambda resolver: True
            reason, _ = tray.resolver_health("asuvpn0", "192.0.2.53")
            r.check("a resolver in force the old way is still in force",
                    reason is None,
                    f"got {reason!r} — asuvpn-notify hands DNS back to"
                    " vpnc-script whenever it cannot configure the link, and"
                    " demoting that tunnel would walk the ladder to a Duo push"
                    " every five minutes with nothing wrong")
        finally:
            tray.resolv_conf_lists = original_conf

        tray.link_resolvers = lambda device: []
        reason, _ = tray.resolver_health("asuvpn0", "")
        r.check("a VPN that pushed no resolver is not judged for losing one",
                reason is None, f"got {reason!r}")
    finally:
        tray.link_resolvers = original

    r.check("every check source has its own demotion wording",
            set(tray.DEMOTION_TEXT) == set(tray.CHECK_SOURCES)
            and tray.DEMOTION_TEXT["dns"] != tray.DEMOTION_TEXT["probe"],
            f"{tray.DEMOTION_TEXT} — a tunnel whose DNS was taken off the link"
            " is carrying traffic perfectly; saying it is not would be a lie")
    r.check("a fresh strike count covers every source",
            set(tray.fresh_strikes()) == set(tray.CHECK_SOURCES),
            f"got {tray.fresh_strikes()} — a source missing here raises"
            " KeyError in a worker thread on its first bad verdict")


def check_notify_roundtrip(C, r):
    """Run asuvpn-notify for real: does the event arrive, and does chaining work?

    This is the whole state framework in miniature, and it needs no privileges
    and no network — so there is no excuse for not checking it on the machine
    that will run it.

    The wire shape below is asserted as literal bytes on purpose — a golden
    reading of the datagram, so a format change fails here loudly — while the
    variable names come from the contract, so a rename follows through.
    """
    notify = os.path.join(HERE, "asuvpn-notify")
    if not os.access(notify, os.X_OK):
        r.fail("asuvpn-notify reports an event and chains to the real script",
               f"{notify} is missing or not executable")
        return
    with tempfile.TemporaryDirectory() as d:
        os.chmod(d, 0o700)
        sock_path = os.path.join(d, "events")
        marker = os.path.join(d, "chained")
        chained = os.path.join(d, "real-script")
        with open(chained, "w", encoding="utf-8") as fh:
            token_var = "${" + C.EVENT_TOKEN_VAR + ":-scrubbed}"
            fh.write("#!/bin/sh\n"
                     f'printf "%s\\n%s\\n" "$reason" "{token_var}" > {marker}\n')
        os.chmod(chained, 0o700)

        server = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
        server.bind(sock_path)
        server.settimeout(10)
        env = dict(os.environ)
        env.update({
            C.EVENT_SOCKET_VAR: sock_path,
            C.EVENT_TOKEN_VAR: "0123456789abcdef",
            C.REAL_SCRIPT_VAR: chained,
            "reason": "connect",
            "TUNDEV": "asuvpn0",
            "INTERNAL_IP4_ADDRESS": "192.0.2.31",
            "INTERNAL_IP4_DNS": "192.0.2.53 192.0.2.54",
        })
        try:
            proc = subprocess.run([notify], env=env, stdin=subprocess.DEVNULL,
                                  capture_output=True, text=True, timeout=20)
        except (OSError, subprocess.SubprocessError) as exc:
            r.fail("asuvpn-notify reports an event and chains to the real script",
                   str(exc))
            server.close()
            return
        try:
            data = server.recvfrom(4096)[0].decode("utf-8", "replace")
        except OSError:
            data = ""
        finally:
            server.close()

        fields = data.split("\t")
        r.check("the event reaches the helper's socket",
                len(fields) == 6 and fields[0] == "0123456789abcdef",
                f"received {data!r}")
        r.check("the VPN's own resolvers come across, for the liveness probe",
                fields[5:6] == ["192.0.2.53 192.0.2.54"],
                f"received {fields[5:6]} — without this there is nothing to"
                " probe, and a black-holed tunnel looks perfectly healthy")
        r.check("the event carries reason, device and address",
                fields[1:4] == ["connect", "asuvpn0", "192.0.2.31"],
                f"received {fields[1:4] if len(fields) > 3 else data!r}")
        try:
            with open(marker, encoding="utf-8") as fh:
                got = fh.read().split("\n")
        except OSError:
            got = []
        r.check("the real vpnc-script still runs, with its environment intact",
                got[:1] == ["connect"],
                f"marker={got!r} rc={proc.returncode} err={proc.stderr.strip()!r}"
                "\n            routing and DNS depend on this chaining")
        r.check("the event token is scrubbed before the real script sees it",
                got[1:2] == ["scrubbed"],
                f"the chained script saw the token: {got[1:2]}")


def check_dns_takeover(notify, C, r):
    """Drive the DNS handover against a stand-in resolvectl.

    The real one is found at a hardcoded absolute path, which is what a program
    running as root should do and also what makes it unstubbable from outside --
    so this reaches in and passes the stand-in, exercising the calls and the
    decisions rather than the lookup.

    What matters here is not that the happy path works. It is that every way it
    can fail leaves the old behaviour intact: the real vpnc-script guards both
    its DNS branches on INTERNAL_IP4_DNS, so that variable surviving is the
    whole of "we did not take this over, carry on as before".
    """
    with tempfile.TemporaryDirectory() as d:
        os.chmod(d, 0o700)
        log = os.path.join(d, "calls")
        faildir = os.path.join(d, "fail")
        os.mkdir(faildir)
        stub = os.path.join(d, "resolvectl")
        with open(stub, "w", encoding="utf-8") as fh:
            fh.write("#!/bin/sh\n"
                     f'printf "%s\\n" "$*" >> {log}\n'
                     f'[ -e {faildir}/"$1" ] && exit 1\n'
                     "exit 0\n")
        os.chmod(stub, 0o700)

        def run(env_extra, reason="connect"):
            for path in (log,):
                if os.path.exists(path):
                    os.remove(path)
            env = {
                "reason": reason,
                "TUNDEV": "asuvpn0",
                "INTERNAL_IP4_DNS": "192.0.2.53 192.0.2.54",
                C.DNS_DOMAINS_VAR: "example.com",
            }
            env.update(env_extra)
            notify.handle_dns(env, binary=stub)
            try:
                with open(log, encoding="utf-8") as fh:
                    calls = [line.strip() for line in fh if line.strip()]
            except OSError:
                calls = []
            return calls, env

        calls, env = run({})
        r.check("a connect scopes the link before it gives it any resolver",
                calls[:3] == ["default-route asuvpn0 no",
                              "domain asuvpn0 example.com",
                              "dns asuvpn0 192.0.2.53 192.0.2.54"],
                f"got {calls} — servers before scope would make the tunnel a"
                " candidate for every lookup on the machine, briefly")
        r.check("a link this configured is handed over completely",
                "INTERNAL_IP4_DNS" not in env,
                "the real vpnc-script would still rewrite /etc/resolv.conf,"
                " and lose the change at the next link change")

        calls, env = run({"CISCO_DEF_DOMAIN": "ad.example.com",
                          "CISCO_SPLIT_DNS": "one.example.com,two.example.com"})
        scoped = ("domain asuvpn0 ad.example.com one.example.com"
                  " two.example.com")
        r.check("what the gateway pushes beats what was configured locally",
                scoped in calls,
                f"got {calls} — the gateway is the authority on which names"
                " live behind it")

        calls, env = run({C.DNS_DOMAINS_VAR: ""})
        r.check("a tunnel with no domains at all takes every lookup, not none",
                "default-route asuvpn0 yes" in calls
                and "dns asuvpn0 192.0.2.53 192.0.2.54" in calls,
                f"got {calls} — scoping to nothing would resolve nothing,"
                " which is worse than the behaviour being replaced")

        open(os.path.join(faildir, "domain"), "w", encoding="utf-8").close()
        calls, env = run({})
        r.check("a link that cannot be scoped is reverted, not left half done",
                "revert asuvpn0" in calls,
                f"got {calls} — servers without scope is the one state worse"
                " than doing nothing")
        r.check("a failed handover leaves the real script its DNS job",
                "INTERNAL_IP4_DNS" in env,
                "falling back to the old behaviour is always better than"
                " half-configuring DNS")
        os.remove(os.path.join(faildir, "domain"))

        open(os.path.join(faildir, "dns"), "w", encoding="utf-8").close()
        calls, env = run({})
        r.check("a link whose resolver will not take is reverted too",
                "revert asuvpn0" in calls and "INTERNAL_IP4_DNS" in env,
                f"got {calls}")
        os.remove(os.path.join(faildir, "dns"))

        calls, env = run({}, reason="disconnect")
        r.check("a disconnect reverts the link",
                calls == ["revert asuvpn0"], f"got {calls}")
        r.check("a disconnect never strips INTERNAL_IP4_DNS",
                "INTERNAL_IP4_DNS" in env,
                "a previous connect may have fallen back, and the real"
                " script needs this set to put /etc/resolv.conf back")

        calls, env = run({}, reason="attempt-reconnect")
        r.check("a retry in progress leaves the resolver where it is",
                calls == [],
                f"got {calls} — the link is coming back and reconnect will"
                " configure it again; tearing it down in between would send"
                " tunnel lookups out over the public resolver")

        env = {"reason": "connect", "TUNDEV": "asuvpn0",
               "INTERNAL_IP4_DNS": "192.0.2.53"}
        notify.handle_dns(env, binary=stub)
        r.check("DNS is left alone entirely when the user turned it off",
                "INTERNAL_IP4_DNS" in env and not os.path.exists(log),
                "the helper omits the option, and its absence is the switch")

        calls, env = run({"TUNDEV": "not a device"})
        r.check("a device name that cannot name a device reaches no root command",
                calls == [] and "INTERNAL_IP4_DNS" in env, f"got {calls}")


def check_fatal_reporting(r):
    """A refusal must reach the user as a sentence, not as an exit number.

    Runs the helper on paths that stop before openconnect is ever spawned, so
    nothing here starts a tunnel or touches the real binary. The marker matters:
    the tray used to pick these out by testing whether the line began with
    "refusing", and four of the seven refusals did not, so those arrived as a
    bare "openconnect exited with status N".
    """
    path = os.path.join(HERE, "asuvpn-helper")
    if not os.access(path, os.X_OK):
        r.warn("a refused connect explains itself", f"{path} not executable")
        return
    cases = [
        (["--", "--background"], "an option that would detach openconnect", 25),
        (["--", "--interface", "../../etc/passwd"],
         "an impossible device name", 22),
        # lo exists on every machine this runs on, and refusing to take over
        # a device this session did not create is what keeps `--interface
        # docker0` from having root delete the caller's bridge at teardown.
        (["--", "--interface", "lo"], "a device this session did not create", 23),
    ]
    unmarked = []
    for extra, label, expected_rc in cases:
        try:
            out = subprocess.run(
                [path, "--host", "https://example.invalid",
                 "--fingerprint", "pin-sha256:x", *extra],
                input="not-a-real-cookie\n", capture_output=True,
                text=True, timeout=30)
        except (OSError, subprocess.SubprocessError) as exc:
            r.warn("a refused connect explains itself", f"{label}: {exc}")
            return
        marked = any(line.startswith("[helper] FATAL ")
                     for line in out.stdout.splitlines())
        # Each refusal pinned to its own documented code: accepting either
        # for either would let two exit paths swap meanings unnoticed.
        if not (marked and out.returncode == expected_rc):
            unmarked.append(f"{label}: rc={out.returncode}"
                            f" (expected {expected_rc}) out={out.stdout!r}")
    r.check("a refused connect explains itself rather than exiting a number",
            not unmarked, "; ".join(unmarked))


def check_event_translation(helper, tray, C, r):
    """A datagram in, a state line out — including several hostile ones.

    Built through C.encode_event rather than by hand, so this exercises the
    contract both ends actually use instead of a test's idea of it. An earlier
    version assembled the fields itself and kept passing after the wire format
    gained one.

    Two independent guards are covered. A device name that could not name a
    device is rejected outright, because it is about to be used as a path under
    /sys/class/net. Everything that does get printed is then collapsed to a
    single token, because a newline in any field would end the state line early
    and a space would quietly add a second one.
    """
    token = "a" * 32
    expected_lines = 4
    emitted: list[str] = []
    original = helper.emit
    helper.emit = emitted.append
    server = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
    try:
        with tempfile.TemporaryDirectory() as directory:
            path = os.path.join(directory, "events")
            server.bind(path)
            threading.Thread(target=helper.serve_events,
                             args=(server, token, "asuvpn0", {}),
                             daemon=True).start()
            client = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
            try:
                for datagram in (
                    # discarded: wrong token, too few fields, unmapped reason
                    C.encode_event("wrong-token", {"reason": "connect",
                                                   "TUNDEV": "asuvpn0"}),
                    b"only\ttwo",
                    C.encode_event(token, {"reason": "pre-init",
                                           "TUNDEV": "asuvpn0"}),
                    # a token that decodes to non-ASCII text: discarded, not
                    # a crash. compare_digest raises TypeError on non-ASCII
                    # str input, and one hostile datagram killing the reader
                    # thread would freeze the badge for the tunnel's life —
                    # every event after this one would then go unseen.
                    C.encode_event("töken", {"reason": "connect",
                                             "TUNDEV": "asuvpn0"}),
                    # accepted
                    C.encode_event(token, {"reason": "connect",
                                           "TUNDEV": "asuvpn0",
                                           "INTERNAL_IP4_ADDRESS": "192.0.2.31"}),
                    # a name that cannot be a device: rejected, and the line
                    # falls back to the name we gave openconnect
                    C.encode_event(token, {"reason": "connect",
                                           "TUNDEV": "evil dev\nmore",
                                           "INTERNAL_IP4_ADDRESS": "203.0.113.4"}),
                    # the address is not an interface name, so nothing validates
                    # it — one_token is all that stands between this and both a
                    # forged second line and a forged second field. The space
                    # matters as much as the newline: the tray splits on
                    # whitespace and the last addr= wins.
                    C.encode_event(token, {"reason": "reconnect",
                                           "TUNDEV": "asuvpn0",
                                           "INTERNAL_IP4_ADDRESS":
                                               "203.0.113.99 addr=203.0.113.66\nmore"}),
                    # A resolver list that leads with something that is not an
                    # address. The tray is told this value and then holds the
                    # tunnel to it -- it is the probe target and it is what the
                    # DNS check hunts for on the link -- while what actually
                    # gets installed is whatever asuvpn-notify's split_resolvers
                    # accepted. Reading the variable a second way here is how
                    # the two ends come to disagree, and the tunnel that pays
                    # for it is a working one.
                    C.encode_event(token, {"reason": "connect",
                                           "TUNDEV": "asuvpn0",
                                           "INTERNAL_IP4_ADDRESS": "192.0.2.7",
                                           "INTERNAL_IP4_DNS":
                                               "--listen 192.0.2.53"}),
                ):
                    client.sendto(datagram, path)
            finally:
                client.close()
            # Wait for the expected output rather than sleeping a fixed amount:
            # a loaded machine would otherwise fail this spuriously.
            deadline = time.monotonic() + 10
            while time.monotonic() < deadline:
                if len([ln for ln in emitted
                        if C.decode_message(ln)
                        and C.decode_message(ln)[0] == C.KIND_STATE
                        ]) >= expected_lines:
                    break
                time.sleep(0.05)
    finally:
        server.close()
        time.sleep(0.05)  # let the reader see the closed socket and return
        helper.emit = original

    decoded = [C.decode_message(ln) for ln in emitted]
    accepted = [payload for message in decoded if message
                for kind, payload in [message] if kind == C.KIND_STATE]
    r.check("unauthenticated and malformed events are discarded",
            len(accepted) == expected_lines,
            f"expected {expected_lines} accepted, got {len(accepted)}: {emitted}")
    r.check("an authenticated event becomes a state line",
            any("connected dev=asuvpn0 addr=192.0.2.31" in p for p in accepted),
            f"got {accepted}")
    r.check("a device name that cannot name a device is rejected",
            any(m and "impossible device name" in m[1] for m in decoded)
            and not any("evil" in p for p in accepted),
            f"got {emitted}")
    r.check("the resolver the tray is told is one the link could really hold",
            any("addr=192.0.2.7 dns=192.0.2.53" in p for p in accepted)
            and not any("--listen" in p for p in accepted),
            f"got {accepted} — the tray probes this value and asserts it is"
            " configured on the link, so it has to be the same one"
            " asuvpn-notify would install, not the raw first word")
    r.check("no event can emit more than one line",
            all(ln.count("\n") == 1 and ln.endswith("\n") for ln in emitted),
            f"got {emitted}")
    injected = []
    for payload in accepted:
        _, fields = tray.parse_state_payload(payload)
        # Exactly four words: the state, and one each of dev=, addr= and dns=.
        # Checking the parsed field *names* is not enough — a second "addr="
        # would leave the set unchanged while overwriting the value, because
        # parsing keeps the last occurrence. The word count is what holds.
        if len(payload.split()) != 4 or set(fields) != {"dev", "addr", "dns"}:
            injected.append(payload)
    r.check("no event can inject an extra field into a state line", not injected,
            f"these carry more than one dev and one addr: {injected}")


# ---------------------------------------------------------------------- main


def main():
    parser = argparse.ArgumentParser(
        prog="asuvpn selftest",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        description="Check the applet against the machine it is installed on.",
        epilog="Exit codes: 0 nothing failed, 1 at least one check failed.",
    )
    parser.add_argument("--tier", action="append", choices=TIERS,
                        help="run only this tier (may be repeated)")
    parser.add_argument("--quiet", action="store_true",
                        help="print only failures and warnings")
    args = parser.parse_args()
    tiers = args.tier or list(TIERS)

    r = Results(quiet=args.quiet)
    try:
        C = load("asuvpn_contract", "asuvpn_contract.py")
        helper = load("asuvpn_helper", "asuvpn-helper")
        tray = load("asuvpn_tray", "asuvpn-tray")
        notify = load("asuvpn_notify", "asuvpn-notify")
    except Exception as exc:  # nothing else can run either
        print(f"cannot load the installed programs from {HERE}: {exc}",
              file=sys.stderr)
        return 1

    def guarded(check, *args):
        """One broken check must not hide every result after it.

        A check that raises is itself a failure, recorded as one; the
        remaining checks still run and the summary still prints, so the
        accounting survives a bug in the suite.
        """
        try:
            return check(*args)
        except Exception as exc:
            r.fail(f"{check.__name__} did not finish",
                   f"{exc.__class__.__name__}: {exc} — the check itself"
                   " broke, so everything it would have verified is"
                   " unverified")
            return None

    if "logic" in tiers:
        r.section("logic — this project's own rules, driven in-process")
        guarded(check_contract, C, r)
        guarded(check_option_blocklist, helper, r)
        guarded(check_interface_names, C, r)
        guarded(check_option_parsing, helper, r)
        guarded(check_reason_contract, C, r)
        guarded(check_split_dns, C, r)
        guarded(check_permission_rules, C, r)
        guarded(check_state_payload, tray, r)
        guarded(check_log_scrubbing, tray, r)
        guarded(check_secret_redaction, tray, r)
        guarded(check_log_write_path, tray, r)
        guarded(check_tunnel_health, tray, r)
        guarded(check_demotion_rules, tray, r)
        guarded(check_teardown_rows, tray, r)
        guarded(check_signin_deadline, tray, r)
        guarded(check_rebuild_rules, tray, r)
        guarded(check_signin_races, tray, r)
        guarded(check_start_refusals_keep_promise, tray, r)
        guarded(check_log_rotation, tray, r)
        guarded(check_config_editing, tray, r)
        guarded(check_liveness_probe, tray, r)
        guarded(check_resolver_health, tray, r)
        guarded(check_notifications, tray, r)
        guarded(check_output_framing, helper, r)
        guarded(check_closing_flag, helper, r)
        guarded(check_teardown_ownership, helper, r)

    if "environment" in tiers:
        r.section("environment — our assumptions, put to the installed binaries")
        r.info(f"this is asuvpn {C.VERSION}, contract version"
               f" {C.CONTRACT_VERSION}; every program loaded the one copy"
               " beside itself")
        binary = guarded(check_openconnect, helper, r)
        guarded(check_gives_up, helper, r, binary)
        guarded(check_default_script, helper, C, binary, r)
        guarded(check_message_catalogue, tray, binary, r)
        guarded(check_standin_catalogue, binary, r)
        guarded(check_script_variables, binary, r)
        guarded(check_sso_method, C, r)
        guarded(check_desktop_stack, tray, r)
        guarded(check_installed_permissions, C, r)
        guarded(check_install_scripts, C, r)
        guarded(check_no_bytecode, r)

    if "wiring" in tiers:
        r.section("wiring — asuvpn-notify executed end to end")
        guarded(check_notify_roundtrip, C, r)
        guarded(check_dns_takeover, notify, C, r)
        guarded(check_event_translation, helper, tray, C, r)
        guarded(check_fatal_reporting, r)

    counts = r.counts
    print(f"\n{counts[PASS]} passed, {counts[FAIL]} failed, {counts[WARN]} warnings")
    if counts[FAIL]:
        print("Something above will bite at connect time. See the detail lines.",
              file=sys.stderr)
    return 1 if counts[FAIL] else 0


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