"""对齐 Muse P2 测试：客机 nspawn 单元 + 浏览器子代理独立。 只测本轨交付物（chv-guest/chv-cell.py、chv-broker.py、chv-browser.py、 build.sh 干跑、buildinfo.py 内容项）。全部用假 socket/假进程/假 playwright，绝不连真实网络、不起真实 VM、不改生产行为。 """ import importlib.util import json import os import shutil import socket import struct import subprocess import sys import tempfile import unittest from pathlib import Path from unittest import mock ROOT = Path(__file__).resolve().parents[2] def _load_script(name): """把 chv-guest 下的独立脚本当模块加载（无包结构）。""" path = ROOT / "chv-guest" / name spec = importlib.util.spec_from_file_location( "chv_guest_p2_" + name[:-3], str(path)) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module CELL = _load_script("chv-cell.py") BROKER = _load_script("chv-broker.py") BROWSER = _load_script("chv-browser.py") 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"]) self.assertIn("--private-users=1000:65536", argv) self.assertIn("--machine=chv-cell", argv) self.assertIn("--directory=/", argv) self.assertIn("--as-pid2", argv) # round-11 真机修复：nspawn 255 拒裸 --no-new-privileges #（当作带必填参数的选项）与裸 --directory=/（须 # --ephemeral/--volatile），见 p2.log 的解析报错。 self.assertIn("--no-new-privileges=yes", argv) self.assertNotIn("--no-new-privileges", argv) self.assertIn("--ephemeral", argv) def test_muse_named_capabilities_dropped(self): argv = CELL.build_nspawn_argv(["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) # 绝不能出现“保留这些能力”的写法。 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"]) filt = [a for a in argv if a.startswith("--system-call-filter=")] self.assertEqual(len(filt), 1) self.assertIn("@system-service", filt[0]) for call in ("io_uring_setup", "io_uring_enter", "io_uring_register"): self.assertIn("~%s" % call, filt[0]) def test_secrets_and_browser_hidden_from_cell(self): argv = CELL.build_nspawn_argv( ["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) def test_writable_binds_and_tmpfs(self): argv = CELL.build_nspawn_argv(["true"]) self.assertIn("--bind=/home/chv/workspace", argv) self.assertIn("--bind=/home/chv/.codex", argv) self.assertIn("--bind=/run/chv", argv) self.assertIn("--tmpfs=/tmp", argv) self.assertIn("--chdir=/home/chv/workspace", argv) self.assertIn("--setenv=CHV_CELL=1", argv) self.assertIn("--setenv=HOME=/home/chv", argv) def test_veth_default_and_shared_opt_in(self): self.assertIn("--network-veth", CELL.build_nspawn_argv(["true"])) self.assertNotIn("--network-veth", CELL.build_nspawn_argv(["true"], network="shared")) with self.assertRaises(ValueError): CELL.build_nspawn_argv(["true"], network="macvlan") with self.assertRaises(ValueError): CELL.build_nspawn_argv([]) def test_env_forward_allowlist(self): env = {"SUB2API_API_KEY": "sk-test", "SUB2API_BASE_URL": "https://example.invalid", "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"]) self.assertIn("--setenv=SUB2API_API_KEY=sk-test", argv) self.assertIn("--setenv=XCM_API_KEY=xk-test", argv) self.assertFalse(any("GITEA_TOKEN" in a for a in argv)) class CellNetworkTest(unittest.TestCase): def test_host_network_commands_nat_via_uplink(self): cmds = CELL.host_network_commands("enp0s5") flat = [" ".join(c) for c in cmds] self.assertIn("ip addr add 10.200.0.1/24 dev ve-chv-cell", flat) self.assertIn("ip link set ve-chv-cell up", flat) self.assertIn("sysctl -w net.ipv4.ip_forward=1", flat) self.assertTrue(any("iptables -t nat -A POSTROUTING -s 10.200.0.0/24 " "-o enp0s5 -j MASQUERADE" in line for line in flat)) self.assertTrue(any("-A FORWARD -i ve-chv-cell -o enp0s5 -j ACCEPT" in line for line in flat)) with self.assertRaises(ValueError): CELL.host_network_commands("") def test_host_network_cleanup_mirrors_with_delete(self): cmds = CELL.host_network_cleanup_commands("enp0s5") self.assertTrue(cmds) for cmd in cmds: self.assertIn("iptables", cmd) self.assertIn("-D", cmd) self.assertNotIn("-A", cmd) def test_cell_network_commands_via_nsenter(self): cmds = CELL.cell_network_commands(4321) flat = [" ".join(c) for c in cmds] self.assertIn("nsenter -t 4321 -n ip addr add 10.200.0.2/24 " "dev host0", flat) self.assertIn("nsenter -t 4321 -n ip link set host0 up", flat) self.assertIn("nsenter -t 4321 -n ip route add default via " "10.200.0.1", flat) with self.assertRaises(ValueError): CELL.cell_network_commands(0) def test_parse_default_route_iface(self): out = ("default via 192.168.10.1 dev enp0s5 proto dhcp metric 100\n" "192.168.10.0/24 dev enp0s5 proto kernel scope link src " "192.168.10.10\n") self.assertEqual(CELL.parse_default_route_iface(out), "enp0s5") self.assertIsNone(CELL.parse_default_route_iface("")) self.assertIsNone(CELL.parse_default_route_iface( "192.168.10.0/24 dev enp0s5 proto kernel scope link\n")) def test_find_uplink_iface_uses_injected_runner(self): class FakeProc: returncode = 0 stdout = "default via 192.168.10.1 dev tap0 proto static\n" def fake_run(cmd, **kwargs): self.assertEqual(cmd, ["ip", "route", "show", "default"]) return FakeProc() 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): """mapping: pid -> ns 标识。建出 /proc//ns/pid 软链。""" for pid, ns in mapping.items(): d = os.path.join(root, str(pid), "ns") os.makedirs(d) os.symlink("pid:[%s]" % ns, os.path.join(d, "pid")) def test_peer_allowed_same_pid_ns(self): with tempfile.TemporaryDirectory() as tmp: self._fake_proc(tmp, {100: "111", 200: "111"}) ok, _ = BROKER.peer_allowed(1000, 200, cell_pid=100, proc_root=tmp) self.assertTrue(ok) def test_peer_denied_uid_outside_cell_range(self): with tempfile.TemporaryDirectory() as tmp: self._fake_proc(tmp, {100: "111", 200: "111"}) for uid in (0, 999, 66536, 100000): ok, reason = BROKER.peer_allowed(uid, 200, cell_pid=100, proc_root=tmp) self.assertFalse(ok, uid) self.assertIn("单元映射范围", reason) def test_peer_denied_different_pid_ns(self): with tempfile.TemporaryDirectory() as tmp: self._fake_proc(tmp, {100: "111", 200: "222"}) ok, reason = BROKER.peer_allowed(1000, 200, cell_pid=100, proc_root=tmp) self.assertFalse(ok) self.assertIn("pid ns", reason) def test_peer_denied_when_cell_ns_unreadable(self): with tempfile.TemporaryDirectory() as tmp: self._fake_proc(tmp, {200: "111"}) ok, _ = BROKER.peer_allowed(1000, 200, cell_pid=100, proc_root=tmp) self.assertFalse(ok) def test_peer_uid_only_when_cell_pid_missing(self): ok, reason = BROKER.peer_allowed(1000, 200, cell_pid=None) self.assertTrue(ok) self.assertIn("cell.pid 缺失", reason) def test_ping_and_unknown_op(self): self.assertEqual(BROKER.handle_request({"op": "ping"}), {"ok": True, "cell": True}) resp = BROKER.handle_request({"op": "rm"}) self.assertFalse(resp["ok"]) resp = BROKER.handle_request("not-a-dict") self.assertFalse(resp["ok"]) def test_sync_rejects_bad_action_and_outside_repo(self): resp = BROKER.handle_request({"op": "sync", "action": "reset"}) self.assertFalse(resp["ok"]) for bad in ("/etc/passwd", "/home/chv/workspace/../..", "/home/chv", "/tmp/x"): resp = BROKER.handle_request( {"op": "sync", "action": "pull", "repo": bad}) self.assertFalse(resp["ok"], bad) def test_sync_runs_chv_sync_as_chv_without_shell(self): seen = {} class FakeProc: returncode = 0 stdout = "已同步\n" stderr = "" def fake_run(cmd, **kwargs): seen["cmd"] = cmd self.assertFalse(kwargs.get("shell", False)) return FakeProc() with mock.patch.object(BROKER.subprocess, "run", fake_run): resp = BROKER.handle_request( {"op": "sync", "action": "status", "repo": "/home/chv/workspace/demo"}) self.assertTrue(resp["ok"]) self.assertEqual(seen["cmd"][:6], ["runuser", "-u", "chv", "--", "env", "HOME=/home/chv"]) self.assertIn("/usr/local/bin/chv-sync", seen["cmd"]) self.assertIn("status", seen["cmd"]) self.assertIn("/home/chv/workspace/demo", seen["cmd"]) def test_sync_failure_reported_not_swallowed(self): class FakeProc: returncode = 1 stdout = "" stderr = "分叉\n" with mock.patch.object(BROKER.subprocess, "run", lambda *a, **k: FakeProc()): resp = BROKER.handle_request({"op": "sync", "action": "pull"}) self.assertFalse(resp["ok"]) self.assertEqual(resp["returncode"], 1) def test_browser_forward_and_down(self): payload = {"ok": True, "title": "Example Domain"} class FakeConn: def __init__(self, *args, **kwargs): self.sent = b"" def settimeout(self, timeout): pass def connect(self, path): self.path = path def sendall(self, data): self.sent = data def recv(self, n): data, self._buf = self._buf, b"" return data def close(self): pass fake = FakeConn() fake._buf = (json.dumps(payload) + "\n").encode() with mock.patch.object(BROKER.socket, "socket", lambda *a, **k: fake): resp = BROKER.handle_request( {"op": "browser", "action": "open"}, browser_sock="/run/chv/browser.sock") self.assertEqual(resp, payload) self.assertEqual(json.loads(fake.sent.decode())["op"], "browser") def boom(*args, **kwargs): raise OSError("no socket") with mock.patch.object(BROKER.socket, "socket", boom): resp = BROKER.handle_request({"op": "browser"}) self.assertFalse(resp["ok"]) self.assertIn("浏览器子代理未就绪", resp["error"]) def test_peer_creds_unpack(self): # struct ucred = pid/uid/gid 三个 int；broker 用它验对端。 raw = struct.pack("3i", 4321, 1000, 1000) class FakeConn: def getsockopt(self, level, opt, size): self.seen = (level, opt, size) return raw fake = FakeConn() self.assertEqual(BROKER._peer_creds(fake), (4321, 1000, 1000)) self.assertEqual(fake.seen[1], socket.SO_PEERCRED) class BrowserTest(unittest.TestCase): def test_validate_url(self): self.assertTrue(BROWSER.validate_url("https://example.com")[0]) self.assertTrue(BROWSER.validate_url("http://example.com/x")[0]) self.assertTrue( BROWSER.validate_url("https://user@example.com/")[0]) for bad in ("ftp://example.com", "file:///etc/passwd", "https://localhost:8080/", "http://127.0.0.1/", "https://192.168.1.1/", "https://10.0.0.5/", "https://172.17.0.2/", "", None): ok, _ = BROWSER.validate_url(bad) self.assertFalse(ok, bad) def test_truncate_a11y_small_tree_untouched(self): tree = {"role": "WebArea", "name": "Example", "children": [{"role": "heading", "name": "Hi"}]} out, truncated = BROWSER.truncate_a11y(tree) self.assertFalse(truncated) self.assertEqual(out, tree) def test_truncate_a11y_node_cap(self): tree = {"role": "WebArea", "children": [{"role": "link", "name": "l%d" % i} for i in range(300)]} out, truncated = BROWSER.truncate_a11y(tree) self.assertTrue(truncated) self.assertLessEqual(len(out["children"]), 200) def test_truncate_a11y_char_cap(self): tree = {"role": "WebArea", "name": "x" * 70000, "children": []} out, truncated = BROWSER.truncate_a11y(tree) self.assertTrue(truncated) self.assertTrue(out.get("truncated_chars")) def test_forbidden_ops_never_execute(self): agent = mock.Mock() for op in ("evaluate", "content", "devtools", "js"): resp = BROWSER.handle_request(agent, {"op": op}) self.assertFalse(resp["ok"]) self.assertIn("不暴露", resp["error"]) agent.open.assert_not_called() resp = BROWSER.handle_request(agent, {"op": "bogus"}) self.assertFalse(resp["ok"]) def test_open_rejects_bad_url_without_loading_playwright(self): def exploding_factory(): raise AssertionError("坏 URL 不应加载 playwright") agent = BROWSER.BrowserAgent(playwright_factory=exploding_factory) resp = agent.open("https://127.0.0.1/secret") self.assertFalse(resp["ok"]) def test_snapshot_and_screenshot_need_open_first(self): agent = BROWSER.BrowserAgent( playwright_factory=lambda: (_ for _ in ()).throw( AssertionError("不应加载 playwright"))) self.assertFalse(agent.snapshot()["ok"]) self.assertFalse(agent.screenshot()["ok"]) self.assertTrue(agent.close()["ok"]) def test_open_returns_a11y_snapshot_with_fake_playwright(self): page = FakePage() with tempfile.TemporaryDirectory() as tmp: agent = BROWSER.BrowserAgent( playwright_factory=lambda: FakePlaywright(page), shots_dir=tmp) resp = agent.open("https://example.com") self.assertTrue(resp["ok"]) self.assertEqual(resp["title"], "Example Domain") self.assertEqual(resp["url"], "https://example.com/") self.assertEqual(resp["a11y"]["role"], "WebArea") snap = agent.snapshot() self.assertTrue(snap["ok"]) shot = agent.screenshot() self.assertTrue(shot["ok"]) self.assertTrue(shot["path"].endswith(".png")) self.assertTrue(os.path.exists(shot["path"])) self.assertTrue(agent.close()["ok"]) class FakePage: def __init__(self): self.url = "https://example.com/" self._closed = False def goto(self, url, timeout=None): self.url = url.rstrip("/") + "/" def wait_for_load_state(self, state, timeout=None): pass def title(self): return "Example Domain" @property def accessibility(self): return self def snapshot(self): return {"role": "WebArea", "name": "Example Domain", "children": [{"role": "heading", "name": "Example"}]} def screenshot(self, path=None): with open(path, "wb") as fh: fh.write(b"\x89PNG") def close(self): self._closed = True class FakeBrowser: def __init__(self, page): self._page = page def new_page(self): return self._page def close(self): pass class FakeEntered: def __init__(self, page): self.chromium = mock.Mock() self.chromium.launch.return_value = FakeBrowser(page) class FakePlaywright: def __init__(self, page): self._page = page def __enter__(self): return FakeEntered(self._page) def __exit__(self, *exc): return False class CellCliTest(unittest.TestCase): def test_cli_argv_prints_json(self): import contextlib import io buf = io.StringIO() with contextlib.redirect_stdout(buf): rc = CELL.main(["argv", "--", "echo", "hi"]) self.assertEqual(rc, 0) argv = json.loads(buf.getvalue()) self.assertIn("--private-users=1000:65536", argv) self.assertEqual(argv[-2:], ["echo", "hi"]) def test_cli_run_requires_double_dash(self): self.assertEqual(CELL.main(["run", "echo", "hi"]), 2) self.assertEqual(CELL.main(["bogus"]), 2) self.assertEqual(CELL.main([]), 2) def test_cli_network_commands_prints_tables(self): import contextlib import io buf = io.StringIO() with contextlib.redirect_stdout(buf): rc = CELL.main(["network-commands", "--uplink", "enp0s5", "--cell-pid", "99"]) self.assertEqual(rc, 0) out = json.loads(buf.getvalue()) self.assertIn("host", out) self.assertIn("cell", out) self.assertIn("host_cleanup", out) def test_cli_unknown_args_rejected(self): self.assertEqual(BROKER.main(["--bogus"]), 2) self.assertEqual(BROWSER.main(["--bogus"]), 2) class BuildInfoTest(unittest.TestCase): def _synth_out(self, tmp): out = os.path.join(tmp, "out") os.makedirs(out) for name, content in (("disk.raw", b"D"), ("vmlinux", b"K")): with open(os.path.join(out, name), "wb") as fh: fh.write(content) for name, lines in ( ("guest-versions-dpkg.txt", ["git 9.9.9-9ubuntu9"]), ("guest-versions-pip.txt", ["playwright==9.9.9"]), ("guest-versions-gt.txt", ["gt version 1.0.0"]), ("guest-versions-chromium.txt", ["chromium_headless_shell-1248"])): with open(os.path.join(out, name), "w") as fh: fh.write("\n".join(lines) + "\n") return out def test_contents_include_p2_items(self): with tempfile.TemporaryDirectory() as tmp: out = self._synth_out(tmp) info = BUILDINFO.build( out, "chv-guest-24.04-0.164.0-alpha.9-abc1234", "24.04", "0.164.0-alpha.9", "abc1234", "0" * 64, True, "1248") for item in ("systemd-container", "chv-cell", "chv-broker", "chv-browser"): self.assertIn(item, info["contents"]) def test_extra_contents_appended_once(self): with tempfile.TemporaryDirectory() as tmp: out = self._synth_out(tmp) info = BUILDINFO.build( out, "chv-guest-24.04-0.164.0-alpha.9-abc1234", "24.04", "0.164.0-alpha.9", "abc1234", "0" * 64, True, "1248", ["vsock_notify"]) self.assertIn("vsock_notify", info["contents"]) info2 = BUILDINFO.build( out, "chv-guest-24.04-0.164.0-alpha.9-abc1234", "24.04", "0.164.0-alpha.9", "abc1234", "0" * 64, True, "1248", ["vsock_notify", "chv-cell"]) self.assertEqual(info2["contents"].count("chv-cell"), 1) self.assertIn("vsock_notify", info2["contents"]) def test_main_accepts_tenth_arg(self): with tempfile.TemporaryDirectory() as tmp: out = self._synth_out(tmp) args = ["buildinfo.py", out, "chv-guest-24.04-0.164.0-alpha.9-abc1234", "24.04", "0.164.0-alpha.9", "abc1234", "0" * 64, "1", "1248", "vsock_notify"] self.assertEqual(BUILDINFO.main(args), 0) with open(os.path.join(out, "BUILDINFO.json")) as fh: info = json.load(fh) self.assertIn("vsock_notify", info["contents"]) self.assertEqual(BUILDINFO.main(args[:9]), 0) class BuildDryRunTest(unittest.TestCase): def test_dry_run_contains_p2_steps(self): # 在隔离的临时 git 仓里跑 build.sh --dry-run：工作树干净、 # HEAD 与 REPO_COMMIT 一致、假 codex 二进制报对版本。 with tempfile.TemporaryDirectory() as tmp: repo = os.path.join(tmp, "repo") guest = os.path.join(repo, "chv-guest") bindir = os.path.join(tmp, "bin") os.makedirs(guest) os.makedirs(os.path.join(bindir, "codex-resources")) for name in ("build.sh", "buildinfo.py", "check_lockfile.py", "generate_lockfile.py", "chv-sync.sh", "chv-cell.py", "chv-broker.py", "chv-browser.py", "authd.py", "git-credential-chv-authd", "chv-authd.service"): shutil.copy2(ROOT / "chv-guest" / name, os.path.join(guest, name)) shutil.copy2(ROOT / "host.py", os.path.join(repo, "host.py")) # round-11：build.sh 要求 P3 云端叶子模块在仓里（fail-fast）， # 夹具须一并拷入，否则干跑在 P3 预检就失败。 for rel in ("cloud/chv/authd.py", "cloud/chv/sentinel.py", "cloud/__init__.py", "cloud/chv/__init__.py"): dest = os.path.join(repo, rel) os.makedirs(os.path.dirname(dest), exist_ok=True) shutil.copy2(ROOT / rel, dest) codex = os.path.join(bindir, "codex") with open(codex, "w") as fh: fh.write("#!/bin/sh\necho 'codex-cli 0.164.0-alpha.9'\n") os.chmod(codex, 0o755) host_bin = os.path.join(bindir, "codex-code-mode-host") with open(host_bin, "w") as fh: fh.write("#!/bin/sh\nexit 0\n") os.chmod(host_bin, 0o755) subprocess.run(["git", "init", "-q"], cwd=repo, check=True) subprocess.run(["git", "add", "-A"], cwd=repo, check=True) subprocess.run( ["git", "-c", "user.email=t@t", "-c", "user.name=t", "commit", "-qm", "p2 dry-run fixture"], cwd=repo, check=True) head = subprocess.run( ["git", "rev-parse", "--short", "HEAD"], cwd=repo, capture_output=True, text=True, check=True).stdout.strip() env = dict(os.environ) env.update({ "CODEX_BIN_DIR": bindir, "UBUNTU_VERSION": "24.04", "CODEX_VERSION": "0.164.0-alpha.9", "REPO_COMMIT": head, "UBUNTU_IMAGE_SHA256": "a" * 64, }) proc = subprocess.run( ["bash", os.path.join(guest, "build.sh"), "--dry-run", "out"], cwd=repo, capture_output=True, text=True, env=env) self.assertEqual(proc.returncode, 0, proc.stderr) out = proc.stdout + proc.stderr for needle in ("systemd-container", "chv-cell", "chv-broker", "chv-browser", "chv-browser.service", "chv-broker.service", "useradd --system", "systemctl enable chv-broker chv-browser", "chv-authd", "git-credential-chv-authd", "systemctl enable chv-authd", "systemctl start chv-authd"): self.assertIn(needle, out, needle) # P1 文件不在本分支：干跑只警告、不失败。 self.assertIn("vsock_notify.py 不在仓库里", out) if __name__ == "__main__": unittest.main()