Corp/backend/api/auth/endpoints/passwords.py

from __future__ import annotations

import logging

from django.conf import settings
from django.contrib.auth import authenticate, password_validation
from django.contrib.auth.tokens import default_token_generator
from django.core.exceptions import ValidationError as DjangoValidationError
from django.utils.encoding import force_bytes, force_str
from django.utils.http import urlsafe_base64_decode, urlsafe_base64_encode
from drf_spectacular.utils import (
    OpenApiResponse,
    extend_schema,
)
from rest_framework import status
from rest_framework.response import Response

from api.v1.employees.views._helpers import Employee
from api.v1.employees.views.mixins import LdapPasswordMixin
from common.emails import send_templated_mail

from ..base import (
    AnonymousAPIView,
    SessionBaseAPIView,
    _build_frontend_password_reset_link,
    _resolve_employee_by_login,
)
from ..serializers import (
    ChangePasswordRequestSerializer,
    ChangePasswordResponseSerializer,
    PasswordResetConfirmRequestSerializer,
    PasswordResetConfirmResponseSerializer,
    PasswordResetRequestResponseSerializer,
    PasswordResetRequestSerializer,
)

logger = logging.getLogger(__name__)


class ChangePasswordAPIView(LdapPasswordMixin, SessionBaseAPIView):
    @extend_schema(
        tags=["Auth"],
        summary="Сменить пароль текущего пользователя",
        request=ChangePasswordRequestSerializer,
        responses={
            200: ChangePasswordResponseSerializer,
            400: OpenApiResponse(
                description="Текущий пароль неверен или новый пароль невалиден."
            ),
            502: OpenApiResponse(description="Не удалось обновить пароль в LDAP."),
        },
    )
    def post(self, request):
        serializer = ChangePasswordRequestSerializer(
            data=request.data,
            context={"request": request},
        )
        serializer.is_valid(raise_exception=True)

        current_password = serializer.validated_data["current_password"]
        new_password = serializer.validated_data["new_password"]
        user = request.user

        verified_user = authenticate(
            request=request,
            email=user.email,
            password=current_password,
        )
        if verified_user is None or verified_user.pk != user.pk:
            return Response(
                {"current_password": ["Текущий пароль указан неверно."]},
                status=status.HTTP_400_BAD_REQUEST,
            )

        try:
            updated, error_response = self.update_ldap_password(user, new_password)
        except DjangoValidationError as exc:
            return Response(
                {"new_password": exc.messages},
                status=status.HTTP_400_BAD_REQUEST,
            )
        if not updated:
            return error_response

        return Response({"ok": True}, status=status.HTTP_200_OK)


class PasswordResetAPIView(AnonymousAPIView):
    throttle_scope = "auth_email"

    @extend_schema(
        tags=["Auth"],
        summary="Запросить письмо для сброса пароля",
        request=PasswordResetRequestSerializer,
        responses={200: PasswordResetRequestResponseSerializer},
    )
    def post(self, request):
        serializer = PasswordResetRequestSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)

        user = _resolve_employee_by_login(serializer.validated_data["login"])
        if user and user.email:
            uid = urlsafe_base64_encode(force_bytes(user.pk))
            token = default_token_generator.make_token(user)
            reset_link = _build_frontend_password_reset_link(uid=uid, token=token)

            send_templated_mail(
                subject="Восстановление пароля",
                to=[user.email],
                template_base="emails/password_reset_link",
                context={
                    "user": user,
                    "reset_link": reset_link,
                    "site_url": getattr(
                        settings, "SITE_URL", "https://corp.robotail.pro"
                    ),
                },
            )

        # Намеренно не раскрываем, существует ли пользователь.
        return Response({"ok": True}, status=status.HTTP_200_OK)


class PasswordResetConfirmAPIView(LdapPasswordMixin, AnonymousAPIView):
    @extend_schema(
        tags=["Auth"],
        summary="Подтвердить сброс пароля по uid и token",
        request=PasswordResetConfirmRequestSerializer,
        responses={
            200: PasswordResetConfirmResponseSerializer,
            400: OpenApiResponse(
                description="Ссылка недействительна или пароль невалиден."
            ),
            502: OpenApiResponse(description="Не удалось обновить пароль в LDAP."),
        },
    )
    def post(self, request):
        serializer = PasswordResetConfirmRequestSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)

        try:
            user_id = force_str(urlsafe_base64_decode(serializer.validated_data["uid"]))
        except (TypeError, ValueError, OverflowError):
            return Response(
                {"token": ["Ссылка для восстановления недействительна."]},
                status=status.HTTP_400_BAD_REQUEST,
            )

        user = Employee.objects.filter(pk=user_id).first()
        if user is None or not default_token_generator.check_token(
            user, serializer.validated_data["token"]
        ):
            return Response(
                {"token": ["Ссылка для восстановления недействительна или устарела."]},
                status=status.HTTP_400_BAD_REQUEST,
            )

        try:
            password_validation.validate_password(
                serializer.validated_data["new_password"],
                user=user,
            )
        except DjangoValidationError as exc:
            return Response(
                {"new_password": exc.messages},
                status=status.HTTP_400_BAD_REQUEST,
            )

        try:
            updated, error_response = self.update_ldap_password(
                user,
                serializer.validated_data["new_password"],
            )
        except DjangoValidationError as exc:
            return Response(
                {"new_password": exc.messages},
                status=status.HTTP_400_BAD_REQUEST,
            )
        if not updated:
            return error_response

        return Response({"ok": True}, status=status.HTTP_200_OK)