From 2a82b203658080e3b9aeb20fd6b5402fbbd0d817 Mon Sep 17 00:00:00 2001 From: guke Date: Mon, 13 Jul 2026 14:40:47 +0800 Subject: [PATCH] =?UTF-8?q?feat(auth):=20=E5=BE=AE=E4=BF=A1=E7=99=BB?= =?UTF-8?q?=E5=BD=95=20bind=5Fticket=20=E4=BB=A4=E7=89=8C=20+=20=E9=85=8D?= =?UTF-8?q?=E7=BD=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.8 --- app/core/config.py | 3 +++ app/core/security.py | 39 +++++++++++++++++++++++++++++++++ tests/test_wechat_login.py | 44 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 86 insertions(+) create mode 100644 tests/test_wechat_login.py diff --git a/app/core/config.py b/app/core/config.py index bf27932..b8d554d 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -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 无法越权访问后台。 diff --git a/app/core/security.py b/app/core/security.py index 8e852be..03bcd6f 100644 --- a/app/core/security.py +++ b/app/core/security.py @@ -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 登录。 diff --git a/tests/test_wechat_login.py b/tests/test_wechat_login.py new file mode 100644 index 0000000..f60b158 --- /dev/null +++ b/tests/test_wechat_login.py @@ -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)