695 lines
14 KiB
Python
695 lines
14 KiB
Python
|
||
#!/usr/bin/env python3
|
||
"""
|
||
Gitea Actions Status Dashboard
|
||
Kompatibel mit Gitea 1.27.x
|
||
Nur Python-Standardbibliothek erforderlich.
|
||
"""
|
||
|
||
import base64
|
||
import datetime as dt
|
||
import json
|
||
import os
|
||
import sys
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
|
||
from zoneinfo import ZoneInfo
|
||
|
||
|
||
# ==========================================
|
||
# Konfiguration
|
||
# ==========================================
|
||
|
||
BASE = os.environ["GITEA_URL"].rstrip("/")
|
||
API = BASE + "/api/v1"
|
||
|
||
TOKEN = os.environ["GITEA_API_TOKEN"]
|
||
|
||
OWNER = os.environ.get("STATUS_OWNER", "VULN")
|
||
STATUS_REPO = os.environ.get("STATUS_REPOSITORY", "status")
|
||
BRANCH = os.environ.get("STATUS_BRANCH", "main")
|
||
|
||
NOW = dt.datetime.now(dt.timezone.utc)
|
||
TIMEZONE = ZoneInfo("Europe/Berlin")
|
||
|
||
PAGE_SIZE = 50
|
||
MAX_PAGES = 5
|
||
|
||
# Markdown-Emoji-Codes
|
||
ICONS = {
|
||
"success": ":green_circle:",
|
||
"failure": ":red_circle:",
|
||
"running": ":large_blue_circle:",
|
||
"waiting": ":yellow_circle:",
|
||
"unknown": ":white_circle:",
|
||
"warning": ":warning:",
|
||
"error": ":x:",
|
||
}
|
||
|
||
|
||
# ==========================================
|
||
# Hilfsfunktionen
|
||
# ==========================================
|
||
|
||
def log(message):
|
||
print(message, flush=True)
|
||
|
||
|
||
def quote(value):
|
||
return urllib.parse.quote(str(value), safe="")
|
||
|
||
|
||
def req(path, method="GET", data=None):
|
||
body = (
|
||
json.dumps(data).encode("utf-8")
|
||
if data is not None
|
||
else None
|
||
)
|
||
|
||
request = urllib.request.Request(
|
||
API + path,
|
||
data=body,
|
||
method=method,
|
||
headers={
|
||
"Authorization": "token " + TOKEN,
|
||
"Accept": "application/json",
|
||
"Content-Type": "application/json",
|
||
"User-Agent": "vuln-status/2.0",
|
||
},
|
||
)
|
||
|
||
try:
|
||
with urllib.request.urlopen(
|
||
request,
|
||
timeout=35
|
||
) as response:
|
||
|
||
payload = response.read()
|
||
|
||
if not payload:
|
||
return {}
|
||
|
||
return json.loads(payload)
|
||
|
||
except urllib.error.HTTPError as exc:
|
||
|
||
message = exc.read(1000).decode(
|
||
errors="replace"
|
||
)
|
||
|
||
raise RuntimeError(
|
||
f"HTTP {exc.code}: {path}: {message}"
|
||
) from exc
|
||
|
||
except urllib.error.URLError as exc:
|
||
|
||
raise RuntimeError(
|
||
f"Verbindungsfehler: {path}: {exc}"
|
||
) from exc
|
||
|
||
|
||
# ==========================================
|
||
# Repository-Dateien
|
||
# ==========================================
|
||
|
||
def content_path(filename):
|
||
|
||
return (
|
||
f"/repos/{quote(OWNER)}/"
|
||
f"{quote(STATUS_REPO)}/contents/"
|
||
f"{urllib.parse.quote(filename, safe='/')}"
|
||
f"?ref={quote(BRANCH)}"
|
||
)
|
||
|
||
|
||
def get_content(filename):
|
||
|
||
obj = req(content_path(filename))
|
||
|
||
content = base64.b64decode(
|
||
obj["content"]
|
||
).decode("utf-8")
|
||
|
||
return content, obj["sha"]
|
||
|
||
|
||
def update_readme(content, sha):
|
||
|
||
path = (
|
||
f"/repos/{quote(OWNER)}/"
|
||
f"{quote(STATUS_REPO)}/contents/README.md"
|
||
)
|
||
|
||
data = {
|
||
"branch": BRANCH,
|
||
"sha": sha,
|
||
"content": base64.b64encode(
|
||
content.encode("utf-8")
|
||
).decode("ascii"),
|
||
"message": (
|
||
"docs: update vulnerability "
|
||
"sync status [skip ci]"
|
||
),
|
||
}
|
||
|
||
return req(path, "PUT", data)
|
||
|
||
|
||
# ==========================================
|
||
# Datum und Laufzeit
|
||
# ==========================================
|
||
|
||
def parse_time(raw):
|
||
|
||
if not raw or str(raw).startswith("0001-"):
|
||
return None
|
||
|
||
try:
|
||
value = dt.datetime.fromisoformat(
|
||
str(raw).replace("Z", "+00:00")
|
||
)
|
||
|
||
if value.tzinfo is None:
|
||
value = value.replace(
|
||
tzinfo=dt.timezone.utc
|
||
)
|
||
|
||
return value.astimezone(dt.timezone.utc)
|
||
|
||
except (ValueError, TypeError):
|
||
return None
|
||
|
||
|
||
def fmt(value):
|
||
|
||
if value is None:
|
||
return "–"
|
||
|
||
return value.astimezone(TIMEZONE).strftime(
|
||
"%d.%m.%Y %H:%M %Z"
|
||
)
|
||
|
||
|
||
def elapsed(start, end):
|
||
|
||
if not start or not end or end < start:
|
||
return "–"
|
||
|
||
seconds = int(
|
||
(end - start).total_seconds()
|
||
)
|
||
|
||
hours, remaining = divmod(seconds, 3600)
|
||
minutes, seconds = divmod(remaining, 60)
|
||
|
||
if hours:
|
||
return f"{hours}h {minutes:02d}m"
|
||
|
||
return f"{minutes}m {seconds:02d}s"
|
||
|
||
|
||
def safe_cell(value):
|
||
|
||
return (
|
||
str(value)
|
||
.replace("|", "\\|")
|
||
.replace("\n", " ")
|
||
.replace("\r", " ")
|
||
)
|
||
|
||
|
||
# ==========================================
|
||
# Gitea Actions API
|
||
# ==========================================
|
||
|
||
def api_runs(owner, repo, page=1):
|
||
|
||
params = urllib.parse.urlencode({
|
||
"page": page,
|
||
"limit": PAGE_SIZE,
|
||
})
|
||
|
||
path = (
|
||
f"/repos/{quote(owner)}/"
|
||
f"{quote(repo)}/actions/runs?{params}"
|
||
)
|
||
|
||
result = req(path)
|
||
|
||
return result.get("workflow_runs", [])
|
||
|
||
|
||
def workflow_matches(run, workflow):
|
||
|
||
if not workflow:
|
||
return True
|
||
|
||
path = str(run.get("path") or "")
|
||
|
||
# Beispiel:
|
||
# .gitea/workflows/sync.yml@refs/heads/main
|
||
|
||
filename = (
|
||
path.split("@", 1)[0]
|
||
.rsplit("/", 1)[-1]
|
||
)
|
||
|
||
return filename == workflow
|
||
|
||
|
||
def get_runs(owner, repo, workflow=None):
|
||
|
||
runs = []
|
||
|
||
for page in range(1, MAX_PAGES + 1):
|
||
|
||
# Erst API-Seite abrufen
|
||
chunk = api_runs(owner, repo, page)
|
||
|
||
if not chunk:
|
||
break
|
||
|
||
# Danach nach Workflow filtern
|
||
matched = [
|
||
run for run in chunk
|
||
if workflow_matches(run, workflow)
|
||
]
|
||
|
||
runs.extend(matched)
|
||
|
||
# Pagination anhand der ungefilterten
|
||
# API-Ergebnisse prüfen.
|
||
if len(chunk) < PAGE_SIZE:
|
||
break
|
||
|
||
runs.sort(
|
||
key=lambda run: (
|
||
parse_time(run.get("created_at"))
|
||
or dt.datetime.min.replace(
|
||
tzinfo=dt.timezone.utc
|
||
),
|
||
run.get("id") or 0,
|
||
),
|
||
reverse=True,
|
||
)
|
||
|
||
return runs
|
||
|
||
|
||
# ==========================================
|
||
# Jobstatus
|
||
# ==========================================
|
||
|
||
def get_state(run):
|
||
|
||
if not run:
|
||
return (
|
||
ICONS["unknown"] + " Kein Lauf"
|
||
)
|
||
|
||
status = str(
|
||
run.get("status") or ""
|
||
).lower()
|
||
|
||
conclusion = str(
|
||
run.get("conclusion") or ""
|
||
).lower()
|
||
|
||
if status in (
|
||
"in_progress",
|
||
"running",
|
||
):
|
||
return ICONS["running"] + " Läuft"
|
||
|
||
if status in (
|
||
"queued",
|
||
"pending",
|
||
"requested",
|
||
"waiting",
|
||
):
|
||
return ICONS["waiting"] + " Wartet"
|
||
|
||
if conclusion == "success":
|
||
return ICONS["success"] + " Erfolgreich"
|
||
|
||
if conclusion in (
|
||
"failure",
|
||
"timed_out",
|
||
"startup_failure",
|
||
):
|
||
return ICONS["failure"] + " Fehlgeschlagen"
|
||
|
||
if conclusion in (
|
||
"cancelled",
|
||
"skipped",
|
||
):
|
||
return (
|
||
ICONS["unknown"]
|
||
+ " "
|
||
+ conclusion.capitalize()
|
||
)
|
||
|
||
return (
|
||
ICONS["unknown"]
|
||
+ " "
|
||
+ (status or "Unbekannt")
|
||
)
|
||
|
||
|
||
def source_status(cfg, source):
|
||
|
||
owner = source.get(
|
||
"owner",
|
||
cfg.get("owner", OWNER)
|
||
)
|
||
|
||
repo = source["repo"]
|
||
|
||
workflow = source.get("workflow")
|
||
|
||
max_age_hours = source.get(
|
||
"max_age_hours",
|
||
cfg.get("max_age_hours", 26)
|
||
)
|
||
|
||
limit_age = dt.timedelta(
|
||
hours=max_age_hours
|
||
)
|
||
|
||
runs = get_runs(
|
||
owner,
|
||
repo,
|
||
workflow
|
||
)
|
||
|
||
latest = runs[0] if runs else None
|
||
|
||
# Letzten erfolgreichen Lauf bestimmen.
|
||
success = next(
|
||
(
|
||
run for run in runs
|
||
if str(
|
||
run.get("conclusion") or ""
|
||
).lower() == "success"
|
||
),
|
||
None
|
||
)
|
||
|
||
latest_success = None
|
||
|
||
if success:
|
||
latest_success = (
|
||
parse_time(success.get("completed_at"))
|
||
or parse_time(success.get("updated_at"))
|
||
or parse_time(success.get("created_at"))
|
||
)
|
||
|
||
# Status ermitteln
|
||
state = get_state(latest)
|
||
|
||
# Überfälligkeit überprüfen
|
||
overdue = (
|
||
latest_success is None
|
||
or NOW - latest_success > limit_age
|
||
)
|
||
|
||
if overdue:
|
||
state += (
|
||
" / "
|
||
+ ICONS["warning"]
|
||
+ " Überfällig"
|
||
)
|
||
|
||
# Laufzeit ermitteln
|
||
started = None
|
||
ended = None
|
||
|
||
if latest:
|
||
|
||
started = (
|
||
parse_time(latest.get("started_at"))
|
||
or parse_time(latest.get("run_started_at"))
|
||
)
|
||
|
||
status = str(
|
||
latest.get("status") or ""
|
||
).lower()
|
||
|
||
if status == "completed":
|
||
|
||
ended = (
|
||
parse_time(latest.get("completed_at"))
|
||
or parse_time(latest.get("updated_at"))
|
||
)
|
||
|
||
if not started:
|
||
started = parse_time(
|
||
latest.get("created_at")
|
||
)
|
||
|
||
duration = elapsed(started, ended)
|
||
|
||
# Workflow-Link
|
||
link = "–"
|
||
|
||
if latest:
|
||
|
||
run_id = latest.get("id")
|
||
|
||
url = latest.get("html_url")
|
||
|
||
if not url and run_id:
|
||
url = (
|
||
f"{BASE}/{quote(owner)}/"
|
||
f"{quote(repo)}/actions/runs/"
|
||
f"{run_id}"
|
||
)
|
||
|
||
if url and run_id:
|
||
link = f"[#{run_id}]({url})"
|
||
|
||
return {
|
||
"state": state,
|
||
"last_success": fmt(latest_success),
|
||
"duration": duration,
|
||
"link": link,
|
||
"overdue": overdue,
|
||
}
|
||
|
||
|
||
# ==========================================
|
||
# README erzeugen
|
||
# ==========================================
|
||
|
||
def build_readme(cfg, rows, stats):
|
||
|
||
timestamp = fmt(NOW)
|
||
|
||
content = [
|
||
"# Vulnerability Database – Sync Status",
|
||
"",
|
||
"Automatisch generiertes Dashboard "
|
||
"für Gitea Actions.",
|
||
"",
|
||
f"**Letzte Prüfung:** {timestamp}",
|
||
"",
|
||
"## Übersicht",
|
||
"",
|
||
f"- **Datenquellen:** {stats['total']}",
|
||
f"- **Erfolgreich:** {stats['success']}",
|
||
f"- **Laufend:** {stats['running']}",
|
||
f"- **Wartend:** {stats['waiting']}",
|
||
f"- **Fehlgeschlagen:** {stats['failed']}",
|
||
f"- **Überfällig:** {stats['overdue']}",
|
||
f"- **API-Fehler:** {stats['errors']}",
|
||
"",
|
||
"## Synchronisierungsstatus",
|
||
"",
|
||
"| Datenquelle | Aktueller Status | "
|
||
"Letzter Erfolg | Dauer letzter Lauf | "
|
||
"Letzter Job |",
|
||
"|---|---|---|---|---|",
|
||
]
|
||
|
||
content.extend(rows)
|
||
|
||
content.extend([
|
||
"",
|
||
"## Statusbeschreibung",
|
||
"",
|
||
f"- {ICONS['success']} **Erfolgreich:** "
|
||
"Letzter Job erfolgreich abgeschlossen.",
|
||
f"- {ICONS['running']} **Läuft:** "
|
||
"Job wird aktuell ausgeführt.",
|
||
f"- {ICONS['waiting']} **Wartet:** "
|
||
"Job wartet auf einen Runner.",
|
||
f"- {ICONS['failure']} **Fehlgeschlagen:** "
|
||
"Letzter Job wurde mit Fehler beendet.",
|
||
f"- {ICONS['warning']} **Überfällig:** "
|
||
"Kein ausreichend aktueller erfolgreicher Lauf.",
|
||
f"- {ICONS['error']} **API-Fehler:** "
|
||
"Repository, Tokenrechte oder API prüfen.",
|
||
"",
|
||
"## Konfiguration",
|
||
"",
|
||
f"- **Standard-Maximalalter:** "
|
||
f"{cfg.get('max_age_hours', 26)} Stunden",
|
||
f"- **Zeitzone:** Europe/Berlin",
|
||
f"- **Workflow-Abfrage:** Gitea REST API",
|
||
f"- **Aktualisierung:** Alle 15 Minuten "
|
||
"oder manuell",
|
||
"",
|
||
"## Links",
|
||
"",
|
||
"- [Vulnerability Database]"
|
||
"(https://vulnerability.apex-project.de)",
|
||
"- [Projektwebsite]"
|
||
"(https://www.apex-project.de)",
|
||
"",
|
||
])
|
||
|
||
return "\n".join(content)
|
||
|
||
|
||
# ==========================================
|
||
# Hauptprogramm
|
||
# ==========================================
|
||
|
||
def main():
|
||
|
||
log("======================================")
|
||
log("Vulnerability Status Dashboard")
|
||
log("======================================")
|
||
|
||
config_text, _ = get_content(
|
||
"sources.json"
|
||
)
|
||
|
||
cfg = json.loads(config_text)
|
||
|
||
rows = []
|
||
errors = []
|
||
|
||
stats = {
|
||
"total": len(cfg["sources"]),
|
||
"success": 0,
|
||
"running": 0,
|
||
"waiting": 0,
|
||
"failed": 0,
|
||
"overdue": 0,
|
||
"errors": 0,
|
||
}
|
||
|
||
for src in cfg["sources"]:
|
||
|
||
owner = src.get(
|
||
"owner",
|
||
cfg.get("owner", OWNER)
|
||
)
|
||
|
||
repo = src["repo"]
|
||
name = src.get("name", repo)
|
||
|
||
url = (
|
||
f"{BASE}/{quote(owner)}/"
|
||
f"{quote(repo)}/actions"
|
||
)
|
||
|
||
try:
|
||
|
||
result = source_status(cfg, src)
|
||
|
||
state = result["state"]
|
||
last_success = result["last_success"]
|
||
duration = result["duration"]
|
||
link = result["link"]
|
||
|
||
if result["overdue"]:
|
||
stats["overdue"] += 1
|
||
|
||
if state.startswith(ICONS["success"]):
|
||
stats["success"] += 1
|
||
|
||
elif state.startswith(ICONS["running"]):
|
||
stats["running"] += 1
|
||
|
||
elif state.startswith(ICONS["waiting"]):
|
||
stats["waiting"] += 1
|
||
|
||
elif state.startswith(ICONS["failure"]):
|
||
stats["failed"] += 1
|
||
|
||
except Exception as exc:
|
||
|
||
state = (
|
||
ICONS["error"] + " API-Fehler"
|
||
)
|
||
|
||
last_success = "–"
|
||
duration = "–"
|
||
link = "–"
|
||
|
||
stats["errors"] += 1
|
||
|
||
errors.append(
|
||
f"{repo}: {exc}"
|
||
)
|
||
|
||
rows.append(
|
||
f"| [{safe_cell(name)}]({url}) "
|
||
f"| {state} "
|
||
f"| {last_success} "
|
||
f"| {duration} "
|
||
f"| {link} |"
|
||
)
|
||
|
||
log(f"{repo}: {state}")
|
||
|
||
# ======================================
|
||
# README aktualisieren
|
||
# ======================================
|
||
|
||
content = build_readme(
|
||
cfg,
|
||
rows,
|
||
stats
|
||
)
|
||
|
||
existing, sha = get_content(
|
||
"README.md"
|
||
)
|
||
|
||
if existing == content:
|
||
|
||
log("README unverändert.")
|
||
|
||
else:
|
||
|
||
update_readme(content, sha)
|
||
|
||
log("README aktualisiert.")
|
||
|
||
# ======================================
|
||
# Fehlerbehandlung
|
||
# ======================================
|
||
|
||
if errors:
|
||
|
||
for error in errors:
|
||
print(
|
||
"ERROR: " + error,
|
||
file=sys.stderr
|
||
)
|
||
|
||
sys.exit(1)
|
||
|
||
log("======================================")
|
||
log("Dashboard erfolgreich aktualisiert")
|
||
log("======================================")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|