188 lines
8.0 KiB
Python
188 lines
8.0 KiB
Python
#!/usr/bin/env python3
|
||
"""Gitea Actions status dashboard. Standard library only; reads config from same repo."""
|
||
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)
|
||
|
||
|
||
def req(path, method='GET', data=None):
|
||
body = json.dumps(data).encode() 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
|
||
|
||
|
||
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=''))
|
||
|
||
|
||
def get_content(filename):
|
||
obj = req(content_path(filename))
|
||
return base64.b64decode(obj['content']).decode('utf-8'), obj['sha']
|
||
|
||
|
||
def parse_time(raw):
|
||
if not raw or raw.startswith('0001-'):
|
||
return None
|
||
try:
|
||
return dt.datetime.fromisoformat(raw.replace('Z', '+00:00')).astimezone(dt.timezone.utc)
|
||
except (ValueError, AttributeError):
|
||
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')
|
||
|
||
|
||
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'
|
||
|
||
|
||
def safe_cell(text):
|
||
return str(text).replace('|', '\\|').replace('\n', ' ')
|
||
|
||
|
||
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}))
|
||
result = req(path)
|
||
runs = result.get('workflow_runs', [])
|
||
if workflow:
|
||
# Gitea run.path is typically .gitea/workflows/<filename>@<branch>.
|
||
# 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 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
|
||
|
||
|
||
def main():
|
||
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'
|
||
try:
|
||
state, last_success, duration, link = source_status(cfg, src)
|
||
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')
|
||
if existing == content:
|
||
print('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.')
|
||
if errors:
|
||
for error in errors:
|
||
print('ERROR: ' + error, file=sys.stderr)
|
||
sys.exit(1)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|