codex-app-server/chv-guest/chv-cell.py

#!/usr/bin/env python3
"""chv-cell：客机内 codex 运行时单元（Hatch 式）启动器。

对齐 Muse P2（见 docs/chv-muse-p2-gap.md）：codex 跑在单元内，不再
直接跑在客机用户态。单元是单命令沙箱——单元即沙箱，单元内 codex
用 danger-full-access，不再套 bwrap（bwrap 的 --proc 挂载在嵌套命名
空间里走不通，见 AGENTS.md 内层 proc 定位）。

关键映射（与 Muse“单元 root 映射为 host 侧非特权用户”同款）：
  单元内 uid 0 → 客机侧 chv（uid 1000）。
  效果：workspace / ~/.codex（chv 所有）单元内仍可写；host-root 0600
  文件（如 /etc/chv/guest-env）在单元内显示为 overflow uid、不可读；
  再加挂载遮蔽作纵深防御。

网络：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 是显式调试逃生口，须显式
设置并打日志，绝不默认。

环境透传：本脚本只转发白名单变量（模型网关与代理变量）。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...>
      # 例：chv-cell run -- python3 /app/host.py --state
      #   /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...>
      # 只打印单元规约（JSON：映射、挂载、网络、环境名），不启动。
      # 环境值只印 <redacted>，绝不印真值。
  chv-cell network-commands [--uplink IFACE] [--cell-pid PID]
      # 只打印配网命令表（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 grp
import json
import os
import pwd
import re
import select
import struct
import subprocess
import sys
import time

MACHINE = "chv-cell"
# 单元映射：单元内 uid 0 → 客机侧 chv（uid 1000，非特权用户）。
UID_BASE = 1000
CELL_USER = "chv"
CELL_UID = 1000
# 单元内 DNS：客机 /etc/resolv.conf 是 127.0.0.53 stub，单元独立
# netns 里自己的 lo 上没有 resolver；上游文件里是真实上游。
RESOLV_UPSTREAM = "/run/systemd/resolve/resolv.conf"
RESOLV_CONF = "/etc/resolv.conf"
CELL_GID = 1000
WORKSPACE = "/home/chv/workspace"
CODEX_HOME = "/home/chv/.codex"
BROKER_DIR = "/run/chv"
CELL_PID_FILE = "/run/chv/cell.pid"

# 单元网络（与客机 tap 网 192.168.10.0/24 不重叠的独立 /24）。
CELL_NET = "10.200.0.0/24"
HOST_IP = "10.200.0.1"
CELL_IP = "10.200.0.2"
HOST_IFACE = "ve-chv-cell"
CELL_IFACE = "host0"

# Muse 点名要丢的能力 + 同批丢的危险能力（单元内 root 只在
# 自己的 userns 里有能力，host 侧本就没有；这里再从 bounding
# 集里丢掉，防 setuid/提权路径，见 _drop_capabilities）。
DROP_CAPABILITIES = [
    "CAP_SYS_ADMIN",
    "CAP_SYS_PTRACE",
    "CAP_NET_ADMIN",
    "CAP_NET_RAW",
    "CAP_SYS_MODULE",
    "CAP_SYS_RAWIO",
    "CAP_SYS_BOOT",
    "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）。新后端用 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）。目录盖 tmpfs，文件绑 /dev/null。
INACCESSIBLE_BASE = [
    "/etc/chv",
    "/home/chv/.git-credentials",
    "/usr/local/share/ms-playwright",
]

# 只转发白名单环境变量进单元。凭证类变量（GITEA_TOKEN 等）
# 不在白名单里，绝不进单元。
FORWARD_ENV_ALLOWLIST = (
    "SUB2API_BASE_URL",
    "SUB2API_API_KEY",
    "XCM_API_KEY",
    "HTTP_PROXY",
    "HTTPS_PROXY",
    "NO_PROXY",
    "http_proxy",
    "https_proxy",
    "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 默认路径。"""
    try:
        import sysconfig
        purelib = sysconfig.get_path("purelib")
        if purelib:
            return os.path.join(purelib, "playwright")
    except Exception:
        pass
    return "/usr/local/lib/python3.12/dist-packages/playwright"


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):
    """单元规约（纯函数，可单测）。command 是单元内要跑的命令
    （list，如 ["python3", "/app/host.py", ...]）。只描述“会怎么
    跑”，不启动；环境值在规约里一律 <redacted>。"""
    if not command:
        raise ValueError("chv-cell: 单元内命令不能为空")
    if network not in ("veth", "shared"):
        raise ValueError("chv-cell: network 只能是 veth|shared：%r"
                         % network)
    if playwright_package_dir is None:
        playwright_package_dir = _playwright_package_dir()
    env = _forwarded_env()
    display_env = {k: ("<redacted>" if k in FORWARD_ENV_ALLOWLIST
                       else v) for k, v in env.items()}
    return {
        "backend": "userns",
        "cell": MACHINE,
        "uid_map": "0 %d 1" % CELL_UID,
        "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"],
        "dns_resolv_conf": "upstream-bind",
        "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] 其余放行
    ]
    return b"".join(insns)


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）。
    只有映射自己的 euid（或在父 userns 有 CAP_SETUID）时才
    写得进；一般情况下由父进程用 _write_id_maps_for_pid 写
    （见下）。本函数供单测与自映射场景用。"""
    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 _write_id_maps_for_pid(pid, uid=CELL_UID, gid=CELL_GID):
    """父进程（仍在初始 userns 的 root）给已 unshare 用户
    ns 的子进程写映射：内 0 → 外 uid（默认 chv 1000）。
    子进程自己写不进——unshare 后它的 euid 变成 overflow，
    连 /proc/self/uid_map 的 open 都 EPERM（2026-11-07 真机
    实测）。这是标准做法（unshare/bwrap 都是父进程写子进程
    的映射）。注意：中间进程保持 root 外层身份 unshare，
    映射写上后它在新 ns 里仍是 overflow uid，但 creator
    caps 在、能继续 unshare 其它 ns；切内 0 由孙进程在
    ns 内 setuid(0) 完成（见 _grandchild_main）。中间进程
    绝不能先 setuid(1000) 再 unshare——实测那样后续
    unshare(CLONE_NEWNS) 得 EPERM。失败即抛 RuntimeError。"""
    try:
        with open("/proc/%d/setgroups" % pid, "w") as fh:
            fh.write("deny")
        with open("/proc/%d/uid_map" % pid, "w") as fh:
            fh.write("0 %d 1" % uid)
        with open("/proc/%d/gid_map" % pid, "w") as fh:
            fh.write("0 %d 1" % gid)
    except OSError as exc:
        raise RuntimeError(
            "chv-cell: 给 pid=%d 写 id 映射失败：%s" % (pid, exc))


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 _resolv_bind(source=RESOLV_UPSTREAM, target=RESOLV_CONF):
    """返回把上游 resolv.conf 绑到单元 /etc/resolv.conf 的 mount
    参数，或 None（无上游文件/无 nameserver 时跳过，单元内
    DNS 保持现状）。客机 /etc/resolv.conf 是 127.0.0.53 的
    stub（systemd-resolved），单元有独立 netns、自己的 lo 上
    没有 resolver，直接沿用则 DNS 全灭；上游文件
    /run/systemd/resolve/resolv.conf 里是真实上游（8.8.8.8
    等），经 veth NAT 可达。target 是 symlink 时挂到真实目标
    （stub 文件），挂载只在单元的私有 mount ns 里，不动宿主。"""
    if not os.path.isfile(source):
        return None
    try:
        with open(source, "r", encoding="utf-8",
                  errors="replace") as f:
            content = f.read()
    except OSError:
        return None
    if "nameserver" not in content:
        return None
    real_target = os.path.realpath(target)
    if not os.path.isfile(real_target):
        return None
    return ["--bind", source, real_target]


def _setup_mounts(playwright_package_dir=None):
    """单元内挂载：根 rprivate、fresh /proc、/tmp tmpfs、
    敏感路径遮蔽（目录盖 tmpfs，文件绑 /dev/null）、上游
    resolv.conf 绑到 /etc/resolv.conf（单元内 DNS）。
    只在新 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]:
        if os.path.isdir(path):
            _mount(["-t", "tmpfs", "tmpfs", path], "遮蔽 %s" % path)
        elif os.path.exists(path):
            _mount(["--bind", "/dev/null", path], "遮蔽 %s" % path)
        else:
            print("chv-cell: 遮蔽目标不存在，跳过：%s" % path,
                  file=sys.stderr)
    bind = _resolv_bind()
    if bind:
        _mount(bind, "DNS resolv.conf")
    else:
        print("chv-cell: 无上游 resolv.conf，单元内 DNS 可能不通",
              file=sys.stderr)


def parse_default_route_iface(ip_route_output):
    """从 `ip route show default` 的输出里解析上行口名。
    无默认路由时返回 None（调用方大声失败）。"""
    for line in (ip_route_output or "").splitlines():
        line = line.strip()
        if not line.startswith("default "):
            continue
        m = re.search(r"\bdev\s+(\S+)", line)
        if m:
            return m.group(1)
    return None


def find_uplink_iface(run=None):
    """发现客机上行口（默认路由的 dev）。run 供单测注入。"""
    run = run or subprocess.run
    proc = run(["ip", "route", "show", "default"],
               capture_output=True, text=True)
    if proc.returncode != 0:
        return None
    return parse_default_route_iface(proc.stdout)


def host_network_commands(uplink_iface, cell_pid=None):
    """客机侧配网命令表（纯函数）：建 veth 对、把 host0 移进
    单元 netns、ve 接口起 IP + 转发 + NAT。cell_pid 为空时
    只返回后半段（ve 起 IP + 转发 + NAT，兼容旧调用与单测）；
    veth 模式调用方须传 cell_pid，建对与移口才完整。
    调用方按顺序执行，任一失败即停。"""
    if not uplink_iface:
        raise ValueError("chv-cell: 上行口名为空，配不出 NAT")
    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"],
        ["iptables", "-t", "nat", "-A", "POSTROUTING",
         "-s", CELL_NET, "-o", uplink_iface, "-j", "MASQUERADE"],
        ["iptables", "-A", "FORWARD", "-i", HOST_IFACE,
         "-o", uplink_iface, "-j", "ACCEPT"],
        ["iptables", "-A", "FORWARD", "-i", 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，
    再删 ve 接口）。清理是 best-effort：删不掉（规则已不在）
    不报错。"""
    if not uplink_iface:
        raise ValueError("chv-cell: 上行口名为空，清不了 NAT")
    cmds = []
    for cmd in host_network_commands(uplink_iface):
        if "iptables" not in cmd:
            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 cell_network_commands(cell_pid):
    """单元内配网命令表（纯函数）：nsenter 进单元 init 的 netns，
    给 host0 配 IP、起接口、加默认路由。"""
    if not cell_pid:
        raise ValueError("chv-cell: 单元 init pid 为空，配不了单元内网络")
    prefix = ["nsenter", "-t", str(cell_pid), "-n"]
    return [
        prefix + ["ip", "addr", "add", "%s/24" % CELL_IP,
                  "dev", CELL_IFACE],
        prefix + ["ip", "link", "set", CELL_IFACE, "up"],
        prefix + ["ip", "route", "add", "default", "via", HOST_IP],
    ]


def _broker_dir_gid():
    """broker 目录（/run/chv）的组：chv-browser 要在里面建 socket，
    组须是 chv-browser（0775 时组可写）；该组不存在时回落 root（0）。

    2026-11-08 构建 #8 live 实证：这里曾写死 gid=0，每次跑 cell 都
    把 /run/chv 的组改回 root，抹掉 tmpfiles 的 chv-browser 组；
    之后 chv-browser 重启时 os.remove(browser.sock) 报 Permission
    denied、无限 crash-loop（62 次重启才被发现）。"""
    try:
        return grp.getgrnam("chv-browser").gr_gid
    except KeyError:
        return 0


def _ensure_bind_source(path, uid=CELL_UID, gid=CELL_GID):
    """bind 源目录须存在且属 chv（单元内 root 即客机 chv，属主不对
    就写不进 workspace）。root 建的目录要 chown 回 chv。"""
    os.makedirs(path, exist_ok=True)
    try:
        st = os.stat(path)
        if st.st_uid != uid or st.st_gid != gid:
            os.chown(path, uid, gid)
    except OSError as e:
        print("chv-cell: chown %s 失败：%s" % (path, e), file=sys.stderr)
        raise


def _wait_for_iface(iface, timeout=10.0):
    deadline = time.time() + timeout
    while time.time() < deadline:
        proc = subprocess.run(["ip", "link", "show", iface],
                              capture_output=True)
        if proc.returncode == 0:
            return True
        time.sleep(0.2)
    return False


def _iface_has_ip(iface, ip):
    """接口上是否有所给 IPv4。配网后复核：2026-11-07 实测
    networkd（80-container-ve.network）会在接口建好后几秒内
    刷掉静态 10.200.0.1、改 DHCP/链路本地，致单元网关不可达；
    镜像里有 79-chv-cell.network（Unmanaged）兜底，这里是
    第二道防线——地址不见即大声失败，不跑无网单元。"""
    proc = subprocess.run(["ip", "addr", "show", iface],
                          capture_output=True, text=True)
    if proc.returncode != 0:
        return False
    return ("inet %s/" % ip) in proc.stdout


def _run_cmds(cmds, what):
    for cmd in cmds:
        print("chv-cell: + %s" % " ".join(cmd), file=sys.stderr)
        proc = subprocess.run(cmd)
        if proc.returncode != 0:
            raise RuntimeError(
                "chv-cell: %s 失败（rc=%d）：%s"
                % (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）：切内 0、挂载、加固、
    报 READY、等放行、exec payload。任何一步失败写 ERR 后退出
    （父进程收到 ERR 即大声失败，不跑 payload）。"""
    try:
        _set_pdeathsig(9)  # SIGKILL：中间进程死即自杀，不留孤儿
        # 在 ns 内切到内 0（映射到外层 chv 1000）：此时进程是
        # creator、有新 ns 的全部 caps，setgid/setuid(0) 合法；
        # 必须先 setgid 再 setuid。绝不能在 unshare 之前以外层
        # 身份 setuid——那会弄丢后续 unshare 的能力（见上）。
        os.setgid(0)
        os.setuid(0)
        _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, mapme_w, map_go_r):
    """中间进程：建 userns（等父进程写映射）与 mount/net/pid ns，
    fork 出单元 init，报 pid、等它退出、报退出码。自己绝不 exec，
    只做命名空间搭建与收割。"""
    try:
        _set_pdeathsig(15)  # SIGTERM：父进程死即退出，带走单元
        # 只设附加组（让它映射到内 0），不改 uid/gid：中间进程
        # 保持 root 外层身份 unshare（creator caps 在，后续
        # unshare 其它 ns 才通）。绝不能先 setuid(1000) 再
        # unshare——2026-11-07 真机对照实测：那样后续
        # unshare(CLONE_NEWNS) 得 EPERM；不 setuid 则通。
        os.setgroups([CELL_GID])
        os.unshare(CLONE_NEWUSER)
        # 映射由父进程写（子进程自己写不进，见
        # _write_id_maps_for_pid）；先报“已 unshare”，再等放行。
        try:
            os.write(mapme_w, b"1")
        except OSError:
            pass
        try:
            os.close(mapme_w)
        except OSError:
            pass
        try:
            got = os.read(map_go_r, 1)
        except OSError:
            os._exit(1)
        try:
            os.close(map_go_r)
        except OSError:
            pass
        if not got:
            os._exit(1)
        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 —— 全部转状态上报
        import traceback
        traceback.print_exc()
        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 运行（userns + 配网都要 root）",
              file=sys.stderr)
        return 2
    _ensure_bind_source(WORKSPACE)
    _ensure_bind_source(CODEX_HOME)
    # broker 目录不写死 gid=0：组须留给 chv-browser（见 _broker_dir_gid）。
    _ensure_bind_source(BROKER_DIR, uid=0, gid=_broker_dir_gid())
    spec = build_sandbox_spec(command, network=network)
    print("chv-cell: 启动单元：%s"
          % json.dumps(spec, ensure_ascii=False), file=sys.stderr)
    pid_r, pid_w = os.pipe()
    ready_r, ready_w = os.pipe()
    go_r, go_w = os.pipe()
    status_r, status_w = os.pipe()
    mapme_r, mapme_w = os.pipe()
    map_go_r, map_go_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,
                       mapme_r, map_go_w):
                try:
                    os.close(fd)
                except OSError:
                    pass
            _intermediate_main(command, network, pid_w, ready_r,
                               ready_w, go_r, status_w,
                               mapme_w, map_go_r)
            os._exit(0)
        # 父进程：关掉子进程用的端。
        for fd in (pid_w, ready_w, go_r, status_w,
                   mapme_w, map_go_r):
            try:
                os.close(fd)
            except OSError:
                pass
        # 等中间进程 unshare 完用户 ns，再由父进程（仍在初始
        # ns 的 root）给它写 id 映射，然后放行。
        got = os.read(mapme_r, 1)
        if not got:
            raise RuntimeError("chv-cell: 中间进程在 unshare 后已死")
        _write_id_maps_for_pid(mid_pid)
        try:
            os.write(map_go_w, b"1")
        except OSError as exc:
            raise RuntimeError("chv-cell: 放行映射等待失败：%s" % exc)
        try:
            os.close(map_go_w)
            os.close(mapme_r)
        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":
            # 先清掉上次泄漏的 ve 接口（父进程被 SIGKILL 时
            # finally 跑不到，接口会残留，建对前先删）。
            subprocess.run(["ip", "link", "del", HOST_IFACE],
                           capture_output=True)
            uplink = find_uplink_iface()
            if not uplink:
                raise RuntimeError(
                    "chv-cell: 客机无默认路由，配不出 NAT")
            _run_cmds(host_network_commands(uplink, cell_pid),
                      "客机侧配网")
            if not _wait_for_iface(HOST_IFACE):
                raise RuntimeError(
                    "chv-cell: veth 接口 %s 未出现，拒绝回落共享网络"
                    % HOST_IFACE)
            _run_cmds(cell_network_commands(cell_pid), "单元内配网")
            if not _iface_has_ip(HOST_IFACE, HOST_IP):
                raise RuntimeError(
                    "chv-cell: veth %s 上 %s 不见了（疑似 networkd "
                    "刷掉），拒绝跑无网单元" % (HOST_IFACE, HOST_IP))
        # 放行单元 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, mapme_r, map_go_w):
            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)
        try:
            if os.path.exists(CELL_PID_FILE):
                os.remove(CELL_PID_FILE)
        except OSError:
            pass


def main(argv):
    if len(argv) < 1:
        print("用法：chv-cell run|argv|network-commands …",
              file=sys.stderr)
        return 2
    verb = argv[0]
    rest = argv[1:]
    network = os.environ.get("CHV_CELL_NETWORK", "veth")
    # --network 只在 run/argv 前接受，其它位置一律不认。
    if rest and rest[0] == "--network":
        if len(rest) < 2:
            print("chv-cell: --network 缺值", file=sys.stderr)
            return 2
        network = rest[1]
        rest = rest[2:]
    if network not in ("veth", "shared"):
        print("chv-cell: network 只能是 veth|shared：%r" % network,
              file=sys.stderr)
        return 2
    if network == "shared":
        print("chv-cell: 警告：共享网络模式（调试逃生口），单元与"
              "客机共用网络命名空间", file=sys.stderr)
    if verb == "run":
        if not rest or rest[0] != "--":
            print("chv-cell: run 用法：chv-cell run -- <cmd...>",
                  file=sys.stderr)
            return 2
        command = rest[1:]
        if not command:
            print("chv-cell: 单元内命令不能为空", file=sys.stderr)
            return 2
        try:
            return run_cell(command, network=network)
        except RuntimeError as e:
            print(str(e), file=sys.stderr)
            return 1
    if verb == "argv":
        if not rest or rest[0] != "--":
            print("chv-cell: argv 用法：chv-cell argv -- <cmd...>",
                  file=sys.stderr)
            return 2
        print(json.dumps(build_sandbox_spec(rest[1:], network=network),
                         ensure_ascii=False))
        return 0
    if verb == "network-commands":
        uplink = None
        cell_pid = None
        i = 0
        while i < len(rest):
            if rest[i] == "--uplink" and i + 1 < len(rest):
                uplink = rest[i + 1]
                i += 2
            elif rest[i] == "--cell-pid" and i + 1 < len(rest):
                cell_pid = int(rest[i + 1])
                i += 2
            else:
                print("chv-cell: network-commands 未知参数：%s"
                      % rest[i], file=sys.stderr)
                return 2
        out = {}
        if 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)
        print(json.dumps(out, ensure_ascii=False))
        return 0
    print("chv-cell: 未知子命令：%s" % verb, file=sys.stderr)
    return 2


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))