cc-ci-orchestrator/cc-ci-plan/audit-advisory-scan.py

#!/usr/bin/env python3
"""Independent audit of advisory-scan's counts.

Deliberately does NOT reuse the scanner's classifier. Re-parses patched versions with a separate
semver implementation and re-derives membership, then diffs against what the scanner concluded.
Anything the two disagree on is a miscategorization in one of them.
"""
import importlib.util, json, re, sys

spec = importlib.util.spec_from_file_location("A", "/srv/cc-ci-orch/cc-ci-plan/advisory-scan.py")
A = importlib.util.module_from_spec(spec); spec.loader.exec_module(A)
REG = "/srv/cc-ci-orch/cc-ci-plan/upstream"


def sv(s):
    """Independent semver parse: strict 3-tuple, missing parts are 0."""
    m = re.match(r"^\s*v?(\d+)(?:\.(\d+))?(?:\.(\d+))?", s or "")
    if not m:
        return None
    return tuple(int(x) if x else 0 for x in m.groups())


def in_window(f, t, patched_expr):
    """Independent membership: any patched token strictly above f and at most t."""
    kf, kt = sv(f), sv(t)
    for tok in re.findall(r"\d+(?:\.\d+)*", patched_expr or ""):
        c = sv(tok)
        if c and kf and kt and kf < c <= kt:
            return True
    return False


_RELS = {}


def fetch_releases(owner_repo):
    """Independent releases fetch — deliberately NOT the scanner's cache or pagination helper."""
    if owner_repo in _RELS:
        return _RELS[owner_repo]
    import urllib.request
    tok = None
    try:
        tok = open("/srv/cc-ci/.github-token").read().strip()
    except OSError:
        pass
    h = {"User-Agent": "audit", "Accept": "application/vnd.github+json"}
    if tok:
        h["Authorization"] = f"Bearer {tok}"
    out, url, pages = [], f"https://api.github.com/repos/{owner_repo}/releases?per_page=100", 0
    while url and pages < 4:
        req = urllib.request.Request(url, headers=h)
        with urllib.request.urlopen(req, timeout=45) as r:
            rows = json.load(r)
            link = r.headers.get("Link", "") or ""
        out += [(x.get("tag_name") or "", (x.get("body") or "") + " " + (x.get("name") or ""))
                for x in rows]
        url = None
        for part in link.split(","):
            if 'rel="next"' in part:
                url = part.split(";")[0].strip().strip("<>")
        pages += 1
    _RELS[owner_repo] = out
    return out


def audit(recipe, vf, vt, images=None, label=""):
    rep = A.scan(recipe, vf, vt, REG, images)
    print(f"\n{'='*78}\n{recipe} {vf} → {vt} {label}\n{'='*78}")
    print(f"scanner count = {rep['cve_count_fixed']}  known={rep['count_known']}  "
          f"failed_sources={rep['sources_failed']}")
    counted = set(rep["fixed_by_this_upgrade"])

    by_src = {}
    for cve, e in rep["cves"].items():
        by_src.setdefault(e["sources"][0], []).append((cve, e))

    windows = rep["windows"]
    total_mismatch = 0
    for src, win in windows.items():
        f, t = win["from"], win["to"]
        method = rep["classified_by"][src]
        rows = by_src.get(src, [])
        scanner_here = {c for c, e in rows if c in counted}
        print(f"\n  ── {src}  ({f} → {t})  method={method}  advisories={len(rows)}")
        if "publish date" in method:
            dw = rep["date_window"][src]
            indep = {c for c, e in rows
                     if e.get("published_at") and dw["from"] < e["published_at"] <= dw["to"]}
            print(f"     date window {dw['from'][:10]} → {dw['to'][:10]}")
        else:
            indep = {c for c, e in rows if in_window(f, t, e.get("patched"))}
            # Independently redo the release-note method: fetch the repo's releases ourselves and
            # confirm a tag NAMING the CVE really does fall inside (f, t].
            kf, kt = sv(f), sv(t)
            owner_repo = src.split(":", 1)[1]
            rels = fetch_releases(owner_repo)
            for c, e in rows:
                if c in indep:
                    continue
                naming = [tag for tag, body in rels if c in body]
                if any(kf < sv(tag) <= kt for tag in naming if sv(tag)):
                    indep.add(c)
        missed = indep - scanner_here
        extra = scanner_here - indep
        print(f"     scanner counted {len(scanner_here)} | independent {len(indep)}"
              f" | missed_by_scanner {len(missed)} | over_counted {len(extra)}")
        if missed:
            print(f"     !! MISSED: {sorted(missed)}")
            for c in sorted(missed):
                print(f"        {c} patched={dict(rows)[c].get('patched')!r}")
        if extra:
            print(f"     !! OVER-COUNTED: {sorted(extra)}")
            for c in sorted(extra):
                print(f"        {c} patched={dict(rows)[c].get('patched')!r}")
        total_mismatch += len(missed) + len(extra)

    # Anything counted that belongs to NO window would be a leak.
    leaked = {c for c in counted if not any(s in rep["cves"][c]["sources"] for s in windows)}
    if leaked:
        print(f"\n  !! COUNTED BUT OUTSIDE EVERY WINDOW: {sorted(leaked)}")
        total_mismatch += len(leaked)

    # Unclassified entries that belong to a WINDOWED source would mean a judged CVE was dropped.
    dropped = [c for c in rep["unclassified"]
               if any(s in rep["cves"][c]["sources"] for s in windows)]
    if dropped:
        print(f"\n  !! UNCLASSIFIED DESPITE HAVING A WINDOW: {sorted(dropped)[:10]}")
        total_mismatch += len(dropped)

    unwindowed = {}
    for cve in rep["unclassified"]:
        unwindowed.setdefault(rep["cves"][cve]["sources"][0], []).append(cve)
    if unwindowed:
        print("\n  unclassified by source (expected: images with no --image given):")
        for s, cs in sorted(unwindowed.items()):
            print(f"     {len(cs):4d}  {s}")

    print(f"\n  VERDICT: {'CLEAN' if total_mismatch == 0 else f'{total_mismatch} DISAGREEMENTS'}")
    return total_mismatch, rep


if __name__ == "__main__":
    bad = 0
    bad += audit("gitea", "1.27.0", "1.27.1")[0]
    bad += audit("discourse", "3.5.3", "2026.7.1")[0]
    bad += audit("discourse", "3.5.3", "2026.7.1", [("redis", "7.4", "8.10")], "+redis sidecar")[0]
    bad += audit("keycloak", "26.7.0", "26.7.1")[0]
    bad += audit("mailu", "2024.06.55", "2024.06.57", [("redis", "8.8.0", "8.10.0")], "+redis")[0]
    bad += audit("n8n", "1.123.0", "2.18.1")[0]
    print(f"\n\n{'#'*78}\nOVERALL: {'CLEAN — no disagreements' if bad == 0 else f'{bad} DISAGREEMENTS'}")
    sys.exit(1 if bad else 0)