Skip to content
Draft
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
434 changes: 401 additions & 33 deletions astrbot/core/computer/booters/local.py

Large diffs are not rendered by default.

2 changes: 1 addition & 1 deletion astrbot/core/config/default.py
Original file line number Diff line number Diff line change
Expand Up @@ -3442,7 +3442,7 @@
"provider_settings.computer_use_require_admin": {
"description": "需要 AstrBot 管理员权限",
"type": "bool",
"hint": "开启后,需要 AstrBot 管理员权限才能调用使用电脑能力。在平台配置->管理员中可添加管理员。使用 /sid 指令查看管理员 ID。",
"hint": "开启后,需要 AstrBot 管理员权限才能调用使用电脑能力。在平台配置->管理员中可添加管理员。关闭后,Local 模式允许 Linux 普通成员通过 bubblewrap、macOS 普通成员通过 Seatbelt 在工作区沙箱内运行 Shell 和 Python;其他 Local 系统仍只允许管理员。使用 /sid 指令查看管理员 ID。",
},
"provider_settings.sandbox.booter": {
"description": "沙箱环境驱动器",
Expand Down
22 changes: 12 additions & 10 deletions astrbot/core/tools/computer_tools/fs.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,8 +24,9 @@
- Member + sandbox: read/write/edit/grep are also not path-restricted by this
module. Upload/download are denied by `check_admin_permission` if invoked.

When `computer_use_require_admin=False`, member behavior in this module matches
admin behavior.
When `computer_use_require_admin=False`, local members remain path-restricted.
This is required because their shell and Python tools may be enabled inside a
workspace-only operating-system sandbox.

Local path resolution rule:
- In local runtime, relative paths are resolved under the primary workspace.
Expand Down Expand Up @@ -149,14 +150,15 @@ def _write_allowed_roots(


def _is_restricted_env(context: ContextWrapper[AstrAgentContext]) -> bool:
if not is_local_runtime(context):
return False
cfg = context.context.context.get_config(
umo=context.context.event.unified_msg_origin
)
provider_settings = cfg.get("provider_settings", {})
require_admin = provider_settings.get("computer_use_require_admin", True)
return require_admin and context.context.event.role != "admin"
"""Return whether Local file access must stay within approved roots.

Args:
context: Tool call context.

Returns:
True for every non-admin Local tool call.
"""
return is_local_runtime(context) and context.context.event.role != "admin"


def _resolve_tool_path(
Expand Down
16 changes: 14 additions & 2 deletions astrbot/core/tools/computer_tools/python.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,12 +7,14 @@
from astrbot.core.agent.run_context import ContextWrapper
from astrbot.core.agent.tool import ToolExecResult
from astrbot.core.astr_agent_context import AstrAgentContext, AstrMessageEvent
from astrbot.core.computer.booters.local import LocalPythonComponent
from astrbot.core.computer.computer_client import get_booter, get_local_booter
from astrbot.core.message.message_event_result import MessageChain

from ..registry import builtin_tool
from .util import (
check_admin_permission,
check_local_execution_permission,
workspace_root_for_context,
)

Expand Down Expand Up @@ -119,7 +121,8 @@ class LocalPythonTool(FunctionTool):
name: str = "astrbot_execute_python"
description: str = (
f"Execute codes in a Python environment. Current OS: {_OS_NAME}. "
"Use system-compatible commands."
"Use system-compatible commands. Non-admin Linux and macOS calls run "
"without network access in a workspace-restricted sandbox."
)

parameters: dict = field(default_factory=lambda: param_schema)
Expand All @@ -131,14 +134,22 @@ async def call(
silent: bool = False,
timeout: int = 30,
) -> ToolExecResult:
if permission_error := check_admin_permission(context, "Python execution"):
sandboxed, permission_error = check_local_execution_permission(
context,
"Python execution",
)
if permission_error:
return permission_error
sb = get_local_booter()
if not isinstance(sb.python, LocalPythonComponent):
return "Error executing code: local Python component is unavailable."
effective_timeout = (
min(timeout, context.tool_call_timeout)
if timeout > 0
else context.tool_call_timeout
)
if sandboxed:
effective_timeout = min(effective_timeout, 300)
try:
current_workspace_root = await workspace_root_for_context(context)
current_workspace_root.mkdir(parents=True, exist_ok=True)
Expand All @@ -147,6 +158,7 @@ async def call(
timeout=effective_timeout,
silent=silent,
cwd=str(current_workspace_root),
sandboxed=sandboxed,
)
return await handle_result(result, context.context.event)
except Exception as e:
Expand Down
20 changes: 14 additions & 6 deletions astrbot/core/tools/computer_tools/shell.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@

from ..registry import builtin_tool
from .util import (
check_admin_permission,
check_local_execution_permission,
is_local_runtime,
workspace_root_for_context,
)
Expand Down Expand Up @@ -98,7 +98,11 @@ async def call(
env: dict[str, Any] | None = None,
yield_time_ms: int = 10_000,
) -> ToolExecResult:
if permission_error := check_admin_permission(context, "Shell execution"):
sandboxed, permission_error = check_local_execution_permission(
context,
"Shell execution",
)
if permission_error:
return permission_error

sb = await get_booter(
Expand All @@ -125,8 +129,9 @@ async def call(
owner_id=context.context.event.unified_msg_origin,
cwd=cwd,
env=env,
timeout=timeout,
timeout=min(timeout or 300, 300) if sandboxed else timeout,
yield_time_ms=0 if background else yield_time_ms,
sandboxed=sandboxed,
),
ensure_ascii=False,
)
Expand Down Expand Up @@ -168,7 +173,9 @@ class LocalExecuteShellTool(ExecuteShellTool):

description: str = (
"Execute a command in the shell. If it is still running after "
"yield_time_ms, the tool returns a managed shell session ID."
"yield_time_ms, the tool returns a managed shell session ID. "
"Non-admin Linux and macOS calls run without network access in a "
"workspace-restricted sandbox."
)
parameters: dict = field(
default_factory=lambda: {
Expand Down Expand Up @@ -307,10 +314,11 @@ async def call(
Returns:
JSON session operation result or a user-facing error.
"""
if permission_error := check_admin_permission(
_, permission_error = check_local_execution_permission(
context,
"Shell session management",
):
)
if permission_error:
return permission_error
if not is_local_runtime(context):
return "Error managing shell session: only local runtime is supported."
Expand Down
26 changes: 26 additions & 0 deletions astrbot/core/tools/computer_tools/util.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import sys
from pathlib import Path

from astrbot.core.agent.run_context import ContextWrapper
Expand Down Expand Up @@ -69,3 +70,28 @@ def check_admin_permission(
f"User's ID is: {context.context.event.get_sender_id()}. User's ID can be found by using /sid command."
)
return None


def check_local_execution_permission(
context: ContextWrapper[AstrAgentContext],
operation_name: str,
) -> tuple[bool, str | None]:
"""Resolve whether an execution tool needs an operating-system sandbox.

Args:
context: Tool call context.
operation_name: User-facing name included in permission errors.

Returns:
A tuple of whether bubblewrap is required and an optional error.
"""
if permission_error := check_admin_permission(context, operation_name):
return False, permission_error
if not is_local_runtime(context) or context.context.event.role == "admin":
return False, None
if not (sys.platform.startswith("linux") or sys.platform == "darwin"):
return False, (
"error: Permission denied. Non-admin Local execution is only supported "
"on Linux with bubblewrap or macOS with Seatbelt."
)
return True, None
18 changes: 18 additions & 0 deletions tests/test_computer_fs_tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -343,6 +343,24 @@ async def test_restricted_local_member_can_read_plugin_skill_inventory_even_if_p
assert result == "# Demo Skill\n"


@pytest.mark.asyncio
async def test_local_member_stays_restricted_when_admin_requirement_is_disabled(
monkeypatch: pytest.MonkeyPatch,
tmp_path,
):
_setup_local_fs_tools(monkeypatch, tmp_path)
outside_file = tmp_path / "outside.txt"
outside_file.write_text("host secret\n", encoding="utf-8")

result = await fs_tools.FileReadTool().call(
_make_context(role="member", require_admin=False),
path=str(outside_file),
)

assert "Read access is restricted for this user." in result
assert "host secret" not in result


@pytest.mark.asyncio
async def test_restricted_local_member_cannot_write_plugin_provided_skill(
monkeypatch: pytest.MonkeyPatch,
Expand Down
Loading
Loading