fix(chv): cell 映射改由父进程写，中间进程先 setuid(1000)

两个真机教训（2026-11-07）：
- 子进程自己写不进 uid_map：unshare 后 euid 变 overflow，
  连 /proc/self/uid_map 的 open 都 EPERM——改由父进程
  （仍在初始 ns 的 root）经 /proc/<pid>/uid_map 写。
- 映射按外层 uid 翻译：中间进程外层是 root(0) 时
  “0 1000 1” 盖不住它、新 ns 里仍是 overflow——中间进程
  须先 setgroups/setgid/setuid(1000) 再 unshare（实测
  chv 身份 unshare 在客机上可行，AppArmor 不挡）。
另：中间进程异常加 traceback（调试用）；userns 单测改测
父写映射真实路径（映射行按 split 比，内核会多空格对齐）。

chv-guest/chv-cell.py

@@ -335,9 +335,9 @@ def _drop_capabilities():
 
 def _write_id_maps():
     """写当前进程的 uid_map/gid_map：内 0 → 外 chv（1000）。
-    须在 unshare(CLONE_NEWUSER) 之后、其它 unshare 之前；
+    只有映射自己的 euid（或在父 userns 有 CAP_SETUID）时才
-    gid_map 之前须先写 setgroups=deny。调用方须是 root（或
+    写得进；一般情况下由父进程用 _write_id_maps_for_pid 写
-    在父 userns 里有 CAP_SETUID），否则映射写不进。"""
+    （见下）。本函数供单测与自映射场景用。"""
     with open("/proc/self/setgroups", "w") as fh:
         fh.write("deny")
     with open("/proc/self/uid_map", "w") as fh:
@@ -346,6 +346,27 @@ def _write_id_maps():
         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 都是父进程写子进程
+    的映射）。注意：映射的外层 uid 必须是子进程当前的外层
+    uid——中间进程须先 setuid(1000) 再 unshare，否则映射盖
+    不住它、它在新 ns 里仍是 overflow。失败即抛 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,
@@ -568,14 +589,40 @@ def _grandchild_main(command, network, ready_w, go_r):
 
 
 def _intermediate_main(command, network, pid_w, ready_r, ready_w,
-                       go_r, status_w):
+                       go_r, status_w, mapme_w, map_go_r):
-    """中间进程：建 userns（写映射）与 mount/net/pid ns，fork
+    """中间进程：建 userns（等父进程写映射）与 mount/net/pid ns，
-    出单元 init，报 pid、等它退出、报退出码。自己绝不 exec，
+    fork 出单元 init，报 pid、等它退出、报退出码。自己绝不 exec，
     只做命名空间搭建与收割。"""
     try:
         _set_pdeathsig(15)  # SIGTERM：父进程死即退出，带走单元
+        # 先以外层 chv 身份再 unshare：映射按外层 uid 翻译，
+        # 外层不是 1000 时 “0 1000 1” 盖不住进程（2026-11-07
+        # 真机实测：root 外层 unshare 后映射写上了、进程在新
+        # ns 里仍是 overflow）。setuid 前须先清附加组、设 gid。
+        os.setgroups([CELL_GID])
+        os.setgid(CELL_GID)
+        os.setuid(CELL_UID)
         os.unshare(CLONE_NEWUSER)
-        _write_id_maps()
+        # 映射由父进程写（子进程自己写不进，见
+        # _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)
@@ -611,6 +658,8 @@ def _intermediate_main(command, network, pid_w, ready_r, ready_w,
         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())
@@ -639,26 +688,46 @@ def run_cell(command, network="veth"):
     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):
+            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)
+                               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):
+        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 超时或中间进程已死")
@@ -712,7 +781,7 @@ def run_cell(command, network="veth"):
             os.close(go_w)
         except OSError:
             pass
-        for fd in (pid_r, ready_r, status_r):
+        for fd in (pid_r, ready_r, status_r, mapme_r, map_go_w):
             try:
                 os.close(fd)
             except OSError:

cloud/tests/test_chv_muse_p2.py

@@ -155,23 +155,55 @@ class CellSeccompTest(unittest.TestCase):
 
 class CellUsernsTest(unittest.TestCase):
     def test_write_id_maps_live(self):
-        """子进程里 unshare 用户 ns 并写映射：内 0 → 外 1000。
+        """父进程给已 unshare 的子进程写映射：内 0 → 子进程
-        环境建不了 userns（容器/AppArmor 限制）则跳过。"""
+        的外层 uid。子进程自己写不进（unshare 后 euid 变
+        overflow，连 /proc/self/uid_map 的 open 都 EPERM，
+        2026-11-07 真机实测）。父进程没 CAP_SETUID 时跳过。
+        注：真实路径里中间进程先 setuid(1000) 再 unshare，
+        映射的外层 uid 就是 1000；单测里子进程不改 uid，
+        映射它自己的 euid/egid，测的是同一套父写映射机制。"""
+        me_r, me_w = os.pipe()
+        go_r, go_w = os.pipe()
+        uid = os.geteuid()
+        gid = os.getegid()
         pid = os.fork()
         if pid == 0:
             try:
+                os.close(me_r)
+                os.close(go_w)
                 os.unshare(CELL.CLONE_NEWUSER)
-                CELL._write_id_maps()
+                os.write(me_w, b"1")
+                os.close(me_w)
+                got = os.read(go_r, 1)
+                if not got:
+                    os._exit(2)
                 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"
+                ok = (uid_map.split() == ["0", str(uid), "1"]
-                      and gid_map == "0 1000 1"
+                      and gid_map.split() == ["0", str(gid), "1"]
                       and os.getuid() == 0)
                 os._exit(0 if ok else 1)
             except Exception:
                 os._exit(2)
+        os.close(me_w)
+        os.close(go_r)
+        try:
+            got = os.read(me_r, 1)
+            if not got:
+                self.skipTest("子进程 unshare 失败，跳过 live 验证")
+            try:
+                CELL._write_id_maps_for_pid(pid, uid=uid, gid=gid)
+            except Exception:
+                self.skipTest("父进程写映射无权限，跳过 live 验证")
+            os.write(go_w, b"1")
+        finally:
+            for fd in (me_r, go_w):
+                try:
+                    os.close(fd)
+                except OSError:
+                    pass
         _, status = os.waitpid(pid, 0)
         code = os.waitstatus_to_exitcode(status)
         if code == 2: