AskTable

AskTable Agent 沙箱执行环境：如何安全地运行 AI 生成的代码

AskTable 团队
AskTable 团队 2026年3月5日

在 AI Canvas 中，用户可以让 AI 生成并执行 Python 代码、SQL 查询和 JSX 图表。如何保证这些代码的安全执行？如何防止恶意代码破坏系统？如何限制资源消耗？

AskTable 的 Agent 沙箱执行环境，通过 资源限制 + 超时管理 + 权限控制 的多层防护，实现了安全可靠的代码执行。

一、为什么需要沙箱？

1. AI 生成代码的风险

恶意代码：

# AI 可能生成的危险代码
import os
os.system("rm -rf /")  # 删除系统文件

资源耗尽：

# 无限循环
while True:
    pass

# 内存耗尽
data = [0] * (10 ** 10)

数据泄露：

# 读取敏感文件
with open("/etc/passwd") as f:
    print(f.read())

2. 沙箱的核心目标

✅ 隔离性：代码执行不影响宿主系统 ✅ 资源限制：限制 CPU、内存、执行时间 ✅ 权限控制：禁止访问敏感资源 ✅ 错误隔离：异常不影响其他任务

二、Python 沙箱实现

1. RestrictedPython 限制

from RestrictedPython import compile_restricted, safe_globals

def execute_python_code(code: str, context: dict) -> Any:
    """在受限环境中执行 Python 代码"""
    # 1. 编译受限代码
    byte_code = compile_restricted(
        code,
        filename="<user_code>",
        mode="exec"
    )

    # 2. 准备安全的全局变量
    restricted_globals = {
        **safe_globals,
        "__builtins__": {
            "print": print,
            "len": len,
            "range": range,
            "sum": sum,
            # 只暴露安全的内置函数
        },
        "pd": pd,  # pandas
        "np": np,  # numpy
    }

    # 3. 执行代码
    exec(byte_code, restricted_globals, context)

    return context.get("result")

限制内容：

2. 超时管理

import signal
from contextlib import contextmanager

@contextmanager
def timeout(seconds: int):
    """超时上下文管理器"""
    def timeout_handler(signum, frame):
        raise TimeoutError(f"Execution timeout after {seconds}s")

    # 设置超时信号
    signal.signal(signal.SIGALRM, timeout_handler)
    signal.alarm(seconds)

    try:
        yield
    finally:
        signal.alarm(0)  # 取消超时

# 使用
with timeout(30):
    execute_python_code(code, context)

3. 资源限制

import resource

def set_resource_limits():
    """设置资源限制"""
    # 限制内存（512MB）
    resource.setrlimit(
        resource.RLIMIT_AS,
        (512 * 1024 * 1024, 512 * 1024 * 1024)
    )

    # 限制 CPU 时间（30 秒）
    resource.setrlimit(
        resource.RLIMIT_CPU,
        (30, 30)
    )

    # 限制文件大小（10MB）
    resource.setrlimit(
        resource.RLIMIT_FSIZE,
        (10 * 1024 * 1024, 10 * 1024 * 1024)
    )

三、SQL 沙箱实现

1. 只读权限

async def execute_sql_readonly(sql: str, datasource: DataSourceAdmin) -> pd.DataFrame:
    """只读执行 SQL"""
    # 1. 检查 SQL 类型
    if not is_select_query(sql):
        raise errors.SecurityError("Only SELECT queries are allowed")

    # 2. 使用只读连接
    async with datasource.get_readonly_connection() as conn:
        df = await conn.execute(sql)

    return df

def is_select_query(sql: str) -> bool:
    """检查是否为 SELECT 查询"""
    parsed = sqlglot.parse_one(sql)
    return isinstance(parsed, exp.Select)

2. 超时控制

async def execute_sql_with_timeout(
    sql: str, datasource: DataSourceAdmin, timeout_seconds: int = 30
) -> pd.DataFrame:
    """带超时的 SQL 执行"""
    try:
        df = await asyncio.wait_for(
            datasource.execute_sql(sql),
            timeout=timeout_seconds
        )
        return df
    except asyncio.TimeoutError:
        raise errors.QueryTimeout(f"Query timeout after {timeout_seconds}s")

3. 行数限制

async def execute_sql_with_limit(
    sql: str, datasource: DataSourceAdmin, max_rows: int = 10000
) -> pd.DataFrame:
    """限制返回行数"""
    # 1. 解析 SQL
    parsed = sqlglot.parse_one(sql, read=datasource.dialect)

    # 2. 添加 LIMIT 子句
    if not parsed.find(exp.Limit):
        parsed = parsed.limit(max_rows)

    # 3. 执行
    modified_sql = parsed.sql(dialect=datasource.dialect)
    df = await datasource.execute_sql(modified_sql)

    return df

四、JSX 沙箱实现

1. JSX 编译

import subprocess

def compile_jsx(jsx_code: str) -> str:
    """编译 JSX 为 JavaScript"""
    # 使用 esbuild 编译
    result = subprocess.run(
        ["esbuild", "--loader=jsx", "--format=iife"],
        input=jsx_code.encode(),
        capture_output=True,
        timeout=5,
    )

    if result.returncode != 0:
        raise errors.CompilationError(result.stderr.decode())

    return result.stdout.decode()

2. 浏览器沙箱

// 前端执行 JSX（在 iframe 中隔离）
function executeJSX(compiledCode) {
    const iframe = document.createElement('iframe');
    iframe.sandbox = 'allow-scripts';  // 只允许脚本执行
    iframe.style.display = 'none';
    document.body.appendChild(iframe);

    const iframeDoc = iframe.contentDocument;
    const script = iframeDoc.createElement('script');
    script.textContent = compiledCode;
    iframeDoc.body.appendChild(script);
}

五、实战案例

案例 1：Python 数据处理

# 用户代码
code = """
import pandas as pd

# 加载父节点数据
df = parent_dataframes[0]

# 数据清洗
df_clean = df.dropna()
df_clean['amount'] = df_clean['amount'].astype(float)

# 计算统计
result = df_clean.groupby('category')['amount'].sum()
"""

# 安全执行
context = {"parent_dataframes": [df1, df2]}
with timeout(30):
    set_resource_limits()
    result = execute_python_code(code, context)

案例 2：SQL 查询

# 用户 SQL
sql = "SELECT * FROM orders WHERE amount > 1000"

# 安全执行
df = await execute_sql_with_timeout(
    sql,
    datasource,
    timeout_seconds=30
)

六、总结

AskTable 的沙箱执行环境，通过多层防护实现了安全可靠的代码执行：

✅ Python：RestrictedPython + 超时 + 资源限制 ✅ SQL：只读权限 + 超时 + 行数限制 ✅ JSX：编译隔离 + 浏览器沙箱

相关阅读：

技术交流：