#!/usr/bin/env python3 """Loopback IPP relay for a PaperCut Mobility Print queue. CUPS sends IPP requests to this relay (for example a queue whose device URI is ``ipp://127.0.0.1:9123/printers/Follow-Me``). The relay forwards every POST to the fixed upstream endpoint from the YAML written by ``acquire_queue_info.py``, adds ``Authorization: Bearer ``, and relays the IPP response back. The incoming request URL is *not* used to choose the upstream, the queue, or the credential. It is logged (and ignored) only so that the operator can see what CUPS sent. The upstream endpoint, canonical IPP ``printer-uri`` and JWT all come from the config file; everything that is not in the config (listen address, port, TLS policy, timeouts, user agent) is a hardcoded default or an explicit command-line override. Config file (flat YAML produced by acquire_queue_info.py): printer_jwt required bearer credential for the upstream POSTs ipp_endpoint required full upstream URL, e.g. https://172.25.1.14:9164/ipp/.../printers/Follow-Me printer_uri optional canonical IPP printer-uri (default: ipp_endpoint) printer optional name shown in logs and the health response mobility_base optional shown in logs queue_url optional not used by the relay, never logged jwt_expires_* optional informational; the JWT is decoded to get exp The relay: * binds to 127.0.0.1:9123 by default and refuses non-loopback addresses unless --allow-non-loopback is given; * substitutes the canonical IPP ``printer-uri`` from the config in every forwarded request, so a CUPS queue whose device URI is a placeholder (``papercut-relay:/``) is still routed to the right queue upstream; * handles Content-Length and Transfer-Encoding: chunked request bodies; * answers Expect: 100-continue (via BaseHTTPRequestHandler); * relays the upstream HTTP status, IPP body and Content-Type; * verifies the upstream TLS certificate only with --verify/--cacert (the captured PaperCut certificate is self-signed); * logs a startup summary, one line per IPP request (including the media / print-color-mode / PPD option attributes CUPS sent), and one line per upstream result to stderr (journal under systemd); * serves a small plain-text health page on GET (always non-secret); * never logs the Printer JWT or the per-user queue URL. Usage (service): python3 -I -B ipp_relay.py --config /etc/papercut/queue.yaml python3 -I -B ipp_relay.py --config queue.yaml --check python3 -I -B ipp_relay.py --config queue.yaml --debug Quick test in the foreground, without systemd: 1. Get the config, or reuse /etc/papercut/queue.yaml: python3 -I acquire_queue_info.py --out queue.yaml 2. Validate it; --check also prints the JWT expiry: python3 -I -B ipp_relay.py --config queue.yaml --check 3. Start the relay and leave it running: python3 -I -B ipp_relay.py --config queue.yaml --debug curl -s http://127.0.0.1:9123/ # non-secret health page 4. In another terminal, create a temporary CUPS queue that sends its jobs to the relay: sudo lpadmin -p RelayTest -E \\ -v ipp://127.0.0.1:9123/ \\ -P /usr/share/ppd/papercut-relay.ppd \\ -o printer-is-shared=false -o auth-info-required=none sudo cupsenable RelayTest sudo cupsaccept RelayTest 5. Print; the relay adds the Bearer JWT and forwards the IPP request: lp -d RelayTest file.pdf lp -d RelayTest -o media=A3 -o print-color-mode=color \\ -o sides=two-sided-long-edge file.pdf Watch the relay output for the "IPP request:" and "upstream result:" lines. The lp options above are already the IPP names the queue honours; the wrapper is only needed for PPD/UI options. 6. Remove the temporary queue and stop the relay with Ctrl-C: sudo lpadmin -x RelayTest The relay is an IPP forwarder, not a spooler. It accepts only Content-Type: application/ipp POSTs from CUPS or another IPP client, so it is not itself an lp destination. The incoming URL path is logged and ignored; the upstream endpoint, printer-uri and JWT all come from the config. Service installation (long-running): sudo install -D -m 0755 ipp_relay.py /usr/local/lib/papercut/ipp_relay.py sudo install -D -m 0600 queue.yaml /etc/papercut/queue.yaml sudo install -m 0644 papercut-ipp-relay.service /etc/systemd/system/ sudo systemctl daemon-reload sudo systemctl enable --now papercut-ipp-relay journalctl -u papercut-ipp-relay -f After replacing the queue YAML, restart the service so the new JWT/endpoint is loaded: sudo systemctl restart papercut-ipp-relay """ from __future__ import annotations import argparse import base64 import gzip import http.client import json import logging import os import re import signal import socket import ssl import stat import sys import threading import time from dataclasses import dataclass from datetime import datetime, timezone from email.utils import formatdate from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from urllib.parse import urlsplit CLIENT_VERSION = "1.10.3838" DEFAULT_CONFIG = "/etc/papercut/queue.yaml" DEFAULT_LISTEN = "127.0.0.1" DEFAULT_PORT = 9123 DEFAULT_USER_AGENT = f"Print Deploy Client/{CLIENT_VERSION}/Debian" DEFAULT_TIMEOUT = 120.0 LOOPBACK_NAMES = {"127.0.0.1", "::1", "localhost"} LOG = logging.getLogger("papercut-ipp-relay") class RelayError(RuntimeError): """Configuration or request handling failed.""" # --------------------------------------------------------------------------- # IPP helpers (operation/status/job-id only; the body is forwarded verbatim) # --------------------------------------------------------------------------- IPP_OPS = { 0x0002: "Print-Job", 0x0003: "Print-URI", 0x0004: "Validate-Job", 0x0005: "Create-Job", 0x0006: "Send-Document", 0x0007: "Send-URI", 0x0008: "Cancel-Job", 0x0009: "Get-Job-Attributes", 0x000A: "Get-Jobs", 0x000B: "Get-Printer-Attributes", } IPP_STATUS = { 0x0000: "successful-ok", 0x0001: "successful-ok-ignored-or-substituted-attributes", 0x0002: "successful-ok-conflicting-attributes", 0x0400: "client-error-bad-request", 0x0401: "client-error-forbidden", 0x0402: "client-error-not-authenticated", 0x0403: "client-error-not-authorized", 0x0404: "client-error-not-possible", 0x0405: "client-error-timeout", 0x0406: "client-error-not-found", 0x0407: "client-error-gone", 0x0408: "client-error-request-entity-too-large", 0x040A: "client-error-document-format-not-supported", 0x040B: "client-error-attributes-or-values-not-supported", 0x0500: "server-error-internal-error", 0x0501: "server-error-operation-not-supported", 0x0502: "server-error-service-unavailable", } def ipp_operation(body: bytes): return int.from_bytes(body[2:4], "big") if len(body) >= 8 else None def ipp_request_id(body: bytes): return int.from_bytes(body[4:8], "big") if len(body) >= 8 else None def ipp_status(body: bytes): return int.from_bytes(body[2:4], "big") if len(body) >= 4 else None def parse_ipp_attributes(data: bytes) -> dict: """Parse top-level IPP attributes, stopping at end-of-attributes.""" out: dict[str, list] = {} pos = 8 last_name = None while pos < len(data): tag = data[pos] if tag == 0x03: # end-of-attributes break if tag < 0x10: # group delimiter pos += 1 continue if pos + 3 > len(data): break name_length = int.from_bytes(data[pos + 1 : pos + 3], "big") name_bytes = data[pos + 3 : pos + 3 + name_length] pos += 3 + name_length name = name_bytes.decode("latin-1") if name_bytes else last_name if name_bytes: last_name = name if pos + 2 > len(data): break value_length = int.from_bytes(data[pos : pos + 2], "big") value = data[pos + 2 : pos + 2 + value_length] pos += 2 + value_length if name: out.setdefault(name, []).append((tag, value)) return out def first_integer(attributes: dict, name: str): values = attributes.get(name) or [] if not values: return None return int.from_bytes(values[0][1], "big") def rewrite_printer_uri(body: bytes, printer_uri: str) -> tuple[bytes, bool]: """Replace the printer-uri operation attribute with the configured value. The stock ``ipp`` backend derives ``printer-uri`` from ``DEVICE_URI``. A CUPS queue whose device URI is a placeholder such as ``papercut-relay:/`` therefore sends a bare ``ipp://127.0.0.1:9123/``; the Mobility Print server identifies the queue from this attribute and rejects the bare form. The relay substitutes the canonical upstream printer-uri from its config before forwarding, which is also what the config's ``printer_uri`` is for. """ if len(body) < 8: return body, False encoded = printer_uri.encode("utf-8") if len(encoded) > 65535: return body, False pos = 8 last_name = None while pos < len(body): tag = body[pos] if tag == 0x03: # end-of-attributes break if tag < 0x10: # group delimiter pos += 1 continue if pos + 3 > len(body): break name_len = int.from_bytes(body[pos + 1 : pos + 3], "big") name_start = pos + 3 name_end = name_start + name_len if name_end > len(body): break name_bytes = body[name_start:name_end] name = name_bytes.decode("latin-1") if name_bytes else last_name if name_bytes: last_name = name if name_end + 2 > len(body): break value_len = int.from_bytes(body[name_end : name_end + 2], "big") value_start = name_end + 2 value_end = value_start + value_len if value_end > len(body): break if name == "printer-uri" and body[value_start:value_end] != encoded: return ( body[:name_end] + len(encoded).to_bytes(2, "big") + encoded + body[value_end:] ), True pos = value_end return body, False JOB_ATTR_NAMES = ( "media", "media-col", "print-color-mode", "ColorModel", "PageSize", "Duplex", "sides", "copies", "document-format", ) def _media_col_dimensions(body: bytes): """Read the first media-col media-size from a request (hundredths of mm).""" x_match = re.search(rb"x-dimension!\x00\x00\x00\x04(.{4})", body, re.S) y_match = re.search(rb"y-dimension!\x00\x00\x00\x04(.{4})", body, re.S) if not x_match or not y_match: return None x = int.from_bytes(x_match.group(1), "big") y = int.from_bytes(y_match.group(1), "big") if not x or not y: return None if x % 100 == 0 and y % 100 == 0: return f"{x // 100}x{y // 100}mm" return f"{x}x{y}" def _attribute_text(value) -> str: tag, raw = value if tag == 0x21 and len(raw) == 4: return str(int.from_bytes(raw, "big")) if tag in (0x41, 0x42, 0x44, 0x45, 0x47, 0x48, 0x49): return raw.decode("utf-8", "replace") return f"tag=0x{tag:02x}" def describe_request(body: bytes) -> str: """One short line about an IPP request, for the log.""" if len(body) < 8: return f"short/non-IPP body ({len(body)} bytes)" op = ipp_operation(body) op_name = IPP_OPS.get(op, f"operation-0x{op:04x}") bits = [op_name, f"request-id={ipp_request_id(body)}"] attributes = parse_ipp_attributes(body) job_id = first_integer(attributes, "job-id") if job_id is not None: bits.append(f"job-id={job_id}") for name in JOB_ATTR_NAMES: values = attributes.get(name) if not values: continue if name == "media-col": dimensions = _media_col_dimensions(body) bits.append( f"media-col={dimensions}" if dimensions else "media-col=collection" ) else: bits.append(f"{name}={_attribute_text(values[0])}") return " ".join(bits) def safe_text(value: str, limit: int = 200) -> str: """Keep log fields printable and bounded (client-controlled input).""" cleaned = "".join(ch if 32 <= ord(ch) < 127 else "?" for ch in value) if len(cleaned) > limit: cleaned = cleaned[:limit] + "..." return cleaned def iso_time(epoch) -> str: try: return datetime.fromtimestamp(int(epoch), tz=timezone.utc).isoformat() except (TypeError, ValueError, OverflowError, OSError): return str(epoch) # --------------------------------------------------------------------------- # Config (the flat YAML emitted by acquire_queue_info.py) # --------------------------------------------------------------------------- @dataclass(frozen=True) class RelayConfig: path: str printer: str mobility_base: str printer_uri: str jwt: str jwt_exp: int endpoint: str scheme: str host: str port: int request_path: str has_queue_url: bool def _parse_scalar(value: str, where: str): if value == "": return "" if value.startswith('"'): try: parsed = json.loads(value) except ValueError as exc: raise RelayError(f"{where}: invalid double-quoted string: {exc}") from exc if not isinstance(parsed, str): raise RelayError(f"{where}: quoted value is not a string") return parsed if value.startswith("'") and value.endswith("'") and len(value) >= 2: return value[1:-1].replace("''", "'") lowered = value.lower() if lowered in ("true", "yes", "on"): return True if lowered in ("false", "no", "off"): return False if lowered in ("null", "~"): return None if re.fullmatch(r"-?[0-9]+", value): return int(value) return value def load_flat_yaml(text: str, path: str) -> dict: """Parse the flat ``key: value`` YAML produced by acquire_queue_info.py.""" data = {} for lineno, raw in enumerate(text.splitlines(), 1): line = raw.strip() if not line or line.startswith("#"): continue if line in ("---", "..."): continue key, sep, value = line.partition(":") if not sep: raise RelayError(f"{path}:{lineno}: expected 'key: value'") key = key.strip() if not re.fullmatch(r"[A-Za-z0-9_][A-Za-z0-9_-]*", key): raise RelayError(f"{path}:{lineno}: unsupported key {key!r}") if key in data: raise RelayError(f"{path}:{lineno}: duplicate key {key!r}") data[key] = _parse_scalar(value.strip(), f"{path}:{lineno}") return data def decode_jwt_payload(token: str) -> dict: parts = token.split(".") if len(parts) != 3: raise RelayError("printer_jwt is not a JWT (expected 3 dot-separated parts)") part = parts[1] + "=" * (-len(parts[1]) % 4) try: payload = json.loads(base64.urlsafe_b64decode(part.encode("ascii"))) except (ValueError, UnicodeDecodeError) as exc: raise RelayError(f"printer_jwt payload cannot be decoded: {exc}") from exc if not isinstance(payload, dict): raise RelayError("printer_jwt payload is not a JSON object") exp = payload.get("exp") if not isinstance(exp, int): raise RelayError("printer_jwt has no integer 'exp' claim") return payload def load_config(path: str) -> RelayConfig: try: info = os.stat(path) except OSError as exc: raise RelayError(f"cannot stat config {path}: {exc}") from exc try: with open(path, encoding="utf-8") as fh: text = fh.read() except (OSError, UnicodeDecodeError) as exc: raise RelayError(f"cannot read config {path}: {exc}") from exc data = load_flat_yaml(text, path) token = data.get("printer_jwt") endpoint = data.get("ipp_endpoint") if not isinstance(token, str) or not token.strip(): raise RelayError(f"{path}: missing required 'printer_jwt'") token = token.strip() if not isinstance(endpoint, str) or not endpoint.strip(): raise RelayError(f"{path}: missing required 'ipp_endpoint'") endpoint = endpoint.strip() try: parts = urlsplit(endpoint) hostname = parts.hostname port = parts.port or (443 if parts.scheme == "https" else 80) except ValueError as exc: raise RelayError(f"{path}: invalid ipp_endpoint: {exc}") from exc if parts.scheme not in ("http", "https"): raise RelayError(f"{path}: ipp_endpoint must be http or https") if not hostname: raise RelayError(f"{path}: ipp_endpoint has no host") if parts.username or parts.password: raise RelayError(f"{path}: ipp_endpoint must not contain credentials") request_path = parts.path or "/" if parts.query: request_path += "?" + parts.query payload = decode_jwt_payload(token) exp = int(payload["exp"]) if exp < time.time(): raise RelayError( f"printer_jwt expired at {iso_time(exp)}; re-run " "acquire_queue_info.py and restart the relay" ) printer_uri = data.get("printer_uri") or endpoint if not isinstance(printer_uri, str): raise RelayError(f"{path}: 'printer_uri' must be a string") printer = data.get("printer") or "" mobility = data.get("mobility_base") or "" for name, value in (("printer", printer), ("mobility_base", mobility)): if not isinstance(value, str): raise RelayError(f"{path}: {name!r} must be a string") return RelayConfig( path=os.path.abspath(path), printer=printer, mobility_base=mobility, printer_uri=printer_uri, jwt=token, jwt_exp=exp, endpoint=endpoint, scheme=parts.scheme, host=hostname, port=port, request_path=request_path, has_queue_url=bool(data.get("queue_url")), ) def log_config(config: RelayConfig, mode: int): LOG.info("config: %s (mode %04o)", config.path, stat.S_IMODE(mode)) if stat.S_IMODE(mode) & 0o077: LOG.warning( "config is group/world-readable; it contains a bearer credential " "and should normally be mode 0600" ) if stat.S_IMODE(mode) & 0o022: LOG.warning( "config is writable by group/others; the JWT could be replaced by " "another user" ) if config.printer: LOG.info("printer: %s", config.printer) if config.mobility_base: LOG.info("mobility base: %s", config.mobility_base) LOG.info("upstream: %s", config.endpoint) LOG.debug("canonical IPP printer-uri attribute (config): %s", config.printer_uri) days = (config.jwt_exp - time.time()) / 86400.0 LOG.info( "printer JWT expires %s (%.1f days remaining)", iso_time(config.jwt_exp), days, ) if config.has_queue_url: LOG.debug("queue_url is present in the config but is not used by the relay") # --------------------------------------------------------------------------- # Upstream request # --------------------------------------------------------------------------- @dataclass class UpstreamResponse: status: int reason: str headers: dict body: bytes elapsed: float def make_ssl_context(verify: bool, cacert): if cacert: return ssl.create_default_context(cafile=cacert) if verify: return ssl.create_default_context() context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) context.check_hostname = False context.verify_mode = ssl.CERT_NONE return context def forward_ipp(config: RelayConfig, ssl_context, timeout: float, user_agent: str, body: bytes, client_ip: str, xff_header) -> UpstreamResponse: """POST one IPP body to the configured upstream; always Bearer JWT.""" if config.scheme == "https": conn = http.client.HTTPSConnection( config.host, config.port, timeout=timeout, context=ssl_context ) else: conn = http.client.HTTPConnection( config.host, config.port, timeout=timeout ) xff = xff_header if xff_header else client_ip xff = xff.replace("\r", "").replace("\n", "")[:400] headers = { "User-Agent": user_agent, "Authorization": f"Bearer {config.jwt}", "Content-Type": "application/ipp", "Accept-Encoding": "identity", "Date": formatdate(time.time(), usegmt=True), "X-Forwarded-For": xff, } started = time.monotonic() try: conn.putrequest("POST", config.request_path, skip_accept_encoding=True) for name, value in headers.items(): conn.putheader(name, value) conn.putheader("Content-Length", str(len(body))) conn.endheaders(body) raw = conn.getresponse() data = raw.read() response_headers = {k.lower(): v for k, v in raw.getheaders()} status, reason = raw.status, raw.reason except (OSError, http.client.HTTPException) as exc: raise RelayError(f"upstream POST failed: {exc}") from exc finally: conn.close() if response_headers.get("content-encoding", "").lower() == "gzip" and data: try: data = gzip.decompress(data) except OSError: LOG.warning("upstream sent unreadable gzip data; relaying it as-is") return UpstreamResponse( status=status, reason=reason, headers=response_headers, body=data, elapsed=time.monotonic() - started, ) # --------------------------------------------------------------------------- # HTTP server/handler # --------------------------------------------------------------------------- class RelayServer(ThreadingHTTPServer): daemon_threads = True allow_reuse_address = True def __init__(self, address, handler, config: RelayConfig, args, ssl_context): super().__init__(address, handler) self.config = config self.args = args self.ssl_context = ssl_context self.started_at = time.time() self.stats_lock = threading.Lock() self.requests = 0 self.errors = 0 def count_request(self): with self.stats_lock: self.requests += 1 def count_error(self): with self.stats_lock: self.errors += 1 class RelayServerV6(RelayServer): address_family = socket.AF_INET6 class RelayHandler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" server_version = "PaperCutIPPRelay/1.0" sys_version = "" def setup(self): super().setup() try: timeout = max(float(self.server.args.timeout), 5.0) self.connection.settimeout(timeout) except (AttributeError, OSError, TypeError, ValueError): pass def log_message(self, fmt, *args): # BaseHTTPRequestHandler access log LOG.debug("http: " + fmt, *args) def log_error(self, fmt, *args): LOG.warning("http: " + fmt, *args) def _send_text(self, status: int, text: str): data = text.encode("utf-8", "replace") self.send_response(status) self.send_header("Content-Type", "text/plain; charset=utf-8") self.send_header("Content-Length", str(len(data))) self.send_header("Connection", "close") self.end_headers() self.close_connection = True try: self.wfile.write(data) except (BrokenPipeError, ConnectionResetError): pass def _send_ipp(self, upstream: UpstreamResponse): body = upstream.body ctype = upstream.headers.get("content-type") or "application/ipp" self.send_response(upstream.status) self.send_header("Content-Type", ctype) self.send_header("Content-Length", str(len(body))) self.send_header("Connection", "close") self.end_headers() self.close_connection = True try: self.wfile.write(body) except (BrokenPipeError, ConnectionResetError): LOG.debug("client disconnected while the response was being sent") def do_GET(self): server = self.server # type: RelayServer config = server.config uptime = int(time.time() - server.started_at) with server.stats_lock: requests = server.requests errors = server.errors text = "\n".join([ "papercut-ipp-relay ok", f"config: {config.path}", f"printer: {config.printer or '(not set)'}", f"upstream: {config.endpoint}", "incoming path: logged and ignored (upstream is fixed by config)", f"jwt_expires_at: {iso_time(config.jwt_exp)}", f"uptime_seconds: {uptime}", f"requests: {requests}", f"errors: {errors}", "", ]) incoming_path = urlsplit(self.path).path or "/" LOG.debug( "GET %s from %s:%s -> health page", safe_text(incoming_path), self.client_address[0], self.client_address[1], ) self._send_text(200, text) def do_POST(self): server = self.server # type: RelayServer config = server.config server.count_request() remote = f"{self.client_address[0]}:{self.client_address[1]}" content_type = self.headers.get("Content-Type", "") media_type = content_type.split(";", 1)[0].strip().lower() if media_type != "application/ipp": LOG.warning( "rejecting POST from %s: Content-Type %r is not application/ipp", remote, safe_text(content_type, 80), ) server.count_error() self._send_text(415, "expected Content-Type: application/ipp\n") return started = time.monotonic() try: body = self._read_body() except RelayError as exc: LOG.warning("cannot read request body from %s: %s", remote, exc) server.count_error() self._send_text(400, f"cannot read IPP body: {exc}\n") return body, uri_rewritten = rewrite_printer_uri(body, config.printer_uri) if uri_rewritten: LOG.debug( "rewrote printer-uri to the configured value %s", config.printer_uri, ) framing = "chunked" if "chunked" in self.headers.get( "Transfer-Encoding", "").lower() else ( f"Content-Length={self.headers.get('Content-Length', '0')}") # Log only the path; a query string could contain sensitive values. incoming_path = urlsplit(self.path).path or "/" LOG.info( "request from %s: POST %s (path ignored), %s, %d bytes", remote, safe_text(incoming_path), framing, len(body), ) LOG.info("IPP request: %s", describe_request(body)) try: upstream = forward_ipp( config, server.ssl_context, server.args.timeout, server.args.user_agent, body, self.client_address[0], self.headers.get("X-Forwarded-For"), ) except RelayError as exc: server.count_error() LOG.error("upstream request failed: %s", exc) self._send_text(502, f"upstream request failed: {exc}\n") return status = ipp_status(upstream.body) status_name = ( IPP_STATUS.get(status, f"unknown-0x{status:04x}") if status is not None else "no IPP status" ) job_id = first_integer(parse_ipp_attributes(upstream.body), "job-id") job_text = f", job-id={job_id}" if job_id is not None else "" if upstream.status in (401, 403): server.count_error() LOG.error( "upstream HTTP %d %s: the Printer JWT may be expired/revoked " "or the queue may have changed; re-run acquire_queue_info.py " "and restart the relay", upstream.status, upstream.reason, ) elif upstream.status != 200: server.count_error() LOG.warning( "upstream HTTP %d %s; relaying the response", upstream.status, upstream.reason, ) elif status is not None and status not in (0x0000, 0x0001, 0x0002): server.count_error() LOG.warning( "upstream IPP status 0x%04x (%s)%s; relaying the response", status, status_name, job_text, ) LOG.info( "upstream result: HTTP %d %s, IPP %s%s, %d bytes in %.2fs", upstream.status, upstream.reason, status_name, job_text, len(upstream.body), upstream.elapsed, ) LOG.info( "relayed response to %s (%d bytes, total %.2fs)", remote, len(upstream.body), time.monotonic() - started, ) self._send_ipp(upstream) def _read_body(self) -> bytes: transfer_encoding = self.headers.get("Transfer-Encoding", "").lower() if "chunked" in transfer_encoding: return self._read_chunked() length_header = self.headers.get("Content-Length") if length_header is None: return b"" try: length = int(length_header) except ValueError as exc: raise RelayError(f"invalid Content-Length {length_header!r}") from exc if length < 0: raise RelayError(f"negative Content-Length {length}") data = self.rfile.read(length) if len(data) != length: raise RelayError( f"truncated body: expected {length} bytes, got {len(data)}" ) return data def _read_chunked(self) -> bytes: chunks = [] while True: line = self.rfile.readline(65537) if not line: raise RelayError("truncated chunk header") size_text = line.split(b";", 1)[0].strip() try: size = int(size_text, 16) except ValueError as exc: raise RelayError(f"invalid chunk size {size_text!r}") from exc if size == 0: # Consume trailers up to the empty line. while True: trailer = self.rfile.readline(65537) if trailer in (b"\r\n", b"\n", b""): break break if size > 256 * 1024 * 1024: raise RelayError(f"chunk too large ({size} bytes)") data = self.rfile.read(size) if len(data) != size: raise RelayError( f"truncated chunk: expected {size} bytes, got {len(data)}" ) chunks.append(data) terminator = self.rfile.read(2) if terminator != b"\r\n": raise RelayError(f"invalid chunk terminator {terminator!r}") return b"".join(chunks) # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser( description="Loopback IPP relay: CUPS -> this relay -> PaperCut Mobility " "Print, using the Printer JWT from the queue YAML.", formatter_class=argparse.ArgumentDefaultsHelpFormatter, ) p.add_argument("--config", default=DEFAULT_CONFIG, help="queue YAML produced by acquire_queue_info.py") p.add_argument("--listen", default=DEFAULT_LISTEN, help="local listen address (loopback only by default)") p.add_argument("--port", type=int, default=DEFAULT_PORT, help="local listen port for the CUPS device URI") p.add_argument("--allow-non-loopback", action="store_true", help="permit binding a non-loopback listen address (unsafe: " "the relay injects a bearer credential)") p.add_argument("--timeout", type=float, default=DEFAULT_TIMEOUT, help="upstream request timeout in seconds") p.add_argument("--verify", action="store_true", help="verify the upstream TLS certificate (off by default)") p.add_argument("--cacert", help="CA bundle to trust (implies verification)") p.add_argument("--user-agent", default=DEFAULT_USER_AGENT, help="User-Agent sent upstream") p.add_argument("--debug", action="store_true", help="enable debug logging (headers are not dumped)") p.add_argument("--check", action="store_true", help="validate the config and exit without binding") return p def configure_logging(debug: bool): logging.basicConfig( stream=sys.stderr, level=logging.DEBUG if debug else logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s", ) def main(argv=None) -> int: args = build_parser().parse_args(argv) configure_logging(args.debug) if not (1 <= args.port <= 65535): LOG.error("--port must be between 1 and 65535") return 2 if args.listen not in LOOPBACK_NAMES and not args.allow_non_loopback: LOG.error( "--listen %r is not loopback; this relay injects a bearer " "credential. Use --allow-non-loopback to override.", args.listen, ) return 2 try: config = load_config(args.config) mode = os.stat(args.config).st_mode except (RelayError, OSError) as exc: LOG.error("%s", exc) return 2 log_config(config, mode) if args.check: LOG.info("config check OK") return 0 ssl_context = make_ssl_context(args.verify, args.cacert) LOG.info("TLS verification: %s", "on" if (args.verify or args.cacert) else "OFF") server_class = RelayServerV6 if ":" in args.listen else RelayServer try: server = server_class( (args.listen, args.port), RelayHandler, config, args, ssl_context ) except OSError as exc: LOG.error("cannot listen on %s:%d: %s", args.listen, args.port, exc) return 2 LOG.info( "listening on %s:%d; CUPS device URI: ipp://%s:%d/printers/%s", args.listen, args.port, args.listen, args.port, config.printer or "printer", ) LOG.info("incoming request paths are logged and ignored; upstream is fixed " "by the config") def stop(signum, _frame): LOG.info("received signal %s; shutting down", signum) threading.Thread(target=server.shutdown, daemon=True).start() signal.signal(signal.SIGTERM, stop) signal.signal(signal.SIGINT, stop) try: server.serve_forever(poll_interval=0.5) finally: server.server_close() LOG.info("stopped") return 0 if __name__ == "__main__": try: raise SystemExit(main()) except KeyboardInterrupt: raise SystemExit(130)