Extend permissions framework to support model-level permissions

ami/base/models.py

@@ -180,7 +180,38 @@ def _get_object_perms(self, user):
         object_perms = [perm for perm in all_perms if perm.endswith(f"_{model_name}")]
         return object_perms
 
+    def check_model_level_permission(self, user: AbstractUser | AnonymousUser, action: str) -> bool:
+        model = self._meta.model_name
+        app_label = "main"  # Assume all model level permissions are in 'main' app
+
+        crud_map = {
+            "create": f"{app_label}.create_{model}",
+            "update": f"{app_label}.update_{model}",
+            "partial_update": f"{app_label}.update_{model}",
+            "destroy": f"{app_label}.delete_{model}",
+            "retrieve": f"{app_label}.view_{model}",
+        }
+
+        perm = crud_map.get(action, f"{app_label}.{action}_{model}")
+        if action == "retrieve":
+            return True  # allow view permission for all users
+        return user.has_perm(perm)
+
     def check_permission(self, user: AbstractUser | AnonymousUser, action: str) -> bool:
+        """
+        Entry point for all permission checks.
+        Decides whether to perform model-level or object-level permission check.
+        """
+        # Get related project accessor
+        accessor = self.get_project_accessor()
+        if accessor is None or accessor == "projects":
+            # If there is no project relation, use model-level permission
+            return self.check_model_level_permission(user, action)
+
+        # If the object is linked to a project then use object-level permission
+        return self.check_object_level_permission(user, action)
+
+    def check_object_level_permission(self, user: AbstractUser | AnonymousUser, action: str) -> bool:
         """
         Check if the user has permission to perform the action
         on this instance.
@@ -189,14 +220,6 @@ def check_permission(self, user: AbstractUser | AnonymousUser, action: str) -> b
         """
         from ami.users.roles import BasicMember
 
-        project = self.get_project() if hasattr(self, "get_project") else None
-        if not project:
-            return False
-        if action == "retrieve":
-            if project.draft:
-                # Allow view permission for members and owners of draft projects
-                return BasicMember.has_role(user, project) or user == project.owner or user.is_superuser
-            return True
         model = self._meta.model_name
         crud_map = {
             "create": f"create_{model}",
@@ -205,14 +228,24 @@ def check_permission(self, user: AbstractUser | AnonymousUser, action: str) -> b
             "destroy": f"delete_{model}",
         }
 
+        project = self.get_project() if hasattr(self, "get_project") else None
+        if not project:
+            # No specific project instance found; fallback to model-level
+            return self.check_model_level_permission(user, action)
+        if action == "retrieve":
+            if project.draft:
+                # Allow view permission for members and owners of draft projects
+                return BasicMember.has_role(user, project) or user == project.owner or user.is_superuser
+            return True
+
         if action in crud_map:
             return user.has_perm(crud_map[action], project)
 
         # Delegate to model-specific logic
-        return self.check_custom_permission(user, action)
+        return self.check_custom_object_level_permission(user, action)
 
-    def check_custom_permission(self, user: AbstractUser | AnonymousUser, action: str) -> bool:
-        """Check custom permissions for the user on this instance.
+    def check_custom_object_level_permission(self, user: AbstractUser | AnonymousUser, action: str) -> bool:
+        """Check custom object level permissions for the user on this instance.
         This is used for actions that are not standard CRUD operations.
         """
         assert self._meta.model_name is not None, "Model must have a model_name defined in Meta class."
@@ -222,41 +255,91 @@ def check_custom_permission(self, user: AbstractUser | AnonymousUser, action: st
 
         return user.has_perm(permission_codename, project)
 
-    def get_user_object_permissions(self, user) -> list[str]:
+    def get_permissions(self, user: AbstractUser | AnonymousUser) -> list[str]:
         """
-        Returns a list of object-level permissions the user has on this instance.
-        This is used by frontend to determine what actions the user can perform.
+        Entry point for retrieving user permissions on this instance.
+        Decides whether to return model-level or object-level permissions.
+        """
+        accessor = self.get_project_accessor()
+
+        if accessor is None or accessor == "projects":
+            # M2M or no project relation, use model-level permissions
+            return self.get_model_level_permissions(user)
+
+        # Otherwise, get object-level permissions
+        return self.get_object_level_permissions(user)
+
+    def get_model_level_permissions(self, user: AbstractUser | AnonymousUser) -> list[str]:
+        """
+        Retrieve model-level permissions for the given user.
+        Returns a list of allowed actions such as ["create", "update", "delete"].
         """
-        # Return all permissions for superusers
         if user.is_superuser:
-            allowed_custom_actions = self.get_custom_user_permissions(user)
-            return ["update", "delete"] + allowed_custom_actions
+            # Superusers get all possible actions
+            return ["update", "delete", "view"]
+
+        model = self._meta.model_name
+        app_label = "main"  # self._meta.app_label
+        crud_map = {
+            "update": f"{app_label}.update_{model}",
+            "delete": f"{app_label}.delete_{model}",
+            "view": f"{app_label}.view_{model}",
+        }
+
+        allowed_actions = [action for action, perm in crud_map.items() if user.has_perm(perm)]
+        return allowed_actions
+
+    def get_object_level_permissions(self, user: AbstractUser | AnonymousUser) -> list[str]:
+        """
+        Retrieve object-level permissions (including custom ones) for this instance.
+        """
+
+        if user.is_superuser:
+            return ["update", "delete"] + self.get_custom_object_level_permissions(user)
+
+        project = self.get_project()
+        if not project:
+            # Fallback to model-level permissions if no related project found
+            return self.get_model_level_permissions(user)
 
         object_perms = self._get_object_perms(user)
-        # Check for update and delete permissions
-        allowed_actions = set()
-        for perm in object_perms:
-            action = perm.split("_", 1)[0]
-            if action in {"update", "delete"}:
-                allowed_actions.add(action)
-
-        allowed_custom_actions = self.get_custom_user_permissions(user)
-        allowed_actions.update(set(allowed_custom_actions))
-        return list(allowed_actions)
-
-    def get_custom_user_permissions(self, user: AbstractUser | AnonymousUser) -> list[str]:
+        allowed_actions = {
+            perm.split("_", 1)[0] for perm in object_perms if perm.split("_", 1)[0] in {"update", "delete"}
+        }
+
+        custom_actions = self.get_custom_object_level_permissions(user)
+        return list(allowed_actions.union(custom_actions))
+
+    def get_custom_object_level_permissions(self, user: AbstractUser | AnonymousUser) -> list[str]:
         """
-        Returns a list of custom permissions (not standard CRUD actions) that the user has on this instance.
+        Retrieve custom (non-CRUD) permissions for this instance.
         """
         object_perms = self._get_object_perms(user)
-        custom_perms = set()
-        # Extract custom permissions that are not standard CRUD actions
-        for perm in object_perms:
-            action = perm.split("_", 1)[0]
-            # Make sure to exclude standard CRUD actions
-            if action not in ["view", "create", "update", "delete"]:
-                custom_perms.add(action)
+        custom_perms = {
+            perm.split("_", 1)[0]
+            for perm in object_perms
+            if perm.split("_", 1)[0] not in ["view", "create", "update", "delete"]
+        }
         return list(custom_perms)
 
+    @classmethod
+    def get_collection_level_permissions(cls, user: AbstractUser | AnonymousUser, project) -> list[str]:
+        """
+        Retrieve collection-level permissions for the given user.
+        """
+        app_label = "main"
+        if user.is_superuser:
+            return ["create"]
+        # If the model is m2m related to projects or has no project relation, use model-level permissions
+        if cls.get_project_accessor() is None or cls.get_project_accessor() == "projects":
+            if user.has_perm(f"{app_label}.create_{cls._meta.model_name}"):
+                return ["create"]
+        # If the model is related to a single project, check create permission at object level
+        if cls.get_project_accessor() is not None and project:
+            if user.has_perm(f"{app_label}.create_{cls._meta.model_name}", project):
+                return ["create"]
+
+        return []
+
     class Meta:
         abstract = True

ami/base/permissions.py

@@ -3,7 +3,6 @@
 import logging
 
 from django.contrib.auth.models import AbstractBaseUser, AnonymousUser, User
-from guardian.shortcuts import get_perms
 from rest_framework import permissions
 
 from ami.main.models import BaseModel
@@ -53,7 +52,7 @@ def add_object_level_permissions(
 
     permissions = response_data.get("user_permissions", set())
     if isinstance(instance, BaseModel):
-        permissions.update(instance.get_user_object_permissions(user))
+        permissions.update(instance.get_permissions(user))
     response_data["user_permissions"] = list(permissions)
     return response_data
 
@@ -67,12 +66,10 @@ def add_collection_level_permissions(user: User | None, response_data: dict, mod
     "create" permission is added to the `user_permissions` set in the `response_data`.
     """
 
-    logger.debug(f"add_collection_level_permissions model {model.__name__}, {type(model)} ")
+    logger.info(f"add_collection_level_permissions model {model.__name__}, {type(model)} ")
     permissions = response_data.get("user_permissions", set())
-    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")
+    collection_level_perms = model.get_collection_level_permissions(user=user, project=project)
+    permissions.update(collection_level_perms)
     response_data["user_permissions"] = list(permissions)
     return response_data
 

ami/base/serializers.py

@@ -58,6 +58,31 @@ def to_representation(self, instance):
         instance_data = self.get_permissions(instance=instance, instance_data=instance_data)
         return instance_data
 
+    def get_instance_for_permission_check(self):
+        """
+        Returns an unsaved model instance built from validated_data,
+        excluding ManyToMany fields and any non-model fields (like 'project').
+        Safe to use for permission checking before saving.
+        """
+        validated_data = getattr(self, "validated_data", {})
+        if not validated_data:
+            raise ValueError("Serializer must be validated before calling this method.")
+
+        model_cls = self.Meta.model
+        model_field_names = {f.name for f in model_cls._meta.get_fields()}
+        m2m_fields = {f.name for f in model_cls._meta.many_to_many}
+
+        safe_data = {}
+        for key, value in validated_data.items():
+            # skip many-to-many and non-model fields
+            if key in m2m_fields:
+                continue
+            if key not in model_field_names:
+                continue
+            safe_data[key] = value
+
+        return model_cls(**safe_data)
+
 
 class MinimalNestedModelSerializer(DefaultSerializer):
     """

ami/jobs/models.py

@@ -949,7 +949,7 @@ def save(self, update_progress=True, *args, **kwargs):
         if self.progress.summary.status != self.status:
             logger.warning(f"Job {self} status mismatches progress: {self.progress.summary.status} != {self.status}")
 
-    def check_custom_permission(self, user, action: str) -> bool:
+    def check_custom_object_level_permission(self, user, action: str) -> bool:
         job_type = self.job_type_key.lower()
         if self.source_image_single:
             action = "run_single_image"
@@ -961,7 +961,7 @@ def check_custom_permission(self, user, action: str) -> bool:
         project = self.get_project() if hasattr(self, "get_project") else None
         return user.has_perm(permission_codename, project)
 
-    def get_custom_user_permissions(self, user) -> list[str]:
+    def get_custom_object_level_permissions(self, user) -> list[str]:
         project = self.get_project()
         if not project:
             return []
@@ -977,7 +977,9 @@ def get_custom_user_permissions(self, user) -> list[str]:
                 # make sure to exclude standard CRUD actions
                 if action not in ["view", "create", "update", "delete"]:
                     custom_perms.add(action)
-        logger.debug(f"Custom permissions for user {user} on project {self}, with jobtype {job_type}: {custom_perms}")
+        logger.debug(
+            f"Custom object permissions for user {user} on project {self}, with jobtype {job_type}: {custom_perms}"
+        )
         return list(custom_perms)
 
     @classmethod

ami/jobs/views.py

@@ -131,7 +131,7 @@ def perform_create(self, serializer):
 
         job: Job = serializer.save()  # type: ignore
         if url_boolean_param(self.request, "start_now", default=False):
-            if job.check_custom_permission(self.request.user, "run"):
+            if job.check_custom_object_level_permission(self.request.user, "run"):
                 # If the user has permission, enqueue the job
                 job.enqueue()
             else:

ami/main/api/views.py

@@ -171,12 +171,12 @@ def get_serializer_class(self):
             return ProjectSerializer
 
     def perform_create(self, serializer):
-        super().perform_create(serializer)
         # Check if user is authenticated
         if not self.request.user or not self.request.user.is_authenticated:
             raise PermissionDenied("You must be authenticated to create a project.")
 
-        # Add current user as project owner
+        # Save once with owner set - this allows the set_project_owner_permissions post_save signal to fire correctly
+        # Signal fires with both created=True AND owner set, assigning ProjectManager role
         serializer.save(owner=self.request.user)
 
     @extend_schema(
@@ -1284,6 +1284,7 @@ class TaxonViewSet(DefaultViewSet, ProjectMixin):
         TaxonTagFilter,
         TagInverseFilter,
     ]
+    permission_classes = [ObjectPermission]
     filterset_fields = [
         "name",
         "rank",