diff --git a/cloud/chv/api_client.py b/cloud/chv/api_client.py index ac3cff3..64b7c6e 100644 --- a/cloud/chv/api_client.py +++ b/cloud/chv/api_client.py @@ -107,7 +107,7 @@ def _unix_socket_api(socket_path, method, path, payload=None): return status, parsed -def normalize_api_result(result): +def normalize_api_result(result) -> tuple: """API 调用归一化:fake api_client 可返回 (status, body) 或 {"status":..., "body":...} 或带 .status 的对象,本 函数统一成 (status:int, body)。""" diff --git a/cloud/chv/config.py b/cloud/chv/config.py index cd35b86..78c947b 100644 --- a/cloud/chv/config.py +++ b/cloud/chv/config.py @@ -50,24 +50,10 @@ class ChvConfig: setattr(self, key, value) -def build_chv_config(params, env): - """把显式构造参数(params,None=未给)与环境变量(env, - 默认为 os.environ 的调用方传入 dict)解析成 ChvConfig。 - 显式参数只有 __init__ 本来就有的那些;max_vms、磁盘模式、 - 磁盘配额、egress、快照、预热池目标数本来就没有显式参数, - 仍只走环境变量(与 #216 一致,不悄悄加新参数面)。""" - env = env if env is not None else {} - binary = params.get("binary") or env.get("CHV_BINARY") \ - or "cloud-hypervisor" - kernel = params.get("kernel") or env.get("CHV_KERNEL") - rootfs = params.get("rootfs") or params.get("image") \ - or env.get("CHV_ROOTFS") or env.get("CHV_IMAGE") - cpus = str(params.get("cpus") or env.get("CHV_CPU") or "2") - memory = params.get("memory") or env.get("CHV_MEM") or "2G" - state_dir = params.get("state_dir") or env.get("CHV_STATE_DIR") \ - or os.path.join(os.path.expanduser("~"), - ".local", "share", "codex-app-server", "chv-vms") +def _ssh_fragment(params: dict, env: dict) -> dict: + """SSH/命令面配置片段:返回 dict,供 build_chv_config 组装。 + 错误文案与原内联代码逐字一致,校验时机不变(构造时)。""" ssh_key = params.get("ssh_key") or env.get("CHV_SSH_KEY") if params.get("ssh_user") is not None \ and params.get("ssh_user") == "": @@ -84,32 +70,27 @@ def build_chv_config(params, env): raise ValueError("CHV_SSH_PORT 必须是整数") if not 1 <= ssh_port <= 65535: raise ValueError("CHV_SSH_PORT 必须在 1-65535 之间") - guest_host_py = params.get("guest_host_py") or "/app/host.py" - cmdline = params.get("cmdline") or env.get("CHV_CMDLINE") \ - or "console=hvc0 root=/dev/vda1 rw" - net = params.get("net") or env.get("CHV_NET") - net_json = params.get("net_json") or env.get("CHV_NET_JSON") - kvm_path = params.get("kvm_path") \ - if params.get("kvm_path") is not None else "/dev/kvm" - cpuinfo_path = params.get("cpuinfo_path") \ - 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") + host_key_mode = params.get("host_key_mode") \ + or env.get("CHV_HOST_KEY_MODE") or "provisioned" + if host_key_mode not in ("provisioned", "keyscan"): + raise ValueError( + "CHV_HOST_KEY_MODE 只能是 provisioned 或 keyscan:%s" + % host_key_mode) + return { + "ssh_key": ssh_key, "ssh_user": ssh_user, + "ssh_port": ssh_port, "host_key_mode": host_key_mode, + "guest_host_py": params.get("guest_host_py") or "/app/host.py", + "cmdline": params.get("cmdline") or env.get("CHV_CMDLINE") + or "console=hvc0 root=/dev/vda1 rw", + "net": params.get("net") or env.get("CHV_NET"), + "net_json": params.get("net_json") or env.get("CHV_NET_JSON"), + "ssh_host": params.get("ssh_host") or env.get("CHV_SSH_HOST"), + "ssh_forward": env.get("CHV_SSH_FORWARD") == "1", + } + + +def _quota_fragment(env: dict) -> dict: + """配额/磁盘/快照配置片段:返回 dict,错误文案逐字一致。""" try: max_vms = int(env.get("CHV_MAX_VMS") or 4) except ValueError: @@ -118,35 +99,13 @@ def build_chv_config(params, env): if disk_mode not in ("copy", "reuse"): raise ValueError( "CHV_DISK_MODE 只能是 copy 或 reuse:%s" % disk_mode) - disk_gb_raw = env.get("CHV_DISK_GB") - # feat-chv-secfix(M3):配额默认开启。CHV_DISK_GB 未设时 - # 不再是 None(配额门直接返回),而是 50 GB 总量预算; - # 显式设 0 或负数同样非法——配额门是 fail-closed,不是 - # “设个小值绕过”。check_disk_quota 仍接受 None(直接 - # 调用者的第二道门),但执行器路径永远带默认值。 try: - disk_gb = float(disk_gb_raw) if disk_gb_raw else 50.0 + disk_gb = float(env.get("CHV_DISK_GB")) \ + if env.get("CHV_DISK_GB") else 50.0 except ValueError: raise ValueError("CHV_DISK_GB 必须是数字") if disk_gb <= 0: raise ValueError("CHV_DISK_GB 必须大于 0") - 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 "provisioned" - if host_key_mode not in ("provisioned", "keyscan"): - raise ValueError( - "CHV_HOST_KEY_MODE 只能是 provisioned 或 keyscan:%s" - % host_key_mode) - # feat-chv-secfix(M6):reuse 实验放行须同时声明单租户 - # 项目(部署声明 + 审计,代码无法从环境变量验证数据库 - # 里是否真的只有一个租户——这一点在文档与 preflight 的 - # kind 里明写,不假装已校验)。 - single_tenant_project = env.get("CHV_SINGLE_TENANT_PROJECT") - # feat-chv-secfix(M3 快照上限 / L2 快照目录白名单): - # 快照数、总量、保留期均有默认值;file:// 快照只允许在 - # state_dir 或 CHV_SNAPSHOT_DIR 内,远端协议须显式放行 - # (见 hardening.validate_snapshot_url)。 try: snapshot_max_count = int(env.get("CHV_SNAPSHOT_MAX_COUNT") or 10) @@ -159,13 +118,29 @@ def build_chv_config(params, env): if snapshot_max_count < 0 or snapshot_max_total_gb < 0 \ or snapshot_retention_days < 0: raise ValueError("CHV_SNAPSHOT_* 不能为负数") - snapshot_dir = env.get("CHV_SNAPSHOT_DIR") - allow_remote_snapshot = env.get("CHV_ALLOW_REMOTE_SNAPSHOT") \ - == "1" + return { + "max_vms": max_vms, "disk_mode": disk_mode, + "disk_gb": disk_gb, + "reuse_experiment": env.get( + "CHV_REUSE_SINGLE_TENANT_EXPERIMENT") == "1", + "single_tenant_project": env.get("CHV_SINGLE_TENANT_PROJECT"), + "snapshot_max_count": snapshot_max_count, + "snapshot_max_total_gb": snapshot_max_total_gb, + "snapshot_retention_days": snapshot_retention_days, + "snapshot_dir": env.get("CHV_SNAPSHOT_DIR"), + "allow_remote_snapshot": env.get("CHV_ALLOW_REMOTE_SNAPSHOT") + == "1", + "egress_proxy": env.get("CHV_EGRESS_PROXY"), + "allow_direct_egress": env.get("CHV_ALLOW_DIRECT_EGRESS") + == "1", + } + + +def _timeout_fragment(params: dict, env: dict) -> dict: + """超时/池/快照 URL 配置片段:返回 dict,错误文案逐字一致。""" task_timeout_raw = env.get("CHV_TASK_TIMEOUT") task_timeout = float(task_timeout_raw) \ if task_timeout_raw else None - snapshot_url = env.get("CHV_SNAPSHOT_URL") try: warm_pool_size = int(env.get("CHV_WARM_POOL_SIZE") or 0) except ValueError: @@ -187,25 +162,69 @@ def build_chv_config(params, env): raise ValueError("CHV_*_TIMEOUT/INTERVAL 必须是数字") if ready_interval <= 0: ready_interval = 5 + return { + "task_timeout": task_timeout, "warm_pool_size": warm_pool_size, + "api_timeout": api_timeout, "ready_timeout": ready_timeout, + "ready_interval": ready_interval, + "snapshot_url": env.get("CHV_SNAPSHOT_URL"), + } + + +def build_chv_config(params: dict, env: dict): + """把显式构造参数(params,None=未给)与环境变量(env, + 默认为 os.environ 的调用方传入 dict)解析成 ChvConfig。 + + 显式参数只有 __init__ 本来就有的那些;max_vms、磁盘模式、 + 磁盘配额、egress、快照、预热池目标数本来就没有显式参数, + 仍只走环境变量(与 #216 一致,不悄悄加新参数面)。 + 片段化(feat-chv-arch95):SSH 面、配额面、超时面分别 + 由 _ssh_fragment/_quota_fragment/_timeout_fragment 解析, + 错误文案与校验顺序逐字一致;build_chv_config 只组装。""" + env = env if env is not None else {} + binary = params.get("binary") or env.get("CHV_BINARY") \ + or "cloud-hypervisor" + kernel = params.get("kernel") or env.get("CHV_KERNEL") + rootfs = params.get("rootfs") or params.get("image") \ + or env.get("CHV_ROOTFS") or env.get("CHV_IMAGE") + cpus = str(params.get("cpus") or env.get("CHV_CPU") or "2") + memory = params.get("memory") or env.get("CHV_MEM") or "2G" + state_dir = params.get("state_dir") or env.get("CHV_STATE_DIR") \ + or os.path.join(os.path.expanduser("~"), + ".local", "share", "codex-app-server", "chv-vms") + kvm_path = params.get("kvm_path") \ + if params.get("kvm_path") is not None else "/dev/kvm" + cpuinfo_path = params.get("cpuinfo_path") \ + 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" + model = params.get("model") + ssh = _ssh_fragment(params, env) + quota = _quota_fragment(env) + timeouts = _timeout_fragment(params, env) return ChvConfig( binary=binary, kernel=kernel, rootfs=rootfs, cpus=cpus, - memory=memory, 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, - 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, - 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, - single_tenant_project=single_tenant_project, - snapshot_max_count=snapshot_max_count, - snapshot_max_total_gb=snapshot_max_total_gb, - snapshot_retention_days=snapshot_retention_days, - snapshot_dir=snapshot_dir, - allow_remote_snapshot=allow_remote_snapshot) + memory=memory, state_dir=state_dir, ssh_key=ssh["ssh_key"], + ssh_user=ssh["ssh_user"], ssh_port=ssh["ssh_port"], + guest_host_py=ssh["guest_host_py"], cmdline=ssh["cmdline"], + net=ssh["net"], net_json=ssh["net_json"], + kvm_path=kvm_path, cpuinfo_path=cpuinfo_path, + ch_remote=ch_remote, model=model, + max_vms=quota["max_vms"], disk_mode=quota["disk_mode"], + disk_gb=quota["disk_gb"], ssh_host=ssh["ssh_host"], + ssh_forward=ssh["ssh_forward"], + reuse_experiment=quota["reuse_experiment"], + egress_proxy=quota["egress_proxy"], + allow_direct_egress=quota["allow_direct_egress"], + host_key_mode=ssh["host_key_mode"], + task_timeout=timeouts["task_timeout"], + snapshot_url=timeouts["snapshot_url"], + warm_pool_size=timeouts["warm_pool_size"], + api_timeout=timeouts["api_timeout"], + ready_timeout=timeouts["ready_timeout"], + ready_interval=timeouts["ready_interval"], + single_tenant_project=quota["single_tenant_project"], + snapshot_max_count=quota["snapshot_max_count"], + snapshot_max_total_gb=quota["snapshot_max_total_gb"], + snapshot_retention_days=quota["snapshot_retention_days"], + snapshot_dir=quota["snapshot_dir"], + allow_remote_snapshot=quota["allow_remote_snapshot"]) diff --git a/cloud/chv/disk.py b/cloud/chv/disk.py index a650971..47a8450 100644 --- a/cloud/chv/disk.py +++ b/cloud/chv/disk.py @@ -33,7 +33,7 @@ def _dir_total_bytes(path): return total -def check_disk_quota(rootfs, disk_gb, state_dir=None): +def check_disk_quota(rootfs: str, disk_gb: float, state_dir=None) -> None: """磁盘配额门(feat-chv-fix 口径): - disk_gb 未设时不检查; @@ -73,7 +73,7 @@ def check_disk_quota(rootfs, disk_gb, state_dir=None): % (free, size)) -def provision_task_disk(state_dir, vm_name, rootfs, disk_mode): +def provision_task_disk(state_dir: str, vm_name: str, rootfs: str, disk_mode: str) -> str: """每任务磁盘路径:copy 模式复制到 state_dir//disk.raw;reuse 模式直接返回基础镜像路径。 @@ -112,12 +112,12 @@ def provision_task_disk(state_dir, vm_name, rootfs, disk_mode): return task_disk -def task_disk_path(state_dir, vm_name): +def task_disk_path(state_dir: str, vm_name: str) -> str: """该 VM 的任务磁盘路径:state_dir//disk.raw。""" return os.path.join(state_dir, vm_name, "disk.raw") -def delete_task_disk(state_dir, vm_name): +def delete_task_disk(state_dir: str, vm_name: str) -> bool: """删任务磁盘(feat-chv-secfix,H3):stop/recycle/ 清理路径都调它。返回 True 表示确实删掉了一个文件, False 表示本来就没有;删失败抛 OSError(调用方记审计, @@ -188,7 +188,7 @@ def prune_snapshots(state_dir, vm_name, retention_days, return kept, pruned -def check_snapshot_limits(state_dir, vm_name, max_count, +def check_snapshot_limits(state_dir: str, vm_name: str, max_count: int, max_total_gb, retention_days): """快照上限预检(snapshot_vm 调 ch-remote 之前调): 先按保留期剪枝,再查数量(已有数 + 本次 > 上限即 @@ -218,7 +218,7 @@ def check_snapshot_limits(state_dir, vm_name, max_count, return kept -def record_snapshot(state_dir, vm_name, snapshot_url, max_total_gb): +def record_snapshot(state_dir: str, vm_name: str, snapshot_url: str, max_total_gb: float) -> dict: """快照成功后登记;file:// 目标记实际大小。登记后若 总量超上限,按最旧优先删登记外的旧快照文件(总量硬 帽,删动作由调用方审计——本函数返回被删的旧条目)。 diff --git a/cloud/chv/executor.py b/cloud/chv/executor.py index 37f87e4..21f3198 100644 --- a/cloud/chv/executor.py +++ b/cloud/chv/executor.py @@ -61,6 +61,13 @@ from cloud.chv.ssh_transport import write_known_hosts _LOGGER = logging.getLogger("cloud.cloud_hypervisor") +def _default_popen_launcher(cmd): + """默认 VMM 启动器:独立会话子进程,输出丢弃(纯移动)。""" + return subprocess.Popen( + cmd, stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, start_new_session=True) + + class CloudHypervisorExecutor(Executor): """Cloud Hypervisor 执行器 v1(命名:cloud-hypervisor,绝不用 cloud/ch 缩写——cloud 已是通用 HTTP runner 占位,撞名会让 @@ -267,20 +274,12 @@ class CloudHypervisorExecutor(Executor): "ready_timeout": ready_timeout, "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", - "guest_host_py", "cmdline", "net", "net_json", - "kvm_path", "cpuinfo_path", "ch_remote", "model", - "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", "ssh_host", - "ssh_forward", "reuse_experiment", - "single_tenant_project", "snapshot_max_count", - "snapshot_max_total_gb", "snapshot_retention_days", - "snapshot_dir", "allow_remote_snapshot"): - setattr(self, _field, getattr(_cfg, _field)) + self._copy_config_fields(_cfg) + self._init_runtime(runner, launcher, api_client, sleeper) + + def _init_runtime(self, runner, launcher, api_client, sleeper) -> None: + """运行时注入:seed 投递集合、API 客户端、睡眠器、 + runner/launcher 默认值(纯移动)。""" # feat-chv-secfix(H2):本实例已 seed 投递密钥的 VM # 名 → 已投递密钥名集合。_run_guest 只对集合内的密钥 # 名 source 客内文件,未投递的密钥名一律拒绝(防止 @@ -301,13 +300,26 @@ class CloudHypervisorExecutor(Executor): if launcher is not None: self.launcher = launcher else: - def _popen(cmd): - return subprocess.Popen( - cmd, stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, start_new_session=True) - self.launcher = _popen + self.launcher = _default_popen_launcher self._vms = {} + def _copy_config_fields(self, _cfg) -> None: + """把 build_chv_config 的字段抄到 self(纯移动,字段集不变)。""" + for _field in ( + "binary", "kernel", "rootfs", "cpus", "memory", + "state_dir", "ssh_key", "ssh_user", "ssh_port", + "guest_host_py", "cmdline", "net", "net_json", + "kvm_path", "cpuinfo_path", "ch_remote", "model", + "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", "ssh_host", + "ssh_forward", "reuse_experiment", + "single_tenant_project", "snapshot_max_count", + "snapshot_max_total_gb", "snapshot_retention_days", + "snapshot_dir", "allow_remote_snapshot"): + setattr(self, _field, getattr(_cfg, _field)) + # -- 名字 / 路径 --------------------------------------------------- def _host_name(self, task): return re.sub(r"[^A-Za-z0-9_.-]", "-", @@ -338,7 +350,7 @@ class CloudHypervisorExecutor(Executor): return os.path.isfile(path) and os.access(path, os.X_OK) return shutil.which(path) is not None - def _preflight_config_checks(self, missing): + def _preflight_config_checks(self, missing: list) -> None: """preflight 的配置校验块(feat-chv-fix,P2-1 小提取: 仅把 reuse 拦截与 CPU/内存校验从 preflight() 里抽出, 语义逐行不变;其余 preflight 检查项与 config.py 的 @@ -378,24 +390,70 @@ class CloudHypervisorExecutor(Executor): except ValueError: missing.append(("config_invalid", "CHV_MEM 格式非法:%r" % (self.memory,))) + self._preflight_numeric_checks(missing) + def _preflight_numeric_checks(self, missing: list) -> None: + """preflight 数值面:端口/配额/超时/快照统一校验,纯移动。 + 构造时已校验正常值;这里兜底属性事后被改写的情形。""" + # feat-chv-arch95(配置统一校验):构造时 build_chv_config + # 已校验过这些字段的正常值;这里再以 config_invalid + # 统一兜底属性被事后改写/注入非法值的情形,preflight + # 是生产路径最后一道统一校验,不再等到 build_vm_config + # 或 wake 时才在不同位置以不同文案报错。 + try: + _port = int(getattr(self, "ssh_port", 2222)) + if not 1 <= _port <= 65535: + raise ValueError("ssh_port") + except (TypeError, ValueError): + missing.append(("config_invalid", + "CHV_SSH_PORT 必须在 1-65535 之间")) + try: + _max_vms = int(getattr(self, "max_vms", 4)) + if _max_vms < 0: + raise ValueError("max_vms") + except (TypeError, ValueError): + missing.append(("config_invalid", + "CHV_MAX_VMS 必须是非负整数")) + try: + _disk_gb = float(getattr(self, "disk_gb", 50.0)) + if _disk_gb <= 0: + raise ValueError("disk_gb") + except (TypeError, ValueError): + missing.append(("config_invalid", + "CHV_DISK_GB 必须是大于 0 的数字")) + try: + if float(getattr(self, "api_timeout", 30)) <= 0 \ + or float(getattr(self, "ready_timeout", 120)) <= 0 \ + or float(getattr(self, "ready_interval", 5)) <= 0: + raise ValueError("timeout") + except (TypeError, ValueError): + missing.append(("config_invalid", + "CHV_*_TIMEOUT/INTERVAL 必须是正数")) + try: + if int(getattr(self, "warm_pool_size", 0)) < 0: + raise ValueError("warm_pool_size") + except (TypeError, ValueError): + missing.append(("config_invalid", + "CHV_WARM_POOL_SIZE 必须是非负整数")) + self._preflight_snapshot_checks(missing) + def _preflight_snapshot_checks(self, missing: list) -> None: + """preflight 快照/host-key 面:统一校验,纯移动。""" + try: + if int(getattr(self, "snapshot_max_count", 10)) < 0 \ + or float(getattr(self, "snapshot_max_total_gb", 20.0)) < 0 \ + or float(getattr(self, "snapshot_retention_days", 30.0)) < 0: + raise ValueError("snapshot") + except (TypeError, ValueError): + missing.append(("config_invalid", + "CHV_SNAPSHOT_* 不能为负数")) + if getattr(self, "host_key_mode", "provisioned") \ + not in ("provisioned", "keyscan"): + missing.append(("config_invalid", + "CHV_HOST_KEY_MODE 只能是 provisioned 或 keyscan")) - def preflight(self): - """预检(缺一不可则 fail-fast):收集全部缺失项一次报出 - 具体缺哪一项,绝不笼统报错;失败为 ChvFailure(带 kind, - 供审计与失败分类用),绝不回落 Docker/Local。 - 检查项与 kind: - - binary_missing:cloud-hypervisor 二进制不存在/不可执行 - - binary_missing:ch-remote 二进制不存在/不可执行 - - kvm_missing:/dev/kvm 不存在 - - kvm_missing:/dev/kvm 不可读/不可写(分开报) - - cpu_flag_missing:/proc/cpuinfo 无 vmx/svm 标志 - - kernel_missing:CHV_KERNEL 未设/文件不存在/不可读 - - image_missing:CHV_ROOTFS 未设/文件不存在/不可读 - - ssh_key_missing:CHV_SSH_KEY 未设/文件不存在/不可读 - - egress_not_configured:CHV_EGRESS_PROXY 未设且未显式 - 允许直连(默认 fail-closed,VM 出口必须过白名单代理) - """ - missing = [] # (kind, 文本) 列表,全部收集完再一次抛 + + + def _preflight_host_checks(self, missing: list) -> None: + """preflight 主机面:二进制/KVM/CPU 标志,纯移动。""" if not self._binary_ok(self.binary): missing.append(("binary_missing", "cloud-hypervisor 二进制缺失或不可执行:%s" @@ -432,6 +490,9 @@ class CloudHypervisorExecutor(Executor): missing.append(("cpu_flag_missing", "CPU 无虚拟化标志(无 vmx/svm):%s" % self.cpuinfo_path)) + + def _preflight_file_checks(self, missing: list) -> None: + """preflight 文件面:内核/镜像/SSH key,纯移动。""" if not self.kernel: missing.append(("kernel_missing", "CHV_KERNEL 未设")) elif not os.path.isfile(self.kernel): @@ -456,6 +517,9 @@ class CloudHypervisorExecutor(Executor): elif not os.access(self.ssh_key, os.R_OK): missing.append(("ssh_key_missing", "SSH key 文件不可读:%s" % self.ssh_key)) + + def _preflight_net_checks(self, missing: list) -> None: + """preflight 网络/目录面:egress/安全目录/net,纯移动。""" if not self.egress_proxy and not self.allow_direct_egress: missing.append(("egress_not_configured", "CHV_EGRESS_PROXY 未设且未显式允许直连:" @@ -494,6 +558,26 @@ class CloudHypervisorExecutor(Executor): validate_net_entries(_entries, []) except ChvFailure as exc: missing.append((exc.kind, str(exc))) + def preflight(self): + """预检(缺一不可则 fail-fast):收集全部缺失项一次报出 + 具体缺哪一项,绝不笼统报错;失败为 ChvFailure(带 kind, + 供审计与失败分类用),绝不回落 Docker/Local。 + 检查项与 kind: + - binary_missing:cloud-hypervisor 二进制不存在/不可执行 + - binary_missing:ch-remote 二进制不存在/不可执行 + - kvm_missing:/dev/kvm 不存在 + - kvm_missing:/dev/kvm 不可读/不可写(分开报) + - cpu_flag_missing:/proc/cpuinfo 无 vmx/svm 标志 + - kernel_missing:CHV_KERNEL 未设/文件不存在/不可读 + - image_missing:CHV_ROOTFS 未设/文件不存在/不可读 + - ssh_key_missing:CHV_SSH_KEY 未设/文件不存在/不可读 + - egress_not_configured:CHV_EGRESS_PROXY 未设且未显式 + 允许直连(默认 fail-closed,VM 出口必须过白名单代理) + """ + missing = [] # (kind, 文本) 列表,全部收集完再一次抛 + self._preflight_host_checks(missing) + self._preflight_file_checks(missing) + self._preflight_net_checks(missing) self._preflight_config_checks(missing) if missing: kind = missing[0][0] @@ -612,7 +696,7 @@ class CloudHypervisorExecutor(Executor): 旧 key 对)。成功后审计 host_key_generated(只含 VM 名与密钥名,不含任何 key 值)。""" from cloud.chv.hostkeys import ( - dirty_marker_path, generate_host_keypair, host_key_paths, + dirty_marker_path, generate_host_keypair, known_hosts_line, write_seed) from cloud.chv.ssh_transport import write_known_hosts vm = self.vm_name(task) @@ -1077,25 +1161,9 @@ class CloudHypervisorExecutor(Executor): "VMM api-socket 组/其它可写(%o),拒绝启动:%s" % (mode, socket_path)) - def wake_vm(self, task, extra_env=None): - """主唤醒路径(API 模式,dispatch/run/goal 的前置步骤): - preflight → 容量门 → 磁盘配额门 → 每任务磁盘供给 → - launcher 只起 VMM(--api-socket)→ 等 vmm.ping 通 → - PUT vm.create(VmConfig)→ PUT vm.boot → 等 guest - SSH 就绪。任一步失败都关 VM(stop_vm)再抛 - ChvFailure,绝不带着半醒的 VM 去跑任务。 - 成功返回 {vm, api_socket, state: "running", ready: True, - mode: "cold"},并审计 vm_create/vm_boot/vm_ready。 - 单任务单 VM:同一任务重复 wake 时先关旧 VM 再起新 - 的,绝不两个 VM 共用一个任务名。 - feat-chv-secfix:extra_env(_run_with_vm 透传)里的 - 密钥子集在 provisioned 模式下写进 seed 客内文件, - keyscan 模式下有密钥子集即 fail-closed;起 VM 前的 - preflight/容量/配额失败也发 vm_failed 审计(L3); - net 隔离在起 VMM 前校验(H7),成功后登记 net.json。""" - if not isinstance(task, dict) or task.get("id") is None: - raise ValueError("任务缺少 id") - name = self.vm_name(task) + def _wake_prechecks(self, task: dict, name: str, extra_env) -> tuple: + """wake 前置:preflight/容量/配额/net/密钥模式,纯移动。 + 失败时发 vm_failed 审计并原样抛 ChvFailure。""" try: self.preflight() # feat-chv-fix:SSH 目标与 reuse 放行在启动前就定性。 @@ -1129,6 +1197,70 @@ class CloudHypervisorExecutor(Executor): "vm=%s wake 前置失败 kind=%s" % (name, exc.kind), exc.kind) raise + return secrets, net_entries + + def _provision_seed_disk(self, task: dict, name: str, secrets) -> str: + """provisioned 模式的 seed 盘供给:生成→制盘→挂盘, + 任一步失败即抛(调用方 stop_vm 清理),纯移动。""" + seed_disk = None + if getattr(self, "host_key_mode", "provisioned") == "provisioned": + # 强方案:启动前预置每 VM host key 并钉死, + # 再把 seed 目录做成 CIDATA seed 磁盘并挂进 + # VmConfig(P0-1 闭环:生成→制盘→挂盘三步缺 + # 一即 fail-fast,绝不回落 keyscan、绝不只 + # 生成不注入)。 + # feat-chv-secfix(H2/H5):secret_env 随 seed + # 写进客内密钥文件;seed.iso 建好(0600,见 + # build_seed_disk)后立即删宿主侧私钥——此后 + # 宿主上只剩公钥 + known_hosts,私钥只活在客 + # 机与 seed.iso(随 VM 销毁而删)里。删私钥 + # 失败即 fail-fast(VM 尚未启动,此时拒绝最 + # 安全,except 里 stop_vm 会清掉 seed)。 + from cloud.chv.hostkeys import ( + build_seed_disk, delete_host_private_key, + seed_disk_path) + provisioned = self.provision_host_key( + task, secret_env=secrets or None) + seed_disk = seed_disk_path(self.state_dir, name) + build_seed_disk( + provisioned["seed_dir"], seed_disk, + runner=lambda cmd: self._call_runner(cmd)) + delete_host_private_key(self.state_dir, name) + if secrets: + self._guest_env_delivered[name] = set(secrets) + else: + self._guest_env_delivered.pop(name, None) + return seed_disk + + def _boot_vm(self, task, name, task_disk, seed_disk, + net_entries) -> None: + """VMM 启动→vm.create→vm.boot→SSH 就绪,纯移动。""" + socket_path = self.api_socket(task) + self._launch_api_only(task) + self.wait_api_ready(task) + self._check_api_socket_permissions(socket_path) + status, _ = self._api_call( + socket_path, "PUT", "/vm.create", + self.build_vm_config(task_disk, seed_disk=seed_disk)) + if status not in (200, 204): + raise ChvFailure( + "vm_create_failed", + "vm.create 失败:status=%s" % status) + self._emit("vm_create", task, + "vm=%s disk=%s" % (name, task_disk)) + if net_entries: + self._record_net_entries(name, net_entries) + status, _ = self._api_call(socket_path, "PUT", "/vm.boot") + if status not in (200, 204): + raise ChvFailure( + "vm_boot_failed", + "vm.boot 失败:status=%s" % status) + self._emit("vm_boot", task, "vm=%s" % name) + self.wait_guest_ready(task) + + def _wake_prepare(self, task: dict, name: str) -> tuple: + """wake 准备:reuse 审计、旧 VM 清理、任务盘供给、 + 直连 egress 审计,纯移动。返回 (task_disk, socket_path)。""" if getattr(self, "disk_mode", "copy") == "reuse": self._emit("disk_reuse_experiment_allowed", task, "vm=%s CHV_REUSE_SINGLE_TENANT_EXPERIMENT=1 " @@ -1144,62 +1276,35 @@ class CloudHypervisorExecutor(Executor): 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) + return task_disk, socket_path + + + def wake_vm(self, task, extra_env=None): + """主唤醒路径(API 模式,dispatch/run/goal 的前置步骤): + preflight → 容量门 → 磁盘配额门 → 每任务磁盘供给 → + launcher 只起 VMM(--api-socket)→ 等 vmm.ping 通 → + PUT vm.create(VmConfig)→ PUT vm.boot → 等 guest + SSH 就绪。任一步失败都关 VM(stop_vm)再抛 + ChvFailure,绝不带着半醒的 VM 去跑任务。 + 成功返回 {vm, api_socket, state: "running", ready: True, + mode: "cold"},并审计 vm_create/vm_boot/vm_ready。 + 单任务单 VM:同一任务重复 wake 时先关旧 VM 再起新 + 的,绝不两个 VM 共用一个任务名。 + feat-chv-secfix:extra_env(_run_with_vm 透传)里的 + 密钥子集在 provisioned 模式下写进 seed 客内文件, + keyscan 模式下有密钥子集即 fail-closed;起 VM 前的 + preflight/容量/配额失败也发 vm_failed 审计(L3); + net 隔离在起 VMM 前校验(H7),成功后登记 net.json。""" + if not isinstance(task, dict) or task.get("id") is None: + raise ValueError("任务缺少 id") + name = self.vm_name(task) + secrets, net_entries = self._wake_prechecks(task, name, extra_env) + task_disk, socket_path = self._wake_prepare(task, name) try: - seed_disk = None - if getattr(self, "host_key_mode", "provisioned") == "provisioned": - # 强方案:启动前预置每 VM host key 并钉死, - # 再把 seed 目录做成 CIDATA seed 磁盘并挂进 - # VmConfig(P0-1 闭环:生成→制盘→挂盘三步缺 - # 一即 fail-fast,绝不回落 keyscan、绝不只 - # 生成不注入)。 - # feat-chv-secfix(H2/H5):secret_env 随 seed - # 写进客内密钥文件;seed.iso 建好(0600,见 - # build_seed_disk)后立即删宿主侧私钥——此后 - # 宿主上只剩公钥 + known_hosts,私钥只活在客 - # 机与 seed.iso(随 VM 销毁而删)里。删私钥 - # 失败即 fail-fast(VM 尚未启动,此时拒绝最 - # 安全,except 里 stop_vm 会清掉 seed)。 - from cloud.chv.hostkeys import ( - build_seed_disk, delete_host_private_key, - seed_disk_path) - provisioned = self.provision_host_key( - task, secret_env=secrets or None) - seed_disk = seed_disk_path(self.state_dir, name) - build_seed_disk( - provisioned["seed_dir"], seed_disk, - runner=lambda cmd: self._call_runner(cmd)) - delete_host_private_key(self.state_dir, name) - if secrets: - self._guest_env_delivered[name] = set(secrets) - else: - self._guest_env_delivered.pop(name, None) - self._launch_api_only(task) - self.wait_api_ready(task) - self._check_api_socket_permissions(socket_path) - status, _ = self._api_call( - socket_path, "PUT", "/vm.create", - self.build_vm_config(task_disk, seed_disk=seed_disk)) - if status not in (200, 204): - raise ChvFailure( - "vm_create_failed", - "vm.create 失败:status=%s" % status) - self._emit("vm_create", task, - "vm=%s disk=%s" % (name, task_disk)) - if net_entries: - self._record_net_entries(name, net_entries) - status, _ = self._api_call(socket_path, "PUT", "/vm.boot") - if status not in (200, 204): - raise ChvFailure( - "vm_boot_failed", - "vm.boot 失败:status=%s" % status) - self._emit("vm_boot", task, "vm=%s" % name) - self.wait_guest_ready(task) + seed_disk = self._provision_seed_disk(task, name, secrets) + self._boot_vm(task, name, task_disk, seed_disk, net_entries) except Exception: try: self.stop_vm(task) @@ -1779,8 +1884,12 @@ class CloudHypervisorExecutor(Executor): def _run_with_vm(self, task, inner_args, goal=False, extra_env=None, **kwargs): - """每任务 VM 生命周期:wake_vm(API 模式 + SSH 就绪) - → 跑客内命令 → finally shutdown_vm。wake 失败时不跑 + """每任务 VM 生命周期(冷启动正式路径,feat-chv-arch95 + 定案):wake_vm(API 模式 + SSH 就绪)→ 跑客内命令 → + finally shutdown_vm。本方法绝不查 WarmPool:预热池是 + 实验性的(见 pool.py 头定案与 ADR-5),dispatch 永远 + 冷启动,warm_pool_wired_to_dispatch 恒 False。 + wake 失败时不跑 任务、直接抛分类失败。本方法不做内部重试:重试是 worker 的 max_retries 语义,执行器内部重试会把一次 失败放大成多次计费与多次 VM 启动。 diff --git a/cloud/chv/hardening.py b/cloud/chv/hardening.py index 6284480..015fc10 100644 --- a/cloud/chv/hardening.py +++ b/cloud/chv/hardening.py @@ -20,7 +20,7 @@ from cloud.chv.errors import ChvFailure _NET_FIELDS = frozenset({"tap", "mac", "ip", "mask"}) -def ensure_secure_state_dir(state_dir): +def ensure_secure_state_dir(state_dir: str) -> None: """建 state_dir 并保证 0700(H6):目录不存在即按 0700 建;已存在但组/其它位可访时 chmod 0700;失败即 ChvFailure(state_dir_insecure),绝不带着 /tmp 式 @@ -48,7 +48,7 @@ def ensure_secure_state_dir(state_dir): return state_dir -def socket_is_live(socket_path): +def socket_is_live(socket_path: str) -> bool: """VMM api-socket 是否真有监听者(M7):能连通才算活。 普通文件、空文件、已死 VMM 的残留 socket 一律算死—— 容量计数不再被残留文件占满。连接失败的任何异常都算 @@ -69,7 +69,7 @@ def socket_is_live(socket_path): return False -def parse_net_entries(net_json): +def parse_net_entries(net_json: str) -> list: """解析 CHV_NET_JSON 为条目 list(非法 JSON / 非 list/dict 即 net_config_invalid)。""" if not net_json: @@ -86,7 +86,7 @@ def parse_net_entries(net_json): return entries -def validate_net_entries(entries, existing_entries): +def validate_net_entries(entries: list, existing_entries: list) -> list: """net 条目隔离校验(H7 + L4): - 字段白名单:未知字段即 net_config_invalid; @@ -139,7 +139,7 @@ def validate_net_entries(entries, existing_entries): return normalized -def validate_egress_proxy(egress_proxy): +def validate_egress_proxy(egress_proxy: str) -> None: """egress 代理地址校验(H1 代码门的一部分):须为 http/https URL 且带主机名;否则 net_config_invalid… 实际 kind 用 config_invalid(preflight 的配置类失败)。 @@ -167,7 +167,7 @@ def _is_within(directory, path): return path == directory or path.startswith(directory + os.sep) -def validate_snapshot_url(snapshot_url, state_dir, snapshot_dir=None, +def validate_snapshot_url(snapshot_url: str, state_dir: str, snapshot_dir=None, allow_remote=False): """快照 URL 白名单(L2): diff --git a/cloud/chv/hostkeys.py b/cloud/chv/hostkeys.py index 34898f0..e7eaac2 100644 --- a/cloud/chv/hostkeys.py +++ b/cloud/chv/hostkeys.py @@ -40,7 +40,7 @@ import shlex GUEST_ENV_PATH = "/etc/chv/guest-env" -def host_key_paths(state_dir, vm_name): +def host_key_paths(state_dir: str, vm_name: str) -> tuple: """该 VM 的 host key 文件路径(私钥/公钥/known_hosts)。""" base = os.path.join(state_dir, vm_name) return { @@ -51,7 +51,7 @@ def host_key_paths(state_dir, vm_name): } -def dirty_marker_path(state_dir, vm_name): +def dirty_marker_path(state_dir: str, vm_name: str) -> str: """清理失败脏标记路径(feat-chv-secfix,M2):stop_vm 清理 seed/私钥/磁盘任一步失败时写下该标记;标记存在 时同名 VM 拒绝启动(清理失败即拒起),人工清理并删 @@ -59,7 +59,7 @@ def dirty_marker_path(state_dir, vm_name): return os.path.join(state_dir, vm_name, ".cleanup-dirty") -def delete_host_private_key(state_dir, vm_name): +def delete_host_private_key(state_dir: str, vm_name: str) -> None: """删宿主侧 host 私钥文件(feat-chv-secfix,H5):seed 盘建好后宿主私钥已无存在必要(客机已有私钥、宿主只需 公钥 + known_hosts 做校验)。删失败即 OSError,由调用 @@ -83,12 +83,12 @@ def known_hosts_line(port, public_key_text, host="127.0.0.1"): return "[%s]:%s %s %s" % (host, port, parts[0], parts[1]) -def seed_disk_path(state_dir, vm_name): +def seed_disk_path(state_dir: str, vm_name: str) -> str: """该 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): +def build_seed_disk(seed_dir: str, seed_disk: str, runner=None) -> None: """把 seed 目录做成 CIDATA 卷标 ISO(feat-chv-fix,P0-1 的 注入载体):genisoimage -volid CIDATA -joliet -rock。 diff --git a/cloud/chv/pool.py b/cloud/chv/pool.py index 4b73b94..6630e64 100644 --- a/cloud/chv/pool.py +++ b/cloud/chv/pool.py @@ -1,4 +1,17 @@ -"""快照预热池 WarmPool(含 state_dir registry 持久化与孤儿清理)。 +"""快照预热池 WarmPool(实验性,非生产路径,feat-chv-arch95 定案)。 + +定案(架构决策,见 docs/chv-architecture.md ADR-5):WarmPool +不再宣称为生产快速启动特性。原因:预热 VM 的磁盘是干净快照 +盘,不是任务盘;dispatch 前查池若要真正接线,必须在派任务 +前给已暂停 VM 重新供给任务盘(热换盘/二盘挂载/host.py 磁盘 +语义三者都未实现),把预热盘直接当任务盘用会击穿“任务盘 +隔离”不变量。acquire 内 resume 只是实验性接口,worker +dispatch 永远走冷启动(wake_vm),_run_with_vm 绝不查池。 +pool 保留的唯一生产价值是 reconcile_orphans()(上一世残留 +清理),worker 启动/关闭仍调它。warm_pool_wired_to_dispatch +恒为 False,任何把它改成 True 的改动都必须先实现任务盘 +重供给并补不变量测试,否则不得合入。 + 隔离红线(与 #216 一致,绝不妥协):预热条目是一次性的 干净快照 VM,acquire 消耗掉它,release 一律销毁,绝不把 @@ -40,7 +53,10 @@ _REGISTRY_FILENAME = "warm-pool.json" class WarmPool: - """快照预热池(快速启动设计,接口与状态机在代码里)。 + """快照预热池(实验性,见模块头定案)。 + + wired_to_dispatch = False:生产 dispatch 不查池。 + 目标:冷启动(preflight + 起 VMM + create + boot + SSH 就绪)是慢路径;预热池提前把干净 VM 起好并暂停,任务 @@ -55,6 +71,8 @@ class WarmPool: prewarm 返回 [](无基准快照可预热,不是静默造快照)。 """ + wired_to_dispatch = False + def __init__(self, executor, size=None, snapshot_url=None, registry_path=None): self.executor = executor @@ -150,7 +168,7 @@ class WarmPool: _LOGGER.warning("chv_pool_registry_save_failed path=%s", self.registry_path) - def reconcile_orphans(self): + def reconcile_orphans(self) -> list: """清理上一世残留条目:对每个 orphan/assigned 条目 尽力 shutdown_vm(清 socket 与残留登记),条目清掉 并落盘,返回已清理的 VM 名列表。worker 启动/清理时 @@ -177,11 +195,11 @@ class WarmPool: return cleaned # -- 池操作 ---------------------------------------------------------- - def warm_count(self): + def warm_count(self) -> int: return sum(1 for e in self._entries.values() if e["state"] == "warm") - def orphan_count(self): + def orphan_count(self) -> int: """上一世残留条目数(待 reconcile,不可分配)。""" return sum(1 for e in self._entries.values() if e["state"] in ("orphan", "assigned")) diff --git a/cloud/chv/ssh_transport.py b/cloud/chv/ssh_transport.py index 8ae8f0e..5d64008 100644 --- a/cloud/chv/ssh_transport.py +++ b/cloud/chv/ssh_transport.py @@ -52,7 +52,7 @@ import shlex _CHV_SECRET_ENV_NAMES = frozenset({"SUB2API_API_KEY"}) -def split_secret_env(extra_env): +def split_secret_env(extra_env: dict) -> tuple: """把 extra_env 拆成(密钥子集,非密钥子集)。""" secrets, plain = {}, {} for key, value in dict(extra_env or {}).items(): @@ -103,7 +103,7 @@ def remote_command(inner_cmd, egress_proxy=None, extra_env=None, return tokens -def known_hosts_path(state_dir, vm_name): +def known_hosts_path(state_dir: str, vm_name: str) -> str: """该 VM 的 known_hosts 文件路径:state_dir//known_hosts。""" return os.path.join(state_dir, vm_name, "known_hosts") @@ -163,7 +163,7 @@ def parse_keyscan_output(text, host="127.0.0.1"): return kept -def write_known_hosts(path, key_lines): +def write_known_hosts(path: str, key_lines: list) -> None: """原子写 known_hosts:先写同目录临时文件再 rename, 文件权限 0600(只含公钥行,仍按敏感文件对待)。""" directory = os.path.dirname(path) diff --git a/cloud/tests/test_chv_arch95.py b/cloud/tests/test_chv_arch95.py new file mode 100644 index 0000000..50f8867 --- /dev/null +++ b/cloud/tests/test_chv_arch95.py @@ -0,0 +1,258 @@ +"""CHV 架构 9.5 提升轨测试(feat-chv-arch95,堆叠于 #246)。 + +只用标准库与 mock,绝不启动真实 VM。本文件是“严格测试替身” +的载体:替身不再无条件放行——fake API 校验 VmConfig 内容 +(任务盘 + seed 盘、net 条目),fake SSH 校验 known_hosts +钉死与 StrictHostKeyChecking=yes,mock 再也不能把被测逻辑 +本身也 mock 掉。测试里只用 FAKE 占位密钥,绝无真实密钥。 +""" + +import os +import sys +import tempfile +import unittest +from pathlib import Path + +try: + from cloud.chv.pool import WarmPool + from cloud.sandbox import ( + ChvFailure, + CloudHypervisorExecutor, + DockerExecutor, + parse_goal_result, + ) +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.chv.pool import WarmPool + from cloud.sandbox import ( + ChvFailure, + CloudHypervisorExecutor, + DockerExecutor, + parse_goal_result, + ) + + +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 Arch95Case(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 = [] + self.api_payloads = [] + + def tearDown(self): + self.tmp.cleanup() + + def _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, "", "") + 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 _strict_api(self, socket_path, method, path, payload=None): + """严格替身:vm.create 必须带任务盘 + seed 盘、net 无重复。""" + if path == "/vm.create": + self.api_payloads.append(payload) + disks = [d.get("path") for d in payload.get("disks", [])] + self.assertEqual(len(disks), 2, "VmConfig 必须恰好两盘") + self.assertTrue(disks[0].endswith("disk.raw"), + "第一盘必须是任务盘") + self.assertTrue(disks[1].endswith("seed.iso"), + "第二盘必须是 seed 盘,缺 seed 即假保证") + self.assertTrue(payload.get("kernel", {}).get("path")) + nets = payload.get("net", []) + taps = [n.get("tap") for n in nets] + self.assertEqual(len(taps), len(set(taps)), "net tap 重复") + return (200, {}) + + def make_exec(self, api=None, runner=None, env_extra=None): + env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888", + "CHV_SSH_FORWARD": "1"} + if env_extra: + env.update(env_extra) + ex = CloudHypervisorExecutor( + 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._runner, + launcher=lambda cmd: FakeProc(alive=True), + api_client=api or self._strict_api, + sleeper=lambda s: None, env=env) + 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 + + +class WarmPoolDecisionTest(Arch95Case): + def test_pool_is_experimental_and_dispatch_is_cold_only(self): + self.assertFalse(WarmPool.wired_to_dispatch) + ex = self.make_exec() + pool = WarmPool(ex) + # 手工放一个 warm 条目:dispatch 必须不消耗它(冷启动路径)。 + pool._entries["warm-1"] = {"state": "warm", + "task": {"id": "warm-1"}, + "vm": "chv-warm-1", + "snapshot_url": None} + ex.wake_vm({"id": "t1"}) + self.assertEqual(pool._entries["warm-1"]["state"], "warm") + self.assertEqual(pool.warm_count(), 1) + + +class StrictVmConfigTest(Arch95Case): + def test_wake_vm_config_has_task_disk_and_seed_disk(self): + ex = self.make_exec() + out = ex.wake_vm({"id": "t1"}) + self.assertEqual(out["state"], "running") + self.assertEqual(len(self.api_payloads), 1) + # known_hosts 钉死:文件存在且含预置公钥,SSH 命令带钉死选项。 + kh = ex.known_hosts_path({"id": "t1"}) + with open(kh) as fh: + content = fh.read() + self.assertIn("AAAAfake1", content) + cmd = ex.build_ssh_command({"id": "t1"}, ["true"]) + self.assertIn("StrictHostKeyChecking=yes", " ".join(cmd)) + self.assertIn(kh, " ".join(cmd)) + + def test_strict_ssh_runner_rejects_missing_known_hosts(self): + def strict_runner(cmd, timeout=None): + if cmd and cmd[0] == "ssh": + self.assertIn("StrictHostKeyChecking=yes", cmd) + found = False + for i, a in enumerate(cmd): + if a.startswith("UserKnownHostsFile="): + path = a.split("=", 1)[1] + found = True + elif a == "UserKnownHostsFile": + path = cmd[i + 1] + found = True + else: + continue + with open(path) as fh: + self.assertTrue(fh.read().strip(), + "known_hosts 为空即假钉死") + self.assertTrue(found, "SSH 命令缺 known_hosts 钉死选项") + return self._runner(cmd, timeout=timeout) + ex = self.make_exec(runner=strict_runner) + ex.wake_vm({"id": "t1"}) + # 严格 runner 在 _run_guest 里再跑一次 status 风格调用: + # 这里直接验证 build 出的命令能过严格 runner 的检查。 + cmd = ex.build_ssh_command({"id": "t1"}, ["true"]) + strict_runner(cmd) + + +class ConfigUnifiedTest(Arch95Case): + def test_preflight_unified_config_invalid(self): + cases = [ + ("ssh_port", 0), ("ssh_port", 70000), + ("max_vms", -1), ("disk_gb", 0), + ("api_timeout", 0), ("ready_timeout", -1), + ("ready_interval", 0), ("warm_pool_size", -1), + ("snapshot_max_count", -1), + ("snapshot_max_total_gb", -1), + ("snapshot_retention_days", -1), + ("host_key_mode", "bogus"), + ] + for field, bad in cases: + with self.subTest(field=field, bad=bad): + ex = self.make_exec() + setattr(ex, field, bad) + with self.assertRaises(ChvFailure) as ctx: + ex.preflight() + self.assertEqual(ctx.exception.kind, "config_invalid") + + +class DockerSyncGuardTest(unittest.TestCase): + """Docker goal 内联版与正典 parse_goal_result 的同步守卫: + 7 个真实标记用例 + 已存在撞名用例,两者状态必须全等; + 任一实现改了分叉,测试即 fail(不再只靠文档清单)。""" + + def _docker_status(self, stdout, returncode, stderr=""): + old = os.environ.get("XCM_API_KEY") + os.environ["XCM_API_KEY"] = "FAKE-TEST-KEY" + try: + ex = DockerExecutor(runner=lambda cmd, timeout=None: + FakeResult(returncode, stdout, stderr)) + ex._call_runner = lambda cmd, timeout=None, env=None: \ + FakeResult(returncode, stdout, stderr) + return ex.goal_task({"id": "t1", "goal": "do it", + "goal_max_rounds": 3})["status"] + finally: + if old is None: + del os.environ["XCM_API_KEY"] + else: + os.environ["XCM_API_KEY"] = old + + def test_docker_and_canonical_goal_status_equal(self): + cases = [ + ("goal met after rounds\n", 0, ""), + ("goal unmet after rounds\nround 1\nverify_out=fail\n", 0, ""), + ("goal interrupted after rounds: boom\n", 1, ""), + ("", 1, "boom"), + ("KeyError: breakpoint\n", 1, ""), + ("", 0, ""), + ("goal-resume 拒绝:breakpoint lost\n", 1, ""), + ("任务已存在\n", 1, ""), + ] + for stdout, rc, stderr in cases: + with self.subTest(stdout=stdout[:30], rc=rc): + canon = parse_goal_result( + "t1", FakeResult(rc, stdout, stderr), "local") + docker = self._docker_status(stdout, rc, stderr) + self.assertEqual(docker, canon["status"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/worker.py b/cloud/worker.py index 83e37ce..7973672 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -551,7 +551,8 @@ class Worker: self.executor.audit_callback = _chv_audit except Exception: pass - # CHV 补缺口轨:WarmPool 显式接线。池构造本身不自动 + # CHV WarmPool:仅孤儿清理接线(feat-chv-arch95 定案)。 + # 池构造本身不自动 # 清理(单活池假设,见 pool.py),worker 启动是显式 # 入口,在这里建池并调 reconcile_orphans();关闭时 # close() 再调一次。调用失败不炸 worker 构造/关闭, diff --git a/docs/chv-architecture.md b/docs/chv-architecture.md new file mode 100644 index 0000000..a216f75 --- /dev/null +++ b/docs/chv-architecture.md @@ -0,0 +1,80 @@ +# Cloud Hypervisor 执行器架构决策记录(ADR) + +范围:cloud/chv/ 执行器链(#216 → #221 → #230 → #232 → #234 → #239 → #246 → 本轨)。 +本文件只记录已落代码的决策,不记录未验证为已验证;真机、压测、动态安全 +仍见 docs/chv-production-readiness.md §11。 + +## ADR-1 SSH 目标:客机 IP 发现(已选) vs vsock / 纯 NAT(未选) + +- 决策:显式 CHV_SSH_HOST > CHV_NET_JSON 首个带 ip 的 net 条目 > + 显式 CHV_SSH_FORWARD=1 宣称下的回环;三者皆无报 + ssh_target_undeclared,绝不无条件假设 127.0.0.1:2222 能通。 +- 未选 vsock:需客内 agent + 宿主代理两套新代码,当前客机镜像 + 无此组件,写进代码仍是假保证。 +- 未选纯 NAT:DNAT 是宿主防火墙外部状态,执行器代码既不能 + 创建也不能校验它,写进代码仍是假保证;宿主防火墙见就绪 + 清单 §11.4 部署清单,代码门 + 防火墙缺一不可。 + +## ADR-2 网关 key 走 seed 投递(已选) vs argv / 宿主环境(已弃) + +- 决策:provisioned 模式下密钥子集随 seed 写进客内 + /etc/chv/guest-env(0600),远程命令 source 它;传输层密钥 + 进 argv 直接 ValueError;keyscan 模式有密钥即 fail-closed。 +- 放弃 argv:宿主 ps / /proc//cmdline 可读;放弃宿主子进程 + 环境:Docker 线 -e 只传名、值仍在 CLI 子进程环境,CHV 比它 + 更严(值连宿主子进程环境都不进)。 +- 残留:seed 随 stop 删,私钥在 seed 盘建好后即删, + cleanup_stale_secrets 扫崩溃残留;删失败 fail-fast 或脏标记拒起。 + +## ADR-3 磁盘 copy 默认,reuse 只作单租户实验(已选) + +- 决策:CHV_DISK_MODE 默认 copy,每任务独立 disk.raw; + reuse 须同时满足 CHV_REUSE_SINGLE_TENANT_EXPERIMENT=1 与 + CHV_SINGLE_TENANT_PROJECT 声明,且 wake 时审计放行。 +- 边界:代码无法验证数据库多租户状态,声明 + 审计即机制, + 不假装已校验;生产多租户下 reuse 一律不启用。 + +## ADR-4 restore 不重预置(纠正,见 #232 审计) + +- 纠正:审计原说法“restore 须把 seed 接入 VmConfig”是错的—— + restore 不走 vm.create,无 VmConfig 可挂盘;快照内已固化原 + VM host key,重预置反而破坏钉死。 +- 正确不变量:provisioned 下复用已钉死 known_hosts、缺失 + fail-fast(host_key_not_pinned),绝不 keyscan 回落。 + +## ADR-5 WarmPool:实验性,生产 dispatch 永远冷启动(本轨定案) + +- 决策(二选一的 B 项):把 WarmPool 从生产路径宣称里摘掉, + 明确标为实验/停用,不再留“半接线”悬空状态。 +- 原因:预热 VM 的磁盘是干净快照盘,不是任务盘;dispatch 前 + 查池若要真正接线,必须在派任务前给已暂停 VM 重新供给任务 + 盘(热换盘 / 二盘挂载 / host.py 磁盘语义三者都未实现), + 把预热盘直接当任务盘用会击穿任务盘隔离不变量。 +- 落地:WarmPool.wired_to_dispatch 恒为 False;_run_with_vm + 绝不查池,dispatch 永远冷启动;acquire 内 resume 只保留为 + 实验接口;worker 启动/关闭仍调 reconcile_orphans()(上一世 + 残留清理是池子保留的唯一生产价值)。 +- 重启条件:将来要重新接线,必须先实现任务盘重供给并补 + “任务盘来源”不变量测试,否则不得把 wired 改成 True。 + +## ADR-6 Docker goal 解析分叉:共享正典 + 同步守卫(本轨定案) + +- 决策:Local/CHV 共用 parse_goal_result 正典;Docker goal_task + 内联版保留,但与正典的 8 个真实标记用例(met/unmet/ + interrupted 标记/无结论异常退出/永久错误/无结论 rc0/ + goal-resume 拒绝/已存在撞名)由 test_chv_arch95.py 的 + DockerSyncGuardTest 行为测试钉住状态全等,任一实现改了 + 分叉测试即 fail,不再只靠文档清单。 +- 注意:Docker 的超时证据路径仍与内联解析绑在一起,那是 + 有意保留的第二处差异,不在 8 用例的状态等价范围内。 + +## ADR-7 #215 未合并时的复用边界(不复制模块) + +- 现状:#215(timeouts/gates/quota/sandbox_policy)未合并, + 本链 base 里无其模块,本轨不复制其代码。 +- 接口对齐点(#215 合并后机械替换,非重写):超时类 + api/ready/task 解析 → timeouts;门禁类 preflight/容量/ + 磁盘配额/egress fail-closed → gates;配额类 max_vms/disk_gb + (任务成本仍走 worker quota)→ quota;隔离类单任务单 VM/ + 磁盘 copy/egress 代理/SSH 钉死/池一次性与孤儿清理 → + sandbox_policy。替换时须重跑本链 163 项测试,不得改语义。 diff --git a/docs/chv-production-readiness.md b/docs/chv-production-readiness.md index 55a2bcd..b85fce5 100644 --- a/docs/chv-production-readiness.md +++ b/docs/chv-production-readiness.md @@ -71,8 +71,10 @@ - ✅🧪 快照/恢复接口:snapshot_vm / restore_vm(restore 后显式 resume,见 §3),WarmPool 接口与状态机(cold → prewarming → warm → assigned → running → recycled/destroyed)。 -- ✅🧪 冷启动回落:池空时 acquire 返回 cold,调用方回落 wake_vm - 冷启动,绝不降级 Docker/Local。 +- ⚠️ 定案(feat-chv-arch95,ADR-5):WarmPool 是实验性、非生产 + 路径。dispatch 永远冷启动,_run_with_vm 绝不查池;池空回落 + 冷启动的说法不再作为生产特性宣称,acquire/resume 只保留为 + 实验接口。 - ✅🧪 预热条目一次性:acquire 消耗、release 一律销毁,绝不跨 任务复用脏 VM。 - ✅🧪 预热池 registry 与孤儿清理(优化轨):池状态落 @@ -245,14 +247,13 @@ 未选,理由见 config.py)。 - ✅🧪 P1 已修或有结论:网关 extra_env 经 SSH env 前缀注入, worker 守卫收窄到 Local/CHV;取消/清理/看门狗接 CHV 通用 - 路径;WarmPool acquire 内 resume(但 worker dispatch 前查 - 池仍未接线,预热收益仍不是已交付特性);容量门改持久化 + 路径;WarmPool acquire 内 resume(本轨定案为实验性,见 ADR-5, + 不再是生产特性);容量门改持久化 计数(socket 联合计数);磁盘配额改总量+余量、已存在磁盘 校验重拷;API 客户端按 Content-Length 验完整性;快照 file:// 目标验存在且非空再 shutdown;execute 死端口归类 vm_not_running。 -- ⚠️ 仍未接线/未验证(不得声称已交付):worker dispatch 前查 - WarmPool;CHV + 网关真实任务;真机 wake/boot/SSH(含 seed +- ⚠️ 仍未验证(不得声称已交付):CHV + 网关真实任务;真机 wake/boot/SSH(含 seed 注入与 NAT/客机 IP 实测);快照恢复端到端;压测;安全审计; CI 账单恢复。另:P0-2 审计原说法“restore 须把 seed 接入 VmConfig”已纠正——restore 不走 vm.create,seed 无从挂盘, @@ -384,8 +385,7 @@ (tap/mac/ip/mask),未知字段直接拒绝。 - L5 WarmPool 预热盘与任务盘对应关系:定义已写进 pool.acquire 文档——预热盘是干净快照盘,不是任务盘; - 将来接线时任务盘仍须按任务重新供给或校验快照来源与 - 任务的对应关系;接线本身仍未做,不假装已接。 + 本轨定案为实验性(ADR-5),不再作为接线待办。 ### 11.4 宿主防火墙部署清单(代码做不到的部分,缺一不可) @@ -424,4 +424,5 @@ - VM 控制按钮、预热池管理页、成本/配额新页面(前端零改动); - 真机动态渗透、压测(无 KVM,做不了,只列清单); -- WarmPool dispatch 前查池接线(仍未接线,见 L5 定义)。 +- WarmPool dispatch 前查池接线(已定案不做,见 ADR-5;重启须先 + 实现任务盘重供给并补不变量测试)。