feat(backend): exchange miniprogram webview tickets for JWT sessions.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
wuzilongandCursor committed 2026-09-14 16:11:36 +08:00
1 parent 98292cb914
commit 5eb7b0b193
3 files changed
+97

No files matched your search

+74
View File
@@ -31,6 +31,7 @@ from app.services.web_auth_roles import (
resolve_web_role_for_user,
user_has_bound_roles,
)
from app.services.mp_webview_ticket import TICKET_TTL_SEC, consume_ticket, issue_ticket
from app.crud.crm.quote import view_by_share_code
router = APIRouter(prefix="", tags=["公共接口"])
@@ -207,6 +208,79 @@ def refresh_token(body: dict, db: Session = Depends(get_db)):
return error_response(401, "refreshToken已过期")
@router.post("/web/auth/mp-webview-ticket", response_model=dict)
def mp_webview_ticket(request: Request, db: Session = Depends(get_db)):
"""小程序内嵌 WebView:签发一次性短时 ticket(需 Bearer)。"""
authorization = request.headers.get("Authorization", "")
if not authorization.startswith("Bearer "):
return error_response(401, "未授权")
token = authorization.replace("Bearer ", "").strip()
try:
current_user = get_current_user_by_token(token=token, db=db)
payload = decode_token(token)
except HTTPException:
return error_response(401, "Token无效或已过期")
session_id = str(payload.get("sid") or "").strip()
if not session_id:
return error_response(401, "Token无效或已过期")
try:
ticket = issue_ticket(
user_id=str(current_user.get("user_id") or ""),
session_id=session_id,
refresh_jti=str(payload.get("jti") or ""),
)
except RuntimeError:
return error_response(503, "暂时无法打开内嵌页,请稍后重试")
return success_response({"ticket": ticket, "expiresIn": TICKET_TTL_SEC})
@router.post("/web/auth/mp-webview-exchange", response_model=dict)
def mp_webview_exchange(body: dict, db: Session = Depends(get_db)):
"""小程序内嵌 WebView:用 ticket 兑换 Web JWT(不吊销原 refresh)。"""
ticket = str((body or {}).get("ticket") or "").strip()
if not ticket:
return error_response(400, "ticket不能为空")
try:
data = consume_ticket(ticket)
except RuntimeError:
return error_response(503, "暂时无法打开内嵌页,请稍后重试")
if not data:
return error_response(401, "ticket无效或已过期")
user_id = str(data.get("userId") or "").strip()
session_id = str(data.get("sessionId") or "").strip()
if not user_id or not session_id:
return error_response(401, "ticket无效或已过期")
if not is_session_valid(session_id):
return error_response(401, "登录已失效,请重新登录")
user = user_crud.get(db, user_id)
if not user or getattr(user, "del_flag", 0) != 0:
return error_response(401, "用户不存在或已删除")
if user.status != "1":
return error_response(403, "账号已被停用,请联系管理员")
web_role = resolve_web_role_for_user(db, user.id, user.tenant_no)
access_jti = generate_uuid()
refresh_jti = generate_uuid()
access_token = create_access_token(
data=_jwt_user_payload(user, web_role),
session_id=session_id,
token_jti=access_jti,
)
refresh_token_val = create_refresh_token(
data={"sub": user.id, "username": user.user_no, "role": web_role},
session_id=session_id,
refresh_jti=refresh_jti,
)
store_refresh_token(refresh_jti=refresh_jti, user_id=user.id)
return success_response(
{
"token": access_token,
"refreshToken": refresh_token_val,
"expiresIn": settings.ACCESS_TOKEN_EXPIRE_MINUTES * 60,
"userInfo": _build_user_info_payload(user, web_role),
}
)
@router.post("/web/auth/logout", response_model=dict)
def logout(request: Request):
"""登出:吊销 Redis 会话(若已配置 REDIS_HOST)。无 Token 或已失效时仍返回成功(幂等)。"""
+1
View File
@@ -160,6 +160,7 @@ class Settings(BaseSettings):
"/api/web/auth/login-rsa-public-key,"
"/api/web/auth/login,"
"/api/web/auth/refresh-token,"
"/api/web/auth/mp-webview-exchange,"
"/api/web/admin/attachment/upload,"
"/api/web/admin/products/upload-image,"
"/api/web/admin/categories/upload-image,"
+22
View File
@@ -0,0 +1,22 @@
"""小程序 WebView ticket 签发 / 兑换 HTTP 接口。"""
from fastapi.testclient import TestClient
from app.main import app
client = TestClient(app)
def test_mp_webview_ticket_requires_bearer():
resp = client.post("/api/web/auth/mp-webview-ticket")
assert resp.status_code == 200
body = resp.json()
assert body["code"] == 401
assert body["message"] == "未授权"
def test_mp_webview_exchange_requires_ticket():
resp = client.post("/api/web/auth/mp-webview-exchange", json={})
assert resp.status_code == 200
body = resp.json()
assert body["code"] == 400
assert body["message"] == "ticket不能为空"