feat(auth): 微信登录 bind_ticket 令牌 + 配置

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
guke
2026-07-13 14:40:47 +08:00
parent 824045dd19
commit 2a82b20365
3 changed files with 86 additions and 0 deletions
+3
View File
@@ -44,6 +44,9 @@ class Settings(BaseSettings):
JWT_ALGORITHM: str = "HS256"
JWT_ACCESS_TOKEN_EXPIRE_MINUTES: int = 120
JWT_REFRESH_TOKEN_EXPIRE_DAYS: int = 30
# 微信登录未命中 openid 时签发的"待绑手机"令牌有效期(分钟)。
# 见 auth.wechat_login / security.create_bind_ticket。
WECHAT_BIND_TICKET_EXPIRE_MINUTES: int = 10
# ===== Admin 后台 =====
# admin 用独立 JWT secret(≠ JWT_SECRET_KEY),App 用户 token 无法越权访问后台。
+39
View File
@@ -87,6 +87,45 @@ def issue_token_pair(user_id: int) -> dict[str, Any]:
}
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 登录。
+44
View File
@@ -0,0 +1,44 @@
"""微信登录 M1 测试:bind_ticket 令牌、wechat-login(openid 命中/未命中)、
bind-phone(建号/占用/令牌过期)。
沿用 tests/test_auth.py 风格:HTTP 走 client fixture;微信 code→openid 用 monkeypatch
拦掉(conftest 里 WECHAT_APP_ID/SECRET 是 dummy,不真连微信);短信走 SMS_MOCK(任意 6 位通过)。
"""
from __future__ import annotations
import pytest
from app.api.v1 import auth
from app.core import security
from app.integrations import wxpay
def _fake_userinfo(openid: str, nickname: str | None = "微信昵称", avatar: str | None = "http://x/a.png"):
"""返回一个可传给 monkeypatch 的假 code_to_userinfo(忽略 code,固定返回给定 openid)。"""
def _f(code: str) -> dict:
return {"openid": openid, "nickname": nickname, "avatar_url": avatar, "raw": {}}
return _f
# ===== Task 1: bind_ticket 令牌 =====
def test_bind_ticket_roundtrip() -> None:
token = security.create_bind_ticket(openid="oid1", wechat_nickname="", wechat_avatar_url="http://a")
claims = security.decode_bind_ticket(token)
assert claims["openid"] == "oid1"
assert claims["wnk"] == ""
assert claims["wav"] == "http://a"
def test_bind_ticket_wrong_type_rejected() -> None:
# 用 access token 冒充 bind_ticket → TokenError(typ 不匹配)
access, _ = security.create_token(user_id=1, token_type="access")
with pytest.raises(security.TokenError):
security.decode_bind_ticket(access)
def test_bind_ticket_expired_rejected(monkeypatch) -> None:
monkeypatch.setattr(security.settings, "WECHAT_BIND_TICKET_EXPIRE_MINUTES", -1)
token = security.create_bind_ticket(openid="oid", wechat_nickname=None, wechat_avatar_url=None)
with pytest.raises(security.TokenError):
security.decode_bind_ticket(token)