Own the network: details, VPN, enterprise Wi-Fi, and a firewall that can also allow
Claude-Session: https://claude.ai/code/session_01Ms2FbjQy31TVf3CEvQhGM8
This commit is contained in:
Gabriel Brown
@@ -13,6 +13,7 @@ publish on all interfaces.
|
||||
Changes go through firewall-cmd, which is polkit-aware, so they prompt.
|
||||
|
||||
panama-firewall snapshot
|
||||
panama-firewall zone-info ZONE
|
||||
panama-firewall add-service NAME | remove-service NAME
|
||||
panama-firewall add-port PORT/PROTO | remove-port PORT/PROTO
|
||||
panama-firewall set-zone INTERFACE ZONE
|
||||
@@ -305,6 +306,57 @@ def snapshot() -> dict:
|
||||
}
|
||||
|
||||
|
||||
# Zone targets, as a sentence rather than firewalld's vocabulary. "default" is
|
||||
# the one that catches people out: it does not mean "the default zone", it means
|
||||
# "reject anything no rule allowed", which is the answer someone browsing zones
|
||||
# is actually looking for.
|
||||
TARGETS = {
|
||||
"": "anything no rule allows is rejected",
|
||||
"default": "anything no rule allows is rejected",
|
||||
"%%REJECT%%": "anything no rule allows is rejected, with a refusal sent back",
|
||||
"REJECT": "anything no rule allows is rejected, with a refusal sent back",
|
||||
"DROP": "anything no rule allows is dropped without an answer",
|
||||
"ACCEPT": "anything not explicitly blocked is allowed in",
|
||||
}
|
||||
|
||||
|
||||
def zone_info(name: str) -> dict:
|
||||
"""One zone, described -- read-only, for browsing before choosing.
|
||||
|
||||
Separate from `snapshot` because the zone browser asks about zones this
|
||||
machine is not using, and a snapshot only ever describes the active ones.
|
||||
Nothing here changes anything, so it needs no authorization and no confirm.
|
||||
"""
|
||||
require(ZONE, name, "That is not a zone.")
|
||||
known = firewall("--get-zones").split()
|
||||
if name not in known:
|
||||
raise BoundaryError("There is no zone by that name.")
|
||||
|
||||
detail = zone_detail(name)
|
||||
if not detail:
|
||||
raise BoundaryError("That zone could not be read.")
|
||||
|
||||
services = detail.get("services", [])
|
||||
ports = detail.get("ports", [])
|
||||
parts = []
|
||||
parts.append(f"{len(services)} service{'' if len(services) == 1 else 's'}")
|
||||
parts.append(f"{len(ports)} port rule{'' if len(ports) == 1 else 's'}")
|
||||
summary = ", ".join(parts) + "; " + TARGETS.get(
|
||||
detail.get("target", ""), "custom handling for anything no rule allows")
|
||||
|
||||
return {
|
||||
"zone": name,
|
||||
"services": services,
|
||||
"ports": ports,
|
||||
"interfaces": detail.get("interfaces", []),
|
||||
"target": detail.get("target", ""),
|
||||
"richRules": detail.get("richRules", []),
|
||||
"isDefault": name == firewall("--get-default-zone"),
|
||||
"summary": summary,
|
||||
"error": "",
|
||||
}
|
||||
|
||||
|
||||
def require(pattern: re.Pattern, value: str, message: str) -> str:
|
||||
if not pattern.fullmatch(value or ""):
|
||||
raise BoundaryError(message)
|
||||
@@ -333,6 +385,18 @@ def main(arguments: list[str]) -> int:
|
||||
print(json.dumps(snapshot(), separators=(",", ":")))
|
||||
return 0
|
||||
|
||||
# Read-only, so it answers with its own shape rather than a snapshot --
|
||||
# the browser wants one zone described, not the machine's exposure.
|
||||
if len(arguments) == 2 and arguments[0] == "zone-info":
|
||||
try:
|
||||
answer = zone_info(arguments[1])
|
||||
except BoundaryError as error:
|
||||
answer = {"zone": arguments[1], "services": [], "ports": [],
|
||||
"interfaces": [], "target": "", "richRules": [],
|
||||
"isDefault": False, "summary": "", "error": str(error)}
|
||||
print(json.dumps(answer, separators=(",", ":")))
|
||||
return 0
|
||||
|
||||
if len(arguments) == 2 and arguments[0] in ("add-service", "remove-service"):
|
||||
name = require(SERVICE, arguments[1], "That is not a service name.")
|
||||
verb = "--add-service" if arguments[0] == "add-service" else "--remove-service"
|
||||
@@ -351,9 +415,9 @@ def main(arguments: list[str]) -> int:
|
||||
firewall(f"--set-default-zone={zone}", timeout=120)
|
||||
else:
|
||||
raise BoundaryError(
|
||||
"Usage: panama-firewall snapshot | add-service NAME | remove-service NAME | "
|
||||
"add-port PORT/PROTO | remove-port PORT/PROTO | set-zone INTERFACE ZONE | "
|
||||
"set-default-zone ZONE")
|
||||
"Usage: panama-firewall snapshot | zone-info ZONE | add-service NAME | "
|
||||
"remove-service NAME | add-port PORT/PROTO | remove-port PORT/PROTO | "
|
||||
"set-zone INTERFACE ZONE | set-default-zone ZONE")
|
||||
except BoundaryError as error:
|
||||
try:
|
||||
state = snapshot()
|
||||
|
||||
Executable
+866
@@ -0,0 +1,866 @@
|
||||
#!/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:]))
|
||||
@@ -17,6 +17,9 @@ says so rather than pretending.
|
||||
panama-printers set-default NAME
|
||||
panama-printers pause NAME | resume NAME
|
||||
panama-printers cancel JOB_ID
|
||||
panama-printers hold JOB_ID | release JOB_ID
|
||||
panama-printers get-options NAME
|
||||
panama-printers set-option NAME KEY VALUE
|
||||
panama-printers test-page NAME
|
||||
"""
|
||||
|
||||
@@ -41,6 +44,24 @@ QUEUE = re.compile(r"^[A-Za-z0-9_.-]{1,127}$")
|
||||
# is checkable rather than scattered.
|
||||
DRIVERLESS_MODEL = "everywhere"
|
||||
|
||||
# The printer defaults this page will change, and every value it will accept.
|
||||
#
|
||||
# Closed on purpose. `lpadmin -o anything=anything` is a passthrough into a
|
||||
# daemon running as root, and the two settings people actually reach for --
|
||||
# paper size and double-siding -- are worth exactly two dropdowns. Anything not
|
||||
# in this table is refused rather than forwarded, so what this page can do to a
|
||||
# print queue is readable in one place.
|
||||
OPTION_VOCABULARY: dict[str, tuple[str, ...]] = {
|
||||
"media": ("Letter", "A4", "Legal"),
|
||||
"sides": ("one-sided", "two-sided-long-edge", "two-sided-short-edge"),
|
||||
}
|
||||
|
||||
# How each choice looks when the printer names it, rather than when a PPD does.
|
||||
# IPP spells paper sizes as self-describing keywords ("na_letter_8.5x11in"),
|
||||
# PPDs as short names ("Letter"), and a driverless queue can report either --
|
||||
# so a choice is recognised by the token both spellings share.
|
||||
OPTION_TOKENS = {"Letter": "letter", "A4": "a4", "Legal": "legal"}
|
||||
|
||||
|
||||
class BoundaryError(RuntimeError):
|
||||
"""A user-visible validation or CUPS failure."""
|
||||
@@ -235,6 +256,86 @@ def cancel(job_id: str) -> None:
|
||||
raise BoundaryError("That job could not be cancelled.") from error
|
||||
|
||||
|
||||
def matches(choice: str, reported: str) -> bool:
|
||||
"""Whether a value the printer reported is the choice this page offers."""
|
||||
token = OPTION_TOKENS.get(choice, choice).lower()
|
||||
return token in str(reported).lower()
|
||||
|
||||
|
||||
def get_options(name: str) -> dict:
|
||||
"""The two defaults this page can change, and which values are on offer.
|
||||
|
||||
Asked of the printer rather than assumed: a queue that cannot do two-sided
|
||||
should not be offered a two-sided dropdown, and a printer loaded with A4
|
||||
should not have to be told it is A4 every time. When the printer reports no
|
||||
opinion, the whole vocabulary is offered -- an empty dropdown is worse than
|
||||
a choice that might be refused.
|
||||
"""
|
||||
require_queue(name)
|
||||
cups, connection = connect()
|
||||
try:
|
||||
attributes = connection.getPrinterAttributes(name)
|
||||
except Exception as error: # noqa: BLE001
|
||||
raise BoundaryError("That printer's settings could not be read.") from error
|
||||
|
||||
options: dict[str, str] = {}
|
||||
choices: dict[str, list[str]] = {}
|
||||
raw: dict[str, str] = {}
|
||||
for key, vocabulary in OPTION_VOCABULARY.items():
|
||||
supported = attributes.get(f"{key}-supported") or []
|
||||
if isinstance(supported, (str, bytes)):
|
||||
supported = [supported]
|
||||
supported = [str(entry) for entry in supported]
|
||||
offered = [choice for choice in vocabulary
|
||||
if any(matches(choice, entry) for entry in supported)]
|
||||
choices[key] = offered or list(vocabulary)
|
||||
|
||||
current = str(attributes.get(f"{key}-default") or "")
|
||||
raw[key] = current
|
||||
options[key] = next(
|
||||
(choice for choice in vocabulary if current and matches(choice, current)), "")
|
||||
|
||||
return {"printer": name, "options": options, "choices": choices,
|
||||
"reported": raw, "error": ""}
|
||||
|
||||
|
||||
def set_option(name: str, key: str, value: str) -> None:
|
||||
"""One printer default, from the closed table above and nowhere else."""
|
||||
require_queue(name)
|
||||
vocabulary = OPTION_VOCABULARY.get(key)
|
||||
if vocabulary is None:
|
||||
raise BoundaryError("That is not a setting this page changes.")
|
||||
if value not in vocabulary:
|
||||
raise BoundaryError("That is not a value this setting accepts.")
|
||||
|
||||
cups, connection = connect()
|
||||
try:
|
||||
connection.addPrinterOptionDefault(name, key, value)
|
||||
except Exception as error: # noqa: BLE001
|
||||
message = str(error).lower()
|
||||
if "not-authorized" in message or "forbidden" in message:
|
||||
raise BoundaryError("Changing that setting was not authorized.") from error
|
||||
raise BoundaryError("That setting could not be changed.") from error
|
||||
|
||||
|
||||
def set_held(job_id: str, held: bool) -> None:
|
||||
"""Hold a job where it is, or let it go.
|
||||
|
||||
A held job stays in the queue rather than leaving it, which is the whole
|
||||
point: cancelling to stop a print and then reprinting is how a fifty-page
|
||||
document gets printed twice.
|
||||
"""
|
||||
if not job_id.isdigit():
|
||||
raise BoundaryError("That is not a job.")
|
||||
cups, connection = connect()
|
||||
try:
|
||||
connection.setJobHoldUntil(int(job_id), "indefinite" if held else "no-hold")
|
||||
except Exception as error: # noqa: BLE001
|
||||
raise BoundaryError(
|
||||
"That job could not be held." if held
|
||||
else "That job could not be released.") from error
|
||||
|
||||
|
||||
def test_page(name: str) -> None:
|
||||
require_queue(name)
|
||||
result = run(["lp", "-d", name, "/usr/share/cups/data/testprint"], timeout=30)
|
||||
@@ -251,6 +352,26 @@ def main(arguments: list[str]) -> int:
|
||||
print(json.dumps(discover(), separators=(",", ":")))
|
||||
return 0
|
||||
|
||||
# Read-only, and about one printer, so it answers with its own shape.
|
||||
if len(arguments) == 2 and arguments[0] == "get-options":
|
||||
try:
|
||||
answer = get_options(arguments[1])
|
||||
except BoundaryError as error:
|
||||
answer = {"printer": arguments[1], "options": {}, "choices": {},
|
||||
"reported": {}, "error": str(error)}
|
||||
print(json.dumps(answer, separators=(",", ":")))
|
||||
return 0
|
||||
|
||||
# A mutation, so it answers with the fresh snapshot -- carrying the
|
||||
# printer's re-read options alongside, so the dropdown that made the
|
||||
# change updates from the reply rather than from a second round trip.
|
||||
if len(arguments) == 4 and arguments[0] == "set-option":
|
||||
set_option(arguments[1], arguments[2], arguments[3])
|
||||
state = snapshot()
|
||||
state["options"] = get_options(arguments[1])
|
||||
print(json.dumps(state, separators=(",", ":")))
|
||||
return 0
|
||||
|
||||
if len(arguments) == 3 and arguments[0] == "add":
|
||||
add(arguments[1], arguments[2])
|
||||
elif len(arguments) == 2 and arguments[0] == "remove":
|
||||
@@ -263,12 +384,18 @@ def main(arguments: list[str]) -> int:
|
||||
set_paused(arguments[1], False)
|
||||
elif len(arguments) == 2 and arguments[0] == "cancel":
|
||||
cancel(arguments[1])
|
||||
elif len(arguments) == 2 and arguments[0] == "hold":
|
||||
set_held(arguments[1], True)
|
||||
elif len(arguments) == 2 and arguments[0] == "release":
|
||||
set_held(arguments[1], False)
|
||||
elif len(arguments) == 2 and arguments[0] == "test-page":
|
||||
test_page(arguments[1])
|
||||
else:
|
||||
raise BoundaryError(
|
||||
"Usage: panama-printers snapshot | discover | add URI NAME | remove NAME | "
|
||||
"set-default NAME | pause NAME | resume NAME | cancel JOB_ID | test-page NAME")
|
||||
"set-default NAME | pause NAME | resume NAME | cancel JOB_ID | "
|
||||
"hold JOB_ID | release JOB_ID | get-options NAME | "
|
||||
"set-option NAME KEY VALUE | test-page NAME")
|
||||
except BoundaryError as error:
|
||||
try:
|
||||
state = snapshot()
|
||||
|
||||
Reference in New Issue
Block a user