✨(models/api) allow inviting external users to a document by their email

We want to be able to share a document with a person even if this person
does not have an account in impress yet.

This code is ported from https://github.com/numerique-gouv/people.

src/backend/core/admin.py

@@ -78,15 +78,45 @@ class TemplateAdmin(admin.ModelAdmin):
 
     inlines = (TemplateAccessInline,)
 
+
 class DocumentAccessInline(admin.TabularInline):
     """Inline admin class for template accesses."""
 
     model = models.DocumentAccess
     extra = 0
 
+
 @admin.register(models.Document)
 class DocumentAdmin(admin.ModelAdmin):
     """Document admin interface declaration."""
 
     inlines = (DocumentAccessInline,)
-    
+    
+
+@admin.register(models.Invitation)
+class InvitationAdmin(admin.ModelAdmin):
+    """Admin interface to handle invitations."""
+
+    fields = (
+        "email",
+        "document",
+        "role",
+        "created_at",
+        "issuer",
+    )
+    readonly_fields = (
+        "created_at",
+        "is_expired",
+        "issuer",
+    )
+    list_display = (
+        "email",
+        "document",
+        "created_at",
+        "is_expired",
+    )
+
+    def save_model(self, request, obj, form, change):
+        obj.issuer = request.user
+        obj.save()
+

src/backend/core/api/serializers.py

@@ -66,7 +66,6 @@ class BaseAccessSerializer(serializers.ModelSerializer):
 
         # Create
         else:
-            teams = user.get_teams()
             try:
                 resource_id = self.context["resource_id"]
             except KeyError as exc:
@@ -74,6 +73,7 @@ class BaseAccessSerializer(serializers.ModelSerializer):
                     "You must set a resource ID in kwargs to create a new access."
                 ) from exc
 
+            teams = user.get_teams()
             if not self.Meta.model.objects.filter(  # pylint: disable=no-member
                 Q(user=user) | Q(team__in=teams),
                 role__in=[models.RoleChoices.OWNER, models.RoleChoices.ADMIN],
@@ -172,3 +172,82 @@ class DocumentGenerationSerializer(serializers.Serializer):
         required=False,
         default="html",
     )
+
+
+class InvitationSerializer(serializers.ModelSerializer):
+    """Serialize invitations."""
+
+    abilities = serializers.SerializerMethodField(read_only=True)
+
+    class Meta:
+        model = models.Invitation
+        fields = [
+            "id",
+            "abilities",
+            "created_at",
+            "email",
+            "document",
+            "role",
+            "issuer",
+            "is_expired",
+        ]
+        read_only_fields = [
+            "id",
+            "abilities",
+            "created_at",
+            "document",
+            "issuer",
+            "is_expired",
+        ]
+
+    def get_abilities(self, invitation) -> dict:
+        """Return abilities of the logged-in user on the instance."""
+        request = self.context.get("request")
+        if request:
+            return invitation.get_abilities(request.user)
+        return {}
+
+    def validate(self, attrs):
+        """Validate and restrict invitation to new user based on email."""
+
+        request = self.context.get("request")
+        user = getattr(request, "user", None)
+        role = attrs.get("role")
+
+        try:
+            document_id = self.context["resource_id"]
+        except KeyError as exc:
+            raise exceptions.ValidationError(
+                "You must set a document ID in kwargs to create a new document invitation."
+            ) from exc
+
+        if not user and user.is_authenticated:
+            raise exceptions.PermissionDenied(
+                "Anonymous users are not allowed to create invitations."
+            )
+
+        teams = user.get_teams()
+        if not models.DocumentAccess.objects.filter(
+            Q(user=user) | Q(team__in=teams),
+            document=document_id,
+            role__in=[models.RoleChoices.OWNER, models.RoleChoices.ADMIN],
+        ).exists():
+            raise exceptions.PermissionDenied(
+                "You are not allowed to manage invitations for this document."
+            )
+
+        if (
+            role == models.RoleChoices.OWNER
+            and not models.DocumentAccess.objects.filter(
+                Q(user=user) | Q(team__in=teams),
+                document=document_id,
+                role=models.RoleChoices.OWNER,
+            ).exists()
+        ):
+            raise exceptions.PermissionDenied(
+                "Only owners of a document can invite other users as owners."
+            )
+
+        attrs["document_id"] = document_id
+        attrs["issuer"] = user
+        return attrs

src/backend/core/api/viewsets.py

@@ -479,3 +479,78 @@ class TemplateAccessViewSet(
     queryset = models.TemplateAccess.objects.select_related("user").all()
     resource_field_name = "template"
     serializer_class = serializers.TemplateAccessSerializer
+
+
+class InvitationViewset(
+    mixins.CreateModelMixin,
+    mixins.ListModelMixin,
+    mixins.RetrieveModelMixin,
+    mixins.DestroyModelMixin,
+    viewsets.GenericViewSet,
+):
+    """API ViewSet for user invitations to document.
+
+    GET /api/v1.0/documents/<document_id>/invitations/:<invitation_id>/
+        Return list of invitations related to that document or one
+        document access if an id is provided.
+
+    POST /api/v1.0/documents/<document_id>/invitations/ with expected data:
+        - email: str
+        - role: str [owner|admin|member]
+        Return newly created invitation (issuer and document are automatically set)
+
+    PUT / PATCH : Not permitted. Instead of updating your invitation,
+        delete and create a new one.
+
+    DELETE  /api/v1.0/documents/<document_id>/invitations/<invitation_id>/
+        Delete targeted invitation
+    """
+
+    lookup_field = "id"
+    pagination_class = Pagination
+    permission_classes = [permissions.IsAuthenticated, permissions.AccessPermission]
+    queryset = (
+        models.Invitation.objects.all()
+        .select_related("document")
+        .order_by("-created_at")
+    )
+    serializer_class = serializers.InvitationSerializer
+
+    def get_serializer_context(self):
+        """Extra context provided to the serializer class."""
+        context = super().get_serializer_context()
+        context["resource_id"] = self.kwargs["resource_id"]
+        return context
+
+    def get_queryset(self):
+        """Return the queryset according to the action."""
+        queryset = super().get_queryset()
+        queryset = queryset.filter(document=self.kwargs["resource_id"])
+
+        if self.action == "list":
+            user = self.request.user
+            teams = user.get_teams()
+
+            # Determine which role the logged-in user has in the document
+            user_roles_query = (
+                models.DocumentAccess.objects.filter(
+                    Q(user=user) | Q(team__in=teams),
+                    document=self.kwargs["resource_id"],
+                )
+                .values("document")
+                .annotate(roles_array=ArrayAgg("role"))
+                .values("roles_array")
+            )
+
+            queryset = (
+                # The logged-in user should be part of a document to see its accesses
+                queryset.filter(
+                    Q(document__accesses__user=user)
+                    | Q(document__accesses__team__in=teams),
+                )
+                # Abilities are computed based on logged-in user's role and
+                # the user role on each document access
+                .annotate(user_roles=Subquery(user_roles_query))
+                .distinct()
+            )
+        return queryset

src/backend/core/factories.py

@@ -112,3 +112,15 @@ class TeamTemplateAccessFactory(factory.django.DjangoModelFactory):
     template = factory.SubFactory(TemplateFactory)
     team = factory.Sequence(lambda n: f"team{n}")
     role = factory.fuzzy.FuzzyChoice([r[0] for r in models.RoleChoices.choices])
+
+
+class InvitationFactory(factory.django.DjangoModelFactory):
+    """A factory to create invitations for a user"""
+
+    class Meta:
+        model = models.Invitation
+
+    email = factory.Faker("email")
+    document = factory.SubFactory(DocumentFactory)
+    role = factory.fuzzy.FuzzyChoice([role[0] for role in models.RoleChoices.choices])
+    issuer = factory.SubFactory(UserFactory)

src/backend/core/migrations/0002_alter_user_language_invitation_and_more.py

@@ -0,0 +1,42 @@
+# Generated by Django 5.0.3 on 2024-05-12 19:02
+
+import django.db.models.deletion
+import uuid
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('core', '0001_initial'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='user',
+            name='language',
+            field=models.CharField(choices="(('en-us', 'English'), ('fr-fr', 'French'))", default='en-us', help_text='The language in which the user wants to see the interface.', max_length=10, verbose_name='language'),
+        ),
+        migrations.CreateModel(
+            name='Invitation',
+            fields=[
+                ('id', models.UUIDField(default=uuid.uuid4, editable=False, help_text='primary key for the record as UUID', primary_key=True, serialize=False, verbose_name='id')),
+                ('created_at', models.DateTimeField(auto_now_add=True, help_text='date and time at which a record was created', verbose_name='created on')),
+                ('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated on')),
+                ('email', models.EmailField(max_length=254, verbose_name='email address')),
+                ('role', models.CharField(choices=[('member', 'Member'), ('administrator', 'Administrator'), ('owner', 'Owner')], default='member', max_length=20)),
+                ('document', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='invitations', to='core.document')),
+                ('issuer', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='invitations', to=settings.AUTH_USER_MODEL)),
+            ],
+            options={
+                'verbose_name': 'Document invitation',
+                'verbose_name_plural': 'Document invitations',
+                'db_table': 'impress_invitation',
+            },
+        ),
+        migrations.AddConstraint(
+            model_name='invitation',
+            constraint=models.UniqueConstraint(fields=('email', 'document'), name='email_and_document_unique_together'),
+        ),
+    ]

src/backend/core/models.py

@@ -3,21 +3,27 @@ Declare and configure the models for the impress core application
 """
 import hashlib
 import json
+import smtplib
 import textwrap
 import uuid
+from datetime import timedelta
+from logging import getLogger
 
 from django.conf import settings
 from django.contrib.auth import models as auth_models
 from django.contrib.auth.base_user import AbstractBaseUser
-from django.core import mail, validators
+from django.contrib.sites.models import Site
+from django.core import exceptions, mail, validators
 from django.core.files.base import ContentFile
 from django.core.files.storage import default_storage
 from django.db import models
 from django.template.base import Template as DjangoTemplate
 from django.template.context import Context
+from django.template.loader import render_to_string
+from django.utils import html, timezone
 from django.utils.functional import lazy
-from django.utils.html import format_html
 from django.utils.translation import gettext_lazy as _
+from django.utils.translation import override
 
 import frontmatter
 import markdown
@@ -26,6 +32,8 @@ from timezone_field import TimeZoneField
 from weasyprint import CSS, HTML
 from weasyprint.text.fonts import FontConfiguration
 
+logger = getLogger(__name__)
+
 
 def get_resource_roles(resource, user):
     """Compute the roles a user has on a resource."""
@@ -164,6 +172,42 @@ class User(AbstractBaseUser, BaseModel, auth_models.PermissionsMixin):
     def __str__(self):
         return self.email or self.admin_email or str(self.id)
 
+    def save(self, *args, **kwargs):
+        """
+        If it's a new user, give its user access to the documents to which s.he was invited.
+        """
+        is_adding = self._state.adding
+        super().save(*args, **kwargs)
+
+        if is_adding:
+            self._convert_valid_invitations()
+
+    def _convert_valid_invitations(self):
+        """
+        Convert valid invitations to document accesses.
+        Expired invitations are ignored.
+        """
+        valid_invitations = Invitation.objects.filter(
+            email=self.email,
+            created_at__gte=(
+                timezone.now()
+                - timedelta(seconds=settings.INVITATION_VALIDITY_DURATION)
+            ),
+        ).select_related("document")
+
+        if not valid_invitations.exists():
+            return
+
+        DocumentAccess.objects.bulk_create(
+            [
+                DocumentAccess(
+                    user=self, document=invitation.document, role=invitation.role
+                )
+                for invitation in valid_invitations
+            ]
+        )
+        valid_invitations.delete()
+
     def email_user(self, subject, message, from_email=None, **kwargs):
         """Email this user."""
         if not self.email:
@@ -523,7 +567,7 @@ class Template(BaseModel):
 
         document_html = HTML(
             string=DjangoTemplate(self.code).render(
-                Context({"body": format_html(body_html), **metadata})
+                Context({"body": html.format_html(body_html), **metadata})
             )
         )
         css = CSS(
@@ -576,3 +620,110 @@ class TemplateAccess(BaseAccess):
         Compute and return abilities for a given user on the template access.
         """
         return self._get_abilities(self.template, user)
+
+
+class Invitation(BaseModel):
+    """User invitation to a document."""
+
+    email = models.EmailField(_("email address"), null=False, blank=False)
+    document = models.ForeignKey(
+        Document,
+        on_delete=models.CASCADE,
+        related_name="invitations",
+    )
+    role = models.CharField(
+        max_length=20, choices=RoleChoices.choices, default=RoleChoices.MEMBER
+    )
+    issuer = models.ForeignKey(
+        User,
+        on_delete=models.CASCADE,
+        related_name="invitations",
+    )
+
+    class Meta:
+        db_table = "impress_invitation"
+        verbose_name = _("Document invitation")
+        verbose_name_plural = _("Document invitations")
+        constraints = [
+            models.UniqueConstraint(
+                fields=["email", "document"], name="email_and_document_unique_together"
+            )
+        ]
+
+    def __str__(self):
+        return f"{self.email} invited to {self.document}"
+
+    def save(self, *args, **kwargs):
+        """Make invitations read-only."""
+        if self.created_at:
+            raise exceptions.PermissionDenied()
+
+        super().save(*args, **kwargs)
+        self.email_invitation()
+
+    def clean(self):
+        """Validate fields."""
+        super().clean()
+
+        # Check if an identity already exists for the provided email
+        if User.objects.filter(email=self.email).exists():
+            raise exceptions.ValidationError(
+                {"email": _("This email is already associated to a registered user.")}
+            )
+
+    @property
+    def is_expired(self):
+        """Calculate if invitation is still valid or has expired."""
+        if not self.created_at:
+            return None
+
+        validity_duration = timedelta(seconds=settings.INVITATION_VALIDITY_DURATION)
+        return timezone.now() > (self.created_at + validity_duration)
+
+    def get_abilities(self, user):
+        """Compute and return abilities for a given user."""
+        can_delete = False
+        roles = []
+
+        if user.is_authenticated:
+            teams = user.get_teams()
+            try:
+                roles = self.user_roles or []
+            except AttributeError:
+                try:
+                    roles = self.document.accesses.filter(
+                        models.Q(user=user) | models.Q(team__in=teams),
+                    ).values_list("role", flat=True)
+                except (self._meta.model.DoesNotExist, IndexError):
+                    roles = []
+
+            can_delete = bool(
+                set(roles).intersection({RoleChoices.OWNER, RoleChoices.ADMIN})
+            )
+
+        return {
+            "destroy": can_delete,
+            "update": False,
+            "partial_update": False,
+            "retrieve": bool(roles),
+        }
+
+    def email_invitation(self):
+        """Email invitation to the user."""
+        try:
+            with override(self.issuer.language):
+                title = _("Invitation to join Impress!")
+                template_vars = {"title": title, "site": Site.objects.get_current()}
+                msg_html = render_to_string("mail/html/invitation.html", template_vars)
+                msg_plain = render_to_string("mail/text/invitation.txt", template_vars)
+                mail.send_mail(
+                    title,
+                    msg_plain,
+                    settings.EMAIL_FROM,
+                    [self.email],
+                    html_message=msg_html,
+                    fail_silently=False,
+                )
+
+        except smtplib.SMTPException as exception:
+            logger.error("invitation to %s was not sent: %s", self.email, exception)

src/backend/core/templatetags/extra_tags.py

@@ -0,0 +1,58 @@
+"""Custom template tags for the core application of People."""
+
+import base64
+
+from django import template
+from django.contrib.staticfiles import finders
+
+from PIL import ImageFile as PillowImageFile
+
+register = template.Library()
+
+
+def image_to_base64(file_or_path, close=False):
+    """
+    Return the src string of the base64 encoding of an image represented by its path
+    or file opened or not.
+
+    Inspired by Django's "get_image_dimensions"
+    """
+    pil_parser = PillowImageFile.Parser()
+    if hasattr(file_or_path, "read"):
+        file = file_or_path
+        if file.closed and hasattr(file, "open"):
+            file_or_path.open()
+        file_pos = file.tell()
+        file.seek(0)
+    else:
+        try:
+            # pylint: disable=consider-using-with
+            file = open(file_or_path, "rb")
+        except OSError:
+            return ""
+        close = True
+
+    try:
+        image_data = file.read()
+        if not image_data:
+            return ""
+        pil_parser.feed(image_data)
+        if pil_parser.image:
+            mime_type = pil_parser.image.get_format_mimetype()
+            encoded_string = base64.b64encode(image_data)
+            return f"data:{mime_type:s};base64, {encoded_string.decode('utf-8'):s}"
+        return ""
+    finally:
+        if close:
+            file.close()
+        else:
+            file.seek(file_pos)
+
+
+@register.simple_tag
+def base64_static(path):
+    """Return a static file into a base64."""
+    full_path = finders.find(path)
+    if full_path:
+        return image_to_base64(full_path, True)
+    return ""

src/backend/core/tests/test_api_document_invitations.py

@@ -0,0 +1,494 @@
+"""
+Unit tests for the Invitation model
+"""
+import random
+import time
+
+import pytest
+from rest_framework import status
+from rest_framework.test import APIClient
+
+from core import factories, models
+from core.tests.conftest import TEAM, USER, VIA
+
+pytestmark = pytest.mark.django_db
+
+
+def test_api_document_invitations__create__anonymous():
+    """Anonymous users should not be able to create invitations."""
+    document = factories.DocumentFactory()
+    invitation_values = {
+        "email": "guest@example.com",
+        "role": random.choice(models.RoleChoices.choices)[0],
+    }
+
+    response = APIClient().post(
+        f"/api/v1.0/documents/{document.id}/invitations/",
+        invitation_values,
+        format="json",
+    )
+
+    assert response.status_code == status.HTTP_401_UNAUTHORIZED
+    assert response.json() == {
+        "detail": "Authentication credentials were not provided."
+    }
+
+
+def test_api_document_invitations__create__authenticated_outsider():
+    """Users outside of document should not be permitted to invite to document."""
+    user = factories.UserFactory()
+    document = factories.DocumentFactory()
+    invitation_values = {
+        "email": "guest@example.com",
+        "role": random.choice(models.RoleChoices.choices)[0],
+    }
+
+    client = APIClient()
+    client.force_login(user)
+    response = client.post(
+        f"/api/v1.0/documents/{document.id}/invitations/",
+        invitation_values,
+        format="json",
+    )
+
+    assert response.status_code == status.HTTP_403_FORBIDDEN
+
+
+@pytest.mark.parametrize(
+    "inviting,invited,is_allowed",
+    (
+        ["member", "member", False],
+        ["member", "administrator", False],
+        ["member", "owner", False],
+        ["administrator", "member", True],
+        ["administrator", "administrator", True],
+        ["administrator", "owner", False],
+        ["owner", "member", True],
+        ["owner", "administrator", True],
+        ["owner", "owner", True],
+    ),
+)
+@pytest.mark.parametrize("via", VIA)
+def test_api_document_invitations__create__privileged_members(
+    via, inviting, invited, is_allowed, mock_user_get_teams
+):
+    """
+    Only owners and administrators should be able to invite new users.
+    Only owners can invite owners.
+    """
+    user = factories.UserFactory()
+    document = factories.DocumentFactory()
+    if via == USER:
+        factories.UserDocumentAccessFactory(document=document, user=user, role=inviting)
+    elif via == TEAM:
+        mock_user_get_teams.return_value = ["lasuite", "unknown"]
+        factories.TeamDocumentAccessFactory(
+            document=document, team="lasuite", role=inviting
+        )
+
+    invitation_values = {
+        "email": "guest@example.com",
+        "role": invited,
+    }
+
+    client = APIClient()
+    client.force_login(user)
+    response = client.post(
+        f"/api/v1.0/documents/{document.id}/invitations/",
+        invitation_values,
+        format="json",
+    )
+    if is_allowed:
+        assert response.status_code == status.HTTP_201_CREATED
+        assert models.Invitation.objects.count() == 1
+    else:
+        assert response.status_code == status.HTTP_403_FORBIDDEN
+        assert models.Invitation.objects.exists() is False
+
+
+def test_api_document_invitations__create__issuer_and_document_override():
+    """It should not be possible to set the "document" and "issuer" fields."""
+    user = factories.UserFactory()
+    document = factories.DocumentFactory(users=[(user, "owner")])
+    other_document = factories.DocumentFactory(users=[(user, "owner")])
+    invitation_values = {
+        "document": str(other_document.id),
+        "issuer": str(factories.UserFactory().id),
+        "email": "guest@example.com",
+        "role": random.choice(models.RoleChoices.choices)[0],
+    }
+
+    client = APIClient()
+    client.force_login(user)
+    response = client.post(
+        f"/api/v1.0/documents/{document.id}/invitations/",
+        invitation_values,
+        format="json",
+    )
+
+    assert response.status_code == status.HTTP_201_CREATED
+    # document and issuer automatically set
+    assert response.json()["document"] == str(document.id)
+    assert response.json()["issuer"] == str(user.id)
+
+
+def test_api_document_invitations__create__cannot_duplicate_invitation():
+    """An email should not be invited multiple times to the same document."""
+    existing_invitation = factories.InvitationFactory()
+    document = existing_invitation.document
+
+    # Grant privileged role on the Document to the user
+    user = factories.UserFactory()
+    models.DocumentAccess.objects.create(
+        document=document, user=user, role="administrator"
+    )
+
+    # Create a new invitation to the same document with the exact same email address
+    invitation_values = {
+        "email": existing_invitation.email,
+        "role": random.choice(["administrator", "member"]),
+    }
+
+    client = APIClient()
+    client.force_login(user)
+    response = client.post(
+        f"/api/v1.0/documents/{document.id}/invitations/",
+        invitation_values,
+        format="json",
+    )
+
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+    assert response.json()["__all__"] == [
+        "Document invitation with this Email address and Document already exists."
+    ]
+
+
+def test_api_document_invitations__create__cannot_invite_existing_users():
+    """
+    It should not be possible to invite already existing users.
+    """
+    user = factories.UserFactory()
+    document = factories.DocumentFactory(users=[(user, "owner")])
+    existing_user = factories.UserFactory()
+
+    # Build an invitation to the email of an exising identity in the db
+    invitation_values = {
+        "email": existing_user.email,
+        "role": random.choice(models.RoleChoices.choices)[0],
+    }
+
+    client = APIClient()
+    client.force_login(user)
+    response = client.post(
+        f"/api/v1.0/documents/{document.id}/invitations/",
+        invitation_values,
+        format="json",
+    )
+
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+    assert response.json()["email"] == [
+        "This email is already associated to a registered user."
+    ]
+
+
+def test_api_document_invitations__list__anonymous_user():
+    """Anonymous users should not be able to list invitations."""
+    document = factories.DocumentFactory()
+    response = APIClient().get(f"/api/v1.0/documents/{document.id}/invitations/")
+    assert response.status_code == status.HTTP_401_UNAUTHORIZED
+
+
+@pytest.mark.parametrize("via", VIA)
+def test_api_document_invitations__list__authenticated(
+    via, mock_user_get_teams, django_assert_num_queries
+):
+    """
+    Authenticated users should be able to list invitations for documents to which they are
+    related, whatever the role and including invitations issued by other users.
+    """
+    user = factories.UserFactory()
+    other_user = factories.UserFactory()
+    document = factories.DocumentFactory()
+    role = random.choice(models.RoleChoices.choices)[0]
+    if via == USER:
+        factories.UserDocumentAccessFactory(document=document, user=user, role=role)
+    elif via == TEAM:
+        mock_user_get_teams.return_value = ["lasuite", "unknown"]
+        factories.TeamDocumentAccessFactory(
+            document=document, team="lasuite", role=role
+        )
+
+    invitation = factories.InvitationFactory(
+        document=document, role="administrator", issuer=user
+    )
+    other_invitations = factories.InvitationFactory.create_batch(
+        2, document=document, role="member", issuer=other_user
+    )
+
+    # invitations from other documents should not be listed
+    other_document = factories.DocumentFactory()
+    factories.InvitationFactory.create_batch(2, document=other_document, role="member")
+
+    client = APIClient()
+    client.force_login(user)
+    with django_assert_num_queries(3):
+        response = client.get(
+            f"/api/v1.0/documents/{document.id}/invitations/",
+        )
+    assert response.status_code == status.HTTP_200_OK
+    assert response.json()["count"] == 3
+    assert sorted(response.json()["results"], key=lambda x: x["created_at"]) == sorted(
+        [
+            {
+                "id": str(i.id),
+                "created_at": i.created_at.isoformat().replace("+00:00", "Z"),
+                "email": str(i.email),
+                "document": str(document.id),
+                "role": i.role,
+                "issuer": str(i.issuer.id),
+                "is_expired": False,
+                "abilities": {
+                    "destroy": role in ["administrator", "owner"],
+                    "update": False,
+                    "partial_update": False,
+                    "retrieve": True,
+                },
+            }
+            for i in [invitation, *other_invitations]
+        ],
+        key=lambda x: x["created_at"],
+    )
+
+
+def test_api_document_invitations__list__expired_invitations_still_listed(settings):
+    """
+    Expired invitations are still listed.
+    """
+    user = factories.UserFactory()
+    other_user = factories.UserFactory()
+
+    document = factories.DocumentFactory(
+        users=[(user, "administrator"), (other_user, "owner")]
+    )
+
+    # override settings to accelerate validation expiration
+    settings.INVITATION_VALIDITY_DURATION = 1  # second
+    expired_invitation = factories.InvitationFactory(
+        document=document,
+        role="member",
+        issuer=user,
+    )
+    time.sleep(1)
+
+    client = APIClient()
+    client.force_login(user)
+    response = client.get(
+        f"/api/v1.0/documents/{document.id}/invitations/",
+    )
+    assert response.status_code == status.HTTP_200_OK
+    assert response.json()["count"] == 1
+    assert sorted(response.json()["results"], key=lambda x: x["created_at"]) == sorted(
+        [
+            {
+                "id": str(expired_invitation.id),
+                "created_at": expired_invitation.created_at.isoformat().replace(
+                    "+00:00", "Z"
+                ),
+                "email": str(expired_invitation.email),
+                "document": str(document.id),
+                "role": expired_invitation.role,
+                "issuer": str(expired_invitation.issuer.id),
+                "is_expired": True,
+                "abilities": {
+                    "destroy": True,
+                    "update": False,
+                    "partial_update": False,
+                    "retrieve": True,
+                },
+            },
+        ],
+        key=lambda x: x["created_at"],
+    )
+
+
+def test_api_document_invitations__retrieve__anonymous_user():
+    """
+    Anonymous users should not be able to retrieve invitations.
+    """
+
+    invitation = factories.InvitationFactory()
+    response = APIClient().get(
+        f"/api/v1.0/documents/{invitation.document.id}/invitations/{invitation.id}/",
+    )
+
+    assert response.status_code == status.HTTP_401_UNAUTHORIZED
+
+
+def test_api_document_invitations__retrieve__unrelated_user():
+    """
+    Authenticated unrelated users should not be able to retrieve invitations.
+    """
+    user = factories.UserFactory()
+    invitation = factories.InvitationFactory()
+
+    client = APIClient()
+    client.force_login(user)
+    response = client.get(
+        f"/api/v1.0/documents/{invitation.document.id!s}/invitations/{invitation.id!s}/",
+    )
+
+    assert response.status_code == status.HTTP_403_FORBIDDEN
+
+
+@pytest.mark.parametrize("via", VIA)
+def test_api_document_invitations__retrieve__document_member(via, mock_user_get_teams):
+    """
+    Authenticated users related to the document should be able to retrieve invitations
+    whatever their role in the document.
+    """
+    user = factories.UserFactory()
+    invitation = factories.InvitationFactory()
+    role = random.choice(models.RoleChoices.choices)[0]
+    if via == USER:
+        factories.UserDocumentAccessFactory(
+            document=invitation.document, user=user, role=role
+        )
+    elif via == TEAM:
+        mock_user_get_teams.return_value = ["lasuite", "unknown"]
+        factories.TeamDocumentAccessFactory(
+            document=invitation.document, team="lasuite", role=role
+        )
+
+    client = APIClient()
+    client.force_login(user)
+    response = client.get(
+        f"/api/v1.0/documents/{invitation.document.id}/invitations/{invitation.id}/",
+    )
+
+    assert response.status_code == status.HTTP_200_OK
+    assert response.json() == {
+        "id": str(invitation.id),
+        "created_at": invitation.created_at.isoformat().replace("+00:00", "Z"),
+        "email": invitation.email,
+        "document": str(invitation.document.id),
+        "role": str(invitation.role),
+        "issuer": str(invitation.issuer.id),
+        "is_expired": False,
+        "abilities": {
+            "destroy": role in ["administrator", "owner"],
+            "update": False,
+            "partial_update": False,
+            "retrieve": True,
+        },
+    }
+
+
+@pytest.mark.parametrize("via", VIA)
+@pytest.mark.parametrize(
+    "method",
+    ["put", "patch"],
+)
+def test_api_document_invitations__update__forbidden(method, via, mock_user_get_teams):
+    """
+    Update of invitations is currently forbidden.
+    """
+    user = factories.UserFactory()
+    invitation = factories.InvitationFactory()
+    if via == USER:
+        factories.UserDocumentAccessFactory(
+            document=invitation.document, user=user, role="owner"
+        )
+    elif via == TEAM:
+        mock_user_get_teams.return_value = ["lasuite", "unknown"]
+        factories.TeamDocumentAccessFactory(
+            document=invitation.document, team="lasuite", role="owner"
+        )
+
+    client = APIClient()
+    client.force_login(user)
+    url = f"/api/v1.0/documents/{invitation.document.id}/invitations/{invitation.id}/"
+    if method == "put":
+        response = client.put(url)
+    if method == "patch":
+        response = client.patch(url)
+
+    assert response.status_code == status.HTTP_405_METHOD_NOT_ALLOWED
+    assert response.json()["detail"] == f'Method "{method.upper()}" not allowed.'
+
+
+def test_api_document_invitations__delete__anonymous():
+    """Anonymous user should not be able to delete invitations."""
+    invitation = factories.InvitationFactory()
+
+    response = APIClient().delete(
+        f"/api/v1.0/documents/{invitation.document.id}/invitations/{invitation.id}/",
+    )
+    assert response.status_code == status.HTTP_401_UNAUTHORIZED
+
+
+def test_api_document_invitations__delete__authenticated_outsider():
+    """Members unrelated to a document should not be allowed to cancel invitations."""
+    user = factories.UserFactory()
+
+    document = factories.DocumentFactory()
+    invitation = factories.InvitationFactory(document=document)
+
+    client = APIClient()
+    client.force_login(user)
+    response = client.delete(
+        f"/api/v1.0/documents/{document.id}/invitations/{invitation.id}/",
+    )
+    assert response.status_code == status.HTTP_403_FORBIDDEN
+
+
+@pytest.mark.parametrize("via", VIA)
+@pytest.mark.parametrize("role", ["owner", "administrator"])
+def test_api_document_invitations__delete__privileged_members(
+    role, via, mock_user_get_teams
+):
+    """Privileged member should be able to cancel invitation."""
+    user = factories.UserFactory()
+    document = factories.DocumentFactory()
+    if via == USER:
+        factories.UserDocumentAccessFactory(document=document, user=user, role=role)
+    elif via == TEAM:
+        mock_user_get_teams.return_value = ["lasuite", "unknown"]
+        factories.TeamDocumentAccessFactory(
+            document=document, team="lasuite", role=role
+        )
+
+    invitation = factories.InvitationFactory(document=document)
+
+    client = APIClient()
+    client.force_login(user)
+    response = client.delete(
+        f"/api/v1.0/documents/{document.id}/invitations/{invitation.id}/",
+    )
+    assert response.status_code == status.HTTP_204_NO_CONTENT
+
+
+@pytest.mark.parametrize("via", VIA)
+def test_api_document_invitations__delete__members(via, mock_user_get_teams):
+    """Member should not be able to cancel invitation."""
+    user = factories.UserFactory()
+    document = factories.DocumentFactory()
+    if via == USER:
+        factories.UserDocumentAccessFactory(document=document, user=user, role="member")
+    elif via == TEAM:
+        mock_user_get_teams.return_value = ["lasuite", "unknown"]
+        factories.TeamDocumentAccessFactory(
+            document=document, team="lasuite", role="member"
+        )
+
+    invitation = factories.InvitationFactory(document=document)
+
+    client = APIClient()
+    client.force_login(user)
+    response = client.delete(
+        f"/api/v1.0/documents/{document.id}/invitations/{invitation.id}/",
+    )
+    assert response.status_code == status.HTTP_403_FORBIDDEN
+    assert (
+        response.json()["detail"]
+        == "You do not have permission to perform this action."
+    )

src/backend/core/tests/test_models_invitations.py

@@ -0,0 +1,312 @@
+"""
+Unit tests for the Invitation model
+"""
+
+import smtplib
+import time
+from logging import Logger
+from unittest import mock
+
+from django.contrib.auth.models import AnonymousUser
+from django.core import exceptions, mail
+
+import pytest
+from faker import Faker
+from freezegun import freeze_time
+
+from core import factories, models
+from core.tests.conftest import TEAM, USER, VIA
+
+pytestmark = pytest.mark.django_db
+
+
+fake = Faker()
+
+
+def test_models_invitations_readonly_after_create():
+    """Existing invitations should be readonly."""
+    invitation = factories.InvitationFactory()
+    with pytest.raises(exceptions.PermissionDenied):
+        invitation.save()
+
+
+def test_models_invitations_email_no_empty_mail():
+    """The "email" field should not be empty."""
+    with pytest.raises(exceptions.ValidationError, match="This field cannot be blank"):
+        factories.InvitationFactory(email="")
+
+
+def test_models_invitations_email_no_null_mail():
+    """The "email" field is required."""
+    with pytest.raises(exceptions.ValidationError, match="This field cannot be null"):
+        factories.InvitationFactory(email=None)
+
+
+def test_models_invitations_document_required():
+    """The "document" field is required."""
+    with pytest.raises(exceptions.ValidationError, match="This field cannot be null"):
+        factories.InvitationFactory(document=None)
+
+
+def test_models_invitations_document_should_be_document_instance():
+    """The "document" field should be a document instance."""
+    with pytest.raises(
+        ValueError, match='Invitation.document" must be a "Document" instance'
+    ):
+        factories.InvitationFactory(document="ee")
+
+
+def test_models_invitations_role_required():
+    """The "role" field is required."""
+    with pytest.raises(exceptions.ValidationError, match="This field cannot be blank"):
+        factories.InvitationFactory(role="")
+
+
+def test_models_invitations_role_among_choices():
+    """The "role" field should be a valid choice."""
+    with pytest.raises(
+        exceptions.ValidationError, match="Value 'boss' is not a valid choice"
+    ):
+        factories.InvitationFactory(role="boss")
+
+
+def test_models_invitations__is_expired(settings):
+    """
+    The 'is_expired' property should return False until validity duration
+    is exceeded and True afterwards.
+    """
+    expired_invitation = factories.InvitationFactory()
+    assert expired_invitation.is_expired is False
+
+    settings.INVITATION_VALIDITY_DURATION = 1
+    time.sleep(1)
+
+    assert expired_invitation.is_expired is True
+
+
+def test_models_invitation__new_user__convert_invitations_to_accesses():
+    """
+    Upon creating a new user, invitations linked to the email
+    should be converted to accesses and then deleted.
+    """
+    # Two invitations to the same mail but to different documents
+    invitation_to_document1 = factories.InvitationFactory()
+    invitation_to_document2 = factories.InvitationFactory(
+        email=invitation_to_document1.email
+    )
+
+    other_invitation = factories.InvitationFactory(
+        document=invitation_to_document2.document
+    )  # another person invited to document2
+
+    new_user = factories.UserFactory(email=invitation_to_document1.email)
+
+    # The invitation regarding
+    assert models.DocumentAccess.objects.filter(
+        document=invitation_to_document1.document, user=new_user
+    ).exists()
+    assert models.DocumentAccess.objects.filter(
+        document=invitation_to_document2.document, user=new_user
+    ).exists()
+    assert not models.Invitation.objects.filter(
+        document=invitation_to_document1.document, email=invitation_to_document1.email
+    ).exists()  # invitation "consumed"
+    assert not models.Invitation.objects.filter(
+        document=invitation_to_document2.document, email=invitation_to_document2.email
+    ).exists()  # invitation "consumed"
+    assert models.Invitation.objects.filter(
+        document=invitation_to_document2.document, email=other_invitation.email
+    ).exists()  # the other invitation remains
+
+
+def test_models_invitation__new_user__filter_expired_invitations():
+    """
+    Upon creating a new identity, valid invitations should be converted into accesses
+    and expired invitations should remain unchanged.
+    """
+    document = factories.DocumentFactory()
+    with freeze_time("2020-01-01"):
+        expired_invitation = factories.InvitationFactory(document=document)
+    user_email = expired_invitation.email
+    valid_invitation = factories.InvitationFactory(email=user_email)
+
+    new_user = factories.UserFactory(email=user_email)
+
+    # valid invitation should have granted access to the related document
+    assert models.DocumentAccess.objects.filter(
+        document=valid_invitation.document, user=new_user
+    ).exists()
+    assert not models.Invitation.objects.filter(
+        document=valid_invitation.document, email=user_email
+    ).exists()
+
+    # expired invitation should not have been consumed
+    assert not models.DocumentAccess.objects.filter(
+        document=expired_invitation.document, user=new_user
+    ).exists()
+    assert models.Invitation.objects.filter(
+        document=expired_invitation.document, email=user_email
+    ).exists()
+
+
+@pytest.mark.parametrize("num_invitations, num_queries", [(0, 3), (1, 6), (20, 6)])
+def test_models_invitation__new_user__user_creation_constant_num_queries(
+    django_assert_num_queries, num_invitations, num_queries
+):
+    """
+    The number of queries executed during user creation should not be proportional
+    to the number of invitations being processed.
+    """
+    user_email = fake.email()
+
+    if num_invitations != 0:
+        factories.InvitationFactory.create_batch(num_invitations, email=user_email)
+
+    # with no invitation, we skip an "if", resulting in 8 requests
+    # otherwise, we should have 11 queries with any number of invitations
+    with django_assert_num_queries(num_queries):
+        models.User.objects.create(email=user_email, password="!")
+
+
+def test_models_document_invitations_email():
+    """Check email invitation during invitation creation."""
+    member_access = factories.UserDocumentAccessFactory(role="member")
+    document = member_access.document
+
+    # pylint: disable-next=no-member
+    assert len(mail.outbox) == 0
+
+    factories.UserDocumentAccessFactory(document=document)
+    invitation = factories.InvitationFactory(document=document, email="john@people.com")
+
+    # pylint: disable-next=no-member
+    assert len(mail.outbox) == 1
+
+    # pylint: disable-next=no-member
+    email = mail.outbox[0]
+
+    assert email.to == [invitation.email]
+    assert email.subject == "Invitation to join Impress!"
+
+    email_content = " ".join(email.body.split())
+    assert "Invitation to join Impress!" in email_content
+    assert "[//example.com]" in email_content
+
+
+@mock.patch(
+    "django.core.mail.send_mail",
+    side_effect=smtplib.SMTPException("Error SMTPException"),
+)
+@mock.patch.object(Logger, "error")
+def test_models_document_invitations_email_failed(mock_logger, _mock_send_mail):
+    """Check invitation behavior when an SMTP error occurs during invitation creation."""
+
+    member_access = factories.UserDocumentAccessFactory(role="member")
+    document = member_access.document
+
+    # pylint: disable-next=no-member
+    assert len(mail.outbox) == 0
+
+    factories.UserDocumentAccessFactory(document=document)
+
+    # No error should be raised
+    invitation = factories.InvitationFactory(document=document, email="john@people.com")
+
+    # No email has been sent
+    # pylint: disable-next=no-member
+    assert len(mail.outbox) == 0
+
+    # Logger should be called
+    mock_logger.assert_called_once()
+
+    (
+        _,
+        email,
+        exception,
+    ) = mock_logger.call_args.args
+
+    assert email == invitation.email
+    assert isinstance(exception, smtplib.SMTPException)
+
+
+# get_abilities
+
+
+def test_models_document_invitations_get_abilities_anonymous():
+    """Check abilities returned for an anonymous user."""
+    access = factories.InvitationFactory()
+    abilities = access.get_abilities(AnonymousUser())
+    assert abilities == {
+        "destroy": False,
+        "retrieve": False,
+        "partial_update": False,
+        "update": False,
+    }
+
+
+def test_models_document_invitations_get_abilities_authenticated():
+    """Check abilities returned for an authenticated user."""
+    access = factories.InvitationFactory()
+    user = factories.UserFactory()
+    abilities = access.get_abilities(user)
+    assert abilities == {
+        "destroy": False,
+        "retrieve": False,
+        "partial_update": False,
+        "update": False,
+    }
+
+
+@pytest.mark.parametrize("via", VIA)
+@pytest.mark.parametrize("role", ["administrator", "owner"])
+def test_models_document_invitations_get_abilities_privileged_member(
+    role, via, mock_user_get_teams
+):
+    """Check abilities for a document member with a privileged role."""
+
+    user = factories.UserFactory()
+    document = factories.DocumentFactory()
+    if via == USER:
+        factories.UserDocumentAccessFactory(document=document, user=user, role=role)
+    elif via == TEAM:
+        mock_user_get_teams.return_value = ["lasuite", "unknown"]
+        factories.TeamDocumentAccessFactory(
+            document=document, team="lasuite", role=role
+        )
+
+    factories.UserDocumentAccessFactory(document=document)  # another one
+
+    invitation = factories.InvitationFactory(document=document)
+    abilities = invitation.get_abilities(user)
+
+    assert abilities == {
+        "destroy": True,
+        "retrieve": True,
+        "partial_update": False,
+        "update": False,
+    }
+
+
+@pytest.mark.parametrize("via", VIA)
+def test_models_document_invitations_get_abilities_member(via, mock_user_get_teams):
+    """Check abilities for a document member with 'member' role."""
+
+    user = factories.UserFactory()
+    document = factories.DocumentFactory()
+    if via == USER:
+        factories.UserDocumentAccessFactory(document=document, user=user, role="member")
+    elif via == TEAM:
+        mock_user_get_teams.return_value = ["lasuite", "unknown"]
+        factories.TeamDocumentAccessFactory(
+            document=document, team="lasuite", role="member"
+        )
+
+    invitation = factories.InvitationFactory(document=document)
+    abilities = invitation.get_abilities(user)
+
+    assert abilities == {
+        "destroy": False,
+        "retrieve": True,
+        "partial_update": False,
+        "update": False,
+    }

src/backend/core/urls.py

@@ -20,6 +20,12 @@ document_related_router.register(
     viewsets.DocumentAccessViewSet,
     basename="document_accesses",
 )
+document_related_router.register(
+    "invitations",
+    viewsets.InvitationViewset,
+    basename="invitations",
+)
+
 
 # - Routes nested under a template
 template_related_router = DefaultRouter()

src/backend/impress/settings.py

@@ -281,6 +281,7 @@ class Base(Configuration):
     EMAIL_FROM = values.Value("from@example.com")
 
     AUTH_USER_MODEL = "core.User"
+    INVITATION_VALIDITY_DURATION = 604800  # 7 days, in seconds
 
     # CORS
     CORS_ALLOW_CREDENTIALS = True

src/backend/pyproject.toml

@@ -42,6 +42,7 @@ dependencies = [
     "dockerflow==2022.8.0",
     "easy_thumbnails==2.8.5",
     "factory_boy==3.3.0",
+    "freezegun==1.5.0",
     "gunicorn==22.0.0",
     "jsonschema==4.20.0",
     "markdown==3.5.1",