Merge pull request #171 from avantifellows/feature/api-permissions

Api permissions scaffolding

organizations/permissions.py

@@ -0,0 +1,11 @@
+from rest_framework import permissions
+
+
+class OrganizationPermission(permissions.BasePermission):
+    """
+    Permission check for organizations.
+    """
+
+    def has_permission(self, request, view):
+        """View-level permissions for organization. This determines whether the request can access organization instances or not."""
+        return request.user.is_superuser

organizations/views.py

@@ -1,6 +1,8 @@
 from rest_framework import viewsets
 from organizations.models import Organization
 from organizations.serializers import OrganizationSerializer
+from rest_framework.permissions import IsAuthenticated
+from organizations.permissions import OrganizationPermission
 
 
 class OrganizationViewSet(viewsets.ModelViewSet):
@@ -15,5 +17,6 @@ class OrganizationViewSet(viewsets.ModelViewSet):
     destroy: Soft delete an organization
     """
 
+    permission_classes = [IsAuthenticated, OrganizationPermission]
     queryset = Organization.objects.all()
     serializer_class = OrganizationSerializer

plio/permissions.py

@@ -0,0 +1,40 @@
+from rest_framework import permissions
+from organizations.middleware import OrganizationTenantMiddleware
+from plio.settings import DEFAULT_TENANT_SHORTCODE
+from users.models import OrganizationUser
+
+
+class PlioPermission(permissions.BasePermission):
+    """
+    Permission check for plios.
+    """
+
+    def has_permission(self, request, view):
+        """View-level permissions for plio. This determines whether the request can access plio instances or not."""
+        return True
+
+    def has_object_permission(self, request, view, obj):
+        """Object-level permissions for plio. This determines whether the request can access a plio instance or not."""
+
+        if request.user.is_superuser:
+            return True
+
+        organization_shortcode = (
+            OrganizationTenantMiddleware.get_organization_shortcode(request)
+        )
+        if organization_shortcode == DEFAULT_TENANT_SHORTCODE:
+            # this is the user's personal workspace in the public DB schema
+            # only plios created by the user can be accessed
+            return request.user == obj.created_by
+
+        # checking if user is a member of the organization
+        user_belongs_to_organization = OrganizationUser.objects.filter(
+            organization__shortcode=organization_shortcode,
+            user=request.user.id,
+        ).exists()
+
+        if user_belongs_to_organization and obj.is_public:
+            return True
+
+        # user doesn't belong to organization or plio isn't public
+        return request.user == obj.created_by

plio/views.py

@@ -25,6 +25,7 @@
     get_responses_dump_query,
     get_events_query,
 )
+from plio.permissions import PlioPermission
 
 
 class StandardResultsSetPagination(PageNumberPagination):
@@ -81,6 +82,7 @@ class PlioViewSet(viewsets.ModelViewSet):
     play: Retrieve a plio in order to play
     """
 
+    permission_classes = [IsAuthenticated, PlioPermission]
     serializer_class = PlioSerializer
     lookup_field = "uuid"
 
@@ -133,6 +135,7 @@ def list_uuid(self, request):
         queryset = self.filter_queryset(self.get_queryset())
         uuid_list = queryset.values_list("uuid", flat=True)
         page = self.paginate_queryset(uuid_list)
+
         if page is not None:
             return self.get_paginated_response(page)
 
@@ -147,7 +150,11 @@ def list_uuid(self, request):
             }
         )
 
-    @action(methods=["get"], detail=True, permission_classes=[IsAuthenticated])
+    @action(
+        methods=["get"],
+        detail=True,
+        permission_classes=[IsAuthenticated, PlioPermission],
+    )
     def play(self, request, uuid):
         queryset = Plio.objects.filter(uuid=uuid)
         queryset = queryset.filter(is_public=True) | queryset.filter(
@@ -162,7 +169,11 @@ def play(self, request, uuid):
         serializer = self.get_serializer(plio)
         return Response(serializer.data)
 
-    @action(methods=["post"], detail=True, permission_classes=[IsAuthenticated])
+    @action(
+        methods=["post"],
+        detail=True,
+        permission_classes=[IsAuthenticated, PlioPermission],
+    )
     def duplicate(self, request, uuid):
         """Creates a clone of the plio with the given uuid"""
         plio = self.get_object()
@@ -173,7 +184,11 @@ def duplicate(self, request, uuid):
         plio.save()
         return Response(self.get_serializer(plio).data)
 
-    @action(methods=["get"], detail=True, permission_classes=[IsAuthenticated])
+    @action(
+        methods=["get"],
+        detail=True,
+        permission_classes=[IsAuthenticated, PlioPermission],
+    )
     def download_data(self, request, uuid):
         # return 404 if user cannot access the object
         # else fetch the object

users/permissions.py

@@ -0,0 +1,19 @@
+from rest_framework import permissions
+
+
+class UserPermission(permissions.BasePermission):
+    """
+    Permission check for users.
+    """
+
+    def has_permission(self, request, view):
+        """View-level permissions for user. This determines whether the request can access user instances or not."""
+        if view.action in ["list", "create"]:
+            # only superuser can list or create users
+            return request.user.is_superuser
+
+        return True
+
+    def has_object_permission(self, request, view, obj):
+        """Object-level permissions for user. This determines whether the request can access a user instance or not."""
+        return request.user.is_superuser or request.user == obj

users/views.py

@@ -2,6 +2,7 @@
 from rest_framework.permissions import AllowAny
 from rest_framework import viewsets, status
 from rest_framework.response import Response
+from rest_framework.permissions import IsAuthenticated
 
 import datetime
 import string
@@ -26,6 +27,7 @@
 
 from users.models import User, OneTimePassword, OrganizationUser
 from users.serializers import UserSerializer, OtpSerializer, OrganizationUserSerializer
+from users.permissions import UserPermission
 
 from .services import SnsService
 import requests
@@ -43,11 +45,16 @@ class UserViewSet(viewsets.ModelViewSet):
     destroy: Soft delete a user
     """
 
+    permission_classes = [IsAuthenticated, UserPermission]
     queryset = User.objects.all()
     serializer_class = UserSerializer
 
-    @action(detail=True, methods=["patch", "get"])
-    def config(self, request, pk=True):
+    @action(
+        detail=True,
+        methods=["patch", "get"],
+        permission_classes=[IsAuthenticated, UserPermission],
+    )
+    def config(self, request, pk):
         user = self.get_object()
         if request.method == "GET":
             return Response(user.config)