From f8b5d31d2e3de06ed0099fe2637f0f4432abe45f Mon Sep 17 00:00:00 2001 From: guke Date: Tue, 14 Jul 2026 17:21:52 +0800 Subject: [PATCH] =?UTF-8?q?feat(auth):=20M2=20conflict=5Fticket=20?= =?UTF-8?q?=E4=BB=A4=E7=89=8C(=E7=BC=96=E7=A0=81=E5=B7=B2=E9=AA=8C?= =?UTF-8?q?=E8=AF=81=20openid+phone)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Fable 5 --- app/core/security.py | 46 +++++++++++++++++++++++++++++++++++ tests/test_wechat_conflict.py | 29 ++++++++++++++++++++++ 2 files changed, 75 insertions(+) diff --git a/app/core/security.py b/app/core/security.py index 03bcd6f..ebbbac0 100644 --- a/app/core/security.py +++ b/app/core/security.py @@ -126,6 +126,52 @@ def decode_bind_ticket(token: str) -> dict[str, Any]: return {"openid": payload["sub"], "wnk": payload.get("wnk"), "wav": payload.get("wav")} +def create_conflict_ticket( + *, openid: str, wechat_nickname: str | None, wechat_avatar_url: str | None, phone: str +) -> str: + """手机号占用时签发的短时"冲突处理"令牌。 + + 比 bind_ticket 多编码 **已验证的手机号 phone** —— 换绑/继续绑定只认它,证明"这对 + openid/手机号刚在绑号时验证通过",免用户重输验证码,又堵住"拿自己 openid + 任意手机号 + 去夺号"的接管漏洞。typ='wechat_conflict';有效期复用 WECHAT_BIND_TICKET_EXPIRE_MINUTES。 + """ + now = _now() + expire = now + timedelta(minutes=settings.WECHAT_BIND_TICKET_EXPIRE_MINUTES) + payload: dict[str, Any] = { + "sub": openid, + "typ": "wechat_conflict", + "wnk": wechat_nickname, + "wav": wechat_avatar_url, + "phn": phone, + "iat": int(now.timestamp()), + "exp": int(expire.timestamp()), + } + return jwt.encode(payload, settings.JWT_SECRET_KEY, algorithm=settings.JWT_ALGORITHM) + + +def decode_conflict_ticket(token: str) -> dict[str, Any]: + """解析"冲突处理"令牌,校验签名/过期/类型。失败抛 TokenError。 + + 返回 {'openid': str, 'wnk': str|None, 'wav': str|None, 'phone': str}。 + """ + try: + payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=[settings.JWT_ALGORITHM]) + except jwt.ExpiredSignatureError as e: + raise TokenError("conflict ticket expired") from e + except jwt.InvalidTokenError as e: + raise TokenError(f"invalid conflict ticket: {e}") from e + if payload.get("typ") != "wechat_conflict": + raise TokenError(f"wrong token type: want=wechat_conflict got={payload.get('typ')}") + if "sub" not in payload or "phn" not in payload: + raise TokenError("conflict ticket missing sub/phn") + return { + "openid": payload["sub"], + "wnk": payload.get("wnk"), + "wav": payload.get("wav"), + "phone": payload["phn"], + } + + # ===================== 密码 hash(admin 后台账号用)===================== # 用户侧是手机号+验证码登录,不存密码;仅 admin 账号用 username+password 登录。 diff --git a/tests/test_wechat_conflict.py b/tests/test_wechat_conflict.py index 093bc9c..a1fa77d 100644 --- a/tests/test_wechat_conflict.py +++ b/tests/test_wechat_conflict.py @@ -47,3 +47,32 @@ def _occupy_via_conflict(client, monkeypatch, openid: str, phone: str, device_id def test_phone_rebind_log_model_importable() -> None: assert PhoneRebindLog.__tablename__ == "phone_rebind_log" + + +# ===== Task 2: conflict_ticket 令牌 ===== + +def test_conflict_ticket_roundtrip() -> None: + token = security.create_conflict_ticket( + openid="oid1", wechat_nickname="昵", wechat_avatar_url="http://a", phone="13900139000" + ) + claims = security.decode_conflict_ticket(token) + assert claims["openid"] == "oid1" + assert claims["wnk"] == "昵" + assert claims["wav"] == "http://a" + assert claims["phone"] == "13900139000" + + +def test_conflict_ticket_wrong_type_rejected() -> None: + # bind_ticket 冒充 conflict_ticket → TokenError(typ 不匹配) + bind = security.create_bind_ticket(openid="oid", wechat_nickname=None, wechat_avatar_url=None) + with pytest.raises(security.TokenError): + security.decode_conflict_ticket(bind) + + +def test_conflict_ticket_expired_rejected(monkeypatch) -> None: + monkeypatch.setattr(security.settings, "WECHAT_BIND_TICKET_EXPIRE_MINUTES", -1) + token = security.create_conflict_ticket( + openid="oid", wechat_nickname=None, wechat_avatar_url=None, phone="13900139000" + ) + with pytest.raises(security.TokenError): + security.decode_conflict_ticket(token)