src/cli_agent/tools/shell_ops.py - master - Watashicuvu/agentic-tools

src/cli_agent/tools/shell_ops.py
97 строк4 KB
Your Name
added come roles
28 май 2026, 15:16
28 май 2026, 15:16745363b
"""Операции с оболочкой: выполнение команд."""
import asyncio
import os
import shlex
from pathlib import Path
from typing import Dict, Any, Optional
from dataclasses import dataclass

@dataclass
class CommandResult:
    """Результат выполнения команды."""
    success: bool
    stdout: str
    stderr: str
    exit_code: int
    command: str
    timeout_reached: bool = False


async def run_shell_command(
    command: str,
    timeout: int = 120000,
    cwd: Optional[str] = None,
    is_background: bool = False
) -> Dict[str, Any]:
    """Выполнить команду в оболочке.
    Блокирует всё, кроме явно разрешённых в ALLOWED_SHELL_CMDS (через запятую).

    Args:
        command: Команда для выполнения
        timeout: Таймаут в миллисекундах (по умолчанию 120с)
        cwd: Рабочая директория
        is_background: Запустить в фоне
        
    Returns:
        Dict с результатом выполнения команды
    """
    # 1. Проверка allowlist
    allowed_str = os.environ.get("ALLOWED_SHELL_CMDS", "").strip()
    allowed_cmds = {c.strip() for c in allowed_str.split(",") if c.strip()} if allowed_str else set()
    
    if not allowed_cmds:
        return {
            "success": False, "command": command, "stdout": "",
            "stderr": "Shell execution is strictly disabled. Set ALLOWED_SHELL_CMDS to enable specific utilities.",
            "exit_code": -1, "timeout_reached": False
        }

    # Безопасное извлечение исполняемого файла (игнорирует флаги/пайпы)
    try:
        base_cmd = shlex.split(command.strip())[0]
    except ValueError:
        base_cmd = command.split()[0] if command.strip() else ""

    if base_cmd not in allowed_cmds:
        return {
            "success": False, "command": command, "stdout": "",
            "stderr": f"Command '{base_cmd}' is blocked. Allowed: {', '.join(sorted(allowed_cmds))}",
            "exit_code": -1, "timeout_reached": False
        }

    # 2. Исполнение
    try:
        process = await asyncio.create_subprocess_shell(
            command,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            cwd=Path(cwd).resolve() if cwd else None
        )

        if is_background:
            return {
                "success": True, "command": command, "pid": process.pid,
                "message": f"Command started in background (PID: {process.pid})", "is_background": True
            }

        timeout_sec = timeout / 1000.0
        try:
            stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=timeout_sec)
            return {
                "success": process.returncode == 0, "command": command,
                "stdout": stdout.decode('utf-8', errors='replace'),
                "stderr": stderr.decode('utf-8', errors='replace'),
                "exit_code": process.returncode, "timeout_reached": False
            }
        except asyncio.TimeoutError:
            try: process.kill()
            except ProcessLookupError: pass
            return {
                "success": False, "command": command, "stdout": "",
                "stderr": f"Command timed out after {timeout_sec}s", "exit_code": -1, "timeout_reached": True
            }
    except Exception as e:
        return {
            "success": False, "command": command, "stdout": "",
            "stderr": f"Execution error: {str(e)}", "exit_code": -1, "timeout_reached": False
        }