808 lines
32 KiB
Python
808 lines
32 KiB
Python
"""对齐 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
|
||
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):
|
||
spec = CELL.build_sandbox_spec(["true"])
|
||
self.assertEqual(spec["backend"], "userns")
|
||
self.assertEqual(spec["uid_map"], "0 1000 1")
|
||
self.assertEqual(spec["gid_map"], "0 1000 1")
|
||
self.assertEqual(spec["cell"], "chv-cell")
|
||
self.assertTrue(spec["no_new_privileges"])
|
||
|
||
def test_muse_named_capabilities_dropped(self):
|
||
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(cap, spec["capabilities_dropped"])
|
||
|
||
def test_io_uring_blocked_in_syscall_filter(self):
|
||
spec = CELL.build_sandbox_spec(["true"])
|
||
filt = spec["syscall_filter"]
|
||
self.assertIn("@system-service", filt)
|
||
for call in ("io_uring_setup", "io_uring_enter",
|
||
"io_uring_register"):
|
||
self.assertIn("~%s" % call, filt)
|
||
|
||
def test_secrets_and_browser_hidden_from_cell(self):
|
||
spec = CELL.build_sandbox_spec(
|
||
["true"],
|
||
playwright_package_dir="/usr/local/lib/python3.12/dist-packages/playwright")
|
||
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, spec["hide"])
|
||
|
||
def test_writable_binds_and_tmpfs(self):
|
||
spec = CELL.build_sandbox_spec(["true"])
|
||
# 新后端不 chroot：根就是客机根，workspace/.codex/broker
|
||
# 直接就是同名目录（单元内 root 即 chv，属主不变）。
|
||
self.assertEqual(spec["workspace"], "/home/chv/workspace")
|
||
self.assertEqual(spec["codex_home"], "/home/chv/.codex")
|
||
self.assertEqual(spec["broker_dir"], "/run/chv")
|
||
self.assertEqual(spec["tmpfs"], ["/tmp"])
|
||
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.assertEqual(
|
||
CELL.build_sandbox_spec(["true"])["network"], "veth")
|
||
self.assertEqual(
|
||
CELL.build_sandbox_spec(["true"], network="shared")["network"],
|
||
"shared")
|
||
with self.assertRaises(ValueError):
|
||
CELL.build_sandbox_spec(["true"], network="macvlan")
|
||
with self.assertRaises(ValueError):
|
||
CELL.build_sandbox_spec([])
|
||
|
||
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):
|
||
spec = CELL.build_sandbox_spec(["true"])
|
||
real = CELL._forwarded_env()
|
||
# 规约只印名不印值：白名单里的值一律 <redacted>。
|
||
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)
|
||
|
||
|
||
def test_dns_resolv_conf_upstream_bind(self):
|
||
spec = CELL.build_sandbox_spec(["true"])
|
||
self.assertEqual(spec["dns_resolv_conf"], "upstream-bind")
|
||
|
||
|
||
class CellResolvTest(unittest.TestCase):
|
||
def setUp(self):
|
||
self.tmp = tempfile.mkdtemp(prefix="cell-resolv-")
|
||
|
||
def tearDown(self):
|
||
shutil.rmtree(self.tmp, ignore_errors=True)
|
||
|
||
def test_upstream_bind_args(self):
|
||
upstream = os.path.join(self.tmp, "resolv.conf")
|
||
with open(upstream, "w", encoding="utf-8") as f:
|
||
f.write("nameserver 8.8.8.8\nnameserver 1.1.1.1\n")
|
||
target = os.path.join(self.tmp, "target-resolv.conf")
|
||
with open(target, "w", encoding="utf-8") as f:
|
||
f.write("nameserver 127.0.0.53\n")
|
||
self.assertEqual(CELL._resolv_bind(upstream, target),
|
||
["--bind", upstream, target])
|
||
|
||
def test_symlink_target_resolved(self):
|
||
upstream = os.path.join(self.tmp, "resolv.conf")
|
||
with open(upstream, "w", encoding="utf-8") as f:
|
||
f.write("nameserver 8.8.8.8\n")
|
||
real = os.path.join(self.tmp, "stub-resolv.conf")
|
||
with open(real, "w", encoding="utf-8") as f:
|
||
f.write("nameserver 127.0.0.53\n")
|
||
link = os.path.join(self.tmp, "resolv-link")
|
||
os.symlink(real, link)
|
||
self.assertEqual(CELL._resolv_bind(upstream, link),
|
||
["--bind", upstream, real])
|
||
|
||
def test_missing_or_no_nameserver_returns_none(self):
|
||
target = os.path.join(self.tmp, "target-resolv.conf")
|
||
with open(target, "w", encoding="utf-8") as f:
|
||
f.write("nameserver 127.0.0.53\n")
|
||
self.assertIsNone(CELL._resolv_bind(
|
||
os.path.join(self.tmp, "no-such-file"), target))
|
||
empty = os.path.join(self.tmp, "empty-resolv.conf")
|
||
with open(empty, "w", encoding="utf-8") as f:
|
||
f.write("search .\n")
|
||
self.assertIsNone(CELL._resolv_bind(empty, target))
|
||
|
||
|
||
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 的子进程写映射：内 0 → 子进程
|
||
的外层 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)
|
||
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.split() == ["0", str(uid), "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:
|
||
self.skipTest("userns 不可用，跳过 live 验证")
|
||
self.assertEqual(code, 0)
|
||
|
||
|
||
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_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)
|
||
ipt = [c for c in cmds if "iptables" in c]
|
||
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)
|
||
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_iface_has_ip_uses_real_loopback(self):
|
||
# 真实接口断言（lo 一定有 127.0.0.1）：校验 _iface_has_ip
|
||
# 的解析逻辑；不存在的接口/地址返回 False。
|
||
self.assertTrue(CELL._iface_has_ip("lo", "127.0.0.1"))
|
||
self.assertFalse(CELL._iface_has_ip("lo", "10.200.0.1"))
|
||
self.assertFalse(CELL._iface_has_ip("no-such-iface-xyz",
|
||
"127.0.0.1"))
|
||
|
||
|
||
class BrokerTest(unittest.TestCase):
|
||
def _fake_proc(self, root, mapping):
|
||
"""mapping: pid -> ns 标识。建出 /proc/<pid>/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.assertIsInstance(resp["a11y"], str)
|
||
self.assertIn("Example Domain", resp["a11y"])
|
||
snap = agent.snapshot()
|
||
self.assertTrue(snap["ok"])
|
||
shot = agent.screenshot()
|
||
self.assertTrue(shot["ok"])
|
||
self.assertTrue(shot["path"].endswith(".jpg"))
|
||
self.assertEqual(shot["format"], "jpeg")
|
||
self.assertEqual(shot["quality"], 70)
|
||
self.assertEqual(shot["width"], 1280)
|
||
self.assertTrue(shot["size"] > 0)
|
||
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 aria_snapshot(self):
|
||
# 当前 Playwright（1.64.1）：快照是字符串，不是 dict 树。
|
||
return ("- heading \"Example Domain\" [level=1]\n"
|
||
"- paragraph \"Example\"")
|
||
|
||
def screenshot(self, path=None, type=None, quality=None):
|
||
data = b"\xff\xd8\xff\xe0FAKEJPEG"
|
||
if path:
|
||
with open(path, "wb") as fh:
|
||
fh.write(data)
|
||
return data
|
||
|
||
def close(self):
|
||
self._closed = True
|
||
|
||
|
||
class FakeContext:
|
||
"""阶段 3：持久上下文（launch_persistent_context 的返回值）。"""
|
||
|
||
def __init__(self, page):
|
||
self._page = page
|
||
|
||
def new_page(self):
|
||
return self._page
|
||
|
||
def close(self):
|
||
pass
|
||
|
||
|
||
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)
|
||
self.chromium.launch_persistent_context.return_value = \
|
||
FakeContext(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)
|
||
spec = json.loads(buf.getvalue())
|
||
self.assertEqual(spec["backend"], "userns")
|
||
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)
|
||
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",
|
||
"chv-data-mount.sh", "chv-daemon.py",
|
||
"chv-execd.py", "chv-syncd.py"):
|
||
shutil.copy2(ROOT / "chv-guest" / name,
|
||
os.path.join(guest, name))
|
||
shutil.copy2(ROOT / "host.py", os.path.join(repo, "host.py"))
|
||
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-daemon chv-data-mount chv-execd chv-syncd",
|
||
"chv-daemon.service", "chv-execd.service",
|
||
"chv-syncd.service"):
|
||
self.assertIn(needle, out, needle)
|
||
# P1 文件不在本分支：干跑只警告、不失败。
|
||
self.assertIn("vsock_notify.py 不在仓库里", out)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|