PCTAMALOU

🔬 Mission pratique – Laboratoire (2 heures)

Scénario : Création d'un analyseur de logs de sécurité

Vous êtes responsable de la surveillance d'un serveur. Vous devez créer un outil qui analyse automatiquement les logs à la recherche d'activités suspectes.

Étapes détaillées : De zéro à script fonctionnel

1. Configuration de l'environnement
   Organisation professionnelle des dossiers :

   # Structure de projet professionnelle
   cd ~/mon_lab
   mkdir -p semaine4_python/{scripts,logs,tests,output}
   cd semaine4_python
   
   # Vérification Python
   python3 --version
   which python3
   
   # Création d'un environnement virtuel (bonne pratique)
   python3 -m venv venv  # Optionnel mais recommandé
   # source venv/bin/activate  # Pour activerCopier

2. Création de logs réalistes
   Générons des logs simulés mais réalistes pour notre analyse :

   # Fichier de logs d'authentification simulé
   cat > logs/auth.log << 'EOF'
   Jan 10 08:30:15 server sshd[1234]: Accepted password for kali from 192.168.1.100 port 54322
   Jan 10 08:31:22 server sshd[1235]: Failed password for invalid user admin from 192.168.1.200 port 54323
   Jan 10 08:31:25 server sshd[1235]: Failed password for invalid user admin from 192.168.1.200 port 54323
   Jan 10 08:31:28 server sshd[1235]: Failed password for invalid user admin from 192.168.1.200 port 54323
   Jan 10 08:32:10 server sshd[1236]: Accepted publickey for root from 10.0.0.5 port 54324
   Jan 10 08:33:45 server sshd[1237]: Failed password for kali from 192.168.1.150 port 54325
   Jan 10 08:34:20 server sudo:     kali : TTY=pts/0 ; PWD=/home/kali ; USER=root ; COMMAND=/bin/ls
   Jan 10 08:35:00 server sshd[1238]: Accepted password for www-data from 172.16.0.10 port 54326
   Jan 10 08:36:15 server sshd[1239]: Invalid user hacker from 203.0.113.5 port 54327
   Jan 10 08:36:18 server sshd[1239]: Failed password for invalid user hacker from 203.0.113.5 port 54327
   Jan 10 08:40:00 server CRON[1240]: (root) CMD (/usr/bin/updatedb)
   EOF
   
   # Fichier de logs système
   cat > logs/system.log << 'EOF'
   INFO: Service ssh started successfully
   ERROR: Disk usage at 95% on /dev/sda1
   WARNING: High memory usage detected (85%)
   INFO: Backup completed successfully
   ERROR: Could not connect to database server
   WARNING: CPU temperature above threshold
   INFO: Security scan started
   ERROR: Firewall rule failed to apply
   WARNING: Unusual network activity from 192.168.1.200
   INFO: System updated to latest version
   EOFCopier

3. Écriture du script d'analyse basique
   Créez votre premier script avec un éditeur (nano, vim, ou code) :

   #!/usr/bin/env python3
   """
   PCTAMALOU - Semaine 4 : Analyseur de logs de sécurité
   Script d'analyse de logs SSH pour détecter des tentatives de bruteforce
   """
   
   import sys
   import os
   from datetime import datetime
   from collections import defaultdict
   
   # ============================================================================
   # CONFIGURATION
   # ============================================================================
   LOG_FILE = "logs/auth.log"
   OUTPUT_FILE = "output/security_report.txt"
   FAILED_THRESHOLD = 3  # Nombre d'échecs avant alerte
   
   # ============================================================================
   # FONCTIONS D'ANALYSE
   # ============================================================================
   def analyse_ssh_logs(log_path):
       """
       Analyse les logs SSH pour détecter des patterns suspects
       Retourne un dictionnaire avec les statistiques
       """
       stats = {
           'total_lines': 0,
           'accepted_logins': 0,
           'failed_logins': 0,
           'invalid_users': 0,
           'suspicious_ips': defaultdict(int),
           'failed_by_user': defaultdict(int)
       }
       
       try:
           with open(log_path, 'r') as file:
               for line in file:
                   stats['total_lines'] += 1
                   
                   # Détection des connexions acceptées
                   if "Accepted" in line and "sshd" in line:
                       stats['accepted_logins'] += 1
                       
                   # Détection des échecs de connexion
                   if "Failed password" in line:
                       stats['failed_logins'] += 1
                       
                       # Extraction de l'IP source
                       import re
                       ip_match = re.search(r'from (\d+\.\d+\.\d+\.\d+)', line)
                       if ip_match:
                           ip = ip_match.group(1)
                           stats['suspicious_ips'][ip] += 1
                       
                       # Extraction du nom d'utilisateur
                       user_match = re.search(r'for (\w+) from', line)
                       if user_match:
                           user = user_match.group(1)
                           stats['failed_by_user'][user] += 1
                   
                   # Détection des utilisateurs invalides
                   if "Invalid user" in line:
                       stats['invalid_users'] += 1
                       
       except FileNotFoundError:
           print(f"[ERREUR] Fichier {log_path} non trouvé")
           return None
       except Exception as e:
           print(f"[ERREUR] Problème de lecture: {e}")
           return None
       
       return stats
   
   def generate_report(stats, output_path):
       """Génère un rapport d'analyse formaté"""
       if not stats:
           return False
       
       try:
           with open(output_path, 'w') as report:
               report.write("=" * 60 + "\n")
               report.write("RAPPORT D'ANALYSE DE SÉCURITÉ\n")
               report.write(f"Généré le: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
               report.write("=" * 60 + "\n\n")
               
               # Statistiques générales
               report.write("📊 STATISTIQUES GÉNÉRALES\n")
               report.write(f"  • Lignes analysées: {stats['total_lines']}\n")
               report.write(f"  • Connexions acceptées: {stats['accepted_logins']}\n")
               report.write(f"  • Échecs de connexion: {stats['failed_logins']}\n")
               report.write(f"  • Utilisateurs invalides: {stats['invalid_users']}\n\n")
               
               # IPs suspectes
               report.write("🔍 IPS SUSPECTES (plus de 2 échecs)\n")
               suspicious_count = 0
               for ip, count in stats['suspicious_ips'].items():
                   if count >= 2:
                       report.write(f"  • {ip}: {count} échecs\n")
                       suspicious_count += 1
               if suspicious_count == 0:
                   report.write("  Aucune IP suspecte détectée\n")
               report.write("\n")
               
               # Utilisateurs avec échecs
               report.write("👤 UTILISATEURS AVEC ÉCHECS\n")
               for user, count in stats['failed_by_user'].items():
                   report.write(f"  • {user}: {count} échecs\n")
               
               # Recommandations
               report.write("\n💡 RECOMMANDATIONS\n")
               if stats['failed_logins'] > 10:
                   report.write("  ⚠️  Nombre élevé d'échecs - Vérifier les règles firewall\n")
               if stats['invalid_users'] > 5:
                   report.write("  ⚠️  Nombre élevé d'utilisateurs invalides - Possible scan\n")
               if suspicious_count > 0:
                   report.write(f"  ⚠️  {suspicious_count} IP(s) suspectes détectées\n")
               
               report.write("\n" + "=" * 60 + "\n")
           
           return True
           
       except Exception as e:
           print(f"[ERREUR] Échec génération rapport: {e}")
           return False
   
   # ============================================================================
   # FONCTION PRINCIPALE
   # ============================================================================
   def main():
       print("[*] PCTAMALOU - Analyseur de logs de sécurité")
       print(f"[*] Analyse du fichier: {LOG_FILE}")
       
       # Analyse des logs
       stats = analyse_ssh_logs(LOG_FILE)
       if not stats:
           print("[!] Échec de l'analyse")
           sys.exit(1)
       
       # Génération du rapport
       print(f"[*] Génération du rapport: {OUTPUT_FILE}")
       if generate_report(stats, OUTPUT_FILE):
           print("[+] Rapport généré avec succès")
           
           # Affichage des points clés
           print(f"\n📋 RÉSUMÉ:")
           print(f"  • Échecs de connexion: {stats['failed_logins']}")
           print(f"  • IPs uniques avec échecs: {len(stats['suspicious_ips'])}")
           
           # Détection d'IPs très suspectes
           high_risk_ips = [ip for ip, count in stats['suspicious_ips'].items() if count >= FAILED_THRESHOLD]
           if high_risk_ips:
               print(f"  ⚠️  IPs à haut risque: {', '.join(high_risk_ips)}")
       else:
           print("[!] Échec génération rapport")
           sys.exit(1)
   
   # ============================================================================
   # POINT D'ENTRÉE
   # ============================================================================
   if __name__ == "__main__":
       main()Copier

4. Exécution et test du script
   Testez votre création :

   # Rendre le script exécutable
   chmod +x scripts/log_analyzer.py
   
   # Exécution
   python3 scripts/log_analyzer.py
   
   # Vérification du rapport
   cat output/security_report.txt
   
   # Test avec un fichier manquant (gestion d'erreur)
   mv logs/auth.log logs/auth.log.backup
   python3 scripts/log_analyzer.py  # Doit afficher une erreur
   mv logs/auth.log.backup logs/auth.logCopier

5. Améliorations et extensions
   Ajoutez des fonctionnalités avancées :

   # Extension 1: Ajout d'arguments en ligne de commande
   import argparse
   
   parser = argparse.ArgumentParser(description="Analyseur de logs de sécurité")
   parser.add_argument("-f", "--file", help="Fichier log à analyser", default="logs/auth.log")
   parser.add_argument("-o", "--output", help="Fichier de sortie", default="output/report.txt")
   parser.add_argument("-t", "--threshold", type=int, help="Seuil d'alerte", default=3)
   args = parser.parse_args()
   
   # Extension 2: Export JSON pour intégration avec d'autres outils
   import json
   
   def export_json(stats, filename):
       with open(filename, 'w') as f:
           json.dump(stats, f, indent=2)
   
   # Extension 3: Envoi d'alertes par email (simulé)
   def send_alert(ip, count):
       print(f"[ALERTE] IP {ip} a {count} échecs de connexion")Copier

6. Création d'un deuxième script : analyseur de logs système
   Pour pratiquer, créez un script complémentaire :

   #!/usr/bin/env python3
   """
   Analyseur de logs système - Détection de problèmes
   """
   
   def analyse_system_logs(log_file):
       errors = []
       warnings = []
       
       with open(log_file, 'r') as f:
           for line in f:
               if "ERROR" in line:
                   errors.append(line.strip())
               elif "WARNING" in line:
                   warnings.append(line.strip())
       
       return {
           'errors': errors,
           'warnings': warnings,
           'total_errors': len(errors),
           'total_warnings': len(warnings)
       }
   
   # Testez ce script sur logs/system.logCopier