from __future__ import annotations from collections import Counter import os import threading from pathlib import Path from typing import List, Optional, Sequence import tkinter as tk from tkinter import filedialog, messagebox, ttk from ...checks.source_code.scanner import ( DEFAULT_EXCLUSIONS, LANGUAGE_EXTENSIONS, SourceCodeScanner, SourceScanConfig, ) from ...logging_config import get_logger from ...model import ScanRequest, ScanResult from ...scanner import SentinelScanner from ..i18n import localization LOGGER = get_logger("gui.sast") class SASTSection: """Gestion du volet SAST décorrélée de la classe principale GUI.""" def __init__(self, gui: "ModernWebSentinelGUI") -> None: self.gui = gui self._scan_messages: List[str] = [] # Messages du scan authentifié # ------------------------------------------------------------------ # Permissions & informations # ------------------------------------------------------------------ def can_use_sast(self) -> bool: gui = self.gui try: features = gui._get_runtime_features() if getattr(features, "allow_source_scan", False): return True # Vérifier le tier de l'utilisateur (priorité au tier stocké dans user) user = gui.auth_manager.get_current_user() if user: user_tier = str(getattr(user, "tier", "")).upper() if user_tier in {"PRO", "ENTERPRISE", "SYSOP"}: return True # Fallback: vérifier license_status status = getattr(gui, "license_status", {}) or {} tier_label = str(status.get("tier", "")).upper() if tier_label in {"PRO", "ENTERPRISE", "SYSOP"}: return True if gui.license_validator: return gui.license_validator.can_use_feature("source_scan") except Exception as exc: # pragma: no cover - robustesse LOGGER.warning(gui._text("sast_check_error", error=exc)) return False def update_info(self) -> None: gui = self.gui frame = getattr(gui, "sast_info_frame", None) if frame is None: return for widget in frame.winfo_children(): widget.destroy() if self.can_use_sast(): features = gui._get_runtime_features() max_files_text = gui._format_limit_value( getattr(features, "max_source_files", -1), unlimited_key="unlimited_files" ) max_size_text = gui._format_size_value(getattr(features, "max_source_size_mb", -1)) info_text = gui._text("sast_info_limits", files=max_files_text, size=max_size_text) color = "#059669" else: info_text = gui._text("sast_info_disabled") color = "#dc2626" ttk.Label( frame, text=info_text, foreground=color, font=("Segoe UI", 10, "bold"), ).pack() scan_button = getattr(gui, "sast_scan_button", None) if scan_button is not None: scan_button.config(state=tk.NORMAL if self.can_use_sast() else tk.DISABLED) try: # Ajuster l'état de l'onglet index = gui.notebook.index(gui.sast_tab) gui.notebook.tab(index, state="normal" if self.can_use_sast() else "disabled") except Exception: pass # ------------------------------------------------------------------ # Sélection des fichiers # ------------------------------------------------------------------ def browse_folder(self) -> None: gui = self.gui folder = filedialog.askdirectory(title=gui._text("dialog_select_folder")) if folder: gui.folder_var.set(folder) self.populate_files_from_folder(folder) def browse_files(self) -> None: gui = self.gui files = filedialog.askopenfilenames( title=gui._text("dialog_select_files"), filetypes=[ (gui._text("source_files_filter"), "*.php *.js *.py *.java *.cs *.go *.cpp *.c *.h"), ("Tous les fichiers", "*.*"), ], ) if files: for file in files: if file not in gui.files_listbox.get(0, tk.END): gui.files_listbox.insert(tk.END, file) def populate_files_from_folder(self, folder: str) -> None: gui = self.gui gui.files_listbox.delete(0, tk.END) # Utiliser TOUTES les extensions supportées par le scanner all_extensions = set() for lang_extensions in LANGUAGE_EXTENSIONS.values(): all_extensions.update(lang_extensions) folder_path = Path(folder) iterator = folder_path.rglob("*") if gui.sast_recursive_var.get() else folder_path.glob("*") # Patterns d'exclusion (synchronisés avec DEFAULT_EXCLUSIONS du scanner) exclusion_folders = { 'node_modules', 'vendor', '.venv', 'venv', '__pycache__', '.git', '.svn', '.hg', 'build', 'dist', 'target', 'bin', 'obj', '.vscode', '.idea' } for file_path in iterator: # Vérifier si le fichier est dans un dossier exclu if any(excluded in file_path.parts for excluded in exclusion_folders): continue # Vérifier l'extension et que c'est un fichier if file_path.is_file() and file_path.suffix.lower() in all_extensions: # Exclure les fichiers minifiés/compilés if not any(pattern in file_path.name.lower() for pattern in ['.min.', '.bundle.', 'chunk-', 'vendor']): gui.files_listbox.insert(tk.END, str(file_path)) def remove_selected_file(self) -> None: gui = self.gui selection = gui.files_listbox.curselection() if selection: gui.files_listbox.delete(selection[0]) def clear_files_list(self) -> None: self.gui.files_listbox.delete(0, tk.END) # ------------------------------------------------------------------ # Scan SAST # ------------------------------------------------------------------ def start_scan(self) -> None: gui = self.gui if not self.can_use_sast(): messagebox.showwarning(gui._text("attention_title"), gui._text("sast_info_disabled")) return if gui.current_sast_thread and gui.current_sast_thread.is_alive(): messagebox.showinfo(gui._text("info_sast_running_title"), gui._text("info_sast_running_body")) return folder = gui.folder_var.get().strip() raw_files = list(gui.files_listbox.get(0, tk.END)) if not folder and not raw_files: messagebox.showwarning(gui._text("attention_title"), gui._text("sast_scan_incomplete_body")) return source_path = Path(folder).expanduser().resolve() if folder else None config, candidate_files = self._build_config(source_path, raw_files) if not candidate_files: warning_message = gui._text("sast_no_files_warning") gui.sast_results_text.config(state=tk.NORMAL) gui.sast_results_text.delete("1.0", tk.END) gui.sast_results_text.insert(tk.END, warning_message + "\n\n") gui.sast_results_text.config(state=tk.DISABLED) gui._set_status("status_no_source") messagebox.showwarning(gui._text("attention_title"), warning_message) return gui._expected_sast_files = len(candidate_files) self.clear_results() folder_display = str(source_path) if source_path else gui._text("manual_selection_label") info_line = ( f"{gui._text('sast_selected_folder', path=folder_display)}\n" f"{gui._text('sast_detected_files', count=len(candidate_files))}\n" ) gui.sast_results_text.config(state=tk.NORMAL) gui.sast_results_text.insert(tk.END, info_line) gui.sast_results_text.insert( tk.END, f"{gui._text('sast_scan_started_line', count=len(candidate_files))}\n\n" ) gui.sast_results_text.config(state=tk.DISABLED) gui.sast_results_text.see(tk.END) gui.sast_progress_var.set(5) gui.sast_scan_button.config(state=tk.DISABLED) gui._set_status("status_scan_running") request = ScanRequest( domain=config.source_path.name if config.source_path else "SAST", source_path=str(config.source_path) if config.source_path else None, source_files=tuple(str(path) for path in candidate_files), source_languages=tuple(config.languages), source_exclude=tuple(config.exclude_patterns), source_recursive=config.recursive, source_detailed_report=config.detailed_report, source_min_severity=config.min_severity, source_max_files=config.max_files, source_max_size_mb=config.max_size_mb, source_advanced_rules=config.advanced_rules, ) gui.save_config() thread = threading.Thread(target=self._run_scan_thread, args=(request,), daemon=True) gui.current_sast_thread = thread thread.start() def _build_config( self, source_path: Optional[Path], raw_files: Sequence[str], ) -> tuple[SourceScanConfig, List[Path]]: gui = self.gui resolved_files = [Path(path).expanduser().resolve() for path in raw_files] # Détection automatique des langages depuis les extensions des fichiers detected_languages = set() for file_path in resolved_files: file_extension = file_path.suffix.lower() for language, extensions in LANGUAGE_EXTENSIONS.items(): if file_extension in extensions: detected_languages.add(language) break # Fallback sur "auto" si aucun langage détecté languages = detected_languages if detected_languages else {"auto"} exclude_patterns = set(DEFAULT_EXCLUSIONS) features = gui._get_runtime_features() config = SourceScanConfig( source_path=source_path, source_files=resolved_files, languages=languages, exclude_patterns=exclude_patterns, min_severity="medium", max_files=getattr(features, "max_source_files", -1), max_size_mb=getattr(features, "max_source_size_mb", -1), advanced_rules=getattr(features, "advanced_rules", False), recursive=bool(gui.sast_recursive_var.get()), detailed_report=bool(gui.sast_detailed_var.get()), ) preview_scanner = SourceCodeScanner(config) candidate_files = preview_scanner._discover_files() # Garder 'auto' pour scanner tous les fichiers reconnus # Ne pas remplacer par des langages spécifiques pour éviter de perdre des fichiers # L'auto-détection se fera lors du scan de chaque fichier detected_languages = set() for file_path in candidate_files: ext = file_path.suffix.lower() for lang, extensions in LANGUAGE_EXTENSIONS.items(): if ext in extensions: detected_languages.add(lang) break # Ne pas remplacer config.languages - garder 'auto' # if detected_languages: # config.languages = detected_languages if source_path: gui._last_sast_root = source_path elif candidate_files: common = os.path.commonpath([str(path.parent) for path in candidate_files]) gui._last_sast_root = Path(common) else: gui._last_sast_root = None return config, candidate_files def _run_scan_thread(self, request: ScanRequest) -> None: self._scan_messages.clear() # 🔄 Nettoyer les messages du scan précédent scanner = SentinelScanner() try: # Mesurer durée du scan import time scan_start = time.time() result = scanner.run(request, enabled_modules=["source-code"]) scan_duration = time.time() - scan_start # 📊 DOUBLE ENVOI : Télémétrie anonyme ET scan authentifié # 1. Télémétrie anonyme (stats globales avec détails des vulnérabilités) try: from ...telemetry import send_scan_telemetry from collections import Counter # Comptage par sévérité severity_counts = dict(Counter(f.severity for f in result.findings)) # 🔍 Extraction du langage depuis le champ `check` de chaque Finding # Exemples: "source-code-php" → "php", "source-code-js" → "javascript" # Mapping complet des 27 langages supportés par Web Sentinel language_mapping = { # Backend Languages (Advanced Analysis) "php": "php", "js": "javascript", "python": "python", "java": "java", "csharp": "c#", "cpp": "c++", "go": "go", "ruby": "ruby", "rust": "rust", "kotlin": "kotlin", "scala": "scala", "swift": "swift", "objc": "objective-c", "dart": "dart", # DevOps & Infrastructure "dockerfile": "dockerfile", "shell": "shell", "powershell": "powershell", "terraform": "terraform", # Data & Config Languages "sql": "sql", "html": "html", "css": "css", "xml": "xml", "json": "json", "yaml": "yaml", } # Construire un dictionnaire { langage: { type_vuln: count } } vulnerabilities_by_lang = {} for finding in result.findings: # 1. Essayer d'extraire depuis check="source-code-{lang}" check_parts = finding.check.split("-") lang = None if len(check_parts) >= 3 and check_parts[2] in language_mapping: lang = language_mapping[check_parts[2]] # 2. Si pas trouvé, extraire depuis l'extension du fichier dans evidence if not lang and finding.evidence: import re # Format: "Fichier: example.php, Ligne: 42" file_match = re.search(r'Fichier:\s*([^\s,]+)', finding.evidence) if file_match: filename = file_match.group(1) file_ext = Path(filename).suffix.lower() # Trouver le langage correspondant à cette extension for lang_key, extensions in LANGUAGE_EXTENSIONS.items(): if file_ext in extensions: lang = language_mapping.get(lang_key, lang_key) break # 3. Fallback: langage inconnu if not lang: lang = "other" # Initialiser si nécessaire if lang not in vulnerabilities_by_lang: vulnerabilities_by_lang[lang] = {} # Compter les vulnérabilités par type pour ce langage vuln_title = finding.title vulnerabilities_by_lang[lang][vuln_title] = \ vulnerabilities_by_lang[lang].get(vuln_title, 0) + 1 # Comptage global par type de vulnérabilité (pour compatibilité) vulnerability_types = dict(Counter(f.title for f in result.findings)) # Comptage global par langage findings_by_language = { lang: sum(counts.values()) for lang, counts in vulnerabilities_by_lang.items() } # Liste des langages détectés (pour compatibilité) languages_detected = list(findings_by_language.keys()) if findings_by_language else ["auto"] send_scan_telemetry( findings_count=len(result.findings), scan_duration=scan_duration, modules_used=["source-code"], severity_counts=severity_counts, vulnerability_types=vulnerability_types, languages_detected=languages_detected, findings_by_language=findings_by_language, vulnerabilities_by_lang=vulnerabilities_by_lang # 🆕 Structure précise ) except Exception: pass # Télémétrie optionnelle, ne pas bloquer si erreur # 2. Scan authentifié (stats personnelles, si utilisateur connecté) is_auth = self.gui.auth_manager.is_authenticated() if self.gui.auth_manager else False if self.gui.auth_manager and is_auth: try: # Récupérer l'API key depuis le user object (pas le JWT token) user = self.gui.auth_manager.get_current_user() if not user or not hasattr(user, 'api_key') or not user.api_key: LOGGER.error(f"❌ API key manquante! User: {vars(user) if user else None}") self._append_to_results(f"❌ Impossible d'enregistrer le scan: API key manquante\n") elif user.api_key: api_key = user.api_key # 🔑 Utiliser l'API key, pas le JWT token # Envoyer le scan complet via l'API /api/v1/scan import requests from collections import Counter # Compter les vulnérabilités par sévérité severity_counts = dict(Counter(f.severity for f in result.findings)) # Préparer les données du scan scan_data = { "target": request.domain or "SAST Scan", "scan_type": "sast", "findings": [ { "check": f.check, "title": f.title, "severity": f.severity, "description": f.description, "remediation": f.remediation, "impact": f.impact, "evidence": f.evidence } for f in result.findings ], "duration": scan_duration } # Envoyer à l'API api_url = "https://web-sentinel.taaazzz-prog.fr/api/v1/scan" headers = {"X-API-Key": api_key, "Content-Type": "application/json"} try: response = requests.post(api_url, json=scan_data, headers=headers, timeout=10) if response.status_code == 201: LOGGER.info(f"✅ Scan authentifié envoyé avec succès") self._append_to_results(f"✅ Scan enregistré dans vos statistiques personnelles ({len(result.findings)} vulnérabilités)\n") else: response_text = response.text[:200] if response.text else "Pas de réponse" LOGGER.error(f"⚠️ Échec envoi scan: HTTP {response.status_code} - {response_text}") self._append_to_results(f"⚠️ Erreur lors de l'enregistrement du scan (HTTP {response.status_code})\n") self._append_to_results(f"💬 Réponse serveur: {response_text}\n") except Exception as e: LOGGER.exception(f"❌ Erreur réseau lors de l'envoi du scan") self._append_to_results(f"⚠️ Impossible de contacter le serveur: {e}\n") except Exception as exc: LOGGER.exception(f"❌ Erreur scan authentifié: {exc}") self._append_to_results(f"❌ Erreur lors de l'enregistrement du scan: {exc}\n") else: LOGGER.info(f"ℹ️ Scan non authentifié (utilisateur non connecté) - Stats personnelles non enregistrées") self._append_to_results(f"ℹ️ Scan non authentifié (utilisateur non connecté) - Stats personnelles non enregistrées\n") self.gui.root.after(0, lambda: self._handle_result(result)) except Exception as exc: LOGGER.exception("Error during SAST scan") self.gui.root.after(0, lambda: self._scan_failed(str(exc))) def _handle_result(self, result: ScanResult) -> None: gui = self.gui gui.sast_results.append(result) gui.history_entries.append(result) if len(gui.history_entries) > 50: gui.history_entries = gui.history_entries[-50:] if gui.history_var.get(): try: gui.history_store.record(result) except Exception as exc: LOGGER.warning(gui._text("history_sast_save_failed", error=exc)) summary = gui._format_result_summary(result, include_details=True) if gui._expected_sast_files is not None: summary = f"{summary}\n{gui._text('sast_files_analyzed_line', count=gui._expected_sast_files)}" # Ajouter les messages du scan authentifié if self._scan_messages: scan_info = "\n" + "="*60 + "\n� Enregistrement du scan\n" + "="*60 + "\n" scan_info += "".join(self._scan_messages) summary = summary + scan_info self._scan_messages.clear() gui.sast_results_text.config(state=tk.NORMAL) gui.sast_results_text.delete("1.0", tk.END) gui.sast_results_text.insert(tk.END, summary + "\n\n") gui.sast_results_text.config(state=tk.DISABLED) gui.sast_results_text.see(tk.END) self._render_stats(result) gui.refresh_history_tab() gui.sast_progress_var.set(100) gui.sast_scan_button.config(state=tk.NORMAL) gui._set_status("status_sast_done") gui.current_sast_thread = None gui._expected_sast_files = None gui.last_sast_result = None gui.last_sast_result = result def _render_stats(self, result: ScanResult) -> None: gui = self.gui frame = getattr(gui, "sast_stats_frame", None) if frame is not None: for widget in frame.winfo_children(): widget.destroy() counts = Counter(finding.severity for finding in result.findings) if frame is not None: if not counts: ttk.Label( frame, text=gui._text("summary_no_issue"), foreground="#059669", font=("Segoe UI", 10, 'bold'), ).pack(anchor=tk.W) else: # Frame horizontal pour les statistiques côte à côte stats_frame = ttk.Frame(frame) stats_frame.pack(anchor=tk.W, pady=5) color_map = { "critical": "#dc2626", "high": "#f97316", "medium": "#facc15", "low": "#0ea5e9", "info": "#64748b", } col = 0 for severity in ("critical", "high", "medium", "low", "info"): count = counts.get(severity, 0) if count: label = localization.translate_severity(severity) ttk.Label( stats_frame, text=f"{label}: {count}", foreground=color_map.get(severity, "#1f2937"), font=("Segoe UI", 10, 'bold'), ).grid(row=0, column=col, padx=10, sticky=tk.W) col += 1 languages = getattr(result.request, "source_languages", ()) or ("auto",) lang_set = sorted({lang for lang in languages if lang}) lang_text = ", ".join(lang_set) if lang_set else gui._text("language_auto") ttk.Label( frame, text=gui._text("sast_languages_label", languages=lang_text), foreground="#334155", font=("Segoe UI", 9, 'italic'), ).pack(anchor=tk.W, pady=(8, 0)) def _scan_failed(self, error_message: str) -> None: gui = self.gui gui._set_status("status_ready") gui.sast_progress_var.set(0) gui.sast_scan_button.config(state=tk.NORMAL) gui.current_sast_thread = None gui._expected_sast_files = None gui.sast_results_text.config(state=tk.NORMAL) gui.sast_results_text.insert( tk.END, f"❌ {gui._text('error_sast_failed', error=error_message)}\n" ) gui.sast_results_text.config(state=tk.DISABLED) gui.sast_results_text.see(tk.END) messagebox.showerror(gui._text("error_title"), gui._text("sast_scan_failed", error=error_message)) LOGGER.error("SAST scan failed: %s", error_message) # ------------------------------------------------------------------ # Exports # ------------------------------------------------------------------ def export_json(self) -> None: gui = self.gui if not gui.sast_results: messagebox.showinfo(gui._text("info_title"), gui._text("no_sast_results")) return path = filedialog.asksaveasfilename( defaultextension=".json", filetypes=[(gui._text("json_files"), "*.json"), (gui._text("all_files"), "*.*")], title=gui._text("dialog_export_sast_json"), ) if not path: return export_result = gui.sast_results[-1] context = gui.report_engine.build_context(export_result) json_payload = gui.report_engine.render_json(context) Path(path).write_text(json_payload, encoding="utf-8") messagebox.showinfo( gui._text("export_success_title"), gui._text("export_sast_json_message", filename=path), ) def export_html(self) -> None: gui = self.gui if not gui.sast_results: messagebox.showinfo(gui._text("info_title"), gui._text("no_sast_results")) return path = filedialog.asksaveasfilename( defaultextension=".html", filetypes=[(gui._text("html_files"), "*.html"), (gui._text("all_files"), "*.*")], title=gui._text("dialog_export_sast_html"), ) if not path: return export_result = gui.sast_results[-1] context = gui.report_engine.build_context(export_result) html_payload = gui.report_engine.render_html(context) Path(path).write_text(html_payload, encoding="utf-8") messagebox.showinfo( gui._text("export_success_title"), gui._text("export_sast_html_message", filename=path), ) def copy_results(self) -> None: gui = self.gui gui.root.clipboard_clear() gui.root.clipboard_append(gui.sast_results_text.get("1.0", tk.END)) messagebox.showinfo(gui._text("export_success_title"), gui._text("export_sast_copy_message")) def export_ai(self) -> None: gui = self.gui if not gui.sast_results: messagebox.showinfo(gui._text("info_title"), gui._text("no_sast_results")) return gui.export_ai_analysis(gui.sast_results[-1], mode="sast") def clear_results(self) -> None: gui = self.gui gui.sast_results_text.config(state=tk.NORMAL) gui.sast_results_text.delete("1.0", tk.END) gui.sast_results_text.config(state=tk.DISABLED) gui.sast_results.clear() gui.last_sast_result = None frame = getattr(gui, "sast_stats_frame", None) if frame is not None: for widget in frame.winfo_children(): widget.destroy() def _append_to_results(self, text: str) -> None: """Stocke les messages du scan authentifié pour affichage.""" self._scan_messages.append(text)