Let taxa lists and processing services be shared with every project, with safe rules for who can edit them

ami/base/models.py

@@ -1,6 +1,7 @@
 from django.contrib.auth.models import AbstractUser, AnonymousUser
+from django.core.exceptions import FieldDoesNotExist
 from django.db import models
-from django.db.models import Q, QuerySet
+from django.db.models import Exists, OuterRef, Q, QuerySet
 from guardian.shortcuts import get_perms
 
 import ami.tasks
@@ -39,6 +40,23 @@ def has_many_to_many_project_relation(model: type[models.Model]) -> bool:
     return False
 
 
+class PublicScopedModel(models.Model):
+    """
+    Base for M2M-to-project models that can be marked public. for_project() and
+    visible_for_user() check issubclass(model, PublicScopedModel) for their
+    public-row bypass, instead of duck-typing an `is_public` attribute that any
+    model could grow by coincidence and silently pick up that behavior.
+    """
+
+    is_public = models.BooleanField(
+        default=False,
+        help_text="Public rows are shown to every project, not just the ones linked via 'projects'.",
+    )
+
+    class Meta:
+        abstract = True
+
+
 class BaseQuerySet(QuerySet):
     def visible_for_user(self, user: User | AnonymousUser) -> QuerySet:
         """
@@ -85,8 +103,37 @@ def visible_for_user(self, user: User | AnonymousUser) -> QuerySet:
         if not is_anonymous:
             filter_condition |= Q(**{f"{project_field}owner": user}) | Q(**{f"{project_field}members": user})
 
+        # Public rows (e.g. public TaxaLists) are visible to everyone, draft or not.
+        if issubclass(model, PublicScopedModel):
+            filter_condition |= Q(is_public=True)
+
         return self.filter(filter_condition).distinct()
 
+    def for_project(self, project: models.Model, include_public: bool = True) -> QuerySet:
+        """
+        Filter to rows in the model's M2M ``projects`` field for the given project,
+        plus every public row when the model defines ``is_public`` and ``include_public``
+        is set.
+
+        Membership is checked with an ``Exists`` subquery against the M2M through table
+        instead of filtering on ``projects=project`` directly, so a row linked to the
+        project through multiple paths cannot appear twice and no ``.distinct()`` is
+        needed downstream.
+        """
+        model = self.model
+        try:
+            field = model._meta.get_field("projects")
+        except FieldDoesNotExist:
+            field = None
+        if not isinstance(field, models.ManyToManyField):
+            raise TypeError(f"{model.__name__} has no ManyToMany 'projects' field; for_project() is not applicable.")
+
+        condition = Q(Exists(model._default_manager.filter(pk=OuterRef("pk"), projects=project)))
+        if include_public and issubclass(model, PublicScopedModel):
+            condition |= Q(is_public=True)
+
+        return self.filter(condition)
+
 
 class BaseModel(models.Model):
     """ """

ami/base/permissions.py

@@ -77,6 +77,67 @@ def add_collection_level_permissions(user: User | None, response_data: dict, mod
     return response_data
 
 
+def user_can_manage_public(user: AbstractBaseUser | AnonymousUser, model_or_instance) -> bool:
+    """
+    A superuser, or a user holding <app_label>.manage_public_<model_name> for
+    the given model (or an instance of it) — the platform permission gating
+    write access to a public row, in place of project membership or staff status.
+    """
+    if not user or not user.is_authenticated:
+        return False
+    if user.is_superuser:  # type: ignore[union-attr]
+        return True
+    meta = model_or_instance._meta
+    return user.has_perm(f"{meta.app_label}.manage_public_{meta.model_name}")  # type: ignore[union-attr]
+
+
+def check_public_scoped_write_permission(user, instance, non_public_fallback) -> bool:
+    """
+    True if `user` may write to `instance`: a public row needs the platform
+    manage_public_<model> permission (superusers always pass, via
+    user_can_manage_public); a non-public row falls back to `non_public_fallback()`,
+    the model's own write rule (project membership, active-staff status, ...).
+    """
+    if getattr(instance, "is_public", False):
+        return user_can_manage_public(user, instance)
+    return non_public_fallback()
+
+
+def check_taxalist_write_permission(user, taxa_list, project) -> bool:
+    """Thin alias kept so branches stacked on this one still import this name."""
+    return check_public_scoped_write_permission(
+        user, taxa_list, lambda: bool(user.is_superuser or (project and project.members.filter(pk=user.pk).exists()))
+    )
+
+
+def check_processingservice_write_permission(user, processing_service) -> bool:
+    """Thin alias kept so branches stacked on this one still import this name."""
+    return check_public_scoped_write_permission(
+        user, processing_service, lambda: bool(user.is_superuser or is_active_staff(user))
+    )
+
+
+def add_processingservice_permissions(user, instance, response_data: dict) -> dict:
+    """
+    Add update/delete to user_permissions for a ProcessingService.
+
+    Unlike add_m2m_object_permissions, this skips the M2M membership check and
+    the guardian lookup entirely: no per-project *_processingservice guardian
+    permission exists anywhere in this codebase (see Project.Permissions and
+    ami/users/roles.py), so that branch could only ever fire for a superuser,
+    which a plain attribute check already covers for free. A public instance
+    still checks the platform manage_public_processingservice permission.
+    """
+    perms = set(response_data.get("user_permissions", []))
+    if getattr(instance, "is_public", False):
+        if user_can_manage_public(user, instance):
+            perms.update(["update", "delete"])
+    elif user.is_superuser:
+        perms.update(["update", "delete"])
+    response_data["user_permissions"] = list(perms)
+    return response_data
+
+
 def add_m2m_object_permissions(
     user, instance, project, response_data: dict, project_perms: set[str] | None = None
 ) -> dict:
@@ -96,13 +157,22 @@ def add_m2m_object_permissions(
     (user, project) pass in `guardian.get_perms(user, project)` once instead
     of once per instance; pass None to look it up here as before.
 
+    A public instance is the one exception: its update/delete permissions come
+    from the model's manage_public_<model> permission, not project membership.
+
     This is a temporary approach for the M2M permission gap described in #1120.
     Once that issue is resolved, this should be replaced by a generic permission
     class (Pattern B: Bare M2M) that handles TaxaList, Taxon, ProcessingService,
     Pipeline, and other M2M-to-Project models uniformly.
     """
     perms = set(response_data.get("user_permissions", []))
 
+    if getattr(instance, "is_public", False):
+        if user_can_manage_public(user, instance):
+            perms.update(["update", "delete"])
+        response_data["user_permissions"] = list(perms)
+        return response_data
+
     if not project:
         response_data["user_permissions"] = list(perms)
         return response_data
@@ -160,6 +230,92 @@ def has_permission(self, request, view):
         return project.members.filter(pk=request.user.pk).exists()
 
 
+class _BaseGateOrPublicManager(permissions.BasePermission):
+    """
+    Shared shape for M2M-to-project models with a public flag: safe methods are
+    open to everyone; unsafe methods need the model's own base gate (project
+    membership, active-staff status, ...) or the manage_public_<model> platform
+    permission. `exclude_create_from_bypass` forces a plain "create a new row"
+    action through the base gate only, since there's no object yet to tell
+    whether it will be public. Subclasses implement get_model() and
+    get_base_gate() — get_model() does a local import to avoid a module-level
+    circular import between this file and the app that owns the model.
+    """
+
+    exclude_create_from_bypass = False
+
+    def get_model(self):
+        raise NotImplementedError
+
+    def get_base_gate(self, request, view) -> bool:
+        raise NotImplementedError
+
+    def has_permission(self, request, view):
+        if request.method in permissions.SAFE_METHODS:
+            return True
+
+        if not request.user or not request.user.is_authenticated:
+            return False
+
+        if request.user.is_superuser:  # type: ignore[union-attr]
+            return True
+
+        if self.exclude_create_from_bypass and getattr(view, "action", None) == "create":
+            return self.get_base_gate(request, view)
+
+        if user_can_manage_public(request.user, self.get_model()):
+            return True
+
+        return self.get_base_gate(request, view)
+
+    def has_object_permission(self, request, view, obj):
+        if request.method in permissions.SAFE_METHODS:
+            return True
+        return check_public_scoped_write_permission(request.user, obj, lambda: self.get_base_gate(request, view))
+
+
+class IsProjectMemberOrPublicListManager(_BaseGateOrPublicManager):
+    """
+    Used by the nested add/remove-taxon route: serves both public and
+    project-scoped lists, and has no object to check yet at has_permission()
+    time.
+    """
+
+    def get_model(self):
+        from ami.main.models import TaxaList
+
+        return TaxaList
+
+    def get_base_gate(self, request, view):
+        get_active_project = getattr(view, "get_active_project", None)
+        project = get_active_project() if get_active_project else None
+        return bool(project and project.members.filter(pk=request.user.pk).exists())
+
+
+class IsProjectMemberOrPublicListManagerOrReadOnly(IsProjectMemberOrPublicListManager):
+    """For TaxaListViewSet: creating a brand-new list always needs real project membership."""
+
+    exclude_create_from_bypass = True
+
+
+class IsActiveStaffOrPublicManager(_BaseGateOrPublicManager):
+    """Used by ProcessingServiceViewSet's non-create actions and any future nested route."""
+
+    def get_model(self):
+        from ami.ml.models.processing_service import ProcessingService
+
+        return ProcessingService
+
+    def get_base_gate(self, request, view):
+        return is_active_staff(request.user)
+
+
+class IsActiveStaffOrPublicManagerOrReadOnly(IsActiveStaffOrPublicManager):
+    """For ProcessingServiceViewSet: creating a brand-new service always needs active-staff status."""
+
+    exclude_create_from_bypass = True
+
+
 class ObjectPermission(permissions.BasePermission):
     """
     Generic permission class that delegates to the model's `check_permission(user, action)` method.

ami/base/views.py

@@ -94,3 +94,17 @@ def get_active_project(self) -> Project | None:
             raise Http404("Project not found.")
 
         return project
+
+    def get_include_public(self) -> bool:
+        """
+        The ?include_public query param: whether to include public rows alongside
+        a model's own project-scoped ones. Defaults to true; an invalid value
+        raises ValidationError (400) via SingleParamSerializer.
+        """
+        from ami.base.serializers import SingleParamSerializer
+
+        return SingleParamSerializer[bool].clean(
+            param_name="include_public",
+            field=serializers.BooleanField(required=False, default=True),
+            data=self.request.query_params,
+        )

ami/main/admin.py

@@ -736,7 +736,7 @@ def parent_names(self, obj) -> str:
 class TaxaListAdmin(admin.ModelAdmin[TaxaList]):
     """Admin panel example for ``TaxaList`` model."""
 
-    list_display = ("name", "taxa_count", "created_at", "updated_at")
+    list_display = ("name", "is_public", "taxa_count", "created_at", "updated_at")
 
     def taxa_count(self, obj) -> int:
         return obj.taxa.count()
@@ -746,7 +746,7 @@ def taxa_count(self, obj) -> int:
         "projects",
     )
 
-    list_filter = ("projects",)
+    list_filter = ("is_public", "projects")
 
 
 @admin.register(Device)

ami/main/api/schemas.py

@@ -13,3 +13,11 @@
     required=False,
     type=int,
 )
+
+include_public_doc_param = OpenApiParameter(
+    name="include_public",
+    description="Include rows that are public (available to every project), not just the ones "
+    "belonging to this project. Defaults to true.",
+    required=False,
+    type=bool,
+)

ami/main/api/serializers.py

@@ -708,6 +708,7 @@ class TaxaListSerializer(DefaultSerializer):
     taxa = serializers.SerializerMethodField()
     taxa_count = serializers.SerializerMethodField()
     projects = serializers.SerializerMethodField()
+    is_public = serializers.BooleanField(read_only=True)
 
     class Meta:
         model = TaxaList
@@ -718,6 +719,7 @@ class Meta:
             "taxa",
             "taxa_count",
             "projects",
+            "is_public",
             "created_at",
             "updated_at",
         ]
@@ -763,11 +765,19 @@ def get_permissions(self, instance, instance_data):
 
     def get_projects(self, obj):
         """
-        Return list of project IDs this taxa list belongs to, sorted for a
-        deterministic response. Reads the `projects` prefetched by
+        Return the ids of this list's linked projects that are visible to the
+        requester, sorted for a deterministic response. A public list can be linked
+        to a draft project it's otherwise not visible in; without this filter, an
+        outsider retrieving the public list would learn that draft project's id even
+        though they can't see the project itself. Reads the `projects` prefetched by
         TaxaListViewSet.get_queryset instead of querying per row.
         """
-        return sorted(project.pk for project in obj.projects.all())
+        request = self.context["request"]
+        if not hasattr(self, "_visible_project_ids"):
+            self._visible_project_ids = set(
+                Project.objects.visible_for_user(request.user).values_list("id", flat=True)
+            )
+        return sorted(project.pk for project in obj.projects.all() if project.pk in self._visible_project_ids)
 
 
 class TaxaListTaxonInputSerializer(serializers.Serializer):