diff --git a/chv-guest/README.md b/chv-guest/README.md index 94222cb..a73c5e4 100644 --- a/chv-guest/README.md +++ b/chv-guest/README.md @@ -1,11 +1,12 @@ # Cloud Hypervisor 客机镜像（chv-guest）——内容清单与版本化构建说明 -> 本目录只有清单与构建脚本骨架，镜像本身不在本轨内实际构建 -> 发布。原因写在前面，不藏着：当前没有带 /dev/kvm 的宿主 -> 可以构建并实测 VM 镜像（生产主机是 CVM、无嵌套虚拟化； -> 本 VM 容器内同样无 /dev/kvm）。构建与实测见 -> `docs/chv-production-readiness.md` 的“需镜像构建 / 需 KVM -> 宿主实测”项，绝不在本轨声称镜像已建成、已发布、已跑通。 +> 本目录有清单与真实构建步骤（build.sh，补缺口轨已实做）， +> 但镜像本身不在本轨内实际构建发布。原因写在前面，不藏着： +> 当前没有带 /dev/kvm 的宿主可以构建并实测 VM 镜像（生产 +> 主机是 CVM、无嵌套虚拟化；本 VM 容器内同样无 /dev/kvm）。 +> 构建与实测见 `docs/chv-production-readiness.md` 的“需镜像 +> 构建 / 需 KVM 宿主实测”项，绝不在本轨声称镜像已建成、 +> 已发布、已跑通。 ## 1. 镜像定位 @@ -53,12 +54,35 @@ chv-guest---<仓库 commit 短 hash> - `BUILDINFO.json`：`{ubuntu, codex_version, repo_commit, built_at, contents: [清单逐项版本]}` -## 4. 构建脚本骨架 +## 4. 构建脚本（补缺口轨：已从骨架改成真实步骤） -见 `build.sh`：只做参数校验、产物路径与版本 tag 计算、以及 -“需要 KVM 宿主 / virt-customize 才能执行的步骤”占位。脚本在 -无 KVM 的机器上运行时应在第一步就明确退出（`CHV_GUEST_BUILD -needs a KVM host`），绝不假装构建成功。 +见 `build.sh`：参数校验 → 下载 Ubuntu 云镜像 → qemu-img +转 raw → virt-customize 装 §2 清单、拷入 codex 成套二进制 +与仓库代码到 /app、写 host key 不重生成配置 → +virt-get-kernel 取内核 → 生成并校验 BUILDINFO.json。 +`./build.sh --dry-run out/` 只打印命令、不执行、不要求 +KVM；全量模式在无 /dev/kvm 的机器上第一步就明确退出 +（`CHV_GUEST_BUILD needs a KVM host`），绝不假装构建成功。 +本轨未在真机上跑过全量模式，镜像未建成、未发布，仍属 +“需 KVM 宿主实测”项。 + +## 6. Host key 首启注入硬性要求（补缺口轨新增） + +执行器的 per-VM 预置方案（见 `cloud/chv/hostkeys.py`）能 +跑通的前提是客机镜像必须满足以下全部，否则预置在真机 +上跑不通，执行器会 fail-fast，不回落成弱方案： +- 装 cloud-init 且支持 NoCloud seed（seed 磁盘/目录由 + 执行器在 VM 启动前生成，含 meta-data 与 user-data）； +- 首启时先应用 seed 再启动 sshd，seed 里的 user-data + 把该 VM 的 host key 对写进 /etc/ssh/ 并重启 sshd； +- 镜像里已钉死 `ssh_deletekeys: false` 与 + `ssh_genkeytypes: []`（build.sh 已写），首启不得无条件 + 重新生成 host key——重生成会使预置的 known_hosts 立即 + 失效，SSH 钉死直接拒绝，这是设计，不是 bug； +- sshd 只允许 key 登录，禁止密码登录； +- seed 磁盘/目录是 per-VM 敏感物，执行器在 VM 关闭/ + 回收时删除，镜像构建方不得把任何一把 host key 固化 + 进镜像（所有 VM 同 key 是安全事故）。 ## 5. 与执行器的对接 diff --git a/chv-guest/build.sh b/chv-guest/build.sh index b1c13fd..1fbd6df 100755 --- a/chv-guest/build.sh +++ b/chv-guest/build.sh @@ -1,26 +1,52 @@ #!/bin/bash -# chv-guest 镜像构建脚本骨架（未在无 KVM 机器上实际执行过）。 +# chv-guest 镜像构建脚本（补缺口轨：从 TODO 骨架改成真实步骤）。 # -# 用法（需要带 /dev/kvm 的宿主，见 README）： +# 用法（必须在带 /dev/kvm 的宿主上跑全量模式）： # CODEX_BIN_DIR=bin UBUNTU_VERSION=24.04 CODEX_VERSION=0.164.0-alpha.9 \ # REPO_COMMIT=$(git rev-parse --short HEAD) ./build.sh out/ +# ./build.sh --dry-run out/ # 只打印命令，不执行、不要求 KVM # -# 本脚本只做三件事，绝不假装已建成镜像： -# 1. 参数校验（缺 CODEX_BIN_DIR 成套二进制、缺版本、缺 /dev/kvm -# 即退出，退出码非 0）； -# 2. 算出版本 tag（chv-guest---）； -# 3. 列出仍需在 KVM 宿主上执行的步骤（下载 Ubuntu 云镜像、 -# virt-customize 装清单软件、拷入 /app 代码、生成 vmlinux / -# disk.raw / BUILDINFO.json），每一步在骨架里以 TODO 标记， -# 不在本轨内静默跳过或谎称完成。 +# 产物（out/ 内）： +# vmlinux # virt-get-kernel 从云镜像取出的内核（CHV --kernel 指向它） +# disk.raw # qemu-img 转成的 raw 根文件系统镜像（CHV_ROOTFS 指向它） +# BUILDINFO.json # {tag, ubuntu, codex_version, repo_commit, built_at, contents} +# 绝不进镜像的东西：任何模型 key、系统 key、用户账号数据、 +# 生产凭证；模型 key 由客机 cloud-init 预置的运行环境变量名 +# 注入，不写进镜像层。 +# +# 客机 host key 硬性要求（见 README §6，补缺口轨新增）： +# 客机必须装 cloud-init 且支持 NoCloud seed，首启先应用 +# seed 再启动 sshd，且不得无条件重新生成 host key，否则 +# 执行器的 per-VM 预置钉死即失效。本脚本在镜像里写 +# /etc/cloud/cloud.cfg.d/99-chv-hostkey.cfg（ssh_deletekeys: +# false、ssh_genkeytypes: []），把“不重生成”钉进镜像。 set -eu +DRY_RUN=0 +if [ "${1:-}" = "--dry-run" ]; then + DRY_RUN=1 + shift +fi OUT="${1:-out}" : "${CODEX_BIN_DIR:?CODEX_BIN_DIR 未设：需 npm 安装 @openai/codex 后的配对二进制目录}" : "${UBUNTU_VERSION:?UBUNTU_VERSION 未设}" : "${CODEX_VERSION:?CODEX_VERSION 未设}" : "${REPO_COMMIT:?REPO_COMMIT 未设：构建时仓库 commit 短 hash}" +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" +REPO_DIR="${REPO_DIR:-$(cd "$SCRIPT_DIR/.." && pwd)}" +TAG="chv-guest-${UBUNTU_VERSION}-${CODEX_VERSION}-${REPO_COMMIT}" +UBUNTU_IMAGE_URL="${UBUNTU_IMAGE_URL:-https://cloud-images.ubuntu.com/releases/${UBUNTU_VERSION}/release/ubuntu-${UBUNTU_VERSION}-server-cloudimg-amd64.img}" +run() { + if [ "$DRY_RUN" = "1" ]; then + echo "DRY-RUN: $*" + else + echo "+ $*" + "$@" + fi +} + +# 参数校验（干跑与全量都做，不依赖 KVM）。 for bin in codex codex-code-mode-host; do if [ ! -x "$CODEX_BIN_DIR/$bin" ]; then echo "缺少配对二进制：$CODEX_BIN_DIR/$bin，构建失败" >&2 @@ -31,17 +57,82 @@ if [ ! -d "$CODEX_BIN_DIR/codex-resources" ]; then echo "缺少 codex-resources（bwrap 载体），构建失败" >&2 exit 1 fi -if [ ! -e /dev/kvm ]; then - echo "CHV_GUEST_BUILD needs a KVM host: /dev/kvm 不存在" >&2 +if [ ! -f "$REPO_DIR/host.py" ] || [ ! -d "$REPO_DIR/cloud" ]; then + echo "REPO_DIR 不是 codex-app-server 仓库根（缺 host.py/cloud/）：$REPO_DIR" >&2 exit 1 fi +if [ "$DRY_RUN" = "0" ]; then + if [ ! -e /dev/kvm ]; then + echo "CHV_GUEST_BUILD needs a KVM host: /dev/kvm 不存在" >&2 + exit 1 + fi + for tool in curl qemu-img virt-customize virt-get-kernel python3; do + if ! command -v "$tool" >/dev/null 2>&1; then + echo "缺少构建工具：$tool，构建失败" >&2 + exit 1 + fi + done +fi -TAG="chv-guest-${UBUNTU_VERSION}-${CODEX_VERSION}-${REPO_COMMIT}" -mkdir -p "$OUT" +run mkdir -p "$OUT" +IMG="$OUT/ubuntu-${UBUNTU_VERSION}-server-cloudimg-amd64.img" +if [ "$DRY_RUN" = "1" ]; then + echo "DRY-RUN: curl -fL $UBUNTU_IMAGE_URL -o $IMG" + echo "DRY-RUN: [ -s $IMG ] || exit 1" +else + if [ ! -s "$IMG" ]; then + curl -fL "$UBUNTU_IMAGE_URL" -o "$IMG" + fi + [ -s "$IMG" ] || { echo "下载的云镜像为空：$IMG" >&2; exit 1; } + if [ -n "${UBUNTU_IMAGE_SHA256:-}" ]; then + echo "$UBUNTU_IMAGE_SHA256 $IMG" | sha256sum -c - + fi +fi +run qemu-img convert -f qcow2 -O raw "$IMG" "$OUT/disk.raw" +# 清单软件（照抄 cloud/docker/Dockerfile 的 apt/pip 清单， +# 中文 LaTeX xelatex+CJK 故意不装，见 README；Playwright +# 默认不装，PLAYWRIGHT=1 时才加装，属可选步骤）。 +run virt-customize -a "$OUT/disk.raw" \ + --install git,pandoc,poppler-utils,latexmk,texlive-latex-base,openssh-server,cloud-init,qemu-guest-agent,python3,python3-pip \ + --run-command 'pip3 install --no-cache-dir --break-system-packages python-docx openpyxl python-pptx PyMuPDF Pillow || pip3 install --no-cache-dir python-docx openpyxl python-pptx PyMuPDF Pillow' \ + --copy-in "$CODEX_BIN_DIR/codex:/usr/local/bin/" \ + --copy-in "$CODEX_BIN_DIR/codex-code-mode-host:/usr/local/bin/" \ + --copy-in "$CODEX_BIN_DIR/codex-resources:/usr/local/bin/" \ + --run-command 'chmod +x /usr/local/bin/codex /usr/local/bin/codex-code-mode-host; ln -sf /usr/local/bin/codex-resources/bwrap /usr/local/bin/bwrap' \ + --mkdir /app \ + --copy-in "$REPO_DIR/host.py:/app/" \ + --copy-in "$REPO_DIR/cloud:/app/" \ + --run-command 'printf "ssh_deletekeys: false\nssh_genkeytypes: []\n" > /etc/cloud/cloud.cfg.d/99-chv-hostkey.cfg; systemctl enable ssh || true' +if [ "${PLAYWRIGHT:-0}" = "1" ]; then + run virt-customize -a "$OUT/disk.raw" \ + --run-command 'pip3 install --no-cache-dir playwright; python3 -m playwright install --with-deps chromium' +fi +run virt-get-kernel -a "$OUT/disk.raw" -o "$OUT/vmlinux" +# 产物校验（干跑只打印，不执行校验命令的副作用部分）。 +if [ "$DRY_RUN" = "1" ]; then + echo "DRY-RUN: test -s $OUT/vmlinux && test -s $OUT/disk.raw" + echo "DRY-RUN: python3 生成 $OUT/BUILDINFO.json 并校验 tag=$TAG" +else + test -s "$OUT/vmlinux" || { echo "内核产物为空：$OUT/vmlinux" >&2; exit 1; } + test -s "$OUT/disk.raw" || { echo "磁盘产物为空：$OUT/disk.raw" >&2; exit 1; } + python3 - "$OUT/BUILDINFO.json" "$TAG" "$UBUNTU_VERSION" "$CODEX_VERSION" "$REPO_COMMIT" <<'PYEOF' +import json, sys, time +path, tag, ubuntu, codex, commit = sys.argv[1:6] +info = {"tag": tag, "ubuntu": ubuntu, "codex_version": codex, + "repo_commit": commit, "built_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), + "contents": ["git", "pandoc", "poppler-utils", "latexmk", "texlive-latex-base", + "openssh-server", "cloud-init", "qemu-guest-agent", + "python-docx", "openpyxl", "python-pptx", "PyMuPDF", "Pillow", + "codex", "codex-code-mode-host", "codex-resources/bwrap", "/app/host.py"]} +with open(path, "w") as fh: + json.dump(info, fh, indent=2) +PYEOF + python3 -c "import json; d=json.load(open('$OUT/BUILDINFO.json')); assert d['tag'] == '$TAG', d" +fi echo "tag=$TAG" echo "out=$OUT" -echo "TODO(KVM host): 下载 Ubuntu ${UBUNTU_VERSION} 云镜像并转 raw" -echo "TODO(KVM host): virt-customize 装 README §2 清单（git/python/pandoc/poppler/latex/codex/playwright/openssh）" -echo "TODO(KVM host): 拷入仓库代码到 /app（版本 $REPO_COMMIT），绝不拷入任何 key" -echo "TODO(KVM host): 生成 vmlinux + disk.raw + BUILDINFO.json，校验 tag 与清单版本" -echo "构建骨架结束：以上 TODO 步骤未执行，镜像未建成（见 README 与生产就绪清单）" +if [ "$DRY_RUN" = "1" ]; then + echo "干跑结束：未执行任何构建步骤，镜像未建成" +else + echo "构建完成：$OUT/vmlinux $OUT/disk.raw $OUT/BUILDINFO.json" +fi diff --git a/cloud/chv/config.py b/cloud/chv/config.py index f444069..2f38a0a 100644 --- a/cloud/chv/config.py +++ b/cloud/chv/config.py @@ -106,6 +106,12 @@ def build_chv_config(params, env): disk_gb = float(disk_gb_raw) if disk_gb_raw else None egress_proxy = env.get("CHV_EGRESS_PROXY") allow_direct_egress = env.get("CHV_ALLOW_DIRECT_EGRESS") == "1" + host_key_mode = params.get("host_key_mode") \ + or env.get("CHV_HOST_KEY_MODE") or "keyscan" + if host_key_mode not in ("provisioned", "keyscan"): + raise ValueError( + "CHV_HOST_KEY_MODE 只能是 provisioned 或 keyscan：%s" + % host_key_mode) task_timeout_raw = env.get("CHV_TASK_TIMEOUT") task_timeout = float(task_timeout_raw) \ if task_timeout_raw else None @@ -141,6 +147,7 @@ def build_chv_config(params, env): max_vms=max_vms, disk_mode=disk_mode, disk_gb=disk_gb, egress_proxy=egress_proxy, allow_direct_egress=allow_direct_egress, + host_key_mode=host_key_mode, task_timeout=task_timeout, snapshot_url=snapshot_url, warm_pool_size=warm_pool_size, api_timeout=api_timeout, ready_timeout=ready_timeout, ready_interval=ready_interval) diff --git a/cloud/chv/executor.py b/cloud/chv/executor.py index f013300..25b629f 100644 --- a/cloud/chv/executor.py +++ b/cloud/chv/executor.py @@ -164,10 +164,10 @@ class CloudHypervisorExecutor(Executor): preflight 直接失败。完整清单见 docs/chv-production-readiness.md。 - worker 层现状：通用路径已能走 dispatch_task/goal_task； - 断点续跑（goal-resume-breakpoint）在 worker 层对非 - Local 执行器仍明确失败（见 worker.py），本类的 - goal_resume_breakpoint_task 已实现，供直接调用与后续 - worker 接线，不通过 worker 迂回假装已通。 + 断点续跑（goal-resume-breakpoint）已由补缺口轨接线： + worker 通用分支对任何提供该方法的执行器调用它， + Local 仍走原路径；网关任务仍被 worker 侧 loud fail + 守卫拦住，不放宽网关规则。 - 网关现状：worker 通用路径在 sub2api 网关下对非 Local 执行器直接 loud fail（与 CloudExecutor 同款），v1 不 改这条门，所以 CHV + sub2api 组合目前在 worker 层就 @@ -231,7 +231,7 @@ class CloudHypervisorExecutor(Executor): cpuinfo_path="/proc/cpuinfo", env=None, model=None, ch_remote=None, api_client=None, sleeper=None, api_timeout=None, ready_timeout=None, - ready_interval=None): + ready_interval=None, host_key_mode=None): self._env = env if env is not None else os.environ # 配置解析已拆到 cloud/chv/config.py（参数优先于环境 # 变量，默认值与 #216 一致；显式空用户/越界端口的收紧 @@ -247,7 +247,8 @@ class CloudHypervisorExecutor(Executor): "cpuinfo_path": cpuinfo_path, "ch_remote": ch_remote, "model": model, "api_timeout": api_timeout, "ready_timeout": ready_timeout, - "ready_interval": ready_interval}, self._env) + "ready_interval": ready_interval, + "host_key_mode": host_key_mode}, self._env) for _field in ( "binary", "kernel", "rootfs", "cpus", "memory", "state_dir", "ssh_key", "ssh_user", "ssh_port", @@ -256,7 +257,7 @@ class CloudHypervisorExecutor(Executor): "max_vms", "disk_mode", "disk_gb", "egress_proxy", "allow_direct_egress", "task_timeout", "snapshot_url", "warm_pool_size", "api_timeout", "ready_timeout", - "ready_interval"): + "ready_interval", "host_key_mode"): setattr(self, _field, getattr(_cfg, _field)) self.api_client = api_client if api_client is not None \ else _unix_socket_api @@ -391,6 +392,22 @@ class CloudHypervisorExecutor(Executor): missing.append(("egress_not_configured", "CHV_EGRESS_PROXY 未设且未显式允许直连：" "VM 出口必须过白名单代理，默认不给直连外网")) + # 补缺口轨 (4)：tap/网络配置校验前置到 preflight。 + # CHV_NET_JSON 非法 JSON 或非 list/dict 时直接报 + # net_config_invalid，不等到 vm.create 才炸；直连 + # 只在显式 CHV_ALLOW_DIRECT_EGRESS=1 时放行，且 + # wake_vm 里发 egress_direct_allowed 审计。tap 桥接 + # 本身是否真只走代理仍需宿主实测，代码不声称已验证。 + if self.net_json: + try: + _net = json.loads(self.net_json) + except ValueError: + missing.append(("net_config_invalid", + "CHV_NET_JSON 不是合法 JSON")) + else: + if not isinstance(_net, (list, dict)): + missing.append(("net_config_invalid", + "CHV_NET_JSON 须为 JSON list 或 dict")) if missing: kind = missing[0][0] detail = "; ".join(text for _, text in missing) @@ -446,12 +463,45 @@ class CloudHypervisorExecutor(Executor): return _known_hosts_path(self.state_dir, self.vm_name(task_or_name)) + def provision_host_key(self, task): + """按 VM 预置 host key（补缺口轨强方案，见 hostkeys.py）： + 启动前为该 VM 生成一次性 ed25519 host key 对，公钥 + 立即写进该 VM 的 known_hosts（钉死），私钥只写进 + per-VM seed 目录的 cloud-init user-data，随 VM 销毁 + 而删。返回 {paths, seed_dir, known_hosts}。keygen + 失败即 raise，provisioned 模式下绝不回落 keyscan； + 不同 VM 的 key 对不同（state_dir/ 隔离），同一 + VM 重启复用已有对（否则钉死立即失效）。""" + from cloud.chv.hostkeys import ( + generate_host_keypair, host_key_paths, known_hosts_line, + write_seed) + from cloud.chv.ssh_transport import write_known_hosts + vm = self.vm_name(task) + paths = generate_host_keypair( + self.state_dir, vm, + runner=lambda cmd: self._call_runner(cmd)) + with open(paths["public"], "r") as fh: + public_text = fh.read() + with open(paths["private"], "r") as fh: + private_text = fh.read() + write_known_hosts( + paths["known_hosts"], + [known_hosts_line(self.ssh_port, public_text)]) + seed_dir = write_seed(paths["seed_dir"], vm, + private_text, public_text) + return {"paths": paths, "seed_dir": seed_dir, + "known_hosts": paths["known_hosts"]} + def capture_host_key(self, task): - """首次捕获客机 host key（TOFU，边界见 ssh_transport.py）： - ssh-keyscan 经本机回环端口抓 key 行，写入该 VM 的 - known_hosts 文件（0600）。抓不到可用 key 行时返回 - False（不写文件、不报错），由 wait_guest_ready 下一轮 - 再试；sshd 还没起来时这就是常态，不是失败。""" + """首次捕获客机 host key（TOFU，弱回落，边界见 + ssh_transport.py 与 hostkeys.py）：ssh-keyscan 经本机 + 回环端口抓 key 行，写入该 VM 的 known_hosts 文件 + （0600）。生产部署必须用 provisioned 预置模式， + 本方法只在 CHV_HOST_KEY_MODE=keyscan 时作回落， + 它只能防首次之后的替换，不能验证首次本身。 + 抓不到可用 key 行时返回 False（不写文件、不报错）， + 由 wait_guest_ready 下一轮再试；sshd 还没起来时 + 这就是常态，不是失败。""" try: result = self._call_runner( ["ssh-keyscan", "-p", str(self.ssh_port), @@ -699,7 +749,18 @@ class CloudHypervisorExecutor(Executor): self._kill_proc(old) task_disk = self._provision_task_disk(task) socket_path = self.api_socket(task) + if self.allow_direct_egress: + # (4) 直连 egress 只在显式 CHV_ALLOW_DIRECT_EGRESS=1 + # 时放行，且必须审计；默认 fail-closed 在 preflight + # 里已拦（无代理且未显式允许即缺项）。tap 桥接 + # 本身是否真只走代理仍需宿主实测，代码不声称已验证。 + self._emit("egress_direct_allowed", task, + "vm=%s CHV_ALLOW_DIRECT_EGRESS=1 explicit" % name) try: + if getattr(self, "host_key_mode", "keyscan") == "provisioned": + # 强方案：启动前预置每 VM host key 并钉死， + # 失败即 fail-fast，绝不回落 keyscan。 + self.provision_host_key(task) self._launch_api_only(task) self.wait_api_ready(task) status, _ = self._api_call( @@ -865,6 +926,22 @@ class CloudHypervisorExecutor(Executor): os.remove(socket_path) except Exception: pass + # 补缺口轨：per-VM seed 目录与 host 私钥随 VM + # 关闭而删，绝不留给下一个 VM 复用；known_hosts + # 公钥文件保留供事后审计，删失败不炸关闭。下次 + # 同名 VM 预置时若私钥已删即重新生成并覆盖 + # known_hosts，不会用旧钉死去校验新 VM。 + try: + seed_dir = os.path.join(self.state_dir, name, "seed") + if os.path.isdir(seed_dir): + shutil.rmtree(seed_dir) + for _suffix in ("ssh_host_ed25519_key", + "ssh_host_ed25519_key.pub"): + _p = os.path.join(self.state_dir, name, _suffix) + if os.path.exists(_p): + os.remove(_p) + except Exception: + pass task = task_or_name if isinstance(task_or_name, dict) \ else {"id": task_or_name} if had: @@ -1195,17 +1272,22 @@ class CloudHypervisorExecutor(Executor): task, args, goal=True, model=task.get("model"), effort=task.get("effort"), approval=task.get("approval")) - def goal_resume_breakpoint_task(self, task) -> dict: + def goal_resume_breakpoint_task(self, task, + model_provider=None) -> dict: """客内 host.py goal-resume 断点续跑：无 --goal-text、 无 --max-rounds，host 沿用原任务从断点轮次续跑。 - 注意 worker 层对非 Local 执行器的断点续跑目前仍明确 - 失败，本方法供直接调用与后续 worker 接线（见类 - docstring），不要理解成 worker 路径已通。""" + worker 层已接线（补缺口轨）：非网关的 CHV 断点任务 + 经 worker 的通用断点分支调到这里，model_provider + 按名透传给客内 host.py（与 Local/Docker 同款口径， + 首轮 thread 的 provider 必须一致）；网关任务在 worker + 侧仍被 loud fail 守卫拦住，不到这里。模型 key + 本身绝不经本方法传值，只按环境变量名注入。""" if not isinstance(task, dict) or task.get("id") is None: raise ValueError("任务缺少 id") return self._run_with_vm( task, ("goal-resume", "--task", self._host_name(task)), goal=True, model=task.get("model"), - effort=task.get("effort"), approval=task.get("approval")) + effort=task.get("effort"), approval=task.get("approval"), + model_provider=model_provider) diff --git a/cloud/chv/hostkeys.py b/cloud/chv/hostkeys.py new file mode 100644 index 0000000..0e2a1c8 --- /dev/null +++ b/cloud/chv/hostkeys.py @@ -0,0 +1,147 @@ +"""按 VM 生成的 SSH host key 预置（替代 TOFU 的强方案）。 + +安全模型（补缺口轨，替代 #221 的 ssh-keyscan 首次捕获）： +- 绝不用镜像固定同一把 host key：所有 VM 共用一把 key 时， + 一台 VM 泄漏即全网可冒充，属安全事故。 +- 每 VM 一次性生成：宿主在 VMM 启动前为该 VM 生成一对 + ed25519 host key（私钥只活在该 VM 的 seed 里，随 VM + 销毁而删，绝不跨 VM 复用、绝不写进镜像、绝不进已知 + 主机文件之外的共享位置）。 +- 公钥预置钉死：生成后立即把公钥写进该 VM 的 + state_dir//known_hosts（格式 [127.0.0.1]: ...）， + 执行器在 VM 启动前就已钉死，启动后 SSH 直接用该文件 + 校验，换 key 即拒绝——不再靠启动后 ssh-keyscan 抓到的 + key 来“事后钉死”。 +- 私钥注入通道（选 cloud-init NoCloud，写清唯一通道）： + seed 目录 state_dir//seed/ 含 meta-data（instance-id + 即 VM 名）、user-data（write_files 写 /etc/ssh/ 的 host + key 对并重启 sshd）。客机镜像硬性要求（见 + chv-guest/README.md §6）：必须装 cloud-init 且支持 + NoCloud seed、首启时先应用 seed 再启动 sshd、且不得在 + seed 应用前/后无条件重新生成 host key（否则预置即失效）。 + seed 磁盘/目录是 per-VM 敏感物，VM 关闭/回收时必须删。 +- 回落边界：ssh-keyscan 首次捕获只作弱回落 + （CHV_HOST_KEY_MODE=keyscan，且文档标为弱方案）： + 它只能防首次之后的替换，不能验证首次本身。生产部署 + 必须用 provisioned 模式；provisioned 模式下 seed 工具 + 或 keygen 不可用时直接 fail-fast，绝不静默回落 keyscan。 + +本模块只做纯函数与文件操作，不跑 VM、不联网、不含 +任何私钥值（测试用假 runner 时私钥由测试自己写文件）。 +""" + +import os + + +def host_key_paths(state_dir, vm_name): + """该 VM 的 host key 文件路径（私钥/公钥/known_hosts）。""" + base = os.path.join(state_dir, vm_name) + return { + "private": os.path.join(base, "ssh_host_ed25519_key"), + "public": os.path.join(base, "ssh_host_ed25519_key.pub"), + "known_hosts": os.path.join(base, "known_hosts"), + "seed_dir": os.path.join(base, "seed"), + } + + +def known_hosts_line(port, public_key_text): + """公钥文件内容转 known_hosts 行：只取前两列（类型+数据）。 + + 公钥文件形如 'ssh-ed25519 AAAA... [comment]'，known_hosts + 行形如 '[127.0.0.1]: ssh-ed25519 AAAA...'。非法 + 内容 ValueError，绝不把陌生行原样写进文件。""" + parts = (public_key_text or "").strip().split() + if len(parts) < 2 or not parts[0].startswith("ssh-"): + raise ValueError("host 公钥格式非法") + return "[127.0.0.1]:%s %s %s" % (port, parts[0], parts[1]) + + +def build_seed_user_data(private_key_text, public_key_text): + """cloud-init user-data：把 host key 对写进客机 /etc/ssh。 + + 注意：user-data 本身含私钥，只写进 per-VM seed 目录 + （0600），随 VM 销毁而删，绝不进镜像、绝不进日志、 + 绝不进任务 prompt 落库。""" + if not private_key_text or not public_key_text: + raise ValueError("host key 对缺失，无法生成 seed") + lines = [ + "#cloud-config", + "ssh_deletekeys: false", + "write_files:", + " - path: /etc/ssh/ssh_host_ed25519_key", + " permissions: '0600'", + " owner: root:root", + " content: |", + ] + for line in private_key_text.rstrip("\n").split("\n"): + lines.append(" " + line) + lines += [ + " - path: /etc/ssh/ssh_host_ed25519_key.pub", + " permissions: '0644'", + " owner: root:root", + " content: |", + ] + for line in public_key_text.rstrip("\n").split("\n"): + lines.append(" " + line) + lines += [ + "runcmd:", + " - chmod 600 /etc/ssh/ssh_host_ed25519_key", + " - systemctl restart ssh || service ssh restart || true", + ] + return "\n".join(lines) + "\n" + + +def write_seed(seed_dir, vm_name, private_key_text, public_key_text): + """写 seed 目录：meta-data + user-data（均为 0600）。 + + 返回 seed 目录路径。seed 目录本身不含已知主机之外的 + 共享数据，调用方（执行器关闭/回收）负责删除它。""" + os.makedirs(seed_dir, exist_ok=True) + meta_path = os.path.join(seed_dir, "meta-data") + user_path = os.path.join(seed_dir, "user-data") + with open(meta_path, "w") as fh: + fh.write("instance-id: %s\nlocal-hostname: %s\n" + % (vm_name, vm_name)) + os.chmod(meta_path, 0o600) + with open(user_path, "w") as fh: + fh.write(build_seed_user_data(private_key_text, public_key_text)) + os.chmod(user_path, 0o600) + return seed_dir + + +def generate_host_keypair(state_dir, vm_name, runner=None, + keygen="ssh-keygen"): + """为该 VM 生成（或复用本 VM 已有的）host key 对。 + + 同一 VM 重启/恢复时复用已有对（否则钉死的 known_hosts + 立即失效）；不同 VM 的 state_dir/ 不同，绝不共用。 + runner 不给时用 subprocess，测试可注入假 runner（假 + runner 负责把私钥/公钥文件写出来）。keygen 失败即 + RuntimeError，调用方不得回落 keyscan（provisioned + 模式）或必须显式标弱回落（keyscan 模式，见执行器）。""" + paths = host_key_paths(state_dir, vm_name) + if (os.path.exists(paths["private"]) + and os.path.exists(paths["public"])): + return paths + base = os.path.dirname(paths["private"]) + os.makedirs(base, exist_ok=True) + cmd = [keygen, "-t", "ed25519", "-N", "", + "-f", paths["private"]] + if runner is not None: + result = runner(cmd) + rc = getattr(result, "returncode", 0) + if rc not in (0, None): + raise RuntimeError("ssh-keygen 失败：returncode=%s" % rc) + else: + import subprocess + result = subprocess.run(cmd, capture_output=True, text=True) + if result.returncode != 0: + raise RuntimeError("ssh-keygen 失败：%s" % result.stderr) + if not (os.path.exists(paths["private"]) + and os.path.exists(paths["public"])): + raise RuntimeError("ssh-keygen 未生成 host key 对") + try: + os.chmod(paths["private"], 0o600) + except OSError: + pass + return paths diff --git a/cloud/tests/test_chv_gaps.py b/cloud/tests/test_chv_gaps.py new file mode 100644 index 0000000..e54f671 --- /dev/null +++ b/cloud/tests/test_chv_gaps.py @@ -0,0 +1,417 @@ +"""CHV 补缺口轨测试（堆叠于 #221，base=feat-chv-optimize）。 + +只用标准库与 mock runner/launcher/api_client，绝不启动真实 +VM：当前环境没有可跑真 VM 的宿主（生产主机是 CVM、无 +/dev/kvm、无嵌套虚拟化；本 VM 容器内同样无 /dev/kvm）。 +真实 VM 启动、快照恢复、压测一律不在本文件声称已验证。 +""" + +import json +import os +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path +from types import SimpleNamespace + +try: + from cloud.chv.hostkeys import ( + build_seed_user_data, + generate_host_keypair, + host_key_paths, + known_hosts_line, + write_seed, + ) + from cloud.chv.pool import WarmPool + from cloud.sandbox import ( + ChvFailure, + CloudHypervisorExecutor, + _parse_memory_bytes, + ) + from cloud.worker import Worker, _call_goal_resume_breakpoint +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.chv.hostkeys import ( + build_seed_user_data, + generate_host_keypair, + host_key_paths, + known_hosts_line, + write_seed, + ) + from cloud.chv.pool import WarmPool + from cloud.sandbox import ( + ChvFailure, + CloudHypervisorExecutor, + _parse_memory_bytes, + ) + from cloud.worker import Worker, _call_goal_resume_breakpoint + + +class FakeResult: + def __init__(self, returncode=0, stdout="", stderr=""): + self.returncode = returncode + self.stdout = stdout + self.stderr = stderr + + +class FakeProc: + def __init__(self, alive=True): + self._alive = alive + + def poll(self): + return None if self._alive else 1 + + def terminate(self): + self._alive = False + + def wait(self, timeout=None): + return 0 + + def kill(self): + self._alive = False + + +def _write(path, content=b"x"): + with open(path, "wb") as fh: + fh.write(content) + os.chmod(path, 0o755) + return path + + +class ChvGapCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.dir = self.tmp.name + self.binary = _write(os.path.join(self.dir, "cloud-hypervisor")) + self.ch_remote = _write(os.path.join(self.dir, "ch-remote")) + self.kernel = _write(os.path.join(self.dir, "vmlinux")) + self.rootfs = _write(os.path.join(self.dir, "disk.raw"), + b"R" * 100) + self.ssh_key = _write(os.path.join(self.dir, "id_chv")) + self.kvm = _write(os.path.join(self.dir, "kvm")) + self.cpuinfo = os.path.join(self.dir, "cpuinfo") + with open(self.cpuinfo, "w") as fh: + fh.write("flags : fpu vmx sse\n") + self.state_dir = os.path.join(self.dir, "state") + self.calls = [] + self.audit = [] + + def tearDown(self): + self.tmp.cleanup() + + def _default_runner(self, cmd, timeout=None): + self.calls.append(list(cmd)) + if cmd and cmd[0] == "ssh": + if cmd[-1] == "true": + return FakeResult(0, "", "") + if "goal-resume" in cmd: + return FakeResult(0, "goal met after rounds\n", "") + if "goal" in cmd: + return FakeResult(0, "goal met after rounds\n", "") + return FakeResult(0, "", "") + return FakeResult(0, "", "") + + def _default_api(self, socket_path, method, path, payload=None): + return (200, {}) + + def _default_launcher(self, cmd): + return FakeProc(alive=True) + + def make_exec(self, runner=None, env_extra=None, **overrides): + env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888"} + if env_extra: + env.update(env_extra) + kwargs = dict( + binary=self.binary, ch_remote=self.ch_remote, + kernel=self.kernel, rootfs=self.rootfs, + ssh_key=self.ssh_key, kvm_path=self.kvm, + cpuinfo_path=self.cpuinfo, state_dir=self.state_dir, + runner=runner or self._default_runner, + launcher=self._default_launcher, + api_client=self._default_api, + sleeper=lambda s: None, env=env) + kwargs.update(overrides) + ex = CloudHypervisorExecutor(**kwargs) + ex.audit_callback = lambda e, t, d, k=None: self.audit.append( + (e, t.get("id") if isinstance(t, dict) else t, d, k)) + return ex + + +# -- (1) worker 断点接线 --------------------------------------------- +class BreakpointWiringTest(ChvGapCase): + def test_call_helper_passes_model_provider_when_accepted(self): + seen = {} + + class FakeChv: + def goal_resume_breakpoint_task(self, task, model_provider=None): + seen["provider"] = model_provider + return {"status": "completed", "task_id": task["id"]} + + out = _call_goal_resume_breakpoint( + FakeChv(), {"id": "bp1"}, model_provider="prov-x") + self.assertEqual(out["status"], "completed") + self.assertEqual(seen["provider"], "prov-x") + + def test_call_helper_single_arg_executor_still_works(self): + class FakeOld: + def goal_resume_breakpoint_task(self, task): + return {"status": "completed", "task_id": task["id"]} + + out = _call_goal_resume_breakpoint( + FakeOld(), {"id": "bp2"}, model_provider="prov-x") + self.assertEqual(out["status"], "completed") + + def test_chv_breakpoint_accepts_model_provider(self): + ex = self.make_exec() + out = ex.goal_resume_breakpoint_task( + {"id": "g1", "prompt": "续跑"}, model_provider="prov-x") + self.assertEqual(out["status"], "completed") + self.assertTrue(out["goal_met"]) + # 客内命令带 --model-provider（网关 key 本身不进命令值）。 + ssh_calls = [c for c in self.calls if c and c[0] == "ssh"] + self.assertTrue(any("--model-provider" in c for c in ssh_calls)) + + def test_worker_branch_uses_hasattr_not_only_local(self): + # worker.py 通用分支的文字证据：Local 优先、其次 hasattr。 + worker_path = Path(__file__).resolve().parents[1] / "worker.py" + text = worker_path.read_text() + self.assertIn("_call_goal_resume_breakpoint", text) + self.assertIn("hasattr(", text) + self.assertIn("goal_resume_breakpoint_task", text) + + +# -- (2) reconcile_orphans 接线 --------------------------------------- +class ReconcileWiringTest(ChvGapCase): + def test_worker_startup_and_close_call_reconcile(self): + from cloud.server import init_db + db_path = os.path.join(self.dir, "control.db") + init_db(db_path) + ex = self.make_exec() + # 手工造一个 orphan：先建池写 registry，再新开池载入。 + pool1 = WarmPool(ex) + pool1._entries["chv-old"] = { + "state": "warm", "vm": "chv-old", + "task": {"id": "old"}, "snapshot_url": None} + pool1._save_registry() + self.assertTrue( + os.path.exists(os.path.join(self.state_dir, "warm-pool.json"))) + # Worker 构造即 startup reconcile：用假 shutdown 记录。 + shutdowns = [] + orig_shutdown = ex.shutdown_vm + ex.shutdown_vm = lambda name: shutdowns.append(name) or True + worker = Worker(db_path, "/fake/host.py", executor=ex, + env={"XCM_API_KEY": "test-placeholder"}) + self.assertIn("chv-old", shutdowns) + self.assertEqual(worker._warm_pool.orphan_count(), 0) + worker.close() + # close 时再次 reconcile 不炸（池已空，返回 [] 即可）。 + self.assertEqual(worker._reconcile_warm_pool("shutdown"), []) + + def test_reconcile_failure_does_not_break_close(self): + from cloud.server import init_db + db_path = os.path.join(self.dir, "control.db") + init_db(db_path) + ex = self.make_exec() + worker = Worker(db_path, "/fake/host.py", executor=ex, + env={"XCM_API_KEY": "test-placeholder"}) + # 池的 reconcile 抛错时 close 仍不炸。 + worker._warm_pool.reconcile_orphans = lambda: (_ for _ in ()).throw( + RuntimeError("boom")) + worker.close() # 不抛即通过 + + +# -- (3) SSH host key 预置 --------------------------------------------- +class HostKeyProvisionTest(ChvGapCase): + def _fake_keygen_runner(self, cmd, timeout=None): + self.calls.append(list(cmd)) + if cmd and cmd[0] == "ssh-keygen": + priv = cmd[cmd.index("-f") + 1] + with open(priv, "w") as fh: + fh.write("-----BEGIN OPENSSH PRIVATE KEY-----\nFAKE\n") + os.chmod(priv, 0o600) + with open(priv + ".pub", "w") as fh: + fh.write("ssh-ed25519 AAAAfake1 vm-key\n") + return FakeResult(0, "", "") + return self._default_runner(cmd, timeout) + + def test_per_vm_keys_differ_and_known_hosts_pinned(self): + ex = self.make_exec(runner=self._fake_keygen_runner) + r1 = ex.provision_host_key({"id": "vm-a"}) + r2 = ex.provision_host_key({"id": "vm-b"}) + # 不同 VM 的私钥文件不同路径、known_hosts 不同文件。 + self.assertNotEqual(r1["paths"]["private"], + r2["paths"]["private"]) + self.assertNotEqual(r1["known_hosts"], r2["known_hosts"]) + for r in (r1, r2): + with open(r["known_hosts"]) as fh: + line = fh.read().strip() + self.assertTrue(line.startswith("[127.0.0.1]:")) + self.assertIn("ssh-ed25519 AAAAfake1", line) + self.assertNotIn("PRIVATE KEY", line) + # 同一 VM 再次预置复用已有对（不重生成、不改 known_hosts）。 + with open(r1["known_hosts"]) as fh: + before = fh.read() + ex.provision_host_key({"id": "vm-a"}) + with open(r1["known_hosts"]) as fh: + self.assertEqual(fh.read(), before) + + def test_seed_contains_private_key_and_is_deleted_on_stop(self): + ex = self.make_exec(runner=self._fake_keygen_runner) + r = ex.provision_host_key({"id": "vm-c"}) + seed_user = os.path.join(r["seed_dir"], "user-data") + self.assertTrue(os.path.exists(seed_user)) + with open(seed_user) as fh: + seed_text = fh.read() + self.assertIn("ssh_host_ed25519_key", seed_text) + self.assertIn("ssh_deletekeys: false", seed_text) + # stop 时 seed 目录与私钥删掉，known_hosts 公钥保留审计。 + ex.stop_vm({"id": "vm-c"}) + self.assertFalse(os.path.exists(r["seed_dir"])) + self.assertFalse(os.path.exists(r["paths"]["private"])) + self.assertTrue(os.path.exists(r["known_hosts"])) + + def test_known_hosts_line_rejects_bad_key(self): + with self.assertRaises(ValueError): + known_hosts_line(2222, "not-a-key") + self.assertEqual( + known_hosts_line(2222, "ssh-ed25519 AAAAx comment"), + "[127.0.0.1]:2222 ssh-ed25519 AAAAx") + + def test_seed_user_data_shape(self): + text = build_seed_user_data("PRIV\n", "ssh-ed25519 AAAAx\n") + self.assertIn("/etc/ssh/ssh_host_ed25519_key", text) + self.assertIn("ssh_deletekeys: false", text) + with self.assertRaises(ValueError): + build_seed_user_data("", "ssh-ed25519 AAAAx") + + def test_keyscan_is_weak_fallback_only(self): + # provisioned 模式下 wake 前必须预置；keyscan 不应被 + # 当成生产路径：配置值本身可解析，且文档/代码里标弱。 + ex = self.make_exec(env_extra={"CHV_HOST_KEY_MODE": "provisioned"}) + self.assertEqual(ex.host_key_mode, "provisioned") + with self.assertRaises(ValueError): + self.make_exec(env_extra={"CHV_HOST_KEY_MODE": "bogus"}) + + +# -- (4) egress tap 校验 ------------------------------------------------- +class EgressValidationTest(ChvGapCase): + def test_direct_egress_without_flag_is_fail_closed(self): + ex = self.make_exec() + ex.egress_proxy = None + ex.allow_direct_egress = False + with self.assertRaises(ChvFailure) as ctx: + ex.preflight() + self.assertEqual(ctx.exception.kind, "egress_not_configured") + + def test_direct_egress_with_flag_passes_and_is_audited(self): + ex = self.make_exec(env_extra={"CHV_ALLOW_DIRECT_EGRESS": "1"}) + # preflight 本身不拦显式直连（守卫在 wake 的审计里）。 + self.assertTrue(ex.preflight()) + out = ex.goal_task({"id": "g1", "prompt": "做目标", + "goal_mode": True}) + self.assertEqual(out["status"], "completed") + self.assertTrue(any(e == "egress_direct_allowed" + for e, _, _, _ in self.audit)) + + def test_bad_net_json_is_preflight_error(self): + ex = self.make_exec(env_extra={"CHV_NET_JSON": "{not json"}) + with self.assertRaises(ChvFailure) as ctx: + ex.preflight() + self.assertEqual(ctx.exception.kind, "net_config_invalid") + + +# -- (5) 构建脚本 -------------------------------------------------------- +class BuildScriptTest(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.dir = self.tmp.name + self.bin_dir = os.path.join(self.dir, "bin") + os.makedirs(os.path.join(self.bin_dir, "codex-resources")) + for name in ("codex", "codex-code-mode-host"): + p = os.path.join(self.bin_dir, name) + with open(p, "w") as fh: + 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_prints_real_steps_and_builds_nothing(self): + out_dir = os.path.join(self.dir, "out") + r = self._run(["--dry-run", out_dir]) + self.assertEqual(r.returncode, 0, r.stderr) + self.assertIn("qemu-img convert", r.stdout) + self.assertIn("virt-customize", r.stdout) + self.assertIn("virt-get-kernel", r.stdout) + self.assertIn("BUILDINFO", r.stdout) + self.assertIn("chv-guest-24.04-0.164.0-alpha.9-test123", r.stdout) + self.assertFalse(os.path.exists(os.path.join(out_dir, "disk.raw"))) + + def test_full_mode_without_kvm_fails_fast(self): + # 本 VM 容器内无 /dev/kvm，全量模式必须明确失败， + # 绝不声称已建成镜像。 + r = self._run([os.path.join(self.dir, "out2")]) + self.assertNotEqual(r.returncode, 0) + self.assertIn("/dev/kvm", r.stderr + r.stdout) + + def test_missing_binary_fails_fast(self): + os.remove(os.path.join(self.bin_dir, "codex")) + r = self._run(["--dry-run", os.path.join(self.dir, "out3")]) + self.assertNotEqual(r.returncode, 0) + + +# -- (6) 中途死亡 → interrupted → 断点续跑闭环 --------------------------- +class MidDeathLoopTest(ChvGapCase): + def test_vm_death_is_interrupted_then_breakpoint_resumes(self): + state = {"dead": False} + + def runner(cmd, timeout=None): + self.calls.append(list(cmd)) + if cmd and cmd[0] == "ssh": + if cmd[-1] == "true": + return FakeResult(0, "", "") + if "goal-resume" in cmd and state["dead"]: + # 第二次断点调用：VM 已换新，健康返回。 + return FakeResult(0, "goal met after rounds\n", "") + if "goal" in cmd: + # 第一次 goal：模拟传输中断，并把 VMM 进程标死。 + state["dead"] = True + # 找到登记的 VMM 进程并标死（wake 已登记）。 + for proc in list(ex._vms.values()): + proc._alive = False + raise RuntimeError("connection lost mid-task") + return FakeResult(0, "", "") + + ex = self.make_exec(runner=runner) + # 第一次 goal：中途死应判 interrupted，不是 failed 死信。 + out = ex.goal_task({"id": "g9", "prompt": "做目标", + "goal_mode": True}) + self.assertEqual(out["status"], "interrupted") + # 断点续跑：同一执行器的断点方法能走通并完成。 + state["dead"] = True # 标记已进入断点阶段（runner 按此分支） + out2 = ex.goal_resume_breakpoint_task( + {"id": "g9", "prompt": "续跑"}) + self.assertEqual(out2["status"], "completed") + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/worker.py b/cloud/worker.py index 83f8d2e..e0bc1cc 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -120,6 +120,7 @@ Docker 下 repos_dir 必须放在 extra-mounts 已挂载的目录下， import argparse import hashlib +import inspect import json import os from datetime import datetime, timezone @@ -332,6 +333,24 @@ def _merge_env(base, override): return merged +def _call_goal_resume_breakpoint(executor, task, model_provider=None): + """通用断点调用：执行器提供 goal_resume_breakpoint_task 时调用它。 + + Local 仍走原路径（调用方直接传 model_provider，不经过这里， + 以免改生产语义）；这里只用于非 Local 执行器：签名接受 + model_provider 时传它（CHV/Docker 均接受），否则只传 task + （旧假执行器单参）。网关 key 绝不经这里传值：worker 侧 + gateway_env 只按名注入，见 _run_attempt 的网关守卫。""" + method = getattr(executor, "goal_resume_breakpoint_task") + try: + params = inspect.signature(method).parameters + except (TypeError, ValueError): + return method(task) + if "model_provider" in params: + return method(task, model_provider=model_provider) + return method(task) + + def _resolve_model(model=None, config_path=None): """模型名：显式参数 > host 配置文件里的 model > host 默认值。""" if model: @@ -525,6 +544,19 @@ class Worker: self.executor.audit_callback = _chv_audit except Exception: pass + # CHV 补缺口轨：WarmPool 显式接线。池构造本身不自动 + # 清理（单活池假设，见 pool.py），worker 启动是显式 + # 入口，在这里建池并调 reconcile_orphans()；关闭时 + # close() 再调一次。调用失败不炸 worker 构造/关闭， + # 但要进审计（_audit 本身 best-effort）。 + self._warm_pool = None + if isinstance(self.executor, CloudHypervisorExecutor): + try: + from cloud.chv.pool import WarmPool as _WarmPool + self._warm_pool = _WarmPool(self.executor) + self._reconcile_warm_pool("startup") + except Exception: + pass self._cost_store = cost_store self._cancel_requested = set() # 真取消用：process_factory 可注入（单测用它返回一个假长跑 @@ -759,8 +791,49 @@ class Worker: self._cost_store = CostStore(self.db_path) return self._cost_store + def _reconcile_warm_pool(self, when): + """显式调 WarmPool.reconcile_orphans()，返回已清理 VM 名。 + + 仅 CHV worker 且池已建时有效；池未建时若执行器是 CHV + 则现建现调（供测试与手动清理用）。调用失败不炸主流程， + 但进审计（失败本身也审计，不静默丢弃）。orphan 条目 + 绝不发给任务，reconcile 只做清理。""" + pool = getattr(self, "_warm_pool", None) + if pool is None: + executor = getattr(self, "executor", None) + if not isinstance(executor, CloudHypervisorExecutor): + return [] + try: + from cloud.chv.pool import WarmPool as _WarmPool + pool = _WarmPool(executor) + self._warm_pool = pool + except Exception: + return [] + try: + cleaned = pool.reconcile_orphans() or [] + except Exception as exc: + try: + self._audit(None, "chv_warm_pool_reconcile_failed", + "when=%s error=%s" % (when, exc)) + except Exception: + pass + return [] + if cleaned: + try: + self._audit(None, "chv_warm_pool_reconcile", + "when=%s cleaned=%s" % (when, ",".join(cleaned))) + except Exception: + pass + return cleaned + def close(self): - """关闭惰性建的成本存储连接（注入的 cost_store 由调用方管）。""" + """关闭惰性建的成本存储连接（注入的 cost_store 由调用方管）。 + CHV worker 关闭时先显式调一次 warm-pool reconcile， + 失败不炸关闭流程。""" + try: + self._reconcile_warm_pool("shutdown") + except Exception: + pass if self._cost_store is not None: try: self._cost_store._conn.close() @@ -4712,6 +4785,18 @@ class Worker: self.executor, LocalExecutor): result = self.executor.goal_resume_breakpoint_task( task, model_provider=gateway_provider) + elif task.get("_goal_resume_breakpoint") and hasattr( + self.executor, + "goal_resume_breakpoint_task"): + # CHV 补缺口轨：任何提供断点方法的执行器 + # 都可走断点（CHV 的 goal_resume_breakpoint_task + # 已接线；Docker 走上面 is_docker 分支，不到 + # 这里）。网关任务在进入这里之前已被 + # “网关不支持该执行器”守卫 loud fail， + # 故这里不放宽网关规则。 + result = _call_goal_resume_breakpoint( + self.executor, task, + model_provider=gateway_provider) elif task.get("_goal_resume_breakpoint"): # 假执行器/CloudExecutor 没有断点方法时 # 绝不回落成 goal（会撞已存在）或 diff --git a/docs/chv-production-readiness.md b/docs/chv-production-readiness.md index f92ac28..8636a83 100644 --- a/docs/chv-production-readiness.md +++ b/docs/chv-production-readiness.md @@ -49,10 +49,15 @@ 非 root 客机用户（CHV_SSH_USER，显式空串直接拒绝）与端口 （CHV_SSH_PORT，1-65535）为显式配置。mock 只验证捕获→ 写文件→钉死选项的管线，不验证首次密钥真实性。 -- 🔒 首次信任边界未闭合：首次捕获是 TOFU，只防首次之后的 - 替换，不验证首次本身；更强的方案是镜像构建/cloud-init 时 - 把客机 host key 预置进 known_hosts，仍是待办，安全审计时 - 须单独立项确认。 +- ✅🧪 Host key 预置方案（补缺口轨，替代 TOFU 为生产路径）： + 宿主在启动前为每 VM 生成一次性 ed25519 host key 对， + 公钥立即写进该 VM 的 known_hosts 钉死，私钥只进 per-VM + cloud-init seed（随 VM 关闭/回收而删），客机镜像硬性 + 要求见 chv-guest/README.md §6（cloud-init NoCloud、 + 首启先应用 seed 再起 sshd、不得无条件重生成 host key）。 + ssh-keyscan 首次捕获只保留为弱回落（CHV_HOST_KEY_MODE + =keyscan），生产部署必须用 provisioned 模式；TOFU 首次 + 真实性仍待预置 key 方案在真机上验证，安全审计时须单独立项确认。 - 🔒 需安全审计：VM 逃逸面（virtio 设备、tap 桥接、SSH 配置）、 客机 root 提权面、租户间东西向隔离的宿主网络配置，须由 安全审计逐项确认，代码自查不能代替审计。 @@ -114,8 +119,10 @@ pandoc/poppler/latex、python-docx/openpyxl/pptx/PyMuPDF/ Pillow、codex 0.164.0-alpha.9 成套、Playwright + Chromium、 openssh-server、cloud-init）。 -- ✅ 构建脚本骨架：`chv-guest/build.sh`（参数校验、版本 tag - 格式、KVM 宿主检查、TODO 步骤清单）。 +- ✅ 构建脚本实做（补缺口轨）：`chv-guest/build.sh`（参数校验、 + KVM 宿主检查、下载 Ubuntu 云镜像、qemu-img 转 raw、 + virt-customize 装清单、拷入 /app 代码、virt-get-kernel、 + BUILDINFO.json 生成与 tag 校验；--dry-run 只打印命令）。 - 🏗️ 需镜像构建：镜像本身未构建、未发布；BUILDINFO 版本核对、 浏览器版本钉死、中文 LaTeX 单独一版，均待构建时做。 - ⏳ 需 KVM 宿主实测：镜像在真 VM 里启动、codex/Playwright/ @@ -168,11 +175,14 @@ egress_not_configured/api_timeout/guest_not_ready/vm_create_failed/ vm_boot_failed/host_capacity_exceeded/disk_quota_exceeded）， 不笼统 failed。 -- ✅ 断点对齐：VM 快照/休眠与现有 task_checkpoints/task_attempts - 语义对齐（recycle 保留两者，丢失临时文件）；goal 断点续跑 - 方法（goal_resume_breakpoint_task）已实现，worker 层对非 - Local 执行器的断点续跑目前仍明确失败（见执行器 docstring）， - 后续 worker 接线是单独一轨，不在本轨假装已通。 +- ✅ 断点对齐（补缺口轨已接线）：VM 快照/休眠与现有 + task_checkpoints/task_attempts 语义对齐（recycle 保留两者， + 丢失临时文件）；worker 通用分支对任何提供 + goal_resume_breakpoint_task 的执行器调用它（Local 仍走 + 原路径，网关任务仍被 loud fail 守卫拦住）；WarmPool + reconcile_orphans 已由 worker 启动与关闭时显式调用， + 失败不炸主流程但进审计；VM 中途死亡改判 interrupted + 后走同一断点路径，不再落 failed 死信（见 §3 与新增测试）。 - ⏳ 需 KVM 宿主实测：审计事件在真机任务 timeline 里的完整 呈现、失败分类与真实故障的对应关系，均未验证。