host
06c3e172fe
restore_vm 此前既不校验 net 冲突、也不登记 net.json： 已登记 VM 的 tap/mac/ip 会被恢复出的 VM 静默复用，且 恢复出的 VM 自身不登记、后来的 wake 也看不到它，两边 都漏判。现与 wake 同款：_validate_restore 在起 VMM 前 校验冲突（net_tap_conflict + vm_failed 审计）， _restore_boot 在 restore 成功后登记 net.json，登记失败 审计 net_registry_write_failed + 拒绝恢复（P2-19 语义， stop_vm 清半恢复 VM）。 hostkeys.write_seed/_apply_seed_extras 的 meta-data、 user-data、network-config 改走 atomic_file（mkstemp 0600 → 写 → fsync → rename）：user-data 含 host 私钥明文， 直接写 + 事后 chmod 会在 chmod 前以 umask 权限短暂可读； 半截文件本身到不了 ISO（唯一消费者 genisoimage 只在 write_seed 返回后跑），原子写同时堵住权限窗口与半截残留。 回归测试 6 个：restore 冲突拒起/恢复后登记/登记失败 拒起/无 net_json 不登记；seed 走原子写（0600、无 tmp 残留）/写失败不留半截文件。CHV 全套件 395 测试通过。
760 lines
34 KiB
Python
760 lines
34 KiB
Python
"""CHV 补缺口轨测试（堆叠于 #221，base=feat-chv-optimize）。
|
||
|
||
只用标准库与 mock runner/launcher/api_client，绝不启动真实
|
||
VM：当前环境没有可跑真 VM 的宿主（生产主机是 CVM、无
|
||
/dev/kvm、无嵌套虚拟化；本 VM 容器内同样无 /dev/kvm）。
|
||
真实 VM 启动、快照恢复、压测一律不在本文件声称已验证。
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import unittest
|
||
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
|
||
import yaml
|
||
|
||
try:
|
||
from cloud.chv.hostkeys import (
|
||
build_seed_user_data,
|
||
generate_host_keypair,
|
||
host_key_paths,
|
||
known_hosts_line,
|
||
write_seed,
|
||
)
|
||
from cloud.chv.experimental_pool import WarmPool
|
||
from cloud.sandbox import (
|
||
ChvFailure,
|
||
CloudHypervisorExecutor,
|
||
_parse_memory_bytes,
|
||
)
|
||
from cloud.worker import Worker, _call_goal_resume_breakpoint
|
||
except ImportError:
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
||
from cloud.chv.hostkeys import (
|
||
build_seed_user_data,
|
||
generate_host_keypair,
|
||
host_key_paths,
|
||
known_hosts_line,
|
||
write_seed,
|
||
)
|
||
from cloud.chv.experimental_pool import WarmPool
|
||
from cloud.sandbox import (
|
||
ChvFailure,
|
||
CloudHypervisorExecutor,
|
||
_parse_memory_bytes,
|
||
)
|
||
from cloud.worker import Worker, _call_goal_resume_breakpoint
|
||
|
||
|
||
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 terminate(self):
|
||
self._alive = False
|
||
|
||
def wait(self, timeout=None):
|
||
return 0
|
||
|
||
def kill(self):
|
||
self._alive = False
|
||
|
||
|
||
def _write(path, content=b"x"):
|
||
with open(path, "wb") as fh:
|
||
fh.write(content)
|
||
os.chmod(path, 0o755)
|
||
return path
|
||
|
||
|
||
class ChvGapCase(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")
|
||
self.calls = []
|
||
self.audit = []
|
||
|
||
def tearDown(self):
|
||
self.tmp.cleanup()
|
||
|
||
def _default_runner(self, cmd, timeout=None):
|
||
self.calls.append(list(cmd))
|
||
if cmd and cmd[0] == "ssh":
|
||
if cmd[-1] == "true":
|
||
return FakeResult(0, "", "")
|
||
if "goal-resume" in cmd:
|
||
return FakeResult(0, "goal met after rounds\n", "")
|
||
if "goal" in cmd:
|
||
return FakeResult(0, "goal met after rounds\n", "")
|
||
return FakeResult(0, "", "")
|
||
return FakeResult(0, "", "")
|
||
|
||
def _default_api(self, socket_path, method, path, payload=None):
|
||
return (200, {})
|
||
|
||
def _default_launcher(self, cmd):
|
||
return FakeProc(alive=True)
|
||
|
||
def make_exec(self, runner=None, env_extra=None, **overrides):
|
||
env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888",
|
||
# 实验室宣称：mock runner 假装回环端口已转发到客机
|
||
#（feat-chv-fix 起，执行器不再无条件假设回环能通）。
|
||
# 实验室显式 opt-in keyscan：生产默认已是
|
||
# provisioned；本套件旧 fake 不实现 ssh-keygen/
|
||
# genisoimage，测的不是 host key 模式，故显式
|
||
# 声明实验室路径，绝不把它写成生产默认。
|
||
"CHV_HOST_KEY_MODE": "keyscan",
|
||
"CHV_SSH_FORWARD": "1"}
|
||
if env_extra:
|
||
env.update(env_extra)
|
||
kwargs = dict(
|
||
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=runner or self._default_runner,
|
||
launcher=self._default_launcher,
|
||
api_client=self._default_api,
|
||
sleeper=lambda s: None, env=env)
|
||
kwargs.update(overrides)
|
||
ex = CloudHypervisorExecutor(**kwargs)
|
||
ex.audit_callback = lambda e, t, d, k=None: self.audit.append(
|
||
(e, t.get("id") if isinstance(t, dict) else t, d, k))
|
||
return ex
|
||
|
||
|
||
# -- (1) worker 断点接线 ---------------------------------------------
|
||
class BreakpointWiringTest(ChvGapCase):
|
||
def test_call_helper_passes_model_provider_when_accepted(self):
|
||
seen = {}
|
||
|
||
class FakeChv:
|
||
def goal_resume_breakpoint_task(self, task, model_provider=None):
|
||
seen["provider"] = model_provider
|
||
return {"status": "completed", "task_id": task["id"]}
|
||
|
||
out = _call_goal_resume_breakpoint(
|
||
FakeChv(), {"id": "bp1"}, model_provider="prov-x")
|
||
self.assertEqual(out["status"], "completed")
|
||
self.assertEqual(seen["provider"], "prov-x")
|
||
|
||
def test_call_helper_single_arg_executor_still_works(self):
|
||
class FakeOld:
|
||
def goal_resume_breakpoint_task(self, task):
|
||
return {"status": "completed", "task_id": task["id"]}
|
||
|
||
out = _call_goal_resume_breakpoint(
|
||
FakeOld(), {"id": "bp2"}, model_provider="prov-x")
|
||
self.assertEqual(out["status"], "completed")
|
||
|
||
def test_chv_breakpoint_accepts_model_provider(self):
|
||
ex = self.make_exec()
|
||
out = ex.goal_resume_breakpoint_task(
|
||
{"id": "g1", "prompt": "续跑"}, model_provider="prov-x")
|
||
self.assertEqual(out["status"], "completed")
|
||
self.assertTrue(out["goal_met"])
|
||
# 客内命令带 --model-provider（网关 key 本身不进命令值）。
|
||
ssh_calls = [c for c in self.calls if c and c[0] == "ssh"]
|
||
self.assertTrue(any("--model-provider" in c for c in ssh_calls))
|
||
|
||
def test_worker_breakpoint_branch_calls_generic_executor(self):
|
||
# feat-chv-fix（P2-3）：不再用源码 grep 充数，改为行为
|
||
# 断言——真实 Worker._run_attempt 走通用断点分支，调到
|
||
# 非 Local 执行器的 goal_resume_breakpoint_task，并把
|
||
# 结果原样返回（参数、守卫顺序、返回值都覆盖）。
|
||
from cloud.server import init_db
|
||
db_path = os.path.join(self.dir, "control.db")
|
||
init_db(db_path)
|
||
seen = {}
|
||
|
||
class FakeGeneric:
|
||
def goal_resume_breakpoint_task(self, task,
|
||
model_provider=None,
|
||
extra_env=None):
|
||
seen["provider"] = model_provider
|
||
seen["task_id"] = task["id"]
|
||
return {"executor": "generic", "task_id": task["id"],
|
||
"status": "completed", "stdout": "",
|
||
"stderr": "", "returncode": 0,
|
||
"host_status": None, "goal_met": True,
|
||
"goal_error": None}
|
||
|
||
worker = Worker(db_path, "/fake/host.py",
|
||
executor=FakeGeneric(),
|
||
env={"XCM_API_KEY": "test-placeholder"})
|
||
task = {"id": "bp-behav", "project_id": "p1",
|
||
"prompt": "续跑", "goal_mode": 1,
|
||
"_goal_resume_breakpoint": True}
|
||
result, error = worker._run_attempt(task)
|
||
self.assertIsNone(error)
|
||
self.assertEqual(result["status"], "completed")
|
||
self.assertTrue(result["goal_met"])
|
||
self.assertEqual(seen["task_id"], "bp-behav")
|
||
self.assertNotIn("executor 不支持", str(result))
|
||
|
||
|
||
# -- (2) reconcile_orphans 接线 ---------------------------------------
|
||
class ReconcileWiringTest(ChvGapCase):
|
||
def test_worker_startup_and_close_call_reconcile(self):
|
||
from cloud.server import init_db
|
||
db_path = os.path.join(self.dir, "control.db")
|
||
init_db(db_path)
|
||
ex = self.make_exec()
|
||
# 手工造一个 orphan：先建池写 registry，再新开池载入。
|
||
pool1 = WarmPool(ex)
|
||
pool1._entries["chv-old"] = {
|
||
"state": "warm", "vm": "chv-old",
|
||
"task": {"id": "old"}, "snapshot_url": None}
|
||
pool1._save_registry()
|
||
self.assertTrue(
|
||
os.path.exists(os.path.join(self.state_dir, "warm-pool.json")))
|
||
# Worker 构造即 startup reconcile：用假 shutdown 记录。
|
||
# 本轨 worker 不再持有 _warm_pool（生产走 orphans
|
||
# facade）；实验池只用于本测试造 registry 种子。
|
||
shutdowns = []
|
||
orig_shutdown = ex.shutdown_vm
|
||
ex.shutdown_vm = lambda name: shutdowns.append(name) or True
|
||
worker = Worker(db_path, "/fake/host.py", executor=ex,
|
||
env={"XCM_API_KEY": "test-placeholder"})
|
||
self.assertIn("chv-old", shutdowns)
|
||
self.assertFalse(hasattr(worker, "_warm_pool"))
|
||
with open(os.path.join(self.state_dir, "warm-pool.json")) as fh:
|
||
self.assertEqual(json.load(fh), {})
|
||
worker.close()
|
||
# close 时再次 reconcile 不炸（池已空，返回 [] 即可）。
|
||
self.assertEqual(worker._reconcile_warm_pool("shutdown"), [])
|
||
|
||
def test_reconcile_failure_does_not_break_close(self):
|
||
from cloud.server import init_db
|
||
import cloud.chv.orphans as orphans
|
||
db_path = os.path.join(self.dir, "control.db")
|
||
init_db(db_path)
|
||
ex = self.make_exec()
|
||
worker = Worker(db_path, "/fake/host.py", executor=ex,
|
||
env={"XCM_API_KEY": "test-placeholder"})
|
||
# facade 抛错时 close 仍不炸（worker 方法内每次现取
|
||
# facade，猴子补丁模块属性即生效）。
|
||
orig = orphans.reconcile_orphans_for_executor
|
||
orphans.reconcile_orphans_for_executor = lambda ex: (
|
||
(_ for _ in ()).throw(RuntimeError("boom")))
|
||
try:
|
||
worker.close() # 不抛即通过
|
||
finally:
|
||
orphans.reconcile_orphans_for_executor = orig
|
||
|
||
|
||
# -- (3) SSH host key 预置 ---------------------------------------------
|
||
class HostKeyProvisionTest(ChvGapCase):
|
||
def _fake_keygen_runner(self, cmd, timeout=None):
|
||
self.calls.append(list(cmd))
|
||
if cmd and cmd[0] == "ssh-keygen":
|
||
priv = cmd[cmd.index("-f") + 1]
|
||
with open(priv, "w") as fh:
|
||
fh.write("-----BEGIN OPENSSH PRIVATE KEY-----\nFAKE\n")
|
||
os.chmod(priv, 0o600)
|
||
with open(priv + ".pub", "w") as fh:
|
||
fh.write("ssh-ed25519 AAAAfake1 vm-key\n")
|
||
return FakeResult(0, "", "")
|
||
return self._default_runner(cmd, timeout)
|
||
|
||
def test_per_vm_keys_differ_and_known_hosts_pinned(self):
|
||
ex = self.make_exec(runner=self._fake_keygen_runner)
|
||
r1 = ex.provision_host_key({"id": "vm-a"})
|
||
r2 = ex.provision_host_key({"id": "vm-b"})
|
||
# 不同 VM 的私钥文件不同路径、known_hosts 不同文件。
|
||
self.assertNotEqual(r1["paths"]["private"],
|
||
r2["paths"]["private"])
|
||
self.assertNotEqual(r1["known_hosts"], r2["known_hosts"])
|
||
for r in (r1, r2):
|
||
with open(r["known_hosts"]) as fh:
|
||
line = fh.read().strip()
|
||
self.assertTrue(line.startswith("[127.0.0.1]:"))
|
||
self.assertIn("ssh-ed25519 AAAAfake1", line)
|
||
self.assertNotIn("PRIVATE KEY", line)
|
||
# 同一 VM 再次预置复用已有对（不重生成、不改 known_hosts）。
|
||
with open(r1["known_hosts"]) as fh:
|
||
before = fh.read()
|
||
ex.provision_host_key({"id": "vm-a"})
|
||
with open(r1["known_hosts"]) as fh:
|
||
self.assertEqual(fh.read(), before)
|
||
|
||
def test_seed_contains_private_key_and_is_deleted_on_stop(self):
|
||
ex = self.make_exec(runner=self._fake_keygen_runner)
|
||
r = ex.provision_host_key({"id": "vm-c"})
|
||
seed_user = os.path.join(r["seed_dir"], "user-data")
|
||
self.assertTrue(os.path.exists(seed_user))
|
||
with open(seed_user) as fh:
|
||
seed_text = fh.read()
|
||
self.assertIn("ssh_host_ed25519_key", seed_text)
|
||
self.assertIn("ssh_deletekeys: false", seed_text)
|
||
# stop 时 seed 目录与私钥删掉，known_hosts 公钥保留审计。
|
||
ex.stop_vm({"id": "vm-c"})
|
||
self.assertFalse(os.path.exists(r["seed_dir"]))
|
||
self.assertFalse(os.path.exists(r["paths"]["private"]))
|
||
self.assertTrue(os.path.exists(r["known_hosts"]))
|
||
|
||
def test_known_hosts_line_port_22_is_bare(self):
|
||
# 22 端口写裸 host 行：OpenSSH 默认端口只查裸名，
|
||
# '[host]:22' 永远匹配不上（2026-11-07 WSL 真机）。
|
||
self.assertEqual(
|
||
known_hosts_line(22, "ssh-ed25519 AAAAx comment",
|
||
host="192.168.10.10"),
|
||
"192.168.10.10 ssh-ed25519 AAAAx")
|
||
self.assertEqual(
|
||
known_hosts_line(2222, "ssh-ed25519 AAAAx comment"),
|
||
"[127.0.0.1]:2222 ssh-ed25519 AAAAx")
|
||
|
||
def test_known_hosts_line_rejects_bad_key(self):
|
||
with self.assertRaises(ValueError):
|
||
known_hosts_line(2222, "not-a-key")
|
||
self.assertEqual(
|
||
known_hosts_line(2222, "ssh-ed25519 AAAAx comment"),
|
||
"[127.0.0.1]:2222 ssh-ed25519 AAAAx")
|
||
|
||
def test_seed_user_data_shape(self):
|
||
text = build_seed_user_data("PRIV\n", "ssh-ed25519 AAAAx\n")
|
||
self.assertIn("/etc/ssh/ssh_host_ed25519_key", text)
|
||
self.assertIn("ssh_deletekeys: false", text)
|
||
with self.assertRaises(ValueError):
|
||
build_seed_user_data("", "ssh-ed25519 AAAAx")
|
||
|
||
def test_seed_user_data_has_no_runcmd(self):
|
||
# 根因回归（sync-test 残留，2026-11-06 实测）：seed
|
||
# 的 user-data 里绝不能带 runcmd——cloud-init 不跨配
|
||
# 置类型合并，user-data 的 runcmd 会整体替换镜像
|
||
# /etc/cloud/cloud.cfg.d/ 里的 runcmd，旧版 seed 的
|
||
# 两行就这样顶掉了 99-chv-sync.cfg 的 chv-sync boot
|
||
# 钩子，首轮启动 clone 静默不跑。sshd 重启已搬到
|
||
# 镜像自己的 runcmd（build.sh 写 99-chv-sync.cfg）。
|
||
text = build_seed_user_data("PRIV\n", "ssh-ed25519 AAAAx\n")
|
||
parsed = yaml.safe_load(text)
|
||
self.assertNotIn("runcmd", parsed)
|
||
self.assertNotIn("runcmd:", text)
|
||
self.assertNotIn("systemctl restart", text)
|
||
self.assertNotIn("merge_how", parsed)
|
||
# write_files 两条 key 条目仍在，chmod 靠其 0600 保证。
|
||
self.assertEqual(len(parsed["write_files"]), 2)
|
||
self.assertEqual(
|
||
parsed["write_files"][0]["permissions"], "0600")
|
||
|
||
def test_write_seed_with_secret_has_no_runcmd(self):
|
||
# 带 secret_env 时 guest-env 条目仍须并进同一个
|
||
# write_files 列表（单 write_files 键），且 user-data
|
||
# 里仍不能出现 runcmd。
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
seed_dir = os.path.join(tmp, "seed")
|
||
write_seed(seed_dir, "vm1", "PRIV", "ssh-ed25519 AAAAx",
|
||
secret_env={"GITEA_TOKEN": "s3cr3t v"})
|
||
with open(os.path.join(seed_dir, "user-data")) as fh:
|
||
content = fh.read()
|
||
self.assertEqual(content.count("write_files:"), 1)
|
||
parsed = yaml.safe_load(content)
|
||
self.assertNotIn("runcmd", parsed)
|
||
self.assertEqual(len(parsed["write_files"]), 3)
|
||
paths = [e["path"] for e in parsed["write_files"]]
|
||
self.assertIn("/etc/chv/guest-env", paths)
|
||
# 密钥值经 shlex.quote 引用后才进文件（值本身在 guest-env
|
||
# 条目内是设计内的，见 hostkeys.write_seed）。
|
||
self.assertIn("GITEA_TOKEN='s3cr3t v'", content)
|
||
|
||
def test_write_seed_uses_atomic_write(self):
|
||
# feat-chv-cover100：seed 暂存文件（user-data 含 host
|
||
# 私钥明文）此前直接 open("w") 再 chmod 0600——崩溃
|
||
# 半截文件本身到不了 ISO（唯一消费者 genisoimage 只
|
||
# 在 write_seed 返回后跑，失败重试会整文件重写），
|
||
# 但 chmod 之前文件以 umask 权限（常为 0644）可读，
|
||
# 同 uid 进程能在该窗口读到私钥；现走 atomic_file
|
||
#（mkstemp 0600 → 写 → fsync → rename），文件从
|
||
# 创建起就是 0600，且写失败不留半截文件。
|
||
import stat as _stat
|
||
from unittest import mock
|
||
|
||
import cloud.chv.hostkeys as hostkeys_mod
|
||
from cloud.chv.atomic_file import atomic_write_text as _real
|
||
calls = []
|
||
|
||
def _spy(path, text, mode=0o600):
|
||
calls.append(path)
|
||
return _real(path, text, mode=mode)
|
||
|
||
with (tempfile.TemporaryDirectory() as tmp,
|
||
mock.patch.object(hostkeys_mod, "atomic_write_text",
|
||
side_effect=_spy)):
|
||
seed_dir = os.path.join(tmp, "seed")
|
||
write_seed(seed_dir, "vm1", "PRIV",
|
||
"ssh-ed25519 AAAAx",
|
||
network_config="version: 2\n")
|
||
self.assertEqual(
|
||
sorted(os.path.basename(p) for p in calls),
|
||
["meta-data", "network-config", "user-data"])
|
||
for name in ("meta-data", "user-data", "network-config"):
|
||
mode = _stat.S_IMODE(os.stat(
|
||
os.path.join(seed_dir, name)).st_mode)
|
||
self.assertEqual(mode, 0o600, name)
|
||
leftovers = [n for n in os.listdir(seed_dir)
|
||
if n.startswith(".tmp-")]
|
||
self.assertEqual(leftovers, [])
|
||
|
||
def test_write_seed_failure_leaves_no_partial_file(self):
|
||
# feat-chv-cover100：原子写的回归——写 user-data 中途
|
||
# 失败时不留半截 user-data、不留 tmp 残留；旧的直接
|
||
# 写会在崩溃后留半截文件（虽到不了 ISO，但含私钥明
|
||
# 文的半截文件滞留 seed 目录本身即泄漏面）。
|
||
from unittest import mock
|
||
|
||
import cloud.chv.hostkeys as hostkeys_mod
|
||
real = hostkeys_mod.atomic_write_text
|
||
|
||
def _fail_on_user_data(path, text, mode=0o600):
|
||
if path.endswith("user-data"):
|
||
raise OSError("disk boom")
|
||
return real(path, text, mode=mode)
|
||
|
||
with (tempfile.TemporaryDirectory() as tmp,
|
||
mock.patch.object(hostkeys_mod, "atomic_write_text",
|
||
side_effect=_fail_on_user_data)):
|
||
seed_dir = os.path.join(tmp, "seed")
|
||
os.makedirs(seed_dir)
|
||
|
||
with self.assertRaises(OSError):
|
||
write_seed(seed_dir, "vm1", "PRIV",
|
||
"ssh-ed25519 AAAAx")
|
||
self.assertFalse(os.path.exists(
|
||
os.path.join(seed_dir, "user-data")))
|
||
leftovers = [n for n in os.listdir(seed_dir)
|
||
if n.startswith(".tmp-")]
|
||
self.assertEqual(leftovers, [])
|
||
|
||
def test_keyscan_is_weak_fallback_only(self):
|
||
# provisioned 模式下 wake 前必须预置；keyscan 不应被
|
||
# 当成生产路径：配置值本身可解析，且文档/代码里标弱。
|
||
ex = self.make_exec(env_extra={"CHV_HOST_KEY_MODE": "provisioned"})
|
||
self.assertEqual(ex.host_key_mode, "provisioned")
|
||
with self.assertRaises(ValueError):
|
||
self.make_exec(env_extra={"CHV_HOST_KEY_MODE": "bogus"})
|
||
|
||
|
||
# -- (4) egress tap 校验 -------------------------------------------------
|
||
class EgressValidationTest(ChvGapCase):
|
||
def test_direct_egress_without_flag_is_fail_closed(self):
|
||
ex = self.make_exec()
|
||
ex.egress_proxy = None
|
||
ex.allow_direct_egress = False
|
||
with self.assertRaises(ChvFailure) as ctx:
|
||
ex.preflight()
|
||
self.assertEqual(ctx.exception.kind, "egress_not_configured")
|
||
|
||
def test_direct_egress_with_flag_passes_and_is_audited(self):
|
||
ex = self.make_exec(env_extra={"CHV_ALLOW_DIRECT_EGRESS": "1"})
|
||
# preflight 本身不拦显式直连（守卫在 wake 的审计里）。
|
||
self.assertTrue(ex.preflight())
|
||
out = ex.goal_task({"id": "g1", "prompt": "做目标",
|
||
"goal_mode": True})
|
||
self.assertEqual(out["status"], "completed")
|
||
self.assertTrue(any(e == "egress_direct_allowed"
|
||
for e, _, _, _ in self.audit))
|
||
|
||
def test_bad_net_json_is_preflight_error(self):
|
||
ex = self.make_exec(env_extra={"CHV_NET_JSON": "{not json"})
|
||
with self.assertRaises(ChvFailure) as ctx:
|
||
ex.preflight()
|
||
self.assertEqual(ctx.exception.kind, "net_config_invalid")
|
||
|
||
|
||
# -- (5) 构建脚本 --------------------------------------------------------
|
||
class BuildScriptTest(unittest.TestCase):
|
||
def setUp(self):
|
||
self.tmp = tempfile.TemporaryDirectory()
|
||
self.dir = self.tmp.name
|
||
self.bin_dir = os.path.join(self.dir, "bin")
|
||
os.makedirs(os.path.join(self.bin_dir, "codex-resources"))
|
||
for name in ("codex", "codex-code-mode-host"):
|
||
p = os.path.join(self.bin_dir, name)
|
||
with open(p, "w") as fh:
|
||
# B-09：build.sh 会核对 codex --version 含 CODEX_VERSION，
|
||
# 假二进制须报出真实版本号。
|
||
fh.write("#!/bin/sh\necho \"codex-cli 0.164.0-alpha.9\"\n")
|
||
os.chmod(p, 0o755)
|
||
self.repo = os.path.join(self.dir, "repo")
|
||
os.makedirs(os.path.join(self.repo, "cloud"))
|
||
with open(os.path.join(self.repo, "host.py"), "w") as fh:
|
||
fh.write("# host\n")
|
||
# B-09：REPO_COMMIT 须等于仓库 HEAD 短 hash 且工作树干净，
|
||
# 假仓库须是真 git 仓库。
|
||
subprocess.run(["git", "init", "-q", self.repo],
|
||
check=True, capture_output=True)
|
||
subprocess.run(["git", "-C", self.repo, "config",
|
||
"user.email", "t@t"], check=True,
|
||
capture_output=True)
|
||
subprocess.run(["git", "-C", self.repo, "config",
|
||
"user.name", "t"], check=True,
|
||
capture_output=True)
|
||
subprocess.run(["git", "-C", self.repo, "add", "-A"],
|
||
check=True, capture_output=True)
|
||
subprocess.run(["git", "-C", self.repo, "commit", "-qm",
|
||
"init"], check=True, capture_output=True)
|
||
self.repo_commit = subprocess.run(
|
||
["git", "-C", self.repo, "rev-parse", "--short", "HEAD"],
|
||
check=True, capture_output=True, text=True).stdout.strip()
|
||
self.script = str(Path(__file__).resolve().parents[2]
|
||
/ "chv-guest" / "build.sh")
|
||
|
||
def tearDown(self):
|
||
self.tmp.cleanup()
|
||
|
||
def _run(self, args, env_extra=None):
|
||
env = dict(os.environ)
|
||
env.update({"CODEX_BIN_DIR": self.bin_dir,
|
||
"UBUNTU_VERSION": "24.04",
|
||
"CODEX_VERSION": "0.164.0-alpha.9",
|
||
"REPO_COMMIT": self.repo_commit,
|
||
# feat-chv-secfix（M4）：SHA256 必填，
|
||
# 干跑只验“已设”，用 FAKE 占位。
|
||
"UBUNTU_IMAGE_SHA256": "0" * 64,
|
||
"REPO_DIR": self.repo})
|
||
if env_extra:
|
||
env.update(env_extra)
|
||
return subprocess.run(["bash", self.script] + args,
|
||
capture_output=True, text=True, env=env)
|
||
|
||
def test_dry_run_prints_real_steps_and_builds_nothing(self):
|
||
out_dir = os.path.join(self.dir, "out")
|
||
r = self._run(["--dry-run", out_dir])
|
||
self.assertEqual(r.returncode, 0, r.stderr)
|
||
self.assertIn("qemu-img convert", r.stdout)
|
||
self.assertIn("virt-customize", r.stdout)
|
||
self.assertIn("virt-get-kernel", r.stdout)
|
||
self.assertIn("BUILDINFO", r.stdout)
|
||
self.assertIn("chv-guest-24.04-0.164.0-alpha.9-%s"
|
||
% self.repo_commit, r.stdout)
|
||
self.assertFalse(os.path.exists(os.path.join(out_dir, "disk.raw")))
|
||
|
||
def test_hostkey_dropin_generates_missing_keys_at_first_boot(self):
|
||
# B-23：Minimal 底座的 cloud.cfg 不含 ssh 模块，build.sh
|
||
# 曾写 ssh_genkeytypes: [] 彻底关掉首启 host key 生成，
|
||
# 客机无 key、sshd -t 失败、ssh.service 连败 5 次，SSH
|
||
# 全程不通。修正：首启 bootcmd 只在缺 key 时生成，绝不
|
||
# 重生成已预置的 per-VM key；写盘的命令里 [] 绝不再出现
|
||
#（注释里提历史可以，功能行里不行）。
|
||
with open(self.script) as fh:
|
||
text = fh.read()
|
||
hostkey_lines = [ln for ln in text.splitlines()
|
||
if "99-chv-hostkey.cfg" in ln
|
||
and "printf" in ln]
|
||
self.assertEqual(len(hostkey_lines), 1)
|
||
cmd = hostkey_lines[0]
|
||
self.assertIn("ssh_deletekeys: false", cmd)
|
||
self.assertIn("[ -f /etc/ssh/ssh_host_ed25519_key ] "
|
||
"|| ssh-keygen -A", cmd)
|
||
self.assertNotIn("ssh_genkeytypes", cmd)
|
||
|
||
def test_full_mode_without_kvm_fails_fast(self):
|
||
# 本 VM 容器内无 /dev/kvm，全量模式必须明确失败，
|
||
# 绝不声称已建成镜像。
|
||
r = self._run([os.path.join(self.dir, "out2")])
|
||
self.assertNotEqual(r.returncode, 0)
|
||
self.assertIn("/dev/kvm", r.stderr + r.stdout)
|
||
|
||
def test_missing_binary_fails_fast(self):
|
||
os.remove(os.path.join(self.bin_dir, "codex"))
|
||
r = self._run(["--dry-run", os.path.join(self.dir, "out3")])
|
||
self.assertNotEqual(r.returncode, 0)
|
||
|
||
|
||
# -- (5b) chv-sync 双向同步 ------------------------------------------------
|
||
# 轻量口径 + Gitea 集成（2026-11-06）：本地 bare 仓当远端替身，
|
||
# 不碰真实 Gitea、不需 token，测同步机制本身（双向通 + 分叉/
|
||
# 脏树时失败关闭）。boot 的 clone 一步需 root + guest-env，
|
||
# 不在单元测试里跑，只测 boot 无 guest-env 时静默成功与
|
||
# guest-env 缺字段时明确失败；真机验收见构建记录。
|
||
class ChvSyncTest(unittest.TestCase):
|
||
def setUp(self):
|
||
self.tmp = tempfile.TemporaryDirectory()
|
||
self.dir = self.tmp.name
|
||
self.script = str(Path(__file__).resolve().parents[2]
|
||
/ "chv-guest" / "chv-sync.sh")
|
||
self.ws = os.path.join(self.dir, "workspace")
|
||
os.makedirs(self.ws)
|
||
# 远端替身：bare 仓 + 一个已推入的初始提交。
|
||
self.remote = os.path.join(self.dir, "remote.git")
|
||
subprocess.run(["git", "init", "-q", "--bare", self.remote],
|
||
check=True, capture_output=True)
|
||
seed = os.path.join(self.dir, "seed-clone")
|
||
subprocess.run(["git", "clone", "-q", self.remote, seed],
|
||
check=True, capture_output=True)
|
||
self._git(seed, "config", "user.email", "t@t")
|
||
self._git(seed, "config", "user.name", "t")
|
||
with open(os.path.join(seed, "a.txt"), "w") as fh:
|
||
fh.write("v1\n")
|
||
self._git(seed, "add", "-A")
|
||
self._git(seed, "commit", "-qm", "v1")
|
||
self._git(seed, "push", "-q", "origin", "HEAD:main")
|
||
# 本地工作区：模拟 boot 已 clone 的结果。
|
||
self.repo = os.path.join(self.ws, "remote")
|
||
subprocess.run(["git", "clone", "-q", "--branch", "main",
|
||
self.remote, self.repo],
|
||
check=True, capture_output=True)
|
||
self._git(self.repo, "config", "user.email", "t@t")
|
||
self._git(self.repo, "config", "user.name", "t")
|
||
|
||
def tearDown(self):
|
||
self.tmp.cleanup()
|
||
|
||
def _git(self, cwd, *args):
|
||
return subprocess.run(["git", "-C", cwd] + list(args),
|
||
check=True, capture_output=True, text=True)
|
||
|
||
def _sync(self, *args, env_extra=None):
|
||
env = dict(os.environ)
|
||
env["CHV_WORKSPACE"] = self.ws
|
||
env["CHV_GUEST_ENV"] = os.path.join(self.dir, "no-guest-env")
|
||
if env_extra:
|
||
env.update(env_extra)
|
||
return subprocess.run(["bash", self.script] + list(args),
|
||
capture_output=True, text=True, env=env)
|
||
|
||
def _head(self, cwd):
|
||
return self._git(cwd, "rev-parse", "HEAD").stdout.strip()
|
||
|
||
def test_push_then_pull_round_trip(self):
|
||
# 本地改 → push → 远端可见。
|
||
with open(os.path.join(self.repo, "a.txt"), "a") as fh:
|
||
fh.write("v2-local\n")
|
||
self._git(self.repo, "add", "-A")
|
||
self._git(self.repo, "commit", "-qm", "v2-local")
|
||
r = self._sync("push", self.repo)
|
||
self.assertEqual(r.returncode, 0, r.stderr)
|
||
other = os.path.join(self.dir, "other-clone")
|
||
subprocess.run(["git", "clone", "-q", "--branch", "main",
|
||
self.remote, other],
|
||
check=True, capture_output=True)
|
||
self.assertEqual(self._head(other), self._head(self.repo))
|
||
# 远端改 → pull → 本地同步回来。
|
||
self._git(other, "config", "user.email", "t@t")
|
||
self._git(other, "config", "user.name", "t")
|
||
with open(os.path.join(other, "a.txt"), "a") as fh:
|
||
fh.write("v3-remote\n")
|
||
self._git(other, "add", "-A")
|
||
self._git(other, "commit", "-qm", "v3-remote")
|
||
self._git(other, "push", "-q", "origin", "HEAD:main")
|
||
r = self._sync("pull", self.repo)
|
||
self.assertEqual(r.returncode, 0, r.stderr)
|
||
with open(os.path.join(self.repo, "a.txt")) as fh:
|
||
self.assertIn("v3-remote", fh.read())
|
||
|
||
def test_push_new_branch_sets_upstream(self):
|
||
self._git(self.repo, "checkout", "-qb", "feature")
|
||
with open(os.path.join(self.repo, "b.txt"), "w") as fh:
|
||
fh.write("f\n")
|
||
self._git(self.repo, "add", "-A")
|
||
self._git(self.repo, "commit", "-qm", "feature")
|
||
r = self._sync("push", self.repo)
|
||
self.assertEqual(r.returncode, 0, r.stderr)
|
||
out = subprocess.run(["git", "--git-dir", self.remote,
|
||
"branch", "--list", "feature"],
|
||
check=True, capture_output=True,
|
||
text=True).stdout
|
||
self.assertIn("feature", out)
|
||
|
||
def test_diverged_pull_fails_closed(self):
|
||
# 本地与远端同分支各进一步 → pull 必须失败，不自动合并。
|
||
with open(os.path.join(self.repo, "a.txt"), "a") as fh:
|
||
fh.write("local-diverge\n")
|
||
self._git(self.repo, "add", "-A")
|
||
self._git(self.repo, "commit", "-qm", "local-diverge")
|
||
other = os.path.join(self.dir, "other2")
|
||
subprocess.run(["git", "clone", "-q", "--branch", "main",
|
||
self.remote, other],
|
||
check=True, capture_output=True)
|
||
self._git(other, "config", "user.email", "t@t")
|
||
self._git(other, "config", "user.name", "t")
|
||
with open(os.path.join(other, "a.txt"), "a") as fh:
|
||
fh.write("remote-diverge\n")
|
||
self._git(other, "add", "-A")
|
||
self._git(other, "commit", "-qm", "remote-diverge")
|
||
self._git(other, "push", "-q", "origin", "HEAD:main")
|
||
r = self._sync("pull", self.repo)
|
||
self.assertNotEqual(r.returncode, 0)
|
||
with open(os.path.join(self.repo, "a.txt")) as fh:
|
||
self.assertNotIn("remote-diverge", fh.read())
|
||
|
||
def test_dirty_tree_pull_refused(self):
|
||
with open(os.path.join(self.repo, "a.txt"), "a") as fh:
|
||
fh.write("uncommitted\n")
|
||
r = self._sync("pull", self.repo)
|
||
self.assertNotEqual(r.returncode, 0)
|
||
|
||
def test_boot_without_guest_env_is_noop(self):
|
||
r = self._sync("boot")
|
||
self.assertEqual(r.returncode, 0, r.stderr)
|
||
|
||
def test_boot_with_incomplete_guest_env_fails(self):
|
||
env_file = os.path.join(self.dir, "guest-env")
|
||
with open(env_file, "w") as fh:
|
||
fh.write("GITEA_URL=https://gitea.tose.top\n")
|
||
r = self._sync("boot", env_extra={"CHV_GUEST_ENV": env_file})
|
||
self.assertNotEqual(r.returncode, 0)
|
||
|
||
|
||
# -- (6) 中途死亡 → interrupted → 断点续跑闭环 ---------------------------
|
||
class MidDeathLoopTest(ChvGapCase):
|
||
def test_vm_death_is_interrupted_then_breakpoint_resumes(self):
|
||
state = {"dead": False}
|
||
|
||
def runner(cmd, timeout=None):
|
||
self.calls.append(list(cmd))
|
||
if cmd and cmd[0] == "ssh":
|
||
if cmd[-1] == "true":
|
||
return FakeResult(0, "", "")
|
||
if "goal-resume" in cmd and state["dead"]:
|
||
# 第二次断点调用：VM 已换新，健康返回。
|
||
return FakeResult(0, "goal met after rounds\n", "")
|
||
if "goal" in cmd:
|
||
# 第一次 goal：模拟传输中断，并把 VMM 进程标死。
|
||
state["dead"] = True
|
||
# 找到登记的 VMM 进程并标死（wake 已登记）。
|
||
for proc in list(ex._vms.values()):
|
||
proc._alive = False
|
||
raise RuntimeError("connection lost mid-task")
|
||
return FakeResult(0, "", "")
|
||
|
||
ex = self.make_exec(runner=runner)
|
||
# 第一次 goal：中途死应判 interrupted，不是 failed 死信。
|
||
out = ex.goal_task({"id": "g9", "prompt": "做目标",
|
||
"goal_mode": True})
|
||
self.assertEqual(out["status"], "interrupted")
|
||
# 断点续跑：同一执行器的断点方法能走通并完成。
|
||
state["dead"] = True # 标记已进入断点阶段（runner 按此分支）
|
||
out2 = ex.goal_resume_breakpoint_task(
|
||
{"id": "g9", "prompt": "续跑"})
|
||
self.assertEqual(out2["status"], "completed")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|