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

"""Cloud Hypervisor 客机 SSH 传输（叶子模块，不依赖
cloud.sandbox，避免循环导入）。

安全模型（本轨的核心变化，替代 #216 的
StrictHostKeyChecking=no + UserKnownHostsFile=/dev/null）：

- 每 VM 一个独立 known_hosts 文件：
  state_dir/<vm 名>/known_hosts（与每任务磁盘同目录，
  任务销毁时不自动删——known_hosts 留在宿主 state_dir 里，
  供事后审计该 VM 当时钉死的 key；VM 名本身已含任务 id，
  不会串到别的 VM 上）。
- 首次信任（TOFU，边界必须写清，不假装是强验证）：
  首次启动时客机 host key 未知，执行器用 ssh-keyscan 经
  宿主转发端口（127.0.0.1:<port>，只走本机回环）抓一次
  当时呈现的 key，写入该 VM 的 known_hosts 文件；此后
  同一 VM 的每次 SSH 都用 StrictHostKeyChecking=yes +
  该文件校验，VM 被替换 / 中间人换 key 时 SSH 直接拒绝，
  不再静默接受。首次捕获本身不验证“这台客机就是我们要
  的那台”——它只防首次之后的替换；更强的方案是镜像构建
  /cloud-init 时把客机 host key 预置进 known_hosts（未来
  工作，见 docs/chv-production-readiness.md 的待办），本轨
  在 mock 测试里只能验证“捕获→写文件→后续命令带钉死
  选项”的管线，绝不声称 mock 验证了首次密钥的真实性。
- 密钥红线不变且收紧（feat-chv-secfix，H2/M1）：argv 里只有
  key 文件路径，绝无 key 值；网关密钥类变量（见
  _CHV_SECRET_ENV_NAMES，与 Docker 的 _DOCKER_SECRET_ENV_NAMES
  同款“只传名”口径，但 CHV 更进一步：值连宿主子进程环境
  都不进，改走 seed 盘里的客内文件，见 hostkeys.write_seed）
  出现在 extra_env 里时本函数直接 ValueError——传输层
  fail-closed，任何调用方都不能再把密钥值拼回 argv。
  非密钥 extra_env 仍可走 env 前缀（值在 argv 里可见，
  但按定义不是密钥）。
- 远程 shell 转义（feat-chv-secfix，M1）：SSH 协议的 exec
  请求本身是一条字符串，ssh 客户端把多个参数用空格拼成
  远程 shell 命令——此前 inner 参数原样进 argv，prompt 里
  的 ;/$()/引号/换行会在远程 shell 里被二次解析。现在每个
  远程 token 都经 shlex.quote 引用后仍以列表元素形式返回
  （ssh 拼接后远程解析结果与本地 list 语义一致）；客内
  agent 本来就是 root，本项修的是“任务执行语义不被 prompt
  改写 + 审计不失真”，不是“防客内提权”（客内 root 无
  需注入即可为所欲为，边界见执行器文档）。
"""

import os
import shlex

# CHV 密钥类环境变量名：值绝不进 ssh argv。Docker 线是
# “-e 只传名、值由 CLI 子进程环境提供”（值仍在宿主进程
# 环境里，/proc/<pid>/environ 对同用户可见）；CHV 线是
# “值只进 seed 盘客内文件”（宿主进程表与宿主进程环境里
# 都没有值），比 Docker 线更严——差异见执行器文档。
_CHV_SECRET_ENV_NAMES = frozenset({"SUB2API_API_KEY"})


def split_secret_env(extra_env: dict) -> tuple:
    """把 extra_env 拆成（密钥子集，非密钥子集）。"""
    secrets, plain = {}, {}
    for key, value in dict(extra_env or {}).items():
        if key in _CHV_SECRET_ENV_NAMES:
            secrets[key] = value
        else:
            plain[key] = value
    return secrets, plain


def remote_command(inner_cmd: list, egress_proxy: str | None = None,
                   extra_env: dict | None = None,
                   guest_env_file: str | None = None) -> list:
    """构造远程 shell 命令的 token 列表（不含 ssh 前缀）。

    每个 token 都经 shlex.quote：ssh 客户端把 argv 用空格拼
    成一条远程命令时，远程 shell 解析回来的恰好是本地 list
    语义。guest_env_file 给定时，远程命令先 `set -a; . <文件>;
    set +a`（把 seed 投递的密钥文件 source 进环境并 export），
    再接 env 前缀与客内命令；文件路径本身不是密钥。
    extra_env 含密钥名时 ValueError（调用方须先 split 出
    密钥子集走 seed 投递，见 split_secret_env）。"""
    if extra_env:
        secrets, _ = split_secret_env(extra_env)
        if secrets:
            raise ValueError(
                "密钥类环境变量 %s 不得进 ssh argv，须走 seed "
                "文件投递" % ",".join(sorted(secrets)))
    tokens = []
    if guest_env_file:
        tokens += ["set", "-a;", ".",
                   shlex.quote(guest_env_file) + ";", "set", "+a;"]
    env_assignments = []
    if extra_env:
        for key in sorted(extra_env):
            env_assignments.append("%s=%s"
                                   % (key, shlex.quote(
                                       str(extra_env[key]))))
    if egress_proxy:
        env_assignments += [
            "HTTPS_PROXY=%s" % shlex.quote(egress_proxy),
            "HTTP_PROXY=%s" % shlex.quote(egress_proxy),
            "NO_PROXY=localhost,127.0.0.1"]
    inner = [shlex.quote(str(arg)) for arg in inner_cmd]
    if env_assignments:
        tokens += ["env"] + env_assignments + inner
    else:
        tokens += inner
    return tokens


def known_hosts_path(state_dir: str, vm_name: str) -> str:
    """该 VM 的 known_hosts 文件路径：state_dir/<vm>/known_hosts。"""
    return os.path.join(state_dir, vm_name, "known_hosts")


def build_ssh_command(ssh_port: int, ssh_key: str | None,
                      ssh_user: str, egress_proxy: str | None,
                      known_hosts_file: str, inner_cmd: list,
                      ssh_host: str = "127.0.0.1",
                      extra_env: dict | None = None,
                      guest_env_file: str | None = None) -> list:
    """构造 SSH 命令：ssh -p <端口> -i <key 文件> + 钉死选项
    + user@<ssh_host> + 客内命令（远程 token 均经 shlex.quote，
    见 remote_command）。

    feat-chv-fix（P0-4）：ssh_host 不再在传输层写死回环，
    由执行器解析传入（显式 CHV_SSH_HOST / net 客机 IP /
    显式转发宣称下的回环）；本函数缺省回环仅为旧调用点
    兼容，执行器路径绝不依赖该缺省。
    feat-chv-secfix（H2）：extra_env 只接受非密钥变量；
    密钥类变量须先经 split_secret_env 拆出、走 seed 盘客内
    文件投递，调用方把 guest_env_file 传进来（客内 seed
    写出的 /etc/chv/guest-env），本函数在远程命令前 source
    它；密钥值绝不出现在本函数返回的任何 argv 元素里。
    egress_proxy 已设时客内命令前加 env 前缀（代理地址本身
    不是密钥；客机出口只走该白名单代理——但这只是 env 约
    定，网络层强制见宿主防火墙部署清单，代码不声称已隔离）。"""
    remote = remote_command(inner_cmd, egress_proxy=egress_proxy,
                            extra_env=extra_env,
                            guest_env_file=guest_env_file)
    return ["ssh", "-p", str(ssh_port),
            "-i", ssh_key,
            "-o", "StrictHostKeyChecking=yes",
            "-o", "UserKnownHostsFile=%s" % known_hosts_file,
            "-o", "BatchMode=yes",
            "%s@%s" % (ssh_user, ssh_host)] + remote


def parse_keyscan_output(text: str, host: str = "127.0.0.1") -> list:
    """从 ssh-keyscan 的 stdout 里挑出目标主机的 host key
    行：形如 "[<host>]:<port> ssh-ed25519 AAAA..."（-p 时
    的标准写法）或 "<host> ssh-rsa AAAA..."。其它主机行、
    注释行、空行一律丢掉，绝不把陌生主机的 key 写进文件。
    host 缺省回环仅为旧调用点兼容；客机 IP 方案下应传
    解析出的客机 IP。"""
    kept = []
    for line in (text or "").splitlines():
        line = line.strip()
        if not line or line.startswith("#"):
            continue
        parts = line.split()
        if len(parts) < 3:
            continue
        line_host, keytype = parts[0], parts[1]
        if line_host != host \
                and not line_host.startswith("[%s]:" % host):
            continue
        if not keytype.startswith("ssh-"):
            continue
        kept.append(line)
    return kept


def write_known_hosts(path: str, key_lines: list) -> None:
    """原子写 known_hosts：先写同目录临时文件再 rename，
    文件权限 0600（只含公钥行，仍按敏感文件对待）。"""
    directory = os.path.dirname(path)
    if directory:
        os.makedirs(directory, exist_ok=True)
    tmp_path = path + ".tmp"
    with open(tmp_path, "w") as fh:
        fh.write("\n".join(key_lines) + "\n")
    os.chmod(tmp_path, 0o600)
    os.replace(tmp_path, path)