Blog Post
Threat Modeling for Databases
October 07, 2025
35 min read
Introduction
Database threat modeling is a systematic approach to identifying, analyzing, and mitigating security threats that could compromise your database systems and the sensitive data they contain. As cyber threats become increasingly sophisticated and data breaches more costly, organizations must proactively assess and address potential vulnerabilities before they can be exploited.
Effective threat modeling goes beyond traditional security measures by providing a structured methodology to understand your database's attack surface, identify potential threat vectors, and implement appropriate countermeasures. This process enables organizations to make informed security investment decisions, prioritize remediation efforts, and build more resilient database architectures.
This comprehensive guide covers the complete threat modeling lifecycle for database systems, from initial asset identification through ongoing threat landscape monitoring. You'll learn proven methodologies, practical tools, and real-world techniques that have been refined through years of experience protecting critical database systems across diverse industries and threat environments.
Table of Contents
Threat Modeling Fundamentals
Core Concepts and Definitions
- Threat: A potential danger that could exploit a vulnerability to cause harm
- Vulnerability: A weakness that can be exploited by threats
- Asset: Valuable database components that need protection
- Risk: The potential for loss when a threat exploits a vulnerability
- Attack Vector: The path or method used to access or attack database systems
- Impact: The potential consequences of a successful attack
Threat Modeling Methodologies
STRIDE Framework:
- Spoofing: Impersonating legitimate users or systems
- Tampering: Unauthorized modification of data or code
- Repudiation: Denying actions or transactions
- Information Disclosure: Unauthorized access to confidential data
- Denial of Service: Making systems unavailable
- Elevation of Privilege: Gaining unauthorized access levels
PASTA (Process for Attack Simulation and Threat Analysis): Seven-stage risk-centric methodology focusing on business objectives and compliance requirements.
OCTAVE (Operationally Critical Threat, Asset, and Vulnerability Evaluation): Risk-based approach emphasizing organizational planning and decision making.
class ThreatModelingFramework:
"""Core threat modeling implementation"""
def __init__(self, methodology='STRIDE'):
self.methodology = methodology
self.assets = {}
self.threats = {}
self.vulnerabilities = {}
self.controls = {}
self.risk_assessments = {}
def add_asset(self, asset_id, name, asset_type, criticality,
data_classification):
"""Add database asset to threat model"""
self.assets[asset_id] = {
'name': name,
'type': asset_type,
'criticality': criticality,
'data_classification': data_classification,
'threats': set(),
'vulnerabilities': set(),
'controls': set(),
'created_at': datetime.utcnow()
}
def add_threat(self, threat_id, category, description, likelihood,
impact, threat_actor):
"""Add threat to threat model"""
self.threats[threat_id] = {
'category': category, # STRIDE category
'description': description,
'likelihood': likelihood, # 1-5 scale
'impact': impact, # 1-5 scale
'threat_actor': threat_actor,
'affected_assets': set(),
'attack_vectors': [],
'created_at': datetime.utcnow()
}
def add_vulnerability(self, vuln_id, description, severity,
affected_components, exploit_difficulty):
"""Add vulnerability to threat model"""
self.vulnerabilities[vuln_id] = {
'description': description,
'severity': severity, # Critical, High, Medium, Low
'affected_components': affected_components,
'exploit_difficulty': exploit_difficulty, # 1-5 scale
'related_threats': set(),
'mitigation_status': 'open',
'created_at': datetime.utcnow()
}
def calculate_risk_score(self, threat_id, vuln_id):
"""Calculate risk score for threat-vulnerability pair"""
threat = self.threats.get(threat_id)
vulnerability = self.vulnerabilities.get(vuln_id)
if not threat or not vulnerability:
return 0
# Risk = Likelihood × Impact × Exploitability
likelihood = threat['likelihood']
impact = threat['impact']
exploitability = 6 - vulnerability['exploit_difficulty']
risk_score = (likelihood * impact * exploitability) / 3
return min(risk_score, 5.0) # Cap at maximum score
def generate_threat_matrix(self):
"""Generate comprehensive threat matrix"""
matrix = {
'assets': {},
'threat_summary': {},
'risk_levels': {
'critical': [],
'high': [],
'medium': [],
'low': []
},
'generated_at': datetime.utcnow()
}
for asset_id, asset in self.assets.items():
asset_threats = []
for threat_id in asset['threats']:
threat = self.threats[threat_id]
# Find related vulnerabilities
related_vulns = []
for vuln_id, vuln in self.vulnerabilities.items():
if asset_id in vuln['affected_components']:
risk_score = self.calculate_risk_score(
threat_id, vuln_id
)
related_vulns.append({
'vulnerability_id': vuln_id,
'risk_score': risk_score,
'severity': vuln['severity']
})
asset_threats.append({
'threat_id': threat_id,
'threat': threat,
'vulnerabilities': related_vulns
})
matrix['assets'][asset_id] = {
'asset_info': asset,
'threats': asset_threats
}
return matrixDatabase-Specific Threat Categories
Data Threats:
- Unauthorized data access
- Data exfiltration
- Data manipulation/corruption
- Data destruction
- Data leakage through backups or logs
Access Control Threats:
- Privilege escalation
- Account compromise
- Weak authentication
- Session hijacking
- Insider threats
Infrastructure Threats:
- SQL injection attacks
- Network-based attacks
- Operating system vulnerabilities
- Database software vulnerabilities
- Configuration weaknesses
Operational Threats:
- Backup/recovery failures
- Inadequate monitoring
- Poor change management
- Insufficient disaster recovery
- Human error
Database Asset Identification
Comprehensive Asset Inventory
Identifying and cataloging all database assets is the foundation of effective threat modeling. This includes data assets, system components, network resources, and human elements that interact with your database infrastructure.
class DatabaseAssetInventory:
"""Comprehensive database asset identification and classification"""
def __init__(self):
self.assets = {
'data_assets': {},
'system_assets': {},
'network_assets': {},
'human_assets': {},
'process_assets': {}
}
def identify_data_assets(self, database_connections):
"""Identify and classify data assets"""
for db_name, connection in database_connections.items():
with connection.cursor() as cursor:
# Get schema and table information
cursor.execute("""
SELECT table_schema, table_name, table_type,
is_insertable_into, is_typed
FROM information_schema.tables
WHERE table_schema NOT IN ('information_schema', 'pg_catalog')
""")
tables = cursor.fetchall()
for table in tables:
schema, table_name, table_type, insertable, typed = table
# Analyze table data sensitivity
sensitivity = self.analyze_data_sensitivity(
connection, schema, table_name
)
# Identify PII and sensitive columns
sensitive_columns = self.identify_sensitive_columns(
connection, schema, table_name
)
asset_id = f"{db_name}.{schema}.{table_name}"
self.assets['data_assets'][asset_id] = {
'database': db_name,
'schema': schema,
'table_name': table_name,
'table_type': table_type,
'data_classification': sensitivity,
'sensitive_columns': sensitive_columns,
'record_count': self.get_record_count(
connection, schema, table_name
),
'access_frequency': self.analyze_access_patterns(
connection, schema, table_name
),
'business_criticality': self.assess_business_criticality(
schema, table_name
)
}
def analyze_data_sensitivity(self, connection, schema, table_name):
"""Analyze data sensitivity based on content and metadata"""
with connection.cursor() as cursor:
# Get column information
cursor.execute("""
SELECT column_name, data_type, is_nullable
FROM information_schema.columns
WHERE table_schema = %s AND table_name = %s
""", (schema, table_name))
columns = cursor.fetchall()
sensitivity_indicators = {
'public': 0,
'internal': 0,
'confidential': 0,
'restricted': 0
}
# Analyze column names and types for sensitivity indicators
for column_name, data_type, nullable in columns:
column_lower = column_name.lower()
# PII indicators
if any(indicator in column_lower for indicator in [
'ssn', 'social_security', 'tax_id', 'passport'
]):
sensitivity_indicators['restricted'] += 3
# Financial indicators
if any(indicator in column_lower for indicator in [
'salary', 'income', 'credit_card', 'bank_account'
]):
sensitivity_indicators['confidential'] += 2
# Personal information
if any(indicator in column_lower for indicator in [
'email', 'phone', 'address', 'birth_date'
]):
sensitivity_indicators['confidential'] += 1
# Internal business data
if any(indicator in column_lower for indicator in [
'employee_id', 'customer_id', 'revenue', 'profit'
]):
sensitivity_indicators['internal'] += 1
# Public information
if any(indicator in column_lower for indicator in [
'name', 'title', 'description', 'category'
]):
sensitivity_indicators['public'] += 1
# Determine overall classification
max_sensitivity = max(
sensitivity_indicators.items(),
key=lambda x: x[1]
)
return max_sensitivity[0]
def identify_sensitive_columns(self, connection, schema, table_name):
"""Identify columns containing sensitive data"""
sensitive_patterns = {
'pii': [
'ssn', 'social_security', 'tax_id', 'passport', 'drivers_license'
],
'financial': [
'credit_card', 'bank_account', 'routing_number', 'iban'
],
'contact': ['email', 'phone', 'address', 'zip_code'],
'health': [
'medical_record', 'diagnosis', 'treatment', 'medication'
],
'authentication': ['password', 'token', 'secret', 'key']
}
sensitive_columns = {}
with connection.cursor() as cursor:
cursor.execute("""
SELECT column_name, data_type
FROM information_schema.columns
WHERE table_schema = %s AND table_name = %s
""", (schema, table_name))
columns = cursor.fetchall()
for column_name, data_type in columns:
column_lower = column_name.lower()
for category, patterns in sensitive_patterns.items():
if any(pattern in column_lower for pattern in patterns):
if category not in sensitive_columns:
sensitive_columns[category] = []
sensitive_columns[category].append({
'column_name': column_name,
'data_type': data_type
})
return sensitive_columnsData Flow Analysis
Understanding how data flows through your systems is critical for identifying potential attack vectors and trust boundaries where security controls should be enforced.
class DataFlowAnalyzer:
"""Analyze data flows to identify potential attack vectors"""
def __init__(self):
self.data_flows = {}
self.trust_boundaries = {}
self.entry_points = {}
def map_data_flows(self, applications, databases):
"""Map data flows between applications and databases"""
for app_name, app_config in applications.items():
for db_connection in app_config.get('database_connections', []):
flow_id = f"{app_name}_to_{db_connection['database']}"
self.data_flows[flow_id] = {
'source': {
'type': 'application',
'name': app_name,
'host': app_config.get('host'),
'port': app_config.get('port')
},
'destination': {
'type': 'database',
'name': db_connection['database'],
'host': db_connection['host'],
'port': db_connection['port']
},
'protocol': db_connection.get('protocol', 'tcp'),
'encryption': db_connection.get('ssl_enabled', False),
'authentication': db_connection.get('auth_method', 'password'),
'data_types': self.identify_data_types(
app_name, db_connection
),
'trust_level': self.assess_trust_level(
app_config, db_connection
)
}
def identify_trust_boundaries(self):
"""Identify trust boundaries in data flows"""
trust_boundaries = {
'internet_to_dmz': {
'description': 'External internet traffic to DMZ',
'risk_level': 'high',
'controls_required': [
'firewall', 'waf', 'ddos_protection'
]
},
'dmz_to_internal': {
'description': 'DMZ to internal network',
'risk_level': 'medium',
'controls_required': [
'firewall', 'proxy', 'authentication'
]
},
'application_to_database': {
'description': 'Application tier to database tier',
'risk_level': 'medium',
'controls_required': [
'network_segmentation', 'encryption', 'auth'
]
},
'database_to_backup': {
'description': 'Database to backup systems',
'risk_level': 'low',
'controls_required': ['encryption', 'access_control']
}
}
self.trust_boundaries = trust_boundaries
return trust_boundaries
def identify_entry_points(self):
"""Identify potential attack entry points"""
entry_points = {
'web_applications': {
'attack_vectors': [
'sql_injection', 'xss', 'csrf', 'auth_bypass'
],
'exposure_level': 'high',
'mitigation_priority': 'critical'
},
'api_endpoints': {
'attack_vectors': [
'injection', 'broken_authentication', 'api_abuse'
],
'exposure_level': 'high',
'mitigation_priority': 'critical'
},
'database_ports': {
'attack_vectors': [
'brute_force', 'privilege_escalation', 'network_sniffing'
],
'exposure_level': 'medium',
'mitigation_priority': 'high'
},
'backup_systems': {
'attack_vectors': [
'unauthorized_access', 'data_theft', 'tampering'
],
'exposure_level': 'low',
'mitigation_priority': 'medium'
},
'admin_interfaces': {
'attack_vectors': [
'credential_theft', 'privilege_escalation', 'social_engineering'
],
'exposure_level': 'medium',
'mitigation_priority': 'high'
}
}
self.entry_points = entry_points
return entry_pointsThreat Identification Methodologies
STRIDE-Based Threat Analysis
STRIDE provides a comprehensive framework for categorizing and analyzing database threats across six key dimensions. This systematic approach ensures no major threat categories are overlooked during the modeling process.
class DatabaseThreatAnalyzer:
"""Comprehensive database threat analysis using STRIDE methodology"""
def __init__(self):
self.identified_threats = {
'spoofing': [],
'tampering': [],
'repudiation': [],
'information_disclosure': [],
'denial_of_service': [],
'elevation_of_privilege': []
}
def analyze_spoofing_threats(self, assets):
"""Identify spoofing threats against database assets"""
spoofing_threats = [
{
'threat_id': 'SPOOF-001',
'name': 'Database User Impersonation',
'description': 'Attacker impersonates legitimate database user',
'attack_vectors': [
'Credential theft through phishing',
'Password brute force attacks',
'Session token hijacking',
'Man-in-the-middle attacks'
],
'affected_assets': ['user_accounts', 'authentication_system'],
'likelihood': 4,
'impact': 5,
'threat_actor': 'external_attacker'
},
{
'threat_id': 'SPOOF-002',
'name': 'Application Identity Spoofing',
'description': 'Malicious application impersonates legitimate service',
'attack_vectors': [
'Service account compromise',
'Certificate theft or forgery',
'DNS poisoning',
'Network protocol manipulation'
],
'affected_assets': ['service_accounts', 'certificates'],
'likelihood': 3,
'impact': 4,
'threat_actor': 'advanced_persistent_threat'
},
{
'threat_id': 'SPOOF-003',
'name': 'Database Server Impersonation',
'description': 'Fake database server presented to applications',
'attack_vectors': [
'DNS hijacking',
'Network infrastructure compromise',
'SSL certificate spoofing',
'Routing table manipulation'
],
'affected_assets': ['database_servers', 'network_infrastructure'],
'likelihood': 2,
'impact': 5,
'threat_actor': 'nation_state'
}
]
self.identified_threats['spoofing'] = spoofing_threats
return spoofing_threats
def analyze_tampering_threats(self, assets):
"""Identify data tampering threats"""
tampering_threats = [
{
'threat_id': 'TAMP-001',
'name': 'Unauthorized Data Modification',
'description': 'Malicious modification of database records',
'attack_vectors': [
'SQL injection attacks',
'Privilege escalation',
'Direct database access',
'Backup tampering'
],
'affected_assets': ['data_tables', 'application_interfaces'],
'likelihood': 4,
'impact': 5,
'threat_actor': 'malicious_insider'
},
{
'threat_id': 'TAMP-002',
'name': 'Schema Manipulation',
'description': 'Unauthorized changes to database structure',
'attack_vectors': [
'Administrative privilege abuse',
'DDL injection attacks',
'Configuration file tampering',
'System-level compromise'
],
'affected_assets': ['database_schema', 'configuration_files'],
'likelihood': 3,
'impact': 4,
'threat_actor': 'privileged_insider'
},
{
'threat_id': 'TAMP-003',
'name': 'Audit Log Tampering',
'description': 'Modification or deletion of audit trails',
'attack_vectors': [
'Log file direct access',
'Administrative privilege abuse',
'System compromise',
'Storage system attack'
],
'affected_assets': ['audit_logs', 'monitoring_systems'],
'likelihood': 3,
'impact': 4,
'threat_actor': 'malicious_insider'
}
]
self.identified_threats['tampering'] = tampering_threats
return tampering_threats
def analyze_information_disclosure_threats(self, assets):
"""Identify information disclosure threats"""
disclosure_threats = [
{
'threat_id': 'INFO-001',
'name': 'SQL Injection Data Exfiltration',
'description': 'Data extraction through SQL injection vulnerabilities',
'attack_vectors': [
'Union-based SQL injection',
'Boolean-based blind injection',
'Time-based blind injection',
'Error-based injection'
],
'affected_assets': ['sensitive_data', 'web_applications'],
'likelihood': 5,
'impact': 5,
'threat_actor': 'external_attacker'
},
{
'threat_id': 'INFO-002',
'name': 'Backup Data Theft',
'description': 'Unauthorized access to database backups',
'attack_vectors': [
'Backup system compromise',
'Storage location exposure',
'Unencrypted backup access',
'Cloud storage misconfiguration'
],
'affected_assets': ['backup_systems', 'archived_data'],
'likelihood': 3,
'impact': 5,
'threat_actor': 'external_attacker'
},
{
'threat_id': 'INFO-003',
'name': 'Insider Data Theft',
'description': 'Authorized users extracting sensitive data for malicious purposes',
'attack_vectors': [
'Excessive privilege abuse',
'Data export functionality misuse',
'Direct database queries',
'Screen capture or photography'
],
'affected_assets': ['sensitive_data', 'user_access_controls'],
'likelihood': 4,
'impact': 4,
'threat_actor': 'malicious_insider'
}
]
self.identified_threats['information_disclosure'] = disclosure_threats
return disclosure_threats
def analyze_denial_of_service_threats(self, assets):
"""Identify denial of service threats"""
dos_threats = [
{
'threat_id': 'DOS-001',
'name': 'Database Resource Exhaustion',
'description': 'Overwhelming database with resource-intensive operations',
'attack_vectors': [
'Complex query injection',
'Connection pool exhaustion',
'Storage space consumption',
'CPU-intensive operations'
],
'affected_assets': ['database_servers', 'application_performance'],
'likelihood': 4,
'impact': 4,
'threat_actor': 'external_attacker'
},
{
'threat_id': 'DOS-002',
'name': 'Network-Level DDoS',
'description': 'Network flooding attacks against database infrastructure',
'attack_vectors': [
'Volumetric attacks',
'Protocol attacks',
'Application layer attacks',
'Botnet-based attacks'
],
'affected_assets': ['network_infrastructure', 'database_endpoints'],
'likelihood': 3,
'impact': 4,
'threat_actor': 'cybercriminal_group'
},
{
'threat_id': 'DOS-003',
'name': 'Ransomware Data Encryption',
'description': 'Malicious encryption of database files and backups',
'attack_vectors': [
'Malware deployment',
'Credential compromise',
'Network lateral movement',
'Backup system targeting'
],
'affected_assets': ['database_files', 'backup_systems'],
'likelihood': 3,
'impact': 5,
'threat_actor': 'ransomware_group'
}
]
self.identified_threats['denial_of_service'] = dos_threats
return dos_threatsAttack Tree Analysis
Attack trees provide a structured way to model how attackers might achieve specific goals, helping teams understand attack paths and identify critical control points.
class AttackTreeAnalyzer:
"""Attack tree analysis for specific threat scenarios"""
def __init__(self):
self.attack_trees = {}
def build_sql_injection_attack_tree(self):
"""Build comprehensive SQL injection attack tree"""
attack_tree = {
'root_goal': 'Execute SQL Injection Attack',
'attack_paths': {
'input_validation_bypass': {
'probability': 0.7,
'prerequisites': [
'web_application_access',
'input_parameter_discovery'
],
'attack_vectors': {
'union_based': {
'description': 'Extract data using UNION queries',
'techniques': [
'Identify injectable parameters',
'Determine number of columns',
'Extract database schema',
'Extract sensitive data'
],
'success_probability': 0.8,
'detection_probability': 0.6
},
'boolean_blind': {
'description': 'Extract data through boolean responses',
'techniques': [
'Identify boolean-based responses',
'Enumerate database structure',
'Extract data bit by bit',
'Automate extraction process'
],
'success_probability': 0.6,
'detection_probability': 0.3
},
'time_based_blind': {
'description': 'Extract data through time delays',
'techniques': [
'Identify time-based responses',
'Create extraction queries',
'Implement timing analysis',
'Extract sensitive information'
],
'success_probability': 0.5,
'detection_probability': 0.2
}
}
},
'stored_procedure_exploitation': {
'probability': 0.4,
'prerequisites': [
'database_access',
'stored_procedure_knowledge'
],
'attack_vectors': {
'dynamic_sql_injection': {
'description': 'Exploit dynamic SQL in stored procedures',
'techniques': [
'Identify dynamic SQL procedures',
'Analyze parameter handling',
'Inject malicious SQL',
'Execute unauthorized commands'
],
'success_probability': 0.7,
'detection_probability': 0.4
}
}
},
'second_order_injection': {
'probability': 0.3,
'prerequisites': [
'data_storage_capability',
'data_retrieval_trigger'
],
'attack_vectors': {
'stored_data_exploitation': {
'description': 'Exploit previously stored malicious data',
'techniques': [
'Store malicious payload',
'Trigger data retrieval',
'Execute injected code',
'Escalate privileges'
],
'success_probability': 0.6,
'detection_probability': 0.2
}
}
}
}
}
self.attack_trees['sql_injection'] = attack_tree
return attack_tree
def calculate_attack_probability(self, attack_tree):
"""Calculate overall attack success probability"""
total_probability = 0
for path_name, path_data in attack_tree['attack_paths'].items():
path_probability = path_data['probability']
# Calculate vector probabilities within path
vector_probabilities = []
for vector_name, vector_data in path_data['attack_vectors'].items():
vector_prob = vector_data['success_probability']
detection_prob = vector_data['detection_probability']
# Adjust for detection probability
adjusted_prob = vector_prob * (1 - detection_prob)
vector_probabilities.append(adjusted_prob)
# Use maximum vector probability for path
max_vector_prob = max(vector_probabilities) if vector_probabilities else 0
path_total_prob = path_probability * max_vector_prob
total_probability += path_total_prob
# Cap at maximum probability
return min(total_probability, 1.0)Risk Analysis and Prioritization
Risk Assessment Framework
Effective risk assessment combines threat likelihood, vulnerability severity, and asset value to calculate inherent risk, then applies control effectiveness to determine residual risk.
class DatabaseRiskAssessment:
"""Comprehensive database risk assessment framework"""
def __init__(self):
self.risk_matrix = {}
self.risk_factors = {}
self.mitigation_priorities = {}
def calculate_inherent_risk(self, threat, vulnerability, asset):
"""Calculate inherent risk before controls"""
# Threat factors
threat_likelihood = threat.get('likelihood', 3)
threat_sophistication = threat.get('sophistication', 3)
# Vulnerability factors
vuln_severity = self.convert_severity_to_numeric(
vulnerability.get('severity', 'medium')
)
exploit_difficulty = vulnerability.get('exploit_difficulty', 3)
# Asset factors
asset_value = self.assess_asset_value(asset)
asset_exposure = self.assess_asset_exposure(asset)
# Risk calculation
likelihood_score = (threat_likelihood + (5 - exploit_difficulty)) / 2
impact_score = (vuln_severity + asset_value + asset_exposure) / 3
inherent_risk = likelihood_score * impact_score
return {
'likelihood_score': likelihood_score,
'impact_score': impact_score,
'inherent_risk_score': inherent_risk,
'risk_level': self.categorize_risk_level(inherent_risk)
}
def calculate_residual_risk(self, inherent_risk, controls):
"""Calculate residual risk after applying controls"""
control_effectiveness = self.assess_control_effectiveness(controls)
# Apply control effectiveness to reduce risk
risk_reduction = inherent_risk['inherent_risk_score'] * (
control_effectiveness / 5.0
)
residual_risk_score = inherent_risk['inherent_risk_score'] - risk_reduction
return {
'residual_risk_score': max(residual_risk_score, 0.5),
'risk_reduction': risk_reduction,
'control_effectiveness': control_effectiveness,
'residual_risk_level': self.categorize_risk_level(residual_risk_score)
}
def assess_asset_value(self, asset):
"""Assess business value of asset"""
value_factors = {
'data_classification': {
'public': 1,
'internal': 2,
'confidential': 4,
'restricted': 5
},
'business_criticality': {
'low': 1,
'medium': 2,
'high': 4,
'critical': 5
},
'regulatory_requirements': {
'none': 1,
'low': 2,
'medium': 3,
'high': 4,
'critical': 5
}
}
classification_score = value_factors['data_classification'].get(
asset.get('data_classification', 'internal'), 2
)
criticality_score = value_factors['business_criticality'].get(
asset.get('business_criticality', 'medium'), 2
)
regulatory_score = value_factors['regulatory_requirements'].get(
asset.get('regulatory_impact', 'low'), 2
)
return (classification_score + criticality_score + regulatory_score) / 3
def assess_control_effectiveness(self, controls):
"""Assess effectiveness of security controls"""
control_weights = {
'preventive': 0.4,
'detective': 0.3,
'corrective': 0.2,
'compensating': 0.1
}
total_effectiveness = 0
for control in controls:
control_type = control.get('type', 'preventive')
control_maturity = control.get('maturity', 3) # 1-5 scale
control_coverage = control.get('coverage', 0.8) # 0-1 scale
# Calculate control effectiveness
base_effectiveness = control_maturity * control_coverage
weighted_effectiveness = base_effectiveness * control_weights.get(
control_type, 0.25
)
total_effectiveness += weighted_effectiveness
return min(total_effectiveness, 5.0)
def prioritize_risks(self, risk_assessments):
"""Prioritize risks for remediation"""
prioritized_risks = []
for risk_id, assessment in risk_assessments.items():
priority_score = self.calculate_priority_score(assessment)
prioritized_risks.append({
'risk_id': risk_id,
'assessment': assessment,
'priority_score': priority_score,
'recommended_actions': self.recommend_actions(assessment)
})
# Sort by priority score (highest first)
prioritized_risks.sort(key=lambda x: x['priority_score'], reverse=True)
return prioritized_risks
def calculate_priority_score(self, assessment):
"""Calculate risk prioritization score"""
factors = {
'residual_risk': assessment['residual_risk']['residual_risk_score'] * 0.4,
'asset_value': assessment['asset_value'] * 0.3,
'exploit_likelihood': assessment['threat']['likelihood'] * 0.2,
'business_impact': assessment.get('business_impact', 3) * 0.1
}
return sum(factors.values())
def recommend_actions(self, assessment):
"""Recommend risk mitigation actions"""
recommendations = []
residual_risk = assessment['residual_risk']['residual_risk_score']
if residual_risk >= 4.0:
recommendations.extend([
'Immediate remediation required',
'Implement additional preventive controls',
'Consider risk transfer or avoidance',
'Executive notification required'
])
elif residual_risk >= 3.0:
recommendations.extend([
'Priority remediation within 30 days',
'Enhance monitoring and detection',
'Review and strengthen existing controls'
])
elif residual_risk >= 2.0:
recommendations.extend([
'Standard remediation timeline',
'Monitor for changes in threat landscape',
'Periodic control effectiveness review'
])
else:
recommendations.extend([
'Monitor and maintain current controls',
'Include in routine security reviews'
])
return recommendationsBusiness Impact Analysis
Quantifying the potential business impact of database threats enables informed decision-making about security investments and risk acceptance.
class BusinessImpactAnalyzer:
"""Analyze business impact of database security threats"""
def __init__(self):
self.impact_categories = {}
self.cost_models = {}
def analyze_financial_impact(self, threat_scenario, affected_assets):
"""Analyze financial impact of threat realization"""
impact_components = {
'direct_costs': self.calculate_direct_costs(
threat_scenario, affected_assets
),
'indirect_costs': self.calculate_indirect_costs(
threat_scenario, affected_assets
),
'regulatory_costs': self.calculate_regulatory_costs(
threat_scenario, affected_assets
),
'opportunity_costs': self.calculate_opportunity_costs(
threat_scenario, affected_assets
)
}
total_financial_impact = sum(impact_components.values())
return {
'total_impact': total_financial_impact,
'impact_breakdown': impact_components,
'confidence_level': self.assess_estimate_confidence(
threat_scenario
)
}
def calculate_direct_costs(self, threat_scenario, affected_assets):
"""Calculate direct incident response costs"""
cost_factors = {
'incident_response': {
'internal_hours': 200, # Average IR hours
'hourly_rate': 150, # Blended rate
'external_consultant_hours': 80,
'external_hourly_rate': 300
},
'forensic_investigation': {
'hours': 120,
'hourly_rate': 250
},
'system_recovery': {
'downtime_hours': threat_scenario.get('estimated_downtime', 8),
'hourly_revenue_loss': 10000 # Organization-specific
},
'notification_costs': {
'customers': len(affected_assets.get('customer_records', [])),
'regulatory': 5000 # Fixed regulatory notification cost
}
}
direct_costs = (
cost_factors['incident_response']['internal_hours'] *
cost_factors['incident_response']['hourly_rate'] +
cost_factors['incident_response']['external_consultant_hours'] *
cost_factors['incident_response']['external_hourly_rate'] +
cost_factors['forensic_investigation']['hours'] *
cost_factors['forensic_investigation']['hourly_rate'] +
cost_factors['system_recovery']['downtime_hours'] *
cost_factors['system_recovery']['hourly_revenue_loss'] +
cost_factors['notification_costs']['customers'] +
cost_factors['notification_costs']['regulatory']
)
return direct_costs
def calculate_regulatory_costs(self, threat_scenario, affected_assets):
"""Calculate potential regulatory fines and penalties"""
regulatory_frameworks = {
'gdpr': {
'max_fine_percentage': 0.04, # 4% of annual revenue
'per_record_fine': 0,
'triggers': ['eu_personal_data']
},
'hipaa': {
'max_fine_percentage': 0,
'per_record_fine': 408, # Average HIPAA fine per record
'triggers': ['health_information']
},
'pci_dss': {
'max_fine_percentage': 0,
'per_record_fine': 200, # Average PCI fine per record
'triggers': ['payment_card_data']
},
'sox': {
'max_fine_percentage': 0,
'per_record_fine': 0,
'fixed_penalties': 1000000, # SOX financial reporting violations
'triggers': ['financial_data']
}
}
total_regulatory_cost = 0
annual_revenue = 100000000 # Organization-specific
for framework, parameters in regulatory_frameworks.items():
if self.check_regulatory_triggers(affected_assets, parameters):
if parameters['max_fine_percentage'] > 0:
percentage_fine = annual_revenue * parameters['max_fine_percentage']
total_regulatory_cost += percentage_fine
if parameters['per_record_fine'] > 0:
affected_records = self.count_affected_records(affected_assets)
record_fine = affected_records * parameters['per_record_fine']
total_regulatory_cost += record_fine
if 'fixed_penalties' in parameters:
total_regulatory_cost += parameters['fixed_penalties']
return total_regulatory_cost
def analyze_operational_impact(self, threat_scenario, affected_assets):
"""Analyze operational impact of security incident"""
operational_impacts = {
'service_availability': {
'affected_services': self.identify_affected_services(
affected_assets
),
'downtime_duration': threat_scenario.get('estimated_downtime', 0),
'recovery_time': threat_scenario.get('estimated_recovery', 0),
'user_impact': self.calculate_user_impact(affected_assets)
},
'data_integrity': {
'corrupted_records': threat_scenario.get('corrupted_records', 0),
'data_recovery_effort': threat_scenario.get('recovery_effort', 0),
'data_validation_required': threat_scenario.get(
'validation_required', False
)
},
'business_process_disruption': {
'critical_processes': self.identify_critical_processes(
affected_assets
),
'process_downtime': threat_scenario.get('process_downtime', 0),
'manual_workaround_effort': threat_scenario.get(
'manual_effort', 0
)
}
}
return operational_impactsFAQ Section
Q: How often should threat modeling be performed for databases?
Conduct initial comprehensive threat modeling during design phase, annual reviews for mature systems, and triggered reviews for significant changes (new applications, infrastructure changes, major vulnerabilities). High-risk systems may require quarterly reviews.
Q: What's the difference between threats, vulnerabilities, and risks?
Threats are potential dangers that could cause harm, vulnerabilities are weaknesses that can be exploited, and risks are the potential for loss when threats exploit vulnerabilities. Risk = Threat × Vulnerability × Asset Value.
Q: Should we use automated tools for threat modeling?
Automated tools help with asset discovery and vulnerability scanning, but human expertise is essential for threat identification, risk assessment, and mitigation strategy development. Use tools to augment, not replace, human analysis.
Q: How do we prioritize threats when resources are limited?
Focus on high-impact, high-likelihood threats affecting critical assets first. Consider regulatory requirements, business criticality, and exploit difficulty. Use risk matrices to guide decision-making and ensure consistent prioritization.
Q: What's the role of threat intelligence in database threat modeling?
Threat intelligence provides current information about attack trends, new vulnerabilities, and threat actor capabilities. This helps update threat likelihood assessments, identify emerging attack vectors, and adjust security controls accordingly.
Q: How do we handle threats from insider access?
Implement principle of least privilege, segregation of duties, comprehensive monitoring, background checks, and regular access reviews. Design controls that detect unusual behavior patterns and limit damage from compromised insider accounts.
Need Expert Database Threat Modeling Help?
At Udu Labs, we specialize in comprehensive database threat modeling and security assessments. Our expert team helps organizations identify vulnerabilities, prioritize risks, and implement robust security controls using proven STRIDE methodologies and risk assessment frameworks.
Our threat modeling services include:
- Complete database asset identification and classification
- STRIDE-based threat analysis and attack tree modeling
- Risk assessment and prioritization with business impact analysis
- Security control implementation and validation
Conclusion
Database threat modeling is a critical security practice that enables organizations to proactively identify and mitigate risks before they can be exploited. By following the systematic approaches outlined in this guide, organizations can build more secure database systems and make informed security investment decisions.
Success in threat modeling requires combining technical expertise with business understanding, regular practice with continuous improvement, and proactive thinking with reactive learning. The methodologies and frameworks provided offer a foundation for developing comprehensive threat modeling capabilities tailored to your specific environment and requirements.
Remember that threat modeling is an ongoing process, not a one-time activity. As threats evolve and systems change, your threat models must be updated to maintain their effectiveness and relevance.
About UduLabs
UduLabs provides expert database security consulting, including comprehensive threat modeling, risk assessment, and security architecture services. Our specialists help organizations identify and mitigate database security risks through proven methodologies and practical solutions.