""" Django management command to rollback permission migration """ from django.core.management.base import BaseCommand, CommandError from django.db import transaction from users.enhanced_permissions_models import PagePermission, RolePermissionTemplate, UserPagePermission import json import os from datetime import datetime class Command(BaseCommand): help = 'Rollback permission migration from granular system to legacy system' def add_arguments(self, parser): parser.add_argument( '--backup-file', type=str, help='Path to backup file to restore from', ) parser.add_argument( '--dry-run', action='store_true', help='Run rollback in dry-run mode without making changes', ) parser.add_argument( '--confirm', action='store_true', help='Confirm that you want to rollback (required for safety)', ) def handle(self, *args, **options): self.dry_run = options['dry_run'] self.backup_file = options['backup_file'] if not options['confirm'] and not self.dry_run: self.stdout.write( self.style.ERROR( 'This operation will delete all granular permissions. ' 'Use --confirm to proceed or --dry-run to test.' ) ) return if not self.backup_file: # Find the most recent backup file backup_files = [f for f in os.listdir('.') if f.startswith('permission_backup_')] if not backup_files: raise CommandError('No backup file specified and no backup files found') self.backup_file = sorted(backup_files)[-1] self.stdout.write(f'Using most recent backup: {self.backup_file}') if not os.path.exists(self.backup_file): raise CommandError(f'Backup file not found: {self.backup_file}') try: with open(self.backup_file, 'r') as f: backup_data = json.load(f) except Exception as e: raise CommandError(f'Failed to load backup file: {str(e)}') self.stdout.write( self.style.WARNING('Starting permission rollback...') ) try: with transaction.atomic(): # Step 1: Clear granular permissions self.clear_granular_permissions() # Step 2: Restore legacy permissions from backup self.restore_legacy_permissions(backup_data) # Step 3: Validate rollback self.validate_rollback(backup_data) if self.dry_run: self.stdout.write( self.style.WARNING('DRY RUN: Rolling back transaction...') ) raise transaction.TransactionManagementError("Dry run - rolling back") self.stdout.write( self.style.SUCCESS('Permission rollback completed successfully!') ) except Exception as e: self.stdout.write( self.style.ERROR(f'Rollback failed: {str(e)}') ) raise CommandError(f'Rollback failed: {str(e)}') def clear_granular_permissions(self): """Clear all granular permission data""" self.stdout.write('Clearing granular permissions...') # Count before deletion user_perms_count = UserPagePermission.objects.count() role_templates_count = RolePermissionTemplate.objects.count() page_perms_count = PagePermission.objects.count() if not self.dry_run: # Delete in order to respect foreign key constraints UserPagePermission.objects.all().delete() RolePermissionTemplate.objects.all().delete() PagePermission.objects.all().delete() self.stdout.write(f' Cleared {user_perms_count} user page permissions') self.stdout.write(f' Cleared {role_templates_count} role permission templates') self.stdout.write(f' Cleared {page_perms_count} page permissions') def restore_legacy_permissions(self, backup_data): """Restore legacy permissions from backup data""" self.stdout.write('Restoring legacy permissions...') from users.models import RolePermission, DefaultRolePermission, UserPermission, CustomUser # Restore role permissions role_perms_restored = 0 for perm_data in backup_data.get('role_permissions', []): if not self.dry_run: RolePermission.objects.get_or_create( role=perm_data['role'], module=perm_data['module'], action=perm_data['action'], defaults={ 'is_allowed': perm_data['is_allowed'], } ) role_perms_restored += 1 # Restore default role permissions default_perms_restored = 0 for perm_data in backup_data.get('default_role_permissions', []): if not self.dry_run: DefaultRolePermission.objects.get_or_create( role=perm_data['role'], module=perm_data['module'], action=perm_data['action'], defaults={ 'is_allowed': perm_data['is_allowed'], 'description': perm_data.get('description', ''), } ) default_perms_restored += 1 # Restore user permissions user_perms_restored = 0 for perm_data in backup_data.get('user_permissions', []): try: if not self.dry_run: user = CustomUser.objects.get(id=perm_data['user_id']) granted_by = None if perm_data.get('granted_by_id'): try: granted_by = CustomUser.objects.get(id=perm_data['granted_by_id']) except CustomUser.DoesNotExist: pass expires_at = None if perm_data.get('expires_at'): expires_at = datetime.fromisoformat(perm_data['expires_at'].replace('Z', '+00:00')) UserPermission.objects.get_or_create( user=user, module=perm_data['module'], action=perm_data['action'], defaults={ 'is_allowed': perm_data['is_allowed'], 'granted_by': granted_by, 'reason': perm_data.get('reason', ''), 'expires_at': expires_at, } ) user_perms_restored += 1 except CustomUser.DoesNotExist: self.stdout.write( self.style.WARNING( f'User not found for permission: {perm_data["user_username"]}' ) ) self.stdout.write(f' Restored {role_perms_restored} role permissions') self.stdout.write(f' Restored {default_perms_restored} default role permissions') self.stdout.write(f' Restored {user_perms_restored} user permissions') def validate_rollback(self, backup_data): """Validate the rollback was successful""" self.stdout.write('Validating rollback...') from users.models import RolePermission, DefaultRolePermission, UserPermission # Check counts match backup expected_role_perms = len(backup_data.get('role_permissions', [])) expected_default_perms = len(backup_data.get('default_role_permissions', [])) expected_user_perms = len(backup_data.get('user_permissions', [])) actual_role_perms = RolePermission.objects.count() actual_default_perms = DefaultRolePermission.objects.count() actual_user_perms = UserPermission.objects.count() self.stdout.write(f'Role permissions: {actual_role_perms}/{expected_role_perms}') self.stdout.write(f'Default role permissions: {actual_default_perms}/{expected_default_perms}') self.stdout.write(f'User permissions: {actual_user_perms}/{expected_user_perms}') # Check granular permissions are cleared granular_perms_remaining = ( PagePermission.objects.count() + RolePermissionTemplate.objects.count() + UserPagePermission.objects.count() ) if granular_perms_remaining > 0: self.stdout.write( self.style.WARNING( f'Warning: {granular_perms_remaining} granular permissions still exist' ) ) else: self.stdout.write( self.style.SUCCESS('All granular permissions successfully cleared') ) self.stdout.write(self.style.SUCCESS('Rollback validation completed'))