Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 31 additions & 0 deletions astrbot/dashboard/api/plugins.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
70 changes: 70 additions & 0 deletions tests/test_fastapi_v1_dashboard.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
Loading