对于项目的后端，我们正在结合 FastAPI 和 Django。
为什么选择 FastAPI？
为什么选择 Django？
为什么不是姜戈忍者？
为什么不使用 Django 以及用于 API 的 Django Rest Framework 和用于 WebSockets 的通道？
我们希望根据 Django 具有的身份验证功能（内置用户会话和权限）对 FastAPI 端点进行身份验证和授权。我正在努力寻找一种使用 Django 会话来保护 FastAPI 端点的好方法。
在下面的代码中您可以看到当前的设置：
main.pyrouter.py## 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 端点？
正如评论中所讨论的，这个答案不使用基于会话的身份验证，而是使用基于令牌的身份验证。
DRFsimplejwtsimplejwtpip install djangorestframework-simplejwt[crypto]
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
}
# 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)
从您的前端、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/您可以对其他端点执行相同的操作 - 如果令牌未验证，您可以根据刷新令牌获取新的访问令牌，然后查看是否会验证。
这意味着前端不必跟踪两个不同的服务器，也不必跟踪令牌生命周期。因此，在前端看来，会话身份验证似乎已就位，只不过它是任意时间限制的 - 但只要客户端在刷新令牌的生命周期内发出请求，它们的“会话”就会不断更新。
但这也增加了 FastAPI 端点的复杂性，因此您必须解决这个问题。
# 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")