fix: allow plugin page asset token fallback (#8970)

Co-authored-by: beidou-lab-macmini <beidou-lab-macmini@beidou-lab-macminideMac-mini.local>
This commit is contained in:
FuShang114
2026-06-24 22:11:49 +08:00
committed by GitHub
parent 6a343b7656
commit b7cadfe704
2 changed files with 67 additions and 24 deletions

View File

@@ -279,37 +279,74 @@ class AstrBotDashboard:
):
return None
is_plugin_page_path = PluginPageAuth.is_protected_path(path)
token = self._extract_dashboard_jwt(current_request)
if not token and is_plugin_page_path:
token = PluginPageAuth.extract_asset_token(current_request.query_params)
if not token:
dashboard_token = self._extract_dashboard_jwt(current_request)
asset_token = (
PluginPageAuth.extract_asset_token(current_request.query_params)
if is_plugin_page_path
else None
)
token_candidates = []
if dashboard_token:
token_candidates.append(dashboard_token)
if asset_token and asset_token != dashboard_token:
token_candidates.append(asset_token)
if not token_candidates:
r = JSONResponse(error("未授权"))
r.status_code = 401
return r
token_errors: list[str] = []
for token in token_candidates:
payload, token_error = self._validate_dashboard_token(token, path)
if payload is not None:
current_request.state.dashboard_g.username = cast(
str, payload["username"]
)
return None
token_errors.append(token_error)
error_message = (
"Token 过期"
if token_errors and all(item == "Token 过期" for item in token_errors)
else "Token 无效"
)
r = JSONResponse(error(error_message))
r.status_code = 401
return r
def _validate_dashboard_token(
self,
token: str,
path: str,
) -> tuple[dict[str, Any] | None, str]:
"""Validate a dashboard JWT or scoped plugin page asset token.
Args:
token: JWT value from the Authorization header, cookie, or query string.
path: Current request path used for plugin page asset token scope checks.
Returns:
A tuple of the decoded payload and an error message. The payload is
present only when the token is valid for the current request path.
"""
try:
payload = jwt.decode(token, self._jwt_secret, algorithms=["HS256"])
if PluginPageAuth.is_asset_token(
payload
) and not PluginPageAuth.is_scope_valid(
payload,
path,
):
r = JSONResponse(error("Token 无效"))
r.status_code = 401
return r
username = payload.get("username")
if not isinstance(username, str) or not username.strip():
raise jwt.InvalidTokenError("missing username in token payload")
current_request.state.dashboard_g.username = username
except jwt.ExpiredSignatureError:
r = JSONResponse(error("Token 过期"))
r.status_code = 401
return r
return None, "Token 过期"
except jwt.InvalidTokenError:
r = JSONResponse(error("Token 无效"))
r.status_code = 401
return r
return None, "Token 无效"
if PluginPageAuth.is_asset_token(payload) and not PluginPageAuth.is_scope_valid(
payload,
path,
):
return None, "Token 无效"
username = payload.get("username")
if not isinstance(username, str) or not username.strip():
return None, "Token 无效"
return payload, ""
async def _apply_auth_rate_limit(
self,

View File

@@ -1773,6 +1773,12 @@ async def test_plugin_page_content_issues_scoped_asset_token(
css_response = await anonymous_client.get(css_url.group(1))
assert css_response.status_code == 200
stale_cookie_response = await anonymous_client.get(
app_js_url.group(1),
headers={"Cookie": f"{DASHBOARD_JWT_COOKIE_NAME}=stale.dashboard.token"},
)
assert stale_cookie_response.status_code == 200
out_of_scope_response = await anonymous_client.get(
f"/api/plugin/get?asset_token={asset_token}"
)