414 lines
17 KiB
Python
Executable File
414 lines
17 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
|
|
"""Printers, through CUPS' own API rather than by parsing lpstat.
|
|
|
|
Driverless only, on purpose. This adds printers that describe their own
|
|
capabilities over IPP -- IPP Everywhere, which is every printer sold in roughly
|
|
the last decade -- and does not choose PPDs or download vendor drivers. Driver
|
|
selection is most of what the panel this replaces does, and getting it wrong
|
|
produces a printer that accepts jobs and silently prints nothing. A printer old
|
|
enough to need a PPD is better served by system-config-printer, and the page
|
|
says so rather than pretending.
|
|
|
|
panama-printers snapshot
|
|
panama-printers discover
|
|
panama-printers add URI NAME
|
|
panama-printers remove NAME
|
|
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
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
|
|
# What may be printed to. Everything else -- file:, pipe:, a shell fragment --
|
|
# is refused: this validates the URI rather than trusting a settings page,
|
|
# because a device URI is handed to a backend that runs as root.
|
|
SAFE_SCHEMES = ("ipp", "ipps", "socket", "dnssd", "http", "https")
|
|
URI = re.compile(r"^(%s)://[A-Za-z0-9._~:/?#\[\]@!$&'()*+,;=%%-]+$" % "|".join(SAFE_SCHEMES))
|
|
|
|
# CUPS queue names: no spaces, slashes, or '#'.
|
|
QUEUE = re.compile(r"^[A-Za-z0-9_.-]{1,127}$")
|
|
|
|
# The only model this adds. Naming it in one place means the driverless promise
|
|
# 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."""
|
|
|
|
|
|
def connect():
|
|
try:
|
|
import cups
|
|
|
|
return cups, cups.Connection()
|
|
except Exception as error: # noqa: BLE001 - no cups is a legitimate state
|
|
raise BoundaryError("The printing service is not answering.") from error
|
|
|
|
|
|
def run(command: list[str], timeout: float = 30.0) -> subprocess.CompletedProcess:
|
|
try:
|
|
return subprocess.run(command, capture_output=True, text=True,
|
|
timeout=timeout, check=False)
|
|
except (OSError, subprocess.TimeoutExpired) as error:
|
|
raise BoundaryError(f"{command[0]} did not answer.") from error
|
|
|
|
|
|
def service_state() -> dict:
|
|
active = run(["systemctl", "is-active", "cups.service"]).stdout.strip()
|
|
enabled = run(["systemctl", "is-enabled", "cups.service"]).stdout.strip()
|
|
avahi = run(["systemctl", "is-active", "avahi-daemon.service"]).stdout.strip()
|
|
return {
|
|
"running": active == "active",
|
|
# "on demand" is a real answer and a reasonable configuration, not a
|
|
# problem to report: socket activation starts CUPS when something prints.
|
|
"startsAtBoot": enabled == "enabled",
|
|
"startMode": enabled or "unknown",
|
|
"discoveryAvailable": avahi == "active",
|
|
}
|
|
|
|
|
|
def printer_state(attributes: dict) -> str:
|
|
"""CUPS reports state as a number; people read words."""
|
|
return {3: "idle", 4: "printing", 5: "stopped"}.get(
|
|
int(attributes.get("printer-state", 0)), "unknown")
|
|
|
|
|
|
def snapshot() -> dict:
|
|
cups, connection = connect()
|
|
try:
|
|
raw = connection.getPrinters()
|
|
default = connection.getDefault()
|
|
jobs = connection.getJobs(which_jobs="not-completed", my_jobs=False,
|
|
requested_attributes=[
|
|
"job-id", "job-name", "job-printer-uri",
|
|
"job-state", "job-originating-user-name",
|
|
"job-impressions", "time-at-creation"])
|
|
except Exception as error: # noqa: BLE001
|
|
raise BoundaryError("The printing service could not be read.") from error
|
|
|
|
printers = []
|
|
for name, attributes in raw.items():
|
|
printers.append({
|
|
"name": name,
|
|
"description": str(attributes.get("printer-info") or name),
|
|
"location": str(attributes.get("printer-location") or ""),
|
|
"makeAndModel": str(attributes.get("printer-make-and-model") or ""),
|
|
"uri": str(attributes.get("device-uri") or ""),
|
|
"state": printer_state(attributes),
|
|
"stateMessage": str(attributes.get("printer-state-message") or ""),
|
|
"accepting": bool(attributes.get("printer-is-accepting-jobs", True)),
|
|
"shared": bool(attributes.get("printer-is-shared", False)),
|
|
"isDefault": name == default,
|
|
})
|
|
printers.sort(key=lambda entry: (not entry["isDefault"], entry["name"].lower()))
|
|
|
|
queue = []
|
|
for job in jobs.values() if isinstance(jobs, dict) else []:
|
|
printer_uri = str(job.get("job-printer-uri") or "")
|
|
queue.append({
|
|
"id": int(job.get("job-id") or 0),
|
|
"name": str(job.get("job-name") or "Untitled"),
|
|
"printer": printer_uri.rsplit("/", 1)[-1] if printer_uri else "",
|
|
"state": {3: "pending", 4: "held", 5: "processing",
|
|
6: "stopped", 7: "cancelled"}.get(int(job.get("job-state", 0)), "unknown"),
|
|
"user": str(job.get("job-originating-user-name") or ""),
|
|
"pages": int(job.get("job-impressions") or 0),
|
|
"createdAt": int(job.get("time-at-creation") or 0),
|
|
})
|
|
queue.sort(key=lambda entry: entry["createdAt"])
|
|
|
|
return {"printers": printers, "jobs": queue, "service": service_state(), "error": ""}
|
|
|
|
|
|
def discover() -> dict:
|
|
"""Printers announcing themselves on the network, driverless ones only.
|
|
|
|
A printer that does not advertise IPP Everywhere is reported as found but
|
|
not addable, rather than offered and then failing at the point of adding.
|
|
"""
|
|
if not shutil.which("avahi-browse"):
|
|
return {"found": [], "error": "Network discovery is not available."}
|
|
|
|
found: dict[str, dict] = {}
|
|
for service in ("_ipps._tcp", "_ipp._tcp"):
|
|
result = run(["avahi-browse", "-rtp", service], timeout=20)
|
|
for line in result.stdout.splitlines():
|
|
if not line.startswith("="):
|
|
continue
|
|
parts = line.split(";")
|
|
if len(parts) < 10:
|
|
continue
|
|
name, host, port, text = parts[3], parts[6], parts[8], parts[9]
|
|
scheme = "ipps" if service == "_ipps._tcp" else "ipp"
|
|
resource = ""
|
|
for field in re.findall(r"\"([^\"]*)\"", text):
|
|
if field.startswith("rp="):
|
|
resource = field[3:]
|
|
# Only IPP Everywhere. The attribute a driverless printer publishes
|
|
# is its PDF/JPEG support; without it, CUPS would need a driver.
|
|
driverless = "application/pdf" in text or "URF=" in text or "urf=" in text
|
|
uri = f"{scheme}://{host}:{port}/{resource}" if resource else f"{scheme}://{host}:{port}/"
|
|
found[uri] = {
|
|
"name": name.replace("\\032", " "),
|
|
"uri": uri,
|
|
"host": host,
|
|
"driverless": driverless,
|
|
}
|
|
return {"found": sorted(found.values(), key=lambda entry: entry["name"]), "error": ""}
|
|
|
|
|
|
def require_queue(name: str) -> str:
|
|
if not QUEUE.fullmatch(name or ""):
|
|
raise BoundaryError("That is not a printer name.")
|
|
return name
|
|
|
|
|
|
def add(uri: str, name: str) -> None:
|
|
if not URI.fullmatch(uri or ""):
|
|
raise BoundaryError("That address cannot be used to reach a printer.")
|
|
require_queue(name)
|
|
|
|
cups, connection = connect()
|
|
try:
|
|
# ppdname is the driverless model and nothing else. There is no branch
|
|
# here that selects a PPD, which is what keeps the promise checkable.
|
|
connection.addPrinter(name, device=uri, ppdname=DRIVERLESS_MODEL)
|
|
connection.enablePrinter(name)
|
|
connection.acceptJobs(name)
|
|
except Exception as error: # noqa: BLE001
|
|
message = str(error)
|
|
if "device-error" in message or "1284" in message:
|
|
raise BoundaryError(
|
|
"The printer did not answer, or does not support driverless printing.") from error
|
|
if "not-authorized" in message or "forbidden" in message.lower():
|
|
raise BoundaryError("Adding a printer was not authorized.") from error
|
|
raise BoundaryError("That printer could not be added.") from error
|
|
|
|
|
|
def remove(name: str) -> None:
|
|
require_queue(name)
|
|
cups, connection = connect()
|
|
try:
|
|
connection.deletePrinter(name)
|
|
except Exception as error: # noqa: BLE001
|
|
raise BoundaryError("That printer could not be removed.") from error
|
|
|
|
|
|
def set_default(name: str) -> None:
|
|
require_queue(name)
|
|
cups, connection = connect()
|
|
try:
|
|
connection.setDefault(name)
|
|
except Exception as error: # noqa: BLE001
|
|
raise BoundaryError("That printer could not be made the default.") from error
|
|
|
|
|
|
def set_paused(name: str, paused: bool) -> None:
|
|
require_queue(name)
|
|
cups, connection = connect()
|
|
try:
|
|
if paused:
|
|
connection.disablePrinter(name)
|
|
else:
|
|
connection.enablePrinter(name)
|
|
except Exception as error: # noqa: BLE001
|
|
raise BoundaryError("That printer could not be changed.") from error
|
|
|
|
|
|
def cancel(job_id: str) -> None:
|
|
if not job_id.isdigit():
|
|
raise BoundaryError("That is not a job.")
|
|
cups, connection = connect()
|
|
try:
|
|
connection.cancelJob(int(job_id))
|
|
except Exception as error: # noqa: BLE001
|
|
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)
|
|
if result.returncode != 0:
|
|
raise BoundaryError((result.stderr.strip() or "The test page could not be sent.")[:200])
|
|
|
|
|
|
def main(arguments: list[str]) -> int:
|
|
try:
|
|
if arguments == ["snapshot"]:
|
|
print(json.dumps(snapshot(), separators=(",", ":")))
|
|
return 0
|
|
if arguments == ["discover"]:
|
|
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":
|
|
remove(arguments[1])
|
|
elif len(arguments) == 2 and arguments[0] == "set-default":
|
|
set_default(arguments[1])
|
|
elif len(arguments) == 2 and arguments[0] == "pause":
|
|
set_paused(arguments[1], True)
|
|
elif len(arguments) == 2 and arguments[0] == "resume":
|
|
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 | "
|
|
"hold JOB_ID | release JOB_ID | get-options NAME | "
|
|
"set-option NAME KEY VALUE | test-page NAME")
|
|
except BoundaryError as error:
|
|
try:
|
|
state = snapshot()
|
|
except BoundaryError:
|
|
state = {"printers": [], "jobs": [], "service": service_state()}
|
|
state["error"] = str(error)
|
|
print(json.dumps(state, separators=(",", ":")))
|
|
return 0
|
|
|
|
print(json.dumps(snapshot(), separators=(",", ":")))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main(sys.argv[1:]))
|