#!/usr/bin/env python3 """Host CUPS backend: forward a job to a CUPS queue inside an LXC container. The typical case is a host that has CUPS but cannot reach the container over TCP: the container is on the work WiFi, while the host has its own network. This backend runs the container's ``lp`` through ``lxc-attach`` and feeds it the spool file on stdin, so the container's CUPS queue (for example ``Papercut-Default`` with its loopback PaperCut relay and Printer JWT) does the protocol work. Nothing needs to listen on the work network. Device URI (container + queue): lxc-papercut://CONTAINER/QUEUE Install and create the host queue: sudo install -m 0755 lxc_papercut_backend.py \\ /usr/lib/cups/backend/lxc-papercut sudo systemctl restart cups sudo lpadmin -p Papercut-LXC -E \\ -v lxc-papercut://CONTAINER/Papercut-Default \\ -P /usr/share/ppd/papercut-relay.ppd \\ -o printer-is-shared=false -o auth-info-required=none sudo cupsenable Papercut-LXC sudo cupsaccept Papercut-LXC lp -d Papercut-LXC -o media=A3 -o print-color-mode=color \\ -o sides=two-sided-long-edge document.pdf The backend maps PPD option names to the IPP names the queue honours, exactly like papercut_relay_backend.py: ColorModel=RGB|CMYK|CMY -> print-color-mode=color ColorModel=Gray|FastGray -> print-color-mode=monochrome SelectColor=Color|Grayscale -> print-color-mode=color|monochrome PageSize=A3 (or any PPD size) -> media-col={media-size={...}} Duplex=None / 1Sided -> sides=one-sided Duplex=DuplexNoTumble / 2Sided -> sides=two-sided-long-edge Duplex=DuplexTumble -> sides=two-sided-short-edge KMDuplex=1Sided|2Sided -> sides=one-sided|two-sided-long-edge Explicit print-color-mode / media / media-col / sides options are preserved and win over the PPD names. Environment overrides (mainly for testing or unusual setups): LXC_PAPERCUT_CONTAINER container name (overrides the device URI host) LXC_PAPERCUT_QUEUE queue name (overrides the device URI path) LXC_PAPERCUT_ATTACH lxc-attach executable (default: lxc-attach) LXC_PAPERCUT_LP lp path inside the container (default: /usr/bin/lp) LXC_PAPERCUT_TIMEOUT submission timeout in seconds (default: 600) CUPS normally runs backends as root, which ``lxc-attach`` needs. If the CUPS service is sandboxed in a way that blocks entering the container, the backend fails with the lxc-attach error in the CUPS error log. Keep this file root-owned and not group/world-writable; CUPS refuses to run backends otherwise. """ from __future__ import annotations import os import re import shlex import shutil import subprocess import sys from urllib.parse import unquote, urlsplit LOGGER = "lxc-papercut" DEFAULT_ATTACH = "lxc-attach" DEFAULT_LP = "/usr/bin/lp" DEFAULT_TIMEOUT = 600.0 CONTAINER_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,63}") QUEUE_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.@ -]{0,126}") # --------------------------------------------------------------------------- # Option mapping (kept in sync with papercut_relay_backend.py) # --------------------------------------------------------------------------- # PWG media names for PPD sizes when the PPD has no PaperDimension entry. MEDIA_NAMES = { "A3": "iso_a3", "A4": "iso_a4", "A5": "iso_a5", "A6": "iso_a6", "B4": "iso_b4", "B5": "iso_b5", "Letter": "na_letter", "Legal": "na_legal", "Tabloid": "na_ledger", "Ledger": "na_ledger", "Executive": "na_executive", "Statement": "na_statement", } # Canonical PWG dimensions (hundredths of a millimetre) for common sizes. MEDIA_DIMENSIONS = { "A3": (29700, 42000), "A4": (21000, 29700), "A5": (14800, 21000), "A6": (10500, 14800), "Letter": (21590, 27940), "Legal": (21590, 35560), "Tabloid": (27940, 43180), "Ledger": (27940, 43180), "Executive": (18410, 26670), "Statement": (13970, 21590), } COLOR_MONO = {"Gray", "FastGray", "DeviceGray", "Grayscale"} COLOR_COLOR = {"RGB", "CMYK", "CMY", "Color"} DUPLEX_SIDES = { "None": "one-sided", "1Sided": "one-sided", "DuplexNoTumble": "two-sided-long-edge", "2Sided": "two-sided-long-edge", "DuplexTumble": "two-sided-short-edge", } def log(message: str): print(f"[{LOGGER}] {message}", file=sys.stderr, flush=True) def split_options(text: str): """Split a CUPS options string, respecting quotes and {...} collections.""" tokens = [] buf = [] quote = None depth = 0 i = 0 while i < len(text): ch = text[i] if quote: if ch == "\\" and i + 1 < len(text): buf.append(text[i + 1]) i += 2 continue if ch == quote: quote = None else: buf.append(ch) i += 1 continue if ch in ('"', "'"): quote = ch i += 1 continue if ch == "{": depth += 1 buf.append(ch) i += 1 continue if ch == "}": depth = max(0, depth - 1) buf.append(ch) i += 1 continue if ch.isspace() and depth == 0: if buf: tokens.append("".join(buf)) buf = [] i += 1 continue if ch == "\\" and i + 1 < len(text): buf.append(text[i + 1]) i += 2 continue buf.append(ch) i += 1 if buf: tokens.append("".join(buf)) return tokens def quote_option(token: str) -> str: """Quote one name=value token for cupsParseOptions().""" name, sep, value = token.partition("=") if not sep or not any(ch.isspace() for ch in value): return token if value.startswith("{"): return f"{name}={value}" escaped = value.replace("\\", "\\\\").replace('"', '\\"') return f'{name}="{escaped}"' def load_ppd_sizes(path): """Return {PageSize choice: (x, y) in hundredths of a millimetre}.""" sizes = {} if not path: return sizes try: with open(path, encoding="utf-8", errors="replace") as fh: for line in fh: match = re.match( r'^\*PaperDimension\s+(\S+):\s*"([^"]+)"', line ) if not match: continue name = match.group(1).split("/", 1)[0] value = match.group(2).split() if len(value) < 2: continue try: width_pt, height_pt = float(value[0]), float(value[1]) except ValueError: continue width = round(width_pt * 100 * 25.4 / 72) height = round(height_pt * 100 * 25.4 / 72) if width > 0 and height > 0: sizes[name] = (width, height) except OSError as exc: log(f"cannot read PPD {path}: {exc}") return sizes def media_col(width: int, height: int) -> str: return ( "media-col={media-size={x-dimension=%d y-dimension=%d}}" % (width, height) ) def parse_custom_mm(value: str): match = re.fullmatch(r"(\d+(?:\.\d+)?)x(\d+(?:\.\d+)?)mm", value) if not match: return None return (round(float(match.group(1)) * 100), round(float(match.group(2)) * 100)) def map_options(options: str, ppd_sizes): """Return (new_options, changes) with PPD options mapped to IPP names.""" tokens = split_options(options) values = {} for token in tokens: name, sep, value = token.partition("=") if sep and name not in values: values[name] = value explicit_color = "print-color-mode" in values explicit_media = "media" in values or "media-col" in values explicit_sides = "sides" in values out = [] changes = [] seen = set() for token in tokens: name, sep, value = token.partition("=") if not sep: out.append(token) continue if name in ("ColorModel", "SelectColor"): if explicit_color or "print-color-mode" in seen: continue if value in COLOR_MONO: color = "monochrome" elif value in COLOR_COLOR: color = "color" else: log(f"no color mapping for {name}={value!r}; leaving it unchanged") out.append(token) continue out.append(f"print-color-mode={color}") seen.add("print-color-mode") changes.append(f"{name}={value} -> print-color-mode={color}") continue if name == "PageSize": if explicit_media or "media-col" in seen or "media" in seen: continue size = ( MEDIA_DIMENSIONS.get(value) or ppd_sizes.get(value) or parse_custom_mm(value) ) if size: out.append(media_col(*size)) seen.add("media-col") changes.append( f"PageSize={value} -> media-col={size[0]//100}x{size[1]//100}mm" ) continue if value in MEDIA_NAMES: out.append(f"media={MEDIA_NAMES[value]}") seen.add("media") changes.append(f"PageSize={value} -> media={MEDIA_NAMES[value]}") continue log(f"no media mapping for PageSize={value!r}; leaving it unchanged") out.append(token) continue if name in ("Duplex", "KMDuplex"): if explicit_sides or "sides" in seen: continue sides = DUPLEX_SIDES.get(value) if name == "KMDuplex" and value == "2Sided" and \ values.get("Binding") == "TopBinding": sides = "two-sided-short-edge" if sides: out.append(f"sides={sides}") seen.add("sides") changes.append(f"{name}={value} -> sides={sides}") continue log(f"no duplex mapping for {name}={value!r}; leaving it unchanged") out.append(token) continue out.append(token) return " ".join(quote_option(t) for t in out), changes # --------------------------------------------------------------------------- # Backend # --------------------------------------------------------------------------- def parse_destination(): parts = urlsplit(os.environ.get("DEVICE_URI", "")) container = ( os.environ.get("LXC_PAPERCUT_CONTAINER") or unquote(parts.hostname or "") ) queue = ( os.environ.get("LXC_PAPERCUT_QUEUE") or unquote(parts.path.lstrip("/")) ) return container.strip(), queue.strip() def run_lp(container: str, queue: str, title: str, copies: int, option_tokens, document_path, document_data): attach = os.environ.get("LXC_PAPERCUT_ATTACH", DEFAULT_ATTACH) resolved = shutil.which(attach) or attach lp_path = os.environ.get("LXC_PAPERCUT_LP", DEFAULT_LP) try: timeout = float(os.environ.get("LXC_PAPERCUT_TIMEOUT", DEFAULT_TIMEOUT)) except ValueError: log("LXC_PAPERCUT_TIMEOUT is not a number; using the default") timeout = DEFAULT_TIMEOUT command = [resolved, "-n", container, "--", lp_path, "-d", queue] if title: command += ["-t", title] if copies > 1 and not any(t.startswith("copies=") for t in option_tokens): command += ["-n", str(copies)] for token in option_tokens: command += ["-o", token] # No file argument: lp reads the document from stdin. log("forwarding to container %r queue %r" % (container, queue)) if option_tokens: log("IPP options: " + " ".join(option_tokens)) if copies > 1: log("copies: %d" % copies) try: if document_data is None: with open(document_path, "rb") as fh: proc = subprocess.run( command, stdin=fh, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=timeout, ) else: proc = subprocess.run( command, input=document_data, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=timeout, ) except FileNotFoundError as exc: log(f"cannot run {resolved}: {exc}") return 1 except subprocess.TimeoutExpired: log(f"container lp timed out after {timeout:.0f}s") return 1 except OSError as exc: log(f"cannot forward job: {exc}") return 1 output = (proc.stdout or b"") + (proc.stderr or b"") text = " ".join(output.decode("utf-8", "replace").split()) if text: log("container lp: " + text[:400]) if proc.returncode != 0: log(f"container lp failed with status {proc.returncode}") return 1 return 0 def main(argv) -> int: if not argv: # CUPS device discovery (lpinfo -v); no job is being submitted. container = os.environ.get("LXC_PAPERCUT_CONTAINER", "CONTAINER") queue = os.environ.get("LXC_PAPERCUT_QUEUE", "Papercut-Default") print( 'direct lxc-papercut://%s/%s "PaperCut via LXC container" ' '"MFG:PaperCut;MDL:Mobility Print LXC;"' % ( container, queue.replace(" ", "%20")), ) return 0 if argv[0] in ("--help", "-h"): print(__doc__) return 0 if len(argv) < 5: log("usage: lxc-papercut job-id user title copies options [file]") return 1 job_id, _user, title, copies_arg, options = argv[:5] document_path = argv[5] if len(argv) > 5 else None container, queue = parse_destination() if not CONTAINER_RE.fullmatch(container): log(f"invalid or missing container name {container!r}; use " "lxc-papercut://CONTAINER/QUEUE") return 1 if not QUEUE_RE.fullmatch(queue): log(f"invalid or missing queue name {queue!r}; use " "lxc-papercut://CONTAINER/QUEUE") return 1 try: copies = max(1, int(copies_arg)) except (TypeError, ValueError): copies = 1 if document_path in (None, "", "-"): document_path = None document_data = sys.stdin.buffer.read() if not document_data: log("no document data on stdin") return 1 else: document_data = None if not os.path.isfile(document_path): log(f"spool file not found: {document_path}") return 1 ppd_sizes = load_ppd_sizes(os.environ.get("PPD")) mapped, changes = map_options(options, ppd_sizes) for change in changes: log(change) mapped_tokens = split_options(mapped) return run_lp( container=container, queue=queue, title=title or f"lxc-papercut-{job_id}", copies=copies, option_tokens=mapped_tokens, document_path=document_path, document_data=document_data, ) if __name__ == "__main__": raise SystemExit(main(sys.argv[1:]))