from __future__ import annotations """Bridge entre la licence embarquée (.lic) et le runtime (GUI/CLI).""" from dataclasses import dataclass from pathlib import Path from typing import Optional from ..subscription import SubscriptionManager, SubscriptionTier from ..subscription.models import SubscriptionFeatures, get_features_for_tier from ..subscription.validator import LicenseValidator from .license_manager import LicenceError from .defaults import ( detect_active_session, generate_default_free_license, generate_license_for_tier, generate_sysop_license, resolve_user_subscription_tier, ) from .runtime import ( LicenceRuntimeState, install_licence_file, load_runtime_state, resolve_license_path, ) @dataclass class _AlertInfo: level: Optional[str] threshold: Optional[int] class LicenseRuntimeBridge: """Gestionnaire centralisé du statut de licence pour l'application.""" def __init__(self, licence_path: Optional[Path] = None): self.licence_path = licence_path or resolve_license_path() self.hardware_fingerprint = "" self.state: Optional[LicenceRuntimeState] = None # Détecter l'utilisateur actuel depuis la session active active_user_id = self._detect_current_user() self.subscription_manager = SubscriptionManager(user_id=active_user_id) self.runtime_validator = LicenseValidator(self.subscription_manager) self.initialized = False self.initialization_error: Optional[str] = None # --- Cycle de vie ------------------------------------------------- def initialize(self, hardware_fingerprint: str = "") -> bool: """Charger la licence embarquée. Retourne True si licence valide.""" self.hardware_fingerprint = hardware_fingerprint return self._refresh_state() def refresh(self) -> bool: """Actualiser le statut licence (après import ou timer).""" return self._refresh_state() def install_new_license(self, source_path: Path) -> bool: """Installer une nouvelle licence et recharger l'état.""" install_licence_file(source_path, self.licence_path) return self.refresh() def apply_subscription(self, tier: SubscriptionTier, owner_email: Optional[str] = None) -> bool: """Forcer la licence selon le tier courant.""" owner_email = (owner_email or "").strip() try: if tier == SubscriptionTier.SYSOP: account = [ { "email": owner_email or "sysop@web-sentinel.dev", "role": "sysop", "is_active": True, } ] generate_sysop_license(self.licence_path, account) elif tier == SubscriptionTier.FREE: generate_default_free_license(self.licence_path) else: generate_license_for_tier(self.licence_path, tier, owner_email) except Exception as exc: # pragma: no cover - génération licence self.initialized = False self.initialization_error = str(exc) self.state = None return False return self.initialize(self.hardware_fingerprint) # --- Accès runtime ------------------------------------------------ def get_runtime_manager(self) -> SubscriptionManager: if not self.initialized: raise RuntimeError("Système de licence non initialisé.") return self.subscription_manager def get_runtime_validator(self) -> LicenseValidator: if not self.initialized: raise RuntimeError("Système de licence non initialisé.") return self.runtime_validator def validate_feature_access(self, feature: str) -> bool: """Valider l'accès à une fonctionnalité spécifique.""" if feature == "invasiveTests": return self.subscription_manager.can_use_invasive_tests() if feature == "htmlExport": return self.subscription_manager.can_export_html() if feature == "apiAccess": return self.subscription_manager.can_use_api() return False # --- Informations ------------------------------------------------- def get_license_status(self) -> dict: """Retourner un dictionnaire prêt pour l'UI/CLI.""" if not self.initialized: return { "status": "ERROR", "tier": "FREE", "valid": False, "error": self.initialization_error or "Licence non initialisée", } features = self.subscription_manager.get_features() features_enabled = { "invasive_tests": self.subscription_manager.can_use_invasive_tests(), "html_export": self.subscription_manager.can_export_html(), "multi_user": self.subscription_manager.can_manage_users(), "api_access": self.subscription_manager.can_use_api(), "source_scan": self.subscription_manager.can_scan_source(), } if not self.state: # Utiliser le tier du subscription_manager qui est résolu depuis SQLite actual_tier = self.subscription_manager.tier status_mode = "FREE_MODE" if actual_tier == SubscriptionTier.FREE else "SUBSCRIPTION_MODE" return { "status": status_mode, "tier": actual_tier.value.upper(), "valid": True, "license_id": None, "expires_in_days": None, "grace_hours": 0, "max_domains": features.domain_limit, "max_users": features.max_users, "max_source_files": features.max_source_files, "max_source_size_mb": features.max_source_size_mb, "features": {}, "features_enabled": features_enabled, "alert_level": None, "alert_threshold": None, "error": self.initialization_error, "backend": { "name": "none", "verified": False, }, } info = self.state.status.info expires_in_days = int(self.state.status.expires_in_days) alert = _compute_alert(self.state.alert_threshold, self.state.status.expires_in_days) return { "status": "LICENSED", "tier": self.subscription_manager.tier.value.upper(), "valid": not self.state.is_expired, "license_id": info.licence_id, "valid_until": info.valid_until.isoformat(), "expires_in_days": expires_in_days, "grace_hours": info.grace_period_hours, "max_domains": features.domain_limit, "max_users": features.max_users, "max_source_files": features.max_source_files, "max_source_size_mb": features.max_source_size_mb, "features": info.features, "features_enabled": features_enabled, "alert_level": alert.level, "alert_threshold": alert.threshold, "error": None, "backend": { "name": getattr(self.state, "backend_name", "none"), "verified": getattr(self.state, "backend_verified", False), }, } def get_super_admin_credentials(self) -> Optional[tuple[str, str]]: if self.state: return self.state.status.info.super_admin_email, "" return None def get_database_encryption_seed(self) -> Optional[str]: if self.state: return self.state.status.info.db_seed return None # --- Internes ----------------------------------------------------- def _detect_current_user(self) -> str: """Détecter l'utilisateur actuel depuis la session active ou défaut.""" # Essayer d'abord la session active standard active_session = detect_active_session() if active_session: email = active_session[0].get("email", "") if email: return email # Si pas de session "remember_me", chercher directement dans auth-config.json from pathlib import Path import json session_path = Path.home() / ".web-sentinel" / "auth-config.json" if session_path.exists(): try: session_data = json.loads(session_path.read_text(encoding="utf-8")) user_data = session_data.get("user_data", {}) if isinstance(user_data, dict) and user_data.get("is_active", True): email = user_data.get("email", "") if email: return email except (json.JSONDecodeError, OSError): pass return "licensed_user" # Fallback par défaut def _refresh_state(self) -> bool: try: self.state = load_runtime_state(self.licence_path, self.hardware_fingerprint) except LicenceError as exc: self.initialization_error = str(exc) self.state = None self._apply_features(get_features_for_tier(SubscriptionTier.FREE)) self.initialized = True return False else: self.initialization_error = None self._apply_state(self.state) return True def _apply_state(self, state: LicenceRuntimeState) -> None: self._apply_features(state.features) self.initialized = True def _apply_features(self, features: SubscriptionFeatures) -> None: self.subscription_manager.apply_features(features) self.runtime_validator = LicenseValidator(self.subscription_manager) def _compute_alert(threshold: Optional[int], expires_in_days: float) -> _AlertInfo: if threshold is None: return _AlertInfo(level=None, threshold=None) if expires_in_days < 0: return _AlertInfo(level="CRITICAL", threshold=0) if threshold <= 3: return _AlertInfo(level="CRITICAL", threshold=threshold) if threshold <= 14: return _AlertInfo(level="WARNING", threshold=threshold) return _AlertInfo(level="INFO", threshold=threshold) _GLOBAL_BRIDGE: Optional[LicenseRuntimeBridge] = None def get_license_bridge() -> LicenseRuntimeBridge: global _GLOBAL_BRIDGE if _GLOBAL_BRIDGE is None: _GLOBAL_BRIDGE = LicenseRuntimeBridge() return _GLOBAL_BRIDGE def initialize_license_system(hardware_fingerprint: str = "") -> bool: bridge = get_license_bridge() licence_path = bridge.licence_path def _attempt_initialize() -> bool: try: return bridge.initialize(hardware_fingerprint) except Exception as exc: # pragma: no cover - defensive fallback bridge.initialized = False bridge.initialization_error = str(exc) bridge.state = None return False active_session = detect_active_session() if active_session: payload = active_session[0] role = str(payload.get("role", "member")).lower() email = payload.get("email", "") if role == SubscriptionTier.SYSOP.value: try: generate_sysop_license(licence_path, active_session) except Exception: return _attempt_initialize() return _attempt_initialize() user_tier = resolve_user_subscription_tier(email) if user_tier and user_tier != SubscriptionTier.FREE: try: generate_license_for_tier(licence_path, user_tier, email) except Exception: return _attempt_initialize() return _attempt_initialize() try: generate_default_free_license(licence_path) except Exception: return _attempt_initialize() return _attempt_initialize()