Как выполнить аутентификацию на основе сеанса

Для бэкенда проекта мы объединяем FastAPI и Django.

  • Почему FastAPI?

    1. Нам нужно использовать WebSockets, а FastAPI поддерживает это нативно
      2. .
    2. Нам нравится Pydantic валидация и автодокументация, которую она включает

  • Почему Django?

    1. Мы хотим использовать встроенные функции аутентификации и авторизации (разрешения)
      2. .
    2. Мы хотим иметь возможность использовать его ORM

  • Почему не Django-ninja?

    1. Не поддерживает WebSockets

  • Почему бы не использовать Django с Django Rest Framework для API & с Channels для WebSockets?

    1. Мы предпочитаем API-функции FastAPIs, а не DRF
      2. .
    2. Нам нравится, что FastAPI поддерживает WebSockets нативно

Мы хотим аутентифицировать и авторизировать конечные точки FastAPI, основываясь на возможностях аутентификации, которые есть в Django (встроенные сессии и разрешения пользователей). Я пытаюсь найти хороший способ защиты конечных точек FastAPI с помощью сессий из Django.

В приведенном ниже коде вы можете увидеть текущую настройку:

  1. main.py: Здесь настроен FastAPI: (1) маршрутизатор, содержащий конечные точки FastAPI, и (2) установленное приложение Django
    2. .
  2. router.py: Содержит конечные точки FastAPI, я хотел бы использовать встроенные в Django пользовательские сессии для аутентификации и авторизации конечных точек FastAPI.
## main.py (FastAPI)

import os

from django.conf import settings
from django.core.asgi import get_asgi_application
from django.apps import apps

from fastapi import FastAPI

from service_using_websockets.endpoints import router

# Setup FastAPI w/ and include the api router
api = FastAPI()
api.include_router(router, prefix="/api")

# Mount the Django backend application to the api
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "backend.settings")
apps.populate(settings.INSTALLED_APPS)
api.mount("/backend", get_asgi_application())


## router.py (FastAPI, using Django auth)

from fastapi import APIRouter, Response
from django.contrib.sessions.models import Session
from django.contrib.auth import authenticate
from django.utils import timezone
import datetime

router = APIRouter()

@router.post("/login")
def login(credentials: HTTPBasicCredentials = Depends(security)):
    user = authenticate(username=credentials.username, password=credentials.password)
    
    # User authenticated w/ Django, now create a session
    if user is not None:
        # TODO: this works, but might be bad security-wise as we
        # are working around Django's security middleware this way
        session = Session()
        session.session_key = Session.objects.generate_session_key()
        session.session_data = {}  # You can store additional session data here
        session.expire_date = timezone.now() + datetime.timedelta(days=1)  # Set session expiry
        session.save()

        # Set the session ID in the cookie
        response.set_cookie(key="sessionid", value=session.session_key, httponly=True)
        return {"message": "User logged in successfully"}
    else:
        return {"message": "Invalid username or password"}

@router.get("/protected-endpoint")
def example_protected_endpoint(request: Request):
    session_id = request.cookies.get("session_id")
    if session_id is None or int(session_id) not in sessions_in_db:
        raise HTTPException(
            status_code=401,
            detail="Login and get a valid session",
        )
    # Get the user from the session
    user = get_user_from_session(int(session_id))
    # ... do some endpoint logic for this user

Мой текущий подход может сработать, но я думаю, что с точки зрения безопасности это может быть не лучший подход для прямого создания сессий таким образом. Я считаю, что мы пропускаем целую кучу промежуточного программного обеспечения безопасности Django, делая это таким образом. Есть ли лучший способ повторно использовать пользовательские сессии и разрешения Django для защиты конечных точек FastAPI?

Как обсуждалось в комментарии, в этом ответе не используется аутентификация на основе сеанса, вместо этого используется аутентификация на основе маркера.

(A) Создайте экземпляр Django с помощью DRF + simplejwt

  1. Установите simplejwt с зависимостью crypto (для безопасной проверки в вашем экземпляре FastAPI):
pip install djangorestframework-simplejwt[crypto]
  1. Настройте Django на использование simplejwt
# settings.py
from datetime import timedelta

SECRET_KEY = "" # Add key here, whether you read it from .env or w/e

INSTALLED_APPS = [
    # [...]
    'rest_framework',
]

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

SIMPLE_JWT = {
    # Lifecycle for access token
    'ACCESS_TOKEN_LIFETIME': timedelta(minutes=60),

    # Lifecycle for refresh token - if this expires, user has to log in again
    'REFRESH_TOKEN_LIFETIME': timedelta(days=7),

    'ALGORITHM'   : 'HS512',
    'SIGNING_KEY' : SECRET_KEY
}
  1. Добавьте конечные точки аутентификации
# urls.py
from rest_framework_simplejwt.views import (
    TokenObtainPairView,
    TokenRefreshView,
)

# if simplejwt's default flow is good enough:
urlpatterns = [
    path('token/obtain/',  TokenObtainPairView.as_view(), name='token_obtain'),
    path('token/refresh/', TokenRefreshView.as_view(), name='token_refresh'),
]

Если вам нужен собственный поток входа в систему, вот базовый вариант для начала:

# auth_views.py
from django.contrib.auth                      import authenticate

from rest_framework.views                     import APIView
from rest_framework.permissions               import AllowAny
from rest_framework.response                  import Response


class LoginView(APIView):
    '''
    Authenticates the user, and if successful, returns the user object as well as tokens for access & refresh
    '''
    permission_classes = (permissions.AllowAny,)
    serializer_class   = LoginResponseSerializer
    http_method_names  = ['post']

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

        if user:
            refresh = RefreshToken.for_user(user)
            serializer = UserLoginSerializer(user)
            return Response({
                'refresh': str(refresh),
                'access': str(refresh.access_token),
                'user_details': serializer.data
            }, status=status.HTTP_200_OK)
        else:
            return Response({'error': 'Invalid credentials'}, status=status.HTTP_401_UNAUTHORIZED)

(B) Как использовать этот аутентификатор

С вашего фронтенда, CLI, в зависимости от того, что вам нужно, сделайте запрос, эквивалентный simplejwt docs, только убедитесь, что выбрали правильную конечную точку из вашего urls.py.

"Первый раз" вход: 

curl \
  -X POST \
  -H "Content-Type: application/json" \
  -d '{"username": "davidattenborough", "password": "boatymcboatface"}' \
  http://localhost:8000/api/token/obtain/

Получение нового маркера доступа на основе действительного маркера обновления:

curl \
  -X POST \
  -H "Content-Type: application/json" \
  -d '{"refresh":"eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyX3BrIjoxLCJ0b2tlbl90eXBlIjoicmVmcmVzaCIsImNvbGRfc3R1ZmYiOiLimIMiLCJleHAiOjIzNDU2NywianRpIjoiZGUxMmY0ZTY3MDY4NDI3ODg5ZjE1YWMyNzcwZGEwNTEifQ.aEoAYkSJjoWH1boshQAaTkf8G3yn0kapko6HFRt7Rh4"}' \
  http://localhost:8000/api/token/refresh/

Если вам нужно упростить это для потребителя фронтенда, вы можете "обернуть" это в конечную точку FastAPI, так что ваш FastAPI /login/, скажем, будет под капотом передавать токен(ы) или имя пользователя+пароль в Django, когда это необходимо.

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

Это означает, что фронтенду не нужно отслеживать два разных сервера и цикл жизни токена. Таким образом, фронтенду будет казаться, что действует сессионный аутентификатор, за исключением того, что он произвольно ограничен по времени - но до тех пор, пока клиенты делают запрос в течение срока действия маркера обновления, их "сессия" будет продолжать обновляться.

Но это также добавляет сложности вашей конечной точке (точкам) FastAPI, так что вам придется решить эту проблему.

(C) Проверка токенов внутри FastAPI, используя PyJWT

# jwt.py
import jwt

# Has to be the same key that you used as SIGNING_KEY for simplejwt in Django
SECRET_KEY = ""

# Has to include the same algorithm that you used for simplejwt config in Django
ALGORITHM  = ["HS256"]

def verify(access_token):
    try:
        decoded_token = jwt.decode(access_token, SECRET_KEY, algorithms=ALGORITHM)
        return decoded_token
    except jwt.ExpiredSignatureError:
        # Token has expired - call Django's refresh-endpoint to get a new token set and repeat verification?
    except jwt.InvalidTokenError:
        # Token was invalid for other reasons - maybe the token never was valid. Redirect to login? Up to you.

# fastapi.py or wherever
from .jwt import verify

# on incoming request:
token = request.headers.get('Authorization')
if verify(token):
    pass
else:
    raise Exception("Auth was not valid")
Вернуться на верх