First pass at rest of views

Author: AlanCoding

ansible_base/lib/dynamic_config/settings_logic.py

@@ -108,6 +108,7 @@ def get_mergeable_dab_settings(settings: dict) -> dict:  # NOSONAR
             'user_ansible_id',
             'team_ansible_id',
             'object_ansible_id',
+            'assignment',  # for RoleAssignmentFilterBackend, assignment filtering
         )
 
     # SPECTACULAR SETTINGS

ansible_base/rbac/api/serializers.py

@@ -8,13 +8,14 @@
 from rest_framework.serializers import ValidationError
 
 from ansible_base.lib.abstract_models.common import get_url_for_object
-from ansible_base.lib.serializers.common import CommonModelSerializer, ImmutableCommonModelSerializer
+from ansible_base.lib.serializers.common import AbstractCommonModelSerializer, CommonModelSerializer, ImmutableCommonModelSerializer
+from ansible_base.lib.utils.auth import get_team_model
 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
+from ..models import DABContentType, DABPermission, get_evaluation_model
 
 
 class RoleDefinitionSerializer(CommonModelSerializer):
@@ -241,3 +242,56 @@ def get_actor_queryset(self, requesting_user):
 
 class RoleMetadataSerializer(serializers.Serializer):
     allowed_permissions = serializers.DictField(help_text=_('A List of permissions allowed for a role definition, given its content type.'))
+
+
+class AccessListMixin(AbstractCommonModelSerializer):
+    role_assignments = serializers.SerializerMethodField()
+
+    @staticmethod
+    def summarize_role_definition(role_definition):
+        return {"name": role_definition.name, "url": get_url_for_object(role_definition)}
+
+    def get_role_assignments(self, actor):
+        obj = self.context.get("related_object")
+        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)
+        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})
+
+        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:
+                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
+
+
+class UserAccessListMixin(AccessListMixin):
+    _expected_fields = ['username', 'role_assignments']
+
+
+class TeamAccessListMixin(AccessListMixin):
+    _expected_fields = ['name', 'organization', 'role_assignments']

ansible_base/rbac/api/views.py

@@ -5,11 +5,12 @@
 from django.db.models import Model
 from django.utils.translation import gettext_lazy as _
 from rest_framework import permissions
-from rest_framework.exceptions import ValidationError
+from rest_framework.exceptions import NotFound, ValidationError
 from rest_framework.generics import GenericAPIView
 from rest_framework.response import Response
-from rest_framework.viewsets import ModelViewSet
+from rest_framework.viewsets import GenericViewSet, ModelViewSet, mixins
 
+from ansible_base.lib.utils.auth import get_team_model, get_user_model
 from ansible_base.lib.utils.views.django_app_api import AnsibleBaseDjangoAppApiView
 from ansible_base.lib.utils.views.permissions import try_add_oauth2_scope_permission
 from ansible_base.rbac.api.permissions import RoleDefinitionPermissions
@@ -19,16 +20,18 @@
     RoleMetadataSerializer,
     RoleTeamAssignmentSerializer,
     RoleUserAssignmentSerializer,
+    TeamAccessListMixin,
+    UserAccessListMixin,
 )
 from ansible_base.rbac.evaluations import has_super_permission
 from ansible_base.rbac.models import RoleDefinition
 from ansible_base.rbac.permission_registry import permission_registry
 from ansible_base.rbac.policies import check_can_remove_assignment
 from ansible_base.rbac.validators import check_locally_managed, permissions_allowed_for_role, system_roles_enabled
-from ansible_base.rest_filters.rest_framework.ansible_id_backend import TeamAnsibleIdAliasFilterBackend, UserAnsibleIdAliasFilterBackend
+from ansible_base.rest_filters.rest_framework import ansible_id_backend
 
-from ..models.content_type import DABContentType
-from ..remote import get_resource_prefix
+from ..models import DABContentType, DABPermission, get_evaluation_model
+from ..remote import RemoteObject, get_resource_prefix
 
 
 def list_combine_values(data: dict[Type[Model], list[str]]) -> list[str]:
@@ -174,7 +177,8 @@ class RoleTeamAssignmentViewSet(BaseAssignmentViewSet):
     serializer_class = RoleTeamAssignmentSerializer
     prefetch_related = ('team',)
     filter_backends = BaseAssignmentViewSet.filter_backends + [
-        TeamAnsibleIdAliasFilterBackend,
+        ansible_id_backend.TeamAnsibleIdAliasFilterBackend,
+        ansible_id_backend.RoleAssignmentFilterBackend,
     ]
 
 
@@ -193,5 +197,108 @@ class RoleUserAssignmentViewSet(BaseAssignmentViewSet):
     serializer_class = RoleUserAssignmentSerializer
     prefetch_related = ('user',)
     filter_backends = BaseAssignmentViewSet.filter_backends + [
-        UserAnsibleIdAliasFilterBackend,
+        ansible_id_backend.UserAnsibleIdAliasFilterBackend,
+        ansible_id_backend.RoleAssignmentFilterBackend,
     ]
+
+
+class UserAccessViewSet(
+    AnsibleBaseDjangoAppApiView,
+    mixins.ListModelMixin,
+    GenericViewSet,
+):
+    """
+    Use this endpoint to get a list of users who have access to a resource.
+    This is a list-only view that provides a list of users, plus extra data.
+    """
+
+    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)
+
+            if not self.request.user.has_obj_perm(self.related_object, 'view'):
+                raise NotFound
+
+        return (self.permission, self.content_type, self.related_object)
+
+    def get_queryset(self):
+        permission, ct, obj = self.get_data_from_url()
+
+        evaluation_cls = get_evaluation_model(obj)
+        reverse_name = evaluation_cls._meta.get_field('role').remote_field.name
+        actor_cls = self.get_actor_model()
+        assignment_cls = actor_cls._meta.get_field('role_assignments').related_model
+
+        if permission:
+            obj_eval_qs = evaluation_cls.objects.filter(codename=permission.codename, object_id=obj.pk, content_type_id=ct.id)
+        else:
+            # All relevant evaluations for the object
+            obj_eval_qs = evaluation_cls.objects.filter(object_id=obj.pk, content_type_id=ct.id)
+        obj_assignment_qs = assignment_cls.objects.filter(**{f'object_role__{reverse_name}__in': obj_eval_qs})
+
+        if permission:
+            global_assignment_qs = assignment_cls.objects.filter(content_type=None, role_definition__permissions=permission)
+        else:
+            global_assignment_qs = assignment_cls.objects.filter(content_type=None, role_definition__permissions__content_type=ct)
+
+        assignment_qs = obj_assignment_qs | global_assignment_qs
+        actor_qs = actor_cls.objects.filter(role_assignments__in=assignment_qs)
+        if actor_cls._meta.model_name == 'user':
+            actor_qs |= actor_qs.filter(is_superuser=True)
+        return actor_qs
+
+    def get_serializer_class(self):
+        actor_cls = self.get_actor_model()
+
+        class DynamicActorSerializer(self.serializer_mixin):
+            class Meta:
+                model = actor_cls
+                fields = self.serializer_mixin.Meta.fields + self.serializer_mixin._expected_fields
+
+        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()

ansible_base/rbac/service_api/serializers.py

@@ -1,3 +1,4 @@
+from django.apps import apps
 from rest_framework import serializers
 
 from ..models import DABContentType, DABPermission, RoleTeamAssignment, RoleUserAssignment
@@ -28,6 +29,8 @@ class BaseAssignmentSerializer(serializers.ModelSerializer):
     role_definition = serializers.SlugRelatedField(read_only=True, slug_field='name')
     created_by_ansible_id = serializers.SerializerMethodField()
     object_ansible_id = serializers.SerializerMethodField()
+    # TODO: use the from_service to control what we sync back to
+    from_service = serializers.CharField(write_only=True)
 
     def get_created_by_ansible_id(self, obj):
         return str(obj.created_by.resource.ansible_id)
@@ -40,9 +43,20 @@ def get_object_ansible_id(self, obj):
             return str(content_object.resource.ansible_id)
         return None
 
+    def find_existing_assignment(self, queryset):
+        actor_ansible_id = self.validated_data[f'{self.actor_field}_ansible_id']
+        object_id = self.validated_data['object_id']
+        role_definition = self.validated_data['role_definition']
+
+        resource_cls = apps.get_model('dab_resource_registry', 'Resource')
+        actor_resource = resource_cls.objects.get(ansible_id=actor_ansible_id)
+        actor = actor_resource.content_object
+        return queryset.filter(object_id=object_id, role_definition=role_definition, **{self.actor_field: actor}).first()
+
 
 class RoleUserAssignmentSerializer(BaseAssignmentSerializer):
     user_ansible_id = serializers.SerializerMethodField()
+    actor_field = 'user'
 
     class Meta:
         model = RoleUserAssignment
@@ -54,6 +68,7 @@ def get_user_ansible_id(self, obj):
 
 class RoleTeamAssignmentSerializer(BaseAssignmentSerializer):
     user_ansible_id = serializers.SerializerMethodField()
+    actor_field = 'team'
 
     class Meta:
         model = RoleTeamAssignment

ansible_base/rbac/service_api/views.py

@@ -1,6 +1,10 @@
+from rest_framework import status
+from rest_framework.decorators import action
+from rest_framework.response import Response
 from rest_framework.viewsets import GenericViewSet, mixins
 
 from ansible_base.lib.utils.views.django_app_api import AnsibleBaseDjangoAppApiView
+from ansible_base.rest_filters.rest_framework import ansible_id_backend
 
 from ..models import DABContentType, DABPermission, RoleTeamAssignment, RoleUserAssignment
 from . import serializers as service_serializers
@@ -43,6 +47,42 @@ class ServiceRoleUserAssignmentViewSet(
 
     queryset = RoleUserAssignment.objects.prefetch_related('user__resource', *prefetch_related)
     serializer_class = service_serializers.RoleUserAssignmentSerializer
+    filter_backends = AnsibleBaseDjangoAppApiView.filter_backends + [
+        ansible_id_backend.UserAnsibleIdAliasFilterBackend,
+        ansible_id_backend.RoleAssignmentFilterBackend,
+    ]
+
+    @action(detail=False, methods=['post'], url_path='assign')
+    def assign(self, request):
+        serializer = self.get_serializer(data=request.data)
+        serializer.is_valid(raise_exception=True)
+
+        existing = serializer.find_existing_assignment(self.get_queryset())
+        if existing:
+            return Response(
+                {"detail": "This assignment already exists."},
+                status=status.HTTP_409_CONFLICT,
+            )
+
+        instance = serializer.save()
+        output_serializer = self.get_serializer(instance)
+        return Response(output_serializer.data, status=status.HTTP_201_CREATED)
+
+    @action(detail=False, methods=['post'], url_path='unassign')
+    def unassign(self, request):
+        serializer = self.get_serializer(data=request.data)
+        serializer.is_valid(raise_exception=True)
+
+        existing = serializer.find_existing_assignment(self.get_queryset())
+        if not existing:
+            return Response(
+                {"detail": "No such assignment exists."},
+                status=status.HTTP_409_CONFLICT,
+            )
+
+        # Use standard DRF delete logic
+        self.perform_destroy(existing)
+        return Response(status=status.HTTP_204_NO_CONTENT)
 
 
 class ServiceRoleTeamAssignmentViewSet(
@@ -54,3 +94,7 @@ class ServiceRoleTeamAssignmentViewSet(
 
     queryset = RoleTeamAssignment.objects.prefetch_related('team__resource', *prefetch_related)
     serializer_class = service_serializers.RoleTeamAssignmentSerializer
+    filter_backends = AnsibleBaseDjangoAppApiView.filter_backends + [
+        ansible_id_backend.TeamAnsibleIdAliasFilterBackend,
+        ansible_id_backend.RoleAssignmentFilterBackend,
+    ]

ansible_base/rbac/urls.py

@@ -1,7 +1,7 @@
 from django.urls import include, path
 
 from ansible_base.rbac.api.router import router
-from ansible_base.rbac.api.views import RoleMetadataView
+from ansible_base.rbac.api.views import RoleMetadataView, TeamAccessViewSet, UserAccessViewSet
 from ansible_base.rbac.apps import AnsibleRBACConfig
 
 from .service_api.router import service_router
@@ -13,10 +13,15 @@
     path('', include(service_router.urls)),
 ]
 
+user_access_view = UserAccessViewSet.as_view({'get': 'list'})
+team_access_view = TeamAccessViewSet.as_view({'get': 'list'})
+
 api_version_urls = [
     path('', include(router.urls)),
     path('service-index/', include(service_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"),
 ]
 
 root_urls = []