feat(chv-guest): Gitea 同步变量经 seed 进 guest-env(执行器集成)

- 新增 cloud/chv/gitea_sync.py:6 键白名单,三源合并
  (宿主 CHV_GITEA_* < 任务 gitea_sync < extra_env),
  必填 4 键缺失即 fail-closed,空即不同步
- ssh_transport:GITEA_TOKEN 进密钥集(argv 拒绝),其余
  5 键进 guest-env 集(split_guest_env)
- hostkeys.write_seed:guest_env 与 secret_env 合并写同一
  条目,仍无 runcmd、0600
- config:_gitea_fragment 产 gitea_sync_env(函数内 import,
  叶子模块顶层只许标准库)
- executor.wake_vm:起 VMM 前合并+校验(gitea_sync_incomplete),
  guest-env 随 seed 投递;provision_host_key 透传 guest_env
- worker:_run_attempt 仅对 CHV 执行器把同步变量并进 gateway_env
- 测试:cloud/tests/test_chv_gitea_sync.py 20 项
- 边界:tasks 表无 JSON 列,DB 往返丢任务 gitea_sync;
  生产主来源是宿主 CHV_GITEA_* 环境变量
This commit is contained in:
host committed 2026-11-06 16:55:32 +00:00
1 parent 67ca0b86cd
commit e7f699c62b
7 files changed
+785 -37

No files matched your search

+15
View File
@@ -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": def _assemble_chv_config(fragments: list) -> "ChvConfig":
"""把各片段 dict 合并成 ChvConfig(纯合并,不改值)。 """把各片段 dict 合并成 ChvConfig(纯合并,不改值)。
@@ -277,4 +291,5 @@ def build_chv_config(params: dict, env: dict) -> "ChvConfig":
_disk_fragment(env), _disk_fragment(env),
_network_fragment(params, env), _network_fragment(params, env),
_timeout_fragment(params, env), _timeout_fragment(params, env),
_gitea_fragment(env),
]) ])
+55 -7
View File
@@ -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 known_hosts_path as _known_hosts_path
from cloud.chv.ssh_transport import ( from cloud.chv.ssh_transport import (
parse_keyscan_output, parse_keyscan_output,
split_guest_env,
split_secret_env, split_secret_env,
write_known_hosts, write_known_hosts,
) )
@@ -398,7 +399,8 @@ class CloudHypervisorExecutor(Executor):
"ssh_forward", "reuse_experiment", "ssh_forward", "reuse_experiment",
"single_tenant_project", "snapshot_max_count", "single_tenant_project", "snapshot_max_count",
"snapshot_max_total_gb", "snapshot_retention_days", "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)) setattr(self, _field, getattr(_cfg, _field))
# -- 名字 / 路径 --------------------------------------------------- # -- 名字 / 路径 ---------------------------------------------------
@@ -827,7 +829,8 @@ class CloudHypervisorExecutor(Executor):
[known_hosts_line(self.ssh_port, public_text, [known_hosts_line(self.ssh_port, public_text,
host=ssh_host)]) 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 预置 host key(补缺口轨强方案,见 hostkeys.py):
启动前为该 VM 生成一次性 ed25519 host key 对,公钥 启动前为该 VM 生成一次性 ed25519 host key 对,公钥
立即写进该 VM 的 known_hosts(钉死),私钥只写进 立即写进该 VM 的 known_hosts(钉死),私钥只写进
@@ -837,7 +840,9 @@ class CloudHypervisorExecutor(Executor):
不同 VM 的 key 对不同(state_dir/<vm> 隔离),同一 不同 VM 的 key 对不同(state_dir/<vm> 隔离),同一
VM 重启复用已有对(否则钉死立即失效)。 VM 重启复用已有对(否则钉死立即失效)。
feat-chv-secfix(H2/M2):secret_env 给定时写进 feat-chv-secfix(H2/M2):secret_env 给定时写进
seed 的客内密钥文件(0600);上次清理失败留下的 seed 的客内密钥文件(0600);guest_env 给定时与
secret_env 合并写同一客内文件(boot 时要用的非密钥
同步变量,见 gitea_sync);上次清理失败留下的
脏标记存在时直接拒绝(清理失败即拒起,不再复用 脏标记存在时直接拒绝(清理失败即拒起,不再复用
旧 key 对)。成功后审计 host_key_generated(只含 旧 key 对)。成功后审计 host_key_generated(只含
VM 名与密钥名,不含任何 key 值)。""" VM 名与密钥名,不含任何 key 值)。"""
@@ -862,7 +867,8 @@ class CloudHypervisorExecutor(Executor):
self._write_provisioned_known_hosts(paths, public_text) self._write_provisioned_known_hosts(paths, public_text)
seed_dir = write_seed(paths["seed_dir"], vm, seed_dir = write_seed(paths["seed_dir"], vm,
private_text, public_text, private_text, public_text,
secret_env=secret_env) secret_env=secret_env,
guest_env=guest_env)
self._emit("host_key_generated", task, self._emit("host_key_generated", task,
"vm=%s secret_names=%s" "vm=%s secret_names=%s"
% (vm, ",".join(sorted(secret_env or {})))) % (vm, ",".join(sorted(secret_env or {}))))
@@ -1305,6 +1311,37 @@ class CloudHypervisorExecutor(Executor):
"VMM api-socket 组/其它可写(%o),拒绝启动:%s" "VMM api-socket 组/其它可写(%o),拒绝启动:%s"
% (mode, socket_path)) % (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: def _wake_prechecks(self, task: dict, name: str, extra_env) -> tuple:
"""wake 前置:preflight/容量/配额/net/密钥模式,纯移动。 """wake 前置:preflight/容量/配额/net/密钥模式,纯移动。
失败时发 vm_failed 审计并原样抛 ChvFailure。""" 失败时发 vm_failed 审计并原样抛 ChvFailure。"""
@@ -1343,7 +1380,8 @@ class CloudHypervisorExecutor(Executor):
raise raise
return secrets, net_entries 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 盘供给:生成→制盘→挂盘, """provisioned 模式的 seed 盘供给:生成→制盘→挂盘,
任一步失败即抛(调用方 stop_vm 清理),纯移动。""" 任一步失败即抛(调用方 stop_vm 清理),纯移动。"""
seed_disk = None seed_disk = None
@@ -1366,7 +1404,8 @@ class CloudHypervisorExecutor(Executor):
seed_disk_path, seed_disk_path,
) )
provisioned = self.provision_host_key( 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) seed_disk = seed_disk_path(self.state_dir, name)
build_seed_disk( build_seed_disk(
provisioned["seed_dir"], 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: if not isinstance(task, dict) or task.get("id") is None:
raise ValueError("任务缺少 id") raise ValueError("任务缺少 id")
name = self.vm_name(task) 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) 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) task_disk, socket_path = self._wake_prepare(task, name)
try: 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) self._boot_vm(task, name, task_disk, seed_disk, net_entries)
except Exception: except Exception:
try: try:
+126
View File
@@ -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))
+36 -29
View File
@@ -168,20 +168,37 @@ def build_seed_user_data(private_key_text: str, public_key_text: str) -> str:
return "\n".join(lines) + "\n" 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, def write_seed(seed_dir: str, vm_name: str, private_key_text: str,
public_key_text: str, public_key_text: str,
secret_env: dict | None = None) -> str: secret_env: dict | None = None,
"""写 seed 目录:meta-data + user-data(均为 0600)。 guest_env: dict | None = None) -> str:
"""写 seed 目录:meta-data + user-data(均为 0600),
返回 seed 目录路径。调用方负责随 VM 销毁删掉它。
返回 seed 目录路径。seed 目录本身不含已知主机之外的 secret_env 给定时写进客内 GUEST_ENV_PATH(0600 root,
共享数据,调用方(执行器关闭/回收)负责删除它。 远程命令 source 它得密钥环境,值不进宿主 argv/环境/
feat-chv-secfix(H2):secret_env 给定时,user-data 再 审计明文)。guest_env 给定时与 secret_env 合并写同一
写一个 write_files 条目,把密钥环境变量写进客内 个条目(secret 优先于同名键,防两行同名);guest_env
GUEST_ENV_PATH(0600 root),值经 shlex.quote 引用后 装 boot 时要用的非密钥同步变量(GITEA_URL 等,见
才进文件内容——远程命令 source 该文件即可得到密钥 gitea_sync):只走 SSH 前缀到不了 cloud-init 时运行
环境,密钥值从不出现在宿主 ssh argv、宿主进程环境与 的 chv-sync boot。两者皆空时不写该条目。"""
审计明文里。secret_env 为空时不写该条目(客内文件不
存在时远程命令也不 source 它,见执行器)。"""
os.makedirs(seed_dir, exist_ok=True) os.makedirs(seed_dir, exist_ok=True)
meta_path = os.path.join(seed_dir, "meta-data") meta_path = os.path.join(seed_dir, "meta-data")
user_path = os.path.join(seed_dir, "user-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)) % (vm_name, vm_name))
os.chmod(meta_path, 0o600) os.chmod(meta_path, 0o600)
user_data = build_seed_user_data(private_key_text, public_key_text) user_data = build_seed_user_data(private_key_text, public_key_text)
if secret_env: merged_env = dict(guest_env or {})
# 注意:只能把条目并进已有的 write_files 列表,绝不能 merged_env.update(secret_env or {})
# 再写一个 write_files: 键(YAML 重复键会覆盖掉 host if merged_env:
# key 的两条写文件条目,预置即失效)。seed 自 2026-11-07 lines = _guest_env_entry(merged_env)
# 起不再带 runcmd 段(见 build_seed_user_data 的根因 # 条目插在 runcmd 段之前、仍属同一 write_files 列表
# 禁令),所以正常走 else 分支追加到 write_files 列表 #(seed 自 2026-11-07 起不再带 runcmd,见
# 末尾;marker 分支只作防御(将来模板若恢复 runcmd, # build_seed_user_data 的根因禁令;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 列表。
marker = "runcmd:\n" marker = "runcmd:\n"
if marker in user_data: if marker in user_data:
head, tail = user_data.split(marker, 1) head, tail = user_data.split(marker, 1)
+31 -1
View File
@@ -44,12 +44,25 @@ StrictHostKeyChecking=no + UserKnownHostsFile=/dev/null):
import os import os
import shlex import shlex
from cloud.chv.gitea_sync import (
GITEA_SYNC_PLAIN_KEYS,
GITEA_SYNC_SECRET_KEYS,
)
# CHV 密钥类环境变量名:值绝不进 ssh argv。Docker 线是 # CHV 密钥类环境变量名:值绝不进 ssh argv。Docker 线是
# “-e 只传名、值由 CLI 子进程环境提供”(值仍在宿主进程 # “-e 只传名、值由 CLI 子进程环境提供”(值仍在宿主进程
# 环境里,/proc/<pid>/environ 对同用户可见);CHV 线是 # 环境里,/proc/<pid>/environ 对同用户可见);CHV 线是
# “值只进 seed 盘客内文件”(宿主进程表与宿主进程环境里 # “值只进 seed 盘客内文件”(宿主进程表与宿主进程环境里
# 都没有值),比 Docker 线更严——差异见执行器文档。 # 都没有值),比 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]: 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 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, def remote_command(inner_cmd: list, egress_proxy: str | None = None,
extra_env: dict | None = None, extra_env: dict | None = None,
guest_env_file: str | None = None) -> list: guest_env_file: str | None = None) -> list:
+495
View File
@@ -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()
+27
View File
@@ -2575,6 +2575,25 @@ class Worker:
} }
return "sub2api", env 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): def _ensure_gateway(self, task):
"""任务 dict 上的网关缓存:已算过直接取,否则算一次并存回。 """任务 dict 上的网关缓存:已算过直接取,否则算一次并存回。
@@ -4684,6 +4703,14 @@ class Worker:
message="worker exec start: %s" % _exec_detail) message="worker exec start: %s" % _exec_detail)
try: try:
gateway_provider, gateway_env = self._ensure_gateway(task) 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 self._use_popen():
if goal_mode and task.get("_goal_resume_breakpoint"): if goal_mode and task.get("_goal_resume_breakpoint"):
# G10 断点:无 --goal-text,不开新循环。 # G10 断点:无 --goal-text,不开新循环。