[RBAC] Rename managed role definitions, and move migration logic here (…

…#15087)

* Rename managed role definitions, and move migration logic here

* Fix naming capitalization

awx/main/migrations/0192_custom_roles.py

@@ -2,9 +2,7 @@
 
 from django.db import migrations
 
-from awx.main.migrations._dab_rbac import migrate_to_new_rbac, create_permissions_as_operation
-
-from ansible_base.rbac.migrations._managed_definitions import setup_managed_role_definitions
+from awx.main.migrations._dab_rbac import migrate_to_new_rbac, create_permissions_as_operation, setup_managed_role_definitions
 
 
 class Migration(migrations.Migration):

awx/main/migrations/_dab_rbac.py

@@ -3,12 +3,15 @@
 
 from django.apps import apps as global_apps
 from django.db.models import ForeignKey
+from django.conf import settings
 from ansible_base.rbac.migrations._utils import give_permissions
 from ansible_base.rbac.management import create_dab_permissions
 
 from awx.main.fields import ImplicitRoleField
 from awx.main.constants import role_name_to_perm_mapping
 
+from ansible_base.rbac.permission_registry import permission_registry
+
 
 logger = logging.getLogger('awx.main.migrations._dab_rbac')
 
@@ -194,7 +197,7 @@ def migrate_to_new_rbac(apps, schema_editor):
             role_definition = managed_definitions[permissions]
         else:
             action = role.role_field.rsplit('_', 1)[0]  # remove the _field ending of the name
-            role_definition_name = f'{role.content_type.model}-{action}'
+            role_definition_name = f'{role.content_type.model_class().__name__} {action.title()}'
 
             description = role_descriptions[role.role_field]
             if type(description) == dict:
@@ -241,3 +244,100 @@ def migrate_to_new_rbac(apps, schema_editor):
             ct += 1
         if ct:
             logger.info(f'Migrated {ct} users to new system auditor flag')
+
+
+def get_or_create_managed(name, description, ct, permissions, RoleDefinition):
+    role_definition, created = RoleDefinition.objects.get_or_create(name=name, defaults={'managed': True, 'description': description, 'content_type': ct})
+    role_definition.permissions.set(list(permissions))
+
+    if not role_definition.managed:
+        role_definition.managed = True
+        role_definition.save(update_fields=['managed'])
+
+    if created:
+        logger.info(f'Created RoleDefinition {role_definition.name} pk={role_definition} with {len(permissions)} permissions')
+
+    return role_definition
+
+
+def setup_managed_role_definitions(apps, schema_editor):
+    """
+    Idepotent method to create or sync the managed role definitions
+    """
+    to_create = settings.ANSIBLE_BASE_ROLE_PRECREATE
+
+    ContentType = apps.get_model('contenttypes', 'ContentType')
+    Permission = apps.get_model('dab_rbac', 'DABPermission')
+    RoleDefinition = apps.get_model('dab_rbac', 'RoleDefinition')
+    Organization = apps.get_model(settings.ANSIBLE_BASE_ORGANIZATION_MODEL)
+    org_ct = ContentType.objects.get_for_model(Organization)
+    managed_role_definitions = []
+
+    org_perms = set()
+    for cls in permission_registry._registry:
+        ct = ContentType.objects.get_for_model(cls)
+        object_perms = set(Permission.objects.filter(content_type=ct))
+        # Special case for InstanceGroup which has an organiation field, but is not an organization child object
+        if cls._meta.model_name != 'instancegroup':
+            org_perms.update(object_perms)
+
+        if 'object_admin' in to_create and cls != Organization:
+            indiv_perms = object_perms.copy()
+            add_perms = [perm for perm in indiv_perms if perm.codename.startswith('add_')]
+            if add_perms:
+                for perm in add_perms:
+                    indiv_perms.remove(perm)
+
+            managed_role_definitions.append(
+                get_or_create_managed(
+                    to_create['object_admin'].format(cls=cls), f'Has all permissions to a single {cls._meta.verbose_name}', ct, indiv_perms, RoleDefinition
+                )
+            )
+
+        if 'org_children' in to_create and cls != Organization:
+            org_child_perms = object_perms.copy()
+            org_child_perms.add(Permission.objects.get(codename='view_organization'))
+
+            managed_role_definitions.append(
+                get_or_create_managed(
+                    to_create['org_children'].format(cls=cls),
+                    f'Has all permissions to {cls._meta.verbose_name_plural} within an organization',
+                    org_ct,
+                    org_child_perms,
+                    RoleDefinition,
+                )
+            )
+
+        if 'special' in to_create:
+            special_perms = []
+            for perm in object_perms:
+                if perm.codename.split('_')[0] not in ('add', 'change', 'update', 'delete', 'view'):
+                    special_perms.append(perm)
+            for perm in special_perms:
+                action = perm.codename.split('_')[0]
+                view_perm = Permission.objects.get(content_type=ct, codename__startswith='view_')
+                managed_role_definitions.append(
+                    get_or_create_managed(
+                        to_create['special'].format(cls=cls, action=action.title()),
+                        f'Has {action} permissions to a single {cls._meta.verbose_name}',
+                        ct,
+                        [perm, view_perm],
+                        RoleDefinition,
+                    )
+                )
+
+    if 'org_admin' in to_create:
+        managed_role_definitions.append(
+            get_or_create_managed(
+                to_create['org_admin'].format(cls=Organization),
+                'Has all permissions to a single organization and all objects inside of it',
+                org_ct,
+                org_perms,
+                RoleDefinition,
+            )
+        )
+
+    unexpected_role_definitions = RoleDefinition.objects.filter(managed=True).exclude(pk__in=[rd.pk for rd in managed_role_definitions])
+    for role_definition in unexpected_role_definitions:
+        logger.info(f'Deleting old managed role definition {role_definition.name}, pk={role_definition.pk}')
+        role_definition.delete()

awx/main/models/rbac.py

@@ -553,7 +553,7 @@ def get_role_definition(role):
         return
     f = obj._meta.get_field(role.role_field)
     action_name = f.name.rsplit("_", 1)[0]
-    rd_name = f'{obj._meta.model_name}-{action_name}-compat'
+    rd_name = f'{type(obj).__name__} {action_name.title()} Compat'
     perm_list = get_role_codenames(role)
     defaults = {'content_type_id': role.content_type_id}
     try:
@@ -573,23 +573,26 @@ def get_role_from_object_role(object_role):
     reverses naming from get_role_definition, and the ANSIBLE_BASE_ROLE_PRECREATE setting.
     """
     rd = object_role.role_definition
-    if rd.name.endswith('-compat'):
-        model_name, role_name, _ = rd.name.split('-')
+    if rd.name.endswith(' Compat'):
+        model_name, role_name, _ = rd.name.split()
+        role_name = role_name.lower()
         role_name += '_role'
-    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('-')
+    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)
         if target_model_name == 'notification_template':
             target_model_name = 'notification'  # total exception
         role_name = f'{target_model_name}_admin_role'
-    elif rd.name.endswith('-admin'):
+    elif rd.name.endswith(' Admin'):
         # cases like "project-admin"
-        model_name, _ = rd.name.rsplit('-', 1)
         role_name = 'admin_role'
     else:
-        model_name, role_name = rd.name.split('-')
+        print(rd.name)
+        model_name, role_name = rd.name.split()
+        role_name = role_name.lower()
         role_name += '_role'
     return getattr(object_role.content_object, role_name)
 

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

@@ -0,0 +1,10 @@
+import pytest
+from django.apps import apps
+
+from awx.main.migrations._dab_rbac import setup_managed_role_definitions
+
+
+@pytest.fixture
+def managed_roles():
+    "Run the migration script to pre-create managed role definitions"
+    setup_managed_role_definitions(apps, None)

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

@@ -0,0 +1,45 @@
+import pytest
+from django.apps import apps
+from django.test.utils import override_settings
+
+from awx.main.migrations._dab_rbac import setup_managed_role_definitions
+
+from ansible_base.rbac.models import RoleDefinition
+
+INVENTORY_OBJ_PERMISSIONS = ['view_inventory', 'adhoc_inventory', 'use_inventory', 'change_inventory', 'delete_inventory', 'update_inventory']
+
+
+@pytest.mark.django_db
+def test_managed_definitions_precreate():
+    with override_settings(
+        ANSIBLE_BASE_ROLE_PRECREATE={
+            'object_admin': '{cls._meta.model_name}-admin',
+            'org_admin': 'organization-admin',
+            'org_children': 'organization-{cls._meta.model_name}-admin',
+            'special': '{cls._meta.model_name}-{action}',
+        }
+    ):
+        setup_managed_role_definitions(apps, None)
+    rd = RoleDefinition.objects.get(name='inventory-admin')
+    assert rd.managed is True
+    # add permissions do not go in the object-level admin
+    assert set(rd.permissions.values_list('codename', flat=True)) == set(INVENTORY_OBJ_PERMISSIONS)
+
+    # test org-level object admin permissions
+    rd = RoleDefinition.objects.get(name='organization-inventory-admin')
+    assert rd.managed is True
+    assert set(rd.permissions.values_list('codename', flat=True)) == set(['add_inventory', 'view_organization'] + INVENTORY_OBJ_PERMISSIONS)
+
+
+@pytest.mark.django_db
+def test_managed_definitions_custom_obj_admin_name():
+    with override_settings(
+        ANSIBLE_BASE_ROLE_PRECREATE={
+            'object_admin': 'foo-{cls._meta.model_name}-foo',
+        }
+    ):
+        setup_managed_role_definitions(apps, None)
+    rd = RoleDefinition.objects.get(name='foo-inventory-foo')
+    assert rd.managed is True
+    # add permissions do not go in the object-level admin
+    assert set(rd.permissions.values_list('codename', flat=True)) == set(INVENTORY_OBJ_PERMISSIONS)

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

@@ -10,19 +10,19 @@
 
 
 @pytest.mark.django_db
-def test_managed_roles_created():
+def test_managed_roles_created(managed_roles):
     "Managed RoleDefinitions are created in post_migration signal, we expect to see them here"
     for cls in (JobTemplate, Inventory):
         ct = ContentType.objects.get_for_model(cls)
         rds = list(RoleDefinition.objects.filter(content_type=ct))
         assert len(rds) > 1
-        assert f'{cls._meta.model_name}-admin' in [rd.name for rd in rds]
+        assert f'{cls.__name__} Admin' in [rd.name for rd in rds]
         for rd in rds:
             assert rd.managed is True
 
 
 @pytest.mark.django_db
-def test_custom_read_role(admin_user, post):
+def test_custom_read_role(admin_user, post, managed_roles):
     rd_url = django_reverse('roledefinition-list')
     resp = post(
         url=rd_url, data={"name": "read role made for test", "content_type": "awx.inventory", "permissions": ['view_inventory']}, user=admin_user, expect=201
@@ -40,8 +40,8 @@ def test_custom_system_roles_prohibited(admin_user, post):
 
 
 @pytest.mark.django_db
-def test_assign_managed_role(admin_user, alice, rando, inventory, post):
-    rd = RoleDefinition.objects.get(name='inventory-admin')
+def test_assign_managed_role(admin_user, alice, rando, inventory, post, managed_roles):
+    rd = RoleDefinition.objects.get(name='Inventory Admin')
     rd.give_permission(alice, inventory)
     # Now that alice has full permissions to the inventory, she will give rando permission
     url = django_reverse('roleuserassignment-list')
@@ -63,7 +63,7 @@ def test_assign_custom_delete_role(admin_user, rando, inventory, delete, patch):
 
 
 @pytest.mark.django_db
-def test_assign_custom_add_role(admin_user, rando, organization, post):
+def test_assign_custom_add_role(admin_user, rando, organization, post, managed_roles):
     rd, _ = RoleDefinition.objects.get_or_create(
         name='inventory-add', permissions=['add_inventory', 'view_organization'], content_type=ContentType.objects.get_for_model(Organization)
     )

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

@@ -14,7 +14,7 @@
     'role_name',
     ['execution_environment_admin_role', 'project_admin_role', 'admin_role', 'auditor_role', 'read_role', 'execute_role', 'notification_admin_role'],
 )
-def test_round_trip_roles(organization, rando, role_name):
+def test_round_trip_roles(organization, rando, role_name, managed_roles):
     """
     Make an assignment with the old-style role,
     get the equivelent new role
@@ -28,7 +28,7 @@ def test_round_trip_roles(organization, rando, role_name):
 
 
 @pytest.mark.django_db
-def test_organization_level_permissions(organization, inventory):
+def test_organization_level_permissions(organization, inventory, managed_roles):
     u1 = User.objects.create(username='alice')
     u2 = User.objects.create(username='bob')
 
@@ -58,14 +58,14 @@ def test_organization_level_permissions(organization, inventory):
 
 
 @pytest.mark.django_db
-def test_organization_execute_role(organization, rando):
+def test_organization_execute_role(organization, rando, managed_roles):
     organization.execute_role.members.add(rando)
     assert rando in organization.execute_role
     assert set(Organization.accessible_objects(rando, 'execute_role')) == set([organization])
 
 
 @pytest.mark.django_db
-def test_workflow_approval_list(get, post, admin_user):
+def test_workflow_approval_list(get, post, admin_user, managed_roles):
     workflow_job_template = WorkflowJobTemplate.objects.create()
     approval_node = WorkflowJobTemplateNode.objects.create(workflow_job_template=workflow_job_template)
     url = reverse('api:workflow_job_template_node_create_approval', kwargs={'pk': approval_node.pk, 'version': 'v2'})
@@ -79,14 +79,14 @@ def test_workflow_approval_list(get, post, admin_user):
 
 
 @pytest.mark.django_db
-def test_creator_permission(rando, admin_user, inventory):
+def test_creator_permission(rando, admin_user, inventory, managed_roles):
     give_creator_permissions(rando, inventory)
     assert rando in inventory.admin_role
     assert rando in inventory.admin_role.members.all()
 
 
 @pytest.mark.django_db
-def test_team_team_read_role(rando, team, admin_user, post):
+def test_team_team_read_role(rando, team, admin_user, post, managed_roles):
     orgs = [Organization.objects.create(name=f'foo-{i}') for i in range(2)]
     teams = [Team.objects.create(name=f'foo-{i}', organization=orgs[i]) for i in range(2)]
     teams[1].member_role.members.add(rando)

awx/settings/defaults.py

@@ -1144,14 +1144,17 @@
 
 # Settings for the ansible_base RBAC system
 
-# Settings for the RBAC system, override as necessary in app
+# Only used internally, names of the managed RoleDefinitions to create
 ANSIBLE_BASE_ROLE_PRECREATE = {
-    'object_admin': '{cls._meta.model_name}-admin',
-    'org_admin': 'organization-admin',
-    'org_children': 'organization-{cls._meta.model_name}-admin',
-    'special': '{cls._meta.model_name}-{action}',
+    'object_admin': '{cls.__name__} Admin',
+    'org_admin': 'Organization Admin',
+    'org_children': 'Organization {cls.__name__} Admin',
+    'special': '{cls.__name__} {action}',
 }
 
+# Name for auto-created roles that give users permissions to what they create
+ANSIBLE_BASE_ROLE_CREATOR_NAME = '{cls.__name__} Creator'
+
 # Use the new Gateway RBAC system for evaluations? You should. We will remove the old system soon.
 ANSIBLE_BASE_ROLE_SYSTEM_ACTIVATED = True