Merge pull request #842 from fao89/7311

Remove permissions associated with groups

Author: bmbouter

CHANGES/7310.feature

@@ -0,0 +1,3 @@
+Extended endpoint `/pulp/api/v3/groups/:pk/users` to add and remove users from a group.
+
+NOTE: this endpoint is in tech-preview and may change in backwards incompatible ways in the future.

CHANGES/7311.feature

@@ -0,0 +1,4 @@
+Extended endpoints `/pulp/api/v3/groups/:pk/model_permissions` and
+`/pulp/api/v3/groups/:pk/object_permissions` to add and remove permissions from a group.
+
+NOTE: this endpoint is in tech-preview and may change in backwards incompatible ways in the future.

pulpcore/app/serializers/user.py

@@ -1,7 +1,9 @@
 from gettext import gettext as _
 
 from django.contrib.auth import get_user_model
-from django.contrib.auth.models import Group
+from django.contrib.auth.models import Group, Permission
+from django.urls import reverse
+from guardian.models.models import GroupObjectPermission
 from rest_framework import serializers
 
 from pulpcore.app.serializers import IdentityField
@@ -27,15 +29,39 @@ def to_representation(self, obj):
             return serializer.data.get("pulp_href")
 
 
-class ObjectPermissionSerializer(serializers.Serializer):
-    """Serializer for user/group object permission."""
+class PermissionSerializer(serializers.Serializer):
+    """Serializer for User/Group object permission."""
 
+    pulp_href = serializers.SerializerMethodField(read_only=True)
+    id = serializers.SerializerMethodField(read_only=True)
     permission = PermissionField(source="*", read_only=True)
     obj = ContentObjectField(help_text=_("Content object."), source="*", read_only=True)
 
+    def get_id(self, obj):
+        """Get model/object permission id."""
+        return obj.id
+
+    def get_pulp_href(self, obj):
+        """Get model/object permission pulp_href."""
+        group_pk = self.context.get("group_pk")
+
+        if group_pk and isinstance(obj, Permission):
+            return reverse("model_permissions-detail", args=[group_pk, obj.pk])
+
+        if group_pk and isinstance(obj, GroupObjectPermission):
+            return reverse("object_permissions-detail", args=[group_pk, obj.pk])
+
+    def to_representation(self, obj):
+        representation = super().to_representation(obj)
+
+        if not self.context.get("group_pk"):
+            representation.pop("pulp_href")
+
+        return representation
+
 
 class UserGroupSerializer(serializers.ModelSerializer):
-    """Serializer for user groups."""
+    """Serializer for Groups that belong to an User."""
 
     name = serializers.CharField(help_text=_("Name."), max_length=150,)
     pulp_href = IdentityField(view_name="groups-detail")
@@ -46,7 +72,7 @@ class Meta:
 
 
 class UserSerializer(serializers.ModelSerializer):
-    """Serializer for user."""
+    """Serializer for User."""
 
     pulp_href = IdentityField(view_name="users-detail")
     id = serializers.IntegerField(read_only=True)
@@ -83,7 +109,7 @@ class Meta:
 
 
 class GroupUserSerializer(serializers.ModelSerializer):
-    """Serializer for group users."""
+    """Serializer for Users that belong to a Group."""
 
     username = serializers.CharField(
         help_text=_("Required. 150 characters or fewer. Letters, digits and @/./+/-/_ only."),
@@ -97,11 +123,11 @@ class Meta:
 
 
 class GroupSerializer(serializers.ModelSerializer):
-    """Serializer for group."""
+    """Serializer for Group."""
 
     pulp_href = IdentityField(view_name="groups-detail")
     id = serializers.IntegerField(read_only=True)
-    name = serializers.CharField(help_text=_("First name"), max_length=150)
+    name = serializers.CharField(help_text=_("Name"), max_length=150)
 
     class Meta:
         model = Group

pulpcore/app/viewsets/__init__.py

@@ -49,4 +49,10 @@
 )
 from .task import TaskViewSet, TaskGroupViewSet, WorkerViewSet  # noqa
 from .upload import UploadViewSet  # noqa
-from .user import GroupViewSet, UserViewSet  # noqa
+from .user import (  # noqa
+    GroupViewSet,
+    GroupUserViewSet,
+    GroupModelPermissionViewSet,
+    GroupObjectPermissionViewSet,
+    UserViewSet,
+)

pulpcore/app/viewsets/user.py

@@ -1,15 +1,20 @@
+from gettext import gettext as _
+
 from django.contrib.auth import get_user_model
-from django.contrib.auth.models import Group
+from django.contrib.auth.models import Group, Permission
+from django.shortcuts import get_object_or_404
 from drf_spectacular.utils import extend_schema
-from rest_framework import mixins
+from guardian.models.models import GroupObjectPermission
+from rest_framework import mixins, status
 from rest_framework.decorators import action
 from rest_framework.response import Response
+from rest_framework.serializers import ValidationError
 
 from pulpcore.app.viewsets.base import NamedModelViewSet
 from pulpcore.app.serializers.user import (
     GroupSerializer,
     GroupUserSerializer,
-    ObjectPermissionSerializer,
+    PermissionSerializer,
     UserSerializer,
 )
 
@@ -29,16 +34,14 @@ class UserViewSet(
     queryset = get_user_model().objects.all()
 
     @extend_schema(description="List user permissions.",)
-    @action(detail=True, methods=["get"], serializer_class=ObjectPermissionSerializer)
+    @action(detail=True, methods=["get"], serializer_class=PermissionSerializer)
     def permissions(self, request, pk):
         """
         List user permissions.
         """
         user = self.get_object()
-        object_permission = ObjectPermissionSerializer(
-            user.userobjectpermission_set.all(), many=True
-        )
-        permissions = ObjectPermissionSerializer(user.user_permissions.all(), many=True)
+        object_permission = PermissionSerializer(user.userobjectpermission_set.all(), many=True)
+        permissions = PermissionSerializer(user.user_permissions.all(), many=True)
         return Response(object_permission.data + permissions.data)
 
 
@@ -58,28 +61,220 @@ class GroupViewSet(
     """
 
     endpoint_name = "groups"
+    router_lookup = "group"
     serializer_class = GroupSerializer
     queryset = Group.objects.all()
 
-    @extend_schema(description="List group permissions.",)
-    @action(detail=True, methods=["get"], serializer_class=ObjectPermissionSerializer)
-    def permissions(self, request, pk):
+
+class GroupModelPermissionViewSet(NamedModelViewSet):
+    """
+    ViewSet for Model Permissions that belongs to a Group.
+
+    NOTE: This API endpoint is in "tech preview" and subject to change
+
+    """
+
+    endpoint_name = "model_permissions"
+    nest_prefix = "groups"
+    router_lookup = "model_permission"
+    parent_viewset = GroupViewSet
+    parent_lookup_kwargs = {"group_pk": "group__pk"}
+    serializer_class = PermissionSerializer
+    queryset = Permission.objects.all()
+
+    def get_model_permission(self, request):
+        """Get model permission"""
+        data = {}
+        for key, value in request.data.items():
+            if key == "permission":
+                data["codename"] = value.split(".")[-1]
+                continue
+
+            if key == "pulp_href":
+                if "id" in data.keys():
+                    continue
+                data["id"] = value.strip("/").split("/")[-1]
+                continue
+
+            data[key] = value
+
+        return get_object_or_404(Permission, **data)
+
+    @extend_schema(description="Retrieve a model permission from a group.")
+    def retrieve(self, request, group_pk, pk):
+        instance = get_object_or_404(Permission, pk=pk)
+        serializer = PermissionSerializer(instance, context={"group_pk": group_pk})
+        return Response(serializer.data)
+
+    @extend_schema(description="List group permissions.", responses={200: PermissionSerializer})
+    def list(self, request, group_pk):
+        """
+        List group model permissions.
+        """
+        group = Group.objects.get(pk=group_pk)
+        queryset = group.permissions.all()
+
+        page = self.paginate_queryset(queryset)
+        if page is not None:
+            serializer = PermissionSerializer(page, context={"group_pk": group_pk}, many=True)
+            return self.get_paginated_response(serializer.data)
+
+        serializer = PermissionSerializer(queryset, context={"group_pk": group_pk}, many=True)
+        return Response(serializer.data)
+
+    @extend_schema(
+        description="Add a model permission to a group.", responses={201: PermissionSerializer},
+    )
+    def create(self, request, group_pk):
+        """
+        Add a model permission to a group.
+        """
+        group = Group.objects.get(pk=group_pk)
+        permission = self.get_model_permission(request)
+        group.permissions.add(permission)
+        group.save()
+        serializer = PermissionSerializer(permission, context={"group_pk": group_pk})
+        return Response(serializer.data, status=status.HTTP_201_CREATED)
+
+    @extend_schema(description="Remove a model permission from a group.")
+    def destroy(self, request, group_pk, pk):
+        """
+        Remove a model permission from a group.
+        """
+        group = Group.objects.get(pk=group_pk)
+        permission = get_object_or_404(Permission, pk=pk)
+        group.permissions.remove(permission)
+        group.save()
+        return Response(status=status.HTTP_204_NO_CONTENT)
+
+
+class GroupObjectPermissionViewSet(NamedModelViewSet):
+    """
+    ViewSet for Object Permissions that belongs to a Group.
+
+    NOTE: This API endpoint is in "tech preview" and subject to change
+
+    """
+
+    endpoint_name = "object_permissions"
+    nest_prefix = "groups"
+    router_lookup = "object_permission"
+    parent_viewset = GroupViewSet
+    parent_lookup_kwargs = {"group_pk": "group__pk"}
+    serializer_class = PermissionSerializer
+    queryset = Permission.objects.all()
+
+    def get_model_permission(self, request):
+        """Get model permission"""
+        codename = request.data["permission"].split(".")[-1]
+        return get_object_or_404(Permission, codename=codename)
+
+    @extend_schema(description="Retrieve a model permission from a group.")
+    def retrieve(self, request, group_pk, pk):
+        instance = get_object_or_404(GroupObjectPermission, pk=pk)
+        serializer = PermissionSerializer(instance, context={"group_pk": group_pk})
+        return Response(serializer.data)
+
+    @extend_schema(
+        description="List group object permissions.", responses={200: PermissionSerializer}
+    )
+    def list(self, request, group_pk):
         """
-        List group permissions.
+        List group object permissions.
         """
-        group = self.get_object()
-        object_permission = ObjectPermissionSerializer(
-            group.groupobjectpermission_set.all(), many=True
+        group = Group.objects.get(pk=group_pk)
+        queryset = group.groupobjectpermission_set.all()
+
+        page = self.paginate_queryset(queryset)
+        if page is not None:
+            serializer = PermissionSerializer(page, context={"group_pk": group_pk}, many=True)
+            return self.get_paginated_response(serializer.data)
+
+        serializer = PermissionSerializer(queryset, context={"group_pk": group_pk}, many=True)
+        return Response(serializer.data)
+
+    @extend_schema(
+        description="Add an object permission to a group.", responses={201: PermissionSerializer},
+    )
+    def create(self, request, group_pk):
+        """
+        Create an object permission to a group.
+        """
+        group = Group.objects.get(pk=group_pk)
+        permission = self.get_model_permission(request)
+        object_pk = request.data["obj"].strip("/").split("/")[-1]
+        object_permission = GroupObjectPermission(
+            group=group,
+            permission=permission,
+            content_type_id=permission.content_type_id,
+            object_pk=object_pk,
         )
-        permissions = ObjectPermissionSerializer(group.permissions.all(), many=True)
-        return Response(object_permission.data + permissions.data)
+        object_permission.save()
+        serializer = PermissionSerializer(object_permission, context={"group_pk": group_pk})
+        return Response(serializer.data, status=status.HTTP_201_CREATED)
+
+    @extend_schema(description="Remove an object permission from a group.")
+    def destroy(self, request, group_pk, pk):
+        """
+        Delete an object permission from a group.
+        """
+        object_permission = get_object_or_404(GroupObjectPermission, pk=pk, group_id=group_pk)
+        object_permission.delete()
+        return Response(status=status.HTTP_204_NO_CONTENT)
+
 
-    @extend_schema(description="List group users.",)
-    @action(detail=True, methods=["get"], serializer_class=GroupUserSerializer)
-    def users(self, request, pk):
+class GroupUserViewSet(NamedModelViewSet):
+    """
+    ViewSet for Users that belongs to a Group.
+
+    NOTE: This API endpoint is in "tech preview" and subject to change
+
+    """
+
+    endpoint_name = "users"
+    nest_prefix = "groups"
+    router_lookup = "user"
+    parent_viewset = GroupViewSet
+    parent_lookup_kwargs = {"group_pk": "groups__pk"}
+    serializer_class = GroupUserSerializer
+    queryset = get_user_model().objects.all()
+
+    def list(self, request, group_pk):
         """
         List group users.
         """
-        group = self.get_object()
-        serializer = GroupUserSerializer(group.user_set.all(), many=True)
+        group = Group.objects.get(pk=group_pk)
+        queryset = group.user_set.all()
+
+        page = self.paginate_queryset(queryset)
+        if page is not None:
+            serializer = GroupUserSerializer(page, context={"request": None}, many=True)
+            return self.get_paginated_response(serializer.data)
+
+        serializer = self.get_serializer(queryset, many=True)
         return Response(serializer.data)
+
+    def create(self, request, group_pk):
+        """
+        Add a user to a group.
+        """
+        group = Group.objects.get(pk=group_pk)
+        if not request.data:
+            raise ValidationError(
+                _("Please provide one of the following values for User: 'pk', 'id', 'username'")
+            )
+        user = get_object_or_404(get_user_model(), **request.data)
+        group.user_set.add(user)
+        group.save()
+        serializer = GroupUserSerializer(user, context={"request": None})
+        return Response(serializer.data, status=status.HTTP_201_CREATED)
+
+    def destroy(self, request, group_pk, pk):
+        """
+        Remove a user from a group.
+        """
+        group = Group.objects.get(pk=group_pk)
+        user = get_object_or_404(get_user_model(), pk=pk)
+        group.user_set.remove(user)
+        group.save()
+        return Response(status=status.HTTP_204_NO_CONTENT)