diff --git a/backend/app/services/mp_webview_ticket.py b/backend/app/services/mp_webview_ticket.py new file mode 100644 index 000000000..320256acf --- /dev/null +++ b/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 diff --git a/backend/tests/test_mp_webview_ticket.py b/backend/tests/test_mp_webview_ticket.py new file mode 100644 index 000000000..e419b0df1 --- /dev/null +++ b/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")