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.
41 lines
1.5 KiB
Python
41 lines
1.5 KiB
Python
"""Minimal SOC Center API client + report helpers (stdlib only)."""
|
|
import datetime
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
|
|
def _env(name, default=None, required=False):
|
|
v = os.environ.get(name, default)
|
|
if required and not v:
|
|
sys.exit(f"[soc-collector] missing required env var: {name}")
|
|
return v
|
|
|
|
|
|
def today():
|
|
return datetime.datetime.utcnow().strftime("%Y-%m-%d")
|
|
|
|
|
|
def finding(fid, sev, title, host="", evidence="", impact=None, fix=None):
|
|
"""One finding in the SOC schema. sev in crit|high|med|low."""
|
|
return {"id": fid, "sev": sev, "title": title, "host": host,
|
|
"evidence": evidence, "impact": impact or [], "fix": fix or []}
|
|
|
|
|
|
def post_report(url, token, slug, title, data, visibility="private", tags=None, date=None):
|
|
"""POST a report to SOC Center. Returns (http_status, json_body)."""
|
|
body = {"slug": slug, "title": title, "date": date or today(),
|
|
"visibility": visibility, "tags": tags or [], "data": data}
|
|
req = urllib.request.Request(
|
|
url.rstrip("/") + "/api/reports", data=json.dumps(body).encode(), method="POST",
|
|
headers={"Authorization": "Bearer " + token, "Content-Type": "application/json"})
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=60) as r:
|
|
return r.status, json.loads(r.read().decode() or "{}")
|
|
except urllib.error.HTTPError as e:
|
|
return e.code, {"error": e.read().decode()}
|
|
except Exception as e: # network / DNS / timeout
|
|
return 0, {"error": str(e)}
|