from __future__ import annotations """Backends de validation externe des licences.""" import os import sqlite3 from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path from typing import Optional, Protocol import requests from .license_manager import LicenceError, LicenceInfo OFFLINE_TIERS = {"GOVERNMENT", "MEDICAL", "FINANCE", "DEFENSE", "AIRGAP"} DEFAULT_SQLITE_PATH = Path.home() / ".web-sentinel" / "auth.db" class LicenceBackend(Protocol): """Interface minimale pour les backends de validation.""" name: str def verify(self, info: LicenceInfo) -> None: """Valider une licence. Lève `LicenceError` en cas d'échec.""" @dataclass class NullBackend: """Backend qui n'applique aucune validation supplémentaire.""" name: str = "none" def verify(self, info: LicenceInfo) -> None: # noqa: D401 - interface simple return @dataclass class SQLiteBackend: """Validation sur un fichier SQLite (auth.db).""" db_path: Path name: str = "sqlite" def verify(self, info: LicenceInfo) -> None: if not self.db_path.exists(): raise LicenceError(f"Base SQLite introuvable: {self.db_path}") if not info.super_admin_email: raise LicenceError("La licence ne contient pas d'email super admin pour vérification.") with sqlite3.connect(self.db_path) as conn: cursor = conn.execute( "SELECT tier, expires_at FROM user_subscriptions WHERE email = ?", (info.super_admin_email.lower(),), ) row = cursor.fetchone() if row is None: raise LicenceError(f"Aucune souscription externe pour {info.super_admin_email}.") db_tier, db_expires_at = row tier_mismatch = db_tier and db_tier.lower() != info.tier.lower() if tier_mismatch: raise LicenceError( f"Incohérence de tier: licence={info.tier} / base={db_tier}." ) if db_expires_at: try: db_expiry = datetime.fromisoformat(str(db_expires_at).replace("Z", "+00:00")) except ValueError: db_expiry = None if db_expiry and db_expiry < datetime.now(timezone.utc): raise LicenceError("La souscription externe est expirée.") @dataclass class PostgreSQLBackend: """Validation sur une base PostgreSQL distante.""" db_url: str name: str = "postgresql" def verify(self, info: LicenceInfo) -> None: if not info.super_admin_email: raise LicenceError("La licence ne contient pas d'email super admin pour vérification.") try: import psycopg2 except ImportError: raise LicenceError("Dépendance psycopg2 manquante pour PostgreSQL backend") try: with psycopg2.connect(self.db_url) as conn: with conn.cursor() as cursor: cursor.execute( "SELECT tier, expires_at FROM api_key WHERE email = %s AND is_active = TRUE", (info.super_admin_email.lower(),) ) row = cursor.fetchone() if row is None: raise LicenceError(f"Aucune souscription PostgreSQL pour {info.super_admin_email}.") db_tier, db_expires_at = row tier_mismatch = db_tier and db_tier.lower() != info.tier.lower() if tier_mismatch: raise LicenceError( f"Incohérence de tier: licence={info.tier} / base={db_tier}." ) if db_expires_at: try: db_expiry = datetime.fromisoformat(str(db_expires_at).replace("Z", "+00:00")) except ValueError: db_expiry = None if db_expiry and db_expiry < datetime.now(timezone.utc): raise LicenceError("La souscription PostgreSQL est expirée.") except Exception as e: if isinstance(e, LicenceError): raise raise LicenceError(f"Erreur PostgreSQL: {str(e)}") @dataclass class HTTPBackend: """Validation via API REST distante.""" api_url: str api_key: str name: str = "http" def verify(self, info: LicenceInfo) -> None: if not info.super_admin_email: raise LicenceError("La licence ne contient pas d'email super admin pour vérification.") try: response = requests.post( f"{self.api_url}/api/v1/license/verify", json={ "email": info.super_admin_email, "tier": info.tier, "license_id": getattr(info, 'license_id', None) }, headers={ "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json" }, timeout=10 ) if response.status_code == 200: data = response.json() if not data.get("valid", False): raise LicenceError(f"Licence invalide: {data.get('reason', 'Unknown')}") elif response.status_code == 404: raise LicenceError(f"Aucune souscription API pour {info.super_admin_email}.") elif response.status_code == 401: raise LicenceError("Clé API invalide pour la validation de licence.") else: raise LicenceError(f"Erreur API: HTTP {response.status_code}") except requests.RequestException as e: raise LicenceError(f"Erreur réseau lors de la validation: {str(e)}") def requires_backend(info: LicenceInfo) -> bool: """Déterminer si une licence doit être validée par un backend externe.""" tier = info.tier.upper() if tier in OFFLINE_TIERS: return False if info.features.get("air_gap"): return False return tier in {"FREE", "PRO", "ENTERPRISE"} def load_backend() -> LicenceBackend: """Charger le backend depuis la configuration d'environnement.""" mode = os.getenv("WEB_SENTINEL_SUBSCRIPTION_BACKEND", "auto").lower() disable_local = os.getenv("WEB_SENTINEL_DISABLE_LOCAL_DB", "").lower() in {"1", "true", "yes"} if mode == "none": return NullBackend() if mode == "postgresql": db_url = os.getenv("WEB_SENTINEL_POSTGRES_URL") if not db_url: raise LicenceError("WEB_SENTINEL_POSTGRES_URL requis pour le backend PostgreSQL") return PostgreSQLBackend(db_url=db_url) if mode == "http": api_url = os.getenv("WEB_SENTINEL_LICENSE_API_URL", "https://api.web-sentinel.taaazzz-prog.fr") api_key = os.getenv("WEB_SENTINEL_LICENSE_API_KEY") if not api_key: raise LicenceError("WEB_SENTINEL_LICENSE_API_KEY requis pour le backend HTTP") return HTTPBackend(api_url=api_url, api_key=api_key) if mode in {"sqlite", "auto"}: db_path = Path(os.getenv("WEB_SENTINEL_SUBSCRIPTION_DB", str(DEFAULT_SQLITE_PATH))).expanduser() if disable_local and mode == "sqlite": raise LicenceError("Le backend SQLite est d�sactiv� (WEB_SENTINEL_DISABLE_LOCAL_DB=1).") if mode == "auto" and (disable_local or not db_path.exists()): # En mode auto, essayer d'abord HTTP puis PostgreSQL puis SQLite try: api_url = os.getenv("WEB_SENTINEL_LICENSE_API_URL", "https://api.web-sentinel.taaazzz-prog.fr") api_key = os.getenv("WEB_SENTINEL_LICENSE_API_KEY") if api_key: return HTTPBackend(api_url=api_url, api_key=api_key) except: pass try: db_url = os.getenv("WEB_SENTINEL_POSTGRES_URL") if db_url: return PostgreSQLBackend(db_url=db_url) except: pass return NullBackend() if disable_local: return NullBackend() return SQLiteBackend(db_path=db_path) return NullBackend()