admin/codex-app-server
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(chv): 安全审计修复 High/Medium/Low（堆叠于 #239）

Browse Source 
审计报告 ~/workspace/chv-security-audit/REPORT.md（No-Go，
Critical 0 / High 7 / Medium 8 / Low 5）代码层修复，mock
测试覆盖，真机动态验证一律未做、只列清单（见
docs/chv-production-readiness.md §11）。

High：
- H1 egress 代码门：代理地址校验、直连显式允许+审计、
  每 VM 独立 tap/mac/ip 校验、net 字段白名单；宿主防火墙
  列为部署清单，代码门+防火墙缺一不可。
- H2 网关 key 移出 argv：provisioned seed 盘客内文件
  /etc/chv/guest-env（0600）投递，传输层密钥进 argv 直接
  ValueError；keyscan 模式密钥投递 fail-closed。
- H3 任务盘删除：stop/recycle/cleanup 删 disk.raw（保留期
  0）+ 残留盘 GC，删失败审计+脏标记。
- H4 默认值改 provisioned：keyscan 只作显式 opt-in。
- H5 seed.iso 生成即 0600；宿主私钥 seed 盘建好后即删；
  崩溃残留扫描 cleanup_stale_secrets。
- H6 state_dir 默认出 /tmp（~/.local/share/.../chv-vms，
  preflight 保 0700）；api-socket 组/其它可写即 fail-fast。
- H7 tap 隔离校验：net.json 登记+stop 释放，共享即拒绝审计。
Medium：M1 shlex 远程转义；M2 清理失败即拒起（脏标记）；
M3 配额默认 50GB + 快照数/总量/保留期上限；M4 SHA256 必填
+ BUILDINFO versions 实测记录（事前 lockfile 列清单）；
M5 host key/ preflight/失败审计补齐；M6 reuse 须声明单租户
项目；M7 容量计数只认连通 socket；M8 sshd 加固进 build.sh
断言。Low：L1 链接不跟随；L2 路径白名单；L3 失败审计；
L4 双变量警告+字段白名单；L5 预热盘/任务盘对应关系定义。
旧测试：实验室默认值与生产默认值区分标注；行为已变的旧
测试按新语义更新（容量死 socket、快照白名单、磁盘路径
白名单、网关 argv、reuse 租户声明）。
测试：新 test_chv_secfix 39 项；旧 117 项 + 2 项新增断言 =
158 项；worker 相关 129 项；前端 vitest 9 项；cloud 全量
1588 项（skipped 4）。
This commit is contained in:
host committed 2026-10-31 03:02:10 +00:00
1 parent 003ea357f9
commit baf26beb78
16 files changed
+2106 -132
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

cloud/tests/test_chv_fix.py
+65 -13
View File
@@ -150,14 +150,23 @@ class ChvFixCase(unittest.TestCase):
class SeedDiskTest(ChvFixCase):
def test_vm_config_carries_seed_disk_readonly(self):
ex = self.make_exec()
cfg = ex.build_vm_config("/tmp/x/disk.raw",
seed_disk="/tmp/x/seed.iso")
# feat-chv-secfix（L2）：磁盘路径须在 state_dir 内，
# 旧 /tmp/x 路径不再合法，改用 state_dir 下路径。
disk = os.path.join(self.state_dir, "x", "disk.raw")
seed = os.path.join(self.state_dir, "x", "seed.iso")
cfg = ex.build_vm_config(disk, seed_disk=seed)
self.assertEqual(len(cfg["disks"]), 2)
self.assertEqual(cfg["disks"][1]["path"], "/tmp/x/seed.iso")
self.assertEqual(cfg["disks"][1]["path"], seed)
self.assertTrue(cfg["disks"][1]["readonly"])
cfg2 = ex.build_vm_config("/tmp/x/disk.raw")
cfg2 = ex.build_vm_config(disk)
self.assertEqual(len(cfg2["disks"]), 1)
def test_vm_config_rejects_disk_outside_state_dir(self):
ex = self.make_exec()
with self.assertRaises(ChvFailure) as ctx:
ex.build_vm_config("/tmp/x/disk.raw")
self.assertEqual(ctx.exception.kind, "disk_path_not_allowed")
def test_provisioned_wake_mounts_seed_disk(self):
ex = self.make_exec(
env_extra={"CHV_HOST_KEY_MODE": "provisioned"})
@@ -208,8 +217,11 @@ class RestoreNoFallbackTest(ChvFixCase):
env_extra={"CHV_HOST_KEY_MODE": "provisioned"})
launches = []
ex.launcher = lambda cmd: launches.append(cmd) or FakeProc(True)
# feat-chv-secfix（L2）：恢复源须在白名单目录内，
# 本测试的意图是“起 VMM 前失败”，用 state_dir 下路径。
snap_url = "file://" + os.path.join(self.state_dir, "snap", "x")
with self.assertRaises(ChvFailure) as ctx:
ex.restore_vm({"id": "r1"}, "file:///snap/x")
ex.restore_vm({"id": "r1"}, snap_url)
self.assertEqual(ctx.exception.kind, "host_key_not_pinned")
self.assertEqual(launches, [])
@@ -226,7 +238,9 @@ class ReuseBlockTest(ChvFixCase):
def test_reuse_passes_with_experiment_flag_and_is_audited(self):
ex = self.make_exec(env_extra={
"CHV_DISK_MODE": "reuse",
"CHV_REUSE_SINGLE_TENANT_EXPERIMENT": "1"})
"CHV_REUSE_SINGLE_TENANT_EXPERIMENT": "1",
# feat-chv-secfix（M6）：实验放行还须声明单租户项目。
"CHV_SINGLE_TENANT_PROJECT": "p1"})
self.assertTrue(ex.preflight())
out = ex.wake_vm({"id": "reuse1"})
self.assertEqual(out["state"], "running")
@@ -264,12 +278,19 @@ class SshTargetTest(ChvFixCase):
# -- P1-1 网关 extra_env ------------------------------------------------
class GatewayExtraEnvTest(ChvFixCase):
def test_chv_ssh_command_carries_extra_env(self):
# feat-chv-secfix（H2）：密钥值绝不再进 argv。
# 非密钥 extra_env 仍走 env 前缀；密钥子集须 seed
# 投递后才 source 客内文件（见 worker 网关测试）。
ex = self.make_exec()
cmd = ex.build_ssh_command({"id": "t"}, ["true"],
extra_env={"SUB2API_API_KEY": "k1"})
extra_env={"SUB2API_BASE_URL":
"http://gw/"})
joined = " ".join(cmd)
self.assertIn("SUB2API_API_KEY=k1", joined)
self.assertIn("SUB2API_BASE_URL=http://gw/", joined)
self.assertIn("HTTPS_PROXY=http://10.0.0.1:8888", joined)
with self.assertRaises(ChvFailure):
ex.build_ssh_command({"id": "t"}, ["true"],
extra_env={"SUB2API_API_KEY": "k1"})
def test_worker_gateway_guard_allows_chv(self):
from cloud.server import init_db
@@ -287,8 +308,20 @@ class GatewayExtraEnvTest(ChvFixCase):
self.assertIsNone(error)
self.assertEqual(result["status"], "completed")
ssh_calls = [c for c in self.calls if c and c[0] == "ssh"]
self.assertTrue(any("SUB2API_API_KEY=gwkey" in " ".join(c)
# feat-chv-secfix（H2）：密钥值不得出现在任何 ssh
# argv 里；远程命令改为 source 客内 seed 密钥文件。
self.assertFalse(any("SUB2API_API_KEY=gwkey" in " ".join(c)
for c in ssh_calls))
self.assertTrue(any("/etc/chv/guest-env" in " ".join(c)
for c in ssh_calls))
# 任务结束 stop 后 seed 目录与宿主私钥均已清（H3/H5；
# seed 内容本身的断言见 test_chv_secfix 的直接
# provision_host_key 测试，stop 前 seed 仍在）。
self.assertFalse(os.path.exists(
os.path.join(self.state_dir, "chv-gw1", "seed")))
self.assertFalse(os.path.exists(
os.path.join(self.state_dir, "chv-gw1",
"ssh_host_ed25519_key")))
# -- P1-2 取消/清理/看门狗 ----------------------------------------------
@@ -420,14 +453,30 @@ class PoolAcquireResumeTest(unittest.TestCase):
# -- P1-4 容量持久化计数 -------------------------------------------------
class CapacityPersistentTest(ChvFixCase):
def test_socket_counts_after_restart(self):
# feat-chv-secfix（M7）：只有真能连通的 socket 才
# 计数——用真实监听 socket 代表活 VMM。
ex = self.make_exec(env_extra={"CHV_MAX_VMS": "1"})
os.makedirs(self.state_dir, exist_ok=True)
sock_path = os.path.join(self.state_dir, "chv-other.sock")
srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
srv.bind(sock_path)
srv.listen(1)
try:
with self.assertRaises(ChvFailure) as ctx:
ex.wake_vm({"id": "new1"})
self.assertEqual(ctx.exception.kind, "host_capacity_exceeded")
finally:
srv.close()
def test_dead_socket_file_does_not_block(self):
# M7 的另一面：空文件/死 socket 不再占满容量。
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-other.sock"),
"w") as fh:
fh.write("")
with self.assertRaises(ChvFailure) as ctx:
ex.wake_vm({"id": "new1"})
self.assertEqual(ctx.exception.kind, "host_capacity_exceeded")
out = ex.wake_vm({"id": "new1"})
self.assertEqual(out["state"], "running")
def test_own_socket_does_not_block_rewake(self):
ex = self.make_exec(env_extra={"CHV_MAX_VMS": "1"})
@@ -523,8 +572,11 @@ class ApiTruncationTest(unittest.TestCase):
class SnapshotPersistTest(ChvFixCase):
def test_snapshot_missing_file_does_not_shutdown(self):
ex = self.make_exec()
# feat-chv-secfix（L2）：快照须在白名单目录内，
# 用 state_dir 下的不存在路径，仍验 snapshot_not_persisted。
missing = os.path.join(self.state_dir, "missing.snap")
with self.assertRaises(ChvFailure) as ctx:
ex.hibernate_vm({"id": "h1"}, "file:///nonexistent/snap")
ex.hibernate_vm({"id": "h1"}, "file://" + missing)
self.assertEqual(ctx.exception.kind, "snapshot_not_persisted")
kinds = [c[-1] for c in self.calls if c[0] == self.ch_remote]
self.assertIn("pause", kinds)
cloud/tests/test_chv_gaps.py
+8
View File
@@ -122,6 +122,11 @@ class ChvGapCase(unittest.TestCase):
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)
@@ -379,6 +384,9 @@ class BuildScriptTest(unittest.TestCase):
"UBUNTU_VERSION": "24.04",
"CODEX_VERSION": "0.164.0-alpha.9",
"REPO_COMMIT": "test123",
# feat-chv-secfix（M4）：SHA256 必填，
# 干跑只验“已设”，用 FAKE 占位。
"UBUNTU_IMAGE_SHA256": "0" * 64,
"REPO_DIR": self.repo})
if env_extra:
env.update(env_extra)
cloud/tests/test_chv_optimize.py
+5
View File
@@ -111,6 +111,11 @@ class ChvOptCase(unittest.TestCase):
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)
cloud/tests/test_chv_secfix.py
+701
View File
@@ -0,0 +1,701 @@
"""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.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.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()
cloud/tests/test_cloud_hypervisor.py
+17 -4
View File
@@ -101,6 +101,11 @@ class ChvCase(unittest.TestCase):
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)
@@ -232,12 +237,14 @@ class PreflightTest(ChvCase):
class ConfigCommandTest(ChvCase):
def test_vm_config_shape_and_memory_bytes(self):
ex = self.make_exec()
cfg = ex.build_vm_config("/tmp/x/disk.raw")
# feat-chv-secfix（L2）：磁盘须在 state_dir 内。
disk = os.path.join(self.state_dir, "x", "disk.raw")
cfg = ex.build_vm_config(disk)
self.assertEqual(cfg["cpus"], {"boot_vcpus": 2,
"max_vcpus": 2})
self.assertEqual(cfg["memory"]["size"], 2 * 1024 ** 3)
self.assertEqual(cfg["kernel"]["path"], self.kernel)
self.assertEqual(cfg["disks"], [{"path": "/tmp/x/disk.raw"}])
self.assertEqual(cfg["disks"], [{"path": disk}])
self.assertEqual(cfg["cmdline"]["args"],
"console=hvc0 root=/dev/vda1 rw")
@@ -391,7 +398,10 @@ class LifecycleTest(ChvCase):
self.assertEqual(kinds, ["pause", "resume"])
def test_restore_then_explicit_resume_order(self):
ex = self.make_exec()
# feat-chv-secfix（L2/H4）：本测试走实验室 keyscan
# 路径 + 声明快照目录（生产默认 provisioned + 白名单
# 目录，旧 file:///snap/1 不再直通）。
ex = self.make_exec(env_extra={"CHV_SNAPSHOT_DIR": "/snap"})
out = ex.restore_vm(self._task(), "file:///snap/1")
self.assertEqual(out["mode"], "restored")
self.assertTrue(out["ready"])
@@ -416,7 +426,10 @@ class LifecycleTest(ChvCase):
ex = self.make_exec()
# feat-chv-fix（P1-7）：file:// 快照须真实存在且非空，
# 否则 hibernate 不 shutdown；用临时文件作快照目标。
snap = os.path.join(self.dir, "h1.snap")
# feat-chv-secfix（L2）：快照须在白名单目录内，放在
# state_dir 下。
snap = os.path.join(self.state_dir, "h1.snap")
os.makedirs(self.state_dir, exist_ok=True)
with open(snap, "wb") as fh:
fh.write(b"S" * 64)
snap_url = "file://" + snap