Ransomware in 2026: Modern Attack Flow Explained for Begi...

#!/usr/bin/env python3
"""
Ransomware Detection System
Production-ready ransomware detection
"""

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

class RansomwareIndicator(Enum):
    FILE_ENCRYPTION = "file_encryption"
    EXTENSION_CHANGE = "extension_change"
    RANSOM_NOTE = "ransom_note"
    SUSPICIOUS_PROCESS = "suspicious_process"

@dataclass
class RansomwareDetection:
    detection_id: str
    indicator: RansomwareIndicator
    file_path: str
    timestamp: datetime
    process_name: str
    confidence: float

class RansomwareDetector:
    """Ransomware detection system."""
    
    def __init__(self):
        self.detections: List[RansomwareDetection] = []
        self.suspicious_extensions = ['.encrypted', '.locked', '.crypto', '.vault']
        self.ransom_note_names = ['README.txt', 'DECRYPT_INSTRUCTIONS.txt', 'HOW_TO_DECRYPT.txt']
    
    def detect_file_encryption(self, file_path: str, process_name: str) -> bool:
        """Detect file encryption activity."""
        # Check for extension changes
        for ext in self.suspicious_extensions:
            if file_path.endswith(ext):
                detection = RansomwareDetection(
                    detection_id=f"DET-{len(self.detections)+1}",
                    indicator=RansomwareIndicator.EXTENSION_CHANGE,
                    file_path=file_path,
                    timestamp=datetime.now(),
                    process_name=process_name,
                    confidence=0.9
                )
                self.detections.append(detection)
                return True
        return False
    
    def detect_ransom_note(self, file_path: str) -> bool:
        """Detect ransom note creation."""
        import os
        filename = os.path.basename(file_path)
        if filename in self.ransom_note_names:
            detection = RansomwareDetection(
                detection_id=f"DET-{len(self.detections)+1}",
                indicator=RansomwareIndicator.RANSOM_NOTE,
                file_path=file_path,
                timestamp=datetime.now(),
                process_name="unknown",
                confidence=0.95
            )
            self.detections.append(detection)
            return True
        return False
    
    def get_detections(self) -> List[RansomwareDetection]:
        """Get all detections."""
        return self.detections

# Usage
detector = RansomwareDetector()
if detector.detect_file_encryption("/path/to/file.encrypted", "malicious.exe"):
    print("Ransomware activity detected!")

#!/usr/bin/env python3
"""
Advanced Ransomware Detection System
Production-ready ransomware detection with behavioral analysis
"""

from typing import List, Dict, Optional, Set
from dataclasses import dataclass, field, asdict
from enum import Enum
from datetime import datetime, timedelta
import logging
import json
import re
from collections import defaultdict
from pathlib import Path

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

class RansomwareIndicator(Enum):
    """Ransomware detection indicators."""
    FILE_ENCRYPTION = "file_encryption"
    EXTENSION_CHANGE = "extension_change"
    RANSOM_NOTE = "ransom_note"
    SUSPICIOUS_PROCESS = "suspicious_process"
    RAPID_FILE_MODIFICATION = "rapid_file_modification"
    NETWORK_EXFILTRATION = "network_exfiltration"
    REGISTRY_MODIFICATION = "registry_modification"

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

@dataclass
class RansomwareDetection:
    """Comprehensive ransomware detection result."""
    detection_id: str
    indicator: RansomwareIndicator
    file_path: Optional[str]
    process_name: Optional[str]
    timestamp: datetime
    confidence: float
    threat_level: ThreatLevel
    metadata: Dict = field(default_factory=dict)
    
    def to_dict(self) -> Dict:
        """Convert to dictionary."""
        return {
            **asdict(self),
            'indicator': self.indicator.value,
            'threat_level': self.threat_level.value,
            'timestamp': self.timestamp.isoformat()
        }

class RansomwareDetector:
    """Production-ready ransomware detection system."""
    
    def __init__(self):
        self.detections: List[RansomwareDetection] = []
        self.file_monitor: Dict[str, datetime] = {}
        self.process_monitor: Dict[str, List[datetime]] = defaultdict(list)
        
        # Known ransomware indicators
        self.suspicious_extensions = {
            '.encrypted', '.locked', '.crypto', '.vault', '.xxx',
            '.zzz', '.aaa', '.micro', '.encryptedRSA', '.ecc',
            '.ezz', '.exx', '.xyz', '.zzz', '.aaa', '.abc', '.ccc'
        }
        
        self.ransom_note_names = {
            'readme.txt', 'decrypt_instructions.txt', 'how_to_decrypt.txt',
            'recover_files.txt', 'restore_files.txt', 'decrypt_my_files.txt',
            'your_files.txt', 'recover_data.txt'
        }
        
        self.suspicious_processes = {
            'vssadmin.exe', 'bcdedit.exe', 'wmic.exe', 'powershell.exe',
            'cmd.exe', 'taskkill.exe', 'net.exe', 'schtasks.exe'
        }
    
    def monitor_file_activity(self, file_path: str, operation: str) -> Optional[RansomwareDetection]:
        """Monitor file activity for ransomware indicators.
        
        Args:
            file_path: Path to file
            operation: File operation (read, write, delete, rename)
            
        Returns:
            Detection if ransomware activity detected
        """
        file_path_lower = file_path.lower()
        
        # Check for extension change
        if operation == 'rename':
            for ext in self.suspicious_extensions:
                if file_path_lower.endswith(ext):
                    detection = RansomwareDetection(
                        detection_id=f"DET-{len(self.detections)+1}",
                        indicator=RansomwareIndicator.EXTENSION_CHANGE,
                        file_path=file_path,
                        process_name=None,
                        timestamp=datetime.now(),
                        confidence=0.9,
                        threat_level=ThreatLevel.CRITICAL,
                        metadata={'new_extension': ext}
                    )
                    self.detections.append(detection)
                    logger.critical(f"Ransomware detected: Extension change to {ext}")
                    return detection
        
        # Check for ransom note
        file_name = Path(file_path).name.lower()
        if file_name in self.ransom_note_names:
            detection = RansomwareDetection(
                detection_id=f"DET-{len(self.detections)+1}",
                indicator=RansomwareIndicator.RANSOM_NOTE,
                file_path=file_path,
                process_name=None,
                timestamp=datetime.now(),
                confidence=0.95,
                threat_level=ThreatLevel.CRITICAL
            )
            self.detections.append(detection)
            logger.critical(f"Ransomware detected: Ransom note found at {file_path}")
            return detection
        
        # Monitor rapid file modifications
        if operation == 'write':
            current_time = datetime.now()
            if file_path in self.file_monitor:
                time_diff = (current_time - self.file_monitor[file_path]).total_seconds()
                if time_diff < 1.0:  # Multiple writes in < 1 second
                    detection = RansomwareDetection(
                        detection_id=f"DET-{len(self.detections)+1}",
                        indicator=RansomwareIndicator.RAPID_FILE_MODIFICATION,
                        file_path=file_path,
                        process_name=None,
                        timestamp=datetime.now(),
                        confidence=0.7,
                        threat_level=ThreatLevel.HIGH,
                        metadata={'modification_rate': 1.0 / time_diff}
                    )
                    self.detections.append(detection)
                    logger.warning(f"Rapid file modification detected: {file_path}")
                    return detection
            
            self.file_monitor[file_path] = current_time
        
        return None
    
    def monitor_process_activity(self, process_name: str, command_line: str) -> Optional[RansomwareDetection]:
        """Monitor process activity for ransomware indicators.
        
        Args:
            process_name: Process name
            command_line: Command line arguments
            
        Returns:
            Detection if suspicious activity detected
        """
        process_lower = process_name.lower()
        command_lower = command_line.lower()
        
        # Check for suspicious processes
        if process_lower in self.suspicious_processes:
            # Check for ransomware-related commands
            ransomware_commands = [
                'delete shadow', 'vssadmin delete shadows',
                'bcdedit /set', 'recoveryenabled no',
                'cipher /d', 'cipher /e'
            ]
            
            for cmd in ransomware_commands:
                if cmd in command_lower:
                    detection = RansomwareDetection(
                        detection_id=f"DET-{len(self.detections)+1}",
                        indicator=RansomwareIndicator.SUSPICIOUS_PROCESS,
                        file_path=None,
                        process_name=process_name,
                        timestamp=datetime.now(),
                        confidence=0.85,
                        threat_level=ThreatLevel.CRITICAL,
                        metadata={'command': command_line}
                    )
                    self.detections.append(detection)
                    logger.critical(f"Ransomware detected: Suspicious process {process_name}")
                    return detection
        
        # Monitor process frequency
        current_time = datetime.now()
        self.process_monitor[process_name].append(current_time)
        
        # Check for rapid process execution
        recent_executions = [
            t for t in self.process_monitor[process_name]
            if (current_time - t).total_seconds() < 60
        ]
        
        if len(recent_executions) > 50:  # More than 50 executions per minute
            detection = RansomwareDetection(
                detection_id=f"DET-{len(self.detections)+1}",
                indicator=RansomwareIndicator.SUSPICIOUS_PROCESS,
                file_path=None,
                process_name=process_name,
                timestamp=datetime.now(),
                confidence=0.6,
                threat_level=ThreatLevel.MEDIUM,
                metadata={'execution_count': len(recent_executions)}
            )
            self.detections.append(detection)
            logger.warning(f"High process execution rate: {process_name}")
            return detection
        
        return None
    
    def analyze_network_activity(self, network_event: Dict) -> Optional[RansomwareDetection]:
        """Analyze network activity for data exfiltration.
        
        Args:
            network_event: Network event dictionary
            
        Returns:
            Detection if exfiltration detected
        """
        # Check for large data transfers
        bytes_sent = network_event.get('bytes_sent', 0)
        destination = network_event.get('destination', '')
        
        if bytes_sent > 100 * 1024 * 1024:  # > 100MB
            detection = RansomwareDetection(
                detection_id=f"DET-{len(self.detections)+1}",
                indicator=RansomwareIndicator.NETWORK_EXFILTRATION,
                file_path=None,
                process_name=network_event.get('process_name'),
                timestamp=datetime.now(),
                confidence=0.7,
                threat_level=ThreatLevel.HIGH,
                metadata={
                    'bytes_sent': bytes_sent,
                    'destination': destination
                }
            )
            self.detections.append(detection)
            logger.warning(f"Large data transfer detected: {bytes_sent} bytes to {destination}")
            return detection
        
        return None
    
    def get_detection_summary(self) -> Dict:
        """Get summary of all detections.
        
        Returns:
            Summary dictionary
        """
        critical_detections = [d for d in self.detections if d.threat_level == ThreatLevel.CRITICAL]
        
        return {
            'total_detections': len(self.detections),
            'critical_detections': len(critical_detections),
            'by_indicator': {
                indicator.value: len([d for d in self.detections if d.indicator == indicator])
                for indicator in RansomwareIndicator
            },
            'by_threat_level': {
                level.value: len([d for d in self.detections if d.threat_level == level])
                for level in ThreatLevel
            },
            'latest_detection': self.detections[-1].to_dict() if self.detections else None
        }
    
    def export_detections(self, format: str = "json") -> str:
        """Export detections.
        
        Args:
            format: Export format (json)
            
        Returns:
            Exported detections string
        """
        detections_dict = [d.to_dict() for d in self.detections]
        return json.dumps(detections_dict, indent=2)
    
    def cleanup(self):
        """Clean up resources."""
        logger.info("Cleaning up ransomware detector resources")
        # In production, close connections, save state, etc.

# Example usage
if __name__ == "__main__":
    detector = RansomwareDetector()
    
    # Monitor file activity
    detector.monitor_file_activity("/path/to/file.txt", "rename")
    detector.monitor_file_activity("/path/to/file.txt.encrypted", "write")
    
    # Monitor process activity
    detector.monitor_process_activity("vssadmin.exe", "vssadmin delete shadows /all")
    
    # Get summary
    summary = detector.get_detection_summary()
    print(f"Detection Summary: {json.dumps(summary, indent=2)}")

#!/usr/bin/env python3
"""
Ransomware Prevention System
Production-ready prevention and response automation
"""

from typing import List, Dict, Optional
from dataclasses import dataclass
from enum import Enum
from datetime import datetime
import logging

logger = logging.getLogger(__name__)

class PreventionAction(Enum):
    """Prevention actions."""
    ISOLATE_ENDPOINT = "isolate_endpoint"
    BLOCK_PROCESS = "block_process"
    QUARANTINE_FILE = "quarantine_file"
    DISABLE_SHARES = "disable_shares"
    ALERT_ADMIN = "alert_admin"

@dataclass
class PreventionRule:
    """Prevention rule definition."""
    rule_id: str
    name: str
    condition: str
    action: PreventionAction
    enabled: bool = True

class RansomwarePrevention:
    """Ransomware prevention and response system."""
    
    def __init__(self):
        self.rules: List[PreventionRule] = []
        self.blocked_processes: Set[str] = set()
        self.quarantined_files: Set[str] = set()
        self._load_default_rules()
    
    def _load_default_rules(self):
        """Load default prevention rules."""
        self.rules = [
            PreventionRule(
                rule_id="RULE-001",
                name="Block Shadow Copy Deletion",
                condition="process == 'vssadmin.exe' AND command contains 'delete shadows'",
                action=PreventionAction.BLOCK_PROCESS
            ),
            PreventionRule(
                rule_id="RULE-002",
                name="Quarantine Encrypted Files",
                condition="file_extension in suspicious_extensions",
                action=PreventionAction.QUARANTINE_FILE
            ),
            PreventionRule(
                rule_id="RULE-003",
                name="Isolate on Critical Detection",
                condition="threat_level == 'critical'",
                action=PreventionAction.ISOLATE_ENDPOINT
            )
        ]
    
    def evaluate_detection(self, detection: RansomwareDetection) -> List[PreventionAction]:
        """Evaluate detection and determine prevention actions.
        
        Args:
            detection: Ransomware detection
            
        Returns:
            List of prevention actions to take
        """
        actions = []
        
        for rule in self.rules:
            if not rule.enabled:
                continue
            
            # Evaluate rule condition (simplified)
            if self._evaluate_condition(rule.condition, detection):
                actions.append(rule.action)
                logger.info(f"Rule {rule.rule_id} triggered: {rule.name}")
        
        return actions
    
    def _evaluate_condition(self, condition: str, detection: RansomwareDetection) -> bool:
        """Evaluate rule condition (simplified)."""
        # In production, would use proper expression evaluator
        if "critical" in condition.lower() and detection.threat_level == ThreatLevel.CRITICAL:
            return True
        if "suspicious_extensions" in condition and detection.indicator == RansomwareIndicator.EXTENSION_CHANGE:
            return True
        return False
    
    def execute_prevention(self, action: PreventionAction, context: Dict):
        """Execute prevention action.
        
        Args:
            action: Action to execute
            context: Action context
        """
        try:
            if action == PreventionAction.BLOCK_PROCESS:
                process_name = context.get('process_name')
                if process_name:
                    self.blocked_processes.add(process_name)
                    logger.warning(f"Blocked process: {process_name}")
            
            elif action == PreventionAction.QUARANTINE_FILE:
                file_path = context.get('file_path')
                if file_path:
                    self.quarantined_files.add(file_path)
                    logger.warning(f"Quarantined file: {file_path}")
            
            elif action == PreventionAction.ISOLATE_ENDPOINT:
                endpoint_id = context.get('endpoint_id')
                logger.critical(f"Isolating endpoint: {endpoint_id}")
                # In production, would send isolation command
            
            elif action == PreventionAction.ALERT_ADMIN:
                message = context.get('message', 'Ransomware detected')
                logger.critical(f"Alerting admin: {message}")
                # In production, would send alert
            
        except Exception as e:
            logger.error(f"Failed to execute prevention action: {e}", exc_info=True)

# Example usage
if __name__ == "__main__":
    detector = RansomwareDetector()
    prevention = RansomwarePrevention()
    
    # Detect ransomware
    detection = detector.monitor_file_activity("/file.txt.encrypted", "rename")
    
    if detection:
        # Evaluate and execute prevention
        actions = prevention.evaluate_detection(detection)
        for action in actions:
            prevention.execute_prevention(action, {
                'file_path': detection.file_path,
                'process_name': detection.process_name
            })

#!/usr/bin/env python3
"""
Unit tests for Ransomware Detection System
"""

import pytest
from datetime import datetime
from ransomware_detector import (
    RansomwareDetector, RansomwareIndicator, ThreatLevel,
    RansomwarePrevention, PreventionAction
)

class TestRansomwareDetector:
    """Tests for RansomwareDetector."""
    
    @pytest.fixture
    def detector(self):
        return RansomwareDetector()
    
    def test_extension_detection(self, detector):
        """Test extension change detection."""
        detection = detector.monitor_file_activity("/file.txt.encrypted", "rename")
        assert detection is not None
        assert detection.indicator == RansomwareIndicator.EXTENSION_CHANGE
    
    def test_ransom_note_detection(self, detector):
        """Test ransom note detection."""
        detection = detector.monitor_file_activity("/readme.txt", "write")
        assert detection is not None
        assert detection.indicator == RansomwareIndicator.RANSOM_NOTE
    
    def test_process_detection(self, detector):
        """Test suspicious process detection."""
        detection = detector.monitor_process_activity(
            "vssadmin.exe",
            "vssadmin delete shadows /all"
        )
        assert detection is not None
        assert detection.indicator == RansomwareIndicator.SUSPICIOUS_PROCESS

class TestRansomwarePrevention:
    """Tests for RansomwarePrevention."""
    
    @pytest.fixture
    def prevention(self):
        return RansomwarePrevention()
    
    def test_prevention_evaluation(self, prevention):
        """Test prevention rule evaluation."""
        from ransomware_detector import RansomwareDetection
        detection = RansomwareDetection(
            detection_id="TEST-001",
            indicator=RansomwareIndicator.EXTENSION_CHANGE,
            file_path="/test.encrypted",
            process_name=None,
            timestamp=datetime.now(),
            confidence=0.9,
            threat_level=ThreatLevel.CRITICAL
        )
        
        actions = prevention.evaluate_detection(detection)
        assert len(actions) > 0

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

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

import logging
from datetime import datetime, timedelta

logger = logging.getLogger(__name__)

class RansomwareDetectorCleanup:
    """Handles cleanup operations for ransomware detector."""
    
    def __init__(self, detector):
        """Initialize cleanup handler."""
        self.detector = detector
    
    def cleanup_old_detections(self, days: int = 90):
        """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_file_monitor(self, hours: int = 24):
        """Clean up old file monitor entries."""
        cutoff_time = datetime.now() - timedelta(hours=hours)
        initial_count = len(self.detector.file_monitor)
        
        self.detector.file_monitor = {
            path: timestamp for path, timestamp in self.detector.file_monitor.items()
            if timestamp >= cutoff_time
        }
        
        removed = initial_count - len(self.detector.file_monitor)
        logger.info(f"Cleaned up {removed} old file monitor entries")
        return removed
    
    def cleanup(self):
        """Perform complete cleanup."""
        logger.info("Starting ransomware detector cleanup")
        
        # Clean up old detections
        self.cleanup_old_detections()
        
        # Clean up file monitor
        self.cleanup_file_monitor()
        
        # Clean up detector resources
        self.detector.cleanup()
        
        logger.info("Ransomware detector cleanup complete")