app_whitelisting.py

#!/usr/bin/env python3"""Application Whitelisting for Ransomware PreventionThis script implements application whitelisting to prevent unauthorizedexecutables from running, effectively blocking ransomware and other malware.""" import osimport sysimport hashlibimport jsonimport timeimport threadingimport argparseimport subprocessimport psutilimport signalfrom pathlib import Pathfrom datetime import datetime, timedeltafrom collections import defaultdictfrom colorama import init, Fore, Style init(autoreset=True) class ApplicationWhitelist:    """    Application Whitelisting System        Only allows approved applications to run. All other executables are blocked.    """        def __init__(self, whitelist_file="whitelist.json", policy="enforce",                 monitor_mode=False, log_file="app_whitelist.log"):        """        Initialize Application Whitelisting                Args:            whitelist_file: File containing whitelisted applications            policy: 'enforce' (block non-whitelisted) or 'audit' (only log)            monitor_mode: Monitor but don't block            log_file: Log file path        """        self.whitelist_file = whitelist_file        self.policy = policy        self.monitor_mode = monitor_mode        self.log_file = log_file                # Whitelist storage        self.whitelist = self.load_whitelist()        self.temp_whitelist = {}  # Temporary approvals                # Process tracking        self.running_processes = {}        self.blocked_attempts = defaultdict(int)                # Statistics        self.stats = {            'total_processes': 0,            'whitelisted': 0,            'blocked': 0,            'temp_approved': 0,            'alerts': 0,            'start_time': time.time()        }                # Control flags        self.running = True        self.monitoring = False                # Setup logging        self.setup_logging()                # Common system paths that are usually safe        self.system_paths = [            'C:\\Windows\\System32\\',            'C:\\Windows\\SysWOW64\\',            '/usr/bin/',            '/bin/',            '/sbin/',            '/usr/sbin/'        ]                # High-risk paths to monitor closely        self.high_risk_paths = [            str(Path.home() / 'Downloads'),            str(Path.home() / 'Desktop'),            str(Path.home() / 'AppData' / 'Local' / 'Temp'),            '/tmp',            '/var/tmp'        ]        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_whitelist(self):        """Load whitelist from file"""        if os.path.exists(self.whitelist_file):            try:                with open(self.whitelist_file, 'r') as f:                    return json.load(f)            except Exception as e:                self.logger.error(f"Error loading whitelist: {e}")                # Default whitelist        return {            'paths': [],            'hashes': [],            'signers': [],            'categories': {                'system': [],                'browsers': ['chrome.exe', 'firefox.exe', 'msedge.exe'],                'office': ['winword.exe', 'excel.exe', 'powerpnt.exe'],                'development': ['code.exe', 'pycharm64.exe', 'eclipse.exe']            },            'temp_approvals': {}        }        def save_whitelist(self):        """Save whitelist to file"""        try:            with open(self.whitelist_file, 'w') as f:                json.dump(self.whitelist, f, indent=2)            self.logger.info("Whitelist saved successfully")        except Exception as e:            self.logger.error(f"Error saving whitelist: {e}")        def calculate_file_hash(self, filepath):        """Calculate SHA-256 hash of executable"""        try:            sha256 = hashlib.sha256()            with open(filepath, 'rb') as f:                for chunk in iter(lambda: f.read(4096), b''):                    sha256.update(chunk)            return sha256.hexdigest()        except:            return None        def is_path_whitelisted(self, filepath):        """Check if path is whitelisted"""        # Check exact path matches        if filepath in self.whitelist.get('paths', []):            return True                # Check directory whitelisting        for whitelisted_path in self.whitelist.get('paths', []):            if os.path.dirname(filepath).startswith(whitelisted_path):                return True                # Check system paths (implicitly whitelisted)        for sys_path in self.system_paths:            if filepath.startswith(sys_path):                return True                return False        def is_hash_whitelisted(self, filepath):        """Check if file hash is whitelisted"""        file_hash = self.calculate_file_hash(filepath)        if file_hash and file_hash in self.whitelist.get('hashes', []):            return True        return False        def is_temp_approved(self, filepath):        """Check if temporarily approved"""        return filepath in self.whitelist.get('temp_approvals', {})        def is_whitelisted(self, filepath):        """Check if application is whitelisted"""        # Check path        if self.is_path_whitelisted(filepath):            return True                # Check hash        if self.is_hash_whitelisted(filepath):            return True                # Check temp approval        if self.is_temp_approved(filepath):            # Check if approval expired            approval_time = self.whitelist['temp_approvals'].get(filepath, {}).get('timestamp', 0)            if time.time() - approval_time < 3600:  # 1 hour expiry                return True            else:                # Remove expired approval                del self.whitelist['temp_approvals'][filepath]                return False        def check_process(self, proc):        """Check if process should be allowed to run"""        try:            proc_name = proc.name()            proc_path = proc.exe() if hasattr(proc, 'exe') else None            proc_cmdline = ' '.join(proc.cmdline())                        if not proc_path:                return True  # Can't determine path, allow (should be rare)                        # Check if already tracked            if proc.pid in self.running_processes:                return True                        # Check if whitelisted            if self.is_whitelisted(proc_path):                self.running_processes[proc.pid] = {                    'name': proc_name,                    'path': proc_path,                    'cmdline': proc_cmdline,                    'start_time': time.time()                }                self.stats['whitelisted'] += 1                return True                        # Check if in high-risk path            is_high_risk = any(proc_path.startswith(path) for path in self.high_risk_paths)                        # Log non-whitelisted process            self.logger.warning(f"Non-whitelisted process detected: {proc_name} ({proc_path})")            self.stats['blocked'] += 1            self.stats['alerts'] += 1                        # Track blocked attempts            self.blocked_attempts[proc_path] += 1                        # Alert message            alert_msg = f"⚠️ BLOCKED: {proc_name} from {proc_path}"            if is_high_risk:                alert_msg += " [HIGH RISK LOCATION]"                        print(f"{Fore.RED}{alert_msg}{Style.RESET_ALL}")                        # Take action based on policy            if self.policy == 'enforce' and not self.monitor_mode:                self.terminate_process(proc)                return False                        return True  # In audit/monitor mode, allow but log                    except (psutil.NoSuchProcess, psutil.AccessDenied):            return True        def terminate_process(self, proc):        """Terminate non-whitelisted process"""        try:            proc.terminate()            # Give it time to terminate gracefully            time.sleep(1)            if proc.is_running():                proc.kill()  # Force kill if still running            self.logger.info(f"Terminated process: {proc.name()} (PID: {proc.pid})")            print(f"{Fore.RED}  └─ Process terminated{Style.RESET_ALL}")        except Exception as e:            self.logger.error(f"Error terminating process: {e}")        def monitor_processes(self):        """Monitor running processes"""        while self.running:            try:                # Check new processes                for proc in psutil.process_iter(['pid', 'name', 'exe', 'cmdline']):                    try:                        if proc.pid not in self.running_processes:                            self.stats['total_processes'] += 1                            self.check_process(proc)                    except (psutil.NoSuchProcess, psutil.AccessDenied):                        pass                                # Clean up dead processes                dead_pids = []                for pid in self.running_processes:                    if not psutil.pid_exists(pid):                        dead_pids.append(pid)                                for pid in dead_pids:                    del self.running_processes[pid]                                time.sleep(1)  # Check every second                            except Exception as e:                self.logger.error(f"Error in process monitoring: {e}")                time.sleep(5)        def add_to_whitelist(self, filepath, category="user"):        """Add application to whitelist"""        if not os.path.exists(filepath):            self.logger.error(f"File not found: {filepath}")            return False                # Add by path        if filepath not in self.whitelist['paths']:            self.whitelist['paths'].append(filepath)                # Add by hash        file_hash = self.calculate_file_hash(filepath)        if file_hash and file_hash not in self.whitelist['hashes']:            self.whitelist['hashes'].append(file_hash)                # Add to category        if category not in self.whitelist['categories']:            self.whitelist['categories'][category] = []                basename = os.path.basename(filepath)        if basename not in self.whitelist['categories'][category]:            self.whitelist['categories'][category].append(basename)                self.save_whitelist()        self.logger.info(f"Added {filepath} to whitelist")        return True        def temp_approve(self, filepath, duration=3600):        """Temporarily approve an application"""        self.whitelist['temp_approvals'][filepath] = {            'timestamp': time.time(),            'duration': duration,            'approved_by': 'user'        }        self.save_whitelist()        self.stats['temp_approved'] += 1        self.logger.info(f"Temporarily approved {filepath} for {duration} seconds")        def remove_from_whitelist(self, filepath):        """Remove application from whitelist"""        if filepath in self.whitelist['paths']:            self.whitelist['paths'].remove(filepath)                # Remove by hash        file_hash = self.calculate_file_hash(filepath)        if file_hash and file_hash in self.whitelist['hashes']:            self.whitelist['hashes'].remove(file_hash)                self.save_whitelist()        self.logger.info(f"Removed {filepath} from whitelist")        def scan_directory_for_whitelist(self, directory, category="scanned"):        """Scan directory and add all executables to whitelist"""        added = 0        for root, dirs, files in os.walk(directory):            for file in files:                if file.endswith(('.exe', '.bat', '.cmd', '.ps1', '.sh', '.bin')):                    filepath = os.path.join(root, file)                    if self.add_to_whitelist(filepath, category):                        added += 1                self.logger.info(f"Added {added} executables from {directory} to whitelist")        return added        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}📊 APPLICATION WHITELIST STATUS{Style.RESET_ALL}")        print(f"{Fore.CYAN}{'='*60}{Style.RESET_ALL}")        print(f"  Uptime: {int(uptime)} seconds")        print(f"  Policy: {self.policy.upper()}")        print(f"  Mode: {'MONITOR' if self.monitor_mode else 'ENFORCE'}")        print(f"  Total Processes: {self.stats['total_processes']}")        print(f"  Whitelisted: {self.stats['whitelisted']}")        print(f"  Blocked: {self.stats['blocked']}")        print(f"  Temp Approved: {self.stats['temp_approved']}")        print(f"  Alerts: {self.stats['alerts']}")        print(f"{Fore.CYAN}{'='*60}{Style.RESET_ALL}")                # Show whitelist stats        print(f"\n{Fore.GREEN}Whitelist Contents:{Style.RESET_ALL}")        print(f"  Paths: {len(self.whitelist.get('paths', []))}")        print(f"  Hashes: {len(self.whitelist.get('hashes', []))}")        print(f"  Categories: {len(self.whitelist.get('categories', {}))}")        print(f"  Temp Approvals: {len(self.whitelist.get('temp_approvals', {}))}")                # Show blocked attempts        if self.blocked_attempts:            print(f"\n{Fore.RED}Top Blocked Applications:{Style.RESET_ALL}")            for path, count in sorted(self.blocked_attempts.items(),                                      key=lambda x: x[1], reverse=True)[:5]:                print(f"  {os.path.basename(path)}: {count} attempts")                # Show running whitelisted processes        if self.running_processes:            print(f"\n{Fore.GREEN}Running Whitelisted Processes:{Style.RESET_ALL}")            for pid, info in list(self.running_processes.items())[:10]:                print(f"  {pid}: {info['name']}")        def stats_loop(self):        """Periodically print statistics"""        while self.running:            time.sleep(10)            self.print_stats()        def start_monitoring(self):        """Start application whitelisting"""        print(f"""        {Fore.CYAN}╔═══════════════════════════════════════╗        ║   Application Whitelisting System   ║        ║     Ransomware Prevention Tool      ║        ╚═══════════════════════════════════════╝{Style.RESET_ALL}        """)                print(f"{Fore.GREEN}[*] Starting application whitelisting...{Style.RESET_ALL}")        print(f"{Fore.GREEN}    Policy: {self.policy}{Style.RESET_ALL}")        print(f"{Fore.GREEN}    Mode: {'Monitoring only' if self.monitor_mode else 'Enforcing'}{Style.RESET_ALL}")        print(f"{Fore.GREEN}    Whitelist: {len(self.whitelist['paths'])} paths, {len(self.whitelist['hashes'])} hashes{Style.RESET_ALL}")                self.monitoring = True                # 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.YELLOW}[*] Monitoring active. Press Ctrl+C to stop.{Style.RESET_ALL}\n")                try:            while self.running:                time.sleep(1)        except KeyboardInterrupt:            self.stop()        def stop(self):        """Stop monitoring"""        self.running = False        print(f"\n{Fore.GREEN}[✓] Application whitelisting stopped{Style.RESET_ALL}")        self.print_stats() def interactive_menu(whitelist):    """Interactive menu for managing whitelist"""    while True:        print(f"\n{Fore.CYAN}{'='*50}{Style.RESET_ALL}")        print(f"{Fore.CYAN}APPLICATION WHITELIST MENU{Style.RESET_ALL}")        print(f"{Fore.CYAN}{'='*50}{Style.RESET_ALL}")        print("1. Start monitoring")        print("2. Add application to whitelist")        print("3. Remove from whitelist")        print("4. Temporarily approve application")        print("5. Scan directory for whitelist")        print("6. View whitelist")        print("7. View statistics")        print("8. Toggle policy (enforce/audit)")        print("9. Toggle mode (monitor/enforce)")        print("0. Exit")                choice = input(f"\n{Fore.YELLOW}Enter choice: {Style.RESET_ALL}")                if choice == '1':            whitelist.start_monitoring()                elif choice == '2':            path = input("Enter application path: ")            category = input("Enter category (default: user): ") or "user"            whitelist.add_to_whitelist(path, category)                elif choice == '3':            path = input("Enter application path: ")            whitelist.remove_from_whitelist(path)                elif choice == '4':            path = input("Enter application path: ")            duration = int(input("Enter approval duration (seconds, default 3600): ") or "3600")            whitelist.temp_approve(path, duration)                elif choice == '5':            directory = input("Enter directory to scan: ")            category = input("Enter category: ")            whitelist.scan_directory_for_whitelist(directory, category)                elif choice == '6':            print(f"\n{Fore.GREEN}Whitelist Contents:{Style.RESET_ALL}")            print(json.dumps(whitelist.whitelist, indent=2))                elif choice == '7':            whitelist.print_stats()                elif choice == '8':            whitelist.policy = 'audit' if whitelist.policy == 'enforce' else 'enforce'            print(f"Policy changed to: {whitelist.policy}")                elif choice == '9':            whitelist.monitor_mode = not whitelist.monitor_mode            print(f"Mode changed to: {'Monitor' if whitelist.monitor_mode else 'Enforce'}")                elif choice == '0':            break def main():    parser = argparse.ArgumentParser(description='Application Whitelisting for Ransomware Prevention')    parser.add_argument('--whitelist-file', default='whitelist.json', help='Whitelist file')    parser.add_argument('--policy', choices=['enforce', 'audit'], default='enforce',                       help='Enforcement policy')    parser.add_argument('--monitor', action='store_true', help="Monitor mode (don't block)")    parser.add_argument('--interactive', action='store_true', help='Interactive mode')    parser.add_argument('--add', help='Add application to whitelist')    parser.add_argument('--scan', help='Scan directory and add to whitelist')        args = parser.parse_args()        whitelist = ApplicationWhitelist(        whitelist_file=args.whitelist_file,        policy=args.policy,        monitor_mode=args.monitor    )        if args.add:        whitelist.add_to_whitelist(args.add)        elif args.scan:        whitelist.scan_directory_for_whitelist(args.scan)        elif args.interactive:        interactive_menu(whitelist)        else:        whitelist.start_monitoring() if __name__ == "__main__":    main()