/
viewsets.py
236 lines (193 loc) · 8.34 KB
/
viewsets.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
"""Admin API."""
from django import http
from django.contrib.contenttypes.models import ContentType
from django.utils.translation import ugettext as _
from django_filters import rest_framework as dj_filters
from drf_spectacular.utils import extend_schema, extend_schema_view
from rest_framework import filters, renderers, status, viewsets
from rest_framework.decorators import action
from rest_framework.exceptions import ParseError
from rest_framework.permissions import DjangoModelPermissions, IsAuthenticated
from rest_framework.response import Response
from modoboa.core import models as core_models
from modoboa.core import sms_backends
from modoboa.lib import renderers as lib_renderers
from modoboa.lib import viewsets as lib_viewsets
from modoboa.lib.throttle import GetThrottleViewsetMixin, PasswordResetRequestThrottle
from ... import lib, models
from . import serializers
@extend_schema_view(
retrieve=extend_schema(
description="Retrieve a particular domain",
summary="Retrieve a particular domain"
),
list=extend_schema(
description="Retrieve a list of domains",
summary="Retrieve a list of domains"
),
create=extend_schema(
description="Create a new domain",
summary="Create a new domain"
)
)
class DomainViewSet(GetThrottleViewsetMixin, lib_viewsets.RevisionModelMixin, viewsets.ModelViewSet):
"""Domain viewset."""
permission_classes = [IsAuthenticated, DjangoModelPermissions, ]
serializer_class = serializers.DomainSerializer
def get_queryset(self):
"""Filter queryset based on current user."""
return models.Domain.objects.get_for_admin(self.request.user)
def perform_destroy(self, instance):
"""Add custom args to delete call."""
instance.delete(self.request.user)
class DomainAliasFilterSet(dj_filters.FilterSet):
"""Custom FilterSet for DomainAlias."""
domain = dj_filters.CharFilter(field_name="target__name")
class Meta:
model = models.DomainAlias
fields = ["domain"]
class DomainAliasViewSet(GetThrottleViewsetMixin, lib_viewsets.RevisionModelMixin,
viewsets.ModelViewSet):
"""ViewSet for DomainAlias."""
filter_backends = (dj_filters.DjangoFilterBackend, )
filterset_class = DomainAliasFilterSet
permission_classes = [IsAuthenticated, DjangoModelPermissions, ]
renderer_classes = (renderers.JSONRenderer, lib_renderers.CSVRenderer)
serializer_class = serializers.DomainAliasSerializer
def get_queryset(self):
"""Filter queryset based on current user."""
return models.DomainAlias.objects.get_for_admin(self.request.user)
def get_renderer_context(self):
context = super().get_renderer_context()
context["headers"] = ["name", "target__name", "enabled"]
return context
class AccountViewSet(GetThrottleViewsetMixin, lib_viewsets.RevisionModelMixin, viewsets.ModelViewSet):
"""ViewSet for User/Mailbox."""
filter_backends = (filters.SearchFilter, )
permission_classes = [IsAuthenticated, DjangoModelPermissions, ]
search_fields = ("^first_name", "^last_name", "^email")
def get_throttles(self):
throttles = super().get_throttles()
if self.action == "reset_password":
throttles.append(PasswordResetRequestThrottle())
return throttles
def get_serializer_class(self):
"""Return a serializer."""
action_dict = {
"list": serializers.AccountSerializer,
"retrieve": serializers.AccountSerializer,
"password": serializers.AccountPasswordSerializer,
"reset_password": serializers.ResetPasswordSerializer,
}
return action_dict.get(
self.action, serializers.WritableAccountSerializer)
def get_queryset(self):
"""Filter queryset based on current user."""
user = self.request.user
ids = user.objectaccess_set \
.filter(content_type=ContentType.objects.get_for_model(user)) \
.values_list("object_id", flat=True)
queryset = core_models.User.objects.filter(pk__in=ids)
domain = self.request.query_params.get("domain")
if domain:
queryset = queryset.filter(mailbox__domain__name=domain)
return queryset
@action(methods=["put"], detail=True)
def password(self, request, pk=None):
"""Change account password."""
try:
user = core_models.User.objects.get(pk=pk)
except core_models.User.DoesNotExist:
raise http.Http404
serializer = self.get_serializer(user, data=request.data)
if serializer.is_valid():
serializer.save()
return Response()
return Response(
serializer.errors, status=status.HTTP_400_BAD_REQUEST)
@action(detail=False)
def exists(self, request):
"""Check if account exists.
Requires a valid email address as argument. Example:
GET /exists/?email=user@test.com
"""
email = request.GET.get("email")
if not email:
raise ParseError("email not provided")
if not core_models.User.objects.filter(email=email).exists():
data = {"exists": False}
else:
data = {"exists": True}
serializer = serializers.AccountExistsSerializer(data)
return Response(serializer.data)
@action(methods=["post"], detail=False)
def reset_password(self, request):
"""Reset account password and send a new one by SMS."""
sms_password_recovery = (
request.localconfig.parameters
.get_value("sms_password_recovery", app="core")
)
if not sms_password_recovery:
return Response(status=404)
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
user = core_models.User.objects.filter(
email=serializer.validated_data["email"]).first()
if not user or not user.phone_number:
return Response(status=404)
backend = sms_backends.get_active_backend(
request.localconfig.parameters)
if not backend:
return Response(status=404)
password = lib.make_password()
content = _("Here is your new Modoboa password: {}").format(
password)
if not backend.send(content, [str(user.phone_number)]):
body = {"status": "ko"}
else:
# SMS was sent, now we can set the new password.
body = {"status": "ok"}
user.set_password(password)
user.save(update_fields=["password"])
return Response(body)
class AliasViewSet(GetThrottleViewsetMixin, lib_viewsets.RevisionModelMixin, viewsets.ModelViewSet):
"""
create:
Create a new alias instance.
"""
permission_classes = [IsAuthenticated, DjangoModelPermissions, ]
serializer_class = serializers.AliasSerializer
def get_queryset(self):
"""Filter queryset based on current user."""
user = self.request.user
ids = (
user.objectaccess_set.filter(
content_type=ContentType.objects.get_for_model(models.Alias))
.values_list("object_id", flat=True)
)
queryset = models.Alias.objects.filter(pk__in=ids)
domain = self.request.query_params.get("domain")
if domain:
queryset = queryset.filter(domain__name=domain)
return queryset
class SenderAddressFilterSet(dj_filters.FilterSet):
"""Custom FilterSet for SenderAddress."""
class Meta:
model = models.SenderAddress
fields = ["mailbox"]
class SenderAddressViewSet(GetThrottleViewsetMixin, lib_viewsets.RevisionModelMixin,
viewsets.ModelViewSet):
"""View set for SenderAddress model."""
filter_backends = (dj_filters.DjangoFilterBackend, )
filterset_class = SenderAddressFilterSet
permission_classes = [IsAuthenticated, DjangoModelPermissions, ]
serializer_class = serializers.SenderAddressSerializer
def get_queryset(self):
"""Filter queryset based on current user."""
user = self.request.user
mb_ids = (
user.objectaccess_set.filter(
content_type=ContentType.objects.get_for_model(models.Mailbox))
.values_list("object_id", flat=True)
)
return models.SenderAddress.objects.filter(mailbox__pk__in=mb_ids)