1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import matplotlib.pyplot as plt

class SecurityMetrics:
    def __init__(self):
        self.metrics = {
            'operational': {
                'incident_count': 0,
                'vulnerability_count': 0,
                'patch_compliance': 0.0,
                'user_training_completion': 0.0,
                'system_uptime': 0.0
            },
            'strategic': {
                'risk_reduction_rate': 0.0,
                'budget_utilization': 0.0,
                'policy_compliance': 0.0,
                'stakeholder_satisfaction': 0.0,
                'security_maturity': 0.0
            },
            'tactical': {
                'control_effectiveness': 0.0,
                'threat_detection_rate': 0.0,
                'response_time': 0.0,
                'false_positive_rate': 0.0,
                'recovery_time': 0.0
            }
        }
        
        self.historical_data = {}
        self.thresholds = {
            'incident_count': {'warning': 10, 'critical': 20},
            'vulnerability_count': {'warning': 50, 'critical': 100},
            'patch_compliance': {'warning': 0.8, 'critical': 0.6},
            'response_time': {'warning': 4, 'critical': 8}  # hours
        }
    
    def update_metric(self, category, metric_name, value, timestamp=None):
        """Update metric"""
        if timestamp is None:
            timestamp = datetime.now()
        
        if category in self.metrics and metric_name in self.metrics[category]:
            self.metrics[category][metric_name] = value
            
            # Save historical data
            if category not in self.historical_data:
                self.historical_data[category] = {}
            if metric_name not in self.historical_data[category]:
                self.historical_data[category][metric_name] = []
            
            self.historical_data[category][metric_name].append({
                'value': value,
                'timestamp': timestamp
            })
    
    def get_metric_status(self, category, metric_name):
        """Get metric status"""
        if category not in self.metrics or metric_name not in self.metrics[category]:
            return 'Unknown'
        
        value = self.metrics[category][metric_name]
        
        if metric_name in self.thresholds:
            thresholds = self.thresholds[metric_name]
            if value >= thresholds['critical']:
                return 'Critical'
            elif value >= thresholds['warning']:
                return 'Warning'
            else:
                return 'Normal'
        
        return 'Normal'
    
    def calculate_trend(self, category, metric_name, days=30):
        """Calculate metric trend"""
        if (category not in self.historical_data or 
            metric_name not in self.historical_data[category]):
            return None
        
        data = self.historical_data[category][metric_name]
        cutoff_date = datetime.now() - timedelta(days=days)
        
        recent_data = [d for d in data if d['timestamp'] >= cutoff_date]
        
        if len(recent_data) < 2:
            return None
        
        values = [d['value'] for d in recent_data]
        
        # Calculate trend using simple linear regression
        x = np.arange(len(values))
        y = np.array(values)
        
        if len(x) > 1:
            slope = np.polyfit(x, y, 1)[0]
            return {
                'trend': 'Increasing' if slope > 0 else 'Decreasing' if slope < 0 else 'Stable',
                'slope': slope,
                'data_points': len(values)
            }
        
        return None
    
    def generate_metrics_report(self):
        """Generate metrics report"""
        report = {
            'timestamp': datetime.now(),
            'summary': {},
            'alerts': [],
            'trends': {}
        }
        
        # Summary by category
        for category, metrics in self.metrics.items():
            category_summary = {
                'total_metrics': len(metrics),
                'normal': 0,
                'warning': 0,
                'critical': 0
            }
            
            for metric_name in metrics.keys():
                status = self.get_metric_status(category, metric_name)
                category_summary[status.lower()] += 1
                
                if status in ['Warning', 'Critical']:
                    report['alerts'].append({
                        'category': category,
                        'metric': metric_name,
                        'status': status,
                        'value': metrics[metric_name]
                    })
            
            report['summary'][category] = category_summary
        
        # Trends
        for category in self.metrics.keys():
            report['trends'][category] = {}
            for metric_name in self.metrics[category].keys():
                trend = self.calculate_trend(category, metric_name)
                if trend:
                    report['trends'][category][metric_name] = trend
        
        return report

# Usage example
security_metrics = SecurityMetrics()

# Update metrics
security_metrics.update_metric('operational', 'incident_count', 5)
security_metrics.update_metric('operational', 'vulnerability_count', 25)
security_metrics.update_metric('operational', 'patch_compliance', 0.85)
security_metrics.update_metric('strategic', 'risk_reduction_rate', 0.15)
security_metrics.update_metric('tactical', 'response_time', 2.5)

# Generate report
report = security_metrics.generate_metrics_report()
print(f"Metrics report: {report['summary']}")

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105

class SecurityAlertSystem:
    def __init__(self):
        self.alerts = {}
        self.alert_rules = {}
        self.notification_channels = {
            'email': [],
            'sms': [],
            'slack': [],
            'dashboard': True
        }
    
    def define_alert_rule(self, rule_id, rule_config):
        """Define alert rule"""
        self.alert_rules[rule_id] = {
            'name': rule_config['name'],
            'condition': rule_config['condition'],
            'threshold': rule_config['threshold'],
            'severity': rule_config['severity'],
            'enabled': rule_config.get('enabled', True),
            'notification_channels': rule_config.get('channels', ['email'])
        }
    
    def check_alerts(self, metric_data):
        """Check alerts based on metrics"""
        triggered_alerts = []
        
        for rule_id, rule in self.alert_rules.items():
            if not rule['enabled']:
                continue
            
            # Evaluate condition
            if self.evaluate_condition(rule['condition'], metric_data, rule['threshold']):
                alert = {
                    'alert_id': f"ALERT-{len(self.alerts) + 1}",
                    'rule_id': rule_id,
                    'rule_name': rule['name'],
                    'severity': rule['severity'],
                    'timestamp': datetime.now(),
                    'metric_data': metric_data,
                    'status': 'Active'
                }
                
                self.alerts[alert['alert_id']] = alert
                triggered_alerts.append(alert)
                
                # Send notifications
                self.send_notifications(alert, rule['notification_channels'])
        
        return triggered_alerts
    
    def evaluate_condition(self, condition, data, threshold):
        """Evaluate alert condition"""
        if condition == 'greater_than':
            return data > threshold
        elif condition == 'less_than':
            return data < threshold
        elif condition == 'equals':
            return data == threshold
        elif condition == 'not_equals':
            return data != threshold
        else:
            return False
    
    def send_notifications(self, alert, channels):
        """Send notifications"""
        for channel in channels:
            if channel in self.notification_channels:
                # Simulate notification sending
                print(f"Notification {channel}: {alert['rule_name']} - {alert['severity']}")
    
    def get_active_alerts(self):
        """Get active alerts"""
        return [alert for alert in self.alerts.values() if alert['status'] == 'Active']
    
    def resolve_alert(self, alert_id, resolution_notes):
        """Resolve alert"""
        if alert_id in self.alerts:
            self.alerts[alert_id]['status'] = 'Resolved'
            self.alerts[alert_id]['resolution_notes'] = resolution_notes
            self.alerts[alert_id]['resolved_at'] = datetime.now()

# Usage example
alert_system = SecurityAlertSystem()

# Define alert rules
alert_system.define_alert_rule('RULE-001', {
    'name': 'High Incident Count',
    'condition': 'greater_than',
    'threshold': 10,
    'severity': 'High',
    'channels': ['email', 'slack']
})

alert_system.define_alert_rule('RULE-002', {
    'name': 'Low Patch Compliance',
    'condition': 'less_than',
    'threshold': 0.8,
    'severity': 'Medium',
    'channels': ['email']
})

# Check alerts
metric_data = {'incident_count': 15, 'patch_compliance': 0.75}
triggered = alert_system.check_alerts(metric_data)
print(f"Alerts triggered: {len(triggered)}")

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147

class SecurityReviewSystem:
    def __init__(self):
        self.reviews = {}
        self.review_templates = {
            'monthly': {
                'name': 'Monthly Security Review',
                'scope': ['Operational metrics', 'Incidents', 'Vulnerabilities'],
                'participants': ['CISO', 'Security_Manager', 'IT_Manager'],
                'duration': '2 hours'
            },
            'quarterly': {
                'name': 'Quarterly Security Review',
                'scope': ['Strategy', 'Budget', 'Risks', 'Controls'],
                'participants': ['CISO', 'CEO', 'CTO', 'CFO', 'Legal'],
                'duration': '4 hours'
            },
            'annual': {
                'name': 'Annual Security Review',
                'scope': ['Complete strategy', 'Compliance', 'Maturity', 'Roadmap'],
                'participants': ['Board', 'CISO', 'CEO', 'CTO', 'CFO', 'Legal', 'Audit'],
                'duration': '8 hours'
            }
        }
    
    def schedule_review(self, review_id, review_type, scheduled_date, participants):
        """Schedule review"""
        template = self.review_templates.get(review_type, {})
        
        self.reviews[review_id] = {
            'review_id': review_id,
            'type': review_type,
            'name': template.get('name', 'Security Review'),
            'scheduled_date': scheduled_date,
            'participants': participants,
            'scope': template.get('scope', []),
            'duration': template.get('duration', '2 hours'),
            'status': 'Scheduled',
            'agenda': [],
            'minutes': None,
            'action_items': []
        }
    
    def create_review_agenda(self, review_id, custom_items=None):
        """Create review agenda"""
        if review_id not in self.reviews:
            return None
        
        review = self.reviews[review_id]
        template = self.review_templates.get(review['type'], {})
        
        agenda = [
            'Opening and welcome',
            'Review of previous minutes',
            'Current security status'
        ]
        
        # Add template items
        agenda.extend(template.get('scope', []))
        
        # Add custom items
        if custom_items:
            agenda.extend(custom_items)
        
        agenda.extend([
            'Discussion of issues and opportunities',
            'Actions and responsibilities',
            'Next review'
        ])
        
        review['agenda'] = agenda
        return agenda
    
    def conduct_review(self, review_id, findings, decisions, action_items):
        """Conduct review"""
        if review_id not in self.reviews:
            return None
        
        review = self.reviews[review_id]
        review['status'] = 'Completed'
        review['completed_date'] = datetime.now()
        review['findings'] = findings
        review['decisions'] = decisions
        review['action_items'] = action_items
        
        # Generate minutes
        review['minutes'] = {
            'participants': review['participants'],
            'findings': findings,
            'decisions': decisions,
            'action_items': action_items,
            'next_review': self.calculate_next_review(review['type'])
        }
        
        return review['minutes']
    
    def calculate_next_review(self, review_type):
        """Calculate next review"""
        if review_type == 'monthly':
            return datetime.now() + timedelta(days=30)
        elif review_type == 'quarterly':
            return datetime.now() + timedelta(days=90)
        elif review_type == 'annual':
            return datetime.now() + timedelta(days=365)
        else:
            return datetime.now() + timedelta(days=30)
    
    def get_pending_action_items(self, review_id):
        """Get pending action items"""
        if review_id not in self.reviews:
            return []
        
        review = self.reviews[review_id]
        return [item for item in review['action_items'] if item.get('status') != 'Completed']

# Usage example
review_system = SecurityReviewSystem()

# Schedule monthly review
review_system.schedule_review(
    'REV-001',
    'monthly',
    datetime(2025, 11, 1),
    ['CISO', 'Security_Manager', 'IT_Manager']
)

# Create agenda
agenda = review_system.create_review_agenda('REV-001', [
    'Review of month's security incident',
    'Status of control implementation'
])

# Conduct review
findings = [
    '20% increase in phishing incidents',
    'Delay in firewall implementation'
]
decisions = [
    'Implement additional phishing training',
    'Accelerate firewall implementation'
]
action_items = [
    {'item': 'Phishing training', 'owner': 'Security_Manager', 'due_date': '2025-11-15'},
    {'item': 'Implement firewall', 'owner': 'IT_Manager', 'due_date': '2025-11-30'}
]

minutes = review_system.conduct_review('REV-001', findings, decisions, action_items)
print(f"Review minutes: {minutes['decisions']}")

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106

class SecurityDashboard:
    def __init__(self):
        self.widgets = {}
        self.layouts = {
            'executive': ['risk_summary', 'incident_trends', 'compliance_status'],
            'operational': ['incident_alerts', 'vulnerability_status', 'system_health'],
            'tactical': ['threat_intelligence', 'control_effectiveness', 'response_metrics']
        }
    
    def create_widget(self, widget_id, widget_type, data_source, config):
        """Create dashboard widget"""
        self.widgets[widget_id] = {
            'widget_id': widget_id,
            'type': widget_type,
            'data_source': data_source,
            'config': config,
            'last_updated': datetime.now(),
            'status': 'Active'
        }
    
    def generate_dashboard_data(self, layout_type):
        """Generate dashboard data"""
        if layout_type not in self.layouts:
            return None
        
        dashboard_data = {
            'layout': layout_type,
            'widgets': [],
            'timestamp': datetime.now(),
            'refresh_interval': 300  # 5 minutes
        }
        
        for widget_id in self.layouts[layout_type]:
            if widget_id in self.widgets:
                widget = self.widgets[widget_id]
                widget_data = self.get_widget_data(widget)
                dashboard_data['widgets'].append({
                    'widget_id': widget_id,
                    'type': widget['type'],
                    'data': widget_data,
                    'last_updated': widget['last_updated']
                })
        
        return dashboard_data
    
    def get_widget_data(self, widget):
        """Get widget data"""
        widget_type = widget['type']
        
        if widget_type == 'risk_summary':
            return {
                'total_risks': 25,
                'high_risks': 5,
                'medium_risks': 12,
                'low_risks': 8,
                'trend': 'Decreasing'
            }
        elif widget_type == 'incident_trends':
            return {
                'current_month': 15,
                'previous_month': 12,
                'trend': 'Increasing',
                'severity_breakdown': {
                    'Critical': 2,
                    'High': 5,
                    'Medium': 6,
                    'Low': 2
                }
            }
        elif widget_type == 'compliance_status':
            return {
                'overall_score': 85,
                'iso_27001': 90,
                'gdpr': 80,
                'pci_dss': 85,
                'trend': 'Stable'
            }
        else:
            return {'message': 'No data available'}
    
    def create_executive_dashboard(self):
        """Create executive dashboard"""
        # Create executive widgets
        self.create_widget('risk_summary', 'summary_card', 'risk_management', {
            'title': 'Risk Summary',
            'color': 'blue'
        })
        
        self.create_widget('incident_trends', 'line_chart', 'incident_management', {
            'title': 'Incident Trends',
            'x_axis': 'time',
            'y_axis': 'count'
        })
        
        self.create_widget('compliance_status', 'gauge', 'compliance', {
            'title': 'Compliance Status',
            'max_value': 100,
            'thresholds': [60, 80, 90]
        })
        
        return self.generate_dashboard_data('executive')

# Usage example
dashboard = SecurityDashboard()
executive_dashboard = dashboard.create_executive_dashboard()
print(f"Executive dashboard: {executive_dashboard['layout']}")