Monitoreo y Revisión

  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):
        """Actualizar métrica"""
        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
            
            # Guardar datos históricos
            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):
        """Obtener estado de métrica"""
        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):
        """Calcular tendencia de métrica"""
        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]
        
        # Calcular tendencia usando regresión lineal simple
        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):
        """Generar reporte de métricas"""
        report = {
            'timestamp': datetime.now(),
            'summary': {},
            'alerts': [],
            'trends': {}
        }
        
        # Resumen por categoría
        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
        
        # Tendencias
        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

# Ejemplo de uso
security_metrics = SecurityMetrics()

# Actualizar métricas
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)

# Generar reporte
report = security_metrics.generate_metrics_report()
print(f"Reporte de métricas: {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):
        """Definir regla de alerta"""
        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):
        """Verificar alertas basadas en métricas"""
        triggered_alerts = []
        
        for rule_id, rule in self.alert_rules.items():
            if not rule['enabled']:
                continue
            
            # Evaluar condición
            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)
                
                # Enviar notificaciones
                self.send_notifications(alert, rule['notification_channels'])
        
        return triggered_alerts
    
    def evaluate_condition(self, condition, data, threshold):
        """Evaluar condición de alerta"""
        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):
        """Enviar notificaciones"""
        for channel in channels:
            if channel in self.notification_channels:
                # Simular envío de notificación
                print(f"Notificación {channel}: {alert['rule_name']} - {alert['severity']}")
    
    def get_active_alerts(self):
        """Obtener alertas activas"""
        return [alert for alert in self.alerts.values() if alert['status'] == 'Active']
    
    def resolve_alert(self, alert_id, resolution_notes):
        """Resolver alerta"""
        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()

# Ejemplo de uso
alert_system = SecurityAlertSystem()

# Definir reglas de alerta
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']
})

# Verificar alertas
metric_data = {'incident_count': 15, 'patch_compliance': 0.75}
triggered = alert_system.check_alerts(metric_data)
print(f"Alertas activadas: {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': 'Revisión Mensual de Seguridad',
                'scope': ['Métricas operativas', 'Incidentes', 'Vulnerabilidades'],
                'participants': ['CISO', 'Security_Manager', 'IT_Manager'],
                'duration': '2 hours'
            },
            'quarterly': {
                'name': 'Revisión Trimestral de Seguridad',
                'scope': ['Estrategia', 'Presupuesto', 'Riesgos', 'Controles'],
                'participants': ['CISO', 'CEO', 'CTO', 'CFO', 'Legal'],
                'duration': '4 hours'
            },
            'annual': {
                'name': 'Revisión Anual de Seguridad',
                'scope': ['Estrategia completa', 'Cumplimiento', 'Maturidad', 'Roadmap'],
                'participants': ['Board', 'CISO', 'CEO', 'CTO', 'CFO', 'Legal', 'Audit'],
                'duration': '8 hours'
            }
        }
    
    def schedule_review(self, review_id, review_type, scheduled_date, participants):
        """Programar revisión"""
        template = self.review_templates.get(review_type, {})
        
        self.reviews[review_id] = {
            'review_id': review_id,
            'type': review_type,
            'name': template.get('name', 'Revisión de Seguridad'),
            '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):
        """Crear agenda de revisión"""
        if review_id not in self.reviews:
            return None
        
        review = self.reviews[review_id]
        template = self.review_templates.get(review['type'], {})
        
        agenda = [
            'Apertura y bienvenida',
            'Revisión de minutos anteriores',
            'Estado actual de la seguridad'
        ]
        
        # Añadir items del template
        agenda.extend(template.get('scope', []))
        
        # Añadir items personalizados
        if custom_items:
            agenda.extend(custom_items)
        
        agenda.extend([
            'Discusión de problemas y oportunidades',
            'Acciones y responsabilidades',
            'Próxima revisión'
        ])
        
        review['agenda'] = agenda
        return agenda
    
    def conduct_review(self, review_id, findings, decisions, action_items):
        """Realizar revisión"""
        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
        
        # Generar minutos
        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):
        """Calcular próxima revisión"""
        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):
        """Obtener items de acción pendientes"""
        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']

# Ejemplo de uso
review_system = SecurityReviewSystem()

# Programar revisión mensual
review_system.schedule_review(
    'REV-001',
    'monthly',
    datetime(2025, 11, 1),
    ['CISO', 'Security_Manager', 'IT_Manager']
)

# Crear agenda
agenda = review_system.create_review_agenda('REV-001', [
    'Revisión de incidente de seguridad del mes',
    'Estado de implementación de controles'
])

# Realizar revisión
findings = [
    'Aumento del 20% en incidentes de phishing',
    'Retraso en implementación de firewall'
]
decisions = [
    'Implementar capacitación adicional en phishing',
    'Acelerar implementación de firewall'
]
action_items = [
    {'item': 'Capacitación en phishing', 'owner': 'Security_Manager', 'due_date': '2025-11-15'},
    {'item': 'Implementar firewall', 'owner': 'IT_Manager', 'due_date': '2025-11-30'}
]

minutes = review_system.conduct_review('REV-001', findings, decisions, action_items)
print(f"Minutos de revisión: {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):
        """Crear widget de dashboard"""
        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):
        """Generar datos del dashboard"""
        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):
        """Obtener datos del widget"""
        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):
        """Crear dashboard ejecutivo"""
        # Crear widgets ejecutivos
        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')

# Ejemplo de uso
dashboard = SecurityDashboard()
executive_dashboard = dashboard.create_executive_dashboard()
print(f"Dashboard ejecutivo: {executive_dashboard['layout']}")