Skip to content
Merged
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
16 changes: 4 additions & 12 deletions astrbot/core/computer/tools/browser.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down
11 changes: 3 additions & 8 deletions astrbot/core/computer/tools/neo_skills.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand All @@ -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]:
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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."
Expand Down
98 changes: 98 additions & 0 deletions tests/test_computer_tool_permissions.py
Original file line number Diff line number Diff line change
@@ -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
17 changes: 15 additions & 2 deletions tests/test_neo_skill_tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down
Loading