fix(chv): cell 后端由 nspawn 改为纯 stdlib userns 启动器

nspawn --directory=/ 在 systemd 255.4 上走不通（2026-11-07 真机
实测，修一个冒一个）：裸 --no-new-privileges 被当成要值吞掉
下一个参数；--directory=/ 被根目录检查挡住；--volatile=state
又要 /run/host/os-release 挂载点与空 /dev（--tmpfs=/dev 不
生效）。cell 从未 live 跑通过（阶段 1 只测参数表纯函数）。
bwrap 作 chv 被 AppArmor userns 门挡住，newuidmap 客机未装。

新后端（同文件、同 CLI、同调用方）：root 起进程，中间进程
unshare 用户 ns 并自己写 uid_map/gid_map（内 0 → 外 chv
1000），再进 mount/pid/net ns、挂 fresh /proc、/tmp tmpfs、
遮蔽敏感路径（目录盖 tmpfs、文件绑 /dev/null）、设
no_new_privileges + seccomp（禁 io_uring 三件套）+ 丢
capability，最后 exec payload；父进程配 veth/NAT、等退出码、
清 NAT。veth/NAT/nsenter 流程与旧版一致，seccomp BPF 与
capability 语义与 P2 设计一致。无新依赖、无 sysctl 改动。

chv-guest/chv-cell.py

@@ -1,27 +1,35 @@
 #!/usr/bin/env python3
-"""chv-cell：客机内 systemd-nspawn 运行时单元（Hatch 式）启动器。
+"""chv-cell：客机内 codex 运行时单元（Hatch 式）启动器。
 
 对齐 Muse P2（见 docs/chv-muse-p2-gap.md）：codex 跑在单元内，不再
-直接跑在客机用户态。单元是单命令容器（--as-pid2，不 boot 第二个
+直接跑在客机用户态。单元是单命令沙箱——单元即沙箱，单元内 codex
-systemd），单元即沙箱——单元内 codex 用 danger-full-access，不再
+用 danger-full-access，不再套 bwrap（bwrap 的 --proc 挂载在嵌套命名
-套 bwrap（bwrap 的 --proc 挂载在嵌套命名空间里走不通，见 AGENTS.md
+空间里走不通，见 AGENTS.md 内层 proc 定位）。
-内层 proc 定位）。
 
 关键映射（与 Muse“单元 root 映射为 host 侧非特权用户”同款）：
-  --private-users=1000:65536 —— 单元内 uid 0 → 客机侧 chv（uid 1000）。
+  单元内 uid 0 → 客机侧 chv（uid 1000）。
   效果：workspace / ~/.codex（chv 所有）单元内仍可写；host-root 0600
   文件（如 /etc/chv/guest-env）在单元内显示为 overflow uid、不可读；
-  再加 --inaccessible 遮蔽作纵深防御。
+  再加挂载遮蔽作纵深防御。
 
-网络：--network-veth。宿主侧 ve-chv-cell（10.200.0.1/24），单元侧
+网络：veth。宿主侧 ve-chv-cell（10.200.0.1/24），单元侧
 host0（10.200.0.2/24，默认路由经 10.200.0.1），客机侧 NAT 出站
 （MASQUERADE 走客机默认路由的上行口，动态发现）。单元内配网经
 nsenter 进单元 init 的 netns 做。veth 建不起来即大声失败，不静默
 回落共享网络；CHV_CELL_NETWORK=shared 是显式调试逃生口，须显式
 设置并打日志，绝不默认。
 
-环境透传：nspawn 不透传调用方环境，本脚本只转发白名单变量（模型
+环境透传：本脚本只转发白名单变量（模型网关与代理变量）。GITEA_TOKEN
-网关与代理变量）。GITEA_TOKEN 等凭证变量不在白名单里，绝不进单元。
+等凭证变量不在白名单里，绝不进单元。
+
+实现（2026-11-07 阶段 2 改写，见末尾“后端说明”）：
+  不再调 systemd-nspawn，改用纯标准库 userns 启动器：root 起进程，
+  子进程自己 unshare 用户命名空间并写 uid_map/gid_map（内 0 →
+  外 chv 1000），再进 mount/pid/net 命名空间、挂 fresh /proc、
+  遮蔽敏感路径、设 no_new_privileges + seccomp（禁 io_uring 三件套）
+  + 丢 capability，最后 exec payload。父进程在外配 veth/NAT、等
+  payload 退出码、清 NAT。只用标准库，无新依赖、无新包、无 sysctl
+  改动。
 
 用法（客机侧 root 运行，执行器经 SSH 调）：
   chv-cell run [--network veth|shared] -- <cmd...>
@@ -29,15 +37,32 @@ nsenter 进单元 init 的 netns 做。veth 建不起来即大声失败，不静
       #   /home/chv/.codex/host-state.json --codex-home /home/chv/.codex
       #   --sandbox danger-full-access dispatch --name t --task ...
   chv-cell argv [--network veth|shared] -- <cmd...>
-      # 只打印 nspawn 参数表（JSON），不启动
+      # 只打印单元规约（JSON：映射、挂载、网络、环境名），不启动。
+      # 环境值只印 <redacted>，绝不印真值。
   chv-cell network-commands [--uplink IFACE] [--cell-pid PID]
-      # 只打印配网命令表（JSON），不执行
+      # 只打印配网命令表（JSON），不执行。
-只用标准库。单元启动失败、配网失败一律非零退出，不静默降级。
+单元启动失败、配网失败一律非零退出，不静默降级。
+
+后端说明——为什么不用 nspawn（2026-11-07 真机实测，systemd 255.4）：
+  nspawn 的 --directory=/ 在本版 systemd 上走不通，修一个冒一个：
+  ① 裸 --no-new-privileges 被当成要值、吞掉下一个参数
+     （Failed to parse --no-new-privileges= argument）；
+  ② --directory=/ 被根目录检查挡住，须加 --volatile；
+  ③ --volatile=state 又要 /run/host/os-release 挂载点（须预建）
+     与空 /dev（--tmpfs=/dev 不生效，staging 根的 /dev 已填充）。
+  结论：nspawn-on-/ 在此 systemd 上不可行，cell 从未 live 跑通过
+  （阶段 1 只测了参数表纯函数，未 live 起过单元）。bwrap 作 chv
+  也被 Ubuntu 24.04 的 AppArmor userns 门（
+  apparmor_restrict_unprivileged_userns=1）挡住，newuidmap 客机
+  未装；本启动器由 root 写映射，不碰这两处限制。
 """
+import ctypes
 import json
 import os
 import pwd
 import re
+import select
+import struct
 import subprocess
 import sys
 import time
@@ -45,7 +70,6 @@ import time
 MACHINE = "chv-cell"
 # 单元映射：单元内 uid 0 → 客机侧 chv（uid 1000，非特权用户）。
 UID_BASE = 1000
-UID_COUNT = 65536
 CELL_USER = "chv"
 CELL_UID = 1000
 CELL_GID = 1000
@@ -61,8 +85,9 @@ CELL_IP = "10.200.0.2"
 HOST_IFACE = "ve-chv-cell"
 CELL_IFACE = "host0"
 
-# Muse 点名要丢的能力 + 同批丢的危险能力（nspawn 默认集已有限，
+# Muse 点名要丢的能力 + 同批丢的危险能力（单元内 root 只在
-# 这里是显式再丢，--drop-capability 逐项写，测试逐项断言）。
+# 自己的 userns 里有能力，host 侧本就没有；这里再从 bounding
+# 集里丢掉，防 setuid/提权路径，见 _drop_capabilities）。
 DROP_CAPABILITIES = [
     "CAP_SYS_ADMIN",
     "CAP_SYS_PTRACE",
@@ -74,13 +99,58 @@ DROP_CAPABILITIES = [
     "CAP_MKNOD",
 ]
 
+# capability 号（Linux cap 表，prctl PR_CAPBSET_DROP 用）。
+CAP_NUMBERS = {
+    "CAP_CHOWN": 0,
+    "CAP_DAC_OVERRIDE": 1,
+    "CAP_DAC_READ_SEARCH": 2,
+    "CAP_FOWNER": 3,
+    "CAP_FSETID": 4,
+    "CAP_KILL": 5,
+    "CAP_SETGID": 6,
+    "CAP_SETUID": 7,
+    "CAP_SETPCAP": 8,
+    "CAP_LINUX_IMMUTABLE": 9,
+    "CAP_NET_BIND_SERVICE": 10,
+    "CAP_NET_BROADCAST": 11,
+    "CAP_NET_ADMIN": 12,
+    "CAP_NET_RAW": 13,
+    "CAP_IPC_LOCK": 14,
+    "CAP_IPC_OWNER": 15,
+    "CAP_SYS_MODULE": 16,
+    "CAP_SYS_RAWIO": 17,
+    "CAP_SYS_CHROOT": 18,
+    "CAP_SYS_PTRACE": 19,
+    "CAP_SYS_PACCT": 20,
+    "CAP_SYS_ADMIN": 21,
+    "CAP_SYS_BOOT": 22,
+    "CAP_SYS_NICE": 23,
+    "CAP_SYS_RESOURCE": 24,
+    "CAP_SYS_TIME": 25,
+    "CAP_SYS_TTY_CONFIG": 26,
+    "CAP_MKNOD": 27,
+    "CAP_LEASE": 28,
+    "CAP_AUDIT_WRITE": 29,
+    "CAP_AUDIT_CONTROL": 30,
+    "CAP_SETFCAP": 31,
+}
+
 # syscall 过滤：@system-service 集合再扣掉 io_uring 三件套
-#（Muse 点名禁用 io_uring）。
+#（Muse 点名禁用 io_uring）。新后端用 seccomp BPF 实现
+#（见 build_seccomp_filter），语义与旧 nspawn 参数表一致。
 SYSCALL_FILTER = ("@system-service ~io_uring_setup "
                   "~io_uring_enter ~io_uring_register")
 
+# x86_64 调用号（BPF 按号拦，只认 x86_64；其它架构一律放行，
+# 见 build_seccomp_filter 的 arch 检查）。
+SYSCALL_NUMBERS_X86_64 = {
+    "io_uring_setup": 425,
+    "io_uring_enter": 426,
+    "io_uring_register": 427,
+}
+
 # 单元内遮掉的路径：凭证与浏览器（单元内跑不起自己的浏览器，
-# 只能走 broker 的 chv-browser）。
+# 只能走 broker 的 chv-browser）。目录盖 tmpfs，文件绑 /dev/null。
 INACCESSIBLE_BASE = [
     "/etc/chv",
     "/home/chv/.git-credentials",
@@ -101,6 +171,32 @@ FORWARD_ENV_ALLOWLIST = (
     "no_proxy",
 )
 
+# unshare 标志（os.CLONE_*，Python 3.12+；老版本回落到数值）。
+CLONE_NEWNS = getattr(os, "CLONE_NEWNS", 0x00020000)
+CLONE_NEWUSER = getattr(os, "CLONE_NEWUSER", 0x10000000)
+CLONE_NEWPID = getattr(os, "CLONE_NEWPID", 0x20000000)
+CLONE_NEWNET = getattr(os, "CLONE_NEWNET", 0x40000000)
+
+# prctl 号。
+_PR_SET_NO_NEW_PRIVILEGES = 38
+_PR_SET_SECCOMP = 22
+_PR_CAPBSET_DROP = 24
+_PR_SET_PDEATHSIG = 1
+_SECCOMP_MODE_FILTER = 2
+_SECCOMP_RET_ERRNO = 0x00050000
+_SECCOMP_RET_ALLOW = 0x7fff0000
+_AUDIT_ARCH_X86_64 = 0xC000003E
+_EPERM = 1
+
+_libc_cached = None
+
+
+def _libc():
+    global _libc_cached
+    if _libc_cached is None:
+        _libc_cached = ctypes.CDLL("libc.so.6", use_errno=True)
+    return _libc_cached
+
 
 def _playwright_package_dir():
     """playwright pip 包目录（遮蔽用）。算不出时回落到 noble 默认路径。"""
@@ -114,12 +210,26 @@ def _playwright_package_dir():
     return "/usr/local/lib/python3.12/dist-packages/playwright"
 
 
-def build_nspawn_argv(command, network="veth",
+def _forwarded_env():
+    """进单元的环境：固定两项 + 白名单（只取当前进程环境里有的）。
+    真值只在 exec 时用；打印规约时一律 <redacted>。"""
+    env = {"CHV_CELL": "1", "HOME": "/home/chv",
+           "PATH": os.environ.get(
+               "PATH", "/usr/local/bin:/usr/bin:/bin")}
+    for name in FORWARD_ENV_ALLOWLIST:
+        value = os.environ.get(name)
+        if value:
+            env[name] = value
+    return env
+
+
+def build_sandbox_spec(command, network="veth",
                        workspace=WORKSPACE, codex_home=CODEX_HOME,
                        broker_dir=BROKER_DIR,
                        playwright_package_dir=None):
-    """产 nspawn 参数表（纯函数，可单测）。command 是单元内要跑的命令
+    """单元规约（纯函数，可单测）。command 是单元内要跑的命令
-   （list，如 ["python3", "/app/host.py", ...]）。"""
+    （list，如 ["python3", "/app/host.py", ...]）。只描述“会怎么
+    跑”，不启动；环境值在规约里一律 <redacted>。"""
     if not command:
         raise ValueError("chv-cell: 单元内命令不能为空")
     if network not in ("veth", "shared"):
@@ -127,38 +237,142 @@ def build_nspawn_argv(command, network="veth",
                          % network)
     if playwright_package_dir is None:
         playwright_package_dir = _playwright_package_dir()
-    argv = [
+    env = _forwarded_env()
-        "systemd-nspawn", "--quiet",
+    display_env = {k: ("<redacted>" if k in FORWARD_ENV_ALLOWLIST
-        "--machine=%s" % MACHINE,
+                       else v) for k, v in env.items()}
-        "--directory=/",
+    return {
-        "--as-pid2",
+        "backend": "userns",
-        "--private-users=%d:%d" % (UID_BASE, UID_COUNT),
+        "cell": MACHINE,
-        "--no-new-privileges",
+        "uid_map": "0 %d 1" % CELL_UID,
-        "--system-call-filter=%s" % SYSCALL_FILTER,
+        "gid_map": "0 %d 1" % CELL_GID,
+        "command": list(command),
+        "chdir": workspace,
+        "workspace": workspace,
+        "codex_home": codex_home,
+        "broker_dir": broker_dir,
+        "env": display_env,
+        "hide": list(INACCESSIBLE_BASE) + [playwright_package_dir],
+        "tmpfs": ["/tmp"],
+        "network": network,
+        "host_ip": HOST_IP,
+        "cell_ip": CELL_IP,
+        "host_iface": HOST_IFACE,
+        "cell_iface": CELL_IFACE,
+        "capabilities_dropped": list(DROP_CAPABILITIES),
+        "syscall_filter": SYSCALL_FILTER,
+        "no_new_privileges": True,
+    }
+
+
+def build_seccomp_filter():
+    """禁 io_uring 三件套的 seccomp BPF（经典 BPF 字节串）。
+    只认 x86_64：arch 不符一律放行（读不到正确的调用号，不乱拦）；
+    命中三件套返回 EPERM，其余放行。共 10 条指令。"""
+    def stmt(code, jt, jf, k):
+        return struct.pack("HBBI", code, jt, jf, k)
+    bpf_ld = 0x20    # BPF_LD | BPF_W | BPF_ABS
+    bpf_jeq = 0x15   # BPF_JMP | BPF_JEQ | BPF_K
+    bpf_ret = 0x06   # BPF_RET | BPF_K（立即数返回；0x16 是 RET|A，
+                     # 会把 A 寄存器的调用号当返回值解码成 KILL）
+    deny = _SECCOMP_RET_ERRNO | _EPERM
+    insns = [
+        stmt(bpf_ld, 0, 0, 4),                    # [0] arch
+        stmt(bpf_jeq, 0, 7, _AUDIT_ARCH_X86_64),  # [1] 非 x86_64 跳去放行
+        stmt(bpf_ld, 0, 0, 0),                    # [2] nr
+        stmt(bpf_jeq, 0, 1,
+             SYSCALL_NUMBERS_X86_64["io_uring_setup"]),  # [3]
+        stmt(bpf_ret, 0, 0, deny),                # [4] 命中 setup → EPERM
+        stmt(bpf_jeq, 0, 1,
+             SYSCALL_NUMBERS_X86_64["io_uring_enter"]),  # [5]
+        stmt(bpf_ret, 0, 0, deny),                # [6] 命中 enter → EPERM
+        stmt(bpf_jeq, 0, 1,
+             SYSCALL_NUMBERS_X86_64["io_uring_register"]),  # [7]
+        stmt(bpf_ret, 0, 0, deny),                # [8] 命中 register → EPERM
+        stmt(bpf_ret, 0, 0, _SECCOMP_RET_ALLOW),   # [9] 其余放行
     ]
-    for cap in DROP_CAPABILITIES:
+    return b"".join(insns)
-        argv.append("--drop-capability=%s" % cap)
+
+
+def _apply_seccomp_filter():
+    """在当前进程装 seccomp 过滤（须已设 no_new_privileges，
+    或在自有 userns 里有 CAP_SYS_ADMIN）。失败即抛错——
+    拦不住 io_uring 的单元不许跑 payload，大声失败。"""
+    prog = build_seccomp_filter()
+
+    class SockFprog(ctypes.Structure):
+        _fields_ = [("len", ctypes.c_ushort),
+                    ("filter", ctypes.c_void_p)]
+
+    buf = ctypes.create_string_buffer(prog, len(prog))
+    fprog = SockFprog(len(prog) // 8, ctypes.cast(buf, ctypes.c_void_p))
+    ret = _libc().prctl(_PR_SET_SECCOMP, _SECCOMP_MODE_FILTER,
+                        ctypes.byref(fprog), 0, 0)
+    if ret != 0:
+        err = ctypes.get_errno()
+        raise RuntimeError(
+            "chv-cell: 装 seccomp 过滤失败（errno=%d）" % err)
+
+
+def _set_no_new_privileges():
+    """prctl PR_SET_NO_NEW_PRIVILEGES。失败即抛错，不许跑 payload。"""
+    ret = _libc().prctl(_PR_SET_NO_NEW_PRIVILEGES, 1, 0, 0, 0)
+    if ret != 0:
+        err = ctypes.get_errno()
+        raise RuntimeError(
+            "chv-cell: 设 no_new_privileges 失败（errno=%d）" % err)
+
+
+def _drop_capabilities():
+    """从 bounding 集丢掉 DROP_CAPABILITIES。失败即抛错。"""
+    for name in DROP_CAPABILITIES:
+        num = CAP_NUMBERS[name]
+        ret = _libc().prctl(_PR_CAPBSET_DROP, num, 0, 0, 0)
+        if ret != 0:
+            err = ctypes.get_errno()
+            raise RuntimeError(
+                "chv-cell: 丢 %s 失败（errno=%d）" % (name, err))
+
+
+def _write_id_maps():
+    """写当前进程的 uid_map/gid_map：内 0 → 外 chv（1000）。
+    须在 unshare(CLONE_NEWUSER) 之后、其它 unshare 之前；
+    gid_map 之前须先写 setgroups=deny。调用方须是 root（或
+    在父 userns 里有 CAP_SETUID），否则映射写不进。"""
+    with open("/proc/self/setgroups", "w") as fh:
+        fh.write("deny")
+    with open("/proc/self/uid_map", "w") as fh:
+        fh.write("0 %d 1" % CELL_UID)
+    with open("/proc/self/gid_map", "w") as fh:
+        fh.write("0 %d 1" % CELL_GID)
+
+
+def _mount(args, what):
+    """调 mount 二进制做一次挂载。失败即抛 RuntimeError。"""
+    proc = subprocess.run(["mount"] + args,
+                          capture_output=True, text=True)
+    if proc.returncode != 0:
+        raise RuntimeError(
+            "chv-cell: 挂载失败 %s：%s"
+            % (what, (proc.stderr or proc.stdout or "").strip()))
+
+
+def _setup_mounts(playwright_package_dir=None):
+    """单元内挂载：根 rprivate、fresh /proc、/tmp tmpfs、
+    敏感路径遮蔽（目录盖 tmpfs，文件绑 /dev/null）。
+    只在新 mount ns 里做，不动宿主挂载表。"""
+    if playwright_package_dir is None:
+        playwright_package_dir = _playwright_package_dir()
+    _mount(["--make-rprivate", "/"], "根 rprivate")
+    _mount(["-t", "proc", "proc", "/proc"], "fresh /proc")
+    _mount(["-t", "tmpfs", "tmpfs", "/tmp"], "/tmp tmpfs")
     for path in INACCESSIBLE_BASE + [playwright_package_dir]:
-        argv.append("--inaccessible=%s" % path)
+        if os.path.isdir(path):
-    argv += [
+            _mount(["-t", "tmpfs", "tmpfs", path], "遮蔽 %s" % path)
-        "--bind=%s" % workspace,
+        elif os.path.exists(path):
-        "--bind=%s" % codex_home,
+            _mount(["--bind", "/dev/null", path], "遮蔽 %s" % path)
-        "--bind=%s" % broker_dir,
+        else:
-        "--tmpfs=/tmp",
+            print("chv-cell: 遮蔽目标不存在，跳过：%s" % path,
-    ]
+                  file=sys.stderr)
-    if network == "veth":
-        argv.append("--network-veth")
-    argv += [
-        "--setenv=CHV_CELL=1",
-        "--setenv=HOME=/home/chv",
-    ]
-    for name in FORWARD_ENV_ALLOWLIST:
-        value = os.environ.get(name)
-        if value:
-            argv.append("--setenv=%s=%s" % (name, value))
-    argv += ["--chdir=%s" % workspace]
-    argv += list(command)
-    return argv
 
 
 def parse_default_route_iface(ip_route_output):
@@ -184,12 +398,22 @@ def find_uplink_iface(run=None):
     return parse_default_route_iface(proc.stdout)
 
 
-def host_network_commands(uplink_iface):
+def host_network_commands(uplink_iface, cell_pid=None):
-    """客机侧配网命令表（纯函数）：ve 接口起 IP + 转发 + NAT。
+    """客机侧配网命令表（纯函数）：建 veth 对、把 host0 移进
+    单元 netns、ve 接口起 IP + 转发 + NAT。cell_pid 为空时
+    只返回后半段（ve 起 IP + 转发 + NAT，兼容旧调用与单测）；
+    veth 模式调用方须传 cell_pid，建对与移口才完整。
     调用方按顺序执行，任一失败即停。"""
     if not uplink_iface:
         raise ValueError("chv-cell: 上行口名为空，配不出 NAT")
-    return [
+    cmds = []
+    if cell_pid:
+        cmds += [
+            ["ip", "link", "add", HOST_IFACE, "type", "veth",
+             "peer", "name", CELL_IFACE],
+            ["ip", "link", "set", CELL_IFACE, "netns", str(cell_pid)],
+        ]
+    cmds += [
         ["ip", "addr", "add", "%s/24" % HOST_IP, "dev", HOST_IFACE],
         ["ip", "link", "set", HOST_IFACE, "up"],
         ["sysctl", "-w", "net.ipv4.ip_forward=1"],
@@ -201,11 +425,13 @@ def host_network_commands(uplink_iface):
          "-o", HOST_IFACE, "-m", "state", "--state",
          "RELATED,ESTABLISHED", "-j", "ACCEPT"],
     ]
+    return cmds
 
 
 def host_network_cleanup_commands(uplink_iface):
-    """配网清理命令表（与 host_network_commands 配对，-A 改 -D）。
+    """配网清理命令表（与 host_network_commands 配对，-A 改 -D，
-    清理是 best-effort：删不掉（规则已不在）不报错。"""
+    再删 ve 接口）。清理是 best-effort：删不掉（规则已不在）
+    不报错。"""
     if not uplink_iface:
         raise ValueError("chv-cell: 上行口名为空，清不了 NAT")
     cmds = []
@@ -214,29 +440,10 @@ def host_network_cleanup_commands(uplink_iface):
             continue
         cleaned = [c if c != "-A" else "-D" for c in cmd]
         cmds.append(cleaned)
+    cmds.append(["ip", "link", "del", HOST_IFACE])
     return cmds
 
 
-def find_cell_init_pid(nspawn_pid, proc_root="/proc"):
-    """找单元 init 的 pid：nspawn 进程的第一个子进程
-   （--as-pid2 时它是 stub init，payload 是 PID 2；两者同 netns）。
-    找不到返回 None。"""
-    try:
-        task_dir = os.path.join(proc_root, str(nspawn_pid), "task")
-        for tid in os.listdir(task_dir):
-            children_path = os.path.join(task_dir, tid, "children")
-            try:
-                with open(children_path) as fh:
-                    children = fh.read().split()
-            except OSError:
-                continue
-            if children:
-                return int(children[0])
-    except (OSError, ValueError):
-        return None
-    return None
-
-
 def cell_network_commands(cell_pid):
     """单元内配网命令表（纯函数）：nsenter 进单元 init 的 netns，
     给 host0 配 IP、起接口、加默认路由。"""
@@ -285,39 +492,236 @@ def _run_cmds(cmds, what):
                 % (what, proc.returncode, " ".join(cmd)))
 
 
+def _read_line(fd, timeout=10.0):
+    """带超时的读一行（select）。timeout=None 一直等。超时或
+    EOF 返回 None。"""
+    buf = b""
+    deadline = None if timeout is None else time.time() + timeout
+    while True:
+        if deadline is None:
+            r, _, _ = select.select([fd], [], [])
+        else:
+            left = deadline - time.time()
+            if left <= 0:
+                return None
+            r, _, _ = select.select([fd], [], [], left)
+        if not r:
+            return None
+        chunk = os.read(fd, 4096)
+        if not chunk:
+            return buf.decode(errors="replace") if buf else None
+        buf += chunk
+        if b"\n" in buf:
+            line, _ = buf.split(b"\n", 1)
+            return line.decode(errors="replace")
+
+
+def _set_pdeathsig(sig):
+    """父进程死即给自己发 sig（防孤儿单元）。设完核对父 pid：
+    若父已死（ppid==1），直接退出，不留孤儿。"""
+    libc = _libc()
+    libc.prctl(_PR_SET_PDEATHSIG, sig, 0, 0, 0)
+    if os.getppid() == 1:
+        os._exit(1)
+
+
+def _grandchild_main(command, network, ready_w, go_r):
+    """单元 init（新 pid ns 里的 pid 1）：挂载、加固、报 READY、
+    等放行、exec payload。任何一步失败写 ERR 后退出（父进程
+    收到 ERR 即大声失败，不跑 payload）。"""
+    try:
+        _set_pdeathsig(9)  # SIGKILL：中间进程死即自杀，不留孤儿
+        _setup_mounts()
+        _set_no_new_privileges()
+        _apply_seccomp_filter()
+        _drop_capabilities()
+    except Exception as exc:  # noqa: BLE001 —— 全部转 ERR 上报
+        try:
+            os.write(ready_w, ("ERR %s\n" % exc).encode())
+        except OSError:
+            pass
+        os._exit(1)
+    try:
+        os.write(ready_w, b"READY\n")
+    except OSError:
+        os._exit(1)
+    # 等父进程配完网（shared 模式父进程立刻放行）。
+    try:
+        got = os.read(go_r, 1)
+    except OSError:
+        os._exit(1)
+    if not got:
+        os._exit(1)
+    try:
+        os.chdir(WORKSPACE)
+    except OSError as exc:
+        print("chv-cell: chdir %s 失败：%s" % (WORKSPACE, exc),
+              file=sys.stderr)
+        os._exit(1)
+    env = _forwarded_env()
+    try:
+        os.execvpe(command[0], list(command), env)
+    except OSError as exc:
+        print("chv-cell: exec %s 失败：%s" % (command[0], exc),
+              file=sys.stderr)
+        os._exit(127)
+
+
+def _intermediate_main(command, network, pid_w, ready_r, ready_w,
+                       go_r, status_w):
+    """中间进程：建 userns（写映射）与 mount/net/pid ns，fork
+    出单元 init，报 pid、等它退出、报退出码。自己绝不 exec，
+    只做命名空间搭建与收割。"""
+    try:
+        _set_pdeathsig(15)  # SIGTERM：父进程死即退出，带走单元
+        os.unshare(CLONE_NEWUSER)
+        _write_id_maps()
+        os.unshare(CLONE_NEWNS)
+        if network == "veth":
+            os.unshare(CLONE_NEWNET)
+        os.unshare(CLONE_NEWPID)
+        grandchild_pid = os.fork()
+        if grandchild_pid == 0:
+            # 单元 init：关掉不用的管道端，进沙箱。
+            for fd in (pid_w, status_w):
+                try:
+                    os.close(fd)
+                except OSError:
+                    pass
+            _grandchild_main(command, network, ready_w, go_r)
+            os._exit(1)
+        # 中间进程：关掉孙进程用的端，只留 pid 上报与收割。
+        for fd in (ready_w, go_r, ready_r):
+            try:
+                os.close(fd)
+            except OSError:
+                pass
+        try:
+            os.write(pid_w, ("%d\n" % grandchild_pid).encode())
+        except OSError:
+            pass
+        try:
+            os.close(pid_w)
+        except OSError:
+            pass
+        _, status = os.waitpid(grandchild_pid, 0)
+        code = os.waitstatus_to_exitcode(status)
+        try:
+            os.write(status_w, ("rc=%d\n" % code).encode())
+        except OSError:
+            pass
+    except Exception as exc:  # noqa: BLE001 —— 全部转状态上报
+        print("chv-cell: 单元搭建失败：%s" % exc, file=sys.stderr)
+        try:
+            os.write(status_w, ("ERR %s\n" % exc).encode())
+        except OSError:
+            pass
+    try:
+        os.close(status_w)
+    except OSError:
+        pass
+    os._exit(0)
+
+
 def run_cell(command, network="veth"):
     """起单元、配网、等 payload 退出、清 NAT，返回 payload 退出码。"""
     if os.geteuid() != 0:
-        print("chv-cell: 须以 root 运行（nspawn + 配网都要 root）",
+        print("chv-cell: 须以 root 运行（userns + 配网都要 root）",
               file=sys.stderr)
         return 2
     _ensure_bind_source(WORKSPACE)
     _ensure_bind_source(CODEX_HOME)
     _ensure_bind_source(BROKER_DIR, uid=0, gid=0)
-    argv = build_nspawn_argv(command, network=network)
+    spec = build_sandbox_spec(command, network=network)
-    print("chv-cell: 启动单元：%s" % " ".join(argv), file=sys.stderr)
+    print("chv-cell: 启动单元：%s"
-    proc = subprocess.Popen(argv)
+          % json.dumps(spec, ensure_ascii=False), file=sys.stderr)
-    uplink = None
+    pid_r, pid_w = os.pipe()
+    ready_r, ready_w = os.pipe()
+    go_r, go_w = os.pipe()
+    status_r, status_w = os.pipe()
     cell_pid = None
+    uplink = None
     try:
+        mid_pid = os.fork()
+        if mid_pid == 0:
+            # 中间进程：关掉父进程用的端。
+            for fd in (pid_r, ready_r, go_w, status_r):
+                try:
+                    os.close(fd)
+                except OSError:
+                    pass
+            _intermediate_main(command, network, pid_w, ready_r,
+                               ready_w, go_r, status_w)
+            os._exit(0)
+        # 父进程：关掉子进程用的端。
+        for fd in (pid_w, ready_w, go_r, status_w):
+            try:
+                os.close(fd)
+            except OSError:
+                pass
+        line = _read_line(pid_r, timeout=15.0)
+        if line is None:
+            raise RuntimeError("chv-cell: 等单元 pid 超时或中间进程已死")
+        try:
+            cell_pid = int(line.strip())
+        except ValueError:
+            raise RuntimeError("chv-cell: 单元 pid 非法：%r" % line)
+        with open(CELL_PID_FILE, "w") as fh:
+            fh.write(str(cell_pid))
+        line = _read_line(ready_r, timeout=15.0)
+        if line is None:
+            raise RuntimeError("chv-cell: 等单元 READY 超时")
+        if line.startswith("ERR "):
+            raise RuntimeError("chv-cell: 单元搭建失败：%s"
+                               % line[4:].strip())
+        if line.strip() != "READY":
+            raise RuntimeError("chv-cell: 单元 READY 非法：%r" % line)
         if network == "veth":
-            if not _wait_for_iface(HOST_IFACE):
+            # 先清掉上次泄漏的 ve 接口（父进程被 SIGKILL 时
-                raise RuntimeError(
+            # finally 跑不到，接口会残留，建对前先删）。
-                    "chv-cell: veth 接口 %s 未出现，拒绝回落共享网络"
+            subprocess.run(["ip", "link", "del", HOST_IFACE],
-                    % HOST_IFACE)
+                           capture_output=True)
             uplink = find_uplink_iface()
             if not uplink:
                 raise RuntimeError(
                     "chv-cell: 客机无默认路由，配不出 NAT")
-            _run_cmds(host_network_commands(uplink), "客机侧配网")
+            _run_cmds(host_network_commands(uplink, cell_pid),
-            cell_pid = find_cell_init_pid(proc.pid)
+                      "客机侧配网")
-            if not cell_pid:
+            if not _wait_for_iface(HOST_IFACE):
-                raise RuntimeError("chv-cell: 找不到单元 init pid")
+                raise RuntimeError(
-            with open(CELL_PID_FILE, "w") as fh:
+                    "chv-cell: veth 接口 %s 未出现，拒绝回落共享网络"
-                fh.write(str(cell_pid))
+                    % HOST_IFACE)
             _run_cmds(cell_network_commands(cell_pid), "单元内配网")
-        return proc.wait()
+        # 放行单元 init 去 exec payload。
+        try:
+            os.write(go_w, b"1")
+        except OSError as exc:
+            raise RuntimeError("chv-cell: 放行单元失败：%s" % exc)
+        line = _read_line(status_r, timeout=None)
+        if line is None:
+            raise RuntimeError("chv-cell: 单元异常退出，未报退出码")
+        if line.startswith("ERR "):
+            raise RuntimeError("chv-cell: 单元搭建失败：%s"
+                               % line[4:].strip())
+        m = re.match(r"rc=(-?\d+)", line.strip())
+        if not m:
+            raise RuntimeError("chv-cell: 退出码非法：%r" % line)
+        return int(m.group(1))
     finally:
+        try:
+            os.close(go_w)
+        except OSError:
+            pass
+        for fd in (pid_r, ready_r, status_r):
+            try:
+                os.close(fd)
+            except OSError:
+                pass
+        if cell_pid:
+            try:
+                os.kill(cell_pid, 9)
+            except OSError:
+                pass
         if uplink:
             for cmd in host_network_cleanup_commands(uplink):
                 subprocess.run(cmd, capture_output=True)
@@ -326,12 +730,6 @@ def run_cell(command, network="veth"):
                 os.remove(CELL_PID_FILE)
         except OSError:
             pass
-        if proc.poll() is None:
-            proc.terminate()
-            try:
-                proc.wait(timeout=10)
-            except subprocess.TimeoutExpired:
-                proc.kill()
 
 
 def main(argv):
@@ -375,7 +773,7 @@ def main(argv):
             print("chv-cell: argv 用法：chv-cell argv -- <cmd...>",
                   file=sys.stderr)
             return 2
-        print(json.dumps(build_nspawn_argv(rest[1:], network=network),
+        print(json.dumps(build_sandbox_spec(rest[1:], network=network),
                          ensure_ascii=False))
         return 0
     if verb == "network-commands":
@@ -395,7 +793,7 @@ def main(argv):
                 return 2
         out = {}
         if uplink:
-            out["host"] = host_network_commands(uplink)
+            out["host"] = host_network_commands(uplink, cell_pid)
             out["host_cleanup"] = host_network_cleanup_commands(uplink)
         if cell_pid:
             out["cell"] = cell_network_commands(cell_pid)

cloud/tests/test_chv_muse_p2.py

@@ -1,9 +1,10 @@
-"""对齐 Muse P2 测试：客机 nspawn 单元 + 浏览器子代理独立。
+"""对齐 Muse P2 测试：客机 userns 单元 + 浏览器子代理独立。
 
 只测本轨交付物（chv-guest/chv-cell.py、chv-broker.py、chv-browser.py、
 build.sh 干跑、buildinfo.py 内容项）。全部用假 socket/假进程/假
 playwright，绝不连真实网络、不起真实 VM、不改生产行为。
 """
+import ctypes
 import importlib.util
 import json
 import os
@@ -38,63 +39,58 @@ BUILDINFO = _load_script("buildinfo.py")
 
 class CellArgvTest(unittest.TestCase):
     def test_private_users_maps_cell_root_to_chv(self):
-        argv = CELL.build_nspawn_argv(["true"])
+        spec = CELL.build_sandbox_spec(["true"])
-        self.assertIn("--private-users=1000:65536", argv)
+        self.assertEqual(spec["backend"], "userns")
-        self.assertIn("--machine=chv-cell", argv)
+        self.assertEqual(spec["uid_map"], "0 1000 1")
-        self.assertIn("--directory=/", argv)
+        self.assertEqual(spec["gid_map"], "0 1000 1")
-        self.assertIn("--as-pid2", argv)
+        self.assertEqual(spec["cell"], "chv-cell")
-        self.assertIn("--no-new-privileges", argv)
+        self.assertTrue(spec["no_new_privileges"])
 
     def test_muse_named_capabilities_dropped(self):
-        argv = CELL.build_nspawn_argv(["true"])
+        spec = CELL.build_sandbox_spec(["true"])
         for cap in ("CAP_SYS_PTRACE", "CAP_NET_ADMIN", "CAP_SYS_ADMIN",
                     "CAP_NET_RAW", "CAP_SYS_MODULE"):
-            self.assertIn("--drop-capability=%s" % cap, argv)
+            self.assertIn(cap, spec["capabilities_dropped"])
-        # 绝不能出现“保留这些能力”的写法。
-        for token in argv:
-            self.assertFalse(token.startswith("--capability=CAP_SYS_PTRACE"))
-            self.assertFalse(token.startswith("--capability=CAP_NET_ADMIN"))
 
     def test_io_uring_blocked_in_syscall_filter(self):
-        argv = CELL.build_nspawn_argv(["true"])
+        spec = CELL.build_sandbox_spec(["true"])
-        filt = [a for a in argv if a.startswith("--system-call-filter=")]
+        filt = spec["syscall_filter"]
-        self.assertEqual(len(filt), 1)
+        self.assertIn("@system-service", filt)
-        self.assertIn("@system-service", filt[0])
         for call in ("io_uring_setup", "io_uring_enter",
                      "io_uring_register"):
-            self.assertIn("~%s" % call, filt[0])
+            self.assertIn("~%s" % call, filt)
 
     def test_secrets_and_browser_hidden_from_cell(self):
-        argv = CELL.build_nspawn_argv(
+        spec = CELL.build_sandbox_spec(
             ["true"],
             playwright_package_dir="/usr/local/lib/python3.12/dist-packages/playwright")
-        hidden = [a.split("=", 1)[1] for a in argv
-                  if a.startswith("--inaccessible=")]
         for path in ("/etc/chv", "/home/chv/.git-credentials",
                      "/usr/local/share/ms-playwright",
                      "/usr/local/lib/python3.12/dist-packages/playwright"):
-            self.assertIn(path, hidden)
+            self.assertIn(path, spec["hide"])
 
     def test_writable_binds_and_tmpfs(self):
-        argv = CELL.build_nspawn_argv(["true"])
+        spec = CELL.build_sandbox_spec(["true"])
-        self.assertIn("--bind=/home/chv/workspace", argv)
+        # 新后端不 chroot：根就是客机根，workspace/.codex/broker
-        self.assertIn("--bind=/home/chv/.codex", argv)
+        # 直接就是同名目录（单元内 root 即 chv，属主不变）。
-        self.assertIn("--bind=/run/chv", argv)
+        self.assertEqual(spec["workspace"], "/home/chv/workspace")
-        self.assertIn("--tmpfs=/tmp", argv)
+        self.assertEqual(spec["codex_home"], "/home/chv/.codex")
-        self.assertIn("--chdir=/home/chv/workspace", argv)
+        self.assertEqual(spec["broker_dir"], "/run/chv")
-        self.assertIn("--setenv=CHV_CELL=1", argv)
+        self.assertEqual(spec["tmpfs"], ["/tmp"])
-        self.assertIn("--setenv=HOME=/home/chv", argv)
+        self.assertEqual(spec["chdir"], "/home/chv/workspace")
+        self.assertEqual(spec["env"]["CHV_CELL"], "1")
+        self.assertEqual(spec["env"]["HOME"], "/home/chv")
 
     def test_veth_default_and_shared_opt_in(self):
-        self.assertIn("--network-veth",
+        self.assertEqual(
-                      CELL.build_nspawn_argv(["true"]))
+            CELL.build_sandbox_spec(["true"])["network"], "veth")
-        self.assertNotIn("--network-veth",
+        self.assertEqual(
-                         CELL.build_nspawn_argv(["true"],
+            CELL.build_sandbox_spec(["true"], network="shared")["network"],
-                                                network="shared"))
+            "shared")
         with self.assertRaises(ValueError):
-            CELL.build_nspawn_argv(["true"], network="macvlan")
+            CELL.build_sandbox_spec(["true"], network="macvlan")
         with self.assertRaises(ValueError):
-            CELL.build_nspawn_argv([])
+            CELL.build_sandbox_spec([])
 
     def test_env_forward_allowlist(self):
         env = {"SUB2API_API_KEY": "sk-test",
@@ -102,10 +98,85 @@ class CellArgvTest(unittest.TestCase):
                "GITEA_TOKEN": "must-never-enter-cell",
                "XCM_API_KEY": "xk-test"}
         with mock.patch.dict(os.environ, env, clear=False):
-            argv = CELL.build_nspawn_argv(["true"])
+            spec = CELL.build_sandbox_spec(["true"])
-        self.assertIn("--setenv=SUB2API_API_KEY=sk-test", argv)
+            real = CELL._forwarded_env()
-        self.assertIn("--setenv=XCM_API_KEY=xk-test", argv)
+        # 规约只印名不印值：白名单里的值一律 <redacted>。
-        self.assertFalse(any("GITEA_TOKEN" in a for a in argv))
+        self.assertEqual(spec["env"]["SUB2API_API_KEY"], "<redacted>")
+        self.assertEqual(spec["env"]["SUB2API_BASE_URL"], "<redacted>")
+        self.assertEqual(spec["env"]["XCM_API_KEY"], "<redacted>")
+        self.assertNotIn("GITEA_TOKEN", spec["env"])
+        # 真实转发：白名单进、GITEA_TOKEN 绝不进。
+        self.assertEqual(real["SUB2API_BASE_URL"],
+                         "https://example.invalid")
+        self.assertNotIn("GITEA_TOKEN", real)
+
+
+class CellSeccompTest(unittest.TestCase):
+    def test_filter_structure(self):
+        raw = CELL.build_seccomp_filter()
+        self.assertEqual(len(raw), 80)  # 10 条指令 × 8 字节
+        insns = [struct.unpack_from("HBBI", raw, i * 8)
+                 for i in range(10)]
+        # [0] 读 arch；[1] 非 x86_64 跳去放行；[2] 读 nr。
+        self.assertEqual(insns[0], (0x20, 0, 0, 4))
+        self.assertEqual(insns[1], (0x15, 0, 7, 0xC000003E))
+        self.assertEqual(insns[2], (0x20, 0, 0, 0))
+        # 三个 io_uring 号各拦一次，命中回 EPERM，末尾放行。
+        for idx, nr in ((3, 425), (5, 426), (7, 427)):
+            self.assertEqual(insns[idx], (0x15, 0, 1, nr))
+        for idx in (4, 6, 8):
+            self.assertEqual(insns[idx], (0x06, 0, 0, 0x00050001))
+        self.assertEqual(insns[9], (0x06, 0, 0, 0x7fff0000))
+
+    def test_filter_blocks_io_uring_setup_live(self):
+        """子进程里装过滤后调 io_uring_setup 须得 EPERM，
+        getpid 仍正常。装不上（环境无 seccomp）则跳过。"""
+        pid = os.fork()
+        if pid == 0:
+            try:
+                CELL._set_no_new_privileges()
+                CELL._apply_seccomp_filter()
+                libc = ctypes.CDLL("libc.so.6", use_errno=True)
+                ctypes.set_errno(0)
+                ret = libc.syscall(425, 0, 0, 0, 0, 0, 0)
+                err = ctypes.get_errno()
+                ok = (ret == -1 and err == 1)
+                ret2 = libc.syscall(39)
+                ok = ok and ret2 > 0
+                os._exit(0 if ok else 1)
+            except Exception:
+                os._exit(2)
+        _, status = os.waitpid(pid, 0)
+        code = os.waitstatus_to_exitcode(status)
+        if code == 2:
+            self.skipTest("seccomp 不可用，跳过 live 验证")
+        self.assertEqual(code, 0)
+
+
+class CellUsernsTest(unittest.TestCase):
+    def test_write_id_maps_live(self):
+        """子进程里 unshare 用户 ns 并写映射：内 0 → 外 1000。
+        环境建不了 userns（容器/AppArmor 限制）则跳过。"""
+        pid = os.fork()
+        if pid == 0:
+            try:
+                os.unshare(CELL.CLONE_NEWUSER)
+                CELL._write_id_maps()
+                with open("/proc/self/uid_map") as fh:
+                    uid_map = fh.read().strip()
+                with open("/proc/self/gid_map") as fh:
+                    gid_map = fh.read().strip()
+                ok = (uid_map == "0 1000 1"
+                      and gid_map == "0 1000 1"
+                      and os.getuid() == 0)
+                os._exit(0 if ok else 1)
+            except Exception:
+                os._exit(2)
+        _, status = os.waitpid(pid, 0)
+        code = os.waitstatus_to_exitcode(status)
+        if code == 2:
+            self.skipTest("userns 不可用，跳过 live 验证")
+        self.assertEqual(code, 0)
 
 
 class CellNetworkTest(unittest.TestCase):
@@ -123,13 +194,24 @@ class CellNetworkTest(unittest.TestCase):
         with self.assertRaises(ValueError):
             CELL.host_network_commands("")
 
+    def test_host_network_commands_with_cell_pid_builds_veth(self):
+        cmds = CELL.host_network_commands("enp0s5", cell_pid=4321)
+        flat = [" ".join(c) for c in cmds]
+        self.assertEqual(
+            flat[0], "ip link add ve-chv-cell type veth peer name host0")
+        self.assertEqual(flat[1], "ip link set host0 netns 4321")
+        self.assertIn("ip addr add 10.200.0.1/24 dev ve-chv-cell", flat)
+
     def test_host_network_cleanup_mirrors_with_delete(self):
         cmds = CELL.host_network_cleanup_commands("enp0s5")
         self.assertTrue(cmds)
-        for cmd in cmds:
+        ipt = [c for c in cmds if "iptables" in c]
-            self.assertIn("iptables", cmd)
+        self.assertTrue(ipt)
+        for cmd in ipt:
             self.assertIn("-D", cmd)
             self.assertNotIn("-A", cmd)
+        # 清理最后删 ve 接口（防父进程被 SIGKILL 后残留）。
+        self.assertEqual(cmds[-1], ["ip", "link", "del", "ve-chv-cell"])
 
     def test_cell_network_commands_via_nsenter(self):
         cmds = CELL.cell_network_commands(4321)
@@ -162,17 +244,6 @@ class CellNetworkTest(unittest.TestCase):
 
         self.assertEqual(CELL.find_uplink_iface(run=fake_run), "tap0")
 
-    def test_find_cell_init_pid_from_fake_proc(self):
-        with tempfile.TemporaryDirectory() as tmp:
-            task_dir = os.path.join(tmp, "777", "task", "777")
-            os.makedirs(task_dir)
-            with open(os.path.join(task_dir, "children"), "w") as fh:
-                fh.write("778 779")
-            self.assertEqual(
-                CELL.find_cell_init_pid(777, proc_root=tmp), 778)
-            self.assertIsNone(
-                CELL.find_cell_init_pid(999, proc_root=tmp))
-
 
 class BrokerTest(unittest.TestCase):
     def _fake_proc(self, root, mapping):
@@ -473,9 +544,10 @@ class CellCliTest(unittest.TestCase):
         with contextlib.redirect_stdout(buf):
             rc = CELL.main(["argv", "--", "echo", "hi"])
         self.assertEqual(rc, 0)
-        argv = json.loads(buf.getvalue())
+        spec = json.loads(buf.getvalue())
-        self.assertIn("--private-users=1000:65536", argv)
+        self.assertEqual(spec["backend"], "userns")
-        self.assertEqual(argv[-2:], ["echo", "hi"])
+        self.assertEqual(spec["uid_map"], "0 1000 1")
+        self.assertEqual(spec["command"], ["echo", "hi"])
 
     def test_cli_run_requires_double_dash(self):
         self.assertEqual(CELL.main(["run", "echo", "hi"]), 2)