1201 lines
49 KiB
Python
Executable File
1201 lines
49 KiB
Python
Executable File
#!/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 saved
|
|
panama-network set-autoconnect CONNECTION true|false
|
|
panama-network set-mac-random CONNECTION true|false
|
|
panama-network set-metered CONNECTION yes|no|auto
|
|
panama-network set-ip CONNECTION 4|6 auto
|
|
panama-network set-ip CONNECTION 4|6 manual ADDR/PREFIX GATEWAY DNS[,DNS...]
|
|
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 join-hidden SSID PROFILE wpa-psk|sae|none (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}$")
|
|
|
|
# Static addressing. Written out per octet rather than as \d{1,3}, because
|
|
# 999.999.999.999 is four groups of three digits and is not an address -- and a
|
|
# static address that NetworkManager refuses is a connection that comes up with
|
|
# no address at all, which is a worse failure than being told to retype it.
|
|
IPV4 = re.compile(r"(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}")
|
|
|
|
# IPv6 in every form a person types one: full, compressed at either end, and
|
|
# "::" alone. One alternative per position the elision can take, which is long
|
|
# but is the only shape that accepts fd00::42 and refuses fd00:::42.
|
|
IPV6 = re.compile(
|
|
r"([0-9A-Fa-f]{1,4}:){7}[0-9A-Fa-f]{1,4}"
|
|
r"|([0-9A-Fa-f]{1,4}:){1,7}:"
|
|
r"|([0-9A-Fa-f]{1,4}:){1,6}:[0-9A-Fa-f]{1,4}"
|
|
r"|([0-9A-Fa-f]{1,4}:){1,5}(:[0-9A-Fa-f]{1,4}){1,2}"
|
|
r"|([0-9A-Fa-f]{1,4}:){1,4}(:[0-9A-Fa-f]{1,4}){1,3}"
|
|
r"|([0-9A-Fa-f]{1,4}:){1,3}(:[0-9A-Fa-f]{1,4}){1,4}"
|
|
r"|([0-9A-Fa-f]{1,4}:){1,2}(:[0-9A-Fa-f]{1,4}){1,5}"
|
|
r"|[0-9A-Fa-f]{1,4}:(:[0-9A-Fa-f]{1,4}){1,6}"
|
|
r"|:((:[0-9A-Fa-f]{1,4}){1,7}|:)"
|
|
)
|
|
|
|
# The prefix lengths, which are different numbers on the two stacks: /24 means
|
|
# something on both, /64 only on one.
|
|
IPV4_PREFIX = re.compile(r"3[0-2]|[12]?[0-9]")
|
|
IPV6_PREFIX = re.compile(r"12[0-8]|1[01][0-9]|[1-9]?[0-9]")
|
|
|
|
# How many nameservers a person may list. Not a NetworkManager limit -- a
|
|
# resolver stops trying long before this, and a field with twenty addresses in
|
|
# it is a typo rather than a configuration.
|
|
MAX_DNS = 6
|
|
|
|
# What a connection's metered flag is called, in each direction. NetworkManager
|
|
# spells "decide for yourself" as unknown; a settings page says "automatic",
|
|
# and the two must never be confused with "no", which is a claim.
|
|
METERED_NAMES = {"yes": "yes", "no": "no", "unknown": "auto", "": "auto"}
|
|
METERED_VALUES = {"yes": "yes", "no": "no", "auto": "unknown"}
|
|
|
|
# Key management for a hidden network, by the name the page offers. A closed
|
|
# set: an arbitrary key-mgmt string produces a profile that never associates,
|
|
# with nothing to say why.
|
|
WIFI_SECURITY = {"wpa-psk": "wpa-psk", "sae": "sae", "none": ""}
|
|
|
|
# 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.")
|
|
|
|
|
|
def require_family(value: str) -> str:
|
|
if value not in ("4", "6"):
|
|
raise BoundaryError("An address is either IPv4 or IPv6.")
|
|
return value
|
|
|
|
|
|
def address_pattern(family: str) -> re.Pattern:
|
|
return IPV4 if family == "4" else IPV6
|
|
|
|
|
|
def require_cidr(family: str, value: str) -> str:
|
|
"""An address with its prefix, which NetworkManager will not take without.
|
|
|
|
Split from the right, because an IPv6 address is mostly colons and one
|
|
slash: rpartition finds the prefix wherever the address ends.
|
|
"""
|
|
address, separator, prefix = (value or "").rpartition("/")
|
|
if not separator:
|
|
raise BoundaryError("A static address needs a prefix, like 192.168.1.50/24."
|
|
if family == "4"
|
|
else "A static address needs a prefix, like fd00::42/64.")
|
|
require(address_pattern(family), address,
|
|
"That is not an IPv4 address." if family == "4" else "That is not an IPv6 address.")
|
|
require(IPV4_PREFIX if family == "4" else IPV6_PREFIX, prefix,
|
|
"An IPv4 prefix is a number from 0 to 32." if family == "4"
|
|
else "An IPv6 prefix is a number from 0 to 128.")
|
|
return f"{address}/{prefix}"
|
|
|
|
|
|
def require_gateway(family: str, value: str) -> str:
|
|
"""A gateway, or nothing. Empty is a real answer: a network segment with no
|
|
router is not a broken configuration, it is a network with no way out."""
|
|
text = (value or "").strip()
|
|
if text == "":
|
|
return ""
|
|
return require(address_pattern(family), text, "That is not a gateway address.")
|
|
|
|
|
|
def require_dns(family: str, value: str) -> list[str]:
|
|
"""The nameserver list, comma-separated as it is typed and as nmcli takes it.
|
|
|
|
Validated per stack rather than per address: NetworkManager stores ipv6.dns
|
|
as IPv6 addresses, and an IPv4 nameserver typed into the IPv6 box is a
|
|
profile it refuses to save with an error nobody can act on.
|
|
"""
|
|
servers = [part.strip() for part in (value or "").split(",") if part.strip()]
|
|
if len(servers) > MAX_DNS:
|
|
raise BoundaryError(f"That is more than {MAX_DNS} nameservers.")
|
|
for server in servers:
|
|
require(address_pattern(family), server, f"{server} is not a nameserver address.")
|
|
return servers
|
|
|
|
|
|
def plain(value: str) -> str:
|
|
"""nmcli's way of saying a property is unset, in either of its spellings."""
|
|
text = (value or "").strip()
|
|
return "" if text in ("--", "(none)") else text
|
|
|
|
|
|
def comma_list(value: str) -> list[str]:
|
|
return [part.strip() for part in plain(value).split(",") if part.strip()]
|
|
|
|
|
|
# ---------------------------------------------------------------- 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,
|
|
"metered": "auto",
|
|
# What the PROFILE says, which is not what the four fields above say.
|
|
# Those are the addresses on the wire; these are the addresses the
|
|
# profile asks for -- and the editor has to show the second, or a static
|
|
# address that has not been applied yet looks like it was never typed.
|
|
"ip4Method": "",
|
|
"ip4Addresses": [],
|
|
"ip4Gateway": "",
|
|
"ip4Dns": [],
|
|
"ip6Method": "",
|
|
"ip6Addresses": [],
|
|
"ip6Gateway": "",
|
|
"ip6Dns": [],
|
|
"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["metered"] = METERED_NAMES.get(plain(found.get("connection.metered", "")), "auto")
|
|
|
|
for family, stack in (("4", "ipv4"), ("6", "ipv6")):
|
|
state[f"ip{family}Method"] = plain(found.get(f"{stack}.method", ""))
|
|
state[f"ip{family}Addresses"] = comma_list(found.get(f"{stack}.addresses", ""))
|
|
state[f"ip{family}Gateway"] = plain(found.get(f"{stack}.gateway", ""))
|
|
state[f"ip{family}Dns"] = comma_list(found.get(f"{stack}.dns", ""))
|
|
|
|
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.")
|
|
|
|
|
|
def set_metered(name: str, mode: str) -> dict:
|
|
"""Whether this connection costs money by the byte.
|
|
|
|
Three states, not two. "Automatic" is NetworkManager deciding from what the
|
|
network told it, and it is not the same claim as "no" -- a page that folded
|
|
the two together would report a guess as a fact.
|
|
"""
|
|
require_connection(name)
|
|
value = METERED_VALUES.get(mode)
|
|
if value is None:
|
|
raise BoundaryError("A connection is metered, not metered, or left to NetworkManager.")
|
|
nmcli("connection", "modify", name, "connection.metered", value, timeout=60)
|
|
return connection_state(name)
|
|
|
|
|
|
def set_ip(name: str, family: str, method: str, address: str = "",
|
|
gateway: str = "", dns_text: str = "") -> dict:
|
|
"""One stack's addressing, written in a single nmcli call.
|
|
|
|
Manual and automatic are one setting rather than two: switching back to
|
|
automatic has to clear the addresses manual mode left behind, or
|
|
NetworkManager keeps them and the connection comes up holding both. So every
|
|
property this verb can write is written on every call, with the empty string
|
|
where one should go back to unset.
|
|
"""
|
|
require_connection(name)
|
|
require_family(family)
|
|
stack = "ipv4" if family == "4" else "ipv6"
|
|
|
|
# Every argument is checked before NetworkManager is asked anything at all.
|
|
# Reading the connection first would mean a typo in an address costs an
|
|
# nmcli invocation before it is refused -- and, worse, would make "did
|
|
# something error" pass with the validation deleted.
|
|
if method == "auto":
|
|
settings = [f"{stack}.method", "auto",
|
|
f"{stack}.addresses", "",
|
|
f"{stack}.gateway", "",
|
|
f"{stack}.dns", "",
|
|
f"{stack}.ignore-auto-dns", "no"]
|
|
elif method == "manual":
|
|
servers = require_dns(family, dns_text)
|
|
settings = [f"{stack}.method", "manual",
|
|
f"{stack}.addresses", require_cidr(family, address),
|
|
f"{stack}.gateway", require_gateway(family, gateway),
|
|
f"{stack}.dns", ",".join(servers),
|
|
# Without this NetworkManager appends the ones DHCP handed
|
|
# out to the ones just typed, which is not what a person
|
|
# choosing "manual" is asking for.
|
|
f"{stack}.ignore-auto-dns", "yes" if servers else "no"]
|
|
else:
|
|
raise BoundaryError("An address is either automatic or manual.")
|
|
|
|
before = connection_state(name)
|
|
if not before["exists"]:
|
|
raise BoundaryError("That connection no longer exists.")
|
|
|
|
nmcli("connection", "modify", name, *settings, timeout=60)
|
|
|
|
# Brought back up only if it was up. Activating a connection that was down
|
|
# is a different action, and doing it here would join a network on the
|
|
# strength of somebody editing its addresses.
|
|
if not before["active"]:
|
|
return connection_state(name, "Applies the next time this connection comes up.")
|
|
|
|
activation = run(["nmcli", "-w", "45", "connection", "up", name], timeout=60)
|
|
if activation.returncode != 0:
|
|
raise BoundaryError(refusal(
|
|
activation, "The addresses were saved, but the connection would not come back up."))
|
|
return connection_state(name, "Applied.")
|
|
|
|
|
|
# ---------------------------------------------------------------- saved profiles
|
|
|
|
|
|
# NAME is read last and split with a limit, because a network legitimately
|
|
# called "Cafe: Guest" would otherwise be cut in half by the field separator.
|
|
# Every field before it is a UUID, a keyword or a number, none of which can
|
|
# contain a colon.
|
|
SAVED_FIELDS = "UUID,TYPE,AUTOCONNECT,ACTIVE,TIMESTAMP,NAME"
|
|
|
|
|
|
def visible_ssids() -> set[str]:
|
|
"""What the last scan saw, for the in-range flag.
|
|
|
|
`--rescan no`, deliberately: listing the profiles this machine remembers
|
|
must not make the radio go looking, or opening a settings page would cost
|
|
airtime and drop throughput on the connection being looked at.
|
|
"""
|
|
result = run(["nmcli", "-t", "-e", "no", "-f", "SSID",
|
|
"device", "wifi", "list", "--rescan", "no"])
|
|
if result.returncode != 0:
|
|
return set()
|
|
return {line.strip() for line in result.stdout.splitlines()
|
|
if line.strip() and line.strip() != "--"}
|
|
|
|
|
|
def saved_connections() -> dict:
|
|
"""Every profile NetworkManager holds, including the ones nowhere near here.
|
|
|
|
The scan is what makes this more than `nmcli connection show`: a saved
|
|
network is otherwise invisible until you are standing next to it, which is
|
|
exactly when you are least able to go and tidy it up.
|
|
"""
|
|
tool("nmcli", "NetworkManager is not available.")
|
|
result = run(["nmcli", "-t", "-e", "no", "-f", SAVED_FIELDS, "connection", "show"])
|
|
if result.returncode != 0:
|
|
raise BoundaryError(refusal(result, "NetworkManager would not list the saved networks."))
|
|
|
|
in_range = visible_ssids()
|
|
entries = []
|
|
for line in result.stdout.splitlines():
|
|
parts = line.split(":", 5)
|
|
if len(parts) != 6:
|
|
continue
|
|
uuid, kind, autoconnect, active, timestamp, name = (part.strip() for part in parts)
|
|
if not name:
|
|
continue
|
|
wireless = "wireless" in kind or kind == "wifi"
|
|
entries.append({
|
|
"name": name,
|
|
"uuid": uuid,
|
|
"type": kind,
|
|
"wifi": wireless,
|
|
"autoconnect": autoconnect in ("yes", "true"),
|
|
"active": active in ("yes", "true"),
|
|
"lastUsed": int(timestamp) if timestamp.isdigit() else 0,
|
|
# Only a Wi-Fi profile can be out of range. A wired profile is not
|
|
# somewhere else; it is a cable, and saying "out of range" about one
|
|
# would be inventing a fact. None means the question does not apply.
|
|
"inRange": (name in in_range) if wireless else None,
|
|
})
|
|
|
|
entries.sort(key=lambda entry: (not entry["active"], not entry["autoconnect"],
|
|
entry["name"].lower()))
|
|
return {"connections": entries, "error": ""}
|
|
|
|
|
|
# ---------------------------------------------------------------- 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."))
|
|
|
|
|
|
# ---------------------------------------------------------------- hidden Wi-Fi
|
|
|
|
|
|
def join_hidden(ssid: str, profile_name: str, security: str) -> dict:
|
|
"""A network that does not broadcast its name.
|
|
|
|
The only thing that makes this different from an ordinary join is
|
|
`802-11-wireless.hidden yes`: without it NetworkManager waits to be told the
|
|
network exists and never probes for it, so the profile saves and never
|
|
connects.
|
|
|
|
Validated before stdin is read, for the same reason join-enterprise is: a
|
|
request that was always going to be refused must not sit waiting for a
|
|
password first. The passphrase goes down the editor's stdin, never argv.
|
|
"""
|
|
require(SSID, ssid, "That is not a network name.")
|
|
require(NAME, profile_name, "That is not a connection name.")
|
|
if security not in WIFI_SECURITY:
|
|
raise BoundaryError("A hidden network is WPA2, WPA3, or open.")
|
|
|
|
secured = security != "none"
|
|
password = read_password() if secured else ""
|
|
if secured and not password:
|
|
raise BoundaryError("That network needs a password.")
|
|
|
|
script = [
|
|
f"set connection.id {profile_name}",
|
|
f"set 802-11-wireless.ssid {ssid}",
|
|
"set 802-11-wireless.hidden yes",
|
|
]
|
|
if secured:
|
|
script.append(f"set 802-11-wireless-security.key-mgmt {WIFI_SECURITY[security]}")
|
|
script.append(f"set 802-11-wireless-security.psk {password}")
|
|
script += ["save", "quit", ""]
|
|
|
|
nmcli("connection", "edit", "type", "wifi", "con-name", profile_name,
|
|
timeout=60, stdin_text="\n".join(script))
|
|
|
|
activation = run(["nmcli", "-w", "45", "connection", "up", profile_name], timeout=60)
|
|
if activation.returncode != 0:
|
|
raise BoundaryError(refusal(activation, "That network did not answer."))
|
|
|
|
state = connection_state(profile_name, "Joined.")
|
|
state["joined"] = state["exists"]
|
|
return state
|
|
|
|
|
|
# ---------------------------------------------------------------- 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, "metered": "auto",
|
|
"ip4Method": "", "ip4Addresses": [], "ip4Gateway": "", "ip4Dns": [],
|
|
"ip6Method": "", "ip6Addresses": [], "ip6Gateway": "", "ip6Dns": [],
|
|
"note": ""},
|
|
"saved": {"connections": []},
|
|
"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 | saved | "
|
|
"set-autoconnect CONNECTION true|false | set-mac-random CONNECTION true|false | "
|
|
"set-metered CONNECTION yes|no|auto | "
|
|
"set-ip CONNECTION 4|6 auto | "
|
|
"set-ip CONNECTION 4|6 manual ADDR/PREFIX GATEWAY DNS[,DNS...] | "
|
|
"import-vpn FILE | hotspot start SSID|stop|status | "
|
|
"join-enterprise SSID PROFILE IDENTITY [CA_CERT] | "
|
|
"join-hidden SSID PROFILE wpa-psk|sae|none | "
|
|
"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 == "saved" and not rest:
|
|
return "saved", saved_connections()
|
|
if verb == "set-metered" and len(rest) == 2:
|
|
return "connection", set_metered(rest[0], rest[1])
|
|
if verb == "set-ip" and len(rest) == 3 and rest[2] == "auto":
|
|
return "connection", set_ip(rest[0], rest[1], "auto")
|
|
if verb == "set-ip" and len(rest) == 6 and rest[2] == "manual":
|
|
return "connection", set_ip(rest[0], rest[1], "manual", rest[3], rest[4], rest[5])
|
|
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 == "join-hidden" and len(rest) == 3:
|
|
return "connection", join_hidden(rest[0], rest[1], rest[2])
|
|
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",
|
|
"set-metered": "connection", "set-ip": "connection",
|
|
"join-enterprise": "connection", "join-hidden": "connection",
|
|
"saved": "saved", "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:]))
|