Thore CimbalandClaude Fable 5
f45e01219b
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PKhFj1S3UdD6xL2fbWPeYj
100 lines
4.0 KiB
Python
100 lines
4.0 KiB
Python
#!/usr/bin/env python3
|
|
# Release-/Advisory-Watch fuer den Element-Stack (gitops#22). Gleiche Machart wie
|
|
# matrix-alerts.py: purer Stdlib-Daemon, Bot-Token aus der Umgebung, Nachricht per
|
|
# Client-Server-API in den Alerts-Raum.
|
|
#
|
|
# Quelle sind die oeffentlichen GitHub-Release-Atom-Feeds (kein API-Token noetig).
|
|
# Element veroeffentlicht Security-Fixes als Releases - der Feed ist damit der
|
|
# praktikable Advisory-Kanal; echte GHSA-Advisories haben keinen oeffentlichen Feed.
|
|
# Neue Eintraege werden einmalig als 📦-Notiz gemeldet; Security-verdaechtige
|
|
# Titel/Inhalte (CVE/security/vulnerab...) bekommen 🚨 und stehen vorn.
|
|
#
|
|
# Abgrenzung (Frage sorb 2026-08-01): REPOS unten ist eine HANDGEPFLEGTE Liste -
|
|
# de facto ein Mini-SBOM auf Repo-Granularitaet, ohne Versions-/Dependency-Wissen.
|
|
# Bei Stack-Aenderungen (neue Komponente, Fork-Wechsel) muss sie mitgezogen werden.
|
|
# Trivy (gitops#31/#47) braucht dagegen keine Pflege: es leitet sein SBOM selbst
|
|
# aus den Images ab. Beide ergaenzen sich, ersetzen sich nicht.
|
|
import json
|
|
import os
|
|
import re
|
|
import time
|
|
import urllib.parse
|
|
import urllib.request
|
|
import uuid
|
|
import xml.etree.ElementTree as ET
|
|
|
|
HOMESERVER = os.environ["MATRIX_HOMESERVER"]
|
|
# Eigener CVE-/Release-Raum (Entscheidung sorb 2026-08-01, gitops#47);
|
|
# Fallback auf den Alerts-Raum, solange der Bot dort noch nicht eingeladen ist.
|
|
ROOM_ID = os.environ.get("MATRIX_RELEASE_ROOM_ID") or os.environ["MATRIX_ROOM_ID"]
|
|
TOKEN = os.environ["MATRIX_TOKEN"]
|
|
STATE_FILE = os.environ.get("RELEASE_WATCH_STATE_FILE", "/state/release-watch.json")
|
|
INTERVAL = int(os.environ.get("RELEASE_WATCH_INTERVAL", "21600")) # 6h
|
|
|
|
REPOS = [
|
|
"element-hq/synapse",
|
|
"element-hq/ess-helm",
|
|
"element-hq/element-web",
|
|
"element-hq/matrix-authentication-service",
|
|
"element-hq/element-call",
|
|
]
|
|
SECURITY_RE = re.compile(r"cve|security|vulnerab|advisory", re.IGNORECASE)
|
|
ATOM = "{http://www.w3.org/2005/Atom}"
|
|
|
|
try:
|
|
with open(STATE_FILE) as f:
|
|
seen = json.load(f) # repo -> [entry ids]
|
|
except Exception:
|
|
seen = {}
|
|
|
|
|
|
def send_notice(text):
|
|
url = (f"{HOMESERVER}/_matrix/client/v3/rooms/{urllib.parse.quote(ROOM_ID)}"
|
|
f"/send/m.room.message/{uuid.uuid4()}")
|
|
req = urllib.request.Request(url, data=json.dumps({"msgtype": "m.notice", "body": text}).encode(),
|
|
method="PUT", headers={"Authorization": f"Bearer {TOKEN}",
|
|
"Content-Type": "application/json"})
|
|
urllib.request.urlopen(req, timeout=10).read()
|
|
|
|
|
|
def check(repo):
|
|
feed = urllib.request.urlopen(f"https://github.com/{repo}/releases.atom", timeout=20).read()
|
|
root = ET.fromstring(feed)
|
|
entries = root.findall(f"{ATOM}entry")
|
|
known = set(seen.get(repo, []))
|
|
first_run = repo not in seen
|
|
new = []
|
|
for e in entries:
|
|
eid = e.findtext(f"{ATOM}id", "")
|
|
title = e.findtext(f"{ATOM}title", "?").strip()
|
|
link = ""
|
|
le = e.find(f"{ATOM}link")
|
|
if le is not None:
|
|
link = le.get("href", "")
|
|
content = e.findtext(f"{ATOM}content", "") or ""
|
|
if eid and eid not in known:
|
|
new.append((eid, title, link, bool(SECURITY_RE.search(title + " " + content[:2000]))))
|
|
# Erstlauf: nur Stand merken, nicht den Raum mit Historie fluten
|
|
seen[repo] = [e.findtext(f"{ATOM}id", "") for e in entries][:30]
|
|
if first_run:
|
|
return
|
|
for eid, title, link, is_sec in new:
|
|
icon = "\U0001F6A8" if is_sec else "\U0001F4E6"
|
|
kind = "Security-verdaechtiges Release" if is_sec else "Neues Release"
|
|
send_notice(f"{icon} {kind}: {repo} — {title}\n{link}")
|
|
|
|
|
|
while True:
|
|
for repo in REPOS:
|
|
try:
|
|
check(repo)
|
|
except Exception as exc:
|
|
print(f"{repo}: {exc}", flush=True)
|
|
try:
|
|
os.makedirs(os.path.dirname(STATE_FILE), exist_ok=True)
|
|
with open(STATE_FILE, "w") as f:
|
|
json.dump(seen, f)
|
|
except Exception as exc:
|
|
print(f"state: {exc}", flush=True)
|
|
time.sleep(INTERVAL)
|