#!/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:
        for target in (here, path):
            info = os.stat(target)
            if info.st_mode & stat.S_IWOTH or (
                    info.st_mode & stat.S_IWGRP and info.st_gid != info.st_uid):
                raise SystemExit(f"refusing to load {target}: writable by others")
    loader = importlib.machinery.SourceFileLoader("asuvpn_contract", path)
    spec = importlib.util.spec_from_loader("asuvpn_contract", loader)
    module = importlib.util.module_from_spec(spec)
    loader.exec_module(module)
    return module


C = _contract()

OPENCONNECT_CANDIDATES = ("/usr/sbin/openconnect", "/usr/bin/openconnect")
PR_SET_PDEATHSIG = 1
PR_GET_PDEATHSIG = 2
IP_CANDIDATES = ("/usr/sbin/ip", "/sbin/ip", "/usr/bin/ip")
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",
)
# 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,}$")

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


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 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
    for name in entries:
        pid = name.partition("-")[0]
        if pid.isdigit() and not os.path.exists(f"/proc/{pid}"):
            shutil.rmtree(os.path.join(RUNTIME_ROOT, name), 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):
    """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: the state 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
        # 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.
    """
    for path in OPENCONNECT_CANDIDATES:
        if os.access(path, os.X_OK):
            return path
    return shutil.which("openconnect")


def find_ip():
    for path in IP_CANDIDATES:
        if os.access(path, os.X_OK):
            return path
    return None


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 watch_for_interface(ifname, holder, deadline=60):
    """Record the ifindex of our device once openconnect creates it."""
    end = time.monotonic() + deadline
    while time.monotonic() < end:
        index = C.interface_index(ifname)
        if index is not None:
            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("/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"/sys/class/net/{name}")


def verify_teardown(ifname, before, was_killed, owned):
    """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.

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

    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:
            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'")
    if was_killed:
        log("WARNING: openconnect was killed, so DNS may still need attention")


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.SETTINGS_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("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().
    for path in (HERE, os.path.abspath(__file__), NOTIFY, CONTRACT):
        reason = C.unsafe_write_access(path)
        if reason:
            fatal(f"refusing to run: {path} is {reason}")
            return 26

    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
        script_env[C.REAL_SCRIPT_VAR] = 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))
        extra = [*extra, "--script", 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")

    command = [
        openconnect,
        "--useragent",
        f"AnyConnect Linux_64 {args.ac_version}",
        "--version-string",
        args.ac_version,
        "--cookie-on-stdin",
        "--servercert",
        args.fingerprint,
        *extra,
        args.host,
    ]

    # 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] = {}

    # 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), daemon=True
        ).start()
    # Kept as a second, independent source: if the script never fires we still
    # learn the device's identity from the kernel.
    threading.Thread(
        target=watch_for_interface, args=(ifname, owned), 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}")
    verify_teardown(ifname, interfaces_before, tunnel.killed, owned.get("ifindex"))
    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())
