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
46 changes: 44 additions & 2 deletions astrbot/core/computer/booters/local.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
import subprocess
import sys
from dataclasses import dataclass
from typing import Any
from typing import Any, Callable

from astrbot.api import logger
from astrbot.core.utils.astrbot_path import (
Expand Down Expand Up @@ -53,6 +53,45 @@ def _ensure_safe_path(path: str) -> str:
return abs_path


def _resolve_working_dir(configured_path: str | None, fallback_func: Callable[[], str]) -> tuple[str, bool]:
"""Resolve working directory with fallback to default.

Args:
configured_path: The configured working directory path, or None
fallback_func: A callable that returns the fallback path (e.g., get_astrbot_root)

Returns:
A tuple of (resolved_path, was_fallback) where was_fallback indicates if fallback was used
"""
if not configured_path:
return fallback_func(), True

try:
abs_path = _ensure_safe_path(configured_path)
except PermissionError:
Comment on lines +56 to +71
Copy link

Copilot AI Apr 1, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

_resolve_working_dir() relies on _ensure_safe_path() for root checks, but _ensure_safe_path() uses os.path.abspath() + startswith(), which can be bypassed via symlinks (and also via prefix paths like /allowed_root_tmp). Since this PR exposes configurable working dirs, consider switching to realpath/Path.resolve() and an os.path.commonpath()-based containment check to enforce allowed roots correctly.

Copilot uses AI. Check for mistakes.
logger.warning(
f"[Computer] Configured path '{configured_path}' is outside allowed roots, "
f"falling back to default directory."
)
return fallback_func(), True

if not os.path.exists(abs_path):
logger.warning(
f"[Computer] Configured path '{configured_path}' does not exist, "
f"falling back to default directory."
)
return fallback_func(), True

if not os.access(abs_path, os.R_OK | os.W_OK):
logger.warning(
f"[Computer] Configured path '{configured_path}' is not accessible (no read/write permission), "
f"falling back to default directory."
)
return fallback_func(), True

return abs_path, False
Comment on lines +78 to +92
Copy link

Copilot AI Apr 1, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

_resolve_working_dir() checks os.path.exists(abs_path) but does not ensure the resolved path is a directory. If a file path is configured, subprocess cwd= will raise NotADirectoryError. Consider validating os.path.isdir(abs_path) (and falling back) to make behavior predictable.

Copilot uses AI. Check for mistakes.


def _decode_bytes_with_fallback(
output: bytes | None,
*,
Expand Down Expand Up @@ -110,7 +149,7 @@ def _run() -> dict[str, Any]:
run_env = os.environ.copy()
if env:
run_env.update({str(k): str(v) for k, v in env.items()})
working_dir = _ensure_safe_path(cwd) if cwd else get_astrbot_root()
working_dir, _ = _resolve_working_dir(cwd, get_astrbot_root)
if background:
# `command` is intentionally executed through the current shell so
# local computer-use behavior matches existing tool semantics.
Expand Down Expand Up @@ -152,14 +191,17 @@ async def exec(
kernel_id: str | None = None,
timeout: int = 30,
silent: bool = False,
cwd: str | None = None,
) -> dict[str, Any]:
def _run() -> dict[str, Any]:
try:
working_dir, _ = _resolve_working_dir(cwd, get_astrbot_root)
result = subprocess.run(
[os.environ.get("PYTHON", sys.executable), "-c", code],
timeout=timeout,
capture_output=True,
text=True,
cwd=working_dir,
)
stdout = "" if silent else result.stdout
stderr = result.stderr if result.returncode != 0 else ""
Expand Down
1 change: 1 addition & 0 deletions astrbot/core/computer/olayer/python.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ async def exec(
kernel_id: str | None = None,
timeout: int = 30,
silent: bool = False,
cwd: str | None = None,
) -> dict[str, Any]:
"""Execute Python code"""
...
9 changes: 9 additions & 0 deletions astrbot/core/computer/tools/permissions.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,15 @@
from astrbot.core.astr_agent_context import AstrAgentContext


def get_configured_cwd(
context: ContextWrapper[AstrAgentContext], config_key: str
) -> str | None:
cfg = context.context.context.get_config(
umo=context.context.event.unified_msg_origin
)
return cfg.get("provider_settings", {}).get(config_key, None)


def check_admin_permission(
context: ContextWrapper[AstrAgentContext], operation_name: str
) -> str | None:
Expand Down
12 changes: 9 additions & 3 deletions astrbot/core/computer/tools/python.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@
from astrbot.core.agent.tool import ToolExecResult
from astrbot.core.astr_agent_context import AstrAgentContext, AstrMessageEvent
from astrbot.core.computer.computer_client import get_booter, get_local_booter
from astrbot.core.computer.tools.permissions import check_admin_permission
from astrbot.core.computer.tools.permissions import check_admin_permission, get_configured_cwd
from astrbot.core.message.message_event_result import MessageChain

_OS_NAME = platform.system()
Expand Down Expand Up @@ -61,6 +61,10 @@ async def handle_result(result: dict, event: AstrMessageEvent) -> ToolExecResult
return resp


def _get_configured_python_cwd(context: ContextWrapper[AstrAgentContext]) -> str | None:
return get_configured_cwd(context, "computer_use_local_python_cwd")


@dataclass
class PythonTool(FunctionTool):
name: str = "astrbot_execute_ipython"
Expand All @@ -77,7 +81,8 @@ async def call(
context.context.event.unified_msg_origin,
)
try:
result = await sb.python.exec(code, silent=silent)
cwd = _get_configured_python_cwd(context)
result = await sb.python.exec(code, silent=silent, cwd=cwd)
return await handle_result(result, context.context.event)
Comment on lines 83 to 86
Copy link

Copilot AI Apr 1, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

sb.python.exec(..., cwd=cwd) will raise TypeError: exec() got an unexpected keyword argument 'cwd' for non-local booters that haven't been updated to accept the new cwd parameter (e.g. Shipyard Neo's NeoPythonComponent.exec currently lacks it). To keep sandbox runtimes working, either update all PythonComponent implementations to accept cwd, or only pass the kwarg when cwd is not None and the target implementation supports it.

Copilot uses AI. Check for mistakes.
except Exception as e:
return f"Error executing code: {str(e)}"
Expand All @@ -100,7 +105,8 @@ async def call(
return permission_error
sb = get_local_booter()
try:
result = await sb.python.exec(code, silent=silent)
cwd = _get_configured_python_cwd(context)
result = await sb.python.exec(code, silent=silent, cwd=cwd)
return await handle_result(result, context.context.event)
except Exception as e:
return f"Error executing code: {str(e)}"
8 changes: 6 additions & 2 deletions astrbot/core/computer/tools/shell.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
from astrbot.core.astr_agent_context import AstrAgentContext

from ..computer_client import get_booter, get_local_booter
from .permissions import check_admin_permission
from .permissions import check_admin_permission, get_configured_cwd


@dataclass
Expand Down Expand Up @@ -40,6 +40,9 @@ class ExecuteShellTool(FunctionTool):

is_local: bool = False

def _get_configured_cwd(self, context: ContextWrapper[AstrAgentContext]) -> str | None:
return get_configured_cwd(context, "computer_use_local_shell_cwd")

async def call(
self,
context: ContextWrapper[AstrAgentContext],
Expand All @@ -58,7 +61,8 @@ async def call(
context.context.event.unified_msg_origin,
)
try:
result = await sb.shell.exec(command, background=background, env=env)
cwd = self._get_configured_cwd(context)
result = await sb.shell.exec(command, cwd=cwd, background=background, env=env)
return json.dumps(result)
except Exception as e:
return f"Error executing command: {str(e)}"
16 changes: 16 additions & 0 deletions astrbot/core/config/default.py
Original file line number Diff line number Diff line change
Expand Up @@ -3147,6 +3147,22 @@ class ChatProviderTemplate(TypedDict):
"type": "bool",
"hint": "开启后,需要 AstrBot 管理员权限才能调用使用电脑能力。在平台配置->管理员中可添加管理员。使用 /sid 指令查看管理员 ID。",
},
"provider_settings.computer_use_local_shell_cwd": {
"description": "本地 Shell 默认工作目录 / Local Shell Default Working Directory",
"type": "string",
"hint": "zh: 设置本地 shell 命令执行时的默认工作目录。仅在 computer_use_runtime=local 时生效。留空则使用 AstrBot 根目录。\nen: Set the default working directory for local shell command execution. Only effective when computer_use_runtime=local. If empty, uses AstrBot root directory.",
"condition": {
"provider_settings.computer_use_runtime": "local",
},
},
"provider_settings.computer_use_local_python_cwd": {
"description": "本地 Python 默认工作目录 / Local Python Default Working Directory",
"type": "string",
"hint": "zh: 设置本地 Python 代码执行时的默认工作目录。仅在 computer_use_runtime=local 时生效。留空则使用 AstrBot 根目录。\nen: Set the default working directory for local Python code execution. Only effective when computer_use_runtime=local. If empty, uses AstrBot root directory.",
"condition": {
"provider_settings.computer_use_runtime": "local",
},
},
Comment on lines +3150 to +3165
Copy link

Copilot AI Apr 1, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

WebUI hints say the cwd options are "only effective when computer_use_runtime=local", but the local runtime also restricts working dirs to allowed roots (root/data/temp) and will silently fall back with a warning when outside/invalid. Consider documenting these restrictions here to reduce user confusion when a configured path doesn't take effect.

Copilot uses AI. Check for mistakes.
"provider_settings.sandbox.booter": {
"description": "沙箱环境驱动器",
"type": "string",
Expand Down
3 changes: 2 additions & 1 deletion astrbot/core/provider/sources/bailian_rerank_source.py
Original file line number Diff line number Diff line change
Expand Up @@ -142,7 +142,8 @@ def _parse_results(self, data: dict) -> list[RerankResult]:
f"百炼 API 错误: {data.get('code')} – {data.get('message', '')}"
)

results = data.get("output", {}).get("results", [])
# 兼容旧版 API (output.results) 和新版 compatible API (results)
Copy link

Copilot AI Apr 1, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Comment mixes Chinese/English in a confusing way ("新版 compatible API"). Consider rephrasing to something consistent like "兼容旧版 API (output.results) 和新版 API (results)" to improve readability.

Suggested change
# 兼容旧版 API (output.results) 和新版 compatible API (results)
# 兼容旧版 APIoutput.results和新版 APIresults

Copilot uses AI. Check for mistakes.
results = (data.get("output") or {}).get("results") or data.get("results") or []
if not results:
logger.warning(f"百炼 Rerank 返回空结果: {data}")
return []
Expand Down
Loading