""" Gestionnaire d'authentification principal pour Web Sentinel Gère l'authentification locale et cloud de manière unifiée """ import json import os import platform import requests from pathlib import Path from typing import Optional, Dict, Any, List, Tuple from datetime import datetime, timedelta import sqlite3 import threading import uuid from .models import User, Session, AuthToken, Organization, LoginAttempt, AuthProvider, UserRole from ..subscription import SubscriptionManager, SubscriptionTier class AuthManager: """Gestionnaire d'authentification principal.""" def __init__(self, config_dir: Optional[Path] = None): self.config_dir = config_dir or (Path.home() / ".web-sentinel") self.config_dir.mkdir(parents=True, exist_ok=True) # Base de données locale pour cache/offline self.db_path = self.config_dir / "auth.db" self.auth_config_path = self.config_dir / "auth-config.json" # Session courante self.current_session: Optional[Session] = None self._session_lock = threading.Lock() # Configuration self.cloud_api_url = os.getenv('WEB_SENTINEL_AUTH_URL', 'https://api.web-sentinel.taaazzz-prog.fr/api/v1') self.offline_mode = False # Initialiser la base locale self._init_local_db() # Restaurer la session si possible self._restore_session() def _init_local_db(self) -> None: """Initialiser la base de données locale.""" with sqlite3.connect(self.db_path) as conn: conn.execute(""" CREATE TABLE IF NOT EXISTS users ( id TEXT PRIMARY KEY, email TEXT UNIQUE NOT NULL, username TEXT, data TEXT NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) conn.execute(""" CREATE TABLE IF NOT EXISTS sessions ( session_id TEXT PRIMARY KEY, user_id TEXT NOT NULL, data TEXT NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, last_activity TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (user_id) REFERENCES users(id) ) """) conn.execute(""" CREATE TABLE IF NOT EXISTS login_attempts ( id INTEGER PRIMARY KEY AUTOINCREMENT, email TEXT NOT NULL, ip_address TEXT, success BOOLEAN NOT NULL, error_reason TEXT, timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) conn.commit() def _get_device_info(self) -> Dict[str, str]: """Obtenir les informations de l'appareil.""" return { 'device_name': platform.node(), 'os': f"{platform.system()} {platform.release()}", 'python_version': platform.python_version(), 'app_version': '2.0.0' # Version Web Sentinel } def _log_login_attempt(self, email: str, success: bool, error_reason: Optional[str] = None) -> None: """Enregistrer une tentative de connexion.""" attempt = LoginAttempt( email=email, success=success, error_reason=error_reason, ip_address=self._get_local_ip() ) with sqlite3.connect(self.db_path) as conn: conn.execute(""" INSERT INTO login_attempts (email, ip_address, success, error_reason, timestamp) VALUES (?, ?, ?, ?, ?) """, ( attempt.email, attempt.ip_address, attempt.success, attempt.error_reason, attempt.timestamp.isoformat() )) conn.commit() def _get_local_ip(self) -> Optional[str]: """Obtenir l'IP locale (pour les logs).""" try: import socket s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.connect(("8.8.8.8", 80)) ip = s.getsockname()[0] s.close() return ip except Exception: return None def _save_session(self) -> None: """Sauvegarder la session courante.""" if not self.current_session: return session_data = { 'session_id': self.current_session.session_id, 'user_data': self.current_session.user.to_dict() if self.current_session.user else None, 'access_token': self.current_session.access_token.token if self.current_session.access_token else None, 'refresh_token': self.current_session.refresh_token.token if self.current_session.refresh_token else None, 'created_at': self.current_session.created_at.isoformat(), 'last_activity': self.current_session.last_activity.isoformat(), 'remember_me': self.current_session.remember_me, 'device_info': self._get_device_info() } with open(self.auth_config_path, 'w', encoding='utf-8') as f: json.dump(session_data, f, indent=2, ensure_ascii=False) def _restore_session(self) -> None: """Restaurer la session depuis le stockage local.""" try: if not self.auth_config_path.exists(): return with open(self.auth_config_path, 'r', encoding='utf-8') as f: session_data = json.load(f) # Reconstituer la session session = Session() session.session_id = session_data.get('session_id', str(uuid.uuid4())) session.created_at = datetime.fromisoformat(session_data['created_at']) session.last_activity = datetime.fromisoformat(session_data['last_activity']) session.remember_me = session_data.get('remember_me', False) # Reconstituer l'utilisateur if session_data.get('user_data'): session.user = User.from_dict(session_data['user_data']) # Reconstituer les tokens if session_data.get('access_token'): session.access_token = AuthToken( token=session_data['access_token'], user_id=session.user.id if session.user else "" ) # Vérifier si la session est encore valide if not session.is_expired and session.user: self.current_session = session # Mettre à jour l'activité self.current_session.update_activity() self._save_session() else: # Session expirée, la supprimer self._clear_session() except Exception as e: print(f"Erreur lors de la restauration de session: {e}") self._clear_session() def _clear_session(self) -> None: """Effacer la session courante.""" self.current_session = None if self.auth_config_path.exists(): self.auth_config_path.unlink() def is_authenticated(self) -> bool: """Vérifier si l'utilisateur est authentifié.""" with self._session_lock: return ( self.current_session is not None and self.current_session.is_authenticated and not self.current_session.is_expired ) def get_current_user(self) -> Optional[User]: """Obtenir l'utilisateur courant.""" if self.is_authenticated() and self.current_session: return self.current_session.user return None def login_email(self, email: str, password: str, remember_me: bool = False) -> Tuple[bool, Optional[str]]: """Connexion par email/mot de passe via l'API cloud.""" try: if self.offline_mode: return False, "Mode hors ligne non supporté" success, error = self._cloud_login_email(email, password, remember_me) if success: self._log_login_attempt(email, True) return True, None else: self._log_login_attempt(email, False, error) return False, error except Exception as e: error_msg = f"Erreur de connexion: {str(e)}" self._log_login_attempt(email, False, error_msg) return False, error_msg def _cloud_login_email(self, email: str, password: str, remember_me: bool) -> Tuple[bool, Optional[str]]: """Connexion cloud via API.""" try: response = requests.post( f"{self.cloud_api_url}/auth/login", json={ 'email': email, 'password': password, 'remember_me': remember_me, 'device_info': self._get_device_info() }, timeout=10 ) if response.status_code == 200: data = response.json() # Créer l'utilisateur depuis la réponse API user = User( id=data.get('user_id', email), email=data.get('email', email), username=data.get('name', email.split('@')[0]), api_key=data.get('api_key'), tier=data.get('tier') ) # Créer la session session = Session( user=user, remember_me=remember_me, device_name=self._get_device_info()['device_name'] ) # Créer les tokens - utiliser le JWT token pour access_token session.access_token = AuthToken( token=data.get('token'), # JWT token user_id=user.id, token_type="access" ) if data.get('refresh_token'): session.refresh_token = AuthToken( token=data['refresh_token'], user_id=user.id, token_type="refresh" ) # Stocker la session with self._session_lock: self.current_session = session self._save_session() return True, None elif response.status_code == 401: return False, "invalid_credentials" elif response.status_code == 403: return False, "account_disabled" else: return False, f"server_error_{response.status_code}" except requests.RequestException: return False, "network_error" def logout(self) -> None: """Déconnexion.""" if self.current_session: # Notifier le serveur si en ligne if not self.offline_mode and self.current_session.access_token: try: requests.post( f"{self.cloud_api_url}/auth/logout", headers={'Authorization': f'Bearer {self.current_session.access_token.token}'}, timeout=5 ) except requests.RequestException: pass # Ignorer les erreurs réseau lors de la déconnexion # Terminer la session locale self.current_session.logout() with self._session_lock: self._clear_session() def register_local_user(self, email: str, password: str, first_name: str = "", last_name: str = "") -> Tuple[bool, Optional[str]]: """Créer un compte utilisateur local.""" try: # Vérifier si l'utilisateur existe déjà with sqlite3.connect(self.db_path) as conn: cursor = conn.execute("SELECT id FROM users WHERE email = ?", (email,)) if cursor.fetchone(): return False, "user_exists" # Créer l'utilisateur user = User( email=email, first_name=first_name, last_name=last_name, auth_provider=AuthProvider.EMAIL, email_verified=True # Local = vérifié par défaut ) user.set_password(password) # Sauvegarder en base locale with sqlite3.connect(self.db_path) as conn: conn.execute(""" INSERT INTO users (id, email, username, data) VALUES (?, ?, ?, ?) """, ( user.id, user.email, user.username, json.dumps(user.to_dict()) )) conn.commit() return True, None except Exception as e: return False, f"registration_error: {str(e)}" def get_subscription_manager(self) -> SubscriptionManager: """Obtenir le gestionnaire d'abonnement pour l'utilisateur courant.""" if self.is_authenticated() and self.current_session and self.current_session.user: # Utiliser le tier de l'utilisateur (FREE, PRO, ENTERPRISE, SYSOP) user_tier = getattr(self.current_session.user, "tier", None) if user_tier: tier_str = user_tier.lower() if isinstance(user_tier, str) else str(user_tier).lower() try: tier = SubscriptionTier(tier_str) return SubscriptionManager( user_id=self.current_session.user.id, tier=tier, ) except ValueError: pass # Fallback: interroger l'API cloud tier = self._get_cloud_subscription_tier() return SubscriptionManager( user_id=self.current_session.user.id, tier=tier ) else: # Utilisateur non connecté = tier FREE return SubscriptionManager(user_id="anonymous", tier=SubscriptionTier.FREE) def _get_cloud_subscription_tier(self) -> SubscriptionTier: """Récupérer le tier d'abonnement depuis l'API cloud.""" try: if not self.current_session or not self.current_session.access_token: return SubscriptionTier.FREE response = requests.get( f"{self.cloud_api_url}/subscription/status", headers={'Authorization': f'Bearer {self.current_session.access_token.token}'}, timeout=5 ) if response.status_code == 200: data = response.json() tier_name = data.get('tier', 'free') try: return SubscriptionTier(tier_name.lower()) except ValueError: return SubscriptionTier.FREE except requests.RequestException: pass return SubscriptionTier.FREE def switch_to_offline_mode(self) -> None: """Basculer en mode hors ligne.""" self.offline_mode = True def switch_to_online_mode(self) -> None: """Basculer en mode en ligne.""" self.offline_mode = False def get_session_info(self) -> Dict[str, Any]: """Obtenir les informations de session.""" if not self.current_session: return {'authenticated': False} return { 'authenticated': self.is_authenticated(), 'user': self.current_session.user.to_dict() if self.current_session.user else None, 'session_id': self.current_session.session_id, 'created_at': self.current_session.created_at.isoformat(), 'last_activity': self.current_session.last_activity.isoformat(), 'offline_mode': self.offline_mode }