""" Enhanced management command to view and manage rate limited IPs with bot detection. """ from django.core.management.base import BaseCommand from django.core.cache import cache from django.conf import settings import time class Command(BaseCommand): help = 'Manage enhanced rate limited IP addresses with bot detection' def add_arguments(self, parser): parser.add_argument( '--list-blocked', action='store_true', help='List information about blocked IP management', ) parser.add_argument( '--unblock', type=str, help='Unblock a specific IP address for all request types', metavar='IP_ADDRESS' ) parser.add_argument( '--block', type=str, help='Block a specific IP address as suspicious', metavar='IP_ADDRESS' ) parser.add_argument( '--clear-all', action='store_true', help='Clear all rate limiting data (unblock all IPs)', ) parser.add_argument( '--stats', type=str, help='Show request statistics for a specific IP across all request types', metavar='IP_ADDRESS' ) parser.add_argument( '--recent-blocks', action='store_true', help='Show recent blocking activity from logs', ) def handle(self, *args, **options): if options['list_blocked']: self.list_blocked_info() elif options['unblock']: self.unblock_ip(options['unblock']) elif options['block']: self.block_ip(options['block']) elif options['clear_all']: self.clear_all_limits() elif options['stats']: self.show_ip_stats(options['stats']) elif options['recent_blocks']: self.show_recent_blocks() else: self.stdout.write(self.style.WARNING('No action specified. Use --help for options.')) def get_cache_key(self, ip_address, key_type, request_type='normal'): """Generate cache key for IP address and request type.""" return f"rate_limit_{key_type}_{request_type}_{ip_address}" def list_blocked_info(self): """Show information about the blocking system.""" self.stdout.write(self.style.SUCCESS('Enhanced Rate Limiting System Info:')) self.stdout.write('') # Configuration info self.stdout.write('Configuration:') self.stdout.write(f' Normal requests: {getattr(settings, "RATE_LIMIT_REQUESTS", 100)}/hour') self.stdout.write(f' Bot requests: {getattr(settings, "RATE_LIMIT_BOT_REQUESTS", 20)}/hour') self.stdout.write(f' Suspicious requests: {getattr(settings, "RATE_LIMIT_SUSPICIOUS_REQUESTS", 5)}/hour') self.stdout.write('') self.stdout.write('Block Durations:') self.stdout.write(f' Normal: {getattr(settings, "RATE_LIMIT_BLOCK_DURATION", 1800)} seconds') self.stdout.write(f' Bot/Suspicious: {getattr(settings, "RATE_LIMIT_BOT_BLOCK_DURATION", 3600)} seconds') self.stdout.write('') self.stdout.write('To check specific IPs, use: --stats IP_ADDRESS') def unblock_ip(self, ip_address): """Unblock a specific IP address for all request types.""" request_types = ['normal', 'bot', 'suspicious'] unblocked_count = 0 for request_type in request_types: block_key = self.get_cache_key(ip_address, "block", request_type) count_key = self.get_cache_key(ip_address, "count", request_type) if cache.get(block_key): unblocked_count += 1 # Remove both block and count entries cache.delete(block_key) cache.delete(count_key) if unblocked_count > 0: self.stdout.write( self.style.SUCCESS(f'Successfully unblocked IP address: {ip_address} ({unblocked_count} block types removed)') ) else: self.stdout.write( self.style.WARNING(f'IP address {ip_address} was not blocked, but cleared any existing data') ) def block_ip(self, ip_address): """Block a specific IP address as suspicious.""" block_key = self.get_cache_key(ip_address, "block", "suspicious") block_duration = getattr(settings, 'RATE_LIMIT_BOT_BLOCK_DURATION', 3600) cache.set(block_key, True, block_duration) self.stdout.write( self.style.WARNING(f'Blocked IP address: {ip_address} as suspicious for {block_duration} seconds') ) def clear_all_limits(self): """Clear all rate limiting data.""" # This will clear the entire cache, which might affect other cached data # In production, consider a more targeted approach cache.clear() self.stdout.write( self.style.SUCCESS('Cleared all rate limiting data') ) def show_ip_stats(self, ip_address): """Show statistics for a specific IP address across all request types.""" request_types = ['normal', 'bot', 'suspicious'] self.stdout.write(f'Enhanced statistics for IP: {ip_address}') self.stdout.write('=' * 50) any_data = False for request_type in request_types: block_key = self.get_cache_key(ip_address, "block", request_type) count_key = self.get_cache_key(ip_address, "count", request_type) is_blocked = cache.get(block_key, False) request_data = cache.get(count_key, None) if is_blocked or request_data: any_data = True self.stdout.write(f'\n{request_type.upper()} Traffic:') self.stdout.write(f' Currently blocked: {"Yes" if is_blocked else "No"}') if request_data: current_time = time.time() window_start = request_data['window_start'] count = request_data['count'] time_window = getattr(settings, 'RATE_LIMIT_WINDOW', 3600) # Get limits for this request type if request_type == 'suspicious': limit = getattr(settings, 'RATE_LIMIT_SUSPICIOUS_REQUESTS', 5) elif request_type == 'bot': limit = getattr(settings, 'RATE_LIMIT_BOT_REQUESTS', 20) else: limit = getattr(settings, 'RATE_LIMIT_REQUESTS', 100) time_in_window = current_time - window_start remaining_time = max(0, time_window - time_in_window) self.stdout.write(f' Requests in current window: {count}/{limit}') self.stdout.write(f' Window started: {time.ctime(window_start)}') self.stdout.write(f' Time remaining in window: {remaining_time:.0f} seconds') if count >= limit: self.stdout.write(self.style.ERROR(f' WARNING: Limit exceeded for {request_type} requests!')) if not any_data: self.stdout.write('No request data found for this IP across any request type') def show_recent_blocks(self): """Show information about recent blocking activity.""" self.stdout.write(self.style.SUCCESS('Recent Blocking Activity Info:')) self.stdout.write('') self.stdout.write('To see recent blocks, check your application logs for entries containing:') self.stdout.write(' - "BLOCKED BOT request"') self.stdout.write(' - "BLOCKED SUSPICIOUS request"') self.stdout.write(' - "RATE LIMIT EXCEEDED"') self.stdout.write('') self.stdout.write('Log locations to check:') self.stdout.write(' - Application logs (Django logging)') self.stdout.write(' - System logs (/var/log/messages, /var/log/syslog)') self.stdout.write(' - Web server logs (nginx/apache access/error logs)') self.stdout.write('') self.stdout.write('Example log analysis commands:') self.stdout.write(' grep "BLOCKED" /path/to/your/logs | tail -20') self.stdout.write(' grep "RATE LIMIT" /path/to/your/logs | tail -20')