""" 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() # Sentinel-Ablaufdatum fuer "Lifetime"-Lizenzen. Bewusst ein fixes, weit in # der Zukunft liegendes Datum statt eines Sonderwerts wie None/0 fuer # expires_at -- dadurch bleibt JEDE bestehende Datumsvergleichs-/ # Differenzrechnung (hier, im Master-Dashboard, UND im TESM-Client) ohne # jede Sonderbehandlung korrekt: "in > 90 Tagen faellig" o.ae. ist fuer ein # Datum im Jahr 2099 schlicht immer wahr. Ob eine Lizenz als "Lifetime" # geflaggt ist, wird zusaetzlich explizit in der Master-DB gespeichert # (licenses.lifetime) -- dieses Sentinel-Datum ist nur die tatsaechlich in # die signierte Lizenzdatei eingebettete, fuer den Client sichtbare Reprae- # sentation davon. LIFETIME_EXPIRES_AT = "2099-12-31T00:00:00Z" # ================================================================ 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, lifetime=False): """Vom MASTER aufgerufen: erzeugt eine neue, signierte Lizenz samt frischem Pro-Lizenz-Schlüsselpaar. lifetime=True ignoriert valid_days komplett und setzt stattdessen das feste Sentinel-Datum LIFETIME_EXPIRES_AT -- siehe dortigen Kommentar. 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": LIFETIME_EXPIRES_AT if lifetime else _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(), # Rein informativ fuer die Dashboard-Kacheln des Masters (siehe # dortiges index()) -- fliesst NICHT in irgendeine Sicherheits- # entscheidung ein (die bleibt allein Sache des Fingerprints), wird # aber wie alle anderen Felder mitsigniert, damit ein Angreifer # ohne den privaten Lizenzschluessel keinen falschen Hostnamen # unterschieben kann. "hostname": socket.gethostname(), "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"))