#!/usr/bin/env python3 """ Bondrucker-Bruecke — laeuft auf dem Raspberry Pi am Kassendrucker. Warum es diesen Dienst ueberhaupt gibt: der Bondrucker haengt per USB am Pi (/dev/usb/lp0) und ist damit vom Timer-Server aus nicht erreichbar. Dieser Dienst nimmt fertige ESC/POS-Daten per HTTP entgegen und schiebt sie in das Geraet. Die Timer-App baut den Bon, der Pi druckt ihn nur noch. Bewusst nur Python-Standardbibliothek: auf dem Pi muss dafuer nichts nachinstalliert werden (kein pip, kein CUPS, kein node). Welcher Drucker genommen wird, entscheidet die Timer-App: sie fragt ueber /devices ab, was angeschlossen ist, und schickt den gewaehlten Pfad bei jedem Auftrag im Kopf X-Print-Device mit. Dieser Dienst haelt dazu keinen Zustand - die Einstellung lebt an einer Stelle, naemlich dort, wo sie bedient wird. Endpunkte GET /status Zustand des Druckers als JSON (fuer Health-Checks) GET /devices angeschlossene Drucker mit Klartextnamen (Token noetig) POST /print Rumpf = rohe ESC/POS-Bytes, wird 1:1 an den Drucker gereicht /status und /print beachten den Kopf X-Print-Device. Fehlt er, wird das eingestellte Geraet genommen und sonst das erste gefundene. Konfiguration ueber Umgebungsvariablen (siehe bondrucker.service): PRINT_DEVICE Rueckfall, wenn die App nichts vorgibt; Standard /dev/usb/lp0 PRINT_HOST Lausch-Adresse, Standard 0.0.0.0 PRINT_PORT Lausch-Port, Standard 9100 PRINT_TOKEN gemeinsames Geheimnis; leer = ohne Anmeldung (nur im LAN!) PRINT_MAX_BYTES Obergrenze je Auftrag, Standard 262144 """ import glob import hmac import json import os import re import socketserver import sys import threading import time from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer VERSION = "1.2" DEVICE = os.environ.get("PRINT_DEVICE", "/dev/usb/lp0") # Welcher Drucker genommen wird, entscheidet die Timer-App und schickt es bei # jedem Auftrag mit. Der Pfad kommt damit von aussen - ohne diese Schranke # liesse sich der Dienst dazu bringen, in eine beliebige Datei zu schreiben. DEVICE_MUSTER = re.compile(r"^/dev/(usb/)?lp[0-9]{1,3}$") HOST = os.environ.get("PRINT_HOST", "0.0.0.0") PORT = int(os.environ.get("PRINT_PORT", "9100")) TOKEN = os.environ.get("PRINT_TOKEN", "").strip() MAX_BYTES = int(os.environ.get("PRINT_MAX_BYTES", str(256 * 1024))) # Zwei Auftraege duerfen sich nicht ineinander schieben - sonst kommt Buchstaben- # salat aus dem Drucker. Ein Prozess, ein Schloss. _print_lock = threading.Lock() _last_error = None _last_ok = None def log(msg): print(msg, flush=True) def _lies(pfad): try: with open(pfad) as f: return f.read().strip() except OSError: return "" def usb_name(geraet): """Klartextname eines USB-Druckers, etwa "EPSON TM-T88IV". Dahinter steckt ein Aufstieg im sysfs: die Geraetedatei zeigt auf die USB-Schnittstelle, die Bezeichnungen haengen aber am Geraet darueber. Ohne diesen Namen steht in der Auswahl nur "lp0" und "lp1" - bei zwei Druckern ist das Raten. """ basis = os.path.basename(geraet) start = "/sys/class/usbmisc/%s/device" % basis if not os.path.exists(start): start = "/sys/class/printer/%s/device" % basis if not os.path.exists(start): return "" pfad = os.path.realpath(start) while pfad and pfad != "/": if os.path.isfile(os.path.join(pfad, "product")): name = _lies(os.path.join(pfad, "manufacturer")) + " " + \ _lies(os.path.join(pfad, "product")) return " ".join(name.split()) pfad = os.path.dirname(pfad) return "" def list_devices(): """Alle angeschlossenen Druckergeraete - die Auswahl fuer die Timer-App.""" pfade = sorted(set(glob.glob("/dev/usb/lp*")) | set(glob.glob("/dev/lp*"))) return [ { "path": p, "name": usb_name(p), "writable": os.access(p, os.W_OK), } for p in pfade if DEVICE_MUSTER.match(p) ] def find_device(requested=None): """Welches Geraet bedient werden soll. Mit Angabe: genau dieses, sofern der Pfad ueberhaupt ein Druckergeraet sein kann. Ohne Angabe das eingestellte, und wenn es das nicht gibt, das erste vorhandene - nach dem Aus- und Wiedereinstecken des USB-Kabels kann aus lp0 naemlich ein lp1 werden, und ohne diesen Rueckfall druckt danach nichts mehr, bis jemand den Dienst anfasst. """ if requested: if not DEVICE_MUSTER.match(requested): raise ValueError("Unerlaubter Geraetepfad: %s" % requested) return requested if os.path.exists(DEVICE): return DEVICE found = sorted(glob.glob("/dev/usb/lp*")) return found[0] if found else None def device_status(token_accepted=None, requested=None): try: dev = find_device(requested) except ValueError as err: return { "ok": False, "version": VERSION, "tokenRequired": bool(TOKEN), "tokenAccepted": token_accepted, "device": requested, "deviceExists": False, "writable": False, "lastError": str(err), "lastPrintAt": _last_ok, } status = { "ok": False, "version": VERSION, # Ob das mitgeschickte Token passt. Der Zustand des Druckers selbst ist # nicht geheim und bleibt fuer Health-Checks ohne Token abfragbar - aber # wer einrichtet, muss pruefen koennen, ob auch das DRUCKEN klappen # wird. Ohne diese Angabe meldet ein Dienst, der noch mit dem alten # Token laeuft, faelschlich "alles gut". "tokenRequired": bool(TOKEN), "tokenAccepted": token_accepted, "device": dev or DEVICE, "deviceConfigured": DEVICE, "deviceRequested": requested or None, "deviceExists": dev is not None and os.path.exists(dev), "writable": False, "lastError": _last_error, "lastPrintAt": _last_ok, } if dev and os.path.exists(dev): status["writable"] = os.access(dev, os.W_OK) status["ok"] = status["writable"] return status def write_to_printer(data, requested=None): """Schreibt einen Auftrag auf den Drucker. Wirft bei Fehlern OSError.""" global _last_error, _last_ok with _print_lock: dev = find_device(requested) if dev is None: raise OSError("Kein Druckergeraet gefunden (/dev/usb/lp*). Kabel? Strom?") # os.open statt open(): kein Puffer dazwischen, und Fehler des Druckers # (Papier leer, Deckel offen) kommen direkt als OSError zurueck. fd = os.open(dev, os.O_WRONLY) try: written = 0 while written < len(data): written += os.write(fd, data[written:]) finally: os.close(fd) _last_error = None _last_ok = time.strftime("%Y-%m-%d %H:%M:%S") return written def token_ok(supplied): if not TOKEN: return True return hmac.compare_digest(TOKEN, (supplied or "").strip()) class Handler(BaseHTTPRequestHandler): server_version = "bondrucker/" + VERSION protocol_version = "HTTP/1.1" def _send(self, code, payload): body = json.dumps(payload, ensure_ascii=False).encode("utf-8") self.send_response(code) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def do_GET(self): pfad = self.path.split("?")[0].rstrip("/") if pfad in ("/status", "/health", ""): st = device_status( token_ok(self.headers.get("X-Print-Token")), self.headers.get("X-Print-Device"), ) self._send(200 if st["ok"] else 503, st) elif pfad == "/devices": # Anders als /status mit Token: welche Geraete am Pi haengen, ist # zwar kein Geheimnis, aber auch keine Auskunft fuer jeden im Netz. if not token_ok(self.headers.get("X-Print-Token")): return self._send(403, {"ok": False, "error": "Token ungueltig"}) self._send(200, {"ok": True, "devices": list_devices(), "fallback": find_device()}) else: self._send(404, {"ok": False, "error": "Unbekannter Pfad"}) def do_POST(self): global _last_error if self.path.split("?")[0].rstrip("/") != "/print": return self._send(404, {"ok": False, "error": "Unbekannter Pfad"}) if not token_ok(self.headers.get("X-Print-Token")): log(f"[abgelehnt] falsches Token von {self.client_address[0]}") return self._send(403, {"ok": False, "error": "Token ungueltig"}) try: length = int(self.headers.get("Content-Length") or 0) except ValueError: return self._send(400, {"ok": False, "error": "Content-Length fehlt"}) if length <= 0: return self._send(400, {"ok": False, "error": "Leerer Druckauftrag"}) if length > MAX_BYTES: return self._send(413, {"ok": False, "error": f"Auftrag groesser als {MAX_BYTES} Bytes"}) data = self.rfile.read(length) try: written = write_to_printer(data, self.headers.get("X-Print-Device")) except ValueError as err: log("[abgelehnt] %s" % err) return self._send(400, {"ok": False, "error": str(err)}) except OSError as err: _last_error = str(err) log(f"[fehler] {self.client_address[0]}: {err}") return self._send(502, {"ok": False, "error": f"Drucker meldet: {err}"}) log(f"[gedruckt] {written} Bytes von {self.client_address[0]}") self._send(200, {"ok": True, "bytes": written}) # Zugriffszeilen selbst formatieren; das Standardformat von # BaseHTTPRequestHandler geht auf stderr und verrauscht das Journal. def log_message(self, fmt, *args): pass class Server(ThreadingHTTPServer): daemon_threads = True allow_reuse_address = True def main(): st = device_status() log(f"Bondrucker-Bruecke {VERSION}") log(f" Geraet: {st['device']} (vorhanden: {st['deviceExists']}, beschreibbar: {st['writable']})") for g in list_devices(): log(f" gefunden: {g['path']} {g['name']}") log(f" Lauscht: {HOST}:{PORT}") log(f" Token: {'gesetzt' if TOKEN else 'NICHT gesetzt - jeder im Netz darf drucken'}") if not st["deviceExists"]: log(" Hinweis: Drucker gerade nicht da. Der Dienst laeuft trotzdem und") log(" findet das Geraet, sobald es wieder angesteckt ist.") try: Server((HOST, PORT), Handler).serve_forever() except KeyboardInterrupt: log("Beendet.") sys.exit(0) if __name__ == "__main__": main()