tesm-license/packages/tesm-licensing-server/src/tesm_licensing_server/__init__.py

"""Die Serverhaelfte des Lizenzprotokolls: ausstellen, erneuern, antworten.

Dieses Paket liegt **nur** beim Lizenzserver. Ein Client verifiziert Lizenzen
und stellt Anfragen -- ausstellen kann er nichts, und er soll den Code dafuer
auch nicht mitbringen. Die gemeinsame Haelfte (kanonisches JSON, Signieren und
Pruefen, Fingerabdruck, Statusbewertung, Client-Anfragen) steht in
``tesm_licensing`` und wird von beiden Seiten benutzt.

Schluesselerzeugung gehoert ebenfalls hierher: der Server erzeugt seinen
Signaturschluessel und je Lizenz ein Client-Schluesselpaar. Der Client
*empfaengt* seinen Schluessel mit dem Bundle; er erzeugt nie einen.
"""

from __future__ import annotations

import base64
import uuid
from datetime import datetime, timedelta
from typing import Any, Callable, Iterable, Mapping, Sequence

from cryptography.hazmat.primitives.asymmetric import ed25519

from tesm_licensing import (
    ACTIONS,
    ALL_MODULES,
    FORMAT_VERSION,
    LICENSE_ROLES,
    LICENSE_TYPES,
    LIFETIME_EXPIRES_AT,
    MAX_REPORT_ROWS,
    RESPONSE_STATES,
    SIG_CONTEXT_LICENSE,
    SIG_CONTEXT_REQUEST,
    SIG_CONTEXT_RESPONSE,
    ProtocolViolation,
    ReplayDetected,
    SignatureInvalid,
    canonical_bytes,
    iso,
    key_id,
    new_nonce,
    parse_iso,
    utcnow,
)
from tesm_licensing.canonical import _b64d, sign, verify
from tesm_licensing.protocol import _FINGERPRINT_RE, _NONCE_RE, check_freshness

__all__ = [
    "build_master_response",
    "generate_keypair",
    "issue_license",
    "public_from_private",
    "resign_license",
    "verify_client_request",
]


def generate_keypair() -> tuple[str, str]:
    """Erzeugt ein Ed25519-Schluesselpaar. Rueckgabe: ``(private_b64, public_b64)``."""
    private = ed25519.Ed25519PrivateKey.generate()
    return (
        base64.b64encode(private.private_bytes_raw()).decode("ascii"),
        base64.b64encode(private.public_key().public_bytes_raw()).decode("ascii"),
    )


def public_from_private(private_key_b64: str) -> str:
    """Leitet den oeffentlichen Teil aus einem privaten Schluessel ab."""
    key = ed25519.Ed25519PrivateKey.from_private_bytes(
        _b64d(private_key_b64, field="private_key")
    )
    return base64.b64encode(key.public_key().public_bytes_raw()).decode("ascii")


def _validate_issue_args(
    license_type: str, role: str, modules: Sequence[str], lifetime: bool
) -> None:
    if license_type not in LICENSE_TYPES:
        raise ValueError(f"Unbekannter Lizenztyp: {license_type!r}")
    if role not in LICENSE_ROLES:
        raise ValueError(f"Unbekannte Rolle: {role!r}")
    unknown = sorted(set(modules) - set(ALL_MODULES))
    if unknown:
        raise ValueError(f"Unbekannte Module: {', '.join(unknown)}")
    if lifetime and license_type == "trial":
        # Dritte und letzte Absicherung derselben Regel (UI + Service pruefen
        # vorher schon): eine unbefristete Trial-Lizenz waere eine permanente
        # Lizenz ohne Module -- ein sinnloser Zustand.
        raise ValueError("Trial-Lizenzen koennen nicht unbefristet sein")
    if role != "customer" and modules:
        raise ValueError("Lizenzserver-Lizenzen tragen keine TESM-Module")


def issue_license(
    *,
    customer: Mapping[str, Any],
    license_type: str,
    modules: Iterable[str],
    valid_days: int,
    master_private_key_b64: str,
    master_public_key_b64: str,
    master_endpoint: str,
    vendor: Mapping[str, Any],
    license_id: str | None = None,
    role: str = "customer",
    lifetime: bool = False,
    seq: int = 1,
    now: datetime | None = None,
) -> tuple[dict[str, Any], str]:
    """Stellt eine Lizenz aus.

    Rueckgabe: ``(bundle, license_public_key)``. ``bundle`` ist die Datei, die
    an den Kunden geht; ``license_public_key`` wandert in die Master-Datenbank,
    damit der Master spaeter Client-Anfragen dieser Lizenz verifizieren kann.
    """
    module_list = sorted({str(m) for m in modules})
    _validate_issue_args(license_type, role, module_list, lifetime)
    if not lifetime and int(valid_days) <= 0:
        raise ValueError("valid_days muss positiv sein")

    moment = now or utcnow()
    client_private, client_public = generate_keypair()
    payload: dict[str, Any] = {
        "fmt": FORMAT_VERSION,
        "license_id": license_id or str(uuid.uuid4()),
        "seq": int(seq),
        "customer": dict(customer),
        "type": license_type,
        "role": role,
        "modules": module_list,
        "lifetime": bool(lifetime),
        "issued_at": iso(moment),
        "not_before": iso(moment),
        "expires_at": LIFETIME_EXPIRES_AT
        if lifetime
        else iso(moment + timedelta(days=int(valid_days))),
        "license_pubkey": client_public,
        "master_pubkey": master_public_key_b64,
        "master_key_id": key_id(master_public_key_b64),
        "master_endpoint": (master_endpoint or "").rstrip("/"),
        "vendor": dict(vendor),
    }
    signature = sign(payload, master_private_key_b64, SIG_CONTEXT_LICENSE)
    document = {**payload, "signature": signature}
    bundle = {
        "kind": "tesm-license-bundle",
        "fmt": FORMAT_VERSION,
        "license": document,
        "client_key": client_private,
    }
    return bundle, client_public


def resign_license(
    document: Mapping[str, Any],
    *,
    master_private_key_b64: str,
    changes: Mapping[str, Any],
    now: datetime | None = None,
) -> dict[str, Any]:
    """Erzeugt eine neu signierte Revision eines bestehenden Lizenzdokuments.

    Wird fuer ``license_update`` im Heartbeat gebraucht (z. B. nachtraeglich
    erweiterte Module oder verlaengerte Laufzeit). ``seq`` steigt dabei zwingend,
    damit der Client eine aeltere Revision nicht als Update akzeptiert.
    """
    payload = {k: v for k, v in document.items() if k != "signature"}
    payload.update(changes)
    payload["seq"] = int(document.get("seq") or 1) + 1
    payload["issued_at"] = iso(now or utcnow())
    _validate_issue_args(
        str(payload.get("type")),
        str(payload.get("role")),
        [str(m) for m in payload.get("modules") or ()],
        bool(payload.get("lifetime")),
    )
    payload["signature"] = sign(payload, master_private_key_b64, SIG_CONTEXT_LICENSE)
    return payload


def verify_client_request(
    request: Mapping[str, Any],
    license_public_key: str,
    *,
    expected_action: str | None = None,
    now: datetime | None = None,
    seen_nonce: Callable[[str], bool] | None = None,
) -> None:
    """Verifiziert eine Client-Anfrage serverseitig. Wirft bei Problemen.

    ``seen_nonce`` ist ein Callback, der ``True`` zurueckgibt, wenn die Nonce
    bereits verbraucht ist. Der Aufrufer entscheidet ueber die Persistenz
    (in dieser Codebasis: Tabelle ``license_nonces``).
    """
    if not isinstance(request, Mapping):
        raise ProtocolViolation("Anfrage ist kein Objekt")
    if int(request.get("fmt") or 0) != FORMAT_VERSION:
        raise ProtocolViolation("Nicht unterstuetzte Protokollversion")

    action = str(request.get("action") or "")
    if action not in ACTIONS:
        raise ProtocolViolation("Unbekannte Aktion")
    if expected_action and action != expected_action:
        raise ProtocolViolation("Aktion passt nicht zum Endpunkt")

    nonce = str(request.get("nonce") or "")
    if not _NONCE_RE.match(nonce):
        raise ProtocolViolation("Nonce hat ein unerwartetes Format")
    fingerprint = str(request.get("fingerprint") or "")
    if not _FINGERPRINT_RE.match(fingerprint):
        raise ProtocolViolation("Fingerprint hat ein unerwartetes Format")

    payload = {k: v for k, v in request.items() if k != "signature"}
    if not verify(
        payload, str(request.get("signature") or ""), license_public_key, SIG_CONTEXT_REQUEST
    ):
        raise SignatureInvalid("Signatur der Anfrage ist ungueltig")

    # Frische und Replay erst *nach* der Signaturpruefung: unsignierte Muellanfragen
    # sollen keine Nonce-Eintraege in der Datenbank erzeugen koennen.
    moment = now or utcnow()
    check_freshness(str(request.get("timestamp") or ""), moment, "Anfrage")
    if seen_nonce is not None and seen_nonce(nonce):
        raise ReplayDetected("Diese Anfrage wurde bereits verarbeitet")

    report = request.get("report")
    if report is not None:
        if not isinstance(report, Mapping):
            raise ProtocolViolation("Bericht ist kein Objekt")
        total = sum(len(report.get(key) or ()) for key in ("customers", "tickets", "licenses"))
        if total > MAX_REPORT_ROWS:
            raise ProtocolViolation("Bericht ueberschreitet die Groessengrenze")


def build_master_response(
    action: str,
    *,
    license_id: str,
    fingerprint: str,
    request_nonce: str,
    master_private_key: str,
    status: str = "ok",
    detail: str = "",
    license_update: Mapping[str, Any] | None = None,
    now: datetime | None = None,
) -> dict[str, Any]:
    """Baut eine signierte Master-Antwort, gebunden an die Nonce der Anfrage."""
    if status not in RESPONSE_STATES:
        raise ProtocolViolation(f"Unbekannter Antwortstatus: {status!r}")
    payload: dict[str, Any] = {
        "fmt": FORMAT_VERSION,
        "license_id": license_id,
        "fingerprint": fingerprint,
        "action": action,
        "status": status,
        "detail": detail,
        "nonce": request_nonce,
        "timestamp": iso(now or utcnow()),
    }
    if license_update is not None:
        payload["license_update"] = dict(license_update)
    payload["signature"] = sign(payload, master_private_key, SIG_CONTEXT_RESPONSE)
    return payload