Agent代码执行沙箱与安全隔离方案

引言

Agent系统在执行任务时，经常需要运行用户提供的代码或自动生成的代码。无论是数据分析、算法验证还是自动化脚本执行，代码执行能力都是Agent不可或缺的核心功能。然而，执行不受信任的代码带来了巨大的安全风险：恶意代码可能访问敏感数据、破坏系统文件、发起网络攻击或消耗全部系统资源。因此，构建一个安全可靠的代码执行沙箱是Agent系统设计中最重要的工程挑战之一。

沙箱是一种隔离机制，它为代码执行提供一个受控的、受限的环境，使得代码在这个环境中运行时无法影响外部系统。一个完善的沙箱方案需要在多个层面实施隔离：进程隔离、文件系统隔离、网络隔离和资源限制。本文将深入探讨Docker沙箱、WASM沙箱两种主流方案，并详细讲解资源限制策略、安全审计机制，最后给出完整的代码实现。

一、代码执行环境隔离的层次

代码执行隔离可以分为多个层次，每个层次提供不同强度的安全保障。理解这些层次的特性和适用场景，是选择合适沙箱方案的基础。

第一层是语言层面的隔离。Python等解释型语言提供了一些隔离机制，如受限的执行环境RestrictedPython、沙箱模块sandbox等。这种方式的优点是轻量、快速，缺点是隔离强度有限，存在多种已知的逃逸路径。它适用于低风险场景，如执行可信度较高的内部代码。

第二层是进程级隔离。通过创建子进程来执行代码，利用操作系统的进程隔离机制来限制代码的影响范围。可以结合seccomp、AppArmor或SELinux等安全模块，进一步限制子进程的系统调用能力。这种方式比语言层面隔离更强，但仍然共享文件系统和网络。

第三层是容器级隔离。使用Docker等容器技术，为代码执行创建一个完整的隔离环境，包括独立的文件系统、网络栈和进程空间。容器级隔离提供了良好的安全性和合理的性能开销，是目前生产环境中最常用的沙箱方案。

第四层是虚拟机级隔离。使用虚拟机或微虚拟机（如Firecracker）提供硬件级别的隔离。这种方式安全性最高，因为虚拟机与宿主机之间有硬件级别的隔离边界。缺点是启动慢、资源开销大，适用于对安全性要求极高的场景。

第五层是WASM隔离。WebAssembly提供了一个轻量级的沙箱执行环境，代码在WASM虚拟机中运行，天然与宿主系统隔离。WASM沙箱启动极快、资源开销极小，是近年来备受关注的新兴方案。

二、Docker沙箱方案

Docker是最常用的容器化平台，它利用Linux内核的namespace和cgroup特性，为应用提供轻量级的隔离环境。使用Docker构建代码执行沙箱，可以实现文件系统隔离、网络隔离和资源限制。

Docker沙箱的基本工作流程是：首先准备一个包含执行环境的Docker镜像，然后为每次代码执行创建一个临时容器，将代码注入容器中执行，最后收集执行结果并销毁容器。这种方式确保了每次执行都在一个干净的环境中运行，不会受到之前执行的影响。

import docker
import tarfile
import io
import json
import time
import uuid
from typing import Optional, Dict, Any


class DockerSandbox:
    """基于Docker的代码执行沙箱"""

    def __init__(self, image: str = "python:3.11-slim", network: str = "none",
                 memory_limit: str = "256m", cpu_quota: int = 50000,
                 timeout: int = 30, work_dir: str = "/sandbox"):
        self.client = docker.from_env()
        self.image = image
        self.network = network
        self.memory_limit = memory_limit
        self.cpu_quota = cpu_quota
        self.timeout = timeout
        self.work_dir = work_dir

    def _create_container(self) -> docker.models.containers.Container:
        """创建隔离容器"""
        container = self.client.containers.create(
            self.image,
            command=["sleep", str(self.timeout + 10)],
            working_dir=self.work_dir,
            network=self.network,
            mem_limit=self.memory_limit,
            cpu_quota=self.cpu_quota,
            cpu_period=100000,
            read_only=False,
            tmpfs={"/tmp": "size=64m"},
            security_opt=["no-new-privileges"],
            cap_drop=["ALL"],
            labels={"sandbox": "true"},
        )
        return container

    def _inject_code(self, container, filename: str, code: str):
        """将代码注入容器"""
        tar_stream = io.BytesIO()
        with tarfile.open(fileobj=tar_stream, mode="w") as tar:
            data = code.encode("utf-8")
            info = tarfile.TarInfo(name=filename)
            info.size = len(data)
            tar.addfile(info, io.BytesIO(data))
        tar_stream.seek(0)
        container.put_archive(self.work_dir, tar_stream)

    def _extract_output(self, container, path: str) -> Optional[str]:
        """从容器中提取文件内容"""
        try:
            stream, _ = container.get_archive(path)
            tar_data = io.BytesIO()
            for chunk in stream:
                tar_data.write(chunk)
            tar_data.seek(0)
            with tarfile.open(fileobj=tar_data, mode="r") as tar:
                members = tar.getmembers()
                if members:
                    f = tar.extractfile(members[0])
                    return f.read().decode("utf-8") if f else None
        except Exception:
            return None

    def execute(self, code: str, language: str = "python") -> Dict[str, Any]:
        """在沙箱中执行代码"""
        container = None
        try:
            container = self._create_container()
            container.start()
            if language == "python":
                filename = "main.py"
                cmd = ["python", "-u", filename]
            elif language == "javascript":
                filename = "main.js"
                cmd = ["node", filename]
            else:
                return {"success": False, "error": f"不支持的语言: {language}"}

            self._inject_code(container, filename, code)
            result = container.exec_run(cmd, workdir=self.work_dir, demux=True)
            stdout = result.output[0].decode("utf-8") if result.output[0] else ""
            stderr = result.output[1].decode("utf-8") if result.output[1] else ""
            return {
                "success": result.exit_code == 0,
                "exit_code": result.exit_code,
                "stdout": stdout,
                "stderr": stderr,
            }
        except Exception as e:
            return {"success": False, "error": f"沙箱执行异常: {str(e)}"}
        finally:
            if container:
                try:
                    container.kill()
                    container.remove(force=True)
                except Exception:
                    pass

    def execute_with_files(self, code: str, input_files: Dict[str, str],
                           output_files: List[str]) -> Dict[str, Any]:
        """支持输入输出文件的沙箱执行"""
        container = None
        try:
            container = self._create_container()
            container.start()
            self._inject_code(container, "main.py", code)
            for fname, content in input_files.items():
                self._inject_code(container, fname, content)
            result = container.exec_run(["python", "-u", "main.py"],
                                        workdir=self.work_dir, demux=True)
            stdout = result.output[0].decode("utf-8") if result.output[0] else ""
            stderr = result.output[1].decode("utf-8") if result.output[1] else ""
            outputs = {}
            for ofile in output_files:
                content = self._extract_output(container, f"{self.work_dir}/{ofile}")
                if content is not None:
                    outputs[ofile] = content
            return {
                "success": result.exit_code == 0,
                "exit_code": result.exit_code,
                "stdout": stdout,
                "stderr": stderr,
                "output_files": outputs,
            }
        except Exception as e:
            return {"success": False, "error": f"沙箱执行异常: {str(e)}"}
        finally:
            if container:
                try:
                    container.kill()
                    container.remove(force=True)
                except Exception:
                    pass


from typing import List

DockerSandbox类封装了基于Docker的代码执行沙箱。在创建容器时，它设置了多项安全限制：network="none"禁用网络访问，mem_limit限制内存使用，cpu_quota限制CPU使用，cap_drop=["ALL"]移除所有Linux能力，security_opt=["no-new-privileges"]防止权限提升。这些设置共同构成了一个严格的隔离环境。

execute方法是核心执行入口。它创建容器、注入代码、执行并收集输出。每次执行完成后，容器会被立即销毁，确保不会留下任何痕迹。execute_with_files方法扩展了基本功能，支持输入文件和输出文件，适用于需要处理文件的代码执行场景。

Docker沙箱的优点是隔离性强、生态成熟、支持多种语言。缺点是每个容器有一定的启动开销（通常在数百毫秒到数秒），不适合需要频繁执行的轻量级任务。此外，Docker守护进程本身需要较高的权限，在多租户环境中需要额外的安全措施。

三、WASM沙箱方案

WebAssembly（WASM）是一种二进制指令格式，最初为浏览器设计，但因其出色的安全性和性能，逐渐被用于服务端沙箱场景。WASM代码在一个受限的虚拟机中运行，默认情况下无法访问文件系统、网络或宿主系统的任何资源。所有外部交互都必须通过显式定义的导入接口进行。

WASM沙箱相比Docker沙箱有几个显著优势：启动速度极快（毫秒级），资源开销极小（不需要完整的操作系统环境），安全性由设计保证（不需要额外配置安全策略）。缺点是WASM目前对某些高级语言的支持还不够完善，且调试工具相对有限。

import wasmer
from typing import Any


class WASMSandbox:
    """基于WASM的代码执行沙箱"""

    def __init__(self):
        self._store = wasmer.Store()
        self._import_object = wasmer.ImportObject()
        self._setup_imports()

    def _setup_imports(self):
        """设置WASM导入对象，定义沙箱允许的外部调用"""
        def clock_time_get():
            import time
            return int(time.time())

        def proc_exit(code: int):
            raise RuntimeError(f"WASM程序退出，代码: {code}")

        self._import_object.register(
            "wasi_snapshot_preview1",
            {
                "clock_time_get": wasmer.Function(self._store, clock_time_get),
                "proc_exit": wasmer.Function(self._store, proc_exit),
            },
        )

    def execute_wasm(self, wasm_bytes: bytes, args: list = None) -> Dict[str, Any]:
        """执行WASM字节码"""
        try:
            module = wasmer.Module(self._store, wasm_bytes)
            instance = module.instantiate(self._import_object)
            main_func = instance.exports.main
            if main_func is None:
                return {"success": False, "error": "未找到main函数"}
            result = main_func(*args) if args else main_func()
            return {"success": True, "result": result}
        except Exception as e:
            return {"success": False, "error": str(e)}

    def execute_python_in_wasm(self, python_code: str) -> Dict[str, Any]:
        """在WASM环境中执行Python代码（通过Pyodide等方案）"""
        try:
            import subprocess
            import tempfile
            import os
            with tempfile.NamedTemporaryFile(mode="w", suffix=".py",
                                              delete=False, encoding="utf-8") as f:
                f.write(python_code)
                temp_path = f.name
            result = subprocess.run(
                ["pyodide", "run", temp_path],
                capture_output=True, text=True, timeout=30,
            )
            os.unlink(temp_path)
            return {
                "success": result.returncode == 0,
                "stdout": result.stdout,
                "stderr": result.stderr,
            }
        except FileNotFoundError:
            return {"success": False, "error": "Pyodide运行时未安装"}
        except Exception as e:
            return {"success": False, "error": str(e)}


class WasmRuntime:
    """轻量级WASM运行时封装"""

    def __init__(self, fuel_limit: int = 1000000, memory_limit_pages: int = 256):
        self.fuel_limit = fuel_limit
        self.memory_limit_pages = memory_limit_pages
        self._modules: Dict[str, bytes] = {}

    def load_module(self, name: str, wasm_bytes: bytes):
        """加载WASM模块"""
        self._modules[name] = wasm_bytes

    def run_module(self, name: str, func_name: str = "main",
                   params: list = None) -> Dict[str, Any]:
        """运行已加载的模块"""
        if name not in self._modules:
            return {"success": False, "error": f"模块 {name} 未加载"}
        store = wasmer.Store()
        module = wasmer.Module(store, self._modules[name])
        import_obj = wasmer.ImportObject()
        instance = module.instantiate(import_obj)
        func = getattr(instance.exports, func_name, None)
        if func is None:
            return {"success": False, "error": f"函数 {func_name} 未找到"}
        try:
            result = func(*(params or []))
            return {"success": True, "result": result}
        except Exception as e:
            return {"success": False, "error": str(e)}

WASMSandbox类封装了基于WASM的代码执行环境。_setup_imports方法定义了沙箱允许的外部调用，这是WASM沙箱安全模型的核心：代码只能通过这些预定义的接口与外部交互，无法直接访问文件系统或网络。execute_wasm方法直接执行WASM字节码，execute_python_in_wasm方法通过Pyodide在WASM环境中执行Python代码。

WasmRuntime类提供了更轻量级的封装，支持模块加载和按需执行。fuel_limit参数限制了WASM代码可以执行的指令数量，防止无限循环。memory_limit_pages限制了WASM线性内存的大小，每个页面为64KB。

WASM沙箱特别适合执行计算密集型任务和需要快速启动的场景。例如，在Agent需要频繁执行短小的代码片段进行数据转换或计算时，WASM沙箱的毫秒级启动速度远优于Docker沙箱的秒级启动。

四、资源限制策略

无论使用哪种沙箱方案，资源限制都是安全隔离的重要组成部分。没有资源限制的沙箱是不完整的，因为恶意代码可以通过消耗资源来造成拒绝服务攻击。资源限制需要覆盖CPU、内存、磁盘、网络和执行时间等多个维度。

import resource
import signal
from contextlib import contextmanager


class ResourceLimiter:
    """资源限制器"""

    @staticmethod
    @contextmanager
    def limit(memory_mb: int = 256, cpu_seconds: int = 10,
              file_size_mb: int = 10, process_count: int = 1):
        """设置进程级资源限制"""
        memory_bytes = memory_mb * 1024 * 1024
        file_size_bytes = file_size_mb * 1024 * 1024

        old_limits = {}
        try:
            old_limits["memory"] = resource.getrlimit(resource.RLIMIT_AS)
            resource.setrlimit(resource.RLIMIT_AS, (memory_bytes, memory_bytes))
        except (ValueError, resource.error):
            pass

        try:
            old_limits["cpu"] = resource.getrlimit(resource.RLIMIT_CPU)
            resource.setrlimit(resource.RLIMIT_CPU, (cpu_seconds, cpu_seconds))
        except (ValueError, resource.error):
            pass

        try:
            old_limits["file_size"] = resource.getrlimit(resource.RLIMIT_FSIZE)
            resource.setrlimit(resource.RLIMIT_FSIZE, (file_size_bytes, file_size_bytes))
        except (ValueError, resource.error):
            pass

        try:
            old_limits["processes"] = resource.getrlimit(resource.RLIMIT_NPROC)
            resource.setrlimit(resource.RLIMIT_NPROC, (process_count, process_count))
        except (ValueError, resource.error):
            pass

        def _cpu_handler(signum, frame):
            raise TimeoutError("CPU时间超限")
        old_handler = signal.signal(signal.SIGXCPU, _cpu_handler)

        try:
            yield
        finally:
            for key, limits in old_limits.items():
                rlimit_map = {
                    "memory": resource.RLIMIT_AS,
                    "cpu": resource.RLIMIT_CPU,
                    "file_size": resource.RLIMIT_FSIZE,
                    "processes": resource.RLIMIT_NPROC,
                }
                try:
                    resource.setrlimit(rlimit_map[key], limits)
                except (ValueError, resource.error):
                    pass
            signal.signal(signal.SIGXCPU, old_handler)


class ExecutionGuard:
    """执行守卫 - 综合安全控制"""

    def __init__(self, timeout: float = 30.0, memory_mb: int = 256,
                 max_output_size: int = 1024 * 1024):
        self.timeout = timeout
        self.memory_mb = memory_mb
        self.max_output_size = max_output_size
        self._forbidden_modules = {
            "os", "sys", "subprocess", "shutil", "ctypes",
            "multiprocessing", "socket", "http", "urllib",
            "pickle", "marshal", "importlib",
        }

    def check_code_safety(self, code: str) -> tuple:
        """静态检查代码安全性"""
        import ast
        try:
            tree = ast.parse(code)
        except SyntaxError as e:
            return False, f"语法错误: {str(e)}"

        for node in ast.walk(tree):
            if isinstance(node, ast.Import):
                for alias in node.names:
                    module = alias.name.split(".")[0]
                    if module in self._forbidden_modules:
                        return False, f"禁止导入模块: {module}"
            elif isinstance(node, ast.ImportFrom):
                if node.module:
                    module = node.module.split(".")[0]
                    if module in self._forbidden_modules:
                        return False, f"禁止导入模块: {module}"
            elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                for decorator in node.decorator_list:
                    if isinstance(decorator, ast.Name) and decorator.id == "staticmethod":
                        continue
        return True, "检查通过"

    def safe_execute(self, code: str) -> Dict[str, Any]:
        """安全执行代码"""
        is_safe, message = self.check_code_safety(code)
        if not is_safe:
            return {"success": False, "error": f"安全检查失败: {message}"}

        import multiprocessing
        queue = multiprocessing.Queue()
        proc = multiprocessing.Process(target=self._run_code, args=(code, queue))
        proc.start()
        proc.join(timeout=self.timeout)
        if proc.is_alive():
            proc.terminate()
            proc.join()
            return {"success": False, "error": "执行超时"}
        if queue.empty():
            return {"success": False, "error": "执行异常终止"}
        return queue.get()

    def _run_code(self, code: str, queue: multiprocessing.Queue):
        """在子进程中执行代码"""
        with ResourceLimiter.limit(memory_mb=self.memory_mb):
            import io
            from contextlib import redirect_stdout, redirect_stderr
            stdout_buf = io.StringIO()
            stderr_buf = io.StringIO()
            try:
                restricted_globals = {"__builtins__": self._get_safe_builtins()}
                with redirect_stdout(stdout_buf), redirect_stderr(stderr_buf):
                    exec(code, restricted_globals)
                stdout = stdout_buf.getvalue()
                stderr = stderr_buf.getvalue()
                if len(stdout) > self.max_output_size:
                    stdout = stdout[:self.max_output_size] + "\n... [输出被截断]"
                queue.put({"success": True, "stdout": stdout, "stderr": stderr})
            except Exception as e:
                queue.put({"success": False, "error": str(e), "stdout": stdout_buf.getvalue()})

    def _get_safe_builtins(self):
        """获取安全的内建函数集合"""
        import builtins
        safe = {
            "print": print, "len": len, "range": range, "enumerate": enumerate,
            "zip": zip, "map": map, "filter": filter, "sorted": sorted,
            "reversed": reversed, "sum": sum, "min": min, "max": max,
            "abs": abs, "round": round, "pow": pow, "divmod": divmod,
            "isinstance": isinstance, "type": type, "int": int, "float": float,
            "str": str, "bool": bool, "list": list, "dict": dict,
            "set": set, "tuple": tuple, "frozenset": frozenset,
            "True": True, "False": False, "None": None,
        }
        for name in ("math", "json", "re", "datetime", "collections", "itertools"):
            try:
                safe[name] = __import__(name)
            except ImportError:
                pass
        return safe

ResourceLimiter类使用操作系统级别的资源限制机制来约束代码执行。它通过resource.setrlimit设置内存、CPU时间、文件大小和进程数量的上限。当代码超出这些限制时，操作系统会发送相应的信号（如SIGXCPU或SIGSEGV），从而终止执行。

ExecutionGuard类提供了更全面的安全控制。check_code_safety方法使用AST分析来检查代码是否导入了危险模块。safe_execute方法在子进程中执行代码，结合资源限制和超时控制，确保即使代码出现无限循环或内存泄漏，也不会影响主进程。_get_safe_builtins方法构建了一个受限的内建函数集合，只暴露安全的函数和模块。

五、安全审计机制

安全审计是沙箱系统不可或缺的组成部分。它记录所有代码执行活动，为安全分析和事件追踪提供数据支持。一个完善的审计系统应该记录代码内容、执行参数、执行结果、资源消耗和异常信息。

import hashlib
import logging
from datetime import datetime


class SecurityAuditor:
    """安全审计器"""

    def __init__(self, log_file: str = "sandbox_audit.log"):
        self.logger = logging.getLogger("sandbox_audit")
        self.logger.setLevel(logging.INFO)
        handler = logging.FileHandler(log_file)
        handler.setFormatter(logging.Formatter(
            "%(asctime)s | %(levelname)s | %(message)s"
        ))
        self.logger.addHandler(handler)
        self._suspicious_patterns = [
            "rm -rf", "mkfs", "dd if=", "chmod 777",
            "nc -l", "bash -i", "python -c 'import socket'",
            "eval(", "exec(", "__import__",
        ]

    def audit_execution(self, code: str, result: Dict[str, Any],
                        executor: str = "unknown", session_id: str = ""):
        """记录执行审计日志"""
        code_hash = hashlib.sha256(code.encode()).hexdigest()[:16]
        is_suspicious, suspicious_items = self._detect_suspicious(code)
        entry = {
            "timestamp": datetime.now().isoformat(),
            "session_id": session_id,
            "executor": executor,
            "code_hash": code_hash,
            "code_length": len(code),
            "success": result.get("success", False),
            "exit_code": result.get("exit_code"),
            "stdout_length": len(result.get("stdout", "")),
            "stderr_length": len(result.get("stderr", "")),
            "error": result.get("error"),
            "is_suspicious": is_suspicious,
            "suspicious_items": suspicious_items,
        }
        level = logging.WARNING if is_suspicious else logging.INFO
        self.logger.log(level, json.dumps(entry, ensure_ascii=False))
        if is_suspicious:
            self.logger.warning(f"检测到可疑代码执行! hash={code_hash}, items={suspicious_items}")
        return entry

    def _detect_suspicious(self, code: str) -> tuple:
        """检测可疑代码模式"""
        found = []
        code_lower = code.lower()
        for pattern in self._suspicious_patterns:
            if pattern.lower() in code_lower:
                found.append(pattern)
        return (len(found) > 0, found)

    def get_statistics(self) -> Dict[str, Any]:
        """获取审计统计"""
        total = 0
        suspicious = 0
        failed = 0
        try:
            with open(self.logger.handlers[0].baseFilename, "r") as f:
                for line in f:
                    if "WARNING" in line:
                        suspicious += 1
                    total += 1
        except FileNotFoundError:
            pass
        return {
            "total_executions": total,
            "suspicious_count": suspicious,
            "suspicious_rate": suspicious / total if total else 0,
        }

SecurityAuditor类实现了完整的审计功能。audit_execution方法记录每次代码执行的详细信息，包括时间戳、执行者、代码哈希、执行结果和可疑检测结果。代码哈希用于追踪重复执行的代码，而不需要存储完整的代码内容。_detect_suspicious方法通过模式匹配检测已知的恶意代码模式，如文件删除命令、反向Shell构造等。

审计日志应该持久化存储，并设置合理的保留策略。在高安全要求的场景中，审计日志应该写入只追加的存储中，防止被篡改。可以接入SIEM（安全信息和事件管理）系统，实现实时告警和自动化响应。

六、完整安全执行环境实现

将上述所有组件整合在一起，我们构建一个完整的安全代码执行环境。这个环境支持多种沙箱后端，提供统一的执行接口，并内置资源限制和安全审计功能。

class SecureExecutionEnvironment:
    """完整的安全代码执行环境"""

    def __init__(self, backend: str = "docker", timeout: int = 30,
                 memory_mb: int = 256, audit: bool = True):
        self.backend = backend
        self.timeout = timeout
        self.memory_mb = memory_mb
        self.auditor = SecurityAuditor() if audit else None
        self._setup_backend()

    def _setup_backend(self):
        if self.backend == "docker":
            self.sandbox = DockerSandbox(timeout=self.timeout, memory_limit=f"{self.memory_mb}m")
        elif self.backend == "wasm":
            self.sandbox = WASMSandbox()
        elif self.backend == "process":
            self.guard = ExecutionGuard(timeout=self.timeout, memory_mb=self.memory_mb)
        else:
            raise ValueError(f"不支持的执行后端: {self.backend}")

    def run(self, code: str, language: str = "python",
            executor: str = "agent", session_id: str = "") -> Dict[str, Any]:
        """执行代码的统一入口"""
        start_time = time.time()
        if self.backend in ("docker", "wasm"):
            result = self.sandbox.execute(code, language=language)
        else:
            result = self.guard.safe_execute(code)
        result["execution_time"] = time.time() - start_time
        result["backend"] = self.backend
        if self.auditor:
            self.auditor.audit_execution(code, result, executor, session_id)
        return result

    def run_batch(self, tasks: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """批量执行代码"""
        results = []
        for task in tasks:
            result = self.run(
                task["code"],
                language=task.get("language", "python"),
                executor=task.get("executor", "agent"),
                session_id=task.get("session_id", ""),
            )
            results.append(result)
        return results


def demo():
    env = SecureExecutionEnvironment(backend="process", timeout=10, memory_mb=128)
    print("=== 安全执行环境演示 ===")
    safe_code = """
x = list(range(100))
total = sum(x)
print(f"1到99的和: {total}")
print(f"平均值: {total / len(x)}")
"""
    result = env.run(safe_code, executor="demo_agent", session_id="sess_001")
    print(f"成功: {result['success']}")
    print(f"输出: {result.get('stdout', '')}")
    print(f"耗时: {result['execution_time']:.3f}s")
    print(f"后端: {result['backend']}")

    print("\n=== 危险代码拦截演示 ===")
    dangerous_code = """
import os
os.system("rm -rf /tmp/test")
"""
    result = env.run(dangerous_code, executor="demo_agent", session_id="sess_002")
    print(f"成功: {result['success']}")
    print(f"错误: {result.get('error', '')}")

    print("\n=== 审计统计 ===")
    if env.auditor:
        stats = env.auditor.get_statistics()
        for k, v in stats.items():
            print(f"  {k}: {v}")


if __name__ == "__main__":
    demo()

SecureExecutionEnvironment类是安全执行环境的统一入口。它支持三种执行后端：Docker沙箱、WASM沙箱和进程级隔离。run方法是执行代码的统一接口，无论使用哪种后端，调用方式完全一致。run_batch方法支持批量执行，适用于需要运行多个代码片段的场景。

demo函数展示了安全执行环境的完整使用流程。首先执行了一段安全的数据计算代码，然后尝试执行一段包含危险操作（删除文件）的代码，安全检查成功拦截了危险代码。最后展示了审计统计功能。

七、沙箱方案对比与选择

不同的沙箱方案各有优劣，选择时需要根据具体场景权衡。Docker沙箱提供最强的隔离性和最广泛的语言支持，适合执行复杂的不受信任代码。它的缺点是启动开销较大，不适合高频轻量级执行。

WASM沙箱提供极快的启动速度和极低的资源开销，适合执行短小的计算任务。它的安全性由设计保证，不需要额外的安全配置。缺点是目前主要支持C/C++/Rust等编译型语言，对Python等解释型语言的支持需要借助Pyodide等中间层。

进程级隔离是最轻量的方案，不需要额外的运行时环境。它通过操作系统的进程隔离和资源限制来提供安全保障。适合执行可信度较高的代码，或作为其他沙箱方案的补充。

在实际应用中，可以采用多层沙箱策略：首先使用静态分析进行代码安全检查，然后根据代码的风险等级选择合适的沙箱后端。低风险代码使用进程级隔离快速执行，高风险代码使用Docker沙箱严格隔离。这种分层策略可以在安全性和性能之间取得良好的平衡。

八、高级安全特性

除了基本的隔离和限制功能外，一个成熟的沙箱系统还可以包含以下高级安全特性。

代码签名验证：在执行代码前，验证代码的数字签名，确保代码来自可信的来源且未被篡改。这可以有效防止供应链攻击，即攻击者在代码传输过程中注入恶意代码。

行为分析：通过机器学习模型分析代码的执行行为，识别异常模式。例如，一个排序算法不应该产生网络请求，如果检测到网络请求行为，可以判定为异常并终止执行。

速率限制：限制单个用户或会话的代码执行频率，防止资源滥用。可以设置每分钟最大执行次数、每天最大执行时间等限制。

输出过滤：检查代码执行的输出，过滤掉可能包含敏感信息的输出。例如，过滤掉匹配信用卡号、身份证号等敏感模式的输出内容。

网络策略：对于需要网络访问的代码，实施严格的网络策略。可以设置白名单，只允许访问特定的域名和端口；可以设置流量限制，防止大量数据外泄。

九、总结

Agent代码执行沙箱是保障系统安全的关键基础设施。本文从隔离层次、Docker沙箱、WASM沙箱、资源限制、安全审计等多个维度，详细阐述了如何构建一个安全可靠的代码执行环境。

核心设计原则是纵深防御：不依赖单一的安全机制，而是通过多层隔离、多重限制和持续监控来构建完整的安全体系。从语言层面的安全检查到操作系统级别的资源限制，从容器隔离到审计追踪，每一层都为系统安全增加一道屏障。

在实际部署中，沙箱方案的选择应该基于威胁模型和性能需求。对于面向公众的服务，应该采用最严格的隔离方案；对于内部使用的开发工具，可以适当放宽限制以提高性能。无论选择哪种方案，安全审计和持续监控都是不可或缺的，它们是发现和响应安全事件的最后一道防线。

随着Agent技术的不断发展，代码执行沙箱也将持续演进。未来的方向包括基于硬件可信执行环境（TEE）的沙箱方案、基于形式化验证的安全证明、以及更加智能的行为分析和威胁检测能力。这些技术的进步将使Agent代码执行更加安全、高效和可靠。