anonymisation/gui_v6/license_client.py

"""Client licence de la GUI V6 (mockable, sans clé privée).

Contrat final aligné sur le portail ``app_aivanov`` :

- ``activate(token, machine_id)`` → ``POST /api/v1/activate``
- ``check(license_ref, machine_id)`` → ``POST /api/v1/check``

Principes :

- La session HTTP est **injectable** (paramètre ``session``) : en test on
  fournit une fausse session, aucun appel réseau réel n'est effectué.
- Le client **ne contient aucune clé privée** : la vérification de signature
  est faite côté serveur ; le client stocke seulement la licence signée reçue.
- Serveur indisponible / réponse illisible → statut ``unavailable``, jamais
  d'exception remontée à l'appelant (la GUI ne doit pas crasher).
- Aucun token n'est journalisé.
"""

from __future__ import annotations

from dataclasses import dataclass
from typing import Any, Optional, Protocol

from gui_v6.license_store import LicenseStore

# Statuts métier exposés à l'UI.
ACTIVE = "active"
GRACE = "grace"
EXPIRED = "expired"
REVOKED = "revoked"
NONE = "none"
UNAVAILABLE = "unavailable"
INVALID = "invalid"

_VALID_STATES = {ACTIVE, GRACE}


class _HttpResponse(Protocol):
    status_code: int

    def json(self) -> Any: ...


class _HttpSession(Protocol):
    def post(self, url: str, json: dict, timeout: float) -> _HttpResponse: ...


@dataclass
class LicenseStatus:
    """État de licence destiné à l'affichage et aux décisions de la GUI."""

    valid: bool
    status: str
    message: str = ""
    expires_at: Optional[str] = None
    grace_days: int = 0
    machine_id: Optional[str] = None
    license_ref: Optional[str] = None

    @classmethod
    def none(cls, message: str = "Aucune licence active") -> "LicenseStatus":
        return cls(valid=False, status=NONE, message=message)

    @classmethod
    def unavailable(
        cls, message: str = "Serveur de licence indisponible"
    ) -> "LicenseStatus":
        return cls(valid=False, status=UNAVAILABLE, message=message)

    @classmethod
    def invalid(cls, message: str = "Licence ou jeton invalide") -> "LicenseStatus":
        return cls(valid=False, status=INVALID, message=message)

    @classmethod
    def from_payload(
        cls, payload: dict[str, Any], machine_id: Optional[str] = None
    ) -> "LicenseStatus":
        """Construit un statut depuis la réponse API ou un ancien payload plat.

        Le portail renvoie ``{"state": "...", "license": {"payload": ...}}``.
        Les payloads plats ``{"status": "...", "license_ref": ...}`` restent
        acceptés pour garder les tests/unités locales simples et robustes.
        """
        status = str(payload.get("state") or payload.get("status") or NONE)
        license_envelope = payload.get("license")
        signed_payload = (
            license_envelope.get("payload")
            if isinstance(license_envelope, dict)
            and isinstance(license_envelope.get("payload"), dict)
            else {}
        )
        source = signed_payload if signed_payload else payload
        try:
            grace = int(source.get("grace_days", payload.get("grace_days", 0)) or 0)
        except (TypeError, ValueError):
            grace = 0
        return cls(
            valid=status in _VALID_STATES,
            status=status,
            message=str(payload.get("message", "") or ""),
            expires_at=source.get("expires_at") or payload.get("expires_at"),
            grace_days=grace,
            machine_id=source.get("machine_id") or payload.get("machine_id") or machine_id,
            license_ref=source.get("license_ref") or payload.get("license_ref"),
        )


class LicenseClient:
    """Appelle le portail licence via une session HTTP injectable."""

    def __init__(
        self,
        base_url: str,
        session: Optional[_HttpSession] = None,
        store: Optional[LicenseStore] = None,
        timeout: float = 10.0,
    ) -> None:
        self._base_url = base_url.rstrip("/")
        self._session = session
        self._store = store if store is not None else LicenseStore()
        self._timeout = timeout

    def _get_session(self) -> _HttpSession:
        if self._session is None:
            # Import paresseux : pas de dépendance dure si une session est injectée.
            import requests

            self._session = requests.Session()
        return self._session

    def _post(self, endpoint: str, payload: dict) -> Optional[_HttpResponse]:
        try:
            session = self._get_session()
            return session.post(
                f"{self._base_url}{endpoint}", json=payload, timeout=self._timeout
            )
        except Exception:
            # Réseau indisponible, DNS, timeout, requests absent… : pas de crash.
            return None

    @staticmethod
    def _parse(response: Optional[_HttpResponse]) -> Optional[dict]:
        if response is None:
            return None
        try:
            data = response.json()
        except Exception:
            return None
        return data if isinstance(data, dict) else None

    def activate(self, token: str, machine_id: str) -> LicenseStatus:
        """Active une licence à partir d'un jeton. Stocke la licence si valide."""
        response = self._post(
            "/api/v1/activate", {"token": token, "machine_id": machine_id}
        )
        if response is None:
            return LicenseStatus.unavailable()
        if getattr(response, "status_code", 500) >= 400:
            payload = self._parse(response) or {}
            message = str(payload.get("detail") or payload.get("message") or "")
            return LicenseStatus.invalid(message or "Jeton refusé par le serveur")
        payload = self._parse(response)
        if payload is None:
            return LicenseStatus.unavailable()
        status = LicenseStatus.from_payload(payload, machine_id)
        if status.valid:
            # Stocke l'enveloppe signée reçue du serveur, jamais le jeton d'activation.
            self._store.save(payload)
        return status

    def check(self, license_ref: str, machine_id: str) -> LicenseStatus:
        """Vérifie l'état d'une licence existante auprès du portail."""
        response = self._post(
            "/api/v1/check", {"license_ref": license_ref, "machine_id": machine_id}
        )
        if response is None:
            return LicenseStatus.unavailable()
        if getattr(response, "status_code", 500) >= 400:
            payload = self._parse(response) or {}
            message = str(payload.get("detail") or payload.get("message") or "")
            return LicenseStatus.invalid(message or "Licence refusée par le serveur")
        payload = self._parse(response)
        if payload is None:
            return LicenseStatus.unavailable()
        status = LicenseStatus.from_payload(payload, machine_id)
        if status.valid:
            self._store.save(payload)
        return status

    def local_status(self) -> LicenseStatus:
        """État de licence depuis le stockage local, sans appel réseau."""
        data = self._store.load()
        if not data:
            return LicenseStatus.none()
        return LicenseStatus.from_payload(data)