309 lines
12 KiB
Python
309 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
"""Gitea Actions dashboard with self-hosted SVG status icons (stdlib only)."""
|
||
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
|
||
|
||
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)
|
||
TZ = ZoneInfo('Europe/Berlin')
|
||
PAGE_SIZE, MAX_PAGES = 50, 5
|
||
|
||
# The URLs use Gitea's raw file route; no externally hosted graphics.
|
||
ICONS = {
|
||
'success': ('#208637', 'check'),
|
||
'failure': ('#d1242f', 'cross'),
|
||
'running': ('#0969da', 'play'),
|
||
'waiting': ('#b78103', 'clock'),
|
||
'unknown': ('#6e7781', 'question'),
|
||
'warning': ('#b78103', 'alert'),
|
||
'error': ('#d1242f', 'cross'),
|
||
}
|
||
|
||
|
||
def q(value):
|
||
return urllib.parse.quote(str(value), safe='')
|
||
|
||
|
||
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=40) as response:
|
||
payload = response.read()
|
||
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
|
||
|
||
|
||
def file_path(filename):
|
||
return f'/repos/{q(OWNER)}/{q(STATUS_REPO)}/contents/{urllib.parse.quote(filename, safe="/")}'
|
||
|
||
|
||
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 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
|
||
|
||
|
||
def parse_time(raw):
|
||
if not raw or str(raw).startswith('0001-'):
|
||
return None
|
||
try:
|
||
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 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())
|
||
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 cell(value):
|
||
return str(value).replace('|', '\\|').replace('\n', ' ').replace('\r', ' ')
|
||
|
||
|
||
def svg(name):
|
||
color, shape = ICONS[name]
|
||
graphic = {
|
||
'check': '<circle cx="12" cy="12" r="10"/><path d="m7.3 12.3 3.2 3.1 6.3-6.6"/>',
|
||
'cross': '<circle cx="12" cy="12" r="10"/><path d="m8 8 8 8M16 8l-8 8"/>',
|
||
'clock': '<circle cx="12" cy="12" r="10"/><path d="M12 6v6l4 2"/>',
|
||
'play': '<circle cx="12" cy="12" r="10"/><path d="m10 7 6 5-6 5z"/>',
|
||
'question': '<circle cx="12" cy="12" r="10"/><path d="M9 9a3 3 0 1 1 4 3c-1 .6-1 1-1 2M12 17h.01"/>',
|
||
'alert': '<path d="M12 2 23 21H1z"/><path d="M12 9v5M12 18h.01"/>',
|
||
}[shape]
|
||
return (f'<svg xmlns="http://www.w3.org/2000/svg" width="24" height="24" '
|
||
f'viewBox="0 0 24 24" fill="none" stroke="{color}" '
|
||
f'stroke-width="2" stroke-linecap="round" stroke-linejoin="round">'
|
||
f'{graphic}</svg>\n')
|
||
|
||
|
||
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''
|
||
|
||
|
||
def get_runs(owner, repo, workflow):
|
||
found = []
|
||
for page in range(1, MAX_PAGES + 1):
|
||
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
|
||
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
|
||
|
||
|
||
def current_state(run):
|
||
if not run:
|
||
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 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
|
||
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_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
|
||
|
||
|
||
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)
|
||
|
||
# Separate Builder status at the top (not included in source totals).
|
||
# Optional overrides via sources.json: "builder": {"workflow": "build.yml", "max_age_hours": 26}
|
||
builder_cfg = cfg.get('builder', {})
|
||
builder_source = {
|
||
'owner': builder_cfg.get('owner', OWNER),
|
||
'repo': builder_cfg.get('repo', 'vuln-builder'),
|
||
'max_age_hours': builder_cfg.get('max_age_hours', cfg.get('max_age_hours', 26)),
|
||
}
|
||
if builder_cfg.get('workflow'):
|
||
builder_source['workflow'] = builder_cfg['workflow']
|
||
builder_owner, builder_repo = builder_source['owner'], builder_source['repo']
|
||
builder_url = f'{BASE}/{q(builder_owner)}/{q(builder_repo)}'
|
||
builder_error = None
|
||
try:
|
||
_, _, builder_state, builder_success, builder_duration, builder_run = inspect_source(cfg, builder_source)
|
||
except Exception as exc:
|
||
builder_state = f'{icon_md("error")} API-Fehler'
|
||
builder_success = builder_duration = builder_run = '–'
|
||
builder_error = f'{builder_repo}: {exc}'
|
||
print(f'Builder {builder_repo}: {builder_state.split(" ", 1)[-1]}', flush=True)
|
||
|
||
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:
|
||
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)
|
||
|
||
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).',
|
||
'', '## SQLite-Builder', '',
|
||
f'### {builder_state} · [vuln-builder]({builder_url})',
|
||
'',
|
||
'| Letzter erfolgreicher Build | Laufzeit des letzten Jobs | Letzter Job |',
|
||
'|:---|:---|:---|',
|
||
f'| **{cell(builder_success)}** | **{cell(builder_duration)}** | {builder_run} |',
|
||
'',
|
||
'> Der Builder erzeugt die SQLite-Datenbank aus den synchronisierten Quellen. '
|
||
'Der Status wird automatisch über Gitea Actions aktualisiert.',
|
||
'', '---', '', '## Ü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 builder_error:
|
||
errors.append(builder_error)
|
||
if errors:
|
||
for err in errors:
|
||
print('ERROR:', err, file=sys.stderr)
|
||
sys.exit(1)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|