guke
148 lines
5.6 KiB
Python
148 lines
5.6 KiB
Python
"""JWT 签发与校验。
|
|
|
|
设计:
|
|
- access_token : 短期(默认 2 小时),每个业务请求 Bearer 携带
|
|
- refresh_token: 长期(默认 30 天),过期时 POST /refresh 换新的 access
|
|
两者结构一致(都是 JWT),靠 payload 里的 `typ` 字段区分,防止用 refresh 当 access 用。
|
|
|
|
为什么不用 OAuth2 + session 表?
|
|
- 占坑 + 早期阶段,JWT 自包含、零状态,够用
|
|
- 后续需要"主动作废"(改密码/账号被盗)时再加 jti 黑名单表
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Any, Literal
|
|
|
|
import bcrypt
|
|
import jwt
|
|
|
|
from app.core.config import settings
|
|
|
|
TokenType = Literal["access", "refresh"]
|
|
|
|
|
|
class TokenError(Exception):
|
|
"""token 解析/校验失败时统一抛这个,api 层捕获后返回 401。"""
|
|
|
|
|
|
def _now() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
def create_token(*, user_id: int, token_type: TokenType) -> tuple[str, datetime]:
|
|
"""签发 token,返回 (token_string, expires_at_utc)。"""
|
|
now = _now()
|
|
if token_type == "access":
|
|
expire = now + timedelta(minutes=settings.JWT_ACCESS_TOKEN_EXPIRE_MINUTES)
|
|
else:
|
|
expire = now + timedelta(days=settings.JWT_REFRESH_TOKEN_EXPIRE_DAYS)
|
|
|
|
payload: dict[str, Any] = {
|
|
"sub": str(user_id),
|
|
"typ": token_type,
|
|
"iat": int(now.timestamp()),
|
|
"exp": int(expire.timestamp()),
|
|
}
|
|
token = jwt.encode(payload, settings.JWT_SECRET_KEY, algorithm=settings.JWT_ALGORITHM)
|
|
return token, expire
|
|
|
|
|
|
def decode_token(token: str, *, expected_type: TokenType | None = None) -> dict[str, Any]:
|
|
"""解析并校验 token。
|
|
|
|
- 签名错/过期 → TokenError
|
|
- expected_type 不匹配 → TokenError (防 refresh 当 access 用)
|
|
"""
|
|
try:
|
|
payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=[settings.JWT_ALGORITHM])
|
|
except jwt.ExpiredSignatureError as e:
|
|
raise TokenError("token expired") from e
|
|
except jwt.InvalidTokenError as e:
|
|
raise TokenError(f"invalid token: {e}") from e
|
|
|
|
if expected_type is not None and payload.get("typ") != expected_type:
|
|
raise TokenError(f"wrong token type: want={expected_type} got={payload.get('typ')}")
|
|
|
|
if "sub" not in payload:
|
|
raise TokenError("token missing sub")
|
|
|
|
return payload
|
|
|
|
|
|
def issue_token_pair(user_id: int) -> dict[str, Any]:
|
|
"""登录/刷新成功后调,一次性发回 access + refresh。
|
|
|
|
返回字段对齐 OAuth2 习惯,前端易接:
|
|
access_token / refresh_token / token_type / expires_in
|
|
"""
|
|
access, access_exp = create_token(user_id=user_id, token_type="access")
|
|
refresh, refresh_exp = create_token(user_id=user_id, token_type="refresh")
|
|
return {
|
|
"access_token": access,
|
|
"refresh_token": refresh,
|
|
"token_type": "Bearer",
|
|
"expires_in": int((access_exp - _now()).total_seconds()),
|
|
"refresh_expires_in": int((refresh_exp - _now()).total_seconds()),
|
|
}
|
|
|
|
|
|
def create_bind_ticket(
|
|
*, openid: str, wechat_nickname: str | None, wechat_avatar_url: str | None
|
|
) -> str:
|
|
"""微信登录未命中 openid 时,签发短时"待绑手机"令牌,承载 openid + 微信昵称头像。
|
|
|
|
typ='wechat_bind'、sub=openid;有效期 settings.WECHAT_BIND_TICKET_EXPIRE_MINUTES 分钟。
|
|
与 access/refresh 用同一 JWT_SECRET_KEY 签名,靠 typ 区分,decode_bind_ticket 校验 typ。
|
|
"""
|
|
now = _now()
|
|
expire = now + timedelta(minutes=settings.WECHAT_BIND_TICKET_EXPIRE_MINUTES)
|
|
payload: dict[str, Any] = {
|
|
"sub": openid,
|
|
"typ": "wechat_bind",
|
|
"wnk": wechat_nickname,
|
|
"wav": wechat_avatar_url,
|
|
"iat": int(now.timestamp()),
|
|
"exp": int(expire.timestamp()),
|
|
}
|
|
return jwt.encode(payload, settings.JWT_SECRET_KEY, algorithm=settings.JWT_ALGORITHM)
|
|
|
|
|
|
def decode_bind_ticket(token: str) -> dict[str, Any]:
|
|
"""解析"待绑手机"令牌,校验签名/过期/类型。失败抛 TokenError。
|
|
|
|
返回 {'openid': str, 'wnk': str|None, 'wav': str|None}。
|
|
"""
|
|
try:
|
|
payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=[settings.JWT_ALGORITHM])
|
|
except jwt.ExpiredSignatureError as e:
|
|
raise TokenError("bind ticket expired") from e
|
|
except jwt.InvalidTokenError as e:
|
|
raise TokenError(f"invalid bind ticket: {e}") from e
|
|
if payload.get("typ") != "wechat_bind":
|
|
raise TokenError(f"wrong token type: want=wechat_bind got={payload.get('typ')}")
|
|
if "sub" not in payload:
|
|
raise TokenError("bind ticket missing sub")
|
|
return {"openid": payload["sub"], "wnk": payload.get("wnk"), "wav": payload.get("wav")}
|
|
|
|
|
|
# ===================== 密码 hash(admin 后台账号用)=====================
|
|
# 用户侧是手机号+验证码登录,不存密码;仅 admin 账号用 username+password 登录。
|
|
|
|
def hash_password(plain: str) -> str:
|
|
"""bcrypt 加盐 hash,返回可入库的字符串。
|
|
|
|
bcrypt 限制明文 ≤72 字节,超出抛 ValueError(实测 bcrypt 5.0:25 个中文=75 字节即触发)。
|
|
统一截断到 72 字节(bcrypt 标准做法)——72 字节仍是强密码,verify 同样截断保证一致。
|
|
"""
|
|
return bcrypt.hashpw(plain.encode("utf-8")[:72], bcrypt.gensalt()).decode("utf-8")
|
|
|
|
|
|
def verify_password(plain: str, hashed: str) -> bool:
|
|
"""校验明文与 bcrypt hash。明文同样截断 72 字节(与 hash_password 一致,否则超长密码永远不匹配);
|
|
hash 串损坏/格式非法时返回 False,不抛异常。"""
|
|
try:
|
|
return bcrypt.checkpw(plain.encode("utf-8")[:72], hashed.encode("utf-8"))
|
|
except (ValueError, TypeError):
|
|
return False
|