core/engine/src/tools/code_execution.py - main - alexefan136/flowstack

core/engine/src/tools/code_execution.py
1 340 строк48 KB
Alexander Efanov
Обновление репозитория
15 июл 2026, 12:19
15 июл 2026, 12:1976704c6
"""
Code Execution Tool — безопасное выполнение кода в изолированном окружении.

Реализует MCP Tool specification 2024-11-05:
https://modelcontextprotocol.io/specification/2024-11-05/server/tools

Поддерживаемые языки:
- Python (полная поддержка, sandboxed)
- JavaScript/Node.js (если установлен node)
- Shell (требует явного подтверждения, очень опасно)

Архитектура безопасности (defense in depth):
1. AST pre-validation — статический анализ Python кода на запрещённые операции
2. Process isolation — выполнение в отдельном процессе через subprocess
3. Resource limits — CPU time и memory через resource module (Unix)
4. Temp directory isolation — код работает только в временной директории
5. Timeout — принудительное завершение при превышении времени
6. Output capture — перехват stdout/stderr
7. Whitelist imports — разрешены только безопасные модули

НЕ разрешено (для Python):
- import os, subprocess, sys, shutil, socket, http, urllib, и т.д.
- __import__(), exec(), eval() внутри кода
- Доступ к __builtins__, __class__, __mro__
- Сетевые соединения (по умолчанию)
- Чтение/запись файлов вне temp directory

Примеры:
    execute(code="print(2 + 2)", language="python") → "4"
    execute(code="[x**2 for x in range(5)]", language="python") → "[0, 1, 4, 9, 16]"
"""
from __future__ import annotations

import ast
import asyncio
import json
import logging
import os
import shutil
import sys
import tempfile
import uuid
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, ClassVar

from src.tools.base import TextContent, Tool, ToolResult


logger = logging.getLogger(__name__)


# ============================================================================
# Exceptions
# ============================================================================


class CodeExecutionError(Exception):
    """Базовое исключение для code execution."""


class CodeValidationError(CodeExecutionError):
    """Код не прошёл валидацию (статический анализ)."""

    def __init__(self, message: str, violations: list[str] | None = None):
        super().__init__(message)
        self.violations = violations or []


class ExecutionTimeoutError(CodeExecutionError):
    """Превышен таймаут выполнения."""


class MemoryLimitError(CodeExecutionError):
    """Превышен лимит памяти."""


class ForbiddenOperationError(CodeExecutionError):
    """Попытка выполнить запрещённую операцию."""


class UnsupportedLanguageError(CodeExecutionError):
    """Неподдерживаемый язык программирования."""


# ============================================================================
# Configuration
# ============================================================================


@dataclass
class CodeExecutionConfig:
    """
    Конфигурация для code execution tool.

    Все лимиты — defense in depth против DoS и sandbox escape.
    """

    # Языки
    enabled_languages: list[str] = field(
        default_factory=lambda: ["python", "javascript"]
    )
    shell_enabled: bool = False  # Shell очень опасен — по умолчанию выключен

    # Лимиты выполнения
    timeout_seconds: float = 10.0
    max_memory_mb: int = 256  # 256 MB
    max_output_length: int = 100_000  # 100 KB
    max_code_length: int = 100_000  # 100 KB

    # Working directory
    working_dir: str | None = None  # None = использовать tempdir

    # Python-specific
    python_executable: str = sys.executable  # тот же Python, что и у нас
    python_allowed_modules: list[str] = field(
        default_factory=lambda: [
            # Safe standard library modules
            "math",
            "statistics",
            "random",
            "decimal",
            "fractions",
            "itertools",
            "functools",
            "collections",
            "copy",
            "string",
            "re",
            "json",
            "datetime",
            "time",
            "calendar",
            "textwrap",
            "unicodedata",
            "base64",
            "hashlib",
            "hmac",
            "secrets",
            "typing",
            "dataclasses",
            "enum",
            "abc",
            "numbers",
            "operator",
            "pprint",
        ]
    )
    python_forbidden_modules: list[str] = field(
        default_factory=lambda: [
            # Dangerous modules
            "os",
            "sys",
            "subprocess",
            "shutil",
            "socket",
            "http",
            "urllib",
            "requests",
            "ftplib",
            "smtplib",
            "telnetlib",
            "xmlrpc",
            "webbrowser",
            "ctypes",
            "multiprocessing",
            "threading",
            "signal",
            "importlib",
            "pickle",
            "shelve",
            "marshal",
            "code",
            "codeop",
            "compile",
            "compileall",
            "py_compile",
            "zipimport",
            "pkgutil",
            "pdb",
            "profile",
            "cProfile",
            "timeit",
            "trace",
            "tracemalloc",
            "gc",
            "weakref",
            "io",  # Может использоваться для filesystem access
            "pathlib",
            "glob",
            "fnmatch",
            "sqlite3",
            "dbm",
        ]
    )
    python_forbidden_builtins: list[str] = field(
        default_factory=lambda: [
            "__import__",
            "exec",
            "eval",
            "compile",
            "globals",
            "locals",
            "vars",
            "dir",
            "getattr",
            "setattr",
            "delattr",
            "open",  # Filesystem access
            "input",
            "breakpoint",
            "exit",
            "quit",
        ]
    )

    # Cleanup
    auto_cleanup: bool = True  # Удалять temp dir после выполнения

    def validate(self) -> list[str]:
        """Валидировать конфигурацию. Чистая функция."""
        errors: list[str] = []

        if self.timeout_seconds <= 0:
            errors.append("timeout_seconds must be positive")

        if self.max_memory_mb <= 0:
            errors.append("max_memory_mb must be positive")

        if self.max_output_length <= 0:
            errors.append("max_output_length must be positive")

        if self.max_code_length <= 0:
            errors.append("max_code_length must be positive")

        return errors


# ============================================================================
# Response Models
# ============================================================================


@dataclass
class ExecutionResult:
    """Результат выполнения кода."""

    language: str
    stdout: str = ""
    stderr: str = ""
    return_code: int = 0
    execution_time_ms: float = 0.0
    memory_used_bytes: int = 0
    timed_out: bool = False
    memory_limited: bool = False

    # Файлы, созданные в working directory
    created_files: list[str] = field(default_factory=list)

    def is_success(self) -> bool:
        """Проверить успешность выполнения. Чистая функция."""
        return self.return_code == 0 and not self.timed_out and not self.memory_limited

    def to_dict(self) -> dict[str, Any]:
        """Чистая функция."""
        return {
            "language": self.language,
            "stdout": self.stdout,
            "stderr": self.stderr,
            "return_code": self.return_code,
            "execution_time_ms": self.execution_time_ms,
            "memory_used_bytes": self.memory_used_bytes,
            "timed_out": self.timed_out,
            "memory_limited": self.memory_limited,
            "created_files": self.created_files,
            "success": self.is_success(),
        }


# ============================================================================
# Python AST Validator
# ============================================================================


class PythonCodeValidator:
    """
    Статический анализ Python кода на запрещённые операции.

    Использует ast для парсинга и проверки:
    - Import statements (разрешены только из whitelist)
    - Запрещённые builtin функции (__import__, exec, eval, open, etc.)
    - Запрещённый attribute access (__builtins__, __class__, __mro__, etc.)
    - Dunder methods, которые могут использоваться для escape

    Это defense in depth — в runtime sandbox всё равно изолирует процесс.
    """

    # Запрещённые dunder attributes (могут использоваться для sandbox escape)
    FORBIDDEN_DUNDERS: ClassVar[set[str]] = {
        "__builtins__",
        "__class__",
        "__bases__",
        "__mro__",
        "__subclasses__",
        "__globals__",
        "__code__",
        "__closure__",
        "__func__",
        "__self__",
        "__module__",
        "__dict__",
        "__init_subclass__",
        "__setattr__",
        "__getattr__",
        "__delattr__",
        "__import__",
    }

    def __init__(self, config: CodeExecutionConfig):
        self.config = config

    def validate(self, code: str) -> list[str]:
        """
        Валидировать Python код.

        Returns:
            Список нарушений (пустой если код безопасен).
        """
        violations: list[str] = []

        # Парсинг
        try:
            tree = ast.parse(code)
        except SyntaxError as e:
            return [f"Syntax error: {e.msg} at line {e.lineno}"]

        # Обход AST
        for node in ast.walk(tree):
            violation = self._check_node(node)
            if violation:
                violations.append(violation)

        return violations

    def _check_node(self, node: ast.AST) -> str | None:
        """Проверить один AST узел. Чистая функция."""
        # Import statement: import X, import X.Y
        if isinstance(node, ast.Import):
            for alias in node.names:
                module_name = alias.name.split(".")[0]
                violation = self._check_module(module_name)
                if violation:
                    return violation

        # From import: from X import Y
        elif isinstance(node, ast.ImportFrom):
            if node.module:
                module_name = node.module.split(".")[0]
                violation = self._check_module(module_name)
                if violation:
                    return violation

        # Вызов функции: foo(), __import__('os'), exec('...')
        elif isinstance(node, ast.Call):
            if isinstance(node.func, ast.Name):
                func_name = node.func.id
                if func_name in self.config.python_forbidden_builtins:
                    return f"Forbidden builtin call: {func_name}()"

        # Attribute access: obj.__class__, obj.__builtins__
        elif isinstance(node, ast.Attribute):
            if node.attr in self.FORBIDDEN_DUNDERS:
                return f"Forbidden attribute access: .{node.attr}"

        return None

    def _check_module(self, module_name: str) -> str | None:
        """Проверить, разрешён ли модуль. Чистая функция."""
        # Явно запрещённые модули (приоритет над whitelist)
        if module_name in self.config.python_forbidden_modules:
            return f"Forbidden module: {module_name}"

        # Если whitelist не пустой — разрешены только из него
        if self.config.python_allowed_modules:
            if module_name not in self.config.python_allowed_modules:
                return (
                    f"Module not in whitelist: {module_name}. "
                    f"Allowed: {', '.join(sorted(self.config.python_allowed_modules))}"
                )

        return None


# ============================================================================
# Python Sandbox
# ============================================================================


class PythonSandbox:
    """
    Sandbox для выполнения Python кода.

    Использует subprocess + wrapper script для изоляции:
    1. Создаёт temp directory
    2. Записывает wrapper script, который:
       - Устанавливает resource limits (CPU, memory)
       - Ограничивает builtins
       - Перехватывает stdout/stderr
       - Выполняет user code
       - Выводит результат как JSON
    3. Запускает subprocess с timeout
    4. Читает результат
    5. Очищает temp directory
    """

    def __init__(self, config: CodeExecutionConfig):
        self.config = config
        self.validator = PythonCodeValidator(config)

    async def execute(self, code: str) -> ExecutionResult:
        """
        Выполнить Python код в sandbox.

        Returns:
            ExecutionResult с stdout/stderr/return_code.
        """
        # Проверка длины
        if len(code) > self.config.max_code_length:
            raise CodeExecutionError(
                f"Code too long: {len(code)} > {self.config.max_code_length}"
            )

        # Статический анализ
        violations = self.validator.validate(code)
        if violations:
            raise CodeValidationError(
                f"Code validation failed: {len(violations)} violation(s)",
                violations=violations,
            )

        # Создаём temp directory
        work_dir = self._create_work_dir()
        start_time = asyncio.get_event_loop().time()

        try:
            # Записываем wrapper и user code
            user_code_path = work_dir / "user_code.py"
            wrapper_path = work_dir / "wrapper.py"

            user_code_path.write_text(code, encoding="utf-8")
            wrapper_path.write_text(
                self._generate_wrapper_script(user_code_path),
                encoding="utf-8",
            )

            # Запускаем subprocess
            result = await self._run_subprocess(wrapper_path, work_dir)

            execution_time_ms = (
                asyncio.get_event_loop().time() - start_time
            ) * 1000
            result.execution_time_ms = execution_time_ms

            # Обрезаем вывод если слишком длинный
            if len(result.stdout) > self.config.max_output_length:
                result.stdout = (
                    result.stdout[: self.config.max_output_length]
                    + f"\n... [truncated, {len(result.stdout)} chars total]"
                )
            if len(result.stderr) > self.config.max_output_length:
                result.stderr = (
                    result.stderr[: self.config.max_output_length]
                    + f"\n... [truncated, {len(result.stderr)} chars total]"
                )

            # Список созданных файлов
            result.created_files = self._list_created_files(work_dir)

            return result

        finally:
            if self.config.auto_cleanup:
                self._cleanup_work_dir(work_dir)

    def _create_work_dir(self) -> Path:
        """Создать временную рабочую директорию."""
        if self.config.working_dir:
            base = Path(self.config.working_dir)
            base.mkdir(parents=True, exist_ok=True)
            work_dir = base / f"code_{uuid.uuid4().hex[:12]}"
        else:
            work_dir = Path(tempfile.mkdtemp(prefix="code_exec_"))
        work_dir.mkdir(parents=True, exist_ok=True)
        return work_dir

    def _cleanup_work_dir(self, work_dir: Path) -> None:
        """Удалить рабочую директорию."""
        try:
            shutil.rmtree(work_dir, ignore_errors=True)
        except Exception as e:
            logger.warning(f"Failed to cleanup {work_dir}: {e}")

    def _list_created_files(self, work_dir: Path) -> list[str]:
        """Список файлов, созданных кодом (исключая wrapper и user_code)."""
        excluded = {"wrapper.py", "user_code.py", "result.json"}
        files: list[str] = []
        for path in work_dir.rglob("*"):
            if path.is_file() and path.name not in excluded:
                try:
                    files.append(str(path.relative_to(work_dir)))
                except ValueError:
                    files.append(str(path))
        return files

    def _generate_wrapper_script(self, user_code_path: Path) -> str:
        """
        Генерация wrapper script.

        Wrapper:
        1. Устанавливает resource limits (Unix)
        2. Restricts builtins
        3. Перехватывает stdout/stderr
        4. Выполняет user code
        5. Выводит результат как JSON
        """
        max_memory_bytes = self.config.max_memory_mb * 1024 * 1024
        allowed_modules_json = json.dumps(self.config.python_allowed_modules)
        forbidden_builtins_json = json.dumps(self.config.python_forbidden_builtins)

        return f'''#!/usr/bin/env python3
"""Wrapper script для изолированного выполнения Python кода."""
import sys
import json
import io
import traceback

# === Resource limits (Unix only) ===
try:
    import resource
    # CPU time limit
    resource.setrlimit(resource.RLIMIT_CPU, ({int(self.config.timeout_seconds)}, {int(self.config.timeout_seconds)}))
    # Address space limit (memory)
    resource.setrlimit(resource.RLIMIT_AS, ({max_memory_bytes}, {max_memory_bytes}))
    # Disable core dumps
    try:
        resource.setrlimit(resource.RLIMIT_CORE, (0, 0))
    except Exception:
        pass
except ImportError:
    pass  # Windows или другая OS без resource
except Exception as e:
    print(json.dumps({{"wrapper_error": f"Failed to set resource limits: {{e}}"}}), file=sys.__stderr__)

# === Restricted builtins ===
FORBIDDEN = set({forbidden_builtins_json})
original_builtins = {{name: getattr(__builtins__, name) if hasattr(__builtins__, name) else __builtins__[name]
                    for name in dir(__builtins__) if isinstance(__builtins__, dict) or hasattr(__builtins__, name)}}

if isinstance(__builtins__, dict):
    restricted_builtins = {{k: v for k, v in __builtins__.items() if k not in FORBIDDEN}}
else:
    restricted_builtins = {{k: getattr(__builtins__, k) for k in dir(__builtins__)
                          if not k.startswith("_") and k not in FORBIDDEN}}

# === Capture stdout/stderr ===
captured_stdout = io.StringIO()
captured_stderr = io.StringIO()
sys.stdout = captured_stdout
sys.stderr = captured_stderr

# === Execute user code ===
result = {{"success": False, "output": "", "error": None, "error_type": None}}

try:
    with open({str(user_code_path)!r}, "r", encoding="utf-8") as f:
        user_code = f.read()

    # Ограниченный globals dict
    restricted_globals = {{"__builtins__": restricted_builtins, "__name__": "__main__"}}

    compiled = compile(user_code, {str(user_code_path)!r}, "exec")
    exec(compiled, restricted_globals)

    result["success"] = True
    result["output"] = captured_stdout.getvalue()

except MemoryError:
    result["error"] = "Memory limit exceeded"
    result["error_type"] = "memory_limit"
    result["output"] = captured_stdout.getvalue()
except Exception as e:
    result["error"] = str(e)
    result["error_type"] = type(e).__name__
    result["output"] = captured_stdout.getvalue()
    result["traceback"] = traceback.format_exc()

# === Output result as JSON to real stdout ===
sys.stdout = sys.__stdout__
sys.stderr = sys.__stderr__

# Выводим результат
print(json.dumps(result, ensure_ascii=False, default=str))
'''

    async def _run_subprocess(
        self, wrapper_path: Path, work_dir: Path
    ) -> ExecutionResult:
        """Запустить wrapper script в subprocess."""
        try:
            proc = await asyncio.create_subprocess_exec(
                self.config.python_executable,
                str(wrapper_path),
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
                cwd=str(work_dir),
                env=self._get_safe_env(),
            )

            try:
                stdout_bytes, stderr_bytes = await asyncio.wait_for(
                    proc.communicate(),
                    timeout=self.config.timeout_seconds,
                )
            except asyncio.TimeoutError:
                # Убиваем процесс
                try:
                    proc.kill()
                    await proc.wait()
                except Exception:
                    pass
                return ExecutionResult(
                    language="python",
                    stdout="",
                    stderr=f"Execution timed out after {self.config.timeout_seconds}s",
                    return_code=-1,
                    timed_out=True,
                )

            stdout = stdout_bytes.decode("utf-8", errors="replace")
            stderr = stderr_bytes.decode("utf-8", errors="replace")
            return_code = proc.returncode if proc.returncode is not None else -1

            # Парсим JSON результат от wrapper
            return self._parse_wrapper_output(stdout, stderr, return_code)

        except FileNotFoundError:
            raise CodeExecutionError(
                f"Python executable not found: {self.config.python_executable}"
            )
        except Exception as e:
            raise CodeExecutionError(f"Failed to run subprocess: {e}") from e

    def _parse_wrapper_output(
        self, stdout: str, stderr: str, return_code: int
    ) -> ExecutionResult:
        """Парсить JSON результат от wrapper script."""
        # Wrapper выводит JSON в stdout
        # Ищем последнюю строку, которая является JSON
        lines = stdout.strip().split("\n")
        result_json: dict[str, Any] | None = None

        # Идём с конца, ищем JSON
        for line in reversed(lines):
            line = line.strip()
            if not line:
                continue
            try:
                parsed = json.loads(line)
                if isinstance(parsed, dict):
                    result_json = parsed
                    break
            except (json.JSONDecodeError, ValueError):
                continue

        if result_json is None:
            # Wrapper не выдал JSON — значит что-то пошло не так
            return ExecutionResult(
                language="python",
                stdout=stdout,
                stderr=stderr or "Wrapper failed to produce JSON output",
                return_code=return_code,
            )

        # Проверяем wrapper error
        if "wrapper_error" in result_json:
            return ExecutionResult(
                language="python",
                stdout="",
                stderr=f"Wrapper error: {result_json['wrapper_error']}",
                return_code=-1,
            )

        # Извлекаем результат
        user_stdout = str(result_json.get("output", ""))
        error = result_json.get("error")
        error_type = result_json.get("error_type")

        if error:
            error_msg = f"{error_type or 'Error'}: {error}"
            if result_json.get("traceback"):
                error_msg += f"\n\n{result_json['traceback']}"
            final_stderr = error_msg
            final_return_code = 1
        else:
            final_stderr = stderr
            final_return_code = 0

        # Определяем memory limit
        memory_limited = error_type == "memory_limit" or error_type == "MemoryError"

        return ExecutionResult(
            language="python",
            stdout=user_stdout,
            stderr=final_stderr,
            return_code=final_return_code,
            memory_limited=memory_limited,
        )

    def _get_safe_env(self) -> dict[str, str]:
        """Получить безопасное окружение для subprocess."""
        env = os.environ.copy()

        # Удаляем потенциально опасные переменные
        dangerous_vars = [
            "PYTHONSTARTUP",
            "PYTHONPATH",  # Может содержать пути к модулям
            "PYTHONHOME",
            "PYTHONDONTWRITEBYTECODE",
        ]
        for var in dangerous_vars:
            env.pop(var, None)

        # Отключаем .pyc файлы
        env["PYTHONDONTWRITEBYTECODE"] = "1"

        # Устанавливаем PYTHONNOUSERSITE для изоляции от user site-packages
        env["PYTHONNOUSERSITE"] = "1"

        return env


# ============================================================================
# Node.js Sandbox (упрощённая версия)
# ============================================================================


class NodeSandbox:
    """
    Sandbox для выполнения JavaScript кода через Node.js.

    Менее строгий чем Python sandbox (Node.js изоляция сложнее),
    но всё равно использует:
    - Subprocess с timeout
    - Temp directory
    - Output capture
    """

    def __init__(self, config: CodeExecutionConfig):
        self.config = config

    async def execute(self, code: str) -> ExecutionResult:
        """Выполнить JavaScript код."""
        if len(code) > self.config.max_code_length:
            raise CodeExecutionError(
                f"Code too long: {len(code)} > {self.config.max_code_length}"
            )

        # Проверка наличия node
        node_path = shutil.which("node")
        if not node_path:
            raise UnsupportedLanguageError(
                "Node.js is not installed. Install from https://nodejs.org/"
            )

        # Создаём temp directory
        work_dir = Path(tempfile.mkdtemp(prefix="code_exec_js_"))
        start_time = asyncio.get_event_loop().time()

        try:
            # Записываем код
            code_path = work_dir / "code.js"
            code_path.write_text(code, encoding="utf-8")

            # Запускаем node
            try:
                proc = await asyncio.create_subprocess_exec(
                    node_path,
                    str(code_path),
                    stdout=asyncio.subprocess.PIPE,
                    stderr=asyncio.subprocess.PIPE,
                    cwd=str(work_dir),
                )

                try:
                    stdout_bytes, stderr_bytes = await asyncio.wait_for(
                        proc.communicate(),
                        timeout=self.config.timeout_seconds,
                    )
                except asyncio.TimeoutError:
                    try:
                        proc.kill()
                        await proc.wait()
                    except Exception:
                        pass
                    return ExecutionResult(
                        language="javascript",
                        stdout="",
                        stderr=f"Execution timed out after {self.config.timeout_seconds}s",
                        return_code=-1,
                        timed_out=True,
                    )

                stdout = stdout_bytes.decode("utf-8", errors="replace")
                stderr = stderr_bytes.decode("utf-8", errors="replace")
                return_code = proc.returncode if proc.returncode is not None else -1

                # Обрезаем вывод
                if len(stdout) > self.config.max_output_length:
                    stdout = (
                        stdout[: self.config.max_output_length]
                        + f"\n... [truncated, {len(stdout)} chars total]"
                    )
                if len(stderr) > self.config.max_output_length:
                    stderr = (
                        stderr[: self.config.max_output_length]
                        + f"\n... [truncated, {len(stderr)} chars total]"
                    )

                execution_time_ms = (
                    asyncio.get_event_loop().time() - start_time
                ) * 1000

                return ExecutionResult(
                    language="javascript",
                    stdout=stdout,
                    stderr=stderr,
                    return_code=return_code,
                    execution_time_ms=execution_time_ms,
                )

            except FileNotFoundError:
                raise UnsupportedLanguageError(
                    "Node.js executable not found"
                )

        finally:
            if self.config.auto_cleanup:
                try:
                    shutil.rmtree(work_dir, ignore_errors=True)
                except Exception as e:
                    logger.warning(f"Failed to cleanup {work_dir}: {e}")


# ============================================================================
# Shell Sandbox (ОЧЕНЬ ОПАСНО — только с explicit confirmation)
# ============================================================================


class ShellSandbox:
    """
    Sandbox для shell команд.

    ⚠️ ВНИМАНИЕ: Shell execution крайне опасен. Даже с timeout и tempdir,
    злонамеренная команда может нанести вред системе.

    Используйте ТОЛЬКО если:
    - Пользователь явно подтвердил выполнение
    - Команды проверены
    - Tool используется в контролируемой среде
    """

    def __init__(self, config: CodeExecutionConfig):
        self.config = config

    async def execute(self, command: str) -> ExecutionResult:
        """Выполнить shell команду."""
        if not self.config.shell_enabled:
            raise ForbiddenOperationError(
                "Shell execution is disabled. Set shell_enabled=True in config to enable."
            )

        if len(command) > self.config.max_code_length:
            raise CodeExecutionError(
                f"Command too long: {len(command)} > {self.config.max_code_length}"
            )

        # Проверка опасных паттернов (очень базовая)
        dangerous_patterns = [
            "rm -rf /",
            "rm -rf /*",
            "mkfs",
            "dd if=",
            ":(){:|:&};:",  # fork bomb
            "> /dev/sda",
            "chmod -R 777 /",
            "wget",  # download and execute
            "curl",
            "nc -",  # netcat reverse shell
        ]
        for pattern in dangerous_patterns:
            if pattern in command:
                raise ForbiddenOperationError(
                    f"Dangerous pattern detected: {pattern!r}"
                )

        work_dir = Path(tempfile.mkdtemp(prefix="code_exec_sh_"))
        start_time = asyncio.get_event_loop().time()

        try:
            # Используем /bin/sh для portability
            shell = shutil.which("sh") or "/bin/sh"

            proc = await asyncio.create_subprocess_exec(
                shell,
                "-c",
                command,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
                cwd=str(work_dir),
            )

            try:
                stdout_bytes, stderr_bytes = await asyncio.wait_for(
                    proc.communicate(),
                    timeout=self.config.timeout_seconds,
                )
            except asyncio.TimeoutError:
                try:
                    proc.kill()
                    await proc.wait()
                except Exception:
                    pass
                return ExecutionResult(
                    language="shell",
                    stdout="",
                    stderr=f"Command timed out after {self.config.timeout_seconds}s",
                    return_code=-1,
                    timed_out=True,
                )

            stdout = stdout_bytes.decode("utf-8", errors="replace")
            stderr = stderr_bytes.decode("utf-8", errors="replace")
            return_code = proc.returncode if proc.returncode is not None else -1

            # Обрезаем вывод
            if len(stdout) > self.config.max_output_length:
                stdout = (
                    stdout[: self.config.max_output_length]
                    + f"\n... [truncated, {len(stdout)} chars total]"
                )
            if len(stderr) > self.config.max_output_length:
                stderr = (
                    stderr[: self.config.max_output_length]
                    + f"\n... [truncated, {len(stderr)} chars total]"
                )

            execution_time_ms = (
                asyncio.get_event_loop().time() - start_time
            ) * 1000

            # Список созданных файлов
            created_files: list[str] = []
            for path in work_dir.rglob("*"):
                if path.is_file():
                    try:
                        created_files.append(str(path.relative_to(work_dir)))
                    except ValueError:
                        created_files.append(str(path))

            return ExecutionResult(
                language="shell",
                stdout=stdout,
                stderr=stderr,
                return_code=return_code,
                execution_time_ms=execution_time_ms,
                created_files=created_files,
            )

        finally:
            if self.config.auto_cleanup:
                try:
                    shutil.rmtree(work_dir, ignore_errors=True)
                except Exception as e:
                    logger.warning(f"Failed to cleanup {work_dir}: {e}")


# ============================================================================
# Code Execution Tool (MCP)
# ============================================================================


class CodeExecutionTool(Tool):
    """
    MCP Tool для безопасного выполнения кода.

    Поддерживаемые языки:
    - python: полная sandbox с AST validation и resource limits
    - javascript: sandbox через Node.js subprocess
    - shell: требует shell_enabled=True в конфиге

    MCP Tool:
    - input_schema определяет параметры для clients
    - Результат возвращается как TextContent + metadata
    - is_error сигнализирует об ошибках выполнения
    - requires_confirmation=True для shell (по умолчанию)
    """

    name: ClassVar[str] = "code_execution"
    description: ClassVar[str] = (
        "Выполняет код в изолированном окружении (sandbox). "
        "Поддерживает: python (рекомендуется), javascript, shell. "
        "Код выполняется с ограничениями CPU/памяти/времени. "
        "Пример Python: 'def factorial(n): return 1 if n <= 1 else n * factorial(n-1)\\nprint(factorial(10))'"
    )

    input_schema: ClassVar[dict[str, Any] | None] = {
        "type": "object",
        "properties": {
            "code": {
                "type": "string",
                "description": "Код для выполнения",
            },
            "language": {
                "type": "string",
                "enum": ["python", "javascript", "shell"],
                "default": "python",
                "description": "Язык программирования",
            },
        },
        "required": ["code"],
        "additionalProperties": False,
    }

    parameters_schema: ClassVar[dict[str, Any] | None] = input_schema

    examples: ClassVar[list[dict[str, Any]]] = [
        {
            "code": "print(sum(range(10)))",
            "language": "python",
            "result": "45",
        },
        {
            "code": "def fib(n):\n    a, b = 0, 1\n    for _ in range(n):\n        a, b = b, a + b\n    return a\nprint(fib(10))",
            "language": "python",
            "result": "55",
        },
        {
            "code": "console.log([1,2,3].map(x => x**2))",
            "language": "javascript",
            "result": "[ 1, 4, 9 ]",
        },
    ]

    tags: ClassVar[list[str]] = ["code", "execution", "sandbox", "utility"]
    is_read_only: ClassVar[bool] = False  # Код может создавать файлы
    requires_confirmation: ClassVar[bool] = False

    def __init__(self, config: CodeExecutionConfig | None = None):
        self.config = config or CodeExecutionConfig()
        self._sandboxes: dict[str, Any] = {}

    def _get_sandbox(self, language: str) -> Any:
        """Получить или создать sandbox для языка (ленивая инициализация)."""
        if language not in self._sandboxes:
            if language == "python":
                self._sandboxes[language] = PythonSandbox(self.config)
            elif language == "javascript":
                self._sandboxes[language] = NodeSandbox(self.config)
            elif language == "shell":
                self._sandboxes[language] = ShellSandbox(self.config)
            else:
                raise UnsupportedLanguageError(
                    f"Unsupported language: {language}. "
                    f"Supported: python, javascript, shell"
                )
        return self._sandboxes[language]

    async def execute(self, **kwargs: Any) -> ToolResult:
        """
        Выполнить код в sandbox.

        Args:
            code: Строка с кодом
            language: Язык программирования (python/javascript/shell)

        Returns:
            ToolResult с выводом программы или ошибкой.
        """
        code = kwargs.get("code", "")
        language = str(kwargs.get("language", "python")).lower()

        # Валидация
        if not isinstance(code, str):
            return ToolResult.failure(
                f"Parameter 'code' must be a string, got {type(code).__name__}"
            )

        if not code.strip():
            return ToolResult.failure("Code is empty")

        if language not in self.config.enabled_languages:
            if language == "shell" and not self.config.shell_enabled:
                return ToolResult.failure(
                    "Shell execution is disabled for security reasons. "
                    "Enable shell_enabled=True in config if you trust the input."
                )
            return ToolResult.failure(
                f"Language '{language}' is not enabled. "
                f"Enabled: {', '.join(self.config.enabled_languages)}"
            )

        try:
            sandbox = self._get_sandbox(language)
            result: ExecutionResult = await sandbox.execute(code)

            # Формируем текстовый вывод
            text_parts: list[str] = []

            if result.stdout:
                text_parts.append("=== Output ===")
                text_parts.append(result.stdout)

            if result.stderr:
                text_parts.append("=== Errors ===")
                text_parts.append(result.stderr)

            if not result.stdout and not result.stderr:
                if result.is_success():
                    text_parts.append("(no output)")
                else:
                    text_parts.append(f"Process exited with code {result.return_code}")

            # Добавляем metadata в конце
            text_parts.append("")
            text_parts.append(
                f"--- Execution: {result.execution_time_ms:.1f}ms, "
                f"exit code: {result.return_code}"
            )
            if result.created_files:
                text_parts.append(
                    f"Created files: {', '.join(result.created_files)}"
                )

            text = "\n".join(text_parts)

            if result.is_success():
                return ToolResult.success_result(
                    [TextContent(text=text)],
                    metadata=result.to_dict(),
                )
            else:
                return ToolResult.failure(
                    error=result.stderr or f"Exit code: {result.return_code}",
                    content=[TextContent(text=text)],
                    metadata=result.to_dict(),
                )

        except CodeValidationError as e:
            violations_text = "\n".join(f"  - {v}" for v in e.violations)
            error_msg = f"Code validation failed:\n{violations_text}"
            return ToolResult.failure(
                error_msg,
                metadata={
                    "language": language,
                    "violations": e.violations,
                },
            )
        except ExecutionTimeoutError as e:
            return ToolResult.failure(
                f"Execution timed out: {e}",
                metadata={"language": language, "timed_out": True},
            )
        except MemoryLimitError as e:
            return ToolResult.failure(
                f"Memory limit exceeded: {e}",
                metadata={"language": language, "memory_limited": True},
            )
        except ForbiddenOperationError as e:
            return ToolResult.failure(
                f"Forbidden operation: {e}",
                metadata={"language": language, "forbidden": True},
            )
        except UnsupportedLanguageError as e:
            return ToolResult.failure(
                str(e),
                metadata={"language": language},
            )
        except CodeExecutionError as e:
            return ToolResult.failure(
                f"Code execution error: {e}",
                metadata={"language": language},
            )
        except Exception as e:
            logger.exception(f"Unexpected error in code_execution tool: {e}")
            return ToolResult.failure(
                f"Unexpected error: {type(e).__name__}: {e}",
                metadata={"language": language},
            )


# ============================================================================
# Self-test
# ============================================================================


if __name__ == "__main__":
    import asyncio

    async def _self_test() -> None:
        """Проверка корректности code execution tool."""
        tool = CodeExecutionTool()

        print("🐍 Code Execution Tool self-test:\n")

        # === Python: успешные случаи ===
        print("=== Python: success cases ===")
        success_cases = [
            ("print(2 + 2)", "4"),
            ("print([x**2 for x in range(5)])", "[0, 1, 4, 9, 16]"),
            (
                "def fib(n):\n    a, b = 0, 1\n    for _ in range(n):\n        a, b = b, a+b\n    return a\nprint(fib(10))",
                "55",
            ),
            ("import math\nprint(math.pi)", "3.14159"),
            ("import json\nprint(json.dumps({'a': 1}))", '{"a": 1}'),
        ]

        passed = 0
        failed = 0

        for code, expected_output in success_cases:
            result = await tool.execute(code=code, language="python")
            actual_output = ""
            if result.metadata:
                actual_output = str(result.metadata.get("stdout", "")).strip()

            is_ok = expected_output in actual_output
            status = "✅" if is_ok else "❌"
            if is_ok:
                passed += 1
            else:
                failed += 1
            print(
                f"  {status} code={code[:50]!r} → expected {expected_output!r}, "
                f"got {actual_output[:80]!r}"
            )

        # === Python: ошибки безопасности ===
        print("\n=== Python: security violations ===")
        security_cases = [
            ("import os", "Forbidden module"),
            ("import subprocess", "Forbidden module"),
            ("os.system('ls')", "Forbidden module"),
            ("__import__('os')", "Forbidden builtin"),
            ("exec('print(1)')", "Forbidden builtin"),
            ("eval('1+1')", "Forbidden builtin"),
            ("open('/etc/passwd')", "Forbidden builtin"),
            ("().__class__.__bases__", "Forbidden attribute"),
        ]

        for code, expected_violation in security_cases:
            result = await tool.execute(code=code, language="python")
            is_ok = result.is_failure() and expected_violation.lower() in result.get_text().lower()
            status = "✅" if is_ok else "❌"
            if is_ok:
                passed += 1
            else:
                failed += 1
            error_text = result.get_text()[:80] if result.is_failure() else "NO ERROR!"
            print(f"  {status} {code!r:40} → {error_text}")

        # === Python: runtime ошибки ===
        print("\n=== Python: runtime errors ===")
        runtime_cases = [
            "1 / 0",
            "undefined_var",
            "raise ValueError('test')",
            "syntax error here",
        ]

        for code in runtime_cases:
            result = await tool.execute(code=code, language="python")
            is_ok = result.is_failure()
            status = "✅" if is_ok else "❌"
            if is_ok:
                passed += 1
            else:
                failed += 1
            error_text = result.get_text()[:80] if result.is_failure() else "NO ERROR!"
            print(f"  {status} {code!r:40} → {error_text}")

        # === Python: timeout ===
        print("\n=== Python: timeout test ===")
        result = await tool.execute(
            code="while True: pass",
            language="python",
        )
        is_timeout = (
            result.is_failure()
            and result.metadata
            and (result.metadata.get("timed_out") or "timed out" in result.get_text().lower())
        )
        status = "✅" if is_timeout else "❌"
        if is_timeout:
            passed += 1
        else:
            failed += 1
        print(f"  {status} infinite loop → timed out: {is_timeout}")

        # === JavaScript (если установлен Node) ===
        if "javascript" in tool.config.enabled_languages and shutil.which("node"):
            print("\n=== JavaScript: success cases ===")
            js_cases = [
                ("console.log(2 + 2)", "4"),
                ("console.log([1,2,3].map(x => x**2))", "1,4,9"),
            ]
            for code, expected in js_cases:
                result = await tool.execute(code=code, language="javascript")
                actual = ""
                if result.metadata:
                    actual = str(result.metadata.get("stdout", "")).strip()
                is_ok = expected.replace(",", "") in actual.replace(" ", "").replace(
                    ",", ""
                )
                status = "✅" if is_ok else "❌"
                if is_ok:
                    passed += 1
                else:
                    failed += 1
                print(
                    f"  {status} {code!r:50} → expected {expected!r}, got {actual[:60]!r}"
                )

        # === MCP Schema ===
        print("\n📝 MCP Tool Schema:")
        print(json.dumps(tool.to_mcp_tool(), indent=2, ensure_ascii=False))

        print(f"\n📊 Results: {passed} passed, {failed} failed")

    asyncio.run(_self_test())