from __future__ import annotations import hashlib import json import os import socket import ssl from dataclasses import dataclass from typing import Any, Optional from urllib import error, parse, request from SRC.Services.license_models import LicenseEnvelope from SRC.Services.license_runtime_config import ( LICENSE_API_ACTIVATE_ENDPOINT, LICENSE_API_BASE_URL, LICENSE_API_CHECK_ENDPOINT, LICENSE_API_KEY, LICENSE_API_TIMEOUT_SECONDS, LICENSE_API_VERIFY_SSL, LICENSE_TLS_CERT_SHA256, ) DEFAULT_TIMEOUT_SECONDS = LICENSE_API_TIMEOUT_SECONDS DEFAULT_ACTIVATE_ENDPOINT = LICENSE_API_ACTIVATE_ENDPOINT DEFAULT_CHECK_ENDPOINT = LICENSE_API_CHECK_ENDPOINT BUSINESS_HTTP_STATUS_CODES = {400, 401, 403, 404, 409, 410, 422} class LicenseApiError(RuntimeError): pass @dataclass(slots=True) class LicenseApiConfig: base_url: str activate_endpoint: str = DEFAULT_ACTIVATE_ENDPOINT check_endpoint: str = DEFAULT_CHECK_ENDPOINT timeout_seconds: int = DEFAULT_TIMEOUT_SECONDS verify_ssl: bool = True api_key: Optional[str] = None pinned_cert_sha256: Optional[str] = None @property def is_configured(self) -> bool: return bool(self.base_url.strip()) def build_url(self, endpoint: str) -> str: base = self.base_url.rstrip("/") endpoint = endpoint.strip() if not endpoint.startswith("/"): endpoint = "/" + endpoint return base + endpoint @dataclass(slots=True) class LicenseApiResponse: success: bool message: str envelope: Optional[LicenseEnvelope] = None server_time_utc: Optional[str] = None raw: Optional[dict[str, Any]] = None class LicenseApiClient: def __init__(self, config: Optional[LicenseApiConfig] = None) -> None: self.config = config or load_license_api_config() def is_configured(self) -> bool: return self.config.is_configured def activate_license( self, *, license_key: str, installation_id: str, machine_fingerprint: str, fingerprint_data: dict[str, Any], ) -> LicenseApiResponse: payload = { "license_key": license_key.strip(), "installation_id": installation_id.strip(), "machine_fingerprint": machine_fingerprint, "fingerprint_data": fingerprint_data, "client_version": "lot-05-compat", } return self._post_json(self.config.activate_endpoint, payload) def check_license( self, *, license_key: str, installation_id: str, machine_fingerprint: str, current_license_id: Optional[str], fingerprint_data: dict[str, Any], ) -> LicenseApiResponse: payload = { "license_key": license_key.strip(), "installation_id": installation_id.strip(), "machine_fingerprint": machine_fingerprint, "fingerprint_data": fingerprint_data, "current_license_id": (current_license_id or "").strip() or None, "client_version": "lot-05-compat", } return self._post_json(self.config.check_endpoint, payload) def _post_json(self, endpoint: str, payload: dict[str, Any]) -> LicenseApiResponse: if not self.is_configured(): raise LicenseApiError( "Le service de licence n'est pas configure sur ce poste." ) url = self.config.build_url(endpoint) self._validate_server_certificate(url) data = json.dumps(payload, ensure_ascii=False).encode("utf-8") headers = { "Content-Type": "application/json; charset=utf-8", "Accept": "application/json", } if self.config.api_key: headers["X-Api-Key"] = self.config.api_key req = request.Request(url, data=data, headers=headers, method="POST") context = None if not self.config.verify_ssl: context = ssl._create_unverified_context() try: with request.urlopen(req, timeout=self.config.timeout_seconds, context=context) as resp: body = resp.read().decode("utf-8") return self._build_response(body) except error.HTTPError as exc: detail = exc.read().decode("utf-8", errors="replace") if exc.code in BUSINESS_HTTP_STATUS_CODES: parsed = self._try_build_response(detail) if parsed is not None: return parsed raise LicenseApiError(f"Reponse HTTP {exc.code} depuis le service de licence : {detail}") from exc except error.URLError as exc: reason = exc.reason if isinstance(reason, socket.timeout): raise LicenseApiError("Le service de licence n'a pas repondu dans le delai imparti.") from exc raise LicenseApiError(f"Service de licence injoignable : {reason}") from exc except TimeoutError as exc: raise LicenseApiError("Le service de licence n'a pas repondu dans le delai imparti.") from exc def _validate_server_certificate(self, url: str) -> None: pinned = (self.config.pinned_cert_sha256 or "").strip().lower() if not pinned: return parsed = parse.urlparse(url) if parsed.scheme.lower() != "https": raise LicenseApiError("Le pinning TLS exige une URL HTTPS.") hostname = parsed.hostname or "" if not hostname: raise LicenseApiError("Nom d'hote du service de licence introuvable.") port = parsed.port or 443 actual = fetch_server_certificate_sha256( hostname=hostname, port=port, timeout_seconds=self.config.timeout_seconds, verify_ssl=self.config.verify_ssl, ) if actual.lower() != pinned: raise LicenseApiError( "Le certificat TLS du service de licence ne correspond pas a l'empreinte attendue." ) def _try_build_response(self, body: str) -> Optional[LicenseApiResponse]: try: return self._build_response(body) except LicenseApiError: return None def _build_response(self, body: str) -> LicenseApiResponse: try: response_payload = json.loads(body) except json.JSONDecodeError as exc: raise LicenseApiError("La reponse du service de licence n'est pas un JSON valide.") from exc envelope_data = response_payload.get("license") or response_payload.get("envelope") envelope = LicenseEnvelope.from_dict(envelope_data) if isinstance(envelope_data, dict) else None message = str(response_payload.get("message", "")).strip() if not message: message = self._translate_error_code(str(response_payload.get("error_code", "")).strip()) return LicenseApiResponse( success=bool(response_payload.get("success", response_payload.get("ok"))), message=message, envelope=envelope, server_time_utc=(str(response_payload.get("server_time_utc", "")).strip() or None), raw=response_payload, ) @staticmethod def _translate_error_code(error_code: str) -> str: mapping = { "license_revoked": "La licence a ete revoquee.", "license_expired": "La licence est expiree.", "license_not_found": "La cle de licence est inconnue.", "license_not_activated": "La licence n'est pas encore activee.", "machine_mismatch": "Cette licence est liee a une autre machine.", "activation_already_bound": "Cette licence est deja activee sur une autre machine.", "invalid_api_key": "La cle d'acces au service de licence est invalide.", "invalid_request": "La requete envoyee au service de licence est invalide.", "invalid_json": "La reponse du serveur de licences est invalide.", "server_error": "Le service de licence a rencontre une erreur interne.", } return mapping.get(error_code, error_code or "Reponse du service de licence non documentee.") def fetch_server_certificate_sha256( *, hostname: str, port: int, timeout_seconds: int, verify_ssl: bool, ) -> str: context = ssl.create_default_context() if verify_ssl else ssl._create_unverified_context() with socket.create_connection((hostname, port), timeout=timeout_seconds) as tcp_sock: with context.wrap_socket(tcp_sock, server_hostname=hostname) as tls_sock: cert_der = tls_sock.getpeercert(binary_form=True) return hashlib.sha256(cert_der).hexdigest() def load_license_api_config() -> LicenseApiConfig: env_base_url = os.environ.get("TRAIL_DPS_LICENSE_API_BASE_URL", "").strip() env_timeout = os.environ.get("TRAIL_DPS_LICENSE_API_TIMEOUT_S", "").strip() timeout = int(env_timeout) if env_timeout.isdigit() else DEFAULT_TIMEOUT_SECONDS return LicenseApiConfig( base_url=env_base_url or LICENSE_API_BASE_URL, timeout_seconds=timeout, activate_endpoint=os.environ.get("TRAIL_DPS_LICENSE_API_ACTIVATE_PATH", DEFAULT_ACTIVATE_ENDPOINT).strip() or DEFAULT_ACTIVATE_ENDPOINT, check_endpoint=os.environ.get("TRAIL_DPS_LICENSE_API_CHECK_PATH", DEFAULT_CHECK_ENDPOINT).strip() or DEFAULT_CHECK_ENDPOINT, verify_ssl=os.environ.get("TRAIL_DPS_LICENSE_API_VERIFY_SSL", "1" if LICENSE_API_VERIFY_SSL else "0").strip() not in {"0", "false", "False"}, api_key=(os.environ.get("TRAIL_DPS_LICENSE_API_KEY", "").strip() or LICENSE_API_KEY), pinned_cert_sha256=(os.environ.get("TRAIL_DPS_LICENSE_TLS_CERT_SHA256", "").strip() or LICENSE_TLS_CERT_SHA256 or None), )