(request: Request)
| 83 | |
| 84 | |
| 85 | async def require_dashboard_user(request: Request) -> str: |
| 86 | if username := _get_dashboard_state_username(request): |
| 87 | return username |
| 88 | |
| 89 | token = _extract_dashboard_jwt(request) |
| 90 | if not token: |
| 91 | raise ApiError("未授权", status_code=401) |
| 92 | |
| 93 | try: |
| 94 | payload = jwt.decode( |
| 95 | token, |
| 96 | request.app.state.jwt_secret, |
| 97 | algorithms=["HS256"], |
| 98 | ) |
| 99 | except jwt.ExpiredSignatureError as exc: |
| 100 | raise ApiError("Token 过期", status_code=401) from exc |
| 101 | except jwt.InvalidTokenError as exc: |
| 102 | raise ApiError("Token 无效", status_code=401) from exc |
| 103 | |
| 104 | username = payload.get("username") |
| 105 | if not isinstance(username, str) or not username.strip(): |
| 106 | raise ApiError("Token 无效", status_code=401) |
| 107 | return username |
| 108 | |
| 109 | |
| 110 | async def _require_api_key_scope( |
nothing calls this directly
no test coverage detected