273ebf4657
后端 FastAPI 六层结构（api/services/repositories/models/schemas/core）， 统一响应信封、全局异常、日志脱敏；6 张表经 Alembic 迁移建库，SQL 留存 db/migration；登录鉴权含图形验证码（Redis GETDEL 防重放）、JWT、登录限流、 登出黑名单（TTL=令牌剩余有效期）、登录日志。 前端 Vue3 + TS + Element Plus 骨架，含登录页、路由守卫、请求封装与 Vite 代理。 工程配置：uv 锁定 Python 3.12、ruff 检查与格式化、editorconfig、gitattributes。 测试 184 项（单元 155 / 集成 29）全绿，浏览器端到端 8 项通过。 Co-Authored-By: Claude Code <noreply@anthropic.com>
103 lines
4.0 KiB
Python
103 lines
4.0 KiB
Python
"""密码哈希与 JWT 签发校验。
|
||
|
||
**为什么不用 passlib**：passlib 1.7.4 自 2020 年起停止维护，且会读取
|
||
bcrypt 4.1 起已删除的 `bcrypt.__about__` 属性（表现为 "(trapped) error
|
||
reading bcrypt version"），与 bcrypt 5.x 组合没有任何官方支持。
|
||
FastAPI 官方文档亦已改用 pwdlib。
|
||
|
||
**BCrypt 的 72 字节硬限制**：bcrypt 5.0 起对超过 72 字节的密码**直接抛
|
||
ValueError**（此前是静默截断）。中文一个字符占 3 字节，24 个汉字即到顶 ——
|
||
因此必须按字节数校验，按字符数校验是错的。
|
||
本模块做防御性兜底，调用方（schemas 层）应更早拦截以给出友好提示。
|
||
"""
|
||
|
||
import uuid
|
||
from datetime import UTC, datetime, timedelta
|
||
from typing import Any
|
||
|
||
import jwt
|
||
from pwdlib import PasswordHash
|
||
from pwdlib.hashers.bcrypt import BcryptHasher
|
||
|
||
from app.core.config import get_settings
|
||
from app.core.constants import JWT_ALGORITHM, PASSWORD_MAX_BYTES
|
||
from app.core.exceptions import AuthError
|
||
|
||
# **显式指定 bcrypt，不用 `PasswordHash.recommended()`**：
|
||
# 后者在 pwdlib 0.3.x 推荐的是 Argon2，且会随库版本变化 —— 结果是
|
||
# 同一批密码用不同算法哈希、与规范要求的「密码 BCrypt」不符。
|
||
# 哈希算法属于会被写入数据、影响长期兼容性的决策，必须钉死。
|
||
_password_hasher = PasswordHash((BcryptHasher(),))
|
||
|
||
|
||
def hash_password(plain: str) -> str:
|
||
"""计算密码哈希。返回值为 BCrypt 格式，定长 60 字符。
|
||
|
||
:raises ValueError: 密码超过 72 字节（bcrypt 算法上限）
|
||
"""
|
||
if len(plain.encode("utf-8")) > PASSWORD_MAX_BYTES:
|
||
raise ValueError(f"密码超过 {PASSWORD_MAX_BYTES} 字节上限，无法哈希")
|
||
return _password_hasher.hash(plain)
|
||
|
||
|
||
def verify_password(plain: str, hashed: str) -> bool:
|
||
"""校验密码是否匹配哈希。
|
||
|
||
超长密码直接判为不匹配而不抛异常 —— 校验路径不应因输入长度而报错，
|
||
否则会给攻击者提供区分"密码过长"与"密码错误"的侧信道。
|
||
"""
|
||
if len(plain.encode("utf-8")) > PASSWORD_MAX_BYTES:
|
||
return False
|
||
try:
|
||
return _password_hasher.verify(plain, hashed)
|
||
except Exception:
|
||
# 哈希串损坏或格式不符时视为校验失败，不向上抛
|
||
return False
|
||
|
||
|
||
# ==================== JWT ====================
|
||
|
||
|
||
def create_access_token(
|
||
*, user_id: int, token_version: int, roles: list[str]
|
||
) -> tuple[str, str, int]:
|
||
"""签发访问令牌。
|
||
|
||
载荷含 `ver`（对应用户的 `token_version`）—— 鉴权时与库中值比对，
|
||
使「禁用账号 / 改密 / 强制下线」能立即让已签发令牌失效，**不依赖 Redis**。
|
||
|
||
:returns: ``(token, jti, 有效期秒数)``。jti 供登出时拉黑使用。
|
||
"""
|
||
settings = get_settings()
|
||
now = datetime.now(UTC)
|
||
expires_in = settings.jwt_expire_minutes * 60
|
||
jti = uuid.uuid4().hex
|
||
|
||
payload = {
|
||
"sub": str(user_id), # PyJWT 要求 sub 为字符串
|
||
"jti": jti,
|
||
"ver": token_version,
|
||
"roles": roles, # 仅供前端展示菜单，鉴权时以数据库为准
|
||
"iat": now,
|
||
"exp": now + timedelta(seconds=expires_in),
|
||
}
|
||
token = jwt.encode(payload, settings.app_secret_key, algorithm=JWT_ALGORITHM)
|
||
return token, jti, expires_in
|
||
|
||
|
||
def decode_token(token: str) -> dict[str, Any]:
|
||
"""解析并校验令牌的签名与有效期。
|
||
|
||
时间一律使用 **UTC 感知时间**：JWT 的 exp 是 UTC 时间戳，若与本地时间
|
||
（容器 TZ=Asia/Shanghai）比较会出现 8 小时偏差，表现为「刚登录就过期」
|
||
或「过期了还能用」。
|
||
|
||
:raises AuthError: 签名无效、格式错误或已过期
|
||
"""
|
||
try:
|
||
return jwt.decode(token, get_settings().app_secret_key, algorithms=[JWT_ALGORITHM])
|
||
except jwt.ExpiredSignatureError as exc:
|
||
raise AuthError("登录已过期，请重新登录") from exc
|
||
except jwt.PyJWTError as exc:
|
||
raise AuthError("登录凭证无效，请重新登录") from exc
|