1
1
"""Admin ViewSets for MailDomain and Mailbox management."""
2
2
3
+ from django .conf import settings
4
+ from django .db .models import Q
3
5
from django .shortcuts import get_object_or_404
4
6
from django .utils .translation import gettext_lazy as _ # For user-facing error messages
5
7
6
- from rest_framework import mixins , status , viewsets
8
+ from drf_spectacular .utils import (
9
+ OpenApiParameter ,
10
+ OpenApiResponse ,
11
+ OpenApiTypes ,
12
+ extend_schema ,
13
+ inline_serializer ,
14
+ )
15
+ from rest_framework import (
16
+ mixins ,
17
+ response ,
18
+ status ,
19
+ viewsets ,
20
+ )
21
+ from rest_framework import (
22
+ serializers as drf_serializers ,
23
+ )
7
24
from rest_framework .response import Response
8
25
9
26
from core import models
10
27
from core .api import permissions as core_permissions
11
28
from core .api import serializers as core_serializers
29
+ from core .identity .keycloak import reset_keycloak_user_password
12
30
13
31
14
- class MailDomainAdminViewSet (mixins .ListModelMixin , viewsets .GenericViewSet ):
32
+ class AdminMailDomainViewSet (mixins .ListModelMixin , viewsets .GenericViewSet ):
15
33
"""
16
34
ViewSet for listing MailDomains the user administers.
17
35
Provides a top-level entry for mail domain administration.
@@ -29,16 +47,12 @@ def get_queryset(self):
29
47
if user .is_superuser and user .is_staff :
30
48
return models .MailDomain .objects .all ().order_by ("name" )
31
49
32
- accessible_maildomain_ids = models .MailDomainAccess .objects .filter (
33
- user = user , role = models .MailDomainAccessRoleChoices .ADMIN
34
- ).values_list ("maildomain_id" , flat = True )
35
-
36
50
return models .MailDomain .objects .filter (
37
- id__in = list ( accessible_maildomain_ids )
51
+ accesses__user = user , accesses__role = models . MailDomainAccessRoleChoices . ADMIN
38
52
).order_by ("name" )
39
53
40
54
41
- class MailboxAdminViewSet (
55
+ class AdminMailDomainMailboxViewSet (
42
56
mixins .CreateModelMixin ,
43
57
mixins .RetrieveModelMixin ,
44
58
mixins .UpdateModelMixin ,
@@ -68,10 +82,45 @@ def get_queryset(self):
68
82
"local_part"
69
83
)
70
84
85
+ @extend_schema (
86
+ description = "Create new mailbox in a specific maildomain." ,
87
+ request = inline_serializer (
88
+ name = "MailboxAdminCreatePayload" ,
89
+ fields = {
90
+ "local_part" : drf_serializers .CharField (required = True ),
91
+ "alias_of" : drf_serializers .UUIDField (required = False ),
92
+ "metadata" : inline_serializer (
93
+ name = "MailboxAdminCreateMetadata" ,
94
+ fields = {
95
+ "type" : drf_serializers .ChoiceField (
96
+ choices = ("personal" , "shared" , "redirect" ), required = True
97
+ ),
98
+ "first_name" : drf_serializers .CharField (
99
+ required = False , allow_blank = True
100
+ ),
101
+ "last_name" : drf_serializers .CharField (
102
+ required = False , allow_blank = True
103
+ ),
104
+ },
105
+ ),
106
+ },
107
+ ),
108
+ responses = {
109
+ 200 : OpenApiResponse (
110
+ response = core_serializers .MailboxAdminCreateSerializer (),
111
+ description = (
112
+ "The new mailbox with one extra field `one_time_password` "
113
+ "if identity provider is keycloak."
114
+ ),
115
+ ),
116
+ },
117
+ )
71
118
def create (self , request , * args , ** kwargs ):
72
119
maildomain_pk = self .kwargs .get ("maildomain_pk" )
73
120
domain = get_object_or_404 (models .MailDomain , pk = maildomain_pk )
121
+ metadata = request .data .get ("metadata" , {})
74
122
123
+ mailbox_type = metadata .get ("type" )
75
124
local_part = request .data .get ("local_part" )
76
125
alias_of_id = request .data .get ("alias_of" )
77
126
@@ -124,8 +173,90 @@ def create(self, request, *args, **kwargs):
124
173
domain = domain , local_part = local_part , alias_of = alias_of
125
174
)
126
175
176
+ # --- Create user and mailbox access if type is personal ---
177
+ if mailbox_type == "personal" :
178
+ email = f"{ local_part } @{ domain .name } "
179
+ first_name = metadata .get ("first_name" )
180
+ last_name = metadata .get ("last_name" )
181
+ user , _created = models .User .objects .get_or_create (
182
+ email = email ,
183
+ defaults = {
184
+ "full_name" : f"{ first_name } { last_name } " ,
185
+ "short_name" : first_name ,
186
+ "password" : "?" ,
187
+ },
188
+ )
189
+ models .MailboxAccess .objects .create (
190
+ mailbox = mailbox ,
191
+ user = user ,
192
+ role = models .MailboxRoleChoices .ADMIN ,
193
+ )
194
+
127
195
serializer = self .get_serializer (mailbox )
128
196
headers = self .get_success_headers (serializer .data )
129
- return Response (
130
- serializer .data , status = status .HTTP_201_CREATED , headers = headers
197
+ payload = serializer .data
198
+ if mailbox_type == "personal" and settings .IDENTITY_PROVIDER == "keycloak" :
199
+ mailbox_password = reset_keycloak_user_password (email )
200
+ payload ["one_time_password" ] = mailbox_password
201
+ return Response (payload , status = status .HTTP_201_CREATED , headers = headers )
202
+
203
+
204
+ class AdminMailDomainUserViewSet (mixins .ListModelMixin , viewsets .GenericViewSet ):
205
+ """
206
+ ViewSet for listing users in a specific MailDomain.
207
+ Nested under /maildomains/{maildomain_pk}/users/
208
+ Permissions are checked by IsMailDomainAdmin for the maildomain_pk.
209
+ """
210
+
211
+ permission_classes = [
212
+ core_permissions .IsAuthenticated ,
213
+ core_permissions .IsMailDomainAdmin ,
214
+ ]
215
+ serializer_class = core_serializers .UserSerializer
216
+ pagination_class = None
217
+
218
+ def get_queryset (self ):
219
+ """
220
+ Get all users having an access to a mailbox or an admin access to the maildomain.
221
+ """
222
+ maildomain_pk = self .kwargs .get ("maildomain_pk" )
223
+ # Get all users with an email ending with maildomain.name or with an admin access to the maildomain
224
+ return (
225
+ models .User .objects .filter (
226
+ Q (mailbox_accesses__mailbox__domain_id = maildomain_pk )
227
+ | Q (
228
+ maildomain_accesses__maildomain_id = maildomain_pk ,
229
+ maildomain_accesses__role = models .MailDomainAccessRoleChoices .ADMIN ,
230
+ )
231
+ )
232
+ .distinct ()
233
+ .order_by ("full_name" , "short_name" , "email" )
131
234
)
235
+
236
+ @extend_schema (
237
+ tags = ["admin-maildomain-user" ],
238
+ parameters = [
239
+ OpenApiParameter (
240
+ name = "q" ,
241
+ type = OpenApiTypes .STR ,
242
+ location = OpenApiParameter .QUERY ,
243
+ description = "Search maildomains user by full name, short name or email." ,
244
+ ),
245
+ ],
246
+ responses = core_serializers .UserSerializer (many = True ),
247
+ )
248
+ def list (self , request , * args , ** kwargs ):
249
+ """
250
+ Search users by email, first name and last name.
251
+ """
252
+ queryset = self .get_queryset ()
253
+
254
+ if query := request .query_params .get ("q" , "" ):
255
+ queryset = queryset .filter (
256
+ Q (email__unaccent__icontains = query )
257
+ | Q (full_name__unaccent__icontains = query )
258
+ | Q (short_name__unaccent__icontains = query )
259
+ )
260
+
261
+ serializer = core_serializers .UserSerializer (queryset , many = True )
262
+ return response .Response (serializer .data )
0 commit comments