FastAPI 本身不内置 session 机制（它推崇 OAuth2/JWT），但实际项目中 session 认证仍然是最简单、最常用的方案之一——尤其适合内部系统、管理后台、SSO 对接。下面从原理到完整实现讲透。
---
一、Session 认证 vs JWT 认证
| 维度 | Session（服务端存储） | JWT（客户端存储） |
|------|----------------------|-------------------|
| 状态 | 有状态，服务端保存会话 | 无状态，令牌自包含 |
| 存储 | 服务端内存/Redis/数据库 | 客户端 Cookie/localStorage |
| 注销 | 删服务端记录即生效 | 需黑名单或等令牌过期 |
| 扩展性 | 需共享 session 存储（Redis） | 天然支持多实例 |
| 安全性 | Cookie HttpOnly，XSS 防护好 | localStorage 易被 XSS 窃取 |
| 适用场景 | 内部系统、管理后台、传统 Web | API 服务、微服务、移动端 |
| 复杂度 | 低 | 中 |
---
二、最小可用实现（内存 Session）
2.1 安装依赖
pip install fastapi uvicorn itsdangerous python-multipart
| 包 | 用途 |
|---|------|
| fastapi | Web 框架 |
| uvicorn | ASGI 服务器 |
| itsdangerous | 签名 Cookie（防篡改） |
| python-multipart | 解析表单数据 |
2.2 完整代码
from fastapi import FastAPI, Request, Form, Depends, HTTPException
from fastapi.responses import HTMLResponse, RedirectResponse
from itsdangerous import URLSafeTimedSerializer
from datetime import timedelta
import secrets
app = FastAPI()
── 配置 ──
SECRETKEY = "your-secret-key-change-in-production" # 生产环境用 secrets.tokenhex(32)
SESSIONCOOKIENAME = "session_id"
SESSIONMAXAGE = 86400 # 24 小时（秒）
── Session 存储（内存版，生产环境换 Redis）──
sessionstore: dict = {} # { sessionid: {"user_id": 1, "username": "admin", "role": "admin"} }
── Cookie 签名器 ──
serializer = URLSafeTimedSerializer(SECRET_KEY, salt="session")
def create_session(data: dict) -> str:
"""创建 session，返回签名后的 session_id"""
sessionid = secrets.tokenhex(32)
sessionstore[sessionid] = data
return session_id
def getsessionid(request: Request) -> str | None:
"""从 Cookie 中提取并验证 session_id"""
signed = request.cookies.get(SESSIONCOOKIENAME)
if not signed:
return None
try:
# 验证签名 + 过期时间
sessionid = serializer.loads(signed, maxage=SESSIONMAXAGE)
return session_id
except Exception:
return None
def getcurrentuser(request: Request):
"""依赖项：获取当前登录用户，未登录则 401"""
sessionid = getsession_id(request)
if not sessionid or sessionid not in session_store:
raise HTTPException(status_code=401, detail="未登录，请先登录")
return sessionstore[sessionid]
def require_role(role: str):
"""依赖项：要求特定角色"""
def checker(user = Depends(getcurrentuser)):
if user.get("role") != role:
raise HTTPException(status_code=403, detail=f"需要 {role} 权限")
return user
return checker
── 模拟用户数据库 ──
USERS = {
"admin": {"id": 1, "username": "admin", "password": "admin123", "role": "admin"},
"user": {"id": 2, "username": "user", "password": "user123", "role": "user"},
}
── 登录页面 ──
@app.get("/login", response_class=HTMLResponse)
async def login_page():
return """
<h2>登录</h2>
<form method="post" action="/login">
<input name="username" placeholder="用户名"><br>
<input name="password" type="password" placeholder="密码"><br>
<button type="submit">登录</button>
</form>
"""
── 登录处理 ──
@app.post("/login")
async def login(request: Request, username: str = Form(...), password: str = Form(...)):
user = USERS.get(username)
if not user or user["password"] != password:
raise HTTPException(status_code=401, detail="用户名或密码错误")
# 创建 session
sessionid = createsession({
"user_id": user["id"],
"username": user["username"],
"role": user["role"],
})
# 签名后写入 Cookie
signed = serializer.dumps(session_id)
response = RedirectResponse(url="/dashboard", status_code=303)
response.set_cookie(
key=SESSIONCOOKIENAME,
value=signed,
maxage=SESSIONMAX_AGE,
httponly=True, # JS 不可读，防 XSS
secure=False, # 生产环境 True（仅 HTTPS）
samesite="lax", # 防 CSRF
)
return response
── 登出 ──
@app.post("/logout")
async def logout(request: Request):
sessionid = getsession_id(request)
if sessionid and sessionid in session_store:
del sessionstore[sessionid] # 服务端删除，立即失效
response = RedirectResponse(url="/login", status_code=303)
response.deletecookie(SESSIONCOOKIE_NAME)
return response
── 受保护页面 ──
@app.get("/dashboard", response_class=HTMLResponse)
async def dashboard(user = Depends(getcurrentuser)):
return f"""
<h2>欢迎，{user['username']}！</h2>
<p>角色：{user['role']}</p>
<p>用户ID：{user['user_id']}</p>
<form method="post" action="/logout">
<button type="submit">退出登录</button>
</form>
"""
── 仅管理员可访问 ──
@app.get("/admin")
async def adminpage(user = Depends(requirerole("admin"))):
return {"message": f"管理员 {user['username']} 访问了管理后台"}
── 公开接口 ──
@app.get("/")
async def index():
return {"message": "首页，无需登录"}
2.3 运行
uvicorn main:app --reload
打开 http://localhost:8000/login
账号 admin / admin123 或 user / user123
---
三、生产级实现（Redis Session）
内存版重启即丢失、多进程不共享。生产环境用 Redis：
pip install redis aioredis
import redis
import json
import secrets
from datetime import timedelta
── Redis 连接 ──
redisclient = redis.Redis(host="localhost", port=6379, db=0, decoderesponses=True)
SESSION_PREFIX = "session:"
SESSION_TTL = 86400 # 24 小时
class SessionManager:
"""Redis-backed Session 管理"""
def create(self, data: dict) -> str:
sessionid = secrets.tokenhex(32)
redis_client.setex(
f"{SESSIONPREFIX}{sessionid}",
SESSION_TTL,
json.dumps(data),
)
return session_id
def get(self, session_id: str) -> dict | None:
data = redisclient.get(f"{SESSIONPREFIX}{session_id}")
if data:
# 续期（滑动过期）
redisclient.expire(f"{SESSIONPREFIX}{sessionid}", SESSIONTTL)
return json.loads(data)
return None
def delete(self, session_id: str):
redisclient.delete(f"{SESSIONPREFIX}{session_id}")
def deletebyuser(self, user_id: int):
"""踢掉某用户所有 session（改密码后调用）"""
for key in redisclient.scaniter(f"{SESSION_PREFIX}*"):
data = json.loads(redis_client.get(key))
if data and data.get("userid") == userid:
redis_client.delete(key)
session_manager = SessionManager()
替换依赖项：
def getcurrentuser(request: Request):
sessionid = getsession_id(request)
if not session_id:
raise HTTPException(status_code=401, detail="未登录")
userdata = sessionmanager.get(session_id)
if not user_data:
raise HTTPException(status_code=401, detail="会话已过期，请重新登录")
return user_data
---
四、使用 Starlette SessionMiddleware（更简洁）
FastAPI 基于 Starlette，可以直接用其内置的 SessionMiddleware（基于 itsdangerous 签名 Cookie）：
from fastapi import FastAPI, Request, Form, Depends, HTTPException
from fastapi.responses import HTMLResponse, RedirectResponse
from starlette.middleware.sessions import SessionMiddleware
from datetime import timedelta
app = FastAPI()
添加 Session 中间件
app.add_middleware(
SessionMiddleware,
secret_key="your-secret-key-change-in-production",
sessioncookie="sessionid",
max_age=86400, # 24 小时
httponly=True,
secure=False, # 生产环境 True
samesite="lax",
)
def getcurrentuser(request: Request):
"""从 Starlette session 中获取用户"""
user = request.session.get("user")
if not user:
raise HTTPException(status_code=401, detail="未登录")
return user
def require_role(role: str):
def checker(user = Depends(getcurrentuser)):
if user.get("role") != role:
raise HTTPException(status_code=403, detail=f"需要 {role} 权限")
return user
return checker
USERS = {
"admin": {"id": 1, "username": "admin", "password": "admin123", "role": "admin"},
"user": {"id": 2, "username": "user", "password": "user123", "role": "user"},
}
@app.get("/login", response_class=HTMLResponse)
async def login_page():
return """
<h2>登录</h2>
<form method="post" action="/login">
<input name="username" placeholder="用户名"><br>
<input name="password" type="password" placeholder="密码"><br>
<button type="submit">登录</button>
</form>
"""
@app.post("/login")
async def login(request: Request, username: str = Form(...), password: str = Form(...)):
user = USERS.get(username)
if not user or user["password"] != password:
raise HTTPException(status_code=401, detail="用户名或密码错误")
# 直接写入 session（中间件自动签名+设Cookie）
request.session["user"] = {
"user_id": user["id"],
"username": user["username"],
"role": user["role"],
}
return RedirectResponse(url="/dashboard", status_code=303)
@app.post("/logout")
async def logout(request: Request):
request.session.clear() # 清除 session
return RedirectResponse(url="/login", status_code=303)
@app.get("/dashboard", response_class=HTMLResponse)
async def dashboard(user = Depends(getcurrentuser)):
return f"""
<h2>欢迎，{user['username']}！</h2>
<p>角色：{user['role']}</p>
<form method="post" action="/logout">
<button type="submit">退出登录</button>
</form>
"""
@app.get("/admin")
async def admin(user = Depends(require_role("admin"))):
return {"message": f"管理员 {user['username']} 访问了管理后台"}
SessionMiddleware 的原理：session 数据存在签名 Cookie 里（不在服务端存储），每次请求自动解签还原到 request.session。优点是零服务端存储、多实例天然兼容；缺点是 Cookie 大小限制 4KB，不能存大量数据。
---
五、三种方案对比
| 维度 | 手写内存 Session | Redis Session | SessionMiddleware |
|------|-----------------|---------------|-------------------|
| 服务端存储 | 内存 dict | Redis | 无（Cookie 自包含） |
| 多实例共享 | ❌ | ✅ | ✅（无需共享） |
| 重启保持 | ❌ | ✅ | ✅ |
| Cookie 大小限制 | 无 | 无 | 4KB |
| 注销即时生效 | ✅ | ✅ | ❌（需等 Cookie 过期） |
| 踢人/批量下线 | ✅ | ✅ | ❌ |
| 复杂度 | 低 | 中 | 最低 |
| 适用场景 | 开发/测试 | 生产环境 | 小型应用/无状态部署 |
---
六、安全加固清单
| 措施 | 说明 | 代码 |
|------|------|------|
| HttpOnly Cookie | JS 不可读，防 XSS 窃取 | httponly=True |
| Secure Cookie | 仅 HTTPS 传输 | secure=True（生产环境） |
| SameSite=Lax | 防 CSRF | samesite="lax" |
| 签名防篡改 | itsdangerous 签名 | URLSafeTimedSerializer |
| CSRF Token | 表单提交加随机 token | 见下方 |
| 密码哈希 | 不存明文密码 | passlib / bcrypt |
| 登录限流 | 防暴力破解 | slowapi |
| Session 续期 | 滑动过期 | 每次访问刷新 TTL |
| 改密码踢人 | 改密码后清除旧 session | sessionmanager.deleteby_user() |
密码哈希
from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def verify_password(plain: str, hashed: str) -> bool:
return pwd_context.verify(plain, hashed)
def hash_password(password: str) -> str:
return pwd_context.hash(password)
注册时：hash_password("admin123") → 存数据库
登录时：verifypassword(inputpassword, dbhashedpassword)
CSRF 防护
import secrets
@app.get("/form")
async def form_page(request: Request):
csrftoken = secrets.tokenhex(16)
request.session["csrftoken"] = csrftoken
return HTMLResponse(f"""
<form method="post" action="/submit">
<input type="hidden" name="csrftoken" value="{csrftoken}">
<input name="data">
<button type="submit">提交</button>
</form>
""")
@app.post("/submit")
async def submit(request: Request, data: str = Form(...), csrf_token: str = Form(...)):
if csrftoken != request.session.get("csrftoken"):
raise HTTPException(status_code=403, detail="CSRF token 验证失败")
return {"data": data}
---
七、完整路由保护模式
from fastapi import APIRouter
创建受保护的路由器
protected_router = APIRouter(
prefix="/api",
dependencies=[Depends(getcurrentuser)], # 所有路由都需要登录
)
admin_router = APIRouter(
prefix="/admin",
dependencies=[Depends(require_role("admin")], # 所有路由都需要 admin
)
@protected_router.get("/profile")
async def profile(user = Depends(getcurrentuser)):
return {"username": user["username"]}
@protected_router.get("/orders")
async def orders(user = Depends(getcurrentuser)):
return {"orders": [], "user": user["username"]}
@adminrouter.delete("/users/{userid}")
async def deleteuser(userid: int, admin = Depends(require_role("admin"))):
return {"message": f"管理员 {admin['username']} 删除了用户 {user_id}"}
注册路由器
app.includerouter(protectedrouter)
app.includerouter(adminrouter)
---
八、方案选择决策树
需要 Session 认证？
│
├── 小型应用 / 无状态部署 / 不需要踢人
│ └── SessionMiddleware（最简，Cookie 自包含）
│
├── 中大型应用 / 需要踢人 / 需要滑动过期 / 多实例
│ └── Redis Session（生产推荐）
│
└── 开发测试 / 单实例
└── 内存 Session（最快上手）
核心就一条链：登录验证 → 创建 session → 签名写入 HttpOnly Cookie → 后续请求自动带 Cookie → 服务端验证 session → Depends(getcurrentuser) 注入用户信息。生产环境用 Redis 存储 + HttpOnly + Secure + SameSite + CSRF Token + 密码哈希，安全闭环就齐了。
0 评论