add permission framework (#327)

* add permission framework

* add test class

* add permission validation to measurement API

django_project/core/group_email_receiver.py

@@ -1,7 +1,7 @@
 """
 Tomorrow Now GAP.
 
-.. note:: Definition admin
+.. note:: Group for crop plan email receiver
 """
 from django.contrib.auth.models import Group
 

django_project/core/settings/project.py

@@ -37,7 +37,8 @@
     'spw',
     'message',
     'prise',
-    'dcas'
+    'dcas',
+    'permission'
 )
 INSTALLED_APPS = INSTALLED_APPS + PROJECT_APPS
 

django_project/gap_api/api_views/measurement.py

@@ -18,7 +18,7 @@
 from django.http import StreamingHttpResponse
 from drf_yasg import openapi
 from drf_yasg.utils import swagger_auto_schema
-from rest_framework.exceptions import ValidationError
+from rest_framework.exceptions import ValidationError, PermissionDenied
 from rest_framework.permissions import IsAuthenticated
 from rest_framework.response import Response
 from rest_framework.views import APIView
@@ -46,6 +46,7 @@
 from gap_api.serializers.common import APIErrorSerializer
 from gap_api.utils.helper import ApiTag
 from gap_api.mixins import GAPAPILoggingMixin, CounterSlidingWindowThrottle
+from permission.models import PermissionType
 
 
 def product_type_list():
@@ -357,7 +358,6 @@ def _get_accel_redirect_response(
         )
         return response
 
-
     def _read_data_as_netcdf(
             self, reader_dict: Dict[int, BaseDatasetReader],
             start_dt: datetime, end_dt: datetime,
@@ -461,6 +461,32 @@ def _read_data_as_ascii(
             suffix='.txt', separator='\t', content_type='text/ascii'
         )
 
+    def validate_product_type(self, product_filter):
+        """Validate user has access to product type.
+
+        :param product_filter: list of product type
+        :type product_filter: list
+        :raises PermissionDenied: no permission to the product type
+        """
+        dataset_types = DatasetType.objects.filter(
+            variable_name__in=product_filter
+        )
+        has_perm = True
+        for dataset_type in dataset_types:
+            has_perm = (
+                has_perm and
+                self.request.user.has_perm(
+                    PermissionType.VIEW_DATASET_TYPE, dataset_type
+                )
+            )
+
+        if not has_perm:
+            raise PermissionDenied({
+                'Missing Permission': (
+                    f'You don\'t have access to {product_filter}!'
+                )
+            })
+
     def validate_output_format(
             self, dataset: Dataset, product_type: str,
             location: DatasetReaderInput, output_format):
@@ -595,6 +621,10 @@ def get_response_data(self) -> Response:
             attribute__is_active=True
         )
         product_filter = self._get_product_filter()
+
+        # validate product type access
+        self.validate_product_type(product_filter)
+
         dataset_attributes = dataset_attributes.annotate(
             product_name=Lower('dataset__type__variable_name')
         ).filter(

django_project/gap_api/tests/test_measurement_api.py

@@ -12,6 +12,7 @@
 from django.contrib.gis.geos import Polygon, MultiPolygon, Point
 from django.urls import reverse
 from rest_framework.exceptions import ValidationError
+from guardian.shortcuts import assign_perm
 
 from core.tests.common import FakeResolverMatchV1, BaseAPIViewTest
 from gap.factories import (
@@ -27,6 +28,7 @@
 from gap_api.models.api_config import DatasetTypeAPIConfig
 from gap_api.api_views.measurement import MeasurementAPI
 from gap_api.factories import LocationFactory
+from permission.models import PermissionType
 
 
 class MockDatasetReader(BaseDatasetReader):
@@ -423,3 +425,52 @@ def test_validate_date_range(self):
             datetime(2024, 10, 1, 0, 0, 0),
             datetime(2024, 10, 30, 0, 0, 0)
         )
+
+    @patch('gap_api.api_views.measurement.get_reader_from_dataset')
+    def test_access_validation(self, mocked_reader):
+        """Test invalid access API."""
+        view = MeasurementAPI.as_view()
+        mocked_reader.return_value = MockDatasetReader
+        dataset = Dataset.objects.get(name='CBAM Climate Reanalysis')
+        attribute1 = DatasetAttribute.objects.filter(
+            dataset=dataset,
+            attribute__variable_name='max_temperature'
+        ).first()
+        attribute2 = DatasetAttribute.objects.filter(
+            dataset=dataset,
+            attribute__variable_name='total_rainfall'
+        ).first()
+        attribs = [
+            attribute1.attribute.variable_name,
+            attribute2.attribute.variable_name
+        ]
+        request = self._get_measurement_request_point(
+            attributes=','.join(attribs)
+        )
+        request.user = self.user_1
+        response = view(request)
+        self.assertEqual(response.status_code, 403)
+        self.assertIn('Missing Permission', response.data)
+        self.assertIn(
+            dataset.type.variable_name,
+            response.data['Missing Permission']
+        )
+        mocked_reader.assert_not_called()
+
+        # add permission to user
+        assign_perm(
+            PermissionType.VIEW_DATASET_TYPE,
+            self.user_1,
+            dataset.type
+        )
+        request = self._get_measurement_request_point(
+            attributes=','.join(attribs)
+        )
+        request.user = self.user_1
+        response = view(request)
+        self.assertEqual(response.status_code, 200)
+        mocked_reader.assert_called_once_with(attribute1.dataset)
+        self.assertIn('metadata', response.data)
+        self.assertIn('results', response.data)
+        results = response.data['results']
+        self.assertEqual(len(results), 1)

django_project/permission/admin.py

@@ -0,0 +1,52 @@
+"""
+Tomorrow Now GAP.
+
+.. note:: Admin for permission
+"""
+from django.contrib import admin
+from django.contrib.auth.models import Permission
+
+from permission.models import (
+    DatasetTypeGroupObjectPermission,
+    DatasetTypeUserObjectPermission,
+    PermissionType
+)
+
+
+def _view_dataset_type_permission():
+    """Fetch permission to view dataset_type."""
+    return Permission.objects.filter(
+        codename=PermissionType.VIEW_DATASET_TYPE
+    ).first()
+
+
+@admin.register(DatasetTypeGroupObjectPermission)
+class DatasetTypeGroupObjectPermissionAdmin(admin.ModelAdmin):
+    """DatasetTypeGroupObjectPermission admin."""
+
+    list_display = ('group', 'content_object',)
+
+    def get_form(self, request, obj=None, **kwargs):
+        """Override the permission dropdown."""
+        form = super().get_form(request, obj, **kwargs)
+        form.base_fields['permission'].disabled = True
+        form.base_fields['permission'].initial = (
+            _view_dataset_type_permission()
+        )
+        return form
+
+
+@admin.register(DatasetTypeUserObjectPermission)
+class DatasetTypeUserObjectPermissionAdmin(admin.ModelAdmin):
+    """DatasetTypeUserObjectPermission admin."""
+
+    list_display = ('user', 'content_object',)
+
+    def get_form(self, request, obj=None, **kwargs):
+        """Override the permission dropdown."""
+        form = super().get_form(request, obj, **kwargs)
+        form.base_fields['permission'].disabled = True
+        form.base_fields['permission'].initial = (
+            _view_dataset_type_permission()
+        )
+        return form

django_project/permission/apps.py

@@ -0,0 +1,17 @@
+"""
+Tomorrow Now GAP.
+
+.. note:: Config for Permission Module
+"""
+from django.apps import AppConfig
+
+
+class PermissionConfig(AppConfig):
+    """Config for Permission App."""
+
+    default_auto_field = 'django.db.models.BigAutoField'
+    name = 'permission'
+
+    def ready(self):
+        """App ready handler."""
+        from permission.group_default_permission import group_gap_default, post_save_user_signal_handler  # noqa

django_project/permission/group_default_permission.py

@@ -0,0 +1,25 @@
+"""
+Tomorrow Now GAP.
+
+.. note:: Group for default permission
+"""
+from django.contrib.auth.models import Group
+from django.contrib.auth import get_user_model
+from django.db.models.signals import post_save
+from django.dispatch import receiver
+
+
+User = get_user_model()
+
+
+def group_gap_default():
+    """Return group GAP Default."""
+    group, _ = Group.objects.get_or_create(name='GAP Default')
+    return group
+
+
+@receiver(post_save, sender=User)
+def post_save_user_signal_handler(sender, instance, created, **kwargs):
+    """Assign gap_default group to new user."""
+    if created:
+        instance.groups.add(group_gap_default())

django_project/permission/migrations/0001_initial.py

@@ -0,0 +1,49 @@
+# Generated by Django 4.2.7 on 2024-12-23 11:31
+
+from django.conf import settings
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+        ('gap', '0043_cropgrowthstage_cropstagetype_and_more'),
+        ('auth', '0012_alter_user_first_name_max_length'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='DatasetTypeUserObjectPermission',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('content_object', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='gap.datasettype')),
+                ('permission', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='auth.permission')),
+                ('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL)),
+            ],
+            options={
+                'verbose_name': 'User Product Type Permission',
+                'db_table': 'permission_dataset_type_user',
+                'abstract': False,
+                'unique_together': {('user', 'permission', 'content_object')},
+            },
+        ),
+        migrations.CreateModel(
+            name='DatasetTypeGroupObjectPermission',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('content_object', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='gap.datasettype')),
+                ('group', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='auth.group')),
+                ('permission', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='auth.permission')),
+            ],
+            options={
+                'verbose_name': 'Group Product Type Permission',
+                'db_table': 'permission_dataset_type_group',
+                'abstract': False,
+                'unique_together': {('group', 'permission', 'content_object')},
+            },
+        ),
+    ]

django_project/permission/models.py

@@ -0,0 +1,37 @@
+"""
+Tomorrow Now GAP.
+
+.. note:: Models for permission
+"""
+from django.db import models
+from django.utils.translation import gettext_lazy as _
+from guardian.models import UserObjectPermissionBase
+from guardian.models import GroupObjectPermissionBase
+
+from gap.models import DatasetType
+
+
+class PermissionType:
+    """Enum that represents the permission type."""
+
+    VIEW_DATASET_TYPE = 'view_datasettype'
+
+
+class DatasetTypeUserObjectPermission(UserObjectPermissionBase):
+    """Model for storing DatasetType Object Permission by user."""
+
+    content_object = models.ForeignKey(DatasetType, on_delete=models.CASCADE)
+
+    class Meta(UserObjectPermissionBase.Meta):  # noqa
+        db_table = 'permission_dataset_type_user'
+        verbose_name = _('User Product Type Permission')
+
+
+class DatasetTypeGroupObjectPermission(GroupObjectPermissionBase):
+    """Model for storing DatasetType Object Permission by group."""
+
+    content_object = models.ForeignKey(DatasetType, on_delete=models.CASCADE)
+
+    class Meta(GroupObjectPermissionBase.Meta):  # noqa
+        db_table = 'permission_dataset_type_group'
+        verbose_name = _('Group Product Type Permission')