""" Module de télémétrie anonyme pour Web Sentinel. Collecte uniquement des statistiques anonymes : - Nombre de findings - Durée du scan - Modules utilisés - Plateforme (GUI/CLI Windows/Linux) ❌ AUCUNE donnée sensible collectée : - Pas de code source - Pas d'URLs/domaines scannés - Pas de données utilisateur - Pas d'IP """ import logging import os import platform import sys from typing import Dict, List, Optional import requests LOGGER = logging.getLogger(__name__) # URL de l'API de télémétrie TELEMETRY_API_URL = os.getenv( "WEB_SENTINEL_TELEMETRY_URL", "https://api.web-sentinel.taaazzz-prog.fr/api/v1/telemetry" ) # Timeout pour les requêtes télémétrie (ne pas bloquer l'application) TELEMETRY_TIMEOUT = 2.0 def _detect_platform() -> str: """ Détecte la plateforme d'exécution. Returns: str: 'windows-gui', 'windows-cli', 'linux-cli', 'macos-cli', 'unknown' """ system = platform.system().lower() # Détecter si exécution depuis .exe PyInstaller is_frozen = getattr(sys, 'frozen', False) if system == "windows": # Si GUI est importable et en mode frozen, c'est le GUI if is_frozen: try: import tkinter return "windows-gui" except ImportError: return "windows-cli" return "windows-cli" elif system == "linux": return "linux-cli" elif system == "darwin": return "macos-cli" else: return "unknown" def _get_app_version() -> str: """ Récupère la version de l'application. Returns: str: Version de l'application (ex: "1.0.0") """ try: # Essayer d'importer depuis setup.py ou __init__.py from . import __version__ return __version__ except (ImportError, AttributeError): return "1.0.0" def send_scan_telemetry( findings_count: int, scan_duration: float, modules_used: List[str], severity_counts: Optional[Dict[str, int]] = None, vulnerability_types: Optional[Dict[str, int]] = None, languages_detected: Optional[List[str]] = None, findings_by_language: Optional[Dict[str, int]] = None, vulnerabilities_by_lang: Optional[Dict[str, Dict[str, int]]] = None, ) -> bool: """ Envoie des statistiques anonymes de scan. Args: findings_count: Nombre total de findings détectés scan_duration: Durée du scan en secondes modules_used: Liste des modules utilisés (ex: ['headers', 'tls']) severity_counts: Comptage par sévérité (ex: {'critical': 2, 'high': 5}) vulnerability_types: Comptage par type de vulnérabilité (ex: {'SQL Injection': 3, 'XSS': 5}) languages_detected: Langages détectés dans le scan (ex: ['python', 'javascript']) findings_by_language: Comptage de findings par langage (ex: {'php': 2, 'javascript': 5}) vulnerabilities_by_lang: Mapping précis { langage: { type_vuln: count } } Returns: bool: True si envoi réussi, False sinon """ # Vérifier si télémétrie désactivée if os.getenv("WEB_SENTINEL_NO_TELEMETRY", "").lower() in ("1", "true", "yes"): LOGGER.debug("Télémétrie désactivée via WEB_SENTINEL_NO_TELEMETRY") return False try: payload = { "findings_count": findings_count, "scan_duration": round(scan_duration, 2), "modules_used": modules_used, "platform": _detect_platform(), "app_version": _get_app_version(), } if severity_counts: payload["severity_counts"] = severity_counts if vulnerability_types: payload["vulnerability_types"] = vulnerability_types if languages_detected: payload["languages_detected"] = languages_detected if findings_by_language: payload["findings_by_language"] = findings_by_language if vulnerabilities_by_lang: payload["vulnerabilities_by_lang"] = vulnerabilities_by_lang LOGGER.debug("Envoi télémétrie: %s", payload) response = requests.post( TELEMETRY_API_URL, json=payload, timeout=TELEMETRY_TIMEOUT, headers={"Content-Type": "application/json"} ) if response.status_code == 200: LOGGER.debug("✅ Télémétrie envoyée avec succès") return True else: LOGGER.debug("⚠️ Télémétrie rejetée: HTTP %d", response.status_code) return False except requests.RequestException as exc: # Ne pas bloquer l'application si télémétrie échoue LOGGER.debug("⚠️ Erreur télémétrie (non bloquant): %s", exc) return False except Exception as exc: LOGGER.debug("⚠️ Erreur télémétrie inattendue: %s", exc) return False def is_telemetry_enabled() -> bool: """ Vérifie si la télémétrie est activée. Returns: bool: True si activée, False sinon """ return os.getenv("WEB_SENTINEL_NO_TELEMETRY", "").lower() not in ("1", "true", "yes") def get_telemetry_info() -> Dict[str, str]: """ Retourne les informations de télémétrie pour affichage utilisateur. Returns: dict: Informations de configuration télémétrie """ return { "enabled": is_telemetry_enabled(), "api_url": TELEMETRY_API_URL, "platform": _detect_platform(), "app_version": _get_app_version(), "data_collected": "findings_count, scan_duration, modules_used, platform", "data_not_collected": "code, urls, domains, user_data, ip", "disable_method": "Définir WEB_SENTINEL_NO_TELEMETRY=1" } def send_authenticated_scan( api_key: str, code: str, filename: str, language: Optional[str] = None, timeout: float = 30.0 ) -> Dict: """ Envoie un scan AUTHENTIFIÉ vers l'API pour enregistrement dans les stats personnelles. Différence avec send_scan_telemetry() : - send_scan_telemetry() : Anonyme, pour stats globales - send_authenticated_scan() : Authentifié, pour stats personnelles de l'utilisateur Args: api_key: Clé API de l'utilisateur (depuis login) code: Code source à scanner filename: Nom du fichier (pour détecter le langage) language: Langage (optionnel, détecté automatiquement si non fourni) timeout: Timeout de la requête en secondes Returns: dict: Résultat du scan avec { 'success': bool, 'scan_id': int (si succès), 'vulnerabilities': list, 'error': str (si erreur) } """ api_url = os.getenv( "WEB_SENTINEL_API_URL", "https://api.web-sentinel.taaazzz-prog.fr/api/v1/scan" ) try: payload = { "code": code, "filename": filename, } if language: payload["language"] = language LOGGER.debug(f"Envoi scan authentifié: filename={filename}, language={language}") response = requests.post( api_url, json=payload, timeout=timeout, headers={ "Content-Type": "application/json", "X-API-Key": api_key } ) if response.status_code == 200: data = response.json() LOGGER.info(f"✅ Scan authentifié réussi: {data.get('summary', {}).get('total_vulnerabilities', 0)} vulns") return { 'success': True, 'scan_id': data.get('scan_id'), 'vulnerabilities': data.get('vulnerabilities', []), 'summary': data.get('summary', {}), 'data': data } elif response.status_code == 401: LOGGER.warning("❌ Scan authentifié échoué: API Key invalide") return { 'success': False, 'error': 'API Key invalide ou expirée' } else: LOGGER.warning(f"❌ Scan authentifié échoué: HTTP {response.status_code}") return { 'success': False, 'error': f'Erreur serveur: {response.status_code}' } except requests.RequestException as exc: LOGGER.error(f"❌ Erreur réseau scan authentifié: {exc}") return { 'success': False, 'error': f'Erreur réseau: {exc}' } except Exception as exc: LOGGER.error(f"❌ Erreur inattendue scan authentifié: {exc}") return { 'success': False, 'error': f'Erreur: {exc}' }