Risk Assessment

 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

import numpy as np
import pandas as pd
from scipy import stats

class QuantitativeRiskAssessment:
    def __init__(self):
        self.risk_factors = {
            'probability': {
                'Very_Low': 0.1,
                'Low': 0.3,
                'Medium': 0.5,
                'High': 0.7,
                'Very_High': 0.9
            },
            'impact': {
                'Very_Low': 1000,
                'Low': 10000,
                'Medium': 100000,
                'High': 1000000,
                'Very_High': 10000000
            }
        }
    
    def calculate_risk_value(self, probability, impact):
        """Calculate risk value"""
        return probability * impact
    
    def calculate_annual_loss_expectancy(self, single_loss_expectancy, annual_rate_of_occurrence):
        """Calculate ALE (Annual Loss Expectancy)"""
        return single_loss_expectancy * annual_rate_of_occurrence
    
    def monte_carlo_simulation(self, risk_scenarios, iterations=10000):
        """Monte Carlo simulation for risk analysis"""
        results = []
        
        for _ in range(iterations):
            total_risk = 0
            for scenario in risk_scenarios:
                # Generate random values based on distributions
                prob = np.random.beta(scenario['prob_alpha'], scenario['prob_beta'])
                impact = np.random.lognormal(scenario['impact_mean'], scenario['impact_std'])
                total_risk += prob * impact
            results.append(total_risk)
        
        return {
            'mean': np.mean(results),
            'std': np.std(results),
            'percentile_95': np.percentile(results, 95),
            'percentile_99': np.percentile(results, 99),
            'distribution': results
        }
    
    def risk_matrix(self, risks_data):
        """Create risk matrix"""
        matrix = np.zeros((5, 5))  # 5x5 matrix for probability vs impact
        
        for risk in risks_data:
            prob_idx = list(self.risk_factors['probability'].keys()).index(risk['probability'])
            impact_idx = list(self.risk_factors['impact'].keys()).index(risk['impact'])
            matrix[prob_idx][impact_idx] += 1
        
        return matrix

# Usage example
quantitative_assessment = QuantitativeRiskAssessment()

# Calculate risk value
risk_value = quantitative_assessment.calculate_risk_value(0.7, 1000000)
print(f"Risk value: ${risk_value:,.2f}")

# Calculate ALE
ale = quantitative_assessment.calculate_annual_loss_expectancy(100000, 0.3)
print(f"ALE: ${ale:,.2f}")

# Monte Carlo simulation
scenarios = [
    {'prob_alpha': 2, 'prob_beta': 8, 'impact_mean': 10, 'impact_std': 1},
    {'prob_alpha': 5, 'prob_beta': 5, 'impact_mean': 12, 'impact_std': 1.5}
]
simulation_results = quantitative_assessment.monte_carlo_simulation(scenarios)
print(f"Monte Carlo simulation - Mean: ${simulation_results['mean']:,.2f}")

 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

class QualitativeRiskAssessment:
    def __init__(self):
        self.risk_levels = {
            'Very_Low': 1,
            'Low': 2,
            'Medium': 3,
            'High': 4,
            'Very_High': 5
        }
        
        self.risk_categories = {
            'Confidentiality': 'Loss of data confidentiality',
            'Integrity': 'Corruption or unauthorized modification',
            'Availability': 'Service or data interruption',
            'Compliance': 'Regulatory non-compliance',
            'Reputation': 'Organizational reputation damage'
        }
    
    def assess_risk_level(self, probability, impact):
        """Assess risk level"""
        prob_score = self.risk_levels.get(probability, 0)
        impact_score = self.risk_levels.get(impact, 0)
        
        risk_score = prob_score * impact_score
        
        if risk_score <= 2:
            return 'Very_Low'
        elif risk_score <= 4:
            return 'Low'
        elif risk_score <= 9:
            return 'Medium'
        elif risk_score <= 16:
            return 'High'
        else:
            return 'Very_High'
    
    def risk_heatmap(self, risks_data):
        """Create risk heatmap"""
        heatmap_data = {}
        
        for risk in risks_data:
            category = risk['category']
            level = self.assess_risk_level(risk['probability'], risk['impact'])
            
            if category not in heatmap_data:
                heatmap_data[category] = {}
            
            if level not in heatmap_data[category]:
                heatmap_data[category][level] = 0
            
            heatmap_data[category][level] += 1
        
        return heatmap_data
    
    def prioritize_risks(self, risks_data):
        """Prioritize risks"""
        prioritized_risks = []
        
        for risk in risks_data:
            risk_level = self.assess_risk_level(risk['probability'], risk['impact'])
            risk_score = self.risk_levels[risk_level]
            
            prioritized_risks.append({
                **risk,
                'risk_level': risk_level,
                'risk_score': risk_score
            })
        
        # Sort by risk score (descending)
        prioritized_risks.sort(key=lambda x: x['risk_score'], reverse=True)
        
        return prioritized_risks

# Usage example
qualitative_assessment = QualitativeRiskAssessment()

# Assess risk level
risk_level = qualitative_assessment.assess_risk_level('High', 'High')
print(f"Risk level: {risk_level}")

# Example data
risks_data = [
    {'category': 'Confidentiality', 'probability': 'High', 'impact': 'High', 'description': 'Data breach'},
    {'category': 'Availability', 'probability': 'Medium', 'impact': 'High', 'description': 'DDoS attack'},
    {'category': 'Integrity', 'probability': 'Low', 'impact': 'Medium', 'description': 'Data corruption'}
]

prioritized = qualitative_assessment.prioritize_risks(risks_data)
print(f"Prioritized risks: {[(r['description'], r['risk_level']) for r in prioritized]}")

  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

class RiskManagementSystem:
    def __init__(self):
        self.risks = {}
        self.assets = {}
        self.threats = {}
        self.vulnerabilities = {}
        self.controls = {}
        self.assessments = {}
    
    def add_asset(self, asset_id, asset_data):
        """Add asset"""
        self.assets[asset_id] = {
            **asset_data,
            'created_date': '2025-10-26',
            'last_updated': '2025-10-26',
            'status': 'Active'
        }
    
    def add_threat(self, threat_id, threat_data):
        """Add threat"""
        self.threats[threat_id] = {
            **threat_data,
            'created_date': '2025-10-26',
            'last_updated': '2025-10-26',
            'status': 'Active'
        }
    
    def add_vulnerability(self, vuln_id, vuln_data):
        """Add vulnerability"""
        self.vulnerabilities[vuln_id] = {
            **vuln_data,
            'created_date': '2025-10-26',
            'last_updated': '2025-10-26',
            'status': 'Open'
        }
    
    def create_risk_assessment(self, assessment_id, assets, threats, vulnerabilities):
        """Create risk assessment"""
        assessment = {
            'assessment_id': assessment_id,
            'assets': assets,
            'threats': threats,
            'vulnerabilities': vulnerabilities,
            'risks': [],
            'created_date': '2025-10-26',
            'status': 'Draft'
        }
        
        # Generate risks based on assets, threats and vulnerabilities
        for asset_id in assets:
            for threat_id in threats:
                for vuln_id in vulnerabilities:
                    risk_id = f"{asset_id}_{threat_id}_{vuln_id}"
                    risk = self.calculate_risk(asset_id, threat_id, vuln_id)
                    assessment['risks'].append(risk)
        
        self.assessments[assessment_id] = assessment
        return assessment
    
    def calculate_risk(self, asset_id, threat_id, vuln_id):
        """Calculate specific risk"""
        asset = self.assets.get(asset_id, {})
        threat = self.threats.get(threat_id, {})
        vulnerability = self.vulnerabilities.get(vuln_id, {})
        
        # Simplified risk calculation
        probability = threat.get('probability', 0.5)
        impact = asset.get('value', 100000) * vulnerability.get('severity', 0.5)
        risk_value = probability * impact
        
        return {
            'risk_id': f"{asset_id}_{threat_id}_{vuln_id}",
            'asset_id': asset_id,
            'threat_id': threat_id,
            'vulnerability_id': vuln_id,
            'probability': probability,
            'impact': impact,
            'risk_value': risk_value,
            'status': 'Open'
        }
    
    def get_risk_summary(self, assessment_id):
        """Get risk summary"""
        if assessment_id not in self.assessments:
            return None
        
        assessment = self.assessments[assessment_id]
        risks = assessment['risks']
        
        if not risks:
            return {'total_risks': 0}
        
        total_risks = len(risks)
        high_risks = len([r for r in risks if r['risk_value'] > 500000])
        medium_risks = len([r for r in risks if 100000 < r['risk_value'] <= 500000])
        low_risks = len([r for r in risks if r['risk_value'] <= 100000])
        
        total_value = sum(r['risk_value'] for r in risks)
        avg_risk = total_value / total_risks if total_risks > 0 else 0
        
        return {
            'total_risks': total_risks,
            'high_risks': high_risks,
            'medium_risks': medium_risks,
            'low_risks': low_risks,
            'total_value': total_value,
            'average_risk': avg_risk
        }

# Usage example
risk_system = RiskManagementSystem()

# Add assets
risk_system.add_asset('ASSET-001', {
    'name': 'Database Server',
    'type': 'Infrastructure',
    'value': 1000000,
    'owner': 'IT Department'
})

# Add threats
risk_system.add_threat('THREAT-001', {
    'name': 'SQL Injection',
    'type': 'Cyber Attack',
    'probability': 0.3,
    'source': 'External'
})

# Add vulnerabilities
risk_system.add_vulnerability('VULN-001', {
    'name': 'Unpatched Database',
    'type': 'Software',
    'severity': 0.8,
    'cvss_score': 7.5
})

# Create assessment
assessment = risk_system.create_risk_assessment(
    'ASSESS-001',
    ['ASSET-001'],
    ['THREAT-001'],
    ['VULN-001']
)

summary = risk_system.get_risk_summary('ASSESS-001')
print(f"Assessment summary: {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
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120

class ScenarioAnalysis:
    def __init__(self):
        self.scenarios = {}
        self.scenario_templates = {
            'Data_Breach': {
                'description': 'Confidential data breach',
                'probability_factors': ['Security_controls', 'Employee_training', 'System_patches'],
                'impact_factors': ['Data_volume', 'Data_sensitivity', 'Regulatory_requirements']
            },
            'Ransomware': {
                'description': 'Ransomware attack',
                'probability_factors': ['Email_security', 'Backup_systems', 'User_awareness'],
                'impact_factors': ['System_downtime', 'Data_recovery', 'Business_continuity']
            },
            'Insider_Threat': {
                'description': 'Insider threat',
                'probability_factors': ['Access_controls', 'Monitoring_systems', 'Employee_satisfaction'],
                'impact_factors': ['Data_access', 'System_damage', 'Reputation']
            }
        }
    
    def create_scenario(self, scenario_id, scenario_type, custom_factors=None):
        """Create risk scenario"""
        template = self.scenario_templates.get(scenario_type, {})
        
        scenario = {
            'scenario_id': scenario_id,
            'type': scenario_type,
            'description': template.get('description', ''),
            'probability_factors': template.get('probability_factors', []),
            'impact_factors': template.get('impact_factors', []),
            'probability_score': 0,
            'impact_score': 0,
            'risk_score': 0,
            'created_date': '2025-10-26'
        }
        
        if custom_factors:
            scenario.update(custom_factors)
        
        self.scenarios[scenario_id] = scenario
        return scenario
    
    def assess_scenario(self, scenario_id, factor_scores):
        """Assess scenario with factor scores"""
        if scenario_id not in self.scenarios:
            return None
        
        scenario = self.scenarios[scenario_id]
        
        # Calculate probability based on factors
        prob_factors = scenario['probability_factors']
        prob_scores = [factor_scores.get(factor, 0) for factor in prob_factors]
        scenario['probability_score'] = sum(prob_scores) / len(prob_scores) if prob_scores else 0
        
        # Calculate impact based on factors
        impact_factors = scenario['impact_factors']
        impact_scores = [factor_scores.get(factor, 0) for factor in impact_factors]
        scenario['impact_score'] = sum(impact_scores) / len(impact_scores) if impact_scores else 0
        
        # Calculate risk score
        scenario['risk_score'] = scenario['probability_score'] * scenario['impact_score']
        
        return scenario
    
    def compare_scenarios(self, scenario_ids):
        """Compare scenarios"""
        comparison = []
        
        for scenario_id in scenario_ids:
            if scenario_id in self.scenarios:
                scenario = self.scenarios[scenario_id]
                comparison.append({
                    'scenario_id': scenario_id,
                    'type': scenario['type'],
                    'probability': scenario['probability_score'],
                    'impact': scenario['impact_score'],
                    'risk_score': scenario['risk_score']
                })
        
        # Sort by risk score
        comparison.sort(key=lambda x: x['risk_score'], reverse=True)
        
        return comparison

# Usage example
scenario_analysis = ScenarioAnalysis()

# Create scenarios
scenario_analysis.create_scenario('SCEN-001', 'Data_Breach')
scenario_analysis.create_scenario('SCEN-002', 'Ransomware')
scenario_analysis.create_scenario('SCEN-003', 'Insider_Threat')

# Assess scenarios
factor_scores = {
    'Security_controls': 0.7,
    'Employee_training': 0.6,
    'System_patches': 0.8,
    'Data_volume': 0.9,
    'Data_sensitivity': 0.8,
    'Regulatory_requirements': 0.7,
    'Email_security': 0.5,
    'Backup_systems': 0.8,
    'User_awareness': 0.6,
    'System_downtime': 0.7,
    'Data_recovery': 0.6,
    'Business_continuity': 0.8,
    'Access_controls': 0.7,
    'Monitoring_systems': 0.6,
    'Employee_satisfaction': 0.8,
    'Data_access': 0.9,
    'System_damage': 0.5,
    'Reputation': 0.8
}

for scenario_id in ['SCEN-001', 'SCEN-002', 'SCEN-003']:
    scenario_analysis.assess_scenario(scenario_id, factor_scores)

comparison = scenario_analysis.compare_scenarios(['SCEN-001', 'SCEN-002', 'SCEN-003'])
print(f"Scenario comparison: {comparison}")