Xây dựng Code Sandbox an toàn cho AI Agent với E2B và Python

AI tutorial - IT technology blog
AI tutorial - IT technology blog

Cảnh báo Out-of-Memory lúc 2:14 sáng làm tê liệt cả cụm máy chủ

Điện thoại của tôi reo lúc 2:14 sáng với một cảnh báo PagerDuty mức độ ưu tiên cao. Node backend chính 32 core đã ngừng phản hồi, mức tải CPU bị kẹt cứng ở mức 100%, và disk I/O nghẽn hoàn toàn.

Ngay khi tôi mở được phiên terminal qua bastion host để kiểm tra htop, thủ phạm lộ diện rõ ràng. Con agent phân tích dữ liệu tự động của chúng tôi đã cố phân tích một file CSV khách hàng nặng 4 GB bằng cách tạo ra một vòng lặp subprocess đệ quy mất kiểm soát. Tệ hơn nữa, nó còn xả thẳng 18 GB file phân mảnh chưa nén vào /tmp, khiến phân vùng root bị đầy 100% dung lượng.

Gọi lệnh exec() mặc định của Python trực tiếp trên máy chủ ứng dụng sẽ biến mọi hiện tượng hallucination của mô hình thành một nguy cơ sập hạ tầng tiềm ẩn. Trong trường hợp của chúng tôi, chỉ một câu prompt lỗi đã đánh sập hai dịch vụ nội bộ trọng yếu suốt gần 45 phút. Nếu ứng dụng của bạn phải thực thi mã nguồn do AI sinh ra, sandbox là yêu cầu bắt buộc không thể thương lượng.

Phân tích nguyên nhân gốc rễ: Vì sao mô hình chạy trực tiếp trên máy chủ lại sụp đổ

Sau khi chạy kill -9 để hạ tiến trình cha bất trị, buổi họp rút kinh nghiệm của chúng tôi đã chỉ ra ba lỗ hổng mang tính hệ thống khi thực thi cục bộ (nhất là trong các quy trình giám sát hệ thống tương tự như AIOps phân tích cảnh báo và RCA):

  • Thoát tiến trình và cạn kiệt tài nguyên: Việc chạy các script không đáng tin cậy bằng subprocess.Popen hay eval() sẽ dùng chung kernel, vùng nhớ swap và network interface của máy chủ host. Một vòng lặp while True không điểm dừng hay một lệnh cấp phát mảng NumPy không giới hạn sẽ bóp nghẹt các worker thread xung quanh ngay lập tức.
  • Sự thất bại của các bộ lọc AST: Phân tích tĩnh và danh sách đen cú pháp nhắm vào các module như os hay sys hiếm khi hiệu quả. Kẻ tấn công—và cả những LLM có xu hướng “sáng tạo”—dễ dàng vượt qua các bộ lọc chuỗi ngây thơ bằng getattr(), import động như __import__('sub' + 'process'), hoặc mã hóa base64. Để kiểm soát an toàn hơn từ tầng prompt, việc kết hợp bảo mật AI với NeMo Guardrails để chặn Prompt Injection là giải pháp nhiều đội ngũ lựa chọn.
  • Nhiễm bẩn trạng thái dùng chung: Khi một agent ghi file tạm hoặc làm thay đổi biến môi trường trong lượt chạy A, lượt chạy B sẽ vô tình nhận lại các tàn dư đó. Hiện tượng rò rỉ trạng thái này phá vỡ tính nhất quán khi tái lập môi trường và tạo ra những lỗ hổng bảo mật nghiêm trọng giữa các tenant.

Đánh giá các chiến lược thực thi: Từ Subprocess đến MicroVM

Chúng tôi đã thử nghiệm ba mô hình thực thi khác nhau qua 500 lượt chạy agent chuẩn trước khi chốt thiết kế cho môi trường production:

1. Subprocess gốc kèm giới hạn tài nguyên

import subprocess

# Rủi ro: Dùng chung filesystem, bộ nhớ và mạng của máy chủ host
res = subprocess.run(["python3", "-c", agent_code], capture_output=True, timeout=10)

Linux cgroups và ulimit có thể siết bớt chu kỳ CPU và heap bộ nhớ, nhưng chúng không mang lại sự cô lập thực sự. Chúng không thể ngăn chặn các request mạng trái phép ra bên ngoài, việc đọc lén các file nhạy cảm trong /etc, hay các cuộc tấn công làm cạn kiệt dung lượng đĩa.

2. Container Docker tạm thời (Ephemeral Containers)

Đóng gói từng tác vụ vào container giúp giải quyết vấn đề rò rỉ filesystem, nhưng thời gian khởi động lại dao động từ 1,5 đến 3,2 giây cho mỗi lần gọi. Độ trễ đó làm giảm trải nghiệm người dùng trong các vòng lặp LLM nhiều lượt. Ngoài ra, việc chạy Docker-in-Docker trên các worker node của Kubernetes đòi hỏi quyền privileged, điều này vi phạm các tiêu chuẩn tuân thủ bảo mật và mở rộng bề mặt tấn công.

3. Firecracker MicroVM thông qua E2B Cloud Sandbox

E2B khởi động các Firecracker MicroVM cô lập chỉ trong khoảng 150 đến 250 mili-giây. Mỗi sandbox khởi chạy một Linux kernel tinh gọn với CPU, bộ nhớ và card mạng ảo chuyên dụng. Khi quá trình thực thi hoàn tất, môi trường sẽ tự hủy sạch sẽ, không để lại bất kỳ trạng thái vướng bận nào trên máy chủ của bạn.

Các bước triển khai thực tế: Cô lập khối lượng công việc của Agent với E2B

Chuyển giao việc thực thi mã sang các sandbox E2B cô lập đã loại bỏ triệt để các sự cố về độ ổn định máy chủ của chúng tôi. Nếu bạn đang xây dựng các hệ thống như AI Agent tự trị hoặc chuẩn hóa kết nối công cụ qua MCP Server tùy chỉnh bằng Python, hãy làm theo các bước sau để xây dựng một trình chạy thực thi linh hoạt bằng Python.

Bước 1: Cài đặt SDK và cấu hình thông tin xác thực

Cài đặt package code interpreter của E2B cùng với python-dotenv:

pip install e2b-code-interpreter python-dotenv

Thêm API token của bạn vào biến môi trường shell:

export E2B_API_KEY="e2b_your_actual_api_key_here"

Bước 2: Xây dựng wrapper thực thi trong Sandbox

Đoạn mã triển khai này sẽ nạp code thô của agent vào một môi trường cô lập, bắt các luồng stream chuẩn và áp đặt mức trần thời gian thực thi cứng.

import os
from e2b_code_interpreter import Sandbox

def run_untrusted_agent_code(script: str) -> dict:
    """
    Chạy mã Python tùy ý bên trong một E2B microVM cô lập.
    Trả về stdout, stderr và lỗi runtime mà không chạm vào máy chủ host.
    """
    try:
        # Khởi tạo một microVM mới, dùng một lần
        with Sandbox() as sandbox:
            execution = sandbox.run_code(
                script,
                timeout=30  # Giới hạn cứng 30 giây
            )
            
            return {
                "status": "success",
                "stdout": [out.text for out in execution.logs.stdout],
                "stderr": [err.text for err in execution.logs.stderr],
                "error": execution.error.name if execution.error else None,
                "error_value": execution.error.value if execution.error else None
            }
    except Exception as exc:
        return {
            "status": "sandbox_failure",
            "error": str(exc)
        }

# Payload thử nghiệm: tính toán các giá trị lớn trong bộ nhớ
risky_agent_payload = """
import sys
import time

print("Đang thực thi phân tích bên trong sandbox an toàn...")
data = [x ** 2 for x in range(500000)]
print(f"Đã xử lý thành công {len(data)} phần tử.")
"""

result = run_untrusted_agent_code(risky_agent_payload)
print(result)

Bước 3: Nạp file an toàn và lấy lại artifact kết quả

Tuyệt đối không mount trực tiếp các thư mục của máy chủ host vào môi trường agent. Hãy ghi các payload đầu vào vào ổ đĩa ảo của máy ảo thông qua SDK, xử lý dữ liệu, rồi stream các artifact đã tạo ngược trở lại bộ nhớ ứng dụng:

from e2b_code_interpreter import Sandbox

with Sandbox() as sandbox:
    # Tải tập dữ liệu mẫu vào môi trường máy ảo
    csv_payload = "id,metric\n1,42\n2,85\n3,128\n"
    sandbox.files.write("/home/user/input.csv", csv_payload)
    
    # Thực thi các biến đổi dữ liệu bên trong sandbox
    sandbox.run_code("""
import pandas as pd
df = pd.read_csv('/home/user/input.csv')
df['scaled'] = df['metric'] * 1.5
df.to_csv('/home/user/output.csv', index=False)
    """)
    
    # Lấy kết quả đã xử lý về dưới dạng chuỗi hoặc byte
    processed_content = sandbox.files.read("/home/user/output.csv")
    print(processed_content)

Ba nguyên tắc phòng vệ trên môi trường Production cho đội ngũ kỹ thuật

Trước khi đưa tính năng thực thi mã do agent điều khiển tới tay người dùng thực tế, hãy áp dụng các chính sách nền tảng sau:

  1. Timeout nghiêm ngặt: Luôn truyền tham số timeout rõ ràng (chẳng hạn từ 30 đến 60 giây) vào sandbox.run_code(). Mã do AI sinh ra tuyệt đối không được phép chạy vô tận.
  2. Bắt buộc dọn dẹp context: Bọc mọi lệnh gọi trong context manager with Sandbox() của Python. Điều này đảm bảo MicroVM được hủy ngay lập tức khi kết thúc, tránh tình trạng bỏ quên phiên chạy gây phát sinh chi phí thanh toán ngoài ý muốn.
  3. Hạn chế kết nối mạng ra ngoài (Egress): Nếu agent của bạn chỉ phân tích các tập dữ liệu CSV hoặc JSON tĩnh, hãy tắt quyền truy cập mạng bên ngoài trong cấu hình mẫu sandbox. Việc chặn lưu lượng mạng chiều đi sẽ triệt tiêu hoàn toàn nguy cơ rò rỉ dữ liệu và các request trái phép ra ngoài.
Share: