AI Threat Modeling Mastery

AI-STRIDE Framework Implementation

import json
import uuid
from typing import Dict, List, Any, Optional
from dataclasses import dataclass, asdict
from enum import Enum
import networkx as nx
import matplotlib.pyplot as plt

class ThreatCategory(Enum):
    SPOOFING = "spoofing"
    TAMPERING = "tampering"
    REPUDIATION = "repudiation"
    INFORMATION_DISCLOSURE = "information_disclosure"
    DENIAL_OF_SERVICE = "denial_of_service"
    ELEVATION_OF_PRIVILEGE = "elevation_of_privilege"

class Severity(Enum):
    CRITICAL = "critical"
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"

@dataclass
class AIThreat:
    id: str
    category: ThreatCategory
    title: str
    description: str
    ai_component: str
    attack_vector: str
    impact: str
    likelihood: str
    severity: Severity
    mitigation_strategies: List[str]
    detection_methods: List[str]
    
class AIStrideAnalyzer:
    def __init__(self):
        self.threats: List[AIThreat] = []
        self.ai_components = [
            'model_training', 'data_pipeline', 'inference_engine',
            'api_gateway', 'model_storage', 'feedback_loop',
            'monitoring_system', 'user_interface'
        ]
        
        # AI-specific threat templates
        self.threat_templates = {
            ThreatCategory.SPOOFING: {
                'model_impersonation': {
                    'title': 'Model Impersonation Attack',
                    'description': 'Attacker deploys fake AI model to impersonate legitimate service',
                    'attack_vector': 'Model substitution, API endpoint spoofing',
                    'impact': 'Malicious outputs, data harvesting, reputation damage'
                },
                'data_source_spoofing': {
                    'title': 'Training Data Source Spoofing',
                    'description': 'Malicious data sources masquerading as legitimate',
                    'attack_vector': 'Compromised data feeds, DNS spoofing',
                    'impact': 'Model poisoning, biased outputs'
                }
            },
            ThreatCategory.TAMPERING: {
                'model_poisoning': {
                    'title': 'Model Poisoning Attack',
                    'description': 'Malicious modification of training data or model parameters',
                    'attack_vector': 'Data injection, gradient manipulation',
                    'impact': 'Compromised model behavior, backdoor insertion'
                },
                'adversarial_examples': {
                    'title': 'Adversarial Input Tampering',
                    'description': 'Crafted inputs designed to fool AI models',
                    'attack_vector': 'Pixel manipulation, feature perturbation',
                    'impact': 'Misclassification, system bypass'
                }
            },
            ThreatCategory.REPUDIATION: {
                'model_decision_denial': {
                    'title': 'AI Decision Repudiation',
                    'description': 'Lack of audit trails for AI decision-making',
                    'attack_vector': 'Missing logs, tampered audit trails',
                    'impact': 'Legal liability, compliance violations'
                }
            },
            ThreatCategory.INFORMATION_DISCLOSURE: {
                'model_extraction': {
                    'title': 'Model Extraction Attack',
                    'description': 'Unauthorized access to model architecture and parameters',
                    'attack_vector': 'API queries, side-channel analysis',
                    'impact': 'IP theft, competitive disadvantage'
                },
                'training_data_leak': {
                    'title': 'Training Data Inference',
                    'description': 'Extracting sensitive training data through model queries',
                    'attack_vector': 'Membership inference, model inversion',
                    'impact': 'Privacy violations, regulatory penalties'
                }
            },
            ThreatCategory.DENIAL_OF_SERVICE: {
                'inference_overload': {
                    'title': 'Inference Resource Exhaustion',
                    'description': 'Overwhelming AI system with expensive computations',
                    'attack_vector': 'Adversarial queries, resource-intensive inputs',
                    'impact': 'Service unavailability, financial losses'
                }
            },
            ThreatCategory.ELEVATION_OF_PRIVILEGE: {
                'prompt_injection': {
                    'title': 'Prompt Injection Privilege Escalation',
                    'description': 'Manipulating AI prompts to gain unauthorized access',
                    'attack_vector': 'Crafted prompts, system instruction override',
                    'impact': 'Unauthorized data access, system control'
                }
            }
        }
    
    def analyze_component(self, component: str) -> List[AIThreat]:
        """Generate STRIDE threats for a specific AI component"""
        component_threats = []
        
        for category in ThreatCategory:
            threats = self._generate_threats_for_category(component, category)
            component_threats.extend(threats)
        
        return component_threats
    
    def _generate_threats_for_category(self, component: str, category: ThreatCategory) -> List[AIThreat]:
        """Generate threats for a specific category and component"""
        threats = []
        
        if category in self.threat_templates:
            for threat_key, template in self.threat_templates[category].items():
                # Customize threat based on component
                customized_description = f"{template['description']} targeting {component}"
                
                threat = AIThreat(
                    id=str(uuid.uuid4()),
                    category=category,
                    title=template['title'],
                    description=customized_description,
                    ai_component=component,
                    attack_vector=template['attack_vector'],
                    impact=template['impact'],
                    likelihood=self._assess_likelihood(component, category),
                    severity=self._assess_severity(component, category),
                    mitigation_strategies=self._get_mitigations(component, category),
                    detection_methods=self._get_detection_methods(component, category)
                )
                threats.append(threat)
        
        return threats
    
    def _assess_likelihood(self, component: str, category: ThreatCategory) -> str:
        """Assess likelihood based on component and threat category"""
        likelihood_matrix = {
            ('model_training', ThreatCategory.TAMPERING): 'High',
            ('inference_engine', ThreatCategory.SPOOFING): 'Medium',
            ('api_gateway', ThreatCategory.DENIAL_OF_SERVICE): 'High',
            ('data_pipeline', ThreatCategory.INFORMATION_DISCLOSURE): 'Medium'
        }
        
        return likelihood_matrix.get((component, category), 'Medium')
    
    def _assess_severity(self, component: str, category: ThreatCategory) -> Severity:
        """Assess severity based on component and threat category"""
        if category in [ThreatCategory.ELEVATION_OF_PRIVILEGE, ThreatCategory.TAMPERING]:
            return Severity.CRITICAL
        elif category in [ThreatCategory.INFORMATION_DISCLOSURE, ThreatCategory.DENIAL_OF_SERVICE]:
            return Severity.HIGH
        else:
            return Severity.MEDIUM
    
    def _get_mitigations(self, component: str, category: ThreatCategory) -> List[str]:
        """Get mitigation strategies for component and threat category"""
        mitigations = {
            ThreatCategory.SPOOFING: [
                'Model signing and verification',
                'Certificate-based authentication',
                'Cryptographic model sealing'
            ],
            ThreatCategory.TAMPERING: [
                'Input validation and sanitization',
                'Model integrity checks',
                'Differential privacy techniques'
            ],
            ThreatCategory.REPUDIATION: [
                'Comprehensive audit logging',
                'Digital signatures for decisions',
                'Immutable decision trails'
            ],
            ThreatCategory.INFORMATION_DISCLOSURE: [
                'Access controls and authorization',
                'Data anonymization',
                'Query rate limiting'
            ],
            ThreatCategory.DENIAL_OF_SERVICE: [
                'Resource quotas and throttling',
                'Input complexity analysis',
                'Distributed inference architecture'
            ],
            ThreatCategory.ELEVATION_OF_PRIVILEGE: [
                'Prompt sanitization',
                'Context isolation',
                'Least privilege principles'
            ]
        }
        
        return mitigations.get(category, [])
    
    def _get_detection_methods(self, component: str, category: ThreatCategory) -> List[str]:
        """Get detection methods for component and threat category"""
        detection_methods = {
            ThreatCategory.SPOOFING: [
                'Model fingerprinting',
                'Behavioral analysis',
                'Certificate validation'
            ],
            ThreatCategory.TAMPERING: [
                'Statistical analysis of outputs',
                'Model drift detection',
                'Input anomaly detection'
            ],
            ThreatCategory.INFORMATION_DISCLOSURE: [
                'Query pattern analysis',
                'Response correlation monitoring',
                'Access pattern analysis'
            ]
        }
        
        return detection_methods.get(category, ['System monitoring', 'Log analysis'])
    
    def generate_threat_model(self) -> Dict[str, Any]:
        """Generate complete STRIDE threat model for AI system"""
        all_threats = []
        
        for component in self.ai_components:
            component_threats = self.analyze_component(component)
            all_threats.extend(component_threats)
        
        # Organize by severity
        threats_by_severity = {}
        for severity in Severity:
            threats_by_severity[severity.value] = [
                threat for threat in all_threats if threat.severity == severity
            ]
        
        return {
            'total_threats': len(all_threats),
            'threats_by_component': self._group_by_component(all_threats),
            'threats_by_category': self._group_by_category(all_threats),
            'threats_by_severity': threats_by_severity,
            'risk_score': self._calculate_risk_score(all_threats),
            'recommendations': self._generate_recommendations(all_threats)
        }
    
    def _group_by_component(self, threats: List[AIThreat]) -> Dict[str, int]:
        """Group threats by AI component"""
        component_counts = {}
        for threat in threats:
            component_counts[threat.ai_component] = component_counts.get(threat.ai_component, 0) + 1
        return component_counts
    
    def _group_by_category(self, threats: List[AIThreat]) -> Dict[str, int]:
        """Group threats by STRIDE category"""
        category_counts = {}
        for threat in threats:
            category_counts[threat.category.value] = category_counts.get(threat.category.value, 0) + 1
        return category_counts
    
    def _calculate_risk_score(self, threats: List[AIThreat]) -> float:
        """Calculate overall risk score"""
        severity_weights = {
            Severity.CRITICAL: 4,
            Severity.HIGH: 3,
            Severity.MEDIUM: 2,
            Severity.LOW: 1
        }
        
        total_score = sum(severity_weights[threat.severity] for threat in threats)
        max_possible = len(threats) * 4
        
        return (total_score / max_possible) * 100 if max_possible > 0 else 0
    
    def _generate_recommendations(self, threats: List[AIThreat]) -> List[str]:
        """Generate prioritized recommendations"""
        critical_threats = [t for t in threats if t.severity == Severity.CRITICAL]
        high_threats = [t for t in threats if t.severity == Severity.HIGH]
        
        recommendations = []
        
        if critical_threats:
            recommendations.append(f"Immediately address {len(critical_threats)} critical threats")
        
        if high_threats:
            recommendations.append(f"Prioritize {len(high_threats)} high-severity threats")
            
        recommendations.extend([
            "Implement comprehensive monitoring and detection",
            "Establish incident response procedures",
            "Regular threat model updates and reviews"
        ])
        
        return recommendations