74 lines
2.7 KiB
Python
74 lines
2.7 KiB
Python
|
"""FastAPI 公共依赖注入。"""
|
|||
|
|
|
|||
|
|
from typing import Annotated
|
|||
|
|
|
|||
|
|
from fastapi import Depends
|
|||
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|||
|
|
from sqlalchemy.orm import Session
|
|||
|
|
|
|||
|
|
from app.core.db import get_db
|
|||
|
|
from app.core.exceptions import AuthError
|
|||
|
|
from app.core.security import decode_token
|
|||
|
|
from app.models import User
|
|||
|
|
from app.repositories import user_repository
|
|||
|
|
from app.services import token_blacklist
|
|||
|
|
|
|||
|
|
# auto_error=False：令牌缺失时由我们自己抛 AuthError，
|
|||
|
|
# 才能返回统一响应信封，而不是 FastAPI 默认的 {"detail": "Not authenticated"}
|
|||
|
|
_bearer_scheme = HTTPBearer(auto_error=False, description="Bearer 访问令牌")
|
|||
|
|
|
|||
|
|
BearerCredentials = Annotated[HTTPAuthorizationCredentials | None, Depends(_bearer_scheme)]
|
|||
|
|
DbSession = Annotated[Session, Depends(get_db)]
|
|||
|
|
|
|||
|
|
|
|||
|
|
def get_raw_token(credentials: BearerCredentials) -> str:
|
|||
|
|
"""取原始令牌字符串。登出时需要它来解析 jti。"""
|
|||
|
|
if credentials is None or not credentials.credentials:
|
|||
|
|
raise AuthError()
|
|||
|
|
return credentials.credentials
|
|||
|
|
|
|||
|
|
|
|||
|
|
def get_current_user(credentials: BearerCredentials, session: DbSession) -> User:
|
|||
|
|
"""从 Bearer 令牌解析当前登录用户。
|
|||
|
|
|
|||
|
|
**三重校验，缺一不可**：
|
|||
|
|
|
|||
|
|
1. 签名与有效期（`decode_token`）
|
|||
|
|
2. 是否已被登出拉黑（Redis `ys:token:blacklist:{jti}`）
|
|||
|
|
3. **令牌版本与库中是否一致** —— 禁用 / 改密 / 强制下线会把
|
|||
|
|
`users.token_version` 加一，使此前签发的所有令牌立即失效。
|
|||
|
|
这一步**不依赖 Redis**，因此 Redis 故障或数据被清空时，
|
|||
|
|
离职员工依然无法继续访问
|
|||
|
|
|
|||
|
|
Redis 不可用时第 2 步会抛异常而非放行 —— **有意的 fail-closed**：
|
|||
|
|
对能查看全员薪酬的系统，宁可拒绝请求也不能放行状态不明的令牌。
|
|||
|
|
|
|||
|
|
:raises AuthError: 任一项校验未通过（401）
|
|||
|
|
"""
|
|||
|
|
if credentials is None or not credentials.credentials:
|
|||
|
|
raise AuthError()
|
|||
|
|
|
|||
|
|
payload = decode_token(credentials.credentials)
|
|||
|
|
|
|||
|
|
jti = payload.get("jti")
|
|||
|
|
if jti and token_blacklist.is_revoked(jti):
|
|||
|
|
raise AuthError("登录状态已失效，请重新登录")
|
|||
|
|
|
|||
|
|
subject = payload.get("sub")
|
|||
|
|
user = user_repository.get_by_id(session, int(subject)) if _is_digits(subject) else None
|
|||
|
|
if user is None or user.status != 1:
|
|||
|
|
raise AuthError()
|
|||
|
|
|
|||
|
|
if payload.get("ver") != user.token_version:
|
|||
|
|
raise AuthError("登录状态已失效，请重新登录")
|
|||
|
|
|
|||
|
|
return user
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _is_digits(value: object) -> bool:
|
|||
|
|
"""判断令牌中的 sub 是否为合法的主键字符串。
|
|||
|
|
|
|||
|
|
令牌内容不可信，直接 int() 会因非法输入抛 ValueError 而不是 401。
|
|||
|
|
"""
|
|||
|
|
return isinstance(value, str) and value.isdigit()
|