AAP-79229 Custom roles with identical permissions to built-in roles cause HTTP 500/403 resource creation failures (#16639)

andle genuine custom matrices and arbitrary names when no legacy equivalent exists

awx/api/serializers.py

@@ -2964,6 +2964,8 @@ class ResourceAccessListElementSerializer(UserSerializer):
                     continue
                 new_roles_seen.add(new_role.id)
                 old_role = get_role_from_object_role(new_role)
+                if old_role is None:
+                    continue
                 all_permissive_role_ids.add(old_role.id)
 
                 if int(new_role.object_id) == obj.id and new_role.content_type_id == content_type.id:
@@ -2984,6 +2986,8 @@ class ResourceAccessListElementSerializer(UserSerializer):
                     continue
                 new_roles_seen.add(new_role.id)
                 old_role = get_role_from_object_role(new_role)
+                if old_role is None:
+                    continue
                 all_permissive_role_ids.add(old_role.id)
 
             # In DAB RBAC, superuser is strictly a user flag, and global roles are not in the RoleEvaluation table

awx/main/models/rbac.py

@@ -47,6 +47,9 @@ __all__ = [
 
 logger = logging.getLogger('awx.main.models.rbac')
 
+LEGACY_ROLE_NAME_PARTS = 2
+LEGACY_SCOPED_ROLE_NAME_PARTS = 3
+
 ROLE_SINGLETON_SYSTEM_ADMINISTRATOR = 'system_administrator'
 ROLE_SINGLETON_SYSTEM_AUDITOR = 'system_auditor'
 
@@ -586,39 +589,64 @@ def get_role_definition(role):
     return rd
 
 
+def _legacy_role_name_for_compat(rd, name_parts):
+    if len(name_parts) != LEGACY_SCOPED_ROLE_NAME_PARTS:
+        logger.debug(f'Role definition "{rd.name}" has no legacy role equivalent, skipping translation.')
+        return None
+    _model_name, role_name, _suffix = name_parts
+    return role_name.lower() + '_role'
+
+
+def _legacy_role_name_for_org_model_admin(rd, name_parts):
+    # cases like "Organization Project Admin"
+    target_model_name = name_parts[1]
+    try:
+        model_cls = apps.get_model('main', target_model_name)
+    except LookupError:
+        logger.debug(f'Role definition "{rd.name}" has no legacy role equivalent, skipping translation.')
+        return None
+    target_model_name = get_type_for_model(model_cls)
+
+    # exception cases completely specific to one model naming convention
+    if target_model_name == 'notification_template':
+        target_model_name = 'notification'
+    elif target_model_name == 'workflow_job_template':
+        target_model_name = 'workflow'
+
+    return f'{target_model_name}_admin_role'
+
+
+def _legacy_role_name_generic(rd, name_parts):
+    if len(name_parts) != LEGACY_ROLE_NAME_PARTS:
+        logger.debug(f'Role definition "{rd.name}" has no legacy role equivalent, skipping translation.')
+        return None
+    _model_name, role_name = name_parts
+    return role_name.lower() + '_role'
+
+
+def _legacy_role_name(rd, name_parts):
+    if rd.name.endswith(' Compat'):
+        return _legacy_role_name_for_compat(rd, name_parts)
+    if len(name_parts) == LEGACY_SCOPED_ROLE_NAME_PARTS and name_parts[0] == 'Organization' and name_parts[2] == 'Admin':
+        return _legacy_role_name_for_org_model_admin(rd, name_parts)
+    if len(name_parts) == LEGACY_ROLE_NAME_PARTS and name_parts[1] == 'Admin':
+        # cases like "Project Admin"
+        return 'admin_role'
+    if rd.name == 'Organization Audit':
+        return 'auditor_role'
+    return _legacy_role_name_generic(rd, name_parts)
+
+
 def get_role_from_object_role(object_role):
     """
     Given an object role from the new system, return the corresponding role from the old system
     reverses naming from get_role_definition, and the ANSIBLE_BASE_ROLE_PRECREATE setting.
     """
     rd = object_role.role_definition
-    if rd.name.endswith(' Compat'):
+    name_parts = rd.name.split()
-        model_name, role_name, _ = rd.name.split()
+    role_name = _legacy_role_name(rd, name_parts)
-        role_name = role_name.lower()
+    if role_name is None:
-        role_name += '_role'
+        return None
-    elif rd.name.endswith(' Admin') and rd.name.count(' ') == 2:
-        # cases like "Organization Project Admin"
-        model_name, target_model_name, role_name = rd.name.split()
-        role_name = role_name.lower()
-        model_cls = apps.get_model('main', target_model_name)
-        target_model_name = get_type_for_model(model_cls)
-
-        # exception cases completely specific to one model naming convention
-        if target_model_name == 'notification_template':
-            target_model_name = 'notification'
-        elif target_model_name == 'workflow_job_template':
-            target_model_name = 'workflow'
-
-        role_name = f'{target_model_name}_admin_role'
-    elif rd.name.endswith(' Admin'):
-        # cases like "project-admin"
-        role_name = 'admin_role'
-    elif rd.name == 'Organization Audit':
-        role_name = 'auditor_role'
-    else:
-        model_name, role_name = rd.name.split()
-        role_name = role_name.lower()
-        role_name += '_role'
     return getattr(object_role.content_object, role_name, None)
 
 

awx/main/tests/functional/dab_rbac/test_access_list.py

@@ -1,5 +1,7 @@
 import pytest
 
+from ansible_base.rbac.models import RoleDefinition
+
 from awx.main.models import User
 from awx.api.versioning import reverse
 
@@ -58,6 +60,29 @@ def test_access_list_direct_access(get, admin_user, inventory):
     assert sorted(access_entry['descendant_roles']) == sorted(['adhoc_role', 'use_role', 'update_role', 'read_role', 'admin_role'])
 
 
+@pytest.mark.django_db
+def test_access_list_custom_role_without_legacy_equivalent(get, admin_user, inventory, setup_managed_roles):
+    user = User.objects.create(username='custom-role-user')
+    managed_rd = RoleDefinition.objects.get(name='Inventory Admin')
+    custom_rd = RoleDefinition.objects.create_from_permissions(
+        name='Custom Inventory Administrator',
+        permissions=managed_rd.permissions.values_list('codename', flat=True),
+        content_type=managed_rd.content_type,
+    )
+    custom_rd.give_permission(user, inventory)
+
+    response = get(
+        reverse('api:inventory_access_list', kwargs={'pk': inventory.id}),
+        user=admin_user,
+        expect=200,
+    )
+    by_username = {item['username']: item for item in response.data['results']}
+
+    assert user.username in by_username
+    assert by_username[user.username]['summary_fields']['direct_access'] == []
+    assert by_username[user.username]['summary_fields']['indirect_access'] == []
+
+
 @pytest.mark.django_db
 def test_access_list_organization_access(get, admin_user, inventory):
     u2 = User.objects.create(username='u2')

awx/main/tests/functional/dab_rbac/test_translation_layer.py

@@ -42,6 +42,33 @@ def test_round_trip_roles(organization, rando, role_name, setup_managed_roles):
     assert old_role.id == getattr(organization, role_name).id
 
 
+@pytest.mark.django_db
+@pytest.mark.parametrize('custom_name', ('Test Custom 1', 'Test Custom Admin', 'This Is A Custom Admin', 'Organization Widget Admin'))
+def test_custom_role_with_unrecognized_name_has_no_legacy_role(organization, rando, setup_managed_roles, custom_name):
+    managed_rd = RoleDefinition.objects.get(name='Organization Project Admin')
+    custom_rd = RoleDefinition.objects.create_from_permissions(
+        name=custom_name,
+        permissions=managed_rd.permissions.values_list('codename', flat=True),
+        content_type=managed_rd.content_type,
+    )
+    assignment = custom_rd.give_permission(rando, organization)
+
+    assert get_role_from_object_role(assignment.object_role) is None
+
+
+@pytest.mark.django_db
+def test_custom_role_with_legacy_compatible_name_still_maps(project, rando, setup_managed_roles):
+    managed_rd = RoleDefinition.objects.get(name='Project Admin')
+    custom_rd = RoleDefinition.objects.create_from_permissions(
+        name='Custom Admin',
+        permissions=managed_rd.permissions.values_list('codename', flat=True),
+        content_type=managed_rd.content_type,
+    )
+    assignment = custom_rd.give_permission(rando, project)
+
+    assert get_role_from_object_role(assignment.object_role) == project.admin_role
+
+
 @pytest.mark.django_db
 def test_bulk_member_addition_records_one_entry_per_user(organization, rando, alice, setup_managed_roles):
     """

awx/main/tests/functional/test_projects.py

@@ -6,6 +6,8 @@ import pytest
 from awx.api.versioning import reverse
 from awx.main.models import Project, JobTemplate
 
+from ansible_base.rbac.models import RoleDefinition, RoleUserAssignment
+
 from django.core.exceptions import ValidationError
 
 
@@ -424,6 +426,34 @@ def test_create_project(post, organization, org_admin, org_member, admin, rando,
         assert 'do not have permission' in str(result.data['detail'])
 
 
+@pytest.mark.django_db
+def test_create_project_when_creator_role_has_no_legacy_equivalent(post, organization, org_admin, setup_managed_roles):
+    organization.project_admin_role.members.add(org_admin)
+    managed_rd = RoleDefinition.objects.get(name='Project Admin')
+    custom_rd = RoleDefinition.objects.create_from_permissions(
+        name='Test Custom 1',
+        permissions=managed_rd.permissions.values_list('codename', flat=True),
+        content_type=managed_rd.content_type,
+    )
+
+    # Keep the Controller compatibility boundary safe even if DAB returns a
+    # creator assignment that cannot be represented by a legacy role.
+    with mock.patch.object(
+        RoleDefinition.objects,
+        'give_creator_permissions',
+        side_effect=lambda user, obj: custom_rd.give_permission(user, obj),
+    ):
+        result = post(
+            reverse('api:project_list'),
+            {'name': 'Project with redundant custom role', 'organization': organization.id},
+            org_admin,
+            expect=201,
+        )
+
+    project = Project.objects.get(pk=result.data['id'])
+    assert RoleUserAssignment.objects.filter(user=org_admin, object_id=project.pk, role_definition=custom_rd).exists()
+
+
 @pytest.mark.django_db
 def test_project_credential_protection(post, put, project, organization, scm_credential, org_admin):
     project.save()