awx/awx/main/migrations/0202_convert_controller_role_definitions.py

# Generated by Django migration for converting Controller role definitions

from ansible_base.rbac.migrations._utils import give_permissions
from django.db import migrations


def convert_controller_role_definitions(apps, schema_editor):
    """
    Convert Controller role definitions to regular role definitions:
    - Controller Organization Admin -> Organization Admin
    - Controller Organization Member -> Organization Member
    - Controller Team Admin -> Team Admin
    - Controller Team Member -> Team Member
    - Controller System Auditor -> Platform Auditor

    Then delete the old Controller role definitions.
    """
    RoleDefinition = apps.get_model('dab_rbac', 'RoleDefinition')
    RoleUserAssignment = apps.get_model('dab_rbac', 'RoleUserAssignment')
    RoleTeamAssignment = apps.get_model('dab_rbac', 'RoleTeamAssignment')
    Permission = apps.get_model('dab_rbac', 'DABPermission')

    # Mapping of old Controller role names to new role names
    role_mappings = {
        'Controller Organization Admin': 'Organization Admin',
        'Controller Organization Member': 'Organization Member',
        'Controller Team Admin': 'Team Admin',
        'Controller Team Member': 'Team Member',
    }

    for old_name, new_name in role_mappings.items():
        # Find the old Controller role definition
        old_role = RoleDefinition.objects.filter(name=old_name).first()
        if not old_role:
            continue  # Skip if the old role doesn't exist

        # Find the new role definition
        new_role = RoleDefinition.objects.get(name=new_name)

        # Collect all the assignments that need to be migrated
        # Group by object (content_type + object_id) to batch the give_permissions calls
        assignments_by_object = {}

        # Get user assignments
        user_assignments = RoleUserAssignment.objects.filter(role_definition=old_role).select_related('object_role')
        for assignment in user_assignments:
            key = (assignment.object_role.content_type_id, assignment.object_role.object_id)
            if key not in assignments_by_object:
                assignments_by_object[key] = {'users': [], 'teams': []}
            assignments_by_object[key]['users'].append(assignment.user)

        # Get team assignments
        team_assignments = RoleTeamAssignment.objects.filter(role_definition=old_role).select_related('object_role')
        for assignment in team_assignments:
            key = (assignment.object_role.content_type_id, assignment.object_role.object_id)
            if key not in assignments_by_object:
                assignments_by_object[key] = {'users': [], 'teams': []}
            assignments_by_object[key]['teams'].append(assignment.team.id)

        # Use give_permissions to create new assignments with the new role definition
        for (content_type_id, object_id), data in assignments_by_object.items():
            if data['users'] or data['teams']:
                give_permissions(
                    apps,
                    new_role,
                    users=data['users'],
                    teams=data['teams'],
                    object_id=object_id,
                    content_type_id=content_type_id,
                )

        # Delete the old role definition (this will cascade to delete old assignments and ObjectRoles)
        old_role.delete()

    # Create or get Platform Auditor
    auditor_rd, created = RoleDefinition.objects.get_or_create(
        name='Platform Auditor',
        defaults={'description': 'Migrated singleton role giving read permission to everything', 'managed': True},
    )
    if created:
        auditor_rd.permissions.add(*list(Permission.objects.filter(codename__startswith='view')))

    old_rd = RoleDefinition.objects.filter(name='Controller System Auditor').first()
    if old_rd:
        for assignment in RoleUserAssignment.objects.filter(role_definition=old_rd):
            RoleUserAssignment.objects.create(
                user=assignment.user,
                role_definition=auditor_rd,
            )

    # Delete the Controller System Auditor role
    RoleDefinition.objects.filter(name='Controller System Auditor').delete()


class Migration(migrations.Migration):
    dependencies = [
        ('main', '0201_create_managed_creds'),
    ]

    operations = [
        migrations.RunPython(convert_controller_role_definitions, migrations.RunPython.noop),
    ]