[feature] Implemented view permissions by extending DjangoModelPermissions class #249

openwisp_users/api/permissions.py

@@ -1,5 +1,5 @@
 from django.utils.translation import gettext_lazy as _
-from rest_framework.permissions import BasePermission
+from rest_framework.permissions import BasePermission, DjangoModelPermissions
 from swapper import load_model
 
 Organization = load_model('openwisp_users', 'Organization')
@@ -66,3 +66,37 @@ class IsOrganizationOwner(BaseOrganizationPermission):
 
     def validate_membership(self, user, org):
         return org and (user.is_superuser or user.is_owner(org))
+
+
+class ViewDjangoModelPermissions(DjangoModelPermissions):
+    perms_map = {
+        'GET': ['%(app_label)s.view_%(model_name)s'],
+        'OPTIONS': [],
+        'HEAD': ['%(app_label)s.view_%(model_name)s'],
+        'POST': ['%(app_label)s.add_%(model_name)s'],
+        'PUT': ['%(app_label)s.change_%(model_name)s'],
+        'PATCH': ['%(app_label)s.change_%(model_name)s'],
+        'DELETE': ['%(app_label)s.delete_%(model_name)s'],
+    }
+
+    def has_permission(self, request, view):
+        # Workaround to ensure DjangoModelPermissions are not applied
+        # to the root view when using DefaultRouter.
+        if getattr(view, '_ignore_model_permissions', False):
+            return True
+
+        user = request.user
+        if not user or (not user.is_authenticated and self.authenticated_users_only):
+            return False
+
+        queryset = self._queryset(view)
+        perms = self.get_required_permissions(request.method, queryset.model)
+        change_perm = self.get_required_permissions('PUT', queryset.model)
+
+        if request.method == 'GET':
+            if user.has_perms(perms) or user.has_perms(change_perm):
+                return True
+            else:
+                return False
+
+        return user.has_perms(perms)

tests/testapp/__init__.py

@@ -14,3 +14,11 @@ def _create_shelf(self, **kwargs):
         s.full_clean()
         s.save()
         return s
+
+    def _create_template(self, **kwargs):
+        options = dict(name='test-template')
+        options.update(kwargs)
+        t = self.template_model(**options)
+        t.full_clean()
+        t.save()
+        return t

tests/testapp/tests/test_permission_classes.py

@@ -1,14 +1,23 @@
+from django.contrib.auth import get_user_model
+from django.contrib.auth.models import Permission
 from django.test import TestCase
 from django.urls import reverse
+from swapper import load_model
 
 from openwisp_users.api.throttling import AuthRateThrottle
 
+from ..models import Template
 from .mixins import TestMultitenancyMixin
 
+User = get_user_model()
+Group = load_model('openwisp_users', 'Group')
+OrganizationUser = load_model('openwisp_users', 'OrganizationUser')
+
 
 class TestPermissionClasses(TestMultitenancyMixin, TestCase):
     def setUp(self):
         AuthRateThrottle.rate = 0
+        self.template_model = Template
         self.member_url = reverse('test_api_member_view')
         self.manager_url = reverse('test_api_manager_view')
         self.owner_url = reverse('test_api_owner_view')
@@ -117,3 +126,128 @@ def test_organization_field_with_errored_parent(self):
         with self.assertRaises(AttributeError) as error:
             self.client.get(reverse('test_error_field_view'), **auth)
         self.assertIn('Organization not found', str(error.exception))
+
+    def test_view_permission_with_operator(self):
+        user = User.objects.create_user(
+            username='operator', password='tester', email='operator@test.com'
+        )
+        operator_group = Group.objects.filter(name='Operator')
+        user.groups.set(operator_group)
+        org1 = self._get_org()
+        OrganizationUser.objects.create(user=user, organization=org1, is_admin=True)
+        self.client.force_login(user)
+        token = self._obtain_auth_token()
+        auth = dict(HTTP_AUTHORIZATION=f'Bearer {token}')
+        t1 = self._create_template(organization=org1)
+        with self.subTest('Get Template List'):
+            response = self.client.get(reverse('test_template_list'), **auth)
+            self.assertEqual(response.status_code, 403)
+        with self.subTest('Get Template Detail'):
+            response = self.client.get(
+                reverse('test_template_detail', args=[t1.pk]), **auth
+            )
+            self.assertEqual(response.status_code, 403)
+
+    def test_view_permission_with_administrator(self):
+        user = User.objects.create_user(
+            username='operator', password='tester', email='operator@test.com'
+        )
+        administrator_group = Group.objects.get(name='Administrator')
+        change_perm = Permission.objects.get(codename='change_template')
+        administrator_group.permissions.add(change_perm)
+        user.groups.add(administrator_group)
+        org1 = self._get_org()
+        OrganizationUser.objects.create(user=user, organization=org1, is_admin=True)
+        self.client.force_login(user)
+        token = self._obtain_auth_token()
+        auth = dict(HTTP_AUTHORIZATION=f'Bearer {token}')
+        t1 = self._create_template(organization=org1)
+        with self.subTest('Get Template List'):
+            response = self.client.get(reverse('test_template_list'), **auth)
+            self.assertEqual(response.status_code, 200)
+        with self.subTest('Get Template Detail'):
+            response = self.client.get(
+                reverse('test_template_detail', args=[t1.pk]), **auth
+            )
+            self.assertEqual(response.status_code, 200)
+        permissions = administrator_group.permissions.values_list('codename', flat=True)
+        self.assertFalse('view_template' in permissions)
+        self.assertTrue('change_template' in permissions)
+
+    def test_view_permission_with_operator_having_view_perm(self):
+        user = User.objects.create_user(
+            username='operator', password='tester', email='operator@test.com'
+        )
+        operator_group = Group.objects.get(name='Operator')
+        view_perm = Permission.objects.get(codename='view_template')
+        operator_group.permissions.add(view_perm)
+        user.groups.add(operator_group)
+        org1 = self._get_org()
+        OrganizationUser.objects.create(user=user, organization=org1, is_admin=True)
+        self.client.force_login(user)
+        token = self._obtain_auth_token()
+        auth = dict(HTTP_AUTHORIZATION=f'Bearer {token}')
+        t1 = self._create_template(organization=org1)
+        with self.subTest('Get Template List'):
+            response = self.client.get(reverse('test_template_list'), **auth)
+            self.assertEqual(response.status_code, 200)
+        with self.subTest('Get Template Detail'):
+            response = self.client.get(
+                reverse('test_template_detail', args=[t1.pk]), **auth
+            )
+            self.assertEqual(response.status_code, 200)
+        with self.subTest('Change Template Detail'):
+            data = {'name': 'change-template'}
+            response = self.client.patch(
+                reverse('test_template_detail', args=[t1.pk]), data, **auth
+            )
+            self.assertEqual(response.status_code, 403)
+        with self.subTest('Delete Template'):
+            response = self.client.delete(
+                reverse('test_template_detail', args=[t1.pk]), **auth
+            )
+            self.assertEqual(response.status_code, 403)
+
+    def test_view_django_model_permission_with_view_perm(self):
+        user = User.objects.create_user(
+            username='operator', password='tester', email='operator@test.com'
+        )
+        user_permissions = Permission.objects.filter(codename='view_template')
+        user.user_permissions.add(*user_permissions)
+        user.organizations_dict  # force caching
+        org1 = self._get_org()
+        OrganizationUser.objects.create(user=user, organization=org1, is_admin=True)
+        self.client.force_login(user)
+        token = self._obtain_auth_token()
+        auth = dict(HTTP_AUTHORIZATION=f'Bearer {token}')
+        t1 = self._create_template(organization=org1)
+        with self.subTest('Get Template List'):
+            response = self.client.get(reverse('test_template_list'), **auth)
+            self.assertEqual(response.status_code, 200)
+        with self.subTest('Get Template Detail'):
+            response = self.client.get(
+                reverse('test_template_detail', args=[t1.pk]), **auth
+            )
+            self.assertEqual(response.status_code, 200)
+
+    def test_view_django_model_permission_with_change_perm(self):
+        user = User.objects.create_user(
+            username='operator', password='tester', email='operator@test.com'
+        )
+        user_permissions = Permission.objects.filter(codename='change_template')
+        user.user_permissions.add(*user_permissions)
+        user.organizations_dict  # force caching
+        org1 = self._get_org()
+        OrganizationUser.objects.create(user=user, organization=org1, is_admin=True)
+        self.client.force_login(user)
+        token = self._obtain_auth_token()
+        auth = dict(HTTP_AUTHORIZATION=f'Bearer {token}')
+        t1 = self._create_template(organization=org1)
+        with self.subTest('Get Template List'):
+            response = self.client.get(reverse('test_template_list'), **auth)
+            self.assertEqual(response.status_code, 200)
+        with self.subTest('Get Template Detail'):
+            response = self.client.get(
+                reverse('test_template_detail', args=[t1.pk]), **auth
+            )
+            self.assertEqual(response.status_code, 200)

tests/testapp/urls.py

@@ -48,4 +48,5 @@
         name='test_shelf_list_unauthorized_view',
     ),
     path('template/', views.template_list, name='test_template_list',),
+    path('template/<int:pk>/', views.template_detail, name='test_template_detail',),
 ]

tests/testapp/views.py

@@ -1,5 +1,9 @@
 import swapper
-from rest_framework.generics import ListAPIView, ListCreateAPIView
+from rest_framework.generics import (
+    ListAPIView,
+    ListCreateAPIView,
+    RetrieveUpdateDestroyAPIView,
+)
 from rest_framework.response import Response
 from rest_framework.views import APIView
 
@@ -17,6 +21,7 @@
     IsOrganizationManager,
     IsOrganizationMember,
     IsOrganizationOwner,
+    ViewDjangoModelPermissions,
 )
 
 from .models import Book, Config, Shelf, Template
@@ -166,10 +171,23 @@ def get_queryset(self):
         return shelf.book_set.all()
 
 
-class TemplateListCreateView(ListCreateAPIView):
+class TemplateListCreateView(FilterByOrganizationManaged, ListCreateAPIView):
     serializer_class = TemplateSerializer
     authentication_classes = (BearerAuthentication,)
-    permission_classes = (IsOrganizationMember,)
+    permission_classes = (
+        IsOrganizationMember,
+        ViewDjangoModelPermissions,
+    )
+    queryset = Template.objects.all()
+
+
+class TemplateDetailView(FilterByOrganizationManaged, RetrieveUpdateDestroyAPIView):
+    serializer_class = TemplateSerializer
+    authentication_classes = (BearerAuthentication,)
+    permission_classes = (
+        IsOrganizationMember,
+        ViewDjangoModelPermissions,
+    )
     queryset = Template.objects.all()
 
 
@@ -188,3 +206,4 @@ class TemplateListCreateView(ListCreateAPIView):
 shelf_list_manager_view = ShelfListManagerView.as_view()
 shelf_list_owner_view = ShelfListOwnerView.as_view()
 template_list = TemplateListCreateView.as_view()
+template_detail = TemplateDetailView.as_view()