codex-app-server/cloud/chv/hostkeys.py

"""按 VM 生成的 SSH host key 预置（替代 TOFU 的强方案）。

安全模型（补缺口轨，替代 #221 的 ssh-keyscan 首次捕获）：
- 绝不用镜像固定同一把 host key：所有 VM 共用一把 key 时，
  一台 VM 泄漏即全网可冒充，属安全事故。
- 每 VM 一次性生成：宿主在 VMM 启动前为该 VM 生成一对
  ed25519 host key（私钥只活在该 VM 的 seed 里，随 VM
  销毁而删，绝不跨 VM 复用、绝不写进镜像、绝不进已知
  主机文件之外的共享位置）。
- 公钥预置钉死：生成后立即把公钥写进该 VM 的
  state_dir/<vm>/known_hosts（格式 [127.0.0.1]:<port> ...），
  执行器在 VM 启动前就已钉死，启动后 SSH 直接用该文件
  校验，换 key 即拒绝——不再靠启动后 ssh-keyscan 抓到的
  key 来“事后钉死”。
- 私钥注入通道（选 cloud-init NoCloud，写清唯一通道）：
  seed 目录 state_dir/<vm>/seed/ 含 meta-data（instance-id
  即 VM 名）、user-data（write_files 写 /etc/ssh/ 的 host
  key 对并重启 sshd）。客机镜像硬性要求（见
  chv-guest/README.md §6）：必须装 cloud-init 且支持
  NoCloud seed、首启时先应用 seed 再启动 sshd、且不得在
  seed 应用前/后无条件重新生成 host key（否则预置即失效）。
  seed 磁盘/目录是 per-VM 敏感物，VM 关闭/回收时必须删。
- 回落边界：ssh-keyscan 首次捕获只作弱回落
  （CHV_HOST_KEY_MODE=keyscan，且文档标为弱方案）：
  它只能防首次之后的替换，不能验证首次本身。生产部署
  必须用 provisioned 模式；provisioned 模式下 seed 工具
  或 keygen 不可用时直接 fail-fast，绝不静默回落 keyscan。

本模块只做纯函数与文件操作，不跑 VM、不联网、不含
任何私钥值（测试用假 runner 时私钥由测试自己写文件）。
"""

import os


def host_key_paths(state_dir, vm_name):
    """该 VM 的 host key 文件路径（私钥/公钥/known_hosts）。"""
    base = os.path.join(state_dir, vm_name)
    return {
        "private": os.path.join(base, "ssh_host_ed25519_key"),
        "public": os.path.join(base, "ssh_host_ed25519_key.pub"),
        "known_hosts": os.path.join(base, "known_hosts"),
        "seed_dir": os.path.join(base, "seed"),
    }


def known_hosts_line(port, public_key_text):
    """公钥文件内容转 known_hosts 行：只取前两列（类型+数据）。

    公钥文件形如 'ssh-ed25519 AAAA... [comment]'，known_hosts
    行形如 '[127.0.0.1]:<port> ssh-ed25519 AAAA...'。非法
    内容 ValueError，绝不把陌生行原样写进文件。"""
    parts = (public_key_text or "").strip().split()
    if len(parts) < 2 or not parts[0].startswith("ssh-"):
        raise ValueError("host 公钥格式非法")
    return "[127.0.0.1]:%s %s %s" % (port, parts[0], parts[1])


def build_seed_user_data(private_key_text, public_key_text):
    """cloud-init user-data：把 host key 对写进客机 /etc/ssh。

    注意：user-data 本身含私钥，只写进 per-VM seed 目录
    （0600），随 VM 销毁而删，绝不进镜像、绝不进日志、
    绝不进任务 prompt 落库。"""
    if not private_key_text or not public_key_text:
        raise ValueError("host key 对缺失，无法生成 seed")
    lines = [
        "#cloud-config",
        "ssh_deletekeys: false",
        "write_files:",
        "  - path: /etc/ssh/ssh_host_ed25519_key",
        "    permissions: '0600'",
        "    owner: root:root",
        "    content: |",
    ]
    for line in private_key_text.rstrip("\n").split("\n"):
        lines.append("      " + line)
    lines += [
        "  - path: /etc/ssh/ssh_host_ed25519_key.pub",
        "    permissions: '0644'",
        "    owner: root:root",
        "    content: |",
    ]
    for line in public_key_text.rstrip("\n").split("\n"):
        lines.append("      " + line)
    lines += [
        "runcmd:",
        "  - chmod 600 /etc/ssh/ssh_host_ed25519_key",
        "  - systemctl restart ssh || service ssh restart || true",
    ]
    return "\n".join(lines) + "\n"


def write_seed(seed_dir, vm_name, private_key_text, public_key_text):
    """写 seed 目录：meta-data + user-data（均为 0600）。

    返回 seed 目录路径。seed 目录本身不含已知主机之外的
    共享数据，调用方（执行器关闭/回收）负责删除它。"""
    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")
    with open(meta_path, "w") as fh:
        fh.write("instance-id: %s\nlocal-hostname: %s\n"
                 % (vm_name, vm_name))
    os.chmod(meta_path, 0o600)
    with open(user_path, "w") as fh:
        fh.write(build_seed_user_data(private_key_text, public_key_text))
    os.chmod(user_path, 0o600)
    return seed_dir


def generate_host_keypair(state_dir, vm_name, runner=None,
                          keygen="ssh-keygen"):
    """为该 VM 生成（或复用本 VM 已有的）host key 对。

    同一 VM 重启/恢复时复用已有对（否则钉死的 known_hosts
    立即失效）；不同 VM 的 state_dir/<vm> 不同，绝不共用。
    runner 不给时用 subprocess，测试可注入假 runner（假
    runner 负责把私钥/公钥文件写出来）。keygen 失败即
    RuntimeError，调用方不得回落 keyscan（provisioned
    模式）或必须显式标弱回落（keyscan 模式，见执行器）。"""
    paths = host_key_paths(state_dir, vm_name)
    if (os.path.exists(paths["private"])
            and os.path.exists(paths["public"])):
        return paths
    base = os.path.dirname(paths["private"])
    os.makedirs(base, exist_ok=True)
    cmd = [keygen, "-t", "ed25519", "-N", "",
           "-f", paths["private"]]
    if runner is not None:
        result = runner(cmd)
        rc = getattr(result, "returncode", 0)
        if rc not in (0, None):
            raise RuntimeError("ssh-keygen 失败：returncode=%s" % rc)
    else:
        import subprocess
        result = subprocess.run(cmd, capture_output=True, text=True)
        if result.returncode != 0:
            raise RuntimeError("ssh-keygen 失败：%s" % result.stderr)
    if not (os.path.exists(paths["private"])
            and os.path.exists(paths["public"])):
        raise RuntimeError("ssh-keygen 未生成 host key 对")
    try:
        os.chmod(paths["private"], 0o600)
    except OSError:
        pass
    return paths