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

"""Aktivierungs-, Deaktivierungs- und Heartbeat-Protokoll.

Beide Transportwege -- HTTPS-JSON und kopierbarer Offline-Code -- tragen exakt
dieselben signierten Objekte. Es gibt deshalb nur *einen* Satz
Verifikationsfunktionen; der Transport ist reine Verpackung.

Haertungen gegenueber dem Vorgaengerprojekt:

* **Domaenengetrennte Signaturen** (siehe ``canonical``): eine Lizenz-Signatur
  kann nie als Antwort-Signatur durchgehen und umgekehrt.
* **Frischefenster**: Anfragen und Antworten tragen einen Zeitstempel, der
  gegen ein Fenster geprueft wird (``REQUEST_MAX_AGE_SECONDS``, plus
  ``CLOCK_SKEW_SECONDS`` Toleranz in die Zukunft). V1 pruefte den Zeitstempel
  gar nicht -- eine abgefangene Aktivierungsanfrage war unbegrenzt
  wiederverwendbar.
* **Nonce-Replay-Schutz**: der Master kann ueber ``seen_nonce`` eine
  Persistenzschicht einhaengen; ein zweites Mal gesehene Nonce wird abgewiesen.
* **Antwort-an-Anfrage-Bindung**: die Master-Antwort echot die Nonce der Anfrage
  in ihrer signierten Payload. Eine alte "ok"-Antwort kann dadurch nicht gegen
  eine neue Anfrage eingespielt werden (relevant vor allem im Offline-Code-Weg,
  wo Codes per Copy-Paste durch die Gegend wandern).
* **Downgrade-Schutz fuer ``license_update``**: das nachgelieferte Dokument muss
  eine hoehere ``seq`` tragen als das aktuell installierte.
"""

from __future__ import annotations

import re
import secrets
from datetime import datetime, timedelta
from typing import Any, Callable, Mapping

from .canonical import SIG_CONTEXT_REQUEST, SIG_CONTEXT_RESPONSE, sign, verify
from .errors import ProtocolViolation, ReplayDetected, SignatureInvalid
from .license import FORMAT_VERSION, iso, parse_iso, utcnow
from .fingerprint import system_fingerprint

#: Erlaubte Aktionen des Protokolls.
ACTIONS: tuple[str, ...] = ("activate", "deactivate", "heartbeat")

#: Antwortstatus, die der Master senden kann.
RESPONSE_STATES: tuple[str, ...] = ("ok", "revoked", "expired", "rejected")

#: Wie alt eine Anfrage/Antwort maximal sein darf (Offline-Codes brauchen Luft).
REQUEST_MAX_AGE_SECONDS = 7 * 24 * 3600
#: Toleranz fuer Uhren, die leicht vorlaufen.
CLOCK_SKEW_SECONDS = 300
#: Regulaerer Heartbeat-Abstand des Clients.
HEARTBEAT_INTERVAL_SECONDS = 6 * 3600
#: Obergrenze fuer Zeilen in einem Sub-Lizenzserver-Bericht.
MAX_REPORT_ROWS = 5000

_NONCE_RE = re.compile(r"^[A-Za-z0-9_-]{16,64}$")
_FINGERPRINT_RE = re.compile(r"^[0-9a-f]{64}$")


def new_nonce() -> str:
    return secrets.token_urlsafe(18)


def build_client_request(
    action: str,
    document: Mapping[str, Any],
    client_private_key: str,
    *,
    hostname: str = "",
    nonce: str | None = None,
    report: Mapping[str, Any] | None = None,
    now: datetime | None = None,
) -> dict[str, Any]:
    """Baut eine signierte Client-Anfrage.

    ``report`` wird ausschliesslich von Heartbeats einer Instanz mit
    ``role="license_server"`` gesetzt (Sub-Lizenzserver-Reporting). Alle anderen
    Aufrufer lassen das Feld weg.
    """
    if action not in ACTIONS:
        raise ProtocolViolation(f"Unbekannte Aktion: {action!r}")
    if not client_private_key:
        raise ProtocolViolation("Kein Client-Schluessel vorhanden -- Lizenz unvollstaendig importiert")

    payload: dict[str, Any] = {
        "fmt": FORMAT_VERSION,
        "license_id": str(document.get("license_id") or ""),
        "seq": int(document.get("seq") or 1),
        "fingerprint": system_fingerprint(),
        "hostname": hostname,
        "action": action,
        "nonce": nonce or new_nonce(),
        "timestamp": iso(now or utcnow()),
    }
    if report is not None:
        payload["report"] = dict(report)
    payload["signature"] = sign(payload, client_private_key, SIG_CONTEXT_REQUEST)
    return payload


def check_freshness(timestamp: str, now: datetime, what: str) -> None:
    try:
        stamp = parse_iso(timestamp)
    except (ValueError, TypeError) as exc:
        raise ProtocolViolation(f"{what}: Zeitstempel unlesbar") from exc
    if stamp > now + timedelta(seconds=CLOCK_SKEW_SECONDS):
        raise ReplayDetected(f"{what}: Zeitstempel liegt in der Zukunft")
    if stamp < now - timedelta(seconds=REQUEST_MAX_AGE_SECONDS):
        raise ReplayDetected(f"{what}: Zeitstempel ist zu alt")






def verify_master_response(
    response: Mapping[str, Any],
    master_public_key: str,
    *,
    expected_license_id: str,
    expected_fingerprint: str,
    expected_action: str,
    expected_nonce: str,
    current_seq: int = 0,
    now: datetime | None = None,
) -> None:
    """Verifiziert eine Master-Antwort clientseitig, komplett offline.

    Prueft neben der Signatur, dass die Antwort tatsaechlich zu *dieser* Anfrage
    gehoert (Lizenz-ID, Fingerprint, Aktion, Nonce) und dass ein mitgeliefertes
    ``license_update`` keine aeltere Revision ist.
    """
    if not isinstance(response, Mapping):
        raise ProtocolViolation("Antwort ist kein Objekt")
    if int(response.get("fmt") or 0) != FORMAT_VERSION:
        raise ProtocolViolation("Nicht unterstuetzte Protokollversion")

    payload = {k: v for k, v in response.items() if k != "signature"}
    if not verify(
        payload, str(response.get("signature") or ""), master_public_key, SIG_CONTEXT_RESPONSE
    ):
        raise SignatureInvalid("Signatur der Antwort ist ungueltig")

    if str(response.get("license_id") or "") != expected_license_id:
        raise ProtocolViolation("Antwort gehoert zu einer anderen Lizenz")
    if str(response.get("fingerprint") or "") != expected_fingerprint:
        raise ProtocolViolation("Antwort gehoert zu einer anderen Maschine")
    if str(response.get("action") or "") != expected_action:
        raise ProtocolViolation("Antwort gehoert zu einer anderen Aktion")
    if str(response.get("nonce") or "") != expected_nonce:
        raise ReplayDetected("Antwort gehoert nicht zu dieser Anfrage")
    if str(response.get("status") or "") not in RESPONSE_STATES:
        raise ProtocolViolation("Unbekannter Antwortstatus")

    check_freshness(str(response.get("timestamp") or ""), now or utcnow(), "Antwort")

    update = response.get("license_update")
    if update is not None:
        if not isinstance(update, Mapping):
            raise ProtocolViolation("license_update ist kein Objekt")
        if int(update.get("seq") or 0) < int(current_seq or 0):
            raise ReplayDetected("license_update ist aelter als die installierte Lizenz")