diff --git a/CHANGELOG.md b/CHANGELOG.md index 83284360..b5a85684 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,10 @@ and this project adheres to - 🔥(backend) remove unused ServerToServerAuthentication backend +### Fixed + +- 🐛(backend) allow inviting external person on item with no direct access + ## [v0.14.0] - 2026-02-25 ### Added diff --git a/src/backend/core/api/permissions.py b/src/backend/core/api/permissions.py index 450d5684..4747880e 100644 --- a/src/backend/core/api/permissions.py +++ b/src/backend/core/api/permissions.py @@ -1,13 +1,12 @@ """Permission handlers for the drive core app.""" from django.core import exceptions -from django.db.models import Q from django.http import Http404 from lasuite.drf.models.choices import PRIVILEGED_ROLES from rest_framework import permissions -from core.models import ItemAccess, RoleChoices, get_trashbin_cutoff +from core.models import RoleChoices, get_trashbin_cutoff ACTION_FOR_METHOD_TO_PERMISSION = { "versions_detail": {"DELETE": "versions_destroy", "GET": "versions_retrieve"}, @@ -65,47 +64,18 @@ class IsOwnedOrPublic(IsAuthenticated): return False -class InvitationPermission(permissions.BasePermission): - """A permission class for invitations.""" +class CreateWithPriviliegedRolesMixin: + """ + Implement a common has_permission method checking that + the user has privileged role on the item in order to + perform the create action. + This mixin must be used with the IsAuthenticated permission class + """ + + resources = None def has_permission(self, request, view): - user = request.user - - # Ensure the user is authenticated - if not (bool(request.auth) or request.user.is_authenticated): - return False - - # Apply permission checks only for creation (POST requests) - if view.action != "create": - return True - - # Check if resource_id is passed in the context - try: - item_id = view.kwargs["resource_id"] - except KeyError as exc: - raise exceptions.ValidationError( - "You must set a item ID in kwargs to manage item invitations." - ) from exc - - # Check if the user has access to manage invitations (Owner/Admin roles) - return ItemAccess.objects.filter( - Q(user=user) | Q(team__in=user.teams), - item=item_id, - role__in=[RoleChoices.OWNER, RoleChoices.ADMIN], - ).exists() - - def has_object_permission(self, request, view, obj): - """Check permission for a given object.""" - abilities = obj.get_abilities(request.user) - action = view.action - return abilities.get(action, False) - - -class ItemAccessPermission(IsAuthenticated): - """Permission class for the ItemAccessViewSet.""" - - def has_permission(self, request, view): - """check create permission for accesses in documents tree.""" + """Check the current user has privileged roles on the related item.""" if super().has_permission(request, view) is False: return False @@ -113,11 +83,28 @@ class ItemAccessPermission(IsAuthenticated): role = getattr(view, view.resource_field_name).get_role(request.user) if role not in PRIVILEGED_ROLES: raise exceptions.PermissionDenied( - "You are not allowed to manage accesses for this resource." + f"You are not allowed to manage {self.resources} for this resource." ) return True + +class InvitationPermission(CreateWithPriviliegedRolesMixin, IsAuthenticated): + """A permission class for the InvitationViewset.""" + + resources = "invitations" + + def has_object_permission(self, request, view, obj): + """Check permission for a given object.""" + abilities = obj.get_abilities(request.user) + return abilities.get(view.action, False) + + +class ItemAccessPermission(CreateWithPriviliegedRolesMixin, IsAuthenticated): + """Permission class for the ItemAccessViewSet.""" + + resources = "accesses" + def has_object_permission(self, request, view, obj): """Check permission for a given object.""" abilities = obj.get_abilities(request.user) diff --git a/src/backend/core/api/serializers.py b/src/backend/core/api/serializers.py index 8ca2459e..e98bf80c 100644 --- a/src/backend/core/api/serializers.py +++ b/src/backend/core/api/serializers.py @@ -7,7 +7,6 @@ from os.path import splitext from urllib.parse import quote from django.conf import settings -from django.db.models import Q from django.urls import reverse from django.utils.translation import gettext_lazy as _ @@ -840,26 +839,6 @@ class InvitationSerializer(serializers.ModelSerializer): return attrs - def validate_role(self, role): - """Custom validation for the role field.""" - request = self.context.get("request") - user = getattr(request, "user", None) - item_id = self.context["resource_id"] - - # If the role is OWNER, check if the user has OWNER access - if role == models.RoleChoices.OWNER: - if not models.ItemAccess.objects.filter( - Q(user=user) | Q(team__in=user.teams), - item=item_id, - role=models.RoleChoices.OWNER, - ).exists(): - raise serializers.ValidationError( - "Only owners of a item can invite other users as owners.", - code="invitation_role_owner_limited_to_owners", - ) - - return role - # Suppress the warning about not implementing `create` and `update` methods # since we don't use a model and only rely on the serializer for validation diff --git a/src/backend/core/api/viewsets.py b/src/backend/core/api/viewsets.py index 5bf465e8..00a49720 100644 --- a/src/backend/core/api/viewsets.py +++ b/src/backend/core/api/viewsets.py @@ -9,7 +9,6 @@ from io import BytesIO from urllib.parse import quote, unquote, urlparse from django.conf import settings -from django.contrib.postgres.aggregates import ArrayAgg from django.contrib.postgres.search import TrigramSimilarity from django.core.cache import cache from django.core.exceptions import ValidationError @@ -1950,6 +1949,17 @@ class InvitationViewset( models.Invitation.objects.all().select_related("item").order_by("-created_at") ) serializer_class = serializers.InvitationSerializer + resource_field_name = "item" + + @cached_property + def item(self) -> models.Item: + """Get related item from resource ID in url and annotate user roles.""" + try: + return models.Item.objects.annotate_user_roles(self.request.user).get( + pk=self.kwargs["resource_id"] + ) + except models.Item.DoesNotExist as excpt: + raise drf.exceptions.NotFound() from excpt def get_serializer_context(self): """Extra context provided to the serializer class.""" @@ -1962,42 +1972,31 @@ class InvitationViewset( queryset = super().get_queryset() queryset = queryset.filter(item=self.kwargs["resource_id"]) + user = self.request.user + + queryset = queryset.annotate_user_roles(user) + if self.action == "list": - user = self.request.user - teams = user.teams + if self.item.get_role(user) not in PRIVILEGED_ROLES: + return queryset.none() - # Determine which role the logged-in user has in the item - user_roles_query = ( - models.ItemAccess.objects.filter( - db.Q(user=user) | db.Q(team__in=teams), - item=self.kwargs["resource_id"], - ) - .values("item") - .annotate(roles_array=ArrayAgg("role")) - .values("roles_array") - ) - - queryset = ( - # The logged-in user should be administrator or owner to see its accesses - queryset.filter( - db.Q( - item__accesses__user=user, - item__accesses__role__in=PRIVILEGED_ROLES, - ) - | db.Q( - item__accesses__team__in=teams, - item__accesses__role__in=PRIVILEGED_ROLES, - ), - ) - # Abilities are computed based on logged-in user's role and - # the user role on each item access - .annotate(user_roles=db.Subquery(user_roles_query)) - .distinct() - ) return queryset + def _validate_provided_role(self, validated_role): + """Ensure that the validated_role can be used.""" + + if ( + validated_role == models.RoleChoices.OWNER + and self.item.get_role(self.request.user) != models.RoleChoices.OWNER + ): + raise drf.serializers.ValidationError( + "Only owners of an item can invite other users as owners.", + code="invitation_role_owner_limited_to_owners", + ) + def perform_create(self, serializer): - """Save invitation to a item then send an email to the invited user.""" + """Save invitation to an item then send an email to the invited user.""" + self._validate_provided_role(serializer.validated_data.get("role")) invitation = serializer.save() invitation.item.send_invitation_email( @@ -2019,8 +2018,11 @@ class InvitationViewset( def perform_update(self, serializer): """Update the invitation and capture the event.""" + self._validate_provided_role(serializer.validated_data.get("role")) + old_role = serializer.instance.role super().perform_update(serializer) + if serializer.instance.role != old_role: posthog_capture( "item_invitation_updated", @@ -2039,6 +2041,7 @@ class InvitationViewset( item = instance.item role = instance.role super().perform_destroy(instance) + posthog_capture( "item_invitation_deleted", self.request.user, diff --git a/src/backend/core/models.py b/src/backend/core/models.py index 37968b3d..6ebe102e 100644 --- a/src/backend/core/models.py +++ b/src/backend/core/models.py @@ -313,9 +313,38 @@ class User(AbstractBaseUser, BaseModel, auth_models.PermissionsMixin): return [] -class ItemQuerySet(TreeQuerySet): +class AnnotateUserRoleQuerySetMixin: + """Mixin to use in a QuerySet to add user_roles annotation.""" + + def annotate_user_roles(self, user): + """ + Annotate queryset with the roles of the current user + on the item or its ancestors. + """ + output_field = ArrayField(base_field=models.CharField()) + + if user.is_authenticated: + user_roles_subquery = ItemAccess.objects.filter( + models.Q(user=user) | models.Q(team__in=user.teams), + item__path__ancestors=models.OuterRef(self.path_property), + ).values_list("role", flat=True) + + return self.annotate( + user_roles=models.Func( + user_roles_subquery, function="ARRAY", output_field=output_field + ) + ) + + return self.annotate( + user_roles=models.Value([], output_field=output_field), + ) + + +class ItemQuerySet(AnnotateUserRoleQuerySetMixin, TreeQuerySet): """Custom queryset for Item model with additional methods.""" + path_property = "path" + def readable_per_se(self, user): """ Filters the queryset to return documents that the given user has @@ -1141,31 +1170,10 @@ class ItemFavorite(BaseModel): return f"{self.user!s} favorite on item {self.item!s}" -class ItemAccessQuerySet(models.QuerySet): +class ItemAccessQuerySet(AnnotateUserRoleQuerySetMixin, models.QuerySet): """Custom queryset for ItemAccess model with additional methods.""" - def annotate_user_roles(self, user): - """ - Annotate ItemAccess queryset with the roles of the current user - on the item or its ancestors. - """ - output_field = ArrayField(base_field=models.CharField()) - - if user.is_authenticated: - user_roles_subquery = ItemAccess.objects.filter( - models.Q(user=user) | models.Q(team__in=user.teams), - item__path__ancestors=models.OuterRef("item__path"), - ).values_list("role", flat=True) - - return self.annotate( - user_roles=models.Func( - user_roles_subquery, function="ARRAY", output_field=output_field - ) - ) - - return self.annotate( - user_roles=models.Value([], output_field=output_field), - ) + path_property = "item__path" class ItemAccessManager(models.Manager.from_queryset(ItemAccessQuerySet)): @@ -1359,6 +1367,16 @@ class ItemAccess(BaseModel): } +class ItemInvitationQuerySet(AnnotateUserRoleQuerySetMixin, models.QuerySet): + """Custom queryset for ItemInvitation model with additional methods.""" + + path_property = "item__path" + + +class ItemInvitationManager(models.Manager.from_queryset(ItemInvitationQuerySet)): + """Manager for ItemAccess model.""" + + class Invitation(BaseModel): """User invitation to an item.""" @@ -1379,6 +1397,8 @@ class Invitation(BaseModel): null=True, ) + objects = ItemInvitationManager() + class Meta: db_table = "drive_invitation" verbose_name = _("Item invitation") @@ -1420,29 +1440,29 @@ class Invitation(BaseModel): validity_duration = timedelta(seconds=settings.INVITATION_VALIDITY_DURATION) return timezone.now() > (self.created_at + validity_duration) + def get_role(self, user): + """Return the role a user has on an item related to this access..""" + if not user.is_authenticated: + return None + + try: + roles = self.user_roles or [] + except AttributeError: + roles = ItemAccess.objects.filter( + models.Q(user=user) | models.Q(team__in=user.teams), + item__path__ancestors=self.item.path, + ).values_list("role", flat=True) + + return RoleChoices.max(*roles) + def get_abilities(self, user): """Compute and return abilities for a given user.""" - roles = [] - - if user.is_authenticated: - teams = user.teams - try: - roles = self.user_roles or [] - except AttributeError: - try: - roles = self.item.accesses.filter( - models.Q(user=user) | models.Q(team__in=teams), - ).values_list("role", flat=True) - except (self._meta.model.DoesNotExist, IndexError): - roles = [] - - is_admin_or_owner = bool( - set(roles).intersection({RoleChoices.OWNER, RoleChoices.ADMIN}) - ) + user_role = self.get_role(user) + is_owner_or_admin = user_role in PRIVILEGED_ROLES return { - "destroy": is_admin_or_owner, - "update": is_admin_or_owner, - "partial_update": is_admin_or_owner, - "retrieve": is_admin_or_owner, + "destroy": is_owner_or_admin, + "update": is_owner_or_admin, + "partial_update": is_owner_or_admin, + "retrieve": is_owner_or_admin, } diff --git a/src/backend/core/tests/items/test_api_item_invitations.py b/src/backend/core/tests/items/test_api_item_invitations.py index 8e820e48..3b247425 100644 --- a/src/backend/core/tests/items/test_api_item_invitations.py +++ b/src/backend/core/tests/items/test_api_item_invitations.py @@ -60,7 +60,7 @@ def test_api_item_invitations_list_authenticated_privileged( client = APIClient() client.force_login(user) - with django_assert_num_queries(3): + with django_assert_num_queries(4): response = client.get( f"/api/v1.0/items/{item.id!s}/invitations/", ) @@ -404,9 +404,9 @@ def test_api_item_invitations_create_privileged_members( # noqa: PLR0913 assert response.json() == { "errors": [ { - "attr": "role", + "attr": None, "code": "invitation_role_owner_limited_to_owners", - "detail": "Only owners of a item can invite other users as owners.", + "detail": "Only owners of an item can invite other users as owners.", }, ], "type": "validation_error", @@ -591,6 +591,36 @@ def test_api_item_invitations_create_cannot_invite_existing_users_case_insensiti } +def test_api_items_invitations_on_item_without_explicit_access(): + """Creating an invitation on an item without explicit access on it should work""" + + user = factories.UserFactory() + + root = factories.ItemFactory( + creator=user, users=[(user, "owner")], type=models.ItemTypeChoices.FOLDER + ) + + child = factories.ItemFactory( + creator=user, parent=root, type=models.ItemTypeChoices.FOLDER + ) + + invitation_values = { + "email": "JOHN.DOE@EXAMPLE.COM", + "role": random.choice(models.RoleChoices.values), + } + + client = APIClient() + client.force_login(user) + + response = client.post( + f"/api/v1.0/items/{child.id!s}/invitations/", + invitation_values, + format="json", + ) + + assert response.status_code == 201 + + # Update @@ -677,8 +707,8 @@ def test_api_item_invitations_update_authenticated_privileged_role( "errors": [ { "code": "invitation_role_owner_limited_to_owners", - "detail": "Only owners of a item can invite other users as owners.", - "attr": "role", + "detail": "Only owners of an item can invite other users as owners.", + "attr": None, } ], }