"""Management Command: Migrate Legacy Configurations to Standalone This command migrates existing inline VPN and FTP configurations from ChannelZoneRelation to the new standalone configuration models. Usage: python manage.py migrate_to_standalone_configs [--dry-run] [--force] Author: Senior Django Developer Version: 1.0.0 """ from django.core.management.base import BaseCommand, CommandError from django.db import transaction, models from django.utils import timezone from apps.channels.models import ( ChannelZoneRelation, StandaloneVPNConfiguration, StandaloneFTPConfiguration, ZoneVPNConfiguration, ZoneFTPConfiguration, IPSecConfiguration, OpenVPNConfiguration, WireGuardConfiguration ) class Command(BaseCommand): help = 'Migrate legacy inline VPN/FTP configurations to standalone configurations' def add_arguments(self, parser): parser.add_argument( '--dry-run', action='store_true', help='Show what would be migrated without making changes', ) parser.add_argument( '--force', action='store_true', help='Force migration even if standalone configs already exist', ) parser.add_argument( '--channel-id', type=str, help='Migrate configurations for a specific channel only', ) parser.add_argument( '--zone-id', type=str, help='Migrate configurations for a specific zone only', ) def handle(self, *args, **options): dry_run = options['dry_run'] force = options['force'] channel_id = options.get('channel_id') zone_id = options.get('zone_id') self.stdout.write( self.style.SUCCESS('Starting migration of legacy configurations to standalone...') ) if dry_run: self.stdout.write( self.style.WARNING('DRY RUN MODE - No changes will be made') ) # Build queryset queryset = ChannelZoneRelation.objects.filter(is_active=True) if channel_id: queryset = queryset.filter(channel_id=channel_id) if zone_id: queryset = queryset.filter(zone_id=zone_id) # Get relations with legacy configurations from django.db import models legacy_relations = queryset.filter( models.Q(vpn_config__isnull=False) | models.Q(ftp_config__isnull=False) ).select_related('channel', 'zone') if not legacy_relations.exists(): self.stdout.write( self.style.WARNING('No legacy configurations found to migrate.') ) return self.stdout.write( f'Found {legacy_relations.count()} relations with legacy configurations' ) migration_stats = { 'vpn_configs_created': 0, 'ftp_configs_created': 0, 'relations_updated': 0, 'errors': [] } try: with transaction.atomic(): for relation in legacy_relations: try: self._migrate_relation(relation, dry_run, force, migration_stats) except Exception as e: error_msg = f'Error migrating relation {relation.id}: {str(e)}' migration_stats['errors'].append(error_msg) self.stdout.write(self.style.ERROR(error_msg)) if dry_run: # Rollback transaction in dry run mode transaction.set_rollback(True) except Exception as e: raise CommandError(f'Migration failed: {str(e)}') # Print migration summary self._print_summary(migration_stats, dry_run) def _migrate_relation(self, relation, dry_run, force, stats): """Migrate a single ChannelZoneRelation.""" self.stdout.write( f'Processing relation: {relation.channel.name} -> {relation.zone.name}' ) # Migrate VPN configuration if relation.vpn_type and not relation.standalone_vpn_config: vpn_config = self._create_vpn_config(relation, dry_run, force) if vpn_config: if not dry_run: relation.standalone_vpn_config = vpn_config stats['vpn_configs_created'] += 1 # Migrate FTP configuration if relation.ftp_host and not relation.standalone_ftp_config: ftp_config = self._create_ftp_config(relation, dry_run, force) if ftp_config: if not dry_run: relation.standalone_ftp_config = ftp_config stats['ftp_configs_created'] += 1 # Save the relation if changes were made if not dry_run and (relation.standalone_vpn_config or relation.standalone_ftp_config): relation.save() stats['relations_updated'] += 1 def _create_vpn_config(self, relation, dry_run, force): """Create standalone VPN configuration from legacy data.""" config_name = f"{relation.channel.name}_{relation.zone.name}_VPN" # Check if config already exists existing_config = StandaloneVPNConfiguration.objects.filter( name=config_name ).first() if existing_config and not force: self.stdout.write( f' VPN config "{config_name}" already exists, skipping' ) return existing_config if dry_run: self.stdout.write( f' Would create VPN config: {config_name} ({relation.vpn_type})' ) return None # Create the standalone VPN configuration vpn_config = StandaloneVPNConfiguration.objects.create( name=config_name, description=f'Migrated from {relation.channel.name} - {relation.zone.name}', vpn_type=relation.vpn_type, is_active=True ) # Create specific VPN type configuration if relation.vpn_type == 'ipsec': ipsec_config = IPSecConfiguration.objects.create( gateway=getattr(relation, 'ipsec_gateway', ''), psk=getattr(relation, 'ipsec_psk', ''), local_id=getattr(relation, 'ipsec_local_id', ''), remote_id=getattr(relation, 'ipsec_remote_id', ''), encryption_algorithm=getattr(relation, 'ipsec_encryption', 'aes256'), hash_algorithm=getattr(relation, 'ipsec_hash', 'sha256'), dh_group=getattr(relation, 'ipsec_dh_group', 14) ) vpn_config.ipsec_config = ipsec_config elif relation.vpn_type == 'openvpn': openvpn_config = OpenVPNConfiguration.objects.create( server_address=getattr(relation, 'openvpn_server', ''), server_port=getattr(relation, 'openvpn_port', 1194), protocol=getattr(relation, 'openvpn_protocol', 'udp'), ca_certificate=getattr(relation, 'openvpn_ca_cert', ''), client_certificate=getattr(relation, 'openvpn_client_cert', ''), client_key=getattr(relation, 'openvpn_client_key', ''), compression=getattr(relation, 'openvpn_compression', False) ) vpn_config.openvpn_config = openvpn_config elif relation.vpn_type == 'wireguard': wireguard_config = WireGuardConfiguration.objects.create( private_key=getattr(relation, 'wireguard_private_key', ''), public_key=getattr(relation, 'wireguard_public_key', ''), endpoint=getattr(relation, 'wireguard_endpoint', ''), allowed_ips=getattr(relation, 'wireguard_allowed_ips', '0.0.0.0/0'), listen_port=getattr(relation, 'wireguard_port', 51820), persistent_keepalive=getattr(relation, 'wireguard_keepalive', 25) ) vpn_config.wireguard_config = wireguard_config vpn_config.save() # Create zone-vpn configuration link ZoneVPNConfiguration.objects.create( zone=relation.zone, vpn_configuration=vpn_config, is_active=True, priority=1 ) self.stdout.write( f' Created VPN config: {config_name}' ) return vpn_config def _create_ftp_config(self, relation, dry_run, force): """Create standalone FTP configuration from legacy data.""" config_name = f"{relation.channel.name}_{relation.zone.name}_FTP" # Check if config already exists existing_config = StandaloneFTPConfiguration.objects.filter( name=config_name ).first() if existing_config and not force: self.stdout.write( f' FTP config "{config_name}" already exists, skipping' ) return existing_config if dry_run: self.stdout.write( f' Would create FTP config: {config_name} ({relation.ftp_host})' ) return None # Create the standalone FTP configuration ftp_config = StandaloneFTPConfiguration.objects.create( name=config_name, description=f'Migrated from {relation.channel.name} - {relation.zone.name}', host=relation.ftp_host, port=getattr(relation, 'ftp_port', 21), username=getattr(relation, 'ftp_username', ''), password=getattr(relation, 'ftp_password', ''), root_directory=getattr(relation, 'ftp_root_directory', '/'), passive_mode=getattr(relation, 'ftp_passive_mode', True), is_active=True ) # Create zone-ftp configuration link ZoneFTPConfiguration.objects.create( zone=relation.zone, ftp_configuration=ftp_config, is_active=True, priority=1 ) self.stdout.write( f' Created FTP config: {config_name}' ) return ftp_config def _print_summary(self, stats, dry_run): """Print migration summary.""" mode = 'DRY RUN' if dry_run else 'COMPLETED' self.stdout.write('\n' + '='*50) self.stdout.write(f'MIGRATION {mode} SUMMARY') self.stdout.write('='*50) self.stdout.write( f'VPN Configurations Created: {stats["vpn_configs_created"]}' ) self.stdout.write( f'FTP Configurations Created: {stats["ftp_configs_created"]}' ) self.stdout.write( f'Relations Updated: {stats["relations_updated"]}' ) if stats['errors']: self.stdout.write( self.style.ERROR(f'Errors: {len(stats["errors"])}') ) for error in stats['errors']: self.stdout.write(f' - {error}') else: self.stdout.write( self.style.SUCCESS('No errors encountered') ) if not dry_run and (stats['vpn_configs_created'] > 0 or stats['ftp_configs_created'] > 0): self.stdout.write('\n' + self.style.SUCCESS( 'Migration completed successfully! ' 'You can now manage configurations independently in the admin interface.' )) elif dry_run: self.stdout.write('\n' + self.style.WARNING( 'This was a dry run. Use --force to actually perform the migration.' ))