Création et Test de Règles SIEM

siem-rule-creation/
├── siem-platforms/
│   ├── splunk/
│   ├── elk-stack/
│   ├── microsoft-sentinel/
│   └── graylog/
├── detection-rules/
│   ├── authentication/
│   ├── malware/
│   ├── lateral-movement/
│   ├── data-exfiltration/
│   └── privilege-escalation/
├── test-data/
│   ├── benign-traffic/
│   ├── attack-simulations/
│   └── red-team-exercises/
├── scripts/
│   ├── rule-validator.py
│   ├── test-case-generator.py
│   └── performance-analyzer.py
├── documentation/
│   ├── rule-development-guide.md
│   ├── tuning-procedures.md
│   └── testing-framework.md
├── reports/
│   └── rule-evaluation-template.md
└── README.md

index=wineventlog sourcetype="WinEventLog:Security" EventCode=4625
| bucket _time span=5m
| stats count by _time, Account_Name, Workstation_Name, IpAddress
| where count > 10
| eval risk_score=case(count > 50, "High", count > 25, "Medium", count > 10, "Low")
| rename Account_Name as user, Workstation_Name as source_host, IpAddress as source_ip
| table _time, user, source_host, source_ip, count, risk_score
| eval description="Multiple failed login attempts detected for user " + user + " from " + source_ip

{
  "rule_id": "powershell_malicious_execution",
  "name": "Exécution PowerShell Suspecte Détectée",
  "description": "Détecte l'exécution potentiellement malveillante de commandes PowerShell",
  "index": ["winlogbeat-*"],
  "type": "query",
  "query": {
    "bool": {
      "must": [
        {"term": {"event.code": "4104"}},
        {"bool": {
          "should": [
            {"wildcard": {"powershell.command_line.text": "*-enc*"}},
            {"wildcard": {"powershell.command_line.text": "*-nop*"}},
            {"wildcard": {"powershell.command_line.text": "*-w hidden*"}},
            {"wildcard": {"powershell.command_line.text": "*bypass*"}}
          ]
        }}
      ]
    }
  },
  "risk_score": 70,
  "severity": "high",
  "tags": ["powershell", "malware", "defense_evasion"],
  "actions": [
    {
      "type": "webhook",
      "url": "https://your-soc-webhook.com/alert"
    }
  ]
}

// Règle de Détection KQL Microsoft Sentinel
let DataTransferThreshold = 100000000; // 100MB
let TimeWindow = 1h;
let HighVolumeTransfers =
OfficeActivity
| where TimeGenerated >= ago(TimeWindow)
| where Operation in ("FileUploaded", "FileDownloaded")
| extend FileSize = toreal(coalesce(tonumber(CoalescedFileSize), 0))
| where FileSize > DataTransferThreshold
| summarize TotalBytes = sum(FileSize), FileCount = count() by TimeGenerated, UserId, Operation, OfficeWorkload
| where TotalBytes > DataTransferThreshold * 2 // Multiple large files
| extend AlertDetails = strcat("User ", UserId, " transferred ", FileCount, " files totaling ", TotalBytes, " bytes in ", OfficeWorkload)
| project TimeGenerated, AlertDetails, UserId, TotalBytes, FileCount, OfficeWorkload
| order by TimeGenerated desc;
HighVolumeTransfers

import yaml
from datetime import datetime

class ThreatModel:
    def __init__(self, tactic, technique, sub_technique):
        self.tactic = tactic
        self.technique = technique
        self.sub_technique = sub_technique
        self.data_sources = []
        self.detection_logic = []
        self.false_positive_mitigation = []
    
    def add_data_source(self, source):
        """Ajouter la source de données requise pour la détection"""
        self.data_sources.append(source)
    
    def add_detection_logic(self, logic):
        """Ajouter une étape de logique de détection"""
        self.detection_logic.append(logic)
    
    def add_fp_mitigation(self, mitigation):
        """Ajouter une stratégie d'atténuation des faux positifs"""
        self.false_positive_mitigation.append(mitigation)
    
    def generate_rule(self, platform="splunk"):
        """Générer une règle pour la plate-form SIEM spécifiée"""
        rule = {
            'name': f"{self.technique} Detection",
            'description': f"Détecte la technique {self.technique} ({self.tactic})",
            'tactic': self.tactic,
            'technique': self.technique,
            'created': datetime.now().isoformat(),
            'data_sources': self.data_sources,
            'logic': self.detection_logic,
            'fp_mitigation': self.false_positive_mitigation
        }
        
        if platform == "splunk":
            return self._generate_splunk_rule(rule)
        elif platform == "elastic":
            return self._generate_elastic_rule(rule)
        elif platform == "kql":
            return self._generate_kql_rule(rule)
        
        return rule

import unittest
import json
from datetime import datetime, timedelta

class RuleTester:
    def __init__(self, rule, test_cases):
        self.rule = rule
        self.test_cases = test_cases
        self.results = []
    
    def run_test_case(self, test_case):
        """Exécuter un cas de test individuel contre la règle"""
        if self.rule['platform'] == 'splunk':
            return self._test_splunk_rule(test_case)
        elif self.rule['platform'] == 'elastic':
            return self._test_elastic_rule(test_case)
        elif self.rule['platform'] == 'kql':
            return self._test_kql_rule(test_case)
    
    def generate_test_data(self, scenario):
        """Générer des données de test pour un scénario d'attaque spécifique"""
        base_event = {
            'timestamp': datetime.now().isoformat(),
            'source': 'test_data'
        }
        
        if scenario == 'brute_force':
            return self._generate_brute_force_data(base_event)
        elif scenario == 'malware_execution':
            return self._generate_malware_data(base_event)
        elif scenario == 'data_exfiltration':
            return self._generate_exfiltration_data(base_event)
        
        return [base_event]
    
    def _generate_brute_force_data(self, base_event):
        """Générer des données de test pour une attaque par force brute"""
        events = []
        for i in range(15):
            event = base_event.copy()
            event.update({
                'event_code': '4625',
                'account_name': 'administrator',
                'ip_address': '192.168.1.100',
                'timestamp': (datetime.now() + timedelta(seconds=i*30)).isoformat()
            })
            events.append(event)
        return events
    
    def validate_rule_accuracy(self):
        """Valider les métriques de précision de la règle"""
        true_positives = 0
        false_positives = 0
        true_negatives = 0
        false_negatives = 0
        
        for test_case in self.test_cases:
            result = self.run_test_case(test_case)
            
            if test_case['expected'] and result['alert_triggered']:
                true_positives += 1
            elif not test_case['expected'] and result['alert_triggered']:
                false_positives += 1
            elif not test_case['expected'] and not result['alert_triggered']:
                true_negatives += 1
            elif test_case['expected'] and not result['alert_triggered']:
                false_negatives += 1
        
        precision = true_positives / (true_positives + false_positives) if (true_positives + false_positives) > 0 else 0
        recall = true_positives / (true_positives + false_negatives) if (true_positives + false_negatives) > 0 else 0
        
        return {
            'precision': precision,
            'recall': recall,
            'accuracy': (true_positives + true_negatives) / len(self.test_cases),
            'false_positive_rate': false_positives / len(self.test_cases),
            'false_negative_rate': false_negatives / len(self.test_cases)
        }

index=wineventlog (EventCode=4625 OR EventCode=4624 OR EventCode=4688)
| streamstats current=f window=5 last(EventCode) as prev_event by src_ip
| eval attack_stage=case(
    EventCode=4625 and prev_event=4625, "brute_force_in_progress",
    EventCode=4624 and prev_event=4625, "potential_compromise",
    EventCode=4688 and prev_event=4624, "post_compromise_activity"
)
| where isnotnull(attack_stage)
| transaction src_ip maxpause=5m
| where eventcount >= 10
| eval risk_score=case(attack_stage="post_compromise_activity", 90, attack_stage="potential_compromise", 70, 60)

from sklearn.ensemble import IsolationForest
import pandas as pd

class AnomalyDetector:
    def __init__(self):
        self.model = IsolationForest(contamination=0.1, random_state=42)
        self.is_trained = False
    
    def train(self, normal_data):
        """Entraîner le modèle sur les données de comportement normal"""
        features = self._extract_features(normal_data)
        self.model.fit(features)
        self.is_trained = True
    
    def detect_anomalies(self, new_data):
        """Détecter les anomalies dans les nouvelles données"""
        if not self.is_trained:
            raise ValueError("Le modèle doit d'abord être entraîné")
        
        features = self._extract_features(new_data)
        predictions = self.model.predict(features)
        return [p == -1 for p in predictions]  # -1 indique une anomalie
    
    def _extract_features(self, data):
        """Extraire les caractéristiques pour le modèle ML"""
        feature_df = pd.DataFrame()
        
        # Fréquence de connexion
        feature_df['logon_frequency'] = data.groupby('user')['timestamp'].transform('count')
        
        # Caractéristiques basées sur le temps
        feature_df['hour_of_day'] = pd.to_datetime(data['timestamp']).dt.hour
        feature_df['day_of_week'] = pd.to_datetime(data['timestamp']).dt.dayofweek
        
        # Caractéristiques réseau
        feature_df['unique_ips'] = data.groupby('user')['src_ip'].transform('nunique')
        
        return feature_df

rule_optimization:
  time_filters:
    - "Toujours appliquer d'abord les limites de temps"
    - "Utiliser earliest/latest dans Splunk"
    - "Limiter les fenêtres de recherche à la période nécessaire"
  
  index_selection:
    - "Spécifier l'index le plus spécifique"
    - "Éviter de rechercher tous les index"
    - "Utiliser des restrictions de type source"
  
  field_selection:
    - "Extraire uniquement les champs requis"
    - "Utiliser les wildcards avec modération"
    - "Optimiser les regex"
  
  aggregation:
    - "Utiliser les fenêtres de temps appropriées"
    - "Limiter les champs de regroupement"
    - "Optimiser l'utilisation de streamstats"