From c39715659e682cac29d056bbdfa2db650f9459bd Mon Sep 17 00:00:00 2001 From: host Date: Fri, 6 Nov 2026 18:41:50 +0000 Subject: [PATCH] =?UTF-8?q?fix(chv):=20P2-20/P2-19/P2-17/P2-18=20=E6=AE=8B?= =?UTF-8?q?=E7=95=99=E6=94=B6=E5=B0=BE=20+=20NO=5FPROXY=20=E5=90=88?= =?UTF-8?q?=E5=B9=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - P2-20：_guest_env_delivered 加专用锁，读写只走 _delivered_secrets/_mark/_clear 三个入口，并发 hammer 测试钉死 - P2-19：_record_net_entries 失败发 net_registry_write_failed 独立审计并返回 False；_boot_vm 见 False 即拒起（wake 随后清半醒 VM） - P2-17：net.json 删失败发 vm_net_registry_delete_failed 审计； _write_dirty_marker 失败发 dirty_marker_write_failed 审计并返回 False，stop 不抛的契约不变 - P2-18：新增 cloud/chv/atomic_file.py（mkstemp 随机名 + 文件 fsync + rename + 目录 fsync），net.json/known_hosts/ snapshots.json/pool registry/脏标记/spill 全部走它，固定 .tmp 清零 - H1 附带：NO_PROXY 实测确有覆盖 bug（env 重复赋值以后者为准， 调用方 NO_PROXY 被 egress 的 localhost 值静默覆盖），改为合并 单条发出；egress 已设时调用方代理变量（含小写）一律不发 - 测试：新增 cloud/tests/test_chv_residual.py（23 项），旧 closeup 两处钉死旧语义的测试按新语义更新 --- cloud/chv/atomic_file.py | 71 ++++ cloud/chv/disk.py | 7 +- cloud/chv/executor.py | 118 ++++-- cloud/chv/orphans.py | 10 +- cloud/chv/ssh_transport.py | 82 ++++- cloud/tests/test_chv_closeup.py | 20 +- cloud/tests/test_chv_residual.py | 593 +++++++++++++++++++++++++++++++ 7 files changed, 850 insertions(+), 51 deletions(-) create mode 100644 cloud/chv/atomic_file.py create mode 100644 cloud/tests/test_chv_residual.py diff --git a/cloud/chv/atomic_file.py b/cloud/chv/atomic_file.py new file mode 100644 index 0000000..09f6af6 --- /dev/null +++ b/cloud/chv/atomic_file.py @@ -0,0 +1,71 @@ +"""CHV 宿主侧小文件的原子写（P2-18 收尾）。 + +此前 net.json / snapshots.json / pool registry / +known_hosts 各自用固定名 `.tmp` + 直接 os.replace， +存在两个问题： + +1. 固定 tmp 名：同一进程内两个写者（或一次崩溃残留的 + .tmp）会互相覆盖/误读；不同 VM 的文件虽在不同目录， + 但同一目录内的快照登记与 net 登记仍共用可预测的 + tmp 名，崩溃后残留的 .tmp 会被下一次写静默覆盖， + 丢了“上次写了一半”的信号。 +2. 无 fsync：os.replace 只保证目录项原子，不保证内容 + 已落盘；宿主断电/崩溃后新文件可能回到旧内容或 + 0 字节，而调用方已认为登记成功。 + +本模块是唯一写路径：mkstemp 随机名（0600）→ 写 → +flush + fsync 文件 → os.replace → fsync 父目录。 +写失败时删临时文件再原样抛错，调用方按各自语义 +处理（审计/拒绝/脏标记），本模块不吞错、不记日志。 +""" + +import os +import tempfile + + +def atomic_write(path: str, data: bytes, mode: int = 0o600) -> None: + """原子写 bytes 到 path（随机 tmp 名 + 双 fsync）。 + + mode 是最终文件的权限（默认 0600：这些文件都在 + 0700 的 state_dir 内、且含登记/密钥类内容，按敏感 + 文件对待；调用方要放宽时显式传）。 + """ + directory = os.path.dirname(path) or "." + fd, tmp_path = tempfile.mkstemp(prefix=".tmp-", dir=directory) + try: + os.fchmod(fd, mode) + with os.fdopen(fd, "wb") as fh: + fh.write(data) + fh.flush() + os.fsync(fh.fileno()) + os.replace(tmp_path, path) + _fsync_dir(directory) + except BaseException: + try: + os.unlink(tmp_path) + except OSError: + pass + raise + + +def atomic_write_text(path: str, text: str, mode: int = 0o600) -> None: + """atomic_write 的 str 版（UTF-8 编码）。""" + atomic_write(path, text.encode("utf-8"), mode=mode) + + +def _fsync_dir(directory: str) -> None: + """fsync 父目录，让 rename 本身落盘（best-effort： + 某些文件系统不支持目录 fsync，失败即跳过——文件 + 内容的 fsync 已在 atomic_write 里做完，目录 fsync + 丢了只影响“rename 是否在崩溃后可见”，不影响内容）。 + """ + try: + dir_fd = os.open(directory, os.O_RDONLY) + except OSError: + return + try: + os.fsync(dir_fd) + except OSError: + pass + finally: + os.close(dir_fd) diff --git a/cloud/chv/disk.py b/cloud/chv/disk.py index 68addf6..7608067 100644 --- a/cloud/chv/disk.py +++ b/cloud/chv/disk.py @@ -154,12 +154,11 @@ def _load_snapshots(state_dir, vm_name): def _save_snapshots(state_dir, vm_name, entries): import json + from cloud.chv.atomic_file import atomic_write_text path = _snapshots_path(state_dir, vm_name) os.makedirs(os.path.dirname(path), exist_ok=True) - tmp_path = path + ".tmp" - with open(tmp_path, "w") as fh: - json.dump(entries, fh) - os.replace(tmp_path, path) + # feat-chv-closeup（P2-18）：随机 tmp 名 + 双 fsync。 + atomic_write_text(path, json.dumps(entries)) def prune_snapshots(state_dir: str, vm_name: str, diff --git a/cloud/chv/executor.py b/cloud/chv/executor.py index 415c7a9..62dc20d 100644 --- a/cloud/chv/executor.py +++ b/cloud/chv/executor.py @@ -37,6 +37,7 @@ import time from collections.abc import Callable from cloud.chv.api_client import _unix_socket_api, normalize_api_result +from cloud.chv.atomic_file import atomic_write_text from cloud.chv.config import _parse_memory_bytes, build_chv_config from cloud.chv.disk import ( check_disk_quota, @@ -366,6 +367,16 @@ class CloudHypervisorExecutor(Executor): # 名 source 客内文件，未投递的密钥名一律拒绝（防止 # “wake 没投递、执行时却按名 source 一个旧文件”）。 self._guest_env_delivered = {} + # feat-chv-closeup（P2-20）：_guest_env_delivered 是 + # 跨 wake/build_ssh_command/stop 的共享字典，wake 与 + # 执行可并发在不同线程——此前无锁读写，set/get/pop + # 交错会丢投递记录（密钥被误判未投递而拒跑，或已清 + # 的记录被误读为已投递）。专用锁护住全部访问（与 + # _vms_lock 分开：投递记录与进程登记是两把不同的锁， + # 合一把会在 stop 路径上不必要地串行）；所有访问点 + # 必须走下面三个小方法，禁止直接读写字典（新方法 + # 是唯一入口，回归测试钉死它）。 + self._guest_env_lock = threading.Lock() self.api_client = api_client if api_client is not None \ else _unix_socket_api self.sleeper = sleeper if sleeper is not None else time.sleep @@ -385,6 +396,25 @@ class CloudHypervisorExecutor(Executor): self.launcher = _default_popen_launcher self._vms = {} + def _delivered_secrets(self, name: str) -> set: + """P2-20 唯一读入口：该 VM 已 seed 投递的密钥名集合。""" + with self._guest_env_lock: + return set(self._guest_env_delivered.get(name, set())) + + def _mark_guest_env_delivered(self, name: str, secrets) -> None: + """P2-20 唯一写入口：记下该 VM 已投递的密钥名集合 + （secrets 为空即清记录，与旧语义一致）。""" + with self._guest_env_lock: + if secrets: + self._guest_env_delivered[name] = set(secrets) + else: + self._guest_env_delivered.pop(name, None) + + def _clear_guest_env_delivered(self, name: str) -> None: + """P2-20 唯一清除入口：stop 时清掉该 VM 的投递记录。""" + with self._guest_env_lock: + self._guest_env_delivered.pop(name, None) + def _copy_config_fields(self, _cfg) -> None: """把 build_chv_config 的字段抄到 self（纯移动，字段集不变）。""" for _field in ( @@ -736,6 +766,11 @@ class CloudHypervisorExecutor(Executor): "task_id": task_id}) with open(path, "a") as fh: fh.write(line + "\n") + # feat-chv-closeup（P2-18）：spill 是审计丢事 + # 件后的唯一取证，写完即 fsync，否则宿主崩溃 + # 后 spill 本身也可能丢。失败只记日志，不炸任务。 + fh.flush() + os.fsync(fh.fileno()) except Exception as exc: _LOGGER.error("chv_audit_spill_failed error=%s", exc) @@ -942,8 +977,7 @@ class CloudHypervisorExecutor(Executor): guest_env_file = None if secrets: from cloud.chv.hostkeys import GUEST_ENV_PATH - delivered = self._guest_env_delivered.get( - self.vm_name(task), set()) + delivered = self._delivered_secrets(self.vm_name(task)) missing: list = [name for name in secrets if name not in delivered] if missing: raise ChvFailure( @@ -1296,18 +1330,34 @@ class CloudHypervisorExecutor(Executor): return registered def _record_net_entries(self, vm, entries): - """wake 成功后登记本 VM 的 net 条目（best-effort 写， - 失败只记日志——登记缺失的后果是漏判冲突，见上文）。""" + """wake 成功后登记本 VM 的 net 条目（P2-19 收尾）。 + + 返回 True=登记成功，False=登记失败。失败时三件事 + 同时做：warning 日志（旧行为保留）+ 独立审计事件 + net_registry_write_failed（调用方与运维都能看见， + 不再只有一条日志）+ 返回 False 让调用方感知。 + 调用方 _boot_vm 见 False 即拒绝启动（ChvFailure， + wake 随后 stop_vm 清掉半醒 VM）：登记缺失的后果 + 是后来的 wake 漏判 tap/mac/ip 冲突（H7 的第二道 + 防线没了），继续启动等于带着缺口跑，不如不起； + state_dir 可写本就是 preflight 的 0700 门，走到 + 这里才写失败是真故障，不是预期内的可容忍情形。 + 写路径走 atomic_file（P2-18）：随机 tmp 名 + 双 fsync。 + """ try: vm_dir = os.path.join(self.state_dir, vm) os.makedirs(vm_dir, exist_ok=True) - tmp_path = os.path.join(vm_dir, "net.json.tmp") - with open(tmp_path, "w") as fh: - json.dump(entries, fh) - os.replace(tmp_path, os.path.join(vm_dir, "net.json")) + atomic_write_text( + os.path.join(vm_dir, "net.json"), + json.dumps(entries)) + return True except OSError as exc: _LOGGER.warning("chv_net_registry_write_failed vm=%s " "error=%s", vm, exc) + self._emit("net_registry_write_failed", None, + "vm=%s net 登记写失败：%s" % (vm, exc), + "net_registry_write_failed") + return False def _check_api_socket_permissions(self, socket_path): """VMM api-socket 权限门（H6）：socket 文件存在时， @@ -1462,10 +1512,7 @@ class CloudHypervisorExecutor(Executor): provisioned["seed_dir"], seed_disk, runner=lambda cmd: self._call_runner(cmd)) delete_host_private_key(self.state_dir, name) - if secrets: - self._guest_env_delivered[name] = set(secrets) - else: - self._guest_env_delivered.pop(name, None) + self._mark_guest_env_delivered(name, secrets) return seed_disk def _boot_vm(self, task, name, task_disk, seed_disk, @@ -1485,7 +1532,13 @@ class CloudHypervisorExecutor(Executor): self._emit("vm_create", task, "vm=%s disk=%s" % (name, task_disk)) if net_entries: - self._record_net_entries(name, net_entries) + # feat-chv-closeup（P2-19）：登记失败即拒起， + # 见 _record_net_entries（审计 + 返回 False）。 + if not self._record_net_entries(name, net_entries): + raise ChvFailure( + "net_registry_write_failed", + "vm=%s net 登记写失败，拒绝启动（缺登记会" + "漏判 tap/mac/ip 冲突）" % name) status, _ = self._api_call(socket_path, "PUT", "/vm.boot") if status not in (200, 204): raise ChvFailure( @@ -1771,24 +1824,49 @@ class CloudHypervisorExecutor(Executor): _net_path = os.path.join(self.state_dir, name, "net.json") if os.path.exists(_net_path): os.remove(_net_path) - except Exception: + except Exception as _net_exc: + # feat-chv-closeup（P2-17）：net.json 删失败 + # 此前只记 failed=True、无独立审计——删不掉的 + # 登记会让后来的 wake 误判该 tap/mac/ip 仍被 + # 占用（误拒）或与脏标记混在一起查不出；现发 + # 独立审计 vm_net_registry_delete_failed。 failed = True - self._guest_env_delivered.pop(name, None) + self._emit("vm_net_registry_delete_failed", task, + "vm=%s error=%s" % (name, _net_exc)) + self._clear_guest_env_delivered(name) return failed def _write_dirty_marker(self, name, failed): - """M2 脏标记：清理失败写标记、成功清标记；写失败不炸关闭。""" + """M2 脏标记：清理失败写标记、成功清标记（P2-17 收尾）。 + + 返回 True=标记状态已如愿落盘（失败时标记已写、 + 成功时标记已清/本来就没有），False=写/删标记 + 本身失败。失败时发独立审计 dirty_marker_write_failed + + error 日志，不再全静默——标记写不下去时拒起链 + 本身就不存在（provision/wake 读不到标记），部署 + 必须保证 state_dir 可写（preflight 的 0700 门） + 并靠审计告警发现；stop 路径仍不抛错（stop_vm + “尽力而为、不抛异常”的契约不变，False 只作 + 调用方可观测的信号）。写路径走 atomic_file + （P2-18）：随机 tmp 名 + 双 fsync。 + """ from cloud.chv.hostkeys import dirty_marker_path try: _marker = dirty_marker_path(self.state_dir, name) if failed: os.makedirs(os.path.dirname(_marker), exist_ok=True) - with open(_marker, "w") as fh: - fh.write("cleanup failed\n") + atomic_write_text(_marker, "cleanup failed\n") elif os.path.exists(_marker): os.remove(_marker) - except Exception: - pass + return True + except Exception as exc: + _LOGGER.error("chv_dirty_marker_write_failed vm=%s " + "failed=%s error=%s", name, failed, exc) + self._emit("dirty_marker_write_failed", None, + "vm=%s 脏标记写失败（failed=%s）：%s" + % (name, failed, exc), + "dirty_marker_write_failed") + return False def stop_vm(self, task_or_name: dict | str) -> bool: """停止 VM（尽力而为，不抛异常）：已死进程直接清登记， diff --git a/cloud/chv/orphans.py b/cloud/chv/orphans.py index 7c7c22a..bfd607a 100644 --- a/cloud/chv/orphans.py +++ b/cloud/chv/orphans.py @@ -74,15 +74,15 @@ def _read_registry(path: str) -> tuple[dict | None, bool]: def _write_registry(path: str, data: dict) -> None: - """原子落盘 registry（tmp + rename）；写失败只记日志。""" + """原子落盘 registry（随机 tmp 名 + 双 fsync，P2-18）； + 写失败只记日志。""" try: + from cloud.chv.atomic_file import atomic_write_text + import json directory = os.path.dirname(path) if directory: os.makedirs(directory, exist_ok=True) - tmp_path = path + ".tmp" - with open(tmp_path, "w") as fh: - json.dump(data, fh) - os.replace(tmp_path, path) + atomic_write_text(path, json.dumps(data)) except OSError: _LOGGER.warning("chv_pool_registry_save_failed path=%s", path) diff --git a/cloud/chv/ssh_transport.py b/cloud/chv/ssh_transport.py index 2975f60..6d3f1ea 100644 --- a/cloud/chv/ssh_transport.py +++ b/cloud/chv/ssh_transport.py @@ -44,6 +44,7 @@ StrictHostKeyChecking=no + UserKnownHostsFile=/dev/null）： import os import shlex +from cloud.chv.atomic_file import atomic_write_text from cloud.chv.gitea_sync import ( GITEA_SYNC_PLAIN_KEYS, GITEA_SYNC_SECRET_KEYS, @@ -93,6 +94,63 @@ def split_guest_env(extra_env: dict | None) -> tuple[dict, dict]: return guest_env, plain +def _merge_no_proxy(extra_env, egress_proxy): + """调用方 NO_PROXY/no_proxy 与 egress 的 localhost、 + 127.0.0.1 去重合并（单条 NO_PROXY 的值）。""" + parts = [] + if extra_env: + for key in ("NO_PROXY", "no_proxy"): + val = extra_env.get(key) + if val: + parts += [p.strip() for p in str(val).split(",") + if p.strip()] + merged = [] + for part in parts: + if part not in merged: + merged.append(part) + if egress_proxy: + for part in ("localhost", "127.0.0.1"): + if part not in merged: + merged.append(part) + return merged + + +def _env_assignments(extra_env, egress_proxy): + """env 前缀的赋值列表（feat-chv-closeup，H1 附带）。 + + NO_PROXY 覆盖顺序已核实并修掉：旧拼装是 extra_env 的 + 赋值在前、egress 的 NO_PROXY 在后，`env` 遇重复赋值以 + 后者为准（实测 env NO_PROXY=a NO_PROXY=b → b），调用方 + 传的 NO_PROXY 被 egress 的 localhost,127.0.0.1 静默覆盖。 + 现改为合并（见 _merge_no_proxy），单条 NO_PROXY 发出。 + egress 代理本身须赢：egress 已设时调用方的 + HTTPS_PROXY/HTTP_PROXY（含小写，curl 等工具读小写优先， + 留着会架空 egress）一律不发出，只发 egress 的三条； + 无 egress 时调用方的值原样保留（只去重，不丢）。""" + skip = {"NO_PROXY", "no_proxy"} + if egress_proxy: + skip |= {"HTTPS_PROXY", "HTTP_PROXY", + "https_proxy", "http_proxy"} + assignments = [] + if extra_env: + for key in sorted(extra_env): + if key in skip: + continue + assignments.append("%s=%s" + % (key, shlex.quote( + str(extra_env[key])))) + merged = _merge_no_proxy(extra_env, egress_proxy) + if egress_proxy: + assignments += [ + "HTTPS_PROXY=%s" % shlex.quote(egress_proxy), + "HTTP_PROXY=%s" % shlex.quote(egress_proxy), + "NO_PROXY=%s" % shlex.quote(",".join(merged))] + elif merged: + assignments += [ + "NO_PROXY=%s" % shlex.quote(",".join(merged))] + return assignments + + def remote_command(inner_cmd: list, egress_proxy: str | None = None, extra_env: dict | None = None, guest_env_file: str | None = None) -> list: @@ -104,7 +162,8 @@ def remote_command(inner_cmd: list, egress_proxy: str | None = None, set +a`（把 seed 投递的密钥文件 source 进环境并 export）， 再接 env 前缀与客内命令；文件路径本身不是密钥。 extra_env 含密钥名时 ValueError（调用方须先 split 出 - 密钥子集走 seed 投递，见 split_secret_env）。""" + 密钥子集走 seed 投递，见 split_secret_env）。env 前缀的 + 拼装见 _env_assignments（NO_PROXY 合并、egress 优先）。""" if extra_env: secrets, _ = split_secret_env(extra_env) if secrets: @@ -115,17 +174,7 @@ def remote_command(inner_cmd: list, egress_proxy: str | None = None, if guest_env_file: tokens += ["set", "-a;", ".", shlex.quote(guest_env_file) + ";", "set", "+a;"] - env_assignments = [] - if extra_env: - for key in sorted(extra_env): - env_assignments.append("%s=%s" - % (key, shlex.quote( - str(extra_env[key])))) - if egress_proxy: - env_assignments += [ - "HTTPS_PROXY=%s" % shlex.quote(egress_proxy), - "HTTP_PROXY=%s" % shlex.quote(egress_proxy), - "NO_PROXY=localhost,127.0.0.1"] + env_assignments = _env_assignments(extra_env, egress_proxy) inner = [shlex.quote(str(arg)) for arg in inner_cmd] if env_assignments: tokens += ["env"] + env_assignments + inner @@ -198,13 +247,10 @@ def parse_keyscan_output(text: str, host: str = "127.0.0.1") -> list: def write_known_hosts(path: str, key_lines: list) -> None: - """原子写 known_hosts：先写同目录临时文件再 rename， + """原子写 known_hosts：随机 tmp 名 + fsync 后 rename + （feat-chv-closeup，P2-18：旧固定 `.tmp` 无 fsync）， 文件权限 0600（只含公钥行，仍按敏感文件对待）。""" directory = os.path.dirname(path) if directory: os.makedirs(directory, exist_ok=True) - tmp_path = path + ".tmp" - with open(tmp_path, "w") as fh: - fh.write("\n".join(key_lines) + "\n") - os.chmod(tmp_path, 0o600) - os.replace(tmp_path, path) + atomic_write_text(path, "\n".join(key_lines) + "\n", mode=0o600) diff --git a/cloud/tests/test_chv_closeup.py b/cloud/tests/test_chv_closeup.py index edd2a85..9aae550 100644 --- a/cloud/tests/test_chv_closeup.py +++ b/cloud/tests/test_chv_closeup.py @@ -1047,12 +1047,18 @@ class DefensiveBranchTest(CloseupCase): marker = dirty_marker_path(self.state_dir, "chv-abc123") self.assertTrue(os.path.exists(marker)) - def test_write_dirty_marker_failure_swallowed(self): + def test_write_dirty_marker_failure_audited_not_raised(self): + # feat-chv-closeup（P2-17）：脏标记写失败仍不抛（stop + # 契约不变），但不再全静默——返回 False + 独立审计 + # dirty_marker_write_failed。 blocker = os.path.join(self.dir, "blocker2") with open(blocker, "w") as fh: fh.write("x") ex = self.make_exec(state_dir=blocker) - ex._write_dirty_marker("chv-abc123", True) # 不抛 + ok = ex._write_dirty_marker("chv-abc123", True) # 不抛 + self.assertFalse(ok) + self.assertIn("dirty_marker_write_failed", + self._audit_events()) def test_stop_socket_remove_failure_swallowed(self): ex = self.make_exec() @@ -1290,7 +1296,10 @@ class ExecutorLastLinesTest(CloseupCase): ex.wait_guest_ready(self._task()) self.assertEqual(ctx.exception.kind, "guest_not_ready") - def test_record_net_entries_write_failure_only_warns(self): + def test_record_net_entries_write_failure_audited_and_reported(self): + # feat-chv-closeup（P2-19）：登记失败不再“只 warning”—— + # 返回 False + 独立审计 net_registry_write_failed，调用方 + # _boot_vm 据此拒起（见 test_chv_residual）。 ex = self.make_exec() blocker = os.path.join(self.state_dir, "chv-x") os.makedirs(self.state_dir, exist_ok=True) @@ -1298,9 +1307,12 @@ class ExecutorLastLinesTest(CloseupCase): fh.write("x") with self.assertLogs("cloud.cloud_hypervisor", level="WARNING") as logs: - ex._record_net_entries("chv-x", [{"tap": "tap0"}]) + ok = ex._record_net_entries("chv-x", [{"tap": "tap0"}]) + self.assertFalse(ok) self.assertTrue(any("net_registry_write_failed" in line for line in logs.output)) + self.assertIn("net_registry_write_failed", + self._audit_events()) def test_socket_permissions_stat_failure_after_exists(self): ex = self.make_exec() diff --git a/cloud/tests/test_chv_residual.py b/cloud/tests/test_chv_residual.py new file mode 100644 index 0000000..eb23d6a --- /dev/null +++ b/cloud/tests/test_chv_residual.py @@ -0,0 +1,593 @@ +"""CHV 残留收尾轨（feat-chv-closeup）新增测试。 + +覆盖 SECURITY-REAUDIT §5 第 5 点的逐项闭环： +P2-20（_guest_env_delivered 加锁）、P2-19（net 登记失败 +审计 + 拒起）、P2-17（net.json 删失败审计、脏标记写失败 +审计 + 返回信号）、P2-18（固定 tmp 名 + 无 fsync 改随机 +名原子写 + 双 fsync）、H1 附带（NO_PROXY 合并顺序）、 +M4（generate 只读 BUILDINFO.versions、Chromium 单独钉死、 +build.sh /app 最小集）。 + +全部用标准库与 mock，不启动真实 VM、不碰 WSL/真机； +M4 的“真实 lockfile 产出”仍待 V-9 真机构建，本文件只 +验“生成端以 BUILDINFO 为唯一来源 + 钉死校验逻辑”， +版本号全部是合成假值，绝不编造真实版本号。 +""" + +import importlib.util +import json +import os +import subprocess +import sys +import tempfile +import threading +import unittest +from pathlib import Path +from unittest import mock + +try: + from cloud.chv.atomic_file import atomic_write_text + from cloud.chv.disk import _save_snapshots + from cloud.chv.executor import CloudHypervisorExecutor + from cloud.chv.hostkeys import dirty_marker_path + from cloud.chv.orphans import _write_registry + from cloud.chv.ssh_transport import ( + remote_command, + write_known_hosts, + ) + from cloud.sandbox import ChvFailure +except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.chv.atomic_file import atomic_write_text + from cloud.chv.disk import _save_snapshots + from cloud.chv.executor import CloudHypervisorExecutor + from cloud.chv.hostkeys import dirty_marker_path + from cloud.chv.orphans import _write_registry + from cloud.chv.ssh_transport import ( + remote_command, + write_known_hosts, + ) + from cloud.sandbox import ChvFailure + +ROOT = Path(__file__).resolve().parents[2] + + +def _load_script(name): + """把 chv-guest 下的独立脚本当模块加载（无包结构）。""" + path = ROOT / "chv-guest" / name + spec = importlib.util.spec_from_file_location( + "chv_guest_" + name[:-3], str(path)) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +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 ResidualCase(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.api_calls = [] + self.audit = [] + + def tearDown(self): + self.tmp.cleanup() + + def make_exec(self, env_extra=None, **overrides): + 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) + kwargs = { + "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": 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 + + def _default_runner(self, cmd, timeout=None): + self.calls.append(list(cmd)) + return FakeResult(0, "", "") + + def _default_api(self, socket_path, method, path, payload=None): + self.api_calls.append((method, path, payload)) + return (200, {}) + + def _default_launcher(self, cmd): + return FakeProc(alive=True) + + def _task(self, task_id="abc123"): + return {"id": task_id, "prompt": "做一个加法函数", + "project_id": "proj1"} + + def _audit_events(self): + return [e for e, _t, _d, _k in self.audit] + + # -- P2-20 --------------------------------------------------------- + def test_guest_env_lock_exists_and_helpers_roundtrip(self): + ex = self.make_exec() + self.assertIsInstance(ex._guest_env_lock, type(threading.Lock())) + ex._mark_guest_env_delivered("chv-a", {"SUB2API_API_KEY"}) + self.assertEqual(ex._delivered_secrets("chv-a"), + {"SUB2API_API_KEY"}) + ex._mark_guest_env_delivered("chv-a", set()) + self.assertEqual(ex._delivered_secrets("chv-a"), set()) + ex._mark_guest_env_delivered("chv-a", {"K"}) + ex._clear_guest_env_delivered("chv-a") + self.assertEqual(ex._delivered_secrets("chv-a"), set()) + + def test_guest_env_concurrent_mark_read_clear(self): + ex = self.make_exec() + errors = [] + + def worker(n): + try: + for i in range(50): + name = f"chv-w{i % 4}" + ex._mark_guest_env_delivered(name, {f"K{n}"}) + got = ex._delivered_secrets(name) + self.assertIsInstance(got, set) + ex._clear_guest_env_delivered(name) + except Exception as exc: # noqa: BLE001 — 汇总后断言 + errors.append(exc) + + threads = [threading.Thread(target=worker, args=(n,)) + for n in range(8)] + for t in threads: + t.start() + for t in threads: + t.join() + self.assertEqual(errors, []) + # 结束后确定性收尾：最后一次 mark 必须可见，带密钥的 + # build_ssh_command 不再报 secret_env_not_delivered。 + ex._mark_guest_env_delivered("chv-final", {"SUB2API_API_KEY"}) + cmd = ex.build_ssh_command( + self._task("final"), ["true"], + extra_env={"SUB2API_API_KEY": "x"}) + self.assertEqual(cmd[0], "ssh") + self.assertNotIn("x", cmd) + + def test_build_ssh_command_sees_delivered_after_mark(self): + ex = self.make_exec() + ex._mark_guest_env_delivered("chv-abc123", {"SUB2API_API_KEY"}) + cmd = ex.build_ssh_command( + self._task(), ["true"], + extra_env={"SUB2API_API_KEY": "x"}) + self.assertNotIn("x", cmd) + + # -- P2-19 --------------------------------------------------------- + def test_record_net_entries_failure_returns_false_and_audits(self): + ex = self.make_exec() + blocker = os.path.join(self.state_dir, "chv-x") + os.makedirs(self.state_dir, exist_ok=True) + with open(blocker, "w") as fh: + fh.write("x") + with self.assertLogs("cloud.cloud_hypervisor", + level="WARNING") as logs: + ok = ex._record_net_entries("chv-x", [{"tap": "tap0"}]) + self.assertFalse(ok) + self.assertTrue(any("net_registry_write_failed" in line + for line in logs.output)) + self.assertIn("net_registry_write_failed", + self._audit_events()) + + def test_boot_vm_refuses_when_net_registry_fails(self): + ex = self.make_exec(env_extra={ + "CHV_NET_JSON": '[{"tap": "tap0", "ip": "10.0.0.2"}]'}) + os.makedirs(self.state_dir, exist_ok=True) + with open(os.path.join(self.state_dir, "chv-abc123"), + "w") as fh: + fh.write("x") + with self.assertRaises(ChvFailure) as ctx: + ex._boot_vm(self._task(), "chv-abc123", self.rootfs, + None, [{"tap": "tap0", "ip": "10.0.0.2"}]) + self.assertEqual(ctx.exception.kind, + "net_registry_write_failed") + paths = [p for _m, p, _pl in self.api_calls] + self.assertIn("/vm.create", paths) + self.assertNotIn("/vm.boot", paths) + self.assertIn("net_registry_write_failed", + self._audit_events()) + + # -- P2-17 --------------------------------------------------------- + def test_cleanup_net_remove_failure_audited(self): + ex = self.make_exec() + vm_dir = os.path.join(self.state_dir, "chv-abc123") + os.makedirs(vm_dir, exist_ok=True) + os.makedirs(os.path.join(vm_dir, "net.json"), exist_ok=True) + with open(os.path.join(self.state_dir, "chv-abc123.sock"), + "w") as fh: + fh.write("") + self.assertTrue(ex.stop_vm(self._task())) + marker = dirty_marker_path(self.state_dir, "chv-abc123") + self.assertTrue(os.path.exists(marker)) + self.assertIn("vm_net_registry_delete_failed", + self._audit_events()) + + def test_write_dirty_marker_failure_returns_false_and_audits(self): + blocker = os.path.join(self.dir, "blocker3") + with open(blocker, "w") as fh: + fh.write("x") + ex = self.make_exec(state_dir=blocker) + ok = ex._write_dirty_marker("chv-abc123", True) # 不抛 + self.assertFalse(ok) + self.assertIn("dirty_marker_write_failed", + self._audit_events()) + + def test_write_dirty_marker_success_roundtrip(self): + ex = self.make_exec() + self.assertTrue(ex._write_dirty_marker("chv-abc123", True)) + marker = dirty_marker_path(self.state_dir, "chv-abc123") + self.assertTrue(os.path.exists(marker)) + self.assertTrue(ex._write_dirty_marker("chv-abc123", False)) + self.assertFalse(os.path.exists(marker)) + + # -- P2-18 --------------------------------------------------------- + def test_atomic_write_no_fixed_tmp_and_fsync(self): + target = os.path.join(self.dir, "probe.json") + calls = {"n": 0} + real_fsync = os.fsync + + def _counting(fd): + calls["n"] += 1 + return real_fsync(fd) + + with mock.patch.object(os, "fsync", side_effect=_counting): + atomic_write_text(target, '{"a": 1}') + with open(target) as fh: + self.assertEqual(fh.read(), '{"a": 1}') + self.assertEqual(oct(os.stat(target).st_mode & 0o777), "0o600") + leftovers = [p for p in os.listdir(self.dir) + if ".tmp" in p] + self.assertEqual(leftovers, []) + # 文件 fsync + 目录 fsync，至少两次。 + self.assertGreaterEqual(calls["n"], 2) + + def test_net_record_leaves_no_tmp_files(self): + ex = self.make_exec() + self.assertTrue(ex._record_net_entries( + "chv-abc123", [{"tap": "tap0"}])) + vm_dir = os.path.join(self.state_dir, "chv-abc123") + with open(os.path.join(vm_dir, "net.json")) as fh: + self.assertEqual(json.load(fh), [{"tap": "tap0"}]) + self.assertEqual([p for p in os.listdir(vm_dir) + if ".tmp" in p], []) + + def test_known_hosts_atomic_no_fixed_tmp(self): + path = os.path.join(self.state_dir, "chv-abc123", + "known_hosts") + write_known_hosts(path, ["127.0.0.1 ssh-ed25519 AAAA"]) + with open(path) as fh: + self.assertEqual(fh.read(), + "127.0.0.1 ssh-ed25519 AAAA\n") + self.assertEqual(oct(os.stat(path).st_mode & 0o777), "0o600") + vm_dir = os.path.join(self.state_dir, "chv-abc123") + self.assertEqual([p for p in os.listdir(vm_dir) + if ".tmp" in p], []) + + def test_snapshot_save_no_fixed_tmp(self): + _save_snapshots(self.state_dir, "chv-abc123", + [{"url": "file://x"}]) + path = os.path.join(self.state_dir, "chv-abc123", + "snapshots.json") + with open(path) as fh: + self.assertEqual(json.load(fh), [{"url": "file://x"}]) + vm_dir = os.path.join(self.state_dir, "chv-abc123") + self.assertEqual([p for p in os.listdir(vm_dir) + if ".tmp" in p], []) + + def test_orphan_registry_no_fixed_tmp(self): + path = os.path.join(self.state_dir, "pool.json") + _write_registry(path, {"chv-a": {"vm": "chv-a"}}) + with open(path) as fh: + self.assertEqual(json.load(fh), + {"chv-a": {"vm": "chv-a"}}) + self.assertEqual([p for p in os.listdir(self.state_dir) + if ".tmp" in p], []) + + # -- H1 附带：NO_PROXY --------------------------------------------- + def _env_assignments(self, tokens): + """remote_command 返回的 env 赋值（KEY=VALUE 形式）。""" + out = {} + for tok in tokens: + if tok == "env": + continue + if "=" in tok and not tok.startswith("'"): + key, _, value = tok.partition("=") + if key.isidentifier() or key in ( + "NO_PROXY", "no_proxy", "HTTPS_PROXY", + "HTTP_PROXY", "https_proxy", "http_proxy"): + out.setdefault(key, []).append( + value.strip("'")) + return out + + def test_no_proxy_merged_with_egress(self): + tokens = remote_command( + ["true"], egress_proxy="http://10.0.0.1:8888", + extra_env={"NO_PROXY": "internal.example.com,10.0.0.0/8"}) + assigns = self._env_assignments(tokens) + self.assertEqual(len(assigns.get("NO_PROXY", [])), 1) + merged = assigns["NO_PROXY"][0] + for part in ("internal.example.com", "10.0.0.0/8", + "localhost", "127.0.0.1"): + self.assertIn(part, merged) + self.assertEqual( + assigns["HTTPS_PROXY"], ["http://10.0.0.1:8888"]) + + def test_no_proxy_lowercase_merged_and_not_duplicated(self): + tokens = remote_command( + ["true"], egress_proxy="http://10.0.0.1:8888", + extra_env={"no_proxy": "gitea.tose.top", + "NO_PROXY": "localhost"}) + assigns = self._env_assignments(tokens) + self.assertEqual(len(assigns.get("NO_PROXY", [])), 1) + self.assertIn("gitea.tose.top", assigns["NO_PROXY"][0]) + self.assertNotIn("no_proxy", assigns) + # localhost 去重后只出现一次。 + self.assertEqual( + assigns["NO_PROXY"][0].split(",").count("localhost"), 1) + + def test_no_proxy_preserved_without_egress(self): + tokens = remote_command( + ["true"], extra_env={"NO_PROXY": "a.example"}) + assigns = self._env_assignments(tokens) + self.assertEqual(assigns.get("NO_PROXY"), ["a.example"]) + + def test_user_proxy_dropped_when_egress_set(self): + tokens = remote_command( + ["true"], egress_proxy="http://10.0.0.1:8888", + extra_env={"HTTPS_PROXY": "http://evil:1", + "https_proxy": "http://evil:1", + "HTTP_PROXY": "http://evil:1"}) + assigns = self._env_assignments(tokens) + self.assertEqual( + assigns["HTTPS_PROXY"], ["http://10.0.0.1:8888"]) + self.assertEqual( + assigns["HTTP_PROXY"], ["http://10.0.0.1:8888"]) + self.assertNotIn("https_proxy", assigns) + self.assertNotIn("http_proxy", assigns) + + +# -- M4 供应链 ---------------------------------------------------------- +class LockfileTest(unittest.TestCase): + """合成 BUILDINFO 干跑：版本全是假值（如 git 9.9.9）， + 只验“来源是 BUILDINFO、不是宿主现场”与钉死校验逻辑， + 真实 lockfile 产出仍待 V-9 真机构建。""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.dir = self.tmp.name + + def tearDown(self): + self.tmp.cleanup() + + def _synth_buildinfo(self, chromium="chromium_headless_shell-1248", + pinned_rev="1248"): + dpkg = ["git 9.9.9-9ubuntu9", "openssh-server 1:9.9", + "openssh-client 1:9.9", "cloud-init 9.9", + "python3 9.9", "python3-pip 9.9", "curl 9.9", + "ca-certificates 9.9", "xz-utils 9.9", + "fonts-wqy-microhei 9.9", "fontconfig 9.9"] + info = { + "tag": "chv-guest-24.04-0.164.0-alpha.9-abc1234", + "ubuntu": "24.04", + "codex_version": "0.164.0-alpha.9", + "repo_commit": "abc1234", + "gt_version": "1.0.0", + "built_at": "2026-11-07T00:00:00Z", + "contents": ["git", "openssh-server", "openssh-client", + "cloud-init", "python3", "python3-pip", + "curl", "ca-certificates", "xz-utils", + "fonts-wqy-microhei", "fontconfig", "codex", + "codex-code-mode-host", "codex-resources/bwrap", + "/app/host.py", "gt", "chv-sync", + "playwright", "chromium-headless-shell"], + "versions": {"dpkg": dpkg, + "pip": ["playwright==9.9.9"], + "gt": ["gt version 1.0.0"], + "chromium": [chromium]}, + "chromium_pinned_rev": pinned_rev, + "image_sha256": "0" * 64, + } + path = os.path.join(self.dir, "BUILDINFO.json") + with open(path, "w") as fh: + json.dump(info, fh) + return path + + def test_generate_uses_buildinfo_not_host(self): + gen = _load_script("generate_lockfile.py") + buildinfo = self._synth_buildinfo() + lockfile = os.path.join(self.dir, "lockfile.json") + self.assertEqual(gen.generate(buildinfo, lockfile), 0) + with open(lockfile) as fh: + data = json.load(fh) + self.assertEqual(data["unresolved"], []) + # 假值进 lockfile：来源是 BUILDINFO，不是宿主 dpkg。 + self.assertIn("git 9.9.9-9ubuntu9", + data["versions"]["dpkg"]) + self.assertIn("playwright 9.9.9", + data["versions"]["pip"]) + self.assertIn("chromium-headless-shell " + "chromium_headless_shell-1248", + data["versions"]["declared"]) + self.assertEqual( + gen.verify(lockfile, buildinfo), 0) + + def test_generate_verify_rejects_wrong_pinned_rev(self): + gen = _load_script("generate_lockfile.py") + buildinfo = self._synth_buildinfo( + chromium="chromium_headless_shell-9999", + pinned_rev="1248") + lockfile = os.path.join(self.dir, "lockfile.json") + self.assertEqual(gen.generate(buildinfo, lockfile), 0) + # 生成本身不设门（只记版本），--verify 才拦钉死不符。 + self.assertEqual(gen.verify(lockfile, buildinfo), 1) + + def test_check_lockfile_pinned_rev(self): + check = _load_script("check_lockfile.py") + ok = self._synth_buildinfo() + self.assertEqual( + check.main(["check_lockfile.py", ok]), 0) + bad = self._synth_buildinfo( + chromium="chromium_headless_shell-9999", + pinned_rev="1248") + self.assertEqual( + check.main(["check_lockfile.py", bad]), 1) + + def test_buildinfo_records_and_enforces_pin(self): + buildinfo = _load_script("buildinfo.py") + out = os.path.join(self.dir, "out") + os.makedirs(out) + for name, content in ( + ("disk.raw", b"D"), ("vmlinux", b"K")): + with open(os.path.join(out, name), "wb") as fh: + fh.write(content) + for name, lines in ( + ("guest-versions-dpkg.txt", + ["git 9.9.9-9ubuntu9"]), + ("guest-versions-pip.txt", + ["playwright==9.9.9"]), + ("guest-versions-gt.txt", ["gt version 1.0.0"]), + ("guest-versions-chromium.txt", + ["chromium_headless_shell-1248"])): + with open(os.path.join(out, name), "w") as fh: + fh.write("\n".join(lines) + "\n") + args = [out, "chv-guest-24.04-0.164.0-alpha.9-abc1234", + "24.04", "0.164.0-alpha.9", "abc1234", "0" * 64, + "1", "1248"] + info = buildinfo.build(*args[0:7], "1248") + self.assertEqual(info["chromium_pinned_rev"], "1248") + self.assertEqual(buildinfo.main(["buildinfo.py"] + args), 0) + wrong = list(args) + wrong[-1] = "9999" + self.assertEqual( + buildinfo.main(["buildinfo.py"] + wrong), 1) + # 未给钉死值：记 null，不强制。 + info2 = buildinfo.build(*args[0:7]) + self.assertIsNone(info2["chromium_pinned_rev"]) + + +class BuildScriptMinimalAppTest(unittest.TestCase): + """build.sh 干跑：/app 只剩 host.py（不用 cloud/ 也能 + 干跑通过），Chromium 钉死值透传进 buildinfo 回显。""" + + 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\necho \"codex-cli 0.164.0-alpha.9\"\n") + os.chmod(p, 0o755) + self.repo = os.path.join(self.dir, "repo") + os.makedirs(self.repo) + # 故意不建 cloud/：最小集收紧后 build.sh 不再要求它。 + with open(os.path.join(self.repo, "host.py"), "w") as fh: + fh.write("# host\n") + 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(ROOT / "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, + "UBUNTU_IMAGE_SHA256": "0" * 64, + "REPO_DIR": self.repo}) + if env_extra: + env.update(env_extra) + return subprocess.run(["bash", self.script] + args, + check=False, + capture_output=True, text=True, env=env) + + def test_dry_run_without_cloud_dir_and_pin_echo(self): + out_dir = os.path.join(self.dir, "out") + r = self._run(["--dry-run", out_dir], + {"CHROMIUM_HEADLESS_SHELL_REV": "1248"}) + self.assertEqual(r.returncode, 0, r.stderr) + self.assertNotIn("cloud:/app", r.stdout) + self.assertIn("host.py:/app", r.stdout) + self.assertIn("1248", r.stdout) + self.assertFalse(os.path.exists( + os.path.join(out_dir, "disk.raw"))) + + def test_build_sh_text_has_no_cloud_copy(self): + with open(self.script) as fh: + text = fh.read() + self.assertNotIn('"$REPO_DIR/cloud:/app/"', text) + self.assertIn('"$REPO_DIR/host.py:/app/"', text) + + +if __name__ == "__main__": + unittest.main()