From b70ebefb17066f66b0e70d2cb08d087eb134ba25 Mon Sep 17 00:00:00 2001 From: alientim Date: Sat, 22 Aug 2026 23:42:32 +0200 Subject: [PATCH] Lizenzsystem Phase 0: Ed25519-Kryptographie, Aktivierungs-/Heartbeat-Protokoll (licensing.py) Neues gemeinsames Modul fuer Client (TESM) und kuenftigen Master-Lizenzserver: - Zwei-Ebenen-Ed25519-Schluesselhierarchie (Master-Signaturschluessel + Pro-Lizenz-Schluesselpaar), kanonische JSON-Signierung - issue_license/verify_license_file/license_status (inkl. 30-Tage-Gnadenfrist, 30-Tage-Vorwarnung, Modul-Set) - Ein Protokoll fuer Online- (API) und Offline-Aktivierung (Code-Copy-Paste): build_client_request/verify_client_request, build_master_response/verify_master_response, encode_code/decode_code - system_fingerprint (Systembindung), heartbeat_stale_warning (14 Tage, rein informativ) Isoliert getestet (Signatur-Rundlauf, Manipulationserkennung, Ablauf-/ Gnadenfrist-Datumslogik, Online- und Offline-Aktivierung/Deaktivierung inkl. Fingerprint-Bindungspruefung und Reaktivierung nach Deaktivierung, Eingabevalidierung) -- 41/41 Checks erfolgreich. Teil von Phase 0 des Lizenz-/Modulsystems, siehe Plan toasty-twirling-hickey.md. Co-Authored-By: Claude Sonnet 5 --- srv/tesm/licensing.py | 281 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 281 insertions(+) create mode 100644 srv/tesm/licensing.py diff --git a/srv/tesm/licensing.py b/srv/tesm/licensing.py new file mode 100644 index 0000000..0c6cf46 --- /dev/null +++ b/srv/tesm/licensing.py @@ -0,0 +1,281 @@ +""" +TESM Lizenzsystem — gemeinsame Kryptographie- und Protokoll-Bausteine für +Kundeninstanz (Client) UND Lizenzserver (Master). Beide Seiten importieren +exakt dieselbe Datei (auf dem Master 1:1 mitkopiert) — Signieren und +Verifizieren müssen bitidentisch funktionieren, jede Abweichung würde +Lizenzen der jeweils anderen Seite unlesbar machen. + +Vollständiges Design: siehe C:\\Users\\tim\\.claude\\plans\\toasty-twirling-hickey.md +(Phase 0). Kurzfassung der Schlüsselhierarchie: + +- Master-Signaturschlüssel (Ed25519): signiert Lizenzdateien selbst UND + jede Aktivierungs-/Deaktivierungs-/Heartbeat-Antwort. Privater Teil bleibt + ausschließlich auf dem Master, öffentlicher Teil steckt in jeder + ausgestellten Lizenzdatei (ermöglicht rein OFFLINE verifizierbare + Master-Antworten beim Kunden). +- Pro-Lizenz-Schlüsselpaar (Ed25519, einmalig je Lizenz erzeugt): der + private Teil reist in der Lizenzdatei zum Kunden und signiert dessen + Aktivierungs-/Deaktivierungs-/Heartbeat-*Anfragen*; der öffentliche Teil + bleibt beim Master in dessen Datenbank (ermöglicht dem Master, jede + Anfrage einer bestimmten Lizenz zweifelsfrei zuzuordnen, ohne die + ursprüngliche Aktivierung "live" gesehen haben zu müssen). + +Ed25519 statt RSA: kleine Schlüssel/Signaturen (wichtig, da Aktivierungs- +Anfragen/-Antworten für den Offline-Fall als von Hand kopierbarer Code +dargestellt werden), fest vorgegebene, sichere Parameter (keine +Padding-/Hash-Wahl wie bei RSA-PSS nötig). +""" +import base64 +import datetime +import hashlib +import json +import socket +import time +import uuid + +from cryptography.exceptions import InvalidSignature +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey, Ed25519PublicKey + +LICENSE_TYPES = ("trial", "standard", "custom", "enterprise") +ALL_MODULES = ("dhcp", "fileshare", "maintenance") + +GRACE_PERIOD_DAYS = 30 # nach Ablauf, bevor lizenzpflichtige Funktionen tatsächlich abgeschaltet werden +EXPIRY_WARNING_DAYS = 30 # Vorlauf für die orange "läuft bald ab"-Anzeige +HEARTBEAT_WARNING_DAYS = 14 # Nichterreichbarkeits-Hinweis (rein informativ, schaltet nie etwas ab) + + +# ============================================================== Schlüssel == + +def generate_keypair(): + """Erzeugt ein neues Ed25519-Schlüsselpaar. Rückgabe: (private_b64, public_b64).""" + priv = Ed25519PrivateKey.generate() + return _encode_private_key(priv), _encode_public_key(priv.public_key()) + + +def _encode_private_key(priv: Ed25519PrivateKey) -> str: + return base64.urlsafe_b64encode(priv.private_bytes_raw()).decode("ascii") + + +def _encode_public_key(pub: Ed25519PublicKey) -> str: + return base64.urlsafe_b64encode(pub.public_bytes_raw()).decode("ascii") + + +def _decode_private_key(s: str) -> Ed25519PrivateKey: + return Ed25519PrivateKey.from_private_bytes(base64.urlsafe_b64decode(s.encode("ascii"))) + + +def _decode_public_key(s: str) -> Ed25519PublicKey: + return Ed25519PublicKey.from_public_bytes(base64.urlsafe_b64decode(s.encode("ascii"))) + + +# ===================================================== Signieren/Prüfen == + +def _canonical_bytes(payload: dict) -> bytes: + """Deterministische JSON-Kodierung (sortierte Keys, kompakte Trenner) -- + Voraussetzung dafür, dass Signieren und Verifizieren exakt dieselben + Bytes sehen, unabhängig von der Dict-Einfügereihenfolge.""" + return json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8") + + +def sign_payload(payload: dict, private_key_b64: str) -> str: + priv = _decode_private_key(private_key_b64) + return base64.urlsafe_b64encode(priv.sign(_canonical_bytes(payload))).decode("ascii") + + +def verify_payload(payload: dict, signature_b64: str, public_key_b64: str) -> bool: + """Gibt bewusst nur True/False zurück (nie eine Exception nach außen) -- + ein Aufrufer soll "ungültig" nie mit einem Absturz verwechseln können.""" + try: + pub = _decode_public_key(public_key_b64) + pub.verify(base64.urlsafe_b64decode(signature_b64.encode("ascii")), _canonical_bytes(payload)) + return True + except (InvalidSignature, ValueError, TypeError, KeyError): + return False + + +# ============================================================ Zeitformat == + +def _iso(ts: float) -> str: + return datetime.datetime.fromtimestamp(ts, tz=datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def _parse_iso(s: str) -> float: + return datetime.datetime.strptime(s, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=datetime.timezone.utc).timestamp() + + +# ================================================================ Lizenz == + +def issue_license(*, customer, license_type, modules, valid_days, + master_private_key_b64, master_public_key_b64, + master_endpoint, vendor, license_id=None, now=None): + """Vom MASTER aufgerufen: erzeugt eine neue, signierte Lizenz samt + frischem Pro-Lizenz-Schlüsselpaar. + + Rückgabe: (license_file, license_pubkey) -- license_file ist die + komplette, an den Kunden auszuhändigende Datei (inkl. dem PRIVATEN + Lizenzschlüssel); license_pubkey ist NUR für die Master-Datenbank + bestimmt (Verifikation künftiger Aktivierungs-/Heartbeat-Anfragen + dieser Lizenz) und wird nicht an den Kunden weitergegeben.""" + if license_type not in LICENSE_TYPES: + raise ValueError(f"Unbekannter Lizenztyp: {license_type!r} (erlaubt: {LICENSE_TYPES})") + unknown = set(modules) - set(ALL_MODULES) + if unknown: + raise ValueError(f"Unbekannte Module: {sorted(unknown)} (erlaubt: {ALL_MODULES})") + + license_priv, license_pub = generate_keypair() + now = now if now is not None else time.time() + payload = { + "license_id": license_id or str(uuid.uuid4()), + "customer": customer, + "type": license_type, + "modules": sorted(modules), + "issued_at": _iso(now), + "expires_at": _iso(now + valid_days * 86400), + "license_pubkey": license_pub, + "master_pubkey": master_public_key_b64, + "master_endpoint": master_endpoint, + "vendor": vendor, + } + signature = sign_payload(payload, master_private_key_b64) + license_file = {**payload, "license_privkey": license_priv, "signature": signature} + return license_file, license_pub + + +def verify_license_file(license_file: dict) -> bool: + """Prüft die Master-Signatur über die Lizenz-Nutzdaten. license_privkey + und signature selbst sind nicht Teil der signierten Nutzlast (die + Signatur wurde ja gerade über den Rest gebildet, siehe issue_license).""" + payload = {k: v for k, v in license_file.items() if k not in ("license_privkey", "signature")} + return verify_payload(payload, license_file.get("signature", ""), license_file.get("master_pubkey", "")) + + +def license_status(license_file: dict, now=None) -> dict: + """Berechnet den aktuellen Anzeige-/Gate-Status einer (bereits als + signaturgültig geprüften!) Lizenz -- ruft NICHT selbst verify_license_file + auf, das bleibt bewusst Sache des Aufrufers, damit hier niemand versehentlich + den Status einer manipulierten Datei berechnet, ohne die Prüfung + überhaupt gemacht zu haben.""" + now = now if now is not None else time.time() + expires_at = _parse_iso(license_file["expires_at"]) + days_left = (expires_at - now) / 86400 + expired = days_left < 0 + days_since_expiry = -days_left if expired else 0.0 + grace_active = expired and days_since_expiry <= GRACE_PERIOD_DAYS + modules_active = (not expired) or grace_active + + return { + "expired": expired, + "days_left": days_left, + "days_since_expiry": days_since_expiry, + "expiring_soon": (not expired) and days_left <= EXPIRY_WARNING_DAYS, + "grace_active": grace_active, + "modules_active": modules_active, + "modules": set(license_file.get("modules", [])) if modules_active else set(), + "type": license_file.get("type"), + "customer": license_file.get("customer"), + "expires_at": license_file.get("expires_at"), + } + + +def heartbeat_stale_warning(last_heartbeat_ts, now=None) -> bool: + """True, wenn der letzte erfolgreiche Heartbeat HEARTBEAT_WARNING_DAYS + oder länger zurückliegt. Rein informativ (siehe Moduldocstring) -- ein + fehlender/alter Heartbeat schaltet nie ein Modul ab, nur das + eingebettete Ablaufdatum selbst zählt dafür (siehe license_status).""" + if last_heartbeat_ts is None: + return False # noch nie verbunden gewesen ist der normale Ausgangszustand, kein Ausfall + now = now if now is not None else time.time() + return (now - last_heartbeat_ts) / 86400 >= HEARTBEAT_WARNING_DAYS + + +# ========================================== Fingerprint (System-Bindung) == + +def system_fingerprint(): + """Stabiler Identifikator dieses Hosts: /etc/machine-id (von systemd bei + der Erstinstallation einmalig erzeugt, übersteht Reboots und normale + Updates) kombiniert mit dem Hostnamen, gehasht damit die rohe + machine-id nie im Klartext übertragen/gespeichert wird. Unter + Windows/ohne /etc/machine-id fällt dies auf den Hostnamen allein + zurück (nur für lokale Tests relevant, Produktivsysteme sind Linux).""" + machine_id = "" + try: + with open("/etc/machine-id", "r", encoding="utf-8") as f: + machine_id = f.read().strip() + except OSError: + pass + raw = f"{machine_id}:{socket.gethostname()}".encode("utf-8") + return hashlib.sha256(raw).hexdigest() + + +# ============================ Aktivierung / Deaktivierung / Heartbeat ===== +# Ein Protokoll, zwei Transportwege: online automatisch per HTTPS-API, +# offline exakt dasselbe Anfrage/Antwort-Paar als von Hand kopierbarer Code +# (encode_code/decode_code) -- der Master ist dadurch in BEIDEN Fällen die +# einzige Quelle der Wahrheit dafür, an welches System eine Lizenz gerade +# gebunden ist, siehe Plan-Abschnitt "Warum das die Doppelnutzung wirksam +# verhindert". + +def build_client_request(action, license_file, nonce=None) -> dict: + """Vom CLIENT aufgerufen: erzeugt eine signierte Anfrage (action: + "activate"/"deactivate"/"heartbeat"). Das Ergebnis ist unverändert für + einen Online-API-Aufruf nutzbar; für den Offline-Fall wird es + zusätzlich mit encode_code() in einen kopierbaren Code umgewandelt.""" + if action not in ("activate", "deactivate", "heartbeat"): + raise ValueError(f"Unbekannte Aktion: {action!r}") + payload = { + "license_id": license_file["license_id"], + "fingerprint": system_fingerprint(), + "action": action, + "nonce": nonce or uuid.uuid4().hex, + "timestamp": _iso(time.time()), + } + signature = sign_payload(payload, license_file["license_privkey"]) + return {**payload, "signature": signature} + + +def verify_client_request(request: dict, license_public_key_b64: str) -> bool: + """Vom MASTER aufgerufen: prüft eine Client-Anfrage gegen den bei + Ausstellung dieser Lizenz in der Master-DB gespeicherten Public-Key.""" + payload = {k: v for k, v in request.items() if k != "signature"} + return verify_payload(payload, request.get("signature", ""), license_public_key_b64) + + +def build_master_response(action, license_id, fingerprint, master_private_key_b64, + status="ok", license_update=None, now=None) -> dict: + """Vom MASTER aufgerufen: erzeugt eine signierte Antwort/Bestätigung + (z.B. "activated", "deactivated", oder das Ergebnis eines Heartbeats). + license_update: optional eine komplette, neu signierte Lizenzdatei + (z.B. nach nachträglicher Modul-Änderung durch den Admin) -- der Client + übernimmt sie nur nach eigener, erfolgreicher Signaturprüfung.""" + payload = { + "license_id": license_id, + "fingerprint": fingerprint, + "action": action, + "status": status, + "timestamp": _iso(now if now is not None else time.time()), + } + if license_update is not None: + payload["license_update"] = license_update + signature = sign_payload(payload, master_private_key_b64) + return {**payload, "signature": signature} + + +def verify_master_response(response: dict, master_public_key_b64: str) -> bool: + """Vom CLIENT aufgerufen: prüft eine Master-Antwort komplett OFFLINE + gegen den in der eigenen Lizenzdatei eingebetteten master_pubkey.""" + payload = {k: v for k, v in response.items() if k != "signature"} + return verify_payload(payload, response.get("signature", ""), master_public_key_b64) + + +# ==================================================== Codes für Offline == + +def encode_code(data: dict) -> str: + """Kodiert ein Anfrage-/Antwort-dict als kompakten, per Copy-Paste + übertragbaren Code (Base64 einer kanonischen JSON-Darstellung, keine + Zeilenumbrüche/Sonderzeichen, damit Kopieren aus/in ein Textfeld nichts + kaputt macht).""" + return base64.urlsafe_b64encode(_canonical_bytes(data)).decode("ascii") + + +def decode_code(code: str) -> dict: + return json.loads(base64.urlsafe_b64decode(code.strip().encode("ascii")).decode("utf-8"))