"""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`` - ``latest_version()`` → ``GET /api/v1/version`` 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: ... def get(self, url: str, 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 def _get(self, endpoint: str) -> Optional[_HttpResponse]: try: session = self._get_session() return session.get(f"{self._base_url}{endpoint}", timeout=self._timeout) except Exception: 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 latest_version(self) -> Optional[dict[str, Any]]: """Retourne les métadonnées de la version active publiée sur le portail.""" response = self._get("/api/v1/version") if response is None or getattr(response, "status_code", 500) >= 400: return None payload = self._parse(response) return payload if payload is not None else None 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)