#!/usr/bin/python3 -I
"""Root side of the ASU VPN tray app, launched through pkexec.

Run isolated (-I) deliberately. As root, sys.path[0] would otherwise be this
script's own directory under ~/.local, which the unprivileged user owns —
dropping a signal.py or subprocess.py beside this file would get it imported
and executed as root. Isolated mode removes that directory from sys.path and
ignores PYTHON* environment variables. This file imports only the standard
library, so it costs nothing.

Reads the SSO session cookie from the first line of stdin, then runs
openconnect. Afterwards it keeps reading stdin: "quit" or EOF tears the tunnel
down. Tying teardown to the pipe means disconnecting needs no second polkit
prompt, and a crashed tray applet cannot leave a stray root process behind.

openconnect's output is relayed through this process rather than wired straight
to the tray's pipe. That costs a thread but means a dead tray can never hit
openconnect with SIGPIPE, which would kill it before it restored the routing
table and DNS. Writes to a broken pipe are dropped, never raised: teardown has
to survive the tray disappearing, since that is exactly when it matters most.

The cookie is never printed.
"""

import argparse
import ctypes
import importlib.machinery
import importlib.util
import os
import re
import secrets
import shlex
import shutil
import signal
import stat
import subprocess
import sys
import socket
import syslog
import tempfile
import threading
import time
import urllib.parse

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

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

    Not `import`: this runs isolated (-I) so its own directory is deliberately
    not on sys.path — that is what stops a signal.py dropped beside it being
    executed as root. An explicit path bypasses sys.path entirely.

    Verified before it is executed, because running as root out of a directory
    an unprivileged user owns is the whole reason -I is there. The check is
    skipped when unprivileged: loading a file you own, as yourself, crosses no
    boundary, and refusing there would break --link mode from an ordinary
    umask-002 checkout.
    """
    here = os.path.dirname(os.path.abspath(__file__))
    path = os.path.join(here, "asuvpn_contract.py")
    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("asuvpn_contract", path)
    spec = importlib.util.spec_from_loader("asuvpn_contract", loader)
    module = importlib.util.module_from_spec(spec)
    loader.exec_module(module)
    return module


C = _contract()

OPENCONNECT_CANDIDATES = ("/usr/sbin/openconnect", "/usr/bin/openconnect")
PR_SET_PDEATHSIG = 1
PR_GET_PDEATHSIG = 2
TUNNEL_PREFIXES = ("tun", "tap", "vpn")
HERE = os.path.dirname(os.path.abspath(__file__))
NOTIFY = os.path.join(HERE, "asuvpn-notify")
CONTRACT = os.path.join(HERE, "asuvpn_contract.py")
# Root-only, so that nothing else on the machine can inject state events. The
# directory bit is the whole access control: 0700 root:root denies traversal to
# every other uid, and only root could reach the socket inside it anyway.
RUNTIME_ROOT = "/run/asuvpn"

# Options that break supervision rather than extend it. --background makes
# openconnect fork, and fork clears PR_SET_PDEATHSIG — so the daemon outlives
# both the helper and the control pipe, with nothing holding a handle on it,
# while teardown deletes its device out from under it. --syslog diverts the
# progress output the tray reads, so it never sees the tunnel come up.
# Contrast --script, which is a deliberate extension point.
UNSUPPORTED_OPTIONS = (
    "-b", "--background",
    "-l", "--syslog",
    "--pid-file",
    "--cookieonly", "--authenticate",
    # Reads further options out of a file, so it defeats any check made against
    # argv. Everything above could be smuggled back in through it.
    "--config",
    # Replaces the tunnel with a pipe to a program of the caller's choosing;
    # there is then no device for teardown to own and no routing to restore.
    "-S", "--script-tun",
    # Runs an arbitrary program as root during the trojan/CSD phase.
    "--csd-wrapper",
)
# Every long form above, for the prefix test below.
UNSUPPORTED_LONG = tuple(o for o in UNSUPPORTED_OPTIONS if o.startswith("--"))
# Real options that happen to be prefixes of blocked ones, and must survive the
# abbreviation test. getopt_long does the same: an exact match wins even when a
# longer option starts with it, so `--script` reaches openconnect as --script
# and never as an abbreviation of --script-tun. Only --script qualifies today,
# and it has to: chaining a caller's own vpnc-script is a deliberate extension
# point of this helper, not an oversight.
PERMITTED_EXACT = ("--script",)
# getopt_long bundles short options, so -bv is -b -v and slips straight past a
# comparison against whole argv elements -- which is how --background got in
# after all, taking openconnect out of this helper's supervision. Working out
# which letter in a bundle is an option and which is somebody's argument means
# reimplementing getopt against a table that changes between releases, so
# bundles are refused instead and the message says how to rewrite them.
BUNDLED_SHORT_RE = re.compile(r"^-[A-Za-z]{2,}$")

# The same lesson a second time, and it had not been learned: getopt_long also
# accepts any *unambiguous abbreviation* of a long option. Measured against the
# installed binary, `--backgroun`, `--backg` and `--syslo` are all accepted
# while `--zzzz` is rejected -- so an exact-spelling blocklist stopped
# `--background` and waved `--backg` through, and with it a root openconnect
# that fork() detaches from this helper's supervision, PR_SET_PDEATHSIG cleared,
# nothing able to reach it. Abbreviations are refused wholesale rather than
# resolved: deciding whether one is ambiguous means knowing openconnect's whole
# option table, which changes between releases. A user who meant a different
# option can spell it in full.

syslog.openlog("asuvpn-helper", syslog.LOG_PID, syslog.LOG_DAEMON)

# Set once teardown starts, so threads can tell "we are shutting down" from
# "something broke". Without it, closing the event socket on the way out looks
# identical to the socket failing mid-tunnel, and every clean disconnect would
# report a spurious warning.
closing = threading.Event()


# Every caller-controlled value that reaches openconnect's command line, with
# the rule it has to satisfy, in one table.
#
# A table and not a run of `if`s, because a run of `if`s is what was here and
# what let two of these through. --host had no rule at all, and it is the last
# element of the argv -- getopt_long permutes, so `--host=-b` is --background,
# the single option UNSUPPORTED_OPTIONS exists to refuse, and
# `--host=--config=…` re-admits every other one out of a file. The blocklist
# was applied only to the passthrough arguments, so it was guarding one door
# of three. --ac-version still had none after that was fixed, because fixing
# it meant adding two more `if`s rather than asking where the list of things
# to check lived.
#
# openconnect_command() below is the only place the argv is built, so adding
# an element means passing through here; `check_every_argument_has_a_rule` in
# the self-test fails if a value reaches the command line without one.
ARGUMENT_RULES = (
    ("--host", "host", C.valid_gateway, "not a server name"),
    ("--fingerprint", "fingerprint", C.valid_fingerprint,
     "not a certificate pin"),
    ("--ac-version", "ac_version", C.valid_ac_version, "not a client version"),
)


def unacceptable_argument(args):
    """The first caller-controlled value that fails its rule, as a sentence."""
    for flag, attribute, acceptable, why in ARGUMENT_RULES:
        value = getattr(args, attribute)
        if not acceptable(value):
            return f"refusing {flag}: {value!r} is {why}"
    return None


def openconnect_command(binary, args, extra):
    """The one place openconnect's argv is built.

    One place so that "what reaches the command line" and "what has a rule"
    are the same list, and can be checked against each other. `extra` is the
    caller's passthrough, already held to UNSUPPORTED_OPTIONS; everything else
    here is either a constant this program wrote or a value ARGUMENT_RULES
    has accepted.
    """
    return [
        binary,
        "--useragent",
        f"AnyConnect Linux_64 {args.ac_version}",
        "--version-string",
        args.ac_version,
        "--cookie-on-stdin",
        "--servercert",
        args.fingerprint,
        *extra,
        args.host,
    ]


def unsupported_option(option):
    """Why this openconnect option cannot be supervised, or None.

    A function rather than an inline loop so `asuvpn selftest` can put the
    whole blocklist through its paces without spawning anything.
    """
    name = option.split("=", 1)[0]
    if name in UNSUPPORTED_OPTIONS:
        return f"refusing {name}: it would detach openconnect from this helper"
    if name.startswith("--") and len(name) > 2 and name not in PERMITTED_EXACT:
        for blocked in UNSUPPORTED_LONG:
            if blocked.startswith(name):
                return (f"refusing {name}: getopt_long accepts it as an"
                        f" abbreviation of {blocked}, which cannot be"
                        " supervised. Spell the option you meant in full.")
    if BUNDLED_SHORT_RE.match(option):
        return (f"refusing {option}: bundled short options cannot be checked"
                " here, because getopt reads -bv as -b -v. Pass them one at a"
                " time (-i lo, not -ilo) so each can be seen.")
    return None


def vpnc_script_from_binary(openconnect):
    """What openconnect reports as its default script, or None if it did not.

    Separate from default_vpnc_script() so callers can tell "the binary told us"
    from "we fell back" — on Debian the two answers are the same string, which
    makes an equality test useless for the distinction.
    """
    result = run([openconnect, "--version"], timeout=10)
    if result is None:
        return None
    for stream in (result.stdout, result.stderr):
        for line in (stream or "").splitlines():
            head, sep, tail = line.partition(":")
            if sep and head.strip().startswith("Default vpnc-script"):
                path = tail.strip()
                if path:
                    return path
    return None


def default_vpnc_script(openconnect):
    """openconnect's own compiled-in default, asked of the binary.

    `openconnect --version` ends with

        Default vpnc-script (override with --script): /path/to/vpnc-script

    Hardcoding that path is the same mistake as hardcoding log text. Debian and
    Ubuntu use /usr/share/vpnc-scripts/vpnc-script, Fedora and Arch use
    /etc/vpnc/vpnc-script, and a source build can use anything at all. It
    matters because passing --script *replaces* the default: guess it wrong and
    the wrapper execs a path that is not there, sh returns 127, openconnect logs
    "Script ... returned error 127" and carries on -- so the tunnel comes up
    with no routes and no DNS, restores nothing on the way out, and the tray
    reports "Connected" throughout.
    """
    return vpnc_script_from_binary(openconnect) or C.FALLBACK_VPNC_SCRIPT


def prune_stale_channels():
    """Remove channel directories left behind by a helper that was killed.

    The directory name carries the pid that created it, and teardown removes it
    -- but SIGKILL skips teardown, so one is left holding a dead socket every
    time. /run is a tmpfs and these are tiny, but they accumulate until reboot.
    Only entries whose pid is gone are touched: pid reuse means a live one may
    belong to something else entirely.
    """
    try:
        entries = os.listdir(RUNTIME_ROOT)
    except OSError:
        return
    ip = find_ip()
    for name in entries:
        pid = name.partition("-")[0]
        if not pid.isdigit() or os.path.exists(f"/proc/{pid}"):
            continue
        directory = os.path.join(RUNTIME_ROOT, name)
        # The one thing in a stale directory that is not merely litter. A
        # helper killed outright never reached its teardown, so its routing
        # rules are still in the kernel with nothing left that knows about
        # them -- except this file, which is why it is written. Harmless if
        # left (see DESIGN.md: a stale rule is inert or correct, never wrong)
        # but this is the sweep that stops "harmless" accumulating.
        if ip is not None:
            marker = os.path.join(directory, C.REPLY_MARKER)
            if os.path.exists(marker):
                taken = remove_reply_routing(ip, marker)
                if taken:
                    log(f"removed {taken} reply-routing rule(s) left by a"
                        f" helper that did not shut down (pid {pid})")
        shutil.rmtree(directory, ignore_errors=True)


def event_channel():
    """A private datagram socket for script events, or (None, None, None).

    /run when we are root, which is the real case; a 0700 temp directory
    otherwise so the mechanism stays testable without privileges.
    """
    try:
        if os.geteuid() == 0:
            os.makedirs(RUNTIME_ROOT, mode=0o700, exist_ok=True)
            os.chmod(RUNTIME_ROOT, 0o700)
            prune_stale_channels()
            directory = tempfile.mkdtemp(prefix=f"{os.getpid()}-", dir=RUNTIME_ROOT)
        else:
            directory = tempfile.mkdtemp(prefix="asuvpn-")
        os.chmod(directory, 0o700)
        path = os.path.join(directory, "events")
        server = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
        server.bind(path)
        os.chmod(path, 0o600)
        return server, path, directory
    except OSError as exc:
        log(f"no event channel ({exc}); falling back to output matching")
        return None, None, None


def serve_events(server, token, ifname, owned, reply=None):
    """Translate openconnect's script events into state lines for the tray.

    A return from here is one-way: the tray latches on the first real event and
    stops trusting log text, so if this thread dies mid-tunnel no further state
    would ever arrive and the badge would freeze at whatever it last showed.
    Teardown closes the socket deliberately, which is the expected way out and
    says nothing; anything else is worth a line.
    """
    while True:
        try:
            # The contract's size, not a literal: a receive buffer smaller than
            # the datagram silently truncates it, and the cut would land in the
            # last field -- the resolver the probe targets.
            data, _ = server.recvfrom(C.EVENT_MAX_BYTES)
        except OSError as exc:
            if not closing.is_set():
                log(f"WARNING: {C.EVENT_CHANNEL_STOPPED} ({exc});"
                    " the tray will fall back to reading openconnect's output")
            return
        event = C.decode_event(data)
        # Shape and token both checked: a short or unrecognised datagram is
        # discarded rather than indexed into. compare_digest is applied here
        # rather than in the contract, which has no business holding an opinion
        # about how a secret is compared. Compared as bytes, because the str
        # form raises TypeError on non-ASCII input -- and one hostile datagram
        # killing this thread would freeze the badge for the tunnel's life.
        if event is None or not secrets.compare_digest(
                event["token"].encode("utf-8", "replace"), token.encode()):
            continue
        reason = event["reason"]
        device = event["TUNDEV"]
        ip4, ip6 = event["INTERNAL_IP4_ADDRESS"], event["INTERNAL_IP6_ADDRESS"]
        # Through the contract's own parser, not str.split(), because 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 looks for on the link. What
        # actually gets installed there is whatever asuvpn-notify's
        # split_resolvers accepted, so reading the variable a second way here
        # is how the two ends come to disagree -- a gateway whose list leads
        # with something that is not an address would have the tray hunting
        # for it forever, demoting a working tunnel every twenty seconds and
        # walking the ladder to an unattended sign-in.
        resolvers = C.split_resolvers(event["INTERNAL_IP4_DNS"])
        dns = resolvers[0] if resolvers else ""
        state = C.REASON_STATES.get(reason)
        if state is None:
            continue  # pre-init: the tunnel is not configured yet
        address = ip4 or ip6
        # The device name is about to be used as a path component under
        # /sys/class/net, so it is validated with the same rule that guards
        # `ip link delete`. Nothing but root can reach this socket and
        # asuvpn-notify sends what openconnect set, so this is depth rather
        # than a hole being closed — but a name that cannot be a device has no
        # business being turned into a filesystem path either way.
        if device and not C.INTERFACE_RE.match(device):
            log(f"ignoring an event for an impossible device name: {device!r}")
            device = ""
        if state == C.STATE_CONNECTED:
            # The authoritative moment to claim the device: openconnect has just
            # created it, so this ifindex is provably ours.
            index = C.interface_index(device or ifname)
            if index is not None:
                owned["ifindex"] = index
            # A reconnect fires this again, and it may bring a different
            # tunnel address with it -- which would leave the installed rule
            # naming an address that no longer exists. That fails open rather
            # than wrong (a rule matching nothing falls through to main), but
            # failing open is still not working, so the address is compared
            # and the rules are rebuilt when it has moved.
            if reply is not None and address \
                    and reply.get("address") != address:
                if reply.get("installed"):
                    remove_reply_routing(reply["ip"], reply["marker"])
                    reply["installed"] = False
                reply["address"] = address
                threading.Thread(
                    target=reply_routing_install,
                    args=(reply["ip"], reply["snapshot"],
                          device or ifname, reply["marker"], reply,
                          reply["families"]),
                    daemon=True).start()
        # Sanitised here rather than trusted from the sender. asuvpn-notify
        # already strips separators, but the rule has to hold where the line is
        # built: a newline in either field ends the line early and the rest
        # arrives as a second, unprefixed message, while a space silently
        # becomes an extra field. Caught by `asuvpn selftest`.
        emit(C.encode_message(C.KIND_STATE,
                              f"{state} dev={C.one_token(device or ifname)}"
                              f" addr={C.one_token(address)}"
                              f" dns={C.one_token(dns)}"))


def emit(text):
    """Write to the tray's log pipe, tolerating a tray that has gone away."""
    try:
        sys.stdout.write(text)
        sys.stdout.flush()
    except (BrokenPipeError, ValueError, OSError):
        pass


def fatal(message):
    """A refusal that stops the tunnel before openconnect is ever started.

    Marked, not merely worded. The tray shows this sentence in place of
    "openconnect exited with status 24", and it used to find it by testing
    whether the line began with "refusing" — which four of the seven refusals
    did not, so those reached the user as a bare number with the explanation
    sitting unread in the log.
    """
    emit(C.encode_message(C.KIND_FATAL, message))
    try:
        syslog.syslog(syslog.LOG_ERR, message)
    except (OSError, ValueError):
        pass


def log(message):
    kind = C.KIND_WARNING if message.startswith("WARNING") else C.KIND_NOTE
    emit(C.encode_message(kind, message.removeprefix("WARNING: ")))
    # Also to syslog, because the tray's pipe is broken in exactly the case the
    # teardown warnings exist for — a crashed or killed applet. Without this,
    # "no default route after teardown" is written to a pipe nobody is reading
    # and the user is never told their network is broken.
    try:
        level = syslog.LOG_WARNING if message.startswith("WARNING") else syslog.LOG_INFO
        syslog.syslog(level, message)
    except (OSError, ValueError):
        pass


def die_with_parent(parent_pid):
    """Ask the kernel to signal openconnect if this helper dies.

    The control pipe protects against the *tray* dying. This is the other
    direction: if the helper is killed — OOM, an unhandled error — openconnect
    would keep running as root with nothing able to reach it, while the tray
    reports the connection as dropped. SIGINT makes it tear down through
    vpnc-script, so the routing table is still restored.

    Runs in the child between fork and exec. That is only safe because the
    helper has not started any threads yet at this point.
    """
    # Both failure modes here were silent: a libc that will not load, and a
    # prctl that returns -1. Either leaves this function doing nothing at all
    # while the caller believes the guard is armed -- and the guard is the only
    # thing stopping a root openconnect outliving its supervisor. A silent
    # no-op on a safety mechanism is the worst shape a bug can have, so both
    # now say so on stderr, which the helper relays into the session log.
    #
    # os.write rather than log(): this runs in the child between fork and exec,
    # where the helper's logging machinery is not safe to touch.
    try:
        libc = ctypes.CDLL("libc.so.6", use_errno=True)
        armed = libc.prctl(PR_SET_PDEATHSIG, signal.SIGINT)
    except (OSError, AttributeError) as exc:
        os.write(2, f"asuvpn: cannot arm parent-death signal ({exc});"
                    " a helper killed outright would leave openconnect"
                    " running as root\n".encode())
        return
    if armed != 0:
        os.write(2, f"asuvpn: prctl(PR_SET_PDEATHSIG) failed with errno"
                    f" {ctypes.get_errno()}; a helper killed outright would"
                    " leave openconnect running as root\n".encode())
        return
    # Read it back. A zero return says the call was accepted, which is not the
    # same as the guard being armed -- and this is a safety mechanism whose
    # failure is invisible until the day it matters, so "the exit status said
    # fine" is not good enough. PR_GET_PDEATHSIG reports what is actually set.
    check = ctypes.c_int(0)
    if libc.prctl(PR_GET_PDEATHSIG, ctypes.byref(check)) != 0 \
            or check.value != int(signal.SIGINT):
        os.write(2, f"asuvpn: parent-death signal reads back as {check.value},"
                    f" not SIGINT ({int(signal.SIGINT)}); the guard is not"
                    " armed and a helper killed outright would leave"
                    " openconnect running as root\n".encode())
        return
    # The flag is only armed now, so a parent that died during the fork would
    # never be noticed. Comparing against the recorded pid rather than testing
    # for init: under systemd an orphan reparents to the nearest subreaper
    # (systemd --user), never to pid 1, so that test silently never fired.
    if os.getppid() != parent_pid:
        os._exit(1)


def find_openconnect():
    """Locate openconnect. There is deliberately no way to override this.

    An earlier version took the path as an argument, which meant anyone who
    could satisfy the polkit prompt had a one-line root exec of any file. The
    PATH fallback is safe because pkexec resets PATH to a sanitised value.
    """
    return C.find_program("openconnect", OPENCONNECT_CANDIDATES, use_path=True)


def find_ip():
    """The absolute paths only: this runs as root and `ip` deletes devices.

    The list moved to the contract when reply routing became a second caller
    of it. Two copies of "where does ip live" is two chances to disagree about
    whether this machine has one.
    """
    return C.ip_path()


def requested_interface(extra):
    """The device name the caller chose for openconnect, if any.

    Takes the *last* occurrence, because that is the one getopt gives
    openconnect. Taking the first would let `--interface docker0 --interface x`
    point teardown at a device openconnect never touched.
    """
    chosen = None
    for index, arg in enumerate(extra):
        if arg in ("--interface", "-i") and index + 1 < len(extra):
            chosen = extra[index + 1]
        elif arg.startswith("--interface="):
            chosen = arg.split("=", 1)[1]
    return chosen


def requested_dpd(extra):
    """Whether the caller set their own DPD interval, so ours is not imposed.

    Last occurrence, like the other option readers: getopt gives openconnect
    the last one, and a reader with different semantics from its siblings is a
    trap for whoever starts using the value instead of its presence.
    """
    chosen = None
    for index, arg in enumerate(extra):
        if arg == "--force-dpd" and index + 1 < len(extra):
            chosen = extra[index + 1]
        elif arg.startswith("--force-dpd="):
            chosen = arg.split("=", 1)[1]
    return chosen


def requested_script(extra):
    """The caller's --script value, so the wrapper can chain to it."""
    chosen = None
    for index, arg in enumerate(extra):
        if arg in ("--script", "-s") and index + 1 < len(extra):
            chosen = extra[index + 1]
        elif arg.startswith("--script="):
            chosen = arg.split("=", 1)[1]
    return chosen


def strip_script_option(extra):
    """Drop the caller's --script; the wrapper takes its place and chains to it."""
    skip = False
    for index, arg in enumerate(extra):
        if skip:
            skip = False
            continue
        if arg in ("--script", "-s"):
            skip = index + 1 < len(extra)
            continue
        if arg.startswith("--script="):
            continue
        yield arg


def free_interface_name():
    """First unused asuvpnN.

    Deliberately not a fixed name. With a constant, a second user on the same
    machine would find the first user's live device, read it as a leftover, and
    delete it as root — taking down a tunnel that was never theirs. Picking a
    free name means no device ever has to be removed to make room.
    """
    for index in range(100):
        name = f"{C.INTERFACE_PREFIX}{index}"
        if not interface_exists(name):
            return name
    return None


def device_is_held_by(pid, ifname, proc_root="/proc"):
    """Does this process hold the tun device `ifname` open?

    The kernel answers it: a tun file descriptor's fdinfo carries `iff:` and
    the device's name. That is proof of ownership, which nothing else here
    has -- a name existing says only that *somebody* created it.

    False on anything unreadable. The caller treats that as "not proven",
    which is the safe direction: an unclaimed device is left alone at
    teardown, and leaving a stray device behind is a smaller harm than
    deleting somebody else's live tunnel as root.
    """
    fd_dir = os.path.join(proc_root, str(pid), "fd")
    try:
        entries = os.listdir(fd_dir)
    except OSError:
        return False
    for entry in entries:
        try:
            with open(os.path.join(proc_root, str(pid), "fdinfo", entry),
                      encoding="utf-8", errors="replace") as handle:
                for line in handle:
                    if line.startswith("iff:") and line.split()[-1] == ifname:
                        return True
        except (OSError, IndexError):
            continue
    return False


def watch_for_interface(ifname, holder, pid, deadline=60):
    """Record the ifindex of our device, once we can prove it is ours.

    The proof is the point. This used to record whatever ifindex appeared at
    the name, which is not the same claim: `free_interface_name` picks the
    first unused asuvpnN, so two sessions starting together can both choose
    asuvpn0 -- and the loser, whose openconnect then failed to create it,
    latched the *winner's* ifindex and deleted a live tunnel belonging to
    another account at teardown, as root. The ifindex comparison in
    verify_teardown was supposed to prevent exactly that and could not,
    because the number it compared had come from here.

    The event channel claims the device authoritatively when it is available
    (serve_events, on STATE_CONNECTED). This is the fallback for a session
    running on log matching, and it now waits for the same standard of
    evidence rather than a weaker one.
    """
    end = time.monotonic() + deadline
    while time.monotonic() < end:
        index = C.interface_index(ifname)
        if index is not None and device_is_held_by(pid, ifname):
            holder["ifindex"] = index
            return
        time.sleep(0.25)


def run(command, timeout=5):
    # stdin is /dev/null, never inherited. This process's stdin is the control
    # pipe: a child that reads it steals the "quit" line teardown depends on,
    # and one that merely blocks on it hangs the helper. Nothing here has any
    # business reading input.
    try:
        return subprocess.run(
            command,
            stdin=subprocess.DEVNULL,
            capture_output=True,
            text=True,
            timeout=timeout,
        )
    except (OSError, subprocess.SubprocessError):
        return None


def tunnel_interfaces():
    try:
        names = os.listdir(C.SYS_CLASS_NET)
        return {n for n in names if n.startswith(TUNNEL_PREFIXES)}
    except OSError:
        return set()


def interface_exists(name):
    """Ask about one device by name.

    Not via tunnel_interfaces(): our own device is called asuvpn0, and a
    caller-supplied --interface can be called anything, so neither is guaranteed
    to match the tun/tap/vpn prefixes used for spotting strays.
    """
    return bool(name) and os.path.exists(f"{C.SYS_CLASS_NET}/{name}")


def dns_owner(event_dir):
    """Who still owns DNS, as asuvpn-notify last recorded it.

    C.DNS_OWNER_LINK, C.DNS_OWNER_SCRIPT, or None -- and None is the good
    answer, because a clean disconnect removes the file. So finding anything
    here at teardown is evidence that the disconnect transition never ran,
    which is the question this actually needs answered.

    It used to ask "did *we* SIGKILL it", which is a different and narrower
    question: an OOM kill, an administrator's `pkill -9` or a crash all leave
    that false, and the whole DNS half of the teardown was skipped -- no
    revert, and no warning about an /etc/resolv.conf still naming resolvers
    that are now unreachable.
    """
    if not event_dir:
        return None
    try:
        with open(os.path.join(event_dir, C.DNS_MARKER), encoding="utf-8") as fh:
            return fh.read().strip() or None
    except OSError:
        return None


def drop_link_dns(ifname):
    """Take our per-link DNS off a link openconnect did not live to clear.

    Quiet about the ordinary outcome and loud about the odd one: reverting a
    link that is already gone is the common case and means nothing, while a
    revert that fails leaves a resolver pointing into a dead tunnel and is
    worth a line.
    """
    binary = C.resolvectl_path()
    if binary is None:
        return
    result = run([binary, "revert", ifname], timeout=C.RESOLVECTL_TIMEOUT)
    if result is not None and result.returncode == 0:
        log(f"per-link DNS on {ifname} reverted")



# --------------------------------------------------------------- reply routing
#
# Replies leave by the interface their request arrived on. Everything here is
# an *addition* above the main routing table, which is never touched: the stock
# vpnc-script still configures routing exactly as it would have. See
# "Who owns the return path" in DESIGN.md for why this is needed and why it is
# shaped this way.


def ip_json(ip, *argv):
    """One `ip -j` query, parsed. [] on any failure, and it never raises.

    Structured output rather than scraped: a column that moves is a bug nobody
    sees until the day it moves.
    """
    import json

    try:
        done = subprocess.run([ip, "-j", *argv], stdin=subprocess.DEVNULL,
                              stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
                              timeout=C.IP_TIMEOUT, check=False)
    except (OSError, ValueError, subprocess.SubprocessError):
        return []
    if done.returncode != 0:
        return []
    try:
        parsed = json.loads(done.stdout or b"[]")
    except ValueError:
        return []
    return parsed if isinstance(parsed, list) else []


def run_ip(ip, *argv, quiet=False):
    """One ip command. True if it exited 0; never raises.

    `quiet` is for the rollback and the teardown, which run against state that
    is usually already gone -- "No such process" on every clean disconnect is
    noise that teaches the reader to skim these lines.
    """
    try:
        done = subprocess.run([ip, *argv], stdin=subprocess.DEVNULL,
                              stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
                              timeout=C.IP_TIMEOUT, check=False)
    except (OSError, ValueError, subprocess.SubprocessError) as exc:
        if not quiet:
            log(f"reply routing: ip {argv[0]} failed: {exc}")
        return False
    if done.returncode != 0:
        if not quiet:
            detail = done.stdout.decode("utf-8", "replace").strip().replace("\n", "; ")
            log(f"reply routing: ip {' '.join(argv)} exited"
                f" {done.returncode}: {detail}")
        return False
    return True


def has_carrier(link):
    """Is this link actually passing traffic?

    `operstate` is the obvious field and it is the wrong one: a tun device
    reports UNKNOWN for its whole life, so the obvious test excludes the one
    device the tunnel half of this depends on while reading perfectly
    sensibly. LOWER_UP is the flag that means what operstate looks like it
    means.
    """
    flags = link.get("flags") or []
    return "UP" in flags and "LOWER_UP" in flags and "LOOPBACK" not in flags


def source_addresses(ip, device, family):
    """Addresses on `device` that could be the source of a reply.

    Global scope only: a host or link-scope address never appears as the source
    of a reply to somebody off this machine. Tentative and deprecated IPv6
    addresses are skipped because the kernel will not choose them for new
    traffic either, so a rule naming one would match nothing.
    """
    out = []
    for link in ip_json(ip, "-f", family, "addr", "show", "dev", device):
        for info in link.get("addr_info") or []:
            if info.get("scope") != "global":
                continue
            if info.get("tentative") or info.get("deprecated"):
                continue
            local = info.get("local")
            if not local:
                continue
            try:
                import ipaddress
                text = str(ipaddress.ip_address(local))
            except ValueError:
                continue  # about to be an argv to a root command
            if text not in out:
                out.append(text)
    return out[:C.MAX_REPLY_ADDRESSES]


def uplink_snapshot(ip, families=("inet", "inet6")):
    """Which interfaces reach the world, and the routes they do it with.

    Taken **before** openconnect is launched, and that is not merely tidy: in a
    full tunnel the stock script replaces the default route, so this is the only
    moment the pre-tunnel value exists. In a split tunnel it survives in the
    main table and could be read at any time; one path is taken in both cases
    rather than two with one of them rarely exercised.

    An interface is an uplink if the main table has a default route through it.
    Not its name, not its driver, not "the one that is not the tunnel" -- that
    property is precisely the one that makes an interface able to answer a peer
    anywhere on the internet, which is the question being asked. A machine with
    ethernet and wifi gets both, and a machine with neither gets nothing, with
    no special case for either.
    """
    snapshot = {}
    carrier = {link.get("ifname"): has_carrier(link)
               for link in ip_json(ip, "link", "show")}
    for family in families:
        for route in ip_json(ip, "-f", family, "route", "show", "default"):
            device = route.get("dev")
            if not device or not C.INTERFACE_RE.match(device):
                continue
            if not carrier.get(device):
                continue
            # Every route the device owns, not just the default. A table
            # carrying only a default *shadows* the more-specific routes in
            # main -- an on-link neighbour would be sent to the gateway
            # instead of straight out the wire. This is the one thing an
            # implementation can get wrong in a way that matters.
            # `dev` is *absent* from every route here: asking by device
            # makes ip drop the field it would otherwise print, because the
            # query implies it. Put it back, or every route fails to render
            # and the table is built empty -- which is how this first shipped
            # a rule for the tunnel, none for the uplink, and a log line
            # saying it had worked.
            routes = [{**r, "dev": device}
                      for r in ip_json(ip, "-f", family, "route",
                                       "show", "dev", device)
                      if r.get("dst")]
            if routes:
                snapshot.setdefault(device, {})[family] = routes
    return snapshot


# Scope values ip(8) accepts. Compared against rather than passed through, for
# the usual reason: this becomes an argument to a command running as root.
ROUTE_SCOPES = ("global", "link", "host", "site", "nowhere")


def route_args(route):
    """One route from `ip -j` rendered back as `ip route add` arguments.

    None if anything in it is not what it claims to be. Every field is checked
    rather than passed through: these came from a parser reading the machine's
    own state, but they are about to be arguments to a root command, and the
    rule in this project is that such a check lives where the call is.

    `protocol` is deliberately dropped. Re-adding a kernel-proto route as
    kernel-proto is a claim the kernel put it there, which would be false.
    """
    import ipaddress

    dst = route.get("dst")
    if not dst:
        return None
    if dst != "default":
        try:
            dst = str(ipaddress.ip_network(dst, strict=False))
        except ValueError:
            return None
    device = route.get("dev")
    if not device or not C.INTERFACE_RE.match(device):
        return None
    args = [dst]
    gateway = route.get("gateway")
    if gateway:
        try:
            args += ["via", str(ipaddress.ip_address(gateway))]
        except ValueError:
            return None
    args += ["dev", device]
    scope = route.get("scope")
    if scope:
        if scope not in ROUTE_SCOPES:
            return None
        args += ["scope", scope]
    source = route.get("prefsrc")
    if source:
        try:
            args += ["src", str(ipaddress.ip_address(source))]
        except ValueError:
            return None
    metric = route.get("metric")
    if metric is not None:
        if not isinstance(metric, int) or not 0 <= metric < 2 ** 32:
            return None
        args += ["metric", str(metric)]
    return args


# Candidates for the default route's probe, in documentation space (RFC 5737
# and RFC 3849). Nothing is ever sent to them -- this is a routing question,
# not a packet -- and a fixture that is never somebody's machine is the rule
# everywhere else in this project too. Several, because the *first* one is only
# a probe of the default route on a machine that has no more specific route
# covering it, and this cannot know that in advance: it asks.
DEFAULT_PROBES = ("192.0.2.1", "198.51.100.1", "203.0.113.1")
DEFAULT_PROBES6 = ("2001:db8::1", "2001:db8:1::1", "2001:db8:2::1")


def probe_targets(ip, routes, device, family):
    """One destination per route, to ask the kernel where it would go.

    The point of the check these feed is narrow: installing a rule must not
    move anything that was already going somewhere sensible. So the probes are
    the machine's own routes, asked about one at a time.

    The default route's probe is **chosen by asking**, not fixed. A hardcoded
    address is only a probe of the default route while nothing more specific
    covers it -- and on a machine that does route documentation space somewhere
    (a lab, a blackhole route) the fixed address would quietly test a different
    route and the default's own behaviour would go unchecked. So candidates are
    tried until one currently resolves through the device whose table is about
    to be copied, and if none does, the default simply contributes no probe
    rather than a misleading one.
    """
    import ipaddress

    out = []
    for route in routes:
        dst = route.get("dst")
        if dst == "default":
            # A candidate inside one of the device's *own* prefixes resolves
            # through the device by the on-link route, not the default -- it
            # would look like a valid probe and test the wrong thing. The
            # prefixes are right here, so they are checked against.
            covered = []
            for other in routes:
                other_dst = other.get("dst")
                if not other_dst or other_dst == "default":
                    continue
                try:
                    covered.append(ipaddress.ip_network(other_dst, strict=False))
                except ValueError:
                    continue
            pool = DEFAULT_PROBES6 if family == "inet6" else DEFAULT_PROBES
            for candidate in pool:
                address = ipaddress.ip_address(candidate)
                if any(address.version == net.version and address in net
                       for net in covered):
                    continue
                answers = ip_json(ip, "-f", family, "route", "get", candidate)
                if answers and answers[0].get("dev") == device:
                    out.append(candidate)
                    break
            continue
        try:
            out.append(str(ipaddress.ip_network(dst, strict=False).network_address))
        except (ValueError, TypeError):
            continue
    return out


def where_would_it_go(ip, destination, source, family):
    """How the kernel answers "from here, to there" right now, as one string.

    Asked **with** a source, which is the whole point. `ip route get` without
    one issues a single lookup, and the behaviour this guards against only
    appears on the second: with the source unspecified the kernel resolves a
    route, takes the source it implies, and looks up again with that source
    set -- and it is that second lookup the rules are consulted on. A check
    that omitted `from` would pass on exactly the configuration it exists to
    catch.
    """
    answers = ip_json(ip, "-f", family, "route", "get", destination,
                      "from", source)
    if not answers:
        return ""
    first = answers[0]
    return " ".join(str(first.get(k, "")) for k in ("dev", "gateway", "prefsrc"))


def table_is_free(ip, table):
    """Nothing lives in this table, in either family.

    Checked rather than assumed. A table id already carrying somebody else's
    routes is the one place a wrong guess here would quietly steal another
    tool's routing, so it is refused rather than reused.
    """
    for family in ("inet", "inet6"):
        if ip_json(ip, "-f", family, "route", "show", "table", str(table)):
            return False
    return True


def priority_is_free(ip, priority):
    """No rule already sits at this priority, in either family."""
    for family in ("inet", "inet6"):
        for rule in ip_json(ip, "-f", family, "rule", "show"):
            if rule.get("priority") == priority:
                return False
    return True


def device_routes(ip, device, family):
    """Every route the main table has through this device, as installed.

    `dev` is *absent* from each route when ip is asked by device -- the query
    implies it -- so it is put back, or the route cannot be rendered later and
    the table is built empty. Read live, never assumed: whatever shape the
    gateway's own routes have is the shape a copy of them has to take.
    """
    return [{**route, "dev": device}
            for route in ip_json(ip, "-f", family, "route", "show",
                                 "dev", device)
            if route.get("dst")]


def tunnel_catch_all(ip, device, family):
    """A catch-all for the tunnel table, shaped like the tunnel's own routes.

    The tunnel table needs one route that matches everything, because its whole
    job is to catch replies to peers the main table would have sent elsewhere --
    copying the gateway's pushed prefixes would catch only the peers already
    going the right way.

    What that route looks like is **read off the device, not assumed**. A tun
    device is point-to-point and its routes carry no gateway, so `dev` alone is
    the correct form -- but that is a property of this tunnel, not a law, and a
    gateway that does push a next hop would make `dev` alone wrong. So the
    device is asked, and whatever shape its own routes have is the shape this
    takes.
    """
    for route in ip_json(ip, "-f", family, "route", "show", "dev", device):
        gateway = route.get("gateway")
        if gateway:
            return {"dst": "default", "dev": device, "gateway": gateway}
    return {"dst": "default", "dev": device}


def uplink_routes_now(ip, device, family, snapshotted):
    """The uplink's routes as they are *now*, plus the pre-tunnel default.

    Two moments are needed and neither is enough alone.

    The stock script does not only add tunnel routes. A gateway that pushes
    split-*excludes* -- "these destinations must not be tunnelled" -- has them
    installed on the **uplink**, via its gateway, after this program took its
    snapshot. Copying only the snapshot therefore misses them: seen live on ASU,
    where three pre-tunnel routes became fifteen. Harmless there, because every
    exclude pointed at the same gateway the copied default already pointed at,
    so replies reached the same next hop either way -- but a single exclude with
    a different next hop would have been shadowed by that default, and nothing
    would have said so.

    Reading live alone is worse. In a full tunnel the stock script *replaces*
    the uplink's default route, so by install time the pre-tunnel value is gone
    and a table built from the live view would have no catch-all at all.

    So: live routes, and the snapshotted default only when the live view has
    lost one.
    """
    live = device_routes(ip, device, family)
    if any(route.get("dst") == "default" for route in live):
        return live
    return live + [r for r in snapshotted if r.get("dst") == "default"]


def plan_reply_routing(ip, snapshot, device, addresses,
                       families=("inet", "inet6")):
    """What to create: one group per (interface, family), or [] if nothing to do.

    `verify` says whether the group carries a "nothing moves" promise. The
    uplink groups do -- their table is a faithful copy of the routes the main
    table already had for that device, so every probe must answer exactly as it
    did. The tunnel group does not, and must not be held to it: moving replies
    from the uplink back onto the tunnel is the whole point of it, so a check
    demanding nothing changed would reject the fix as a regression.
    """
    groups = []
    for uplink, by_family in sorted(snapshot.items()):
        for family, snapshotted in sorted(by_family.items()):
            sources = source_addresses(ip, uplink, family)
            routes = uplink_routes_now(ip, uplink, family, snapshotted)
            if sources and routes:
                groups.append({"device": uplink, "family": family,
                               "routes": routes, "sources": sources,
                               "verify": True})
    if device and addresses:
        # Which families already had another way out, asked of the snapshot
        # rather than assumed: an interface is in there only because the main
        # table had a default route through it for that family, which is
        # exactly the property that decides the shape of the tunnel's table.
        elsewhere = {family for by_family in snapshot.values()
                     for family in by_family}
        for family in families:
            sources = [a for a in addresses
                       if (":" in a) == (family == "inet6")]
            if not sources:
                continue
            if family in elsewhere:
                # Ordinary outbound traffic has another interface to leave by,
                # so the kernel gives it that interface's address as its source
                # and this rule -- which names the tunnel's -- never sees it.
                # Only genuine replies match, and a reply to a peer anywhere
                # needs the catch-all.
                routes = [tunnel_catch_all(ip, device, family)]
            else:
                # Nothing else carries this family, so the tunnel's address is
                # the only source the kernel can choose: *every* outbound
                # connection matches this rule, not just replies. A catch-all
                # would then hand the whole internet to a tunnel that carries
                # only the gateway's own prefixes, which drops it in silence --
                # and silence costs an application its full connect timeout,
                # while a client holding an address of this family tries that
                # family first. That is how a working machine goes slow instead
                # of erroring: dual-stack destinations stall for seconds and
                # single-stack ones do not, which reads as "some sites are
                # slow" rather than as a routing fault. Measured at 0.2s per
                # new host in a browser and tens of seconds in anything without
                # fallback logic.
                #
                # So the table becomes the tunnel's own routes, read off the
                # device exactly as the gateway installed them. Replies to the
                # peers this tunnel really reaches still return through it;
                # anything else finds no route here, falls through to the main
                # table, and fails at once -- which is what lets a client give
                # up on this family and use the one that works.
                routes = device_routes(ip, device, family)
            if routes:
                groups.append({"device": device, "family": family,
                               "routes": routes, "sources": sources,
                               "verify": False})
                if family not in elsewhere:
                    log(f"reply routing: {device} is this machine's only"
                        f" {family} route out, so its table mirrors the"
                        " tunnel's own routes instead of catching everything:"
                        " what the tunnel does not carry fails fast rather"
                        " than stalling")
            else:
                log(f"reply routing: {device} has no {family} routes to"
                    " mirror, so that family is left to the main table")
    return groups


def install_reply_routing(ip, groups, marker):
    """Create the tables and rules, prove nothing moved, record what was made.

    Returns True if reply routing is now in force. Any failure leaves the
    machine exactly as it was found: this is an *addition* to a working
    routing table, and an addition that cannot be made completely is one that
    is not made at all.
    """
    created = []          # (family, table, priority, address) for the manifest
    tables = []           # (family, table) to flush on the way out
    table = C.REPLY_TABLE_BASE
    priority = C.REPLY_RULE_BASE
    baseline = []

    def undo():
        for family, tbl, prio, address in reversed(created):
            run_ip(ip, "-f", family, "rule", "del", "from", address,
                   "table", str(tbl), "priority", str(prio), quiet=True)
        for family, tbl in reversed(tables):
            run_ip(ip, "-f", family, "route", "flush", "table", str(tbl),
                   quiet=True)

    for group in groups:
        family, sources = group["family"], group["sources"]
        while table < C.REPLY_TABLE_BASE + C.REPLY_BAND_SPAN and \
                not table_is_free(ip, table):
            table += 1
        while priority < C.REPLY_RULE_BASE + C.REPLY_BAND_SPAN and \
                not priority_is_free(ip, priority):
            priority += 1
        if table >= C.REPLY_TABLE_BASE + C.REPLY_BAND_SPAN or \
                priority >= C.REPLY_RULE_BASE + C.REPLY_BAND_SPAN:
            log("reply routing: no free routing table or rule priority in"
                f" {C.REPLY_TABLE_BASE}-{C.REPLY_TABLE_BASE + C.REPLY_BAND_SPAN};"
                " leaving routing alone")
            undo()
            return False

        if group["verify"]:
            for address in sources:
                for target in probe_targets(ip, group["routes"],
                                            group["device"], family):
                    baseline.append((family, address, target,
                                     where_would_it_go(ip, target, address, family)))

        # All of it or none of it. A half-copied table is the dangerous
        # shape: it keeps the default route, loses a more-specific one, and
        # sends traffic that was going straight out the wire to the gateway
        # instead. Skipping the group quietly is no better -- that is what
        # made the first version claim success while protecting nothing.
        rendered = [route_args(route) for route in group["routes"]]
        if not rendered or any(args is None for args in rendered):
            log(f"reply routing: could not read {group['device']}'s"
                f" {family} routes back; leaving routing alone")
            undo()
            return False
        for args in rendered:
            if not run_ip(ip, "-f", family, "route", "add", *args,
                          "table", str(table)):
                log(f"reply routing: could not copy {group['device']}'s"
                    f" {family} routes; leaving routing alone")
                undo()
                return False
        tables.append((family, table))
        for address in sources:
            if run_ip(ip, "-f", family, "rule", "add", "from", address,
                      "table", str(table), "priority", str(priority)):
                created.append((family, table, priority, address))
        table += 1
        priority += 1

    if not created:
        undo()
        return False

    # The effect, not the exit status. Every probe that had an answer before
    # must have the same answer now: these tables are copies of routes the main
    # table already had, so anything that moved means the copy was not faithful
    # -- and an unfaithful copy sends ordinary traffic somewhere it does not
    # belong, which is the failure this whole project is organised around.
    for family, address, target, before in baseline:
        after = where_would_it_go(ip, target, address, family)
        if before and after != before:
            log(f"reply routing: {target} from {address} would move"
                f" ({before!r} -> {after!r}); rolling back and leaving"
                " routing alone")
            undo()
            return False

    try:
        with open(marker, "w", encoding="utf-8") as handle:
            for family, tbl, prio, address in created:
                handle.write(f"{family} {tbl} {prio} {address}\n")
    except OSError as exc:
        log(f"reply routing: could not record what was created ({exc});"
            " rolling back rather than leaving rules nothing will remove")
        undo()
        return False

    where = ", ".join(sorted({f"{address} via {group['device']}"
                              for group in groups
                              for address in group["sources"]
                              if any(c[3] == address for c in created)}))
    log(f"reply routing: {where}; replies leave by the interface their"
        " request arrived on")
    return True


def remove_reply_routing(ip, marker):
    """Take away exactly what the manifest says was created, and nothing else.

    Read from the record rather than re-derived. The machine's addresses can
    change between install and teardown, so a list rebuilt now describes a
    different machine -- and deleting by pattern is how you remove somebody
    else's rule. The same discipline as ifindex ownership for the device.
    """
    try:
        with open(marker, encoding="utf-8") as handle:
            lines = handle.read().splitlines()
    except OSError:
        return 0
    removed = 0
    seen = set()
    for line in lines:
        parts = line.split()
        if len(parts) != 4:
            continue
        family, table, priority, address = parts
        if family not in ("inet", "inet6") or not table.isdigit() \
                or not priority.isdigit():
            continue
        if run_ip(ip, "-f", family, "rule", "del", "from", address,
                  "table", table, "priority", priority, quiet=True):
            removed += 1
        seen.add((family, table))
    for family, table in seen:
        run_ip(ip, "-f", family, "route", "flush", "table", table, quiet=True)
    try:
        os.unlink(marker)
    except OSError:
        pass
    return removed


# How long to wait for the stock script to finish configuring the tunnel before
# installing. asuvpn-notify reports the event and *then* execs the real script,
# so the device exists but is not yet addressed when the event lands. Waiting on
# the observable condition beats sleeping a guessed interval, and a bound means
# a script that never finishes costs nothing but this.
REPLY_READY_TIMEOUT = 10
REPLY_READY_POLL = 0.25


def reply_routing_install(ip, snapshot, device, marker, state,
                          families=("inet", "inet6")):
    """Wait for the tunnel to be configured, then install. Own thread.

    The wait is for a fact rather than a duration: the tunnel's address
    appearing on the tunnel's device is the stock script having done its work,
    and it is the same thing this needs before it can describe the tunnel in a
    rule. A device that never gets one simply never arms this; the uplink half
    would still be worth having, but installing half of a symmetry is how the
    third failure in DESIGN.md survived the first two being fixed.
    """
    deadline = time.monotonic() + REPLY_READY_TIMEOUT
    addresses = []
    while time.monotonic() < deadline:
        links = ip_json(ip, "link", "show", "dev", device)
        if links and has_carrier(links[0]):
            addresses = [a for family in families
                         for a in source_addresses(ip, device, family)]
            if addresses:
                break
        time.sleep(REPLY_READY_POLL)
    if not addresses:
        log(f"reply routing: {device} never took an address; leaving routing alone")
        return
    groups = plan_reply_routing(ip, snapshot, device, addresses, families)
    if not groups:
        log("reply routing: no uplink with a default route was found;"
            " nothing to do")
        return
    if install_reply_routing(ip, groups, marker):
        state["installed"] = True
        # The watcher rebuilds against this when an address goes away, so it
        # has to be the device the rules were actually made for.
        state["device"] = device


# How often to confirm the rules still name addresses this machine has. A DHCP
# lease is renewed on a timescale of hours and only rarely changes the address,
# so this is not a hot loop -- it is one `ip -j addr show` per interval, and the
# point is that the window where reply routing is silently off is bounded by
# this rather than by the life of the tunnel.
REPLY_WATCH_INTERVAL = 20


def installed_addresses(marker):
    """The addresses the manifest says rules were made for."""
    try:
        with open(marker, encoding="utf-8") as handle:
            lines = handle.read().splitlines()
    except OSError:
        return set()
    return {parts[3] for parts in (line.split() for line in lines)
            if len(parts) == 4}


def live_addresses(ip):
    """Every global address this machine currently holds, any device."""
    live = set()
    for link in ip_json(ip, "addr", "show"):
        for info in link.get("addr_info") or []:
            if info.get("scope") == "global" and info.get("local"):
                live.add(info["local"])
    return live


def reply_routing_watch(ip, snapshot, marker, state, families, stop):
    """Rebuild the rules if an address they name stops being ours.

    A rule is written `from <address>`, naming one literal address. When DHCP
    renews a lease onto a different address -- or the tunnel is re-established
    with a new one -- that rule matches nothing. It does not fail: it stops
    applying, silently, and the machine is back to answering inbound
    connections down the wrong interface with nothing anywhere saying so. That
    is the same invisible failure reply routing exists to fix, reintroduced by
    its own fix going stale, which is the kind of thing this project treats as
    a bug rather than a caveat.

    Cheap by construction: one query per interval, comparing what the manifest
    records against what the machine holds. A rebuild only happens when an
    address has actually gone.
    """
    while not stop.wait(REPLY_WATCH_INTERVAL):
        if not state.get("installed"):
            continue
        recorded = installed_addresses(marker)
        if not recorded:
            continue
        missing = recorded - live_addresses(ip)
        if not missing:
            continue
        log(f"reply routing: {', '.join(sorted(missing))} is no longer an"
            " address of this machine, so the rule naming it matches nothing;"
            " rebuilding")
        remove_reply_routing(ip, marker)
        state["installed"] = False
        device = state.get("device")
        if not device:
            continue
        addresses = [a for family in families
                     for a in source_addresses(ip, device, family)]
        if not addresses:
            log("reply routing: the tunnel has no address to rebuild against;"
                " leaving routing alone")
            continue
        groups = plan_reply_routing(ip, snapshot, device, addresses, families)
        if groups and install_reply_routing(ip, groups, marker):
            state["installed"] = True


def verify_teardown(ifname, before, was_killed, owned, dns_on_link=None):
    """Confirm the network came back, and clear up if openconnect did not.

    openconnect restores routes and DNS by running vpnc-script on the way out,
    but only when it exits gracefully. If it had to be killed the script never
    ran, which is what leaves a machine with no working network until it is
    rebooted. Deleting the leftover tunnel interface takes its routes with it.

    Routes and DNS are not symmetrical on that path, which this used to gloss
    over with one vague warning. Deleting the device really does take its
    routes, and a per-link resolver dies with the link too -- but a session
    that fell back to /etc/resolv.conf leaves that file naming the tunnel's
    resolvers, and no amount of interface deletion puts it back. So each case
    is now said in its own words, and the one thing that can be undone from
    here is undone rather than warned about.

    Only *our* device is ever deleted. `owned` is the ifindex captured once
    openconnect created it, so a name that has since been reused by somebody
    else's tunnel is left alone. Anything else that appeared meanwhile belongs
    to something else — a VM's tap, a second VPN — and deleting that as root
    because it matched a name prefix would be its own outage.
    """
    ip = find_ip()
    present = tunnel_interfaces()
    # Whether this run actually took the resolver off the link, as opposed to
    # the link having gone on its own or not being ours to touch. Only the
    # branch that calls drop_link_dns may claim it.
    reverted = False

    if interface_exists(ifname):
        if owned is None or C.interface_index(ifname) != owned:
            log(f"{ifname} is not the device this session created; leaving it alone")
        elif not ip:
            log(f"WARNING: {ifname} outlived openconnect but 'ip' is missing;"
                " it was not removed")
        else:
            reverted = dns_on_link == C.DNS_OWNER_LINK
            if reverted:
                # Before the deletion, while there is still a link to name.
                # Usually redundant -- the resolver goes when the link does --
                # but not when the device is persistent or the delete below
                # fails, and those are exactly the runs nobody is watching.
                drop_link_dns(ifname)
            log(f"WARNING: {ifname} outlived openconnect; deleting it")
            result = run([ip, "link", "delete", ifname])
            if result is None or result.returncode != 0:
                log(f"WARNING: could not delete {ifname}; routes may still point at it")

    strays = sorted(present - before - {ifname})
    if strays:
        log(f"other tunnel interfaces present, left alone: {', '.join(strays)}")

    if not ip:
        # Say so rather than returning quietly: silence here reads exactly like
        # a clean teardown, and the routing table may be in pieces.
        log("WARNING: cannot find 'ip'; the default route could not be checked")
        return
    routes = run([ip, "route", "show", "default"])
    if routes is None:
        log("WARNING: 'ip route show' did not answer; the default route is unverified")
        return
    if routes.stdout.strip():
        for line in routes.stdout.strip().splitlines():
            log(f"default route restored: {line.strip()}")
    else:
        log("WARNING: no default route after teardown — network is likely broken")
        log("WARNING: try 'nmcli networking off && nmcli networking on'")
    # Not gated on was_killed. The marker is only removed by the disconnect
    # transition, so its survival says that transition did not run -- however
    # openconnect died, and including the ways we did not cause.
    if dns_on_link == C.DNS_OWNER_LINK and reverted:
        log("the disconnect never ran, so the tunnel's resolver was still on"
            " its link; it has been taken off")
    elif dns_on_link == C.DNS_OWNER_LINK:
        # Reached whenever the deletion branch was not: the device is gone
        # already (so the resolver went with it), or it is not ours to touch,
        # or there is no `ip`. Claiming the revert unconditionally was wrong
        # in the middle case -- a device we could not prove was ours still
        # carries the tunnel's resolvers, and saying "it has been taken off"
        # is exactly the reassurance that stops anybody looking.
        log("WARNING: the disconnect never ran and the tunnel's resolver was"
            f" on {ifname}; this session could not take it off. If names stop"
            f" resolving, 'resolvectl revert {ifname}' does it")
    elif dns_on_link == C.DNS_OWNER_SCRIPT:
        # Nothing here can fix this: the backup belongs to the stock script
        # and is only put back by the disconnect that never ran.
        log("WARNING: openconnect died before it could restore DNS, and this"
            " session had handed DNS to the stock script. /etc/resolv.conf may"
            " still name the tunnel's resolvers; a clean"
            " connect-then-disconnect, or restarting NetworkManager, puts it"
            " back")
    elif was_killed and dns_on_link is None:
        # Killed, and nothing was recorded -- there was no session channel to
        # record in. Say that rather than pick the reassuring half of it.
        log("WARNING: openconnect was killed before it could restore DNS, and"
            " there is no record of how DNS was configured. If names stop"
            " resolving, check /etc/resolv.conf")


class Tunnel:
    """Owns openconnect's lifetime.

    Exactly one thread may call Popen.wait() on a child. CPython's timed wait
    acquires the internal waitpid lock non-blockingly, so if another thread is
    already parked in a blocking wait, every timed wait times out no matter what
    the child does. That is not academic: it made a clean teardown look like a
    hang and escalated all the way to SIGKILL, which is precisely what stops
    vpnc-script from restoring the routing table. So reap() is the only caller
    of wait(), and everyone else waits on the `exited` event instead.
    """

    def __init__(self, proc):
        self.proc = proc
        self.killed = False
        self.exited = threading.Event()
        self._lock = threading.Lock()
        self._closing = False

    def reap(self):
        """The single owner of Popen.wait()."""
        returncode = self.proc.wait()
        self.exited.set()
        return returncode

    def nudge(self):
        """Ask openconnect to re-establish, keeping the session.

        SIGUSR2 is documented as forcing "an immediate disconnection and
        reconnection ... to quickly recover from LAN IP address changes". It
        reuses the session cookie, so unlike a tray-level reconnect it costs no
        sign-in, no Duo push and no polkit prompt. That makes it the right first
        response to a tunnel that has stopped carrying traffic.
        """
        with self._lock:
            if self._closing or self.exited.is_set():
                return False
        try:
            self.proc.send_signal(signal.SIGUSR2)
        except (ProcessLookupError, OSError) as exc:
            log(f"could not signal openconnect to reconnect: {exc}")
            return False
        log("asked openconnect to re-establish the tunnel (SIGUSR2)")
        return True

    def shutdown(self):
        """Ask openconnect to close the tunnel cleanly, escalating if it refuses.

        SIGINT and SIGTERM both make openconnect run vpnc-script with
        reason=disconnect, which is what puts your routes and DNS back. SIGKILL
        does not, so the graces here are deliberately generous: escalating early
        is exactly what leaves a machine with no network.
        """
        with self._lock:
            if self._closing:
                return
            self._closing = True

        escalation = ((signal.SIGINT, 15), (signal.SIGTERM, 10), (signal.SIGKILL, 5))
        for sig, grace in escalation:
            if self.exited.is_set():
                return
            name = signal.Signals(sig).name
            log(f"sending {name} to openconnect (pid {self.proc.pid})")
            if sig is signal.SIGKILL:
                self.killed = True
                log("WARNING: openconnect ignored two polite signals; killing it")
            try:
                # Safe from PID reuse: the child stays a zombie until reap().
                self.proc.send_signal(sig)
            except (ProcessLookupError, OSError):
                return
            if self.exited.wait(timeout=grace):
                return


def relay_output(proc):
    """Forward openconnect's output, and keep draining it if the tray is gone.

    Prefixed, so the tray can tell our messages from openconnect's. Sharing one
    unframed pipe meant a VPN banner containing "[helper] WARNING:" could raise
    a desktop notification with text the server chose.
    """
    for line in proc.stdout:
        emit(C.encode_relay(line))


def watch_control_channel(tunnel):
    try:
        for line in sys.stdin:
            verb = C.decode_control(line)
            if verb == C.CONTROL_QUIT:
                log("disconnect requested")
                return
            if verb == C.CONTROL_RECONNECT:
                # Deliberately does not return: this re-establishes the tunnel
                # and keeps the control channel open, where quit tears it down.
                tunnel.nudge()
                continue
        log("tray applet closed the control pipe")
    except (BrokenPipeError, ValueError, OSError) as exc:
        log(f"control pipe error: {exc}")
    finally:
        tunnel.shutdown()


def host_domain(url):
    """The domain a gateway's own name sits in: https://vpn.example.com/ -> example.com.

    Only ever a fallback, for a gateway that pushes no domains of its own. The
    parsing is here rather than in the contract because this is the one place
    that holds the gateway as a URL; by the time the contract sees it, it is a
    hostname like any other. A --host given without a scheme is still a host:
    urlsplit reads a bare name as a path, so one is added when it is missing.
    """
    text = str(url or "").strip()
    split = urllib.parse.urlsplit(text if "//" in text else "//" + text)
    try:
        return C.parent_domain(split.hostname or "")
    except ValueError:  # a malformed netloc; nothing to derive, and not fatal
        return ""


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--host", required=True, help="VPN URL from openconnect-sso")
    parser.add_argument("--fingerprint", required=True, help="server certificate hash")
    parser.add_argument("--ac-version", default=C.AC_VERSION)
    # Passed in rather than read from a file. Settings belong to the user and
    # this runs as root, where ~ is root's home -- so the side that owns the
    # config resolves it and hands over a value, and the privileged side never
    # reads anything an unprivileged account can write.
    parser.add_argument("--dpd", type=int,
                        default=C.SCHEMA_BY_NAME["dpd"].default,
                        help="dead peer detection interval; 0 leaves the"
                             " server's choice alone")
    # Presence is the switch, not the value: an empty string still means "do
    # it", and says only that no fallback domain was configured. The tray omits
    # the option entirely when the user has turned DNS handling off, and a
    # direct caller who says nothing gets the stock script's DNS behaviour --
    # which is the right default for someone driving the helper by hand.
    parser.add_argument("--dns-domains",
                        help="domains to resolve through the tunnel when the"
                             " gateway names none; empty derives one from"
                             " --host. Omit to leave DNS to vpnc-script")
    parser.add_argument("--no-ipv6", action="store_true",
                        help="do not ask the gateway for IPv6 connectivity"
                             " (openconnect --disable-ipv6)")
    parser.add_argument("--reply-routing", action="store_true",
                        help="add routing rules so replies leave by the"
                             " interface their request arrived on, and remove"
                             " them with the tunnel")
    parser.add_argument("extra", nargs=argparse.REMAINDER)
    args = parser.parse_args()

    # This runs as root from a directory the unprivileged user owns. That is
    # inherent and documented, but group- or world-writable is not: it would let
    # a *second* account take root at the next connect. Enforced, not just noted.
    # Every file this process executes or loads as root. The contract is on the
    # list because it is executed too: the bootstrap checked it before loading,
    # and stating it here as well keeps the whole set visible in one place
    # rather than split between a loader and a main().
    #
    # Ownership is part of the question, not just permissions. "The
    # unprivileged user owns it" is the design; *which* unprivileged user is
    # the thing to check, because an owner can rewrite their own file whatever
    # its mode. Who that is has one answer, in the contract, so that this and
    # the loader above cannot come to different conclusions about the same
    # process.
    trusted = C.invoking_uids()
    for path in (HERE, os.path.abspath(__file__), NOTIFY, CONTRACT):
        reason = C.unsafe_write_access(path, trusted_uids=trusted)
        if reason:
            fatal(f"refusing to run: {path} is {reason}")
            return 26

    refusal = unacceptable_argument(args)
    if refusal:
        fatal(refusal)
        return 28

    openconnect = find_openconnect()
    if not openconnect:
        fatal("openconnect is not installed")
        return 20

    cookie = sys.stdin.readline().rstrip("\n")
    if not cookie:
        fatal("no session cookie arrived from the sign-in")
        return 21

    extra = [a for a in args.extra if a != "--"]
    for option in extra:
        refusal = unsupported_option(option)
        if refusal:
            fatal(refusal)
            return 25
    # Validated rather than trusted, and up here with the other refusals: a
    # refusal after event_channel() below would leave its socket and directory
    # behind in /run. The config parser refuses a negative, so one can only
    # arrive from a caller invoking this directly -- which needs pkexec and is
    # therefore no escalation -- but a privileged program has no business
    # passing its caller's arithmetic through to a root subprocess.
    if args.dpd < 0:
        fatal(f"refusing a negative dead peer detection interval: {args.dpd}")
        return 27
    # Name the device so teardown can prove which interface is ours before
    # deleting anything as root. A caller-supplied name is honoured instead.
    ifname = requested_interface(extra)
    caller_named = ifname is not None
    if caller_named:
        if not C.INTERFACE_RE.match(ifname):
            fatal(f"refusing an unusable interface name: {ifname!r}")
            return 22
    else:
        ifname = free_interface_name()
        if ifname is None:
            fatal("no free asuvpnN interface name is available")
            return 24
        extra = [*extra, "--interface", ifname]

    # Before anything is allocated. Checking after the event channel was open
    # meant this exit path left its socket and directory behind in /run.
    interfaces_before = tunnel_interfaces()
    if interface_exists(ifname):
        # Only reachable for a caller-supplied name now, since our own is chosen
        # free. Never delete a device we were merely *told* to name: `--interface
        # docker0` would otherwise have root destroy the caller's bridge before
        # openconnect had even started.
        fatal(f"{ifname} already exists; refusing to take over"
              " a device this session did not create")
        return 23

    # State comes from openconnect's script contract rather than its log text.
    # A caller-supplied --script is chained, not discarded.
    #
    # Interposing is the optional half of this. Whatever we chain to has to be
    # at least as good as the default we are displacing, because --script
    # replaces it outright -- so if we cannot establish what openconnect would
    # have run, we step aside entirely and let it run that instead. Losing the
    # event channel costs state precision and falls back to log matching;
    # chaining to a path that is not there costs the user their routing table.
    events, event_path, event_dir = event_channel()
    token = secrets.token_hex(16)
    script_env = dict(os.environ)
    caller_script = requested_script(extra)
    if caller_script:
        # Their command line, their responsibility: --script is documented as a
        # shell command line, so it need not be a path we can stat.
        chained, chain_ok = caller_script, True
    else:
        chained = default_vpnc_script(openconnect)
        chain_ok = os.access(chained, os.X_OK)
        if not chain_ok:
            log(f"WARNING: {chained} is not executable, so openconnect's own"
                " default script is left in place; state will come from its"
                " output instead of the script contract")
    if events is not None and (not chain_ok or not os.access(NOTIFY, os.X_OK)):
        # Nothing will ever arrive on it; do not leave a socket and a thread
        # waiting on one for the life of the tunnel.
        events.close()
        events = None
        shutil.rmtree(event_dir, ignore_errors=True)
        event_dir = None
    if events is not None:
        script_env[C.EVENT_SOCKET_VAR] = event_path
        script_env[C.EVENT_TOKEN_VAR] = token
        # Quoted for the same reason: asuvpn-notify execs this through
        # `sh -c` too. A caller's own --script value is already a command line
        # and is passed on as given -- their command line, their quoting --
        # but the path this program derived is a path and has to be made into
        # one.
        script_env[C.REAL_SCRIPT_VAR] = (
            chained if caller_script else shlex.quote(chained))
        if args.dns_domains is not None:
            # Resolved here, once, rather than at every transition: the script
            # runs several times per tunnel and this answer cannot change
            # between them. What the gateway pushes still wins over it -- this
            # is only what to use when the gateway pushes nothing.
            domains = C.split_domains(args.dns_domains) or \
                C.split_domains(host_domain(args.host))
            script_env[C.DNS_DOMAINS_VAR] = " ".join(domains)
            fallback = " ".join(domains) or "nothing (set dns-domains)"
            log("DNS goes on the tunnel's own link, not in /etc/resolv.conf;"
                f" for whichever names the gateway claims, else {fallback}")
        extra = list(strip_script_option(extra))
        # Quoted, because --script is a shell *command line* and not a path:
        # openconnect hands it to `sh -c`, which is also how asuvpn-notify
        # chains onward. An install under a directory with a space in it --
        # `install.sh --link` from "~/My Projects/…", or any XDG_DATA_HOME
        # with one -- otherwise produced `sh -c /home/u/My Projects/...`,
        # which execs "/home/u/My", fails with 127, and leaves *no* script
        # running at any transition: no routes, no DNS, and nothing restored
        # at disconnect, while the tray fell back to log matching and reported
        # Connected throughout. A path containing ; or $( ) was worse than
        # broken. Every sandbox path is space-free, which is why no scenario
        # ever saw it.
        extra = [*extra, "--script", shlex.quote(NOTIFY)]

    # Says what was done, not what the server did: this runs before the server
    # has said anything, and asserting its behaviour here would be wrong for
    # every gateway that does negotiate DPD. (A negative value was refused
    # above, before anything was allocated.)
    if args.dpd and requested_dpd(extra) is None:
        extra = [*extra, "--force-dpd", str(args.dpd)]
        log(f"forcing dead peer detection every {args.dpd}s;"
            " a server that negotiates it off leaves a dropped tunnel"
            " looking connected")

    # Added here rather than in openconnect_command so it sits beside the DPD
    # forcing above: both are this program deciding something on the user's
    # behalf, and both are said out loud for the same reason. Not added twice
    # if the caller already passed it -- their command line, their choice.
    if args.no_ipv6 and not any(
            a == "--disable-ipv6" or a.startswith("--disable-ipv6=")
            for a in extra):
        extra = [*extra, "--disable-ipv6"]
        log("not asking for IPv6: the tunnel will carry IPv4 only")

    command = openconnect_command(openconnect, args, extra)

    # Announced explicitly rather than left to be parsed out of the command line
    # below. The tray watches this device's health, and it needs the name even
    # when the event channel could not be set up and state is coming from log
    # matching. Scraping it back out of a logged argv would be exactly the kind
    # of magic-string coupling the rest of this design removes.
    emit(C.encode_message(C.KIND_DEVICE, C.one_token(ifname)))

    # Logged only once every check has passed, immediately before the spawn.
    # Logging it earlier meant syslog recorded root command lines that were then
    # refused and never executed — reading, during an incident, exactly like
    # commands that had run.
    log(f"running as uid {os.geteuid()}: {shlex.join(command)!r}")

    # Filled in once the device actually appears. Until then there is nothing
    # this session can claim to own, and teardown deletes nothing.
    owned: dict[str, int] = {}

    # Taken here and nowhere else: this is the last moment the routing table is
    # the one this machine had *without* a tunnel. In a full tunnel the stock
    # script replaces the default route, so afterwards the pre-tunnel value is
    # simply gone. Cheap, and skipped entirely when the feature is off or the
    # session has no channel to trigger the install from.
    reply = None
    # With IPv6 off there is no tunnel IPv6 address and no IPv6 route, so an
    # IPv6 rule would name an address that does not exist and protect a family
    # that is not carried. One switch, followed rather than second-guessed.
    reply_families = ("inet",) if args.no_ipv6 else ("inet", "inet6")
    if args.reply_routing:
        reply_ip = find_ip()
        if reply_ip is None:
            log("reply routing: no 'ip' on this machine; leaving routing alone")
        elif event_dir is None:
            log("reply routing: no event channel, so there is no moment to"
                " install at; leaving routing alone")
        else:
            reply = {"ip": reply_ip,
                     "snapshot": uplink_snapshot(reply_ip,
                                                 families=reply_families),
                     "marker": os.path.join(event_dir, C.REPLY_MARKER),
                     "families": reply_families,
                     "installed": False, "address": None}
            if not reply["snapshot"]:
                log("reply routing: no uplink carries a default route;"
                    " nothing to protect")

    # Handlers go up *before* the spawn. A signal landing between Popen and
    # signal.signal() would take the default action and kill this process,
    # leaving a root openconnect with no control channel — so the handler is
    # installed first and reads the tunnel out of a holder once it exists.
    # Handlers run on the main thread, which is about to park in reap(), so
    # teardown is handed to a worker.
    holder: dict[str, Tunnel] = {}

    def on_signal(_signum, _frame):
        tunnel = holder.get("tunnel")
        if tunnel is None:
            os._exit(1)  # nothing spawned yet; nothing to tear down
        threading.Thread(target=tunnel.shutdown, daemon=True).start()

    for sig in (signal.SIGTERM, signal.SIGINT, signal.SIGHUP):
        signal.signal(sig, on_signal)

    parent_pid = os.getpid()
    # Spawned before any thread is started, which is what makes preexec_fn safe.
    proc = subprocess.Popen(
        command,
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        errors="replace",
        bufsize=1,
        # parent_pid captured *before* the fork: preexec_fn runs in the child, so
        # calling os.getpid() in there yields the child's own pid, the guard
        # below always fires, and openconnect is killed before it ever starts.
        env=script_env,
        preexec_fn=lambda: die_with_parent(parent_pid),  # noqa: PLW1509
    )
    tunnel = Tunnel(proc)
    holder["tunnel"] = tunnel
    if events is not None:
        threading.Thread(
            target=serve_events, args=(events, token, ifname, owned, reply),
            daemon=True
        ).start()
    if reply is not None:
        # Stopped by the same event that says the socket close below was
        # deliberate: there is exactly one moment this session is ending, and
        # two flags meaning that would be two chances to disagree.
        threading.Thread(
            target=reply_routing_watch,
            args=(reply["ip"], reply["snapshot"], reply["marker"], reply,
                  reply["families"], closing),
            daemon=True).start()
    # Kept as a second, independent source: if the script never fires we still
    # learn the device's identity from the kernel -- but only once the kernel
    # can also confirm this openconnect is the process holding it open.
    threading.Thread(
        target=watch_for_interface, args=(ifname, owned, proc.pid), daemon=True
    ).start()

    try:
        proc.stdin.write(cookie + "\n")
        proc.stdin.flush()
        proc.stdin.close()
    except (OSError, ValueError) as exc:
        # openconnect can be gone already — an --interface name in use, a
        # missing vpnc-script. Writing to its closed stdin raises, and an
        # unhandled traceback here would replace its real exit status with a
        # crash the tray cannot explain.
        log(f"openconnect closed its input before the cookie arrived: {exc}")

    relay = threading.Thread(target=relay_output, args=(proc,), daemon=True)
    relay.start()
    threading.Thread(target=watch_control_channel, args=(tunnel,), daemon=True).start()

    returncode = tunnel.reap()
    relay.join(timeout=5)
    if returncode < 0:
        signalled = signal.Signals(-returncode).name
        log(f"openconnect was terminated by {signalled}")
        returncode = 128 - returncode
    else:
        log(f"openconnect exited with status {returncode}")
    # Before verify_teardown, which may delete the device: the tunnel table's
    # routes go with it, and a rule left pointing at a table that emptied
    # itself is inert rather than wrong -- but "inert rather than wrong" is the
    # fallback, not the plan. This is the plan, and it runs on every path out
    # of here including the one where openconnect had to be killed and its
    # script never ran at all.
    if reply is not None and reply.get("installed"):
        # Before the removal, not after: a watcher waking between the remove
        # and the flag being cleared would reinstall the rules this is trying
        # to take off, and leave them behind for good.
        closing.set()
        taken = remove_reply_routing(reply["ip"], reply["marker"])
        log(f"reply routing: {taken} rule(s) removed")
    verify_teardown(ifname, interfaces_before, tunnel.killed,
                    owned.get("ifindex"), dns_owner(event_dir))
    closing.set()  # the socket close below is deliberate, not a failure
    if events is not None:
        events.close()
    if event_dir:
        shutil.rmtree(event_dir, ignore_errors=True)
    return returncode


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