diff --git a/sources/config_drift.py b/sources/config_drift.py index 7d1adab..b4ff26b 100644 --- a/sources/config_drift.py +++ b/sources/config_drift.py @@ -1,15 +1,23 @@ -"""Config-drift source: read-only Docker socket -> 0.0.0.0 binds, privileged, rw-socket mounts.""" +"""Config-drift source: read-only Docker socket -> 0.0.0.0 binds, privileged, rw-socket mounts. + +The designated edge proxy (SOC_EDGE, default 'traefik') is SUPPOSED to own 0.0.0.0:80/443 — +that's its whole job — so its public binds are reported as good, not flagged (false positive).""" +import os + from soc import finding from . import _docker +# Containers that are the public edge: allowed to bind 0.0.0.0 (they front the estate). +EDGE = {x.strip() for x in os.environ.get("SOC_EDGE", "traefik").replace(";", ",").split(",") if x.strip()} + def run(): if not _docker.available(): return None # socket not mounted -> skip conts = _docker.get("/containers/json") or [] - findings = [] + findings, edge_ok = [], [] for c in conts: name = (c.get("Names") or ["?"])[0].lstrip("/") cid = c.get("Id", "")[:12] @@ -25,7 +33,9 @@ def run(): for b in (binds or []): if b.get("HostIp") in ("0.0.0.0", "::", ""): exposed.append(f"{b.get('HostIp') or '0.0.0.0'}:{b.get('HostPort')}->{port}") - if exposed: + if exposed and name in EDGE: + edge_ok.append(f"{name} ({'; '.join(exposed)})") # the edge — public by design + elif exposed: findings.append(finding( f"drift-bind-{name}", "high", f"{name} publishes ports on all interfaces", host=name, evidence="; ".join(exposed), @@ -47,8 +57,11 @@ def run(): impact=["the container can control the host's Docker = host root"], fix=["mount it :ro, or put a scoped socket-proxy in front"])) + good = [f"edge {e} binds 0.0.0.0:80/443 by design (allowed)" for e in edge_ok] + if not findings and not good: + good.append("No 0.0.0.0 binds, privileged, or rw-socket containers") return { "scope": f"{len(conts)} container(s)", "findings": findings, - "good": [] if findings else ["No 0.0.0.0 binds, privileged, or rw-socket containers"], + "good": good, } diff --git a/sources/cve.py b/sources/cve.py index 465527b..acb0e2a 100644 --- a/sources/cve.py +++ b/sources/cve.py @@ -1,8 +1,16 @@ -"""Container-CVE source: Trivy-scan running images, reporting only FIXABLE HIGH/CRITICAL CVEs. +"""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: -`--ignore-unfixed` is the whole point: a CVE with no available patch isn't a to-do item, -it's noise. We report only vulns that have a fix, grouped by package with the version to -bump to — so each finding is a short, actionable list instead of a wall of CVE IDs. + [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 @@ -11,6 +19,9 @@ 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 [] @@ -35,9 +46,42 @@ def _scan(img): 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 # socket not mounted -> skip + return None images = _running_images() if not images: return None @@ -49,7 +93,6 @@ def run(): clean += 1 continue worst = "crit" if any(v.get("Severity") == "CRITICAL" for v in vulns) else "high" - # Group by package: which to bump, installed -> fixed, and how many CVEs it clears. pkgs = {} for v in vulns: p = v.get("PkgName", "?") @@ -60,22 +103,28 @@ def run(): 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)" - findings.append(finding( - f"cve-{img}", worst, - f"{len(vulns)} fixable HIGH/CRITICAL CVE(s) in {img}", host=img, - evidence=evidence, - impact=[f"{len(vulns)} CVEs with an available patch across {len(pkgs)} package(s)"], - fix=[f"bump: {', '.join(p for p, _ in top)}" + (" …" if len(pkgs) > 6 else ""), - f"or pull/rebuild a newer {img.split(':')[0]} base image"])) - findings.sort(key=lambda f: (f["sev"] != "crit", f["title"])) # criticals first + 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 only", + "scope": f"{len(images)} running image(s) — fixable HIGH/CRITICAL, tagged yours/upstream/update", "findings": findings, "good": good, }