"""CHV 安全审计修复轨测试（feat-chv-secfix，堆叠于 #239）。 只用标准库与 mock runner/launcher/api_client，绝不启动真实 VM：当前环境没有可跑真 VM 的宿主。本文件只证明代码层面 的修复（密钥移出 argv、磁盘删除、默认值、校验、审计等） 成立；真机否定测试（tap 防火墙实效、unset 代理后直连、 IP spoof 后 SSH 拒绝等）一律列在就绪清单 §11，不在本 文件声称已验证。测试里只用 FAKE 占位密钥，绝无真实密钥。 """ import json import os import shlex import socket import subprocess import sys import tempfile import unittest from pathlib import Path from types import SimpleNamespace try: from cloud.chv.config import build_chv_config from cloud.chv.disk import ( check_snapshot_limits, delete_task_disk, provision_task_disk, record_snapshot, ) from cloud.chv.hardening import ( ensure_secure_state_dir, socket_is_live, validate_net_entries, validate_snapshot_url, ) from cloud.chv.hostkeys import ( build_seed_disk, dirty_marker_path, write_seed, ) from cloud.chv.experimental_pool import WarmPool from cloud.chv.ssh_transport import ( build_ssh_command, split_secret_env, ) from cloud.sandbox import ( ChvFailure, CloudHypervisorExecutor, ) except ImportError: sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from cloud.chv.config import build_chv_config from cloud.chv.disk import ( check_snapshot_limits, delete_task_disk, provision_task_disk, record_snapshot, ) from cloud.chv.hardening import ( ensure_secure_state_dir, socket_is_live, validate_net_entries, validate_snapshot_url, ) from cloud.chv.hostkeys import ( build_seed_disk, dirty_marker_path, write_seed, ) from cloud.chv.experimental_pool import WarmPool from cloud.chv.ssh_transport import ( build_ssh_command, split_secret_env, ) from cloud.sandbox import ( ChvFailure, 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 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 SecfixCase(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, "", "") return FakeResult(0, "", "") 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, "", "") if cmd and cmd[0] == "genisoimage": out = cmd[cmd.index("-output") + 1] with open(out, "wb") as fh: fh.write(b"CIDATA" * 16) return FakeResult(0, "", "") return FakeResult(0, "", "") def _default_api(self, socket_path, method, path, payload=None): return (200, {}) def make_exec(self, runner=None, env_extra=None, **overrides): # 生产默认：provisioned（不显式设 keyscan），回环仍 # 是实验室宣称（mock runner 假装转发已通）。 env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888", "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=lambda cmd: FakeProc(alive=True), 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 # -- H1 egress 代码门 + H7 tap 隔离 ------------------------------------ class EgressAndNetTest(SecfixCase): def test_egress_proxy_non_http_rejected(self): ex = self.make_exec( env_extra={"CHV_EGRESS_PROXY": "ftp://10.0.0.1:21"}) with self.assertRaises(ChvFailure) as ctx: ex.preflight() self.assertEqual(ctx.exception.kind, "config_invalid") def test_egress_proxy_without_host_rejected(self): ex = self.make_exec( env_extra={"CHV_EGRESS_PROXY": "http:///path"}) with self.assertRaises(ChvFailure) as ctx: ex.preflight() self.assertEqual(ctx.exception.kind, "config_invalid") def test_net_unknown_field_rejected(self): with self.assertRaises(ChvFailure) as ctx: validate_net_entries( [{"tap": "tap0", "mac": "aa:bb:cc:dd:ee:01", "ip": "10.0.0.2", "gateway": "10.0.0.1"}], []) self.assertEqual(ctx.exception.kind, "net_config_invalid") def test_net_intra_config_dup_rejected(self): with self.assertRaises(ChvFailure) as ctx: validate_net_entries( [{"tap": "tap0", "mac": "aa:bb:cc:dd:ee:01", "ip": "10.0.0.2"}, {"tap": "tap0", "mac": "aa:bb:cc:dd:ee:02", "ip": "10.0.0.3"}], []) self.assertEqual(ctx.exception.kind, "net_config_invalid") def test_net_cross_vm_conflict_refuses_second_wake(self): net = '[{"tap": "tap0", "mac": "aa:bb:cc:dd:ee:01",' \ '"ip": "10.0.0.2", "mask": "255.255.255.0"}]' ex = self.make_exec(env_extra={"CHV_NET_JSON": net}) out = ex.wake_vm({"id": "vm1"}) self.assertEqual(out["state"], "running") # 第一台的 net.json 已登记；第二台同 tap 即拒绝， # 且未起 VMM（launcher 只被调过一次）。 launches = [] ex.launcher = lambda cmd: launches.append(cmd) or FakeProc(True) with self.assertRaises(ChvFailure) as ctx: ex.wake_vm({"id": "vm2"}) self.assertEqual(ctx.exception.kind, "net_tap_conflict") self.assertEqual(launches, []) self.assertTrue(any(e == "vm_failed" for e, _, _, _ in self.audit)) # -- H2 密钥移出 argv --------------------------------------------------- class SecretDeliveryTest(SecfixCase): def test_secret_env_split_and_argv_refusal(self): secrets, plain = split_secret_env( {"SUB2API_API_KEY": "FAKE-KEY", "SUB2API_BASE_URL": "http://gw/"}) self.assertEqual(secrets, {"SUB2API_API_KEY": "FAKE-KEY"}) self.assertEqual(plain, {"SUB2API_BASE_URL": "http://gw/"}) with self.assertRaises(ValueError): build_ssh_command(2222, "/k", "root", None, "/kh", ["true"], extra_env={"SUB2API_API_KEY": "FAKE-KEY"}) def test_seed_writes_single_write_files_with_guest_env(self): seed_dir = os.path.join(self.dir, "seed") write_seed(seed_dir, "chv-t1", "FAKE-PRIV", "ssh-ed25519 AAAAx", secret_env={"SUB2API_API_KEY": "FAKE-KEY-123"}) text = "" with open(os.path.join(seed_dir, "user-data")) as fh: text = fh.read() # 只能有一个 write_files 键（重复键会覆盖 host key 条目）。 self.assertEqual(text.count("write_files:"), 1) self.assertIn("/etc/chv/guest-env", text) self.assertIn("SUB2API_API_KEY=", text) self.assertEqual(oct(os.stat( os.path.join(seed_dir, "user-data")).st_mode & 0o777), "0o600") def test_keyscan_with_secret_fails_closed_before_launch(self): ex = self.make_exec( env_extra={"CHV_HOST_KEY_MODE": "keyscan"}) launches = [] ex.launcher = lambda cmd: launches.append(cmd) or FakeProc(True) with self.assertRaises(ChvFailure) as ctx: ex.wake_vm({"id": "k1"}, extra_env={"SUB2API_API_KEY": "FAKE-KEY"}) self.assertEqual(ctx.exception.kind, "secret_env_requires_provisioned") self.assertEqual(launches, []) def test_provisioned_wake_deletes_host_private_key(self): ex = self.make_exec() out = ex.wake_vm({"id": "s1"}, extra_env={"SUB2API_API_KEY": "FAKE-KEY"}) self.assertEqual(out["state"], "running") vm_dir = os.path.join(self.state_dir, "chv-s1") # 宿主私钥已删（seed 盘建好后即删）；seed.iso 在且 0600； # 密钥投递有审计（只含名，不含值）。 self.assertFalse(os.path.exists( os.path.join(vm_dir, "ssh_host_ed25519_key"))) seed_iso = os.path.join(vm_dir, "seed.iso") self.assertTrue(os.path.exists(seed_iso)) self.assertEqual(oct(os.stat(seed_iso).st_mode & 0o777), "0o600") self.assertTrue(any(e == "host_key_generated" for e, _, _, _ in self.audit)) self.assertFalse(any("FAKE-KEY" in (d or "") for _, _, d, _ in self.audit)) def test_seed_iso_build_sets_0600(self): seed_dir = os.path.join(self.dir, "seed2") os.makedirs(seed_dir, exist_ok=True) out = os.path.join(self.dir, "seed2.iso") build_seed_disk(seed_dir, out, runner=self._default_runner) self.assertEqual(oct(os.stat(out).st_mode & 0o777), "0o600") # -- H3 任务盘删除 + GC -------------------------------------------------- class DiskDeletionTest(SecfixCase): def test_stop_deletes_task_disk_and_audits(self): ex = self.make_exec() disk = provision_task_disk(self.state_dir, "chv-d1", self.rootfs, "copy") self.assertTrue(os.path.exists(disk)) ex.stop_vm({"id": "d1"}) self.assertFalse(os.path.exists(disk)) self.assertTrue(any(e == "vm_disk_deleted" for e, _, _, _ in self.audit)) def test_gc_deletes_orphan_disk_only(self): ex = self.make_exec() provision_task_disk(self.state_dir, "chv-orphan", self.rootfs, "copy") live_disk = provision_task_disk(self.state_dir, "chv-live", self.rootfs, "copy") # live VM 有 socket 文件（mock 占位，不代表活 VMM， # 但 GC 的 live 判定含 socket 存在——见 gc_stale_disks # 的 socket 判定，此处只验 orphan 被删、live 不动）。 os.makedirs(self.state_dir, exist_ok=True) with open(os.path.join(self.state_dir, "chv-live.sock"), "w") as fh: fh.write("") result = ex.gc_stale_disks() self.assertIn("chv-orphan", result["deleted"]) self.assertTrue(os.path.exists(live_disk)) def test_disk_delete_failure_audited_and_marks_dirty(self): ex = self.make_exec() provision_task_disk(self.state_dir, "chv-f1", self.rootfs, "copy") from unittest import mock with mock.patch("cloud.chv.executor.delete_task_disk", side_effect=OSError("disk busy")): ex.stop_vm({"id": "f1"}) self.assertTrue(any(e == "vm_disk_delete_failed" for e, _, _, _ in self.audit)) self.assertTrue(os.path.exists( dirty_marker_path(self.state_dir, "chv-f1"))) # -- H4 默认值 ------------------------------------------------------------ class DefaultsTest(unittest.TestCase): def test_production_defaults(self): cfg = build_chv_config({}, {}) self.assertEqual(cfg.host_key_mode, "provisioned") self.assertNotEqual(cfg.state_dir, "/tmp/chv-vms") self.assertTrue(cfg.state_dir.startswith(os.path.expanduser("~"))) self.assertEqual(cfg.disk_gb, 50.0) self.assertEqual(cfg.snapshot_max_count, 10) self.assertEqual(cfg.snapshot_max_total_gb, 20.0) self.assertEqual(cfg.snapshot_retention_days, 30.0) def test_keyscan_still_explicitly_selectable(self): cfg = build_chv_config( {}, {"CHV_HOST_KEY_MODE": "keyscan"}) self.assertEqual(cfg.host_key_mode, "keyscan") def test_disk_gb_zero_or_negative_rejected(self): with self.assertRaises(ValueError): build_chv_config({}, {"CHV_DISK_GB": "0"}) with self.assertRaises(ValueError): build_chv_config({}, {"CHV_DISK_GB": "-5"}) # -- H5 崩溃残留扫描 ------------------------------------------------------- class ResidueScanTest(SecfixCase): def test_cleanup_stale_secrets_removes_residue(self): ex = self.make_exec() vm_dir = os.path.join(self.state_dir, "chv-old") os.makedirs(os.path.join(vm_dir, "seed"), exist_ok=True) with open(os.path.join(vm_dir, "ssh_host_ed25519_key"), "w") as fh: fh.write("FAKE-PRIV") with open(os.path.join(vm_dir, "seed.iso"), "wb") as fh: fh.write(b"CIDATA") cleaned = ex.cleanup_stale_secrets() self.assertIn("chv-old", cleaned) self.assertFalse(os.path.exists( os.path.join(vm_dir, "ssh_host_ed25519_key"))) self.assertFalse(os.path.exists( os.path.join(vm_dir, "seed.iso"))) self.assertTrue(any(e == "host_secret_residue_cleaned" for e, _, _, _ in self.audit)) # -- H6 state_dir 与 socket 权限 ------------------------------------------- class StateDirTest(SecfixCase): def test_preflight_creates_state_dir_0700(self): ex = self.make_exec() self.assertTrue(ex.preflight()) mode = os.stat(self.state_dir).st_mode & 0o777 self.assertEqual(oct(mode), "0o700") def test_ensure_secure_state_dir_fixes_loose_dir(self): os.makedirs(self.state_dir, exist_ok=True) os.chmod(self.state_dir, 0o755) ensure_secure_state_dir(self.state_dir) self.assertEqual( oct(os.stat(self.state_dir).st_mode & 0o777), "0o700") def test_group_writable_socket_fails_fast(self): ex = self.make_exec() os.makedirs(self.state_dir, exist_ok=True) sock_path = os.path.join(self.state_dir, "chv-sock.sock") srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) srv.bind(sock_path) srv.listen(1) os.chmod(sock_path, 0o666) try: with self.assertRaises(ChvFailure) as ctx: ex.wake_vm({"id": "sock"}) self.assertEqual(ctx.exception.kind, "socket_permissions_insecure") finally: srv.close() def test_dead_socket_not_counted_live_socket_counted(self): ex = self.make_exec(env_extra={"CHV_MAX_VMS": "1"}) os.makedirs(self.state_dir, exist_ok=True) with open(os.path.join(self.state_dir, "chv-dead.sock"), "w") as fh: fh.write("") self.assertEqual(ex._persistent_vm_count(), 0) self.assertFalse(socket_is_live( os.path.join(self.state_dir, "chv-dead.sock"))) srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) srv.bind(os.path.join(self.state_dir, "chv-live.sock")) srv.listen(1) try: self.assertTrue(socket_is_live( os.path.join(self.state_dir, "chv-live.sock"))) self.assertEqual(ex._persistent_vm_count(), 1) finally: srv.close() # -- M1 远程转义 ------------------------------------------------------------ class RemoteQuotingTest(unittest.TestCase): def test_metachars_round_trip_through_remote_shell(self): inner = ["python3", "/app/host.py", "dispatch", "--task", "a; rm -rf / $(whoami) 'q' \"d\""] cmd = build_ssh_command(2222, "/k", "root", None, "/kh", inner, ssh_host="10.0.0.5") user_idx = cmd.index("root@10.0.0.5") remote = cmd[user_idx + 1:] # ssh 拼接后远程 shell 解析回来的恰好是本地 list。 self.assertEqual(shlex.split(" ".join(remote)), inner) # 含元字符的 token 在 argv 里是被引用的单个元素， # 不是裸的 shell 元字符。 task_token = [t for t in cmd if "whoami" in t] self.assertEqual(len(task_token), 1) self.assertTrue(task_token[0].startswith("'")) # -- M2 脏标记拒起 ------------------------------------------------------------ class DirtyMarkerTest(SecfixCase): def test_dirty_marker_refuses_provision_and_wake(self): ex = self.make_exec() vm_dir = os.path.join(self.state_dir, "chv-dirty") os.makedirs(vm_dir, exist_ok=True) with open(dirty_marker_path(self.state_dir, "chv-dirty"), "w") as fh: fh.write("cleanup failed\n") with self.assertRaises(ChvFailure) as ctx: ex.provision_host_key({"id": "dirty"}) self.assertEqual(ctx.exception.kind, "vm_dirty_refused") with self.assertRaises(ChvFailure) as ctx2: ex.wake_vm({"id": "dirty"}) self.assertEqual(ctx2.exception.kind, "vm_dirty_refused") # -- M3 快照上限 --------------------------------------------------------------- class SnapshotLimitTest(SecfixCase): def _snap_file(self, name, size=64): path = os.path.join(self.state_dir, name) os.makedirs(self.state_dir, exist_ok=True) with open(path, "wb") as fh: fh.write(b"S" * size) return "file://" + path def test_snapshot_count_limit_refused_before_chremote(self): ex = self.make_exec(env_extra={"CHV_SNAPSHOT_MAX_COUNT": "1"}) url1 = self._snap_file("s1.snap") record_snapshot(self.state_dir, "chv-q", url1, 20.0) calls_before = len(self.calls) with self.assertRaises(ChvFailure) as ctx: ex.snapshot_vm({"id": "q"}, self._snap_file("s2.snap")) self.assertEqual(ctx.exception.kind, "snapshot_quota_exceeded") # 上限在 ch-remote 之前拒绝，runner 未被调新命令。 self.assertEqual(len(self.calls), calls_before) def test_snapshot_retention_prunes_expired(self): import time ex = self.make_exec() old_url = self._snap_file("old.snap") entries = [{"url": old_url, "path": old_url[len("file://"):], "size": 64, "created_at": time.time() - 40 * 86400}] import json as _json vm_dir = os.path.join(self.state_dir, "chv-r") os.makedirs(vm_dir, exist_ok=True) with open(os.path.join(vm_dir, "snapshots.json"), "w") as fh: _json.dump(entries, fh) new_url = self._snap_file("new.snap") ex.snapshot_vm({"id": "r"}, new_url) self.assertFalse(os.path.exists(old_url[len("file://"):])) self.assertTrue(os.path.exists(new_url[len("file://"):])) def test_snapshot_total_cap_prunes_oldest(self): ex = self.make_exec( env_extra={"CHV_SNAPSHOT_MAX_TOTAL_GB": "0.0000001"}) # 总量帽约 107 字节：两块 64 字节快照登记后，旧的被剪。 url1 = self._snap_file("t1.snap") record_snapshot(self.state_dir, "chv-t", url1, 20.0) url2 = self._snap_file("t2.snap") _entries, removed = record_snapshot( self.state_dir, "chv-t", url2, 0.0000001) self.assertEqual(len(removed), 1) self.assertFalse(os.path.exists(url1[len("file://"):])) # -- M4/M8 构建脚本 -------------------------------------------------------------- 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: fh.write("#!/bin/sh\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") 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": "test123", "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_requires_sha_and_prints_sshd_hardening(self): out_dir = os.path.join(self.dir, "out") r = self._run(["--dry-run", out_dir]) self.assertNotEqual(r.returncode, 0) self.assertIn("UBUNTU_IMAGE_SHA256", r.stderr) r2 = self._run(["--dry-run", out_dir], env_extra={"UBUNTU_IMAGE_SHA256": "0" * 64}) self.assertEqual(r2.returncode, 0, r2.stderr) self.assertIn("99-chv.conf", r2.stdout) self.assertIn("PasswordAuthentication no", r2.stdout) self.assertIn("sshd -t", r2.stdout) self.assertIn("virt-cat", r2.stdout) # -- M5 审计补齐 ------------------------------------------------------------------ class AuditCoverageTest(SecfixCase): def test_host_key_mismatch_audited_and_fail_fast(self): ex = self.make_exec( env_extra={"CHV_HOST_KEY_MODE": "keyscan"}) kh = ex.known_hosts_path({"id": "m1"}) os.makedirs(os.path.dirname(kh), exist_ok=True) with open(kh, "w") as fh: fh.write("[127.0.0.1]:2222 ssh-ed25519 AAAAfake\n") def bad_ssh(cmd, timeout=None): self.calls.append(list(cmd)) if cmd and cmd[0] == "ssh": return FakeResult( 255, "", "Host key verification failed.\n" "REMOTE HOST IDENTIFICATION HAS CHANGED\n") return FakeResult(0, "", "") ex.runner = bad_ssh with self.assertRaises(ChvFailure) as ctx: ex.wait_guest_ready({"id": "m1"}) self.assertEqual(ctx.exception.kind, "host_key_mismatch") self.assertTrue(any(e == "host_key_mismatch" for e, _, _, _ in self.audit)) def test_preflight_failure_is_audited(self): ex = self.make_exec(binary="/tmp/no-such-chv-binary") with self.assertRaises(ChvFailure): ex.preflight() self.assertTrue(any(e == "preflight_failed" for e, _, _, _ in self.audit)) def test_audit_callback_failure_does_not_break_task(self): ex = self.make_exec() ex.audit_callback = lambda *args: (_ for _ in ()).throw( RuntimeError("audit store down")) # _emit 只记宿主 error 日志，不抛异常。 ex._emit("vm_create", {"id": "t"}, "vm=chv-t") def test_capture_success_is_audited(self): ex = self.make_exec( env_extra={"CHV_HOST_KEY_MODE": "keyscan"}) def keyscan_runner(cmd, timeout=None): self.calls.append(list(cmd)) if cmd and cmd[0] == "ssh-keyscan": return FakeResult( 0, "[127.0.0.1]:2222 ssh-ed25519 AAAAfake1\n", "") return FakeResult(0, "", "") ex.runner = keyscan_runner self.assertTrue(ex.capture_host_key({"id": "cap1"})) self.assertTrue(any(e == "host_key_captured" for e, _, _, _ in self.audit)) # -- M6 reuse 租户声明 --------------------------------------------------------------- class ReuseTenantTest(SecfixCase): def test_reuse_without_tenant_declaration_refused(self): ex = self.make_exec(env_extra={ "CHV_DISK_MODE": "reuse", "CHV_REUSE_SINGLE_TENANT_EXPERIMENT": "1"}) with self.assertRaises(ChvFailure) as ctx: ex.preflight() self.assertEqual(ctx.exception.kind, "reuse_tenant_undeclared") def test_reuse_with_declaration_passes_preflight(self): ex = self.make_exec(env_extra={ "CHV_DISK_MODE": "reuse", "CHV_REUSE_SINGLE_TENANT_EXPERIMENT": "1", "CHV_SINGLE_TENANT_PROJECT": "p1"}) self.assertTrue(ex.preflight()) # -- L1/L2/L3/L4 ----------------------------------------------------------------------- class LowItemsTest(SecfixCase): def test_task_disk_symlink_not_followed(self): target = os.path.join(self.dir, "victim.bin") with open(target, "wb") as fh: fh.write(b"VICTIM") vm_dir = os.path.join(self.state_dir, "chv-l1") os.makedirs(vm_dir, exist_ok=True) os.symlink(target, os.path.join(vm_dir, "disk.raw")) disk = provision_task_disk(self.state_dir, "chv-l1", self.rootfs, "copy") self.assertFalse(os.path.islink(disk)) with open(disk, "rb") as fh: self.assertEqual(fh.read(), b"R" * 100) with open(target, "rb") as fh: self.assertEqual(fh.read(), b"VICTIM") def test_snapshot_url_outside_allowed_dir_refused(self): with self.assertRaises(ChvFailure) as ctx: validate_snapshot_url("file:///etc/passwd", self.state_dir) self.assertEqual(ctx.exception.kind, "snapshot_path_not_allowed") def test_remote_snapshot_requires_explicit_flag(self): with self.assertRaises(ChvFailure) as ctx: validate_snapshot_url("s3://bucket/snap", self.state_dir) self.assertEqual(ctx.exception.kind, "snapshot_remote_not_allowed") self.assertIsNone(validate_snapshot_url( "s3://bucket/snap", self.state_dir, allow_remote=True)) def test_capacity_failure_is_audited(self): ex = self.make_exec(env_extra={"CHV_MAX_VMS": "0"}) with self.assertRaises(ChvFailure) as ctx: ex.wake_vm({"id": "cap"}) self.assertEqual(ctx.exception.kind, "host_capacity_exceeded") self.assertTrue(any(e == "vm_failed" and k == "host_capacity_exceeded" for e, _, _, k in self.audit)) def test_dual_net_vars_warn_but_pass(self): ex = self.make_exec(env_extra={ "CHV_NET": "tap=tap0", "CHV_NET_JSON": '[{"tap": "tap0"}]'}) self.assertTrue(ex.preflight()) self.assertTrue(any(e == "net_config_dual" for e, _, _, _ in self.audit)) def test_warm_pool_acquire_doc_defines_task_disk_policy(self): # L5 是文档项（接线仍未做）：acquire 文档必须写清 # 预热盘与任务盘的对应关系，不假装已接。 self.assertIn("provision_task_disk", WarmPool.acquire.__doc__) if __name__ == "__main__": unittest.main()