Permissions

backend/organizations/decorators.py

@@ -2,6 +2,9 @@
 from rest_framework.response import Response
 from .models import Organization
 from functools import wraps
+from django.http import HttpResponse
+from workspaces.models import Workspace
+
 
 PERMISSION_ERROR = {
     "message": "You do not have enough permissions to access this view!"
@@ -44,3 +47,51 @@ def wrapper(self, request, pk=None, *args, **kwargs):
             return Response(PERMISSION_ERROR, status=403)
 
     return wrapper
+
+
+def is_admin(f):
+    @wraps(f)
+    def wrapper(self, request, *args, **kwargs):
+        if request.user.is_authenticated and (
+            request.user.role == User.ADMIN or request.user.is_superuser
+        ):
+            return f(self, request, *args, **kwargs)
+        return Response("Permission Denied", status=403)
+
+    return wrapper
+
+
+def is_permitted(f):
+    @wraps(f)
+    def wrapper(self, request, *args, **kwargs):
+        if "organization" not in request.data or "workspace" not in request.data:
+            return Response(
+                {
+                    "message": "Please send the complete request data for organization and workspace"
+                },
+                status=403,
+            )
+        organization = Organization.objects.get(id=request.data["organization"])
+        workspace = Workspace.objects.get(id=request.data["workspace"])
+        if Organization.objects.filter(
+            id=request.user.organization.id
+        ) != Organization.objects.filter(id=int(organization)):
+            return Response(NO_ORGANIZATION_OWNER_ERROR, status=403)
+        if workspace.organization != request.user.organization:
+            Response(NO_ORGANIZATION_OWNER_ERROR, status=403)
+        org_permissions = Organization.objects.filter(
+            id=request.user.organization.id
+        ).permission_json
+        requested_permission = request.data.get("requested_permission")
+        allowed_roles = org_permissions.get(requested_permission, 0)
+        if not allowed_roles:
+            return Response({"message": "Requested Permission is invalid"}, status=403)
+        for a in allowed_roles:
+            if (a == "org_owner" and request.user.role != User.ORGANIZATION_OWNER) or (
+                a == "workspace_manager" and request.user not in workspace.managers
+            ):
+                return Response({"message": "Access Denied"}, status=403)
+            return f(self, request, *args, **kwargs)
+        return Response(PERMISSION_ERROR, status=403)
+
+    return wrapper

backend/organizations/migrations/0009_permission_json_organizations.py

@@ -0,0 +1,24 @@
+# Generated by Django 3.2.14 on 2024-07-31 10:12
+
+from django.db import migrations, models
+import organizations.models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("organizations", "0008_auto_20220930_0451"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="organization",
+            name="permission_json",
+            field=models.JSONField(
+                blank=True,
+                default=organizations.models.default_permissions,
+                help_text="Permissions for user role",
+                null=True,
+                verbose_name="permission json",
+            ),
+        ),
+    ]

backend/organizations/models.py

@@ -13,7 +13,65 @@
 
 from django.conf import settings
 
+
 # Create your models here.
+def default_permissions():
+    return {
+        "PROJECT_PERMISSIONS": {
+            "can_view_add_annotators_to_project": ["org_owner", "workspace_manager"],
+            "can_use_add_annotators_to_project": ["org_owner", "workspace_manager"],
+            "can_view_add_reviewers_to_project": ["org_owner", "workspace_manager"],
+            "can_use_add_reviewers_to_project": ["org_owner", "workspace_manager"],
+            "can_view_add_superchecker_to_project": ["org_owner", "workspace_manager"],
+            "can_use_add_superchecker_to_project": ["org_owner", "workspace_manager"],
+            "can_view_basic_project_settings": ["org_owner", "workspace_manager"],
+            "can_use_basic_project_settings": ["org_owner", "workspace_manager"],
+            "can_view_publish_project": ["org_owner", "workspace_manager"],
+            "can_use_publish_project": ["org_owner", "workspace_manager"],
+            "can_view_archive_project": ["org_owner", "workspace_manager"],
+            "can_use_archive_project": ["org_owner", "workspace_manager"],
+            "can_view_export_project_into_dataset": ["org_owner", "workspace_manager"],
+            "can_use_export_project_into_dataset": ["org_owner", "workspace_manager"],
+            "can_view_pull_new_data_items_from_source_dataset": [
+                "org_owner",
+                "workspace_manager",
+            ],
+            "can_use_pull_new_data_items_from_source_dataset": [
+                "org_owner",
+                "workspace_manager",
+            ],
+            "can_view_download_project": ["org_owner", "workspace_manager"],
+            "can_use_download_project": ["org_owner", "workspace_manager"],
+            "can_view_delete_project_tasks": ["org_owner", "workspace_manager"],
+            "can_use_delete_project_tasks": ["org_owner", "workspace_manager"],
+            "can_view_deallocate_user_tasks": ["org_owner", "workspace_manager"],
+            "can_use_deallocate_user_tasks": ["org_owner", "workspace_manager"],
+            "can_view_project_stage": ["org_owner", "workspace_manager"],
+            "can_use_project_stage": ["org_owner", "workspace_manager"],
+            "can_view_supercheck_settings": ["org_owner", "workspace_manager"],
+            "can_use_supercheck_settings": ["org_owner", "workspace_manager"],
+            "can_view_user_profile_details_of_other_users": [
+                "org_owner",
+                "workspace_manager",
+            ],
+            "can_access_user_profile_details_of_other_users": [
+                "org_owner",
+                "workspace_manager",
+            ],
+        },
+        "DATASET_PERMISSIONS": {
+            "can_view_basic_dataset_settings": ["org_owner", "workspace_manager"],
+            "can_use_basic_dataset_settings": ["org_owner", "workspace_manager"],
+            "can_view_download_dataset": ["org_owner", "workspace_manager"],
+            "can_use_download_dataset": ["org_owner", "workspace_manager"],
+            "can_view_upload_dataset": ["org_owner", "workspace_manager"],
+            "can_use_upload_dataset": ["org_owner", "workspace_manager"],
+            "can_view_delete_data_item": ["org_owner", "workspace_manager"],
+            "can_use_delete_data_item": ["org_owner", "workspace_manager"],
+            "can_view_deduplicate_data_items": ["org_owner", "workspace_manager"],
+            "can_use_deduplicate_data_items": ["org_owner", "workspace_manager"],
+        },
+    }
 
 
 class Organization(models.Model):
@@ -47,6 +105,13 @@ class Organization(models.Model):
 
     created_at = models.DateTimeField(verbose_name="created_at", auto_now_add=True)
     updated_at = models.DateTimeField(verbose_name="updated_at", auto_now=True)
+    permission_json = models.JSONField(
+        verbose_name="permission json",
+        null=True,
+        blank=True,
+        default=default_permissions,
+        help_text=("Permissions for user role"),
+    )
 
     def __str__(self):
         return self.title + ", id=" + str(self.pk)

backend/organizations/permissions.py

@@ -0,0 +1,137 @@
+from rest_framework.views import APIView
+from rest_framework.response import Response
+from rest_framework import status
+from .models import Organization
+from .decorators import is_admin
+
+
+class ProjectPermissionView(APIView):
+    def get(self, request, *args, **kwargs):
+        org = Organization.objects.get(id=request.user.organization.id)
+        project_permissions = org.permission_json["PROJECT_PERMISSIONS"]
+        permission_name = request.query_params.get("permission_name")
+        if permission_name is None:
+            return Response(
+                {"message": "Permission name is required"},
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+        pm = project_permissions.get(permission_name)
+        if pm is None:
+            return Response(
+                {"message": "Permission not found"}, status=status.HTTP_404_NOT_FOUND
+            )
+        return Response({"permission": list(pm)}, status=status.HTTP_200_OK)
+
+    @is_admin
+    def post(self, request, *args, **kwargs):
+        org = Organization.objects.get(id=request.user.organization.id)
+        project_permissions = org.permission_json["PROJECT_PERMISSIONS"]
+        permission_name = request.query_params.get("permission_name")
+        if permission_name is None:
+            return Response(
+                {"message": "Permission name is required"},
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+        new_roles = request.data.get("new_roles")
+        if not new_roles:
+            return Response(
+                {"message": "New Roles are required"},
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+        if permission_name in project_permissions:
+            project_permissions[permission_name].append(new_roles)
+        else:
+            project_permissions[permission_name] = [new_roles]
+        org.permission_json["PROJECT_PERMISSIONS"] = project_permissions
+        org.save()
+        return Response(
+            {"message": "Permission updated"},
+            status=status.HTTP_200_OK,
+        )
+
+    @is_admin
+    def delete(self, request, *args, **kwargs):
+        org = Organization.objects.get(id=request.user.organization.id)
+        project_permissions = org.permission_json["PROJECT_PERMISSIONS"]
+        permission_name = request.query_params.get("permission_name")
+        if permission_name is None:
+            return Response(
+                {"message": "Permission name is required"},
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+        if permission_name in project_permissions:
+            del project_permissions[permission_name]
+        else:
+            print(f"Permission '{permission_name}' not found")
+        org.permission_json["PROJECT_PERMISSIONS"] = project_permissions
+        org.save()
+        return Response(
+            {"message": "Permission deleted"},
+            status=status.HTTP_200_OK,
+        )
+
+
+class DatasetPermissionView(APIView):
+    def get(self, request, *args, **kwargs):
+        org = Organization.objects.get(id=request.user.organization.id)
+        dataset_permissions = org.permission_json["DATASET_PERMISSIONS"]
+        permission_name = request.query_params.get("permission_name")
+        if permission_name is None:
+            return Response(
+                {"message": "Permission name is required"},
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+        pm = dataset_permissions.get(permission_name)
+        if pm is None:
+            return Response(
+                {"message": "Permission not found"}, status=status.HTTP_404_NOT_FOUND
+            )
+        return Response({"permission": list(pm)}, status=status.HTTP_200_OK)
+
+    @is_admin
+    def post(self, request, *args, **kwargs):
+        org = Organization.objects.get(id=request.user.organization.id)
+        dataset_permissions = org.permission_json["DATASET_PERMISSIONS"]
+        permission_name = request.query_params.get("permission_name")
+        if permission_name is None:
+            return Response(
+                {"message": "Permission name is required"},
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+        new_roles = request.data.get("new_roles")
+        if not new_roles:
+            return Response(
+                {"message": "New Roles are required"},
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+        if permission_name in dataset_permissions:
+            dataset_permissions[permission_name].append(new_roles)
+        else:
+            dataset_permissions[permission_name] = [new_roles]
+        org.permission_json["DATASET_PERMISSIONS"] = dataset_permissions
+        org.save()
+        return Response(
+            {"message": "Permission updated"},
+            status=status.HTTP_200_OK,
+        )
+
+    @is_admin
+    def delete(self, request, *args, **kwargs):
+        org = Organization.objects.get(id=request.user.organization.id)
+        dataset_permissions = org.permission_json["DATASET_PERMISSIONS"]
+        permission_name = request.query_params.get("permission_name")
+        if permission_name is None:
+            return Response(
+                {"message": "Permission name is required"},
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+        if permission_name in dataset_permissions:
+            del dataset_permissions[permission_name]
+        else:
+            print(f"Permission '{permission_name}' not found")
+        org.permission_json["DATASET_PERMISSIONS"] = dataset_permissions
+        org.save()
+        return Response(
+            {"message": "Permission deleted"},
+            status=status.HTTP_200_OK,
+        )

backend/organizations/urls.py

@@ -9,3 +9,28 @@
 router.register(r"public", OrganizationPublicViewSet, basename="public")
 
 urlpatterns = router.urls
+
+
+from django.urls import path
+
+# from rest_framework.urlpatterns import format_suffix_patterns
+from .views import *
+from .permissions import *
+from django.urls import path
+
+
+urlpatterns = [
+    path(
+        "project_permission/",
+        ProjectPermissionView.as_view(),
+        name="project_permission",
+    ),
+    path(
+        "dataset_permission/",
+        DatasetPermissionView.as_view(),
+        name="dataset_permission",
+    ),
+]
+
+
+# urlpatterns = format_suffix_patterns(urlpatterns)