Framework for User Permissions & Roles

.envs/.local/.django

@@ -2,6 +2,7 @@
 DJANGO_SETTINGS_MODULE="config.settings.local"
 USE_DOCKER=yes
 DJANGO_DEBUG=True
+DJANGO_LOG_LEVEL=INFO
 IPYTHONDIR=/app/.ipython
 
 # Default superuser for local development

ami/base/models.py

@@ -9,6 +9,10 @@ class BaseModel(models.Model):
     created_at = models.DateTimeField(auto_now_add=True)
     updated_at = models.DateTimeField(auto_now=True)
 
+    def get_project(self):
+        """Get the project associated with the model."""
+        return self.project if hasattr(self, "project") else None
+
     def __str__(self) -> str:
         """All django models should have this method."""
         if hasattr(self, "name"):

ami/base/pagination.py

@@ -5,8 +5,27 @@
 
 class LimitOffsetPaginationWithPermissions(LimitOffsetPagination):
     def get_paginated_response(self, data):
+        model = self._get_current_model()
+        project = self._get_project()
         paginated_response = super().get_paginated_response(data=data)
         paginated_response.data = add_collection_level_permissions(
-            user=self.request.user, response_data=paginated_response.data
+            user=self.request.user, response_data=paginated_response.data, model=model, project=project
         )
         return paginated_response
+
+    def _get_current_model(self):
+        """
+        Retrieve the current model from the view.
+        """
+        view = self.request.parser_context.get("view")
+        if view and hasattr(view, "queryset"):
+            queryset = view.queryset
+            if queryset is not None:
+                return queryset.model
+        return None
+
+    def _get_project(self):
+        view = self.request.parser_context.get("view")
+        if hasattr(view, "get_active_project"):
+            return view.get_active_project()
+        return None

ami/base/permissions.py

@@ -1,11 +1,16 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+import logging
 
+from django.contrib.auth.models import AbstractBaseUser, AnonymousUser, User
+from guardian.shortcuts import get_perms
 from rest_framework import permissions
 
-if TYPE_CHECKING:
-    from django.contrib.auth.models import User
+from ami.jobs.models import Job
+from ami.main.models import BaseModel, Deployment, Device, Project, S3StorageSource, Site, SourceImageCollection
+from ami.users.roles import ProjectManager
+
+logger = logging.getLogger(__name__)
 
 
 def is_active_staff(user: User) -> bool:
@@ -28,32 +33,214 @@ def has_permission(self, request, view):
         )
 
 
-def add_object_level_permissions(user: User | None, response_data: dict) -> dict:
-    """
-    Add placeholder permissions to detail views and nested objects.
+def filter_permissions(permissions, model_name):
+    """Filter and extract only the action part of `action_modelname`"""
 
-    If the user is logged in, they can edit any object type.
-    If the user is a superuser, they can delete any object type.
+    filtered_permissions = {
+        perm.split("_")[0]  # Extract "action" from "action_modelname"
+        for perm in permissions
+        if perm.endswith(f"_{model_name}")  # Ensure it matches the model
+    }
+    return filtered_permissions
 
-    @TODO @IMPORTANT At least check if they are the owner of the project.
+
+def add_object_level_permissions(
+    user: AbstractBaseUser | AnonymousUser, instance: BaseModel, response_data: dict
+) -> dict:
+    """
+    Adds object-level permissions to the response data for a given user and instance.
+    This function updates the `response_data` dictionary with the permissions that the
+    specified `user` has on the given `instance`'s project.
     """
+
     permissions = response_data.get("user_permissions", set())
-    if user and is_active_staff(user):
-        permissions.update(["update"])
-        if user.is_superuser:
-            permissions.update(["delete"])
+    project = instance.get_project() if hasattr(instance, "get_project") else None
+    model_name = instance._meta.model_name  # Get model name
+    if user and user.is_superuser:
+        permissions.update(["update", "delete"])
+
+    if project:
+        user_permissions = get_perms(user, project)
+        # Filter and extract only the action part of "action_modelname" based on instance type
+        filtered_permissions = filter_permissions(permissions=user_permissions, model_name=model_name)
+        # Do not return create, view permissions at object-level
+        filtered_permissions -= {"create", "view"}
+        permissions.update(filtered_permissions)
     response_data["user_permissions"] = permissions
     return response_data
 
 
-def add_collection_level_permissions(user: User | None, response_data: dict) -> dict:
-    """
-    Add placeholder permissions to list view responses.
+def add_collection_level_permissions(user: User | None, response_data: dict, model, project) -> dict:
+    """Add collection-level permissions to the response data for a list view.
 
-    If the user is logged in, they can create new objects of any type.
+    This function modifies the `response_data` dictionary to include user permissions
+    for creating new objects of the specified model type. If the user is logged in and
+    is an active staff member, or if the user has create_model permission, the
+    "create" permission is added to the `user_permissions` set in the `response_data`.
     """
+
+    logger.info(f"add_collection_level_permissions model {model.__name__}, {type(model)} ")
     permissions = response_data.get("user_permissions", set())
-    if user and is_active_staff(user):
+    if user and user.is_superuser:
+        permissions.add("create")
+
+    if user and project and f"create_{model.__name__.lower()}" in get_perms(user, project):
         permissions.add("create")
     response_data["user_permissions"] = permissions
     return response_data
+
+
+class CRUDPermission(permissions.BasePermission):
+    """
+    Generic CRUD permission class that dynamically checks user permissions on an object.
+    Permission names follow the convention: `create_<model>`, `update_<model>`, `delete_<model>`.
+    """
+
+    model = None
+
+    def has_permission(self, request, view):
+        """Handles general permission checks"""
+
+        return True  # Fallback to object level permissions
+
+    def has_object_permission(self, request, view, obj):
+        """Handles object-level permission checks."""
+        model_name = self.model._meta.model_name
+        project = obj.get_project() if hasattr(obj, "get_project") else None
+        # check for create action
+        if view.action == "create":
+            return request.user.is_superuser or request.user.has_perm(f"create_{model_name}", project)
+
+        if view.action == "retrieve":
+            return True  # Allow all users to view objects
+
+        # Map ViewSet actions to permission names
+        action_perms = {
+            "retrieve": f"view_{model_name}",
+            "update": f"update_{model_name}",
+            "partial_update": f"update_{model_name}",
+            "destroy": f"delete_{model_name}",
+        }
+
+        required_perm = action_perms.get(view.action)
+        if not required_perm:
+            return True
+        return request.user.has_perm(required_perm, project)
+
+
+class ProjectCRUDPermission(CRUDPermission):
+    model = Project
+
+
+class JobCRUDPermission(CRUDPermission):
+    model = Job
+
+
+class DeploymentCRUDPermission(CRUDPermission):
+    model = Deployment
+
+
+class SourceImageCollectionCRUDPermission(CRUDPermission):
+    model = SourceImageCollection
+
+
+class CanStarSourceImage(permissions.BasePermission):
+    """Custom permission to check if the user can star a Source image."""
+
+    permission = Project.Permissions.STAR_SOURCE_IMAGE
+
+    def has_object_permission(self, request, view, obj):
+        if view.action in ["unstar", "star"]:
+            project = obj.get_project() if hasattr(obj, "get_project") else None
+            return request.user.has_perm(self.permission, project)
+        return True
+
+
+class S3StorageSourceCRUDPermission(CRUDPermission):
+    model = S3StorageSource
+
+
+class SiteCRUDPermission(CRUDPermission):
+    model = Site
+
+
+class DeviceCRUDPermission(CRUDPermission):
+    model = Device
+
+
+# Identification permission checks
+class CanUpdateIdentification(permissions.BasePermission):
+    """Custom permission to check if the user can update/create an identification."""
+
+    permission = Project.Permissions.UPDATE_IDENTIFICATION
+
+    def has_object_permission(self, request, view, obj):
+        if view.action in ["create", "update", "partial_update"]:
+            project = obj.get_project() if hasattr(obj, "get_project") else None
+            return request.user.has_perm(self.permission, project)
+        return True
+
+
+class CanDeleteIdentification(permissions.BasePermission):
+    """Custom permission to check if the user can delete an identification."""
+
+    permission = Project.Permissions.DELETE_IDENTIFICATION
+
+    def has_object_permission(self, request, view, obj):
+        project = obj.get_project() if hasattr(obj, "get_project") else None
+        # Check if user is superuser or staff or project manager
+        if view.action == "destroy":
+            if request.user.is_superuser or ProjectManager.has_role(request.user, project):
+                return True
+            # Check if the user is the owner of the object
+            return obj.user == request.user
+        return True
+
+
+# Job run permission check
+class CanRunJob(permissions.BasePermission):
+    """Custom permission to check if the user can run a job."""
+
+    permission = Project.Permissions.RUN_JOB
+
+    def has_object_permission(self, request, view, obj):
+        if view.action == "run":
+            project = obj.get_project() if hasattr(obj, "get_project") else None
+            return request.user.has_perm(self.permission, project)
+        return True
+
+
+class CanRetryJob(permissions.BasePermission):
+    """Custom permission to check if the user can retry a job."""
+
+    permission = Project.Permissions.RETRY_JOB
+
+    def has_object_permission(self, request, view, obj):
+        if view.action == "retry":
+            project = obj.get_project() if hasattr(obj, "get_project") else None
+            return request.user.has_perm(self.permission, project)
+        return True
+
+
+class CanCancelJob(permissions.BasePermission):
+    """Custom permission to check if the user can cancel a job."""
+
+    permission = Project.Permissions.CANCEL_JOB
+
+    def has_object_permission(self, request, view, obj):
+        if view.action == "cancel":
+            project = obj.get_project() if hasattr(obj, "get_project") else None
+            return request.user.has_perm(self.permission, project)
+        return True
+
+
+class CanPopulateSourceImageCollection(permissions.BasePermission):
+    """Custom permission to check if the user can populate a collection."""
+
+    permission = Project.Permissions.POPULATE_COLLECTION
+
+    def has_object_permission(self, request, view, obj):
+        if view.action == "populate":
+            project = obj.get_project() if hasattr(obj, "get_project") else None
+            return request.user.has_perm(self.permission, project)
+        return True

ami/base/serializers.py

@@ -1,3 +1,4 @@
+import logging
 import typing
 import urllib.parse
 
@@ -9,6 +10,8 @@
 
 from .permissions import add_object_level_permissions
 
+logger = logging.getLogger(__name__)
+
 
 def reverse_with_params(viewname: str, args=None, kwargs=None, request=None, params: dict = {}, **extra) -> str:
     query_string = urllib.parse.urlencode(params)
@@ -40,14 +43,19 @@ class DefaultSerializer(serializers.HyperlinkedModelSerializer):
     url_field_name = "details"
     id = serializers.IntegerField(read_only=True)
 
-    def get_permissions(self, instance_data):
-        request = self.context.get("request")
-        user = request.user if request else None
-        return add_object_level_permissions(user, instance_data)
+    def get_permissions(self, instance, instance_data):
+        request: Request = self.context["request"]
+        user = request.user
+
+        return add_object_level_permissions(
+            user=user,
+            instance=instance,
+            response_data=instance_data,
+        )
 
     def to_representation(self, instance):
         instance_data = super().to_representation(instance)
-        instance_data = self.get_permissions(instance_data)
+        instance_data = self.get_permissions(instance=instance, instance_data=instance_data)
         return instance_data
 
 

ami/base/views.py

@@ -0,0 +1,38 @@
+import logging
+
+from django.shortcuts import get_object_or_404
+from rest_framework import serializers
+
+from ami.main.models import Project
+
+logger = logging.getLogger(__name__)
+
+
+class ProjectMixin:
+    """
+    Mixin to handle project_id fetching from query parameters or URL parameters.
+    By default, project_id is required, but this can be overridden.
+    """
+
+    require_project = False  # Project is optional
+
+    def get_active_project(self):
+        from ami.base.serializers import SingleParamSerializer
+
+        project_id = None
+        # Extract from URL `/projects/` is in the url path
+        if "/projects/" in self.request.path:
+            project_id = self.kwargs.get("pk")
+
+        # If not in URL, try query parameters
+        if not project_id:
+            if self.require_project:
+                project_id = SingleParamSerializer[int].clean(
+                    param_name="project_id",
+                    field=serializers.IntegerField(required=True, min_value=0),
+                    data=self.request.query_params,
+                )
+            else:
+                project_id = self.request.query_params.get("project_id")  # No validation
+
+        return get_object_or_404(Project, id=project_id) if project_id else None