✨(backend) complete admin maildomain api

- Create user (if needed) and mailbox access on mailbox creation
- If identify provider is keycloak, generate a one time password
at mailbox creation
- At nested endpoint to search users related to a maildomain

Co-authored-by: Sylvain Zimmer <[email protected]>

Author: jbpenrath

src/backend/core/admin.py

@@ -141,6 +141,11 @@ class UserAdmin(auth_admin.UserAdmin):
     )
     search_fields = ("id", "sub", "admin_email", "email", "full_name")
 
+class MailDomainAccessInline(admin.TabularInline):
+    """Inline class for the MailDomainAccess model"""
+
+    model = models.MailDomainAccess
+
 
 class MailDomainAccessInline(admin.TabularInline):
     """Inline class for the MailDomainAccess model"""

src/backend/core/api/openapi.json

@@ -4827,4 +4827,4 @@
             }
         }
     }
-}
+}

src/backend/core/api/serializers.py

@@ -532,6 +532,23 @@ class Meta:
         ]
         read_only_fields = fields
 
+class MailboxAdminCreateSerializer(MailboxAdminSerializer):
+    """
+    Serialize Mailbox details for create admin endpoint, including users with access and
+    metadata.
+    """
+
+    one_time_password = serializers.SerializerMethodField(read_only=True, required=False)
+
+    def get_one_time_password(self, instance) -> str | None:
+        """
+        Fake method just to make the OpenAPI schema valid.
+        """
+
+    class Meta:
+        model = models.Mailbox
+        fields = MailboxAdminSerializer.Meta.fields + ["one_time_password"]
+        read_only_fields = fields
 
 class ImportBaseSerializer(serializers.Serializer):
     """Base serializer for import actions that disables create and update."""

src/backend/core/api/viewsets/maildomain.py

@@ -1,17 +1,27 @@
 """Admin ViewSets for MailDomain and Mailbox management."""
 
+from django.conf import settings
+from django.db.models import Q
 from django.shortcuts import get_object_or_404
 from django.utils.translation import gettext_lazy as _  # For user-facing error messages
 
-from rest_framework import mixins, status, viewsets
+from drf_spectacular.utils import (
+    OpenApiParameter,
+    OpenApiResponse,
+    OpenApiTypes,
+    extend_schema,
+    inline_serializer,
+)
+from rest_framework import mixins, response, status, viewsets, serializers as drf_serializers
 from rest_framework.response import Response
 
 from core import models
 from core.api import permissions as core_permissions
 from core.api import serializers as core_serializers
+from core.identity.keycloak import reset_keycloak_user_password
 
 
-class MailDomainAdminViewSet(mixins.ListModelMixin, viewsets.GenericViewSet):
+class AdminMailDomainViewSet(mixins.ListModelMixin, viewsets.GenericViewSet):
     """
     ViewSet for listing MailDomains the user administers.
     Provides a top-level entry for mail domain administration.
@@ -26,16 +36,13 @@ def get_queryset(self):
         if not user or not user.is_authenticated:
             return models.MailDomain.objects.none()
 
-        accessible_maildomain_ids = models.MailDomainAccess.objects.filter(
-            user=user, role=models.MailDomainAccessRoleChoices.ADMIN
-        ).values_list("maildomain_id", flat=True)
-
         return models.MailDomain.objects.filter(
-            id__in=list(accessible_maildomain_ids)
+            accesses__user=user,
+            accesses__role=models.MailDomainAccessRoleChoices.ADMIN
         ).order_by("name")
 
 
-class MailboxAdminViewSet(
+class AdminMailDomainMailboxViewSet(
     mixins.CreateModelMixin,
     mixins.RetrieveModelMixin,
     mixins.UpdateModelMixin,
@@ -65,10 +72,36 @@ def get_queryset(self):
             "local_part"
         )
 
+    @extend_schema(
+        description="Create new mailbox in a specific maildomain.",
+        request=inline_serializer(
+            name="MailboxAdminCreatePayload",
+            fields={
+                "local_part": drf_serializers.CharField(required=True),
+                "alias_of": drf_serializers.UUIDField(required=False),
+                "metadata": inline_serializer(
+                    name="MailboxAdminCreateMetadata",
+                    fields={
+                        "type": drf_serializers.ChoiceField(choices=("personal", "shared", "redirect"), required=True),
+                        "first_name": drf_serializers.CharField(required=False, allow_blank=True),
+                        "last_name": drf_serializers.CharField(required=False, allow_blank=True),
+                    },
+                ),
+            },
+        ),
+        responses={
+            200: OpenApiResponse(
+                response=core_serializers.MailboxAdminCreateSerializer(),
+                description="The new mailbox with one extra field `one_time_password` if identity provider is keycloak.",
+            ),
+        },
+    )
     def create(self, request, *args, **kwargs):
         maildomain_pk = self.kwargs.get("maildomain_pk")
         domain = get_object_or_404(models.MailDomain, pk=maildomain_pk)
+        metadata = request.data.get("metadata", {})
 
+        mailbox_type = metadata.get("type")
         local_part = request.data.get("local_part")
         alias_of_id = request.data.get("alias_of")
 
@@ -121,8 +154,86 @@ def create(self, request, *args, **kwargs):
             domain=domain, local_part=local_part, alias_of=alias_of
         )
 
+        # --- Create user and mailbox access if type is personal ---
+        if mailbox_type == "personal":
+            email = f"{local_part}@{domain.name}"
+            first_name = metadata.get("first_name")
+            last_name = metadata.get("last_name")
+            user, _created = models.User.objects.get_or_create(
+                email=email,
+                defaults={
+                    "full_name": f"{first_name} {last_name}",
+                    "short_name": first_name,
+                    "password": "?",
+                }
+            )
+            models.MailboxAccess.objects.create(
+                mailbox=mailbox,
+                user=user,
+                role=models.MailboxRoleChoices.ADMIN,
+            )
+
         serializer = self.get_serializer(mailbox)
         headers = self.get_success_headers(serializer.data)
+        payload = serializer.data
+        if mailbox_type == "personal" and settings.IDENTITY_PROVIDER == "keycloak":
+            mailbox_password = reset_keycloak_user_password(email)
+            payload["one_time_password"] = mailbox_password
         return Response(
-            serializer.data, status=status.HTTP_201_CREATED, headers=headers
+            payload, status=status.HTTP_201_CREATED, headers=headers
         )
+
+class AdminMailDomainUserViewSet(mixins.ListModelMixin, viewsets.GenericViewSet):
+    """
+    ViewSet for listing users in a specific MailDomain.
+    Nested under /maildomains/{maildomain_pk}/users/
+    Permissions are checked by IsMailDomainAdmin for the maildomain_pk.
+    """
+
+    permission_classes = [
+        core_permissions.IsAuthenticated,
+        core_permissions.IsMailDomainAdmin,
+    ]
+    serializer_class = core_serializers.UserSerializer
+    pagination_class = None
+
+    def get_queryset(self):
+        """
+        Get all users having an access to a mailbox or an admin access to the maildomain.
+        """
+        maildomain_pk = self.kwargs.get("maildomain_pk")
+        # Get all users with an email ending with maildomain.name or with an admin access to the maildomain
+        return models.User.objects.filter(
+            Q(mailbox_accesses__mailbox__domain_id=maildomain_pk)
+            | Q(
+                maildomain_accesses__maildomain_id=maildomain_pk,
+                maildomain_accesses__role=models.MailDomainAccessRoleChoices.ADMIN
+            )).distinct().order_by("full_name", "short_name", "email")
+
+    @extend_schema(
+        tags=["admin-maildomain-user"],
+        parameters=[
+            OpenApiParameter(
+                name="q",
+                type=OpenApiTypes.STR,
+                location=OpenApiParameter.QUERY,
+                description="Search maildomains user by full name, short name or email.",
+            ),
+        ],
+    responses=core_serializers.UserSerializer(many=True),
+    )
+    def list(self, request, *args, **kwargs):
+        """
+        Search users by email, first name and last name.
+        """
+        queryset = self.get_queryset()
+
+        if query := request.query_params.get("q", ""):
+            queryset = queryset.filter(
+                Q(email__unaccent__icontains=query)
+                | Q(full_name__unaccent__icontains=query)
+                | Q(short_name__unaccent__icontains=query)
+            )
+
+        serializer = core_serializers.UserSerializer(queryset, many=True)
+        return response.Response(serializer.data)

src/backend/core/tests/api/test_admin_maildomains_list.py

@@ -0,0 +1,150 @@
+"""Tests for the MailDomain Admin API endpoints."""
+# pylint: disable=unused-argument
+
+from django.urls import reverse
+
+import pytest
+from rest_framework import status
+
+from core import factories
+from core.enums import MailboxRoleChoices, MailDomainAccessRoleChoices
+
+pytestmark = pytest.mark.django_db
+
+
+@pytest.fixture(name="domain_admin_user")
+def fixture_domain_admin_user():
+    """Create a user for domain administration testing."""
+    return factories.UserFactory()
+
+
+@pytest.fixture(name="other_user")
+def fixture_other_user():
+    """Create another user without admin privileges."""
+    return factories.UserFactory()
+
+
+@pytest.fixture(name="mail_domain1")
+def fixture_mail_domain1():
+    """Create the first mail domain for testing."""
+    return factories.MailDomainFactory(name="admin-domain1.com")
+
+
+@pytest.fixture(name="mail_domain2")
+def fixture_mail_domain2():
+    """Create the second mail domain for testing."""
+    return factories.MailDomainFactory(name="admin-domain2.com")
+
+
+@pytest.fixture(name="unmanaged_domain")
+def fixture_unmanaged_domain():
+    """Create a mail domain that has no admin access set up."""
+    return factories.MailDomainFactory(name="unmanaged-domain.com")
+
+
+@pytest.fixture(name="domain_admin_access1")
+def fixture_domain_admin_access1(domain_admin_user, mail_domain1):
+    """Create admin access for domain_admin_user to mail_domain1."""
+    return factories.MailDomainAccessFactory(
+        user=domain_admin_user,
+        maildomain=mail_domain1,
+        role=MailDomainAccessRoleChoices.ADMIN,
+    )
+
+
+@pytest.fixture(name="domain_admin_access2")
+def fixture_domain_admin_access2(domain_admin_user, mail_domain2):
+    """Create admin access for domain_admin_user to mail_domain2."""
+    return factories.MailDomainAccessFactory(
+        user=domain_admin_user,
+        maildomain=mail_domain2,
+        role=MailDomainAccessRoleChoices.ADMIN,
+    )
+
+
+@pytest.fixture(name="mailbox1_domain1")
+def fixture_mailbox1_domain1(mail_domain1):
+    """Create the first mailbox in mail_domain1."""
+    return factories.MailboxFactory(domain=mail_domain1, local_part="box1")
+
+
+@pytest.fixture(name="mailbox2_domain1")
+def fixture_mailbox2_domain1(mail_domain1):
+    """Create the second mailbox in mail_domain1."""
+    return factories.MailboxFactory(domain=mail_domain1, local_part="box2")
+
+
+@pytest.fixture(name="mailbox1_domain2")
+def fixture_mailbox1_domain2(mail_domain2):
+    """Create a mailbox in mail_domain2."""
+    return factories.MailboxFactory(domain=mail_domain2, local_part="boxA")
+
+
+@pytest.fixture(name="user_for_access1")
+def fixture_user_for_access1():
+    """Create a user for mailbox access testing."""
+    return factories.UserFactory(email="[email protected]")
+
+
+@pytest.fixture(name="user_for_access2")
+def fixture_user_for_access2():
+    """Create another user for mailbox access testing."""
+    return factories.UserFactory(email="[email protected]")
+
+
+@pytest.fixture(name="access_mailbox1_user1")
+def fixture_access_mailbox1_user1(mailbox1_domain1, user_for_access1):
+    """Create EDITOR access for user_for_access1 to mailbox1_domain1."""
+    return factories.MailboxAccessFactory(
+        mailbox=mailbox1_domain1, user=user_for_access1, role=MailboxRoleChoices.EDITOR
+    )
+
+
+@pytest.fixture(name="access_mailbox1_user2")
+def fixture_access_mailbox1_user2(mailbox1_domain1, user_for_access2):
+    """Create VIEWER access for user_for_access2 to mailbox1_domain1."""
+    return factories.MailboxAccessFactory(
+        mailbox=mailbox1_domain1, user=user_for_access2, role=MailboxRoleChoices.VIEWER
+    )
+
+
+class TestAdminMailDomainViewSet:
+    """Tests for the AdminMailDomainViewSet."""
+
+    LIST_DOMAINS_URL = reverse("admin-maildomains-list")
+
+    def test_admin_maildomains_list_administered_maildomains_success(
+        self,
+        api_client,
+        domain_admin_user,
+        domain_admin_access1,
+        domain_admin_access2,
+        mail_domain1,
+        mail_domain2,
+        unmanaged_domain,
+    ):
+        """Test that a domain admin can list domains they have admin access to."""
+        api_client.force_authenticate(user=domain_admin_user)
+        response = api_client.get(self.LIST_DOMAINS_URL)
+
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["count"] == 2
+        domain_ids = [item["id"] for item in response.data["results"]]
+        assert str(mail_domain1.id) in domain_ids
+        assert str(mail_domain2.id) in domain_ids
+        assert str(unmanaged_domain.id) not in domain_ids
+
+    def test_admin_maildomains_list_administered_maildomains_no_admin_access(
+        self, api_client, other_user, mail_domain1
+    ):
+        """Test that users without domain admin access get an empty list."""
+        # other_user has no MailDomainAccess records
+        api_client.force_authenticate(user=other_user)
+        response = api_client.get(self.LIST_DOMAINS_URL)
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["count"] == 0
+
+    def test_admin_maildomains_list_administered_maildomains_unauthenticated(self, api_client):
+        """Test that unauthenticated requests to list domains are rejected."""
+        response = api_client.get(self.LIST_DOMAINS_URL)
+        assert response.status_code == status.HTTP_401_UNAUTHORIZED