#!/usr/bin/env python3 # SPDX-FileCopyrightText: 2024-2026 Medulla, http://www.medulla-tech.io # SPDX-License-Identifier: GPL-3.0-or-later # auteur jfk # file : dev/client_tools/pulse_diagnostic_monolithic.py """ Medulla Diagnostic Bundle - Version Monolithique Programme unique qui collecte tous les diagnostiques du client Medulla Aucune dépendance externe requise - Inclut tout en un seul fichier """ import os import sys import platform import shutil import zipfile import tempfile import subprocess import socket import threading import hashlib import json import ast import py_compile import base64 import glob import importlib.util from datetime import datetime from pathlib import Path from typing import List, Dict, Tuple, Optional, Set # ============================================================================ # CONFIGURATION - VARIABLES CONFIGURABLES # ============================================================================ # Modifier ces variables pour adapter le programme à votre environnement # Autres ports à tester (optionnel) ADDITIONAL_PORTS = [80, 443, 8443] # HTTP, HTTPS, etc. # Timeout pour les tests de connexion (secondes) CONNECTION_TIMEOUT = 3 # ============================================================================ # PORTS MEDULLA - TESTS DE CONNECTIVITÉ # ============================================================================ # Ports TCP/UDP à tester vers MEDULLA et RELAY (du script PowerShell) PORTS_PULSE_TCP = [22, 2002, 5222, 22067, 8443, 111, 2049, 55415, 80, 443, 9990, 9999] PORTS_PULSE_UDP = [67, 69, 111, 2049] PORTS_RELAY_TCP = [22, 2002, 5222, 9990, 111, 2049, 80, 443] PORTS_RELAY_UDP = [67, 69, 111, 2049] PORT_DESCRIPTIONS = { 22: "SSH", 2002: "Medulla API", 5222: "XMPP Client", 22067: "Medulla", 8443: "HTTPS Medulla", 80: "HTTP", 443: "HTTPS", 9990: "Admin Console", 9999: "Admin Alt", 111: "RPC Portmapper", 2049: "NFS", 55415: "Medulla Specific", 67: "DHCP", 69: "TFTP" } # ============================================================================ # CONSTANTES INTERNES (Ne pas modifier) # ============================================================================ _DEFAULT_MEDULLA_SERVER_B64 = "amZrLm1lZHVsbGEtdGVjaC5pbw==" # jfk.medulla-tech.io _DEFAULT_MEDULLA_SERVER = base64.b64decode(_DEFAULT_MEDULLA_SERVER_B64).decode('utf-8') MEDULLA_PORT_XMPP = 5222 # Port XMPP standard # ============================================================================ class MedullaDiagnosticMonolithic: """ Collecteur de diagnostiques Medulla monolithique. Programme unique qui inclut : - Détection OS automatique - Exécution de commandes de diagnostique - Collecte de fichiers - Génération de rapports - Compression ZIP """ def __init__(self, output_dir: str = None, medulla_server: str = None, agent_dir: str = None): """ Initialise le collecteur. Args: output_dir: Répertoire de sortie pour le ZIP medulla_server: Serveur Medulla (optionnel, défaut: agentconf.ini ou demander à l'utilisateur) agent_dir: Chemin au répertoire de l'agent (optionnel, défaut: détection automatique) """ self.system_type = platform.system() self.hostname = platform.node() self.timestamp = datetime.now().strftime("%Y%m%d_%Hh%Mm%Ss") # Déterminer le répertoire de sortie if output_dir: self.output_dir = output_dir elif self.system_type == "Windows": # Sur Windows, utiliser C:\Program Files\Medulla\tmp par défaut (plus convivial) self.output_dir = "C:\\Program Files\\Medulla\\tmp" else: # Sur Linux/macOS, utiliser /tmp par défaut (nettoyé automatiquement) # Créer un sous-répertoire medulla_diagnostics pour mieux s'y retrouver self.output_dir = "/tmp/medulla_diagnostics" # Créer le répertoire de sortie s'il n'existe pas try: os.makedirs(self.output_dir, exist_ok=True) # Sur Linux/macOS, s'assurer que le répertoire est lisible par l'utilisateur if self.system_type != "Windows": os.chmod(self.output_dir, 0o755) # Permissions rwxr-xr-x except Exception as e: # Fallback sur le répertoire courant si création échoue print(f"⚠️ Impossible de créer {self.output_dir}: {e}") self.output_dir = os.getcwd() self.bundle_name = f"medulla_diagnostics_{self.hostname}_{self.timestamp}" self.temp_dir = None self.errors = [] self.collected_files = [] self.commands_output = [] self.agent_dir = agent_dir # Chemin fourni manuellement (optionnel) # Déterminer le serveur Medulla: paramètre CLI > agentconf.ini > DEMANDER À L'UTILISATEUR > défaut if medulla_server: # Paramètre -s fourni par l'utilisateur self.medulla_server = medulla_server else: # Essayer de lire depuis agentconf.ini self.medulla_server = self._get_medulla_server_from_config() # Si pas trouvé en config ET pas en paramètre, demander à l'utilisateur if not self.medulla_server: self.medulla_server = self._prompt_for_medulla_server() self._setup_paths_and_commands() def _setup_paths_and_commands(self): """Configure chemins et commandes selon l'OS.""" if self.system_type == "Windows": # Sur Windows: configs dans C:\Program Files\Medulla, agent dans Python site-packages self.pulse_base = "C:\\Program Files\\Medulla" self.file_paths = { "logs": [ os.path.join(self.pulse_base, "var\\log\\xmpp-agent.log"), os.path.join(self.pulse_base, "tmp\\install.log"), os.path.join(self.pulse_base, "var\\log\\*.log"), # Glob pour capturer tous les .log os.path.join(self.pulse_base, "var\\log\\*.json"), # Pour windows11_compatibility_report.json "C:\\Program Files\\Python3\\Lib\\site-packages\\pulse_xmpp_agent\\logs", ], "configs": [ os.path.join(self.pulse_base, "etc\\agentconf.ini"), os.path.join(self.pulse_base, "etc\\manage_scheduler.ini"), os.path.join(self.pulse_base, "etc\\cluster.ini"), os.path.join(self.pulse_base, "etc\\inventory.ini"), os.path.join(self.pulse_base, "etc\\syncthing\\config.xml"), "C:\\Program Files\\Medulla\\agentconf.ini", "C:\\Program Files\\Medulla\\manage_scheduler.ini", ], "agent_dir": "C:\\Program Files\\Python3\\Lib\\site-packages\\pulse_xmpp_agent", } self.diagnostic_commands = { "network": [ ("Network Configuration", "ipconfig /all"), ("Active Routes", "route print"), ("Established Connections", "netstat -an"), ("Process Network Info", "netstat -ab"), ("DNS Configuration", "ipconfig /displaydns"), ("Listening Ports", "netstat -aon | find \"LISTENING\""), ("IP Statistics", "netstat -s"), ("Network Adapters", "Get-NetAdapter | Select-Object Name,Status,Speed | ConvertTo-Csv"), ], "medulla_connectivity": [ (f"DNS Resolution - Medulla Server ({self.medulla_server})", f"powershell -Command \"Resolve-DnsName {self.medulla_server}\""), (f"Port Connectivity Test - {MEDULLA_PORT_XMPP}", f"powershell -Command \"Test-NetConnection {self.medulla_server} -Port {MEDULLA_PORT_XMPP}\""), (f"Route Trace to Medulla ({self.medulla_server})", f"tracert -h 15 {self.medulla_server}"), (f"Connections to Port {MEDULLA_PORT_XMPP}", f"netstat -ano | findstr :{MEDULLA_PORT_XMPP}"), (f"Established Connections to {MEDULLA_PORT_XMPP}", f"netstat -ano | findstr :{MEDULLA_PORT_XMPP} | findstr ESTABLISHED"), ], "system": [ ("System Info", "systeminfo"), ("Disk Space", "wmic logicaldisk get name,freespace,size"), ("Installed Services", "wmic service list brief"), ("Processes", "tasklist /v"), ("Windows Updates", "wmic qfe list brief"), ("EventLog Errors", "wevtutil qe System /q:*[System[(Level=1 or Level=2)]] /f:text /c:10"), ], "medulla_specific": [ ("Medulla Services", "wmic service list brief | find /i \"pulse\""), ("Medulla Processes", "tasklist | find /i \"pulse\""), ("Medulla Registry", "reg query HKLM\\SOFTWARE\\Pulse /s"), ("Medulla Firewall Rules", "netsh advfirewall firewall show rule name=all | find /i \"pulse\""), ], } elif self.system_type == "Darwin": # macOS self.pulse_base = os.path.expanduser("~/Library/Pulse") alt_pulse_base = "/opt/pulse" self.file_paths = { "logs": [ os.path.join(self.pulse_base, "var/log/xmpp-agent.log"), os.path.join(self.pulse_base, "tmp/install.log"), os.path.join(alt_pulse_base, "var/log/xmpp-agent.log"), ], "configs": [ os.path.join(self.pulse_base, "etc/agentconf.ini"), os.path.join(self.pulse_base, "etc/manage_scheduler.ini"), os.path.join(self.pulse_base, "etc/cluster.ini"), os.path.join(self.pulse_base, "etc/inventory.ini"), os.path.join(self.pulse_base, "etc/syncthing/config.xml"), ], "agent_dir": os.path.join(self.pulse_base, "lib"), } self.diagnostic_commands = { "network": [ ("Network Interfaces", "ifconfig"), ("Network Services", "networksetup -listallnetworkservices"), ("DNS Configuration", "scutil --dns"), ("Routing Table", "netstat -rn"), ("Established Connections", "netstat -an"), ("Process Network", "lsof -i -P -n"), ("WiFi Info", "airport -I"), ("Active Network Info", "networksetup -getinfo Wi-Fi"), ], "medulla_connectivity": [ (f"DNS Resolution - Medulla Server ({self.medulla_server})", f"nslookup {self.medulla_server}"), (f"Port Connectivity Test - {MEDULLA_PORT_XMPP}", f"nc -zv {self.medulla_server} {MEDULLA_PORT_XMPP} 2>&1"), (f"Route Trace to Medulla ({self.medulla_server})", f"traceroute -m 15 {self.medulla_server} 2>&1"), (f"Connections to Port {MEDULLA_PORT_XMPP}", f"netstat -an | grep :{MEDULLA_PORT_XMPP}"), (f"Established Connections to {MEDULLA_PORT_XMPP}", f"netstat -an | grep :{MEDULLA_PORT_XMPP} | grep ESTABLISHED"), ], "system": [ ("System Report", "system_profiler SPSoftwareDataType"), ("Disk Space", "df -h"), ("Memory Info", "vm_stat"), ("Processes", "ps aux"), ("Open Files", "lsof | wc -l"), ("System Uptime", "uptime"), ("Logged In Users", "who"), ], "medulla_specific": [ ("Medulla Processes", "ps aux | grep -i pulse"), ("Medulla LaunchDaemons", "ls -la ~/Library/LaunchDaemons/ | grep -i pulse"), ("Medulla Config Perms", "ls -la ~/Library/Pulse/etc/"), ], } else: # Linux self.pulse_base = "/var/lib/pulse" self.file_paths = { "logs": [ "/var/log/xmpp-agent.log", "/var/log/pulse/xmpp-agent-machine.log", "/var/log/pulse/xmpp-agent-relay.log", "/opt/pulse/var/log/xmpp-agent.log", os.path.join(self.pulse_base, "tmp/install.log"), "/var/log/pulse/*.log", # Glob pour capturer tous les .log ], "configs": [ "/etc/pulse-xmpp-agent/agentconf.ini", # Debian/Ubuntu Pulse-XMPP Agent (nouveau chemin) "/etc/pulse/agentconf.ini", "/etc/pulse/manage_scheduler.ini", "/etc/pulse/cluster.ini", "/etc/pulse/inventory.ini", "/etc/pulse/syncthing/config.xml", os.path.join(self.pulse_base, "etc/agentconf.ini"), ], "agent_dir": os.path.join(self.pulse_base, "lib"), } self.diagnostic_commands = { "network": [ ("Network Interfaces", "ip addr show"), ("Routing Table", "ip route show"), ("DNS Configuration", "cat /etc/resolv.conf"), ("Network Statistics", "netstat -an"), ("Connections", "netstat -np"), ("Open Ports", "ss -tlnp"), ("Listening Services", "ss -an | grep LISTEN"), ("Network Performance", "ethtool eth0 2>/dev/null || echo 'ethtool not available'"), ], "medulla_connectivity": [ (f"DNS Resolution - Medulla Server ({self.medulla_server})", f"nslookup {self.medulla_server} 2>&1 || dig {self.medulla_server}"), (f"Port Connectivity Test - {MEDULLA_PORT_XMPP}", f"nc -zv {self.medulla_server} {MEDULLA_PORT_XMPP} 2>&1 || timeout 3 bash -c '&1"), (f"Connections to Port {MEDULLA_PORT_XMPP}", f"netstat -an | grep :{MEDULLA_PORT_XMPP} || ss -an | grep :{MEDULLA_PORT_XMPP}"), (f"Established Connections to {MEDULLA_PORT_XMPP}", f"netstat -an | grep :{MEDULLA_PORT_XMPP} | grep ESTABLISHED || ss -an | grep :{MEDULLA_PORT_XMPP} | grep ESTAB"), ], "system": [ ("OS Release", "cat /etc/os-release"), ("Disk Space", "df -h"), ("Memory Info", "free -h"), ("CPU Info", "lscpu"), ("Processes", "ps aux"), ("System Uptime", "uptime"), ("Logged In Users", "who"), ("Kernel Info", "uname -a"), ("Load Average", "cat /proc/loadavg"), ], "medulla_specific": [ ("Medulla Service Status", "systemctl status pulse-agent 2>/dev/null || echo 'Medulla agent not found'"), ("Medulla Relay Service Status", "systemctl status pulse-xmpp-master 2>/dev/null || echo 'Medulla relay not found'"), ("Medulla Processes", "ps aux | grep -i pulse"), ("Medulla Log Files - Machine", "ls -lah /var/log/pulse/xmpp-agent-machine.log* 2>/dev/null || echo 'Machine logs not found'"), ("Medulla Log Files - Relay", "ls -lah /var/log/pulse/xmpp-agent-relay.log* 2>/dev/null || echo 'Relay logs not found'"), ("Medulla Config Permissions", "ls -la /etc/pulse/"), ("Medulla Package Info", "dpkg -l | grep -i pulse || rpm -qa | grep -i pulse || pacman -Q | grep -i pulse"), ], } def run(self) -> Tuple[bool, str]: """ Exécute la collecte complète. Returns: Tuple (succès, chemin_du_zip_ou_erreur) """ try: print(f"🔍 Collecte des diagnostiques Medulla (Monolithique)...") print(f" Système: {self.system_type}") print(f" Hostname: {self.hostname}") print(f" Serveur Medulla: {self.medulla_server}\n") # Vérifier les ressources disponibles self._check_resources() self.temp_dir = tempfile.mkdtemp(prefix="pulse_diag_") # Collecte principale self._collect_system_info() self._execute_diagnostics() self._verify_agent_integrity() self._collect_files() # Créer archive zip_path = self._create_zip() self._cleanup() print(f"\n✅ Diagnostiques complétés avec succès!") print(f" Archive: {os.path.basename(zip_path)}") print(f" Localisation: {zip_path}") # Message supplémentaire selon l'OS if self.system_type == "Windows": print(f"\n💡 Pour transférer l'archive:") print(f" scp {zip_path} :/local/path/") else: print(f"\n💡 L'archive est dans /tmp/medulla_diagnostics/ (nettoyage automatique)") print(f" Pour transférer: scp {zip_path} :/local/path/") return True, zip_path except Exception as e: self._cleanup() error_msg = f"❌ Erreur: {str(e)}" print(error_msg) return False, error_msg def _collect_system_info(self): """Collecte les informations système.""" info_file = os.path.join(self.temp_dir, "system_info.txt") with open(info_file, "w", encoding="utf-8") as f: f.write("=" * 70 + "\n") f.write("MEDULLA CLIENT DIAGNOSTIC INFORMATION\n") f.write("=" * 70 + "\n\n") f.write(f"Collection Date: {datetime.now().isoformat()}\n") f.write(f"Hostname: {self.hostname}\n") f.write(f"System: {self.system_type}\n") f.write(f"Platform: {platform.platform()}\n") f.write(f"Python Version: {sys.version}\n") f.write(f"Architecture: {platform.machine()}\n") f.write(f"Processor: {platform.processor()}\n\n") # Espace disque f.write("-" * 70 + "\n") f.write("DISK SPACE\n") f.write("-" * 70 + "\n") try: usage = shutil.disk_usage("/") f.write(f"Total: {self._format_size(usage.total)}\n") f.write(f"Used: {self._format_size(usage.used)}\n") f.write(f"Free: {self._format_size(usage.free)}\n") except Exception as e: f.write(f"Erreur: {e}\n") f.write(f"\nDIAGNOSTIC COMMANDS EXECUTED\n") f.write("-" * 70 + "\n") f.write(f"Network checks: {len(self.diagnostic_commands.get('network', []))}\n") f.write(f"Medulla connectivity: {len(self.diagnostic_commands.get('medulla_connectivity', []))}\n") f.write(f"System checks: {len(self.diagnostic_commands.get('system', []))}\n") f.write(f"Medulla specific: {len(self.diagnostic_commands.get('medulla_specific', []))}\n") total_commands = sum(len(cmds) for cmds in self.diagnostic_commands.values()) f.write(f"TOTAL COMMANDS: {total_commands}\n") if self.errors: f.write(f"\nERRORS ENCOUNTERED\n") f.write("-" * 70 + "\n") for error in self.errors: f.write(f"⚠ {error}\n") print(f"✓ Infos système collectées") def _execute_diagnostics(self): """Exécute les commandes de diagnostique et sauvegarde les résultats.""" output_file = os.path.join(self.temp_dir, "diagnostic_commands_output.txt") with open(output_file, "w", encoding="utf-8") as f: f.write("=" * 70 + "\n") f.write("DIAGNOSTIC COMMANDS OUTPUT\n") f.write("=" * 70 + "\n") f.write(f"System: {self.system_type}\n") f.write(f"Executed: {datetime.now().isoformat()}\n") f.write(f"Medulla Server: {self.medulla_server}\n\n") # Ajouter tests des ports TCP/UDP avant les commandes shell f.write(f"\n{'=' * 70}\n") f.write(f"CATEGORY: PORT CONNECTIVITY TESTS\n") f.write(f"TARGET: {self.medulla_server}\n") f.write(f"{'=' * 70}\n\n") # Tests TCP vers Medulla f.write(f"\n{'-' * 70}\n") f.write(f"TCP Port Tests to {self.medulla_server}\n") f.write(f"{'-' * 70}\n") tcp_results = self._test_tcp_ports(self.medulla_server, PORTS_PULSE_TCP) for port, status, msg in tcp_results: f.write(f"Port {port:5d} ({PORT_DESCRIPTIONS.get(port, 'Unknown'):20s}): {status:6s} - {msg}\n") # Tests UDP vers Medulla f.write(f"\n{'-' * 70}\n") f.write(f"UDP Port Tests to {self.medulla_server}\n") f.write(f"{'-' * 70}\n") udp_results = self._test_udp_ports(self.medulla_server, PORTS_PULSE_UDP) for port, status, msg in udp_results: f.write(f"Port {port:5d} ({PORT_DESCRIPTIONS.get(port, 'Unknown'):20s}): {status:6s} - {msg}\n") # Exécuter les autres commandes shell for category, commands in self.diagnostic_commands.items(): f.write(f"\n{'=' * 70}\n") f.write(f"CATEGORY: {category.upper()}\n") f.write(f"{'=' * 70}\n\n") for desc, cmd in commands: f.write(f"\n{'-' * 70}\n") f.write(f"Command: {desc}\n") f.write(f"Exec: {cmd}\n") f.write(f"{'-' * 70}\n") try: # Exécuter la commande # Utiliser encoding='utf-8' et errors='replace' pour gérer les problèmes d'encodage result = subprocess.run( cmd, shell=True, capture_output=True, text=True, encoding='utf-8', errors='replace', timeout=30 ) output = result.stdout if result.stdout else result.stderr f.write(f"{output}\n") print(f" ✓ {desc}") except subprocess.TimeoutExpired: f.write("⚠ Command timed out (30 seconds)\n") print(f" ⚠ {desc} (timeout)") except Exception as e: f.write(f"⚠ Error: {str(e)}\n") self.errors.append(f"Command '{desc}' failed: {str(e)}") print(f" ⚠ {desc} (error)") print(f"✓ Commandes diagnostiques exécutées") def _get_medulla_server_from_config(self) -> Optional[str]: """ Extrait le serveur Medulla depuis agentconf.ini de l'agent. Cherche dans différents chemins selon l'OS. Si agentconf.ini n'existe pas ou est incomplet, affiche que le configurateur ne travaille pas. Returns: Hostname/IP du serveur Medulla ou None """ config_paths = [] if self.system_type == "Windows": config_paths = [ "C:\\Program Files\\Medulla\\etc\\agentconf.ini", "C:\\Program Files\\Medulla\\agentconf.ini", "C:\\Program Files\\Pulse\\etc\\agentconf.ini", "C:\\Program Files (x86)\\Pulse\\etc\\agentconf.ini", ] elif self.system_type == "Darwin": config_paths = [ os.path.expanduser("~/Library/Pulse/etc/agentconf.ini"), "/opt/pulse/etc/agentconf.ini", "/opt/medulla/etc/agentconf.ini", ] else: # Linux - chercher dans les vrais chemins jfkbase config_paths = [ "/etc/pulse-xmpp-agent/agentconf.ini", # Debian/Ubuntu Pulse-XMPP Agent "/etc/pulse/agentconf.ini", "/var/lib/pulse/etc/agentconf.ini", "/opt/pulse/etc/agentconf.ini", "/opt/medulla/etc/agentconf.ini", "/opt/medulla/lib/python3.11/site-packages/pulse_xmpp_agent/agentconf.ini", "/opt/medulla/lib/python3.10/site-packages/pulse_xmpp_agent/agentconf.ini", "/opt/medulla/lib/python3.9/site-packages/pulse_xmpp_agent/agentconf.ini", ] for config_path in config_paths: if os.path.isfile(config_path): try: with open(config_path, "r", encoding="utf-8") as f: content = f.read() # Parser basique du fichier INI for line in content.split("\n"): line = line.strip() if not line or line.startswith("#") or line.startswith(";"): continue # Chercher la clé "confserver", "server", "host", "xmpp_server", "medulla_server" if "=" in line: key, value = line.split("=", 1) key = key.strip().lower() value = value.strip().strip('"').strip("'") if key in ["confserver", "server", "host", "xmpp_server", "medulla_server"]: if value: print(f"✓ Serveur Medulla détecté depuis agentconf.ini: {value}") return value except Exception as e: # Silencieusement ignorer les erreurs de parsing pass # Pas trouvé - retourner None (l'appelant demandera à l'utilisateur) return None def _prompt_for_medulla_server(self) -> str: """ Demande à l'utilisateur d'entrer le serveur Medulla si la config n'a pas pu être trouvée. Returns: Hostname/IP du serveur Medulla saisi par l'utilisateur ou valeur par défaut """ print("\n" + "=" * 70) print("⚠️ CONFIGURATION MEDULLA NON TROUVÉE") print("=" * 70) print(f"Le fichier agentconf.ini n'a pas pu être trouvé ou ne contient pas d'information serveur.") print(f"Veuillez entrer l'adresse IP ou le FQDN du serveur Medulla:") print() try: user_input = input(">>> ").strip() if user_input: print(f"\n✓ Utilisation du serveur Medulla: {user_input}") return user_input else: # Pas de saisie, utiliser la valeur par défaut print(f"\n⚠️ Pas de saisie - Utilisation de la valeur par défaut: {_DEFAULT_MEDULLA_SERVER}") return _DEFAULT_MEDULLA_SERVER except (EOFError, KeyboardInterrupt): # Entrée standard non disponible (script en background) ou Ctrl+C print(f"\n⚠️ Entrée utilisateur non disponible - Utilisation de la valeur par défaut: {_DEFAULT_MEDULLA_SERVER}") return _DEFAULT_MEDULLA_SERVER def _expand_log_files(self, log_patterns: List[str]) -> List[str]: """ Étend les patterns de fichiers logs (sans les fichiers rotatés). Seuls les fichiers de log courants sont collectés. Args: log_patterns: Liste des chemins/patterns de logs Returns: Liste des fichiers logs courants trouvés (sans .log.1, .log.2, etc.) """ import glob log_files = [] for pattern in log_patterns: if "*" in pattern: # Pattern glob : étendre matches = glob.glob(pattern) # Exclure les fichiers rotatés (.log.1, .log.2, etc.) matches = [m for m in matches if not m[-2:].lstrip(".").isdigit()] log_files.extend(matches) # Fallback : scanner le répertoire si glob ne trouve rien if not matches: dir_part = os.path.dirname(pattern) if os.path.isdir(dir_part): try: for fn in os.listdir(dir_part): full_path = os.path.join(dir_part, fn) # Exclure les rotatés (se terminent par .N) ext = fn.rsplit(".", 1)[-1] if "." in fn else "" if os.path.isfile(full_path) and not ext.isdigit(): log_files.append(full_path) except Exception: pass else: # Chemin spécifique : uniquement le fichier courant (pas de rotatés) if os.path.isfile(pattern): log_files.append(pattern) return log_files def _verify_agent_integrity(self): """ Vérifie l'intégrité des fichiers de l'agent (replicator). Calcule les hashs MD5 des fichiers clés et affiche un rapport. """ output_file = os.path.join(self.temp_dir, "agent_integrity_check.txt") with open(output_file, "w", encoding="utf-8") as f: f.write("=" * 70 + "\n") f.write("AGENT INTEGRITY CHECK (Replicator Verification)\n") f.write("=" * 70 + "\n\n") f.write(f"Checked: {datetime.now().isoformat()}\n") f.write(f"System: {self.system_type}\n\n") # Déterminer le répertoire de base de l'agent selon l'OS agent_base = None candidates = [] if self.system_type == "Windows": candidates = [ "C:\\Program Files\\Python3\\Lib\\site-packages\\pulse_xmpp_agent", "C:\\Program Files\\Python3\\Lib\\site-packages\\pulse_xmpp_agent\\lib", "C:\\Program Files\\Python\\Lib\\site-packages\\pulse_xmpp_agent", "C:\\Program Files\\Python\\Lib\\site-packages\\pulse_xmpp_agent\\lib", "C:\\Program Files (x86)\\Python3\\Lib\\site-packages\\pulse_xmpp_agent", "C:\\Program Files (x86)\\Python3\\Lib\\site-packages\\pulse_xmpp_agent\\lib", "C:\\Program Files\\Pulse", "C:\\Program Files (x86)\\Pulse" ] elif self.system_type == "Darwin": candidates = [ os.path.expanduser("~/Library/Pulse"), "/opt/pulse", "/opt/medulla" ] elif self.system_type == "Linux": candidates = [ # Agent machine - venv /opt/medulla (installation standard Debian) "/opt/medulla/lib/python3.11/site-packages/pulse_xmpp_agent", "/opt/medulla/lib/python3.12/site-packages/pulse_xmpp_agent", "/opt/medulla/lib/python3.10/site-packages/pulse_xmpp_agent", "/opt/medulla/lib/python3.9/site-packages/pulse_xmpp_agent", # Agent relay - dist-packages système "/usr/lib/python3/dist-packages/pulse_xmpp_agent", "/usr/lib/python3.11/dist-packages/pulse_xmpp_agent", "/usr/lib/python3.10/dist-packages/pulse_xmpp_agent", # Anciens chemins / chemins alternatifs "/var/lib/pulse/lib", "/opt/pulse/lib", ] # Si chemin fourni manuellement via -a, utiliser celui-ci en priorité if self.agent_dir: if os.path.isdir(self.agent_dir): agent_base = self.agent_dir f.write(f"Agent Base Directory: {agent_base} (fourni via paramètre -a)\n") f.write(f"Directory Exists: True\n\n") else: f.write(f"❌ Le répertoire fourni via -a n'existe pas: {self.agent_dir}\n") f.write(f"Directory Exists: False\n\n") print(f"⚠ Agent integrity check: répertoire agent fourni inexistant: {self.agent_dir}") return else: # Chercher dans les candidats for candidate in candidates: if os.path.isdir(candidate): agent_base = candidate break # Si pas trouvé avec les chemins standards, chercher recursively if not agent_base: try: if self.system_type == "Windows": # Recherche récursive dans les répertoires Python standard Windows search_roots = [ "C:\\Program Files\\Python3\\Lib\\site-packages", "C:\\Program Files\\Python\\Lib\\site-packages", "C:\\Program Files (x86)\\Python3\\Lib\\site-packages", ] for search_root in search_roots: if not os.path.isdir(search_root): continue for root, dirs, files in os.walk(search_root): if "replicator.py" in files and "pulse_xmpp_agent" in root: agent_base = root f.write(f"✓ Found agent directory via recursive search: {agent_base}\n\n") break if agent_base: break else: # Recherche récursive Linux/macOS for root, dirs, files in os.walk("/opt"): if "replicator.py" in files and ("pulse_xmpp_agent" in root or "rollback_pulse_xmpp_agent" in root): agent_base = root f.write(f"✓ Found agent directory via recursive search: {agent_base}\n\n") break except: pass # Sauvegarder le chemin trouvé pour réutilisation dans _collect_files if agent_base: self._found_agent_base = agent_base f.write(f"Agent Base Directory: {agent_base or candidates[0] if candidates else 'unknown'}\n") f.write(f"Directory Exists: {os.path.isdir(agent_base) if agent_base else False}\n\n") if not agent_base or not os.path.isdir(agent_base): f.write("❌ Agent base directory not found\n\n") f.write("SEARCHED LOCATIONS:\n") f.write("-" * 70 + "\n") for i, candidate in enumerate(candidates, 1): f.write(f" {i}. {candidate}\n") f.write("\nRECURSIVE SEARCH: /opt/ (looking for replicator.py)\n\n") f.write("ACTION REQUIRED:\n") f.write("-" * 70 + "\n") f.write("L'agent Medulla n'a pas pu être localisé automatiquement.\n") f.write(f"Réessayez avec le paramètre -a pour spécifier le chemin:\n\n") f.write(f" python3 pulse_diagnostic_monolithic.py -a \"/path/to/agent\"\n\n") f.write("Exemple:\n") f.write(f" python3 pulse_diagnostic_monolithic.py -a \"/opt/medulla/lib/python3.11/site-packages/pulse_xmpp_agent\"\n") print(f"⚠ Agent integrity check: répertoire agent non trouvé") print(f" Indices: cherche un répertoire contenant agentxmpp.py, launcher.py, replicator.py") print(f" Relancez avec: python3 pulse_diagnostic_monolithic.py -a \"\"") return # Fichiers clés à vérifier (du replicator) files_to_check = { "program_agent": [ "agentxmpp.py", "launcher.py", "connectionagent.py", "replicator.py" ], "lib_agent": [], "script_agent": [], "plugins_common": [], "pluginsmachine": [], "pluginsrelay": [], "pluginsmaster": [], } # Scanner les répertoires lib et script (si ils existent) # Sur Windows: pulse_xmpp_agent peut être le répertoire racine OU lib peut être un sous-répertoire # Chercher lib dans l'agent_base OU dans le parent si agent_base se termine par 'lib' if os.path.basename(agent_base).lower() == "lib": # agent_base EST le répertoire lib - remonter au parent parent_base = os.path.dirname(agent_base) lib_dir = agent_base agent_base = parent_base f.write(f"Note: agent_base remonte au parent: {agent_base}\n") else: lib_dir = os.path.join(agent_base, "lib") script_dir = os.path.join(agent_base, "script") if os.path.isdir(lib_dir): try: lib_files = sorted([f for f in os.listdir(lib_dir) if f.endswith(".py")]) files_to_check["lib_agent"] = lib_files f.write(f"Lib directory: {lib_dir} ({len(lib_files)} fichiers .py)\n") except: pass else: f.write(f"Lib directory: {lib_dir} (non trouvé)\n") if os.path.isdir(script_dir): try: files_to_check["script_agent"] = [ f for f in os.listdir(script_dir) if f.endswith((".ps1", ".py")) ] except: pass # Scanner les répertoires plugins for plugin_dir_name in ["plugins_common", "pluginsmachine", "pluginsrelay", "pluginsmaster"]: plugin_dir = os.path.join(agent_base, plugin_dir_name) if os.path.isdir(plugin_dir): try: plugin_files = sorted([fn for fn in os.listdir(plugin_dir) if fn.endswith(".py")]) files_to_check[plugin_dir_name] = plugin_files f.write(f"{plugin_dir_name}: {plugin_dir} ({len(plugin_files)} fichiers .py)\n") except: pass else: f.write(f"{plugin_dir_name}: {plugin_dir} (non trouvé)\n") # Si on a pas trouvé de lib/script, chercher les .py directement dans agent_base if not files_to_check["lib_agent"] and not files_to_check["script_agent"]: try: py_files = [f for f in os.listdir(agent_base) if f.endswith(".py")] if py_files: files_to_check["program_agent"].extend( [f for f in py_files if f not in files_to_check["program_agent"]] ) except: pass # Vérifier version (chercher agentversion ou VERSION) version_file = None for vfile in ["agentversion", "VERSION", "version.txt"]: candidate = os.path.join(agent_base, vfile) if os.path.isfile(candidate): version_file = candidate break if version_file: try: with open(version_file, "r", encoding="utf-8") as vf: version = vf.read().strip() f.write(f"Agent Version: {version}\n\n") except Exception as e: f.write(f"⚠ Error reading version: {e}\n\n") else: f.write(f"Agent Version: Not found\n\n") # Calculer et afficher les hashs f.write("-" * 70 + "\n") f.write("FILE INTEGRITY (MD5 Hashes)\n") f.write("-" * 70 + "\n\n") all_ok = True total_files = 0 checked_files = 0 python_errors = [] missing_imports = [] import_search_paths = [ agent_base, os.path.join(agent_base, "lib"), os.path.join(agent_base, "script"), os.path.join(agent_base, "plugins_common"), os.path.join(agent_base, "pluginsmachine"), os.path.join(agent_base, "pluginsrelay"), os.path.join(agent_base, "pluginsmaster"), ] for category, files in files_to_check.items(): if not files: continue f.write(f"\n{category.upper()}:\n") for filename in files: total_files += 1 if category == "program_agent": filepath = os.path.join(agent_base, filename) elif category == "lib_agent": filepath = os.path.join(agent_base, "lib", filename) elif category in ("plugins_common", "pluginsmachine", "pluginsrelay", "pluginsmaster"): filepath = os.path.join(agent_base, category, filename) else: # script_agent filepath = os.path.join(agent_base, "script", filename) if os.path.isfile(filepath): try: md5_hash = self._calculate_md5(filepath) f.write(f" ✓ {filename:<40} {md5_hash}") # Vérifier les erreurs Python si c'est un .py if filename.endswith(".py"): error = self._check_python_syntax(filepath) if error: f.write(f" [SYNTAX ERROR]") python_errors.append((filename, error)) all_ok = False else: file_missing_imports = self._check_missing_imports( filepath, import_search_paths, ) if file_missing_imports: missing_imports.extend( [(filename, module) for module in file_missing_imports] ) f.write( f" [MISSING IMPORTS: {len(file_missing_imports)}]" ) all_ok = False else: f.write(f" [OK]") f.write(f"\n") checked_files += 1 except Exception as e: f.write(f" ⚠ {filename:<40} ERROR: {e}\n") all_ok = False else: f.write(f" ✗ {filename:<40} FILE NOT FOUND\n") all_ok = False # Détails des erreurs Python if python_errors: f.write(f"\n{'-' * 70}\n") f.write(f"PYTHON SYNTAX ERRORS FOUND\n") f.write(f"{'-' * 70}\n\n") for filename, error in python_errors: f.write(f"File: {filename}\n") f.write(f"Error: {error}\n") f.write(f"\n") # Détails des imports manquants if missing_imports: f.write(f"\n{'-' * 70}\n") f.write(f"PYTHON MISSING IMPORTS FOUND\n") f.write(f"{'-' * 70}\n\n") per_file_missing = {} for filename, module in missing_imports: per_file_missing.setdefault(filename, set()).add(module) for filename in sorted(per_file_missing.keys()): modules = sorted(per_file_missing[filename]) f.write(f"File: {filename}\n") f.write(f"Missing modules: {', '.join(modules)}\n\n") # Vérification du TYPE des plugins selon agent_type plugin_type_errors = self._check_plugin_types(agent_base, f) # Résumé f.write(f"\n{'-' * 70}\n") f.write(f"SUMMARY\n") f.write(f"{'-' * 70}\n") f.write(f"Total files expected: {total_files}\n") f.write(f"Files found: {checked_files}\n") f.write(f"Python syntax errors: {len(python_errors)}\n") f.write(f"Python missing imports: {len(missing_imports)}\n") f.write(f"Plugin type errors: {plugin_type_errors}\n") f.write(f"Status: {'✓ OK' if all_ok and checked_files == total_files and plugin_type_errors == 0 else '⚠ INCOMPLETE/ERROR'}\n") # Avertissements spécifiques if checked_files < total_files: f.write(f"\n⚠ WARNING: {total_files - checked_files} file(s) missing or unreadable\n") f.write(f"This may indicate agent corruption or installation issues.\n") if python_errors: f.write(f"\n⚠ CRITICAL: {len(python_errors)} Python file(s) have syntax errors!\n") f.write(f"The agent may not execute properly. Contact support.\n") if missing_imports: f.write(f"\n⚠ CRITICAL: {len(missing_imports)} missing Python import(s) detected!\n") f.write( f"Some plugins/libs may fail at runtime with ImportError/ModuleNotFoundError.\n" ) if plugin_type_errors > 0: f.write(f"\n⚠ WARNING: {plugin_type_errors} plugin(s) ont un TYPE incorrect!\n") f.write(f"Des plugins avec un mauvais TYPE peuvent être ignorés ou causer des erreurs.\n") print(f"✓ Vérification intégrité agent effectuée") def _check_plugin_types(self, agent_base: str, f) -> int: """ Vérifie que le TYPE des plugins est cohérent avec agent_type dans agentconf.ini. Règles métier : - plugins_common/ : TYPE doit être "all" (toute autre valeur est une erreur) - pluginsmachine/ : TYPE doit être "all" ou "machine" - pluginsrelay/ : TYPE doit être "all" ou "relay" - pluginsmaster/ : TYPE doit être "all" ou "master" Le TYPE est lu dans la variable plugin = {...} de chaque fichier .py. Args: agent_base: Répertoire racine de l'agent f: Fichier de sortie (déjà ouvert en écriture) Returns: Nombre d'erreurs de TYPE trouvées """ # Lire agent_type depuis agentconf.ini agent_type = None config_paths = [ "/etc/pulse-xmpp-agent/agentconf.ini", "/etc/pulse/agentconf.ini", "C:\\Program Files\\Medulla\\etc\\agentconf.ini", ] for config_path in config_paths: if os.path.isfile(config_path): try: in_type_section = False with open(config_path, "r", encoding="utf-8", errors="replace") as cf: for line in cf: line = line.strip() if line.startswith("["): in_type_section = (line.lower() == "[type]") elif in_type_section and line.startswith("agent_type"): parts = line.split("=", 1) if len(parts) == 2: agent_type = parts[1].strip().lower() break if agent_type: break except Exception: pass f.write(f"\n{'-' * 70}\n") f.write(f"PLUGIN TYPE VALIDATION\n") f.write(f"{'-' * 70}\n\n") f.write(f"Agent type (agentconf.ini): {agent_type or 'non trouvé (défaut: machine)'}\n\n") if not agent_type: agent_type = "machine" # valeur par défaut # Règles: pour chaque répertoire, types autorisés # Sur Windows: plugins_common n'existe jamais - son absence n'est PAS une erreur plugin_rules = { "plugins_common": ["all"], "pluginsmachine": ["all", "machine"], "pluginsrelay": ["all", "relay"], "pluginsmaster": ["all", "master"], } # Répertoires dont l'absence est normale sur Windows # - plugins_common : n'existe jamais sur Windows # - pluginsrelay : Windows ne peut pas être agent relay # - pluginsmaster : Windows ne peut pas être master substitute windows_optional_dirs = {"plugins_common", "pluginsrelay", "pluginsmaster"} total_errors = 0 for plugin_dir_name, allowed_types in plugin_rules.items(): plugin_dir = os.path.join(agent_base, plugin_dir_name) if not os.path.isdir(plugin_dir): if self.system_type == "Windows" and plugin_dir_name in windows_optional_dirs: f.write(f"{plugin_dir_name}/: absent (normal sur Windows)\n\n") else: f.write(f"{plugin_dir_name}/: absent (répertoire non trouvé)\n\n") continue f.write(f"{plugin_dir_name}/ (types autorisés: {allowed_types}):\n") dir_errors = 0 try: py_files = sorted([fn for fn in os.listdir(plugin_dir) if fn.endswith(".py")]) except Exception: f.write(f" ⚠ Impossible de lire le répertoire\n\n") continue for filename in py_files: filepath = os.path.join(plugin_dir, filename) plugin_type = None try: with open(filepath, "r", encoding="utf-8", errors="replace") as pf: for line in pf: line = line.strip() # Chercher: plugin = {...} if line.startswith("plugin") and "=" in line and "{" in line: try: dict_str = line.split("=", 1)[1].strip() plugin_info = ast.literal_eval(dict_str) if isinstance(plugin_info, dict): plugin_type = plugin_info.get("TYPE", "").lower() break except Exception: pass except Exception: f.write(f" ⚠ {filename:<40} ERREUR LECTURE\n") continue if plugin_type is None: f.write(f" ? {filename:<40} TYPE non trouvé (plugin sans descripteur)\n") elif plugin_type in allowed_types: f.write(f" ✓ {filename:<40} TYPE={plugin_type}\n") else: f.write(f" ✗ {filename:<40} TYPE={plugin_type} [ERREUR - attendu: {allowed_types}]\n") dir_errors += 1 total_errors += 1 if dir_errors == 0: f.write(f" → Tous les plugins OK\n") else: f.write(f" → {dir_errors} plugin(s) avec TYPE incorrect\n") f.write("\n") return total_errors def _collect_files(self): """Collecte les fichiers de configuration et logs.""" import glob # Créer structure répertoires logs_dir = os.path.join(self.temp_dir, "logs") configs_dir = os.path.join(self.temp_dir, "configs") agent_dir = os.path.join(self.temp_dir, "agent_files") os.makedirs(logs_dir, exist_ok=True) os.makedirs(configs_dir, exist_ok=True) os.makedirs(agent_dir, exist_ok=True) # Copier logs - utiliser une fonction dédiée pour capturer les fichiers rotatés log_files_to_copy = self._expand_log_files(self.file_paths.get("logs", [])) # Dédupliquer et copier for log_path in set(log_files_to_copy): try: target = os.path.join(logs_dir, os.path.basename(log_path)) # Si un fichier avec le même nom existe déjà, ajouter un suffixe if os.path.exists(target): name, ext = os.path.splitext(os.path.basename(log_path)) target = os.path.join(logs_dir, f"{name}_{os.path.basename(os.path.dirname(log_path))}{ext}") shutil.copy2(log_path, target) self.collected_files.append(log_path) print(f" ✓ Log: {os.path.basename(log_path)}") except Exception as e: self.errors.append(f"Error copying log {log_path}: {e}") # Copier configs for config_path in self.file_paths.get("configs", []): if os.path.isfile(config_path): try: target = os.path.join(configs_dir, os.path.basename(config_path)) shutil.copy2(config_path, target) self.collected_files.append(config_path) print(f" ✓ Config: {os.path.basename(config_path)}") except Exception as e: self.errors.append(f"Error copying config {config_path}: {e}") # Créer un fichier avec le contenu réel des configurations # Cherche TOUS les fichiers .ini dans les répertoires de config (agent, relay, etc.) config_content_file = os.path.join(self.temp_dir, "configuration_files_content.txt") with open(config_content_file, "w", encoding="utf-8") as f: f.write("=" * 70 + "\n") f.write("CONFIGURATION FILES CONTENT (All .ini files)\n") f.write("=" * 70 + "\n\n") config_found = False # Chercher récursivement tous les .ini dans les répertoires de config Linux if self.system_type == "Linux": search_dirs = [ "/etc/pulse-xmpp-agent", "/etc/pulse-xmpp-master", "/etc/pulse", "/var/lib/pulse/etc", "/opt/pulse/etc", "/opt/medulla/etc", ] for search_dir in search_dirs: if os.path.isdir(search_dir): try: for root, dirs, files in os.walk(search_dir): for file in sorted(files): if file.endswith(".ini"): config_path = os.path.join(root, file) config_found = True f.write(f"\n{'=' * 70}\n") f.write(f"File: {config_path}\n") f.write(f"{'=' * 70}\n\n") try: with open(config_path, "r", encoding="utf-8", errors="replace") as cf: content = cf.read() f.write(content) if not content.endswith("\n"): f.write("\n") except Exception as e: f.write(f"ERROR: Cannot read {config_path}: {e}\n") except Exception as e: pass # Ignorer les erreurs d'accès # Fallback: chercher aussi dans les chemins spécifiques (Windows, macOS, etc.) if not config_found: for config_path in self.file_paths.get("configs", []): if os.path.isfile(config_path): config_found = True f.write(f"\n{'=' * 70}\n") f.write(f"File: {config_path}\n") f.write(f"{'=' * 70}\n\n") try: with open(config_path, "r", encoding="utf-8", errors="replace") as cf: content = cf.read() f.write(content) if not content.endswith("\n"): f.write("\n") except Exception as e: f.write(f"ERROR: Cannot read {config_path}: {e}\n") if not config_found: f.write("No configuration files (.ini) found.\n") # Créer arbre des fichiers agent # Utiliser le chemin détecté par _verify_agent_integrity en priorité agent_base = getattr(self, "_found_agent_base", None) or self.file_paths.get("agent_dir") if agent_base and os.path.isdir(agent_base): try: inventory = os.path.join(self.temp_dir, "agent_files_list.txt") with open(inventory, "w", encoding="utf-8") as f: f.write("=" * 70 + "\n") f.write("AGENT FILE TREE\n") f.write("=" * 70 + "\n") f.write(f"Répertoire: {agent_base}\n\n") # Construire l'arbre récursif def write_tree(directory, prefix="", f=f): try: entries = sorted(os.listdir(directory)) except Exception: return dirs = [e for e in entries if os.path.isdir(os.path.join(directory, e))] files = [e for e in entries if os.path.isfile(os.path.join(directory, e))] # Fichiers en premier for i, fn in enumerate(files): fp = os.path.join(directory, fn) is_last_file = (i == len(files) - 1) and not dirs connector = "└── " if is_last_file else "├── " try: size = self._format_size(os.path.getsize(fp)) mtime = datetime.fromtimestamp(os.path.getmtime(fp)).strftime("%Y-%m-%d %H:%M") except Exception: size, mtime = "?", "?" f.write(f"{prefix}{connector}{fn:<45} {size:>10} {mtime}\n") # Sous-répertoires ensuite for i, dn in enumerate(dirs): dp = os.path.join(directory, dn) is_last = (i == len(dirs) - 1) connector = "└── " if is_last else "├── " f.write(f"{prefix}{connector}{dn}/\n") extension = " " if is_last else "│ " write_tree(dp, prefix + extension, f) write_tree(agent_base) # Sur Windows: ajouter aussi l'arbre de C:\Program Files\Medulla if self.system_type == "Windows": medulla_base = "C:\\Program Files\\Medulla" f.write(f"\n{'=' * 70}\n") f.write(f"MEDULLA INSTALLATION TREE\n") f.write(f"{'=' * 70}\n") if os.path.isdir(medulla_base): f.write(f"Répertoire: {medulla_base}\n\n") write_tree(medulla_base) else: f.write(f"✗ {medulla_base} (non trouvé)\n") # Sur Linux: ajouter /tmp/inventory.txt s'il existe if self.system_type != "Windows": inventory_path = "/tmp/inventory.txt" f.write(f"\n{'=' * 70}\n") f.write(f"INVENTORY FILE\n") f.write(f"{'=' * 70}\n") if os.path.isfile(inventory_path): try: size = self._format_size(os.path.getsize(inventory_path)) mtime = datetime.fromtimestamp(os.path.getmtime(inventory_path)).strftime("%Y-%m-%d %H:%M") f.write(f"✓ {inventory_path} {size} {mtime}\n") # Copier le fichier dans l'archive shutil.copy2(inventory_path, os.path.join(self.temp_dir, "inventory.txt")) print(f" ✓ inventory.txt collecté depuis /tmp/") except Exception as e: f.write(f"⚠ {inventory_path} ERREUR: {e}\n") else: f.write(f"✗ {inventory_path} (non trouvé)\n") print(f"✓ Arbre fichiers agent créé") except Exception as e: self.errors.append(f"Error creating agent file tree: {e}") def _create_zip(self) -> str: """ Crée l'archive ZIP avec compression maximale. Avant de créer la nouvelle archive, supprime les anciennes archives pulse_diagnostics_*.zip. Returns: Chemin de l'archive """ # Nettoyer les anciennes archives self._cleanup_old_archives() zip_path = os.path.join( self.output_dir, f"{self.bundle_name}.zip" ) print(f"\n📦 Création archive ZIP...") print(f" Destination: {zip_path}") with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED, compresslevel=9) as zf: for root, dirs, files in os.walk(self.temp_dir): for file in files: file_path = os.path.join(root, file) arcname = os.path.relpath(file_path, self.temp_dir) arcname = os.path.join(self.bundle_name, arcname) zf.write(file_path, arcname) zip_size = os.path.getsize(zip_path) # S'assurer que l'archive est lisible par l'utilisateur try: os.chmod(zip_path, 0o644) # Permissions rw-r--r-- except Exception: pass # Ignorer les erreurs de chmod (Windows notamment) print(f" Taille: {self._format_size(zip_size)}") print(f" ✓ Archive prête: {zip_path}") return zip_path def _cleanup_old_archives(self): """ Supprime les anciennes archives pulse_diagnostics_*.zip du répertoire output. Utile pour ne pas accumuler les archives. """ try: pattern = os.path.join(self.output_dir, "pulse_diagnostics_*.zip") old_archives = glob.glob(pattern) for archive_path in old_archives: try: os.remove(archive_path) print(f" ✓ Archive supprimée: {os.path.basename(archive_path)}") except Exception as e: # Ignorer les erreurs de suppression (fichier peut être en use) pass except Exception as e: # Ignorer les erreurs de glob pass def _cleanup(self): """Nettoie les fichiers temporaires.""" if self.temp_dir and os.path.isdir(self.temp_dir): shutil.rmtree(self.temp_dir) def _check_resources(self): """ Vérifie les ressources disponibles (mémoire, disque). Affiche les avertissements mais continue quand même (robustesse). """ try: # Vérifier disque disk_usage = shutil.disk_usage(self.output_dir) disk_free = disk_usage.free disk_total = disk_usage.total # Calculer le pourcentage utilisé disk_used_pct = (disk_usage.used / disk_total * 100) if disk_total > 0 else 0 print(f"📊 Ressources Disponibles:") print(f" 💾 Disque: {disk_used_pct:.1f}% utilisé ({self._format_size(disk_free)} libre)") # Avertissement si disque faible (< 100 MB libre) if disk_free < 100 * 1024 * 1024: print(f" ⚠️ ATTENTION: Espace disque faible ({self._format_size(disk_free)})") # Vérifier mémoire (Linux/macOS) if self.system_type in ["Linux", "Darwin"]: try: result = subprocess.run( "free -b | grep Mem" if self.system_type == "Linux" else "vm_stat | grep 'Pages free'", shell=True, capture_output=True, text=True, timeout=5 ) if result.stdout: # Essayer de parser la mémoire if self.system_type == "Linux": parts = result.stdout.split() if len(parts) >= 7: mem_total = int(parts[1]) mem_used = int(parts[2]) mem_free = int(parts[3]) mem_pct = (mem_used / mem_total * 100) if mem_total > 0 else 0 print(f" 🧠 Mémoire: {mem_pct:.1f}% utilisée ({self._format_size(mem_free)} libre)") # Avertissement si mémoire faible (< 10%) if mem_free < mem_total * 0.1: print(f" ⚠️ ATTENTION: Mémoire faible ({self._format_size(mem_free)} libre)") except Exception as e: pass # Silencieusement ignorer les erreurs print() except Exception as e: # Ne pas bloquer si vérification des ressources échoue print(f" ⚠️ Impossible de vérifier ressources: {e}\n") pass def _test_tcp_ports(self, host: str, ports: List[int]) -> List[Tuple[int, str, str]]: """ Teste les ports TCP vers un host. Args: host: Hostname ou IP ports: Liste des ports à tester Returns: Liste de tuples (port, statut, message) """ results = [] for port in ports: status = "CLOSED" msg = "" try: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(CONNECTION_TIMEOUT) result = sock.connect_ex((host, port)) sock.close() if result == 0: status = "OPEN" msg = "OK - Connection succeeded" else: status = "CLOSED" msg = f"Connection refused (errno {result})" except socket.timeout: status = "TIMEOUT" msg = f"Timeout after {CONNECTION_TIMEOUT}s" except socket.gaierror as e: status = "ERROR" msg = f"DNS resolution failed: {str(e)}" except Exception as e: status = "ERROR" msg = f"Error: {str(e)}" results.append((port, status, msg)) print(f" ✓ TCP Port {port}: {status}") return results def _test_udp_ports(self, host: str, ports: List[int]) -> List[Tuple[int, str, str]]: """ Teste les ports UDP vers un host (envoi simple paquet). Args: host: Hostname ou IP ports: Liste des ports à tester Returns: Liste de tuples (port, statut, message) """ results = [] for port in ports: status = "UNKNOWN" msg = "" try: sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.settimeout(CONNECTION_TIMEOUT) # Envoyer un paquet UDP vide sock.sendto(b"", (host, port)) status = "SENT" msg = "Packet sent (ICMP unreachable = CLOSED, no response = OPEN|FILTERED)" sock.close() except socket.timeout: status = "TIMEOUT" msg = "No response (likely OPEN|FILTERED)" except socket.gaierror as e: status = "ERROR" msg = f"DNS resolution failed: {str(e)}" except Exception as e: status = "ERROR" msg = f"Error: {str(e)}" results.append((port, status, msg)) print(f" ✓ UDP Port {port}: {status}") return results def _calculate_md5(self, filepath: str) -> str: """Calcule le hash MD5 d'un fichier.""" try: md5 = hashlib.md5() with open(filepath, "rb") as f: for chunk in iter(lambda: f.read(8192), b""): md5.update(chunk) return md5.hexdigest() except Exception as e: return f"ERROR: {str(e)}" def _check_python_syntax(self, filepath: str) -> Optional[str]: """Vérifie la syntaxe Python d'un fichier. Retourne un message d'erreur ou None.""" if not filepath.endswith(".py"): return None try: with open(filepath, "r", encoding="utf-8", errors="ignore") as f: code = f.read() ast.parse(code) return None # Pas d'erreur except (SyntaxError, IndentationError) as e: return str(e) except Exception as e: return None # Autres erreurs ignorées def _extract_imported_modules(self, filepath: str) -> Set[str]: """Extrait les modules importés (niveau top-level) depuis un fichier Python.""" modules = set() if not filepath.endswith(".py"): return modules try: with open(filepath, "r", encoding="utf-8", errors="ignore") as f: tree = ast.parse(f.read(), filename=filepath) except Exception: return modules for node in ast.walk(tree): if isinstance(node, ast.Import): for alias in node.names: if alias.name: modules.add(alias.name.split(".")[0]) elif isinstance(node, ast.ImportFrom): # Ignorer les imports relatifs: from .x import y if getattr(node, "level", 0): continue if node.module: modules.add(node.module.split(".")[0]) return modules def _check_missing_imports(self, filepath: str, extra_paths: List[str]) -> List[str]: """Retourne la liste des modules importés mais introuvables dans l'environnement Python.""" imported_modules = self._extract_imported_modules(filepath) if not imported_modules: return [] missing = [] original_sys_path = list(sys.path) try: # Ajouter les chemins de l'agent pour résoudre les imports locaux prepend_paths = [p for p in extra_paths if p and os.path.isdir(p)] sys.path = prepend_paths + sys.path for module_name in sorted(imported_modules): if module_name in ("__future__",): continue if module_name in sys.builtin_module_names: continue try: spec = importlib.util.find_spec(module_name) if spec is None: missing.append(module_name) except Exception: missing.append(module_name) finally: sys.path = original_sys_path return missing @staticmethod def _format_size(size_bytes: int) -> str: """Formate la taille en bytes.""" for unit in ["B", "KB", "MB", "GB"]: if size_bytes < 1024.0: return f"{size_bytes:.1f} {unit}" size_bytes /= 1024.0 return f"{size_bytes:.1f} TB" def main(): """Fonction principale.""" import argparse parser = argparse.ArgumentParser( description="Collecte diagnostiques Medulla (Version Monolithique)", epilog=""" Exemples: python3 pulse_diagnostic_monolithic.py → Archive créée dans /tmp (serveur lu depuis agentconf.ini ou demandé) python3 pulse_diagnostic_monolithic.py -s "192.168.1.50" → Archive créée dans /tmp (serveur personnalisé: 192.168.1.50) python3 pulse_diagnostic_monolithic.py -a "C:\\Program Files\\Python3\\Lib\\site-packages\\pulse_xmpp_agent" → Spécifier manuellement le répertoire de l'agent (Windows) python3 pulse_diagnostic_monolithic.py -o "/home/user/Downloads" -s "192.168.1.50" → Archive créée dans /home/user/Downloads (serveur personnalisé) """ ) parser.add_argument( "-s", "--server", help="Serveur Medulla (FQDN ou IP). Si absent, lu depuis agentconf.ini ou demandé.", default=None, metavar="" ) parser.add_argument( "-o", "--output", help="Répertoire de sortie pour l'archive ZIP (défaut: répertoire courant puis /tmp)", default=None, metavar="" ) parser.add_argument( "-a", "--agent-dir", help="Chemin au répertoire de l'agent Medulla (détection automatique par défaut)", default=None, metavar="" ) parser.add_argument( "-q", "--quiet", help="Mode silencieux", action="store_true" ) args = parser.parse_args() # Déterminer le répertoire de sortie output_dir = args.output if not output_dir: # Priorité 1: répertoire courant (où l'utilisateur lance le script) output_dir = os.getcwd() if not os.access(output_dir, os.W_OK): # Priorité 2: /tmp si le répertoire courant n'est pas accessible output_dir = tempfile.gettempdir() if not os.path.isdir(output_dir): print(f"❌ Le répertoire {output_dir} n'existe pas") sys.exit(1) if not os.access(output_dir, os.W_OK): print(f"❌ Pas de permission d'écriture sur {output_dir}") sys.exit(1) bundler = MedullaDiagnosticMonolithic(output_dir=output_dir, medulla_server=args.server, agent_dir=args.agent_dir) # Message d'accueil print(f""" {'=' * 70} MEDULLA - OUTIL DE DIAGNOSTIC SUPPORT {'=' * 70} Cet outil collecte automatiquement toutes les informations nécessaires au support technique Medulla pour diagnostiquer votre installation. Il évite les allers-retours avec le support en regroupant en une seule archive horodatée : ✓ Informations système (OS, Python, disque, mémoire) ✓ Tests de connectivité réseau (ports TCP/UDP vers le serveur Medulla) ✓ Fichiers de configuration (agentconf.ini, cluster.ini, ...) ✓ Fichiers de logs courants (xmpp-agent-machine.log, ...) ✓ Vérification d'intégrité de l'agent (MD5 + syntaxe Python) ✓ Validation des types de plugins installés ✓ Arbre complet des fichiers de l'agent avec tailles et dates ✓ Inventaire matériel (/tmp/inventory.txt si présent) L'archive est créée dans : Windows : C:\\Program Files\\Medulla\\tmp\\ Linux : /tmp/medulla_diagnostics/ Exemple : medulla_diagnostics_jfk1_20260717_16h27m36s.zip Transmettez ensuite l'archive au support Medulla. {'=' * 70} """) success, result = bundler.run() if success: abs_path = os.path.abspath(result) print(f"\n{'=' * 70}") print(f"✅ SUCCÈS - Archive créée") print(f"{'=' * 70}") print(f"📁 Chemin: {abs_path}") print(f"📊 Taille: {MedullaDiagnosticMonolithic._format_size(os.path.getsize(abs_path))}") print(f"🔍 Répertoire: {os.path.dirname(abs_path)}") print(f"\n💡 Pour récupérer l'archive:") print(f" scp {abs_path} :/chemin/local/") print(f"{'=' * 70}\n") sys.exit(0) else: print(result) sys.exit(1) if __name__ == "__main__": main()