Threat Hunting Avancé : Guide des Techniques de Détection des Menaces

Le threat hunting est une approche proactive de la cybersécurité qui vise à identifier les menaces avant qu’elles ne causent des dommages. Ce guide explore les techniques avancées utilisées par les experts en threat hunting.

1. Préparation de l’Environnement

Configuration des Outils

# Installation de Zeek (anciennement Bro)
sudo apt install zeek

# Configuration de base
cat > /usr/local/zeek/share/zeek/site/local.zeek << EOF
@load frameworks/files/extract-all-files
@load frameworks/communication/listen
@load frameworks/communication/seen
@load frameworks/communication/notice
@load frameworks/communication/notice/actions/email
EOF

Mise en Place du Monitoring

# Script de monitoring système
import psutil
import time
from datetime import datetime

def monitor_system():
    while True:
        metrics = {
            'cpu_percent': psutil.cpu_percent(),
            'memory_percent': psutil.virtual_memory().percent,
            'disk_usage': psutil.disk_usage('/').percent,
            'network_connections': len(psutil.net_connections()),
            'timestamp': datetime.now().isoformat()
        }
        print(metrics)
        time.sleep(60)

2. Techniques de Détection

Analyse des Processus

# Détection de processus malveillants
import psutil
import re

def detect_malicious_processes():
    suspicious_patterns = [
        r'cryptominer',
        r'miner',
        r'botnet',
        r'backdoor'
    ]
    
    for proc in psutil.process_iter(['pid', 'name', 'cmdline']):
        try:
            cmdline = ' '.join(proc.info['cmdline'])
            for pattern in suspicious_patterns:
                if re.search(pattern, cmdline, re.I):
                    print(f"Processus suspect détecté: {proc.info['pid']} - {proc.info['name']}")
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            continue

Analyse du Trafic Réseau

# Analyse du trafic avec Scapy
from scapy.all import *
import collections

def analyze_network_traffic():
    packet_counts = collections.Counter()
    
    def packet_callback(packet):
        if IP in packet:
            packet_counts[packet[IP].src] += 1
    
    sniff(prn=packet_callback, store=0, timeout=60)
    
    # Analyse des résultats
    for ip, count in packet_counts.most_common():
        if count > 1000:  # Seuil de détection
            print(f"Trafic suspect détecté depuis {ip}: {count} paquets")

3. Analyse des Logs

Analyse des Logs Système

# Analyse des logs avec Python
import re
from datetime import datetime, timedelta

def analyze_system_logs():
    suspicious_patterns = {
        'failed_login': r'Failed password for',
        'sudo_attempt': r'sudo:.*authentication failure',
        'port_scan': r'port scan detected'
    }
    
    with open('/var/log/auth.log', 'r') as f:
        for line in f:
            for pattern_name, pattern in suspicious_patterns.items():
                if re.search(pattern, line):
                    print(f"Événement suspect détecté: {pattern_name} - {line.strip()}")

Analyse des Logs DNS

# Analyse des requêtes DNS
import dnslib
from collections import defaultdict

def analyze_dns_queries():
    query_counts = defaultdict(int)
    suspicious_domains = set()
    
    def process_dns_query(query):
        domain = str(query.qname)
        query_counts[domain] += 1
        
        if query_counts[domain] > 1000:  # Seuil de détection
            suspicious_domains.add(domain)
    
    return suspicious_domains

4. Détection de Malware

Analyse Statique

# Analyse de fichiers avec YARA
import yara

def analyze_file(file_path):
    rules = yara.compile(source='''
        rule MaliciousFile {
            strings:
                $a = "CreateRemoteThread" nocase
                $b = "VirtualAlloc" nocase
                $c = "WriteProcessMemory" nocase
            condition:
                any of them
        }
    ''')
    
    matches = rules.match(file_path)
    return matches

Analyse Dynamique

# Analyse comportementale
import ctypes
import sys

def monitor_process_behavior(pid):
    suspicious_apis = [
        'CreateRemoteThread',
        'VirtualAlloc',
        'WriteProcessMemory',
        'ShellExecute'
    ]
    
    def hook_api(api_name):
        # Implémentation du hook
        pass
    
    for api in suspicious_apis:
        hook_api(api)

5. Investigation Forensique

Analyse de la Mémoire

# Script d'analyse mémoire
import volatility.plugins.common as common
import volatility.plugins.taskmods as taskmods

def analyze_memory(memory_file):
    # Configuration de Volatility
    config = common.Config()
    config.add_option('FILE', short_option='f', default=memory_file)
    
    # Analyse des processus
    pslist = taskmods.PSList(config)
    for proc in pslist.calculate():
        print(f"Processus trouvé: {proc.pid} - {proc.name}")

Analyse du Disque

# Script d'analyse disque
import os
import hashlib
from datetime import datetime

def analyze_disk():
    known_malware_hashes = set()
    
    def calculate_file_hash(file_path):
        sha256_hash = hashlib.sha256()
        with open(file_path, "rb") as f:
            for byte_block in iter(lambda: f.read(4096), b""):
                sha256_hash.update(byte_block)
        return sha256_hash.hexdigest()
    
    for root, dirs, files in os.walk('/'):
        for file in files:
            try:
                file_path = os.path.join(root, file)
                file_hash = calculate_file_hash(file_path)
                if file_hash in known_malware_hashes:
                    print(f"Malware détecté: {file_path}")
            except (PermissionError, FileNotFoundError):
                continue

6. Automatisation et Orchestration

Pipeline de Détection

# Pipeline de threat hunting
from typing import List, Dict
import json

class ThreatHuntingPipeline:
    def __init__(self):
        self.detectors = []
        self.analyzers = []
    
    def add_detector(self, detector):
        self.detectors.append(detector)
    
    def add_analyzer(self, analyzer):
        self.analyzers.append(analyzer)
    
    def run(self, data: Dict) -> List[Dict]:
        alerts = []
        for detector in self.detectors:
            if detector.detect(data):
                alerts.append(detector.get_alert())
        
        for alert in alerts:
            for analyzer in self.analyzers:
                analyzer.analyze(alert)
        
        return alerts

Intégration SIEM

# Intégration avec Splunk
from splunklib.client import Service
import json

def send_to_siem(event_data):
    service = Service(
        host='splunk.example.com',
        port=8089,
        username='admin',
        password='password'
    )
    
    index = service.indexes['threat_hunting']
    index.submit(json.dumps(event_data))

7. Visualisation et Reporting

Dashboard de Monitoring

# Création de dashboard avec Dash
from dash import Dash, html, dcc
import plotly.express as px

app = Dash(__name__)

def create_dashboard():
    app.layout = html.Div([
        html.H1('Dashboard de Threat Hunting'),
        dcc.Graph(
            id='alerts-graph',
            figure=px.line(
                alerts_data,
                x='timestamp',
                y='count',
                title='Alertes par Heure'
            )
        )
    ])

Génération de Rapports

# Génération de rapports
from jinja2 import Template
import pdfkit

def generate_report(alerts, findings):
    template = Template('''
        <html>
            <head>
                <title>Rapport de Threat Hunting</title>
                <style>
                    body { font-family: Arial; }
                    .alert { color: red; }
                    .finding { color: orange; }
                </style>
            </head>
            <body>
                <h1>Rapport de Threat Hunting</h1>
                <h2>Alertes</h2>
                {% for alert in alerts %}
                <div class="alert">
                    {{ alert }}
                </div>
                {% endfor %}
                <h2>Découvertes</h2>
                {% for finding in findings %}
                <div class="finding">
                    {{ finding }}
                </div>
                {% endfor %}
            </body>
        </html>
    ''')
    
    html_content = template.render(alerts=alerts, findings=findings)
    pdfkit.from_string(html_content, 'threat_hunting_report.pdf')

Conclusion

Le threat hunting est une discipline essentielle qui nécessite :

• Une connaissance approfondie des systèmes
• Des outils spécialisés
• Une méthodologie rigoureuse
• Une documentation précise
• Une veille constante

Points clés à retenir :

• Automatiser les tâches répétitives
• Maintenir une base de connaissances
• Collaborer avec les équipes
• Documenter les découvertes
• Mettre à jour régulièrement les outils