# -*- coding: utf-8 -*- """Mini-API zwischen Rechnungstool und Steuerjournal. WARUM ----- Beide Programme stehen nebeneinander offen. Schreibt das Rechnungstool eine Rechnung, soll sie im Journal SOFORT auftauchen - nicht erst beim naechsten Einlesen und nicht nach einer Wartezeit. Seine Ansage dazu: *"machst ein kleinen py server der hin und her pingt mit einem push vom rechnungstool: eh schau mal in die sqlite"* und *"die muessen ja sehen das die beiden auf sind"*. WIE --- Jedes Programm macht beim Start einen winzigen Server auf **127.0.0.1** mit einem Port, den das Betriebssystem aussucht (Port 0). Wo es lauscht, traegt es in die gemeinsame SQLite ein - die ist das Telefonbuch. Wer jemanden erreichen will, schlaegt dort Port und PID nach, prueft die PID und ruft an. Zwei Nachrichten reichen dafuer: {"was": "ping"} -> {"antwort": "hallo, bin noch da"} {"was": "schau_nach"} -> {"antwort": "ok"} (und das Fenster liest nach) Dazu kommt eine dritte, die das Fenster BEDIENT - damit ein Pruefstand die Knoepfe druecken kann, ohne dem Menschen am Rechner Maus und Tastatur wegzunehmen: {"was": "aktion", "schluessel": "...", "name": "vorschau"} -> {"antwort": {...}} DER SCHLUESSEL -------------- Wer bedient, muss sich ausweisen - wie mit einem SSH-Schluessel. Beim ersten Start legt das Programm eine Zufallszahl in `api_schluessel.txt` **im gemeinsamen Ordner** ab (dort steht auch das Nummernbuch). Wer diesen Ordner lesen darf, darf sprechen: das Steuerjournal und der Betreiber - sonst niemand. Ohne gueltigen Schluessel wird eine Aktion verworfen, und zwar ohne zu verraten, ob der Name ueberhaupt existiert. `ping` und `schau_nach` bleiben schluessellos: sie sagen nur "ich lebe" bzw. "schau in die SQLite" und veraendern nichts. Sonst koennte ein aelteres Steuerjournal nicht mehr anklopfen. WAS EINE AKTION DARF -------------------- Alles, was einen BELEG SCHREIBT (Rechnung, Angebot, Storno), geht nur im Testlauf (`--test` / `CAMPINGHOF_TEST=1`). Im normalen Betrieb bleiben nur Aktionen, die nichts in die Buchhaltung schreiben: nachsehen, Felder fuellen, Vorschau zeigen. Eine Fernbedienung, die echte Rechnungen ausstellt, hat in einer Buchhaltung nichts zu suchen. Der Push traegt keine Daten. Er sagt nur "schau in die SQLite" - dort steht die Meldung ohnehin. Damit gibt es genau EINE Wahrheit, und wenn der Anruf nicht durchgeht (Programm zu, Port belegt), findet das andere die Meldung beim naechsten Nachsehen trotzdem. SICHERHEIT ---------- Gebunden wird ausschliesslich auf 127.0.0.1 - von aussen ist da nichts zu erreichen, und Windows fragt bei Loopback nicht nach der Firewall. Angenommen werden nur die zwei Woerter oben; alles andere wird verworfen. Die API fuehrt keine Befehle aus und uebertraegt keine Daten. Diese Datei liegt in BEIDEN Programmen gleich. Wer sie aendert, kopiert sie hinueber. """ import json import logging import os import queue import secrets import socket import socketserver import sys import threading log = logging.getLogger("api") ANTWORT_HALLO = "hallo, bin noch da" SCHLUESSEL_DATEI = "api_schluessel.txt" def schluessel_holen(ordner): """Schluessel aus dem Ordner lesen; ist keiner da, einen anlegen. Der Schluessel ist eine 32 Byte lange Zufallszahl in Hex. Er liegt als Datei neben dem Nummernbuch, damit beide Programme denselben lesen - eine Absprache ueber den gemeinsamen Ordner, ohne Server und ohne Passwort, das sich jemand merken muesste. """ if not ordner: return "" pfad = os.path.join(ordner, SCHLUESSEL_DATEI) try: if os.path.exists(pfad): wert = open(pfad, encoding="utf-8").read().strip() if wert: return wert wert = secrets.token_hex(32) with open(pfad, "w", encoding="utf-8") as f: f.write(wert) log.info("neuer API-Schluessel angelegt: %s", pfad) return wert except OSError as e: log.warning("API-Schluessel nicht nutzbar (%s)", e) return "" class _Handler(socketserver.StreamRequestHandler): timeout = 3 def handle(self): # Nur der eigene Rechner darf reden - gebunden ist ohnehin nur Loopback. if self.client_address[0] not in ("127.0.0.1", "::1"): return try: zeile = self.rfile.readline(4096).decode("utf-8", "replace").strip() except OSError: return if not zeile: return try: nachricht = json.loads(zeile) was = str(nachricht.get("was", "")) except (ValueError, AttributeError): nachricht, was = {}, "" if was == "ping": antwort = ANTWORT_HALLO elif was == "schau_nach": antwort = "ok" self.server.melden() # Fenster benachrichtigen (threadsicher) elif was == "aktion": antwort = self._aktion(nachricht) else: antwort = "haeh?" try: self.wfile.write((json.dumps({"antwort": antwort}) + "\n").encode("utf-8")) except OSError: pass def _aktion(self, nachricht): """Eine Bedienung ausfuehren - nur mit gueltigem Schluessel.""" erwartet = getattr(self.server, "schluessel", "") gegeben = str(nachricht.get("schluessel", "")) # compare_digest statt "==": ein Vergleich, der bei der ersten falschen # Stelle abbricht, verraet ueber die Zeit, wie weit man richtig lag. if not erwartet or not secrets.compare_digest(erwartet, gegeben): log.warning("Aktion ohne gueltigen Schluessel abgewiesen") return "kein zutritt" if self.server.aktion is None: return "keine bedienung moeglich" name = str(nachricht.get("name", "")) daten = nachricht.get("daten") or {} log.info("Aktion ueber die API: %s", name) try: return self.server.aktion(name, daten) except Exception as e: # noqa: BLE001 - der Server darf daran nicht sterben log.warning("Aktion %r fehlgeschlagen: %s", name, e) return {"fehler": str(e)} class _Server(socketserver.ThreadingTCPServer): allow_reuse_address = True daemon_threads = True def __init__(self, adresse, bei_push, aktion=None, schluessel=""): self.melden = bei_push self.aktion = aktion self.schluessel = schluessel super().__init__(adresse, _Handler) class MiniApi: """Der kleine Server eines Programms. `bei_push` wird aus einem FREMDEN Faden gerufen - Tkinter darf man von dort nicht anfassen. Der Aufrufer legt die Meldung deshalb in eine Warteschlange und holt sie im Fenster mit after() ab. """ def __init__(self, programm, bei_push, aktion=None): self.programm = programm self.bei_push = bei_push self.aktion = aktion # (name, daten) -> Antwort; None = keine Bedienung self.server = None self.faden = None self.port = 0 self.schluessel = "" def starten(self, buch, ordner=None): """Server aufmachen und im Telefonbuch (SQLite) eintragen. Der Schluessel kommt aus dem Ordner des Nummernbuchs - dort liegt er fuer beide Programme lesbar. Gibt es kein Buch, gibt es keinen Schluessel und damit auch keine Bedienung von aussen. """ if self.server is not None: return self.port if ordner is None and buch is not None: ordner = os.path.dirname(getattr(buch, "pfad", "") or "") self.schluessel = schluessel_holen(ordner) try: self.server = _Server(("127.0.0.1", 0), self.bei_push, aktion=self.aktion, schluessel=self.schluessel) except OSError as e: log.warning("API konnte nicht starten: %s", e) self.server = None return 0 self.port = self.server.server_address[1] self.faden = threading.Thread(target=self.server.serve_forever, name=f"api-{self.programm}", daemon=True) self.faden.start() log.info("API von %s laeuft auf 127.0.0.1:%s", self.programm, self.port) self.eintragen(buch) return self.port def eintragen(self, buch): """Port und PID ins Telefonbuch. Ohne Buch gibt es keine API-Auskunft.""" if buch is None or not self.port: return try: buch.anmelden(self.programm, os.getpid(), os.path.basename(sys.executable), self.port) except Exception as e: # noqa: BLE001 log.warning("Eintrag ins Telefonbuch fehlgeschlagen: %s", e) def stoppen(self, buch=None): if buch is not None: try: buch.abmelden(self.programm) except Exception: # noqa: BLE001 - beim Schliessen nichts erzwingen pass if self.server is not None: try: self.server.shutdown() self.server.server_close() except Exception: # noqa: BLE001 pass self.server = None self.port = 0 def _reden(port, was, zeitgrenze=1.5, **rest): """Eine Zeile hinschicken und die Antwort lesen. None = niemand da.""" nachricht = {"was": was} nachricht.update(rest) try: with socket.create_connection(("127.0.0.1", int(port)), timeout=zeitgrenze) as s: s.settimeout(zeitgrenze) s.sendall((json.dumps(nachricht) + chr(10)).encode("utf-8")) roh = s.makefile("r", encoding="utf-8").readline() return json.loads(roh).get("antwort", "") except (OSError, ValueError): return None def erreichbar(buch, programm): """Meldet sich das andere Programm? ("hallo, bin noch da") Zuerst die PID - die beantwortet Windows und kostet nichts. Erst wenn die stimmt, wird ueberhaupt angerufen. """ if buch is None: return False try: eintrag = buch.wer_laeuft(programm) except Exception as e: # noqa: BLE001 log.debug("Telefonbuch nicht lesbar: %s", e) return False if not eintrag or not eintrag.get("port"): return False return _reden(eintrag["port"], "ping") == ANTWORT_HALLO def anklopfen(buch, programm): """Push: "schau mal in die sqlite". True, wenn es angekommen ist.""" if buch is None: return False try: eintrag = buch.wer_laeuft(programm) except Exception as e: # noqa: BLE001 log.debug("Telefonbuch nicht lesbar: %s", e) return False if not eintrag or not eintrag.get("port"): return False return _reden(eintrag["port"], "schau_nach") == "ok" def bedienen(buch, programm, name, daten=None, ordner=None, zeitgrenze=15.0): """Einen Knopf im anderen Programm druecken. None = niemand da / kein Zutritt. Fuer Pruefstaende und fuer den Betreiber: damit laesst sich das Fenster bedienen, ohne dem Menschen am Rechner Maus und Tastatur wegzunehmen. Der Schluessel kommt aus demselben Ordner wie das Nummernbuch - wer den lesen darf, darf bedienen. """ if buch is None: return None try: eintrag = buch.wer_laeuft(programm) except Exception as e: # noqa: BLE001 log.debug("Telefonbuch nicht lesbar: %s", e) return None if not eintrag or not eintrag.get("port"): return None if ordner is None: ordner = os.path.dirname(getattr(buch, "pfad", "") or "") return _reden(eintrag["port"], "aktion", zeitgrenze=zeitgrenze, schluessel=schluessel_holen(ordner), name=name, daten=daten or {}) class Warteschlange: """Aktionen aus dem Netzfaden in den Fensterfaden bringen - und zurueck. Tkinter darf nur aus SEINEM Faden angefasst werden. Der API-Server lauscht aber in einem eigenen. Deshalb legt er die Aktion hier ab und wartet auf das Ergebnis; das Fenster holt sie in seiner after()-Schleife ab, fuehrt sie aus und legt die Antwort zurueck. Kommt das Fenster nicht dazu (es haengt in einem Dialog), laeuft die Wartezeit ab und der Anrufer bekommt eine ehrliche Absage statt einer Verbindung, die nie antwortet. """ def __init__(self): self.eingang = queue.Queue() def annehmen(self, name, daten, zeitgrenze=12.0): """Aus dem Netzfaden gerufen: Aktion einreihen und auf das Ergebnis warten.""" fertig = threading.Event() auftrag = {"name": name, "daten": daten, "fertig": fertig, "antwort": None} self.eingang.put(auftrag) if not fertig.wait(zeitgrenze): return {"fehler": "das Fenster hat nicht geantwortet"} return auftrag["antwort"] def abarbeiten(self, ausfuehren): """Im Fensterfaden gerufen: alles erledigen, was eingegangen ist.""" while True: try: auftrag = self.eingang.get_nowait() except queue.Empty: return try: auftrag["antwort"] = ausfuehren(auftrag["name"], auftrag["daten"]) except Exception as e: # noqa: BLE001 - eine Aktion darf das Fenster nicht killen log.warning("Aktion %r fehlgeschlagen: %s", auftrag["name"], e) auftrag["antwort"] = {"fehler": str(e)} finally: auftrag["fertig"].set()