#!/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:]))