💻 Примеры — Урок 44

← К оглавлению урока

⚡ Ключевые файлы проекта

Полный JWT-auth проект состоит из 4 файлов: settings.py, middleware.py, serializers.py, views.py, urls.py.

Основной паттерн: RefreshToken.for_user(user) → response.set_cookie() при логине/регистрации, и response.delete_cookie() при логауте.

Пример 1: Полный settings.py

# myproject/settings.py

from datetime import timedelta

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'rest_framework',
    'rest_framework_simplejwt',
    'rest_framework_simplejwt.token_blacklist',  # обязательно для blacklist
    'myapp',
]

# Middleware — JWTAuthenticationMiddleware ПОСЛЕ SecurityMiddleware
MIDDLEWARE = [
    'django.middleware.security.SecurityMiddleware',
    'django.contrib.sessions.middleware.SessionMiddleware',
    'django.middleware.common.CommonMiddleware',
    'django.middleware.csrf.CsrfViewMiddleware',
    'django.contrib.auth.middleware.AuthenticationMiddleware',
    'django.contrib.messages.middleware.MessageMiddleware',
    'django.middleware.clickjacking.XFrameOptionsMiddleware',
    'myapp.middleware.JWTAuthenticationMiddleware',  # наш middleware
]

REST_FRAMEWORK = {
    'DEFAULT_AUTHENTICATION_CLASSES': [
        'rest_framework_simplejwt.authentication.JWTAuthentication',
    ],
    'DEFAULT_PERMISSION_CLASSES': [
        'rest_framework.permissions.IsAuthenticated',
    ],
}

SIMPLE_JWT = {
    'ACCESS_TOKEN_LIFETIME': timedelta(minutes=5),
    'REFRESH_TOKEN_LIFETIME': timedelta(days=1),
    'ROTATE_REFRESH_TOKENS': True,
    'BLACKLIST_AFTER_ROTATION': True,
    'AUTH_HEADER_TYPES': ('Bearer',),
}

Пример 2: Полный middleware.py

# myapp/middleware.py

from datetime import datetime, timezone
from django.utils.deprecation import MiddlewareMixin
from rest_framework_simplejwt.tokens import RefreshToken, AccessToken
from rest_framework_simplejwt.exceptions import TokenError


class JWTAuthenticationMiddleware(MiddlewareMixin):
    """
    Middleware для автоматического использования JWT-токенов из куки.

    Алгоритм:
    1. Читаем access_token из куки
    2. Если валиден — ставим HTTP_AUTHORIZATION
    3. Если истёк — обновляем через refresh_token и ставим HTTP_AUTHORIZATION
    4. В process_response — обновляем куку если был выпущен новый access
    """

    def process_request(self, request):
        access_token = request.COOKIES.get('access_token')
        refresh_token = request.COOKIES.get('refresh_token')

        if access_token:
            try:
                # Проверяем валидность access-токена
                token = AccessToken(access_token)
                if datetime.fromtimestamp(token['exp'], tz=timezone.utc) < datetime.now(timezone.utc):
                    raise TokenError('Token expired')
                # Токен валиден — ставим в заголовок
                request.META['HTTP_AUTHORIZATION'] = f'Bearer {access_token}'
            except TokenError:
                # Access истёк — пробуем обновить
                new_access_token = self.refresh_access_token(refresh_token)
                if new_access_token:
                    request.META['HTTP_AUTHORIZATION'] = f'Bearer {new_access_token}'
                    request._new_access_token = new_access_token
                else:
                    # Refresh тоже недействителен
                    self.clear_cookies(request)
        elif refresh_token:
            # Access отсутствует — получаем новый
            new_access_token = self.refresh_access_token(refresh_token)
            if new_access_token:
                request.META['HTTP_AUTHORIZATION'] = f'Bearer {new_access_token}'
                request._new_access_token = new_access_token
            else:
                self.clear_cookies(request)

    def refresh_access_token(self, refresh_token):
        """Обновить access-токен через refresh-токен."""
        if not refresh_token:
            return None
        try:
            refresh = RefreshToken(refresh_token)
            return str(refresh.access_token)
        except TokenError:
            return None

    def process_response(self, request, response):
        """Обновить куку access_token если был выпущен новый."""
        new_access_token = getattr(request, '_new_access_token', None)
        if new_access_token:
            access_expiry = AccessToken(new_access_token)['exp']
            response.set_cookie(
                key='access_token',
                value=new_access_token,
                httponly=True,
                secure=False,   # True в продакшне
                samesite='Lax',
                expires=datetime.fromtimestamp(access_expiry, tz=timezone.utc)
            )
        return response

    def clear_cookies(self, request):
        """Очистить истекшие токены из объекта запроса."""
        request.COOKIES.pop('access_token', None)
        request.COOKIES.pop('refresh_token', None)

Пример 3: serializers.py

# myapp/serializers.py

from rest_framework import serializers
from django.contrib.auth.models import User
from django.contrib.auth.password_validation import validate_password
from rest_framework.validators import UniqueValidator


class RegisterSerializer(serializers.ModelSerializer):
    """
    Сериализатор для регистрации нового пользователя.
    Поле password - только для записи, в ответе не возвращается.
    """
    password = serializers.CharField(write_only=True)

    class Meta:
        model = User
        fields = ['username', 'password', 'email']

    def create(self, validated_data):
        # create_user() хэширует пароль через set_password()
        user = User.objects.create_user(
            username=validated_data['username'],
            password=validated_data['password'],
            email=validated_data.get('email', '')
        )
        return user
⚠️ Проверить по документации: для расширенной валидации (уникальность email, сложность пароля) в продакшне следует добавить UniqueValidator для поля email и validate_password из django.contrib.auth.password_validation. Базовый вариант из лекции этого не включает.

Пример 4: views.py — полный файл

# myapp/views.py

from datetime import datetime, timezone
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import status
from django.contrib.auth import authenticate
from rest_framework_simplejwt.tokens import RefreshToken
from rest_framework.permissions import AllowAny, IsAuthenticated
from .serializers import RegisterSerializer


def set_jwt_cookies(response, user):
    """Вспомогательная функция: устанавливает JWT-токены в куки ответа."""
    refresh_token = RefreshToken.for_user(user)
    access_token = refresh_token.access_token

    access_expiry = datetime.fromtimestamp(access_token['exp'], tz=timezone.utc)
    refresh_expiry = datetime.fromtimestamp(refresh_token['exp'], tz=timezone.utc)

    response.set_cookie(
        key='access_token',
        value=str(access_token),
        httponly=True,
        secure=False,   # True в продакшне
        samesite='Lax',
        expires=access_expiry
    )
    response.set_cookie(
        key='refresh_token',
        value=str(refresh_token),
        httponly=True,
        secure=False,
        samesite='Lax',
        expires=refresh_expiry
    )


class LoginView(APIView):
    """Логин: аутентификация и выдача JWT-токенов в куках."""
    permission_classes = [AllowAny]

    def post(self, request, *args, **kwargs):
        username = request.data.get('username')
        password = request.data.get('password')
        user = authenticate(request, username=username, password=password)

        if user:
            refresh = RefreshToken.for_user(user)
            access_token = refresh.access_token

            access_expiry = datetime.fromtimestamp(access_token['exp'], tz=timezone.utc)
            refresh_expiry = datetime.fromtimestamp(refresh['exp'], tz=timezone.utc)

            response = Response(status=status.HTTP_200_OK)
            response.set_cookie(
                key='access_token',
                value=str(access_token),
                httponly=True,
                secure=False,
                samesite='Lax',
                expires=access_expiry
            )
            response.set_cookie(
                key='refresh_token',
                value=str(refresh),
                httponly=True,
                secure=False,
                samesite='Lax',
                expires=refresh_expiry
            )
            return response
        else:
            return Response(
                {"detail": "Invalid credentials"},
                status=status.HTTP_401_UNAUTHORIZED
            )


class LogoutView(APIView):
    """Логаут: удаление токенов из куки."""

    def post(self, request, *args, **kwargs):
        response = Response(status=status.HTTP_204_NO_CONTENT)
        response.delete_cookie('access_token')
        response.delete_cookie('refresh_token')
        return response


class RegisterView(APIView):
    """Регистрация нового пользователя с немедленной выдачей JWT-токенов."""
    permission_classes = [AllowAny]

    def post(self, request):
        serializer = RegisterSerializer(data=request.data)
        if serializer.is_valid():
            user = serializer.save()
            response = Response({
                'user': {
                    'username': user.username,
                    'email': user.email
                }
            }, status=status.HTTP_201_CREATED)
            set_jwt_cookies(response, user)
            return response
        else:
            return Response(
                serializer.errors,
                status=status.HTTP_400_BAD_REQUEST
            )


class ProtectedView(APIView):
    """Пример защищённого эндпоинта (IsAuthenticated по умолчанию)."""
    permission_classes = [IsAuthenticated]

    def get(self, request):
        return Response({
            'message': f'Hello, {request.user.username}! This is a protected resource.'
        })

Пример 5: urls.py

# myapp/urls.py

from django.urls import path
from .views import LoginView, LogoutView, RegisterView, ProtectedView

urlpatterns = [
    path('api/login/', LoginView.as_view(), name='login'),
    path('api/logout/', LogoutView.as_view(), name='logout'),
    path('api/register/', RegisterView.as_view(), name='register'),
    path('protected/', ProtectedView.as_view(), name='protected'),
]
# myproject/urls.py

from django.contrib import admin
from django.urls import path, include

urlpatterns = [
    path('admin/', admin.site.urls),
    path('', include('myapp.urls')),
]

Пример 6: Тестирование через Postman

Шаг 1: Создание пользователя в Django shell

python manage.py shell

from django.contrib.auth.models import User
User.objects.create_superuser('admin', 'admin@example.com', 'password123')

Шаг 2: Логин

  • Метод: POST
  • URL: http://127.0.0.1:8000/api/login/
  • Body → form-data: username=admin, password=password123
  • Ожидаемый ответ: статус 200, в заголовках Set-Cookie — access_token и refresh_token

Шаг 3: Доступ к защищённому ресурсу

  • Метод: GET
  • URL: http://127.0.0.1:8000/protected/
  • Postman автоматически отправит куки (если Cookies включены в настройках)
  • Ожидаемый ответ: 200, JSON с приветствием пользователя

Шаг 4: Регистрация

  • Метод: POST
  • URL: http://127.0.0.1:8000/api/register/
  • Body → form-data: username=newuser, password=secret123, email=new@example.com
  • Ожидаемый ответ: 201 Created, JSON с username/email, Set-Cookie с токенами

Шаг 5: Логаут

  • Метод: POST
  • URL: http://127.0.0.1:8000/api/logout/
  • Ожидаемый ответ: 204 No Content, куки удалены