First pass at drill-down endpoint

Author: AlanCoding

ansible_base/rbac/api/queries.py

@@ -0,0 +1,32 @@
+from django.db.models import Model
+
+from ..models import DABContentType, get_evaluation_model
+
+
+def assignment_qs_user_to_obj(actor: Model, obj: Model):
+    """Querset of assignments (team or user) that grants the actor any form of permission to obj"""
+    evaluation_cls = get_evaluation_model(obj)
+    ct = DABContentType.objects.get_for_model(obj)
+    reverse_name = evaluation_cls._meta.get_field('role').remote_field.name
+
+    # All relevant assignments for the object
+    obj_eval_qs = evaluation_cls.objects.filter(object_id=obj.pk, content_type_id=ct.id)
+    obj_assignment_qs = actor.role_assignments.filter(**{f'object_role__{reverse_name}__in': obj_eval_qs})
+
+    global_assignment_qs = actor.role_assignments.filter(content_type=None, role_definition__permissions__content_type=ct)
+
+    return (global_assignment_qs | obj_assignment_qs).distinct()
+
+
+def assignment_qs_user_to_obj_perm(actor: Model, obj: Model, permission: Model):
+    """Queryset of assignments that grants this specific permission to this specific object"""
+    evaluation_cls = get_evaluation_model(obj)
+    ct = DABContentType.objects.get_for_model(obj)
+    reverse_name = evaluation_cls._meta.get_field('role').remote_field.name
+
+    obj_eval_qs = evaluation_cls.objects.filter(codename=permission.codename, object_id=obj.pk, content_type_id=ct.id)
+    obj_assignment_qs = actor.role_assignments.filter(**{f'object_role__{reverse_name}__in': obj_eval_qs})
+
+    global_assignment_qs = actor.role_assignments.filter(content_type=None, role_definition__permissions=permission)
+
+    return (global_assignment_qs | obj_assignment_qs).distinct()

ansible_base/rbac/api/serializers.py

@@ -10,13 +10,15 @@
 from ansible_base.lib.abstract_models.common import get_url_for_object
 from ansible_base.lib.serializers.common import AbstractCommonModelSerializer, CommonModelSerializer, ImmutableCommonModelSerializer
 from ansible_base.lib.utils.auth import get_team_model
+from ansible_base.lib.utils.response import get_relative_url
 from ansible_base.rbac.models import RoleDefinition, RoleTeamAssignment, RoleUserAssignment
 from ansible_base.rbac.permission_registry import permission_registry  # careful for circular imports
 from ansible_base.rbac.policies import check_content_obj_permission, visible_users
 from ansible_base.rbac.validators import check_locally_managed, validate_permissions_for_model
 
-from ..models import DABContentType, DABPermission, get_evaluation_model
+from ..models import DABContentType, DABPermission
 from ..remote import RemoteObject
+from .queries import assignment_qs_user_to_obj, assignment_qs_user_to_obj_perm
 
 
 class RoleDefinitionSerializer(CommonModelSerializer):
@@ -249,6 +251,17 @@ class RoleMetadataSerializer(serializers.Serializer):
 
 class AccessListMixin:
 
+    def _get_related(self, obj) -> dict[str, str]:
+        if obj is None:
+            return {}
+        related_fields = {}
+        actor_cls = self.Meta.model
+        related_fields['details'] = get_relative_url(
+            f'role-{actor_cls._meta.model_name}-access-assignments',
+            kwargs={'model_name': self.context.get("content_type").api_slug, 'pk': self.context.get("related_object").pk, 'actor_pk': obj.pk},
+        )
+        return related_fields
+
     @staticmethod
     def summarize_role_definition(role_definition):
         return {"name": role_definition.name, "url": get_url_for_object(role_definition)}
@@ -258,36 +271,25 @@ def get_object_role_assignments(self, actor):
         permission = self.context.get("permission")
         ct = self.context.get("content_type")
 
-        assignment_list = []
-
-        evaluation_cls = get_evaluation_model(obj)
-        reverse_name = evaluation_cls._meta.get_field('role').remote_field.name
         if permission:
-            obj_eval_qs = evaluation_cls.objects.filter(codename=permission.codename, object_id=obj.pk, content_type_id=ct.id)
+            assignment_qs = assignment_qs_user_to_obj_perm(actor, obj, permission)
         else:
-            # All relevant assignments for the object
-            obj_eval_qs = evaluation_cls.objects.filter(object_id=obj.pk, content_type_id=ct.id)
-        obj_assignment_qs = actor.role_assignments.filter(**{f'object_role__{reverse_name}__in': obj_eval_qs})
+            assignment_qs = assignment_qs_user_to_obj(actor, obj)
 
         team_ct = DABContentType.objects.get_for_model(get_team_model())
 
-        for assignment in obj_assignment_qs.distinct():
-            if assignment.content_type_id == team_ct.pk:
+        assignment_list = []
+        for assignment in assignment_qs.distinct():
+            if assignment.content_type_id is None:
+                perm_type = "global"
+            elif assignment.content_type_id == team_ct.pk:
                 perm_type = "team"
             elif assignment.content_type_id == ct.pk:
                 perm_type = "direct"
             else:
                 perm_type = "indirect"
             assignment_list.append({"type": perm_type, "role_definition": self.summarize_role_definition(assignment.role_definition)})
 
-        if permission:
-            global_assignment_qs = actor.role_assignments.filter(content_type=None, role_definition__permissions=permission)
-        else:
-            global_assignment_qs = actor.role_assignments.filter(content_type=None, role_definition__permissions__content_type=ct)
-
-        for assignment in global_assignment_qs.distinct():
-            assignment_list.append({"type": "global", "role_definition": self.summarize_role_definition(assignment.role_definition)})
-
         return assignment_list
 
     def get_url(self, obj) -> str:
@@ -298,9 +300,19 @@ class UserAccessListMixin(AccessListMixin, serializers.ModelSerializer):
     "controller uses auth.User model so this needs to be as compatible as possible, thus ModelSerializer"
 
     object_role_assignments = serializers.SerializerMethodField()
-    _expected_fields = ['id', 'url', 'username', 'is_superuser', 'object_role_assignments']
+    related = serializers.SerializerMethodField('_get_related')
+    _expected_fields = ['id', 'url', 'related', 'username', 'is_superuser', 'object_role_assignments']
 
 
 class TeamAccessListMixin(AccessListMixin, AbstractCommonModelSerializer):
     object_role_assignments = serializers.SerializerMethodField()
-    _expected_fields = ['id', 'url', 'name', 'organization', 'object_role_assignments']
+    related = serializers.SerializerMethodField('_get_related')
+    _expected_fields = ['id', 'url', 'related', 'name', 'organization', 'object_role_assignments']
+
+
+class UserAccessAssignmentSerializer(RoleUserAssignmentSerializer):
+    pass
+
+
+class TeamAccessAssignmentSerializer(RoleTeamAssignmentSerializer):
+    pass

ansible_base/rbac/api/views.py

@@ -20,7 +20,9 @@
     RoleMetadataSerializer,
     RoleTeamAssignmentSerializer,
     RoleUserAssignmentSerializer,
+    TeamAccessAssignmentSerializer,
     TeamAccessListMixin,
+    UserAccessAssignmentSerializer,
     UserAccessListMixin,
 )
 from ansible_base.rbac.evaluations import has_super_permission
@@ -34,6 +36,7 @@
 from ..policies import check_content_obj_permission
 from ..remote import RemoteObject, get_resource_prefix
 from ..sync import maybe_reverse_sync_assignment, maybe_reverse_sync_unassignment
+from .queries import assignment_qs_user_to_obj, assignment_qs_user_to_obj_perm
 
 
 def list_combine_values(data: dict[Type[Model], list[str]]) -> list[str]:
@@ -220,7 +223,61 @@ class RoleUserAssignmentViewSet(BaseAssignmentViewSet):
     ]
 
 
+class AccessURLMixin:
+    def get_actor_model(self):
+        return get_user_model()
+
+    def get_url_permission(self):
+        model_name = self.kwargs.get("model_name")
+        # Prefer treating the URL as requesting for some permission
+        return DABPermission.objects.filter(api_slug=model_name).first()
+
+    def get_url_content_type(self):
+        if getattr(self, 'permission', None):
+            # Access list will be all permissions for the given object
+            return self.permission.content_type
+
+        model_name = self.kwargs.get("model_name")
+        content_type = DABContentType.objects.filter(api_slug=model_name).first()
+        if not content_type:
+            raise NotFound(f'The slug {model_name} is not a valid permission or type identifier')
+
+        return content_type
+
+    def get_url_obj(self):
+        model_cls = self.content_type.model_class()
+        object_id = self.kwargs.get("pk")
+        if not issubclass(model_cls, RemoteObject):
+            try:
+                return model_cls.objects.get(pk=object_id)
+            except model_cls.DoesNotExist:
+                raise NotFound(f'The primary key {object_id} was not found for model {model_cls}')
+        else:
+            return model_cls(content_type=self.content_type, object_id=object_id)
+
+    def check_permission_to_object(self, obj):
+        try:
+            if not self.request.user.has_obj_perm(obj, 'view'):
+                raise NotFound
+        except RuntimeError:
+            check_content_obj_permission(self.request.user, obj)
+
+    def get_serializer_context(self):
+        ctx = super().get_serializer_context()
+        self.get_data_from_url()
+
+        ctx.update(
+            {
+                "permission": self.permission,
+                "related_object": self.related_object,
+                "content_type": self.content_type,
+            }
+        )
+        return ctx
+
+
 class UserAccessViewSet(
+    AccessURLMixin,
     AnsibleBaseDjangoAppApiView,
     mixins.ListModelMixin,
     GenericViewSet,
@@ -232,40 +289,12 @@ class UserAccessViewSet(
 
     serializer_mixin = UserAccessListMixin
 
-    def get_actor_model(self):
-        return get_user_model()
-
     def get_data_from_url(self):
         if not hasattr(self, 'related_object'):
-            model_name = self.kwargs.get("model_name")
-            object_id = self.kwargs.get("pk")
-
-            # Prefer treating the URL as requesting for some permission
-            self.permission = DABPermission.objects.filter(api_slug=model_name).first()
-
-            if not self.permission:
-                self.content_type = DABContentType.objects.filter(api_slug=model_name).first()
-                if not self.content_type:
-                    raise NotFound(f'The slug {model_name} is not a valid permission or type identifier')
-            else:
-                # Access list will be all permissions for the given object
-                self.content_type = self.permission.content_type
-
-            model_cls = self.content_type.model_class()
-            if not issubclass(model_cls, RemoteObject):
-                try:
-                    self.related_object = model_cls.objects.get(pk=object_id)
-                except model_cls.DoesNotExist:
-                    raise NotFound
-            else:
-                self.related_object = model_cls(content_type=self.content_type, object_id=object_id)
-
-            try:
-                if not self.request.user.has_obj_perm(self.related_object, 'view'):
-                    raise NotFound
-            except RuntimeError:
-                check_content_obj_permission(self.request.user, self.related_object)
-
+            self.permission = self.get_url_permission()
+            self.content_type = self.get_url_content_type()
+            self.related_object = self.get_url_obj()
+            self.check_permission_to_object(self.related_object)
         return (self.permission, self.content_type, self.related_object)
 
     def get_queryset(self):
@@ -315,22 +344,55 @@ class Meta:
 
         return DynamicActorSerializer
 
-    def get_serializer_context(self):
-        ctx = super().get_serializer_context()
-        permission, ct, obj = self.get_data_from_url()
-
-        ctx.update(
-            {
-                "permission": permission,
-                "related_object": obj,
-                "content_type": ct,
-            }
-        )
-        return ctx
-
 
 class TeamAccessViewSet(UserAccessViewSet):
     serializer_mixin = TeamAccessListMixin
 
     def get_actor_model(self):
         return get_team_model()
+
+
+class UserAccessAssignmentViewSet(
+    AccessURLMixin,
+    AnsibleBaseDjangoAppApiView,
+    mixins.ListModelMixin,
+    GenericViewSet,
+):
+    """
+    This gives drill-down information about the means of inheritance
+    for all the permissions show in the higher-level view of the access list
+    """
+
+    serializer_class = UserAccessAssignmentSerializer
+
+    def get_url_actor(self):
+        actor_pk = self.kwargs.get("actor_pk")
+        actor_cls = self.get_actor_model()
+        try:
+            return actor_cls.objects.get(pk=actor_pk)
+        except actor_cls.DoesNotExist:
+            raise NotFound(f'The {actor_cls._meta.model_name} with pk={actor_pk} can not be found')
+
+    def get_data_from_url(self):
+        if not hasattr(self, 'related_object'):
+            self.permission = self.get_url_permission()
+            self.content_type = self.get_url_content_type()
+            self.related_object = self.get_url_obj()
+            self.actor = self.get_url_actor()
+            self.check_permission_to_object(self.related_object)
+        return (self.permission, self.content_type, self.related_object, self.actor)
+
+    def get_queryset(self):
+        permission, ct, obj, actor = self.get_data_from_url()
+
+        if permission:
+            return assignment_qs_user_to_obj_perm(actor, obj, permission)
+        else:
+            return assignment_qs_user_to_obj(actor, obj)
+
+
+class TeamAccessAssignmentViewSet(UserAccessAssignmentViewSet):
+    serializer_class = TeamAccessAssignmentSerializer
+
+    def get_actor_model(self):
+        return get_team_model()

ansible_base/rbac/urls.py

@@ -1,19 +1,31 @@
 from django.urls import include, path
 
 from ansible_base.rbac.api.router import router
-from ansible_base.rbac.api.views import RoleMetadataView, TeamAccessViewSet, UserAccessViewSet
+from ansible_base.rbac.api.views import RoleMetadataView, TeamAccessAssignmentViewSet, TeamAccessViewSet, UserAccessAssignmentViewSet, UserAccessViewSet
 from ansible_base.rbac.apps import AnsibleRBACConfig
 
 app_name = AnsibleRBACConfig.label
 
 user_access_view = UserAccessViewSet.as_view({'get': 'list'})
 team_access_view = TeamAccessViewSet.as_view({'get': 'list'})
+user_access_assignment_view = UserAccessAssignmentViewSet.as_view({'get': 'list'})
+team_access_assignment_view = TeamAccessAssignmentViewSet.as_view({'get': 'list'})
 
 api_version_urls = [
     path('', include(router.urls)),
     path(r'role_metadata/', RoleMetadataView.as_view(), name="role-metadata"),
     path('role_user_access/<str:model_name>/<int:pk>/', user_access_view, name="role-user-access"),
     path('role_team_access/<str:model_name>/<int:pk>/', team_access_view, name="role-team-access"),
+    path(
+        'role_user_access/<str:model_name>/<int:pk>/<str:actor_pk>/',
+        user_access_assignment_view,
+        name='role-user-access-assignments',
+    ),
+    path(
+        'role_team_access/<str:model_name>/<int:pk>/<str:actor_pk>/',
+        team_access_assignment_view,
+        name='role-team-access-assignments',
+    ),
 ]
 
 root_urls = []