ransomware_behavior.py

#!/usr/bin/env python3"""Ransomware Behavior DetectorThis script monitors system behavior for ransomware-like activities includingmass file encryption, shadow copy deletion, and suspicious process behavior.""" import osimport sysimport timeimport hashlibimport threadingimport argparseimport signalimport jsonfrom datetime import datetimefrom collections import defaultdict, dequeimport subprocessimport reimport psutiltry:    import winreg  # Windows onlyexcept ImportError:    winreg = Nonefrom pathlib import Pathfrom colorama import init, Fore, Style init(autoreset=True) class RansomwareBehaviorDetector:    def __init__(self, watch_paths=None, check_interval=1, entropy_threshold=0.7,                 file_change_threshold=50, log_file="ransomware_alerts.log"):        """        Initialize Ransomware Behavior Detector                Args:            watch_paths: List of paths to monitor            check_interval: Seconds between checks            entropy_threshold: File entropy threshold (0-1)            file_change_threshold: Number of changes to trigger alert            log_file: Log file path        """        self.watch_paths = watch_paths or [str(Path.home()), "C:\\" if os.name == 'nt' else "/home"]        self.check_interval = check_interval        self.entropy_threshold = entropy_threshold        self.file_change_threshold = file_change_threshold        self.log_file = log_file                # Tracking structures        self.file_hashes = {}        self.file_changes = defaultdict(int)        self.process_activity = defaultdict(lambda: defaultdict(int))        self.suspicious_processes = set()                # Known ransomware indicators        self.ransomware_indicators = {            'process_names': [                'wannacry', 'locky', 'cryptolocker', 'teslacrypt',                'cerber', 'ryuk', 'revil', 'maze', 'conti',                'vssadmin.exe', 'wmic.exe', 'bcdedit.exe'            ],            'file_extensions': [                '.encrypted', '.locked', '.crypted', '.wncry',                '.ryk', '.locky', '.revil', '.maze', '.conti'            ],            'registry_keys': [                r'SOFTWARE\Microsoft\Windows\CurrentVersion\Run',                r'SOFTWARE\Microsoft\Windows\CurrentVersion\RunOnce'            ],            'commands': [                'vssadmin delete shadows',                'wmic shadowcopy delete',                'bcdedit /set {default} recoveryenabled no',                'wevtutil cl'            ]        }                # Statistics        self.stats = {            'files_monitored': 0,            'files_changed': 0,            'alerts_triggered': 0,            'suspicious_processes': 0,            'start_time': time.time()        }                # Control flags        self.running = True                # Create baseline        self.create_baseline()        def create_baseline(self):        """Create baseline of file hashes"""        print(f"{Fore.CYAN}[*] Creating file baseline...{Style.RESET_ALL}")                for path in self.watch_paths:            if os.path.exists(path):                for root, dirs, files in os.walk(path):                    for file in files[:100]:  # Limit for demo                        filepath = os.path.join(root, file)                        try:                            self.file_hashes[filepath] = self.calculate_file_hash(filepath)                            self.stats['files_monitored'] += 1                        except:                            pass                print(f"{Fore.GREEN}[✓] Baseline created for {self.stats['files_monitored']} files{Style.RESET_ALL}")        def calculate_file_hash(self, filepath):        """Calculate SHA-256 hash of file"""        sha256 = hashlib.sha256()        try:            with open(filepath, 'rb') as f:                for chunk in iter(lambda: f.read(4096), b''):                    sha256.update(chunk)            return sha256.hexdigest()        except:            return None        def calculate_entropy(self, data):        """Calculate Shannon entropy of data"""        if not data:            return 0                entropy = 0        for x in range(256):            p_x = float(data.count(x)) / len(data)            if p_x > 0:                entropy += - p_x * (p_x.bit_length())                return entropy / 8  # Normalize to 0-1        def check_file_changes(self):        """Check for file changes"""        changed_files = []                for filepath, original_hash in list(self.file_hashes.items()):            if not os.path.exists(filepath):                # File deleted                changed_files.append(('deleted', filepath))                self.file_changes[filepath] += 1                continue                        current_hash = self.calculate_file_hash(filepath)            if current_hash != original_hash:                # File modified                changed_files.append(('modified', filepath))                self.file_changes[filepath] += 1                return changed_files        def check_file_entropy(self, filepath):        """Check file entropy (high entropy indicates encryption)"""        try:            with open(filepath, 'rb') as f:                data = f.read(4096)  # Read first 4KB                entropy = self.calculate_entropy(data)                return entropy        except:            return 0        def monitor_processes(self):        """Monitor running processes for suspicious activity"""        suspicious = []                for proc in psutil.process_iter(['pid', 'name', 'cmdline']):            try:                proc_name = proc.info['name'].lower() if proc.info['name'] else ''                cmdline = ' '.join(proc.info['cmdline'] or []).lower()                                # Check process names                for indicator in self.ransomware_indicators['process_names']:                    if indicator in proc_name or indicator in cmdline:                        suspicious.append(proc.info)                        self.suspicious_processes.add(proc.info['pid'])                        break                                # Check for VSS deletion commands                for cmd in self.ransomware_indicators['commands']:                    if cmd in cmdline:                        self.trigger_alert(                            "VSS_DELETION",                            f"Process {proc_name} attempting to delete shadow copies",                            {'pid': proc.info['pid'], 'cmdline': cmdline}                        )                                # Track process activity                self.process_activity[proc.info['pid']]['cpu'] = proc.cpu_percent()                self.process_activity[proc.info['pid']]['memory'] = proc.memory_percent()                            except (psutil.NoSuchProcess, psutil.AccessDenied):                pass                return suspicious        def check_registry_changes(self):        """Check Windows registry for suspicious changes"""        if os.name != 'nt' or winreg is None:            return []                suspicious = []                for reg_key in self.ransomware_indicators['registry_keys']:            try:                key = winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, reg_key)                                i = 0                while True:                    try:                        value_name, value_data, _ = winreg.EnumValue(key, i)                                                # Check for suspicious autostart entries                        for indicator in self.ransomware_indicators['process_names']:                            if indicator in value_data.lower():                                suspicious.append({                                    'key': reg_key,                                    'value': value_name,                                    'data': value_data                                })                                                i += 1                    except OSError:                        break                                winreg.CloseKey(key)            except:                pass                return suspicious        def detect_mass_file_changes(self):        """Detect mass file changes (encryption pattern)"""        recent_changes = sum(1 for count in self.file_changes.values() if count > 0)                if recent_changes > self.file_change_threshold:            self.trigger_alert(                "MASS_FILE_CHANGE",                f"Mass file changes detected: {recent_changes} files modified",                {'threshold': self.file_change_threshold, 'changes': recent_changes}            )                # Check for specific file extensions        suspicious_extensions = []        for filepath in list(self.file_changes.keys())[:100]:            ext = os.path.splitext(filepath)[1].lower()            if ext in self.ransomware_indicators['file_extensions']:                suspicious_extensions.append(filepath)                if suspicious_extensions:            self.trigger_alert(                "SUSPICIOUS_EXTENSION",                f"Files with ransomware extensions detected",                {'files': suspicious_extensions[:10]}            )        def check_network_connections(self):        """Check for suspicious network connections"""        suspicious = []                for conn in psutil.net_connections():            try:                if conn.status == 'ESTABLISHED' and conn.raddr:                    # Check for TOR connections                    if conn.raddr.port in [9001, 9030, 9040, 9050, 9051, 9150]:                        suspicious.append({                            'pid': conn.pid,                            'remote': f"{conn.raddr.ip}:{conn.raddr.port}",                            'type': 'TOR'                        })                                        # Check for known C2 ports                    if conn.raddr.port in [4444, 1337, 6667, 8080, 8443]:                        suspicious.append({                            'pid': conn.pid,                            'remote': f"{conn.raddr.ip}:{conn.raddr.port}",                            'type': 'POTENTIAL_C2'                        })            except:                pass                return suspicious        def trigger_alert(self, alert_type, message, data=None):        """Trigger security alert"""        self.stats['alerts_triggered'] += 1                timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')        alert = {            'timestamp': timestamp,            'type': alert_type,            'message': message,            'data': data or {}        }                # Print alert        print(f"\n{Fore.RED}🚨 RANSOMWARE ALERT [{timestamp}]{Style.RESET_ALL}")        print(f"{Fore.RED}   Type: {alert_type}{Style.RESET_ALL}")        print(f"{Fore.RED}   Message: {message}{Style.RESET_ALL}")                if data:            print(f"{Fore.YELLOW}   Details: {json.dumps(data, indent=2)}{Style.RESET_ALL}")                # Log to file        with open(self.log_file, 'a') as f:            f.write(f"[{timestamp}] {alert_type}: {message}\n")        def scan_loop(self):        """Main scanning loop"""        print(f"{Fore.GREEN}[*] Starting ransomware behavior monitoring...{Style.RESET_ALL}")                while self.running:            try:                # Check file changes                changed = self.check_file_changes()                if changed:                    for change_type, filepath in changed[:10]:  # Limit output                        print(f"{Fore.YELLOW}  [!] {change_type}: {filepath}{Style.RESET_ALL}")                                        self.stats['files_changed'] += len(changed)                                        # Check entropy of changed files                    for _, filepath in changed[:5]:                        entropy = self.check_file_entropy(filepath)                        if entropy > self.entropy_threshold:                            self.trigger_alert(                                "HIGH_ENTROPY",                                f"High entropy detected in {filepath}: {entropy:.2f}",                                {'file': filepath, 'entropy': entropy}                            )                                # Monitor processes                suspicious_procs = self.monitor_processes()                if suspicious_procs:                    self.stats['suspicious_processes'] += len(suspicious_procs)                    for proc in suspicious_procs:                        self.trigger_alert(                            "SUSPICIOUS_PROCESS",                            f"Suspicious process detected: {proc.get('name')}",                            proc                        )                                # Check registry (Windows only)                if os.name == 'nt':                    reg_changes = self.check_registry_changes()                    if reg_changes:                        self.trigger_alert(                            "REGISTRY_CHANGE",                            "Suspicious registry changes detected",                            {'changes': reg_changes}                        )                                # Check for mass file changes                self.detect_mass_file_changes()                                # Check network connections                net_conns = self.check_network_connections()                if net_conns:                    self.trigger_alert(                        "SUSPICIOUS_NETWORK",                        "Suspicious network connections detected",                        {'connections': net_conns}                    )                                time.sleep(self.check_interval)                            except Exception as e:                print(f"{Fore.RED}[!] Error in scan loop: {e}{Style.RESET_ALL}")                time.sleep(self.check_interval)        def print_stats(self):        """Print current statistics"""        os.system('clear' if os.name == 'posix' else 'cls')                uptime = time.time() - self.stats['start_time']                print(f"{Fore.CYAN}{'='*60}{Style.RESET_ALL}")        print(f"{Fore.CYAN}📊 RANSOMWARE DETECTOR STATISTICS{Style.RESET_ALL}")        print(f"{Fore.CYAN}{'='*60}{Style.RESET_ALL}")        print(f"  Uptime: {int(uptime)} seconds")        print(f"  Files Monitored: {self.stats['files_monitored']}")        print(f"  Files Changed: {self.stats['files_changed']}")        print(f"  Alerts Triggered: {self.stats['alerts_triggered']}")        print(f"  Suspicious Processes: {self.stats['suspicious_processes']}")        print(f"{Fore.CYAN}{'='*60}{Style.RESET_ALL}")                # Show recent alerts        if self.stats['alerts_triggered'] > 0:            print(f"\n{Fore.YELLOW}Recent Alerts:{Style.RESET_ALL}")            with open(self.log_file, 'r') as f:                lines = f.readlines()[-5:]                for line in lines:                    print(f"  {line.strip()}")        def start(self):        """Start the detector"""        print(f"""        {Fore.CYAN}╔═══════════════════════════════════════╗        ║     Ransomware Behavior Detector    ║        ║       FOR EDUCATIONAL USE ONLY       ║        ╚═══════════════════════════════════════╝{Style.RESET_ALL}        """)                print(f"{Fore.YELLOW}[!] WARNING: This tool monitors system behavior{Style.RESET_ALL}")        print(f"{Fore.YELLOW}[!] Only use in controlled environments{Style.RESET_ALL}\n")                # Start scan thread        scan_thread = threading.Thread(target=self.scan_loop)        scan_thread.daemon = True        scan_thread.start()                # Start stats thread        stats_thread = threading.Thread(target=self.print_stats_loop)        stats_thread.daemon = True        stats_thread.start()                print(f"{Fore.GREEN}[✓] Detector started. Press Ctrl+C to stop.{Style.RESET_ALL}")                try:            while self.running:                time.sleep(1)        except KeyboardInterrupt:            self.stop()        def print_stats_loop(self):        """Periodically print statistics"""        while self.running:            time.sleep(10)            self.print_stats()        def stop(self):        """Stop the detector"""        self.running = False        print(f"\n{Fore.GREEN}[✓] Detector stopped{Style.RESET_ALL}")        self.print_stats() def signal_handler(sig, frame):    """Handle Ctrl+C"""    detector.stop()    sys.exit(0) def main():    parser = argparse.ArgumentParser(description='Ransomware Behavior Detector')    parser.add_argument('-p', '--paths', help='Paths to monitor (comma-separated)')    parser.add_argument('-t', '--threshold', type=int, default=50,                        help='File change threshold')    parser.add_argument('-e', '--entropy', type=float, default=0.7,                       help='Entropy threshold (0-1)')    parser.add_argument('-i', '--interval', type=int, default=1,                       help='Check interval in seconds')    parser.add_argument('-l', '--log', default='ransomware_alerts.log',                       help='Log file path')        args = parser.parse_args()        # Parse paths    watch_paths = args.paths.split(',') if args.paths else None        global detector    detector = RansomwareBehaviorDetector(        watch_paths=watch_paths,        check_interval=args.interval,        entropy_threshold=args.entropy,        file_change_threshold=args.threshold,        log_file=args.log    )        signal.signal(signal.SIGINT, signal_handler)    detector.start() if __name__ == "__main__":    main()