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 <noreply@anthropic.com>
2026-08-22 23:42:32 +02:00
parent d90c27b8f8
commit b70ebefb17
1 changed files with 281 additions and 0 deletions
@@ -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"))