file_monitor.py

#!/usr/bin/env python3"""File Monitor for Ransomware DetectionLocation: 03-malware-analysis/ransomware/detection/file_monitor.pyThis script monitors file system activity in real-time to detectransomware-like behavior including rapid file modifications and encryption.""" import osimport sysimport timeimport hashlibimport threadingimport argparseimport jsonfrom datetime import datetimefrom collections import defaultdict, dequefrom watchdog.observers import Observerfrom watchdog.events import FileSystemEventHandlerimport psutilfrom pathlib import Pathfrom colorama import init, Fore, Style init(autoreset=True) class RansomwareFileHandler(FileSystemEventHandler):    """Handle file system events"""        def __init__(self, monitor):        self.monitor = monitor        def on_modified(self, event):        if not event.is_directory:            self.monitor.handle_file_event('modified', event.src_path)        def on_created(self, event):        if not event.is_directory:            self.monitor.handle_file_event('created', event.src_path)        def on_deleted(self, event):        if not event.is_directory:            self.monitor.handle_file_event('deleted', event.src_path)        def on_moved(self, event):        if not event.is_directory:            self.monitor.handle_file_event('moved', f"{event.src_path} -> {event.dest_path}") class FileMonitor:    def __init__(self, watch_paths=None, alert_threshold=50, time_window=60,                 entropy_threshold=0.7, log_file="file_monitor.log"):        """        Initialize File Monitor                Args:            watch_paths: Paths to monitor            alert_threshold: Events per time window to trigger alert            time_window: Time window in seconds            entropy_threshold: File entropy threshold            log_file: Log file path        """        self.watch_paths = watch_paths or [str(Path.home())]        self.alert_threshold = alert_threshold        self.time_window = time_window        self.entropy_threshold = entropy_threshold        self.log_file = log_file                # Event tracking        self.events = deque(maxlen=10000)        self.file_hashes = {}        self.extension_counts = defaultdict(int)        self.process_activity = defaultdict(lambda: defaultdict(int))                # Suspicious patterns        self.suspicious_extensions = [            '.encrypted', '.locked', '.crypted', '.wncry', '.ryk',            '.locky', '.revil', '.maze', '.conti', '.crypt',            '.zzzzz', '.pay', '.odin', '.zeppelin', '.avdn'        ]                self.ransomware_notes = [            'README', 'HOW_TO_DECRYPT', 'DECRYPT_INFO', 'RANSOM_NOTE',            'RECOVER', 'FILES_ENCRYPTED', 'INSTRUCTIONS'        ]                # Statistics        self.stats = {            'total_events': 0,            'creations': 0,            'modifications': 0,            'deletions': 0,            'moves': 0,            'alerts': 0,            'suspicious_files': 0,            'start_time': time.time()        }                # Control flags        self.running = True        def calculate_entropy(self, filepath):        """Calculate Shannon entropy of file"""        try:            with open(filepath, 'rb') as f:                data = f.read(4096)                if not data:                    return 0                                entropy = 0                for x in range(256):                    p_x = data.count(x) / len(data)                    if p_x > 0:                        entropy += - p_x * (p_x.bit_length())                                return entropy / 8  # Normalize to 0-1        except:            return 0        def get_process_info(self):        """Get information about current process"""        try:            proc = psutil.Process()            return {                'pid': proc.pid,                'name': proc.name(),                'cpu_percent': proc.cpu_percent(),                'memory_percent': proc.memory_percent()            }        except:            return {}        def handle_file_event(self, event_type, filepath):        """Handle file system event"""        timestamp = time.time()        self.events.append({            'type': event_type,            'path': filepath,            'time': timestamp,            'process': self.get_process_info()        })                self.stats['total_events'] += 1        stat_key = {'created': 'creations', 'modified': 'modifications',                    'deleted': 'deletions', 'moved': 'moves'}.get(event_type)        if stat_key:            self.stats[stat_key] += 1                # Check for suspicious patterns        self.check_suspicious_file(filepath, event_type)                # Clean old events        self.clean_old_events()                # Check thresholds        self.check_event_thresholds()        def check_suspicious_file(self, filepath, event_type):        """Check if file is suspicious"""        suspicious = False        reasons = []                # Check extension        ext = os.path.splitext(filepath)[1].lower()        if ext in self.suspicious_extensions:            suspicious = True            reasons.append(f"suspicious extension: {ext}")            self.extension_counts[ext] += 1                # Check filename for ransom notes        filename = os.path.basename(filepath).upper()        for note in self.ransomware_notes:            if note in filename:                suspicious = True                reasons.append(f"ransom note pattern: {note}")                # Check entropy for new/modified files        if event_type in ['created', 'modified'] and os.path.exists(filepath):            entropy = self.calculate_entropy(filepath)            if entropy > self.entropy_threshold:                suspicious = True                reasons.append(f"high entropy: {entropy:.2f}")                if suspicious:            self.stats['suspicious_files'] += 1            self.trigger_alert(                "SUSPICIOUS_FILE",                f"Suspicious file activity: {filepath}",                {'reasons': reasons, 'event_type': event_type}            )        def check_event_thresholds(self):        """Check if event thresholds are exceeded"""        current_time = time.time()        recent_events = [e for e in self.events                         if current_time - e['time'] < self.time_window]                if len(recent_events) > self.alert_threshold:            self.trigger_alert(                "HIGH_ACTIVITY",                f"High file activity detected: {len(recent_events)} events in {self.time_window}s",                {'threshold': self.alert_threshold, 'actual': len(recent_events)}            )                # Check for specific patterns        recent_modifications = [e for e in recent_events if e['type'] == 'modified']        if len(recent_modifications) > self.alert_threshold / 2:            self.trigger_alert(                "MASS_MODIFICATION",                f"Mass file modifications: {len(recent_modifications)} files modified",                {}            )        def clean_old_events(self):        """Remove old events"""        current_time = time.time()        while self.events and current_time - self.events[0]['time'] > 3600:  # 1 hour            self.events.popleft()        def trigger_alert(self, alert_type, message, data=None):        """Trigger security alert"""        self.stats['alerts'] += 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}🚨 FILE MONITOR ALERT [{timestamp}]{Style.RESET_ALL}")        print(f"{Fore.RED}   Type: {alert_type}{Style.RESET_ALL}")        print(f"{Fore.RED}   Message: {message}{Style.RESET_ALL}")                # Log to file        with open(self.log_file, 'a') as f:            f.write(f"[{timestamp}] {alert_type}: {message}\n")            if data:                f.write(f"    Data: {json.dumps(data)}\n")        def print_stats(self):        """Print current statistics"""        os.system('clear' if os.name == 'posix' else 'cls')                uptime = time.time() - self.stats['start_time']        rate = self.stats['total_events'] / uptime if uptime > 0 else 0                print(f"{Fore.CYAN}{'='*60}{Style.RESET_ALL}")        print(f"{Fore.CYAN}📊 FILE MONITOR STATISTICS{Style.RESET_ALL}")        print(f"{Fore.CYAN}{'='*60}{Style.RESET_ALL}")        print(f"  Uptime: {int(uptime)} seconds")        print(f"  Event Rate: {rate:.2f} events/sec")        print(f"  Total Events: {self.stats['total_events']}")        print(f"  Creations: {self.stats['creations']}")        print(f"  Modifications: {self.stats['modifications']}")        print(f"  Deletions: {self.stats['deletions']}")        print(f"  Moves: {self.stats['moves']}")        print(f"  Alerts: {self.stats['alerts']}")        print(f"  Suspicious Files: {self.stats['suspicious_files']}")        print(f"{Fore.CYAN}{'='*60}{Style.RESET_ALL}")                # Show suspicious extensions        if self.extension_counts:            print(f"\n{Fore.YELLOW}Suspicious Extensions:{Style.RESET_ALL}")            for ext, count in sorted(self.extension_counts.items(), key=lambda x: x[1], reverse=True)[:5]:                print(f"  {ext}: {count} occurrences")                # Show recent events        recent = list(self.events)[-5:] if self.events else []        if recent:            print(f"\n{Fore.GREEN}Recent Events:{Style.RESET_ALL}")            for event in recent:                print(f"  [{event['type']}] {os.path.basename(event['path'])}")        def start_monitoring(self):        """Start file system monitoring"""        print(f"\n{Fore.CYAN}🚀 Starting File Monitor...{Style.RESET_ALL}")        print(f"{Fore.CYAN}   Watching: {', '.join(self.watch_paths)}{Style.RESET_ALL}")        print(f"{Fore.CYAN}   Alert Threshold: {self.alert_threshold} events/{self.time_window}s{Style.RESET_ALL}")                # Setup watchdog observer        event_handler = RansomwareFileHandler(self)        observer = Observer()                for path in self.watch_paths:            if os.path.exists(path):                observer.schedule(event_handler, path, recursive=True)                print(f"{Fore.GREEN}  [✓] Watching: {path}{Style.RESET_ALL}")                observer.start()                # Start stats thread        stats_thread = threading.Thread(target=self.stats_loop)        stats_thread.daemon = True        stats_thread.start()                print(f"\n{Fore.YELLOW}   Press Ctrl+C to stop{Style.RESET_ALL}\n")                try:            while self.running:                time.sleep(1)        except KeyboardInterrupt:            observer.stop()            self.running = False                observer.join()        def stats_loop(self):        """Periodically print statistics"""        while self.running:            time.sleep(10)            self.print_stats()        def generate_report(self):        """Generate monitoring report"""        report = {            'timestamp': datetime.now().isoformat(),            'duration': time.time() - self.stats['start_time'],            'statistics': dict(self.stats),            'suspicious_extensions': dict(self.extension_counts),            'recent_events': list(self.events)[-100:] if self.events else []        }                report_file = f"file_monitor_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"        with open(report_file, 'w') as f:            json.dump(report, f, indent=2)                print(f"{Fore.GREEN}[✓] Report saved to {report_file}{Style.RESET_ALL}")        return report_file def main():    parser = argparse.ArgumentParser(description='File Monitor for Ransomware Detection')    parser.add_argument('-p', '--paths', help='Paths to monitor (comma-separated)')    parser.add_argument('-t', '--threshold', type=int, default=50,                       help='Event threshold per time window')    parser.add_argument('-w', '--window', type=int, default=60,                       help='Time window in seconds')    parser.add_argument('-e', '--entropy', type=float, default=0.7,                       help='Entropy threshold')    parser.add_argument('-l', '--log', default='file_monitor.log',                       help='Log file')    parser.add_argument('--report', action='store_true',                       help='Generate report on exit')        args = parser.parse_args()        # Parse paths    watch_paths = args.paths.split(',') if args.paths else None        monitor = FileMonitor(        watch_paths=watch_paths,        alert_threshold=args.threshold,        time_window=args.window,        entropy_threshold=args.entropy,        log_file=args.log    )        try:        monitor.start_monitoring()    except KeyboardInterrupt:        print(f"\n{Fore.YELLOW}[*] Stopping monitor...{Style.RESET_ALL}")        if args.report:            monitor.generate_report() if __name__ == "__main__":    main()