fix(chv): restore 路径补 H7 net 校验+登记,seed 暂存文件走 atomic_file

restore_vm 此前既不校验 net 冲突、也不登记 net.json:
已登记 VM 的 tap/mac/ip 会被恢复出的 VM 静默复用,且
恢复出的 VM 自身不登记、后来的 wake 也看不到它,两边
都漏判。现与 wake 同款:_validate_restore 在起 VMM 前
校验冲突(net_tap_conflict + vm_failed 审计),
_restore_boot 在 restore 成功后登记 net.json,登记失败
审计 net_registry_write_failed + 拒绝恢复(P2-19 语义,
stop_vm 清半恢复 VM)。

hostkeys.write_seed/_apply_seed_extras 的 meta-data、
user-data、network-config 改走 atomic_file(mkstemp 0600
→ 写 → fsync → rename):user-data 含 host 私钥明文,
直接写 + 事后 chmod 会在 chmod 前以 umask 权限短暂可读;
半截文件本身到不了 ISO(唯一消费者 genisoimage 只在
write_seed 返回后跑),原子写同时堵住权限窗口与半截残留。

回归测试 6 个:restore 冲突拒起/恢复后登记/登记失败
拒起/无 net_json 不登记;seed 走原子写(0600、无 tmp
残留)/写失败不留半截文件。CHV 全套件 395 测试通过。
This commit is contained in:
host committed 2026-11-06 19:32:41 +00:00
1 parent f448ecb74b
commit 06c3e172fe
5 files changed
+211 -18

No files matched your search

+68
View File
@@ -382,6 +382,74 @@ class HostKeyProvisionTest(ChvGapCase):
# 条目内是设计内的,见 hostkeys.write_seed)。
self.assertIn("GITEA_TOKEN='s3cr3t v'", content)
def test_write_seed_uses_atomic_write(self):
# feat-chv-cover100:seed 暂存文件(user-data 含 host
# 私钥明文)此前直接 open("w") 再 chmod 0600——崩溃
# 半截文件本身到不了 ISO(唯一消费者 genisoimage 只
# 在 write_seed 返回后跑,失败重试会整文件重写),
# 但 chmod 之前文件以 umask 权限(常为 0644)可读,
# 同 uid 进程能在该窗口读到私钥;现走 atomic_file
#(mkstemp 0600 → 写 → fsync → rename),文件从
# 创建起就是 0600,且写失败不留半截文件。
import stat as _stat
from unittest import mock
import cloud.chv.hostkeys as hostkeys_mod
from cloud.chv.atomic_file import atomic_write_text as _real
calls = []
def _spy(path, text, mode=0o600):
calls.append(path)
return _real(path, text, mode=mode)
with (tempfile.TemporaryDirectory() as tmp,
mock.patch.object(hostkeys_mod, "atomic_write_text",
side_effect=_spy)):
seed_dir = os.path.join(tmp, "seed")
write_seed(seed_dir, "vm1", "PRIV",
"ssh-ed25519 AAAAx",
network_config="version: 2\n")
self.assertEqual(
sorted(os.path.basename(p) for p in calls),
["meta-data", "network-config", "user-data"])
for name in ("meta-data", "user-data", "network-config"):
mode = _stat.S_IMODE(os.stat(
os.path.join(seed_dir, name)).st_mode)
self.assertEqual(mode, 0o600, name)
leftovers = [n for n in os.listdir(seed_dir)
if n.startswith(".tmp-")]
self.assertEqual(leftovers, [])
def test_write_seed_failure_leaves_no_partial_file(self):
# feat-chv-cover100:原子写的回归——写 user-data 中途
# 失败时不留半截 user-data、不留 tmp 残留;旧的直接
# 写会在崩溃后留半截文件(虽到不了 ISO,但含私钥明
# 文的半截文件滞留 seed 目录本身即泄漏面)。
from unittest import mock
import cloud.chv.hostkeys as hostkeys_mod
real = hostkeys_mod.atomic_write_text
def _fail_on_user_data(path, text, mode=0o600):
if path.endswith("user-data"):
raise OSError("disk boom")
return real(path, text, mode=mode)
with (tempfile.TemporaryDirectory() as tmp,
mock.patch.object(hostkeys_mod, "atomic_write_text",
side_effect=_fail_on_user_data)):
seed_dir = os.path.join(tmp, "seed")
os.makedirs(seed_dir)
with self.assertRaises(OSError):
write_seed(seed_dir, "vm1", "PRIV",
"ssh-ed25519 AAAAx")
self.assertFalse(os.path.exists(
os.path.join(seed_dir, "user-data")))
leftovers = [n for n in os.listdir(seed_dir)
if n.startswith(".tmp-")]
self.assertEqual(leftovers, [])
def test_keyscan_is_weak_fallback_only(self):
# provisioned 模式下 wake 前必须预置;keyscan 不应被
# 当成生产路径:配置值本身可解析,且文档/代码里标弱。