wip: Consolidate access control logic with django-guardian

setup.py

@@ -42,6 +42,7 @@
         'django-configurations[database,email]==2.5.1',
         'django-extensions==3.2.3',
         'django-filter==24.3',
+        'django-guardian==2.4.0',
         'django-oauth-toolkit==2.4.0',
         'djangorestframework==3.15.2',
         'django-large-image==0.10.0',

uvdat/core/models/chart.py

@@ -13,18 +13,6 @@ class Chart(models.Model):
     chart_options = models.JSONField(blank=True, null=True)
     editable = models.BooleanField(default=False)
 
-    def is_in_project(self, project_id):
-        return self.project.id == project_id
-
-    def readable_by(self, user):
-        return self.project.readable_by(user)
-
-    def editable_by(self, user):
-        return self.project.editable_by(user)
-
-    def deletable_by(self, user):
-        return self.project.owned_by(user)
-
     def spawn_conversion_task(
         self,
         conversion_options=None,

uvdat/core/models/dataset.py

@@ -16,18 +16,6 @@ class DatasetType(models.TextChoices):
         choices=DatasetType.choices,
     )
 
-    def is_in_project(self, project_id):
-        return self.project_set.filter(id=project_id).exists()
-
-    def readable_by(self, user):
-        return True
-
-    def editable_by(self, user):
-        return user.is_superuser
-
-    def deletable_by(self, user):
-        return user.is_superuser
-
     def spawn_conversion_task(
         self,
         style_options=None,

uvdat/core/models/file_item.py

@@ -18,18 +18,6 @@ class FileItem(TimeStampedModel):
     metadata = models.JSONField(blank=True, null=True)
     index = models.IntegerField(null=True)
 
-    def is_in_project(self, project_id):
-        return self.dataset.is_in_project(project_id)
-
-    def readable_by(self, user):
-        return self.dataset.readable_by(user)
-
-    def editable_by(self, user):
-        return self.dataset.editable_by(user)
-
-    def deletable_by(self, user):
-        return self.dataset.deletable_by(user)
-
     def download(self):
         # TODO: download
         pass

uvdat/core/models/map_layers.py

@@ -18,18 +18,6 @@ class AbstractMapLayer(TimeStampedModel):
     default_style = models.JSONField(blank=True, null=True)
     index = models.IntegerField(null=True)
 
-    def is_in_project(self, project_id):
-        return self.dataset.is_in_project(project_id)
-
-    def readable_by(self, user):
-        return self.dataset.readable_by(user)
-
-    def editable_by(self, user):
-        return self.dataset.editable_by(user)
-
-    def deletable_by(self, user):
-        return self.dataset.deletable_by(user)
-
     class Meta:
         abstract = True
 

uvdat/core/models/networks.py

@@ -10,18 +10,6 @@ class Network(models.Model):
     category = models.CharField(max_length=25)
     metadata = models.JSONField(blank=True, null=True)
 
-    def is_in_project(self, project_id):
-        return self.dataset.is_in_project(project_id)
-
-    def readable_by(self, user):
-        return self.dataset.readable_by(user)
-
-    def editable_by(self, user):
-        return self.dataset.editable_by(user)
-
-    def deletable_by(self, user):
-        return self.dataset.deletable_by(user)
-
     def get_graph(self):
         from uvdat.core.tasks.networks import get_network_graph
 
@@ -43,18 +31,6 @@ class NetworkNode(models.Model):
     capacity = models.IntegerField(null=True)
     location = geo_models.PointField()
 
-    def is_in_project(self, project_id):
-        return self.network.is_in_project(project_id)
-
-    def readable_by(self, user):
-        return self.network.readable_by(user)
-
-    def editable_by(self, user):
-        return self.network.editable_by(user)
-
-    def deletable_by(self, user):
-        return self.network.deletable_by(user)
-
     def get_adjacent_nodes(self) -> models.QuerySet:
         entering_node_ids = (
             NetworkEdge.objects.filter(to_node=self.id)
@@ -80,15 +56,3 @@ class NetworkEdge(models.Model):
     directed = models.BooleanField(default=False)
     from_node = models.ForeignKey(NetworkNode, related_name='+', on_delete=models.CASCADE)
     to_node = models.ForeignKey(NetworkNode, related_name='+', on_delete=models.CASCADE)
-
-    def is_in_project(self, project_id):
-        return self.network.is_in_project(project_id)
-
-    def readable_by(self, user):
-        return self.network.readable_by(user)
-
-    def editable_by(self, user):
-        return self.network.editable_by(user)
-
-    def deletable_by(self, user):
-        return self.network.deletable_by(user)

uvdat/core/models/project.py

@@ -14,16 +14,9 @@ class Project(models.Model):
     collaborators = models.ManyToManyField(User, related_name='read_write_projects')
     followers = models.ManyToManyField(User, related_name='read_only_projects')
 
-    def readable_by(self, user):
-        return (
-            user.is_superuser
-            or user == self.owner
-            or user in self.collaborators.all()
-            or user in self.followers.all()
-        )
-
-    def editable_by(self, user):
-        return user.is_superuser or user == self.owner or user in self.collaborators.all()
-
-    def deletable_by(self, user):
-        return user.is_superuser or user == self.owner
+    class Meta:
+        permissions = [
+            ('owner', 'Can read, write, and delete'),
+            ('collaborator', 'Can read and write'),
+            ('follower', 'Can read'),
+        ]

uvdat/core/models/regions.py

@@ -12,18 +12,6 @@ class SourceRegion(models.Model):
     metadata = models.JSONField(blank=True, null=True)
     boundary = geo_models.MultiPolygonField()
 
-    def is_in_project(self, project_id):
-        return self.dataset.is_in_project(project_id)
-
-    def readable_by(self, user):
-        return self.dataset.readable_by(user)
-
-    def editable_by(self, user):
-        return self.dataset.editable_by(user)
-
-    def deletable_by(self, user):
-        return self.dataset.deletable_by(user)
-
     class Meta:
         constraints = [
             # We enforce name uniqueness across datasets
@@ -54,18 +42,6 @@ class VectorOperation(models.TextChoices):
     # They need their own reference to a map representation
     map_layer = models.ForeignKey(VectorMapLayer, on_delete=models.PROTECT)
 
-    def is_in_project(self, project_id):
-        return self.project.id == int(project_id)
-
-    def readable_by(self, user):
-        return self.project.readable_by(user)
-
-    def editable_by(self, user):
-        return self.project.editable_by(user)
-
-    def deletable_by(self, user):
-        return self.project.deletable_by(user)
-
     def get_map_layers(self):
         return [
             {

uvdat/core/models/simulations.py

@@ -24,18 +24,6 @@ class SimulationType(models.TextChoices):
     output_data = models.JSONField(blank=True, null=True)
     error_message = models.TextField(null=True, blank=True)
 
-    def is_in_project(self, project_id):
-        return self.project.id == int(project_id)
-
-    def readable_by(self, user):
-        return self.project.readable_by(user)
-
-    def editable_by(self, user):
-        return self.project.editable_by(user)
-
-    def deletable_by(self, user):
-        return self.project.deletable_by(user)
-
     def get_simulation_type(self):
         if not self.simulation_type or self.simulation_type not in AVAILABLE_SIMULATIONS:
             raise ValueError(f'Simulation type not found: {self.simulation_type}')

uvdat/core/rest/chart.py

@@ -3,14 +3,15 @@
 from rest_framework.viewsets import ModelViewSet
 
 from uvdat.core.models import Chart
-from uvdat.core.rest.filter import AccessControl
+from uvdat.core.rest.guardian import GuardianFilter, GuardianPermission
 from uvdat.core.rest.serializers import ChartSerializer
 
 
 class ChartViewSet(ModelViewSet):
     queryset = Chart.objects.all()
     serializer_class = ChartSerializer
-    filter_backends = [AccessControl]
+    permission_classes = [GuardianPermission]
+    filter_backends = [GuardianFilter]
     lookup_field = 'id'
 
     def validate_editable(self, chart, func, *args, **kwargs):

uvdat/core/rest/dataset.py

@@ -6,7 +6,7 @@
 from rest_framework.viewsets import ModelViewSet
 
 from uvdat.core.models import Dataset, NetworkEdge, NetworkNode
-from uvdat.core.rest.filter import AccessControl
+from uvdat.core.rest.guardian import GuardianFilter, GuardianPermission
 from uvdat.core.rest.serializers import (
     DatasetSerializer,
     NetworkEdgeSerializer,
@@ -20,7 +20,8 @@
 class DatasetViewSet(ModelViewSet):
     queryset = Dataset.objects.all()
     serializer_class = DatasetSerializer
-    filter_backends = [AccessControl]
+    permission_classes = [GuardianPermission]
+    filter_backends = [GuardianFilter]
     lookup_field = 'id'
 
     @action(detail=True, methods=['get'])

uvdat/core/rest/file_item.py

@@ -1,12 +1,13 @@
 from rest_framework.viewsets import ModelViewSet
 
 from uvdat.core.models import FileItem
-from uvdat.core.rest.filter import AccessControl
+from uvdat.core.rest.guardian import GuardianFilter, GuardianPermission
 from uvdat.core.rest.serializers import FileItemSerializer
 
 
 class FileItemViewSet(ModelViewSet):
     queryset = FileItem.objects.all()
     serializer_class = FileItemSerializer
-    filter_backends = [AccessControl]
+    permission_classes = [GuardianPermission]
+    filter_backends = [GuardianFilter]
     lookup_field = 'id'

uvdat/core/rest/guardian.py

@@ -0,0 +1,51 @@
+from guardian.shortcuts import get_objects_for_user
+from rest_framework.filters import BaseFilterBackend
+from rest_framework.permissions import SAFE_METHODS, IsAuthenticated
+
+from uvdat.core import models
+
+
+class GuardianPermission(IsAuthenticated):
+    def get_object_queryset(self, obj):
+        if isinstance(obj, models.Project):
+            return obj
+        elif isinstance(models.Dataset):
+            return obj.project_set
+        elif (
+            isinstance(obj.models.Chart)
+            or isinstance(obj, models.SimulationResult)
+            or isinstance(obj, models.DerivedRegion)
+        ):
+            return obj.project
+        elif (
+            isinstance(obj, models.FileItem)
+            or isinstance(obj, models.VectorMapLayer)
+            or isinstance(obj, models.RasterMapLayer)
+            or isinstance(obj, models.Network)
+            or isinstance(obj, models.SourceRegion)
+        ):
+            return obj.dataset.project_set
+        elif isinstance(obj, models.NetworkEdge) or isinstance(obj, models.NetworkNode):
+            return obj.network.dataset.project_set
+
+    def has_object_permission(self, request, view, obj):
+        if request.user.is_superuser:
+            return True
+
+        perms = ['follower']
+        if request.method not in SAFE_METHODS:
+            perms.append('collaborator')
+        if request.method == 'DELETE':
+            perms.append('owner')
+        return request.user.has_perm(perms, self.get_object_queryset(obj))
+
+
+class GuardianFilter(BaseFilterBackend):
+    def filter_queryset(self, request, queryset, view):
+        if request.user.is_superuser:
+            pass
+        return get_objects_for_user(
+            klass=queryset,
+            user=request.user,
+            perms=['follower', 'collaborator', 'owner'],
+        )

uvdat/core/rest/map_layers.py

@@ -8,7 +8,7 @@
 from rest_framework.viewsets import ModelViewSet
 
 from uvdat.core.models import RasterMapLayer, VectorMapLayer
-from uvdat.core.rest.filter import AccessControl
+from uvdat.core.rest.guardian import GuardianFilter, GuardianPermission
 from uvdat.core.rest.serializers import (
     RasterMapLayerSerializer,
     VectorMapLayerDetailSerializer,
@@ -73,7 +73,8 @@
 class RasterMapLayerViewSet(ModelViewSet, LargeImageFileDetailMixin):
     queryset = RasterMapLayer.objects.select_related('dataset').all()
     serializer_class = RasterMapLayerSerializer
-    filter_backends = [AccessControl]
+    permission_classes = [GuardianPermission]
+    filter_backends = [GuardianFilter]
     lookup_field = 'id'
     FILE_FIELD_NAME = 'cloud_optimized_geotiff'
 
@@ -92,7 +93,8 @@ def get_raster_data(self, request, resolution: str = '1', **kwargs):
 class VectorMapLayerViewSet(ModelViewSet):
     queryset = VectorMapLayer.objects.select_related('dataset').all()
     serializer_class = VectorMapLayerSerializer
-    filter_backends = [AccessControl]
+    permission_classes = [GuardianPermission]
+    filter_backends = [GuardianFilter]
     lookup_field = 'id'
 
     def retrieve(self, request, *args, **kwargs):