codex-app-server/chv-guest/check_lockfile.py

#!/usr/bin/env python3
"""校验已构建镜像的 BUILDINFO 供应链清单（事后 lockfile 检查）。

生成端见 generate_lockfile.py（读 BUILDINFO.versions 生成
lockfile.json，版本唯一来源是 build.sh 从客机读回的
dpkg/pip freeze + chromium 目录名 + codex/repo_commit/gt
声明字段，绝不查构建宿主现场、绝不编造版本号）。
本脚本校验 BUILDINFO：versions 非空、每条有版本、contents
全覆盖、tag 与预期一致、钉死的 Chromium 修订号一致、
可选的冻结清单（--frozen JSON，形如
{"dpkg": ["pkg version", ...]}）双向比对，漂移即退出非 0。

用法：
  python3 check_lockfile.py out/BUILDINFO.json [--tag TAG]
      [--frozen frozen.json]
"""
import json
import sys

# 非包名的内容项：版本来自 BUILDINFO 的声明字段，不是 versions 列表。
DECLARED_SOURCES = {
    "codex": "codex_version",
    "codex-code-mode-host": "codex_version",
    "codex-resources/bwrap": "codex_version",
    "/app/host.py": "repo_commit",
    "gt": "gt_version",
    "chv-sync": "repo_commit",
    # 对齐 Muse P2：nspawn 单元启动器 + broker + 浏览器子代理，
    # 均是仓库内脚本，版本随 repo_commit（与 chv-sync 同款）。
    "chv-cell": "repo_commit",
    "chv-broker": "repo_commit",
    "chv-browser": "repo_commit",
    # feat-chv-opt-landing（阶段 1）：数据盘挂载脚本，版本随
    # repo_commit（与 chv-sync 同款）。
    "chv-data-mount": "repo_commit",
}


def _names(entries, pip_style=False):
    """versions 条目 → 小写名集合。dpkg 行是 "name version"，
    pip freeze 行是 "name==version"。"""
    names = set()
    for entry in entries or []:
        text = str(entry)
        if pip_style and "==" in text:
            names.add(text.split("==", 1)[0].lower())
        else:
            parts = text.split(None, 1)
            if parts:
                names.add(parts[0].lower())
    return names


def _check_format(kind, entries):
    """每条须带非空版本：dpkg "name version"，pip "name==version"。"""
    for entry in entries or []:
        text = str(entry)
        if kind == "pip" and "==" in text:
            if not text.split("==", 1)[1].strip():
                print("版本为空的条目：%s %r" % (kind, entry),
                      file=sys.stderr)
                return False
        else:
            parts = text.split()
            if len(parts) < 2 or not parts[1]:
                print("版本为空的条目：%s %r" % (kind, entry),
                      file=sys.stderr)
                return False
    return True


def main(argv):
    if len(argv) < 2:
        print("用法：check_lockfile.py BUILDINFO.json [--tag TAG] "
              "[--frozen frozen.json]", file=sys.stderr)
        return 2
    path = argv[1]
    tag = None
    frozen_path = None
    i = 2
    while i < len(argv):
        if argv[i] == "--tag" and i + 1 < len(argv):
            tag = argv[i + 1]
            i += 2
        elif argv[i] == "--frozen" and i + 1 < len(argv):
            frozen_path = argv[i + 1]
            i += 2
        else:
            print("未知参数：%s" % argv[i], file=sys.stderr)
            return 2
    with open(path) as fh:
        data = json.load(fh)
    if tag is not None and data.get("tag") != tag:
        print("tag 不一致：%s != %s" % (data.get("tag"), tag),
              file=sys.stderr)
        return 1
    versions = data.get("versions") or {}
    if not versions.get("dpkg") or not versions.get("pip"):
        print("BUILDINFO.versions.dpkg/pip 为空，供应链清单缺失",
              file=sys.stderr)
        return 1
    if not _check_format("dpkg", versions.get("dpkg")):
        return 1
    if not _check_format("pip", versions.get("pip")):
        return 1
    # contents 全覆盖：每项须在 dpkg/pip 名单里，或有声明字段版本。
    covered = _names(versions.get("dpkg")) | _names(
        versions.get("pip"), pip_style=True)
    missing = []
    for name in data.get("contents") or []:
        if name.lower() in covered:
            continue
        source = DECLARED_SOURCES.get(name)
        if source and data.get(source):
            continue
        if name.lower() in ("chromium", "chromium-headless-shell") \
                and versions.get("chromium"):
            continue
        missing.append(name)
    if missing:
        print("contents 未覆盖：%s" % sorted(missing), file=sys.stderr)
        return 1
    # feat-chv-closeup（M4）：钉死的 Chromium 修订号须与读回
    # 的缓存目录名一致；未钉（pinned_rev 为空）时不校验。
    pinned = data.get("chromium_pinned_rev")
    if pinned:
        expected = "chromium_headless_shell-%s" % pinned
        if expected not in (versions.get("chromium") or []):
            print("Chromium 修订号与钉死值不符：读回 %s，钉死 %s"
                  % (versions.get("chromium"), expected),
                  file=sys.stderr)
            return 1
    if frozen_path:
        with open(frozen_path) as fh:
            frozen = json.load(fh)
        for kind in ("dpkg", "pip"):
            want = set(frozen.get(kind) or [])
            got = set(versions.get(kind) or [])
            absent = want - got
            if absent:
                print("%s 漂移：冻结清单里有、构建产物里无：%s"
                      % (kind, sorted(absent)), file=sys.stderr)
                return 1
            extra = got - want
            if extra:
                print("%s 漂移：构建产物里有、冻结清单里无：%s"
                      % (kind, sorted(extra)), file=sys.stderr)
                return 1
    print("lockfile 校验通过：tag=%s dpkg=%d pip=%d contents=%d"
          % (data.get("tag"), len(versions["dpkg"]),
             len(versions["pip"]),
             len(data.get("contents") or [])))
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv))