mirror of
https://github.com/jwinterm/modoboa.git
synced 2026-10-22 06:28:41 +02:00
236 lines
8.3 KiB
Python
236 lines
8.3 KiB
Python
"""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)
|