diff --git a/scripts/update_status.py b/scripts/update_status.py index 8fddc5d..d919aca 100644 --- a/scripts/update_status.py +++ b/scripts/update_status.py @@ -1,11 +1,5 @@ - #!/usr/bin/env python3 -""" -Gitea Actions Status Dashboard -Kompatibel mit Gitea 1.27.x -Nur Python-Standardbibliothek erforderlich. -""" - +"""Gitea Actions dashboard with self-hosted SVG status icons (stdlib only).""" import base64 import datetime as dt import json @@ -14,681 +8,269 @@ 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") - +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") +TZ = ZoneInfo('Europe/Berlin') +PAGE_SIZE, MAX_PAGES = 50, 5 -PAGE_SIZE = 50 -MAX_PAGES = 5 - -# Markdown-Emoji-Codes +# The URLs use Gitea's raw file route; no externally hosted graphics. ICONS = { - "success": ":green_circle:", - "failure": ":red_circle:", - "running": ":large_blue_circle:", - "waiting": ":yellow_circle:", - "unknown": ":white_circle:", - "warning": ":warning:", - "error": ":x:", + 'success': ('#208637', 'check'), + 'failure': ('#d1242f', 'cross'), + 'running': ('#0969da', 'play'), + 'waiting': ('#b78103', 'clock'), + 'unknown': ('#6e7781', 'question'), + 'warning': ('#b78103', 'alert'), + 'error': ('#d1242f', 'cross'), } -# ========================================== -# Hilfsfunktionen -# ========================================== - -def log(message): - print(message, flush=True) +def q(value): + return urllib.parse.quote(str(value), safe='') -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", - }, - ) - +def req(path, method='GET', data=None): + encoded = json.dumps(data).encode('utf-8') if data is not None else None + request = urllib.request.Request(API + path, data=encoded, method=method, headers={ + 'Authorization': 'token ' + TOKEN, + 'Accept': 'application/json', + 'Content-Type': 'application/json', + 'User-Agent': 'vuln-status/3.0', + }) try: - with urllib.request.urlopen( - request, - timeout=35 - ) as response: - + with urllib.request.urlopen(request, timeout=40) 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 + return json.loads(payload) if payload else {} + except urllib.error.HTTPError as e: + detail = e.read(500).decode(errors='replace') + raise RuntimeError(f'HTTP {e.code} {path}: {detail}') from e -# ========================================== -# 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 file_path(filename): + return f'/repos/{q(OWNER)}/{q(STATUS_REPO)}/contents/{urllib.parse.quote(filename, safe="/")}' -def get_content(filename): - - obj = req(content_path(filename)) - - content = base64.b64decode( - obj["content"] - ).decode("utf-8") - - return content, obj["sha"] +def get_file(filename): + try: + result = req(file_path(filename) + '?ref=' + q(BRANCH)) + except RuntimeError as e: + if str(e).startswith('HTTP 404 '): + return None, None + raise + return base64.b64decode(result['content']).decode('utf-8'), result['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]" - ), +def put_file(filename, content, message, old=None): + if old is None: + existing, sha = get_file(filename) + else: + existing, sha = old + if existing == content: + return False + body = { + 'branch': BRANCH, + 'content': base64.b64encode(content.encode('utf-8')).decode('ascii'), + 'message': message + ' [skip ci]', } + if sha: + body['sha'] = sha + req(file_path(filename), 'PUT' if sha else 'POST', body) + return True - return req(path, "PUT", data) - - -# ========================================== -# Datum und Laufzeit -# ========================================== def parse_time(raw): - - if not raw or str(raw).startswith("0001-"): + 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) - + result = dt.datetime.fromisoformat(str(raw).replace('Z', '+00:00')) + if result.tzinfo is None: + result = result.replace(tzinfo=dt.timezone.utc) + return result.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 display_time(value): + return value.astimezone(TZ).strftime('%d.%m.%Y %H:%M %Z') if value else '–' 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" + return '–' + seconds = int((end - start).total_seconds()) + h, rem = divmod(seconds, 3600) + m, s = divmod(rem, 60) + return f'{h}h {m:02d}m' if h else f'{m}m {s:02d}s' -def safe_cell(value): - - return ( - str(value) - .replace("|", "\\|") - .replace("\n", " ") - .replace("\r", " ") - ) +def 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 svg(name): + color, shape = ICONS[name] + graphic = { + 'check': '', + 'cross': '', + 'clock': '', + 'play': '', + 'question': '', + 'alert': '', + }[shape] + return (f'' + f'{graphic}\n') -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 icon_md(name): + # Absolute image URL works in the Gitea README from all link contexts. + # The repository must be readable by the viewer; private repos require login. + url = (f'{BASE}/{q(OWNER)}/{q(STATUS_REPO)}/raw/branch/' + f'{q(BRANCH)}/assets/status-{name}.svg') + return f'![{name}]({url})' -def get_runs(owner, repo, workflow=None): - - runs = [] - +def get_runs(owner, repo, workflow): + found = [] 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. + params = urllib.parse.urlencode({'page': page, 'limit': PAGE_SIZE}) + path = f'/repos/{q(owner)}/{q(repo)}/actions/runs?{params}' + chunk = req(path).get('workflow_runs', []) + for run in chunk: + if workflow: + name = str(run.get('path') or '').split('@', 1)[0].rsplit('/', 1)[-1] + if name != workflow: + continue + found.append(run) 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 + found.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 found -# ========================================== -# Jobstatus -# ========================================== - -def get_state(run): - +def current_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") - ) + return 'unknown', 'Kein Lauf' + status = str(run.get('status') or '').lower() + conclusion = str(run.get('conclusion') or '').lower() + if status in ('running', 'in_progress'): + return 'running', 'Läuft' + if status in ('queued', 'pending', 'requested', 'waiting'): + return 'waiting', 'Wartet' + if conclusion == 'success': + return 'success', 'Erfolgreich' + if conclusion in ('failure', 'timed_out', 'startup_failure'): + return 'failure', 'Fehlgeschlagen' + if conclusion in ('cancelled', 'skipped'): + return 'unknown', conclusion.capitalize() + return '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 - ) - +def inspect_source(cfg, source): + owner = source.get('owner', cfg.get('owner', OWNER)) + repo = source['repo'] + runs = get_runs(owner, repo, source.get('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 - ) - + successful = next((r for r in runs if str(r.get('conclusion') or '').lower() == 'success'), None) + success_at = (parse_time(successful.get('completed_at')) or + parse_time(successful.get('updated_at')) or + parse_time(successful.get('created_at'))) if successful else None + max_age = float(source.get('max_age_hours', cfg.get('max_age_hours', 26))) + overdue = success_at is None or NOW - success_at > dt.timedelta(hours=max_age) + kind, state = current_state(latest) + state_text = f'{icon_md(kind)} {state}' if overdue: - state += ( - " / " - + ICONS["warning"] - + " Überfällig" - ) + state_text += f' / {icon_md("warning")} Überfällig' + started = (parse_time(latest.get('started_at')) or + parse_time(latest.get('run_started_at')) or + parse_time(latest.get('created_at'))) if latest else None + completed = (parse_time(latest.get('completed_at')) or + parse_time(latest.get('updated_at'))) if latest and str(latest.get('status') or '').lower() == 'completed' else None + duration = elapsed(started, completed) + link = '–' + if latest and latest.get('id'): + url = latest.get('html_url') or f'{BASE}/{q(owner)}/{q(repo)}/actions/runs/{latest["id"]}' + link = f'[{latest["id"]}]({url})' + return kind, overdue, state_text, display_time(success_at), duration, link - # 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(): + cfg_raw, _ = get_file('sources.json') + if cfg_raw is None: + raise RuntimeError('sources.json nicht gefunden') + cfg = json.loads(cfg_raw) + # Ensure all SVG graphics exist before referencing them in README. + for name in ICONS: + filename = f'assets/status-{name}.svg' + changed = put_file(filename, svg(name), f'assets: update {name} status icon') + if changed: + print(f'SVG erstellt: {filename}', flush=True) - 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" - ) - + counts = dict(total=len(cfg['sources']), success=0, failure=0, running=0, + waiting=0, overdue=0, error=0, unknown=0) + errors, rows = [], [] + for src in cfg['sources']: + owner = src.get('owner', cfg.get('owner', OWNER)) + repo = src['repo'] + title = src.get('name', repo) + url = f'{BASE}/{q(owner)}/{q(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 - + kind, overdue, state, success, duration, link = inspect_source(cfg, src) + counts[kind] += 1 + if overdue: + counts['overdue'] += 1 except Exception as exc: + state, success, duration, link = f'{icon_md("error")} API-Fehler', '–', '–', '–' + counts['error'] += 1 + errors.append(f'{repo}: {exc}') + rows.append(f'| [{cell(title)}]({url}) | {state} | {success} | {duration} | {link} |') + print(f'{repo}: {state.split(" ", 1)[-1]}', flush=True) - 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 - # ====================================== - + lines = [ + '# Vulnerability Database – Sync Status', '', + 'Automatisch generiertes Dashboard für Gitea Actions.', '', + f'**Letzte Prüfung:** {display_time(NOW)} ', + f'**Überfällig:** Kein erfolgreicher Lauf innerhalb von {cfg.get("max_age_hours", 26)} Stunden (pro Quelle anpassbar).', + '', '## Übersicht', '', + f'- Erfolgreich: **{counts["success"]}**', + f'- Laufend: **{counts["running"]}**', + f'- Wartend: **{counts["waiting"]}**', + f'- Fehlgeschlagen: **{counts["failure"]}**', + f'- Überfällig: **{counts["overdue"]}**', + f'- API-Fehler: **{counts["error"]}**', + '', '## Synchronisierungsstatus', '', + '| Datenquelle | Status | Letzter Erfolg | Dauer letzter Lauf | Letzter Job |', + '|---|---|---|---|---|', + *rows, '', '## Legende', '', + f'- {icon_md("success")} Erfolgreich', + f'- {icon_md("waiting")} Wartet', + f'- {icon_md("running")} Läuft', + f'- {icon_md("failure")} Fehlgeschlagen', + f'- {icon_md("warning")} Überfällig', + f'- {icon_md("error")} API-Fehler', + '', '## Links', '', + '- [Vulnerability Database](https://vulnerability.apex-project.de)', + '- [Projektwebsite](https://www.apex-project.de)', '', + ] + content = '\n'.join(lines) + changed = put_file('README.md', content, 'docs: update vulnerability sync status') + print('README aktualisiert.' if changed else 'README unverändert.', flush=True) if errors: - - for error in errors: - print( - "ERROR: " + error, - file=sys.stderr - ) - + for err in errors: + print('ERROR:', err, file=sys.stderr) sys.exit(1) - log("======================================") - log("Dashboard erfolgreich aktualisiert") - log("======================================") - -if __name__ == "__main__": +if __name__ == '__main__': main()