🐛(backend) allow inviting external person on item with no direct access

When a user with priviledged role on an item but inherited he can not
invite external person. We refacto the invitation viewset, serializer
and permission to check roles against all the tree.
This commit is contained in:
Manuel Raynaud
2026-03-03 14:46:54 +01:00
parent a19d70cb4e
commit bd44d4aab8
6 changed files with 168 additions and 145 deletions
+4
View File
@@ -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
+29 -42
View File
@@ -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)
-21
View File
@@ -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
+35 -32
View File
@@ -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,
+65 -45
View File
@@ -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,
}
@@ -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,
}
],
}