feat(backend): add one-time Redis tickets for miniprogram webview login.

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/app/services/mp_webview_ticket.py

@@ -0,0 +1,50 @@
+"""小程序 WebView 一次性短时 ticket（不把 JWT 放进 URL）。"""
+from __future__ import annotations
+
+import json
+from typing import Optional
+
+from app.core.redis_client import get_sync_redis_client
+from app.core.utils import generate_uuid
+
+TICKET_TTL_SEC = 60
+KEY_PREFIX = "mp_wv_ticket:"
+
+
+def _key(ticket: str) -> str:
+    return f"{KEY_PREFIX}{ticket}"
+
+
+def issue_ticket(user_id: str, session_id: str, refresh_jti: str = "") -> str:
+    r = get_sync_redis_client()
+    if r is None:
+        raise RuntimeError("Redis 不可用，无法签发 WebView ticket")
+    ticket = generate_uuid()
+    body = json.dumps(
+        {"userId": user_id, "sessionId": session_id, "refreshJti": refresh_jti or ""},
+        ensure_ascii=False,
+    )
+    ok = r.set(_key(ticket), body, ex=TICKET_TTL_SEC, nx=True)
+    if not ok:
+        raise RuntimeError("签发 WebView ticket 失败")
+    return ticket
+
+
+def consume_ticket(ticket: str) -> Optional[dict]:
+    raw = (ticket or "").strip()
+    if not raw:
+        return None
+    r = get_sync_redis_client()
+    if r is None:
+        raise RuntimeError("Redis 不可用，无法兑换 WebView ticket")
+    key = _key(raw)
+    payload = r.get(key)
+    r.delete(key)
+    if not payload:
+        return None
+    if isinstance(payload, bytes):
+        payload = payload.decode("utf-8")
+    data = json.loads(payload)
+    if not isinstance(data, dict):
+        return None
+    return data

backend/tests/test_mp_webview_ticket.py

@@ -0,0 +1,46 @@
+import json
+from unittest.mock import patch
+
+import pytest
+
+from app.services import mp_webview_ticket as svc
+
+
+class FakeRedis:
+    def __init__(self):
+        self.data = {}
+
+    def set(self, key, value, ex=None, nx=False):
+        if nx and key in self.data:
+            return False
+        self.data[key] = value
+        return True
+
+    def get(self, key):
+        return self.data.get(key)
+
+    def delete(self, key):
+        return 1 if self.data.pop(key, None) is not None else 0
+
+
+def test_issue_and_consume_once():
+    fake = FakeRedis()
+    with patch.object(svc, "get_sync_redis_client", return_value=fake):
+        ticket = svc.issue_ticket("u1", "sid-1", "rj-1")
+        payload = svc.consume_ticket(ticket)
+        assert payload["userId"] == "u1"
+        assert payload["sessionId"] == "sid-1"
+        assert payload["refreshJti"] == "rj-1"
+        assert svc.consume_ticket(ticket) is None
+
+
+def test_consume_unknown_returns_none():
+    fake = FakeRedis()
+    with patch.object(svc, "get_sync_redis_client", return_value=fake):
+        assert svc.consume_ticket("missing") is None
+
+
+def test_issue_without_redis_raises():
+    with patch.object(svc, "get_sync_redis_client", return_value=None):
+        with pytest.raises(RuntimeError, match="Redis"):
+            svc.issue_ticket("u1", "sid-1")