///
Map identified threats to appropriate security controls and mitigations. Use when prioritizing security investments, creating remediation plans, or validating control effectiveness.
Install
$ npx agentshq add wshobson/agents --agent threat-mitigation-mappingMap identified threats to appropriate security controls and mitigations. Use when prioritizing security investments, creating remediation plans, or validating control effectiveness.
Connect threats to controls for effective security planning.
Preventive ────► Stop attacks before they occur
│ (Firewall, Input validation)
│
Detective ─────► Identify attacks in progress
│ (IDS, Log monitoring)
│
Corrective ────► Respond and recover from attacks
(Incident response, Backup restore)
| Layer | Examples | | --------------- | ------------------------------------ | | Network | Firewall, WAF, DDoS protection | | Application | Input validation, authentication | | Data | Encryption, access controls | | Endpoint | EDR, patch management | | Process | Security training, incident response |
┌──────────────────────┐
│ Perimeter │ ← Firewall, WAF
│ ┌──────────────┐ │
│ │ Network │ │ ← Segmentation, IDS
│ │ ┌────────┐ │ │
│ │ │ Host │ │ │ ← EDR, Hardening
│ │ │ ┌────┐ │ │ │
│ │ │ │App │ │ │ │ ← Auth, Validation
│ │ │ │Data│ │ │ │ ← Encryption
│ │ │ └────┘ │ │ │
│ │ └────────┘ │ │
│ └──────────────┘ │
└──────────────────────┘
from dataclasses import dataclass, field
from enum import Enum
from typing import List, Dict, Optional, Set
from datetime import datetime
class ControlType(Enum):
PREVENTIVE = "preventive"
DETECTIVE = "detective"
CORRECTIVE = "corrective"
class ControlLayer(Enum):
NETWORK = "network"
APPLICATION = "application"
DATA = "data"
ENDPOINT = "endpoint"
PROCESS = "process"
PHYSICAL = "physical"
class ImplementationStatus(Enum):
NOT_IMPLEMENTED = "not_implemented"
PARTIAL = "partial"
IMPLEMENTED = "implemented"
VERIFIED = "verified"
class Effectiveness(Enum):
NONE = 0
LOW = 1
MEDIUM = 2
HIGH = 3
VERY_HIGH = 4
@dataclass
class SecurityControl:
id: str
name: str
description: str
control_type: ControlType
layer: ControlLayer
effectiveness: Effectiveness
implementation_cost: str # Low, Medium, High
maintenance_cost: str
status: ImplementationStatus = ImplementationStatus.NOT_IMPLEMENTED
mitigates_threats: List[str] = field(default_factory=list)
dependencies: List[str] = field(default_factory=list)
technologies: List[str] = field(default_factory=list)
compliance_refs: List[str] = field(default_factory=list)
def coverage_score(self) -> float:
"""Calculate coverage score based on status and effectiveness."""
status_multiplier = {
ImplementationStatus.NOT_IMPLEMENTED: 0.0,
ImplementationStatus.PARTIAL: 0.5,
ImplementationStatus.IMPLEMENTED: 0.8,
ImplementationStatus.VERIFIED: 1.0,
}
return self.effectiveness.value * status_multiplier[self.status]
@dataclass
class Threat:
id: str
name: str
category: str # STRIDE category
description: str
impact: str # Critical, High, Medium, Low
likelihood: str
risk_score: float
@dataclass
class MitigationMapping:
threat: Threat
controls: List[SecurityControl]
residual_risk: str = "Unknown"
notes: str = ""
def calculate_coverage(self) -> float:
"""Calculate how well controls cover the threat."""
if not self.controls:
return 0.0
total_score = sum(c.coverage_score() for c in self.controls)
max_possible = len(self.controls) * Effectiveness.VERY_HIGH.value
return (total_score / max_possible) * 100 if max_possible > 0 else 0
def has_defense_in_depth(self) -> bool:
"""Check if multiple layers are covered."""
layers = set(c.layer for c in self.controls if c.status != ImplementationStatus.NOT_IMPLEMENTED)
return len(layers) >= 2
def has_control_diversity(self) -> bool:
"""Check if multiple control types are present."""
types = set(c.control_type for c in self.controls if c.status != ImplementationStatus.NOT_IMPLEMENTED)
return len(types) >= 2
@dataclass
class MitigationPlan:
name: str
threats: List[Threat] = field(default_factory=list)
controls: List[SecurityControl] = field(default_factory=list)
mappings: List[MitigationMapping] = field(default_factory=list)
def get_unmapped_threats(self) -> List[Threat]:
"""Find threats without mitigations."""
mapped_ids = {m.threat.id for m in self.mappings}
return [t for t in self.threats if t.id not in mapped_ids]
def get_control_coverage(self) -> Dict[str, float]:
"""Get coverage percentage for each threat."""
return {
m.threat.id: m.calculate_coverage()
for m in self.mappings
}
def get_gaps(self) -> List[Dict]:
"""Identify mitigation gaps."""
gaps = []
for mapping in self.mappings:
coverage = mapping.calculate_coverage()
if coverage < 50:
gaps.append({
"threat": mapping.threat.id,
"threat_name": mapping.threat.name,
"coverage": coverage,
"issue": "Insufficient control coverage",
"recommendation": "Add more controls or improve existing ones"
})
if not mapping.has_defense_in_depth():
gaps.append({
"threat": mapping.threat.id,
"threat_name": mapping.threat.name,
"coverage": coverage,
"issue": "No defense in depth",
"recommendation": "Add controls at different layers"
})
if not mapping.has_control_diversity():
gaps.append({
"threat": mapping.threat.id,
"threat_name": mapping.threat.name,
"coverage": coverage,
"issue": "No control diversity",
"recommendation": "Add detective/corrective controls"
})
return gaps
class ControlLibrary:
"""Library of standard security controls."""
STANDARD_CONTROLS = {
# Authentication Controls
"AUTH-001": SecurityControl(
id="AUTH-001",
name="Multi-Factor Authentication",
description="Require MFA for all user authentication",
control_type=ControlType.PREVENTIVE,
layer=ControlLayer.APPLICATION,
effectiveness=Effectiveness.HIGH,
implementation_cost="Medium",
maintenance_cost="Low",
mitigates_threats=["SPOOFING"],
technologies=["TOTP", "WebAuthn", "SMS OTP"],
compliance_refs=["PCI-DSS 8.3", "NIST 800-63B"]
),
"AUTH-002": SecurityControl(
id="AUTH-002",
name="Account Lockout Policy",
description="Lock accounts after failed authentication attempts",
control_type=ControlType.PREVENTIVE,
layer=ControlLayer.APPLICATION,
effectiveness=Effectiveness.MEDIUM,
implementation_cost="Low",
maintenance_cost="Low",
mitigates_threats=["SPOOFING"],
technologies=["Custom implementation"],
compliance_refs=["PCI-DSS 8.1.6"]
),
# Input Validation Controls
"VAL-001": SecurityControl(
id="VAL-001",
name="Input Validation Framework",
description="Validate and sanitize all user input",
control_type=ControlType.PREVENTIVE,
layer=ControlLayer.APPLICATION,
effectiveness=Effectiveness.HIGH,
implementation_cost="Medium",
maintenance_cost="Medium",
mitigates_threats=["TAMPERING", "INJECTION"],
technologies=["Joi", "Yup", "Pydantic"],
compliance_refs=["OWASP ASVS V5"]
),
"VAL-002": SecurityControl(
id="VAL-002",
name="Web Application Firewall",
description="Deploy WAF to filter malicious requests",
control_type=ControlType.PREVENTIVE,
layer=ControlLayer.NETWORK,
effectiveness=Effectiveness.MEDIUM,
implementation_cost="Medium",
maintenance_cost="Medium",
mitigates_threats=["TAMPERING", "INJECTION", "DOS"],
technologies=["AWS WAF", "Cloudflare", "ModSecurity"],
compliance_refs=["PCI-DSS 6.6"]
),
# Encryption Controls
"ENC-001": SecurityControl(
id="ENC-001",
name="Data Encryption at Rest",
description="Encrypt sensitive data in storage",
control_type=ControlType.PREVENTIVE,
layer=ControlLayer.DATA,
effectiveness=Effectiveness.HIGH,
implementation_cost="Medium",
maintenance_cost="Low",
mitigates_threats=["INFORMATION_DISCLOSURE"],
technologies=["AES-256", "KMS", "HSM"],
compliance_refs=["PCI-DSS 3.4", "GDPR Art. 32"]
),
"ENC-002": SecurityControl(
id="ENC-002",
name="TLS Encryption",
description="Encrypt data in transit using TLS 1.3",
control_type=ControlType.PREVENTIVE,
layer=ControlLayer.NETWORK,
effectiveness=Effectiveness.HIGH,
implementation_cost="Low",
maintenance_cost="Low",
mitigates_threats=["INFORMATION_DISCLOSURE", "TAMPERING"],
technologies=["TLS 1.3", "Certificate management"],
compliance_refs=["PCI-DSS 4.1", "HIPAA"]
),
# Logging Controls
"LOG-001": SecurityControl(
id="LOG-001",
name="Security Event Logging",
description="Log all security-relevant events",
control_type=ControlType.DETECTIVE,
layer=ControlLayer.APPLICATION,
effectiveness=Effectiveness.MEDIUM,
implementation_cost="Low",
maintenance_cost="Medium",
mitigates_threats=["REPUDIATION"],
technologies=["ELK Stack", "Splunk", "CloudWatch"],
compliance_refs=["PCI-DSS 10.2", "SOC2"]
),
"LOG-002": SecurityControl(
id="LOG-002",
name="Log Integrity Protection",
description="Protect logs from tampering",
control_type=ControlType.PREVENTIVE,
layer=ControlLayer.DATA,
effectiveness=Effectiveness.MEDIUM,
implementation_cost="Medium",
maintenance_cost="Low",
mitigates_threats=["REPUDIATION", "TAMPERING"],
technologies=["Immutable storage", "Log signing"],
compliance_refs=["PCI-DSS 10.5"]
),
# Access Control
"ACC-001": SecurityControl(
id="ACC-001",
name="Role-Based Access Control",
description="Implement RBAC for authorization",
control_type=ControlType.PREVENTIVE,
layer=ControlLayer.APPLICATION,
effectiveness=Effectiveness.HIGH,
implementation_cost="Medium",
maintenance_cost="Medium",
mitigates_threats=["ELEVATION_OF_PRIVILEGE", "INFORMATION_DISCLOSURE"],
technologies=["RBAC", "ABAC", "Policy engines"],
compliance_refs=["PCI-DSS 7.1", "SOC2"]
),
# Availability Controls
"AVL-001": SecurityControl(
id="AVL-001",
name="Rate Limiting",
description="Limit request rates to prevent abuse",
control_type=ControlType.PREVENTIVE,
layer=ControlLayer.APPLICATION,
effectiveness=Effectiveness.MEDIUM,
implementation_cost="Low",
maintenance_cost="Low",
mitigates_threats=["DENIAL_OF_SERVICE"],
technologies=["API Gateway", "Redis", "Token bucket"],
compliance_refs=["OWASP API Security"]
),
"AVL-002": SecurityControl(
id="AVL-002",
name="DDoS Protection",
description="Deploy DDoS mitigation services",
control_type=ControlType.PREVENTIVE,
layer=ControlLayer.NETWORK,
effectiveness=Effectiveness.HIGH,
implementation_cost="High",
maintenance_cost="Medium",
mitigates_threats=["DENIAL_OF_SERVICE"],
technologies=["Cloudflare", "AWS Shield", "Akamai"],
compliance_refs=["NIST CSF"]
),
}
def get_controls_for_threat(self, threat_category: str) -> List[SecurityControl]:
"""Get all controls that mitigate a threat category."""
return [
c for c in self.STANDARD_CONTROLS.values()
if threat_category in c.mitigates_threats
]
def get_controls_by_layer(self, layer: ControlLayer) -> List[SecurityControl]:
"""Get controls for a specific layer."""
return [c for c in self.STANDARD_CONTROLS.values() if c.layer == layer]
def get_control(self, control_id: str) -> Optional[SecurityControl]:
"""Get a specific control by ID."""
return self.STANDARD_CONTROLS.get(control_id)
def recommend_controls(
self,
threat: Threat,
existing_controls: List[str]
) -> List[SecurityControl]:
"""Recommend additional controls for a threat."""
available = self.get_controls_for_threat(threat.category)
return [c for c in available if c.id not in existing_controls]
class MitigationAnalyzer:
"""Analyze and optimize mitigation strategies."""
def __init__(self, plan: MitigationPlan, library: ControlLibrary):
self.plan = plan
self.library = library
def calculate_overall_risk_reduction(self) -> float:
"""Calculate overall risk reduction percentage."""
if not self.plan.mappings:
return 0.0
weighted_coverage = 0
total_weight = 0
for mapping in self.plan.mappings:
# Weight by threat risk score
weight = mapping.threat.risk_score
coverage = mapping.calculate_coverage()
weighted_coverage += weight * coverage
total_weight += weight
return weighted_coverage / total_weight if total_weight > 0 else 0
def get_critical_gaps(self) -> List[Dict]:
"""Find critical gaps that need immediate attention."""
gaps = self.plan.get_gaps()
critical_threats = {t.id for t in self.plan.threats if t.impact == "Critical"}
return [g for g in gaps if g["threat"] in critical_threats]
def optimize_budget(
self,
budget: float,
cost_map: Dict[str, float]
) -> List[SecurityControl]:
"""Select controls that maximize risk reduction within budget."""
# Simple greedy approach - can be replaced with optimization algorithm
recommended = []
remaining_budget = budget
unmapped = self.plan.get_unmapped_threats()
# Sort controls by effectiveness/cost ratio
all_controls = list(self.library.STANDARD_CONTROLS.values())
controls_with_value = []
for control in all_controls:
if control.status == ImplementationStatus.NOT_IMPLEMENTED:
cost = cost_map.get(control.id, float('inf'))
if cost <= remaining_budget:
# Calculate value as threats covered * effectiveness / cost
threats_covered = len([
t for t in unmapped
if t.category in control.mitigates_threats
])
if threats_covered > 0:
value = (threats_covered * control.effectiveness.value) / cost
controls_with_value.append((control, value, cost))
# Sort by value (higher is better)
controls_with_value.sort(key=lambda x: x[1], reverse=True)
for control, value, cost in controls_with_value:
if cost <= remaining_budget:
recommended.append(control)
remaining_budget -= cost
return recommended
def generate_roadmap(self) -> List[Dict]:
"""Generate implementation roadmap by priority."""
roadmap = []
gaps = self.plan.get_gaps()
# Phase 1: Critical threats with low coverage
phase1 = []
for gap in gaps:
mapping = next(
(m for m in self.plan.mappings if m.threat.id == gap["threat"]),
None
)
if mapping and mapping.threat.impact == "Critical":
controls = self.library.get_controls_for_threat(mapping.threat.category)
phase1.extend([
{
"threat": gap["threat"],
"control": c.id,
"control_name": c.name,
"phase": 1,
"priority": "Critical"
}
for c in controls
if c.status == ImplementationStatus.NOT_IMPLEMENTED
])
roadmap.extend(phase1[:5]) # Top 5 for phase 1
# Phase 2: High impact threats
phase2 = []
for gap in gaps:
mapping = next(
(m for m in self.plan.mappings if m.threat.id == gap["threat"]),
None
)
if mapping and mapping.threat.impact == "High":
controls = self.library.get_controls_for_threat(mapping.threat.category)
phase2.extend([
{
"threat": gap["threat"],
"control": c.id,
"control_name": c.name,
"phase": 2,
"priority": "High"
}
for c in controls
if c.status == ImplementationStatus.NOT_IMPLEMENTED
])
roadmap.extend(phase2[:5]) # Top 5 for phase 2
return roadmap
def defense_in_depth_analysis(self) -> Dict[str, List[str]]:
"""Analyze defense in depth coverage."""
layer_coverage = {layer.value: [] for layer in ControlLayer}
for mapping in self.plan.mappings:
for control in mapping.controls:
if control.status in [ImplementationStatus.IMPLEMENTED, ImplementationStatus.VERIFIED]:
layer_coverage[control.layer.value].append(control.id)
return layer_coverage
def generate_report(self) -> str:
"""Generate comprehensive mitigation report."""
risk_reduction = self.calculate_overall_risk_reduction()
gaps = self.plan.get_gaps()
critical_gaps = self.get_critical_gaps()
layer_coverage = self.defense_in_depth_analysis()
report = f"""
# Threat Mitigation Report
## Executive Summary
- **Overall Risk Reduction:** {risk_reduction:.1f}%
- **Total Threats:** {len(self.plan.threats)}
- **Total Controls:** {len(self.plan.controls)}
- **Identified Gaps:** {len(gaps)}
- **Critical Gaps:** {len(critical_gaps)}
## Defense in Depth Coverage
{self._format_layer_coverage(layer_coverage)}
## Critical Gaps Requiring Immediate Action
{self._format_gaps(critical_gaps)}
## Recommendations
{self._format_recommendations()}
## Implementation Roadmap
{self._format_roadmap()}
"""
return report
def _format_layer_coverage(self, coverage: Dict[str, List[str]]) -> str:
lines = []
for layer, controls in coverage.items():
status = "✓" if controls else "✗"
lines.append(f"- {layer}: {status} ({len(controls)} controls)")
return "\n".join(lines)
def _format_gaps(self, gaps: List[Dict]) -> str:
if not gaps:
return "No critical gaps identified."
lines = []
for gap in gaps:
lines.append(f"- **{gap['threat_name']}**: {gap['issue']}")
lines.append(f" - Coverage: {gap['coverage']:.1f}%")
lines.append(f" - Recommendation: {gap['recommendation']}")
return "\n".join(lines)
def _format_recommendations(self) -> str:
recommendations = []
layer_coverage = self.defense_in_depth_analysis()
for layer, controls in layer_coverage.items():
if not controls:
recommendations.append(f"- Add {layer} layer controls")
gaps = self.plan.get_gaps()
if any(g["issue"] == "No control diversity" for g in gaps):
recommendations.append("- Add more detective and corrective controls")
return "\n".join(recommendations) if recommendations else "Current coverage is adequate."
def _format_roadmap(self) -> str:
roadmap = self.generate_roadmap()
if not roadmap:
return "No additional controls recommended at this time."
lines = []
current_phase = 0
for item in roadmap:
if item["phase"] != current_phase:
current_phase = item["phase"]
lines.append(f"\n### Phase {current_phase}")
lines.append(f"- [{item['priority']}] {item['control_name']} (for {item['threat']})")
return "\n".join(lines)
from dataclasses import dataclass
from typing import List, Callable, Any
import asyncio
@dataclass
class ControlTest:
control_id: str
test_name: str
test_function: Callable[[], bool]
expected_result: bool
description: str
class ControlTester:
"""Test control effectiveness."""
def __init__(self):
self.tests: List[ControlTest] = []
self.results: List[Dict] = []
def add_test(self, test: ControlTest) -> None:
self.tests.append(test)
async def run_tests(self) -> List[Dict]:
"""Run all control tests."""
self.results = []
for test in self.tests:
try:
result = test.test_function()
passed = result == test.expected_result
self.results.append({
"control_id": test.control_id,
"test_name": test.test_name,
"passed": passed,
"actual_result": result,
"expected_result": test.expected_result,
"description": test.description,
"error": None
})
except Exception as e:
self.results.append({
"control_id": test.control_id,
"test_name": test.test_name,
"passed": False,
"actual_result": None,
"expected_result": test.expected_result,
"description": test.description,
"error": str(e)
})
return self.results
def get_effectiveness_score(self, control_id: str) -> float:
"""Calculate effectiveness score for a control."""
control_results = [r for r in self.results if r["control_id"] == control_id]
if not control_results:
return 0.0
passed = sum(1 for r in control_results if r["passed"])
return (passed / len(control_results)) * 100
def generate_test_report(self) -> str:
"""Generate test results report."""
if not self.results:
return "No tests have been run."
total = len(self.results)
passed = sum(1 for r in self.results if r["passed"])
report = f"""
# Control Effectiveness Test Report
## Summary
- **Total Tests:** {total}
- **Passed:** {passed}
- **Failed:** {total - passed}
- **Pass Rate:** {(passed/total)*100:.1f}%
## Results by Control
"""
# Group by control
controls = {}
for result in self.results:
cid = result["control_id"]
if cid not in controls:
controls[cid] = []
controls[cid].append(result)
for control_id, results in controls.items():
score = self.get_effectiveness_score(control_id)
report += f"\n### {control_id} (Effectiveness: {score:.1f}%)\n"
for r in results:
status = "✓" if r["passed"] else "✗"
report += f"- {status} {r['test_name']}\n"
if r["error"]:
report += f" - Error: {r['error']}\n"
return report