FastAPI 自動(dòng)續(xù)簽 Token 與刷新機(jī)制詳解:讓用戶登錄更持久!
在 Web 應(yīng)用中,JWT(JSON Web Token)常用于身份認(rèn)證。但 JWT 通常有固定的過期時(shí)間,一旦過期就會(huì)導(dǎo)致用戶強(qiáng)制重新登錄 —— 這顯然影響體驗(yàn)。
解決方案?
引入 Refresh Token(刷新令牌)機(jī)制,實(shí)現(xiàn) Access Token 的自動(dòng)續(xù)簽!
一句話理解 Refresh Token
Refresh Token 是一種用于刷新 Access Token 的憑證。 它不用于訪問受保護(hù)資源,只用于生成新的 Access Token。
整體流程圖
登錄成功后返回:
├── access_token(短期有效) → 用于訪問接口
└── refresh_token(長(zhǎng)期有效) → 用于刷新 access_token
?? 客戶端邏輯:
1. 每次請(qǐng)求攜帶 access_token
2. 如果接口返回 401(token 過期):
- 使用 refresh_token 請(qǐng)求刷新接口,換一個(gè)新的 access_token
- 自動(dòng)重試原請(qǐng)求
數(shù)據(jù)模型設(shè)計(jì)
新增 refresh_token 存儲(chǔ)字段(也可以存 Redis 中)。
models/user.py 示例: from tortoise import fields, models
from tortoise import fields, models
class User(models.Model):
id = fields.IntField(pk=True)
username = fields.CharField(max_length=50, unique=True)
hashed_password = fields.CharField(max_length=128)
refresh_token = fields.CharField(max_length=512, null=True) # 新增
JWT 工具函數(shù)封裝(包含 refresh 支持)
from jose import jwt
from datetime import datetime, timedelta
SECRET_KEY = "your-access-token-secret"
REFRESH_SECRET = "your-refresh-token-secret"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 15
REFRESH_TOKEN_EXPIRE_DAYS = 7
def create_access_token(data: dict, expires_delta: timedelta = None):
to_encode = data.copy()
expire = datetime.utcnow() + (expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES))
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
def create_refresh_token(data: dict):
expire = datetime.utcnow() + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
to_encode = data.copy()
to_encode.update({"exp": expire})
return jwt.encode(to_encode, REFRESH_SECRET, algorithm=ALGORITHM)
def verify_refresh_token(token: str):
try:
payload = jwt.decode(token, REFRESH_SECRET, algorithms=[ALGORITHM])
return payload
except Exception:
return None
登錄接口返回雙 token
@router.post("/login")
async def login(user: UserLoginSchema):
user_obj = await authenticate(user.username, user.password)
if not user_obj:
raise HTTPException(status_code=401, detail="賬號(hào)或密碼錯(cuò)誤")
data = {"sub": user_obj.username, "uid": user_obj.id}
access_token = create_access_token(data)
refresh_token = create_refresh_token(data)
# 保存 refresh_token 到數(shù)據(jù)庫
user_obj.refresh_token = refresh_token
await user_obj.save()
return {
"access_token": access_token,
"refresh_token": refresh_token,
"token_type": "bearer"
}
提供刷新接口 /refresh
@router.post("/refresh")
async def refresh_token(refresh_token: str = Body(...)):
payload = verify_refresh_token(refresh_token)
if not payload:
raise HTTPException(status_code=401, detail="Refresh token 失效")
# 確認(rèn)數(shù)據(jù)庫中 refresh_token 是否匹配(防止偽造)
user = await User.get(id=payload["uid"])
if user.refresh_token != refresh_token:
raise HTTPException(status_code=401, detail="Refresh token 不合法")
new_token = create_access_token({"sub": user.username, "uid": user.id})
return {"access_token": new_token, "token_type": "bearer"}
小結(jié)
結(jié)合兩個(gè) token,我們實(shí)現(xiàn)了:
- 安全性:短時(shí) Access Token 暴露后影響有限
- 易用性:Refresh Token 自動(dòng)續(xù)期,無需頻繁登錄