Allow refresh of managed roles, use api_slugs

Author: AlanCoding

ansible_base/rbac/managed.py

@@ -46,6 +46,51 @@ def get_content_type(self, apps):
             content_type_cls = apps.get_model('contenttypes', 'ContentType')
         return content_type_cls.objects.get_for_model(model)
 
+    def refresh_permissions(self, rd, apps):
+        """Make role permissions equal what the managed definition specifies"""
+        permission_cls = apps.get_model('dab_rbac', 'DABPermission')
+
+        # Desired permission codenames from managed source
+        desired_codenames = set(self.get_permissions(apps))
+
+        # Resolve to actual permission objects or error
+        desired_permissions = []
+        for codename in desired_codenames:
+            try:
+                if '.' in codename:
+                    perm = permission_cls.objects.get(api_slug=codename)
+                else:
+                    perm = permission_cls.objects.get(codename=codename)
+                desired_permissions.append(perm)
+            except permission_cls.DoesNotExist:
+                db_codenames = list(permission_cls.objects.values_list('codename', flat=True))
+                raise permission_cls.DoesNotExist(
+                    f'Permission codename "{codename}" does not exist.\n'
+                    f'Managed role: {self}\nExpected: {sorted(desired_codenames)}\n'
+                    f'Available in DB: {sorted(db_codenames)}'
+                )
+
+        desired_set = set(desired_permissions)
+        current_set = set(rd.permissions.all())
+
+        to_add = desired_set - current_set
+        to_remove = current_set - desired_set
+
+        if not to_add and not to_remove:
+            logger.info(f'No permission changes needed for role "{self.name}"')
+        else:
+            if to_add:
+                rd.permissions.add(*to_add)
+                added_codenames = sorted(p.codename for p in to_add)
+                logger.info(f'Added permissions to role "{self.name}": {added_codenames}')
+
+            if to_remove:
+                rd.permissions.remove(*to_remove)
+                removed_codenames = sorted(p.codename for p in to_remove)
+                logger.info(f'Removed permissions from role "{self.name}": {removed_codenames}')
+
+        logger.debug(f'Final permissions for role "{self.name}": ' f'{sorted(p.codename for p in rd.permissions.all())}')
+
     def get_or_create(self, apps):
         "Create from a list of text-type permissions and do validation"
         role_definition_cls = apps.get_model('dab_rbac', 'RoleDefinition')
@@ -57,36 +102,33 @@ def get_or_create(self, apps):
         rd, created = role_definition_cls.objects.get_or_create(name=self.name, defaults=defaults)
 
         if created:
-            permissions = self.get_permissions(apps)
-            permission_cls = apps.get_model('dab_rbac', 'DABPermission')
-            perm_list = []
-            for str_perm in permissions:
-                try:
-                    perm_list.append(permission_cls.objects.get(codename=str_perm))
-                except permission_cls.DoesNotExist:
-                    # Better error handling for debugging
-                    db_codenames = list(permission_cls.objects.values_list('codename', flat=True))
-                    raise permission_cls.DoesNotExist(
-                        f'Permission codename {str_perm} does not exist. Manged role {self}\n expected: {permissions}\n Database permissions: {db_codenames}'
-                    )
-            rd.permissions.add(*perm_list)
-            logger.info(f'Created {self.shortname} managed role definition, name={self.name}')
+            self.refresh_permissions(rd, apps=apps)
             logger.debug(f'Data of {self.name} role definition: {defaults}')
-            logger.debug(f'Permissions of {self.name} role definition: {permissions}')
         return rd, created
 
-    def allowed_permissions(self, model: Optional[Type[Model]]) -> set[str]:
-        from ansible_base.rbac.validators import combine_values, permissions_allowed_for_role
+    def allowed_permissions_by_model(self, model: Optional[Type[Model]]) -> dict[Type, list[str]]:
+        from ansible_base.rbac.validators import permissions_allowed_for_role
+
+        return permissions_allowed_for_role(model)
+
+    def allowed_permissions_slug_list(self, model: Optional[Type[Model]]) -> set[str]:
+        "Returns all possible permissions for model in terms format of awx.change_inventory"
+        from ansible_base.rbac.remote import get_resource_prefix
 
-        return combine_values(permissions_allowed_for_role(model))
+        slug_list = set()
+        for child_model, child_codenames in self.allowed_permissions_by_model(model).items():
+            prefix = get_resource_prefix(child_model)
+            for codename in child_codenames:
+                slug_list.add(f'{prefix}.{codename}')
+        return slug_list
 
 
 class ManagedAdminBase(ManagedRoleConstructor):
     description = gettext_noop("Has all permissions to a single {model_name_verbose}")
 
     def get_permissions(self, apps) -> set[str]:
         """All permissions possible for the associated model"""
-        return self.allowed_permissions(self.get_model(apps))
+        return self.allowed_permissions_slug_list(self.get_model(apps))
 
 
 class ManagedActionBase(ManagedRoleConstructor):
@@ -108,7 +150,7 @@ class ManagedReadOnlyBase(ManagedRoleConstructor):
     description = gettext_noop("Has all viewing related permissions that can be delegated via {model_name_verbose}")
 
     def get_permissions(self, apps) -> set[str]:
-        return {codename for codename in self.allowed_permissions(self.get_model(apps)) if codename.startswith('view')}
+        return {api_slug for api_slug in self.allowed_permissions_slug_list(self.get_model(apps)) if '.view' in api_slug}
 
 
 class OrganizationMixin:

ansible_base/rbac/models/content_type.py

@@ -174,7 +174,7 @@ def get_for_id(self, id: int) -> django_models.Model:
             self._add_to_cache(self.db, ct)
             return ct
 
-    def load_remote_types(self, remote_data: list[dict]):
+    def load_remote_objects(self, remote_data: list[dict]):
         parent_mapping: dict[django_models.Model, str] = {}
         for remote_type in remote_data:
             service = remote_type.pop('service')

ansible_base/rbac/models/permission.py

@@ -1,17 +1,23 @@
+from django.apps import apps
 from django.db import models
 from django.utils.translation import gettext_lazy as _
 
 from .content_type import DABContentType
 
 
 class DABPermissionManager(models.Manager):
-    def load_remote_types(self, remote_data: list[dict]):
+    def load_remote_objects(self, remote_data: list[dict], update_managed=False):
         for remote_type in remote_data:
             codename = remote_type.pop('codename')
             ct_slug = remote_type.pop('content_type')
             ct = DABContentType.objects.get(api_slug=ct_slug)
             ct, _ = self.get_or_create(codename=codename, content_type=ct, defaults=remote_type)
 
+        if update_managed:
+            from ansible_base.rbac import permission_registry
+
+            permission_registry.create_managed_roles(apps, update_perms=True)
+
 
 class DABPermission(models.Model):
     "This is a minimal copy of auth.Permission for internal use"

ansible_base/rbac/permission_registry.py

@@ -108,7 +108,7 @@ def register_managed_role_constructors(self) -> None:
         for shortname, constructor in managed_defs.items():
             self.register_managed_role_constructor(shortname, constructor)
 
-    def create_managed_roles(self, apps) -> list[tuple[Model, bool]]:
+    def create_managed_roles(self, apps, update_perms=False) -> list[tuple[Model, bool]]:
         """Safe-ish method to create managed roles inside of a migration
 
         Returns a list with all the managed RoleDefinition objects and whether they were created
@@ -118,6 +118,8 @@ def create_managed_roles(self, apps) -> list[tuple[Model, bool]]:
         ret = []
         for managed_role in self._managed_roles.values():
             rd, created = managed_role.get_or_create(apps)
+            if update_perms and (not created):
+                managed_role.refresh_permissions(rd, apps)
             ret.append((rd, created))
         return ret
 

ansible_base/rbac/remote.py

@@ -163,7 +163,7 @@ def get_resource_prefix(model: Union[Type[models.Model], models.Model, Type[Remo
     """
     if isinstance(model, RemoteObject) or (inspect.isclass(model) and issubclass(model, RemoteObject)):
         # If it is a remote object, it was only ever created from this to begin with
-        return model._meta.model_name
+        return model._meta.service
 
     if registry := get_resource_registry():
         # duplicates logic in ansible_base/resource_registry/apps.py

test_app/tests/rbac/remote/test_managed_role_remote_perms.py

@@ -0,0 +1,83 @@
+import pytest
+from django.apps import apps
+from django.test import override_settings
+
+from ansible_base.rbac import permission_registry
+from ansible_base.rbac.models import DABContentType, DABPermission, RoleDefinition
+from ansible_base.rbac.remote import get_resource_prefix
+
+
+@pytest.mark.django_db
+@override_settings(ANSIBLE_BASE_MANAGED_ROLE_REGISTRY={'org_admin': {}})
+def test_remote_permissions_on_create(foo_permission):
+    rd = RoleDefinition.objects.managed.org_admin
+    assert foo_permission in rd.permissions.all()
+
+
+@pytest.mark.django_db
+def test_remote_permission_refresh(foo_type):
+    rd = RoleDefinition.objects.managed.org_admin
+
+    foo_permission = DABPermission.objects.create(codename='foo_foo', content_type=foo_type)
+
+    # the foo_permission was created after the managed role
+    # so the role is not explicitly created
+    assert foo_permission not in rd.permissions.all()
+
+    # Loading it requires the refresh call
+    permission_registry.create_managed_roles(apps, update_perms=True)
+    assert foo_permission in rd.permissions.all()
+
+
+@pytest.mark.django_db
+def test_remote_permission_load_update_roles():
+    rd = RoleDefinition.objects.managed.org_admin
+    DABContentType.objects.load_remote_objects(
+        [
+            {'service': 'fooland', 'app_label': 'foo', 'model': 'foo', 'api_slug': 'fooland.foo', 'parent_content_type': 'shared.organization'},
+            {'service': 'fooland', 'app_label': 'foo', 'model': 'bar', 'api_slug': 'fooland.bar', 'parent_content_type': None},
+        ]
+    )
+    assert not DABPermission.objects.filter(codename='foo_foo').exists()
+    DABPermission.objects.load_remote_objects(
+        [
+            {'codename': 'foo_foo', 'content_type': 'fooland.foo', 'api_slug': "fooland.foo_foo"},
+            {'codename': 'bar_bar', 'content_type': 'fooland.bar', 'api_slug': "fooland.bar_bar"},
+        ],
+        update_managed=True,
+    )
+    assert DABPermission.objects.filter(codename='foo_foo').exists()
+
+    foo_permission = DABPermission.objects.get(codename='foo_foo')
+    bar_permission = DABPermission.objects.get(codename='bar_bar')
+    assert foo_permission in rd.permissions.all()
+    assert bar_permission not in rd.permissions.all()  # not child type of organization
+
+
+@pytest.mark.django_db
+def test_remote_permission_duplicate_name():
+    rd = RoleDefinition.objects.managed.org_admin
+    DABContentType.objects.load_remote_objects(
+        [
+            {'service': 'foo', 'app_label': 'core', 'model': 'thing', 'api_slug': 'foo.thing', 'parent_content_type': 'shared.organization'},
+            {'service': 'bar', 'app_label': 'core', 'model': 'thing', 'api_slug': 'bar.thing', 'parent_content_type': 'shared.organization'},
+        ]
+    )
+    ct = DABContentType.objects.get(api_slug='foo.thing')
+    assert get_resource_prefix(ct.model_class()) == 'foo'
+
+    assert not DABPermission.objects.filter(codename='view_thing').exists()
+    DABPermission.objects.load_remote_objects(
+        [
+            {'codename': 'view_thing', 'content_type': 'foo.thing', 'api_slug': "foo.view_thing"},
+            {'codename': 'view_thing', 'content_type': 'bar.thing', 'api_slug': "bar.view_thing"},
+        ],
+        update_managed=True,
+    )
+    assert DABPermission.objects.filter(codename='view_thing').count() == 2
+
+    foo_permission = DABPermission.objects.get(api_slug='foo.view_thing')
+    bar_permission = DABPermission.objects.get(api_slug='bar.view_thing')
+    # Both are valid permissions for the org_admin role
+    assert foo_permission in rd.permissions.all()
+    assert bar_permission in rd.permissions.all()

test_app/tests/rbac/remote/test_service_api.py

@@ -68,7 +68,7 @@ def test_reload_types(admin_api_client):
 
     DABContentType.objects.all().delete()  # Delete all types, see if we get them back
 
-    DABContentType.objects.load_remote_types(type_list)
+    DABContentType.objects.load_remote_objects(type_list)
 
     response = admin_api_client.get(url + '?page_size=200', format="json")
     assert response.status_code == 200, response.data
@@ -78,7 +78,7 @@ def test_reload_types(admin_api_client):
 
 @pytest.mark.django_db
 def test_load_child_of_org():
-    DABContentType.objects.load_remote_types([{'service': 'fooland', 'app_label': 'foop', 'model': 'fooser', 'parent_content_type': 'shared.organization'}])
+    DABContentType.objects.load_remote_objects([{'service': 'fooland', 'app_label': 'foop', 'model': 'fooser', 'parent_content_type': 'shared.organization'}])
     ct = DABContentType.objects.get(api_slug='fooland.fooser')
     assert ct.parent_content_type.app_label == 'test_app'  # proves connection to existing
 
@@ -94,7 +94,7 @@ def test_reload_permissions(admin_api_client):
 
     DABPermission.objects.all().delete()  # Delete all permissions, see if we get them back
 
-    DABPermission.objects.load_remote_types(perm_list)
+    DABPermission.objects.load_remote_objects(perm_list)
 
     response = admin_api_client.get(url + '?page_size=200', format="json")
     assert response.status_code == 200, response.data