diff --git a/cloud/chv/config.py b/cloud/chv/config.py index 693cfe1..0fdbf2d 100644 --- a/cloud/chv/config.py +++ b/cloud/chv/config.py @@ -245,6 +245,20 @@ def _timeout_fragment(params: dict, env: dict) -> dict: } +def _gitea_fragment(env: dict) -> dict: + """Gitea 同步片段:宿主配置环境变量 CHV_GITEA_*(见 + gitea_sync.CHV_GITEA_ENV_MAP)映射成的 guest-env 键值 + dict。无配置即空 dict(本次启动不同步);值本身不校验 + 完整性——完整性由 wake_vm 在起 VMM 前 fail-closed + (validate_gitea_sync_env),构造时不拦,免得一个配 + 错的宿主连执行器对象都建不出来、排障信息全丢。 + import 在函数内:本模块是叶子模块,顶层只许 import + 标准库(见 test_leaf_modules_only_import_stdlib), + 与执行器的方法内 import 同款惯例。""" + from cloud.chv.gitea_sync import config_gitea_sync_env + return {"gitea_sync_env": config_gitea_sync_env(env)} + + def _assemble_chv_config(fragments: list) -> "ChvConfig": """把各片段 dict 合并成 ChvConfig(纯合并,不改值)。 @@ -277,4 +291,5 @@ def build_chv_config(params: dict, env: dict) -> "ChvConfig": _disk_fragment(env), _network_fragment(params, env), _timeout_fragment(params, env), + _gitea_fragment(env), ]) diff --git a/cloud/chv/executor.py b/cloud/chv/executor.py index 3bdddc4..764f568 100644 --- a/cloud/chv/executor.py +++ b/cloud/chv/executor.py @@ -59,6 +59,7 @@ from cloud.chv.ssh_transport import build_ssh_command as _build_ssh_cmd from cloud.chv.ssh_transport import known_hosts_path as _known_hosts_path from cloud.chv.ssh_transport import ( parse_keyscan_output, + split_guest_env, split_secret_env, write_known_hosts, ) @@ -398,7 +399,8 @@ class CloudHypervisorExecutor(Executor): "ssh_forward", "reuse_experiment", "single_tenant_project", "snapshot_max_count", "snapshot_max_total_gb", "snapshot_retention_days", - "snapshot_dir", "allow_remote_snapshot"): + "snapshot_dir", "allow_remote_snapshot", + "gitea_sync_env"): setattr(self, _field, getattr(_cfg, _field)) # -- 名字 / 路径 --------------------------------------------------- @@ -827,7 +829,8 @@ class CloudHypervisorExecutor(Executor): [known_hosts_line(self.ssh_port, public_text, host=ssh_host)]) - def provision_host_key(self, task: dict, secret_env: dict | None = None) -> dict: + def provision_host_key(self, task: dict, secret_env: dict | None = None, + guest_env: dict | None = None) -> dict: """按 VM 预置 host key(补缺口轨强方案,见 hostkeys.py): 启动前为该 VM 生成一次性 ed25519 host key 对,公钥 立即写进该 VM 的 known_hosts(钉死),私钥只写进 @@ -837,7 +840,9 @@ class CloudHypervisorExecutor(Executor): 不同 VM 的 key 对不同(state_dir/ 隔离),同一 VM 重启复用已有对(否则钉死立即失效)。 feat-chv-secfix(H2/M2):secret_env 给定时写进 - seed 的客内密钥文件(0600);上次清理失败留下的 + seed 的客内密钥文件(0600);guest_env 给定时与 + secret_env 合并写同一客内文件(boot 时要用的非密钥 + 同步变量,见 gitea_sync);上次清理失败留下的 脏标记存在时直接拒绝(清理失败即拒起,不再复用 旧 key 对)。成功后审计 host_key_generated(只含 VM 名与密钥名,不含任何 key 值)。""" @@ -862,7 +867,8 @@ class CloudHypervisorExecutor(Executor): self._write_provisioned_known_hosts(paths, public_text) seed_dir = write_seed(paths["seed_dir"], vm, private_text, public_text, - secret_env=secret_env) + secret_env=secret_env, + guest_env=guest_env) self._emit("host_key_generated", task, "vm=%s secret_names=%s" % (vm, ",".join(sorted(secret_env or {})))) @@ -1305,6 +1311,37 @@ class CloudHypervisorExecutor(Executor): "VMM api-socket 组/其它可写(%o),拒绝启动:%s" % (mode, socket_path)) + def _effective_extra_env(self, task: dict, + extra_env: dict | None) -> dict | None: + """wake 生效的 extra_env:把 Gitea 同步变量并进来。 + + 优先级(低→高):执行器宿主配置 CHV_GITEA_*(构造时 + 已映射成 gitea_sync_env)< 任务 dict 的 gitea_sync < + 显式 extra_env 同名键(见 gitea_sync.extract_...)。 + 同步已请求(任一同步键出现)但必填键不全时,在起 + VMM 之前就 ChvFailure(gitea_sync_incomplete, + fail-closed;错误只含缺失键名,不含值)——半份同步 + 配置绝不能让 VM 带着“以为同步了”的状态启动。无同步 + 变量时原样返回(本次启动不同步)。 + 必须在 _wake_prechecks 之前调:任务/配置来源的 + GITEA_TOKEN 也要受 keyscan 模式 fail-closed 检查。""" + from cloud.chv.gitea_sync import ( + extract_gitea_sync_env, + validate_gitea_sync_env, + ) + sync = extract_gitea_sync_env( + task=task, extra_env=extra_env, + config_sync=getattr(self, "gitea_sync_env", None)) + try: + validate_gitea_sync_env(sync) + except ValueError as exc: + raise ChvFailure("gitea_sync_incomplete", str(exc)) + if not sync: + return extra_env + merged = dict(extra_env or {}) + merged.update(sync) + return merged + def _wake_prechecks(self, task: dict, name: str, extra_env) -> tuple: """wake 前置:preflight/容量/配额/net/密钥模式,纯移动。 失败时发 vm_failed 审计并原样抛 ChvFailure。""" @@ -1343,7 +1380,8 @@ class CloudHypervisorExecutor(Executor): raise return secrets, net_entries - def _provision_seed_disk(self, task: dict, name: str, secrets) -> str | None: + def _provision_seed_disk(self, task: dict, name: str, secrets, + guest_env: dict | None = None) -> str | None: """provisioned 模式的 seed 盘供给:生成→制盘→挂盘, 任一步失败即抛(调用方 stop_vm 清理),纯移动。""" seed_disk = None @@ -1366,7 +1404,8 @@ class CloudHypervisorExecutor(Executor): seed_disk_path, ) provisioned = self.provision_host_key( - task, secret_env=secrets or None) + task, secret_env=secrets or None, + guest_env=guest_env or None) seed_disk = seed_disk_path(self.state_dir, name) build_seed_disk( provisioned["seed_dir"], seed_disk, @@ -1446,10 +1485,19 @@ class CloudHypervisorExecutor(Executor): if not isinstance(task, dict) or task.get("id") is None: raise ValueError("任务缺少 id") name = self.vm_name(task) + try: + extra_env = self._effective_extra_env(task, extra_env) + except ChvFailure as exc: + self._emit("vm_failed", task, + "vm=%s wake 失败 kind=%s" % (name, exc.kind), + exc.kind) + raise secrets, net_entries = self._wake_prechecks(task, name, extra_env) + guest_env, _plain = split_guest_env(extra_env) task_disk, socket_path = self._wake_prepare(task, name) try: - seed_disk = self._provision_seed_disk(task, name, secrets) + seed_disk = self._provision_seed_disk( + task, name, secrets, guest_env=guest_env or None) self._boot_vm(task, name, task_disk, seed_disk, net_entries) except Exception: try: diff --git a/cloud/chv/gitea_sync.py b/cloud/chv/gitea_sync.py new file mode 100644 index 0000000..7b4ca5f --- /dev/null +++ b/cloud/chv/gitea_sync.py @@ -0,0 +1,126 @@ +"""Gitea 同步变量供给(CHV guest-env 集成,叶子模块)。 + +背景:客内 chv-sync boot 在 cloud-init 时运行、SSH 之前, +只能读 seed 投递的 /etc/chv/guest-env,看不到 SSH env 前缀。 +因此同步所需的 6 个变量必须随 seed 进 guest-env,不能只靠 +extra_env 的 SSH 前缀;其中 GITEA_TOKEN 是密钥,只进 seed, +绝不进 argv/日志/镜像。 + +变量与来源(优先级低→高,后者覆盖前者): +1. 宿主配置环境变量(主来源,生产口径):CHV_GITEA_URL、 + CHV_GITEA_USER、CHV_GITEA_TOKEN、CHV_SYNC_REPO、 + CHV_SYNC_BRANCH、CHV_GITEA_EMAIL。由部署该 worker/执行器 + 的运维在宿主环境里设置,与 CHV_SSH_KEY、CHV_EGRESS_PROXY + 同款口径;CHV 是单租户单 VM 模型,一台宿主对应一个 + Gitea 身份。密钥来源绝不硬编码进代码。 +2. 任务级覆盖(内存态):任务 dict 的 `gitea_sync` 子 dict, + 键即 6 个变量名。仅在调用方以内存态构造任务 dict 时生效 + (直接调 wake_vm、测试、手工驱动);tasks 表目前没有 + JSON/元数据列,DB 往返会丢掉它——这是已知边界,不做 + 静默声称,需要每任务不同仓库时再做任务表迁移。 +3. 显式 extra_env:调用方直接传给 wake_vm/dispatch 的 + extra_env 里的同名键,优先级最高。 + +缺失语义(fail-closed,不跳过同步): +- 6 个键一个都没有 → 本次启动不同步(chv-sync boot 见无 + guest-env 静默成功,VM 照常用)。 +- 任一键出现、但必填 4 键(GITEA_URL、GITEA_USER、 + GITEA_TOKEN、SYNC_REPO)缺失 → 校验失败,wake 在起 VMM + 之前直接拒绝;半份同步配置绝不能让 VM 带着“我以为同步 + 了”的状态启动。SYNC_BRANCH、GITEA_EMAIL 可选。 +- 校验错误只含缺失的键名,绝不含任何变量值。 + +本模块是纯函数,不读文件、不跑 VM、不含任何密钥值。 +""" + +# 同步变量全集(guest-env 里应有的键)。 +GITEA_SYNC_KEYS = frozenset({ + "GITEA_URL", "GITEA_USER", "GITEA_TOKEN", + "SYNC_REPO", "SYNC_BRANCH", "GITEA_EMAIL", +}) + +# 其中密钥:只进 seed,绝不进 argv(见 ssh_transport 的 +# _CHV_SECRET_ENV_NAMES)。 +GITEA_SYNC_SECRET_KEYS = frozenset({"GITEA_TOKEN"}) + +# 其中非密钥、但 boot 时就要用、也必须进 guest-env 的键 +#(SSH env 前缀到不了 cloud-init 时运行的 chv-sync boot)。 +GITEA_SYNC_PLAIN_KEYS = GITEA_SYNC_KEYS - GITEA_SYNC_SECRET_KEYS + +# 必填键:缺一即 fail-closed(见模块头缺失语义)。 +GITEA_SYNC_REQUIRED_KEYS = frozenset({ + "GITEA_URL", "GITEA_USER", "GITEA_TOKEN", "SYNC_REPO", +}) + +# 宿主配置环境变量名 → guest-env 键名。 +CHV_GITEA_ENV_MAP = { + "CHV_GITEA_URL": "GITEA_URL", + "CHV_GITEA_USER": "GITEA_USER", + "CHV_GITEA_TOKEN": "GITEA_TOKEN", + "CHV_SYNC_REPO": "SYNC_REPO", + "CHV_SYNC_BRANCH": "SYNC_BRANCH", + "CHV_GITEA_EMAIL": "GITEA_EMAIL", +} + + +def _clean(mapping: dict | None) -> dict: + """只留白名单键:值须为非空字符串(strip 后),其它 + 类型/空值/未知键一律丢掉——调用方拼错的键名绝不静默 + 进 seed 变成客内无人读的脏变量。""" + cleaned = {} + for key, value in dict(mapping or {}).items(): + if key not in GITEA_SYNC_KEYS: + continue + if not isinstance(value, str): + continue + value = value.strip() + if not value: + continue + cleaned[key] = value + return cleaned + + +def config_gitea_sync_env(env: dict | None) -> dict: + """宿主配置环境变量(CHV_GITEA_*)→ guest-env 键名。""" + mapped = {} + for env_name, key in CHV_GITEA_ENV_MAP.items(): + value = (env or {}).get(env_name) + if value is not None: + mapped[key] = value + return _clean(mapped) + + +def task_gitea_sync_env(task: dict | None) -> dict: + """任务 dict 的 `gitea_sync` 子 dict → 白名单后的键值。 + 不是 dict(或缺失)即空——绝不把任务其它字段误读成 + 同步变量。""" + if not isinstance(task, dict): + return {} + return _clean(task.get("gitea_sync")) + + +def extract_gitea_sync_env(task: dict | None = None, + extra_env: dict | None = None, + config_sync: dict | None = None) -> dict: + """三源合并(优先级:config_sync < 任务 gitea_sync < + extra_env 同名键)。返回白名单后的 dict,可能为空 + (空 = 本次启动不同步)。""" + merged = {} + merged.update(_clean(config_sync)) + merged.update(task_gitea_sync_env(task)) + merged.update(_clean(extra_env)) + return merged + + +def validate_gitea_sync_env(env: dict | None) -> None: + """校验同步变量完整性:空即不同步(通过);非空但必填 + 键缺失即 ValueError(只含缺失键名,不含值)。调用方 + (执行器 wake_vm)在起 VMM 前调本函数做 fail-closed。""" + env = dict(env or {}) + if not env: + return + missing = sorted(GITEA_SYNC_REQUIRED_KEYS - set(env)) + if missing: + raise ValueError( + "Gitea 同步变量不完整,缺失:%s(同步已请求但必填" + "键不全,拒绝启动)" % ",".join(missing)) diff --git a/cloud/chv/hostkeys.py b/cloud/chv/hostkeys.py index 7575b4c..5898308 100644 --- a/cloud/chv/hostkeys.py +++ b/cloud/chv/hostkeys.py @@ -168,20 +168,37 @@ def build_seed_user_data(private_key_text: str, public_key_text: str) -> str: return "\n".join(lines) + "\n" +def _guest_env_entry(merged_env: dict) -> list: + """guest-env 的 write_files 条目行(0600 root,值经 + shlex.quote 引用):只能并进已有的 write_files 列表, + 绝不能再写一个 write_files: 键(YAML 重复键会覆盖掉 + host key 的两条写文件条目,预置即失效)。""" + lines = [ + " - path: %s" % GUEST_ENV_PATH, + " permissions: '0600'", + " owner: root:root", + " content: |", + ] + for key in sorted(merged_env): + lines.append(" %s=%s" + % (key, shlex.quote(str(merged_env[key])))) + return lines + + def write_seed(seed_dir: str, vm_name: str, private_key_text: str, public_key_text: str, - secret_env: dict | None = None) -> str: - """写 seed 目录:meta-data + user-data(均为 0600)。 + secret_env: dict | None = None, + guest_env: dict | None = None) -> str: + """写 seed 目录:meta-data + user-data(均为 0600), + 返回 seed 目录路径。调用方负责随 VM 销毁删掉它。 - 返回 seed 目录路径。seed 目录本身不含已知主机之外的 - 共享数据,调用方(执行器关闭/回收)负责删除它。 - feat-chv-secfix(H2):secret_env 给定时,user-data 再 - 写一个 write_files 条目,把密钥环境变量写进客内 - GUEST_ENV_PATH(0600 root),值经 shlex.quote 引用后 - 才进文件内容——远程命令 source 该文件即可得到密钥 - 环境,密钥值从不出现在宿主 ssh argv、宿主进程环境与 - 审计明文里。secret_env 为空时不写该条目(客内文件不 - 存在时远程命令也不 source 它,见执行器)。""" + secret_env 给定时写进客内 GUEST_ENV_PATH(0600 root, + 远程命令 source 它得密钥环境,值不进宿主 argv/环境/ + 审计明文)。guest_env 给定时与 secret_env 合并写同一 + 个条目(secret 优先于同名键,防两行同名);guest_env + 装 boot 时要用的非密钥同步变量(GITEA_URL 等,见 + gitea_sync):只走 SSH 前缀到不了 cloud-init 时运行 + 的 chv-sync boot。两者皆空时不写该条目。""" os.makedirs(seed_dir, exist_ok=True) meta_path = os.path.join(seed_dir, "meta-data") user_path = os.path.join(seed_dir, "user-data") @@ -190,24 +207,14 @@ def write_seed(seed_dir: str, vm_name: str, private_key_text: str, % (vm_name, vm_name)) os.chmod(meta_path, 0o600) user_data = build_seed_user_data(private_key_text, public_key_text) - if secret_env: - # 注意:只能把条目并进已有的 write_files 列表,绝不能 - # 再写一个 write_files: 键(YAML 重复键会覆盖掉 host - # key 的两条写文件条目,预置即失效)。seed 自 2026-11-07 - # 起不再带 runcmd 段(见 build_seed_user_data 的根因 - # 禁令),所以正常走 else 分支追加到 write_files 列表 - # 末尾;marker 分支只作防御(将来模板若恢复 runcmd, - # 条目仍插在其前、仍属同一列表)。 - lines = [ - " - path: %s" % GUEST_ENV_PATH, - " permissions: '0600'", - " owner: root:root", - " content: |", - ] - for key in sorted(secret_env): - lines.append(" %s=%s" - % (key, shlex.quote(str(secret_env[key])))) - # 条目插在 runcmd 段之前、仍属同一 write_files 列表。 + merged_env = dict(guest_env or {}) + merged_env.update(secret_env or {}) + if merged_env: + lines = _guest_env_entry(merged_env) + # 条目插在 runcmd 段之前、仍属同一 write_files 列表 + #(seed 自 2026-11-07 起不再带 runcmd,见 + # build_seed_user_data 的根因禁令;marker 分支只作 + # 防御,将来模板若恢复 runcmd 仍插在其前)。 marker = "runcmd:\n" if marker in user_data: head, tail = user_data.split(marker, 1) diff --git a/cloud/chv/ssh_transport.py b/cloud/chv/ssh_transport.py index 71899b1..2975f60 100644 --- a/cloud/chv/ssh_transport.py +++ b/cloud/chv/ssh_transport.py @@ -44,12 +44,25 @@ StrictHostKeyChecking=no + UserKnownHostsFile=/dev/null): import os import shlex +from cloud.chv.gitea_sync import ( + GITEA_SYNC_PLAIN_KEYS, + GITEA_SYNC_SECRET_KEYS, +) + # CHV 密钥类环境变量名:值绝不进 ssh argv。Docker 线是 # “-e 只传名、值由 CLI 子进程环境提供”(值仍在宿主进程 # 环境里,/proc//environ 对同用户可见);CHV 线是 # “值只进 seed 盘客内文件”(宿主进程表与宿主进程环境里 # 都没有值),比 Docker 线更严——差异见执行器文档。 -_CHV_SECRET_ENV_NAMES = frozenset({"SUB2API_API_KEY"}) +_CHV_SECRET_ENV_NAMES = frozenset({"SUB2API_API_KEY"}) \ + | GITEA_SYNC_SECRET_KEYS + +# CHV guest-env 类非密钥变量名:值可进 SSH env 前缀(按定义 +# 不是密钥),但 chv-sync boot 在 cloud-init 时运行、SSH 之 +# 前,只能读 seed 投递的 /etc/chv/guest-env,所以这些键也 +# 必须随 seed 写进 guest-env(见 split_guest_env 与执行器 +# wake_vm),只走 SSH 前缀是不够的。 +_CHV_GUEST_ENV_NAMES = frozenset(GITEA_SYNC_PLAIN_KEYS) def split_secret_env(extra_env: dict | None) -> tuple[dict, dict]: @@ -63,6 +76,23 @@ def split_secret_env(extra_env: dict | None) -> tuple[dict, dict]: return secrets, plain +def split_guest_env(extra_env: dict | None) -> tuple[dict, dict]: + """把 extra_env 拆成(须写进 seed guest-env 的子集, + 非密钥子集)。 + + guest-env 子集 = 密钥子集 + _CHV_GUEST_ENV_NAMES 命中的 + 非密钥同步变量(GITEA_URL 等 boot 时就要用的键);第二 + 个返回仍是全部非密钥子集(含 guest-env 命中键——它们 + 任务期也要经 SSH env 前缀进客机环境,seed 只解决 + boot 时可见性,不替代任务期前缀)。""" + secrets, plain = split_secret_env(extra_env) + guest_env = dict(secrets) + for key, value in plain.items(): + if key in _CHV_GUEST_ENV_NAMES: + guest_env[key] = value + return guest_env, plain + + def remote_command(inner_cmd: list, egress_proxy: str | None = None, extra_env: dict | None = None, guest_env_file: str | None = None) -> list: diff --git a/cloud/tests/test_chv_gitea_sync.py b/cloud/tests/test_chv_gitea_sync.py new file mode 100644 index 0000000..fec9fd6 --- /dev/null +++ b/cloud/tests/test_chv_gitea_sync.py @@ -0,0 +1,495 @@ +"""CHV guest-env 集成测试:Gitea 同步变量供给链路。 + +只用标准库与 mock runner/launcher/api_client,绝不启动真实 +VM、不联网、不含任何真实 token(测试值全是假值)。 + +覆盖: +- gitea_sync 纯函数:三源合并优先级、白名单清洗、完整性 + 校验(fail-closed)与空=不同步; +- ssh_transport:GITEA_TOKEN 进密钥集(argv 拒绝)、其余 + 5 键进 guest-env 集; +- hostkeys.write_seed:guest_env 与 secret_env 合并写同一 + 条目、仍无 runcmd、0600; +- 执行器 wake_vm:完整同步变量进 seed user-data;缺失必填 + 键时起 VMM 前 ChvFailure 且不启动;无同步变量时正常启动 + 且不写 guest-env;seed 投递后 build_ssh_command 才能用 + GITEA_TOKEN 且 argv 里绝无 token 值; +- worker:_gitea_sync_env 取值;_run_attempt 只对 CHV 执行器 + 把同步变量并进 gateway_env。 +""" + +import os +import sys +import tempfile +import unittest +from pathlib import Path +from types import SimpleNamespace + +import yaml + +try: + from cloud.chv.gitea_sync import ( + GITEA_SYNC_KEYS, + config_gitea_sync_env, + extract_gitea_sync_env, + validate_gitea_sync_env, + ) + from cloud.chv.hostkeys import write_seed + from cloud.chv.ssh_transport import ( + _CHV_SECRET_ENV_NAMES, + build_ssh_command, + remote_command, + split_guest_env, + split_secret_env, + ) + 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.gitea_sync import ( + GITEA_SYNC_KEYS, + config_gitea_sync_env, + extract_gitea_sync_env, + validate_gitea_sync_env, + ) + from cloud.chv.hostkeys import write_seed + from cloud.chv.ssh_transport import ( + _CHV_SECRET_ENV_NAMES, + build_ssh_command, + remote_command, + split_guest_env, + split_secret_env, + ) + from cloud.sandbox import ChvFailure, CloudHypervisorExecutor + from cloud.worker import Worker + + +FAKE_SYNC = { + "GITEA_URL": "https://gitea.tose.top", + "GITEA_USER": "admin", + "GITEA_TOKEN": "fake-token-123", + "SYNC_REPO": "admin/chv-sync-test", + "SYNC_BRANCH": "main", + "GITEA_EMAIL": "admin@tose.top", +} + + +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 + + +def _seed_user_text(state_dir, vm_name): + with open(os.path.join(state_dir, vm_name, "seed", "user-data")) as fh: + return fh.read() + + +def _guest_env_lines(user_text): + parsed = yaml.safe_load(user_text) + entries = parsed["write_files"] + guest = [e for e in entries if e["path"] == "/etc/chv/guest-env"] + assert len(guest) == 1, entries + return guest[0]["content"], guest[0]["permissions"] + + +# -- (1) gitea_sync 纯函数 --------------------------------------------- +class GiteaSyncPureTest(unittest.TestCase): + def test_keys_are_six(self): + self.assertEqual(GITEA_SYNC_KEYS, set(FAKE_SYNC)) + + def test_config_mapping(self): + env = {"CHV_GITEA_URL": "https://gitea.tose.top", + "CHV_GITEA_USER": "admin", + "CHV_GITEA_TOKEN": "tok", + "CHV_SYNC_REPO": "admin/chv-sync-test", + "CHV_SYNC_BRANCH": "main", + "CHV_GITEA_EMAIL": "admin@tose.top", + "CHV_UNRELATED": "x"} + self.assertEqual(config_gitea_sync_env(env), { + "GITEA_URL": "https://gitea.tose.top", + "GITEA_USER": "admin", "GITEA_TOKEN": "tok", + "SYNC_REPO": "admin/chv-sync-test", + "SYNC_BRANCH": "main", "GITEA_EMAIL": "admin@tose.top"}) + + def test_extract_priority_config_lt_task_lt_extra_env(self): + task = {"id": "t1", "gitea_sync": { + "GITEA_URL": "https://task.example", + "GITEA_USER": "task-user", "GITEA_TOKEN": "task-tok", + "SYNC_REPO": "task/repo"}} + out = extract_gitea_sync_env( + task=task, + extra_env={"SYNC_REPO": "extra/repo"}, + config_sync={"GITEA_URL": "https://config.example", + "GITEA_USER": "config-user", + "GITEA_TOKEN": "config-tok", + "SYNC_REPO": "config/repo", + "SYNC_BRANCH": "dev"}) + self.assertEqual(out["GITEA_URL"], "https://task.example") + self.assertEqual(out["GITEA_USER"], "task-user") + self.assertEqual(out["GITEA_TOKEN"], "task-tok") + self.assertEqual(out["SYNC_REPO"], "extra/repo") + self.assertEqual(out["SYNC_BRANCH"], "dev") + + def test_extract_drops_unknown_and_empty_and_nonstr(self): + task = {"id": "t1", "gitea_sync": { + "GITEA_URL": "https://gitea.tose.top", + "EVIL": "nope", "SYNC_REPO": " ", + "GITEA_TOKEN": 12345}} + out = extract_gitea_sync_env(task=task) + self.assertEqual(out, {"GITEA_URL": "https://gitea.tose.top"}) + self.assertNotIn("gitea_sync", out) + + def test_build_chv_config_maps_gitea_env(self): + from cloud.chv.config import build_chv_config + cfg = build_chv_config({}, { + "CHV_GITEA_URL": "https://gitea.tose.top", + "CHV_GITEA_USER": "admin", + "CHV_GITEA_TOKEN": "tok", + "CHV_SYNC_REPO": "admin/chv-sync-test"}) + self.assertEqual(cfg.gitea_sync_env, { + "GITEA_URL": "https://gitea.tose.top", + "GITEA_USER": "admin", "GITEA_TOKEN": "tok", + "SYNC_REPO": "admin/chv-sync-test"}) + cfg2 = build_chv_config({}, {}) + self.assertEqual(cfg2.gitea_sync_env, {}) + + def test_validate_empty_passes_partial_fails_with_names_only(self): + validate_gitea_sync_env({}) + validate_gitea_sync_env(None) + validate_gitea_sync_env(dict(FAKE_SYNC)) + with self.assertRaises(ValueError) as ctx: + validate_gitea_sync_env({"GITEA_URL": "https://gitea.tose.top", + "SYNC_REPO": "admin/chv-sync-test"}) + msg = str(ctx.exception) + self.assertIn("GITEA_TOKEN", msg) + self.assertIn("GITEA_USER", msg) + self.assertNotIn("gitea.tose.top", msg) + self.assertNotIn("chv-sync-test", msg) + + +# -- (2) ssh_transport ------------------------------------------------- +class SshTransportSyncTest(unittest.TestCase): + def test_gitea_token_is_secret(self): + self.assertIn("GITEA_TOKEN", _CHV_SECRET_ENV_NAMES) + self.assertIn("SUB2API_API_KEY", _CHV_SECRET_ENV_NAMES) + secrets, plain = split_secret_env( + {"GITEA_TOKEN": "tok", "GITEA_URL": "https://x", + "OTHER": "1"}) + self.assertEqual(secrets, {"GITEA_TOKEN": "tok"}) + self.assertEqual(plain, {"GITEA_URL": "https://x", "OTHER": "1"}) + + def test_split_guest_env_covers_all_six_keys(self): + guest, plain = split_guest_env(dict(FAKE_SYNC, OTHER="1")) + self.assertEqual(set(guest), set(FAKE_SYNC)) + self.assertEqual(guest["GITEA_TOKEN"], "fake-token-123") + # 非密钥子集仍含同步普通变量(任务期 SSH 前缀照常带它们)。 + self.assertIn("GITEA_URL", plain) + self.assertNotIn("GITEA_TOKEN", plain) + self.assertIn("OTHER", plain) + self.assertNotIn("OTHER", guest) + + def test_remote_command_rejects_gitea_token(self): + with self.assertRaises(ValueError): + remote_command(["true"], extra_env={"GITEA_TOKEN": "tok"}) + with self.assertRaises(ValueError): + build_ssh_command(2222, "/tmp/key", "root", None, + "/tmp/known_hosts", ["true"], + extra_env={"GITEA_TOKEN": "tok"}) + + +# -- (3) hostkeys.write_seed ------------------------------------------- +class WriteSeedGuestEnvTest(unittest.TestCase): + def test_guest_env_and_secret_merge_into_one_entry_no_runcmd(self): + with tempfile.TemporaryDirectory() as tmp: + seed_dir = os.path.join(tmp, "seed") + write_seed(seed_dir, "vm1", "PRIV", "ssh-ed25519 AAAAx", + secret_env={"SUB2API_API_KEY": "gw-key"}, + guest_env={"GITEA_URL": "https://gitea.tose.top", + "GITEA_USER": "admin", + "SYNC_REPO": "admin/chv-sync-test", + "GITEA_TOKEN": "dup-should-lose"}) + with open(os.path.join(seed_dir, "user-data")) as fh: + content = fh.read() + self.assertEqual(content.count("write_files:"), 1) + parsed = yaml.safe_load(content) + self.assertNotIn("runcmd", parsed) + self.assertEqual(len(parsed["write_files"]), 3) + text, perms = _guest_env_lines(content) + self.assertEqual(perms, "0600") + self.assertIn("GITEA_URL=https://gitea.tose.top", text) + self.assertIn("SUB2API_API_KEY=gw-key", text) + # 同名键以 secret_env 为准,不出现两行 GITEA_TOKEN。 + self.assertEqual(text.count("GITEA_TOKEN="), 1) + self.assertIn("GITEA_TOKEN=dup-should-lose", text) + + def test_guest_env_only_still_no_runcmd(self): + with tempfile.TemporaryDirectory() as tmp: + seed_dir = os.path.join(tmp, "seed") + write_seed(seed_dir, "vm1", "PRIV", "ssh-ed25519 AAAAx", + guest_env={"GITEA_URL": "https://gitea.tose.top"}) + with open(os.path.join(seed_dir, "user-data")) as fh: + content = fh.read() + parsed = yaml.safe_load(content) + self.assertNotIn("runcmd", parsed) + self.assertEqual(len(parsed["write_files"]), 3) + + +# -- (4) 执行器 wake_vm ------------------------------------------------- +class WakeGuestEnvCase(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.launches = [] + + 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"ISO") + return FakeResult(0, "", "") + return FakeResult(0, "", "") + + def _launcher(self, cmd): + self.launches.append(list(cmd)) + return FakeProc(alive=True) + + def make_exec(self, env_extra=None): + env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888", + "CHV_HOST_KEY_MODE": "provisioned", + "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=self._runner, launcher=self._launcher, + api_client=lambda s, m, p, payload=None: (200, {}), + 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 + + def test_wake_with_task_gitea_sync_writes_all_six_to_seed(self): + ex = self.make_exec() + task = {"id": "sync1", "gitea_sync": dict(FAKE_SYNC)} + out = ex.wake_vm(task) + self.assertEqual(out["state"], "running") + text = _seed_user_text(self.state_dir, "chv-sync1") + self.assertNotIn("runcmd", yaml.safe_load(text)) + guest_text, perms = _guest_env_lines(text) + self.assertEqual(perms, "0600") + for key, value in FAKE_SYNC.items(): + self.assertIn("%s=%s" % (key, value), guest_text) + # seed.iso 已制盘且 0600。 + seed_iso = os.path.join(self.state_dir, "chv-sync1", "seed.iso") + self.assertTrue(os.path.exists(seed_iso)) + self.assertEqual(oct(os.stat(seed_iso).st_mode & 0o777), "0o600") + + def test_wake_with_config_env_sync(self): + ex = self.make_exec(env_extra={ + "CHV_GITEA_URL": "https://gitea.tose.top", + "CHV_GITEA_USER": "admin", + "CHV_GITEA_TOKEN": "fake-token-123", + "CHV_SYNC_REPO": "admin/chv-sync-test", + "CHV_SYNC_BRANCH": "main"}) + self.assertEqual(ex.gitea_sync_env["SYNC_REPO"], + "admin/chv-sync-test") + out = ex.wake_vm({"id": "sync2"}) + self.assertEqual(out["state"], "running") + guest_text, _ = _guest_env_lines( + _seed_user_text(self.state_dir, "chv-sync2")) + self.assertIn("GITEA_TOKEN=fake-token-123", guest_text) + self.assertIn("SYNC_BRANCH=main", guest_text) + + def test_wake_partial_sync_fails_closed_before_launch(self): + ex = self.make_exec() + task = {"id": "sync3", "gitea_sync": { + "GITEA_URL": "https://gitea.tose.top", + "SYNC_REPO": "admin/chv-sync-test"}} + with self.assertRaises(ChvFailure) as ctx: + ex.wake_vm(task) + self.assertEqual(ctx.exception.kind, "gitea_sync_incomplete") + self.assertEqual(self.launches, []) + self.assertFalse(os.path.exists( + os.path.join(self.state_dir, "chv-sync3", "seed.iso"))) + kinds = [a[0] for a in self.audit] + self.assertIn("vm_failed", kinds) + + def test_wake_partial_sync_from_config_also_fails_closed(self): + ex = self.make_exec(env_extra={ + "CHV_GITEA_URL": "https://gitea.tose.top"}) + with self.assertRaises(ChvFailure) as ctx: + ex.wake_vm({"id": "sync4"}) + self.assertEqual(ctx.exception.kind, "gitea_sync_incomplete") + self.assertEqual(self.launches, []) + + def test_wake_without_sync_writes_no_guest_env(self): + ex = self.make_exec() + out = ex.wake_vm({"id": "plain1"}) + self.assertEqual(out["state"], "running") + parsed = yaml.safe_load( + _seed_user_text(self.state_dir, "chv-plain1")) + self.assertEqual(len(parsed["write_files"]), 2) + + def test_build_ssh_command_after_wake_sources_guest_env(self): + ex = self.make_exec() + ex.wake_vm({"id": "sync5", "gitea_sync": dict(FAKE_SYNC)}) + cmd = ex.build_ssh_command( + {"id": "sync5"}, ["true"], + extra_env={"GITEA_TOKEN": "fake-token-123"}) + joined = " ".join(cmd) + self.assertNotIn("fake-token-123", joined) + self.assertIn("/etc/chv/guest-env", joined) + # 未 wake 投递的 VM 用密钥即拒绝。 + with self.assertRaises(ChvFailure) as ctx: + ex.build_ssh_command( + {"id": "other"}, ["true"], + extra_env={"GITEA_TOKEN": "fake-token-123"}) + self.assertEqual(ctx.exception.kind, "secret_env_not_delivered") + + def test_wake_keyscan_mode_with_token_fails_closed(self): + ex = self.make_exec(env_extra={"CHV_HOST_KEY_MODE": "keyscan"}) + with self.assertRaises(ChvFailure): + ex.wake_vm({"id": "sync6", "gitea_sync": dict(FAKE_SYNC)}) + self.assertEqual(self.launches, []) + + +# -- (5) worker --------------------------------------------------------- +class WorkerGiteaSyncTest(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + from cloud.server import init_db + init_db(self.db_path) + + def tearDown(self): + self.tmp.cleanup() + + def _worker(self, executor, env_extra=None): + env = {"XCM_API_KEY": "test-placeholder"} + if env_extra: + env.update(env_extra) + return Worker(self.db_path, "/fake/host.py", + executor=executor, env=env) + + def test_gitea_sync_env_from_worker_env_and_task(self): + worker = self._worker(None, env_extra={ + "CHV_GITEA_URL": "https://gitea.tose.top", + "CHV_GITEA_USER": "admin", + "CHV_GITEA_TOKEN": "env-tok", + "CHV_SYNC_REPO": "admin/chv-sync-test"}) + out = worker._gitea_sync_env({"id": "t1"}) + self.assertEqual(out["GITEA_TOKEN"], "env-tok") + out2 = worker._gitea_sync_env( + {"id": "t1", "gitea_sync": {"GITEA_TOKEN": "task-tok"}}) + self.assertEqual(out2["GITEA_TOKEN"], "task-tok") + self.assertEqual(out2["SYNC_REPO"], "admin/chv-sync-test") + worker2 = self._worker(None) + self.assertEqual(worker2._gitea_sync_env({"id": "t1"}), {}) + + def test_run_attempt_merges_sync_env_only_for_chv(self): + import tempfile as _tf + case_dir = _tf.mkdtemp() + binary = _write(os.path.join(case_dir, "cloud-hypervisor")) + ch_remote = _write(os.path.join(case_dir, "ch-remote")) + kernel = _write(os.path.join(case_dir, "vmlinux")) + rootfs = _write(os.path.join(case_dir, "disk.raw"), b"R" * 100) + ssh_key = _write(os.path.join(case_dir, "id_chv")) + kvm = _write(os.path.join(case_dir, "kvm")) + cpuinfo = os.path.join(case_dir, "cpuinfo") + with open(cpuinfo, "w") as fh: + fh.write("flags : fpu vmx sse\n") + ex = CloudHypervisorExecutor( + binary=binary, ch_remote=ch_remote, kernel=kernel, + rootfs=rootfs, ssh_key=ssh_key, kvm_path=kvm, + cpuinfo_path=cpuinfo, + state_dir=os.path.join(case_dir, "state"), + runner=lambda cmd, timeout=None: FakeResult(0, "", ""), + launcher=lambda cmd: FakeProc(alive=True), + api_client=lambda s, m, p, payload=None: (200, {}), + sleeper=lambda s: None, + env={"CHV_EGRESS_PROXY": "http://10.0.0.1:8888", + "CHV_HOST_KEY_MODE": "provisioned", + "CHV_SSH_FORWARD": "1"}) + seen = {} + orig_dispatch = ex.dispatch_task + + def fake_dispatch(task, extra_env=None, model_provider=None): + seen["extra_env"] = dict(extra_env or {}) + return {"executor": "cloud-hypervisor", "task_id": task["id"], + "status": "completed", "stdout": "", "stderr": "", + "returncode": 0, "host_status": "completed", + "usage": None} + ex.dispatch_task = fake_dispatch + worker = self._worker(ex, env_extra={ + "CHV_GITEA_URL": "https://gitea.tose.top", + "CHV_GITEA_USER": "admin", + "CHV_GITEA_TOKEN": "fake-token-123", + "CHV_SYNC_REPO": "admin/chv-sync-test"}) + worker._ensure_gateway = lambda task: (None, {}) + task = {"id": "w1", "project_id": "p1", "prompt": "做事"} + result, error = worker._run_attempt(task) + self.assertIsNone(error) + self.assertEqual(result["status"], "completed") + self.assertEqual(seen["extra_env"]["GITEA_TOKEN"], + "fake-token-123") + self.assertEqual(seen["extra_env"]["SYNC_REPO"], + "admin/chv-sync-test") + worker.close() + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/worker.py b/cloud/worker.py index d3a41b6..49c2d04 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -2575,6 +2575,25 @@ class Worker: } return "sub2api", env + def _gitea_sync_env(self, task): + """Gitea 同步变量(CHV guest-env 集成):任务 gitea_sync + 覆盖宿主 CHV_GITEA_* 配置(见 gitea_sync.extract_...)。 + + 只读本任务与本 worker 的 self._env(默认 os.environ), + 绝不硬编码任何值、不读其它任务、不写日志。返回空 dict + 即本次启动不同步;完整性校验不在这里做——执行器 + wake_vm 在起 VMM 前 fail-closed(gitea_sync_incomplete), + worker 只负责把变量带进 extra_env。 + 注意:tasks 表没有 JSON/元数据列,DB 往返会丢任务 dict + 的 gitea_sync;生产主来源是宿主 CHV_GITEA_* 环境变量。""" + from cloud.chv.gitea_sync import ( + config_gitea_sync_env, + extract_gitea_sync_env, + ) + return extract_gitea_sync_env( + task=task, + config_sync=config_gitea_sync_env(self._env)) + def _ensure_gateway(self, task): """任务 dict 上的网关缓存:已算过直接取,否则算一次并存回。 @@ -4684,6 +4703,14 @@ class Worker: message="worker exec start: %s" % _exec_detail) try: gateway_provider, gateway_env = self._ensure_gateway(task) + if isinstance(self.executor, CloudHypervisorExecutor): + # CHV guest-env 集成:Gitea 同步变量随 extra_env + # 进 wake_vm(密钥走 seed,boot 变量也走 seed; + # 见 gitea_sync)。只对 CHV 合并——Docker/Local + # 的密钥处理口径不同(-e/子进程环境),不在本轮 + # 改动,避免 GITEA_TOKEN 经其它通道外泄。 + gateway_env = _merge_env( + gateway_env, self._gitea_sync_env(task)) if self._use_popen(): if goal_mode and task.get("_goal_resume_breakpoint"): # G10 断点:无 --goal-text,不开新循环。