Вход и аутентификация с помощью REST-фреймворка Django

Мне бы хотелось, чтобы кто-то, кто имеет больший опыт работы с фреймворком Django REST, подсказал мне, соответствует ли созданная мной система логина и аутентификации стандартам фреймворка

Я сомневаюсь, стоит ли использовать функцию Django login() в некоторой части моего кода, в клиентских представлениях я хочу избежать прямого доступа к серверу, поэтому я делаю всю логику с участием базы данных в микросервисных представлениях, в моем микросервисном представлении "UserLoginAPI", должен ли я использовать функцию login(request, user) для входа в систему пользователя? Я видел в некоторых обсуждениях, что я должен заменить login() на update_last_login(None, user), и это то, что я сделал, но я не уверен, что это правильно

Я также хотел бы знать, правильно ли работает мое промежуточное ПО аутентификации, игнорирующее публичные представления и проверяющее токен и обновляющее пользователя в сессии

У меня не так много опыта, и я был бы признателен, если бы вы могли сказать мне, соответствует ли мой код правильным стандартам или есть что-то, что я могу улучшить, определенная проблема, которую я могу избежать.

Мой поток входа:

  1. HTML-форма через POST-запрос отправляет данные "имя пользователя" и "пароль" клиентскому представлению "Логин"
    2. .
  2. Клиентское представление "Login" получает эти данные и отправляет их в API микросервиса "UserLoginAPI"
    3. .
  3. Представление микросервиса "UserLoginAPI" получает эти данные и отправляет их в сериализатор "UserLoginSerializer" для проверки
    4. .
  4. Сериализатор "UserLoginSerializer" выполняет необходимые проверки, такие как: Проверяет, не заблокирован ли пользователь, аутентифицирует с помощью функции Django authenticate(), проверяет, не является ли пользователь "None" и активен ли он, создает токен с oauth2.0 для этого пользователя и возвращает аутентифицированного пользователя и токен в представление "UserLoginAPI"
    5. .
  5. Представление "UserLoginAPI" получает пользователя и токен, обновляет последний вход пользователя и отправляет токен клиентскому представлению "Login"
    6. .
  6. Клиентское представление "Login" получает токен, сохраняет его в сессии и перенаправляет пользователя на главную страницу
    7. .

Мой поток выхода из системы таков:

  1. HTML-форма через POST-запрос отправляет данные "username" клиентскому представлению "logout"
    2. .
  2. Клиентское представление "Выход" получает эти данные и отправляет их в API микросервиса "UserLogoutAPI"
    3. .
  3. Представление микросервиса "UserLogoutAPI" получает эти данные и отправляет их в сериализатор "UserLogoutSerializer" для проверки
    4. .
  4. Сериализатор "UserLogoutSerializer" проверяет пользователя, принимает активный токен, истекает срок действия этого токена и возвращает пользователя в представление "UserLogoutAPI"
    5. .
  5. Представление "UserLogoutAPI" получает пользователя, выходит из системы с помощью функции django logout() и посылает сигнал в django axes, чтобы он знал, что пользователь вышел из системы
    6. .
  6. Клиентское представление "Logout" перенаправляет пользователя на страницу входа
    7. .

В моем промежуточном ПО я проверяю, является ли представление публичным или нет, если оно публичное, то промежуточное ПО не выполняется, потому что это представления, для доступа к которым пользователю не нужно авторизоваться, а если это представление не публичное, то промежуточное ПО выполняется и проверяет, есть ли токен в сессии, оно берет токен из базы данных, соответствующий токену сессии, и если он не существует, то возвращает ошибку. Если токен существует, проверяется, не истекает ли срок его действия, и если истекает, то возвращается ошибка, а если нет, то обновляется пользователь запроса для пользователя, связанного с токеном.

"Login" и "Logout" - это мои представления на стороне клиента, они получают POST-запрос от HTML-формы и отправляют эти данные в API микросервисов

def login(request):

    if request.method == 'POST':
        username = request.POST.get('username')
        password = request.POST.get('password')

        api_url = f'{BASE_URL}/UserLoginAPI/'  
        data = {
            'username': username,
            'password': password,
        }

        try:

            response = api_request.make_post_request(api_url, data=data)

            if response['status'] == status.HTTP_200_OK:
                access_token = response.get('access_token')
                request.session['access_token'] = access_token
                return redirect('home')
            elif response['status'] in [status.HTTP_400_BAD_REQUEST, status.HTTP_500_INTERNAL_SERVER_ERROR]:
                messages.error(request, response['message'], extra_tags='error')   

        except requests.exceptions.RequestException:
            messages.error(request, 'Erro ao fazer login', extra_tags='error')

    return render(request, r'login\login.html')

def logout(request):

    if request.method == 'POST':
        username = request.user

        api_url = f'{BASE_URL}/UserLogoutAPI/'  
        data = {
            'username': username,
        }

        try:

            response = api_request.make_post_request(api_url, data=data)

            if response['status'] == status.HTTP_200_OK:
                return redirect('login')
            elif response['status'] in [status.HTTP_400_BAD_REQUEST, status.HTTP_500_INTERNAL_SERVER_ERROR]:
                messages.error(request, response['message'], extra_tags='error')

        except requests.exceptions.RequestException:
            messages.error(request, 'Erro ao deslogar', extra_tags='error')

"UserLoginAPI" и "UserLogoutAPI" - это мои представления в моем микросервисе аутентификации

@authentication_classes([OAuth2Authentication])
@permission_classes([TokenHasReadWriteScope])
class UserLoginAPI(APIView):
    def post(self, request):
        
        serializer = UserLoginSerializer(data=request.data, context={'request': request})

        try:

            serializer.is_valid(raise_exception=True)
            user = serializer.validated_data['user']
            access_token = serializer.validated_data['access_token']

            update_last_login(None, user)
            return Response({'success': 'Login bem sucedido', 'access_token': access_token}, status=status.HTTP_200_OK)

        except ValidationError as e:
            error_message = str(e.detail['non_field_errors'][0])
            log_error(request.path, error_message, None)
            return Response({'error': error_message}, status=status.HTTP_400_BAD_REQUEST)
        except Exception as e:
            log_error(request.path, str(e), None)
            return Response({'error': 'Erro ao fazer login'}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
        
@authentication_classes([OAuth2Authentication])
@permission_classes([TokenHasReadWriteScope])     
class UserLogoutAPI(APIView):
    def post(self, request):

        serializer = UserLogoutSerializer(data=request.data)

        try:

            serializer.is_valid(raise_exception=True)
            user = serializer.validated_data['user']

            logout(request)
            signals.user_logged_out.send(sender=User, request=request, user=user)
            return Response({'success': 'Usuário deslogado com sucesso'}, status=status.HTTP_200_OK)
        
        except ValidationError as e:
            error_message = str(e.detail['non_field_errors'][0])
            log_error(request.path, error_message, user)
            return Response({'error': str(e)}, status=status.HTTP_400_BAD_REQUEST)
        except Exception as e:
            log_error(request.path, str(e), user)
            return Response({'error': 'Erro ao deslogar'}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)

"UserLoginSerializer" и "UserLogoutSerializer" - это мои сериализаторы

class UserLoginSerializer(serializers.Serializer):
    username = serializers.CharField()
    password = serializers.CharField()

    def validate(self, data):
        username = data.get('username')
        password = data.get('password')
        request = self.context.get('request')
      
        if AxesProxyHandler().is_locked(request):
            raise ValidationError('O usuário está bloqueado por múltiplas tentativas falhas de login, aguarde 30 minutos antes de tentar entrar novamente')

        user = authenticate(request=request, username=username, password=password)

        if user is None or not user.is_active:
            raise ValidationError('Usuário ou senha inválidos')

        existing_token = AccessToken.objects.filter(
                user=user,
                expires__gt=timezone.now()
        ).first()

        if existing_token:
            access_token = existing_token
        else:
            access_token = AccessToken.objects.create(
                user=user,
                token=generate_token(),
                application=Application.objects.get(name='application_login'),
                expires=timezone.now() + timedelta(minutes=60),
            )
        
        return {'user': user, 'access_token': access_token.token}  

class UserLogoutSerializer(serializers.Serializer):
    username = serializers.CharField()

    def validate(self, data):
        username = data.get('username')

        user = User.objects.get(username=username)

        if not user or not user.is_active:
            raise ValidationError('Usuário inválido')
        
        existing_token = AccessToken.objects.filter(
                user = user,
                expires__gt = timezone.now()
            ).first()

        existing_token.expires = timezone.now()  
        existing_token.save()
        
        return {'user': user}

"TokenAuthenticationMiddleware" - это мое промежуточное ПО аутентификации

class TokenAuthenticationMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response
        self.public_views_paths = set(settings.PUBLIC_VIEWS)

    def __call__(self, request):
        request_path = request.path

        if request_path not in self.public_views_paths:
            session_token = request.session.get('access_token')

            if session_token:
                try:
                    access_token = AccessToken.objects.get(token=session_token)
        
                    if access_token.expires < timezone.now():
                        request.user = AnonymousUser()
                        messages.error(request, "Sua sessão expirou, faça o login novamente")
                        return redirect('login')
                    else:
                        request.user = access_token.user
                        log_path(request_path, request.user)
                except AccessToken.DoesNotExist:
                    request.user = AnonymousUser()
                    messages.error(request, "Acesso negado, faça login para continuar")
                    return redirect('login')
  
        response = self.get_response(request)
        return response

Мой settings.py с публичными урлами

PUBLIC_VIEWS = ['/', '/esqueci-a-senha/', '/redefinir-senha/']

Помогите просмотреть мой код и оценить, соответствует ли он стандартам разработки

Вы можете попробовать это,

Попробуйте view.py

from rest_framework.authtoken.models import Token

    class LoginAPIView(APIView):
        serializer_class = LoginSerializer
        authentication_classes = [TokenAuthentication]
    
        def post(self, request):
            serializer = LoginSerializer(data = request.data)
            if serializer.is_valid(raise_exception=True):
                user = authenticate(username=serializer.data['username'], password=serializer.data['password'])
                if user:
                    token, created = Token.objects.get_or_create(user=user)
                    return Response({'token': [token.key], "Sucsses":"Login SucssesFully"}, status=status.HTTP_201_CREATED )
                return Response({'Massage': 'Invalid Username and Password'}, status=401)

Попробуйте serializers.py

class LoginSerializer(serializers.ModelSerializer):
    class Meta:
        model = CustomUser
        fields = ("username", "password")

Попробуйте urls.py

from rest_framework.authtoken.views import obtain_auth_token

urlpatterns = [
    path('auth/', obtain_auth_token, name='auth'),
]

Попробуйте settings.py

REST_FRAMEWORK = {
    'DEFAULT_AUTHENTICATION_CLASSES': [
        'rest_framework.authentication.TokenAuthentication',
    ],
}

Используйте этот метод, и ваша работа станет легкой

Вернуться на верх