用户代码 → nsjail 沙箱 → 宿主系统

┌─────────────────────────────────────────┐
│              nsjail 沙箱                 │
│  ┌─────────────────────────────────┐    │
│  │  用户 UDF 进程                   │    │
│  │  ├── PID Namespace (隔离进程)    │    │
│  │  ├── NET Namespace (隔离网络)    │    │
│  │  ├── MNT Namespace (隔离文件)    │    │
│  │  ├── USER Namespace (隔离用户)   │    │
│  │  ├── UTS Namespace (隔离主机名)  │    │
│  │  ├── IPC Namespace (隔离 IPC)    │    │
│  │  ├── seccomp-bpf (系统调用过滤)  │    │
│  │  └── cgroup (CPU/内存/IO 限制)   │    │
│  └─────────────────────────────────┘    │
│                                          │
│  只暴露：stdin/stdout/stderr + 返回值    │
└─────────────────────────────────────────┘

1. 解析配置（.cfg 文件或命令行参数）
2. 创建 Linux namespaces (clone(2))
3. 配置 cgroup 资源限制
4. 设置文件系统挂载点
5. 加载 seccomp-bpf 过滤器
6. 切换用户（drop privileges）
7. chroot/pivot_root 到沙箱根目录
8. 执行目标程序
9. 监控资源使用和超时
10. 收集退出状态和输出

// coomia-dip UDF 执行沙箱配置
name: "coomia-dip-udf-sandbox"

mode: ONCE  // 执行一次后退出

time_limit: 30        // 30 秒超时
max_cpus: 1           // 1 CPU 核心
rlimit_as: 512        // 最大虚拟内存 512MB
rlimit_cpu: 30        // CPU 时间限制 30 秒
rlimit_fsize: 10      // 最大文件大小 10MB
rlimit_nofile: 64     // 最大打开文件数 64
rlimit_nproc: 1       // 禁止创建子进程

clone_newnet: true     // 网络隔离
clone_newuser: true    // 用户隔离
clone_newns: true      // 文件系统隔离
clone_newpid: true     // PID 隔离
clone_newipc: true     // IPC 隔离
clone_newuts: true     // UTS 隔离
clone_newcgroup: true  // cgroup 隔离

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

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

mount {
    src: "/usr/lib/python3.12"
    dst: "/usr/lib/python3.12"
    is_bind: true
    rw: false
}

mount {
    src: "/tmp/udf-input"
    dst: "/input"
    is_bind: true
    rw: false
}

mount {
    src: "/tmp/udf-output"
    dst: "/output"
    is_bind: true
    rw: true
}

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

mount {
    dst: "/proc"
    fstype: "proc"
    rw: false
}

seccomp_string: "ALLOW { read, write, exit, exit_group, brk, mmap, munmap, mprotect, futex, clock_gettime, close, fstat, openat, lseek, ioctl, sigaltstack, rt_sigaction, rt_sigprocmask, gettid, getpid, getrandom, set_tid_address, set_robust_list, arch_prctl, prlimit64, sched_getaffinity, sched_yield, poll, epoll_create1, epoll_ctl, epoll_wait, pipe2, newfstatat, access, readlink } DEFAULT KILL"

exec_bin {
    path: "/usr/bin/python3"
    arg: "/input/user_function.py"
}

log_level: WARNING

宿主系统（真实 PID）：
  PID 1: systemd
  PID 1234: nsjail
  PID 1235: python3 (UDF)

UDF 视角（隔离 PID）：
  PID 1: python3  ← UDF 认为自己是 PID 1
  (看不到其他进程)

# coomia-dip UDF 允许的系统调用（最小化）
ALLOWED_SYSCALLS = {
    # 基础 I/O
    "read", "write", "close", "fstat", "lseek",
    # 内存管理
    "brk", "mmap", "munmap", "mprotect",
    # 进程控制（只读）
    "exit", "exit_group", "getpid", "gettid",
    # 信号处理
    "rt_sigaction", "rt_sigprocmask", "sigaltstack",
    # 时间
    "clock_gettime", "clock_getres",
    # 线程同步
    "futex", "sched_yield",
    # 文件打开（受限路径）
    "openat", "access", "newfstatat",
    # 随机数
    "getrandom",
    # Python 运行时需要
    "arch_prctl", "set_tid_address", "set_robust_list",
    "prlimit64", "sched_getaffinity",
}

# 禁止的关键系统调用
BLOCKED_SYSCALLS = {
    "execve",      # 禁止执行其他程序
    "fork", "vfork", "clone",  # 禁止创建进程
    "socket", "connect", "bind", "listen",  # 禁止网络
    "ptrace",      # 禁止调试其他进程
    "mount", "umount2",  # 禁止挂载
    "chroot", "pivot_root",  # 禁止改变根目录
    "setuid", "setgid",  # 禁止提权
}

# 当 UDF 尝试调用禁止的系统调用时
# nsjail 行为: KILL (发送 SIGSYS 信号，立即终止进程)

# coomia-dip 捕获并记录
class SandboxExecutor:
    async def execute(self, udf_code: str, input_data: dict) -> dict:
        result = await self._run_in_nsjail(udf_code, input_data)

        if result.exit_signal == signal.SIGSYS:
            # seccomp 违规
            await self.audit_log.record(
                event="SECCOMP_VIOLATION",
                udf_id=udf_code.id,
                user_id=udf_code.owner,
                severity="CRITICAL",
            )
            raise SecurityViolationError(
                "UDF attempted a forbidden system call"
            )

# coomia-dip UDF 资源限制配置
UDF_RESOURCE_LIMITS = {
    "cpu_time_seconds": 30,        # CPU 时间上限
    "wall_time_seconds": 60,       # 总时间上限（含 I/O 等待）
    "memory_mb": 512,              # 内存上限
    "disk_write_mb": 10,           # 磁盘写入上限
    "open_files": 64,              # 打开文件数上限
    "processes": 1,                # 进程数上限（禁止 fork）
    "cpu_cores": 1,                # CPU 核心数
}

# 按 UDF 类型分级
UDF_TIERS = {
    "basic": {
        "cpu_time_seconds": 10,
        "memory_mb": 128,
    },
    "standard": {
        "cpu_time_seconds": 30,
        "memory_mb": 512,
    },
    "premium": {
        "cpu_time_seconds": 120,
        "memory_mb": 2048,
    },
}

# nsjail 自动配置 cgroup v2
# /sys/fs/cgroup/coomia-dip-udf-{id}/

# CPU 限制
echo "100000 1000000" > cpu.max  # 10% CPU (100ms/1000ms)

# 内存限制
echo "536870912" > memory.max    # 512MB
echo "536870912" > memory.swap.max  # 禁止 swap

# PID 限制
echo "1" > pids.max              # 只允许 1 个进程

沙箱内文件系统布局：

/                          (tmpfs, 只读)
├── /usr/lib/python3.12/   (bind mount, 只读 — Python 运行时)
├── /usr/lib/node_modules/ (bind mount, 只读 — Node.js 运行时)
├── /input/                (bind mount, 只读 — 输入数据)
│   └── user_function.py   (用户代码)
├── /output/               (bind mount, 读写 — 输出目录)
├── /tmp/                  (tmpfs 50MB, 读写 — 临时文件)
└── /proc/                 (procfs, 只读 — 最小化)

不可见：
  /etc/          — 系统配置
  /home/         — 用户目录
  /var/          — 系统数据
  /root/         — root 目录
  宿主机其他所有路径

class SandboxIO:
    """沙箱 I/O 协议"""

    @staticmethod
    def prepare_input(udf_id: str, input_data: dict) -> str:
        """准备输入目录"""
        input_dir = f"/tmp/nsjail/input/{udf_id}"
        os.makedirs(input_dir, exist_ok=True)

        # 写入用户代码
        with open(f"{input_dir}/user_function.py", "w") as f:
            f.write(generate_wrapper(udf_id))

        # 写入输入数据（JSON）
        with open(f"{input_dir}/data.json", "w") as f:
            json.dump(input_data, f)

        return input_dir

    @staticmethod
    def collect_output(udf_id: str) -> dict:
        """收集输出结果"""
        output_dir = f"/tmp/nsjail/output/{udf_id}"
        result_file = f"{output_dir}/result.json"

        if os.path.exists(result_file):
            with open(result_file) as f:
                return json.load(f)
        raise UDFExecutionError("No output produced")

默认模式（NET Namespace）：
  UDF 进程无网络接口 → 无法连接任何外部服务

沙箱内 `ip addr` 输出：
  1: lo: <LOOPBACK> mtu 65536
      link/loopback 00:00:00:00:00:00
      (仅有 loopback，且未 UP)

# 某些 UDF 需要访问 coomia-dip 内部 API
# 通过 veth pair 和 iptables 实现受控网络

ALLOWED_ENDPOINTS = [
    "control-Layer:9090",   # gRPC
    "data-Layer:50051",     # Flight SQL
]

# nsjail 配置：
# macvlan_iface: "eth0"
# 配合 iptables 规则只允许访问特定 IP:Port

class FunctionRuntime:
    """coomia-dip 函数运行时"""

    def __init__(self):
        self.nsjail_binary = "/usr/bin/nsjail"
        self.config_template = load_config_template()

    async def execute_function(
        self,
        function_id: str,
        code: str,
        input_data: dict,
        timeout: int = 30,
        memory_mb: int = 512,
    ) -> FunctionResult:

        # 1. 准备沙箱
        sandbox_id = f"{function_id}-{uuid4().hex[:8]}"
        input_dir = SandboxIO.prepare_input(sandbox_id, input_data)
        output_dir = f"/tmp/nsjail/output/{sandbox_id}"
        os.makedirs(output_dir, exist_ok=True)

        # 2. 写入用户代码
        self._write_user_code(input_dir, code, input_data)

        # 3. 生成 nsjail 配置
        config = self._generate_config(
            sandbox_id, input_dir, output_dir, timeout, memory_mb
        )
        config_path = f"/tmp/nsjail/config/{sandbox_id}.cfg"
        with open(config_path, "w") as f:
            f.write(config)

        # 4. 执行 nsjail
        start_time = time.monotonic()
        process = await asyncio.create_subprocess_exec(
            self.nsjail_binary,
            "--config", config_path,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )

        try:
            stdout, stderr = await asyncio.wait_for(
                process.communicate(),
                timeout=timeout + 5,  # 额外 5 秒缓冲
            )
        except asyncio.TimeoutError:
            process.kill()
            raise FunctionTimeoutError(f"Function timed out after {timeout}s")

        duration = time.monotonic() - start_time

        # 5. 收集结果
        if process.returncode == 0:
            output = SandboxIO.collect_output(sandbox_id)
            return FunctionResult(
                status="success",
                output=output,
                duration_ms=duration * 1000,
                stdout=stdout.decode(),
            )
        else:
            return FunctionResult(
                status="error",
                error=stderr.decode(),
                duration_ms=duration * 1000,
                exit_code=process.returncode,
            )

        # 6. 清理
        finally:
            await self._cleanup(sandbox_id)

def generate_wrapper(function_code: str) -> str:
    """生成安全的代码包装器"""
    return f'''
import json
import sys

def main():
    # 读取输入
    with open("/input/data.json") as f:
        input_data = json.load(f)

    # 执行用户函数
    try:
        # 用户代码在受限的全局环境中执行
        user_globals = {{"__builtins__": __builtins__}}
        exec("""{function_code}""", user_globals)

        if "handler" not in user_globals:
            raise RuntimeError("Function must define a 'handler' function")

        result = user_globals["handler"](input_data)

        # 写入输出
        with open("/output/result.json", "w") as f:
            json.dump(result, f)

    except Exception as e:
        error = {{"error": str(e), "type": type(e).__name__}}
        with open("/output/result.json", "w") as f:
            json.dump(error, f)
        sys.exit(1)

if __name__ == "__main__":
    main()
'''

class SandboxAuditLogger:
    async def log_execution(self, event: SandboxEvent) -> None:
        await self.audit_service.record(
            event_type="UDF_EXECUTION",
            metadata={
                "function_id": event.function_id,
                "user_id": event.user_id,
                "world_id": event.world_id,
                "duration_ms": event.duration_ms,
                "memory_peak_mb": event.memory_peak_mb,
                "exit_code": event.exit_code,
                "seccomp_violations": event.seccomp_violations,
                "sandbox_id": event.sandbox_id,
            },
        )

SANDBOX_METRICS = {
    "udf_execution_total": Counter("总执行次数"),
    "udf_execution_duration": Histogram("执行耗时"),
    "udf_execution_errors": Counter("执行错误次数"),
    "udf_memory_peak": Histogram("内存峰值"),
    "udf_seccomp_violations": Counter("seccomp 违规次数"),
    "udf_timeout_total": Counter("超时次数"),
}