Initial Status
This commit is contained in:
root
@@ -0,0 +1,176 @@
|
||||
#!/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):
|
||||
# Workflow-scoped API: avoids misreading another workflow in the repository.
|
||||
path = '/repos/{}/{}/actions/workflows/{}/runs?branch={}&page={}&limit={}'.format(
|
||||
*(urllib.parse.quote(x, safe='') for x in (owner, repo, workflow)),
|
||||
urllib.parse.quote(BRANCH, safe=''), page, limit)
|
||||
return req(path).get('workflow_runs', [])
|
||||
|
||||
|
||||
def source_status(cfg, source):
|
||||
owner = source.get('owner', cfg['owner'])
|
||||
repo = source['repo']
|
||||
workflow = source['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['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: Quelle, Workflowname, 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()
|
||||
Reference in New Issue
Block a user