我正在使用 Fast API 作为后端编写一个简单的 Web 应用程序，并且我希望通过 Azure B2C 实现身份验证。这一切的逻辑现在并不重要，我只是想能够测试一下我是否可以使用不同的方法成功登录。
但是，在尝试了很长一段时间之后，我不明白，为什么我可以重定向后不会从会话中检索用户的数据。
import logging
import secrets
from fastapi import FastAPI, Request, HTTPException, status
from fastapi.responses import RedirectResponse, JSONResponse
from starlette.middleware.sessions import SessionMiddleware
from datetime import datetime
import jwt
import time
import json
from logger import logger
from env import *
from db_helpers import *
from mailer import *
from ad_helpers import *
from models import CreateUserRequest, InviteUserRequest, AcceptInvitationRequest, ProcessGoogleTokenRequest, ResendInvitationRequest, UserResponse, CreateUserResponse, InviteUserResponse
# Logging
app = FastAPI()
app.add_middleware(SessionMiddleware, secret_key="your-very-secret-key", max_age=None)
logger.setLevel(logging.DEBUG)
logger.debug(f"B2C Authority: {B2C_AUTHORITY}")
# Middleware to handle session cookies
@app.middleware("http")
async def session_middleware(request: Request, call_next):
response = await call_next(request)
session = request.cookies.get('session')
if session:
response.set_cookie(key='session', value=session, httponly=True)
return response
# FRONT ROUTES
@app.get('/login')
async def login(request: Request):
"""
Get the authorization URL for login.
"""
logger.debug("Executing login function")
auth_url = build_auth_url()
logger.debug(f"Authorization URL: {auth_url}")
return JSONResponse(content={"auth_url": auth_url})
@app.get('/oauth_callback')
async def oauth_callback(request: Request):
"""
Handle the OAuth2 authorization code and store user information in the session.
"""
if 'code' in request.query_params:
logger.debug("Authorization code received: %s", request.query_params['code'])
cache = load_cache(request)
try:
result = build_msal_app(cache=cache).acquire_token_by_authorization_code(
request.query_params['code'],
scopes=[],
redirect_uri=f"{BASE_URL}/oauth_callback") # Ensure proper URL
logger.debug(f"Token acquisition result: {result}")
if 'error' in result:
logger.error("Error in token acquisition: %s", result['error_description'])
raise HTTPException(status_code=400, detail=f"Error: {result['error_description']}")
user_info = result.get('id_token_claims')
request.session['user'] = user_info
save_cache(request, cache)
logger.debug("Session data saved: %s", request.session['user'])
except Exception as e:
logger.error("Exception during token acquisition: %s", str(e))
raise HTTPException(status_code=500, detail=str(e))
return RedirectResponse("/user-profile")
if not request.session.get('user'):
logger.debug("No user session found, redirecting to login")
return RedirectResponse("/login/")
logger.debug("User session found, redirecting to profile")
return RedirectResponse("/user-profile")
@app.get('/user-profile', response_model=UserResponse)
async def user_profile(request: Request):
"""
Get the user's profile information from the session.
"""
user = request.session.get('user')
logger.debug("user data from session: %s", user)
if not user:
raise HTTPException(status_code=401, detail="User not logged in")
return UserResponse(message="User profile retrieved successfully", email=user['emails'][0], id=user['oid'])
我将非常高兴您的帮助或任何有关我应该查看的内容的提示。
我尝试从 THEsession 中设置和检索用户的数据，但它显示 None每时每刻。
问题在于你试图从
request.cookies
中访问会话，而应该从
request.session
中访问。另外，你的
/oauth_callback
路由在重定向到
/user-profile
之前没有设置会话 cookie。
以下是如何修复代码：
import logging
import secrets
from fastapi import FastAPI, Request, HTTPException, status
from fastapi.responses import RedirectResponse, JSONResponse
from starlette.middleware.sessions import SessionMiddleware
# ... other imports
app = FastAPI()
app.add_middleware(
SessionMiddleware, secret_key="your-very-secret-key", max_age=None
)
logger.setLevel(logging.DEBUG)
# ... other routes
@app.get('/oauth_callback')
async def oauth_callback(request: Request):
"""
Handle the OAuth2 authorization code and store user information in the session.
"""
if 'code' in request.query_params:
logger.debug("Authorization code received: %s", request.query_params['code'])
cache = load_cache(request)
try:
result = build_msal_app(cache=cache).acquire_token_by_authorization_code(
request.query_params['code'],
scopes=[],
redirect_uri=f"{BASE_URL}/oauth_callback", # Ensure proper URL
)
logger.debug(f"Token acquisition result: {result}")
if 'error' in result:
logger.error("Error in token acquisition: %s", result['error_description'])
raise HTTPException(status_code=400, detail=f"Error: {result['error_description']}")
user_info = result.get('id_token_claims')
request.session['user'] = user_info
save_cache(request, cache)
logger.debug("Session data saved: %s", request.session['user'])
# Set the session cookie in the response
response = RedirectResponse("/user-profile")
response.set_cookie(key="session", value=request.session.get("session"))
return response
except Exception as e:
logger.error("Exception during token acquisition: %s", str(e))
raise HTTPException(status_code=500, detail=str(e))
if not request.session.get('user'):
logger.debug("No user session found, redirecting to login")
return RedirectResponse("/login/")
logger.debug("User session found, redirecting to profile")
return RedirectResponse("/user-profile")
@app.get('/user-profile', response_model=UserResponse)
async def user_profile(request: Request):
"""
Get the user's profile information from the session.
"""
user = request.session.get('user')
logger.debug("user data from session: %s", user)
if not user:
raise HTTPException(status_code=401, detail="User not logged in")
return UserResponse(message="User profile retrieved successfully", email=user['emails'][0], id=user['oid'])
解释：
-
从
request.session获取会话数据： 在/user-profile路由中，我们使用request.session.get('user')获取用户数据，而不是request.cookies.get('session')。 -
在重定向之前设置会话 cookie： 在
/oauth_callback路由中，我们在重定向到/user-profile之前 使用response.set_cookie()设置会话 cookie。这确保了在重定向后，浏览器会将会话 cookie 发送回服务器，从而允许我们访问会话数据。
通过这些更改，你现在应该能够在重定向到
/user-profile
后从会话中成功检索用户数据。