rootkit_detector.py

#!/usr/bin/env python3"""Rootkit DetectorThis script detects rootkits using multiple techniques including:- System call table verification- Hidden process detection- Kernel module analysis- Memory scanning- Signature-based detection""" import osimport sysimport timeimport hashlibimport jsonimport argparseimport subprocessimport platformimport reimport ctypesfrom datetime import datetimefrom pathlib import Pathfrom collections import defaultdictimport psutilfrom colorama import init, Fore, Style init(autoreset=True) class RootkitDetector:    """    Comprehensive Rootkit Detection System    """        def __init__(self, signatures_file="rootkit_signatures.json",                  whitelist_file="rootkit_whitelist.json",                 log_file="rootkit_detector.log"):        """        Initialize Rootkit Detector                Args:            signatures_file: Known rootkit signatures            whitelist_file: Whitelisted files/processes            log_file: Log file path        """        self.signatures_file = signatures_file        self.whitelist_file = whitelist_file        self.log_file = log_file        self.system = platform.system()                # Load signatures and whitelist        self.signatures = self.load_signatures()        self.whitelist = self.load_whitelist()                # Rootkit indicators        self.rootkit_indicators = {            'windows': {                'hidden_processes': [],                'ssdt_hooks': [],                'idtr_hooks': [],                'driver_hidden': [],                'registry_hidden': [],                'file_hidden': []            },            'linux': {                'hidden_processes': [],                'syscall_hooks': [],                'kernel_modules': [],                'file_hidden': [],                'ld_preload': []            }        }                # System information        self.system_info = self.gather_system_info()                # Statistics        self.stats = {            'files_scanned': 0,            'processes_scanned': 0,            'modules_scanned': 0,            'detections': 0,            'suspicious_findings': 0,            'start_time': time.time()        }                # Findings storage        self.findings = []                # Setup logging        self.setup_logging()            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 load_signatures(self):        """Load rootkit signatures"""        default_signatures = {            'known_rootkits': {                'fu': ['\\\\.\\procfs\\', 'FU_ROOTKIT'],                'tdl4': ['\\Driver\\TDL', 'TDSS', 'Alureon'],                'stuxnet': ['MRxCls', 'MRxNet'],                'mebroot': ['mbr.sys', 'grd.sys'],                'zeroaccess': ['max++', 'max222'],                'sinowal': ['mfehidk.sys'],                'rustock': ['Rustock', 'C:\\\\Windows\\\\system32\\\\drivers\\\\Rustock.sys'],                'cutwail': ['Cutwail', 'Panda'],                'alureon': ['Alureon', 'TDL'],                'mybot': ['MyBot', 'Rootkit.Win32.MyBot']            },            'suspicious_paths': [                '\\SystemRoot\\System32\\Drivers\\',                '\\??\\C:\\Windows\\System32\\Drivers\\',                '\\Device\\Harddisk0\\DR0\\',                '\\\\?\\GLOBALROOT\\Device\\Harddisk0\\DR0\\'            ],            'suspicious_processes': [                'lsass.exe', 'csrss.exe', 'winlogon.exe', 'services.exe',                'svchost.exe', 'explorer.exe', 'spoolsv.exe'            ],            'suspicious_files': [                'ntoskrnl.exe', 'hal.dll', 'kernel32.dll', 'ntdll.dll',                'win32k.sys', 'tcpip.sys', 'ndis.sys'            ],            'linux_patterns': {                'knark': ['knark', 'modhide'],                'adore': ['adore-ng', 'adore'],                'suterusu': ['suterusu'],                'enyelkm': ['enyelkm'],                'diamorphine': ['diamorphine']            }        }                if os.path.exists(self.signatures_file):            try:                with open(self.signatures_file, 'r') as f:                    return json.load(f)            except:                return default_signatures        return default_signatures        def load_whitelist(self):        """Load whitelist"""        if os.path.exists(self.whitelist_file):            try:                with open(self.whitelist_file, 'r') as f:                    return json.load(f)            except:                return {'processes': [], 'files': [], 'modules': []}        return {'processes': [], 'files': [], 'modules': []}        def gather_system_info(self):        """Gather system information"""        info = {            'hostname': platform.node(),            'os': self.system,            'release': platform.release(),            'version': platform.version(),            'architecture': platform.machine(),            'processor': platform.processor(),            'boot_time': datetime.fromtimestamp(psutil.boot_time()).isoformat(),            'cpu_count': psutil.cpu_count(),            'memory': psutil.virtual_memory().total,            'processes': len(psutil.pids())        }                return info        def detect_hidden_processes_windows(self):        """Detect hidden processes on Windows using multiple methods"""        hidden = []                # Method 1: Compare process list from different sources        api_pids = set(psutil.pids())                # Method 2: Scan process memory directly (simplified)        # In reality, this would need to parse EPROCESS structures        scan_pids = set()        for pid in range(1, 10000):  # Scan PID range            try:                proc = psutil.Process(pid)                scan_pids.add(pid)            except:                pass                # Find hidden processes        hidden_pids = api_pids - scan_pids if len(scan_pids) > 0 else set()                for pid in hidden_pids:            hidden.append({                'pid': pid,                'detection': 'Cross-view comparison',                'confidence': 'HIGH'            })                return hidden        def detect_hidden_processes_linux(self):        """Detect hidden processes on Linux"""        hidden = []                try:            # Method 1: Compare /proc with ps            proc_pids = set()            if os.path.exists('/proc'):                for pid_file in os.listdir('/proc'):                    if pid_file.isdigit():                        proc_pids.add(int(pid_file))                        ps_pids = set(psutil.pids())                        # PIDs in /proc but not in ps (hidden)            hidden_pids = proc_pids - ps_pids                        for pid in hidden_pids:                hidden.append({                    'pid': pid,                    'detection': '/proc vs ps mismatch',                    'confidence': 'HIGH'                })                        # Method 2: Check for PID wrapping/hiding            # This would need kernel module enumeration                    except Exception as e:            self.logger.error(f"Error detecting hidden processes: {e}")                return hidden        def check_ssdt_hooks_windows(self):        """Check for SSDT (System Service Descriptor Table) hooks"""        # This requires kernel-level access        # Simplified detection - in reality, would need to read SSDT from kernel memory                hooks = []                # Common hooked functions        critical_functions = [            'NtCreateFile', 'NtOpenProcess', 'NtOpenThread',            'NtCreateProcess', 'NtCreateThread', 'NtDeleteFile',            'NtQuerySystemInformation', 'NtQueryDirectoryFile',            'NtReadFile', 'NtWriteFile', 'NtDeviceIoControlFile'        ]                # Would need to compare with clean system        # This is a placeholder for demonstration                return hooks        def check_syscall_hooks_linux(self):        """Check for syscall table hooks on Linux"""        hooks = []                # Check /proc/kallsyms for syscall table        try:            if os.path.exists('/proc/kallsyms'):                with open('/proc/kallsyms', 'r') as f:                    kallsyms = f.read()                                # Look for syscall table entries                syscall_pattern = r'sys_([a-zA-Z0-9_]+)'                matches = re.findall(syscall_pattern, kallsyms)                                # Would need to verify against known good addresses                # This is simplified                        except Exception as e:            self.logger.error(f"Error checking syscall hooks: {e}")                return hooks        def detect_hidden_files(self):        """Detect hidden files and directories"""        hidden_files = []                try:            # Check common rootkit hiding locations            suspicious_paths = [                'C:\\Windows\\System32\\Drivers\\' if self.system == 'Windows' else '/lib/modules/',                'C:\\Windows\\Temp\\' if self.system == 'Windows' else '/tmp/',                'C:\\ProgramData\\' if self.system == 'Windows' else '/var/tmp/'            ]                        for path in suspicious_paths:                if os.path.exists(path):                    # List directory contents                    try:                        files = os.listdir(path)                                                # Check for files with hidden attributes                        for file in files:                            filepath = os.path.join(path, file)                                                        # Check for files with suspicious names                            if file.startswith('.') and len(file) > 1:                                if file not in ['.', '..']:                                    hidden_files.append({                                        'file': filepath,                                        'reason': 'Hidden attribute',                                        'confidence': 'MEDIUM'                                    })                                                        # Check for files with suspicious extensions                            if file.endswith('.sys') and '~' in file:                                hidden_files.append({                                    'file': filepath,                                    'reason': 'Suspicious system file',                                    'confidence': 'MEDIUM'                                })                                                    except PermissionError:                        pass                                except Exception as e:            self.logger.error(f"Error detecting hidden files: {e}")                return hidden_files        def check_kernel_modules(self):        """Check loaded kernel modules"""        suspicious_modules = []                if self.system == 'Linux':            try:                # List loaded modules                result = subprocess.run(['lsmod'], capture_output=True, text=True)                modules = result.stdout.split('\n')[1:]  # Skip header                                for module in modules:                    if module.strip():                        parts = module.split()                        if len(parts) >= 1:                            module_name = parts[0]                                                        # Check against known malicious modules                            for rootkit, patterns in self.signatures.get('linux_patterns', {}).items():                                for pattern in patterns:                                    if pattern in module_name.lower():                                        suspicious_modules.append({                                            'module': module_name,                                            'rootkit': rootkit,                                            'detection': 'Known rootkit module'                                        })                                                        # Check for hidden modules                            if module_name.startswith('.'):                                suspicious_modules.append({                                    'module': module_name,                                    'detection': 'Hidden module name',                                    'confidence': 'HIGH'                                })                                            except Exception as e:                self.logger.error(f"Error checking kernel modules: {e}")                elif self.system == 'Windows':            try:                # Use driverquery to list drivers                result = subprocess.run(['driverquery', '/v'],                                        capture_output=True, text=True)                                # Parse output for suspicious drivers                lines = result.stdout.split('\n')                for line in lines:                    # Look for unsigned drivers                    if 'FALSE' in line and 'Driver' in line:                        suspicious_modules.append({                            'driver': line,                            'detection': 'Unsigned driver',                            'confidence': 'MEDIUM'                        })                                    except Exception as e:                self.logger.error(f"Error checking drivers: {e}")                return suspicious_modules        def check_ld_preload_linux(self):        """Check for LD_PRELOAD rootkits on Linux"""        findings = []                try:            # Check LD_PRELOAD environment variable            ld_preload = os.environ.get('LD_PRELOAD', '')            if ld_preload:                findings.append({                    'type': 'LD_PRELOAD',                    'value': ld_preload,                    'detection': 'LD_PRELOAD set in environment',                    'confidence': 'HIGH'                })                        # Check /etc/ld.so.preload            if os.path.exists('/etc/ld.so.preload'):                with open('/etc/ld.so.preload', 'r') as f:                    preload_content = f.read().strip()                    if preload_content:                        findings.append({                            'type': 'ld.so.preload',                            'value': preload_content,                            'detection': 'System-wide preload configured',                            'confidence': 'HIGH'                        })                                except Exception as e:            self.logger.error(f"Error checking LD_PRELOAD: {e}")                return findings        def check_alternate_data_streams(self):        """Check for NTFS alternate data streams (Windows)"""        ads_findings = []                if self.system != 'Windows':            return ads_findings                try:            # Use dir /r to list ADS            result = subprocess.run(['cmd', '/c', 'dir', '/r', 'C:\\'],                                    capture_output=True, text=True)                        lines = result.stdout.split('\n')            for line in lines:                if ':$DATA' in line:                    ads_findings.append({                        'ads': line.strip(),                        'detection': 'Alternate Data Stream detected',                        'confidence': 'MEDIUM'                    })                            except Exception as e:            self.logger.error(f"Error checking ADS: {e}")                return ads_findings        def check_mbr_gpt(self):        """Check MBR/GPT for bootkits"""        findings = []                if self.system == 'Windows':            try:                # Read first sector of disk                with open('\\\\.\\PhysicalDrive0', 'rb') as f:                    mbr = f.read(512)                                # Check for suspicious MBR code                # Known bootkit signatures would be checked here                # This is simplified                                # Check if MBR is modified from standard                standard_mbr_hash = 'a1b2c3d4...'  # Would compare with known good                            except Exception as e:                self.logger.error(f"Error checking MBR: {e}")                return findings        # Names of public Linux rootkits (kernel modules and userland). Matched against whole file names only.    LINUX_ROOTKIT_NAMES = {        'diamorphine', 'reptile', 'suterusu', 'adore-ng', 'adore_ng', 'knark', 'azazel',        'jynx', 'jynx2', 'beurk', 'vlany', 'kovid', 'enyelkm', 'phalanx', 'sebek',    }     def _signature_sets(self):        """Split configured signatures into whole-name matches. Short or generic fragments        (for example 'TDL' or 'Panda') are ignored: as substrings they match harmless files        such as libltdl and produced false 'confirmed' detections."""        file_names, process_names = set(), set()        for rootkit, signatures in self.signatures.get('known_rootkits', {}).items():            for sig in signatures:                low = sig.lower()                if re.search(r'\.(sys|dll|ko|exe)$', low):                    file_names.add(low)                elif len(low) >= 6 and re.fullmatch(r'[a-z0-9_.+-]+', low):                    process_names.add(low)        if self.system != 'Windows':            file_names |= {n for n in self.LINUX_ROOTKIT_NAMES}            process_names |= {n for n in self.LINUX_ROOTKIT_NAMES}        return file_names, process_names     def signature_scan(self):        """Look for processes and files whose NAME exactly matches a known rootkit.        A name match is only an indicator, so confidence is MEDIUM; confirm with content or hash checks."""        findings = []        file_names, process_names = self._signature_sets()         for proc in psutil.process_iter(['pid', 'name', 'exe']):            try:                proc_name = (proc.info['name'] or '').lower()                exe_base = os.path.basename(proc.info['exe'] or '').lower()                for candidate in {proc_name, exe_base}:                    if candidate and candidate in process_names | file_names:                        findings.append({                            'type': 'process_signature', 'rootkit': candidate,                            'process': proc.info['name'], 'pid': proc.info['pid'],                            'signature': candidate, 'confidence': 'MEDIUM',                        })            except (psutil.NoSuchProcess, psutil.AccessDenied):                continue         if self.system == 'Windows':            system_dirs = ['C:\\Windows\\System32\\drivers\\']        else:            system_dirs = ['/lib/modules/', '/usr/lib/modules/']         seen_dirs = set()        for directory in system_dirs:            real = os.path.realpath(directory)           # /lib is often a symlink to /usr/lib            if not os.path.isdir(real) or real in seen_dirs:                continue            seen_dirs.add(real)            for root, _dirs, files in os.walk(real):                for name in files:                    base = re.sub(r'\.(ko|sys)(\.(gz|xz|zst))?$', '', name.lower())                    if name.lower() in file_names or base in file_names or base in self.LINUX_ROOTKIT_NAMES:                        findings.append({                            'type': 'file_signature', 'rootkit': base,                            'file': os.path.join(root, name), 'signature': name,                            'confidence': 'MEDIUM',                        })        return findings     def detect_rootkit_activity(self):        """Detect rootkit-like behavior"""        findings = []                # Check for unusual system behavior        try:            # CPU usage patterns            cpu_percent = psutil.cpu_percent(interval=1)            if cpu_percent > 95:                findings.append({                    'type': 'behavior',                    'detection': 'Extremely high CPU usage',                    'value': cpu_percent,                    'confidence': 'LOW'                })                        # Memory patterns            memory = psutil.virtual_memory()            if memory.used > memory.total * 0.95:                findings.append({                    'type': 'behavior',                    'detection': 'Extremely high memory usage',                    'value': f"{memory.used/memory.total*100:.1f}%",                    'confidence': 'LOW'                })                        # Network connections            connections = len(psutil.net_connections())            if connections > 1000:                findings.append({                    'type': 'behavior',                    'detection': 'Unusually high number of network connections',                    'value': connections,                    'confidence': 'MEDIUM'                })                        except Exception as e:            self.logger.error(f"Error detecting rootkit activity: {e}")                return findings        def run_detection(self):        """Run all detection methods"""        print(f"""        {Fore.CYAN}╔═══════════════════════════════════════╗        ║        Rootkit Detector v1.0         ║        ╚═══════════════════════════════════════╝{Style.RESET_ALL}        """)                print(f"{Fore.YELLOW}[!] WARNING: Rootkit detection is dangerous{Style.RESET_ALL}")        print(f"{Fore.YELLOW}[!] Only run on isolated systems{Style.RESET_ALL}\n")                print(f"{Fore.CYAN}[*] System: {self.system} {platform.release()}{Style.RESET_ALL}")        print(f"{Fore.CYAN}[*] Starting detection...{Style.RESET_ALL}\n")                # 1. Hidden Process Detection        print(f"{Fore.CYAN}[*] Checking for hidden processes...{Style.RESET_ALL}")        if self.system == 'Windows':            hidden_procs = self.detect_hidden_processes_windows()        else:            hidden_procs = self.detect_hidden_processes_linux()                for proc in hidden_procs:            self.findings.append(('HIDDEN_PROCESS', proc))            print(f"{Fore.RED}  [!] Hidden process detected: PID {proc['pid']}{Style.RESET_ALL}")                # 2. Kernel Module Check        print(f"\n{Fore.CYAN}[*] Checking kernel modules...{Style.RESET_ALL}")        suspicious_modules = self.check_kernel_modules()        for module in suspicious_modules:            self.findings.append(('SUSPICIOUS_MODULE', module))            print(f"{Fore.RED}  [!] Suspicious module: {module.get('module', module.get('driver', 'Unknown'))}{Style.RESET_ALL}")                # 3. Hidden Files        print(f"\n{Fore.CYAN}[*] Checking for hidden files...{Style.RESET_ALL}")        hidden_files = self.detect_hidden_files()        for file in hidden_files[:10]:  # Limit output            self.findings.append(('HIDDEN_FILE', file))            print(f"{Fore.YELLOW}  [?] Hidden file: {file['file']}{Style.RESET_ALL}")                # 4. Signature Scan        print(f"\n{Fore.CYAN}[*] Scanning for known rootkit signatures...{Style.RESET_ALL}")        signature_hits = self.signature_scan()        for hit in signature_hits:            self.findings.append(('SIGNATURE', hit))            print(f"{Fore.RED}  [!] Signature match: {hit.get('rootkit', 'Unknown')} in {hit.get('file', hit.get('process', 'Unknown'))}{Style.RESET_ALL}")                # 5. Linux-specific checks        if self.system == 'Linux':            print(f"\n{Fore.CYAN}[*] Checking LD_PRELOAD...{Style.RESET_ALL}")            preload_findings = self.check_ld_preload_linux()            for finding in preload_findings:                self.findings.append(('LD_PRELOAD', finding))                print(f"{Fore.RED}  [!] {finding['detection']}: {finding.get('value', '')}{Style.RESET_ALL}")                # 6. Windows-specific checks        if self.system == 'Windows':            print(f"\n{Fore.CYAN}[*] Checking Alternate Data Streams...{Style.RESET_ALL}")            ads_findings = self.check_alternate_data_streams()            for finding in ads_findings:                self.findings.append(('ADS', finding))                print(f"{Fore.YELLOW}  [?] ADS detected: {finding['ads']}{Style.RESET_ALL}")                # 7. Behavioral Detection        print(f"\n{Fore.CYAN}[*] Analyzing system behavior...{Style.RESET_ALL}")        behavior_findings = self.detect_rootkit_activity()        for finding in behavior_findings:            self.findings.append(('BEHAVIOR', finding))            print(f"{Fore.YELLOW}  [?] {finding['detection']}: {finding.get('value', '')}{Style.RESET_ALL}")                # Update statistics        self.stats['detections'] = len([f for f in self.findings if f[0] in ['HIDDEN_PROCESS', 'SIGNATURE']])        self.stats['suspicious_findings'] = len(self.findings)                # Generate report        self.generate_report()            def generate_report(self):        """Generate detection report"""        report = {            'timestamp': datetime.now().isoformat(),            'system': self.system_info,            'statistics': self.stats,            'findings': [                {                    'type': f[0],                    'details': f[1]                } for f in self.findings            ],            'risk_level': self.calculate_risk_level()        }                report_file = f"rootkit_scan_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"        with open(report_file, 'w') as f:            json.dump(report, f, indent=2)                # Print summary        print(f"\n{Fore.CYAN}{'='*60}{Style.RESET_ALL}")        print(f"{Fore.CYAN}📊 ROOTKIT DETECTION SUMMARY{Style.RESET_ALL}")        print(f"{Fore.CYAN}{'='*60}{Style.RESET_ALL}")        print(f"  Scan ID: {report_file}")        print(f"  Duration: {time.time() - self.stats['start_time']:.2f} seconds")        print(f"  Total Findings: {self.stats['suspicious_findings']}")        print(f"  Confirmed Detections: {self.stats['detections']}")        print(f"  Risk Level: {report['risk_level']}")        print(f"{Fore.CYAN}{'='*60}{Style.RESET_ALL}")                if self.stats['detections'] > 0:            print(f"\n{Fore.RED}[!] ROOTKIT DETECTED! Immediate action required.{Style.RESET_ALL}")        elif self.stats['suspicious_findings'] > 0:            print(f"\n{Fore.YELLOW}[!] Suspicious indicators found. Investigate manually.{Style.RESET_ALL}")        else:            print(f"\n{Fore.GREEN}[✓] No rootkits detected{Style.RESET_ALL}")                print(f"\n{Fore.GREEN}[✓] Report saved to {report_file}{Style.RESET_ALL}")                return report_file        def calculate_risk_level(self):        """Calculate overall risk level"""        high_risk_types = ['HIDDEN_PROCESS', 'SIGNATURE', 'LD_PRELOAD']        medium_risk_types = ['SUSPICIOUS_MODULE', 'ADS']                high_count = sum(1 for f in self.findings if f[0] in high_risk_types)        medium_count = sum(1 for f in self.findings if f[0] in medium_risk_types)                if high_count > 0:            return 'HIGH'        elif medium_count > 2:            return 'MEDIUM'        elif medium_count > 0:            return 'LOW'        else:            return 'NONE' def main():    parser = argparse.ArgumentParser(description='Rootkit Detector')    parser.add_argument('--signatures', default='rootkit_signatures.json', help='Signatures file')    parser.add_argument('--whitelist', default='rootkit_whitelist.json', help='Whitelist file')    parser.add_argument('--log', default='rootkit_detector.log', help='Log file')        args = parser.parse_args()        detector = RootkitDetector(        signatures_file=args.signatures,        whitelist_file=args.whitelist,        log_file=args.log    )        try:        detector.run_detection()    except KeyboardInterrupt:        print(f"\n{Fore.YELLOW}[!] Scan interrupted{Style.RESET_ALL}")    except Exception as e:        print(f"{Fore.RED}[!] Error: {e}{Style.RESET_ALL}") if __name__ == "__main__":    main()