tesm-license/srv/tesm-license/licensing.py

"""
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"))