Cloud Account Takeover Attacks Explained for Beginners (2...

#!/usr/bin/env python3
"""
Cloud Account Takeover Detection
Production-ready account takeover detection
"""

from typing import List, Dict
from dataclasses import dataclass
from enum import Enum
from datetime import datetime, timedelta

class TakeoverIndicator(Enum):
    SUSPICIOUS_LOGIN = "suspicious_login"
    MFA_FATIGUE = "mfa_fatigue"
    LOCATION_ANOMALY = "location_anomaly"
    DEVICE_CHANGE = "device_change"

@dataclass
class AccountEvent:
    timestamp: datetime
    user_id: str
    action: str
    source_ip: str
    location: str
    device_id: str
    success: bool

class AccountTakeoverDetector:
    """Cloud account takeover detection system."""
    
    def __init__(self):
        self.events: List[AccountEvent] = []
        self.user_baselines: Dict[str, Dict] = {}
    
    def add_event(self, event: AccountEvent):
        """Add account event for analysis."""
        self.events.append(event)
        self.update_baseline(event)
    
    def update_baseline(self, event: AccountEvent):
        """Update user behavior baseline."""
        if event.user_id not in self.user_baselines:
            self.user_baselines[event.user_id] = {
                'locations': set(),
                'devices': set(),
                'ips': set()
            }
        
        baseline = self.user_baselines[event.user_id]
        baseline['locations'].add(event.location)
        baseline['devices'].add(event.device_id)
        baseline['ips'].add(event.source_ip)
    
    def detect_takeover(self, user_id: str) -> List[Dict]:
        """Detect account takeover indicators."""
        detections = []
        recent_events = [e for e in self.events if e.user_id == user_id]
        
        if not recent_events:
            return detections
        
        latest_event = recent_events[-1]
        baseline = self.user_baselines.get(user_id, {})
        
        # Check for location anomaly
        if baseline.get('locations') and latest_event.location not in baseline['locations']:
            detections.append({
                'indicator': TakeoverIndicator.LOCATION_ANOMALY,
                'severity': 'high',
                'details': f"Login from new location: {latest_event.location}"
            })
        
        # Check for device change
        if baseline.get('devices') and latest_event.device_id not in baseline['devices']:
            detections.append({
                'indicator': TakeoverIndicator.DEVICE_CHANGE,
                'severity': 'medium',
                'details': f"Login from new device: {latest_event.device_id}"
            })
        
        # Check for MFA fatigue
        recent_mfa_attempts = [e for e in recent_events 
                               if 'mfa' in e.action.lower() 
                               and (datetime.now() - e.timestamp) < timedelta(hours=1)]
        if len(recent_mfa_attempts) > 10:
            detections.append({
                'indicator': TakeoverIndicator.MFA_FATIGUE,
                'severity': 'high',
                'details': f"{len(recent_mfa_attempts)} MFA attempts in last hour"
            })
        
        return detections

# Usage
detector = AccountTakeoverDetector()
event = AccountEvent(
    timestamp=datetime.now(),
    user_id="user123",
    action="login",
    source_ip="192.168.1.100",
    location="US",
    device_id="device1",
    success=True
)
detector.add_event(event)
detections = detector.detect_takeover("user123")
print(f"Detections: {len(detections)}")

#!/usr/bin/env python3
"""
Advanced Cloud Account Takeover Detection System
Production-ready detection with ML and behavioral analysis
"""

from typing import List, Dict, Optional, Set
from dataclasses import dataclass, field, asdict
from enum import Enum
from datetime import datetime, timedelta
from collections import defaultdict
import logging
import json
import numpy as np
from sklearn.ensemble import IsolationForest, RandomForestClassifier

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class TakeoverIndicator(Enum):
    SUSPICIOUS_LOGIN = "suspicious_login"
    MFA_FATIGUE = "mfa_fatigue"
    LOCATION_ANOMALY = "location_anomaly"
    DEVICE_CHANGE = "device_change"
    TIME_ANOMALY = "time_anomaly"
    BEHAVIOR_ANOMALY = "behavior_anomaly"

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

@dataclass
class AccountEvent:
    """Account event record."""
    timestamp: datetime
    user_id: str
    action: str
    source_ip: str
    location: str
    device_id: str
    user_agent: str
    success: bool
    mfa_used: bool = False
    
    def to_dict(self) -> Dict:
        """Convert to dictionary."""
        return {
            **asdict(self),
            'timestamp': self.timestamp.isoformat()
        }

@dataclass
class TakeoverDetection:
    """Account takeover detection result."""
    detection_id: str
    user_id: str
    indicators: List[TakeoverIndicator]
    threat_level: ThreatLevel
    confidence: float
    details: Dict
    timestamp: datetime = field(default_factory=datetime.now)
    recommendation: str = ""
    
    def to_dict(self) -> Dict:
        """Convert to dictionary."""
        return {
            **asdict(self),
            'indicators': [i.value for i in self.indicators],
            'threat_level': self.threat_level.value,
            'timestamp': self.timestamp.isoformat()
        }

class AdvancedAccountTakeoverDetector:
    """Production-ready cloud account takeover detection system."""
    
    def __init__(self):
        self.events: List[AccountEvent] = []
        self.detections: List[TakeoverDetection] = []
        self.user_baselines: Dict[str, Dict] = {}
        self.ml_model = IsolationForest(contamination=0.1, random_state=42)
        self.is_trained = False
    
    def add_event(self, event: AccountEvent):
        """Add account event for analysis.
        
        Args:
            event: Account event to add
        """
        self.events.append(event)
        self._update_baseline(event)
        self._analyze_event(event)
    
    def _update_baseline(self, event: AccountEvent):
        """Update user behavior baseline.
        
        Args:
            event: Account event
        """
        if event.user_id not in self.user_baselines:
            self.user_baselines[event.user_id] = {
                'locations': set(),
                'devices': set(),
                'ips': set(),
                'user_agents': set(),
                'login_times': [],
                'common_actions': defaultdict(int)
            }
        
        baseline = self.user_baselines[event.user_id]
        baseline['locations'].add(event.location)
        baseline['devices'].add(event.device_id)
        baseline['ips'].add(event.source_ip)
        baseline['user_agents'].add(event.user_agent)
        
        if event.action == 'login':
            baseline['login_times'].append(event.timestamp.hour)
        
        baseline['common_actions'][event.action] += 1
    
    def _analyze_event(self, event: AccountEvent):
        """Analyze event for takeover indicators.
        
        Args:
            event: Account event to analyze
        """
        if event.action == 'login' and event.success:
            detections = self.detect_takeover(event.user_id)
            if detections:
                # Create detection record
                detection = TakeoverDetection(
                    detection_id=f"DET-{len(self.detections)+1}",
                    user_id=event.user_id,
                    indicators=[d['indicator'] for d in detections],
                    threat_level=self._calculate_threat_level(detections),
                    confidence=self._calculate_confidence(detections),
                    details={'detections': detections},
                    recommendation=self._generate_recommendation(detections)
                )
                self.detections.append(detection)
                logger.warning(f"Account takeover detected: user={event.user_id}")
    
    def detect_takeover(self, user_id: str) -> List[Dict]:
        """Detect account takeover indicators.
        
        Args:
            user_id: User ID to analyze
            
        Returns:
            List of detection indicators
        """
        detections = []
        recent_events = [
            e for e in self.events
            if e.user_id == user_id and
            (datetime.now() - e.timestamp) < timedelta(hours=24)
        ]
        
        if not recent_events:
            return detections
        
        latest_event = recent_events[-1]
        baseline = self.user_baselines.get(user_id, {})
        
        # Check for location anomaly
        if baseline.get('locations') and latest_event.location not in baseline['locations']:
            detections.append({
                'indicator': TakeoverIndicator.LOCATION_ANOMALY,
                'severity': 'high',
                'details': f"Login from new location: {latest_event.location}"
            })
        
        # Check for device change
        if baseline.get('devices') and latest_event.device_id not in baseline['devices']:
            detections.append({
                'indicator': TakeoverIndicator.DEVICE_CHANGE,
                'severity': 'medium',
                'details': f"Login from new device: {latest_event.device_id}"
            })
        
        # Check for MFA fatigue
        recent_mfa_attempts = [
            e for e in recent_events
            if e.mfa_used and
            (datetime.now() - e.timestamp) < timedelta(hours=1)
        ]
        if len(recent_mfa_attempts) > 10:
            detections.append({
                'indicator': TakeoverIndicator.MFA_FATIGUE,
                'severity': 'high',
                'details': f"{len(recent_mfa_attempts)} MFA attempts in last hour"
            })
        
        # Check for time anomaly
        if baseline.get('login_times'):
            avg_login_hour = np.mean(baseline['login_times'])
            current_hour = latest_event.timestamp.hour
            if abs(current_hour - avg_login_hour) > 6:  # 6 hours difference
                detections.append({
                    'indicator': TakeoverIndicator.TIME_ANOMALY,
                    'severity': 'medium',
                    'details': f"Login at unusual time: {current_hour}:00 (avg: {avg_login_hour:.1f}:00)"
                })
        
        # Check for behavior anomaly
        if baseline.get('common_actions'):
            action_frequency = baseline['common_actions'].get(latest_event.action, 0)
            total_actions = sum(baseline['common_actions'].values())
            if total_actions > 0:
                action_ratio = action_frequency / total_actions
                if action_ratio < 0.1 and latest_event.action not in ['login', 'logout']:
                    detections.append({
                        'indicator': TakeoverIndicator.BEHAVIOR_ANOMALY,
                        'severity': 'medium',
                        'details': f"Unusual action: {latest_event.action}"
                    })
        
        return detections
    
    def _calculate_threat_level(self, detections: List[Dict]) -> ThreatLevel:
        """Calculate threat level from detections.
        
        Args:
            detections: List of detection dictionaries
            
        Returns:
            ThreatLevel
        """
        if not detections:
            return ThreatLevel.LOW
        
        high_severity_count = len([d for d in detections if d['severity'] == 'high'])
        critical_indicators = [TakeoverIndicator.MFA_FATIGUE, TakeoverIndicator.LOCATION_ANOMALY]
        has_critical = any(d['indicator'] in critical_indicators for d in detections)
        
        if has_critical and high_severity_count >= 2:
            return ThreatLevel.CRITICAL
        elif high_severity_count >= 1:
            return ThreatLevel.HIGH
        elif len(detections) >= 2:
            return ThreatLevel.MEDIUM
        else:
            return ThreatLevel.LOW
    
    def _calculate_confidence(self, detections: List[Dict]) -> float:
        """Calculate detection confidence.
        
        Args:
            detections: List of detection dictionaries
            
        Returns:
            Confidence score (0-1)
        """
        if not detections:
            return 0.0
        
        base_confidence = 0.3
        confidence_boost = {
            'high': 0.3,
            'medium': 0.2,
            'low': 0.1
        }
        
        for detection in detections:
            base_confidence += confidence_boost.get(detection['severity'], 0.1)
        
        return min(base_confidence, 1.0)
    
    def _generate_recommendation(self, detections: List[Dict]) -> str:
        """Generate response recommendation.
        
        Args:
            detections: List of detection dictionaries
            
        Returns:
            Recommendation string
        """
        if any(d['indicator'] == TakeoverIndicator.MFA_FATIGUE for d in detections):
            return "Immediately revoke sessions and require password reset"
        elif any(d['indicator'] == TakeoverIndicator.LOCATION_ANOMALY for d in detections):
            return "Verify user identity and review recent activity"
        else:
            return "Monitor account activity and verify user identity"
    
    def get_statistics(self) -> Dict:
        """Get detection statistics.
        
        Returns:
            Statistics dictionary
        """
        return {
            'total_events': len(self.events),
            'total_detections': len(self.detections),
            'users_monitored': len(self.user_baselines),
            'by_threat_level': {
                level.value: len([d for d in self.detections if d.threat_level == level])
                for level in ThreatLevel
            },
            'by_indicator': {
                ind.value: len([d for d in self.detections if ind in d.indicators])
                for ind in TakeoverIndicator
            }
        }
    
    def cleanup(self):
        """Clean up resources."""
        logger.info("Cleaning up account takeover detector resources")

# Example usage
if __name__ == "__main__":
    detector = AdvancedAccountTakeoverDetector()
    
    # Add events
    event = AccountEvent(
        timestamp=datetime.now(),
        user_id="user123",
        action="login",
        source_ip="192.168.1.100",
        location="US",
        device_id="DEV-001",
        user_agent="Mozilla/5.0",
        success=True,
        mfa_used=True
    )
    detector.add_event(event)
    
    # Get statistics
    stats = detector.get_statistics()
    print(f"Statistics: {json.dumps(stats, indent=2)}")

#!/usr/bin/env python3
"""
Unit tests for Account Takeover Detector
"""

import pytest
from datetime import datetime, timedelta
from account_takeover_detector import (
    AdvancedAccountTakeoverDetector, AccountEvent, TakeoverIndicator, ThreatLevel
)

class TestAccountTakeoverDetector:
    """Tests for AdvancedAccountTakeoverDetector."""
    
    @pytest.fixture
    def detector(self):
        return AdvancedAccountTakeoverDetector()
    
    def test_add_event(self, detector):
        """Test event addition."""
        event = AccountEvent(
            timestamp=datetime.now(),
            user_id="test_user",
            action="login",
            source_ip="192.168.1.100",
            location="US",
            device_id="DEV-001",
            user_agent="test",
            success=True
        )
        detector.add_event(event)
        assert len(detector.events) > 0
    
    def test_detect_takeover(self, detector):
        """Test takeover detection."""
        # Add baseline events
        for i in range(5):
            event = AccountEvent(
                timestamp=datetime.now() - timedelta(hours=i),
                user_id="test_user",
                action="login",
                source_ip="192.168.1.100",
                location="US",
                device_id="DEV-001",
                user_agent="test",
                success=True
            )
            detector.add_event(event)
        
        # Add suspicious event
        suspicious_event = AccountEvent(
            timestamp=datetime.now(),
            user_id="test_user",
            action="login",
            source_ip="10.0.0.1",
            location="RU",
            device_id="DEV-999",
            user_agent="test",
            success=True
        )
        detector.add_event(suspicious_event)
        
        detections = detector.detect_takeover("test_user")
        assert len(detections) > 0

if __name__ == "__main__":
    pytest.main([__file__, "-v"])

#!/usr/bin/env python3
"""
Account Takeover Detector Cleanup
Production-ready cleanup and resource management
"""

import logging
from datetime import datetime, timedelta

logger = logging.getLogger(__name__)

class AccountTakeoverDetectorCleanup:
    """Handles cleanup operations."""
    
    def __init__(self, detector):
        self.detector = detector
    
    def cleanup_old_events(self, days: int = 90):
        """Remove events older than specified days."""
        cutoff_date = datetime.now() - timedelta(days=days)
        initial_count = len(self.detector.events)
        
        self.detector.events = [
            e for e in self.detector.events
            if e.timestamp >= cutoff_date
        ]
        
        removed = initial_count - len(self.detector.events)
        logger.info(f"Cleaned up {removed} old events")
        return removed
    
    def cleanup_old_detections(self, days: int = 180):
        """Remove detections older than specified days."""
        cutoff_date = datetime.now() - timedelta(days=days)
        initial_count = len(self.detector.detections)
        
        self.detector.detections = [
            d for d in self.detector.detections
            if d.timestamp >= cutoff_date
        ]
        
        removed = initial_count - len(self.detector.detections)
        logger.info(f"Cleaned up {removed} old detections")
        return removed
    
    def cleanup(self):
        """Perform complete cleanup."""
        logger.info("Starting account takeover detector cleanup")
        self.cleanup_old_events()
        self.cleanup_old_detections()
        self.detector.cleanup()
        logger.info("Account takeover detector cleanup complete")

Real-World Case Study

Challenge: Organization experienced cloud account takeover:

• MFA fatigue successful
• Data exfiltration
• System access
• Business disruption

Solution: Implemented cloud security:

• Passwordless authentication
• Strong MFA policies
• Conditional access
• Identity monitoring
• Incident response

Results:

• Zero account takeovers: Strong authentication effective
• Improved security: Conditional access prevents unauthorized access
• Faster detection: Monitoring identifies threats
• Compliance: Security controls meet requirements
• User experience: Passwordless improves UX

FAQ

Q: How do I prevent cloud account takeover?

A: Implement strong authentication (passwordless, MFA), conditional access, identity monitoring, anomaly detection, and incident response capabilities.

Q: What’s MFA fatigue?

A: Attackers send repeated MFA prompts to exhaust users, who eventually approve from fatigue. Use number matching or time-based limits to prevent.

Q: How do I detect account takeover?

A: Monitor for unusual sign-ins, location anomalies, device changes, privilege escalations, and behavioral anomalies. Use identity monitoring tools.

Conclusion

Cloud account takeover is a serious threat. Implement strong authentication, monitoring, and incident response to protect cloud accounts.

Action Steps

1. Implement passwordless authentication
2. Strengthen MFA policies
3. Configure conditional access
4. Monitor identity activity
5. Detect anomalies
6. Prepare incident response
7. Train users

Related Topics

• Identity-Based Attacks
• Cloud IAM Misconfigurations

Educational Use Only: This content is for educational purposes. Implement cloud security to protect against account takeover.