process_analyzer.py

#!/usr/bin/env python3"""Process Analyzer for Trojan DetectionLocation: 03-malware-analysis/trojans/detection/process_analyzer.pyThis script monitors and analyzes running processes for trojan-like behaviorincluding process injection, suspicious network connections, and anomaly detection.""" import psutilimport timeimport threadingimport argparseimport jsonimport osimport sysimport signalfrom datetime import datetime, timedeltafrom collections import defaultdict, dequeimport subprocessimport reimport hashlibfrom colorama import init, Fore, Style init(autoreset=True) class ProcessAnalyzer:    """    Real-time Process Analyzer for Trojan Detection    """        def __init__(self, interval=1, alert_threshold=5, log_file="process_analyzer.log"):        """        Initialize Process Analyzer                Args:            interval: Sampling interval in seconds            alert_threshold: Number of alerts before notification            log_file: Log file path        """        self.interval = interval        self.alert_threshold = alert_threshold        self.log_file = log_file                # Process tracking        self.process_history = defaultdict(lambda: deque(maxlen=100))        self.process_stats = {}        self.network_connections = defaultdict(list)        self.suspicious_processes = set()                # Known trojan indicators        self.suspicious_names = [            'svchost', 'explorer', 'winlogon', 'services', 'lsass',            'csrss', 'smss', 'wininit', 'taskhost', 'rundll32'        ]                self.suspicious_paths = [            '\\temp\\', '\\tmp\\', '\\appdata\\local\\temp\\',            '\\users\\public\\', '\\programdata\\', '\\windows\\temp\\'        ]                self.suspicious_ports = [4444, 1337, 6667, 8080, 8443, 31337, 12345, 27374]                self.suspicious_domains = [            '.no-ip.org', '.dyndns.org', '.servehttp.com', '.redirectme.net',            '.hopto.org', '.serveftp.net', '.ddns.net', '.zapto.org'        ]                # Process behavior baselines        self.baselines = {}                # Statistics        self.stats = {            'total_processes': 0,            'suspicious_count': 0,            'alerts': 0,            'network_connections': 0,            'start_time': time.time()        }                # Control flags        self.running = True                # Setup logging        self.setup_logging()                # Create baseline        self.create_baseline()        def setup_logging(self):        """Setup logging"""        import logging        logging.basicConfig(            level=logging.INFO,            format='%(asctime)s - %(levelname)s - %(message)s',            handlers=[                logging.FileHandler(self.log_file),                logging.StreamHandler()            ]        )        self.logger = logging.getLogger(__name__)        def create_baseline(self):        """Create baseline of normal process behavior"""        self.logger.info("Creating process behavior baseline...")                for proc in psutil.process_iter(['pid', 'name', 'exe', 'cpu_percent', 'memory_percent']):            try:                proc_info = proc.info                self.baselines[proc_info['pid']] = {                    'name': proc_info['name'],                    'exe': proc_info['exe'],                    'cpu_baseline': proc_info['cpu_percent'] or 0,                    'memory_baseline': proc_info['memory_percent'] or 0,                    'create_time': proc.create_time()                }            except (psutil.NoSuchProcess, psutil.AccessDenied):                pass                self.logger.info(f"Baseline created for {len(self.baselines)} processes")        def get_process_info(self, proc):        """Get detailed process information"""        try:            with proc.oneshot():                return {                    'pid': proc.pid,                    'name': proc.name(),                    'exe': proc.exe(),                    'cmdline': ' '.join(proc.cmdline()),                    'cwd': proc.cwd(),                    'username': proc.username(),                    'cpu_percent': proc.cpu_percent(),                    'memory_percent': proc.memory_percent(),                    'memory_info': proc.memory_info()._asdict() if proc.memory_info() else {},                    'connections': len(proc.connections()),                    'threads': proc.num_threads(),                    'create_time': proc.create_time(),                    'status': proc.status(),                    'nice': proc.nice(),                    'io_counters': proc.io_counters()._asdict() if proc.io_counters() else {}                }        except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):            return None        def check_suspicious_name(self, proc_info):        """Check for suspicious process names"""        if not proc_info:            return None                name_lower = proc_info['name'].lower()        exe_lower = proc_info['exe'].lower() if proc_info['exe'] else ''                # Check for masquerading        for suspicious in self.suspicious_names:            if suspicious in name_lower and suspicious not in exe_lower:                return f"Process masquerading as {suspicious}"                # Check for hidden names        if name_lower.startswith('.') or ' ' in name_lower.strip():            return "Hidden or suspicious process name"                return None        def check_suspicious_path(self, proc_info):        """Check if process is running from suspicious location"""        if not proc_info or not proc_info['exe']:            return None                exe_lower = proc_info['exe'].lower()                for path in self.suspicious_paths:            if path in exe_lower:                return f"Running from suspicious location: {proc_info['exe']}"                return None        def check_network_connections(self, proc_info, proc):        """Check for suspicious network connections"""        if not proc_info:            return []                suspicious = []                try:            for conn in proc.connections(kind='inet'):                if conn.raddr and conn.raddr.port:                    # Check suspicious ports                    if conn.raddr.port in self.suspicious_ports:                        suspicious.append({                            'type': 'suspicious_port',                            'port': conn.raddr.port,                            'address': conn.raddr.ip,                            'status': conn.status                        })                                        # Check for dynamic DNS domains (would require reverse DNS)                    # This is simplified - in practice you'd do reverse DNS lookups                                        self.stats['network_connections'] += 1                            except (psutil.AccessDenied, psutil.NoSuchProcess):            pass                return suspicious        def check_behavior_anomalies(self, proc_info, history):        """Check for behavioral anomalies"""        if not proc_info or not history:            return []                anomalies = []                # Check CPU usage spikes        if len(history) > 5:            recent_cpu = [h.get('cpu_percent', 0) for h in list(history)[-5:]]            avg_cpu = sum(recent_cpu) / len(recent_cpu)                        if avg_cpu > 80 and proc_info['cpu_percent'] > 90:                anomalies.append(f"High CPU usage: {proc_info['cpu_percent']:.1f}%")                # Check memory leaks        if len(history) > 10:            memory_values = [h.get('memory_percent', 0) for h in list(history)[-10:]]            if all(memory_values[i] <= memory_values[i+1] for i in range(len(memory_values)-1)):                anomalies.append("Possible memory leak")                # Check thread count        if proc_info['threads'] > 100:            anomalies.append(f"High thread count: {proc_info['threads']}")                return anomalies        def check_process_injection(self):        """Check for signs of process injection"""        # This is a simplified check - real injection detection is complex        injected = []                for proc in psutil.process_iter(['pid', 'name', 'memory_maps']):            try:                # Check for processes with suspicious memory regions                # This would require deep memory analysis                pass            except:                pass                return injected        def analyze_process(self, proc):        """Analyze a single process"""        proc_info = self.get_process_info(proc)        if not proc_info:            return None                pid = proc_info['pid']                # Store in history        self.process_history[pid].append({            'timestamp': time.time(),            'cpu_percent': proc_info['cpu_percent'],            'memory_percent': proc_info['memory_percent'],            'threads': proc_info['threads'],            'connections': proc_info['connections']        })                findings = []                # Check for suspicious name        name_issue = self.check_suspicious_name(proc_info)        if name_issue:            findings.append({                'type': 'suspicious_name',                'severity': 'HIGH' if 'masquerading' in name_issue else 'MEDIUM',                'description': name_issue            })                # Check for suspicious path        path_issue = self.check_suspicious_path(proc_info)        if path_issue:            findings.append({                'type': 'suspicious_path',                'severity': 'HIGH',                'description': path_issue            })                # Check network connections        net_issues = self.check_network_connections(proc_info, proc)        for issue in net_issues:            findings.append({                'type': 'network',                'severity': 'HIGH',                'description': f"Suspicious connection to port {issue['port']} ({issue['address']})"            })                # Check behavioral anomalies        anomalies = self.check_behavior_anomalies(proc_info, self.process_history[pid])        for anomaly in anomalies:            findings.append({                'type': 'behavioral',                'severity': 'MEDIUM',                'description': anomaly            })                if findings:            self.stats['suspicious_count'] += 1                        return {                'pid': pid,                'name': proc_info['name'],                'exe': proc_info['exe'],                'cmdline': proc_info['cmdline'],                'username': proc_info['username'],                'findings': findings,                'timestamp': datetime.now().isoformat()            }                return None        def monitor_processes(self):        """Main monitoring loop"""        self.logger.info("Starting process monitoring...")                while self.running:            try:                for proc in psutil.process_iter():                    try:                        result = self.analyze_process(proc)                                                if result:                            self.stats['alerts'] += 1                                                        # Print alert                            print(f"\n{Fore.RED}🚨 SUSPICIOUS PROCESS DETECTED{Style.RESET_ALL}")                            print(f"{Fore.RED}   PID: {result['pid']} - {result['name']}{Style.RESET_ALL}")                            print(f"{Fore.RED}   Path: {result['exe']}{Style.RESET_ALL}")                                                        for finding in result['findings']:                                severity_color = Fore.RED if finding['severity'] == 'HIGH' else Fore.YELLOW                                print(f"{severity_color}   - [{finding['severity']}] {finding['description']}{Style.RESET_ALL}")                                                        # Log to file                            self.logger.warning(f"Suspicious process: {result}")                                                        # Save to JSON                            self.save_finding(result)                                                except (psutil.NoSuchProcess, psutil.AccessDenied):                        pass                                time.sleep(self.interval)                            except Exception as e:                self.logger.error(f"Error in monitoring loop: {e}")                time.sleep(self.interval)        def save_finding(self, finding):        """Save finding to JSON file"""        try:            findings_file = f"process_findings_{datetime.now().strftime('%Y%m%d')}.json"                        # Load existing            if os.path.exists(findings_file):                with open(findings_file, 'r') as f:                    findings = json.load(f)            else:                findings = []                        findings.append(finding)                        # Save            with open(findings_file, 'w') as f:                json.dump(findings, f, indent=2)                        except Exception as e:            self.logger.error(f"Error saving finding: {e}")        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}📊 PROCESS ANALYZER STATISTICS{Style.RESET_ALL}")        print(f"{Fore.CYAN}{'='*60}{Style.RESET_ALL}")        print(f"  Uptime: {int(uptime)} seconds")        print(f"  Total Processes: {len(psutil.pids())}")        print(f"  Suspicious Found: {self.stats['suspicious_count']}")        print(f"  Alerts Triggered: {self.stats['alerts']}")        print(f"  Network Connections: {self.stats['network_connections']}")        print(f"{Fore.CYAN}{'='*60}{Style.RESET_ALL}")                # Show top processes by CPU        print(f"\n{Fore.GREEN}Top CPU Processes:{Style.RESET_ALL}")        for proc in sorted(psutil.process_iter(['pid', 'name', 'cpu_percent']),                           key=lambda x: x.info['cpu_percent'] or 0, reverse=True)[:5]:            try:                if proc.info['cpu_percent'] > 0:                    print(f"  {proc.info['pid']}: {proc.info['name']} - {proc.info['cpu_percent']:.1f}% CPU")            except:                pass        def stats_loop(self):        """Periodically print statistics"""        while self.running:            time.sleep(10)            self.print_stats()        def generate_report(self):        """Generate analysis report"""        report = {            'timestamp': datetime.now().isoformat(),            'duration': time.time() - self.stats['start_time'],            'statistics': self.stats,            'suspicious_count': self.stats['suspicious_count'],            'alerts': self.stats['alerts']        }                report_file = f"process_analysis_{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 start(self):        """Start the analyzer"""        print(f"""        {Fore.CYAN}╔═══════════════════════════════════════╗        ║        Process Analyzer v1.0         ║        ║      Trojan Behavior Detection       ║        ╚═══════════════════════════════════════╝{Style.RESET_ALL}        """)                # Start monitoring thread        monitor_thread = threading.Thread(target=self.monitor_processes)        monitor_thread.daemon = True        monitor_thread.start()                # Start stats thread        stats_thread = threading.Thread(target=self.stats_loop)        stats_thread.daemon = True        stats_thread.start()                print(f"{Fore.GREEN}[✓] Process analyzer started{Style.RESET_ALL}")        print(f"{Fore.YELLOW}   Press Ctrl+C to stop{Style.RESET_ALL}\n")                try:            while self.running:                time.sleep(1)        except KeyboardInterrupt:            self.stop()        def stop(self):        """Stop the analyzer"""        self.running = False        print(f"\n{Fore.GREEN}[✓] Process analyzer stopped{Style.RESET_ALL}")        self.print_stats()        self.generate_report() def signal_handler(sig, frame):    """Handle Ctrl+C"""    analyzer.stop()    sys.exit(0) def main():    parser = argparse.ArgumentParser(description='Process Analyzer for Trojan Detection')    parser.add_argument('-i', '--interval', type=int, default=1, help='Sampling interval')    parser.add_argument('-t', '--threshold', type=int, default=5, help='Alert threshold')    parser.add_argument('-l', '--log', default='process_analyzer.log', help='Log file')        args = parser.parse_args()        global analyzer    analyzer = ProcessAnalyzer(        interval=args.interval,        alert_threshold=args.threshold,        log_file=args.log    )        signal.signal(signal.SIGINT, signal_handler)    analyzer.start() if __name__ == "__main__":    main()