"""安全模块 — JWT Token 生成与验证(双 Token 架构)""" from datetime import timedelta from jose import JWTError, jwt from passlib.context import CryptContext from app.core.config import settings from app.core.time_utils import get_beijing_time pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") ALGORITHM = "HS256" # Token 类型声明 TOKEN_TYPE_ACCESS = "access" TOKEN_TYPE_REFRESH = "refresh" def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str: """生成 JWT Access Token(2 小时有效)""" to_encode = data.copy() expire = get_beijing_time() + ( expires_delta or timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES) ) to_encode.update({"exp": expire, "type": TOKEN_TYPE_ACCESS}) return jwt.encode(to_encode, settings.SECRET_KEY, algorithm=ALGORITHM) def create_refresh_token(data: dict, expires_delta: timedelta | None = None) -> str: """生成 JWT Refresh Token(7 天有效,仅用于刷新 Access Token)""" to_encode = data.copy() expire = get_beijing_time() + ( expires_delta or timedelta(days=settings.REFRESH_TOKEN_EXPIRE_DAYS) ) to_encode.update({"exp": expire, "type": TOKEN_TYPE_REFRESH}) return jwt.encode(to_encode, settings.SECRET_KEY, algorithm=ALGORITHM) def decode_token(token: str) -> dict: """解码并验证 JWT Token,返回 payload""" return jwt.decode(token, settings.SECRET_KEY, algorithms=[ALGORITHM]) def peek_token_identity(token: str) -> dict | None: """读出令牌里的用户身份 —— **仅供审计标注,绝不可用于授权**。 与 decode_token 的唯一区别:**关闭过期校验**。 为什么需要它:刷新令牌接口正是"access token 过期了才来"的场景, 请求里不带 Authorization 头,JWT 依赖根本不执行,审计只能记成 「未认证」—— 而"谁在什么时候尝试刷新"恰恰是该留痕的信息。 签名校验照常进行,伪造的令牌解不出任何东西。 ⚠️ 返回值只允许写进 request.state 的审计字段; 任何鉴权判断一律走 get_current_user,不要用本函数。 """ try: return jwt.decode( token, settings.SECRET_KEY, algorithms=[ALGORITHM], options={"verify_exp": False}, ) except JWTError: return None def verify_password(plain_password: str, hashed_password: str) -> bool: """验证明文密码 vs 哈希密码""" return pwd_context.verify(plain_password, hashed_password) def hash_password(password: str) -> str: """对明文密码进行哈希""" return pwd_context.hash(password)