✨(backend) add server-to-server API endpoint to create documents

We want trusted external applications to be able to create documents
via the API on behalf of any user. The user may or may not pre-exist
in our database and should be notified of the document creation by
email.

Author: sampaccoud

CHANGELOG.md

@@ -9,6 +9,11 @@ and this project adheres to
 
 ## [Unreleased]
 
+## Added
+
+- ✨(backend) add server-to-server API endpoint to create documents #467
+
+
 ## [1.9.0] - 2024-12-11
 
 ## Added

src/backend/core/api/serializers.py

@@ -4,13 +4,18 @@
 
 from django.conf import settings
 from django.db.models import Q
+from django.utils.functional import lazy
 from django.utils.translation import gettext_lazy as _
 
 import magic
 from rest_framework import exceptions, serializers
 
 from core import enums, models
 from core.services.ai_services import AI_ACTIONS
+from core.services.converter_services import (
+    ConversionError,
+    YdocConverter,
+)
 
 
 class UserSerializer(serializers.ModelSerializer):
@@ -227,6 +232,96 @@ def validate_id(self, value):
         return value
 
 
+class ServerCreateDocumentSerializer(serializers.Serializer):
+    """
+    Serializer for creating a document from a server-to-server request.
+
+    Expects 'content' as a markdown string, which is converted to our internal format
+    via a Node.js microservice. The conversion is handled automatically, so third parties
+    only need to provide markdown.
+
+    Both "sub" and "email" are required because the external app calling doesn't know
+    if the user will pre-exist in Docs database. If the user pre-exist, we will ignore the
+    submitted "email" field and use the email address set on the user account in our database
+    """
+
+    # Document
+    title = serializers.CharField(required=True)
+    content = serializers.CharField(required=True)
+    # User
+    sub = serializers.CharField(
+        required=True, validators=[models.User.sub_validator], max_length=255
+    )
+    email = serializers.EmailField(required=True)
+    language = serializers.ChoiceField(
+        required=False, choices=lazy(lambda: settings.LANGUAGES, tuple)()
+    )
+    # Invitation
+    message = serializers.CharField(required=False)
+    subject = serializers.CharField(required=False)
+
+    def create(self, validated_data):
+        """Create the document and associate it with the user or send an invitation."""
+        language = validated_data.get("language", settings.LANGUAGE_CODE)
+
+        # Get the user based on the sub (unique identifier)
+        try:
+            user = models.User.objects.get(sub=validated_data["sub"])
+        except (models.User.DoesNotExist, KeyError):
+            user = None
+            email = validated_data["email"]
+        else:
+            email = user.email
+            language = user.language or language
+
+        try:
+            converter_response = YdocConverter().convert_markdown(
+                validated_data["content"]
+            )
+        except ConversionError as err:
+            raise exceptions.APIException(detail="could not convert content") from err
+
+        document = models.Document.objects.create(
+            title=validated_data["title"],
+            content=converter_response["content"],
+            creator=user,
+        )
+
+        if user:
+            # Associate the document with the pre-existing user
+            models.DocumentAccess.objects.create(
+                document=document,
+                role=models.RoleChoices.OWNER,
+                user=user,
+            )
+        else:
+            # The user doesn't exist in our database: we need to invite him/her
+            models.Invitation.objects.create(
+                document=document,
+                email=email,
+                role=models.RoleChoices.OWNER,
+            )
+
+        # Notify the user about the newly created document
+        subject = validated_data.get("subject") or _(
+            "A new document was created on your behalf!"
+        )
+        context = {
+            "message": validated_data.get("message")
+            or _("You have been granted ownership of a new document:"),
+            "title": subject,
+        }
+        document.send_email(subject, [email], context, language)
+
+        return document
+
+    def update(self, instance, validated_data):
+        """
+        This serializer does not support updates.
+        """
+        raise NotImplementedError("Update is not supported for this serializer.")
+
+
 class LinkDocumentSerializer(BaseResourceSerializer):
     """
     Serialize link configuration for documents.

src/backend/core/api/viewsets.py

@@ -25,10 +25,11 @@
 import rest_framework as drf
 from botocore.exceptions import ClientError
 from django_filters import rest_framework as drf_filters
-from rest_framework import filters
+from rest_framework import filters, status
+from rest_framework import response as drf_response
 from rest_framework.permissions import AllowAny
 
-from core import enums, models
+from core import authentication, enums, models
 from core.services.ai_services import AIService
 from core.services.collaboration_services import CollaborationService
 
@@ -430,6 +431,30 @@ def perform_create(self, serializer):
             role=models.RoleChoices.OWNER,
         )
 
+    @drf.decorators.action(
+        authentication_classes=[authentication.ServerToServerAuthentication],
+        detail=False,
+        methods=["post"],
+        permission_classes=[],
+        url_path="create-for-owner",
+    )
+    def create_for_owner(self, request):
+        """
+        Create a document on behalf of a specified owner (pre-existing user or invited).
+        """
+        # Deserialize and validate the data
+        serializer = serializers.ServerCreateDocumentSerializer(data=request.data)
+        if not serializer.is_valid():
+            return drf_response.Response(
+                serializer.errors, status=status.HTTP_400_BAD_REQUEST
+            )
+
+        document = serializer.save()
+
+        return drf_response.Response(
+            {"id": str(document.id)}, status=status.HTTP_201_CREATED
+        )
+
     @drf.decorators.action(detail=True, methods=["get"], url_path="versions")
     def versions_list(self, request, *args, **kwargs):
         """
@@ -813,11 +838,11 @@ def perform_create(self, serializer):
         access = serializer.save()
         language = self.request.headers.get("Content-Language", "en-us")
 
-        access.document.email_invitation(
-            language,
+        access.document.send_invitation_email(
             access.user.email,
             access.role,
             self.request.user,
+            language,
         )
 
     def perform_update(self, serializer):
@@ -1078,8 +1103,8 @@ def perform_create(self, serializer):
 
         language = self.request.headers.get("Content-Language", "en-us")
 
-        invitation.document.email_invitation(
-            language, invitation.email, invitation.role, self.request.user
+        invitation.document.send_invitation_email(
+            invitation.email, invitation.role, self.request.user, language
         )
 
 

src/backend/core/authentication/__init__.py

@@ -0,0 +1,52 @@
+"""Custom authentication classes for the Impress core app"""
+
+from django.conf import settings
+
+from rest_framework.authentication import BaseAuthentication
+from rest_framework.exceptions import AuthenticationFailed
+
+
+class ServerToServerAuthentication(BaseAuthentication):
+    """
+    Custom authentication class for server-to-server requests.
+    Validates the presence and correctness of the Authorization header.
+    """
+
+    AUTH_HEADER = "Authorization"
+    TOKEN_TYPE = "Bearer"  # noqa S105
+
+    def authenticate(self, request):
+        """
+        Authenticate the server-to-server request by validating the Authorization header.
+
+        This method checks if the Authorization header is present in the request, ensures it
+        contains a valid token with the correct format, and verifies the token against the
+        list of allowed server-to-server tokens. If the header is missing, improperly formatted,
+        or contains an invalid token, an AuthenticationFailed exception is raised.
+
+        Returns:
+            None: If authentication is successful
+                  (no user is authenticated for server-to-server requests).
+
+        Raises:
+            AuthenticationFailed: If the Authorization header is missing, malformed,
+            or contains an invalid token.
+        """
+        auth_header = request.headers.get(self.AUTH_HEADER)
+        if not auth_header:
+            raise AuthenticationFailed("Authorization header is missing.")
+
+        # Validate token format and existence
+        auth_parts = auth_header.split(" ")
+        if len(auth_parts) != 2 or auth_parts[0] != self.TOKEN_TYPE:
+            raise AuthenticationFailed("Invalid authorization header.")
+
+        token = auth_parts[1]
+        if token not in settings.SERVER_TO_SERVER_API_TOKENS:
+            raise AuthenticationFailed("Invalid server-to-server token.")
+
+        # Authentication is successful, but no user is authenticated
+
+    def authenticate_header(self, request):
+        """Return the WWW-Authenticate header value."""
+        return f"{self.TOKEN_TYPE} realm='Create document server to server'"

src/backend/core/migrations/0012_make_document_creator_and_invitation_issuer_optional.py

@@ -0,0 +1,30 @@
+# Generated by Django 5.1.2 on 2024-11-30 22:23
+
+import django.db.models.deletion
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('core', '0011_populate_creator_field_and_make_it_required'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='document',
+            name='creator',
+            field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.RESTRICT, related_name='documents_created', to=settings.AUTH_USER_MODEL),
+        ),
+        migrations.AlterField(
+            model_name='invitation',
+            name='issuer',
+            field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, related_name='invitations', to=settings.AUTH_USER_MODEL),
+        ),
+        migrations.AlterField(
+            model_name='user',
+            name='language',
+            field=models.CharField(choices="(('en-us', 'English'), ('fr-fr', 'French'), ('de-de', 'German'))", default='en-us', help_text='The language in which the user wants to see the interface.', max_length=10, verbose_name='language'),
+        ),
+    ]