host
b41fb67026
chv-cell 须 root（nspawn + 配网），执行器经 SSH 以 chv 身份调， 此前无 sudo 即 rc=2。改 _guest_host_command 在单元开时前缀 sudo （探针 p2.log 的手动调用即 sudo chv-cell run；chv 的 NOPASSWD sudo 来自 seed users 块）。关时原命令不变。
181 lines
7.2 KiB
Python
181 lines
7.2 KiB
Python
"""round-11 集成补洞：执行器接 nspawn 单元（P2 差距表 §4）。
|
||
|
||
只测 _guest_host_command 的命令构造与 CHV_CELL_ENABLED 开关：
|
||
单元开（默认）时 host.py 调用改为 sudo chv-cell run -- python3
|
||
/app/host.py --state … --codex-home … --sandbox
|
||
danger-full-access …（sudo 不可省：chv-cell 须 root，执行器以
|
||
chv 身份调；round-11 真机修复）；单元关时走原命令；非法开关值即
|
||
ValueError。只用 mock，不启动真实 VM、不连真实网络。
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import tempfile
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
try:
|
||
from cloud.chv import CloudHypervisorExecutor
|
||
except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
||
from cloud.chv import CloudHypervisorExecutor
|
||
|
||
|
||
class FakeResult:
|
||
def __init__(self, returncode=0, stdout="", stderr=""):
|
||
self.returncode = returncode
|
||
self.stdout = stdout
|
||
self.stderr = stderr
|
||
|
||
|
||
class FakeProc:
|
||
def __init__(self, alive=True):
|
||
self._alive = alive
|
||
|
||
def poll(self):
|
||
return None if self._alive else 1
|
||
|
||
|
||
def _write(path, data=b"x"):
|
||
with open(path, "wb") as fh:
|
||
fh.write(data)
|
||
return path
|
||
|
||
|
||
class CellCase(unittest.TestCase):
|
||
def setUp(self):
|
||
self.tmp = tempfile.TemporaryDirectory()
|
||
self.dir = self.tmp.name
|
||
self.binary = _write(os.path.join(self.dir, "cloud-hypervisor"))
|
||
self.ch_remote = _write(os.path.join(self.dir, "ch-remote"))
|
||
self.kernel = _write(os.path.join(self.dir, "vmlinux"))
|
||
self.rootfs = _write(os.path.join(self.dir, "disk.raw"),
|
||
b"R" * 100)
|
||
self.ssh_key = _write(os.path.join(self.dir, "id_chv"))
|
||
self.kvm = _write(os.path.join(self.dir, "kvm"))
|
||
self.cpuinfo = os.path.join(self.dir, "cpuinfo")
|
||
with open(self.cpuinfo, "w") as fh:
|
||
fh.write("flags : fpu vmx sse\n")
|
||
self.state_dir = os.path.join(self.dir, "state")
|
||
|
||
def tearDown(self):
|
||
self.tmp.cleanup()
|
||
|
||
def make_exec(self, env_extra=None):
|
||
env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888",
|
||
"CHV_HOST_KEY_MODE": "keyscan",
|
||
"CHV_SSH_FORWARD": "1"}
|
||
if env_extra:
|
||
env.update(env_extra)
|
||
return CloudHypervisorExecutor(
|
||
binary=self.binary, ch_remote=self.ch_remote,
|
||
kernel=self.kernel, rootfs=self.rootfs,
|
||
ssh_key=self.ssh_key, kvm_path=self.kvm,
|
||
cpuinfo_path=self.cpuinfo, state_dir=self.state_dir,
|
||
runner=lambda cmd, timeout=None: FakeResult(0, "", ""),
|
||
launcher=lambda cmd: FakeProc(alive=True),
|
||
api_client=lambda *a, **k: (200, {}),
|
||
sleeper=lambda s: None, env=env)
|
||
|
||
|
||
class CellCommandTest(CellCase):
|
||
def test_cell_on_by_default(self):
|
||
# 开关未设即默认开：完整命令与契约示例一致。
|
||
ex = self.make_exec()
|
||
cmd = ex._guest_host_command("status", "--task", "t1")
|
||
self.assertEqual(cmd, [
|
||
"sudo", "chv-cell", "run", "--",
|
||
"python3", "/app/host.py",
|
||
"--state", "/home/chv/.codex/host-state.json",
|
||
"--codex-home", "/home/chv/.codex",
|
||
"--sandbox", "danger-full-access",
|
||
"status", "--task", "t1"])
|
||
|
||
def test_cell_on_explicit_1(self):
|
||
ex = self.make_exec(env_extra={"CHV_CELL_ENABLED": "1"})
|
||
cmd = ex._guest_host_command("status", "--task", "t1")
|
||
self.assertEqual(cmd[:4], ["sudo", "chv-cell", "run", "--"])
|
||
self.assertIn("--sandbox", cmd)
|
||
self.assertEqual(
|
||
cmd[cmd.index("--sandbox") + 1], "danger-full-access")
|
||
|
||
def test_cell_off_returns_original_command(self):
|
||
# 关时走原命令：一个 token 也不多不少（旧镜像逃生口）。
|
||
ex = self.make_exec(env_extra={"CHV_CELL_ENABLED": "0"})
|
||
cmd = ex._guest_host_command("status", "--task", "t1")
|
||
self.assertEqual(cmd, ["python3", "/app/host.py",
|
||
"status", "--task", "t1"])
|
||
self.assertNotIn("chv-cell", cmd)
|
||
self.assertNotIn("--state", cmd)
|
||
self.assertNotIn("--codex-home", cmd)
|
||
self.assertNotIn("--sandbox", cmd)
|
||
|
||
def test_cell_on_preserves_original_params(self):
|
||
# 原参数一个不少：model/provider/effort/approval/
|
||
# 调用方 sandbox 与子命令全部保留且顺序不变；强制的
|
||
# danger-full-access 放全局参数最后，盖过调用方的
|
||
# read-only（argparse 后胜出），read-only 本身仍在。
|
||
ex = self.make_exec()
|
||
cmd = ex._guest_host_command(
|
||
"dispatch", "--name", "t", "--task", "x",
|
||
model="m1", effort="low", approval="ask",
|
||
model_provider="sub2api", sandbox="read-only")
|
||
self.assertEqual(cmd, [
|
||
"sudo", "chv-cell", "run", "--",
|
||
"python3", "/app/host.py",
|
||
"--state", "/home/chv/.codex/host-state.json",
|
||
"--codex-home", "/home/chv/.codex",
|
||
"--model", "m1",
|
||
"--model-provider", "sub2api",
|
||
"--effort", "low",
|
||
"--approval-policy", "untrusted",
|
||
"--sandbox", "read-only",
|
||
"--sandbox", "danger-full-access",
|
||
"dispatch", "--name", "t", "--task", "x"])
|
||
# --state/--codex-home 紧跟 host.py 路径。
|
||
host_idx = cmd.index("/app/host.py")
|
||
self.assertEqual(
|
||
cmd[host_idx + 1:host_idx + 5],
|
||
["--state", "/home/chv/.codex/host-state.json",
|
||
"--codex-home", "/home/chv/.codex"])
|
||
|
||
def test_sandbox_only_in_cell_path(self):
|
||
# 调用方没给 sandbox、approval 也没带时：单元路径
|
||
# 有且仅有一个 --sandbox（danger-full-access），
|
||
# 非单元路径一个也没有。
|
||
cell_on = self.make_exec()
|
||
on_cmd = cell_on._guest_host_command("status", "--task", "t1")
|
||
self.assertEqual(on_cmd.count("--sandbox"), 1)
|
||
self.assertEqual(
|
||
on_cmd[on_cmd.index("--sandbox") + 1],
|
||
"danger-full-access")
|
||
cell_off = self.make_exec(
|
||
env_extra={"CHV_CELL_ENABLED": "0"})
|
||
off_cmd = cell_off._guest_host_command(
|
||
"status", "--task", "t1")
|
||
self.assertEqual(off_cmd.count("--sandbox"), 0)
|
||
|
||
def test_cell_invalid_value_raises(self):
|
||
# 只认 0/1：非法值即 ValueError，不静默 fallback。
|
||
for bad in ("yes", "2", "", " true"):
|
||
ex = self.make_exec(
|
||
env_extra={"CHV_CELL_ENABLED": bad})
|
||
with self.assertRaises(ValueError):
|
||
ex._guest_host_command("status", "--task", "t1")
|
||
|
||
def test_cell_wiring_reaches_ssh_command(self):
|
||
# 接线落到 SSH 层：build_ssh_command 的远端命令里
|
||
# sudo chv-cell 前缀在、python3 在其后。
|
||
ex = self.make_exec()
|
||
inner = ex._guest_host_command("status", "--task", "t1")
|
||
cmd = ex.build_ssh_command({"id": "t1"}, inner)
|
||
self.assertIn("sudo", cmd)
|
||
self.assertIn("chv-cell", cmd)
|
||
self.assertIn("danger-full-access", cmd)
|
||
self.assertLess(cmd.index("sudo"), cmd.index("chv-cell"))
|
||
self.assertLess(cmd.index("chv-cell"), cmd.index("python3"))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|