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