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
|