kernel_patching.py

#!/usr/bin/env python3"""Kernel Patching and Update ManagementThis script manages kernel updates and patches to prevent rootkitexploitation of kernel vulnerabilities.""" import osimport sysimport subprocessimport argparseimport platformimport shutilimport requestsimport hashlibimport tempfilefrom datetime import datetimefrom packaging import versionfrom colorama import init, Fore, Style init(autoreset=True) class KernelPatchManager:    """    Kernel Patch and Update Management System    """        def __init__(self):        self.system = platform.system()        self.kernel_version = platform.release()        self.os_version = platform.version()        self.patches_applied = []        self.vulnerabilities = []            def get_kernel_info(self):        """Get detailed kernel information"""        info = {            'system': self.system,            'kernel_version': self.kernel_version,            'os_version': self.os_version,            'architecture': platform.machine(),            'processor': platform.processor()        }                if self.system == 'Linux':            # Get Linux distribution info            if os.path.exists('/etc/os-release'):                with open('/etc/os-release', 'r') as f:                    for line in f:                        if line.startswith('NAME='):                            info['distro'] = line.split('=')[1].strip().strip('"')                        elif line.startswith('VERSION='):                            info['distro_version'] = line.split('=')[1].strip().strip('"')                elif self.system == 'Windows':            # Get Windows build info            try:                result = subprocess.run(['ver'], capture_output=True, text=True, shell=True)                info['windows_version'] = result.stdout.strip()            except:                pass                return info        def check_kernel_vulnerabilities(self):        """Check for known kernel vulnerabilities"""        vulnerabilities = []                if self.system == 'Linux':            vulnerabilities = self.check_linux_vulnerabilities()        elif self.system == 'Windows':            vulnerabilities = self.check_windows_vulnerabilities()                return vulnerabilities        def check_linux_vulnerabilities(self):        """Check Linux kernel vulnerabilities"""        vulns = []                # Common Linux kernel vulnerabilities        known_vulns = [            {                'name': 'Dirty Cow',                'cve': 'CVE-2016-5195',                'affected_versions': '<4.8.3',                'description': 'Privilege escalation via race condition',                'severity': 'HIGH'            },            {                'name': 'Stack Clash',                'cve': 'CVE-2017-1000364',                'affected_versions': '<4.12',                'description': 'Stack memory corruption',                'severity': 'HIGH'            },            {                'name': 'Spectre',                'cve': 'CVE-2017-5753',                'affected_versions': 'All',                'description': 'Side-channel attack',                'severity': 'MEDIUM'            },            {                'name': 'Meltdown',                'cve': 'CVE-2017-5754',                'affected_versions': '<4.15',                'description': 'Kernel memory leak',                'severity': 'HIGH'            },            {                'name': 'BlueBorne',                'cve': 'CVE-2017-1000251',                'affected_versions': '<4.13',                'description': 'Bluetooth stack vulnerability',                'severity': 'MEDIUM'            }        ]                # Parse kernel version        try:            kernel_ver = version.parse(self.kernel_version.split('-')[0])                        for vuln in known_vulns:                affected = vuln['affected_versions']                if affected != 'All':                    # Simple version comparison                    # This is simplified - real checking would be more complex                    vulns.append({                        **vuln,                        'status': 'CHECK_MANUALLY'                    })                else:                    vulns.append({                        **vuln,                        'status': 'POTENTIALLY_VULNERABLE'                    })                            except Exception as e:            print(f"{Fore.RED}[!] Error parsing kernel version: {e}{Style.RESET_ALL}")                return vulns        def check_windows_vulnerabilities(self):        """Check Windows kernel vulnerabilities"""        vulns = []                # Known Windows kernel vulnerabilities        known_vulns = [            {                'name': 'EternalBlue',                'cve': 'CVE-2017-0144',                'affected': 'Windows 7/8/10 pre-2017',                'description': 'SMB remote code execution',                'severity': 'CRITICAL'            },            {                'name': 'BlueKeep',                'cve': 'CVE-2019-0708',                'affected': 'Windows 7/2008 R2',                'description': 'RDP remote code execution',                'severity': 'CRITICAL'            },            {                'name': 'PrintNightmare',                'cve': 'CVE-2021-34527',                'affected': 'All Windows versions',                'description': 'Print spooler RCE',                'severity': 'HIGH'            },            {                'name': 'Zerologon',                'cve': 'CVE-2020-1472',                'affected': 'Windows Server 2016/2019',                'description': 'Netlogon privilege escalation',                'severity': 'CRITICAL'            }        ]                # Would need to check Windows build number against patch levels        for vuln in known_vulns:            vulns.append({                **vuln,                'status': 'CHECK_WINDOWS_UPDATE'            })                return vulns        def update_kernel_linux(self, method='auto'):        """Update Linux kernel"""        print(f"\n{Fore.CYAN}[*] Updating Linux kernel...{Style.RESET_ALL}")                try:            if os.path.exists('/etc/debian_version'):                # Debian/Ubuntu                print(f"{Fore.CYAN}  [*] Using apt package manager{Style.RESET_ALL}")                                # Update package list                subprocess.run(['sudo', 'apt-get', 'update'], check=True)                                # Check for available kernel updates                result = subprocess.run(                    ['apt', 'list', '--upgradable', 'linux-image*'],                    capture_output=True, text=True                )                                if 'linux-image' in result.stdout:                    print(f"{Fore.YELLOW}  [!] Kernel updates available:{Style.RESET_ALL}")                    print(result.stdout)                                        # Ask for confirmation                    response = input(f"{Fore.YELLOW}[?] Install updates? (yes/no): {Style.RESET_ALL}")                    if response.lower() == 'yes':                        subprocess.run(['sudo', 'apt-get', 'upgrade', '-y'], check=True)                        print(f"{Fore.GREEN}[✓] Kernel updated successfully{Style.RESET_ALL}")                        print(f"{Fore.YELLOW}[!] Reboot required{Style.RESET_ALL}")                else:                    print(f"{Fore.GREEN}[✓] Kernel is up to date{Style.RESET_ALL}")                        elif os.path.exists('/etc/redhat-release'):                # RHEL/CentOS/Fedora                print(f"{Fore.CYAN}  [*] Using yum/dnf package manager{Style.RESET_ALL}")                                if shutil.which('dnf'):                    cmd = ['sudo', 'dnf', 'update', '-y', 'kernel*']                else:                    cmd = ['sudo', 'yum', 'update', '-y', 'kernel*']                                subprocess.run(cmd, check=True)                print(f"{Fore.GREEN}[✓] Kernel updated{Style.RESET_ALL}")                        elif os.path.exists('/etc/arch-release'):                # Arch Linux                print(f"{Fore.CYAN}  [*] Using pacman package manager{Style.RESET_ALL}")                subprocess.run(['sudo', 'pacman', '-Syu', '--noconfirm'], check=True)                print(f"{Fore.GREEN}[✓] System updated{Style.RESET_ALL}")                    except subprocess.CalledProcessError as e:            print(f"{Fore.RED}[✗] Error updating kernel: {e}{Style.RESET_ALL}")            return False                return True        def update_kernel_windows(self):        """Update Windows kernel via Windows Update"""        print(f"\n{Fore.CYAN}[*] Checking Windows updates...{Style.RESET_ALL}")                try:            # Check for updates using PowerShell            ps_script = """            $Session = New-Object -ComObject Microsoft.Update.Session            $Searcher = $Session.CreateUpdateSearcher()            $SearchResult = $Searcher.Search("IsInstalled=0 and Type='Software'")                        Write-Host "Available updates: $($SearchResult.Updates.Count)"                        if ($SearchResult.Updates.Count -gt 0) {                $UpdatesToDownload = New-Object -ComObject Microsoft.Update.UpdateColl                foreach ($Update in $SearchResult.Updates) {                    if ($Update.Title -like "*Security*" -or $Update.Title -like "*Critical*") {                        $UpdatesToDownload.Add($Update) | Out-Null                        Write-Host "Critical: $($Update.Title)"                    }                }                                if ($UpdatesToDownload.Count -gt 0) {                    $Downloader = $Session.CreateUpdateDownloader()                    $Downloader.Updates = $UpdatesToDownload                    $Downloader.Download()                                        $Installer = $Session.CreateUpdateInstaller()                    $Installer.Updates = $UpdatesToDownload                    $InstallationResult = $Installer.Install()                                        Write-Host "Installation result: $($InstallationResult.ResultCode)"                }            }            """                        # Save script to temp file            with tempfile.NamedTemporaryFile(mode='w', suffix='.ps1', delete=False) as f:                f.write(ps_script)                script_path = f.name                        # Run PowerShell script            result = subprocess.run(                ['powershell', '-ExecutionPolicy', 'Bypass', '-File', script_path],                capture_output=True, text=True            )                        os.unlink(script_path)                        print(result.stdout)                        if 'Reboot' in result.stdout:                print(f"{Fore.YELLOW}[!] Reboot required to complete installation{Style.RESET_ALL}")                    except Exception as e:            print(f"{Fore.RED}[✗] Error updating Windows: {e}{Style.RESET_ALL}")            return False                return True        def apply_security_patches(self):        """Apply specific security patches"""        print(f"\n{Fore.CYAN}[*] Applying security patches...{Style.RESET_ALL}")                if self.system == 'Linux':            # Check for available security patches            try:                if os.path.exists('/etc/debian_version'):                    # Install only security updates                    subprocess.run([                        'sudo', 'apt-get', 'update'                    ], check=True)                                        subprocess.run([                        'sudo', 'apt-get', 'upgrade', '-y',                        '-o', 'APT::Get::Only-Source='                    ], check=True)                                        print(f"{Fore.GREEN}[✓] Security patches applied{Style.RESET_ALL}")                                except Exception as e:                print(f"{Fore.RED}[✗] Error applying patches: {e}{Style.RESET_ALL}")                elif self.system == 'Windows':            # Use Windows Update for security patches            self.update_kernel_windows()        def configure_kernel_parameters(self):        """Configure secure kernel parameters"""        print(f"\n{Fore.CYAN}[*] Configuring kernel parameters...{Style.RESET_ALL}")                if self.system == 'Linux':            # Security-focused sysctl settings            sysctl_settings = {                'kernel.kptr_restrict': '2',                'kernel.dmesg_restrict': '1',                'kernel.printk': '3 3 3 3',                'kernel.randomize_va_space': '2',                'kernel.exec-shield': '1',                'kernel.yama.ptrace_scope': '2',                'net.core.bpf_jit_enable': '0',                'kernel.unprivileged_bpf_disabled': '1',                'net.ipv4.tcp_syncookies': '1',                'net.ipv4.conf.all.rp_filter': '1',                'net.ipv4.conf.default.rp_filter': '1',                'net.ipv4.icmp_echo_ignore_broadcasts': '1',                'net.ipv4.icmp_ignore_bogus_error_responses': '1',                'net.ipv4.conf.all.accept_source_route': '0',                'net.ipv4.conf.default.accept_source_route': '0',                'net.ipv6.conf.all.accept_source_route': '0',                'net.ipv6.conf.default.accept_source_route': '0',                'net.ipv4.tcp_timestamps': '0',                'kernel.core_uses_pid': '1',            }                        sysctl_conf = '/etc/sysctl.d/99-security.conf'                        try:                with open(sysctl_conf, 'w') as f:                    f.write('# Security kernel parameters\n')                    f.write(f'# Applied on {datetime.now()}\n\n')                                        for key, value in sysctl_settings.items():                        f.write(f'{key} = {value}\n')                                # Apply settings                subprocess.run(['sudo', 'sysctl', '-p', sysctl_conf], check=True)                print(f"{Fore.GREEN}[✓] Kernel parameters configured{Style.RESET_ALL}")                            except Exception as e:                print(f"{Fore.RED}[✗] Error configuring kernel parameters: {e}{Style.RESET_ALL}")                elif self.system == 'Windows':            # Windows kernel hardening            ps_script = r"""            # Enable DEP            bcdedit /set {current} nx AlwaysOn                        # Enable ASLR            Set-ItemProperty -Path "HKLM:\SYSTEM\CurrentControlSet\Control\Session Manager\Memory Management" -Name "MoveImages" -Value 0                        # Disable guest account            Set-ItemProperty -Path "HKLM:\SYSTEM\CurrentControlSet\Control\Lsa" -Name "LimitBlankPasswordUse" -Value 1                        # Enable UAC            Set-ItemProperty -Path "HKLM:\SOFTWARE\Microsoft\Windows\CurrentVersion\Policies\System" -Name "EnableLUA" -Value 1            """                        with tempfile.NamedTemporaryFile(mode='w', suffix='.ps1', delete=False) as f:                f.write(ps_script)                script_path = f.name                        try:                subprocess.run(['powershell', '-ExecutionPolicy', 'Bypass', '-File', script_path], check=True)                print(f"{Fore.GREEN}[✓] Windows kernel hardening applied{Style.RESET_ALL}")            except Exception as e:                print(f"{Fore.RED}[✗] Error applying kernel hardening: {e}{Style.RESET_ALL}")                        os.unlink(script_path)        def verify_kernel_integrity(self):        """Verify kernel integrity after updates"""        print(f"\n{Fore.CYAN}[*] Verifying kernel integrity...{Style.RESET_ALL}")                if self.system == 'Linux':            # Check kernel module signatures            try:                result = subprocess.run(['lsmod'], capture_output=True, text=True)                modules = result.stdout.split('\n')[1:]                                unsigned = []                for module in modules:                    if module.strip():                        parts = module.split()                        if len(parts) >= 1:                            module_name = parts[0]                            # Check if module is signed                            mod_path = f'/lib/modules/{self.kernel_version}/kernel/{module_name}.ko'                            if os.path.exists(mod_path):                                # Would check signature here                                pass                                print(f"{Fore.GREEN}[✓] Kernel module verification complete{Style.RESET_ALL}")                            except Exception as e:                print(f"{Fore.RED}[✗] Error verifying kernel: {e}{Style.RESET_ALL}")                elif self.system == 'Windows':            # Check Windows kernel integrity            try:                # Check digital signatures of kernel files                kernel_files = [                    'C:\\Windows\\System32\\ntoskrnl.exe',                    'C:\\Windows\\System32\\hal.dll',                    'C:\\Windows\\System32\\win32k.sys'                ]                                for file in kernel_files:                    if os.path.exists(file):                        # Would check digital signature here                        print(f"{Fore.GREEN}  [✓] {os.path.basename(file)} present{Style.RESET_ALL}")                                    except Exception as e:                print(f"{Fore.RED}[✗] Error verifying Windows kernel: {e}{Style.RESET_ALL}")        def generate_patch_report(self):        """Generate kernel patch report"""        kernel_info = self.get_kernel_info()        vulns = self.check_kernel_vulnerabilities()                report = {            'timestamp': datetime.now().isoformat(),            'kernel_info': kernel_info,            'vulnerabilities': vulns,            'patches_applied': self.patches_applied,            'recommendations': self.get_recommendations(vulns)        }                report_file = f"kernel_patch_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"        with open(report_file, 'w') as f:            import json            json.dump(report, f, indent=2)                # Print summary        print(f"\n{Fore.CYAN}{'='*60}{Style.RESET_ALL}")        print(f"{Fore.CYAN}📊 KERNEL PATCH REPORT{Style.RESET_ALL}")        print(f"{Fore.CYAN}{'='*60}{Style.RESET_ALL}")        print(f"  System: {kernel_info['system']}")        print(f"  Kernel: {kernel_info['kernel_version']}")        print(f"  Vulnerabilities found: {len(vulns)}")        print(f"  Patches applied: {len(self.patches_applied)}")        print(f"{Fore.CYAN}{'='*60}{Style.RESET_ALL}")                if vulns:            print(f"\n{Fore.RED}Vulnerabilities:{Style.RESET_ALL}")            for vuln in vulns:                severity_color = Fore.RED if vuln['severity'] == 'CRITICAL' else Fore.YELLOW                print(f"  {severity_color}[{vuln['severity']}]{Style.RESET_ALL} {vuln['name']} - {vuln['cve']}")                print(f"     {vuln['description']}")                print(f"\n{Fore.GREEN}[✓] Report saved to {report_file}{Style.RESET_ALL}")                return report_file        def get_recommendations(self, vulns):        """Get patch recommendations"""        recommendations = []                for vuln in vulns:            if vuln['severity'] in ['CRITICAL', 'HIGH']:                recommendations.append({                    'priority': 'IMMEDIATE',                    'action': f"Patch {vuln['name']} ({vuln['cve']})",                    'details': vuln['description']                })                # General recommendations        if self.system == 'Linux':            recommendations.append({                'priority': 'REGULAR',                'action': 'Enable automatic security updates',                'details': 'Configure unattended-upgrades'            })        elif self.system == 'Windows':            recommendations.append({                'priority': 'REGULAR',                'action': 'Enable automatic Windows updates',                'details': 'Configure Windows Update for automatic installation'            })                return recommendations        def show_status(self):        """Show kernel patch status"""        kernel_info = self.get_kernel_info()        vulns = self.check_kernel_vulnerabilities()                print(f"\n{Fore.CYAN}{'='*60}{Style.RESET_ALL}")        print(f"{Fore.CYAN}🔧 KERNEL PATCH STATUS{Style.RESET_ALL}")        print(f"{Fore.CYAN}{'='*60}{Style.RESET_ALL}")                print(f"\n{Fore.GREEN}System Information:{Style.RESET_ALL}")        for key, value in kernel_info.items():            print(f"  {key}: {value}")                print(f"\n{Fore.GREEN}Security Status:{Style.RESET_ALL}")        critical = sum(1 for v in vulns if v['severity'] == 'CRITICAL')        high = sum(1 for v in vulns if v['severity'] == 'HIGH')        medium = sum(1 for v in vulns if v['severity'] == 'MEDIUM')                print(f"  Critical vulnerabilities: {Fore.RED}{critical}{Style.RESET_ALL}")        print(f"  High vulnerabilities: {Fore.YELLOW}{high}{Style.RESET_ALL}")        print(f"  Medium vulnerabilities: {Fore.CYAN}{medium}{Style.RESET_ALL}")                if critical > 0:            print(f"\n{Fore.RED}[!] CRITICAL: Apply patches immediately!{Style.RESET_ALL}")                print(f"\n{Fore.CYAN}{'='*60}{Style.RESET_ALL}") def main():    parser = argparse.ArgumentParser(description='Kernel Patch Manager')    parser.add_argument('--status', action='store_true', help='Show kernel patch status')    parser.add_argument('--update', action='store_true', help='Update kernel')    parser.add_argument('--patch', action='store_true', help='Apply security patches')    parser.add_argument('--configure', action='store_true', help='Configure kernel parameters')    parser.add_argument('--verify', action='store_true', help='Verify kernel integrity')    parser.add_argument('--report', action='store_true', help='Generate patch report')    parser.add_argument('--auto', action='store_true', help='Run all updates automatically')        args = parser.parse_args()        manager = KernelPatchManager()        if args.auto:        print(f"{Fore.CYAN}[*] Running automatic kernel maintenance...{Style.RESET_ALL}")        manager.update_kernel_linux() if manager.system == 'Linux' else manager.update_kernel_windows()        manager.apply_security_patches()        manager.configure_kernel_parameters()        manager.verify_kernel_integrity()        manager.generate_patch_report()        if args.status:        manager.show_status()        if args.update:        if manager.system == 'Linux':            manager.update_kernel_linux()        elif manager.system == 'Windows':            manager.update_kernel_windows()        if args.patch:        manager.apply_security_patches()        if args.configure:        manager.configure_kernel_parameters()        if args.verify:        manager.verify_kernel_integrity()        if args.report:        manager.generate_patch_report()        if not any([args.status, args.update, args.patch, args.configure, args.verify, args.report, args.auto]):        manager.show_status() if __name__ == "__main__":    main()