fix: honor computer_use_require_admin in shipyard_neo tools (#6951)

Author: 1zzxy1

astrbot/core/computer/tools/browser.py

@@ -8,21 +8,13 @@
 from astrbot.core.astr_agent_context import AstrAgentContext
 
 from ..computer_client import get_booter
+from .permissions import check_admin_permission
 
 
 def _to_json(data: Any) -> str:
     return json.dumps(data, ensure_ascii=False, default=str)
 
 
-def _ensure_admin(context: ContextWrapper[AstrAgentContext]) -> str | None:
-    if context.context.event.role != "admin":
-        return (
-            "error: Permission denied. Browser and skill lifecycle tools are only allowed "
-            "for admin users."
-        )
-    return None
-
-
 async def _get_browser_component(context: ContextWrapper[AstrAgentContext]) -> Any:
     booter = await get_booter(
         context.context.context,
@@ -77,7 +69,7 @@ async def call(
         learn: bool = False,
         include_trace: bool = False,
     ) -> ToolExecResult:
-        if err := _ensure_admin(context):
+        if err := check_admin_permission(context, "Using browser tools"):
             return err
         try:
             browser = await _get_browser_component(context)
@@ -140,7 +132,7 @@ async def call(
         learn: bool = False,
         include_trace: bool = False,
     ) -> ToolExecResult:
-        if err := _ensure_admin(context):
+        if err := check_admin_permission(context, "Using browser tools"):
             return err
         try:
             browser = await _get_browser_component(context)
@@ -187,7 +179,7 @@ async def call(
         description: str | None = None,
         tags: str | None = None,
     ) -> ToolExecResult:
-        if err := _ensure_admin(context):
+        if err := check_admin_permission(context, "Using browser tools"):
             return err
         try:
             browser = await _get_browser_component(context)

astrbot/core/computer/tools/neo_skills.py

@@ -10,6 +10,7 @@
 from astrbot.core.skills.neo_skill_sync import NeoSkillSyncManager
 
 from ..computer_client import get_booter
+from .permissions import check_admin_permission
 
 
 def _to_jsonable(model_like: Any) -> Any:
@@ -26,12 +27,6 @@ def _to_json_text(data: Any) -> str:
     return json.dumps(_to_jsonable(data), ensure_ascii=False, default=str)
 
 
-def _ensure_admin(context: ContextWrapper[AstrAgentContext]) -> str | None:
-    if context.context.event.role != "admin":
-        return "error: Permission denied. Skill lifecycle tools are only allowed for admin users."
-    return None
-
-
 async def _get_neo_context(
     context: ContextWrapper[AstrAgentContext],
 ) -> tuple[Any, Any]:
@@ -59,7 +54,7 @@ async def _run(
         neo_call: Callable[[Any, Any], Awaitable[Any]],
         error_action: str,
     ) -> ToolExecResult:
-        if err := _ensure_admin(context):
+        if err := check_admin_permission(context, "Using skill lifecycle tools"):
             return err
         try:
             client, sandbox = await _get_neo_context(context)
@@ -392,7 +387,7 @@ async def call(
         stage: str = "canary",
         sync_to_local: bool = True,
     ) -> ToolExecResult:
-        if err := _ensure_admin(context):
+        if err := check_admin_permission(context, "Using skill lifecycle tools"):
             return err
         if stage not in {"canary", "stable"}:
             return "Error promoting skill candidate: stage must be canary or stable."

tests/test_computer_tool_permissions.py

@@ -0,0 +1,98 @@
+import json
+from types import SimpleNamespace
+
+import pytest
+
+from astrbot.core.agent.run_context import ContextWrapper
+from astrbot.core.computer.tools.browser import BrowserExecTool
+from astrbot.core.computer.tools.neo_skills import GetExecutionHistoryTool
+
+
+class _FakeBrowser:
+    async def exec(self, **kwargs):
+        return {
+            "ok": True,
+            "cmd": kwargs["cmd"],
+        }
+
+
+class _FakeSandbox:
+    async def get_execution_history(self, **kwargs):
+        return {
+            "items": [],
+            "limit": kwargs["limit"],
+        }
+
+
+def _make_run_context(require_admin: bool, role: str = "member") -> ContextWrapper:
+    config_holder = SimpleNamespace(
+        get_config=lambda umo: {  # noqa: ARG005
+            "provider_settings": {
+                "computer_use_require_admin": require_admin,
+            }
+        }
+    )
+    event = SimpleNamespace(
+        role=role,
+        unified_msg_origin="qq_official:friend:user-1",
+        get_sender_id=lambda: "user-1",
+    )
+    astr_ctx = SimpleNamespace(context=config_holder, event=event)
+    return ContextWrapper(context=astr_ctx)
+
+
+@pytest.mark.asyncio
+async def test_browser_tool_allows_non_admin_when_admin_requirement_disabled(
+    monkeypatch,
+):
+    async def _fake_get_booter(_ctx, _session_id):
+        return SimpleNamespace(browser=_FakeBrowser())
+
+    monkeypatch.setattr(
+        "astrbot.core.computer.tools.browser.get_booter",
+        _fake_get_booter,
+    )
+
+    result = await BrowserExecTool().call(
+        _make_run_context(require_admin=False),
+        cmd="open https://example.com",
+    )
+
+    assert json.loads(result)["ok"] is True
+
+
+@pytest.mark.asyncio
+async def test_neo_skill_tool_allows_non_admin_when_admin_requirement_disabled(
+    monkeypatch,
+):
+    async def _fake_get_booter(_ctx, _session_id):
+        return SimpleNamespace(
+            bay_client=object(),
+            sandbox=_FakeSandbox(),
+        )
+
+    monkeypatch.setattr(
+        "astrbot.core.computer.tools.neo_skills.get_booter",
+        _fake_get_booter,
+    )
+
+    result = await GetExecutionHistoryTool().call(
+        _make_run_context(require_admin=False),
+        limit=5,
+    )
+
+    payload = json.loads(result)
+    assert payload["items"] == []
+    assert payload["limit"] == 5
+
+
+@pytest.mark.asyncio
+async def test_browser_tool_still_denies_non_admin_when_admin_requirement_enabled():
+    result = await BrowserExecTool().call(
+        _make_run_context(require_admin=True),
+        cmd="open https://example.com",
+    )
+
+    assert "Permission denied" in result
+    assert "Using browser tools is only allowed for admin users" in result
+    assert "User's ID is: user-1" in result

tests/test_neo_skill_tools.py

@@ -54,8 +54,21 @@ async def _fake_sync_release(self, client, **kwargs):
         _fake_sync_release,
     )
 
-    event = SimpleNamespace(role="admin", unified_msg_origin="session-1")
-    astr_ctx = SimpleNamespace(context=SimpleNamespace(), event=event)
+    event = SimpleNamespace(
+        role="admin",
+        unified_msg_origin="session-1",
+        get_sender_id=lambda: "admin-user",
+    )
+    astr_ctx = SimpleNamespace(
+        context=SimpleNamespace(
+            get_config=lambda umo: {  # noqa: ARG005
+                "provider_settings": {
+                    "computer_use_require_admin": True,
+                }
+            }
+        ),
+        event=event,
+    )
     run_ctx = ContextWrapper(context=astr_ctx)
 
     tool = PromoteSkillCandidateTool()