Risk-Based Alerting: AI-Powered Priority Scoring and Alert Consolidation#

1
# Alert Fatigue Analysis
2
class AlertFatigueAnalyzer:
3
    def __init__(self):
4
        self.alert_metrics = {
5
            'daily_volume': 11347,
6
            'false_positive_rate': 0.75,
7
            'avg_investigation_time': 23,  # minutes
8
            'analyst_capacity': 480,  # minutes per day
9
            'missed_critical_alerts': 0.12  # 12% of critical alerts missed
10
        }
11

12
    def calculate_alert_overload(self):
13
        """Calculate the severity of alert overload"""
14
        # Time required to investigate all alerts
15
        total_investigation_time = (
16
            self.alert_metrics['daily_volume'] *
17
            self.alert_metrics['avg_investigation_time']
18
        )
19

20
        # Available analyst time
21
        available_time = self.alert_metrics['analyst_capacity']
22

23
        # Overload factor
24
        overload_factor = total_investigation_time / available_time
25

26
        # Effective alerts (after accounting for fatigue)
27
        fatigue_factor = min(1.0, 1.0 / (overload_factor ** 0.5))
28
        effective_alerts_investigated = (
29
            self.alert_metrics['daily_volume'] * fatigue_factor
30
        )
31

32
        return {
33
            'overload_factor': overload_factor,
34
            'alerts_investigated': effective_alerts_investigated,
35
            'alerts_missed': self.alert_metrics['daily_volume'] - effective_alerts_investigated,
36
            'critical_alerts_missed': int(
37
                self.alert_metrics['missed_critical_alerts'] *
38
                self.alert_metrics['daily_volume']
39
            ),
40
            'wasted_time_hours': (
41
                self.alert_metrics['false_positive_rate'] *
42
                effective_alerts_investigated *
43
                self.alert_metrics['avg_investigation_time'] / 60
44
            )
45
        }

1
class RiskScoringEngine:
2
    def __init__(self):
3
        self.risk_factors = {
4
            'threat_severity': {
5
                'weight': 0.25,
6
                'calculator': self.calculate_threat_severity
7
            },
8
            'asset_criticality': {
9
                'weight': 0.20,
10
                'calculator': self.calculate_asset_criticality
11
            },
12
            'attack_sophistication': {
13
                'weight': 0.15,
14
                'calculator': self.calculate_attack_sophistication
15
            },
16
            'exploit_probability': {
17
                'weight': 0.15,
18
                'calculator': self.calculate_exploit_probability
19
            },
20
            'business_impact': {
21
                'weight': 0.15,
22
                'calculator': self.calculate_business_impact
23
            },
24
            'temporal_factors': {
25
                'weight': 0.10,
26
                'calculator': self.calculate_temporal_factors
27
            }
28
        }
29
        self.ml_enhancer = MLRiskEnhancer()
30

31
    def calculate_risk_score(self, alert):
32
        """Calculate comprehensive risk score for alert"""
33
        risk_components = {}
34
        weighted_score = 0
35

36
        # Calculate each risk factor
37
        for factor_name, factor_config in self.risk_factors.items():
38
            factor_score = factor_config['calculator'](alert)
39
            risk_components[factor_name] = factor_score
40
            weighted_score += factor_score * factor_config['weight']
41

42
        # ML enhancement
43
        ml_adjustment = self.ml_enhancer.predict_risk_adjustment(
44
            alert,
45
            risk_components
46
        )
47

48
        # Final score (0-100 scale)
49
        final_score = min(100, weighted_score * ml_adjustment * 100)
50

51
        return {
52
            'risk_score': final_score,
53
            'risk_level': self.categorize_risk(final_score),
54
            'components': risk_components,
55
            'ml_confidence': ml_adjustment,
56
            'priority': self.determine_priority(final_score, alert)
57
        }
58

59
    def calculate_threat_severity(self, alert):
60
        """Calculate threat severity component"""
61
        severity_scores = {
62
            'critical': 1.0,
63
            'high': 0.8,
64
            'medium': 0.5,
65
            'low': 0.3,
66
            'info': 0.1
67
        }
68

69
        base_severity = severity_scores.get(
70
            alert.get('severity', 'low'),
71
            0.1
72
        )
73

74
        # Adjust for threat intelligence
75
        if alert.get('threat_intel_match'):
76
            base_severity *= 1.5
77

78
        # Adjust for kill chain phase
79
        kill_chain_multipliers = {
80
            'initial_access': 1.2,
81
            'execution': 1.3,
82
            'persistence': 1.4,
83
            'privilege_escalation': 1.5,
84
            'defense_evasion': 1.3,
85
            'credential_access': 1.6,
86
            'lateral_movement': 1.5,
87
            'collection': 1.4,
88
            'exfiltration': 1.8,
89
            'impact': 2.0
90
        }
91

92
        kill_chain_phase = alert.get('kill_chain_phase')
93
        if kill_chain_phase:
94
            base_severity *= kill_chain_multipliers.get(kill_chain_phase, 1.0)
95

96
        return min(1.0, base_severity)

1
class AssetCriticalityCalculator:
2
    def __init__(self):
3
        self.asset_db = AssetDatabase()
4
        self.business_context = BusinessContextProvider()
5

6
    def calculate_asset_criticality(self, alert):
7
        """Calculate asset criticality score"""
8
        asset_id = alert.get('asset_id')
9
        if not asset_id:
10
            return 0.5  # Default medium criticality
11

12
        asset_info = self.asset_db.get_asset(asset_id)
13

14
        # Base criticality from asset classification
15
        base_criticality = {
16
            'critical': 1.0,
17
            'high': 0.8,
18
            'medium': 0.5,
19
            'low': 0.3,
20
            'test': 0.1
21
        }.get(asset_info.get('classification', 'medium'), 0.5)
22

23
        # Adjust for business context
24
        adjustments = []
25

26
        # Data sensitivity
27
        if asset_info.get('handles_pii'):
28
            adjustments.append(1.3)
29
        if asset_info.get('handles_financial_data'):
30
            adjustments.append(1.4)
31
        if asset_info.get('handles_ip'):
32
            adjustments.append(1.5)
33

34
        # Operational importance
35
        if asset_info.get('production'):
36
            adjustments.append(1.2)
37
        if asset_info.get('customer_facing'):
38
            adjustments.append(1.3)
39
        if asset_info.get('revenue_generating'):
40
            adjustments.append(1.4)
41

42
        # Compliance requirements
43
        if asset_info.get('compliance_scope'):
44
            adjustments.append(1.2)
45

46
        # Apply adjustments
47
        final_criticality = base_criticality
48
        for adjustment in adjustments:
49
            final_criticality *= adjustment
50

51
        return min(1.0, final_criticality)

1
class AlertConsolidationEngine:
2
    def __init__(self):
3
        self.clustering_model = self.build_clustering_model()
4
        self.similarity_calculator = SimilarityCalculator()
5
        self.correlation_window = 3600  # 1 hour
6

7
    def consolidate_alerts(self, alerts):
8
        """Consolidate related alerts into incidents"""
9
        # Extract features for clustering
10
        features = self.extract_features(alerts)
11

12
        # Perform clustering
13
        clusters = self.clustering_model.fit_predict(features)
14

15
        # Build consolidated incidents
16
        incidents = defaultdict(list)
17
        for alert, cluster_id in zip(alerts, clusters):
18
            incidents[cluster_id].append(alert)
19

20
        # Post-process incidents
21
        consolidated_incidents = []
22
        for cluster_id, cluster_alerts in incidents.items():
23
            incident = self.build_incident(cluster_alerts)
24
            consolidated_incidents.append(incident)
25

26
        return consolidated_incidents
27

28
    def build_incident(self, alerts):
29
        """Build consolidated incident from related alerts"""
30
        incident = {
31
            'id': self.generate_incident_id(),
32
            'alert_count': len(alerts),
33
            'alerts': alerts,
34
            'timespan': {
35
                'start': min(a['timestamp'] for a in alerts),
36
                'end': max(a['timestamp'] for a in alerts)
37
            },
38
            'risk_score': max(a.get('risk_score', 0) for a in alerts),
39
            'attack_pattern': self.identify_attack_pattern(alerts),
40
            'affected_assets': self.extract_affected_assets(alerts),
41
            'recommended_actions': self.generate_recommendations(alerts)
42
        }
43

44
        # Calculate incident priority
45
        incident['priority'] = self.calculate_incident_priority(incident)
46

47
        # Generate incident summary
48
        incident['summary'] = self.generate_incident_summary(incident)
49

50
        return incident
51

52
    def identify_attack_pattern(self, alerts):
53
        """Identify attack pattern from alert sequence"""
54
        # Extract kill chain phases
55
        kill_chain_sequence = [
56
            a.get('kill_chain_phase') for a in alerts
57
            if a.get('kill_chain_phase')
58
        ]
59

60
        # Known attack patterns
61
        attack_patterns = {
62
            'ransomware': [
63
                'initial_access',
64
                'execution',
65
                'privilege_escalation',
66
                'defense_evasion',
67
                'discovery',
68
                'lateral_movement',
69
                'collection',
70
                'impact'
71
            ],
72
            'data_theft': [
73
                'initial_access',
74
                'persistence',
75
                'credential_access',
76
                'discovery',
77
                'collection',
78
                'exfiltration'
79
            ],
80
            'apt_campaign': [
81
                'initial_access',
82
                'execution',
83
                'persistence',
84
                'privilege_escalation',
85
                'defense_evasion',
86
                'credential_access',
87
                'discovery',
88
                'lateral_movement'
89
            ]
90
        }
91

92
        # Match against known patterns
93
        best_match = None
94
        best_score = 0
95

96
        for pattern_name, pattern_sequence in attack_patterns.items():
97
            score = self.sequence_similarity(
98
                kill_chain_sequence,
99
                pattern_sequence
100
            )
101
            if score > best_score:
102
                best_score = score
103
                best_match = pattern_name
104

105
        return {
106
            'pattern': best_match,
107
            'confidence': best_score,
108
            'kill_chain_coverage': len(set(kill_chain_sequence))
109
        }

1
class MLAlertCorrelator:
2
    def __init__(self):
3
        self.correlation_model = self.build_correlation_model()
4
        self.feature_extractor = AlertFeatureExtractor()
5

6
    def build_correlation_model(self):
7
        """Build deep learning model for alert correlation"""
8
        model = Sequential([
9
            # Input layer
10
            Dense(256, activation='relu', input_shape=(128,)),
11
            Dropout(0.3),
12

13
            # Hidden layers
14
            Dense(128, activation='relu'),
15
            BatchNormalization(),
16
            Dropout(0.3),
17

18
            Dense(64, activation='relu'),
19
            BatchNormalization(),
20
            Dropout(0.2),
21

22
            # Output layer (correlation probability)
23
            Dense(1, activation='sigmoid')
24
        ])
25

26
        model.compile(
27
            optimizer='adam',
28
            loss='binary_crossentropy',
29
            metrics=['accuracy', 'precision', 'recall']
30
        )
31

32
        return model
33

34
    def correlate_alerts(self, alert1, alert2):
35
        """Determine if two alerts are correlated"""
36
        # Extract features
37
        features1 = self.feature_extractor.extract(alert1)
38
        features2 = self.feature_extractor.extract(alert2)
39

40
        # Combine features
41
        combined_features = np.concatenate([
42
            features1,
43
            features2,
44
            self.calculate_similarity_features(alert1, alert2)
45
        ])
46

47
        # Predict correlation
48
        correlation_prob = self.correlation_model.predict(
49
            combined_features.reshape(1, -1)
50
        )[0][0]
51

52
        return {
53
            'correlated': correlation_prob > 0.7,
54
            'confidence': correlation_prob,
55
            'correlation_type': self.determine_correlation_type(
56
                alert1,
57
                alert2,
58
                correlation_prob
59
            )
60
        }

1
<!-- Risk-Based Priority Rules -->
2
<group name="risk_based_alerting">
3
  <!-- Critical Risk Alert -->
4
  <rule id="900001" level="15">
5
    <if_sid>100000</if_sid>
6
    <field name="risk_score" compare=">=">85</field>
7
    <field name="asset_criticality" compare=">=">0.8</field>
8
    <description>Critical Risk: Immediate response required</description>
9
    <options>alert_by_email,alert_by_sms</options>
10
    <group>critical_risk,priority_1</group>
11
  </rule>
12

13
  <!-- High Risk with Active Exploitation -->
14
  <rule id="900002" level="14">
15
    <if_sid>100000</if_sid>
16
    <field name="risk_score" compare=">=">70</field>
17
    <field name="threat_intel.actively_exploited">true</field>
18
    <description>High Risk: Active exploitation detected</description>
19
    <group>high_risk,priority_2</group>
20
  </rule>
21

22
  <!-- Consolidated Attack Pattern -->
23
  <rule id="900003" level="13">
24
    <if_sid>100000</if_sid>
25
    <field name="incident.alert_count" compare=">=">5</field>
26
    <field name="incident.attack_pattern.confidence" compare=">=">0.8</field>
27
    <description>Attack Pattern Detected: Multi-stage attack in progress</description>
28
    <group>attack_pattern,priority_2</group>
29
  </rule>
30

31
  <!-- Business Critical Asset Compromise -->
32
  <rule id="900004" level="14">
33
    <if_sid>100000</if_sid>
34
    <field name="asset.business_critical">true</field>
35
    <field name="alert.category">compromise</field>
36
    <description>Business Critical: Revenue-impacting system compromised</description>
37
    <group>business_impact,priority_1</group>
38
  </rule>
39
</group>

1
class DynamicPriorityEngine:
2
    def __init__(self):
3
        self.priority_factors = {
4
            'risk_score': 0.3,
5
            'business_impact': 0.25,
6
            'threat_velocity': 0.15,
7
            'asset_exposure': 0.15,
8
            'historical_accuracy': 0.15
9
        }
10
        self.context_provider = ContextProvider()
11

12
    def calculate_priority(self, incident):
13
        """Calculate dynamic priority for incident"""
14
        priority_score = 0
15
        priority_factors = {}
16

17
        # Risk score factor
18
        risk_factor = incident['risk_score'] / 100
19
        priority_factors['risk'] = risk_factor
20
        priority_score += risk_factor * self.priority_factors['risk_score']
21

22
        # Business impact factor
23
        business_impact = self.calculate_business_impact(incident)
24
        priority_factors['business'] = business_impact
25
        priority_score += business_impact * self.priority_factors['business_impact']
26

27
        # Threat velocity (how fast is it spreading)
28
        velocity = self.calculate_threat_velocity(incident)
29
        priority_factors['velocity'] = velocity
30
        priority_score += velocity * self.priority_factors['threat_velocity']
31

32
        # Asset exposure (how exposed is the asset)
33
        exposure = self.calculate_asset_exposure(incident)
34
        priority_factors['exposure'] = exposure
35
        priority_score += exposure * self.priority_factors['asset_exposure']
36

37
        # Historical accuracy (how accurate have similar alerts been)
38
        accuracy = self.get_historical_accuracy(incident)
39
        priority_factors['accuracy'] = accuracy
40
        priority_score += accuracy * self.priority_factors['historical_accuracy']
41

42
        # Apply contextual adjustments
43
        context_multiplier = self.get_context_multiplier(incident)
44
        final_priority = priority_score * context_multiplier
45

46
        return {
47
            'priority_score': final_priority,
48
            'priority_level': self.score_to_priority_level(final_priority),
49
            'factors': priority_factors,
50
            'context_multiplier': context_multiplier,
51
            'sla': self.determine_sla(final_priority)
52
        }
53

54
    def calculate_threat_velocity(self, incident):
55
        """Calculate how fast the threat is spreading"""
56
        alerts = incident['alerts']
57

58
        if len(alerts) < 2:
59
            return 0.5
60

61
        # Sort by timestamp
62
        sorted_alerts = sorted(alerts, key=lambda x: x['timestamp'])
63

64
        # Calculate spread rate
65
        time_span = (
66
            sorted_alerts[-1]['timestamp'] -
67
            sorted_alerts[0]['timestamp']
68
        ).total_seconds()
69

70
        if time_span == 0:
71
            return 1.0  # Instantaneous spread
72

73
        # Assets affected per hour
74
        unique_assets = len(set(a.get('asset_id') for a in alerts))
75
        spread_rate = (unique_assets / time_span) * 3600
76

77
        # Normalize (0-1 scale)
78
        normalized_velocity = min(1.0, spread_rate / 10)  # 10 assets/hour = max
79

80
        return normalized_velocity

1
class AlertSuppressionEngine:
2
    def __init__(self):
3
        self.suppression_rules = []
4
        self.ml_suppressor = MLAlertSuppressor()
5
        self.feedback_tracker = FeedbackTracker()
6

7
    def should_suppress_alert(self, alert):
8
        """Determine if alert should be suppressed"""
9
        suppression_result = {
10
            'suppress': False,
11
            'reason': None,
12
            'confidence': 0,
13
            'alternative_action': None
14
        }
15

16
        # Check suppression rules
17
        for rule in self.suppression_rules:
18
            if self.evaluate_suppression_rule(rule, alert):
19
                suppression_result['suppress'] = True
20
                suppression_result['reason'] = rule['name']
21
                suppression_result['confidence'] = rule['confidence']
22
                break
23

24
        # ML-based suppression
25
        if not suppression_result['suppress']:
26
            ml_result = self.ml_suppressor.evaluate(alert)
27
            if ml_result['suppress_probability'] > 0.85:
28
                suppression_result['suppress'] = True
29
                suppression_result['reason'] = 'ML prediction'
30
                suppression_result['confidence'] = ml_result['suppress_probability']
31

32
        # Check if similar alerts were false positives
33
        if not suppression_result['suppress']:
34
            fp_rate = self.feedback_tracker.get_false_positive_rate(alert)
35
            if fp_rate > 0.9:
36
                suppression_result['suppress'] = True
37
                suppression_result['reason'] = 'High false positive rate'
38
                suppression_result['confidence'] = fp_rate
39

40
        # Determine alternative action
41
        if suppression_result['suppress']:
42
            suppression_result['alternative_action'] = self.determine_alternative_action(
43
                alert,
44
                suppression_result['reason']
45
            )
46

47
        return suppression_result
48

49
    def create_adaptive_suppression_rule(self, alert_pattern):
50
        """Create suppression rule based on analyst feedback"""
51
        rule = {
52
            'id': self.generate_rule_id(),
53
            'name': f"Auto-generated suppression for {alert_pattern['type']}",
54
            'conditions': self.extract_conditions(alert_pattern),
55
            'confidence': alert_pattern['false_positive_rate'],
56
            'created_at': datetime.now(),
57
            'expires_at': datetime.now() + timedelta(days=30),
58
            'review_count': 0
59
        }
60

61
        # Validate rule impact
62
        impact = self.evaluate_rule_impact(rule)
63

64
        if impact['suppression_rate'] < 0.1:  # Less than 10% suppression
65
            rule['approved'] = True
66
            self.suppression_rules.append(rule)
67
        else:
68
            # Requires manual review
69
            rule['approved'] = False
70
            rule['pending_review'] = True
71

72
        return rule

1
class ContextualEnrichmentEngine:
2
    def __init__(self):
3
        self.enrichment_sources = {
4
            'threat_intel': ThreatIntelEnricher(),
5
            'asset_context': AssetContextEnricher(),
6
            'user_behavior': UserBehaviorEnricher(),
7
            'network_context': NetworkContextEnricher(),
8
            'historical_context': HistoricalContextEnricher()
9
        }
10

11
    def enrich_alert_with_context(self, alert):
12
        """Enrich alert with comprehensive context"""
13
        enriched_alert = alert.copy()
14
        enrichment_metadata = {
15
            'sources_used': [],
16
            'enrichment_time': 0,
17
            'confidence_boost': 0
18
        }
19

20
        start_time = time.time()
21

22
        # Parallel enrichment
23
        with ThreadPoolExecutor(max_workers=5) as executor:
24
            futures = {}
25

26
            for source_name, enricher in self.enrichment_sources.items():
27
                future = executor.submit(enricher.enrich, alert)
28
                futures[source_name] = future
29

30
            # Collect results
31
            for source_name, future in futures.items():
32
                try:
33
                    enrichment_data = future.result(timeout=2)
34
                    enriched_alert[f'context_{source_name}'] = enrichment_data
35
                    enrichment_metadata['sources_used'].append(source_name)
36

37
                    # Update confidence based on enrichment
38
                    if enrichment_data.get('confidence_modifier'):
39
                        enrichment_metadata['confidence_boost'] += (
40
                            enrichment_data['confidence_modifier']
41
                        )
42
                except Exception as e:
43
                    logger.error(f"Enrichment failed for {source_name}: {e}")
44

45
        enrichment_metadata['enrichment_time'] = time.time() - start_time
46
        enriched_alert['_enrichment_metadata'] = enrichment_metadata
47

48
        # Recalculate risk score with enriched context
49
        enriched_alert['risk_score'] = self.recalculate_risk_score(enriched_alert)
50

51
        return enriched_alert

1
class RiskBasedDashboard:
2
    def __init__(self):
3
        self.visualization_engine = VisualizationEngine()
4
        self.real_time_updater = RealTimeUpdater()
5

6
    def generate_dashboard_data(self):
7
        """Generate risk-based dashboard data"""
8
        dashboard = {
9
            'timestamp': datetime.now(),
10
            'summary': self.generate_summary_metrics(),
11
            'priority_distribution': self.get_priority_distribution(),
12
            'risk_heatmap': self.generate_risk_heatmap(),
13
            'incident_timeline': self.generate_incident_timeline(),
14
            'analyst_workload': self.calculate_analyst_workload(),
15
            'effectiveness_metrics': self.calculate_effectiveness_metrics()
16
        }
17

18
        return dashboard
19

20
    def generate_risk_heatmap(self):
21
        """Generate risk heatmap visualization data"""
22
        query = {
23
            "size": 0,
24
            "aggs": {
25
                "risk_by_asset": {
26
                    "terms": {
27
                        "field": "asset.category",
28
                        "size": 20
29
                    },
30
                    "aggs": {
31
                        "risk_over_time": {
32
                            "date_histogram": {
33
                                "field": "@timestamp",
34
                                "calendar_interval": "1h"
35
                            },
36
                            "aggs": {
37
                                "avg_risk": {
38
                                    "avg": {
39
                                        "field": "risk_score"
40
                                    }
41
                                },
42
                                "max_risk": {
43
                                    "max": {
44
                                        "field": "risk_score"
45
                                    }
46
                                }
47
                            }
48
                        }
49
                    }
50
                }
51
            }
52
        }
53

54
        results = self.es.search(index="wazuh-alerts-*", body=query)
55

56
        # Transform to heatmap format
57
        heatmap_data = []
58
        for asset_bucket in results['aggregations']['risk_by_asset']['buckets']:
59
            asset_category = asset_bucket['key']
60

61
            for time_bucket in asset_bucket['risk_over_time']['buckets']:
62
                heatmap_data.append({
63
                    'asset_category': asset_category,
64
                    'timestamp': time_bucket['key_as_string'],
65
                    'avg_risk': time_bucket['avg_risk']['value'],
66
                    'max_risk': time_bucket['max_risk']['value'],
67
                    'alert_count': time_bucket['doc_count']
68
                })
69

70
        return heatmap_data

1
class AlertFatigueMonitor:
2
    def __init__(self):
3
        self.analyst_tracker = AnalystActivityTracker()
4

5
    def calculate_fatigue_metrics(self, time_range='24h'):
6
        """Calculate alert fatigue metrics"""
7
        metrics = {
8
            'total_alerts': 0,
9
            'alerts_viewed': 0,
10
            'alerts_investigated': 0,
11
            'alerts_dismissed': 0,
12
            'avg_time_to_view': 0,
13
            'avg_investigation_time': 0,
14
            'false_positive_rate': 0,
15
            'analyst_satisfaction': 0
16
        }
17

18
        # Get analyst activity data
19
        activity_data = self.analyst_tracker.get_activity(time_range)
20

21
        # Calculate metrics
22
        metrics['total_alerts'] = activity_data['total_alerts']
23
        metrics['alerts_viewed'] = activity_data['alerts_viewed']
24
        metrics['alerts_investigated'] = activity_data['alerts_investigated']
25
        metrics['alerts_dismissed'] = activity_data['alerts_dismissed']
26

27
        # View and investigation rates
28
        if metrics['total_alerts'] > 0:
29
            metrics['view_rate'] = metrics['alerts_viewed'] / metrics['total_alerts']
30
            metrics['investigation_rate'] = (
31
                metrics['alerts_investigated'] / metrics['total_alerts']
32
            )
33
            metrics['dismissal_rate'] = (
34
                metrics['alerts_dismissed'] / metrics['alerts_viewed']
35
            )
36

37
        # Time metrics
38
        metrics['avg_time_to_view'] = np.mean(
39
            activity_data['time_to_view_list']
40
        )
41
        metrics['avg_investigation_time'] = np.mean(
42
            activity_data['investigation_times']
43
        )
44

45
        # Fatigue indicators
46
        metrics['fatigue_score'] = self.calculate_fatigue_score(metrics)
47

48
        return metrics
49

50
    def calculate_fatigue_score(self, metrics):
51
        """Calculate overall fatigue score"""
52
        # Factors indicating fatigue
53
        factors = {
54
            'high_dismissal_rate': min(1.0, metrics.get('dismissal_rate', 0) / 0.5),
55
            'low_investigation_rate': 1.0 - metrics.get('investigation_rate', 0),
56
            'increasing_time_to_view': self.check_increasing_trend('time_to_view'),
57
            'high_volume': min(1.0, metrics['total_alerts'] / 10000)
58
        }
59

60
        # Weighted fatigue score
61
        weights = {
62
            'high_dismissal_rate': 0.3,
63
            'low_investigation_rate': 0.3,
64
            'increasing_time_to_view': 0.2,
65
            'high_volume': 0.2
66
        }
67

68
        fatigue_score = sum(
69
            factors[factor] * weights[factor]
70
            for factor in factors
71
        )
72

73
        return fatigue_score

1
class RiskBasedAlertingRollout:
2
    def __init__(self):
3
        self.phases = [
4
            {
5
                'name': 'Phase 1: Baseline Collection',
6
                'duration_days': 30,
7
                'actions': [
8
                    'Deploy risk scoring in shadow mode',
9
                    'Collect baseline metrics',
10
                    'Train ML models',
11
                    'Gather analyst feedback'
12
                ]
13
            },
14
            {
15
                'name': 'Phase 2: Pilot Deployment',
16
                'duration_days': 30,
17
                'actions': [
18
                    'Enable for 10% of alerts',
19
                    'Monitor false positive reduction',
20
                    'Tune risk factors',
21
                    'Train analysts on new system'
22
                ]
23
            },
24
            {
25
                'name': 'Phase 3: Gradual Expansion',
26
                'duration_days': 45,
27
                'actions': [
28
                    'Expand to 50% of alerts',
29
                    'Enable alert consolidation',
30
                    'Implement suppression rules',
31
                    'Monitor analyst satisfaction'
32
                ]
33
            },
34
            {
35
                'name': 'Phase 4: Full Deployment',
36
                'duration_days': 30,
37
                'actions': [
38
                    'Enable for all alerts',
39
                    'Activate all features',
40
                    'Continuous optimization',
41
                    'Regular effectiveness reviews'
42
                ]
43
            }
44
        ]

1
<!-- Risk-Based Alerting Configuration -->
2
<ossec_config>
3
  <risk_based_alerting>
4
    <enabled>yes</enabled>
5

6
    <!-- Risk Scoring Configuration -->
7
    <risk_scoring>
8
      <factors>
9
        <factor name="threat_severity" weight="0.25"/>
10
        <factor name="asset_criticality" weight="0.20"/>
11
        <factor name="attack_sophistication" weight="0.15"/>
12
        <factor name="exploit_probability" weight="0.15"/>
13
        <factor name="business_impact" weight="0.15"/>
14
        <factor name="temporal_factors" weight="0.10"/>
15
      </factors>
16

17
      <ml_enhancement>
18
        <enabled>yes</enabled>
19
        <model_path>/var/ossec/models/risk_scoring_model.pkl</model_path>
20
        <update_frequency>daily</update_frequency>
21
      </ml_enhancement>
22
    </risk_scoring>
23

24
    <!-- Alert Consolidation -->
25
    <consolidation>
26
      <enabled>yes</enabled>
27
      <correlation_window>3600</correlation_window>
28
      <min_alerts_for_incident>3</min_alerts_for_incident>
29
      <ml_correlation>yes</ml_correlation>
30
    </consolidation>
31

32
    <!-- Suppression Rules -->
33
    <suppression>
34
      <enabled>yes</enabled>
35
      <ml_suppression>yes</ml_suppression>
36
      <false_positive_threshold>0.85</false_positive_threshold>
37
      <auto_create_rules>yes</auto_create_rules>
38
    </suppression>
39

40
    <!-- Priority Assignment -->
41
    <prioritization>
42
      <dynamic>yes</dynamic>
43
      <business_context>yes</business_context>
44
      <sla_enforcement>yes</sla_enforcement>
45
    </prioritization>
46
  </risk_based_alerting>
47
</ossec_config>

1
{
2
  "risk_based_alerting_metrics": {
3
    "alert_reduction": {
4
      "total_alerts_before": 11347,
5
      "total_alerts_after": 2269,
6
      "reduction_percentage": "80%",
7
      "high_priority_alerts": 156,
8
      "incidents_created": 89
9
    },
10
    "accuracy_improvements": {
11
      "false_positive_rate_before": "75%",
12
      "false_positive_rate_after": "12%",
13
      "true_positive_increase": "34%",
14
      "missed_critical_alerts": "0.3%"
15
    },
16
    "analyst_efficiency": {
17
      "avg_investigation_time_before": "23 min",
18
      "avg_investigation_time_after": "8 min",
19
      "alerts_per_analyst_per_day_before": 21,
20
      "alerts_per_analyst_per_day_after": 67,
21
      "analyst_satisfaction_score": "8.4/10"
22
    },
23
    "business_impact": {
24
      "mttr_improvement": "68%",
25
      "critical_incidents_missed_reduction": "94%",
26
      "compliance_violation_detection": "99.2%",
27
      "cost_per_alert_reduction": "73%"
28
    },
29
    "system_performance": {
30
      "risk_scoring_latency": "12ms",
31
      "consolidation_processing_time": "145ms",
32
      "ml_inference_time": "34ms",
33
      "dashboard_load_time": "1.2s"
34
    }
35
  }
36
}

1
class FeedbackLoopManager:
2
    def __init__(self):
3
        self.feedback_collector = FeedbackCollector()
4
        self.model_updater = ModelUpdater()
5
        self.rule_optimizer = RuleOptimizer()
6

7
    def process_analyst_feedback(self, feedback):
8
        """Process analyst feedback to improve system"""
9
        improvement_actions = []
10

11
        # Update ML models based on feedback
12
        if feedback['type'] == 'false_positive':
13
            self.model_updater.add_false_positive_sample(
14
                feedback['alert'],
15
                feedback['reason']
16
            )
17
            improvement_actions.append('Added to FP training set')
18

19
        elif feedback['type'] == 'missed_threat':
20
            self.model_updater.add_false_negative_sample(
21
                feedback['alert'],
22
                feedback['severity']
23
            )
24
            improvement_actions.append('Added to FN training set')
25

26
        # Optimize rules
27
        if feedback.get('suggested_rule_change'):
28
            rule_change = self.rule_optimizer.evaluate_suggestion(
29
                feedback['suggested_rule_change']
30
            )
31
            if rule_change['approved']:
32
                self.rule_optimizer.apply_change(rule_change)
33
                improvement_actions.append(f"Applied rule change: {rule_change['id']}")
34

35
        # Update risk factors
36
        if feedback.get('risk_score_feedback'):
37
            self.update_risk_factors(feedback['risk_score_feedback'])
38
            improvement_actions.append('Updated risk factor weights')
39

40
        return {
41
            'feedback_id': feedback['id'],
42
            'processed_at': datetime.now(),
43
            'actions_taken': improvement_actions
44
        }