integrity_checker.py

#!/usr/bin/env python3"""System Integrity CheckerLocation: 03-malware-analysis/rootkits/detection/integrity_checker.pyThis script verifies system integrity by checking:- File system integrity- Kernel module signatures- System call tables- Critical system files- Registry integrity (Windows)- Boot sector integrity""" import osimport sysimport hashlibimport jsonimport timeimport argparseimport subprocessimport platformimport tempfileimport shutilfrom datetime import datetimefrom pathlib import Pathfrom collections import defaultdictimport psutilfrom colorama import init, Fore, Style init(autoreset=True) class IntegrityChecker:    """    System Integrity Verification System    """        def __init__(self, baseline_file="integrity_baseline.json",                  verify_all=False, log_file="integrity_check.log"):        """        Initialize Integrity Checker                Args:            baseline_file: Baseline file for comparisons            verify_all: Verify all files (slow)            log_file: Log file path        """        self.baseline_file = baseline_file        self.verify_all = verify_all        self.log_file = log_file        self.system = platform.system()                # Critical system files        self.critical_files = self.get_critical_files()                # Baseline storage        self.baseline = self.load_baseline()                # Violations storage        self.violations = []                # Statistics        self.stats = {            'files_checked': 0,            'files_modified': 0,            'files_added': 0,            'files_missing': 0,            'permission_changes': 0,            'signature_failures': 0,            'start_time': time.time()        }                # 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 get_critical_files(self):        """Get list of critical system files"""        critical = {            'Windows': [                'C:\\Windows\\System32\\ntoskrnl.exe',                'C:\\Windows\\System32\\hal.dll',                'C:\\Windows\\System32\\kernel32.dll',                'C:\\Windows\\System32\\ntdll.dll',                'C:\\Windows\\System32\\win32k.sys',                'C:\\Windows\\System32\\drivers\\tcpip.sys',                'C:\\Windows\\System32\\drivers\\ndis.sys',                'C:\\Windows\\System32\\config\\SAM',                'C:\\Windows\\System32\\config\\SYSTEM',                'C:\\Windows\\System32\\config\\SOFTWARE',                'C:\\Windows\\explorer.exe',                'C:\\Windows\\System32\\cmd.exe',                'C:\\Windows\\System32\\powershell.exe'            ],            'Linux': [                '/boot/vmlinuz',                '/boot/initrd.img',                '/etc/passwd',                '/etc/shadow',                '/etc/ld.so.preload',                '/lib/modules/',                '/bin/ls',                '/bin/ps',                '/bin/netstat',                '/usr/bin/sshd',                '/etc/ssh/sshd_config',                '/lib/libc.so*',                '/lib/ld-linux.so*'            ],            'Darwin': [  # macOS                '/System/Library/Kernels/kernel',                '/etc/passwd',                '/etc/shadow',                '/bin/ls',                '/bin/ps',                '/usr/libexec/sshd-keygen-wrapper'            ]        }                return critical.get(self.system, [])        def load_baseline(self):        """Load integrity baseline"""        if os.path.exists(self.baseline_file):            try:                with open(self.baseline_file, 'r') as f:                    return json.load(f)            except Exception as e:                self.logger.error(f"Error loading baseline: {e}")                return {'files': {}, 'created': None}                return {'files': {}, 'created': None}        def save_baseline(self):        """Save integrity baseline"""        try:            with open(self.baseline_file, 'w') as f:                json.dump(self.baseline, f, indent=2)            self.logger.info(f"Baseline saved to {self.baseline_file}")            return True        except Exception as e:            self.logger.error(f"Error saving baseline: {e}")            return False        def calculate_file_hash(self, filepath, hash_type='sha256'):        """Calculate file hash"""        try:            hash_func = hashlib.new(hash_type)            with open(filepath, 'rb') as f:                for chunk in iter(lambda: f.read(8192), b''):                    hash_func.update(chunk)            return hash_func.hexdigest()        except Exception as e:            self.logger.error(f"Error hashing {filepath}: {e}")            return None        def get_file_metadata(self, filepath):        """Get file metadata"""        try:            stat = os.stat(filepath)            return {                'size': stat.st_size,                'modified': stat.st_mtime,                'permissions': oct(stat.st_mode)[-3:],                'owner': stat.st_uid,                'group': stat.st_gid            }        except Exception as e:            self.logger.error(f"Error getting metadata for {filepath}: {e}")            return None        def create_baseline(self):        """Create new integrity baseline"""        print(f"\n{Fore.CYAN}[*] Creating integrity baseline...{Style.RESET_ALL}")                baseline_data = {            'created': datetime.now().isoformat(),            'system': platform.platform(),            'files': {}        }                # Baseline critical files        for filepath in self.critical_files:            if os.path.exists(filepath) and not os.path.islink(filepath):                if os.path.isfile(filepath):                    print(f"  Processing: {filepath}")                    file_hash = self.calculate_file_hash(filepath)                    metadata = self.get_file_metadata(filepath)                                        if file_hash and metadata:                        baseline_data['files'][filepath] = {                            'hash': file_hash,                            'metadata': metadata                        }                        self.stats['files_checked'] += 1                elif os.path.isdir(filepath):                    # Walk directory                    for root, dirs, files in os.walk(filepath):                        for file in files:                            fullpath = os.path.join(root, file)                            if os.path.isfile(fullpath) and not os.path.islink(fullpath):                                print(f"  Processing: {fullpath}")                                file_hash = self.calculate_file_hash(fullpath)                                metadata = self.get_file_metadata(fullpath)                                                                if file_hash and metadata:                                    baseline_data['files'][fullpath] = {                                        'hash': file_hash,                                        'metadata': metadata                                    }                                    self.stats['files_checked'] += 1                self.baseline = baseline_data        self.save_baseline()                print(f"\n{Fore.GREEN}[✓] Baseline created with {len(baseline_data['files'])} files{Style.RESET_ALL}")        return baseline_data        def verify_file_integrity(self, filepath):        """Verify single file integrity"""        if filepath not in self.baseline['files']:            return {'status': 'not_in_baseline', 'file': filepath}                baseline = self.baseline['files'][filepath]                if not os.path.exists(filepath):            return {'status': 'missing', 'file': filepath}                # Check hash        current_hash = self.calculate_file_hash(filepath)        if current_hash != baseline['hash']:            return {                'status': 'modified',                'file': filepath,                'expected_hash': baseline['hash'],                'current_hash': current_hash,                'type': 'hash_mismatch'            }                # Check metadata        current_meta = self.get_file_metadata(filepath)        if current_meta:            changes = []            if current_meta['size'] != baseline['metadata']['size']:                changes.append('size')            if current_meta['permissions'] != baseline['metadata']['permissions']:                changes.append('permissions')            if current_meta['modified'] != baseline['metadata']['modified']:                changes.append('modification_time')                        if changes:                return {                    'status': 'modified',                    'file': filepath,                    'changes': changes,                    'type': 'metadata_changed'                }                return {'status': 'ok', 'file': filepath}        def verify_system_integrity(self):        """Verify entire system integrity"""        print(f"\n{Fore.CYAN}[*] Verifying system integrity...{Style.RESET_ALL}")                if not self.baseline['files']:            print(f"{Fore.RED}[!] No baseline found. Create baseline first.{Style.RESET_ALL}")            return False                # Check all files in baseline        for filepath in self.baseline['files']:            result = self.verify_file_integrity(filepath)            self.stats['files_checked'] += 1                        if result['status'] == 'modified':                self.stats['files_modified'] += 1                self.violations.append(result)                print(f"{Fore.RED}  [!] MODIFIED: {filepath}{Style.RESET_ALL}")                if 'type' in result:                    print(f"{Fore.RED}      Reason: {result['type']}{Style.RESET_ALL}")                        elif result['status'] == 'missing':                self.stats['files_missing'] += 1                self.violations.append(result)                print(f"{Fore.RED}  [!] MISSING: {filepath}{Style.RESET_ALL}")                # Check for new files in critical locations        if self.verify_all:            print(f"\n{Fore.CYAN}[*] Checking for new files...{Style.RESET_ALL}")            for location in self.critical_files:                if os.path.isdir(location):                    for root, dirs, files in os.walk(location):                        for file in files:                            fullpath = os.path.join(root, file)                            if fullpath not in self.baseline['files']:                                self.stats['files_added'] += 1                                self.violations.append({                                    'status': 'new',                                    'file': fullpath                                })                                print(f"{Fore.YELLOW}  [?] NEW: {fullpath}{Style.RESET_ALL}")                return True        def verify_kernel_integrity(self):        """Verify kernel integrity"""        kernel_violations = []                if self.system == 'Linux':            try:                # Check kernel modules                result = subprocess.run(['lsmod'], capture_output=True, text=True)                modules = result.stdout.split('\n')[1:]                                for module in modules:                    if module.strip():                        parts = module.split()                        if len(parts) >= 1:                            module_name = parts[0]                                                        # Check if module is signed (if required)                            if os.path.exists(f'/lib/modules/$(uname -r)/kernel/{module_name}.ko'):                                # Would check module signature here                                pass                                            except Exception as e:                self.logger.error(f"Error verifying kernel: {e}")                elif self.system == 'Windows':            try:                # Check digital signatures of kernel drivers                result = subprocess.run(['driverquery', '/si'],                                        capture_output=True, text=True)                if 'unsigned' in result.stdout.lower():                    kernel_violations.append({                        'type': 'unsigned_drivers',                        'details': 'Unsigned drivers detected'                    })                                except Exception as e:                self.logger.error(f"Error verifying kernel: {e}")                return kernel_violations        def verify_boot_integrity(self):        """Verify boot sector integrity"""        boot_violations = []                if self.system == 'Windows':            try:                # Read MBR                with open('\\\\.\\PhysicalDrive0', 'rb') as f:                    mbr = f.read(512)                                # Check if MBR is in baseline                mbr_hash = hashlib.sha256(mbr).hexdigest()                                # Would compare with known good MBR                # This is simplified                            except Exception as e:                self.logger.error(f"Error verifying boot sector: {e}")                elif self.system == 'Linux':            try:                # Check /boot for modifications                if os.path.exists('/boot'):                    for file in os.listdir('/boot'):                        if file.endswith('.efi') or file.startswith('vmlinuz'):                            filepath = os.path.join('/boot', file)                            if filepath in self.baseline['files']:                                result = self.verify_file_integrity(filepath)                                if result['status'] == 'modified':                                    boot_violations.append(result)                                                except Exception as e:                self.logger.error(f"Error verifying boot: {e}")                return boot_violations        def verify_registry_integrity(self):        """Verify Windows registry integrity"""        registry_violations = []                if self.system != 'Windows':            return registry_violations                try:            # Check critical registry keys            critical_keys = [                r'HKLM\SYSTEM\CurrentControlSet\Services',                r'HKLM\SOFTWARE\Microsoft\Windows\CurrentVersion\Run',                r'HKLM\SOFTWARE\Microsoft\Windows NT\CurrentVersion\Winlogon'            ]                        for key in critical_keys:                # Would compare registry values with baseline                # This is simplified                pass                        except Exception as e:            self.logger.error(f"Error verifying registry: {e}")                return registry_violations        def verify_process_integrity(self):        """Verify running process integrity"""        process_violations = []                try:            for proc in psutil.process_iter(['pid', 'name', 'exe']):                try:                    if proc.info['exe'] and os.path.exists(proc.info['exe']):                        exe_path = proc.info['exe']                                                if exe_path in self.baseline['files']:                            result = self.verify_file_integrity(exe_path)                            if result['status'] == 'modified':                                process_violations.append({                                    'pid': proc.info['pid'],                                    'name': proc.info['name'],                                    'exe': exe_path,                                    'violation': result                                })                                                except (psutil.NoSuchProcess, psutil.AccessDenied):                    pass                            except Exception as e:            self.logger.error(f"Error verifying processes: {e}")                return process_violations        def run_integrity_check(self):        """Run complete integrity check"""        print(f"""        {Fore.CYAN}╔═══════════════════════════════════════╗        ║      System Integrity Checker       ║        ║       Rootkit Detection Tool        ║        ╚═══════════════════════════════════════╝{Style.RESET_ALL}        """)                # Verify main system integrity        self.verify_system_integrity()                # Additional checks        print(f"\n{Fore.CYAN}[*] Verifying kernel integrity...{Style.RESET_ALL}")        kernel_violations = self.verify_kernel_integrity()        for violation in kernel_violations:            self.violations.append(('KERNEL', violation))                print(f"{Fore.CYAN}[*] Verifying boot integrity...{Style.RESET_ALL}")        boot_violations = self.verify_boot_integrity()        for violation in boot_violations:            self.violations.append(('BOOT', violation))                if self.system == 'Windows':            print(f"{Fore.CYAN}[*] Verifying registry integrity...{Style.RESET_ALL}")            registry_violations = self.verify_registry_integrity()            for violation in registry_violations:                self.violations.append(('REGISTRY', violation))                print(f"{Fore.CYAN}[*] Verifying process integrity...{Style.RESET_ALL}")        process_violations = self.verify_process_integrity()        for violation in process_violations:            self.violations.append(('PROCESS', violation))                # Generate report        self.generate_report()        def generate_report(self):        """Generate integrity check report"""        report = {            'timestamp': datetime.now().isoformat(),            'system': platform.platform(),            'baseline_created': self.baseline.get('created', 'Unknown'),            'statistics': self.stats,            'violations': self.violations[:100],  # Limit size            'integrity_score': self.calculate_integrity_score()        }                report_file = f"integrity_check_{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}📊 INTEGRITY CHECK SUMMARY{Style.RESET_ALL}")        print(f"{Fore.CYAN}{'='*60}{Style.RESET_ALL}")        print(f"  Files Checked: {self.stats['files_checked']}")        print(f"  Files Modified: {self.stats['files_modified']}")        print(f"  Files Added: {self.stats['files_added']}")        print(f"  Files Missing: {self.stats['files_missing']}")        print(f"  Permission Changes: {self.stats['permission_changes']}")        print(f"  Signature Failures: {self.stats['signature_failures']}")        print(f"  Total Violations: {len(self.violations)}")        print(f"  Integrity Score: {report['integrity_score']:.1f}%")        print(f"{Fore.CYAN}{'='*60}{Style.RESET_ALL}")                if self.stats['files_modified'] > 0 or self.stats['files_missing'] > 0:            print(f"\n{Fore.RED}[!] System integrity compromised!{Style.RESET_ALL}")        elif len(self.violations) > 0:            print(f"\n{Fore.YELLOW}[!] Suspicious changes detected{Style.RESET_ALL}")        else:            print(f"\n{Fore.GREEN}[✓] System integrity verified{Style.RESET_ALL}")                print(f"\n{Fore.GREEN}[✓] Report saved to {report_file}{Style.RESET_ALL}")                return report_file        def calculate_integrity_score(self):        """Calculate integrity score (0-100%)"""        total_files = len(self.baseline['files'])        if total_files == 0:            return 0                violations_weight = {            'modified': 1.0,            'missing': 1.0,            'new': 0.5,            'permission_changes': 0.3        }                total_penalty = 0        for violation in self.violations:            if isinstance(violation, dict):                status = violation.get('status', '')                total_penalty += violations_weight.get(status, 0.5)                score = max(0, 100 - (total_penalty / total_files * 100))        return score def main():    parser = argparse.ArgumentParser(description='System Integrity Checker')    parser.add_argument('--create-baseline', action='store_true', help='Create new baseline')    parser.add_argument('--check', action='store_true', help='Run integrity check')    parser.add_argument('--verify-all', action='store_true', help='Check all files (slow)')    parser.add_argument('--baseline', default='integrity_baseline.json', help='Baseline file')        args = parser.parse_args()        checker = IntegrityChecker(        baseline_file=args.baseline,        verify_all=args.verify_all    )        if args.create_baseline:        checker.create_baseline()    elif args.check:        checker.run_integrity_check()    else:        print("Use --create-baseline or --check") if __name__ == "__main__":    main()