"""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()