tesm/packages/tesm-licensing/src/tesm_licensing/license.py

"""Lizenzdatei-Format, Ausstellung, Verifikation und Statusberechnung.

Format-Version 2. Unterschiede zur Version 1 des Vorgaengerprojekts, jeweils mit
Begruendung:

1. **Der private Client-Schluessel steckt nicht mehr in der Lizenz-Payload.**
   V1 lieferte ``license_privkey`` als Feld neben der signierten Payload in
   derselben Datei aus. Wer diese Datei zu Supportzwecken weitergab, gab damit
   das Aktivierungs-Credential mit. V2 liefert ein Bundle aus
   ``{"license": <signiertes Dokument>, "client_key": <privater Schluessel>}``,
   das der Client beim Import aufteilt: Lizenzdokument nach ``license.json``,
   Schluessel nach ``license_key.json`` mit Modus 0600.
2. **``lifetime`` ist ein signiertes Feld**, nicht mehr nur aus dem
   Sentinel-Datum abgeleitet. Der Sentinel bleibt zusaetzlich erhalten, damit
   jede Datumsvergleichslogik ohne Sonderfall auskommt.
3. **``not_before``** existiert und wird geprueft -- eine Lizenz, die versehentlich
   mit Zukunftsdatum ausgestellt wurde, gilt nicht rueckwirkend.
4. **``master_key_id``** (Fingerprint des Master-Public-Keys) erlaubt Pinning und
   Schluesselrotation auf Client-Seite.
5. **``seq``** (monoton steigende Revision) erlaubt es dem Client, ein per
   Heartbeat nachgeliefertes ``license_update`` gegen Downgrade-Replay zu
   schuetzen.
6. **Verifikation gegen einen gepinnten Aussteller-Schluessel** ist vorgesehen
   (``trusted_master_keys``). V1 pruefte die Signatur ausschliesslich gegen den in
   derselben Datei eingebetteten ``master_pubkey`` -- das beweist nur innere
   Selbstkonsistenz, nicht die Herkunft.
"""

from __future__ import annotations

import hashlib
import uuid
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from typing import Any, Iterable, Mapping, Sequence

from .canonical import SIG_CONTEXT_LICENSE, sign, verify
from .errors import InvalidLicenseFile, SignatureInvalid, UntrustedIssuer

FORMAT_VERSION = 2

LICENSE_TYPES: tuple[str, ...] = ("trial", "standard", "custom", "enterprise")
LICENSE_ROLES: tuple[str, ...] = ("master", "license_server", "customer")
ALL_MODULES: tuple[str, ...] = ("dhcp", "fileshare", "maintenance")

LICENSE_TYPE_LABELS: dict[str, str] = {
    "trial": "Testlizenz",
    "standard": "Standard",
    "custom": "Individuell",
    "enterprise": "Enterprise",
}

MODULE_LABELS: dict[str, str] = {
    "dhcp": "DHCP-Server (Kea)",
    "fileshare": "Dateifreigaben",
    "maintenance": "Wartung (SSH-Bulk)",
}

ROLE_LABELS: dict[str, str] = {
    "master": "Master-Lizenzserver",
    "license_server": "Lizenzserver",
    "customer": "Kundeninstanz",
}

#: Kulanzzeitraum nach Ablauf, in dem Module weiterlaufen.
GRACE_PERIOD_DAYS = 30
#: Ab wann "laeuft bald ab" gemeldet wird.
EXPIRY_WARNING_DAYS = 30
#: Ab wann ein ausgebliebener Heartbeat gemeldet wird (schaltet nichts ab).
HEARTBEAT_WARNING_DAYS = 14
#: Standardlaufzeit einer Trial-Ausstellung.
TRIAL_DEFAULT_DAYS = 30
#: Sentinel-Ablaufdatum fuer Lifetime-Lizenzen.
LIFETIME_EXPIRES_AT = "2099-12-31T00:00:00Z"

_PAYLOAD_FIELDS = (
    "fmt",
    "license_id",
    "seq",
    "customer",
    "type",
    "role",
    "modules",
    "lifetime",
    "issued_at",
    "not_before",
    "expires_at",
    "license_pubkey",
    "master_pubkey",
    "master_key_id",
    "master_endpoint",
    "vendor",
)


def utcnow() -> datetime:
    return datetime.now(timezone.utc)


def iso(value: datetime) -> str:
    """ISO-8601 in UTC mit ``Z``-Suffix, sekundengenau (keine Mikrosekunden)."""
    return value.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")


def parse_iso(value: str) -> datetime:
    """Parst ISO-8601. Ohne Zeitzone wird UTC angenommen."""
    text = (value or "").strip()
    if text.endswith("Z"):
        text = text[:-1] + "+00:00"
    parsed = datetime.fromisoformat(text)
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=timezone.utc)
    return parsed.astimezone(timezone.utc)


def key_id(public_key_b64: str) -> str:
    """Kurzer, stabiler Fingerprint eines Public Keys -- fuer Pinning und Anzeige."""
    return hashlib.sha256((public_key_b64 or "").encode("ascii")).hexdigest()[:16]


@dataclass(frozen=True, slots=True)
class LicenseStatus:
    """Auswertung einer Lizenzdatei zu einem Zeitpunkt. Rein rechnerisch, ohne I/O."""

    license_id: str
    seq: int
    type: str
    role: str
    customer: Mapping[str, Any]
    vendor: Mapping[str, Any]
    master_endpoint: str
    master_key_id: str
    modules: frozenset[str]
    lifetime: bool
    issued_at: datetime | None
    expires_at: datetime | None
    expired: bool
    not_yet_valid: bool
    days_left: int | None
    days_since_expiry: int | None
    expiring_soon: bool
    grace_active: bool
    modules_active: bool
    warnings: tuple[str, ...] = field(default=())

    def has_module(self, name: str) -> bool:
        return self.modules_active and name in self.modules








def split_bundle(bundle: Mapping[str, Any]) -> tuple[dict[str, Any], str]:
    """Zerlegt ein Auslieferungs-Bundle in ``(lizenzdokument, client_private_key)``.

    Akzeptiert zusaetzlich ein bereits aufgeteiltes Dokument (dann ist der
    zweite Rueckgabewert leer) sowie -- fuer Migrationen -- eine V1-Datei mit
    eingebettetem ``license_privkey``.
    """
    if not isinstance(bundle, Mapping):
        raise InvalidLicenseFile("Lizenzdatei enthaelt kein Objekt")

    inner = bundle.get("license")
    if isinstance(inner, Mapping):
        return dict(inner), str(bundle.get("client_key") or "")

    if "license_privkey" in bundle:  # V1-Kompatibilitaet
        document = {k: v for k, v in bundle.items() if k != "license_privkey"}
        return document, str(bundle.get("license_privkey") or "")

    if "signature" in bundle and "license_id" in bundle:
        return dict(bundle), ""

    raise InvalidLicenseFile("Unbekanntes Lizenzdatei-Format")


def verify_license(
    document: Mapping[str, Any],
    *,
    trusted_master_keys: Sequence[str] | None = None,
) -> None:
    """Verifiziert ein Lizenzdokument. Wirft bei Problemen, gibt sonst ``None``.

    ``trusted_master_keys`` sind Base64-Public-Keys, denen der Client vertraut
    (im Build mitgeliefert oder beim ersten Import per Trust-on-First-Use
    gepinnt). Ist die Liste leer/``None``, wird nur die innere Konsistenz der
    Datei geprueft -- dann traegt der Status eine entsprechende Warnung.
    """
    if not isinstance(document, Mapping):
        raise InvalidLicenseFile("Lizenzdokument ist kein Objekt")

    missing = [name for name in _PAYLOAD_FIELDS if name not in document]
    if missing:
        raise InvalidLicenseFile(f"Lizenzdokument unvollstaendig: {', '.join(missing)}")
    if int(document.get("fmt") or 0) != FORMAT_VERSION:
        raise InvalidLicenseFile(
            f"Nicht unterstuetzte Formatversion: {document.get('fmt')!r}"
            f" (erwartet {FORMAT_VERSION})"
        )
    if document.get("type") not in LICENSE_TYPES:
        raise InvalidLicenseFile("Unbekannter Lizenztyp")
    if document.get("role") not in LICENSE_ROLES:
        raise InvalidLicenseFile("Unbekannte Rolle")

    master_pubkey = str(document.get("master_pubkey") or "")
    if key_id(master_pubkey) != str(document.get("master_key_id") or ""):
        raise InvalidLicenseFile("master_key_id passt nicht zum eingebetteten Schluessel")

    payload = {k: v for k, v in document.items() if k != "signature"}
    if not verify(
        payload, str(document.get("signature") or ""), master_pubkey, SIG_CONTEXT_LICENSE
    ):
        raise SignatureInvalid("Signatur der Lizenzdatei ist ungueltig")

    if trusted_master_keys and master_pubkey not in set(trusted_master_keys):
        raise UntrustedIssuer(
            "Die Lizenz wurde nicht von einem vertrauenswuerdigen Lizenzserver ausgestellt"
        )


def license_status(
    document: Mapping[str, Any],
    *,
    now: datetime | None = None,
    issuer_pinned: bool = False,
) -> LicenseStatus:
    """Berechnet den Status. Verifiziert nicht selbst -- das macht ``verify_license``.

    Bewusst getrennt gehalten, damit hier niemand versehentlich den Status einer
    unverifizierten Datei berechnet und weiterverwendet.
    """
    moment = now or utcnow()
    lifetime = bool(document.get("lifetime"))
    warnings: list[str] = []
    if not issuer_pinned:
        warnings.append("Aussteller nicht gepinnt -- nur innere Konsistenz geprueft")

    issued_at: datetime | None = None
    expires_at: datetime | None = None
    not_before: datetime | None = None
    for name in ("issued_at", "expires_at", "not_before"):
        raw = document.get(name)
        if not raw:
            continue
        try:
            parsed = parse_iso(str(raw))
        except ValueError:
            warnings.append(f"Feld {name} ist kein gueltiges Datum")
            continue
        if name == "issued_at":
            issued_at = parsed
        elif name == "expires_at":
            expires_at = parsed
        else:
            not_before = parsed

    not_yet_valid = bool(not_before and moment < not_before)

    days_left: int | None
    days_since_expiry: int | None
    if lifetime:
        days_left = None
        days_since_expiry = None
        expired = False
    elif expires_at is None:
        # Fail-safe: unlesbares Ablaufdatum gilt als abgelaufen, nicht als gueltig.
        days_left = None
        days_since_expiry = None
        expired = True
        warnings.append("Ablaufdatum fehlt oder ist unlesbar")
    else:
        delta = expires_at - moment
        expired = delta.total_seconds() <= 0
        if expired:
            days_left = 0
            days_since_expiry = int((moment - expires_at).total_seconds() // 86400)
        else:
            days_left = -(-int(delta.total_seconds()) // 86400)
            days_since_expiry = None

    grace_active = bool(
        expired and days_since_expiry is not None and days_since_expiry < GRACE_PERIOD_DAYS
    )
    modules_active = not not_yet_valid and (lifetime or not expired or grace_active)
    expiring_soon = bool(
        not lifetime
        and not expired
        and days_left is not None
        and days_left <= EXPIRY_WARNING_DAYS
    )

    return LicenseStatus(
        license_id=str(document.get("license_id") or ""),
        seq=int(document.get("seq") or 1),
        type=str(document.get("type") or ""),
        # Fehlt die Rolle, gilt "customer" -- niemals ein privilegierterer Wert.
        role=str(document.get("role") or "customer"),
        customer=dict(document.get("customer") or {}),
        vendor=dict(document.get("vendor") or {}),
        master_endpoint=str(document.get("master_endpoint") or ""),
        master_key_id=str(document.get("master_key_id") or ""),
        modules=frozenset(str(m) for m in (document.get("modules") or ())),
        lifetime=lifetime,
        issued_at=issued_at,
        expires_at=expires_at,
        expired=expired,
        not_yet_valid=not_yet_valid,
        days_left=days_left,
        days_since_expiry=days_since_expiry,
        expiring_soon=expiring_soon,
        grace_active=grace_active,
        modules_active=modules_active,
        warnings=tuple(warnings),
    )