沙箱方案对比：

  方案          隔离级别    启动速度    资源开销    安全性
  ──────────────────────────────────────────────────────
  进程级 (无隔离)  无        0ms        0          极低
  nsjail         进程+NS   5-10ms     ~5MB       高
  gVisor         用户态内核  50-100ms   ~30MB      很高
  Docker         容器       200-500ms  ~50MB      高
  VM (Firecracker) 虚拟机   125ms      ~30MB      最高

nsjail 四层安全隔离：

  Layer 1: Linux Namespaces
  ┌─────────────────────────────┐
  │ PID NS: 隔离进程树            │
  │ NET NS: 隔离网络栈            │
  │ MNT NS: 隔离文件系统挂载       │
  │ UTS NS: 隔离主机名            │
  │ IPC NS: 隔离进程间通信         │
  │ USER NS: 隔离用户/组 ID       │
  └─────────────────────────────┘

  Layer 2: seccomp-bpf
  ┌─────────────────────────────┐
  │ 只允许白名单系统调用            │
  │ 阻止: ptrace, mount, reboot  │
  │ 阻止: socket (可选)           │
  └─────────────────────────────┘

  Layer 3: cgroups v2
  ┌─────────────────────────────┐
  │ CPU 时间限制                  │
  │ 内存上限                     │
  │ PID 数量限制                  │
  │ I/O 带宽限制                  │
  └─────────────────────────────┘

  Layer 4: chroot / pivot_root
  ┌─────────────────────────────┐
  │ 只读根文件系统                 │
  │ 最小化挂载点                   │
  │ 无 /proc, /sys 访问           │
  └─────────────────────────────┘

# nsjail 配置文件：python-sandbox.cfg

name: "python-function-sandbox"
description: "Sandbox for Python user-defined functions"

mode: ONCE                          # 执行一次后退出
time_limit: 30                      # 最大执行时间 30s
rlimit_as_type: HARD
rlimit_as: 512                      # 虚拟内存 512MB
rlimit_cpu_type: HARD
rlimit_cpu: 10                      # CPU 时间 10s
rlimit_fsize_type: HARD
rlimit_fsize: 10                    # 文件大小 10MB
rlimit_nofile_type: HARD
rlimit_nofile: 64                   # 文件描述符 64

clone_newnet: true                  # 网络隔离
clone_newuser: true                 # 用户隔离
clone_newns: true                   # 挂载隔离
clone_newpid: true                  # PID 隔离
clone_newipc: true                  # IPC 隔离
clone_newuts: true                  # UTS 隔离

uidmap {
    inside_id: "1000"
    outside_id: "65534"             # nobody
    count: 1
}

gidmap {
    inside_id: "1000"
    outside_id: "65534"
    count: 1
}

mount {
    src: "/usr"
    dst: "/usr"
    is_bind: true
    rw: false
}

mount {
    src: "/lib"
    dst: "/lib"
    is_bind: true
    rw: false
}

mount {
    src: "/lib64"
    dst: "/lib64"
    is_bind: true
    rw: false
}

mount {
    dst: "/tmp"
    fstype: "tmpfs"
    rw: true
    options: "size=50m"
}

mount {
    dst: "/dev"
    fstype: "tmpfs"
    rw: false
}

# seccomp 策略
seccomp_string: "POLICY sandbox {"
seccomp_string: "  ALLOW {"
seccomp_string: "    read, write, close, fstat, lseek,"
seccomp_string: "    mmap, mprotect, munmap, brk,"
seccomp_string: "    ioctl, access, pipe, select,"
seccomp_string: "    clone, execve, wait4, exit_group,"
seccomp_string: "    openat, getdents64, fcntl,"
seccomp_string: "    futex, set_tid_address, set_robust_list,"
seccomp_string: "    rt_sigaction, rt_sigprocmask,"
seccomp_string: "    getpid, getuid, getgid, gettid,"
seccomp_string: "    arch_prctl, prlimit64, getrandom"
seccomp_string: "  }"
seccomp_string: "  DENY {"
seccomp_string: "    ptrace, mount, reboot, swapon,"
seccomp_string: "    init_module, delete_module,"
seccomp_string: "    socket, connect, bind, listen"
seccomp_string: "  }"
seccomp_string: "}"

cgroup_mem_max: 268435456           # 256MB
cgroup_pids_max: 32                 # 最大 32 个进程
cgroup_cpu_ms_per_sec: 500          # 每秒最多使用 500ms CPU

from __future__ import annotations
import subprocess
import json
import tempfile
import os
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any


@dataclass
class NsjailConfig:
    """nsjail 配置"""
    config_path: str                      # .cfg 文件路径
    nsjail_binary: str = "/usr/bin/nsjail"
    work_dir: str = "/tmp/nsjail-work"
    log_level: str = "WARNING"


@dataclass
class ExecutionResult:
    """沙箱执行结果"""
    stdout: str
    stderr: str
    exit_code: int
    duration_ms: float
    memory_peak_mb: float = 0.0
    cpu_time_ms: float = 0.0
    killed_by_signal: int = 0
    oom_killed: bool = False


class NsjailExecutor:
    """nsjail 沙箱执行器"""

    def __init__(self, config: NsjailConfig):
        self._config = config
        self._verify_nsjail()

    def _verify_nsjail(self) -> None:
        """验证 nsjail 可用"""
        result = subprocess.run(
            [self._config.nsjail_binary, "--version"],
            capture_output=True, text=True,
        )
        if result.returncode != 0:
            raise RuntimeError("nsjail not found or not executable")

    def execute(self, command: list[str],
                env: dict[str, str] | None = None,
                stdin_data: str | None = None,
                extra_mounts: list[tuple[str, str, bool]] | None = None,
                timeout_seconds: float = 30.0) -> ExecutionResult:
        """在 nsjail 沙箱中执行命令"""
        start = time.monotonic()

        cmd = [
            self._config.nsjail_binary,
            "--config", self._config.config_path,
            "--log_level", self._config.log_level,
        ]

        # 额外挂载
        if extra_mounts:
            for src, dst, rw in extra_mounts:
                flag = "--bindmount" if not rw else "--bindmount"
                cmd.extend([flag, f"{src}:{dst}"])

        # 环境变量
        if env:
            for key, value in env.items():
                cmd.extend(["--env", f"{key}={value}"])

        cmd.append("--")
        cmd.extend(command)

        try:
            result = subprocess.run(
                cmd,
                input=stdin_data,
                capture_output=True,
                text=True,
                timeout=timeout_seconds,
            )

            duration = (time.monotonic() - start) * 1000

            return ExecutionResult(
                stdout=result.stdout,
                stderr=result.stderr,
                exit_code=result.returncode,
                duration_ms=duration,
            )

        except subprocess.TimeoutExpired:
            duration = (time.monotonic() - start) * 1000
            return ExecutionResult(
                stdout="",
                stderr="Execution timeout",
                exit_code=-1,
                duration_ms=duration,
                killed_by_signal=9,
            )

    def execute_python(self, script_path: str,
                        venv_dir: str | None = None,
                        timeout: float = 30.0) -> ExecutionResult:
        """执行 Python 脚本"""
        python = os.path.join(venv_dir, "bin", "python") if venv_dir else "/usr/bin/python3"

        extra_mounts = []
        if venv_dir:
            extra_mounts.append((venv_dir, venv_dir, False))

        extra_mounts.append((
            os.path.dirname(script_path),
            "/sandbox/scripts",
            False,
        ))

        return self.execute(
            [python, f"/sandbox/scripts/{os.path.basename(script_path)}"],
            extra_mounts=extra_mounts,
            timeout_seconds=timeout,
        )

    def execute_node(self, script_path: str,
                      node_modules: str | None = None,
                      timeout: float = 30.0) -> ExecutionResult:
        """执行 Node.js 脚本"""
        extra_mounts = []
        if node_modules:
            extra_mounts.append((node_modules, "/sandbox/node_modules", False))

        extra_mounts.append((
            os.path.dirname(script_path),
            "/sandbox/scripts",
            False,
        ))

        return self.execute(
            ["/usr/bin/node", f"/sandbox/scripts/{os.path.basename(script_path)}"],
            extra_mounts=extra_mounts,
            timeout_seconds=timeout,
        )

class SandboxedPythonRuntime(LanguageRuntime):
    """基于 nsjail 的 Python 运行时"""

    def __init__(self, nsjail_config: NsjailConfig):
        self._executor = NsjailExecutor(nsjail_config)
        self._venv_cache: dict[str, str] = {}

    def supported_language(self):
        from enum import Enum
        class FL(Enum):
            PYTHON = "python"
        return FL.PYTHON

    async def initialize(self, spec) -> None:
        """在沙箱外创建 venv（安装依赖是受信操作）"""
        venv_dir = f"/var/fn-venvs/{spec.function_id}/{spec.version}"
        os.makedirs(venv_dir, exist_ok=True)

        subprocess.run(
            ["python3", "-m", "venv", venv_dir],
            check=True, timeout=30,
        )

        if spec.dependencies:
            allowed = self._filter_allowed_deps(spec.dependencies)
            pip = os.path.join(venv_dir, "bin", "pip")
            subprocess.run(
                [pip, "install", "--no-cache-dir"] + allowed,
                check=True, timeout=120,
            )

        self._venv_cache[spec.function_id] = venv_dir

    async def execute(self, spec, inputs: dict) -> dict:
        """在 nsjail 沙箱中执行"""
        wrapper = f"""
import json, sys

{spec.source_code}

inputs = json.loads(sys.stdin.read())
result = {spec.entry_point}(**inputs)
print(json.dumps(result))
"""
        with tempfile.NamedTemporaryFile(
            mode="w", suffix=".py", delete=False, dir="/tmp"
        ) as f:
            f.write(wrapper)
            script_path = f.name

        try:
            venv_dir = self._venv_cache.get(spec.function_id)

            result = self._executor.execute(
                command=[
                    os.path.join(venv_dir, "bin", "python") if venv_dir else "/usr/bin/python3",
                    "/sandbox/script.py",
                ],
                stdin_data=json.dumps(inputs),
                extra_mounts=[
                    (script_path, "/sandbox/script.py", False),
                ] + ([(venv_dir, venv_dir, False)] if venv_dir else []),
                timeout_seconds=spec.resource_limits.max_execution_time_ms / 1000,
            )

            if result.exit_code != 0:
                raise RuntimeError(
                    f"Sandbox execution failed (exit={result.exit_code}): "
                    f"{result.stderr}"
                )

            if result.oom_killed:
                raise MemoryError("Function killed due to OOM")

            return json.loads(result.stdout)

        finally:
            os.unlink(script_path)

    def _filter_allowed_deps(self, deps: list[str]) -> list[str]:
        """过滤允许的依赖包"""
        allowlist = {
            "numpy", "pandas", "scipy", "scikit-learn",
            "requests", "httpx", "pydantic", "ortools",
            "jsonschema", "pyyaml", "python-dateutil",
        }
        allowed = []
        for dep in deps:
            name = dep.split("==")[0].split(">=")[0].split("<=")[0].lower()
            if name in allowlist:
                allowed.append(dep)
        return allowed

    async def cleanup(self, spec) -> None:
        import shutil
        venv_dir = self._venv_cache.pop(spec.function_id, None)
        if venv_dir and os.path.exists(venv_dir):
            shutil.rmtree(venv_dir)

class CgroupMonitor:
    """cgroups v2 资源监控"""

    def __init__(self, cgroup_path: str = "/sys/fs/cgroup"):
        self._base = cgroup_path

    def get_memory_usage(self, cgroup_name: str) -> dict:
        path = os.path.join(self._base, cgroup_name)
        current = self._read_int(os.path.join(path, "memory.current"))
        peak = self._read_int(os.path.join(path, "memory.peak"))
        limit = self._read_int(os.path.join(path, "memory.max"))

        return {
            "current_mb": current / (1024 * 1024),
            "peak_mb": peak / (1024 * 1024),
            "limit_mb": limit / (1024 * 1024) if limit < 2**62 else "unlimited",
            "usage_pct": (current / limit * 100) if limit < 2**62 else 0,
        }

    def get_cpu_usage(self, cgroup_name: str) -> dict:
        path = os.path.join(self._base, cgroup_name)
        stat_content = self._read_file(os.path.join(path, "cpu.stat"))
        stats = {}
        for line in stat_content.strip().split("\n"):
            key, value = line.split()
            stats[key] = int(value)

        return {
            "user_usec": stats.get("user_usec", 0),
            "system_usec": stats.get("system_usec", 0),
            "total_usec": stats.get("usage_usec", 0),
            "nr_periods": stats.get("nr_periods", 0),
            "nr_throttled": stats.get("nr_throttled", 0),
        }

    def get_pid_count(self, cgroup_name: str) -> dict:
        path = os.path.join(self._base, cgroup_name)
        current = self._read_int(os.path.join(path, "pids.current"))
        limit = self._read_int(os.path.join(path, "pids.max"))

        return {
            "current": current,
            "limit": limit,
        }

    def _read_int(self, path: str) -> int:
        try:
            with open(path) as f:
                content = f.read().strip()
                return int(content) if content != "max" else 2**63
        except (FileNotFoundError, ValueError):
            return 0

    def _read_file(self, path: str) -> str:
        try:
            with open(path) as f:
                return f.read()
        except FileNotFoundError:
            return ""

class SeccompPolicyBuilder:
    """seccomp-bpf 策略构建器"""

    # 安全的系统调用白名单
    SAFE_SYSCALLS = {
        "read", "write", "close", "fstat", "lseek",
        "mmap", "mprotect", "munmap", "brk",
        "ioctl", "access", "pipe", "select", "poll",
        "clone", "fork", "vfork", "execve",
        "wait4", "exit", "exit_group",
        "openat", "getdents64", "fcntl",
        "futex", "set_tid_address", "set_robust_list",
        "rt_sigaction", "rt_sigprocmask", "rt_sigreturn",
        "getpid", "getppid", "getuid", "getgid", "gettid",
        "arch_prctl", "prlimit64", "getrandom",
        "clock_gettime", "clock_getres", "nanosleep",
        "sched_yield", "sched_getaffinity",
    }

    # 危险的系统调用黑名单
    DANGEROUS_SYSCALLS = {
        "ptrace", "mount", "umount2", "reboot",
        "swapon", "swapoff",
        "init_module", "delete_module", "finit_module",
        "kexec_load", "kexec_file_load",
        "pivot_root", "chroot",
        "setns", "unshare",
        "keyctl", "request_key", "add_key",
    }

    @classmethod
    def build_policy(cls, allow_network: bool = False) -> str:
        allowed = set(cls.SAFE_SYSCALLS)

        if allow_network:
            allowed.update({"socket", "connect", "bind", "listen",
                           "accept", "sendto", "recvfrom",
                           "setsockopt", "getsockopt"})

        lines = [
            "POLICY sandbox {",
            "  ALLOW {",
            "    " + ", ".join(sorted(allowed)),
            "  }",
            "  DENY {",
            "    " + ", ".join(sorted(cls.DANGEROUS_SYSCALLS)),
            "  }",
            "}",
        ]
        return "\n".join(lines)

class EscapeDetector:
    """沙箱逃逸检测"""

    @staticmethod
    def scan_source_code(code: str) -> list[str]:
        """扫描源代码中的危险模式"""
        import re

        dangers = []
        patterns = [
            (r"import\s+os\b", "Importing os module"),
            (r"import\s+subprocess\b", "Importing subprocess module"),
            (r"import\s+shutil\b", "Importing shutil module"),
            (r"import\s+ctypes\b", "Importing ctypes module"),
            (r"__import__\s*\(", "Using __import__"),
            (r"eval\s*\(", "Using eval"),
            (r"exec\s*\(", "Using exec"),
            (r"compile\s*\(", "Using compile"),
            (r"open\s*\(.*/etc/", "Accessing /etc"),
            (r"open\s*\(.*/proc/", "Accessing /proc"),
            (r"open\s*\(.*/sys/", "Accessing /sys"),
            (r"socket\s*\.\s*socket", "Creating socket"),
            (r"os\s*\.\s*system", "Using os.system"),
            (r"os\s*\.\s*popen", "Using os.popen"),
            (r"os\s*\.\s*exec", "Using os.exec*"),
            (r"subprocess\s*\.\s*", "Using subprocess"),
        ]

        for pattern, description in patterns:
            if re.search(pattern, code):
                dangers.append(description)

        return dangers

    @staticmethod
    def validate_before_execution(spec) -> list[str]:
        """执行前验证"""
        issues = EscapeDetector.scan_source_code(spec.source_code)

        if spec.resource_limits.network_enabled:
            issues.append("WARNING: Network access is enabled")

        if not spec.resource_limits.filesystem_read_only:
            issues.append("WARNING: Filesystem is writable")

        if spec.resource_limits.max_memory_mb > 512:
            issues.append("WARNING: High memory limit")

        return issues

import asyncio
from collections import deque


class SandboxPool:
    """沙箱池：预热和复用沙箱环境"""

    def __init__(self, nsjail_config: NsjailConfig,
                 pool_size: int = 10):
        self._config = nsjail_config
        self._pool_size = pool_size
        self._available: deque[NsjailExecutor] = deque()
        self._in_use: set[int] = set()
        self._lock = asyncio.Lock()

    async def initialize(self) -> None:
        """预热沙箱池"""
        for _ in range(self._pool_size):
            executor = NsjailExecutor(self._config)
            self._available.append(executor)

    async def acquire(self) -> NsjailExecutor:
        """获取一个沙箱执行器"""
        async with self._lock:
            if self._available:
                executor = self._available.popleft()
                self._in_use.add(id(executor))
                return executor
            else:
                executor = NsjailExecutor(self._config)
                self._in_use.add(id(executor))
                return executor

    async def release(self, executor: NsjailExecutor) -> None:
        """释放沙箱执行器"""
        async with self._lock:
            self._in_use.discard(id(executor))
            if len(self._available) < self._pool_size:
                self._available.append(executor)

    @property
    def stats(self) -> dict:
        return {
            "available": len(self._available),
            "in_use": len(self._in_use),
            "pool_size": self._pool_size,
        }

沙箱启动延迟分解：

  操作                | 延迟    | 优化后
  ──────────────────|────────|────────
  nsjail 进程创建     | 2ms    | 2ms (不可优化)
  Namespace 创建      | 3ms    | 3ms (不可优化)
  chroot 设置         | 1ms    | 1ms
  seccomp 加载        | 0.5ms  | 0.5ms
  cgroup 配置         | 0.5ms  | 0.5ms (预创建)
  Python 解释器启动    | 200ms  | 50ms (pyc 预编译)
  依赖加载            | 300ms  | 0ms (venv 预热)
  ──────────────────|────────|────────
  总计               | ~507ms | ~57ms

class OverlayFSManager:
    """OverlayFS 优化：共享基础层"""

    def __init__(self, base_dir: str = "/var/fn-overlay"):
        self._base = base_dir

    def create_overlay(self, function_id: str,
                        base_layer: str) -> str:
        """创建 OverlayFS 挂载点"""
        upper = os.path.join(self._base, function_id, "upper")
        work = os.path.join(self._base, function_id, "work")
        merged = os.path.join(self._base, function_id, "merged")

        os.makedirs(upper, exist_ok=True)
        os.makedirs(work, exist_ok=True)
        os.makedirs(merged, exist_ok=True)

        subprocess.run([
            "mount", "-t", "overlay", "overlay",
            "-o", f"lowerdir={base_layer},upperdir={upper},workdir={work}",
            merged,
        ], check=True)

        return merged

    def cleanup_overlay(self, function_id: str) -> None:
        merged = os.path.join(self._base, function_id, "merged")
        subprocess.run(["umount", merged], check=False)

        import shutil
        shutil.rmtree(os.path.join(self._base, function_id), ignore_errors=True)

nsjail 沙箱监控面板：

  活跃沙箱: 23/50
  ┌───────────────────────────┐
  │ Python  ████████████ 15   │
  │ Node.js ████         5    │
  │ WASM    ██           3    │
  └───────────────────────────┘

  资源使用:
  ┌───────────────────────────┐
  │ Avg Memory: 45MB / 256MB  │
  │ Avg CPU:    120ms / 5000ms│
  │ OOM Kills:  2 (0.3%)      │
  │ Timeouts:   5 (0.8%)      │
  └───────────────────────────┘

  安全事件 (24h):
  ┌───────────────────────────┐
  │ seccomp violations:  3    │
  │ escape attempts:     0    │
  │ code scan warnings: 12    │
  └───────────────────────────┘

class HybridSandboxManager:
    """混合沙箱：nsjail + gVisor 按需选择"""

    def __init__(self, nsjail_config: NsjailConfig,
                 gvisor_config: dict):
        self._nsjail = NsjailExecutor(nsjail_config)
        self._gvisor_config = gvisor_config

    def select_sandbox(self, spec) -> str:
        """根据函数特征选择沙箱"""
        if spec.resource_limits.network_enabled:
            return "gvisor"  # 需要网络时用 gVisor

        if spec.resource_limits.max_execution_time_ms > 60000:
            return "gvisor"  # 长运行任务用 gVisor

        if spec.resource_limits.max_memory_mb > 512:
            return "gvisor"  # 高资源需求用 gVisor

        return "nsjail"  # 短生命周期函数用 nsjail

# 场景：在 nsjail 沙箱中安全执行用户自定义函数

# 1. 配置 nsjail
config = NsjailConfig(
    config_path="/etc/nsjail/python-sandbox.cfg",
    nsjail_binary="/usr/bin/nsjail",
)

# 2. 创建沙箱化运行时
runtime = SandboxedPythonRuntime(config)

# 3. 注册函数
spec = FunctionSpec(
    function_id="fn-risk-score",
    name="risk_scorer",
    language=FunctionLanguage.PYTHON,
    version="1.0.0",
    entry_point="calculate_risk",
    source_code="""
import math

def calculate_risk(credit_score, income, debt_ratio):
    base = credit_score / 850
    income_factor = math.log(income + 1) / math.log(1000001)
    risk = 1.0 - (base * 0.5 + income_factor * 0.3 + (1 - debt_ratio) * 0.2)
    return {"risk_score": round(risk, 4), "risk_level": "high" if risk > 0.7 else "low"}
""",
    dependencies=["numpy"],
    resource_limits=ResourceLimits(
        max_memory_mb=128,
        max_execution_time_ms=5000,
        network_enabled=False,
    ),
)

# 4. 安全检查
issues = EscapeDetector.validate_before_execution(spec)
assert len(issues) == 0, f"Security issues: {issues}"

# 5. 初始化并执行
await runtime.initialize(spec)
result = await runtime.execute(spec, {
    "credit_score": 650,
    "income": 200000,
    "debt_ratio": 0.45,
})
# result = {"risk_score": 0.3821, "risk_level": "low"}