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

"""CHV 安全加固 helper（feat-chv-secfix，叶子模块）。

这里只放不依赖执行器类的纯校验/文件操作：state_dir 安全
目录、VMM socket 活性、net 条目隔离校验、快照 URL 白名单、
egress 代理地址校验。宿主防火墙规则、tap 桥接本身是运维
外部状态，代码建不出来也不校验——那部分只写进部署清单
（docs/chv-production-readiness.md §11），绝不在代码里
假装已隔离。
"""

import json
import os
import socket

from cloud.chv.errors import ChvFailure

# NET_JSON 条目字段白名单（L4）：tap/mac/ip/mask 之外的字段
# 一律拒绝——未知字段原样送进 VMM API 的行为不确定，静默
# 透传等于把 VMM 的字段语义赌在运维手写上。
_NET_FIELDS = frozenset({"tap", "mac", "ip", "mask"})


def ensure_secure_state_dir(state_dir: str) -> None:
    """建 state_dir 并保证 0700（H6）：目录不存在即按 0700
    建；已存在但组/其它位可访时 chmod 0700；失败即
    ChvFailure（state_dir_insecure），绝不带着 /tmp 式
    宽松目录启动 VM。umask 影响 mkdir 的默认权限，故建
    后一律显式 chmod 校验一次。"""
    try:
        os.makedirs(state_dir, exist_ok=True)
        os.chmod(state_dir, 0o700)
    except OSError as exc:
        raise ChvFailure(
            "state_dir_insecure",
            "state_dir 无法建成安全目录（0700）：%s：%s"
            % (state_dir, exc))
    try:
        mode = os.stat(state_dir).st_mode & 0o777
    except OSError as exc:
        raise ChvFailure(
            "state_dir_insecure",
            "state_dir 无法校验权限：%s：%s" % (state_dir, exc))
    if mode & 0o077:
        raise ChvFailure(
            "state_dir_insecure",
            "state_dir 权限仍宽松（%o），拒绝启动：%s"
            % (mode, state_dir))
    return state_dir


def socket_is_live(socket_path: str) -> bool:
    """VMM api-socket 是否真有监听者（M7）：能连通才算活。
    普通文件、空文件、已死 VMM 的残留 socket 一律算死——
    容量计数不再被残留文件占满。连接失败的任何异常都算
    死（不区分 refused/超时，计数是尽力而为的读操作）。
    注意竞态：刚启动、尚未 listen 的 VMM 会被判死，容量
    门在并发唤醒时可能短暂低估——文档明写该窗口，容量门
    是“拒绝超限”的护栏，不是精确的分布式锁。"""
    try:
        if not socket_path or not os.path.exists(socket_path):
            return False
        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        try:
            sock.settimeout(0.5)
            return sock.connect_ex(socket_path) == 0
        finally:
            sock.close()
    except OSError:
        return False


def parse_net_entries(net_json: str) -> list:
    """解析 CHV_NET_JSON 为条目 list（非法 JSON / 非
    list/dict 即 net_config_invalid）。"""
    if not net_json:
        return []
    try:
        net = json.loads(net_json)
    except ValueError:
        raise ChvFailure("net_config_invalid",
                         "CHV_NET_JSON 不是合法 JSON")
    entries = net if isinstance(net, list) else [net]
    if not all(isinstance(e, dict) for e in entries):
        raise ChvFailure("net_config_invalid",
                         "CHV_NET_JSON 条目须为 dict")
    return entries


def validate_net_entries(entries: list, existing_entries: list) -> list:
    """net 条目隔离校验（H7 + L4）：

    - 字段白名单：未知字段即 net_config_invalid；
    - 条目内 tap/mac/ip 归一化（mac 小写、空 tap 视为未
      指定，不参与 tap 冲突比较——空 tap 是“VMM 默认网卡”
      的运维写法，代码不编造 tap 名，但空 tap 多 VM 共用
      默认网卡的风险仍在部署清单里明写）；
    - 本次条目内 tap/mac/ip 重复即 net_config_invalid；
    - 与已登记 VM（existing_entries，每项为 {tap, mac, ip,
      vm}）的 tap/mac/ip 任一重复即 net_tap_conflict，
      拒绝启动并由调用方审计。
    返回归一化后的条目 list。"""
    seen_taps, seen_macs, seen_ips = set(), set(), set()
    normalized = []
    for entry in entries:
        unknown = set(entry) - _NET_FIELDS
        if unknown:
            raise ChvFailure(
                "net_config_invalid",
                "CHV_NET_JSON 条目含未知字段：%s（白名单：%s）"
                % (",".join(sorted(unknown)),
                   ",".join(sorted(_NET_FIELDS))))
        tap = str(entry.get("tap") or "").strip()
        mac = str(entry.get("mac") or "").strip().lower()
        ip = str(entry.get("ip") or "").strip()
        for kind, value, seen in (("tap", tap, seen_taps),
                                  ("mac", mac, seen_macs),
                                  ("ip", ip, seen_ips)):
            if not value:
                continue
            if kind == "tap" and not tap:
                continue
            if value in seen:
                raise ChvFailure(
                    "net_config_invalid",
                    "CHV_NET_JSON 内 %s 重复：%s" % (kind, value))
            seen.add(value)
        normalized.append({"tap": tap, "mac": mac, "ip": ip,
                           "mask": str(entry.get("mask") or "")})
    for entry in normalized:
        for other in existing_entries:
            for kind in ("tap", "mac", "ip"):
                value = entry.get(kind)
                if value and value == other.get(kind):
                    raise ChvFailure(
                        "net_tap_conflict",
                        "net %s %s 已被 VM %s 使用，拒绝共享"
                        " tap/mac/ip：每 VM 必须独立"
                        % (kind, value, other.get("vm")))
    return normalized


def validate_egress_proxy(egress_proxy: str) -> None:
    """egress 代理地址校验（H1 代码门的一部分）：须为
    http/https URL 且带主机名；否则 net_config_invalid…
    实际 kind 用 config_invalid（preflight 的配置类失败）。
    空值由调用方另行处理（未设 + 未显式允许直连时
    preflight 已有 egress_not_configured）。"""
    if not egress_proxy:
        return None
    try:
        from urllib.parse import urlparse
        parsed = urlparse(egress_proxy)
    except ValueError:
        parsed = None
    if parsed is None or parsed.scheme not in ("http", "https") \
            or not parsed.hostname:
        raise ChvFailure(
            "config_invalid",
            "CHV_EGRESS_PROXY 必须是 http/https 代理地址：%r"
            % (egress_proxy,))
    return egress_proxy


def _is_within(directory, path):
    directory = os.path.realpath(directory)
    path = os.path.realpath(path)
    return path == directory or path.startswith(directory + os.sep)


def validate_snapshot_url(snapshot_url: str, state_dir: str, snapshot_dir=None,
                          allow_remote=False):
    """快照 URL 白名单（L2）：

    - file://：目标须在 state_dir 或 CHV_SNAPSHOT_DIR 内，
      否则 snapshot_path_not_allowed；
    - 其它协议（含无 :// 的裸路径由调用方先补 file://）：
      须显式 CHV_ALLOW_REMOTE_SNAPSHOT=1，否则
      snapshot_remote_not_allowed。
    返回 file:// 目标的本地路径（非 file:// 时返回 None）。"""
    if snapshot_url.startswith("file://"):
        path = snapshot_url[len("file://"):]
        allowed = [state_dir]
        if snapshot_dir:
            allowed.append(snapshot_dir)
        if not any(_is_within(d, path) for d in allowed):
            raise ChvFailure(
                "snapshot_path_not_allowed",
                "file:// 快照须在 state_dir 或 CHV_SNAPSHOT_DIR "
                "内：%s" % snapshot_url)
        return path
    if not allow_remote:
        raise ChvFailure(
            "snapshot_remote_not_allowed",
            "非 file:// 快照须显式设 CHV_ALLOW_REMOTE_SNAPSHOT=1：%s"
            % snapshot_url)
    return None