#!/usr/bin/python3 -I
"""vpnc-script wrapper: reports openconnect's state, then does the real work.

openconnect runs this at every transition with the state in the environment —
`reason` is one of pre-init, connect, disconnect, attempt-reconnect, reconnect,
alongside TUNDEV and INTERNAL_IP4_ADDRESS. That is a documented, versioned
contract (openconnect.org/vpnc-script.html), inherited from vpnc and unchanged
across the v7 to v8 rename that silently broke log-message matching.

This forwards one datagram to asuvpn-helper, configures DNS itself, and then
execs the real vpnc-script, so routing is configured exactly as it would have
been.

DNS is the one thing not left to the real script, and only where
systemd-resolved is running. The stock vpnc-script decides how to install a
resolver by grepping /etc/nsswitch.conf for `resolve`; on a machine using
systemd-resolved through its stub -- Ubuntu's default, where nss-resolve is not
installed -- that grep fails, the script falls all the way through to its
generic branch, and the generic branch writes the tunnel's resolver into
/etc/resolv.conf. That path is a symlink to a file systemd-resolved owns, so the
change survives only until resolved next rewrites it, which it does at any link
change. The tunnel then keeps carrying traffic while internal names quietly stop
resolving, and nothing about the tunnel looks wrong -- which is exactly how this
went unnoticed for hours at a time.

So this configures the resolver where it belongs: on the link, through
systemd-resolved's own interface, with the tunnel's domains routed to it and
every other name left on the resolver the machine already had. That is what a
split tunnel needs and what a single global resolver list cannot express. Having
done it, this removes INTERNAL_IP4_DNS from the environment the real script
inherits, because the real script guards both its DNS branches on that variable
being set -- so the handover is total and there are two owners of nothing.

If any of that does not work, the variable stays and the real script does
whatever it would have done. Falling back to the old behaviour is always better
than half-configuring DNS.

Reporting is best effort and never fatal: a failure here must not stop the
tunnel being configured. Isolated (-I) for the same reason as the helper — this
runs as root, and its directory is owned by an unprivileged user.

What goes on the wire, and under which names, is asuvpn_contract's to say. This
end used to restate the field order and the variable names, which is exactly how
two ends of a protocol drift apart.
"""

import importlib.machinery
import importlib.util
import os
import stat
import sys


# 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 by explicit path; see asuvpn_contract.py."""
    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()


def report():
    import socket

    path = os.environ.get(C.EVENT_SOCKET_VAR)
    token = os.environ.get(C.EVENT_TOKEN_VAR)
    if not path or not token:
        return  # nobody is listening, which is not an error
    client = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
    try:
        client.settimeout(2)
        client.sendto(C.encode_event(token, os.environ), path)
    except OSError:
        pass
    finally:
        client.close()


def note(message):
    """Say what was done with DNS, on the stream openconnect relays.

    Not silent even when it works. The failure this whole path exists to fix was
    invisible -- a tunnel that looked perfect while name resolution was quietly
    wrong -- so the log gets a line saying which resolver went on which link and
    which names were pointed at it, every time.
    """
    try:
        sys.stderr.write(f"asuvpn: {message}\n")
        sys.stderr.flush()
    except OSError:
        pass


def resolvectl(binary, *argv, quiet=False):
    """Run one resolvectl subcommand. True if it worked; never raises.

    No shell, an explicit argument list, and every value already checked against
    the contract's own patterns before it gets here -- a domain the gateway sent
    is about to be an argument to a program running as root.

    `quiet` is for the reverts. A revert either tidies up after a failure that
    has already been reported, or runs on the way out against a link that is
    usually gone by then -- and "no such device" on every clean disconnect is
    noise that would teach the reader to skim these lines.
    """
    import subprocess

    try:
        done = subprocess.run([binary, *argv], stdin=subprocess.DEVNULL,
                              stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
                              timeout=C.RESOLVECTL_TIMEOUT, check=False)
    except (OSError, ValueError, subprocess.SubprocessError) as exc:
        if not quiet:
            note(f"resolvectl {argv[0]} failed: {exc}")
        return False
    if done.returncode != 0:
        if not quiet:
            detail = done.stdout.decode("utf-8", "replace").strip().replace("\n", "; ")
            note(f"resolvectl {argv[0]} exited {done.returncode}: {detail}")
        return False
    return True


def tunnel_resolvers(env):
    """The resolvers the gateway pushed, IPv4 first, as resolvectl wants them.

    Both families come out of INTERNAL_IP4_DNS despite its name: openconnect
    merges every resolver the gateway pushed into that one variable, IPv6
    included, and does not set INTERNAL_IP6_DNS at all (verified against the
    installed library by `asuvpn selftest`). Reading INTERNAL_IP6_DNS as well
    costs nothing and is what the documented contract says vpnc sets. This is
    also why the addresses are validated as addresses of either family rather
    than as IPv4 -- holding them to the variable's name would throw away
    working resolvers.

    Validated on the way through, like the domains: a gateway chose these and
    they are about to be arguments to a program running as root.
    """
    return C.split_resolvers(env.get("INTERNAL_IP4_DNS"),
                             env.get("INTERNAL_IP6_DNS"))


def tunnel_domains(env):
    """Which names belong behind this tunnel, and where that was decided.

    The gateway's own answer wins: CISCO_DEF_DOMAIN and CISCO_SPLIT_DNS are it
    saying which names it serves, and nothing local should overrule that. Only
    when it says nothing does the helper's value apply -- the user's setting, or
    the domain derived from the gateway's own address.
    """
    pushed = C.split_domains(env.get("CISCO_DEF_DOMAIN"),
                             env.get("CISCO_SPLIT_DNS"))
    if pushed:
        return pushed, "pushed by the gateway"
    return C.split_domains(env.get(C.DNS_DOMAINS_VAR)), "configured locally"


# What a gateway puts in a split-include list to mean "and everything else
# too". Compared against, never bound to; the analysers cannot tell those
# apart from the literal alone, hence the two suppressions.
DEFAULT_ROUTE_ADDR = "0.0.0.0"  # nosec B104  # noqa: S104 - compared, not bound


def full_tunnel(env):
    """Did the gateway route *everything* through this tunnel?

    Asked because scoping DNS to a domain and marking the link
    not-a-default-route is right for a split tunnel and wrong for a full one.
    With every packet already going down the tunnel, every lookup outside the
    scoped domain would still be handed to whatever resolver the local network
    advertised -- in clear, over the one link the VPN exists to stop using.
    The stock script would have made the tunnel's resolver the only one, so
    this was a regression in the configuration where it matters most.

    Answered exactly the way the stock script decides whether to install a
    default route, because that is the same question: no split-include list
    plus an address of that family (vpnc-script's `elif [ -n
    "$INTERNAL_IP4_ADDRESS" ]`), or a list that contains the default route
    itself. Anything unparseable, or no evidence either way, reads as a split
    tunnel -- the narrower answer, and the one this has always given.
    """
    def listed(var):
        raw = (env.get(var) or "").strip()
        if not raw:
            return 0
        try:
            return max(0, int(raw))
        except ValueError:
            return -1

    four = listed("CISCO_SPLIT_INC")
    if four < 0:
        return False
    if four == 0:
        if env.get("INTERNAL_IP4_ADDRESS"):
            return True
    elif any((env.get(f"CISCO_SPLIT_INC_{i}_ADDR") or "").strip()
             == DEFAULT_ROUTE_ADDR for i in range(four)):
        return True

    six = listed("CISCO_IPV6_SPLIT_INC")
    if six < 0:
        return False
    if six == 0:
        return bool(env.get("INTERNAL_IP6_NETMASK")
                    or env.get("INTERNAL_IP6_ADDRESS"))
    return any((env.get(f"CISCO_IPV6_SPLIT_INC_{i}_MASKLEN") or "").strip() == "0"
               for i in range(six))


def marker_path(env):
    """Where to record that this link's DNS is ours, for teardown to read.

    The helper's per-session directory under /run/asuvpn, which is root-only
    and removed with the session. It exists because the helper otherwise had to
    *infer* the answer from whether the user asked for link DNS -- intent, not
    fact -- and said "the resolver went with its link" on runs where DNS had in
    truth been handed to the stock script and left in /etc/resolv.conf.

    None when there is no event channel: then nothing can be recorded, and the
    helper says it cannot tell rather than guessing.
    """
    socket_path = env.get(C.EVENT_SOCKET_VAR)
    if not socket_path:
        return None
    return os.path.join(os.path.dirname(socket_path), C.DNS_MARKER)


def set_marker(env, owner):
    """Record who owns DNS now, or clear it. Best effort by design.

    `owner` is C.DNS_OWNER_LINK, C.DNS_OWNER_SCRIPT, or None to say nobody
    does. The distinction matters at teardown: a resolver on the link is
    something the helper can take off itself, while a rewritten
    /etc/resolv.conf is only ever put back by the stock script's own
    disconnect -- so the two need different words and one of them needs a
    warning.
    """
    path = marker_path(env)
    if path is None:
        return
    try:
        if owner is None:
            os.unlink(path)
        else:
            with open(path, "w", encoding="utf-8") as handle:
                handle.write(owner + "\n")
    except OSError:
        pass


def apply_dns(env, device, binary, revert_on_failure=True):
    """Put the tunnel's resolver on the tunnel's link. True if DNS is now ours.

    False means nothing was changed and the caller must leave the real script's
    DNS handling alone. Every failure returns False, and any failure after the
    first change reverts the link first: a link holding servers but not the
    domains that scope them is worse than one holding neither, because
    systemd-resolved would then treat the tunnel as a candidate for every
    lookup on the machine.

    `binary` is passed in rather than looked up here so that the tool is
    resolved once per run, and so `asuvpn selftest` can drive this against a
    stand-in and check the calls it makes. The lookup itself stays in
    handle_dns, hardcoded: this runs as root and must not take the path to a
    program it executes from anything openconnect could have set.
    """
    servers = tunnel_resolvers(env)
    if not servers:
        return False  # nothing was pushed; there is nothing to install
    domains, source = tunnel_domains(env)

    # Ordered so the link is never briefly wrong. Scope first, servers last: a
    # link with domains and no servers resolves nothing, which is harmless,
    # while a link with servers and no scope catches every query on the machine.
    #
    # Every call is checked, including the default-route one. That was dropped
    # on the grounds that listing a domain implies it -- and systemd-resolved's
    # own documentation says the implication holds for a *route-only* domain
    # ("~name"), while these are search domains, where the implicit answer is
    # the opposite. So the explicit call is the only thing keeping the link off
    # the default-route set, and a version of resolved without
    # SetLinkDefaultRoute, or one transient bus error, left the tunnel's
    # resolvers catching every lookup on the machine while the log reported a
    # neatly scoped success.
    everything = full_tunnel(env)
    # A full tunnel wants every lookup, and "default-route yes" alone does not
    # deliver that. resolved sends a query matching no routing domain to *all*
    # links that are default routes -- so the local network's resolver is still
    # asked, in parallel, in clear, over the one link the VPN exists to stop
    # using, and it can win the race and answer for any name, including one
    # inside the tunnel's own domain. "~." is a routing domain that matches
    # every name, so this link becomes the best match for all of them and the
    # others are not consulted. That is what the stock script's
    # /etc/resolv.conf rewrite achieved, and what this has to achieve to
    # replace it.
    scope = ["~.", *domains] if everything else list(domains)
    default_route = "yes" if everything or not scope else "no"

    def failed():
        # A connect reverts, because the revert *is* the handover: the stock
        # script still has INTERNAL_IP4_DNS and picks DNS up. A reconnect has
        # nobody to hand to, so it keeps what the link already had.
        if revert_on_failure:
            resolvectl(binary, "revert", device, quiet=True)
        return False

    if not resolvectl(binary, "default-route", device, default_route):
        return failed()
    if scope and not resolvectl(binary, "domain", device, *scope):
        return failed()
    if not scope:
        # The gateway named no domains and none could be derived. Scoping to
        # nothing would resolve nothing, so the link takes every query
        # instead. Deliberately *not* "~." here, unlike the full tunnel above:
        # traffic is still split, so names the local network serves -- a
        # printer, a NAS -- should stay with the resolver that knows them.
        note(f"no domains for {device}: every lookup may go to the"
             " tunnel's resolver. Set dns-domains to scope it.")
    if not resolvectl(binary, "dns", device, *servers):
        return failed()
    set_marker(env, C.DNS_OWNER_LINK)

    where = " ".join(domains) if domains else "all names"
    if everything:
        where = f"{where} (the gateway routes everything through this tunnel)"
    note(f"DNS for {where} on {device} via {' '.join(servers)}"
         f" ({source}); /etc/resolv.conf left alone")
    return True


def revert_dns(device, binary, env):
    """Drop whatever this put on the link, on the way out.

    Unconditional and quiet: the link is usually gone by now, reverting one that
    was never configured does nothing, and there is no outcome here worth a line
    in the log.

    INTERNAL_IP4_DNS is deliberately *not* removed on the way out, however this
    turns out. If a previous connect fell back, the real script has a
    /etc/resolv.conf backup to put back and needs that variable set to do it --
    and its restore is already a no-op when there is no backup, so letting it
    run always is the only choice that is right in both cases.
    """
    resolvectl(binary, "revert", device, quiet=True)
    set_marker(env, None)


def handle_dns(environ, binary=None):
    """Take DNS over for this transition, if it is ours to take.

    The helper sets DNS_DOMAINS_VAR whenever the user has this turned on, so its
    presence -- not its value, which is empty whenever no fallback domain could
    be worked out -- is the switch. Absent means the user turned it off, or this
    is being run as a vpnc-script by hand; either way the real script keeps the
    job it has always had.
    """
    if C.DNS_DOMAINS_VAR not in environ:
        return
    binary = C.resolvectl_path() if binary is None else binary
    if binary is None:
        # No systemd-resolved on this machine, so the stock script keeps DNS
        # and will have written /etc/resolv.conf. Recorded for the teardown,
        # which otherwise had no way to know this run had fallen back.
        if environ.get("reason", "") in ("connect", "reconnect"):
            set_marker(environ, C.DNS_OWNER_SCRIPT)
        return
    device = environ.get("TUNDEV", "")
    if not C.INTERFACE_RE.match(device):
        # About to be an argument to a root command. The helper checks this too;
        # a rule that guards a privileged call is checked where the call is.
        note(f"not touching DNS: {device!r} cannot be a device name")
        return
    reason = environ.get("reason", "")
    if reason == "disconnect":
        revert_dns(device, binary, environ)
    elif reason in ("connect", "reconnect") and apply_dns(
            environ, device, binary,
            # Only a connect may revert on failure. On a connect the revert is
            # the handover: the link goes back to untouched and the stock
            # script picks DNS up from INTERNAL_IP4_DNS, which is still set. On
            # a reconnect there is nobody to hand it to -- the stock script's
            # `reconnect` branch runs hooks and nothing else (checked against
            # the installed script) -- so reverting would take a working
            # configuration off the link and leave DNS owned by no one, over a
            # transient resolvectl failure on a tunnel that is perfectly fine.
                revert_on_failure=(reason == "connect")):
        # The real script guards both its DNS branches on this one variable
        # (vpnc-script: `if [ -n "$INTERNAL_IP4_DNS" ]`), so removing it is how
        # it is told the job is done -- routing untouched, DNS entirely ours.
        environ.pop("INTERNAL_IP4_DNS", None)
    elif reason in ("connect", "reconnect"):
        # Handed back. Recorded, because "no marker" would otherwise mean both
        # "DNS was cleanly given back" and "the stock script has been holding
        # /etc/resolv.conf all along", and the teardown has to tell those
        # apart to know whether to warn.
        set_marker(environ, C.DNS_OWNER_SCRIPT)
    # attempt-reconnect and pre-init are left alone on purpose. The link is
    # coming back, `reconnect` will configure it again, and tearing the resolver
    # down in between would only send lookups meant for the tunnel out over the
    # public resolver instead.


def main():
    try:
        report()
    except Exception:  # reporting must never break routing
        pass

    try:
        handle_dns(os.environ)
    except Exception as exc:  # DNS must never break routing either
        note(f"leaving DNS to the real script: {exc.__class__.__name__}: {exc}")

    # Hand over with the environment otherwise intact; the real script needs
    # every var openconnect set. --script is a shell command line, not a path,
    # so this goes through sh exactly as openconnect would have run it. The
    # fallback is only reached when this is run as a vpnc-script by hand: in
    # normal use the helper always sets the real script, derived from
    # `openconnect --version`.
    script = os.environ.get(C.REAL_SCRIPT_VAR) or C.FALLBACK_VPNC_SCRIPT
    for var in C.EVENT_VARS:
        os.environ.pop(var, None)
    os.execv("/bin/sh", ["/bin/sh", "-c", script])  # noqa: S606


# No code after main(): execv replaces this process, and if it fails it
# raises, so the traceback's nonzero exit is the failure report.
if __name__ == "__main__":
    main()
