"""API endpoints""" from django.db.models import Q, Subquery from rest_framework import exceptions, filters, mixins, viewsets from rest_framework.decorators import action from rest_framework.response import Response from core import models as core_models from core.api.client.serializers import UserSerializer from mailbox_manager import enums, models from mailbox_manager.api import permissions from mailbox_manager.api.client import serializers from mailbox_manager.utils.dimail import DimailAPIClient # pylint: disable=too-many-ancestors class MailDomainViewSet( mixins.CreateModelMixin, mixins.ListModelMixin, mixins.RetrieveModelMixin, viewsets.GenericViewSet, ): """ MailDomain viewset. GET /api//mail-domains/ Return a list of mail domains user has access to. GET /api//mail-domains// Return details for a mail domain user has access to. POST /api//mail-domains/ with expected data: - name: str - support_email: str Return newly created domain """ permission_classes = [permissions.AccessPermission] serializer_class = serializers.MailDomainSerializer filter_backends = [filters.OrderingFilter] ordering_fields = ["created_at", "name"] ordering = ["-created_at"] lookup_field = "slug" queryset = models.MailDomain.objects.all() def get_queryset(self): """Restrict results to the current user's team.""" return self.queryset.filter(accesses__user=self.request.user) def perform_create(self, serializer): """Set the current user as owner of the newly created mail domain.""" domain = serializer.save() serializers.MailDomainAccessSerializer().create( validated_data={ "user": self.request.user, "domain": domain, "role": str(core_models.RoleChoices.OWNER), } ) # pylint: disable=too-many-ancestors class MailDomainAccessViewSet( viewsets.GenericViewSet, mixins.ListModelMixin, mixins.CreateModelMixin, mixins.UpdateModelMixin, mixins.RetrieveModelMixin, mixins.DestroyModelMixin, ): """ API ViewSet for all interactions with mail domain accesses. GET /api/v1.0/mail-domains//accesses/: Return list of all domain accesses related to the logged-in user and one domain access if an id is provided. GET /api/v1.0/mail-domains//accesses/users/ Return list of all users who can have an access to the domain POST /api/v1.0/mail-domains//accesses/ with expected data: - user: str - role: str [owner|admin|viewer] Return newly created mail domain access PUT /api/v1.0/mail-domains//accesses// with expected data: - role: str [owner|admin|viewer] Return updated domain access PATCH /api/v1.0/mail-domains//accesses// with expected data: - role: str [owner|admin|viewer] Return partially updated domain access DELETE /api/v1.0/mail-domains//accesses// Delete targeted domain access """ permission_classes = [permissions.MailDomainAccessRolePermission] serializer_class = serializers.MailDomainAccessSerializer filter_backends = [filters.OrderingFilter] ordering_fields = ["role", "user__email", "user__name"] ordering = ["-created_at"] queryset = ( models.MailDomainAccess.objects.all() .select_related("user") .order_by("-created_at") ) list_serializer_class = serializers.MailDomainAccessReadOnlySerializer detail_serializer_class = serializers.MailDomainAccessSerializer def get_serializer_class(self): """Chooses list or detail serializer according to the action.""" if self.action in {"list", "retrieve"}: return self.list_serializer_class return self.detail_serializer_class def get_serializer_context(self): """Extra context provided to the serializer class.""" context = super().get_serializer_context() context["domain_slug"] = self.kwargs["domain_slug"] context["authenticated_user"] = self.request.user return context def get_queryset(self): """Return the queryset according to the action.""" queryset = super().get_queryset() queryset = queryset.filter(domain__slug=self.kwargs["domain_slug"]) if self.action in {"list", "retrieve"}: # Determine which role the logged-in user has in the domain user_role_query = models.MailDomainAccess.objects.filter( user=self.request.user, domain__slug=self.kwargs["domain_slug"] ).values("role")[:1] queryset = ( # The logged-in user should be part of a domain to see its accesses queryset.filter( domain__accesses__user=self.request.user, ) # Abilities are computed based on logged-in user's role and # the user role on each domain access .annotate( user_role=Subquery(user_role_query), ) .select_related("user") .distinct() ) return queryset def perform_update(self, serializer): """Check that we don't change the role if it leads to losing the last owner.""" instance = serializer.instance # Check if the role is being updated and the new role is not "owner" if ( "role" in self.request.data and self.request.data["role"] != enums.MailDomainRoleChoices.OWNER ): domain = instance.domain # Check if the access being updated is the last owner access for the domain if ( instance.role == enums.MailDomainRoleChoices.OWNER and domain.accesses.filter( role=enums.MailDomainRoleChoices.OWNER ).count() == 1 ): message = "Cannot change the role to a non-owner role for the last owner access." raise exceptions.PermissionDenied({"role": message}) serializer.save() def destroy(self, request, *args, **kwargs): """Forbid deleting the last owner access""" instance = self.get_object() domain = instance.domain # Check if the access being deleted is the last owner access for the domain if ( instance.role == enums.MailDomainRoleChoices.OWNER and domain.accesses.filter(role=enums.MailDomainRoleChoices.OWNER).count() == 1 ): message = "Cannot delete the last owner access for the domain." raise exceptions.PermissionDenied({"detail": message}) return super().destroy(request, *args, **kwargs) @action(detail=False, url_path="users", methods=["get"]) def get_available_users(self, request, domain_slug): """API endpoint to search user to give them new access. More filters and permission will be added soon. """ domain = models.MailDomain.objects.get(slug=domain_slug) abilities = domain.get_abilities(request.user) if not abilities["manage_accesses"]: raise exceptions.PermissionDenied() queryset = ( core_models.User.objects.order_by("-created_at") # exclude inactive users and get users from identified user's organization .filter(is_active=True, organization_id=request.user.organization_id) # exclude all users with already an access config .exclude(mail_domain_accesses__domain__slug=domain_slug) ) # Search by case-insensitive and accent-insensitive if query := request.GET.get("q", ""): queryset = queryset.filter( Q(name__unaccent__icontains=query) | Q(email__unaccent__icontains=query) ) return Response(UserSerializer(queryset.all(), many=True).data) class MailBoxViewSet( mixins.CreateModelMixin, mixins.ListModelMixin, viewsets.GenericViewSet, ): """MailBox ViewSet GET /api//mail-domains//mailboxes/ Return a list of mailboxes on the domain POST /api//mail-domains//mailboxes/ with expected data: - first_name: str - last_name: str - local_part: str - secondary_email: str Sends request to email provisioning API and returns newly created mailbox POST /api//mail-domains//mailboxes//disable/ Send a request to dimail to disable mailbox and change status of the mailbox in our DB POST /api//mail-domains//mailboxes//enable/ Send a request to dimail to enable mailbox and change status of the mailbox in our DB """ permission_classes = [permissions.MailBoxPermission] serializer_class = serializers.MailboxSerializer filter_backends = [filters.OrderingFilter] ordering = ["-created_at"] queryset = models.Mailbox.objects.all() def get_queryset(self): """Custom queryset to get mailboxes related to a mail domain.""" domain_slug = self.kwargs.get("domain_slug", "") if domain_slug: return self.queryset.filter(domain__slug=domain_slug) return self.queryset def perform_create(self, serializer): """Create new mailbox.""" domain_slug = self.kwargs.get("domain_slug", "") if domain_slug: serializer.validated_data["domain"] = models.MailDomain.objects.get( slug=domain_slug ) super().perform_create(serializer) @action(detail=True, methods=["post"]) def disable(self, request, domain_slug, pk=None): # pylint: disable=unused-argument """Disable mailbox. Send a request to dimail and change status in our DB""" mailbox = self.get_object() client = DimailAPIClient() client.disable_mailbox(mailbox, request.user.sub) mailbox.status = enums.MailboxStatusChoices.DISABLED mailbox.save() return Response(serializers.MailboxSerializer(mailbox).data) @action(detail=True, methods=["post"]) def enable(self, request, domain_slug, pk=None): # pylint: disable=unused-argument """Enable mailbox. Send a request to dimail and change status in our DB""" mailbox = self.get_object() client = DimailAPIClient() client.enable_mailbox(mailbox, request.user.sub) mailbox.status = enums.MailboxStatusChoices.ENABLED mailbox.save() return Response(serializers.MailboxSerializer(mailbox).data)