Steffen Skui
7336b17f79
Build and push soc-collector / build (push) Successful in 31s
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.
75 lines
2.8 KiB
Python
75 lines
2.8 KiB
Python
#!/usr/bin/env python3
|
|
"""soc-collector — scan an estate and push reports to SOC Center.
|
|
|
|
Runs once and exits (schedule it with cron / a systemd timer / a CI job). Each
|
|
enabled source becomes one report, re-ingested under a stable slug so SOC Center
|
|
versions it, diffs each run, and alerts only on new findings.
|
|
|
|
Configuration is entirely via env vars — see the README. Required: SOC_URL, SOC_TOKEN.
|
|
A source self-skips when its inputs are absent (no SOC_TARGETS -> no nmap; no Docker
|
|
socket -> no Trivy / config-drift), so the same image is safe to run anywhere.
|
|
"""
|
|
import os
|
|
import sys
|
|
|
|
import soc
|
|
from sources import cve, config_drift, surface, tls
|
|
|
|
|
|
def _list(name):
|
|
return [x.strip() for x in os.environ.get(name, "").replace(";", ",").split(",") if x.strip()]
|
|
|
|
|
|
def main():
|
|
url = soc._env("SOC_URL", required=True)
|
|
token = soc._env("SOC_TOKEN", required=True)
|
|
prefix = os.environ.get("SOC_SLUG_PREFIX", "").strip()
|
|
visibility = os.environ.get("SOC_VISIBILITY", "private")
|
|
base_tags = _list("SOC_TAGS") or ["collector"]
|
|
want = set(_list("SOC_SOURCES") or ["surface", "cve", "tls", "config"])
|
|
targets, domains = _list("SOC_TARGETS"), _list("SOC_DOMAINS")
|
|
|
|
def slug(s):
|
|
return f"{prefix}-{s}" if prefix else s
|
|
|
|
plan = []
|
|
if "surface" in want:
|
|
plan.append(("surface", "Attack surface (nmap)", slug("attack-surface"), lambda: surface.run(targets)))
|
|
if "cve" in want:
|
|
plan.append(("cve", "Container CVEs (Trivy)", slug("container-cves"), cve.run))
|
|
if "tls" in want:
|
|
plan.append(("tls", "TLS & web exposure", slug("tls-exposure"), lambda: tls.run(domains)))
|
|
if "config" in want:
|
|
plan.append(("config", "Config drift (Docker)", slug("config-drift"), config_drift.run))
|
|
|
|
print(f"[soc-collector] target={url} sources={sorted(want)}")
|
|
rc = 0
|
|
for key, title, rslug, fn in plan:
|
|
print(f"[soc-collector] source: {key}")
|
|
try:
|
|
data = fn()
|
|
except Exception as e:
|
|
print(f" ! {key} errored: {e}")
|
|
rc = 1
|
|
continue
|
|
if data is None:
|
|
print(f" - {key} skipped (no targets / no Docker socket)")
|
|
continue
|
|
n = len(data.get("findings", []))
|
|
status, resp = soc.post_report(url, token, rslug, title, data,
|
|
visibility=visibility, tags=base_tags + [key])
|
|
if status in (200, 201):
|
|
d = resp.get("diff", {})
|
|
print(f" ✓ {rslug}: {n} finding(s), {'created' if resp.get('created') else 'updated'} "
|
|
f"(+{d.get('added', 0)}/-{d.get('removed', 0)}/~{d.get('changed', 0)}) "
|
|
f"alerted={resp.get('alerted')}")
|
|
else:
|
|
print(f" ! {rslug}: HTTP {status} {resp}")
|
|
rc = 1
|
|
|
|
sys.exit(rc)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|