From f6403426532bb8e414d9a9ba59e366e30d3aff9f Mon Sep 17 00:00:00 2001 From: lxfight <1686540385@qq.com> Date: Wed, 2 Sep 2026 13:54:53 +0800 Subject: [PATCH] fix: reject extension API calls for disabled plugins _call_plugin_extension executed handlers of disabled plugins, while the static plugin pages already return 403 for the same state. Resolve the owning plugin from the handler module (function module or bound-method owner class) and return an error response when it is disabled. Legacy routes whose handlers live outside the plugin module namespaces are unaffected. --- astrbot/dashboard/api/plugins.py | 31 +++++++++++++ tests/test_fastapi_v1_dashboard.py | 70 ++++++++++++++++++++++++++++++ 2 files changed, 101 insertions(+) diff --git a/astrbot/dashboard/api/plugins.py b/astrbot/dashboard/api/plugins.py index 319110a02d..bbb6065e81 100644 --- a/astrbot/dashboard/api/plugins.py +++ b/astrbot/dashboard/api/plugins.py @@ -205,6 +205,37 @@ async def _call_plugin_extension( view_handler, path_values = matched_api plugin_name = plugin_path.strip("/").split("/", 1)[0].strip() or None + + # Gate calls on the owning plugin's activation state, mirroring the + # static plugin pages behavior. Ownership is resolved from the handler + # module so legacy routes without a plugin name prefix keep working. + handler_modules = set() + handler_module = getattr(view_handler, "__module__", None) + if isinstance(handler_module, str): + handler_modules.add(handler_module) + handler_owner = getattr(view_handler, "__self__", None) + if handler_owner is not None: + handler_modules.add(type(handler_owner).__module__) + + plugin = None + if handler_modules: + for ( + star + ) in request.app.state.core_lifecycle.plugin_manager.context.get_all_stars(): + if not star.root_dir_name: + continue + prefix = ( + "astrbot.builtin_stars" if star.reserved else "data.plugins" + ) + f".{star.root_dir_name}" + if any( + module == prefix or module.startswith(f"{prefix}.") + for module in handler_modules + ): + plugin = star + break + + if plugin is not None and not plugin.activated: + return {"status": "error", "message": "插件未启用", "data": {}} plugin_request = PluginRequest( request, path_params=path_values, diff --git a/tests/test_fastapi_v1_dashboard.py b/tests/test_fastapi_v1_dashboard.py index fb0b9087ae..a1e81bcb78 100644 --- a/tests/test_fastapi_v1_dashboard.py +++ b/tests/test_fastapi_v1_dashboard.py @@ -3005,6 +3005,76 @@ def test_astrbot_web_request_requires_plugin_context(): _ = plugin_request.method +@pytest.mark.asyncio +async def test_v1_plugin_extension_rejects_disabled_plugin( + asgi_client: httpx.AsyncClient, + fake_core_lifecycle, +): + from astrbot.api.web import json_response + + async def plugin_owned_extension(): + return json_response({"ok": True}) + + plugin_owned_extension.__module__ = "data.plugins.demo_plugin.main" + + disabled_plugin = SimpleNamespace( + name="astrbot_plugin_demo", + root_dir_name="demo_plugin", + reserved=False, + activated=False, + ) + fake_core_lifecycle.plugin_manager.context.get_all_stars = lambda: [disabled_plugin] + fake_core_lifecycle.star_context.registered_web_apis = [ + ("/demo_plugin/status", plugin_owned_extension, ["GET"], "demo") + ] + + response = await asgi_client.get( + "/api/v1/plugins/extensions/demo_plugin/status", + headers=_jwt_headers(), + ) + + assert response.status_code == 200 + assert response.json() == { + "status": "error", + "message": "插件未启用", + "data": {}, + } + + +@pytest.mark.asyncio +async def test_v1_plugin_extension_allows_activated_plugin( + asgi_client: httpx.AsyncClient, + fake_core_lifecycle, +): + from astrbot.api.web import json_response + + async def plugin_owned_extension(): + return json_response({"ok": True}) + + plugin_owned_extension.__module__ = "data.plugins.demo_plugin.main" + + activated_plugin = SimpleNamespace( + name="astrbot_plugin_demo", + root_dir_name="demo_plugin", + reserved=False, + activated=True, + ) + fake_core_lifecycle.plugin_manager.context.get_all_stars = lambda: [ + activated_plugin + ] + fake_core_lifecycle.star_context.registered_web_apis = [ + ("/demo_plugin/status", plugin_owned_extension, ["GET"], "demo") + ] + + response = await asgi_client.get( + "/api/v1/plugins/extensions/demo_plugin/status", + headers=_jwt_headers(), + ) + + assert response.status_code == 200 + assert response.json() == {"ok": True} + + def test_astrbot_web_request_proxy_exposes_typed_methods(): from typing import get_type_hints