#!/usr/bin/env python3

"""The parts of networking Quickshell has no surface for.

services/Connectivity.qml is pinned pure-native: Wi-Fi scanning, joining and
Bluetooth pairing already work over DBus through Quickshell.Networking, and
nothing there may shell out. But NetworkManager knows a great deal that module
never exposes -- a connection's addresses, whether it comes back by itself,
whether its MAC is randomised, VPN profile import, hotspots, enterprise
authentication -- and the system proxy and the radio kill switches are not
NetworkManager's at all. This helper is where all of that lives, so that the
native service stays native.

    panama-network details CONNECTION
    panama-network forget CONNECTION
    panama-network set-autoconnect CONNECTION true|false
    panama-network set-mac-random CONNECTION true|false
    panama-network import-vpn FILE
    panama-network hotspot start SSID | hotspot stop | hotspot status
    panama-network join-enterprise SSID PROFILE IDENTITY [CA_CERT]   (password on stdin)
    panama-network proxy get
    panama-network proxy set none
    panama-network proxy set manual [HOST PORT]
    panama-network proxy set auto [PAC_URL]
    panama-network airplane status
    panama-network airplane set true|false

SECRETS

Two rules, both enforced in code rather than by care:

  * Nothing a page renders can contain a secret. Every profile NetworkManager
    describes is read through `listing()`, which drops any property named as a
    secret OR merely shaped like one -- so a field a future NetworkManager adds
    is excluded before anybody notices it exists, rather than after.
  * An enterprise password never appears in argv, which is world-readable
    through /proc for the life of the process. It is read from stdin, and only
    after the rest of the request has been validated: reading it first would
    mean waiting on a password for a request that was always going to be
    refused.

HOW THE ENTERPRISE PASSWORD REACHES NetworkManager

Preferred: libnm's GObject-introspection bindings, which build the profile in
memory and hand it to NetworkManager over D-Bus with AddConnection. The secret
is a value in a D-Bus message: never a command line, never a temporary file.
This machine has them (NM 1.56), so this is the live path.

The probe constructs a client rather than only importing the module, because
the question is not "are the bindings installed" but "can the password be
handed over on the bus" -- a machine with the typelib and a NetworkManager that
will not answer must fall through, not fail.

Fallback: a scripted `nmcli connection edit` session driven over stdin. nmcli's
editor takes `set 802-1x.password …` as input rather than as an argument, which
keeps the secret out of ps just the same. Second choice only because it depends
on the editor's prompt behaviour rather than on a stable API.

The hotspot password is different in kind: NetworkManager generates it, and it
is useless unless it is shown. It is returned once, by the verb that created or
looked up the hotspot, and never logged.
"""

from __future__ import annotations

import json
import re
import shutil
import subprocess
import sys
from pathlib import Path

# Connection names. Every command here is an argument list, never a shell
# string, so this is not defending against quoting: it is defending against
# argument confusion. A name beginning with '-' would be read by nmcli as an
# option, which is the one shape that turns a validated list back into an
# injection -- so a name must begin with a letter, digit or underscore. No
# slashes either, so a name can never be mistaken for or built into a path.
# The rest is broad but bounded, because real networks are called things like
# "Cafe: Guest" and "Bob's iPhone" and refusing those helps nobody.
NAME = re.compile(r"^[A-Za-z0-9_][A-Za-z0-9 _.:+()@'&#!-]{0,62}$")

# An SSID is at most 32 bytes on the wire. Names that cannot fit are refused
# here rather than truncated silently into a network nobody can find.
SSID = re.compile(r"^[A-Za-z0-9_][A-Za-z0-9 _.:+()@'&#!-]{0,31}$")

HOSTNAME = re.compile(r"^[A-Za-z0-9]([A-Za-z0-9.-]{0,253}[A-Za-z0-9])?$")
PAC_URL = re.compile(r"^(https?|file)://[A-Za-z0-9._~:/?#\[\]@!$&'()*+,;=%-]{1,500}$")

# The EAP profiles offered. A closed set, because "type your own EAP string"
# produces a profile that fails to authenticate with no way to tell why.
EAP_PROFILES = {
    "peap-mschapv2": {"eap": "peap", "phase2": "mschapv2"},
    "ttls-pap": {"eap": "ttls", "phase2": "pap"},
}

# VPN profile formats, by extension. NetworkManager's importer picks the plugin
# by type, so the extension is what decides -- and nothing else is accepted.
VPN_TYPES = {".conf": "wireguard", ".ovpn": "openvpn"}

# Properties that hold a secret, named so the filter is readable rather than
# only regular.
SECRET_PROPERTIES = frozenset({
    "802-11-wireless-security.psk",
    "802-11-wireless-security.leap-password",
    "802-1x.password",
    "802-1x.private-key-password",
    "802-1x.phase2-private-key-password",
    "802-1x.pin",
    "wireguard.private-key",
    "gsm.password",
    "gsm.pin",
    "ppp.password",
})

# And the shape of one, because the list above can only ever name the fields
# that existed when it was written. A property whose name reads like a
# credential is dropped before anything downstream can render it -- so a field
# NetworkManager adds in a future release is excluded by default rather than
# leaked until somebody notices.
SECRET_SHAPE = re.compile(
    r"(password|passwd|secret|psk|passphrase|private-key|wep-key|leap|\.pin)", re.I)

# The hotspot's own profile. A fixed name so that stopping and restarting reuse
# one profile rather than accumulating "Hotspot 1", "Hotspot 2" forever, and so
# scripts/panama-wifi-qr can be pointed at it by name for the QR code.
HOTSPOT_CONNECTION = "Panama Hotspot"

# The MAC randomisation property, by connection type. nmcli's short aliases,
# because those are what a person reads in a bug report -- and because a profile
# that does not say what type it is still needs an answer, which for anything
# with a MAC worth randomising means Wi-Fi.
CLONED_MAC = {
    "802-3-ethernet": "ethernet.cloned-mac-address",
    "ethernet": "ethernet.cloned-mac-address",
}
CLONED_MAC_DEFAULT = "wifi.cloned-mac-address"

# Where a profile records the same setting when it is read back, either spelling.
CLONED_MAC_KEYS = ("wifi.cloned-mac-address", "802-11-wireless.cloned-mac-address",
                   "ethernet.cloned-mac-address", "802-3-ethernet.cloned-mac-address")


class BoundaryError(RuntimeError):
    """A user-visible validation or NetworkManager failure."""


def run(command: list[str], timeout: float = 15.0,
        stdin_text: str | None = None) -> subprocess.CompletedProcess:
    try:
        return subprocess.run(command, capture_output=True, text=True,
                              timeout=timeout, check=False, input=stdin_text)
    except (OSError, subprocess.TimeoutExpired) as error:
        raise BoundaryError(f"{command[0]} did not answer.") from error


def tool(name: str, absent: str) -> str:
    found = shutil.which(name)
    if not found:
        raise BoundaryError(absent)
    return name


def nmcli(*arguments: str, timeout: float = 15.0,
          stdin_text: str | None = None) -> str:
    tool("nmcli", "NetworkManager is not available.")
    result = run(["nmcli", *arguments], timeout=timeout, stdin_text=stdin_text)
    if result.returncode != 0:
        raise BoundaryError(refusal(result, "NetworkManager refused that."))
    return result.stdout.strip()


def refusal(result: subprocess.CompletedProcess, fallback: str) -> str:
    lines = (result.stderr or result.stdout or "").strip().splitlines()
    if not lines:
        return fallback
    text = lines[-1].strip()
    lowered = text.lower()
    if "not authorized" in lowered or "dismissed" in lowered:
        return "That network change was not authorized."
    if "no such connection profile" in lowered:
        return "That connection no longer exists."
    return text[:200]


def require(pattern: re.Pattern, value: str, message: str) -> str:
    if not pattern.fullmatch(value or ""):
        raise BoundaryError(message)
    return value


def require_bool(value: str) -> bool:
    if value not in ("true", "false"):
        raise BoundaryError("That is not true or false.")
    return value == "true"


def require_connection(value: str) -> str:
    return require(NAME, value, "That is not a connection name.")


# ---------------------------------------------------------------- reading


def is_secret(key: str) -> bool:
    """Whether a property name is one whose value must not travel."""
    bare = re.sub(r"\[\d+\]$", "", key).strip().lower()
    return bare in SECRET_PROPERTIES or SECRET_SHAPE.search(bare) is not None


def listing(*arguments: str) -> dict[str, str]:
    """A terse nmcli listing, parsed to property -> value, secrets dropped.

    Filtered here rather than at each use. `details` renders whatever it is
    handed, so the one function that reads NetworkManager is the one place that
    has to be certain a passphrase never gets that far -- and a filter applied
    once cannot be forgotten by the next field somebody adds.

    `-e no` turns off nmcli's in-field escaping, so a value containing ':' --
    every MAC address, for one -- arrives whole and splits at the first colon
    exactly where the property name ends.
    """
    tool("nmcli", "NetworkManager is not available.")
    result = run(["nmcli", "-t", "-e", "no", *arguments])
    if result.returncode != 0:
        return {}

    found: dict[str, str] = {}
    for line in result.stdout.splitlines():
        key, separator, value = line.partition(":")
        if not separator:
            continue
        key = key.strip()
        if not key or is_secret(key):
            continue
        found[key] = value.strip()
    return found


def indexed(found: dict[str, str], base: str) -> list[str]:
    """Every value of a repeated property, in order. IP4.DNS[1], [2], and so on."""
    pattern = re.compile(rf"^{re.escape(base)}(\[(\d+)\])?$")
    matches = []
    for key, value in found.items():
        hit = pattern.match(key)
        if hit and value:
            matches.append((int(hit.group(2) or 0), value))
    return [value for _, value in sorted(matches)]


def first(found: dict[str, str], *keys: str) -> str:
    for key in keys:
        if found.get(key):
            return found[key]
    return ""


def connection_listing(name: str) -> dict[str, str]:
    return listing("connection", "show", name)


def connection_exists(name: str) -> bool:
    return bool(connection_listing(name))


def connection_state(name: str, note: str = "") -> dict:
    """Everything the details grid shows, and nothing a page should not have.

    This is the shape every per-connection verb answers with, so a mutation and
    a plain read are the same object to the page -- the change is visible in the
    reply rather than in a refresh that may or may not arrive.
    """
    state = {
        "connection": name,
        "exists": False,
        "uuid": "",
        "type": "",
        "interface": "",
        "active": False,
        "ip4": "",
        "ip6": "",
        "gateway": "",
        "dns": [],
        "mac": "",
        "macRandomized": False,
        "autoconnect": False,
        "note": note,
        "error": "",
    }
    found = connection_listing(name)
    if not found:
        return state

    state["exists"] = True
    state["uuid"] = found.get("connection.uuid", "")
    state["type"] = found.get("connection.type", "")
    state["autoconnect"] = found.get("connection.autoconnect", "") in ("yes", "true")
    state["macRandomized"] = first(found, *CLONED_MAC_KEYS).lower() == "random"

    state["active"] = found.get("GENERAL.STATE", "") == "activated"
    state["ip4"] = (indexed(found, "IP4.ADDRESS") or [""])[0]
    state["ip6"] = (indexed(found, "IP6.ADDRESS") or [""])[0]
    state["gateway"] = found.get("IP4.GATEWAY", "")
    state["dns"] = indexed(found, "IP4.DNS") + indexed(found, "IP6.DNS")

    device = (indexed(found, "GENERAL.DEVICES") or [""])[0]
    state["interface"] = device

    # The address actually on the wire, which is the randomised one when
    # randomisation is on -- the profile records only the policy. A profile
    # listing does not carry it, so the device is asked; the listing is checked
    # first because some builds of nmcli do include it.
    state["mac"] = found.get("GENERAL.HWADDR", "")
    if not state["mac"] and device:
        state["mac"] = listing("-f", "GENERAL.HWADDR", "device", "show",
                               device).get("GENERAL.HWADDR", "")

    return state


# ---------------------------------------------------------------- per-connection


def forget(name: str) -> dict:
    require_connection(name)
    if not connection_exists(name):
        raise BoundaryError("That connection no longer exists.")
    nmcli("connection", "delete", name, timeout=60)
    return connection_state(name, "Forgotten.")


def set_autoconnect(name: str, enabled: bool) -> dict:
    require_connection(name)
    nmcli("connection", "modify", name,
          "connection.autoconnect", "yes" if enabled else "no", timeout=60)
    return connection_state(name)


def set_mac_random(name: str, enabled: bool) -> dict:
    require_connection(name)
    kind = connection_listing(name).get("connection.type", "")
    field = CLONED_MAC.get(kind, CLONED_MAC_DEFAULT)
    nmcli("connection", "modify", name, field,
          "random" if enabled else "permanent", timeout=60)
    # NetworkManager applies a cloned address when the connection comes up, so
    # the address on the wire is the old one until it does. Saying so is the
    # difference between "this did nothing" and "this takes effect next time".
    return connection_state(name, "Reconnect for this to take effect.")


# ---------------------------------------------------------------- VPN import


def connection_uuids() -> set[str]:
    """Every stored profile's UUID, for spotting the one an import added."""
    tool("nmcli", "NetworkManager is not available.")
    result = run(["nmcli", "-t", "-e", "no", "-f", "UUID", "connection", "show"])
    return {line.strip() for line in result.stdout.splitlines() if line.strip()}


IMPORTED = re.compile(r"'([^']+)'\s*\(([0-9a-fA-F-]{36})\)")


def import_vpn(path_text: str) -> dict:
    """A WireGuard or OpenVPN profile, by whichever plugin its extension names.

    The extension is what decides, and nothing else: NetworkManager's importer
    picks its plugin by type, and guessing from file contents would mean
    guessing wrong on a file that is neither.

    nmcli names what it made -- "Connection 'x' (uuid) successfully added." --
    and the quoted name and parenthesised UUID survive translation even where
    the sentence around them does not. When they do not, the UUID list either
    side of the import says which profile is new.
    """
    path = Path(path_text).expanduser()
    kind = VPN_TYPES.get(path.suffix.lower())
    if not kind:
        raise BoundaryError("A VPN profile must be a .conf or .ovpn file.")
    if not path.is_file():
        raise BoundaryError("There is no file at that path.")

    before = connection_uuids()
    output = nmcli("connection", "import", "type", kind, "file", str(path), timeout=60)

    name, uuid = "", ""
    announced = IMPORTED.search(output)
    if announced:
        name, uuid = announced.group(1), announced.group(2)
    else:
        added = connection_uuids() - before
        if added:
            uuid = added.pop()
            name = connection_listing(uuid).get("connection.id", "")

    if not name and not uuid:
        raise BoundaryError("The profile imported but no new connection appeared.")

    return {"name": name, "uuid": uuid, "kind": kind, "error": ""}


# ---------------------------------------------------------------- hotspot


def hotspot_state(password: str = "") -> dict:
    state = {
        "active": False,
        "ssid": "",
        "password": password,
        "connection": HOTSPOT_CONNECTION,
        "band": "",
        "interface": "",
        "error": "",
    }
    found = connection_listing(HOTSPOT_CONNECTION)
    if not found:
        return state
    state["ssid"] = first(found, "802-11-wireless.ssid", "wifi.ssid")
    state["band"] = first(found, "802-11-wireless.band", "wifi.band")
    state["active"] = found.get("GENERAL.STATE", "") == "activated"
    state["interface"] = (indexed(found, "GENERAL.DEVICES") or [""])[0]
    return state


def wifi_interface() -> str:
    """The Wi-Fi device to share from, or "" to let NetworkManager pick.

    Naming it is better when there is more than one adapter; not naming it is
    better than refusing to start because this could not work out which was
    which. NetworkManager makes the same choice, and says so if it cannot.
    """
    result = run(["nmcli", "-t", "-e", "no", "-f", "DEVICE,TYPE", "device"])
    for line in result.stdout.splitlines():
        device, _, kind = line.partition(":")
        if kind.strip() == "wifi" and device.strip():
            return device.strip()
    return ""


def hotspot_password() -> str:
    """The passphrase NetworkManager generated, read back once to be shown.

    `nmcli device wifi show-password` exists for exactly this and prints the
    live hotspot's credentials; the stored profile is asked only if that output
    is not in the expected shape. Never logged, never written down here.
    """
    result = run(["nmcli", "device", "wifi", "show-password"], timeout=20)
    if result.returncode != 0:
        return ""
    for line in result.stdout.splitlines():
        key, _, value = line.partition(":")
        if key.strip().lower() == "password":
            return value.strip()
    return ""


def hotspot_start(ssid: str) -> dict:
    require(SSID, ssid, "That is not a network name.")
    device = wifi_interface()
    where = ["ifname", device] if device else []
    nmcli("device", "wifi", "hotspot", *where,
          "con-name", HOTSPOT_CONNECTION, "ssid", ssid, timeout=45)
    return hotspot_state(hotspot_password())


def hotspot_stop() -> dict:
    if not connection_exists(HOTSPOT_CONNECTION):
        return hotspot_state()
    nmcli("connection", "down", HOTSPOT_CONNECTION, timeout=45)
    return hotspot_state()


# ---------------------------------------------------------------- enterprise


def native_bindings():
    """A live libnm client through GObject introspection, or None.

    The probe constructs the client rather than merely importing the module,
    because the question is not "are the bindings installed" but "can the
    password be handed over on the bus". A machine with the typelib and a
    NetworkManager that will not answer must fall through to the command-line
    client, not fail.
    """
    try:
        import gi

        gi.require_version("NM", "1.0")
        from gi.repository import GLib, NM

        return GLib, NM, NM.Client.new(None)
    except Exception:  # noqa: BLE001 - any failure here means "use the other path"
        return None


def join_enterprise(ssid: str, profile_name: str, identity: str,
                    ca_cert: str) -> dict:
    """Validate first, then read the password, then join.

    The order is deliberate. Reading stdin before the arguments are known to be
    good would leave the helper waiting on a password for a request it was
    always going to refuse -- which, when nothing is piped in, is a hang rather
    than an error message.
    """
    require(SSID, ssid, "That is not a network name.")
    if profile_name not in EAP_PROFILES:
        raise BoundaryError("That is not an authentication method this can use.")
    if not identity.strip():
        raise BoundaryError("An enterprise network needs a username.")
    if len(identity) > 128 or "\n" in identity:
        raise BoundaryError("That username cannot be used.")
    if ca_cert and not Path(ca_cert).expanduser().is_file():
        raise BoundaryError("There is no certificate at that path.")

    password = read_password()
    if not password:
        raise BoundaryError("An enterprise network needs a password.")

    bindings = native_bindings()
    if bindings is not None:
        join_enterprise_native(bindings, ssid, profile_name, identity, ca_cert, password)
    else:
        join_enterprise_nmcli(ssid, profile_name, identity, ca_cert, password)

    state = connection_state(ssid, "Joined.")
    state["joined"] = state["exists"]
    return state


def join_enterprise_native(bindings, ssid: str, profile_name: str, identity: str,
                           ca_cert: str, password: str) -> None:
    """Build the profile in memory and hand it over on the bus.

    AddConnection carries the password as a value in a D-Bus message: it is
    never a command-line argument, so it never appears in /proc, and never a
    temporary file, so it never reaches disk unencrypted on its way in.
    """
    GLib, NM, client = bindings
    method = EAP_PROFILES[profile_name]

    connection = NM.SimpleConnection.new()

    setting = NM.SettingConnection.new()
    setting.set_property(NM.SETTING_CONNECTION_ID, ssid)
    setting.set_property(NM.SETTING_CONNECTION_UUID, NM.utils_uuid_generate())
    setting.set_property(NM.SETTING_CONNECTION_TYPE, "802-11-wireless")
    connection.add_setting(setting)

    wireless = NM.SettingWireless.new()
    wireless.set_property(NM.SETTING_WIRELESS_SSID,
                          GLib.Bytes.new(ssid.encode("utf-8")))
    wireless.set_property(NM.SETTING_WIRELESS_MODE, "infrastructure")
    connection.add_setting(wireless)

    security = NM.SettingWirelessSecurity.new()
    security.set_property(NM.SETTING_WIRELESS_SECURITY_KEY_MGMT, "wpa-eap")
    connection.add_setting(security)

    eap = NM.Setting8021x.new()
    eap.add_eap_method(method["eap"])
    eap.set_property(NM.SETTING_802_1X_PHASE2_AUTH, method["phase2"])
    eap.set_property(NM.SETTING_802_1X_IDENTITY, identity)
    eap.set_property(NM.SETTING_802_1X_PASSWORD, password)
    if ca_cert:
        eap.set_ca_cert(str(Path(ca_cert).expanduser()),
                        NM.Setting8021xCKScheme.PATH, None)
    connection.add_setting(eap)

    ip4 = NM.SettingIP4Config.new()
    ip4.set_property(NM.SETTING_IP_CONFIG_METHOD, "auto")
    connection.add_setting(ip4)
    ip6 = NM.SettingIP6Config.new()
    ip6.set_property(NM.SETTING_IP_CONFIG_METHOD, "auto")
    connection.add_setting(ip6)

    loop = GLib.MainLoop()
    outcome: dict = {}

    def added(source, result, _data):
        try:
            outcome["connection"] = source.add_connection_finish(result)
        except GLib.Error as error:  # noqa: BLE001 - reported, not raised, on this thread
            outcome["error"] = error.message
        loop.quit()

    client.add_connection_async(connection, True, None, added, None)
    # Bounded, because a NetworkManager that never answers must not leave a
    # settings page spinning forever.
    GLib.timeout_add_seconds(45, lambda: (loop.quit(), False)[1])
    loop.run()

    if "error" in outcome:
        raise BoundaryError(str(outcome["error"])[:200])
    if "connection" not in outcome:
        raise BoundaryError("NetworkManager did not answer.")

    # Added, now bring it up so "Connect" means connected.
    activation = run(["nmcli", "-w", "45", "connection", "up", ssid], timeout=60)
    if activation.returncode != 0:
        raise BoundaryError(refusal(activation, "That network refused the sign-in."))


def join_enterprise_nmcli(ssid: str, profile_name: str, identity: str,
                          ca_cert: str, password: str) -> None:
    """The fallback for a machine without libnm's bindings.

    nmcli's connection editor takes its input as lines on stdin, so the password
    arrives the same way it would be typed -- as data, not as an argument. That
    is the whole reason this shape is used rather than `nmcli connection add`
    with the secret on the command line.
    """
    method = EAP_PROFILES[profile_name]
    script = [
        f"set connection.id {ssid}",
        f"set 802-11-wireless.ssid {ssid}",
        "set 802-11-wireless-security.key-mgmt wpa-eap",
        f"set 802-1x.eap {method['eap']}",
        f"set 802-1x.phase2-auth {method['phase2']}",
        f"set 802-1x.identity {identity}",
        f"set 802-1x.password {password}",
    ]
    if ca_cert:
        script.append(f"set 802-1x.ca-cert {Path(ca_cert).expanduser()}")
    script += ["save", "quit", ""]

    nmcli("connection", "edit", "type", "wifi", "con-name", ssid,
          timeout=60, stdin_text="\n".join(script))

    activation = run(["nmcli", "-w", "45", "connection", "up", ssid], timeout=60)
    if activation.returncode != 0:
        raise BoundaryError(refusal(activation, "That network refused the sign-in."))


# ---------------------------------------------------------------- proxy


PROXY_SCHEMA = "org.gnome.system.proxy"


def gsettings_get(schema: str, key: str) -> str:
    tool("gsettings", "The desktop settings store is not available.")
    result = run(["gsettings", "get", schema, key])
    return result.stdout.strip() if result.returncode == 0 else ""


def gsettings_set(schema: str, key: str, value: str) -> None:
    tool("gsettings", "The desktop settings store is not available.")
    result = run(["gsettings", "set", schema, key, value], timeout=20)
    if result.returncode != 0:
        raise BoundaryError(refusal(result, "The proxy could not be changed."))


def unquote(raw: str) -> str:
    """gsettings prints strings quoted and everything else bare."""
    text = raw.strip()
    if len(text) >= 2 and text[0] == text[-1] and text[0] in ("'", '"'):
        return text[1:-1]
    return text


def proxy_get() -> dict:
    mode = unquote(gsettings_get(PROXY_SCHEMA, "mode")) or "none"
    port_text = gsettings_get(f"{PROXY_SCHEMA}.http", "port")
    return {
        "mode": mode,
        "host": unquote(gsettings_get(f"{PROXY_SCHEMA}.http", "host")),
        "port": int(port_text) if port_text.isdigit() else 0,
        "pacUrl": unquote(gsettings_get(PROXY_SCHEMA, "autoconfig-url")),
        "error": "",
    }


def proxy_set(mode: str, arguments: list[str]) -> dict:
    if mode == "none":
        if arguments:
            raise BoundaryError("Turning the proxy off takes no address.")
        gsettings_set(PROXY_SCHEMA, "mode", "none")
    elif mode == "manual":
        # No address means "switch to the manual proxy already stored", which
        # is what choosing Manual from a dropdown means: the fields to fill in
        # only appear once the mode is chosen, so demanding them first would be
        # a mode nobody could select.
        if len(arguments) not in (0, 2):
            raise BoundaryError("A manual proxy needs a host and a port.")
        if arguments:
            host, port = arguments
            require(HOSTNAME, host, "That is not a proxy address.")
            if not port.isdigit() or not (1 <= int(port) <= 65535):
                raise BoundaryError("That is not a port number.")
            # http, https and socks together this phase: separate proxies per
            # protocol is a real configuration but not one anybody asks a
            # settings page for, and three fields that must agree is three ways
            # to get it subtly wrong.
            for protocol in ("http", "https", "socks"):
                gsettings_set(f"{PROXY_SCHEMA}.{protocol}", "host", host)
                gsettings_set(f"{PROXY_SCHEMA}.{protocol}", "port", port)
        gsettings_set(PROXY_SCHEMA, "mode", "manual")
    elif mode == "auto":
        if len(arguments) not in (0, 1):
            raise BoundaryError("An automatic proxy needs a configuration URL.")
        if arguments:
            require(PAC_URL, arguments[0], "That is not a proxy configuration URL.")
            gsettings_set(PROXY_SCHEMA, "autoconfig-url", arguments[0])
        gsettings_set(PROXY_SCHEMA, "mode", "auto")
    else:
        raise BoundaryError("A proxy is off, manual, or automatic.")
    return proxy_get()


# ---------------------------------------------------------------- airplane


def airplane_status() -> dict:
    """Every radio's kill switch, read the way the keybind reads it.

    Any radio still unblocked reads as radios-on, matching scripts/panama-osd:
    airplane mode is a claim about all of them, so a mixed state is not it.
    """
    tool("rfkill", "The radio kill switches are not available.")
    radios: list[dict] = []

    result = run(["rfkill", "-J"])
    parsed = None
    if result.returncode == 0:
        try:
            parsed = json.loads(result.stdout or "{}")
        except json.JSONDecodeError:
            parsed = None

    if parsed is not None:
        for entry in parsed.get("rfkilldevices", []):
            radios.append({
                "type": str(entry.get("type") or ""),
                "soft": str(entry.get("soft") or "") == "blocked",
                "hard": str(entry.get("hard") or "") == "blocked",
            })
    else:
        # Older util-linux has no --json; its list output is stable enough.
        current: dict | None = None
        for line in run(["rfkill", "list"]).stdout.splitlines():
            heading = re.match(r"^\d+:\s+\S+:\s+(.+)$", line)
            if heading:
                current = {"type": heading.group(1).strip().lower(),
                           "soft": False, "hard": False}
                radios.append(current)
            elif current is not None:
                key, _, value = line.strip().partition(":")
                if key == "Soft blocked":
                    current["soft"] = value.strip() == "yes"
                elif key == "Hard blocked":
                    current["hard"] = value.strip() == "yes"

    def blocked(kinds: tuple[str, ...]) -> bool:
        matched = [r for r in radios if any(k in r["type"] for k in kinds)]
        return bool(matched) and all(r["soft"] or r["hard"] for r in matched)

    return {
        "on": bool(radios) and all(radio["soft"] for radio in radios),
        "wifiBlocked": blocked(("wlan", "wireless")),
        "bluetoothBlocked": blocked(("bluetooth",)),
        "hardBlocked": any(radio["hard"] for radio in radios),
        "radios": len(radios),
        "error": "",
    }


def airplane_set(enabled: bool) -> dict:
    tool("rfkill", "The radio kill switches are not available.")
    result = run(["rfkill", "block" if enabled else "unblock", "all"], timeout=20)
    if result.returncode != 0:
        raise BoundaryError(refusal(result, "The radios could not be changed."))
    return airplane_status()


# ---------------------------------------------------------------- entry


def read_password() -> str:
    """The enterprise password, from stdin, stripped of its trailing newline.

    Only the newline: a password may legitimately begin or end with a space,
    and quietly trimming it would produce an authentication failure nobody
    could explain.
    """
    return sys.stdin.readline().rstrip("\n").rstrip("\r")


# What to answer with when a verb fails: the same shape it would have answered
# with, so a page never has to branch on whether the reply is an error.
FALLBACKS = {
    "connection": {"connection": "", "exists": False, "uuid": "", "type": "",
                   "interface": "", "active": False, "ip4": "", "ip6": "",
                   "gateway": "", "dns": [], "mac": "", "macRandomized": False,
                   "autoconnect": False, "note": ""},
    "import": {"name": "", "uuid": "", "kind": ""},
    "hotspot": {"active": False, "ssid": "", "password": "",
                "connection": HOTSPOT_CONNECTION, "band": "", "interface": ""},
    "proxy": {"mode": "none", "host": "", "port": 0, "pacUrl": ""},
    "airplane": {"on": False, "wifiBlocked": False, "bluetoothBlocked": False,
                 "hardBlocked": False, "radios": 0},
}

USAGE = ("Usage: panama-network details CONNECTION | forget CONNECTION | "
         "set-autoconnect CONNECTION true|false | set-mac-random CONNECTION true|false | "
         "import-vpn FILE | hotspot start SSID|stop|status | "
         "join-enterprise SSID PROFILE IDENTITY [CA_CERT] | "
         "proxy get | proxy set none|manual [HOST PORT]|auto [PAC_URL] | "
         "airplane status | airplane set true|false")


def dispatch(arguments: list[str]) -> tuple[str, dict]:
    verb = arguments[0] if arguments else ""
    rest = arguments[1:]

    if verb == "details" and len(rest) == 1:
        return "connection", connection_state(require_connection(rest[0]))
    if verb == "forget" and len(rest) == 1:
        return "connection", forget(rest[0])
    if verb == "set-autoconnect" and len(rest) == 2:
        return "connection", set_autoconnect(rest[0], require_bool(rest[1]))
    if verb == "set-mac-random" and len(rest) == 2:
        return "connection", set_mac_random(rest[0], require_bool(rest[1]))
    if verb == "import-vpn" and len(rest) == 1:
        return "import", import_vpn(rest[0])
    if verb == "hotspot" and rest == ["status"]:
        return "hotspot", hotspot_state()
    if verb == "hotspot" and rest == ["stop"]:
        return "hotspot", hotspot_stop()
    if verb == "hotspot" and len(rest) == 2 and rest[0] == "start":
        return "hotspot", hotspot_start(rest[1])
    if verb == "join-enterprise" and len(rest) in (3, 4):
        return "connection", join_enterprise(
            rest[0], rest[1], rest[2], rest[3] if len(rest) == 4 else "")
    if verb == "proxy" and rest == ["get"]:
        return "proxy", proxy_get()
    if verb == "proxy" and len(rest) >= 2 and rest[0] == "set":
        return "proxy", proxy_set(rest[1], rest[2:])
    if verb == "airplane" and rest == ["status"]:
        return "airplane", airplane_status()
    if verb == "airplane" and len(rest) == 2 and rest[0] == "set":
        return "airplane", airplane_set(require_bool(rest[1]))

    raise BoundaryError(USAGE)


def shape_for(arguments: list[str]) -> str:
    verb = arguments[0] if arguments else ""
    return {"details": "connection", "forget": "connection",
            "set-autoconnect": "connection", "set-mac-random": "connection",
            "join-enterprise": "connection", "import-vpn": "import",
            "hotspot": "hotspot", "proxy": "proxy",
            "airplane": "airplane"}.get(verb, "connection")


def main(arguments: list[str]) -> int:
    try:
        _, answer = dispatch(arguments)
    except BoundaryError as error:
        answer = dict(FALLBACKS[shape_for(arguments)])
        answer["error"] = str(error)
    print(json.dumps(answer, separators=(",", ":")))
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))
