#!/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, and nothing it does can change this machine or
the tunnel. It does reach the network, three times and all read-only. The real
openconnect is run only to describe itself — `--version` for its script path,
`--help` for its retry default. Two probe exercises leave nothing behind: a
loopback connect and a SYN to RFC 5737 documentation space, which must never
answer. And one unauthenticated capability handshake asks the configured
gateway which sign-in method it offers — the same first request any VPN client
makes, carrying no credentials, and skipped rather than failed when the server
cannot be reached. 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 contextlib  # noqa: E402
import io  # 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.SCHEMA_BY_NAME["probe-every"].default
            and values["probe-port"] == C.SCHEMA_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.SCHEMA_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:
        # Ownership, not only mode -- and here rather than in main(), which is
        # where this check used to live *alone*. main() runs long after this
        # function has already executed the file as root, so a contract owned
        # by another user was run and only then refused. An owner can rewrite
        # their own file whatever its mode, so the two questions are one
        # question and both have to be asked before exec_module.
        # PKEXEC_UID and SUDO_UID both name the human behind a root process;
        # sudo was missing, and `sudo asuvpn selftest` is a usage this project
        # anticipates in its own code. Restates C.invoking_uids(), which lives
        # in the file this is about to load.
        trusted = {0}
        for variable in ("PKEXEC_UID", "SUDO_UID"):
            try:
                trusted.add(int(os.environ[variable]))
                break
            except (KeyError, ValueError):
                continue
        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")
            if info.st_uid not in trusted:
                raise SystemExit(
                    f"refusing to load {target}: owned by uid {info.st_uid},"
                    " who is neither root nor the user asking for this")
    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",
        # Abbreviations. getopt_long takes any unambiguous prefix, so an
        # exact-spelling blocklist stopped --background and waved --backg
        # through -- measured against the installed binary, which accepts
        # --backgroun, --backg and --syslo while rejecting --zzzz. This whole
        # list existed to keep openconnect supervised and did not.
        "--backgroun", "--backg", "--b", "--syslo", "--pid-fi",
        "--cookieonl", "--authenticat",
        # Reads more options out of a file, so it defeats any argv check.
        "--config", "--config=/tmp/x",
        # No tun device to own, so teardown has nothing to restore.
        "-S", "--script-tun",
        # An arbitrary program run as root during the trojan phase.
        "--csd-wrapper", "--csd-wrapper=/tmp/x",
    ]
    # --script must survive the abbreviation test even though --script-tun is
    # blocked: chaining a caller's own vpnc-script is a deliberate extension
    # point, and getopt_long itself prefers an exact match to a longer option.
    allow = ["-v", "-i", "--interface", "lo", "--script", "--script=/tmp/s",
             "-s", "-4", "--os=linux-64", "--servercert", "--protocol=anyconnect"]
    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 and takes"
        " --backg for --background, so either can smuggle in a daemonised"
        " openconnect that fork() detaches from PR_SET_PDEATHSIG",
    )
    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)} — --script in particular is an"
        " extension point this helper chains through, not an abbreviation of"
        " --script-tun",
    )


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_credentials_probe_fails_closed(tray, r):
    """A keyring that will not answer must never be read as "ready".

    The applet answers openconnect-sso's TOTP prompt with a blank line, which
    means "not required". The same code path asks for the *password* first
    when the keyring has none, and a blank there would be saved as the
    password and quietly break auto-fill -- so this probe exists to refuse to
    guess, and the branch that matters most is the timeout, which is almost
    always a locked keyring blocking on its unlock dialog.

    None of it was covered: the sandbox's stand-in interpreter answers "ready"
    to any `-c` whatsoever, so every scenario takes the one path that needs no
    decision, and nothing asserted the mapping.
    """
    original_run = tray.subprocess.run
    original_interpreter = tray.sso_interpreter

    class Result:
        def __init__(self, out):
            self.stdout = out

    outcomes = {}
    try:
        tray.sso_interpreter = lambda _path: "/fake/python"
        for label, behaviour in (
                ("ready", lambda *a, **k: Result("ready\n")),
                ("no-password", lambda *a, **k: Result("no-password\n")),
                ("no-credentials", lambda *a, **k: Result("no-credentials\n")),
                ("silence", lambda *a, **k: Result("\n")),
                ("gibberish", lambda *a, **k: Result("what?\n")),
                ("locked", _raise(tray.subprocess.TimeoutExpired("x", 1))),
                ("broken", _raise(OSError("no such interpreter")))):
            tray.subprocess.run = behaviour
            outcomes[label] = tray.probe_credentials("/fake/openconnect-sso")
        tray.sso_interpreter = lambda _path: None
        outcomes["no interpreter"] = tray.probe_credentials("/fake/x")
    finally:
        tray.subprocess.run = original_run
        tray.sso_interpreter = original_interpreter

    r.check("a keyring that will not answer is never read as ready",
            outcomes["locked"] == tray.CREDENTIALS_LOCKED
            and outcomes["silence"] == tray.CREDENTIALS_UNKNOWN
            and outcomes["gibberish"] == tray.CREDENTIALS_UNKNOWN
            and outcomes["broken"] == tray.CREDENTIALS_UNKNOWN
            and outcomes["no interpreter"] == tray.CREDENTIALS_UNKNOWN,
            f"{outcomes} — a timeout is a locked keyring, and treating it as"
            " ready feeds our blank line to the password prompt, where it is"
            " saved as the password")
    r.check("and the answers it does understand are not flattened",
            outcomes["ready"] == tray.CREDENTIALS_READY
            and outcomes["no-password"] == tray.CREDENTIALS_NO_PASSWORD
            and outcomes["no-credentials"] == tray.CREDENTIALS_NONE,
            f"{outcomes} — otherwise the check above is satisfied by a probe"
            " that says UNKNOWN to everything")


def check_who_is_asking(C, r):
    """Root, plus the human behind the root process, however they got there.

    Every loader restates this rule because it lives in the file they are
    about to load, and the helper's own check calls it -- so getting it wrong
    is either a hole or a refusal of legitimate callers. It has been both: it
    knew only PKEXEC_UID, so `sudo asuvpn selftest` was refused against the
    user's own checkout, and so were the CI containers.
    """
    original = {name: os.environ.get(name)
                for name in ("PKEXEC_UID", "SUDO_UID")}
    try:
        for name in original:
            os.environ.pop(name, None)
        bare = C.invoking_uids()
        os.environ["SUDO_UID"] = "4242"
        sudo = C.invoking_uids()
        os.environ["PKEXEC_UID"] = "4343"
        both = C.invoking_uids()
        os.environ["PKEXEC_UID"] = "not a number"
        nonsense = C.invoking_uids()
    finally:
        for name, value in original.items():
            if value is None:
                os.environ.pop(name, None)
            else:
                os.environ[name] = value
    r.check("both ways of becoming root name the human behind it",
            sudo == {0, 4242} and both == {0, 4343},
            f"sudo={sudo} both={both} — pkexec exports PKEXEC_UID and sudo"
            " exports SUDO_UID; knowing only the first refused `sudo asuvpn"
            " selftest` against the user's own checkout")
    r.check("with neither, only root is trusted, and never nobody",
            bare == {0, os.getuid()} and 0 in nonsense,
            f"bare={bare} nonsense={nonsense} — an empty set would refuse"
            " every path, and an unparseable value must not widen the set")


def check_owner_is_a_principal(C, r):
    """A file another user owns is not safe to run as root, however tidy it is.

    The permission bits say who may write a file *besides* its owner. They say
    nothing about who the owner is -- and an owner may rewrite their own file
    whatever the mode. So a 0644 file belonging to somebody else passed this
    gate for as long as it only looked at permissions, and `install.sh --link`
    from a checkout another user owns is a documented way to arrive there. The
    next thing they wrote would run as root at the next connect.

    The uid set is passed in because only the caller knows which human is
    asking: under pkexec that is PKEXEC_UID, not geteuid().
    """
    import os
    import tempfile

    with tempfile.TemporaryDirectory() as directory:
        path = os.path.join(directory, "asuvpn-helper")
        with open(path, "w", encoding="utf-8") as handle:
            handle.write("#!/usr/bin/python3\n")
        os.chmod(path, 0o644)
        mine = os.stat(path).st_uid

        # The set deliberately does not contain 0, though every real caller's
        # does: run as root -- which is how the container jobs run, and how
        # this first failed -- the owner *is* root, so {0, mine + 1} trusts it
        # and the check proved nothing. What is being asserted is the rule,
        # "an owner outside the trusted set is refused", so the fixture has to
        # build a set that genuinely excludes this file's owner whoever that
        # turns out to be.
        r.check("a file owned by an untrusted uid is refused",
                C.unsafe_write_access(path, trusted_uids={mine + 1}) is not None,
                "a 0644 file owned by somebody else is writable by them"
                " whenever they like; running it as root is the whole hazard"
                " this gate exists for")
        r.check("our own file, and root's, are accepted",
                C.unsafe_write_access(path, trusted_uids={0, mine}) is None,
                "the documented design is that the invoking user owns these"
                " files; refusing that would refuse every ordinary install")
        r.check("an unprivileged caller is not asked the ownership question",
                C.unsafe_write_access(path) is None,
                "with no trusted set there is no privilege being granted, so"
                " ownership is not the caller's business")

        os.chmod(path, 0o646)  # nosec B103  # noqa: S103 — the hazard is the fixture
        r.check("world-writable still wins over the ownership test",
                C.unsafe_write_access(path, trusted_uids={0, mine})
                == "world-writable",
                "the older, blunter finding must not be masked by the new one")


def first_routed_device():
    """A device this machine has an IPv4 route through, or None.

    `lo` is not one: it has no entry in /proc/net/route at all. Two checks
    need a device that really carries routes, and asking the kernel is the
    only way to name one that is true on this machine rather than on mine.
    """
    try:
        with open("/proc/net/route", encoding="utf-8") as handle:
            next(handle, "")
            for line in handle:
                fields = line.split()
                if fields:
                    return fields[0]
    except OSError:
        pass
    return None


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(f"{tray.C.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
    # A device that actually has routes, which `lo` need not: it has no entry
    # in /proc/net/route at all, so on a machine with IPv6 disabled its counts
    # are (0, None) and the healthy case failed for a reason that had nothing
    # to do with the code. This suite ships to users and runs in containers,
    # where that is a normal configuration.
    healthy, healthy_index = "lo", lo_index
    counts = tray.route_count("lo")
    if not [count for count in counts if count]:
        routed = first_routed_device()
        if routed:
            index = tray.C.interface_index(routed)
            if index is not None:
                healthy, healthy_index = routed, index
    reason, facts = tray.tunnel_health(healthy, healthy_index)
    r.check("a healthy device is not reported as broken", reason is None,
            f"{healthy} 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}")
    # The one verdict no real device can stage: a tunnel that exists and is
    # administratively down. The sandbox shares the host's network namespace
    # and cannot create devices, so this points the kernel-facts directory at
    # one the check builds. Deleting the IFF_UP branch used to change nothing
    # anywhere in the suite.
    original_sys = tray.C.SYS_CLASS_NET
    with tempfile.TemporaryDirectory() as fake_sys:
        os.makedirs(os.path.join(fake_sys, "asuvpn0"))
        with open(os.path.join(fake_sys, "asuvpn0", "ifindex"), "w",
                  encoding="utf-8") as handle:
            handle.write("42\n")

        def flags(value):
            with open(os.path.join(fake_sys, "asuvpn0", "flags"), "w",
                      encoding="utf-8") as handle:
                handle.write(value + "\n")

        try:
            tray.C.SYS_CLASS_NET = fake_sys
            flags("0x1002")          # IFF_BROADCAST, no IFF_UP
            down, _ = tray.tunnel_health("asuvpn0", 42)
            flags("0x1003")          # the same device, up
            up, _ = tray.tunnel_health("asuvpn0", 42)
        finally:
            tray.C.SYS_CLASS_NET = original_sys
    r.check("a device that is administratively down is a verdict",
            down is not None and "down" in down,
            f"got {down!r} — one of the four things the watchdog is built on,"
            " and no real device can be staged to prove it")
    r.check("the same device, up, is not a verdict on its flags",
            up is None or "down" not in up,
            f"got {up!r} — otherwise the check above would pass against a"
            " predicate that always says down")

    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 IPv4 half, separately, because `lo` has no entry in /proc/net/route
    # at all: the assertion above was carried entirely by lo's two ::1 routes,
    # and blinding the IPv4 branch left the whole suite green. IPv4 is the
    # half that matters most -- `ip route flush dev X` removes only IPv4, and
    # that is the break this watchdog was built for.
    routed = first_routed_device()
    if routed is None:
        r.warn("IPv4 routes are counted, not just IPv6 ones",
               "no device on this machine has an IPv4 route to count")
    else:
        r.check("IPv4 routes are counted, not just IPv6 ones",
                tray.route_count(routed)[0] > 0,
                f"{routed} has an IPv4 route in /proc/net/route and"
                f" route_count says {tray.route_count(routed)}")
    # 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 _raise(exception):
    """A stand-in callable that raises. Named because a lambda cannot."""
    def raiser(*_args, **_kwargs):
        raise exception
    return raiser


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.giveup_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.connecting_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.pipes_closed = []
            self.threads = []
            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

        # The teardown workers. The base class raises NotImplementedError for
        # each, which is right for a class that must not be run half-stubbed
        # -- but a handler that starts one does it in a daemon thread, so the
        # exception lands on stderr *after* the check has already passed. It
        # printed a traceback into the self-check that install.sh runs, on a
        # perfectly good install. A stub has to stub every side effect,
        # including the ones that fail quietly.
        def _disconnect_thread(self):
            self.threads.append("disconnect")

        def _reconnect_thread(self):
            self.threads.append("reconnect")

        def _cancel_tunnel_thread(self):
            self.threads.append("cancel")

        def _quit_thread(self):
            self.threads.append("quit")

        def _close_helper_stdin(self, proc):
            self.pipes_closed.append(proc)

        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_verbs_reach_every_state(tray, r):
    """The states a user verb was silently dropped in, and the ones it was not.

    Two gaps, both found by reading the table rather than the code: RECOVERING
    had no row for connect or reconnect, and DEMOTED none for link-lost. The
    first mattered because the CLI is a second surface with no menu to hide
    the verb -- `asuvpn reconnect --wait` was answered "ok", dropped, and then
    polled a busy state for its whole timeout. The second because a link blip
    on an already-demoted tunnel kept striking through openconnect's own free
    re-establish, and the next rung after a spent nudge is a Duo push.
    """
    Stub = machine_stub(tray)

    recovering = Stub()
    recovering.state = tray.RECOVERING
    recovering.helper_proc = None
    recovering.dispatch(tray.MSG_RECONNECT)
    r.check("reconnect is not dropped while openconnect is re-establishing",
            recovering.signins == [False]
            and not any("ignoring" in line for line in recovering.logged),
            f"signins={recovering.signins} logged={recovering.logged} — the"
            " IPC answers ok either way, so a dropped verb is a --wait that"
            " polls a busy state until it times out")

    demoted = Stub()
    demoted.state = tray.DEMOTED
    demoted.strikes = dict(tray.fresh_strikes(), probe=2)
    demoted.dispatch(tray.MSG_LINK_LOST)
    r.check("a link openconnect is rebuilding stops the strikes, not the badge",
            demoted.state == tray.DEMOTED
            and demoted.strikes == tray.fresh_strikes()
            and demoted.signins == [],
            f"state={demoted.state!r} strikes={demoted.strikes} — promoting"
            " here would be the mistake adopt_only exists to prevent, and"
            " striking through it buys a Duo push for a WiFi blip")

    # FAILED does not always mean "no tunnel": a timed-out teardown lands
    # there still holding one. Connect then meant a browser, a Duo approval,
    # and only afterwards "a tunnel is already running".
    class Live:
        @staticmethod
        def poll():
            return None

    stuck = Stub()
    stuck.state = tray.FAILED
    stuck.helper_proc = Live()
    stuck.dispatch(tray.MSG_CONNECT)
    r.check("connect on a tunnel that is still up closes it first",
            stuck.signins == [] and stuck.state == tray.DISCONNECTING,
            f"signins={stuck.signins} state={stuck.state!r} — signing in from"
            " cold spends a Duo push to arrive at a refusal")

    # And the exit of that old helper, arriving while the sign-in it triggered
    # is still in flight. There is no (FAILED, auth-ok) row, so painting
    # FAILED here threw the completed sign-in away.
    signing = Stub()
    signing.state = tray.AUTHENTICATING
    signing.tunnel_ever_connected = True
    signing.autoreconnect = True
    signing.dispatch(tray.MSG_HELPER_EXITED, status=1,
                     generation=signing.helper_generation)
    r.check("a previous tunnel's exit does not discard the sign-in under way",
            signing.state == tray.AUTHENTICATING and not signing.dropped,
            f"state={signing.state!r} dropped={signing.dropped} — the user's"
            " Duo approval would arrive in FAILED, be logged as ignored, and"
            " the tick would start a second sign-in beside the first")


def check_the_fallback_comes_back(tray, r):
    """The helper's promise about the event channel has to be true.

    When its event thread dies mid-tunnel the helper warns that "the tray will
    fall back to reading openconnect's output". The tray did not: the latch
    that disables log matching was set by the first event and cleared only
    when the next tunnel started. So a tunnel that re-established after the
    channel died had no way to say so — the event could not travel, the
    pattern scan was off, and the watchdog stands aside in RECOVERING — and a
    working tunnel showed "Connecting…" for the rest of the session.
    """
    Stub = machine_stub(tray)
    stub = Stub()
    stub.state_events = True
    stub.dispatch(tray.MSG_WARNING,
                  sentence=f"{tray.C.EVENT_CHANNEL_STOPPED} (closed)")
    r.check("a dead event channel re-arms the log-matching fallback",
            not stub.state_events,
            "the helper tells the user this happens; it has to happen")

    # And an unrelated warning must not turn the fallback on behind a channel
    # that is working — the two sources would then both drive the badge.
    other = Stub()
    other.state_events = True
    other.dispatch(tray.MSG_WARNING, sentence="no default route after teardown")
    r.check("an unrelated warning leaves the event channel trusted",
            other.state_events,
            "log matching is the fallback, not a second opinion")


def check_a_teardown_is_not_a_drop(tray, r):
    """A teardown the user asked for must not become a reconnect.

    The helper usually exits *after* the deadline rather than instead of it,
    and teardown_intent is the only record of why the teardown was happening.
    Clearing it on the timeout meant that late exit arrived as an unexplained
    death: `dropped` was armed, and the next health tick opened a browser and
    raised a Duo push -- for a user who had asked to disconnect and been told
    the disconnect had failed.
    """
    Stub = machine_stub(tray)

    class Live:
        @staticmethod
        def poll():
            return None

    for during in ("disconnect", "reconnect"):
        stub = Stub()
        stub.state = tray.DISCONNECTING
        stub.teardown_intent = during
        stub.helper_proc = Live()
        stub.tunnel_ever_connected = True
        stub.autoreconnect = True
        stub.dispatch(tray.MSG_TEARDOWN_TIMEOUT, during=during)
        # The helper finally dies, after the deadline.
        stub.helper_proc = None
        stub.dispatch(tray.MSG_HELPER_EXITED, status=1,
                      generation=stub.helper_generation)
        r.check(f"a {during} that timed out is still a teardown when it ends",
                not stub.dropped and stub.signins == []
                and stub.state == tray.DISCONNECTED,
                f"during={during} dropped={stub.dropped}"
                f" signins={stub.signins} state={stub.state!r} — arming a"
                " rebuild here signs the user back in after they asked to"
                " disconnect, and after being told it had failed")

    # And Disconnect asked for while a reconnect's teardown is running: the
    # teardown carries on, the sign-in after it does not.
    mid = Stub()
    mid.state = tray.DISCONNECTING
    mid.teardown_intent = "reconnect"
    mid.reconnect_keep_log = True
    mid.dispatch(tray.MSG_DISCONNECT)
    mid.dispatch(tray.MSG_HELPER_EXITED, status=0,
                 generation=mid.helper_generation)
    r.check("disconnect during a reconnect stops after the teardown",
            mid.signins == [] and mid.state == tray.DISCONNECTED,
            f"signins={mid.signins} state={mid.state!r} — the verb was"
            " dropped and the reconnect carried on into a browser and a Duo"
            " push, while the IPC answered ok")


def check_a_human_taking_over_gets_the_budget_back(tray, r):
    """Connect and Reconnect are the same intent and must cost the same.

    Both land on the same teardown-then-sign-in, but only the Connect
    spelling refunded the rebuild budget. So after the ladder had spent its
    three attempts, `asuvpn connect` restored automatic recovery for the next
    tunnel and `asuvpn reconnect` left it exhausted -- and the next demotion
    announced "automatic sign-in gave up" without having tried once.
    """
    Stub = machine_stub(tray)
    spent = {}
    for verb, message in (("connect", tray.MSG_CONNECT),
                          ("reconnect", tray.MSG_RECONNECT)):
        stub = Stub()
        stub.state = tray.DEMOTED
        stub.rebuilds = tray.MAX_REBUILDS
        stub.dropped = True
        stub.giveup_notified = True
        stub.dispatch(message)
        spent[verb] = (stub.rebuilds, stub.dropped, stub.giveup_notified)
    r.check("connect and reconnect hand the rebuild budget back alike",
            spent["connect"] == spent["reconnect"] == (0, False, False),
            f"connect={spent['connect']} reconnect={spent['reconnect']} —"
            " a human is taking over either way, and the ladder keeps its"
            " own count by calling the teardown row directly")


def check_a_demotion_can_always_be_lifted(tray, r):
    """No verdict may outlive the check that can overturn it.

    Only the demoting source promotes -- that is what stops one check clearing
    another's finding. But nothing made that source speak again, so a user who
    read "staying demoted until a check clears" and switched off the check
    being complained about was stuck with the badge until Reconnect.
    """
    Stub = machine_stub(tray)
    original = {name: tray.SETTINGS[name]
                for name in ("dns", "probe", "health-interval")}
    try:
        demoted = Stub()
        demoted.state = tray.DEMOTED
        demoted.demoted_by = "dns"
        demoted.tunnel_dns = "192.0.2.53"
        demoted.tunnel_address = "192.0.2.1"
        tray.SETTINGS["dns"] = True
        r.check("a check that still runs keeps holding its demotion",
                demoted._source_can_speak("dns"),
                "a source that can still speak must not be stood down; that"
                " would clear findings the moment they were made")
        tray.SETTINGS["dns"] = False
        r.check("a check that has been switched off cannot hold one",
                not demoted._source_can_speak("dns"),
                "nothing will ever clear it, so the badge would stay demoted"
                " for the rest of the session")
        demoted._stand_down("dns")
        r.check("standing down lifts the badge without claiming a recovery",
                demoted.state == tray.CONNECTED and demoted.demoted_by is None
                and demoted.notes == [],
                f"state={demoted.state!r} demoted_by={demoted.demoted_by!r}"
                f" notes={demoted.notes} — nothing checked, so nothing may"
                " announce that traffic is flowing again")
        tray.SETTINGS["dns"] = True
        tray.SETTINGS["health-interval"] = 0
        r.check("a watchdog switched off holds no demotions either",
                not demoted._source_can_speak("dns")
                and not demoted._source_can_speak("device"),
                "with nothing being inspected, every source has stopped"
                " speaking, including the one that needs only the kernel")
    finally:
        tray.SETTINGS.update(original)


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)

    class Live:
        """A helper process that has not exited."""

        @staticmethod
        def poll():
            return None

    def tearing_down(helper=True):
        # 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.
        #
        # And a live helper, because that is what a teardown timeout *means*.
        # These fixtures used to leave helper_proc None -- a helper that will
        # not exit, modelled by no helper at all -- which passed only because
        # nothing on the path looked. The row now distinguishes a helper that
        # is still there from one whose exit crossed the deadline, so the
        # difference has to be staged.
        stub = Stub()
        stub.state = tray.DISCONNECTING
        stub.teardown_intent = "reconnect"
        if helper:
            stub.helper_proc = Live()
        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}")

    # What the user does next, after that warning. The timeout leaves FAILED
    # beside a helper handle this still holds, and Disconnect used to answer
    # with a Disconnected badge and no teardown at all -- a claim about a root
    # tunnel that may well still be carrying traffic.
    torn = []
    stuck._disconnect_thread = lambda *a, **k: torn.append(True)
    stuck.dispatch(tray.MSG_DISCONNECT)
    r.check("Disconnect after a stuck teardown tries again, and claims nothing",
            stuck.state == tray.DISCONNECTING and torn == [True]
            and stuck.teardown_intent == "disconnect",
            f"state={stuck.state!r} torn={torn}"
            f" intent={stuck.teardown_intent!r} — the badge said Disconnected"
            " while a root openconnect was still up, and nothing had been"
            " asked to close it")
    # And the ordinary FAILED, where there is no helper: this must still be
    # the plain answer it always was, or every dropped tunnel would start a
    # teardown of nothing.
    gone = Stub()
    gone.state = tray.FAILED
    gone.dropped = True
    gone.rebuilds = 2
    gone.dispatch(tray.MSG_DISCONNECT)
    r.check("Disconnect on a tunnel that is really gone still just says so",
            gone.state == tray.DISCONNECTED and not gone.dropped
            and gone.rebuilds == 0,
            f"state={gone.state!r} dropped={gone.dropped}"
            f" rebuilds={gone.rebuilds} — this is also how a pending rebuild"
            " is called off, and a human taking over must end it")

    # The other side of that: a deadline that expires after the helper has
    # already gone must not report a tunnel that would not close. Event.wait
    # returns the condition's answer rather than the flag, so a set() landing
    # exactly on the timeout still returns False and the timeout is posted for
    # a teardown that finished.
    crossed = tearing_down(helper=False)
    crossed.reconnect_keep_log = True
    crossed.dispatch(tray.MSG_TEARDOWN_TIMEOUT, during="reconnect")
    r.check("a deadline that crossed the helper's exit signs in, not fails",
            crossed.signins == [True] and crossed.state != tray.FAILED,
            f"state={crossed.state!r} signins={crossed.signins} — the"
            " reconnect had nothing left to wait for, and stranding it leaves"
            " no tunnel, no helper and no rebuild owed")

    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.SCHEMA_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}")

    # The other end of the same setting, which nothing bounded at all until
    # now: pkexec's dialog. A polkit agent does not time out, so an
    # unattended rebuild raised a password prompt at three in the morning and
    # sat behind it in CONNECTING -- a state the watchdog does not run in --
    # until somebody came back. Driven through the shipping table, like every
    # other row; the sandbox drives the tick that sends this.
    Stub = machine_stub(tray)

    class Terminable:
        """A helper that is still running, and remembers being asked to stop."""

        def __init__(self):
            self.terminated = False

        def poll(self):
            return None

        def terminate(self):
            self.terminated = True

    stub = Stub()
    stub.state = tray.CONNECTING
    stub.dropped = True          # this was a rebuild: one attempt is spent
    stub.rebuilds = 1
    helper = Terminable()
    stub.helper_proc = helper
    generation = stub.helper_generation
    stub.dispatch(tray.MSG_AUTHORIZE_TIMEOUT)
    # The pipe, not just the signal. helper_spoke does not prove the helper is
    # still unprivileged -- it is set from an idle callback while this row is
    # reached from a timeout, and a ready timeout runs first -- so a password
    # typed at the last moment lands here with a root helper, against which
    # terminate() is EPERM. Closing the control pipe is the only teardown that
    # reaches one, and this was the single path that skipped it.
    r.check("an unanswered authorization closes the pipe, not just signals",
            stub.pipes_closed == [helper] and helper.terminated
            and stub.helper_proc is helper,
            f"closed={stub.pipes_closed} terminated={helper.terminated}"
            f" handle={stub.helper_proc!r} — a root helper survives SIGTERM"
            " from us, and dropping the handle made Disconnect and Quit both"
            " report success over a live tunnel")
    r.check("an unanswered authorization ends, and keeps the rebuild's promise",
            stub.state == tray.FAILED and stub.dropped and stub.rebuilds == 1
            and stub.helper_generation == generation + 1
            and stub.detail.endswith("; rebuilding"),
            f"state={stub.state!r} dropped={stub.dropped}"
            f" rebuilds={stub.rebuilds} detail={stub.detail!r}"
            f" generation={stub.helper_generation} (was {generation}) —"
            " handing the budget back here, as Cancel does, would make this a"
            " loop: prompt, wait, refund, prompt; and without the bump the"
            " kill's own exit paints a bare SIGTERM over the explanation")
    # A tunnel that is up is not in CONNECTING, so the table must drop this
    # rather than tear one down on a timer.
    live = Stub()
    live.state = tray.CONNECTED
    live.dispatch(tray.MSG_AUTHORIZE_TIMEOUT)
    r.check("an authorization deadline cannot reach a tunnel that is up",
            live.state == tray.CONNECTED and live.published == [],
            f"state={live.state!r} published={live.published}")


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_ladder_is_bounded(tray, r):
    """The expensive rung of the ladder is counted, not merely spaced.

    The gap this closes: `_tr_rebuild_dropped` was capped and `_act_ladder`
    was not, so the *cheaper* break -- a tunnel that dies outright -- was
    bounded while the one that costs the same sign-in and lasts longer was
    left to spacing alone. A tunnel that comes up, passes as usable, goes
    bad and does it again bought a Duo push and a polkit dialog every
    autoreconnect-min-gap, for as long as the machine stayed awake, with
    nobody at the keyboard to see it.

    Driven as the flap itself arrives: connect, strike to a demotion, let the
    ladder spend its free nudge and then its expensive one, and do it again.
    Nothing here reaches into the ladder; the laps are what the applet would
    have lived through.
    """
    original_gap = tray.SETTINGS["autoreconnect-min-gap"]
    tray.SETTINGS["autoreconnect-min-gap"] = 300
    try:
        Stub = machine_stub(tray)
        strikes = max(1, tray.SETTINGS["health-strikes"])

        def lap(stub, up_for=0.0):
            """One turn of the flap: a tunnel arrives, then stops carrying.

            `up_for` is how long it lasted before going bad -- the whole
            question the refund asks. The rate limit is wound off because the
            real thing waits it out in wall-clock time, and a fixture that
            waited would test the clock instead of the count.
            """
            stub.state = tray.CONNECTING
            stub._apply_state_event(
                "connected dev=asuvpn0 addr=192.0.2.1 dns=192.0.2.53")
            stub.tunnel_up_since = time.monotonic() - up_for
            stub.last_autoreconnect = float("-inf")
            for _ in range(2 * strikes):
                stub.last_nudge = float("-inf")
                stub.dispatch(tray.MSG_CHECK, source="probe",
                              reason="nothing answers", detail="no reply")

        stub = Stub()
        stub.autoreconnect = True
        for _ in range(tray.MAX_REBUILDS + 3):
            lap(stub)
        r.check("a flapping tunnel spends MAX_REBUILDS sign-ins and no more",
                len(stub.signins) == tray.MAX_REBUILDS,
                f"signins={stub.signins} rebuilds={stub.rebuilds}"
                f" logged={stub.logged} — every one of these is a Duo push and"
                " a password dialog raised at nobody")
        # Said once per incident, like every other verdict on this ladder --
        # this row runs on every health tick for as long as the tunnel stays
        # demoted, and the point of giving up is to stop making noise. A
        # reconnect between laps genuinely is a new incident and may speak
        # again; twenty seconds of the same one may not.
        said = stub.notes.count("VPN still not usable")
        signins = len(stub.signins)
        for _ in range(4 * strikes):
            stub.last_nudge = float("-inf")
            stub.last_autoreconnect = float("-inf")
            stub.dispatch(tray.MSG_CHECK, source="probe",
                          reason="nothing answers", detail="no reply")
        r.check("a spent budget goes quiet instead of repeating itself",
                stub.notes.count("VPN still not usable") == said
                and len(stub.signins) == signins,
                f"notes={stub.notes} signins={stub.signins} — the health tick"
                " keeps arriving after the ladder has nothing left to try")
        # And the ladder's free rung is untouched by any of this: it costs
        # nothing, so a spent budget must not disable it.
        r.check("the free re-establish outlives the budget",
                stub.sent.count(tray.C.CONTROL_RECONNECT)
                == tray.MAX_REBUILDS + 3,
                f"sent={stub.sent} — the nudge asks openconnect to reuse the"
                " session it already has: no sign-in, no push, nothing to ration")

        # The other half of the fix. Nothing on this path exits openconnect
        # the ordinary way, so the exit row's refund is unreachable here -- a
        # cap without one would bar unattended recovery for the rest of the
        # session after three unrelated incidents.
        recovered = Stub()
        recovered.autoreconnect = True
        recovered.rebuilds = tray.MAX_REBUILDS
        lap(recovered, up_for=tray.SETTINGS["autoreconnect-min-gap"] + 1)
        r.check("a tunnel that lasted before going bad earns the budget back",
                len(recovered.signins) == 1 and recovered.rebuilds == 1,
                f"signins={recovered.signins} rebuilds={recovered.rebuilds} —"
                " a session that outlived the gap is a recovery that worked and"
                " later met something new, not a flap")

        # One budget across both recoveries, which is why the ladder reads the
        # counter the dropped-tunnel row writes rather than keeping its own.
        shared = Stub()
        shared.autoreconnect = True
        shared.rebuilds = tray.MAX_REBUILDS - 1  # as if drops had spent them
        lap(shared)
        spent_last = len(shared.signins)
        lap(shared)
        r.check("the two automatic sign-ins draw on one budget",
                spent_last == 1 and len(shared.signins) == 1,
                f"signins={shared.signins} after two laps with"
                f" {tray.MAX_REBUILDS - 1} already spent elsewhere — a tunnel"
                " breaking in both shapes at once is not a reason to be asked"
                " twice as often")

        # Opting out still governs: the cap is a second bound, not a
        # replacement for the setting.
        off = Stub()
        off.autoreconnect = False
        lap(off)
        r.check("automatic sign-in stays off when it is off",
                off.signins == [] and off.rebuilds == 0,
                f"signins={off.signins} rebuilds={off.rebuilds}")
    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}")
            # A config that cannot be read must not be replaced with
            # defaults: `asuvpn autoreconnect on` would then discard the
            # user's server, dpd and everything else, silently.
            #
            # Staged as a symlink to itself, and both halves of that matter.
            # Mode 0 stops nobody when euid is 0, so this whole block used to
            # sit under `if os.geteuid() != 0` while still reporting a plain
            # ok -- and the four portable CI jobs run as root, so the one
            # assertion guarding against losing a user's entire config was
            # green there and untested. A loop is ELOOP for root as well. A
            # directory would also be unreadable, but writing to one fails
            # too, so the check would pass either way and prove nothing; a
            # dangling symlink is a path that *can* be written, which is what
            # makes "was it replaced?" a real question.
            unreadable = parent / "asuvpn-unreadable.conf"
            os.symlink(unreadable.name, unreadable)
            was = tray.CONFIG_FILE
            tray.CONFIG_FILE = unreadable
            try:
                tray.write_setting("autoreconnect", True)
            except OSError:
                pass
            else:
                problems.append("an unreadable config was quietly replaced"
                                " with defaults")
            finally:
                tray.CONFIG_FILE = was
            if not os.path.islink(unreadable):
                problems.append("the refused edit replaced the config anyway")
            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")
    # Silence, which is the probe's whole reason to exist and the one verdict
    # the suite never asserted: `is not True` was as far as it went, so
    # turning the timeout branch into "inconclusive" changed nothing here and
    # disabled black-hole detection with only a five-minute display scenario
    # standing between it and shipping. Staged by making the connect time out
    # rather than by aiming at an address and hoping this network drops it —
    # a network that answers or rejects would otherwise decide the result.
    original_connect = tray.socket.create_connection

    def never_answers(*_args, **_kwargs):
        raise TimeoutError("staged")

    try:
        tray.socket.create_connection = never_answers
        silent, silent_detail = tray.probe_tunnel("192.0.2.1", 53, 3)
    finally:
        tray.socket.create_connection = original_connect
    r.check("silence is a failure, not an inconclusive answer", silent is False,
            f"got alive={silent!r} ({silent_detail}) — a tunnel that carries"
            " nothing is exactly what this check is for, and None would be"
            " discarded by _probe_result without ever striking")

    # And the guard that decides what to do with each verdict. Its twin in
    # the DNS source shipped broken this week -- one inconclusive answer read
    # as a pass, which clears the strike count and can promote a broken
    # tunnel -- so the identical guard here is worth a check of its own.
    class Holder:
        def __init__(self):
            self.probe_in_flight = True
            self.quitting = False
            self.helper_generation = 0
            self.dispatched = []

        def dispatch(self, kind, **data):
            self.dispatched.append((kind, data.get("reason")))

    verdicts = {}
    for label, value in (("alive", True), ("silent", False),
                         ("inconclusive", None)):
        holder = Holder()
        tray.VpnTray._probe_result(holder, value, "detail", 0)
        verdicts[label] = (holder.dispatched, holder.probe_in_flight)
    stale = Holder()
    tray.VpnTray._probe_result(stale, False, "detail", 99)
    r.check("an inconclusive probe is not dispatched as a verdict",
            verdicts["inconclusive"][0] == []
            and len(verdicts["alive"][0]) == 1
            and len(verdicts["silent"][0]) == 1,
            f"{verdicts} — dispatching None reaches the clear path, which"
            " zeroes the strikes and can promote a tunnel that is carrying"
            " nothing")
    r.check("every probe reply clears the in-flight flag, verdict or not",
            not any(flag for _, flag in verdicts.values())
            and not stale.probe_in_flight,
            f"{verdicts} stale={stale.probe_in_flight} — a reply that left it"
            " set would disable probing for the applet's lifetime")
    r.check("a reply about a superseded tunnel changes nothing",
            stale.dispatched == [],
            f"{stale.dispatched}")

    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):
    """Both scrubbers must be wired where the log is written, not merely exist.

    The regex checks above prove the patterns; this proves the bindings. A
    hostile line and a session cookie pushed through the real VpnTray.log must
    both 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.

    The secret half was missing, and its absence was invisible: the docstring
    claimed the binding was proven while only the control-character scrubber
    was pushed through. Deleting the redact_secrets() call from log() left the
    whole suite green and every COOKIE= line openconnect prints landing
    verbatim in a file `asuvpn log` prints to a terminal.

    Runs without GTK too, now that log() does. That matters more than it
    looks: the containers and any ssh session are exactly where this used to
    skip, so on those machines *neither* scrubber had a wiring proof.
    """
    name = "the log file receives scrubbed lines only, and is born 0600"

    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")
            # The other scrubber, through the same one call. Spelled the way
            # openconnect prints it, because that is the line this exists for.
            tray.VpnTray.log(holder, "COOKIE=NOT-A-REAL-COOKIE-example-value")
            # 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 "NOT-A-REAL-COOKIE" in text or "<redacted>" not in text:
        problems.append("a session cookie reached the log file:"
                        f" {text!r} — redact_secrets is not wired to log()")
    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
    original_resolvectl = helper.C.resolvectl_path

    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"

        # The killed path. vpnc-script never ran, so nothing put anything
        # back: deleting the device takes its routes, and the per-link
        # resolver is the one remaining thing this can undo itself.
        helper.interface_exists = lambda name: True
        helper.C.resolvectl_path = lambda: "/fake/resolvectl"
        deleted.clear()
        helper.verify_teardown("asuvpn0", set(), True, 7,
                               dns_on_link=helper.C.DNS_OWNER_LINK)
        revert = ["/fake/resolvectl", "revert", "asuvpn0"]
        delete = ["/fake/ip", "link", "delete", "asuvpn0"]
        ordered = (revert in deleted and delete in deleted
                   and deleted.index(revert) < deleted.index(delete))
        # A device that is not this session's incarnation is not ours to
        # revert either -- the same rule the deletion already follows, for the
        # same reason: the name may belong to somebody else's tunnel by now.
        deleted.clear()
        helper.verify_teardown("asuvpn0", set(), True, 9,
                               dns_on_link=helper.C.DNS_OWNER_LINK)
        foreign_revert = [c for c in deleted if "revert" in c]
        # And when the stock script owned DNS there is nothing to revert and
        # nothing this can fix -- only /etc/resolv.conf, whose backup belongs
        # to the disconnect that never ran. Saying nothing would be the worst
        # outcome: it reads exactly like a clean teardown.
        deleted.clear()
        logged.clear()
        helper.verify_teardown("asuvpn0", set(), True, 7,
                               dns_on_link=helper.C.DNS_OWNER_SCRIPT)
        stock_revert = [c for c in deleted if "revert" in c]
        stock_warned = any(line.startswith("WARNING") and "resolv.conf" in line
                           for line in logged)
        # And the case the was_killed gate used to hide entirely: openconnect
        # dying by somebody else's hand -- an OOM kill, a pkill -9, a crash --
        # with the stock script still holding /etc/resolv.conf. The marker
        # says the disconnect never ran; how it died is beside the point.
        deleted.clear()
        logged.clear()
        helper.verify_teardown("asuvpn0", set(), False, 7,
                               dns_on_link=helper.C.DNS_OWNER_SCRIPT)
        crash_warned = any(line.startswith("WARNING") and "resolv.conf" in line
                           for line in logged)
    finally:
        for name, value in originals.items():
            setattr(helper, name, value)
        helper.C.interface_index = original_index
        helper.C.resolvectl_path = original_resolvectl
    # The proof the deletion rests on. `free_interface_name` picks the first
    # unused asuvpnN, so two sessions starting together can both choose
    # asuvpn0 -- and the loser used to latch the winner's ifindex simply
    # because the name had appeared, then delete a live tunnel belonging to
    # another account, as root. A tun fd's fdinfo names the device it holds,
    # which is the one thing that actually settles ownership.
    with tempfile.TemporaryDirectory() as proc_root:
        for pid, iff in (("100", "asuvpn0"), ("200", "asuvpn1")):
            os.makedirs(os.path.join(proc_root, pid, "fd"))
            os.makedirs(os.path.join(proc_root, pid, "fdinfo"))
            for fd, body in (("3", "pos:\t0\nflags:\t02\n"),
                             ("7", f"pos:\t0\niff:\t{iff}\n")):
                open(os.path.join(proc_root, pid, "fd", fd), "w",
                     encoding="utf-8").close()
                with open(os.path.join(proc_root, pid, "fdinfo", fd), "w",
                          encoding="utf-8") as handle:
                    handle.write(body)
        ours = helper.device_is_held_by(100, "asuvpn0", proc_root)
        theirs = helper.device_is_held_by(200, "asuvpn0", proc_root)
        gone = helper.device_is_held_by(999, "asuvpn0", proc_root)
    r.check("a device is only ours if this openconnect is holding it open",
            ours and not theirs and not gone,
            f"ours={ours} the-other-session's={theirs} no-such-process={gone}"
            " — the name existing says only that somebody created it, and the"
            " loser of a name race deleted the winner's tunnel on that basis")

    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")
    # And the helper's half of that agreement. The marker is removed by the
    # disconnect transition, so anything still here at teardown says that
    # transition did not run -- whatever killed openconnect.
    with tempfile.TemporaryDirectory() as session:
        marker = os.path.join(session, helper.C.DNS_MARKER)
        clean = helper.dns_owner(session)
        with open(marker, "w", encoding="utf-8") as handle:
            handle.write(helper.C.DNS_OWNER_SCRIPT + "\n")
        handed_back = helper.dns_owner(session)
        with open(marker, "w", encoding="utf-8") as handle:
            handle.write(helper.C.DNS_OWNER_LINK + "\n")
        on_link = helper.dns_owner(session)
    r.check("the teardown reads who owns DNS, or says it cannot tell",
            on_link == helper.C.DNS_OWNER_LINK
            and handed_back == helper.C.DNS_OWNER_SCRIPT
            and clean is None and helper.dns_owner(None) is None,
            f"link={on_link!r} script={handed_back!r} clean={clean!r}"
            f" no_channel={helper.dns_owner(None)!r} — the two owners need"
            " different words, and only one of them needs a warning")
    r.check("a killed teardown drops the per-link resolver before the device",
            ordered and not foreign_revert,
            f"ordered={ordered} foreign_revert={foreign_revert} — the resolver"
            " usually dies with the link, but not when the device is"
            " persistent or the delete fails, and those are the runs nobody"
            " is watching")
    r.check("a death we did not cause is reported like one we did",
            crash_warned,
            f"logged={logged} — gating this on was_killed meant an OOM kill"
            " or a pkill -9 left /etc/resolv.conf naming a dead tunnel's"
            " resolvers with nothing said at all")
    r.check("DNS the stock script owns is not claimed back, but is reported",
            not stock_revert and stock_warned,
            f"stock_revert={stock_revert} stock_warned={stock_warned}"
            f" logged={logged} — /etc/resolv.conf is restored from a backup"
            " only the disconnect that never ran would have used, so the one"
            " honest thing left is to say so")
    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.info("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
    # Every say()/print() must hand over a double-quoted literal, and this is
    # what makes the rest of the check mean anything: it reads literals at the
    # call site, so one level of indirection --
    #
    #     banner = "Connected as fake-user on a tunnel that does not exist"
    #     say(banner)
    #
    # sailed straight past it, which is the direct successor of the bug the
    # whole sandbox was built around. The rule is not a style preference: this
    # file exists to imitate another program's output, and an imitation that
    # cannot be read is one nothing is holding to the original.
    # The definitions of say() and note() are not imitation -- they are the
    # plumbing that carries it -- so the line that defines each is skipped.
    speaking = "\n".join(line for line in source.splitlines()
                         if not re.match(r"\s*def (?:say|note)\b", line))
    indirect = re.findall(r'(?:\bsay|\bprint)\((?!f?")([^)\n]*)\)', speaking)
    indirect = [call for call in indirect if call.strip()]
    if indirect:
        r.fail(name, "the stand-in prints something this check cannot read"
               " back to the catalogue: "
               + "; ".join(repr(call) for call in indirect[:5])
               + " — pass double-quoted literals at the call site")
        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.SCHEMA_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
    # Warn, not fail. The two branches above are definite -- the gateway named
    # a method, and it was the wrong one. This is the leftover: a body we do
    # not recognise. That is far more likely to be a load balancer, a rate
    # limit, a captive portal or a truncated read than a real change at the far
    # end, and I watched exactly one such run go red here today against a
    # gateway that was fine a second later. The same reasoning already governs
    # the unreachable case a few lines up; an answer we cannot parse is no more
    # of a verdict than no answer at all, and a check that reddens on its own
    # is one people stop reading.
    r.warn(f"{server} still offers the sign-in method this applet uses",
           f"answered with neither an sso-v2 login URL nor a recognisable"
           f" form, so this run proves nothing either way: {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 tray.C.find_program("pkexec", use_path=True):
        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_pipx_copy_is_the_running_one(C, r):
    """After `pipx upgrade asuvpn`, is the copy that runs the new one?

    The PyPI package is a delivery channel: `asuvpn-install` copies the
    programs out of the wheel's payload into ~/.local/share/asuvpn, and that
    copy is what the launcher, the CLI and the tray all run. `pipx upgrade`
    replaces the payload and nothing else. So an upgrade appears to do
    nothing, silently, and the next bug report is about a bug that was fixed
    two releases ago -- the failure mode with no symptom, which is the kind
    this whole suite exists for.

    Found through the console script rather than by guessing at pipx's
    layout: `asuvpn-install` exists on PATH exactly when the package is
    installed, and its shebang names the interpreter that can import the
    payload. Anything unexpected along the way means only that this question
    cannot be answered here, and an unanswerable question is not a verdict.
    """
    name = "the installed programs are as new as the package they came from"
    # PATH first, then the place pipx puts console scripts. Trusting PATH
    # alone gave a cheerful "nothing to fall behind" to exactly the user this
    # check is for: `~/.local/bin` off PATH is the case install.sh warns about
    # and the README documents, and it is no evidence at all that the package
    # is absent.
    script = C.find_program("asuvpn-install", use_path=True, user_local=True)
    if script is None:
        # info, not ok: this is the answer on nearly every machine, and a
        # pass that cannot fail only pads the total -- which is what info's
        # own docstring says.
        r.info("not installed from PyPI; nothing to fall behind")
        return
    try:
        with open(script, encoding="utf-8", errors="replace") as handle:
            shebang = handle.readline()
    except OSError as exc:
        r.warn(name, f"cannot read {script}: {exc}")
        return
    interpreter = shebang[2:].strip() if shebang.startswith("#!") else ""
    if not interpreter or not os.access(interpreter, os.X_OK):
        r.warn(name, f"{script} does not name an interpreter this can run")
        return
    program = ("from importlib.resources import files;"
               "import runpy, sys;"
               "p = files('asuvpn_dist') / 'payload' / 'asuvpn_contract.py';"
               "sys.stdout.write("
               "runpy.run_path(str(p)).get('VERSION', ''))")
    try:
        # -I, and this is not incidental: without it `-c` puts the working
        # directory first on sys.path, so running this from a checkout made
        # the venv's python import the *checkout's* asuvpn_dist instead of
        # its own -- the answer would then have come from the very tree the
        # question is about. Isolated mode is also what keeps a stray
        # PYTHONPATH out of a question whose whole point is "what does this
        # venv carry".
        out = subprocess.run([interpreter, "-I", "-c", program],
                             capture_output=True, text=True, timeout=30)
    except (OSError, subprocess.SubprocessError) as exc:
        r.warn(name, f"could not ask {interpreter} what it carries: {exc}")
        return
    packaged = out.stdout.strip()
    if out.returncode != 0 or not packaged:
        r.warn(name, "the package is installed but does not carry a readable"
                     f" payload version: {out.stderr.strip()[:200]}")
        return
    # Only one direction is a problem, and the code now agrees with that
    # sentence: it used to warn on any difference, so a checkout install
    # sitting beside an older PyPI one -- what a developer's machine looks
    # like -- was told on every run to run asuvpn-install, which would
    # overwrite the newer copy with the older one.
    def ordered(text):
        """Comparable form, or None when this is not a plain dotted number."""
        parts = str(text).split(".")
        try:
            return tuple(int(part) for part in parts)
        except ValueError:
            return None

    here, there = ordered(C.VERSION), ordered(packaged)
    behind = (here < there) if (here and there) else (packaged != C.VERSION)
    if behind:
        r.warn(name,
               f"this copy is {C.VERSION} and the installed package carries"
               f" {packaged}. `pipx upgrade asuvpn` replaces the package but"
               " not the programs it installed; run `asuvpn-install` to"
               " finish the upgrade")
        return
    r.ok(name, f"this copy is {C.VERSION}, the installed package carries"
               f" {packaged}" + ("" if packaged == C.VERSION else
                                 " — older, so nothing is owed"))



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.

    *Its* criteria, which means the ownership half too. This passed no
    trusted uids, and None is exactly what switches that half off -- so on a
    checkout owned by another user the self-check reported "nothing the helper
    runs as root is writable by anyone else" and the first connect then died
    with exit 26. The whole point of running it now is to say what will happen
    then.
    """
    problems = []
    trusted = C.invoking_uids()
    # 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, trusted_uids=trusted)
        if reason:
            problems.append(f"{path} is {reason}")
    r.check("nothing the helper runs as root is writable or owned by another",
            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:
        # True / False / None, like probe_tunnel. The three are distinct on
        # purpose: the caller may only clear a strike on True, because None
        # means the question could not be asked.
        tray.link_resolvers = lambda device: ["192.0.2.53"]
        healthy, _ = tray.resolver_health("asuvpn0", "192.0.2.53")
        r.check("a link still holding the pushed resolver is healthy",
                healthy is True, f"got {healthy!r}, wanted True")

        tray.link_resolvers = lambda device: []
        healthy, detail = tray.resolver_health("asuvpn0", "192.0.2.53")
        r.check("a link the resolver was taken off is a verdict",
                healthy is False and "no resolvers" in detail,
                f"got {healthy!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"]
        healthy, _ = tray.resolver_health("asuvpn0", "192.0.2.53")
        r.check("a link holding somebody else's resolver is a verdict",
                healthy is False, f"got {healthy!r}, wanted False")

        # 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
        healthy, _ = tray.resolver_health("asuvpn0", "192.0.2.53")
        r.check("an unanswerable question is not a verdict",
                healthy is None,
                f"got {healthy!r}, wanted None — and None must never reach"
                " _check_clear, which would read it as a pass")

        original_conf = tray.resolv_conf_lists
        try:
            tray.link_resolvers = lambda device: []
            tray.resolv_conf_lists = lambda resolver: True
            healthy, _ = tray.resolver_health("asuvpn0", "192.0.2.53")
            r.check("a resolver in force the old way is still in force",
                    healthy is True,
                    f"got {healthy!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: []
        healthy, _ = tray.resolver_health("asuvpn0", "")
        r.check("a VPN that pushed no resolver is not judged for losing one",
                healthy is None, f"got {healthy!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".

    Its notes are captured rather than left to escape. `note()` writes to
    stderr because that is the stream openconnect relays into the log, and it
    is right to be loud in production -- but driving it here put seven lines
    of "resolvectl domain exited 1" and "not touching DNS" on the terminal of
    anyone who had just run install.sh, ahead of "self-check passed". A suite
    that says 0 failed while printing what reads as a stack of errors is worse
    than one that says nothing.
    """
    with contextlib.redirect_stderr(io.StringIO()), \
            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")

        # The call that keeps a split tunnel off the default-route set. Its
        # result used to be discarded, on the grounds that listing a domain
        # implies it -- true for a route-only domain, false for the search
        # domains actually passed, which default the other way. So a resolved
        # too old for SetLinkDefaultRoute, or one transient bus error, left
        # the tunnel's resolvers catching every lookup on the machine while
        # the log said "DNS for example.com".
        open(os.path.join(faildir, "default-route"), "w",
             encoding="utf-8").close()
        calls, env = run({})
        r.check("a link that cannot be kept off the default route is not used",
                "revert asuvpn0" in calls and "INTERNAL_IP4_DNS" in env
                and not any(c.startswith("dns ") for c in calls),
                f"got {calls} — servers on a link that answers for everything"
                " is the state worse than doing nothing")
        os.remove(os.path.join(faildir, "default-route"))

        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"))

        # A reconnect is not a connect, and the difference is the fallback.
        # The stock script's `reconnect` branch runs hooks and nothing else
        # (checked against the installed script), so there is nobody to hand
        # DNS back to -- and reverting would take a working configuration off
        # the link over a transient resolvectl failure, leaving DNS owned by
        # no one on a tunnel that is fine.
        open(os.path.join(faildir, "dns"), "w", encoding="utf-8").close()
        calls, env = run({}, reason="reconnect")
        r.check("a reconnect that half fails keeps what the link already had",
                "revert asuvpn0" not in calls,
                f"got {calls} — the previous configuration was working a"
                " second ago and nothing else will put one back")
        calls, env = run({}, reason="connect")
        r.check("a connect that half fails still reverts and hands back",
                "revert asuvpn0" in calls and "INTERNAL_IP4_DNS" in env,
                f"got {calls} — on a connect the revert *is* the handover")
        os.remove(os.path.join(faildir, "dns"))

        # A full tunnel: every packet goes down it, so leaving every lookup
        # outside the scoped domain with the local network's resolver would
        # send it in clear over the one link the VPN exists to stop using.
        # Detected the way the stock script decides to install a default
        # route, because it is the same question.
        calls, env = run({"INTERNAL_IP4_ADDRESS": "192.0.2.9"})
        r.check("a full tunnel takes every lookup exclusively, not in parallel",
                "default-route asuvpn0 yes" in calls
                and "domain asuvpn0 ~. example.com" in calls,
                f"got {calls} — `default-route yes` alone only adds this link"
                " to the set resolved asks *in parallel*, so the local"
                " network's resolver still sees every unmatched query, in"
                " clear, and can win the race. `~.` is what makes this link"
                " the best match for every name and leaves the others out —"
                " which is what the /etc/resolv.conf rewrite it replaces did")
        calls, env = run({"INTERNAL_IP4_ADDRESS": "192.0.2.9",
                          "CISCO_SPLIT_INC": "1",
                          "CISCO_SPLIT_INC_0_ADDR": "198.51.100.0"})
        r.check("a split tunnel is still scoped, address or no address",
                "default-route asuvpn0 no" in calls,
                f"got {calls} — a split tunnel carries named networks and"
                " has no business answering for every name on the machine")
        calls, env = run({"INTERNAL_IP4_ADDRESS": "192.0.2.9",
                          "CISCO_SPLIT_INC": "1",
                          "CISCO_SPLIT_INC_0_ADDR":
                              "0.0.0.0"})  # nosec B104  # noqa: S104
        r.check("a split list containing the default route is a full tunnel",
                "default-route asuvpn0 yes" in calls,
                f"got {calls} — the stock script reads 0.0.0.0 in that list"
                " as set_default_route, and so must this")
        calls, env = run({"INTERNAL_IP4_ADDRESS": "192.0.2.9",
                          "CISCO_SPLIT_INC": "not a number"})
        r.check("an unreadable split list keeps the narrower behaviour",
                "default-route asuvpn0 no" in calls,
                f"got {calls} — an unexpected value is not evidence of a"
                " full tunnel")

        # Whether DNS ended up on the link is a fact, and the teardown has to
        # read it rather than infer it from what the user asked for. The
        # marker is how the two processes agree; without it a killed teardown
        # reported "the resolver went with its link" on runs where DNS was
        # sitting in /etc/resolv.conf with its backup orphaned.
        marker = os.path.join(d, C.DNS_MARKER)

        def owner():
            try:
                with open(marker, encoding="utf-8") as handle:
                    return handle.read().strip()
            except OSError:
                return None

        socket_env = {C.EVENT_SOCKET_VAR: os.path.join(d, "events.sock")}
        run(socket_env)
        took_over = owner()
        run(dict(socket_env), reason="disconnect")
        cleared = owner()
        r.check("taking DNS over is recorded, and giving it back clears it",
                took_over == C.DNS_OWNER_LINK and cleared is None,
                f"after connect={took_over!r} after disconnect={cleared!r}")
        # And the other outcome, which is the one that needs a warning at
        # teardown. "Nothing recorded" used to mean both "handed back" and
        # "cleanly reverted", so the teardown could not tell a rewritten
        # /etc/resolv.conf from a tunnel that had tidied up after itself.
        open(os.path.join(faildir, "dns"), "w", encoding="utf-8").close()
        run(dict(socket_env))
        os.remove(os.path.join(faildir, "dns"))
        r.check("a handover that failed records who took DNS instead",
                owner() == C.DNS_OWNER_SCRIPT,
                f"recorded {owner()!r} — the stock script now holds"
                " /etc/resolv.conf, and only its own disconnect puts it back")

        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_every_argument_has_a_rule(helper, r):
    """Nothing reaches openconnect's command line without a rule.

    The invariant, rather than three separate checks that each rule works.
    Validation used to be a run of `if`s in main(), and the list of things
    validated drifted from the list of things that reach the argv: --host had
    no rule (and is the element getopt_long permutes into an option), and
    --ac-version still had none after that was found.

    This builds the real command from sentinels and insists every element is
    either a constant the program wrote or a value ARGUMENT_RULES vouched for.
    Adding a field to openconnect_command() without adding its rule breaks
    this check -- by AttributeError if nothing else, because the Namespace
    below is built *from* the table.
    """
    rules = helper.ARGUMENT_RULES
    sentinels = {attribute: f"SENTINEL-{attribute}"
                 for _, attribute, _, _ in rules}
    args = argparse.Namespace(**sentinels)
    extra = ["SENTINEL-extra"]
    try:
        command = helper.openconnect_command("/usr/bin/openconnect", args, extra)
    except AttributeError as exc:
        # The Namespace is built from the table, so a field the command reads
        # but the table does not list is missing from it. That is the failure
        # this check exists for, and it deserves the sentence rather than a
        # traceback.
        r.fail("every value on openconnect's command line has a rule",
               f"the command reads {exc.name!r}, which ARGUMENT_RULES does not"
               " list — a caller-controlled value reaches openconnect with"
               " nothing vouching for it")
        return

    unvouched = []
    for element in command[1:]:          # [0] is the binary this program found
        if element.startswith("--"):
            continue                     # a flag this program wrote itself
        if any(sentinel in element for sentinel in sentinels.values()):
            continue                     # a value the table has a rule for
        if element in extra:
            continue                     # the passthrough, held to the blocklist
        unvouched.append(element)
    # The other direction, which this could not see: the assertion above is
    # "no unvouched element", and deleting an element makes that *more* true.
    # Removing --servercert -- the only authentication of the gateway, since
    # this deployment's chain is not in the system trust store -- left the
    # whole suite green, and lifecycle.sh green with it. So the command's
    # required parts are pinned as parts, not merely permitted.
    required = ("--useragent", "--version-string", "--cookie-on-stdin",
                "--servercert")
    absent = [flag for flag in required if flag not in command]
    pinned = ("--servercert" in command
              and command.index("--servercert") + 1 < len(command)
              and sentinels["fingerprint"]
              in command[command.index("--servercert") + 1])
    r.check("the command still carries what makes it safe to run",
            not absent and pinned,
            f"missing={absent} pin_follows_servercert={pinned} —"
            " --servercert is the only thing authenticating the gateway here,"
            " and --cookie-on-stdin is what keeps the session out of argv and"
            " off an interactive prompt")
    r.check("every value on openconnect's command line has a rule",
            not unvouched,
            f"these reach the argv with nothing vouching for them:"
            f" {unvouched} — the rules are a table so that this list and the"
            " command can be compared; a value here means one was added to"
            " the command without one")

    # And the rules are not decorative: each must actually refuse the shape
    # that made it necessary.
    refused = []
    for flag, attribute, _, _ in rules:
        hostile = argparse.Namespace(**dict(sentinels, **{attribute: "-b"}))
        if helper.unacceptable_argument(hostile) is None:
            refused.append(flag)
    r.check("each rule refuses a value that would become an option",
            not refused,
            f"these accepted '-b': {refused} — openconnect permutes, so any"
            " of them would be read as --background")
    # The contrast case, or a validator that refuses everything would satisfy
    # the line above.
    ordinary = argparse.Namespace(host="https://vpn.example.edu",
                                  fingerprint="pin-sha256:AAAA=",
                                  ac_version=helper.C.AC_VERSION)
    r.check("the values a real sign-in produces are accepted",
            helper.unacceptable_argument(ordinary) is None,
            f"an ordinary connect was refused:"
            f" {helper.unacceptable_argument(ordinary)}")


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
    host, pin = "https://example.invalid", "pin-sha256:x"
    cases = [
        ([host, pin, "--", "--background"],
         "an option that would detach openconnect", 25),
        ([host, pin, "--", "--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.
        ([host, pin, "--", "--interface", "lo"],
         "a device this session did not create", 23),
        # The door beside the blocklist. --host is the last element of
        # openconnect's argv and getopt_long permutes, so these are options to
        # the program that receives them -- the first being the very one the
        # blocklist exists for, and the second re-admitting all of them from a
        # file. Neither passed through unsupported_option(), which only ever
        # saw the arguments after "--".
        # The `=` spelling, deliberately: argparse rejects `--host -b` on its
        # own, because it reads the -b as an option of its own, so the space
        # form never reaches the helper's argv at all and testing it would
        # prove nothing about this check.
        (["-b", pin], "a gateway that is really --background", 28),
        (["--config=/tmp/asuvpn-not-a-real-file", pin],
         "a gateway that is really --config", 28),
        ([host, "-b"], "a certificate pin that is really an option", 28),
    ]
    unmarked = []
    for argv, label, expected_rc in cases:
        try:
            out = subprocess.run(
                [path, f"--host={argv[0]}", f"--fingerprint={argv[1]}",
                 *argv[2:]],
                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))

    # And the other direction, without which the six refusals above would be
    # satisfied by a helper that refuses everything: an ordinary gateway and
    # an ordinary pin must get past these checks and fail later, on the
    # missing cookie.
    try:
        ordinary = subprocess.run(
            [path, "--host", "vpn.example.edu", "--fingerprint",
             "pin-sha256:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="],
            input="\n", capture_output=True, text=True, timeout=30)
        r.check("an ordinary gateway and pin are not caught by those refusals",
                ordinary.returncode == 21,
                f"rc={ordinary.returncode} (expected 21, the missing cookie)"
                f" out={ordinary.stdout!r} — a validator that refuses real"
                " sign-ins would break every connect")
    except (OSError, subprocess.SubprocessError) as exc:
        r.warn("an ordinary gateway and pin are not caught by those refusals",
               str(exc))


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_credentials_probe_fails_closed, tray, r)
        guarded(check_who_is_asking, C, r)
        guarded(check_owner_is_a_principal, C, r)
        guarded(check_tunnel_health, tray, r)
        guarded(check_demotion_rules, tray, r)
        guarded(check_verbs_reach_every_state, tray, r)
        guarded(check_the_fallback_comes_back, tray, r)
        guarded(check_a_teardown_is_not_a_drop, tray, r)
        guarded(check_a_human_taking_over_gets_the_budget_back, tray, r)
        guarded(check_a_demotion_can_always_be_lifted, tray, r)
        guarded(check_teardown_rows, tray, r)
        guarded(check_signin_deadline, tray, r)
        guarded(check_rebuild_rules, tray, r)
        guarded(check_ladder_is_bounded, 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_pipx_copy_is_the_running_one, C, 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_every_argument_has_a_rule, helper, r)
        guarded(check_fatal_reporting, r)

    # Last, and that is the whole point. It used to run at the end of the
    # *environment* tier, before the wiring tier had executed asuvpn-notify
    # and asuvpn-helper as subprocesses -- which is when a stray __pycache__
    # would appear. Deleting sys.dont_write_bytecode from both programs left
    # the suite green and a .pyc behind, reported only by the *next* run, as
    # the previous run's damage.
    guarded(check_no_bytecode, 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())
