142 lines
5.3 KiB
Python
142 lines
5.3 KiB
Python
"""Shell 执行: subprocess 跑命令,有黑名单拦明显危险操作。"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
import shlex
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from core import procs
|
|
|
|
from .base import Tool, compact_tool_output
|
|
|
|
|
|
class ShellTool(Tool):
|
|
name = "shell"
|
|
description = (
|
|
"Execute a shell command and return stdout/stderr/exit_code. "
|
|
"Default 60s timeout. Working directory is the agent's base dir. "
|
|
"For long-running commands (expected >~1 min), set background=true and "
|
|
"poll with check_process."
|
|
)
|
|
parameters = {
|
|
"type": "object",
|
|
"properties": {
|
|
"command": {"type": "string"},
|
|
"timeout": {"type": "integer", "default": 60, "description": "Seconds before kill. background=true 时含义变为最长运行时长,默认 7200s。"},
|
|
"background": {
|
|
"type": "boolean",
|
|
"description": (
|
|
"true = 后台执行:立即返回 proc_id,进程 detach 独立运行(不受本轮对话、"
|
|
"服务重启影响),用 check_process 查进度/结果。适用:预计运行超过约 1 分钟的"
|
|
"命令(编译/批处理/长计算)。快命令保持默认 false。"
|
|
),
|
|
},
|
|
},
|
|
"required": ["command"],
|
|
}
|
|
|
|
def __init__(
|
|
self,
|
|
base_dir: Optional[Path] = None,
|
|
user_root: Optional[Path] = None,
|
|
task_id: Optional[str] = None,
|
|
) -> None:
|
|
super().__init__(base_dir, user_root=user_root)
|
|
self.task_id = str(task_id) if task_id else "default"
|
|
|
|
BLOCKED_PATTERNS = (
|
|
"rm -rf /",
|
|
"rm -rf ~",
|
|
"rm -rf $HOME",
|
|
":(){ :|:& };:",
|
|
"mkfs",
|
|
"dd if=/dev/zero",
|
|
"> /dev/sda",
|
|
"format c:",
|
|
)
|
|
|
|
# Windows cmd 不识别 unix flag,常见踩坑命令直接在工具层兜底
|
|
_MKDIR_P_RE = re.compile(r"^\s*mkdir\s+-p\s+(.+?)\s*$")
|
|
|
|
def _windows_compat(self, command: str) -> tuple[str, str | None]:
|
|
"""Windows cmd 下把 unix 风格命令转译为可执行形式。
|
|
|
|
返回 (转译后命令, 转译说明 or None)。无需转译时第二项为 None。
|
|
"""
|
|
if sys.platform != "win32":
|
|
return command, None
|
|
m = self._MKDIR_P_RE.match(command)
|
|
if m:
|
|
paths = shlex.split(m.group(1), posix=False)
|
|
for p in paths:
|
|
p = p.strip('"').strip("'")
|
|
os.makedirs(p, exist_ok=True)
|
|
return (
|
|
"echo [shell-tool] mkdir -p handled in-process (Windows cmd doesn't support -p)",
|
|
f"intercepted `mkdir -p`: created {paths} via os.makedirs",
|
|
)
|
|
return command, None
|
|
|
|
def execute(self, command: str, timeout: int = 60, background: bool = False) -> str:
|
|
normalized = command.lower()
|
|
for pat in self.BLOCKED_PATTERNS:
|
|
if pat in normalized:
|
|
return f"[Error] blocked dangerous command pattern: {pat!r}"
|
|
|
|
command, note = self._windows_compat(command)
|
|
|
|
if background:
|
|
anchor = self.user_root or self.base_dir
|
|
if procs.count_running(anchor) >= procs.MAX_RUNNING_PER_USER:
|
|
return (
|
|
f"[Error] 已有 {procs.MAX_RUNNING_PER_USER} 个后台进程在跑(上限)。"
|
|
f"用 check_process 查看,等待完成或 kill 掉不需要的再启动。"
|
|
)
|
|
timeout_s = procs.clamp_timeout(timeout if timeout and timeout > 60 else None)
|
|
proc_id, _ = procs.launch_host(
|
|
anchor, self.task_id,
|
|
kind="shell",
|
|
command_display=command,
|
|
shell_cmd=command,
|
|
cwd=self.base_dir, timeout_s=timeout_s, env=None,
|
|
)
|
|
return (
|
|
f"[Background] 已启动后台进程 proc_id={proc_id},最长运行 {timeout_s}s。\n"
|
|
f"用 check_process(proc_id=\"{proc_id}\") 查进度和日志。进程独立于本轮对话运行,"
|
|
f"服务重启也不中断。现在可以继续其他工作;若无事可做,结束回合并告知用户稍后询问进度。"
|
|
)
|
|
|
|
try:
|
|
result = subprocess.run(
|
|
command,
|
|
shell=True,
|
|
cwd=str(self.base_dir),
|
|
capture_output=True,
|
|
timeout=timeout,
|
|
text=True,
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
)
|
|
except subprocess.TimeoutExpired:
|
|
return (
|
|
f"[Error] command timed out after {timeout}s. "
|
|
f"若命令本身需要长时间运行,用 background=true 重新发起(后台执行,check_process 查进度)。"
|
|
)
|
|
except FileNotFoundError as e:
|
|
return f"[Error] {e}"
|
|
|
|
if note:
|
|
result.stdout = (result.stdout or "") + f"\n[note] {note}"
|
|
|
|
parts = []
|
|
if result.stdout:
|
|
parts.append(f"[stdout]\n{result.stdout.rstrip()}")
|
|
if result.stderr:
|
|
parts.append(f"[stderr]\n{result.stderr.rstrip()}")
|
|
parts.append(f"[exit {result.returncode}]")
|
|
return compact_tool_output("\n".join(parts))
|