service-catalog/scripts/build/check-rebuild-needed.py

#!/usr/bin/env python3
"""check-rebuild-needed.py — 배포 중인 자체 빌드 이미지가 재빌드로 고쳐지는지 판정한다.

왜 필요한가
-----------
자체 빌드 이미지는 **소스를 안 바꿔도 시간이 지나면 게이트를 벗어난다.** 새 CVE 가 공개되고,
베이스 OS 패키지가 갱신되고, 툴체인 패치가 나오는 동안 우리가 참조해 둔 태그는 그대로다.
의도적으로 보는 장치가 없으면 문서 전용 PR 이 우연히 검증 빌드를 돌릴 때만 발견된다.

`self-build-drift-check.yml` 이 주간으로 이것을 돌린다.

무엇이 "재빌드로 고쳐지는가" 인가
--------------------------------
"핀이 뒤처진 경우" 만으로는 부족하다. 재빌드로 고쳐지는 경우는 셋이다.

    1. 핀이 뒤처졌다              → 핀을 올리고 빌드
    2. 핀은 맞는데 이미지가 그 핀으로 안 빌드됐다 → 그냥 빌드
    3. 베이스 OS 패키지가 뒤처졌다 → 그냥 빌드 (재빌드하면 zypper 가 최신을 깐다)

그래서 트리거는 **"수정 버전이 있는 차단 CVE 가 있는가"** 다. 이 스크립트는 그 판정만
한다 — 핀을 무엇으로 올려야 하는지는 판단하지 않는다(아래 "레포 분리" 참고).

수정 버전이 없는 차단(`affected`·`will_not_fix`)은 재빌드해도 그대로이므로 트리거가 아니다 —
`no-fix` 로 따로 보고해 사람이 다른 레버(태그 교체·예외 승인)를 판단하게 한다.

판정 규칙은 게이트와 같은 출처를 쓴다
-----------------------------------
실효 등급 `max(벤더, NVD)` · 승인 예외(`doc/cve-exceptions.json`) 적용 — 전부
`scripts/pipeline/cve-gate.py` 의 함수를 그대로 가져다 쓴다. 규칙이 두 곳으로 갈리면
"게이트는 PASS 인데 재빌드하라고 한다" 가 나온다.

기준은 카탈로그가 실제로 가리키는 이미지다
----------------------------------------
    catalog/image-map/<image>.env  →  카탈로그 values 의 현재 ref
                                    →  그 ref 의 스캔 리포트

재빌드해 보지 않고도 "지금 배포 중인 것이 규정을 벗어났는가" 를 답한다.

레포 분리 이후 — 이 스크립트는 절반이다
----------------------------------------
커스텀 이미지는 hardened-containers(별도 레포)로 나갔다. "무엇을 배포 중인가"는 이 카탈로그가
알고, "그 이미지를 어떻게 만드는가"(핀 판단·재빌드)는 hardened-containers 가 안다. 이 스크립트는
**탐지만** 한다 — 재빌드가 필요하다고 판단되면 `self-build-drift-check.yml` 이
hardened-containers 의 `build-image.yml` 을 `workflow_dispatch` 로 부른다. 핀을 무엇으로
올릴지는 그 레포의 `suggest-go-upgrades.py --apply` 가 한다.

사용
----
    python3 scripts/build/check-rebuild-needed.py --list-refs          # 무엇을 스캔해야 하나
    python3 scripts/build/check-rebuild-needed.py --reports <디렉토리>

종료 코드
    0  정상 (재빌드 대상이 있어도 0 — --fail-on-drift 를 주면 1)
    1  재빌드 대상 있음 + --fail-on-drift
    2  실행 오류
"""
import argparse
import importlib.util
import json
import os
import pathlib
import re
import sys

HERE = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.normpath(os.path.join(HERE, "..", ".."))
IMAGE_MAP_DIR = os.path.join(ROOT, "catalog", "image-map")
DEFAULT_EXCEPTIONS = os.path.join(ROOT, "doc", "cve-exceptions.json")

# 카탈로그 values 는 이 순서로 찾는다. 자체 빌드 태그는 custom-values.yaml 이 기준이고
# dip-values.yaml 은 같은 값을 쓰는 배포 파라미터화용이다.
VALUES_CANDIDATES = ("custom-values.yaml", "dip-values.yaml")


def load(path, name):
    spec = importlib.util.spec_from_file_location(name, path)
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


PATCH = load(os.path.join(HERE, "patch-catalog-tag.py"), "patch_catalog_tag")
GATE = load(os.path.join(HERE, "..", "pipeline", "cve-gate.py"), "cve_gate")


def read_env(path):
    """`KEY=value` 셸 할당만 읽는다. 값의 따옴표는 벗긴다."""
    out = {}
    if not os.path.isfile(path):
        return out
    for line in pathlib.Path(path).read_text().splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        k, v = line.split("=", 1)
        if not re.match(r"^[A-Z_][A-Z0-9_]*$", k):
            continue
        out[k] = v.strip().strip('"').strip("'")
    return out


def report_path(reports_dir, image_ref):
    """이미지 ref → 리포트 파일 경로. `trivy image <ref>` 스캔 결과를
    `tr '/:@' '___'` 로 만든 파일명으로 저장했다고 가정한다."""
    return os.path.join(reports_dir, re.sub(r"[/:@]", "_", image_ref) + ".json")


def resolve_current_ref(chart_env):
    """카탈로그 values 에서 현재 이미지 ref 를 읽는다. (ref, 읽은 파일) 또는 (None, 이유)."""
    chart_dirs = (chart_env.get("CHART_DIRS") or "").split()
    style = chart_env.get("TAG_STYLE") or ""
    block = chart_env.get("TAG_BLOCK") or "image"
    if not chart_dirs or style not in ("imageName", "split"):
        return None, "catalog/image-map/<image>.env 에 CHART_DIRS/TAG_STYLE 이 없다"
    for d in chart_dirs:
        for fname in VALUES_CANDIDATES:
            p = os.path.join(ROOT, d, fname)
            if not os.path.isfile(p):
                continue
            text = pathlib.Path(p).read_text()
            ref = (PATCH.read_image_name(text) if style == "imageName"
                   else PATCH.read_split_block(text, block))
            if ref:
                return ref, os.path.relpath(p, ROOT)
    return None, f"{chart_dirs} 에서 태그를 읽지 못했다 (style={style}, block={block})"


def blocking_cves(report, image_ref, rank, floor, exceptions):
    """실효 등급 floor 이상 + 승인 예외 제외한 고유 CVE.

    반환: {cve: {"pkg", "sev", "fixable"}}
    """
    out = {}
    for res in report.get("Results") or []:
        for v in res.get("Vulnerabilities") or []:
            cve = v.get("VulnerabilityID")
            if not cve:
                continue
            eff = GATE.effective_severity(
                v.get("Severity"),
                GATE.nvd_severity(((v.get("CVSS") or {}).get("nvd") or {}).get("V3Score")),
            )
            if rank.get(eff, 0) < floor:
                continue
            if any(GATE.exception_applies(e, cve, image_ref) for e in exceptions):
                continue
            prev = out.get(cve)
            fixable = bool(v.get("FixedVersion"))
            if prev is None:
                out[cve] = {"pkg": v.get("PkgName"), "sev": eff, "fixable": fixable}
            elif fixable:
                prev["fixable"] = True
    return out


def check_image(image, reports_dir, rank, floor, exceptions):
    res = {"image": image, "status": "ok", "notes": [], "blocking": 0, "fixable": 0}

    chart_env = read_env(os.path.join(IMAGE_MAP_DIR, f"{image}.env"))
    ref, where = resolve_current_ref(chart_env)
    if not ref:
        res["status"] = "skip"
        res["notes"].append(where)
        return res
    res["ref"] = ref
    res["ref_from"] = where

    rp = report_path(reports_dir, ref)
    if not os.path.isfile(rp):
        res["status"] = "no-report"
        res["notes"].append("이 ref 의 스캔 리포트가 없다 — 스캔 대상에 없었거나 태그가 갱신됐다")
        return res

    with open(rp) as f:
        report = json.load(f)
    cves = blocking_cves(report, ref, rank, floor, exceptions)
    fixable = {c: d for c, d in cves.items() if d["fixable"]}
    res["blocking"] = len(cves)
    res["fixable"] = len(fixable)
    res["packages"] = sorted({d["pkg"] for d in cves.values() if d["pkg"]})

    if not cves:
        return res  # ok

    if fixable:
        res["status"] = "rebuild"
    else:
        res["status"] = "no-fix"
        res["notes"].append("차단 CVE 에 수정 버전이 없다 — 재빌드로 해소되지 않는다. "
                            "상위 태그 교체나 예외 승인을 검토한다")
    return res


def render_md(results):
    rebuild = [r for r in results if r["status"] == "rebuild"]
    nofix = [r for r in results if r["status"] == "no-fix"]
    ok = [r for r in results if r["status"] == "ok"]
    out = ["## 🔁 자체 빌드 이미지 재빌드 판정", ""]

    if not rebuild and not nofix:
        out.append(f"배포 중인 자체 빌드 이미지 {len(ok)}개에 차단 CVE 가 없다.")
    if rebuild:
        out += [f"**{len(rebuild)}개 이미지가 재빌드 대상이다.** 차단 CVE 에 수정 버전이 있다 "
                "— hardened-containers 레포에서 재빌드하면 해소될 가능성이 높다. 핀 조정이 "
                "필요한지는 그 레포의 `suggest-go-upgrades.py` 가 판단한다.", "",
                "| 이미지 | 현재 ref | 차단 | 수정 가능 | 패키지 |",
                "|---|---|---:|---:|---|"]
        for r in rebuild:
            pkgs = ", ".join(f"`{p}`" for p in r["packages"][:4])
            out.append(f"| `{r['image']}` | `{r['ref']}` | {r['blocking']} | {r['fixable']} | {pkgs} |")
    if nofix:
        out += ["", f"**{len(nofix)}개 이미지는 재빌드로 해소되지 않는다.** 다른 레버가 필요하다.",
                "", "| 이미지 | 차단 | 패키지 |", "|---|---:|---|"]
        for r in nofix:
            out.append(f"| `{r['image']}` | {r['blocking']} | "
                       f"{', '.join(f'`{p}`' for p in r['packages'][:4])} |")

    notes = [(r["image"], n) for r in results for n in r["notes"]
             if r["status"] in ("rebuild", "no-fix")]
    if notes:
        out += ["", "**수동 확인 필요**", ""]
        out += [f"- `{img}` — {n}" for img, n in notes]

    skipped = [r for r in results if r["status"] in ("skip", "no-report")]
    if skipped:
        out += ["", f"<details><summary>판정하지 않은 이미지 {len(skipped)}개</summary>", ""]
        out += [f"- `{r['image']}` — {'; '.join(r['notes'])}" for r in skipped]
        out += ["", "</details>"]

    out += ["", "> 판정 기준은 게이트와 같다 — 실효 등급 `max(벤더, NVD)` · 승인 예외 적용 · "
            "기본 HIGH 이상. 재빌드는 hardened-containers 레포의 `build-image.yml` 을 "
            "`workflow_dispatch` 로 부른다.", ""]
    return "\n".join(out)


def main():
    ap = argparse.ArgumentParser(description="배포 중인 자체 빌드 이미지의 재빌드 필요 여부 판정")
    ap.add_argument("--reports", help="trivy 리포트(JSON) 디렉토리 (--list-refs 면 불필요)")
    ap.add_argument("--image", default="", help="이 이미지만 판정 (catalog/image-map/<image>.env)")
    ap.add_argument("--exceptions", default=DEFAULT_EXCEPTIONS,
                    help="승인 예외 목록 (기본 doc/cve-exceptions.json)")
    ap.add_argument("--min-severity", default="HIGH",
                    choices=["CRITICAL", "HIGH", "MEDIUM", "LOW"],
                    help="이 등급 이상만 차단으로 본다 (기본 HIGH — 게이트와 동일)")
    ap.add_argument("--list-refs", action="store_true",
                    help="`이미지<TAB>ref<TAB>리포트파일명` 만 출력하고 종료. CI 가 무엇을 "
                         "스캔해야 하는지 알아야 할 때 쓴다 — ref 해석 규칙이 워크플로로 "
                         "새지 않게 한다")
    ap.add_argument("--summary-md", metavar="FILE", help="마크다운 요약을 이 파일에 쓴다")
    ap.add_argument("--json-out", metavar="FILE", help="결과 JSON 을 이 파일에 쓴다")
    ap.add_argument("--fail-on-drift", action="store_true",
                    help="재빌드 대상이 있으면 1 로 종료 (기본은 보고만 하고 0)")
    args = ap.parse_args()

    if not os.path.isdir(IMAGE_MAP_DIR):
        print(f"::error::catalog/image-map/ 를 찾지 못했다: {IMAGE_MAP_DIR}", file=sys.stderr)
        return 2
    if not args.list_refs and not args.reports:
        print("::error::--reports 가 필요하다", file=sys.stderr)
        return 2
    if args.reports and not os.path.isdir(args.reports):
        print(f"::error::리포트 디렉토리가 없다: {args.reports}", file=sys.stderr)
        return 2

    names = sorted(f[:-4] for f in os.listdir(IMAGE_MAP_DIR) if f.endswith(".env"))
    if args.image:
        names = [n for n in names if n == args.image]
        if not names:
            print(f"::error::catalog/image-map/{args.image}.env 가 없다", file=sys.stderr)
            return 2

    if args.list_refs:
        for n in names:
            ref, why = resolve_current_ref(read_env(os.path.join(IMAGE_MAP_DIR, f"{n}.env")))
            if not ref:
                print(f"::warning::{n}: ref 를 읽지 못했다 — {why}", file=sys.stderr)
                continue
            print(f"{n}\t{ref}\t{os.path.basename(report_path('', ref))}")
        return 0

    rank = GATE.RANK
    floor = rank[args.min_severity]
    # load_exceptions 는 (유효, 만료) 를 낸다. 만료된 예외는 게이트가 인정하지 않으므로
    # 여기서도 인정하지 않는다 — 재검토를 강제하는 것이 그 설계의 의도다.
    exceptions, expired = GATE.load_exceptions(args.exceptions)
    if expired:
        print(f"::warning::만료된 예외 {len(expired)}건은 인정하지 않는다 — "
              f"{', '.join(e.get('id', '?') for e in expired)}")

    results = [check_image(n, args.reports, rank, floor, exceptions) for n in names]

    for r in results:
        if r["status"] == "rebuild":
            print(f"::warning::{r['image']}: 재빌드 대상 — 차단 {r['blocking']}건"
                  f"(수정 가능 {r['fixable']})")
        elif r["status"] == "no-fix":
            print(f"::warning::{r['image']}: 차단 {r['blocking']}건이나 수정 버전 없음 — "
                  "재빌드로 해소되지 않는다")
        for n in r["notes"]:
            if r["status"] in ("rebuild", "no-fix"):
                print(f"::warning::{r['image']}: {n}")

    md = render_md(results)
    print()
    print(md)

    if args.summary_md:
        pathlib.Path(args.summary_md).write_text(md + "\n")
    if args.json_out:
        pathlib.Path(args.json_out).write_text(
            json.dumps(results, ensure_ascii=False, indent=2) + "\n")

    return 1 if ([r for r in results if r["status"] == "rebuild"] and args.fail_on_drift) else 0


if __name__ == "__main__":
    sys.exit(main())