soc-collector/sources/cve.py

"""Container-CVE source: Trivy-scan running images for FIXABLE HIGH/CRITICAL CVEs,
and — critically — label each with an UPDATE STATUS so the report distinguishes
what's yours to fix from what's the maintainer's:

  [yours]    git.skui.io/* image  -> bump deps in its repo + rebuild (CI). Actionable.
  [update]   third-party, a NEWER image is published -> pull + recreate. Actionable.
  [upstream] third-party, already on the latest published image -> clears only when
             the maintainer rebuilds. NOT actionable here; re-pull periodically.

"Fixable" (Trivy --ignore-unfixed) only means a patched package version EXISTS — not
that you can apply it. For an image you don't build and already run :latest of, the
"bump X / rebuild base" advice is the maintainer's, not yours. The status makes that
explicit, and actionable findings sort above the can't-do-anything ones.
"""
import json
import subprocess

from soc import finding

from . import _docker

OWN_PREFIX = "git.skui.io/"
STATUS_RANK = {"yours": 0, "update": 1, "upstream": 2}


def _running_images():
    conts = _docker.get("/containers/json") or []
    seen = []
    for c in conts:
        img = c.get("Image", "")
        if img and img not in seen:
            seen.append(img)
    return seen


def _scan(img):
    """Fixable HIGH/CRITICAL vulns for one image (empty list on error)."""
    try:
        proc = subprocess.run(
            ["trivy", "image", "--quiet", "--scanners", "vuln", "--ignore-unfixed",
             "--severity", "HIGH,CRITICAL", "--format", "json", img],
            capture_output=True, text=True, timeout=900)
        rep = json.loads(proc.stdout or "{}")
    except Exception:
        return []
    return [v for res in (rep.get("Results") or []) for v in (res.get("Vulnerabilities") or [])]


def _registry_digest(img):
    """The registry's current digest for img's tag, via the daemon (no pull). '' on error."""
    try:
        d = _docker.get(f"/distribution/{img}/json") or {}
        return (d.get("Descriptor") or {}).get("digest", "")
    except Exception:
        return ""


def _local_digests(img):
    try:
        info = _docker.get(f"/images/{img}/json") or {}
        return [r.split("@")[-1] for r in (info.get("RepoDigests") or [])]
    except Exception:
        return []


def _classify(img, pkgs):
    """Return (status, fix_lines) — is this image actionable, and how?"""
    bump = ", ".join(pkgs[:6]) + (" …" if len(pkgs) > 6 else "")
    if img.startswith(OWN_PREFIX):
        return "yours", [f"YOURS — bump {bump} in this repo, then rebuild (CI)"]
    reg = _registry_digest(img)
    if reg and reg in _local_digests(img):
        return "upstream", [
            "UPSTREAM — you already run the latest published image; this clears only when "
            f"the maintainer rebuilds (with {bump}). Re-pull periodically; nothing to do here."]
    if reg:
        return "update", ["UPDATE AVAILABLE — a newer image is published; pull + recreate to get it"]
    return "upstream", [
        f"UPSTREAM — third-party image; the fix ({bump}) comes from the maintainer's rebuild"]


def run():
    if not _docker.available():
        return None
    images = _running_images()
    if not images:
        return None

    findings, clean = [], 0
    for img in images:
        vulns = _scan(img)
        if not vulns:
            clean += 1
            continue
        worst = "crit" if any(v.get("Severity") == "CRITICAL" for v in vulns) else "high"
        pkgs = {}
        for v in vulns:
            p = v.get("PkgName", "?")
            d = pkgs.setdefault(p, {"inst": v.get("InstalledVersion", "?"),
                                    "fix": v.get("FixedVersion", "?"), "n": 0})
            d["n"] += 1
        top = sorted(pkgs.items(), key=lambda kv: -kv[1]["n"])[:6]
        evidence = "; ".join(f"{p} {d['inst']}→{d['fix']} ({d['n']})" for p, d in top)
        if len(pkgs) > 6:
            evidence += f"; +{len(pkgs) - 6} more package(s)"

        status, fix = _classify(img, [p for p, _ in sorted(pkgs.items(), key=lambda kv: -kv[1]["n"])])
        f = finding(
            f"cve-{img}", worst,
            f"[{status}] {len(vulns)} fixable HIGH/CRITICAL CVE(s) in {img}", host=img,
            evidence=evidence,
            impact=[f"{len(vulns)} CVEs with a published patch across {len(pkgs)} package(s)"],
            fix=fix)
        f["_status"] = status
        findings.append(f)

    # actionable first (yours, then update), then upstream; criticals first within each
    sev = {"crit": 0, "high": 1, "med": 2, "low": 3}
    findings.sort(key=lambda f: (STATUS_RANK.get(f.pop("_status", "upstream"), 2), sev.get(f["sev"], 9)))

    good = []
    if clean:
        good.append(f"{clean} of {len(images)} image(s) have no fixable HIGH/CRITICAL CVEs")
    if not findings and not good:
        good.append("No fixable HIGH/CRITICAL CVEs in any running image")
    return {
        "scope": f"{len(images)} running image(s) — fixable HIGH/CRITICAL, tagged yours/upstream/update",
        "findings": findings,
        "good": good,
    }