diff --git a/cloud/chv/api_client.py b/cloud/chv/api_client.py index 92c5f53..ac3cff3 100644 --- a/cloud/chv/api_client.py +++ b/cloud/chv/api_client.py @@ -33,25 +33,73 @@ def _unix_socket_api(socket_path, method, path, payload=None): sock.connect(socket_path) sock.sendall(raw) chunks = [] - while True: + # feat-chv-fix(P1-6):读超时不再当“响应结束”。 + # 先读满响应头,再按 Content-Length 读满体部;超时、 + # 提前关流、体部短于 Content-Length 一律 RuntimeError + # (调用方判 api_timeout / 传输失败),绝不解析半截 + # 响应。chunked 与无 Content-Length 响应明确拒绝 + # (Cloud Hypervisor REST 不应发 chunked;无长度则 + # 无法证完整性)。 + buf = b"" + header_end = None + timed_out = False + while header_end is None: try: data = sock.recv(65536) except socket.timeout: + timed_out = True break if not data: break - chunks.append(data) + buf += data + header_end = buf.find(b"\r\n\r\n") + if timed_out or header_end is None: + raise RuntimeError("CHV API 响应头不完整(超时或提前关流)") + header = buf[:header_end].decode("utf-8", errors="replace") + body = buf[header_end + 4:] + content_length = None + is_chunked = False + for line in header.split("\r\n")[1:]: + if ":" not in line: + continue + name, value = line.split(":", 1) + name = name.strip().lower() + value = value.strip() + if name == "content-length": + try: + content_length = int(value) + except ValueError: + raise RuntimeError("CHV API Content-Length 非法") + elif name == "transfer-encoding" \ + and "chunked" in value.lower(): + is_chunked = True + if is_chunked: + raise RuntimeError("CHV API 响应为 chunked,拒绝解析") + if content_length is None: + raise RuntimeError("CHV API 响应无 Content-Length,拒绝解析") + while len(body) < content_length: + try: + data = sock.recv(65536) + except socket.timeout: + raise RuntimeError( + "CHV API 响应体截断:已收 %d / 需 %d 字节" + % (len(body), content_length)) + if not data: + raise RuntimeError( + "CHV API 响应体截断:已收 %d / 需 %d 字节" + % (len(body), content_length)) + body += data + body = body[:content_length] finally: sock.close() - text = b"".join(chunks).decode("utf-8", errors="replace") + text = header + "\r\n\r\n" + body.decode("utf-8", errors="replace") status = 0 if text.startswith("HTTP/"): try: status = int(text.split()[1]) except (IndexError, ValueError): status = 0 - body_text = text.split("\r\n\r\n", 1)[1] if "\r\n\r\n" in text \ - else "" + body_text = body.decode("utf-8", errors="replace") try: parsed = json.loads(body_text) if body_text.strip() else {} except ValueError: diff --git a/cloud/chv/config.py b/cloud/chv/config.py index 2f38a0a..900fb53 100644 --- a/cloud/chv/config.py +++ b/cloud/chv/config.py @@ -93,6 +93,20 @@ def build_chv_config(params, env): if params.get("cpuinfo_path") is not None else "/proc/cpuinfo" ch_remote = params.get("ch_remote") or env.get("CHV_REMOTE") \ or "ch-remote" + # P0-4 SSH 目标解析(feat-chv-fix):显式 CHV_SSH_HOST 优先; + # 其次取 VmConfig net 首个带 ip 的条目(静态 tap 客机 IP, + # 即选定的客机 IP 发现方案);两者皆无时只有显式 + # CHV_SSH_FORWARD=1 才允许回环 127.0.0.1(运维宣称宿主 + # NAT 已把该端口转发到客机 sshd),否则执行器在 wake/ + # wait 时报 ssh_target_undeclared,绝不无条件假设回环能通。 + # vsock 未选:需客内 agent + 宿主代理两套新代码(#216 已 + # 否决);纯 NAT 方案未选:DNAT 是宿主防火墙外部状态, + # 执行器代码既不能创建也不能校验它,写进代码仍是假保证。 + ssh_host = params.get("ssh_host") or env.get("CHV_SSH_HOST") + ssh_forward = env.get("CHV_SSH_FORWARD") == "1" + # P0-3:reuse 实验二次确认(名字即带实验含义)。 + reuse_experiment = env.get( + "CHV_REUSE_SINGLE_TENANT_EXPERIMENT") == "1" model = params.get("model") try: max_vms = int(env.get("CHV_MAX_VMS") or 4) @@ -145,6 +159,8 @@ def build_chv_config(params, env): net_json=net_json, kvm_path=kvm_path, cpuinfo_path=cpuinfo_path, ch_remote=ch_remote, model=model, max_vms=max_vms, disk_mode=disk_mode, disk_gb=disk_gb, + ssh_host=ssh_host, ssh_forward=ssh_forward, + reuse_experiment=reuse_experiment, egress_proxy=egress_proxy, allow_direct_egress=allow_direct_egress, host_key_mode=host_key_mode, diff --git a/cloud/chv/disk.py b/cloud/chv/disk.py index 061a1c2..ad297de 100644 --- a/cloud/chv/disk.py +++ b/cloud/chv/disk.py @@ -2,8 +2,18 @@ 单任务单 VM 的文件系统独立性就靠这里:copy 模式每任务 复制一份基础镜像,两个任务绝不写同一文件;reuse 模式 -直接返回基础镜像路径(仅单租户实验,生产禁用,见 -CloudHypervisorExecutor 的类 docstring)。 +直接返回基础镜像路径(仅单租户实验,生产拦截见执行器 +preflight,须同时设 CHV_REUSE_SINGLE_TENANT_EXPERIMENT=1 +并审计放行)。 + +feat-chv-fix(P1-5)配额口径:不再只比基础镜像大小。 +check_disk_quota 改为三项:state_dir 已用总量 + 本次 +拷贝(基础镜像大小)+ 宿主剩余空间必须容纳本次拷贝; +镜像不可读时不再静默跳过,报 disk_quota_unverified +(preflight 在 wake 路径上会先报 image_missing,这里 +是直接调用者的第二道门)。provision_task_disk 对已存在 +磁盘校验大小:与源不一致即重拷;源不可读即报失败, +绝不复用脏磁盘。 """ import os @@ -12,32 +22,83 @@ import shutil from cloud.chv.errors import ChvFailure -def check_disk_quota(rootfs, disk_gb): - """磁盘配额门:CHV_DISK_GB 已设且基础镜像超过该 GB 数 - 即拒绝(kind=disk_quota_exceeded)。未设则不检查; - 镜像读不到时不拦(preflight 自会报 image_missing, - 这里不抢它的错)。""" +def _dir_total_bytes(path): + total = 0 + for root, _dirs, files in os.walk(path): + for name in files: + try: + total += os.path.getsize(os.path.join(root, name)) + except OSError: + pass + return total + + +def check_disk_quota(rootfs, disk_gb, state_dir=None): + """磁盘配额门(feat-chv-fix 口径): + + - disk_gb 未设时不检查; + - 镜像不可读时报 disk_quota_unverified,不静默跳过; + - 镜像大小超过 disk_gb 即 disk_quota_exceeded; + - state_dir 已给时:已用总量 + 本次拷贝超过 disk_gb + 即 disk_quota_exceeded;宿主剩余空间不足以容纳本次 + 拷贝即 disk_quota_exceeded(宿主余量校验)。 + """ if disk_gb is None or not rootfs: return try: size = os.path.getsize(rootfs) except OSError: - return + raise ChvFailure( + "disk_quota_unverified", + "磁盘镜像不可读,配额无法校验:%s" % rootfs) if size > disk_gb * 1024 ** 3: raise ChvFailure( "disk_quota_exceeded", "磁盘镜像 %d 字节超过配额 %.1f GB" % (size, disk_gb)) + if state_dir and os.path.isdir(state_dir): + used = _dir_total_bytes(state_dir) + if used + size > disk_gb * 1024 ** 3: + raise ChvFailure( + "disk_quota_exceeded", + "磁盘总量 %d 字节 + 本次 %d 字节超过配额 %.1f GB" + % (used, size, disk_gb)) + try: + free = shutil.disk_usage(state_dir).free + except OSError: + free = None + if free is not None and free < size: + raise ChvFailure( + "disk_quota_exceeded", + "宿主剩余空间 %d 字节不足以容纳本次磁盘拷贝 %d 字节" + % (free, size)) def provision_task_disk(state_dir, vm_name, rootfs, disk_mode): """每任务磁盘路径:copy 模式复制到 - state_dir//disk.raw(已存在则复用,不重复复制); - reuse 模式直接返回基础镜像路径。""" + state_dir//disk.raw;reuse 模式直接返回基础镜像路径。 + + feat-chv-fix:已存在磁盘须校验——大小与源不一致即 + 重拷(脏磁盘不复用);源不可读即 ChvFailure,绝不 + 静默返回旧磁盘。""" if disk_mode == "reuse": return rootfs task_dir = os.path.join(state_dir, vm_name) os.makedirs(task_dir, exist_ok=True) task_disk = os.path.join(task_dir, "disk.raw") - if not os.path.exists(task_disk): + if os.path.exists(task_disk): + try: + same = os.path.getsize(task_disk) == os.path.getsize(rootfs) + except OSError: + raise ChvFailure( + "disk_quota_unverified", + "任务磁盘或源镜像不可读,无法校验复用:%s" % task_disk) + if same: + return task_disk + os.remove(task_disk) + try: shutil.copyfile(rootfs, task_disk) + except OSError as exc: + raise ChvFailure( + "disk_quota_unverified", + "任务磁盘拷贝失败:%s" % exc) return task_disk diff --git a/cloud/chv/executor.py b/cloud/chv/executor.py index 25b629f..c28aed6 100644 --- a/cloud/chv/executor.py +++ b/cloud/chv/executor.py @@ -227,7 +227,8 @@ class CloudHypervisorExecutor(Executor): runner=None, launcher=None, ssh_key=None, ssh_user=None, ssh_port=None, guest_host_py="/app/host.py", cmdline=None, - net=None, net_json=None, kvm_path="/dev/kvm", + net=None, net_json=None, ssh_host=None, + kvm_path="/dev/kvm", cpuinfo_path="/proc/cpuinfo", env=None, model=None, ch_remote=None, api_client=None, sleeper=None, api_timeout=None, ready_timeout=None, @@ -243,7 +244,8 @@ class CloudHypervisorExecutor(Executor): "state_dir": state_dir, "ssh_key": ssh_key, "ssh_user": ssh_user, "ssh_port": ssh_port, "guest_host_py": guest_host_py, "cmdline": cmdline, - "net": net, "net_json": net_json, "kvm_path": kvm_path, + "net": net, "net_json": net_json, "ssh_host": ssh_host, + "kvm_path": kvm_path, "cpuinfo_path": cpuinfo_path, "ch_remote": ch_remote, "model": model, "api_timeout": api_timeout, "ready_timeout": ready_timeout, @@ -257,7 +259,8 @@ 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", "host_key_mode"): + "ready_interval", "host_key_mode", "ssh_host", + "ssh_forward", "reuse_experiment"): setattr(self, _field, getattr(_cfg, _field)) self.api_client = api_client if api_client is not None \ else _unix_socket_api @@ -408,6 +411,17 @@ class CloudHypervisorExecutor(Executor): if not isinstance(_net, (list, dict)): missing.append(("net_config_invalid", "CHV_NET_JSON 须为 JSON list 或 dict")) + # feat-chv-fix(P0-3):reuse 生产拦截。一行配置再也 + # 不能击穿租户隔离:只有显式实验二次确认 + # CHV_REUSE_SINGLE_TENANT_EXPERIMENT=1 时才放行, + # 且放行本身在 wake_vm 里审计;否则 preflight 直接 + # 拒绝(kind=disk_mode_reuse_blocked)。 + if getattr(self, "disk_mode", "copy") == "reuse" \ + and not getattr(self, "reuse_experiment", False): + missing.append(("disk_mode_reuse_blocked", + "CHV_DISK_MODE=reuse 未经显式实验确认:" + "须同时设 CHV_REUSE_SINGLE_TENANT_" + "EXPERIMENT=1(仅单租户实验)")) if missing: kind = missing[0][0] detail = "; ".join(text for _, text in missing) @@ -463,6 +477,39 @@ class CloudHypervisorExecutor(Executor): return _known_hosts_path(self.state_dir, self.vm_name(task_or_name)) + def _net_guest_ip(self): + """VmConfig net 首个带 ip 的条目(静态 tap 客机 IP)。""" + if not self.net_json: + return None + try: + net = json.loads(self.net_json) + except ValueError: + return None + entries = net if isinstance(net, list) else [net] + for entry in entries: + if isinstance(entry, dict) and entry.get("ip"): + return str(entry["ip"]) + return None + + def resolve_ssh_host(self): + """SSH 目标解析(feat-chv-fix,P0-4 选定方案:客机 IP): + + 显式 CHV_SSH_HOST > net 静态客机 IP > 显式转发宣称下 + 的回环;三者皆无时报 ChvFailure(ssh_target_undeclared), + 绝不无条件假设 127.0.0.1:2222 能通。""" + if getattr(self, "ssh_host", None): + return self.ssh_host + guest_ip = self._net_guest_ip() + if guest_ip: + return guest_ip + if getattr(self, "ssh_forward", False): + return "127.0.0.1" + raise ChvFailure( + "ssh_target_undeclared", + "SSH 目标未声明:须设 CHV_SSH_HOST、或在 CHV_NET_JSON " + "里给 net 条目配 ip(静态 tap 客机 IP)、或显式设 " + "CHV_SSH_FORWARD=1 宣称宿主 NAT 已转发回环端口") + def provision_host_key(self, task): """按 VM 预置 host key(补缺口轨强方案,见 hostkeys.py): 启动前为该 VM 生成一次性 ed25519 host key 对,公钥 @@ -484,9 +531,16 @@ class CloudHypervisorExecutor(Executor): public_text = fh.read() with open(paths["private"], "r") as fh: private_text = fh.read() + # feat-chv-fix:known_hosts 行的主机部分跟 SSH 目标走 + # (客机 IP 时不再写死回环,否则钉死校验永远对不上)。 + try: + ssh_host = self.resolve_ssh_host() + except ChvFailure: + ssh_host = "127.0.0.1" write_known_hosts( paths["known_hosts"], - [known_hosts_line(self.ssh_port, public_text)]) + [known_hosts_line(self.ssh_port, public_text, + host=ssh_host)]) seed_dir = write_seed(paths["seed_dir"], vm, private_text, public_text) return {"paths": paths, "seed_dir": seed_dir, @@ -502,14 +556,18 @@ class CloudHypervisorExecutor(Executor): 抓不到可用 key 行时返回 False(不写文件、不报错), 由 wait_guest_ready 下一轮再试;sshd 还没起来时 这就是常态,不是失败。""" + try: + ssh_host = self.resolve_ssh_host() + except ChvFailure: + return False try: result = self._call_runner( ["ssh-keyscan", "-p", str(self.ssh_port), - "127.0.0.1"], timeout=10) + ssh_host], timeout=10) stdout = getattr(result, "stdout", "") or "" except Exception: return False - key_lines = parse_keyscan_output(stdout) + key_lines = parse_keyscan_output(stdout, host=ssh_host) if not key_lines: return False try: @@ -518,7 +576,7 @@ class CloudHypervisorExecutor(Executor): return False return True - def build_ssh_command(self, task, inner_cmd): + def build_ssh_command(self, task, inner_cmd, extra_env=None): """构造 SSH 命令:ssh -p <端口> -i + 钉死选项 + user@127.0.0.1 + 客内命令。argv 里只有 key 文件路径, 绝无任何 key 值。egress_proxy 已设时客内命令前加 env @@ -527,13 +585,19 @@ class CloudHypervisorExecutor(Executor): Host key 选项是钉死式(StrictHostKeyChecking=yes + 每 VM known_hosts 文件),#216 的 no + /dev/null 已 在本轨废除,见类 docstring 与 ssh_transport.py。""" + try: + ssh_host = self.resolve_ssh_host() + except ChvFailure: + # 纯命令构造(旧测试/干跑)不炸:回环缺省仅作 + # 构造兼容,wake/wait 路径仍严格解析。 + ssh_host = "127.0.0.1" return _build_ssh_cmd( self.ssh_port, self.ssh_key, self.ssh_user, self.egress_proxy, self.known_hosts_path(task), - inner_cmd) + inner_cmd, ssh_host=ssh_host, extra_env=extra_env) # -- API 模式:VmConfig / API 调用 / 就绪轮询 ----------------------- - def build_vm_config(self, task_disk=None): + def build_vm_config(self, task_disk=None, seed_disk=None): """API 模式的 VmConfig(vm.create 请求体,纯函数): cpus{boot_vcpus,max_vcpus}、memory{size: 字节}、 kernel{path}、cmdline{args}、disks[{path: 每任务磁盘}]、 @@ -553,6 +617,12 @@ class CloudHypervisorExecutor(Executor): "disks": [{"path": task_disk or self.rootfs}], "rng": {"src": "/dev/urandom"}, } + # feat-chv-fix(P0-1):seed 磁盘是 VmConfig 第二块 + # 只读盘(CIDATA 卷标 ISO,客机 cloud-init NoCloud + # 数据源),不是宿主目录——目录本身从不进 VmConfig。 + if seed_disk: + config["disks"].append({"path": seed_disk, + "readonly": True}) if self.net_json: try: net = json.loads(self.net_json) @@ -595,22 +665,51 @@ class CloudHypervisorExecutor(Executor): with self._vms_lock: return len(self._vms) - def _check_capacity(self): - """宿主容量门:已登记 VM 数达到 max_vms 即明确拒绝 - (kind=host_capacity_exceeded,调用方应排队/拒绝, - 绝不降级 Docker/Local)。""" - if self._active_count() >= self.max_vms: + def _persistent_vm_count(self, exclude=None): + """跨重启容量计数(feat-chv-fix,P1-4):本实例登记 + 的活进程名 ∪ state_dir 下 chv-*.sock 对应的 VM 名。 + worker 重启后 _vms 为空,但旧 VMM 的 socket 仍在, + 仍计入容量——重启不再绕过 CHV_MAX_VMS。注意这是 + 保守计数:已死 VMM 的残留 socket 在清理前也计入, + 直到 cleanup_stale_vms / worker reconcile 清掉它。 + exclude:调用方自己的 VM 名(同任务重复 wake 时先 + 关旧 VM,不应被自己的残留 socket 误拒)。""" + names = set() + with self._vms_lock: + for name, proc in self._vms.items(): + try: + alive = proc.poll() is None + except Exception: + alive = False + if alive: + names.add(name) + try: + entries = os.listdir(self.state_dir) + except Exception: + entries = [] + for entry in entries: + if entry.endswith(".sock") and entry[:-5].startswith("chv-"): + names.add(entry[:-5]) + if exclude: + names.discard(exclude) + return len(names) + + def _check_capacity(self, exclude=None): + """宿主容量门:持久化计数(见 _persistent_vm_count) + 达到 max_vms 即明确拒绝(kind=host_capacity_exceeded, + 调用方应排队/拒绝,绝不降级 Docker/Local)。""" + count = self._persistent_vm_count(exclude=exclude) + if count >= self.max_vms: raise ChvFailure( "host_capacity_exceeded", - "宿主 CHV 容量已满:已登记 %d 台,上限 %d," - "请排队或拒绝本任务" % (self._active_count(), - self.max_vms)) + "宿主 CHV 容量已满:已计数 %d 台,上限 %d," + "请排队或拒绝本任务" % (count, self.max_vms)) def _check_disk_quota(self): - """磁盘配额门(实现在 cloud/chv/disk.py): - CHV_DISK_GB 已设且基础镜像超过该 GB 数即拒绝 - (kind=disk_quota_exceeded)。未设则不检查。""" - check_disk_quota(self.rootfs, self.disk_gb) + """磁盘配额门(实现在 cloud/chv/disk.py,feat-chv-fix + 口径:总量 + 本次 + 宿主余量,镜像不可读不静默跳过)。""" + check_disk_quota(self.rootfs, self.disk_gb, + state_dir=self.state_dir) def _provision_task_disk(self, task): """每任务磁盘供给(实现在 cloud/chv/disk.py): @@ -618,8 +717,8 @@ class CloudHypervisorExecutor(Executor): state_dir//disk.raw,VmConfig 用该路径;复制 慢但两个任务绝不写同一文件,测试用小文件可直接验证。 - reuse(仅单租户实验):直接返回基础镜像路径,磁盘 - 不独立,生产禁用(preflight 不拦它,但类 docstring - 与就绪清单里明写 reuse 不可用于生产)。 + 不独立;preflight 拦截(须显式实验二次确认并审计 + 放行,见 preflight,feat-chv-fix)。 返回每任务磁盘路径。""" return provision_task_disk( self.state_dir, self.vm_name(task), self.rootfs, @@ -657,6 +756,19 @@ class CloudHypervisorExecutor(Executor): 不先捕获就永远等不到就绪。超时时先关 VM 再 raise ChvFailure(kind=guest_not_ready),绝不在未就绪时 派任务。""" + # feat-chv-fix(P0-2):provisioned 模式下缺 known_hosts + # 时直接 fail-fast,绝不调用 capture_host_key 回落 TOFU。 + # (keyscan 模式才走下面的捕获分支。) + if getattr(self, "host_key_mode", "keyscan") == "provisioned" \ + and not os.path.exists(self.known_hosts_path(task)): + try: + self.stop_vm(task) + except Exception: + pass + raise ChvFailure( + "host_key_not_pinned", + "provisioned 模式下 known_hosts 缺失,拒绝 keyscan " + "回落:VM %s 未钉死 host key" % self.vm_name(task)) attempts = max(1, int(self.ready_timeout / self.ready_interval)) for attempt in range(attempts): try: @@ -738,9 +850,17 @@ class CloudHypervisorExecutor(Executor): if not isinstance(task, dict) or task.get("id") is None: raise ValueError("任务缺少 id") self.preflight() - self._check_capacity() - self._check_disk_quota() + # feat-chv-fix:SSH 目标与 reuse 放行在启动前就定性。 + # resolve 失败即 fail-fast(不先起 VMM 再超时); + # reuse 实验放行必须审计,留痕给运维复核。 + self.resolve_ssh_host() name = self.vm_name(task) + self._check_capacity(exclude=name) + self._check_disk_quota() + if getattr(self, "disk_mode", "copy") == "reuse": + self._emit("disk_reuse_experiment_allowed", task, + "vm=%s CHV_REUSE_SINGLE_TENANT_EXPERIMENT=1 " + "explicit" % name) old = None with self._vms_lock: if name in self._vms: @@ -757,15 +877,25 @@ class CloudHypervisorExecutor(Executor): self._emit("egress_direct_allowed", task, "vm=%s CHV_ALLOW_DIRECT_EGRESS=1 explicit" % name) try: + seed_disk = None if getattr(self, "host_key_mode", "keyscan") == "provisioned": # 强方案:启动前预置每 VM host key 并钉死, - # 失败即 fail-fast,绝不回落 keyscan。 - self.provision_host_key(task) + # 再把 seed 目录做成 CIDATA seed 磁盘并挂进 + # VmConfig(P0-1 闭环:生成→制盘→挂盘三步缺 + # 一即 fail-fast,绝不回落 keyscan、绝不只 + # 生成不注入)。 + from cloud.chv.hostkeys import ( + build_seed_disk, seed_disk_path) + provisioned = self.provision_host_key(task) + seed_disk = seed_disk_path(self.state_dir, name) + build_seed_disk( + provisioned["seed_dir"], seed_disk, + runner=lambda cmd: self._call_runner(cmd)) self._launch_api_only(task) self.wait_api_ready(task) status, _ = self._api_call( socket_path, "PUT", "/vm.create", - self.build_vm_config(task_disk)) + self.build_vm_config(task_disk, seed_disk=seed_disk)) if status not in (200, 204): raise ChvFailure( "vm_create_failed", @@ -828,6 +958,18 @@ class CloudHypervisorExecutor(Executor): raise ValueError("snapshot_url 不能为空") self._ch_remote(task_or_name, "snapshot", "source_url=%s" % snapshot_url) + # feat-chv-fix(P1-7):只查 returncode 不够。file:// + # 快照须验证目标文件存在且非空,否则报 + # snapshot_not_persisted,调用方(hibernate_vm)据此 + # 不 shutdown、不销毁原 VM;非 file:// 目标无法从宿 + # 主验证落盘,仍只认 returncode,但文档明写该边界。 + if snapshot_url.startswith("file://"): + _snap_path = snapshot_url[len("file://"):] + if not os.path.isfile(_snap_path) \ + or os.path.getsize(_snap_path) == 0: + raise ChvFailure( + "snapshot_not_persisted", + "快照未落盘或为空:%s" % snapshot_url) task = task_or_name if isinstance(task_or_name, dict) \ else {"id": task_or_name} self._emit("vm_snapshot", task, @@ -849,8 +991,27 @@ class CloudHypervisorExecutor(Executor): if "://" not in snapshot_url: snapshot_url = "file://%s" % snapshot_url self.preflight() - self._check_capacity() + self.resolve_ssh_host() + # feat-chv-fix(P0-2 修正版):restore 不走 vm.create, + # 没有 VmConfig 可挂 seed——审计报告“restore 须把 seed + # 接入 VmConfig”的说法在此不适用,纠正为:快照内已 + # 固化原 VM 的 host key,重新预置新 key 反而会破坏 + # 钉死。正确不变量是:provisioned 下复用已钉死的 + # known_hosts(stop 时公钥保留、私钥已删但快照内仍有 + # 原私钥),缺失时 fail-fast、绝不 keyscan、绝不重 + # 生成。file:// 快照须先验存在且非空,否则不启动。 + # 注意:restore 时不重复验快照文件存在——快照落盘 + # 校验是 snapshot_vm/hibernate_vm 创建时的门(P1-7), + # restore 的 ch-remote 本身会在源缺失时失败;在这里 + # 再验一次会把“源在远端/对象存储”的合法路径也挡掉。 + if getattr(self, "host_key_mode", "keyscan") == "provisioned" \ + and not os.path.exists(self.known_hosts_path(task)): + raise ChvFailure( + "host_key_not_pinned", + "provisioned 模式下 restore 缺 known_hosts,拒绝 " + "keyscan 回落:VM %s" % self.vm_name(task)) name = self.vm_name(task) + self._check_capacity(exclude=name) try: self._launch_api_only(task) self.wait_api_ready(task) @@ -936,7 +1097,8 @@ class CloudHypervisorExecutor(Executor): if os.path.isdir(seed_dir): shutil.rmtree(seed_dir) for _suffix in ("ssh_host_ed25519_key", - "ssh_host_ed25519_key.pub"): + "ssh_host_ed25519_key.pub", + "seed.iso"): _p = os.path.join(self.state_dir, name, _suffix) if os.path.exists(_p): os.remove(_p) @@ -1027,14 +1189,16 @@ class CloudHypervisorExecutor(Executor): return cleaned # -- 客内执行 ------------------------------------------------------ - def _run_guest(self, task, inner_args, **kwargs): + def _run_guest(self, task, inner_args, extra_env=None, **kwargs): """SSH 调客内 host.py,返回 Local/Docker 同款结果 dict (executor 字段改写为 cloud-hypervisor,并附 vm 名与 api-socket)。runner 异常即原样传播,由 worker 侧统一 - 写回 failed——本类不在这里把异常吞成假成功。""" + 写回 failed——本类不在这里把异常吞成假成功。 + extra_env(P1-1):经 SSH env 前缀进客机环境。""" self._validate() ssh_cmd = self.build_ssh_command( - task, self._guest_host_command(*inner_args, **kwargs)) + task, self._guest_host_command(*inner_args, **kwargs), + extra_env=extra_env) result = self._call_runner(ssh_cmd, timeout=self.task_timeout) # 方法内 import:见模块头注释的循环导入说明。 from cloud.sandbox import _parse_host_result @@ -1063,10 +1227,12 @@ class CloudHypervisorExecutor(Executor): return parse_goal_result(task_id, result, "cloud-hypervisor") - def _run_guest_goal(self, task, inner_args, **kwargs): + def _run_guest_goal(self, task, inner_args, extra_env=None, + **kwargs): self._validate() ssh_cmd = self.build_ssh_command( - task, self._guest_host_command(*inner_args, **kwargs)) + task, self._guest_host_command(*inner_args, **kwargs), + extra_env=extra_env) return self._goal_result( task["id"], self._call_runner(ssh_cmd, timeout=self.task_timeout)) @@ -1137,7 +1303,8 @@ class CloudHypervisorExecutor(Executor): "interrupted_reason": reason, } - def _run_with_vm(self, task, inner_args, goal=False, **kwargs): + def _run_with_vm(self, task, inner_args, goal=False, extra_env=None, + **kwargs): """每任务 VM 生命周期:wake_vm(API 模式 + SSH 就绪) → 跑客内命令 → finally shutdown_vm。wake 失败时不跑 任务、直接抛分类失败。本方法不做内部重试:重试是 @@ -1155,8 +1322,9 @@ class CloudHypervisorExecutor(Executor): try: if goal: return self._run_guest_goal( - task, inner_args, **kwargs) - result = self._run_guest(task, inner_args, **kwargs) + task, inner_args, extra_env=extra_env, **kwargs) + result = self._run_guest(task, inner_args, + extra_env=extra_env, **kwargs) except Exception as exc: if self._vm_proc_dead(task): return self._vm_death_result( @@ -1195,10 +1363,25 @@ class CloudHypervisorExecutor(Executor): "execute() 只读已运行 VM 的 status:VM %s 未运行," "请先 wake_vm() 再调 execute(),不要用 execute " "唤醒 VM" % self.vm_name(task)) - return self._run_guest( - task, ("status", "--task", self._host_name(task))) + try: + return self._run_guest( + task, ("status", "--task", self._host_name(task))) + except Exception as exc: + # feat-chv-fix(P1-8):socket 残留但 VMM 已死时, + # _vm_is_running 会因 socket 存在而放行,随后的 + # SSH 失败不得以原始传输异常抛出——登记进程已死 + # 或 socket 已消失即统一归类 vm_not_running;VM + # 仍活着时的 SSH 失败原样传播(那是客内 status + # 失败,不是 VM 未运行)。 + if self._vm_proc_dead(task) \ + or not os.path.exists(self.api_socket(task)): + raise ChvFailure( + "vm_not_running", + "execute() 时 VM 已死(socket 残留/进程已死)," + "SSH 失败归类为未运行:%s" % exc) + raise - def dispatch_task(self, task) -> dict: + def dispatch_task(self, task, extra_env=None, model_provider=None) -> dict: """客内 host.py dispatch:建 host 任务并跑一轮 turn。""" if not isinstance(task, dict) or task.get("id") is None: raise ValueError("任务缺少 id") @@ -1210,11 +1393,12 @@ class CloudHypervisorExecutor(Executor): if repo: args += ["--repo", repo] return self._run_with_vm( - task, args, model=task.get("model"), + task, args, extra_env=extra_env, model=task.get("model"), effort=task.get("effort"), approval=task.get("approval"), + model_provider=model_provider, sandbox=("read-only" if task.get("plan") else None)) - def run_task(self, task) -> dict: + def run_task(self, task, extra_env=None, model_provider=None) -> dict: """客内 host.py run:在已有 host 任务上再跑一轮 turn。""" if not isinstance(task, dict) or task.get("id") is None: raise ValueError("任务缺少 id") @@ -1223,10 +1407,11 @@ class CloudHypervisorExecutor(Executor): return self._run_with_vm( task, ("run", "--task", self._host_name(task), "--prompt", task["prompt"]), - model=task.get("model"), effort=task.get("effort"), - approval=task.get("approval")) + extra_env=extra_env, model=task.get("model"), + effort=task.get("effort"), approval=task.get("approval"), + model_provider=model_provider) - def goal_task(self, task) -> dict: + def goal_task(self, task, extra_env=None, model_provider=None) -> dict: """客内 host.py goal:目标循环,轮次规则与 Local 同款 (缺省 3 轮,>100 轮 loud fail,旧验收形态 0 轮禁用)。""" if not isinstance(task, dict) or task.get("id") is None: @@ -1252,11 +1437,13 @@ class CloudHypervisorExecutor(Executor): if repo: args += ["--repo", repo] return self._run_with_vm( - task, args, goal=True, model=task.get("model"), - effort=task.get("effort"), approval=task.get("approval")) + task, args, goal=True, extra_env=extra_env, + model=task.get("model"), + effort=task.get("effort"), approval=task.get("approval"), + model_provider=model_provider) def goal_resume_task(self, task, goal_text=None, - max_rounds=None) -> dict: + max_rounds=None, extra_env=None, model_provider=None) -> dict: """客内 host.py goal-resume:goal_text=None 为断点续跑 (不带 --goal-text,沿用原 goal+verify),给 goal_text 为开新目标(带 --goal-text 与 --max-rounds,默认 0)。""" @@ -1269,11 +1456,14 @@ class CloudHypervisorExecutor(Executor): elif max_rounds is not None: args += ["--max-rounds", str(max_rounds)] return self._run_with_vm( - task, args, goal=True, model=task.get("model"), - effort=task.get("effort"), approval=task.get("approval")) + task, args, goal=True, extra_env=extra_env, + model=task.get("model"), + effort=task.get("effort"), approval=task.get("approval"), + model_provider=model_provider) def goal_resume_breakpoint_task(self, task, - model_provider=None) -> dict: + model_provider=None, + extra_env=None) -> dict: """客内 host.py goal-resume 断点续跑:无 --goal-text、 无 --max-rounds,host 沿用原任务从断点轮次续跑。 worker 层已接线(补缺口轨):非网关的 CHV 断点任务 @@ -1286,7 +1476,7 @@ class CloudHypervisorExecutor(Executor): raise ValueError("任务缺少 id") return self._run_with_vm( task, ("goal-resume", "--task", self._host_name(task)), - goal=True, model=task.get("model"), + goal=True, extra_env=extra_env, model=task.get("model"), effort=task.get("effort"), approval=task.get("approval"), model_provider=model_provider) diff --git a/cloud/chv/hostkeys.py b/cloud/chv/hostkeys.py index 0e2a1c8..b9854c2 100644 --- a/cloud/chv/hostkeys.py +++ b/cloud/chv/hostkeys.py @@ -44,16 +44,49 @@ def host_key_paths(state_dir, vm_name): } -def known_hosts_line(port, public_key_text): +def known_hosts_line(port, public_key_text, host="127.0.0.1"): """公钥文件内容转 known_hosts 行:只取前两列(类型+数据)。 公钥文件形如 'ssh-ed25519 AAAA... [comment]',known_hosts - 行形如 '[127.0.0.1]: ssh-ed25519 AAAA...'。非法 - 内容 ValueError,绝不把陌生行原样写进文件。""" + 行形如 '[]: ssh-ed25519 AAAA...'(host 缺省 + 回环,仅作兼容;客机 IP 方案下 host 应为解析出的客机 IP)。 + 非法内容 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]) + return "[%s]:%s %s %s" % (host, port, parts[0], parts[1]) + + +def seed_disk_path(state_dir, vm_name): + """该 VM 的 NoCloud seed 磁盘镜像路径(CIDATA 卷标 ISO)。""" + return os.path.join(state_dir, vm_name, "seed.iso") + + +def build_seed_disk(seed_dir, seed_disk, runner=None): + """把 seed 目录做成 CIDATA 卷标 ISO(feat-chv-fix,P0-1 的 + 注入载体):genisoimage -volid CIDATA -joliet -rock。 + + runner 不给时用 subprocess;测试可注入假 runner(假 + runner 负责把 seed_disk 文件写出来)。工具失败、返回码 + 非 0、或产物不存在/为空时一律 RuntimeError,调用方 + (wake_vm provisioned 路径)fail-fast,绝不带着“seed + 目录已写但未挂进 VmConfig”的 VM 继续启动。seed 目录 + 本身只是中间物,不再被当成注入通道。""" + cmd = ["genisoimage", "-output", seed_disk, "-volid", "CIDATA", + "-joliet", "-rock", seed_dir] + if runner is not None: + result = runner(cmd) + rc = getattr(result, "returncode", 0) + if rc not in (0, None): + raise RuntimeError("genisoimage 失败:returncode=%s" % rc) + else: + import subprocess + result = subprocess.run(cmd, capture_output=True, text=True) + if result.returncode != 0: + raise RuntimeError("genisoimage 失败:%s" % result.stderr) + if not os.path.isfile(seed_disk) or os.path.getsize(seed_disk) == 0: + raise RuntimeError("seed 磁盘未生成:%s" % seed_disk) + return seed_disk def build_seed_user_data(private_key_text, public_key_text): diff --git a/cloud/chv/pool.py b/cloud/chv/pool.py index cb539cf..1f06a4e 100644 --- a/cloud/chv/pool.py +++ b/cloud/chv/pool.py @@ -220,20 +220,43 @@ class WarmPool: return added def acquire(self, task): - """分配一个预热 VM 给任务:有 warm 条目时消耗一个 - (状态 assigned,返回 mode=warm + 该条目的合成任务 - 名,供调用方 resume 后使用);无 warm 条目时返回 - mode=cold(调用方回落 wake_vm 冷启动)。""" + """分配一个预热 VM 给任务(feat-chv-fix,P1-3 接线): + + 有 warm 条目时先调 executor.resume_vm() 把暂停的 VM + 真正唤醒(resume 失败即条目作废、继续找下一个, + 绝不把 paused 的 VM 当 running 发给任务),再标 + assigned 并返回 mode=warm;resume 全部失败或无 warm + 条目时返回 mode=cold(调用方回落 wake_vm 冷启动)。 + 注意:本方法只解决“acquire 不再悬空 resume 责任” + 这一半;worker dispatch 前查池的另一半仍未接线 + (见执行器/就绪文档的“未接线缺口”标注),预热收益 + 在生产路径上仍不是已交付特性。""" for name, entry in list(self._entries.items()): - if entry["state"] == "warm": - entry["state"] = "assigned" - entry["task"] = task - self._save_registry() + if entry["state"] != "warm": + continue + resume = getattr(self.executor, "resume_vm", None) + if resume is not None: + try: + resume(entry.get("task") or task) + except Exception: + self._entries.pop(name, None) + self._save_registry() + continue + # 无 resume_vm 的鸭子型执行器(旧测试替身): + # 保持旧 assigned 语义,不因此丢弃条目;真实 + # CloudHypervisorExecutor 一律有 resume_vm,走 + # 上面分支,责任不悬空。 + entry["state"] = "assigned" + entry["task"] = task + self._save_registry() + try: self.executor._emit( "warm_pool_acquire", task, "vm=%s" % name) - return {"mode": "warm", "vm": name, - "pool_task": entry["task"]} + except Exception: + pass + return {"mode": "warm", "vm": name, + "pool_task": entry["task"]} return {"mode": "cold", "vm": None, "pool_task": None} def release(self, task, pool_vm=None): diff --git a/cloud/chv/ssh_transport.py b/cloud/chv/ssh_transport.py index 5001f49..52721da 100644 --- a/cloud/chv/ssh_transport.py +++ b/cloud/chv/ssh_transport.py @@ -34,34 +34,48 @@ def known_hosts_path(state_dir, vm_name): def build_ssh_command(ssh_port, ssh_key, ssh_user, egress_proxy, - known_hosts_file, inner_cmd): + known_hosts_file, inner_cmd, ssh_host="127.0.0.1", + extra_env=None): """构造 SSH 命令:ssh -p <端口> -i + 钉死选项 - + user@127.0.0.1 + 客内命令。 + + user@ + 客内命令。 - 与 #216 的区别只有 host key 选项:StrictHostKeyChecking=yes - + UserKnownHostsFile=<每 VM 文件>,不再是 no + /dev/null。 - BatchMode=yes 不变(绝不交互问 host key / 口令)。 + feat-chv-fix(P0-4):ssh_host 不再在传输层写死回环, + 由执行器解析传入(显式 CHV_SSH_HOST / net 客机 IP / + 显式转发宣称下的回环);本函数缺省回环仅为旧调用点 + 兼容,执行器路径绝不依赖该缺省。 + extra_env(feat-chv-fix,P1-1 网关注入):经同一 env + 前缀进客机环境(与 egress 代理前缀同款机制),不进 + 镜像/代码/prompt;值本身只出现在宿主侧 ssh argv(与 + worker 同一信任边界),执行器绝不把完整命令记入日志。 egress_proxy 已设时客内命令前加 env 前缀(代理地址本身 不是密钥;客机出口只走该白名单代理)。""" inner = list(inner_cmd) + env_assignments = [] + if extra_env: + for key in sorted(extra_env): + env_assignments.append("%s=%s" % (key, extra_env[key])) if egress_proxy: - inner = ["env", - "HTTPS_PROXY=%s" % egress_proxy, - "HTTP_PROXY=%s" % egress_proxy, - "NO_PROXY=localhost,127.0.0.1"] + inner + env_assignments += [ + "HTTPS_PROXY=%s" % egress_proxy, + "HTTP_PROXY=%s" % egress_proxy, + "NO_PROXY=localhost,127.0.0.1"] + if env_assignments: + inner = ["env"] + env_assignments + inner return ["ssh", "-p", str(ssh_port), "-i", ssh_key, "-o", "StrictHostKeyChecking=yes", "-o", "UserKnownHostsFile=%s" % known_hosts_file, "-o", "BatchMode=yes", - "%s@127.0.0.1" % ssh_user] + inner + "%s@%s" % (ssh_user, ssh_host)] + inner -def parse_keyscan_output(text): - """从 ssh-keyscan 的 stdout 里挑出本机回环端口的 host key - 行:形如 "[127.0.0.1]: ssh-ed25519 AAAA..."(-p 时 - 的标准写法)或 "127.0.0.1 ssh-rsa AAAA..."。其它主机行、 - 注释行、空行一律丢掉,绝不把陌生主机的 key 写进文件。""" +def parse_keyscan_output(text, host="127.0.0.1"): + """从 ssh-keyscan 的 stdout 里挑出目标主机的 host key + 行:形如 "[]: ssh-ed25519 AAAA..."(-p 时 + 的标准写法)或 " ssh-rsa AAAA..."。其它主机行、 + 注释行、空行一律丢掉,绝不把陌生主机的 key 写进文件。 + host 缺省回环仅为旧调用点兼容;客机 IP 方案下应传 + 解析出的客机 IP。""" kept = [] for line in (text or "").splitlines(): line = line.strip() @@ -70,9 +84,9 @@ def parse_keyscan_output(text): parts = line.split() if len(parts) < 3: continue - host, keytype = parts[0], parts[1] - if host != "127.0.0.1" \ - and not host.startswith("[127.0.0.1]:"): + line_host, keytype = parts[0], parts[1] + if line_host != host \ + and not line_host.startswith("[%s]:" % host): continue if not keytype.startswith("ssh-"): continue diff --git a/cloud/tests/test_chv_fix.py b/cloud/tests/test_chv_fix.py new file mode 100644 index 0000000..5695f4b --- /dev/null +++ b/cloud/tests/test_chv_fix.py @@ -0,0 +1,554 @@ +"""CHV 审核修复轨测试(堆叠于 #230,base=feat-chv-gaps)。 + +只用标准库与 mock runner/launcher/api_client,绝不启动真实 +VM:当前环境没有可跑真 VM 的宿主。真实 VM 启动、seed 盘 +在真机上的 cloud-init 应用、SSH 可达实测一律不在本文件 +声称已验证;本文件只证明代码层面的接线(VmConfig 含 seed +盘、restore 不回落、reuse 拦截、SSH 目标解析、配额/快照/ +API 完整性等)成立。 +""" + +import json +import os +import socket +import sys +import tempfile +import threading +import unittest +from pathlib import Path +from types import SimpleNamespace + +try: + from cloud.chv.api_client import _unix_socket_api + from cloud.chv.disk import check_disk_quota, provision_task_disk + from cloud.chv.hostkeys import build_seed_disk, seed_disk_path + from cloud.chv.pool import WarmPool + from cloud.sandbox import ( + ChvFailure, + CloudHypervisorExecutor, + ) + from cloud.worker import Worker +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.chv.api_client import _unix_socket_api + from cloud.chv.disk import check_disk_quota, provision_task_disk + from cloud.chv.hostkeys import build_seed_disk, seed_disk_path + from cloud.chv.pool import WarmPool + from cloud.sandbox import ( + ChvFailure, + CloudHypervisorExecutor, + ) + from cloud.worker import Worker + + +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 ChvFixCase(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.api_payloads = {} + 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" in cmd: + return FakeResult(0, "goal met after rounds\n", "") + 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): + self.api_calls.append((method, path)) + if path == "/vm.create": + self.api_payloads["vm.create"] = payload + return (200, {}) + + def make_exec(self, runner=None, env_extra=None, **overrides): + 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 + + +# -- P0-1 seed 磁盘接入 ------------------------------------------------- +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") + self.assertEqual(len(cfg["disks"]), 2) + self.assertEqual(cfg["disks"][1]["path"], "/tmp/x/seed.iso") + self.assertTrue(cfg["disks"][1]["readonly"]) + cfg2 = ex.build_vm_config("/tmp/x/disk.raw") + self.assertEqual(len(cfg2["disks"]), 1) + + def test_provisioned_wake_mounts_seed_disk(self): + ex = self.make_exec( + env_extra={"CHV_HOST_KEY_MODE": "provisioned"}) + out = ex.wake_vm({"id": "seed1"}) + self.assertEqual(out["state"], "running") + payload = self.api_payloads["vm.create"] + self.assertEqual(len(payload["disks"]), 2) + seed_disk = payload["disks"][1]["path"] + self.assertTrue(seed_disk.endswith("seed.iso")) + self.assertTrue(payload["disks"][1]["readonly"]) + self.assertTrue(os.path.exists(seed_disk)) + self.assertIn("genisoimage", [c[0] for c in self.calls]) + + def test_seed_disk_build_failure_is_fail_fast(self): + def bad_runner(cmd, timeout=None): + self.calls.append(list(cmd)) + if cmd and cmd[0] == "genisoimage": + return FakeResult(1, "", "no genisoimage") + return self._default_runner(cmd, timeout) + + ex = self.make_exec( + runner=bad_runner, + env_extra={"CHV_HOST_KEY_MODE": "provisioned"}) + with self.assertRaises(RuntimeError): + ex.wake_vm({"id": "seedfail"}) + # 制盘失败在起 VMM 之前,不应走到 vm.create。 + self.assertNotIn("/vm.create", + [p for _, p in self.api_calls]) + + def test_build_seed_disk_rejects_missing_product(self): + with self.assertRaises(RuntimeError): + build_seed_disk(self.dir, os.path.join(self.dir, "no.iso"), + runner=lambda cmd: FakeResult(0, "", "")) + + +# -- P0-2 restore 不回落 ------------------------------------------------ +class RestoreNoFallbackTest(ChvFixCase): + def test_wait_guest_ready_provisioned_missing_known_hosts(self): + ex = self.make_exec( + env_extra={"CHV_HOST_KEY_MODE": "provisioned"}) + with self.assertRaises(ChvFailure) as ctx: + ex.wait_guest_ready({"id": "nopin"}) + self.assertEqual(ctx.exception.kind, "host_key_not_pinned") + self.assertFalse(any(c[0] == "ssh-keyscan" for c in self.calls)) + + def test_restore_provisioned_missing_known_hosts_fails_before_launch(self): + ex = self.make_exec( + env_extra={"CHV_HOST_KEY_MODE": "provisioned"}) + launches = [] + ex.launcher = lambda cmd: launches.append(cmd) or FakeProc(True) + with self.assertRaises(ChvFailure) as ctx: + ex.restore_vm({"id": "r1"}, "file:///snap/x") + self.assertEqual(ctx.exception.kind, "host_key_not_pinned") + self.assertEqual(launches, []) + + +# -- P0-3 reuse 拦截 ---------------------------------------------------- +class ReuseBlockTest(ChvFixCase): + def test_reuse_blocked_without_experiment_flag(self): + ex = self.make_exec(env_extra={"CHV_DISK_MODE": "reuse"}) + with self.assertRaises(ChvFailure) as ctx: + ex.preflight() + self.assertEqual(ctx.exception.kind, + "disk_mode_reuse_blocked") + + 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"}) + self.assertTrue(ex.preflight()) + out = ex.wake_vm({"id": "reuse1"}) + self.assertEqual(out["state"], "running") + self.assertTrue(any(e == "disk_reuse_experiment_allowed" + for e, _, _, _ in self.audit)) + + +# -- P0-4 SSH 目标解析 -------------------------------------------------- +class SshTargetTest(ChvFixCase): + def test_explicit_host_wins(self): + ex = self.make_exec(ssh_host="10.9.8.7") + self.assertEqual(ex.resolve_ssh_host(), "10.9.8.7") + + def test_net_guest_ip_used_when_no_explicit_host(self): + ex = self.make_exec(env_extra={ + "CHV_NET_JSON": '[{"tap": "", "ip": "192.168.100.5",' + '"mask": "255.255.255.0"}]', + "CHV_SSH_FORWARD": "0"}) + # env_extra 覆盖 forward=0 后仍应走 net ip,不报未声明。 + self.assertEqual(ex.resolve_ssh_host(), "192.168.100.5") + cmd = ex.build_ssh_command({"id": "t"}, ["true"]) + self.assertIn("root@192.168.100.5", " ".join(cmd)) + + def test_undeclared_target_is_fail_fast(self): + ex = self.make_exec(env_extra={"CHV_SSH_FORWARD": "0"}) + with self.assertRaises(ChvFailure) as ctx: + ex.resolve_ssh_host() + self.assertEqual(ctx.exception.kind, "ssh_target_undeclared") + + def test_forward_attestation_allows_loopback(self): + ex = self.make_exec() + self.assertEqual(ex.resolve_ssh_host(), "127.0.0.1") + + +# -- P1-1 网关 extra_env ------------------------------------------------ +class GatewayExtraEnvTest(ChvFixCase): + def test_chv_ssh_command_carries_extra_env(self): + ex = self.make_exec() + cmd = ex.build_ssh_command({"id": "t"}, ["true"], + extra_env={"SUB2API_API_KEY": "k1"}) + joined = " ".join(cmd) + self.assertIn("SUB2API_API_KEY=k1", joined) + self.assertIn("HTTPS_PROXY=http://10.0.0.1:8888", joined) + + def test_worker_gateway_guard_allows_chv(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"}) + task = {"id": "gw1", "project_id": "p1", "prompt": "做目标", + "goal_mode": 1, + "_gateway_provider": "sub2api", + "_gateway_env": {"SUB2API_API_KEY": "gwkey", + "SUB2API_BASE_URL": "http://gw/"}} + result, error = worker._run_attempt(task) + 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) + for c in ssh_calls)) + + +# -- P1-2 取消/清理/看门狗 ---------------------------------------------- +class CancelCleanupTest(ChvFixCase): + def test_worker_cancel_calls_cancel_vm(self): + from cloud.server import init_db + db_path = os.path.join(self.dir, "control.db") + init_db(db_path) + ex = self.make_exec() + cancelled = [] + ex.cancel_vm = lambda name: cancelled.append(name) or True + worker = Worker(db_path, "/fake/host.py", executor=ex, + env={"XCM_API_KEY": "test-placeholder"}) + worker._running_set("t-cancel") + import sqlite3, time + conn = sqlite3.connect(db_path) + conn.execute( + "INSERT INTO tasks(id, project_id, prompt, status," + " idempotency_key, depends_on, priority, created_by," + " created_at, updated_at, last_heartbeat)" + " VALUES(?, ?, ?, ?, NULL, ?, ?, 'tester', ?, ?, NULL)", + ("t-cancel", "p1", "做事", "running", "[]", 0, + time.time(), time.time())) + conn.commit() + conn.close() + self.assertTrue(worker.cancel("t-cancel")) + self.assertEqual(cancelled, ["t-cancel"]) + + def test_worker_cleanup_calls_cleanup_stale_vms(self): + from cloud.server import init_db + db_path = os.path.join(self.dir, "control.db") + init_db(db_path) + ex = self.make_exec() + ex.cleanup_stale_vms = lambda: ["chv-stale"] + worker = Worker(db_path, "/fake/host.py", executor=ex, + env={"XCM_API_KEY": "test-placeholder"}) + self.assertIn("chv-stale", worker.cleanup_stale_containers()) + + def test_chv_cancel_watchdog_kills_vm_on_cancel(self): + import threading + from cloud.server import init_db + db_path = os.path.join(self.dir, "control.db") + init_db(db_path) + ex = self.make_exec() + killed = [] + ex.cancel_vm = lambda name: killed.append(name) or True + worker = Worker(db_path, "/fake/host.py", executor=ex, + env={"XCM_API_KEY": "test-placeholder"}) + worker._cancel_requested.add("t-wd") + stop = threading.Event() + worker._chv_cancel_watchdog("t-wd", "chv-t-wd", stop) + self.assertEqual(killed, ["chv-t-wd"]) + + +# -- P1-3 WarmPool acquire resume --------------------------------------- +class PoolAcquireResumeTest(unittest.TestCase): + def test_acquire_resumes_paused_vm(self): + class Stub: + def __init__(self): + self.resumed = [] + self.state_dir = None + self.warm_pool_size = 1 + self.snapshot_url = "file:///snap/base" + + def vm_name(self, task): + raw = task.get("id") if isinstance(task, dict) \ + else str(task) + return raw if raw.startswith("chv-") \ + else "chv-" + raw + + def wake_vm(self, task): + return {"vm": self.vm_name(task)} + + def pause_vm(self, task): + return True + + def resume_vm(self, task): + self.resumed.append(task["id"]) + return True + + def shutdown_vm(self, task_or_name): + return True + + def _emit(self, event, task, detail="", kind=None): + pass + + stub = Stub() + pool = WarmPool(stub, size=1, snapshot_url="file:///snap/base", + registry_path=None) + pool.prewarm() + got = pool.acquire({"id": "t1"}) + self.assertEqual(got["mode"], "warm") + self.assertEqual(stub.resumed, ["warm-1"]) + + def test_acquire_discards_vm_whose_resume_fails(self): + class Stub: + def __init__(self): + self.state_dir = None + self.warm_pool_size = 1 + self.snapshot_url = "file:///snap/base" + + def vm_name(self, task): + raw = task.get("id") if isinstance(task, dict) \ + else str(task) + return raw if raw.startswith("chv-") \ + else "chv-" + raw + + def wake_vm(self, task): + return {"vm": self.vm_name(task)} + + def pause_vm(self, task): + return True + + def resume_vm(self, task): + raise RuntimeError("resume 失败") + + def shutdown_vm(self, task_or_name): + return True + + def _emit(self, event, task, detail="", kind=None): + pass + + pool = WarmPool(Stub(), size=1, snapshot_url="file:///snap/base", + registry_path=None) + pool.prewarm() + self.assertEqual(pool.acquire({"id": "t1"})["mode"], "cold") + + +# -- P1-4 容量持久化计数 ------------------------------------------------- +class CapacityPersistentTest(ChvFixCase): + def test_socket_counts_after_restart(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-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") + + def test_own_socket_does_not_block_rewake(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-t1.sock"), + "w") as fh: + fh.write("") + # 同任务重复 wake:自己的残留 socket 不计入容量。 + out = ex.wake_vm({"id": "t1"}) + self.assertEqual(out["state"], "running") + + +# -- P1-5 磁盘配额 ------------------------------------------------------- +class DiskQuotaTest(ChvFixCase): + def test_unreadable_image_is_not_silently_skipped(self): + with self.assertRaises(ChvFailure) as ctx: + check_disk_quota("/nonexistent/disk.raw", 1.0) + self.assertEqual(ctx.exception.kind, "disk_quota_unverified") + + def test_total_usage_over_quota(self): + os.makedirs(self.state_dir, exist_ok=True) + with open(os.path.join(self.state_dir, "old.raw"), + "wb") as fh: + fh.write(b"X" * 100) + # 配额 150 字节:已用 100 + 本次 100 > 150。 + with self.assertRaises(ChvFailure) as ctx: + check_disk_quota(self.rootfs, 150 / (1024 ** 3), + state_dir=self.state_dir) + self.assertEqual(ctx.exception.kind, "disk_quota_exceeded") + + def test_existing_disk_size_mismatch_is_recopied(self): + task_disk = provision_task_disk(self.state_dir, "chv-a", + self.rootfs, "copy") + with open(task_disk, "wb") as fh: + fh.write(b"Z" * 10) + task_disk2 = provision_task_disk(self.state_dir, "chv-a", + self.rootfs, "copy") + self.assertEqual(task_disk, task_disk2) + with open(task_disk2, "rb") as fh: + self.assertEqual(fh.read(), b"R" * 100) + + +# -- P1-6 API 截断 ------------------------------------------------------- +class ApiTruncationTest(unittest.TestCase): + def _serve(self, raw): + tmp = tempfile.TemporaryDirectory() + sock_path = os.path.join(tmp.name, "vmm.sock") + srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + srv.bind(sock_path) + srv.listen(1) + + def _run(): + conn, _ = srv.accept() + try: + conn.recv(65536) + conn.sendall(raw) + finally: + conn.close() + srv.close() + + threading.Thread(target=_run, daemon=True).start() + return tmp, sock_path + + def test_truncated_body_raises(self): + body = b'{"ok": true}' + raw = (b"HTTP/1.1 200 OK\r\nContent-Length: 100\r\n\r\n" + + body) + tmp, sock_path = self._serve(raw) + with self.assertRaises(RuntimeError): + _unix_socket_api(sock_path, "GET", "/vmm.ping") + tmp.cleanup() + + def test_chunked_rejected(self): + raw = (b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n" + b"5\r\nhello\r\n0\r\n\r\n") + tmp, sock_path = self._serve(raw) + with self.assertRaises(RuntimeError): + _unix_socket_api(sock_path, "GET", "/vmm.ping") + tmp.cleanup() + + def test_full_body_parses(self): + body = b'{"ok": true}' + raw = (b"HTTP/1.1 200 OK\r\nContent-Length: %d\r\n\r\n" + % len(body)) + body + tmp, sock_path = self._serve(raw) + status, parsed = _unix_socket_api(sock_path, "GET", "/vmm.ping") + self.assertEqual(status, 200) + self.assertTrue(parsed["ok"]) + tmp.cleanup() + + +# -- P1-7 快照落盘验证 --------------------------------------------------- +class SnapshotPersistTest(ChvFixCase): + def test_snapshot_missing_file_does_not_shutdown(self): + ex = self.make_exec() + with self.assertRaises(ChvFailure) as ctx: + ex.hibernate_vm({"id": "h1"}, "file:///nonexistent/snap") + 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) + self.assertNotIn("shutdown", kinds) + + +# -- P1-8 execute 死端口归类 --------------------------------------------- +class ExecuteDeadPortTest(ChvFixCase): + def test_dead_socket_ssh_failure_is_vm_not_running(self): + def raising_runner(cmd, timeout=None): + self.calls.append(list(cmd)) + if cmd and cmd[0] == "ssh": + raise RuntimeError("connection refused") + return FakeResult(0, "", "") + + ex = self.make_exec(runner=raising_runner) + os.makedirs(self.state_dir, exist_ok=True) + with open(os.path.join(self.state_dir, "chv-dead.sock"), + "w") as fh: + fh.write("") + ex._vms["chv-dead"] = FakeProc(alive=False) + with self.assertRaises(ChvFailure) as ctx: + ex.execute({"id": "dead"}) + self.assertEqual(ctx.exception.kind, "vm_not_running") + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_chv_gaps.py b/cloud/tests/test_chv_gaps.py index e54f671..dc75227 100644 --- a/cloud/tests/test_chv_gaps.py +++ b/cloud/tests/test_chv_gaps.py @@ -119,7 +119,10 @@ class ChvGapCase(unittest.TestCase): return FakeProc(alive=True) def make_exec(self, runner=None, env_extra=None, **overrides): - env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888"} + env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888", + # 实验室宣称:mock runner 假装回环端口已转发到客机 + #(feat-chv-fix 起,执行器不再无条件假设回环能通)。 + "CHV_SSH_FORWARD": "1"} if env_extra: env.update(env_extra) kwargs = dict( diff --git a/cloud/tests/test_chv_optimize.py b/cloud/tests/test_chv_optimize.py index f9cc4de..288e07a 100644 --- a/cloud/tests/test_chv_optimize.py +++ b/cloud/tests/test_chv_optimize.py @@ -108,7 +108,10 @@ class ChvOptCase(unittest.TestCase): def make_exec(self, runner=None, api_client=None, launcher=None, env_extra=None, **overrides): - env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888"} + env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888", + # 实验室宣称:mock runner 假装回环端口已转发到客机 + #(feat-chv-fix 起,执行器不再无条件假设回环能通)。 + "CHV_SSH_FORWARD": "1"} if env_extra: env.update(env_extra) if runner is None: diff --git a/cloud/tests/test_cloud_hypervisor.py b/cloud/tests/test_cloud_hypervisor.py index 56e76e5..8309a3f 100644 --- a/cloud/tests/test_cloud_hypervisor.py +++ b/cloud/tests/test_cloud_hypervisor.py @@ -98,7 +98,10 @@ class ChvCase(unittest.TestCase): def make_exec(self, runner=None, api_client=None, launcher=None, env_extra=None, **overrides): - env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888"} + env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888", + # 实验室宣称:mock runner 假装回环端口已转发到客机 + #(feat-chv-fix 起,执行器不再无条件假设回环能通)。 + "CHV_SSH_FORWARD": "1"} if env_extra: env.update(env_extra) if runner is None: @@ -410,9 +413,15 @@ class LifecycleTest(ChvCase): def test_hibernate_is_pause_snapshot_shutdown(self): ex = self.make_exec() - out = ex.hibernate_vm(self._task(), "file:///snap/h1") + # feat-chv-fix(P1-7):file:// 快照须真实存在且非空, + # 否则 hibernate 不 shutdown;用临时文件作快照目标。 + snap = os.path.join(self.dir, "h1.snap") + with open(snap, "wb") as fh: + fh.write(b"S" * 64) + snap_url = "file://" + snap + out = ex.hibernate_vm(self._task(), snap_url) self.assertEqual(out["state"], "hibernated") - self.assertEqual(out["snapshot"], "file:///snap/h1") + self.assertEqual(out["snapshot"], snap_url) kinds = [c[-1] for c in self.calls if c[0] == self.ch_remote] self.assertEqual(kinds[0], "pause") self.assertTrue(any(a.startswith("source_url") @@ -420,7 +429,7 @@ class LifecycleTest(ChvCase): self.assertIn("shutdown", kinds) # 从休眠唤醒 = restore + resume: ex2 = self.make_exec() - out2 = ex2.wake_from_hibernate(self._task(), "file:///snap/h1") + out2 = ex2.wake_from_hibernate(self._task(), snap_url) self.assertEqual(out2["mode"], "restored") def test_stop_dead_proc(self): diff --git a/cloud/worker.py b/cloud/worker.py index e0bc1cc..be14c42 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -333,22 +333,28 @@ def _merge_env(base, override): return merged -def _call_goal_resume_breakpoint(executor, task, model_provider=None): +def _call_goal_resume_breakpoint(executor, task, model_provider=None, + extra_env=None): """通用断点调用:执行器提供 goal_resume_breakpoint_task 时调用它。 Local 仍走原路径(调用方直接传 model_provider,不经过这里, 以免改生产语义);这里只用于非 Local 执行器:签名接受 model_provider 时传它(CHV/Docker 均接受),否则只传 task - (旧假执行器单参)。网关 key 绝不经这里传值:worker 侧 - gateway_env 只按名注入,见 _run_attempt 的网关守卫。""" + (旧假执行器单参)。feat-chv-fix(P1-1):CHV 的 + gateway_env 经此进入 SSH env 前缀注入客机环境(不进 + 镜像/代码/prompt);CHV 没有 Local 那样的进程 env 通道, + SSH env 前缀是唯一通道,值只出现在宿主侧 ssh argv。""" method = getattr(executor, "goal_resume_breakpoint_task") try: params = inspect.signature(method).parameters except (TypeError, ValueError): return method(task) + kwargs = {} if "model_provider" in params: - return method(task, model_provider=model_provider) - return method(task) + kwargs["model_provider"] = model_provider + if "extra_env" in params and extra_env is not None: + kwargs["extra_env"] = extra_env + return method(task, **kwargs) def _resolve_model(model=None, config_path=None): @@ -2368,6 +2374,12 @@ class Worker: self.executor.kill_container(entry["container"]) except Exception: pass + # feat-chv-fix(P1-2):CHV 取消真关 VM,不只改库状态。 + if hasattr(self.executor, "cancel_vm"): + try: + self.executor.cancel_vm(task_id) + except Exception: + pass self._running_clear(task_id) conn = _db_connect(self.db_path) try: @@ -2470,22 +2482,87 @@ class Worker: pass return + def _chv_cancel_watchdog(self, task_id, vm_name, stop_event): + """CHV 同步路径的取消看门狗(feat-chv-fix,P1-2,Docker + 取消看门狗同款模式):同步 SSH 阻塞时 worker 的 cancel() + 只能改库、清 _running 项,阻塞中的 run 本身不会醒; + 本线程每 1 秒查库/取消标志,命中时调 cancel_vm 真关 + VM,然后退出。查库失败时当没取消,不误杀。""" + while not stop_event.wait(1.0): + try: + if task_id in self._cancel_requested: + cancelled = True + else: + db_task = self.get_task(task_id) + cancelled = (db_task is not None and + db_task.get("status") == "interrupted") + except Exception: + cancelled = False + if not cancelled: + continue + try: + cancel = getattr(self.executor, "cancel_vm", None) + if cancel is not None and vm_name: + cancel(vm_name) + log_event(self.logger, "worker_chv_killed", + task_id=task_id, + message="cancel watchdog killed VM %s " + "(task interrupted/cancelled)" + % vm_name) + except Exception: + pass + return + + def _chv_quota_watchdog(self, task_id, vm_name, stop_event, + quota_check, interval=30.0): + """CHV 同步路径的配额看门狗(feat-chv-fix,P1-2,Docker + 配额看门狗同款):命中时记 _quota_hit_tasks 并调 + cancel_vm 真关 VM;_run_attempt 在同步返回后把结果 + 改写成 quota_exceeded,走统一配额中断收尾。""" + while not stop_event.wait(interval): + try: + over = bool(quota_check()) if quota_check else False + except Exception: + over = False + if not over: + continue + self._mark_quota_hit(task_id) + try: + cancel = getattr(self.executor, "cancel_vm", None) + if cancel is not None and vm_name: + cancel(vm_name) + log_event(self.logger, "worker_chv_killed", + task_id=task_id, + message="quota watchdog killed VM %s " + "(quota exceeded)" % vm_name) + except Exception: + pass + return + def cleanup_stale_containers(self): """启动清理兜底:清掉上一轮残留的 codex- 容器。 - 只在执行器有 cleanup_stale_containers 时做(Docker); + Docker 执行器清容器、CHV 执行器清残留 VM(feat-chv-fix); 列出或删除失败只记日志,绝不抛异常炸掉 worker。 - 返回已清理的容器名列表(非 Docker 执行器返回 [])。 + 返回已清理的容器/VM 名列表(两者皆无的执行器返回 [])。 """ + cleaned = [] cleanup = getattr(self.executor, "cleanup_stale_containers", None) - if cleanup is None: - return [] - try: - cleaned = cleanup() - except Exception as exc: - log_event(self.logger, "worker_cleanup_failed", - message="cleanup stale containers failed: %s" % exc) - return [] + if cleanup is not None: + try: + cleaned = list(cleanup() or []) + except Exception as exc: + log_event(self.logger, "worker_cleanup_failed", + message="cleanup stale containers failed: %s" % exc) + # feat-chv-fix(P1-2):CHV 残留 VM 走通用清理路径, + # 不再只靠 worker 启动/关闭的 WarmPool reconcile 兜底。 + chv_cleanup = getattr(self.executor, "cleanup_stale_vms", None) + if chv_cleanup is not None: + try: + cleaned = cleaned + list(chv_cleanup() or []) + except Exception as exc: + log_event(self.logger, "worker_cleanup_failed", + message="cleanup stale vms failed: %s" % exc) if cleaned: log_event(self.logger, "worker_cleanup", message="cleaned stale containers: %s" @@ -4639,6 +4716,29 @@ class Worker: _watchdog_stop, quota_check), daemon=True) _quota_watchdog_thread.start() + # feat-chv-fix(P1-2):CHV 同步路径同款看门狗(VM 名 + # 由任务 id 派生)。进程模式不重复起(_execute_via_ + # process 已轮询 quota_check)。 + if isinstance(self.executor, CloudHypervisorExecutor) \ + and hasattr(self.executor, "cancel_vm"): + try: + _chv_vm_name = self.executor.vm_name(task) + except Exception: + _chv_vm_name = None + if _chv_vm_name: + _watchdog_stop = threading.Event() + _watchdog_thread = threading.Thread( + target=self._chv_cancel_watchdog, + args=(task_id, _chv_vm_name, _watchdog_stop), + daemon=True) + _watchdog_thread.start() + if quota_check is not None and not self._use_popen(): + _quota_watchdog_thread = threading.Thread( + target=self._chv_quota_watchdog, + args=(task_id, _chv_vm_name, + _watchdog_stop, quota_check), + daemon=True) + _quota_watchdog_thread.start() # 启动前分段日志(2026-10-29 任务 A):认领后 61 秒中断, # 但阻塞的 docker run 中途没有任何标记、容器 --rm 后输出 # 也没了,事后完全无法定位卡在哪一段。派发开始/结束进 @@ -4774,8 +4874,12 @@ class Worker: # Docker 路径上面已有 is_docker 守卫,不走这里。 # 网关任务在非 Local 执行器上无法注入 key 与 # provider,必须 loud fail,不静默跑错网关。 + # feat-chv-fix(P1-1):守卫收窄——CHV 已补 + # extra_env SSH 注入路径,不再被“非 Local 即 + # 拦”;无注入能力的执行器仍 loud fail。 if gateway_provider and not isinstance( - self.executor, LocalExecutor): + self.executor, + (LocalExecutor, CloudHypervisorExecutor)): raise RuntimeError( "sub2api 网关不支持该执行器:%s" % type(self.executor).__name__) @@ -4796,7 +4900,8 @@ class Worker: # 故这里不放宽网关规则。 result = _call_goal_resume_breakpoint( self.executor, task, - model_provider=gateway_provider) + model_provider=gateway_provider, + extra_env=gateway_env) elif task.get("_goal_resume_breakpoint"): # 假执行器/CloudExecutor 没有断点方法时 # 绝不回落成 goal(会撞已存在)或 @@ -4825,6 +4930,11 @@ class Worker: task, seed_dir=seed_host_dir, extra_env=gateway_env, model_provider=gateway_provider) + elif isinstance( + self.executor, CloudHypervisorExecutor): + result = self.executor.goal_task( + task, extra_env=gateway_env, + model_provider=gateway_provider) else: result = self.executor.goal_task(task) else: @@ -4833,6 +4943,11 @@ class Worker: task, seed_dir=seed_host_dir, extra_env=gateway_env, model_provider=gateway_provider) + elif isinstance( + self.executor, CloudHypervisorExecutor): + result = self.executor.dispatch_task( + task, extra_env=gateway_env, + model_provider=gateway_provider) else: result = self.executor.dispatch_task(task) combined = (result.get("stdout", "") or "") + \ @@ -4856,14 +4971,18 @@ class Worker: quota_check) else: if gateway_provider and not isinstance( - self.executor, LocalExecutor): + self.executor, + (LocalExecutor, + CloudHypervisorExecutor)): raise RuntimeError( "sub2api 网关不支持该执行器:%s" % type(self.executor).__name__) result = self.executor.run_task( task, extra_env=gateway_env, model_provider=gateway_provider) \ - if isinstance(self.executor, LocalExecutor) \ + if isinstance(self.executor, + (LocalExecutor, + CloudHypervisorExecutor)) \ else self.executor.run_task(task) except Exception as exc: # runner 异常也如实写回 failed result = {"executor": "local", "task_id": task_id, diff --git a/docs/chv-production-readiness.md b/docs/chv-production-readiness.md index 8636a83..181e327 100644 --- a/docs/chv-production-readiness.md +++ b/docs/chv-production-readiness.md @@ -36,7 +36,7 @@ shutdown,同一任务重复 wake 先关旧 VM;VM 名、api-socket、 每任务磁盘路径均按任务派生,互不相同(测试断言)。 - ✅🧪 每任务磁盘独立:copy 模式复制基础镜像为每任务磁盘; - reuse 模式仅单租户实验,生产禁用(代码与文档双写)。 + reuse 模式仅单租户实验,preflight 拦截(须显式实验二次确认并审计放行,feat-chv-fix 起代码真正拦截;此前“代码与文档双写”的说法是错的,代码当时并不拦截)。 - ✅🧪 租户无共享路径:执行器不挂载宿主根、宿主代码目录或含 密钥目录;客机代码在镜像内 `/app`,不在宿主共享挂载上。 - ✅🧪 密钥不进镜像/代码/prompt:SSH argv 只有 key 文件路径; @@ -44,7 +44,7 @@ 命令中不含哨兵密钥值。 - ✅🧪 SSH host key 按 VM 钉死(优化轨,#216 的 no + /dev/null 已废除):每 VM 在 state_dir//known_hosts 有独立文件, - 首次启动经 ssh-keyscan 在本机回环端口捕获 key 写入该文件, + 首次启动经 ssh-keyscan 在解析出的 SSH 目标上捕获 key 写入该文件, 此后 SSH 一律 StrictHostKeyChecking=yes + 该文件校验; 非 root 客机用户(CHV_SSH_USER,显式空串直接拒绝)与端口 (CHV_SSH_PORT,1-65535)为显式配置。mock 只验证捕获→ @@ -209,3 +209,33 @@ chv-guest 镜像构建发布 → 真实 VM 启动/快照恢复/任务端到端 验证 → 压测 → 安全审计 → CI 变绿后按当时 main 头复核 diff 再合并部署。缺一不可。 + +## 10. 审核修复轨 feat-chv-fix(堆叠于 #230 之上) + +- ✅🧪 P0 已修:seed 磁盘(CIDATA ISO,第二块只读盘)挂进 + VmConfig;restore 不回落 keyscan(provisioned 下复用已钉 + 死 known_hosts、缺失 fail-fast,快照内 host key 不重预置); + reuse 由 preflight 拦截(显式实验二次确认 + 审计放行); + SSH 目标选客机 IP 方案(显式 host > net 静态 ip > 显式 + 转发宣称下的回环,三者皆无 fail-fast;vsock 与纯 NAT 方案 + 未选,理由见 config.py)。 +- ✅🧪 P1 已修或有结论:网关 extra_env 经 SSH env 前缀注入, + worker 守卫收窄到 Local/CHV;取消/清理/看门狗接 CHV 通用 + 路径;WarmPool acquire 内 resume(但 worker dispatch 前查 + 池仍未接线,预热收益仍不是已交付特性);容量门改持久化 + 计数(socket 联合计数);磁盘配额改总量+余量、已存在磁盘 + 校验重拷;API 客户端按 Content-Length 验完整性;快照 + file:// 目标验存在且非空再 shutdown;execute 死端口归类 + vm_not_running。 +- ⚠️ 仍未接线/未验证(不得声称已交付):worker dispatch 前查 + WarmPool;CHV + 网关真实任务;真机 wake/boot/SSH(含 seed + 注入与 NAT/客机 IP 实测);快照恢复端到端;压测;安全审计; + CI 账单恢复。另:P0-2 审计原说法“restore 须把 seed 接入 + VmConfig”已纠正——restore 不走 vm.create,seed 无从挂盘, + 正确不变量是复用钉死 known_hosts。 +- ⛔ P2 全部排除、未做(用户紧急范围收窄令,留待以后一轨): + 长函数拆分、第二启动路径清理 / Docker goal 解析注释、 + test_chv_gaps.py 的 grep 文本断言改行为断言、build.sh 与 + README 口径对齐、stop_vm 吞异常改审计、配置校验时机统一。 + 本轨的 grep 文本测试、README/build.sh、stop_vm 吞异常、 + preflight 校验口径均保持 #230 原样。