Files
soc-collector/sources/cve.py
T
Steffen Skui 7336b17f79
Build and push soc-collector / build (push) Successful in 31s
soc-collector: run-anywhere scanner that pushes reports to SOC Center
A single Docker image, configured entirely by env vars (SOC_URL + SOC_TOKEN +
targets). Runs once and exits; schedule with cron. Four self-skipping sources:
  - surface     nmap open-port/service scan (SOC_TARGETS)
  - cve         Trivy HIGH/CRITICAL scan of running images (Docker socket)
  - tls         cert-expiry / weak-TLS / missing-headers (SOC_DOMAINS)
  - config      0.0.0.0 binds, --privileged, rw docker-socket mounts (Docker socket)
Stable slugs per source so SOC Center versions/diffs/alerts. Pure stdlib;
image adds nmap + trivy. MIT, intended as a public Ethica repo.
2026-06-24 18:17:39 +02:00

55 lines
1.8 KiB
Python

"""Container-CVE source: Trivy-scan the images of running containers (HIGH/CRITICAL)."""
import json
import subprocess
from soc import finding
from . import _docker
SEV = {"CRITICAL": "crit", "HIGH": "high", "MEDIUM": "med", "LOW": "low"}
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 run():
if not _docker.available():
return None # socket not mounted -> skip
images = _running_images()
if not images:
return None
findings = []
for img in images:
try:
proc = subprocess.run(
["trivy", "image", "--quiet", "--scanners", "vuln",
"--severity", "HIGH,CRITICAL", "--format", "json", img],
capture_output=True, text=True, timeout=900)
rep = json.loads(proc.stdout or "{}")
except Exception:
continue
vulns = [v for res in (rep.get("Results") or []) for v in (res.get("Vulnerabilities") or [])]
if not vulns:
continue
worst = "crit" if any(v.get("Severity") == "CRITICAL" for v in vulns) else "high"
ids = sorted({v.get("VulnerabilityID", "") for v in vulns if v.get("VulnerabilityID")})
findings.append(finding(
f"cve-{img}", worst, f"{len(vulns)} HIGH/CRITICAL CVEs in {img}", host=img,
evidence=", ".join(ids[:10]) + ("" if len(ids) > 10 else ""),
impact=["known-exploitable vulnerabilities in a running image"],
fix=[f"pull/rebuild a patched {img.split(':')[0]} image"]))
return {
"scope": f"{len(images)} running image(s)",
"findings": findings,
"good": [] if findings else ["No HIGH/CRITICAL CVEs in running images"],
}