Create permissions service layer

dandiapi/api/mail.py

@@ -8,6 +8,8 @@
 from django.template.loader import render_to_string
 from django.utils.html import strip_tags
 
+from dandiapi.api.services.permissions.dandiset import get_dandiset_owners
+
 if TYPE_CHECKING:
     from collections.abc import Iterable
 
@@ -207,7 +209,7 @@ def build_dandiset_unembargoed_message(dandiset: Dandiset):
         subject='Your Dandiset has been unembargoed!',
         message=strip_tags(html_message),
         html_message=html_message,
-        to=[owner.email for owner in dandiset.owners],
+        to=[owner.email for owner in get_dandiset_owners(dandiset)],
     )
 
 
@@ -229,7 +231,7 @@ def build_dandiset_unembargo_failed_message(dandiset: Dandiset):
         subject=f'DANDI: Unembargo failed for dandiset {dandiset.identifier}',
         message=strip_tags(html_message),
         html_message=html_message,
-        to=[owner.email for owner in dandiset.owners],
+        to=[owner.email for owner in get_dandiset_owners(dandiset)],
         bcc=[settings.DANDI_DEV_EMAIL],
         reply_to=[ADMIN_EMAIL],
     )

dandiapi/api/management/commands/extract_metadata.py

@@ -15,6 +15,7 @@
 
 from dandiapi.api.models import Asset, Dandiset, Version
 from dandiapi.api.services.asset import change_asset
+from dandiapi.api.services.permissions.dandiset import get_dandiset_owners
 
 if TYPE_CHECKING:
     from django.db.models import QuerySet
@@ -55,7 +56,7 @@ def extract_asset_metadata(asset: Asset, draft_version: Version):
 
     # Use dandiset owner, default to some admin user
     user = (
-        draft_version.dandiset.owners.first()
+        get_dandiset_owners(draft_version.dandiset).first()
         or User.objects.filter(is_superuser=True, is_staff=True).first()
     )
 

dandiapi/api/models/dandiset.py

@@ -3,41 +3,12 @@
 from django.db import models
 from django_extensions.db.models import TimeStampedModel
 from guardian.models import GroupObjectPermissionBase, UserObjectPermissionBase
-from guardian.shortcuts import assign_perm, get_objects_for_user, get_users_with_perms, remove_perm
-
-
-class DandisetManager(models.Manager):
-    def visible_to(self, user) -> models.QuerySet[Dandiset]:
-        """
-        Return a queryset containing all dandisets visible to the given user.
-
-        This is basically all dandisets except embargoed dandisets not owned by the given user.
-        """
-        # We would like to do something like:
-        # Dandiset.filter(embargo_status=OPEN | permission__owned)
-        # but this is not possible with django-guardian; the `get_objects_for_user` shortcut must
-        # be used instead.
-        # We would like to do something like:
-        # queryset = Dandiset.objects.filter(embargo_status=OPEN).union(get_objects_for_user(...))
-        # but you cannot filter or perform many other common queryset operations after using
-        # union().
-        # Therefore, we resort to fetching the list of all primary keys for dandisets owned by the
-        # current user with `get_objects_for_user(...)`, then filter for either those dandisets or
-        # OPEN dandisets. There aren't very many dandisets and most users won't own very many of
-        # them, so there shouldn't be too many dandisets in the list.
-        owned_dandiset_pks = get_objects_for_user(user, 'owner', Dandiset).values('pk').all()
-        return self.filter(
-            models.Q(embargo_status=Dandiset.EmbargoStatus.OPEN)
-            | models.Q(pk__in=owned_dandiset_pks)
-        ).order_by('created')
 
 
 class Dandiset(TimeStampedModel):
     # Don't add beginning and end markers, so this can be embedded in larger regexes
     IDENTIFIER_REGEX = r'\d{6}'
 
-    objects: DandisetManager = DandisetManager()
-
     class EmbargoStatus(models.TextChoices):
         EMBARGOED = 'EMBARGOED', 'Embargoed'
         UNEMBARGOING = 'UNEMBARGOING', 'Unembargoing'
@@ -74,31 +45,6 @@ def most_recent_published_version(self):
     def draft_version(self):
         return self.versions.filter(version='draft').get()
 
-    @property
-    def owners(self):
-        return get_users_with_perms(self, only_with_perms_in=['owner']).order_by('date_joined')
-
-    def set_owners(self, new_owners):
-        old_owners = get_users_with_perms(self, only_with_perms_in=['owner'])
-
-        removed_owners = []
-        added_owners = []
-
-        # Remove old owners
-        for old_owner in old_owners:
-            if old_owner not in new_owners:
-                remove_perm('owner', old_owner, self)
-                removed_owners.append(old_owner)
-
-        # Add new owners
-        for new_owner in new_owners:
-            if new_owner not in old_owners:
-                assign_perm('owner', new_owner, self)
-                added_owners.append(new_owner)
-
-        # Return the owners added/removed so they can be emailed
-        return removed_owners, added_owners
-
     @classmethod
     def published_count(cls):
         """Return the number of Dandisets with published Versions."""

dandiapi/api/services/asset/__init__.py

@@ -17,6 +17,7 @@
     DraftDandisetNotModifiableError,
     ZarrArchiveBelongsToDifferentDandisetError,
 )
+from dandiapi.api.services.permissions.dandiset import is_dandiset_owner
 from dandiapi.api.tasks import remove_asset_blob_embargoed_tag_task
 
 if TYPE_CHECKING:
@@ -98,7 +99,7 @@ def change_asset(  # noqa: PLR0913
     if 'path' not in new_metadata:
         raise ValueError('Path must be present in new_metadata')
 
-    if not user.has_perm('owner', version.dandiset):
+    if not is_dandiset_owner(version.dandiset, user):
         raise DandisetOwnerRequiredError
     if version.version != 'draft':
         raise DraftDandisetNotModifiableError
@@ -148,7 +149,7 @@ def add_asset_to_version(
     if 'path' not in metadata:
         raise RuntimeError('Path must be present in metadata')
 
-    if not user.has_perm('owner', version.dandiset):
+    if not is_dandiset_owner(version.dandiset, user):
         raise DandisetOwnerRequiredError
     if version.version != 'draft':
         raise DraftDandisetNotModifiableError
@@ -194,7 +195,7 @@ def add_asset_to_version(
 
 
 def remove_asset_from_version(*, user, asset: Asset, version: Version) -> Version:
-    if not user.has_perm('owner', version.dandiset):
+    if not is_dandiset_owner(version.dandiset, user):
         raise DandisetOwnerRequiredError
     if version.version != 'draft':
         raise DraftDandisetNotModifiableError

dandiapi/api/services/dandiset/__init__.py

@@ -1,14 +1,14 @@
 from __future__ import annotations
 
 from django.db import transaction
-from guardian.shortcuts import assign_perm
 
 from dandiapi.api.models.dandiset import Dandiset
 from dandiapi.api.models.version import Version
 from dandiapi.api.services import audit
 from dandiapi.api.services.dandiset.exceptions import DandisetAlreadyExistsError
 from dandiapi.api.services.embargo.exceptions import DandisetUnembargoInProgressError
 from dandiapi.api.services.exceptions import AdminOnlyOperationError, NotAllowedError
+from dandiapi.api.services.permissions.dandiset import add_dandiset_owner, is_dandiset_owner
 from dandiapi.api.services.version.metadata import _normalize_version_metadata
 
 
@@ -38,7 +38,7 @@ def create_dandiset(
         dandiset = Dandiset(id=identifier, embargo_status=embargo_status)
         dandiset.full_clean()
         dandiset.save()
-        assign_perm('owner', user, dandiset)
+        add_dandiset_owner(dandiset, user)
         draft_version = Version(
             dandiset=dandiset,
             name=version_name,
@@ -56,7 +56,7 @@ def create_dandiset(
 
 
 def delete_dandiset(*, user, dandiset: Dandiset) -> None:
-    if not user.has_perm('owner', dandiset):
+    if not is_dandiset_owner(dandiset, user):
         raise NotAllowedError('Cannot delete dandisets which you do not own.')
     if dandiset.versions.exclude(version='draft').exists():
         raise NotAllowedError('Cannot delete dandisets with published versions.')

dandiapi/api/services/embargo/__init__.py

@@ -12,6 +12,7 @@
 from dandiapi.api.services.embargo.utils import _delete_object_tags, remove_dandiset_embargo_tags
 from dandiapi.api.services.exceptions import DandiError
 from dandiapi.api.services.metadata import validate_version_metadata
+from dandiapi.api.services.permissions.dandiset import is_dandiset_owner
 from dandiapi.api.storage import get_boto_client
 from dandiapi.api.tasks import unembargo_dandiset_task
 from dandiapi.zarr.models import ZarrArchive
@@ -95,7 +96,7 @@ def kickoff_dandiset_unembargo(*, user: User, dandiset: Dandiset):
     if dandiset.embargo_status != Dandiset.EmbargoStatus.EMBARGOED:
         raise DandisetNotEmbargoedError
 
-    if not user.has_perm('owner', dandiset):
+    if not is_dandiset_owner(dandiset, user):
         raise DandisetOwnerRequiredError
 
     if dandiset.uploads.count():

dandiapi/api/services/permissions/dandiset.py

@@ -0,0 +1,118 @@
+"""Abstracts over django guardian to provide an internal permission framework."""
+
+from __future__ import annotations
+
+import typing
+
+from django.contrib.auth.models import AnonymousUser, User
+from django.db import transaction
+from django.db.models import Q, QuerySet
+from django.utils.decorators import method_decorator
+from guardian.decorators import permission_required
+from guardian.shortcuts import assign_perm, get_objects_for_user, get_users_with_perms
+
+from dandiapi.api.models.dandiset import Dandiset, DandisetUserObjectPermission
+
+if typing.TYPE_CHECKING:
+    from django.contrib.auth.base_user import AbstractBaseUser
+
+    from dandiapi.api.models.asset import Asset
+
+
+def get_dandiset_owners(dandiset: Dandiset) -> QuerySet[User]:
+    qs = typing.cast(QuerySet[User], get_users_with_perms(dandiset, only_with_perms_in=['owner']))
+    return qs.order_by('date_joined')
+
+
+def add_dandiset_owner(dandiset: Dandiset, user: User):
+    assign_perm('owner', user, dandiset)
+
+
+def replace_dandiset_owners(dandiset: Dandiset, users: list[User]):
+    existing_owners = get_dandiset_owners(dandiset)
+    existing_owner_set = set(existing_owners)
+    new_owner_set = set(users)
+
+    with transaction.atomic():
+        # Delete all existing owners
+        DandisetUserObjectPermission.objects.filter(
+            content_object=dandiset.pk, permission__codename='owner'
+        ).delete()
+
+        # Set owners to new list
+        for user in users:
+            add_dandiset_owner(dandiset, user)
+
+    # Return the owners added/removed so they can be emailed
+    removed_owners = existing_owner_set - new_owner_set
+    added_owners = new_owner_set - existing_owner_set
+    return removed_owners, added_owners
+
+
+def is_dandiset_owner(dandiset: Dandiset, user: AbstractBaseUser | AnonymousUser) -> bool:
+    if isinstance(user, AnonymousUser):
+        return False
+
+    user = typing.cast(User, user)
+    return user.has_perm('owner', dandiset)
+
+
+def is_owned_asset(asset: Asset, user: AbstractBaseUser | AnonymousUser) -> bool:
+    """Return `True` if this asset belongs to a dandiset that the user is an owner of."""
+    if user.is_anonymous:
+        return False
+
+    user = typing.cast(User, user)
+    asset_dandisets = Dandiset.objects.filter(versions__in=asset.versions.all())
+    asset_dandisets_owned_by_user = DandisetUserObjectPermission.objects.filter(
+        content_object__in=asset_dandisets,
+        user=user,
+        permission__codename='owner',
+    )
+
+    return asset_dandisets_owned_by_user.exists()
+
+
+def get_owned_dandisets(
+    user: AbstractBaseUser | AnonymousUser,
+    include_superusers=True,  # noqa: FBT002
+) -> QuerySet[Dandiset]:
+    return get_objects_for_user(user, 'owner', Dandiset, with_superuser=include_superusers)
+
+
+def get_visible_dandisets(user: AbstractBaseUser | AnonymousUser) -> QuerySet[Dandiset]:
+    """
+    Return a queryset containing all dandisets visible to the given user.
+
+    This is basically all dandisets except embargoed dandisets not owned by the given user.
+    """
+    # We would like to do something like:
+    # Dandiset.filter(embargo_status=OPEN | permission__owned)
+    # but this is not possible with django-guardian; the `get_objects_for_user` shortcut must
+    # be used instead.
+    # We would like to do something like:
+    # queryset = Dandiset.objects.filter(embargo_status=OPEN).union(get_objects_for_user(...))
+    # but you cannot filter or perform many other common queryset operations after using
+    # union().
+    # Therefore, we resort to fetching the list of all primary keys for dandisets owned by the
+    # current user with `get_objects_for_user(...)`, then filter for either those dandisets or
+    # OPEN dandisets. There aren't very many dandisets and most users won't own very many of
+    # them, so there shouldn't be too many dandisets in the list.
+
+    owned_dandiset_pks = get_owned_dandisets(user).values('pk').all()
+    return Dandiset.objects.filter(
+        Q(embargo_status=Dandiset.EmbargoStatus.OPEN) | Q(pk__in=owned_dandiset_pks)
+    ).order_by('created')
+
+
+def require_dandiset_owner_or_403(pk_path: str):
+    """
+    Decorate viewset methods to only allow access to dandiset owners.
+
+    The `pk_path` argument is the Dandiset ID URL path variable that DRF passes into the request.
+    """
+    return method_decorator(
+        permission_required(
+            perm='owner', lookup_variables=(Dandiset, 'pk', pk_path), return_403=True
+        )
+    )

dandiapi/api/services/publish/__init__.py

@@ -13,6 +13,7 @@
 from dandiapi.api.models import Asset, Dandiset, Version
 from dandiapi.api.services import audit
 from dandiapi.api.services.exceptions import NotAllowedError
+from dandiapi.api.services.permissions.dandiset import is_dandiset_owner
 from dandiapi.api.services.publish.exceptions import (
     DandisetAlreadyPublishedError,
     DandisetAlreadyPublishingError,
@@ -49,7 +50,7 @@ def _lock_dandiset_for_publishing(*, user: User, dandiset: Dandiset) -> None:  #
 
     This function MUST be called before _publish_dandiset is called.
     """
-    if not user.has_perm('owner', dandiset):
+    if not is_dandiset_owner(dandiset, user):
         raise NotAllowedError
 
     if dandiset.embargo_status != Dandiset.EmbargoStatus.OPEN: