Thore CimbalandClaude Fable 5
b6007c50fd
- cve-scan: Trivy-Loop ueber die 29 real deployten Images (Cluster-Inventur 2026-08-01 + Prod-Web-Image); 24h-Intervall, Fehler einzelner Images blockieren nicht - cve-exporter: Stdlib-Exporter mit first_seen-State (Zeitstrahl), Schema trivy_vuln_info/_count/_first_seen/_last_scan gemaess Pflichtfeldern - 3 Alertregeln (CRITICAL sofort, HIGH mit 24h-Daempfung, Scan-Frische) - promtool SUCCESS 9 rules; alle mit room=security - matrix-alerts: Label-basiertes Raum-Routing (MATRIX_ROOM_<NAME>), Edits landen im richtigen Raum via State - Grafana-Dashboard cve-overview: Severity-Stats, CVE-Tabelle mit NVD-Link/Fix-Version/first-seen, Zeitstrahl, Verlauf UNGETESTET bis zum Deploy auf CFGMON (compose up -d). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PKhFj1S3UdD6xL2fbWPeYj
107 lines
4.5 KiB
Python
107 lines
4.5 KiB
Python
#!/usr/bin/env python3
|
|
# Minimaler Alertmanager-Webhook -> Matrix-Raum (gitops#32). Gleiche Machart wie
|
|
# maintenance-notify (Issue #24 im gitops-Repo): purer Stdlib-HTTP-Server, Bot-Token
|
|
# aus der Umgebung, Nachricht per Client-Server-API.
|
|
#
|
|
# Ein Alarm = EINE Nachricht: beim Firing wird pro Alarm (Alertmanager-Fingerprint)
|
|
# eine Nachricht gesendet und deren Event-ID gemerkt; beim Resolved wird dieselbe
|
|
# Nachricht per m.replace-Edit durchgestrichen und abgehakt statt eine neue zu
|
|
# posten (Wunsch sorb 2026-08-01: append-only wird unuebersichtlich). Die Zuordnung
|
|
# ueberlebt Container-Restarts via State-Datei; ohne Zuordnung (z.B. nach Neubau)
|
|
# faellt Resolved auf eine eigenstaendige ✅-Nachricht zurueck.
|
|
import html
|
|
import json
|
|
import os
|
|
import urllib.parse
|
|
import urllib.request
|
|
import uuid
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
|
|
HOMESERVER = os.environ["MATRIX_HOMESERVER"]
|
|
ROOM_ID = os.environ["MATRIX_ROOM_ID"]
|
|
TOKEN = os.environ["MATRIX_TOKEN"]
|
|
STATE_FILE = os.environ.get("MATRIX_STATE_FILE", "/tmp/matrix-alerts-state.json")
|
|
# Raum-Routing (gitops#47): Alerts mit Label room="<name>" landen im Raum aus
|
|
# MATRIX_ROOM_<NAME> (z.B. room="security" -> MATRIX_ROOM_SECURITY), alles
|
|
# andere im Default-Raum. Der Bot muss in jedem Zielraum Mitglied sein.
|
|
ROOM_MAP = {k[len("MATRIX_ROOM_"):].lower(): v
|
|
for k, v in os.environ.items()
|
|
if k.startswith("MATRIX_ROOM_") and k != "MATRIX_ROOM_ID" and v}
|
|
|
|
try:
|
|
with open(STATE_FILE) as f:
|
|
state = json.load(f) # fingerprint -> {"event_id":..., "text":...}
|
|
except Exception:
|
|
state = {}
|
|
|
|
|
|
def save_state():
|
|
try:
|
|
with open(STATE_FILE, "w") as f:
|
|
json.dump(state, f)
|
|
except Exception as e:
|
|
print(f"state save failed: {e}", flush=True)
|
|
|
|
|
|
def matrix_put(content, room=ROOM_ID):
|
|
url = (f"{HOMESERVER}/_matrix/client/v3/rooms/{urllib.parse.quote(room)}"
|
|
f"/send/m.room.message/{uuid.uuid4()}")
|
|
req = urllib.request.Request(url, data=json.dumps(content).encode(), method="PUT",
|
|
headers={"Authorization": f"Bearer {TOKEN}",
|
|
"Content-Type": "application/json"})
|
|
return json.loads(urllib.request.urlopen(req, timeout=10).read()).get("event_id")
|
|
|
|
|
|
def send_text(text, room=ROOM_ID):
|
|
return matrix_put({"msgtype": "m.text", "body": text}, room)
|
|
|
|
|
|
def edit_resolved(event_id, old_text, room=ROOM_ID):
|
|
# m.replace-Edit: Original wird in Element in-place ersetzt (durchgestrichen + Haken)
|
|
new_body = f"✅ ~~{old_text}~~"
|
|
new_html = f"✅ <del>{html.escape(old_text)}</del>"
|
|
matrix_put({
|
|
"msgtype": "m.text",
|
|
"body": f"* {new_body}",
|
|
"m.new_content": {"msgtype": "m.text", "body": new_body,
|
|
"format": "org.matrix.custom.html", "formatted_body": new_html},
|
|
"m.relates_to": {"rel_type": "m.replace", "event_id": event_id},
|
|
}, room)
|
|
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def do_POST(self):
|
|
if self.path != "/alert":
|
|
self.send_response(404); self.end_headers(); return
|
|
data = json.loads(self.rfile.read(int(self.headers.get("Content-Length", 0))))
|
|
try:
|
|
for a in data.get("alerts", []):
|
|
name = a.get("labels", {}).get("alertname", "?")
|
|
summary = a.get("annotations", {}).get("summary", "")
|
|
fp = a.get("fingerprint", "")
|
|
room = ROOM_MAP.get(a.get("labels", {}).get("room", ""), ROOM_ID)
|
|
if a.get("status") == "firing":
|
|
text = f"\U0001F534 [firing] {name}: {summary}"
|
|
if fp in state: # re-notify desselben Alarms -> keine Doppelnachricht
|
|
continue
|
|
event_id = send_text(text, room)
|
|
if fp and event_id:
|
|
state[fp] = {"event_id": event_id, "text": text, "room": room}
|
|
else:
|
|
known = state.pop(fp, None)
|
|
if known:
|
|
edit_resolved(known["event_id"], known["text"], known.get("room", ROOM_ID))
|
|
else:
|
|
send_text(f"✅ [resolved] {name}: {summary}", room)
|
|
save_state()
|
|
except Exception as e:
|
|
self.log_message("matrix send failed: %s", e)
|
|
self.send_response(502); self.end_headers(); return
|
|
self.send_response(200); self.end_headers()
|
|
|
|
def log_message(self, fmt, *args):
|
|
print(fmt % args, flush=True)
|
|
|
|
|
|
ThreadingHTTPServer(("0.0.0.0", 8080), Handler).serve_forever()
|