diff --git a/scripts/update_status.py b/scripts/update_status.py index c3bad0f..8fddc5d 100644 --- a/scripts/update_status.py +++ b/scripts/update_status.py @@ -1,5 +1,11 @@ + #!/usr/bin/env python3 -"""Gitea Actions status dashboard. Standard library only; reads config from same repo.""" +""" +Gitea Actions Status Dashboard +Kompatibel mit Gitea 1.27.x +Nur Python-Standardbibliothek erforderlich. +""" + import base64 import datetime as dt import json @@ -8,180 +14,681 @@ import sys import urllib.error import urllib.parse import urllib.request + from zoneinfo import ZoneInfo -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') + +# ========================================== +# 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:", +} -def req(path, method='GET', data=None): - body = json.dumps(data).encode() if data is not None else None +# ========================================== +# 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/1.0'}) - try: - with urllib.request.urlopen(request, timeout=35) as r: - return json.load(r) - except urllib.error.HTTPError as e: - if e.code == 404: - raise FileNotFoundError(f'HTTP 404: {path}') from e - err = e.read(500).decode(errors='replace') - raise RuntimeError(f'HTTP {e.code} {path}: {err}') from e + 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 '/repos/{}/{}/contents/{}?ref={}'.format( - urllib.parse.quote(OWNER, safe=''), - urllib.parse.quote(STATUS_REPO, safe=''), - urllib.parse.quote(filename, safe='/'), - urllib.parse.quote(BRANCH, safe='')) + + 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)) - return base64.b64decode(obj['content']).decode('utf-8'), obj['sha'] + 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 raw.startswith('0001-'): + + if not raw or str(raw).startswith("0001-"): return None + try: - return dt.datetime.fromisoformat(raw.replace('Z', '+00:00')).astimezone(dt.timezone.utc) - except (ValueError, AttributeError): + 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 '–' - local = value.astimezone(ZoneInfo('Europe/Berlin')) - return local.strftime('%d.%m.%Y %H:%M %Z') + 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 '–' - sec = int((end - start).total_seconds()) - return f'{sec // 3600}h {sec % 3600 // 60}m' if sec >= 3600 else f'{sec // 60}m {sec % 60:02d}s' + 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(text): - return str(text).replace('|', '\\|').replace('\n', ' ') +def safe_cell(value): + + return ( + str(value) + .replace("|", "\\|") + .replace("\n", " ") + .replace("\r", " ") + ) -def api_runs(owner, repo, workflow, page, limit=50): - # Gitea 1.27: Repository-level endpoint; avoids 404 caused by - # guessed/nonexistent workflow filenames. - path = '/repos/{}/{}/actions/runs?{}' .format( - urllib.parse.quote(owner, safe=''), - urllib.parse.quote(repo, safe=''), - urllib.parse.urlencode({'page': page, 'limit': limit})) +# ========================================== +# 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) - runs = result.get('workflow_runs', []) - if workflow: - # Gitea run.path is typically .gitea/workflows/@. - # Restrict to configured workflow, never show a different workflow as success. - def matches(run): - name = str(run.get('path') or '').split('@', 1)[0].rsplit('/', 1)[-1] - return name == workflow - runs = [run for run in runs if matches(run)] + + 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 -def source_status(cfg, source): - owner = source.get('owner', cfg['owner']) - repo = source['repo'] - workflow = source.get('workflow') - limit_age = dt.timedelta(hours=source.get('max_age_hours', cfg.get('max_age_hours', 26))) - all_runs = [] - # Up to 250 runs; sufficient for typical daily sources, can be adjusted. - for page in range(1, 6): - chunk = api_runs(owner, repo, workflow, page) - all_runs.extend(chunk) - if len(chunk) < 50: - break - all_runs.sort(key=lambda r: (parse_time(r.get('created_at')) or dt.datetime.min.replace(tzinfo=dt.timezone.utc), r.get('id') or 0), reverse=True) - latest = all_runs[0] if all_runs else None - success = next((r for r in all_runs if (r.get('conclusion') or '').lower() == 'success'), None) - latest_success = parse_time(success.get('completed_at')) or parse_time(success.get('created_at')) if success else None - if not latest: - state = '⚪ Kein Lauf' - elif (latest.get('status') or '').lower() in ('in_progress', 'running'): - state = '🔵 Läuft' - elif (latest.get('status') or '').lower() in ('queued', 'pending', 'requested', 'waiting'): - state = '🟡 Wartet' - elif (latest.get('conclusion') or '').lower() == 'success': - state = '🟢 Erfolgreich' - elif (latest.get('conclusion') or '').lower() in ('failure', 'timed_out', 'startup_failure'): - state = '🔴 Fehlgeschlagen' - elif (latest.get('conclusion') or '').lower() in ('cancelled', 'skipped'): - state = '⚪ ' + latest['conclusion'] - else: - state = '🟡 ' + str(latest.get('status') or 'Unbekannt') - if latest_success is None or NOW - latest_success > limit_age: - state += ' / ⚠️ Überfällig' - started = parse_time(latest.get('started_at')) if latest else None - ended = parse_time(latest.get('completed_at')) if latest else None - latest_url = latest.get('html_url') if latest else None - link = f'[#{latest["id"]}]({latest_url})' if latest_url and latest else '–' - return state, fmt(latest_success), elapsed(started, ended), link +# ========================================== +# 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(): - config_text, _ = get_content('sources.json') + + log("======================================") + log("Vulnerability Status Dashboard") + log("======================================") + + config_text, _ = get_content( + "sources.json" + ) + cfg = json.loads(config_text) + rows = [] errors = [] - for src in cfg['sources']: - owner = src.get('owner', cfg['owner']) - repo = src['repo'] - workflow = src.get('workflow') - name = src['name'] - url = BASE + '/' + urllib.parse.quote(owner) + '/' + urllib.parse.quote(repo) + '/actions' + + 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: - state, last_success, duration, link = source_status(cfg, src) + + 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, last_success, duration, link = '❌ API-Fehler', '–', '–', '–' - errors.append(f'{repo}: {exc}') - rows.append(f'| [{safe_cell(name)}]({url}) | {state} | {last_success} | {duration} | {link} |') - print(f'{repo}: {state}', flush=True) - content = ('# Vulnerability Database – Sync Status\n\n' - 'Automatisch generiertes Dashboard für Gitea Actions.\n\n' - f'**Letzte Prüfung:** {NOW.strftime("%d.%m.%Y %H:%M UTC")} \n' - f'**Überfällig:** Kein erfolgreicher Lauf innerhalb von {cfg.get("max_age_hours", 26)} Stunden ' - '(pro Quelle anpassbar).\n\n' - '| Datenquelle | Aktueller Status | Letzter Erfolg | Dauer letzter Lauf | Letzter Job |\n' - '|---|---|---|---|---|\n' + '\n'.join(rows) + '\n\n' - '## Hinweise\n\n' - '- 🔴 Fehlgeschlagen: Der neueste Workflow ist gescheitert.\n' - '- ⚠️ Überfällig: Es gibt keinen ausreichend aktuellen erfolgreichen Lauf, auch wenn der letzte alte Lauf grün war.\n' - '- 🔵 Läuft / 🟡 Wartet: Der neueste Lauf ist noch nicht abgeschlossen.\n' - '- ❌ API-Fehler: Repository, Tokenrechte oder API prüfen.\n' - '- Die Dauer bezieht sich auf den zuletzt gestarteten Lauf, soweit Start- und Endzeit vorliegen.\n' - '- Status und Zeitangaben werden alle 15 Minuten oder manuell aktualisiert.\n\n' - '## Links\n\n' - '- [Vulnerability Database](https://vulnerability.apex-project.de)\n' - '- [Projektwebsite](https://www.apex-project.de)\n') - existing, sha = get_content('README.md') + + 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: - print('README unverändert.') + + log("README unverändert.") + else: - path = '/repos/{}/{}/contents/README.md'.format(urllib.parse.quote(OWNER), urllib.parse.quote(STATUS_REPO)) - req(path, 'PUT', {'branch': BRANCH, 'sha': sha, - 'content': base64.b64encode(content.encode()).decode(), - 'message': 'docs: update vulnerability sync status [skip ci]'}) - print('README aktualisiert.') + + update_readme(content, sha) + + log("README aktualisiert.") + + # ====================================== + # Fehlerbehandlung + # ====================================== + if errors: + for error in errors: - print('ERROR: ' + error, file=sys.stderr) + print( + "ERROR: " + error, + file=sys.stderr + ) + sys.exit(1) + log("======================================") + log("Dashboard erfolgreich aktualisiert") + log("======================================") -if __name__ == '__main__': + +if __name__ == "__main__": main()