#!/usr/bin/env python3 """ ABIS Security Audit Agent v1.0 Audit informatique & cybersécurité — Windows & Linux Usage: python abis_audit.py --once # audit immédiat python abis_audit.py --auto # boucle automatique (défaut 24h) python abis_audit.py --auto --interval 6h python abis_audit.py --once --local # pas d'envoi serveur python abis_audit.py --once --json # sortie JSON pure """ import os, sys, json, socket, platform, subprocess, datetime, time, re, argparse import urllib.request, urllib.error # ─── CONFIG ──────────────────────────────────────────────────────────────────── ABIS_SERVER = "https://api.abisinfo.net/audit/report" ABIS_EMAIL_FROM = "audit@abisinfo.net" ABIS_SUPPORT = "support@abisinfo.net" REPORT_DIR = os.path.join(os.path.expanduser("~"), ".abis_audit") VERSION = "1.0.0" # Clé d'identification de l'agent (configurable par client) AGENT_KEY = os.environ.get("ABIS_AGENT_KEY", "") CLIENT_NAME = os.environ.get("ABIS_CLIENT_NAME", socket.gethostname()) RESEND_API_KEY = os.environ.get("RESEND_API_KEY", "") NOTIFY_EMAIL = os.environ.get("ABIS_NOTIFY_EMAIL", "") IS_WINDOWS = platform.system() == "Windows" IS_LINUX = platform.system() == "Linux" # ─── COULEURS TERMINAL ───────────────────────────────────────────────────────── class C: RED = "\033[91m" if not IS_WINDOWS else "" YELLOW = "\033[93m" if not IS_WINDOWS else "" GREEN = "\033[92m" if not IS_WINDOWS else "" BLUE = "\033[94m" if not IS_WINDOWS else "" GREY = "\033[90m" if not IS_WINDOWS else "" BOLD = "\033[1m" if not IS_WINDOWS else "" RESET = "\033[0m" if not IS_WINDOWS else "" SEVERITY_COLOR = { "CRITIQUE": C.RED + C.BOLD, "ÉLEVÉ": C.RED, "MOYEN": C.YELLOW, "FAIBLE": C.BLUE, "INFO": C.GREY, "OK": C.GREEN, } SEVERITY_SCORE = {"CRITIQUE": -20, "ÉLEVÉ": -10, "MOYEN": -5, "FAIBLE": -2, "INFO": 0, "OK": 0} # ─── RÉSULTAT D'UN CHECK ──────────────────────────────────────────────────────── def finding(severity, category, title, detail="", recommendation=""): return { "severity": severity, "category": category, "title": title, "detail": detail, "recommendation": recommendation, "ts": datetime.datetime.now().isoformat(), } def ok(category, title, detail=""): return finding("OK", category, title, detail) # ─── UTILITAIRES ─────────────────────────────────────────────────────────────── def run(cmd, timeout=10): try: r = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=timeout) return r.stdout.strip() except Exception: return "" def file_readable_by_others(path): """Vérifie si un fichier est lisible par tous (Linux).""" try: mode = oct(os.stat(path).st_mode)[-3:] return int(mode[2]) >= 4 # autres ont accès lecture except Exception: return False def check_port_open(host, port, timeout=1): try: s = socket.socket() s.settimeout(timeout) s.connect((host, port)) s.close() return True except Exception: return False # ═══════════════════════════════════════════════════════════════════════════════ # CHECKS COMMUNS (Windows + Linux) # ═══════════════════════════════════════════════════════════════════════════════ def check_system_info(): """Collecte les infos système de base.""" info = { "hostname": socket.gethostname(), "os": platform.system(), "os_version": platform.version(), "os_release": platform.release(), "machine": platform.machine(), "python": platform.python_version(), "cpu_count": os.cpu_count(), } try: import shutil total, used, free = shutil.disk_usage("/") info["disk_total_gb"] = round(total / 1e9, 1) info["disk_used_gb"] = round(used / 1e9, 1) info["disk_pct"] = round(used / total * 100, 1) except Exception: pass return info def check_open_ports(): """Scan des ports courants sur localhost.""" findings = [] DANGEROUS_PORTS = { 21: ("FTP", "ÉLEVÉ", "FTP non chiffré — utiliser SFTP"), 23: ("Telnet", "CRITIQUE", "Telnet non chiffré — désactiver immédiatement"), 25: ("SMTP", "MOYEN", "Port SMTP ouvert — vérifier si nécessaire"), 69: ("TFTP", "ÉLEVÉ", "TFTP non chiffré — désactiver"), 110: ("POP3", "MOYEN", "POP3 non chiffré — utiliser POP3S (995)"), 135: ("RPC", "MOYEN", "RPC exposé — risque d'exploitation"), 139: ("NetBIOS", "ÉLEVÉ", "NetBIOS exposé — risque de partage non sécurisé"), 143: ("IMAP", "MOYEN", "IMAP non chiffré — utiliser IMAPS (993)"), 445: ("SMB", "ÉLEVÉ", "SMB exposé — risque ransomware"), 1433: ("MSSQL", "ÉLEVÉ", "SQL Server exposé — restreindre l'accès"), 3306: ("MySQL", "ÉLEVÉ", "MySQL exposé — restreindre au localhost"), 3389: ("RDP", "ÉLEVÉ", "RDP exposé — utiliser VPN"), 5432: ("PostgreSQL", "ÉLEVÉ", "PostgreSQL exposé — restreindre au localhost"), 5900: ("VNC", "ÉLEVÉ", "VNC exposé — chiffrer ou désactiver"), 6379: ("Redis", "CRITIQUE", "Redis sans auth exposé — risque critique"), 8080: ("HTTP-alt", "MOYEN", "Port HTTP alternatif exposé"), 27017:("MongoDB", "ÉLEVÉ", "MongoDB exposé — risque de fuite de données"), } SAFE_PORTS = { 22: "SSH", 80: "HTTP", 443: "HTTPS", 53: "DNS", } open_dangerous = [] open_safe = [] for port, info in DANGEROUS_PORTS.items(): if check_port_open("127.0.0.1", port): open_dangerous.append((port, info)) for port, name in SAFE_PORTS.items(): if check_port_open("127.0.0.1", port): open_safe.append((port, name)) for port, (name, sev, reco) in open_dangerous: findings.append(finding(sev, "Réseau", f"Port {port} ({name}) ouvert", f"Le port {port} est accessible sur ce système.", reco)) if open_safe: names = ", ".join(f"{p} ({n})" for p, n in open_safe) findings.append(ok("Réseau", f"Ports standards ouverts : {names}")) if not open_dangerous and not open_safe: findings.append(ok("Réseau", "Aucun port dangereux détecté")) return findings def check_sensitive_files_common(): """Vérifie la présence de fichiers sensibles courants.""" findings = [] SENSITIVE_PATTERNS = [ (".env", "Fichier d'environnement (.env)"), ("*.pem", "Certificat/clé PEM"), ("*.key", "Fichier de clé privée"), ("id_rsa", "Clé SSH privée"), ("id_ed25519", "Clé SSH privée Ed25519"), ("*.pfx", "Certificat PKCS#12"), ("credentials.json", "Fichier de credentials"), ("secrets.json", "Fichier de secrets"), ] home = os.path.expanduser("~") for fname, label in SENSITIVE_PATTERNS: if "*" not in fname: fpath = os.path.join(home, fname) if os.path.exists(fpath): findings.append(finding("MOYEN", "Fichiers sensibles", f"{label} trouvé dans le répertoire home", f"Chemin : {fpath}", "Vérifier les permissions et s'assurer qu'il n'est pas exposé")) return findings def check_passwords_in_env(): """Vérifie les variables d'env contenant des mots de passe.""" findings = [] risky_keys = [] for key, val in os.environ.items(): kl = key.lower() if any(w in kl for w in ("password", "passwd", "secret", "token", "api_key", "apikey")): if val and len(val) > 3: risky_keys.append(key) if risky_keys: findings.append(finding("INFO", "Secrets", f"{len(risky_keys)} variable(s) sensible(s) en mémoire", f"Variables : {', '.join(risky_keys[:5])}{'...' if len(risky_keys) > 5 else ''}", "Normal si l'application en a besoin — s'assurer qu'elles ne sont pas loggées")) return findings def check_ssl_certificates(): """Vérifie les certificats SSL des domaines courants.""" findings = [] import ssl domains_to_check = [] # Chercher des domaines dans les variables d'env for val in os.environ.values(): if val and ("http://" in val or "https://" in val): m = re.search(r'https?://([a-zA-Z0-9.\-]+)', val) if m: d = m.group(1) if "." in d and d not in domains_to_check: domains_to_check.append(d) # Limiter à 5 domaines for domain in domains_to_check[:5]: try: ctx = ssl.create_default_context() with ctx.wrap_socket(socket.socket(), server_hostname=domain) as s: s.settimeout(3) s.connect((domain, 443)) cert = s.getpeercert() exp_str = cert.get("notAfter", "") if exp_str: exp = datetime.datetime.strptime(exp_str, "%b %d %H:%M:%S %Y %Z") days = (exp - datetime.datetime.utcnow()).days if days < 0: findings.append(finding("CRITIQUE", "SSL/TLS", f"Certificat expiré : {domain}", f"Expiré depuis {-days} jours", "Renouveler immédiatement")) elif days < 14: findings.append(finding("ÉLEVÉ", "SSL/TLS", f"Certificat expire dans {days} jours : {domain}", f"Expiration : {exp.strftime('%d/%m/%Y')}", "Renouveler avant expiration")) elif days < 30: findings.append(finding("MOYEN", "SSL/TLS", f"Certificat expire dans {days} jours : {domain}", f"Expiration : {exp.strftime('%d/%m/%Y')}", "Planifier le renouvellement")) else: findings.append(ok("SSL/TLS", f"Certificat valide {days} jours : {domain}")) except ssl.SSLCertVerificationError: findings.append(finding("CRITIQUE", "SSL/TLS", f"Certificat invalide ou auto-signé : {domain}", "", "Utiliser un certificat signé par une CA de confiance")) except Exception: pass return findings # ═══════════════════════════════════════════════════════════════════════════════ # CHECKS LINUX # ═══════════════════════════════════════════════════════════════════════════════ def check_linux_file_permissions(): findings = [] sensitive = [ ("/root/.env", "0600", "Fichier .env root"), ("/root/.ssh", "0700", "Répertoire SSH root"), ("/root/.ssh/id_rsa", "0600", "Clé SSH privée root"), ("/root/.ssh/authorized_keys", "0600", "Clés SSH autorisées"), ("/etc/passwd", "0644", "Fichier passwd"), ("/etc/shadow", "0640", "Fichier shadow (mots de passe)"), ] for path, expected, label in sensitive: if not os.path.exists(path): continue mode = oct(os.stat(path).st_mode)[-4:] if mode != expected: sev = "CRITIQUE" if "shadow" in path or "id_rsa" in path else "ÉLEVÉ" findings.append(finding(sev, "Permissions", f"Permissions incorrectes : {label}", f"Actuel : {mode} — Attendu : {expected}", f"Corriger avec : chmod {expected[1:]} {path}")) else: findings.append(ok("Permissions", f"Permissions OK : {label}", f"{mode}")) # Chercher des fichiers .env dans /root avec permissions trop larges for fname in ["/root/api_keys.json", "/root/.env"]: if os.path.exists(fname) and file_readable_by_others(fname): findings.append(finding("ÉLEVÉ", "Permissions", f"Fichier sensible lisible par tous : {fname}", "Permissions world-readable détectées", f"chmod 600 {fname}")) return findings def check_linux_auth_logs(): findings = [] log_files = ["/var/log/auth.log", "/var/log/secure"] for lf in log_files: if not os.path.exists(lf): continue # Compter les échecs SSH des dernières 24h out = run(f"grep 'Failed password' {lf} | tail -200") lines = [l for l in out.splitlines() if l.strip()] if len(lines) > 100: findings.append(finding("CRITIQUE", "Auth", f"Brute-force SSH détecté ({len(lines)} échecs)", f"Source log : {lf}", "Activer fail2ban, changer le port SSH, utiliser des clés uniquement")) elif len(lines) > 20: findings.append(finding("ÉLEVÉ", "Auth", f"Nombreuses tentatives SSH échouées ({len(lines)})", f"Source log : {lf}", "Vérifier fail2ban et envisager de changer le port SSH")) elif len(lines) > 0: findings.append(finding("MOYEN", "Auth", f"Tentatives SSH échouées détectées ({len(lines)})", f"Source log : {lf}", "Surveiller et activer fail2ban si pas déjà fait")) else: findings.append(ok("Auth", "Aucune tentative SSH échouée récente")) break return findings def check_linux_processes(): findings = [] out = run("ps aux --no-headers") lines = out.splitlines() # Vérifier les services qui tournent en root inutilement risky_as_root = ["nginx", "apache2", "httpd", "mysqld", "mongod", "redis-server"] for line in lines: parts = line.split() if len(parts) < 11: continue user, cmd = parts[0], " ".join(parts[10:]) for svc in risky_as_root: if svc in cmd and user == "root": findings.append(finding("MOYEN", "Processus", f"{svc} tourne en tant que root", f"Commande : {cmd[:80]}", f"Créer un utilisateur dédié pour {svc}")) if not findings: findings.append(ok("Processus", "Aucun service sensible tournant en root")) return findings def check_linux_updates(): findings = [] if run("which apt"): out = run("apt list --upgradable 2>/dev/null | wc -l", timeout=30) try: count = int(out.strip()) - 1 # retirer la ligne "Listing..." if count > 50: findings.append(finding("ÉLEVÉ", "Mises à jour", f"{count} paquets en attente de mise à jour", "", "Exécuter : apt update && apt upgrade")) elif count > 10: findings.append(finding("MOYEN", "Mises à jour", f"{count} paquets en attente de mise à jour", "", "Planifier une mise à jour")) elif count > 0: findings.append(finding("FAIBLE", "Mises à jour", f"{count} paquet(s) en attente", "", "Mettre à jour")) else: findings.append(ok("Mises à jour", "Système à jour")) except Exception: pass return findings def check_linux_ssh_config(): findings = [] sshd = "/etc/ssh/sshd_config" if not os.path.exists(sshd): return findings content = open(sshd).read() checks = [ ("PermitRootLogin yes", "ÉLEVÉ", "Connexion SSH root autorisée", "Changer PermitRootLogin en 'no' ou 'without-password'"), ("PasswordAuthentication yes", "MOYEN", "Authentification SSH par mot de passe active", "Préférer les clés SSH : PasswordAuthentication no"), ("PermitEmptyPasswords yes", "CRITIQUE", "Mots de passe vides SSH autorisés", "PermitEmptyPasswords no immédiatement"), ("X11Forwarding yes", "FAIBLE", "Redirection X11 SSH activée", "Désactiver si non nécessaire : X11Forwarding no"), ] for pattern, sev, title, reco in checks: if pattern.lower() in content.lower(): findings.append(finding(sev, "SSH", title, f"Trouvé dans {sshd}", reco)) if not findings: findings.append(ok("SSH", "Configuration SSH correcte")) return findings def check_linux_firewall(): findings = [] ufw = run("ufw status 2>/dev/null") iptables = run("iptables -L INPUT --line-numbers 2>/dev/null | head -5") if "Status: active" in ufw: findings.append(ok("Firewall", "UFW actif", ufw.split("\n")[0])) elif iptables and "Chain INPUT" in iptables: findings.append(ok("Firewall", "iptables configuré")) else: findings.append(finding("ÉLEVÉ", "Firewall", "Aucun firewall détecté", "UFW et iptables semblent inactifs", "Activer UFW : ufw enable && ufw default deny incoming")) return findings def check_linux_disk(): findings = [] out = run("df -h / 2>/dev/null | tail -1") if out: parts = out.split() if len(parts) >= 5: pct = int(parts[4].replace("%", "")) if pct >= 90: findings.append(finding("CRITIQUE", "Stockage", f"Disque root presque plein ({pct}%)", out, "Libérer de l'espace immédiatement")) elif pct >= 80: findings.append(finding("ÉLEVÉ", "Stockage", f"Disque root utilisé à {pct}%", out, "Surveiller et nettoyer")) elif pct >= 70: findings.append(finding("MOYEN", "Stockage", f"Disque root utilisé à {pct}%", out, "Planifier un nettoyage")) else: findings.append(ok("Stockage", f"Espace disque OK ({pct}%)", out)) return findings def check_linux_pm2(): findings = [] out = run("pm2 jlist 2>/dev/null") if not out: return findings try: procs = json.loads(out) restarts_high = [(p["name"], p.get("pm2_env", {}).get("restart_time", 0)) for p in procs if p.get("pm2_env", {}).get("restart_time", 0) > 50] stopped = [p["name"] for p in procs if p.get("pm2_env", {}).get("status") != "online"] if restarts_high: for name, count in restarts_high: findings.append(finding("MOYEN", "Services PM2", f"Service {name} : {count} redémarrages", "", "Investiguer les logs : pm2 logs " + name)) if stopped: for name in stopped: findings.append(finding("ÉLEVÉ", "Services PM2", f"Service PM2 arrêté : {name}", "", f"Redémarrer : pm2 restart {name}")) if not restarts_high and not stopped: findings.append(ok("Services PM2", f"{len(procs)} service(s) PM2 en ligne")) except Exception: pass return findings def check_linux_http_headers(): findings = [] domains = ["localhost", "127.0.0.1"] for domain in domains: for port in [80, 443, 8080]: if not check_port_open(domain, port): continue proto = "https" if port == 443 else "http" try: req = urllib.request.Request( f"{proto}://{domain}:{port}/", headers={"User-Agent": "ABIS-Audit/1.0"} ) ctx = None if proto == "https": import ssl ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE resp = urllib.request.urlopen(req, timeout=3, context=ctx) headers = dict(resp.headers) missing = [] if "X-Frame-Options" not in headers: missing.append("X-Frame-Options") if "X-Content-Type-Options" not in headers: missing.append("X-Content-Type-Options") if "Strict-Transport-Security" not in headers and port == 443: missing.append("HSTS") if "Content-Security-Policy" not in headers: missing.append("Content-Security-Policy") if missing: findings.append(finding("MOYEN", "Headers HTTP", f"Headers de sécurité manquants sur {domain}:{port}", f"Manquants : {', '.join(missing)}", "Ajouter ces headers dans la config Caddy/Nginx")) else: findings.append(ok("Headers HTTP", f"Headers sécurité OK sur {domain}:{port}")) except Exception: pass return findings # ═══════════════════════════════════════════════════════════════════════════════ # CHECKS WINDOWS # ═══════════════════════════════════════════════════════════════════════════════ def check_windows_firewall(): findings = [] out = run("netsh advfirewall show allprofiles state 2>nul") if "ON" in out: findings.append(ok("Firewall", "Pare-feu Windows actif")) else: findings.append(finding("ÉLEVÉ", "Firewall", "Pare-feu Windows désactivé", out[:200], "Activer le pare-feu Windows immédiatement")) return findings def check_windows_antivirus(): findings = [] out = run('powershell -Command "Get-MpComputerStatus | Select-Object AntivirusEnabled,RealTimeProtectionEnabled | ConvertTo-Json" 2>nul') if out: try: data = json.loads(out) av = data.get("AntivirusEnabled", False) rtp = data.get("RealTimeProtectionEnabled", False) if av and rtp: findings.append(ok("Antivirus", "Windows Defender actif avec protection temps réel")) elif av: findings.append(finding("MOYEN", "Antivirus", "Windows Defender actif mais protection temps réel désactivée", "", "Activer la protection temps réel")) else: findings.append(finding("ÉLEVÉ", "Antivirus", "Windows Defender désactivé", "", "Activer Windows Defender ou installer un antivirus")) except Exception: findings.append(finding("INFO", "Antivirus", "Impossible de vérifier le statut antivirus", out[:100], "")) return findings def check_windows_updates(): findings = [] out = run('powershell -Command "(New-Object -ComObject Microsoft.Update.Session).CreateUpdateSearcher().Search(\'IsInstalled=0 and Type=\'Software\'\').Updates.Count" 2>nul') try: count = int(out.strip()) if count > 20: findings.append(finding("ÉLEVÉ", "Mises à jour", f"{count} mises à jour Windows en attente", "", "Installer les mises à jour via Windows Update")) elif count > 0: findings.append(finding("MOYEN", "Mises à jour", f"{count} mise(s) à jour en attente", "", "Planifier les mises à jour")) else: findings.append(ok("Mises à jour", "Windows à jour")) except Exception: pass return findings def check_windows_shares(): findings = [] out = run("net share 2>nul") dangerous_shares = [] for line in out.splitlines(): parts = line.split() if parts and parts[0] not in ("Nom", "Share", "------", "La", ""): share = parts[0] if share not in ("C$", "ADMIN$", "IPC$", "print$"): dangerous_shares.append(share) if dangerous_shares: findings.append(finding("MOYEN", "Partages réseau", f"Partages réseau non standards détectés : {', '.join(dangerous_shares)}", "", "Vérifier que ces partages sont nécessaires et sécurisés")) else: findings.append(ok("Partages réseau", "Aucun partage réseau non standard")) return findings def check_windows_rdp(): findings = [] out = run('reg query "HKLM\\System\\CurrentControlSet\\Control\\Terminal Server" /v fDenyTSConnections 2>nul') if "0x0" in out: if check_port_open("127.0.0.1", 3389): findings.append(finding("ÉLEVÉ", "RDP", "Bureau à distance (RDP) activé", "Port 3389 ouvert et RDP activé", "Utiliser un VPN pour l'accès RDP distant")) else: findings.append(finding("FAIBLE", "RDP", "RDP activé mais port non accessible en local", "", "")) else: findings.append(ok("RDP", "Bureau à distance désactivé")) return findings def check_windows_users(): findings = [] out = run("net localgroup administrators 2>nul") admins = [] in_members = False for line in out.splitlines(): if "---" in line: in_members = True continue if in_members and line.strip() and "The command" not in line: admins.append(line.strip()) if len(admins) > 3: findings.append(finding("MOYEN", "Comptes", f"{len(admins)} comptes administrateurs locaux", f"Admins : {', '.join(admins[:5])}", "Réduire le nombre d'administrateurs au strict nécessaire")) elif admins: findings.append(ok("Comptes", f"{len(admins)} admin(s) local/locaux", ", ".join(admins))) return findings def check_windows_bitlocker(): findings = [] out = run('powershell -Command "Get-BitLockerVolume -MountPoint C: | Select-Object ProtectionStatus | ConvertTo-Json" 2>nul') if out and "1" in out: findings.append(ok("Chiffrement", "BitLocker actif sur C:")) elif out and "0" in out: findings.append(finding("MOYEN", "Chiffrement", "BitLocker non actif sur C:", "", "Activer BitLocker pour protéger les données en cas de vol")) return findings # ═══════════════════════════════════════════════════════════════════════════════ # MOTEUR D'AUDIT # ═══════════════════════════════════════════════════════════════════════════════ def run_audit(): """Lance tous les checks et retourne le rapport complet.""" print(f"\n{C.BOLD}{'═'*60}{C.RESET}") print(f"{C.BLUE}{C.BOLD} ABIS Security Audit v{VERSION}{C.RESET}") print(f"{C.GREY} {datetime.datetime.now().strftime('%d/%m/%Y %H:%M:%S')} — {socket.gethostname()}{C.RESET}") print(f"{C.BOLD}{'═'*60}{C.RESET}\n") all_findings = [] system_info = check_system_info() def run_checks(label, checks_fn): print(f"{C.GREY} ▶ {label}...{C.RESET}", end=" ", flush=True) try: results = checks_fn() all_findings.extend(results) crits = sum(1 for f in results if f["severity"] == "CRITIQUE") highs = sum(1 for f in results if f["severity"] == "ÉLEVÉ") marker = f"{C.RED}✗ {crits}C {highs}H{C.RESET}" if crits or highs else f"{C.GREEN}✓{C.RESET}" print(marker) except Exception as e: print(f"{C.YELLOW}⚠ erreur{C.RESET}") # Checks communs run_checks("Ports réseau", check_open_ports) run_checks("Fichiers sensibles", check_sensitive_files_common) run_checks("Variables d'env", check_passwords_in_env) run_checks("Certificats SSL", check_ssl_certificates) # Checks spécifiques à l'OS if IS_LINUX: run_checks("Permissions fichiers", check_linux_file_permissions) run_checks("Logs authentification", check_linux_auth_logs) run_checks("Processus système", check_linux_processes) run_checks("Mises à jour système", check_linux_updates) run_checks("Configuration SSH", check_linux_ssh_config) run_checks("Firewall", check_linux_firewall) run_checks("Espace disque", check_linux_disk) run_checks("Services PM2", check_linux_pm2) run_checks("Headers HTTP", check_linux_http_headers) elif IS_WINDOWS: run_checks("Pare-feu Windows", check_windows_firewall) run_checks("Antivirus", check_windows_antivirus) run_checks("Mises à jour", check_windows_updates) run_checks("Partages réseau", check_windows_shares) run_checks("Bureau à distance", check_windows_rdp) run_checks("Comptes admin", check_windows_users) run_checks("Chiffrement BitLocker",check_windows_bitlocker) # Calcul du score score = 100 for f in all_findings: score += SEVERITY_SCORE.get(f["severity"], 0) score = max(0, min(100, score)) counts = {s: sum(1 for f in all_findings if f["severity"] == s) for s in ["CRITIQUE", "ÉLEVÉ", "MOYEN", "FAIBLE", "INFO", "OK"]} return { "version": VERSION, "timestamp": datetime.datetime.now().isoformat(), "hostname": system_info["hostname"], "client": CLIENT_NAME, "os": f"{system_info['os']} {system_info.get('os_release','')}", "score": score, "risk_level": "CRITIQUE" if score < 40 else "ÉLEVÉ" if score < 60 else "MOYEN" if score < 75 else "BON" if score < 90 else "EXCELLENT", "counts": counts, "system_info": system_info, "findings": all_findings, } # ═══════════════════════════════════════════════════════════════════════════════ # AFFICHAGE DU RAPPORT # ═══════════════════════════════════════════════════════════════════════════════ def print_report(report): findings = report["findings"] score = report["score"] counts = report["counts"] # Trier par sévérité ORDER = ["CRITIQUE", "ÉLEVÉ", "MOYEN", "FAIBLE", "INFO", "OK"] findings_sorted = sorted(findings, key=lambda f: ORDER.index(f["severity"])) print(f"\n{C.BOLD}{'─'*60}{C.RESET}") print(f"{C.BOLD} RÉSULTATS DE L'AUDIT{C.RESET}") print(f"{'─'*60}") current_cat = None for f in findings_sorted: if f["severity"] == "OK": continue # Afficher les OK en résumé seulement cat = f["category"] if cat != current_cat: print(f"\n {C.BOLD}{cat.upper()}{C.RESET}") current_cat = cat color = SEVERITY_COLOR.get(f["severity"], "") sev_tag = f"[{f['severity']:<8}]" print(f" {color}{sev_tag}{C.RESET} {f['title']}") if f["detail"]: print(f" {C.GREY} {f['detail'][:80]}{C.RESET}") if f["recommendation"]: print(f" {C.BLUE} → {f['recommendation'][:80]}{C.RESET}") # Score final score_color = C.RED if score < 60 else C.YELLOW if score < 80 else C.GREEN risk_color = C.RED if report["risk_level"] in ("CRITIQUE", "ÉLEVÉ") else C.YELLOW if report["risk_level"] == "MOYEN" else C.GREEN print(f"\n{C.BOLD}{'═'*60}{C.RESET}") print(f" {C.BOLD}SCORE DE SÉCURITÉ : {score_color}{score}/100{C.RESET} " f"Niveau : {risk_color}{report['risk_level']}{C.RESET}") print(f" {C.RED}Critiques : {counts.get('CRITIQUE',0)}{C.RESET} " f"{C.RED}Élevés : {counts.get('ÉLEVÉ',0)}{C.RESET} " f"{C.YELLOW}Moyens : {counts.get('MOYEN',0)}{C.RESET} " f"{C.GREEN}OK : {counts.get('OK',0)}{C.RESET}") print(f"{'═'*60}\n") # ═══════════════════════════════════════════════════════════════════════════════ # ENVOI AU SERVEUR ABIS # ═══════════════════════════════════════════════════════════════════════════════ def send_to_server(report): try: payload = json.dumps(report).encode("utf-8") headers = { "Content-Type": "application/json", "User-Agent": f"ABIS-Audit/{VERSION}", "X-Agent-Key": AGENT_KEY, } req = urllib.request.Request(ABIS_SERVER, data=payload, headers=headers, method="POST") resp = urllib.request.urlopen(req, timeout=10) print(f" {C.GREEN}✓ Rapport envoyé au serveur ABIS{C.RESET}") return True except Exception as e: print(f" {C.YELLOW}⚠ Envoi serveur échoué : {e}{C.RESET}") return False def send_email_report(report, to_email): if not RESEND_API_KEY or not to_email: return score = report["score"] counts = report["counts"] risk_color = "#ef4444" if score < 60 else "#f59e0b" if score < 80 else "#22c55e" rows = "" ORDER = ["CRITIQUE", "ÉLEVÉ", "MOYEN", "FAIBLE", "INFO"] SEV_COLORS = { "CRITIQUE": "#ef4444", "ÉLEVÉ": "#ef4444", "MOYEN": "#f59e0b", "FAIBLE": "#3b82f6", "INFO": "#6b7280" } for f in sorted(report["findings"], key=lambda x: ORDER.index(x["severity"]) if x["severity"] in ORDER else 99): if f["severity"] == "OK": continue color = SEV_COLORS.get(f["severity"], "#fff") rows += f""" [{f['severity']}] {f['category']} {f['title']} {f.get('recommendation','')[:60]} """ html = f"""
ABIS
Security Audit Report
{report['timestamp'][:16]} — {report['hostname']}
{score}/100
{report['risk_level']}
● Critiques : {counts.get('CRITIQUE',0)} ● Élevés : {counts.get('ÉLEVÉ',0)} ● Moyens : {counts.get('MOYEN',0)} ● OK : {counts.get('OK',0)}
{rows}
NIVEAU CATÉGORIE CONSTAT ACTION
ABIS Télémaintenance — abisinfo.net — {ABIS_SUPPORT}
""" try: payload = json.dumps({ "from": f"ABIS Audit <{ABIS_EMAIL_FROM}>", "to": [to_email], "subject": f"[ABIS Audit] {report['hostname']} — Score {score}/100 {report['risk_level']}", "html": html, }).encode("utf-8") req = urllib.request.Request( "https://api.resend.com/emails", data=payload, headers={"Authorization": f"Bearer {RESEND_API_KEY}", "Content-Type": "application/json"}, method="POST" ) urllib.request.urlopen(req, timeout=10) print(f" {C.GREEN}✓ Rapport email envoyé à {to_email}{C.RESET}") except Exception as e: print(f" {C.YELLOW}⚠ Email échoué : {e}{C.RESET}") def save_report_local(report): os.makedirs(REPORT_DIR, exist_ok=True) ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") path = os.path.join(REPORT_DIR, f"audit_{ts}.json") with open(path, "w", encoding="utf-8") as f: json.dump(report, f, indent=2, ensure_ascii=False) print(f" {C.GREY}✓ Rapport sauvegardé : {path}{C.RESET}") return path # ═══════════════════════════════════════════════════════════════════════════════ # POINT D'ENTRÉE # ═══════════════════════════════════════════════════════════════════════════════ def parse_interval(s): """Convertit '6h', '30m', '1d' en secondes.""" m = re.match(r'^(\d+)(h|m|d)$', s.lower()) if not m: return 86400 # défaut 24h v, unit = int(m.group(1)), m.group(2) return v * (3600 if unit == "h" else 60 if unit == "m" else 86400) def main(): parser = argparse.ArgumentParser(description="ABIS Security Audit Agent") parser.add_argument("--once", action="store_true", help="Audit unique puis quitter") parser.add_argument("--auto", action="store_true", help="Boucle automatique") parser.add_argument("--interval", default="24h", help="Intervalle en mode auto (ex: 6h, 30m, 1d)") parser.add_argument("--local", action="store_true", help="Ne pas envoyer au serveur ABIS") parser.add_argument("--json", action="store_true", help="Sortie JSON pure (stdout)") parser.add_argument("--email", default=NOTIFY_EMAIL, help="Email de notification") args = parser.parse_args() if not args.once and not args.auto: print(f"\n{C.BOLD}ABIS Security Audit v{VERSION}{C.RESET}") print("Usage:") print(" python abis_audit.py --once # audit unique") print(" python abis_audit.py --auto # boucle 24h") print(" python abis_audit.py --auto --interval 6h") print(" python abis_audit.py --once --local # sans envoi serveur") print(" python abis_audit.py --once --json # sortie JSON\n") sys.exit(0) interval = parse_interval(args.interval) def do_audit(): report = run_audit() if args.json: print(json.dumps(report, indent=2, ensure_ascii=False)) return print_report(report) save_report_local(report) if not args.local: send_to_server(report) if args.email: send_email_report(report, args.email) if args.once: do_audit() elif args.auto: print(f"{C.BLUE}Mode automatique — intervalle : {args.interval}{C.RESET}") while True: do_audit() next_run = datetime.datetime.now() + datetime.timedelta(seconds=interval) print(f" {C.GREY}Prochain audit : {next_run.strftime('%d/%m/%Y %H:%M:%S')}{C.RESET}\n") time.sleep(interval) if __name__ == "__main__": main()