ffsshhttiikk What I do
- Identify and categorize security risks
- Assess likelihood and impact of threats
- Prioritize risks for remediation
- Recommend mitigation strategies
- Track risk over time
- Quantify risk in business terms
When to use me
When evaluating security posture, planning security improvements, or conducting risk assessments for compliance.
Risk Assessment Framework
Risk Model
from enum import Enum
from datetime import datetime
class Likelihood(Enum):
RARE = 1
UNLIKELY = 2
POSSIBLE = 3
LIKELY = 4
CERTAIN = 5
class Impact(Enum):
NEGLIGIBLE = 1
MINOR = 2
MODERATE = 3
MAJOR = 4
CATASTROPHIC = 5
class RiskLevel(Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
class Risk:
def __init__(self, id: str, name: str, description: str):
self.id = id
self.name = name
self.description = description
self.likelihood = None
self.impact = None
self.current_controls = []
self.residual_risk = None
self.treated_risk = None
def calculate_risk_level(self) -> RiskLevel:
score = self.likelihood.value * self.impact.value
if score >= 16:
return RiskLevel.CRITICAL
elif score >= 10:
return RiskLevel.HIGH
elif score >= 5:
return RiskLevel.MEDIUM
else:
return RiskLevel.LOW
def calculate_residual_risk(self) -> int:
"""Risk after current controls"""
base_score = self.likelihood.value * self.impact.value
control_effectiveness = sum(c.effectiveness for c in self.current_controls)
reduction = min(control_effectiveness * 2, 80) # Max 80% reduction
return int(base_score * (100 - reduction) / 100)
Risk Assessment Process
class RiskAssessment:
def __init__(self, name: str):
self.name = name
self.created_at = datetime.now()
self.risks = []
self.assets = []
def add_asset(self, asset_id: str, name: str, value: str,
classification: str):
"""Register an asset to protect"""
self.assets.append({
"id": asset_id,
"name": name,
"value": value,
"classification": classification,
"dependencies": []
})
def identify_risks(self) -> list:
"""Systematic risk identification"""
risk_categories = [
"Physical Security",
"Network Security",
"Application Security",
"Data Security",
"Operational Security",
"Business Continuity"
]
for category in risk_categories:
risks = self._identify_category_risks(category)
self.risks.extend(risks)
return self.risks
def _identify_category_risks(self, category: str) -> list:
"""Category-specific risk identification"""
templates = {
"Application Security": [
("R01", "SQL Injection", "Unsanitized database queries"),
("R02", "XSS Attack", "Unescaped user output"),
("R03", "CSRF", "Missing token validation"),
("R04", "Authentication Bypass", "Weak auth mechanisms"),
],
"Data Security": [
("R05", "Data Breach", "Unauthorized data access"),
("R06", "Data Loss", "Accidental or malicious deletion"),
("R07", "Data Exfiltration", "Unauthorized data transfer"),
]
}
risks = []
for risk_id, name, desc in templates.get(category, []):
risk = Risk(risk_id, name, desc)
risks.append(risk)
return risks
def assess_risk(self, risk_id: str, likelihood: Likelihood,
impact: Impact):
"""Assess individual risk"""
for risk in self.risks:
if risk.id == risk_id:
risk.likelihood = likelihood
risk.impact = impact
risk.residual_risk = risk.calculate_residual_risk()
return risk
raise ValueError(f"Risk {risk_id} not found")
def prioritize_risks(self) -> list:
"""Sort risks by severity"""
return sorted(self.risks,
key=lambda r: r.calculate_risk_level().value,
reverse=True)
Risk Treatment
class RiskTreatment:
def __init__(self):
self.treatments = []
def recommend_treatment(self, risk: Risk) -> dict:
"""Recommend mitigation strategy"""
risk_level = risk.calculate_risk_level()
if risk_level == RiskLevel.CRITICAL:
treatment = "mitigate"
timeline = "immediate"
options = [
"Implement preventive controls",
"Deploy detection mechanisms",
"Transfer risk through insurance"
]
elif risk_level == RiskLevel.HIGH:
treatment = "mitigate"
timeline = "within 30 days"
options = [
"Enhance existing controls",
"Add monitoring",
"Implement compensating controls"
]
elif risk_level == RiskLevel.MEDIUM:
treatment = "accept" # with monitoring
timeline = "within 90 days"
options = [
"Accept with monitoring",
"Implement partial mitigation"
]
else:
treatment = "accept"
timeline = "as scheduled"
options = ["Accept risk"]
return {
"risk_id": risk.id,
"recommended_treatment": treatment,
"timeline": timeline,
"options": options,
"cost_benefit": self._calculate_cost_benefit(risk, options)
}
def _calculate_cost_benefit(self, risk: Risk,
options: list) -> dict:
"""Analyze cost vs benefit of treatment"""
annual_loss_expectancy = self._calculate_ALE(risk)
return {
"annual_loss_expectancy": annual_loss_expectancy,
"mitigation_cost": 0, # Calculate actual cost
"roi": 0 # Return on investment
}
def _calculate_ALE(self, risk: Risk) -> int:
"""Annual Loss Expectancy = Likelihood * Impact"""
return risk.likelihood.value * risk.impact.value * 10000
Continuous Risk Monitoring
class RiskMonitor:
def __init__(self):
self.risk_register = {}
def track_risk(self, risk: Risk):
"""Track risk over time"""
if risk.id not in self.risk_register:
self.risk_register[risk.id] = []
self.risk_register[risk.id].append({
"timestamp": datetime.now(),
"likelihood": risk.likelihood,
"impact": risk.impact,
"residual_risk": risk.residual_risk,
"controls": risk.current_controls
})
def detect_risk_change(self, risk_id: str) -> dict:
"""Detect significant risk changes"""
history = self.risk_register.get(risk_id, [])
if len(history) < 2:
return {"changed": False}
current = history[-1]
previous = history[-2]
if current["residual_risk"] > previous["residual_risk"] * 1.2:
return {
"changed": True,
"direction": "increasing",
"delta": current["residual_risk"] - previous["residual_risk"]
}
return {"changed": False}
def generate_risk_report(self) -> dict:
"""Generate risk status report"""
return {
"total_risks": len(self.risk_register),
"critical_risks": 0,
"high_risks": 0,
"risk_trends": "stable",
"recommendations": []
}