Files
Deploymentcenter/tests/e2e/suite_kern.py
T
Deploymentcenter BotandClaude Opus 5 0c1d269665 test(e2e): whoami absichern und die gemeldete Fassung festnageln
Vier Faelle, die genau die Fehler wiederfinden wuerden, um die es in 2.7.1
ging:

- whoami nennt Bindung und Rechte, und zwar dieselben, die provisioniert
  wurden
- whoami gibt den Klartextwert des Tokens nicht heraus. Eine Auskunft ueber
  ein Token darf das Token nicht weiterreichen
- whoami ohne Token antwortet 401
- Health meldet dieselbe Fassung, die changelog.json fuehrt. Das ist der
  Fall, der #346 nicht wiederkommen laesst: Ein fester Wert in der
  Konfiguration faellt sonst niemandem auf, weil die Datei gitignored ist

Ausserdem: Wird der Shared Key an whoami geschickt, sagt der Endpunkt das
jetzt (not_a_token) statt "Token unbekannt". Der Shared Key ist kein Token -
er hat weder Bindung noch Rechteliste, und ohne den Hinweis sucht der
Aufrufer den Fehler bei seinem Token.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-30 21:09:38 +02:00

421 lines
15 KiB
Python

#!/usr/bin/env python3
"""
Kern: Zustand, Schnittstellenbeschreibung, Installation, Token.
Was hier faellt, macht jede weitere Gruppe bedeutungslos - deshalb laeuft
diese Gruppe vor den Modulen.
"""
from pathlib import Path
from harness import (test, erwarte, erwarte_status, erwarte_erfolg,
erwarte_fehler, ueberspringen)
PROJECT_DIR = Path(__file__).resolve().parents[2]
# ---------------------------------------------------------------------------
# Zustand
# ---------------------------------------------------------------------------
@test('kern', 'Health meldet gesunden Zustand')
def _(ctx):
"""Datenbank erreichbar, Logverzeichnis beschreibbar."""
r = ctx.client.get('/api/health', token=None)
erwarte_status(r, 200)
erwarte(r.feld('healthy') is True, "healthy ist %r | %s" % (r.feld('healthy'), r.kurz(300)))
erwarte(r.feld('checks', 'database', 'ok') is True, "Datenbankpruefung: %s" % r.kurz(300))
erwarte(r.feld('checks', 'log_writable', 'ok') is True, "Logverzeichnis: %s" % r.kurz(300))
@test('kern', 'Health ohne Token nennt keine Interna')
def _(ctx):
"""Ohne Berechtigung nur der Gesamtzustand - keine Schema- oder Kennzahlen."""
r = ctx.client.get('/api/health', token=None)
for feld in ('schema', 'migrations', 'metrics', 'pending_migrations'):
erwarte(feld not in (r.json or {}),
"Health liefert %r ohne Authentifizierung" % feld)
@test('kern', 'Health mit Token liefert Details')
def _(ctx):
r = ctx.client.get('/api/health')
erwarte_status(r, 200)
zusatz = [k for k in (r.json or {}) if k not in
('status', 'healthy', 'app', 'version', 'time_utc', 'checks')]
erwarte(zusatz, "Mit Token kommen keine zusaetzlichen Angaben: %s" % r.kurz(300))
@test('kern', 'Health meldet die Fassung aus dem Changelog')
def _(ctx):
"""
Die Fassung stand frueher als feste Zeichenkette in config/config.php und
blieb dort auf 2.0.0 stehen, waehrend 2.7.0 lief. Zwei gemeldete Fehler
sind daran fehldiagnostiziert worden ("der Server ist zu alt"). Sie kommt
jetzt aus public/docs/changelog.json - und dieser Fall haelt beides
zusammen.
"""
import json
pfad = PROJECT_DIR / 'public' / 'docs' / 'changelog.json'
erwartet = json.loads(pfad.read_text(encoding='utf-8'))['current']
r = ctx.client.get('/api/health', token=None)
erwarte_status(r, 200)
erwarte(r.feld('version') == erwartet,
"Server meldet %r, changelog.json fuehrt %r"
% (r.feld('version'), erwartet))
@test('kern', 'OpenAPI-Beschreibung ist gueltig')
def _(ctx):
"""Die Beschreibung ist der Vertrag mit den SDKs - sie muss ladbar bleiben."""
r = ctx.client.get('/api/openapi.json')
erwarte_status(r, 200)
erwarte(isinstance(r.json, dict), "Antwort ist kein JSON: %s" % r.kurz(200))
erwarte(str(r.feld('openapi', default='')).startswith('3.'),
"openapi-Version: %r" % r.feld('openapi'))
pfade = r.feld('paths', default={})
erwarte(len(pfade) >= 25, "Nur %d Pfade beschrieben" % len(pfade))
ctx['openapi'] = r.json
@test('kern', 'OpenAPI beschreibt alle ausgelieferten Endpunkte')
def _(ctx):
"""
Jede Datei unter public/api/ soll in der Beschreibung vorkommen.
Das ist die Pruefung, die auffaellt, wenn jemand einen Endpunkt ergaenzt
und die Beschreibung vergisst - der haeufigste Weg, auf dem ein SDK und
ein Server auseinanderlaufen.
"""
spec = ctx['openapi']
beschrieben = ' '.join(spec.get('paths', {}).keys())
bekannt = {
'health.php': '/api/health',
'openapi.php': None, # beschreibt sich nicht selbst
'report.php': '/report',
'projects.php': '/projects',
'roadmap.php': '/roadmap',
'provision.php': '/provision',
}
fehlend = []
for datei in sorted((PROJECT_DIR / 'public' / 'api').rglob('*.php')):
rel = datei.relative_to(PROJECT_DIR / 'public' / 'api').as_posix()
name = datei.name
if name in bekannt:
marke = bekannt[name]
if marke is None or marke in beschrieben:
continue
fehlend.append(rel)
continue
# index.php eines Moduls: mindestens ein Pfad des Moduls muss da sein
modul = rel.split('/')[0]
if ('/api/%s/' % modul) not in beschrieben:
fehlend.append(rel)
erwarte(not fehlend, "Nicht in der OpenAPI-Beschreibung: %s" % ', '.join(fehlend))
# ---------------------------------------------------------------------------
# Installation
# ---------------------------------------------------------------------------
@test('kern', 'Alle Migrationen sind angewendet')
def _(ctx):
"""
Der Vergleich zaehlt Dateien im Repository gegen Eintraege in
dc_migrations. Eine nicht ausgefuehrte Migration faellt sonst erst auf,
wenn ein Endpunkt eine Spalte vermisst.
"""
dateien = {p.stem for p in (PROJECT_DIR / 'sql' / 'migrations').glob('*.sql')}
dateien.add('000_schema')
r = ctx.testctl('status')
erwarte_erfolg(r)
angewendet = set(r.feld('migrations', default=[]))
fehlend = sorted(dateien - angewendet)
erwarte(not fehlend, "Nicht angewendet: %s" % ', '.join(fehlend))
ctx['tabellen'] = r.feld('tables')
@test('kern', 'Migration ohne Berechtigung abgewiesen')
def _(ctx):
"""install_db.php war einmal offen erreichbar - das darf nicht wiederkommen."""
r = ctx.client.get('/install_db.php', token=None)
erwarte_fehler(r, http=401, code='unauthorized')
@test('kern', 'Migration ist wiederholbar')
def _(ctx):
"""
Zweiter Aufruf auf demselben Stand: nichts anzuwenden, nichts kaputt.
Ein Deployment loest die Migration aus; sie laeuft also oft auf einem
Stand, der schon aktuell ist.
"""
r = ctx.client.get('/install_db.php', token=ctx.cfg['keys']['shared_key'])
erwarte_erfolg(r)
erwarte(r.feld('applied', default=[]) == [],
"Zweiter Lauf wendet erneut an: %s" % r.feld('applied'))
tabellen = r.feld('tables', default=[])
erwarte(len(tabellen) == ctx['tabellen'],
"Tabellenzahl geaendert: %d statt %d" % (len(tabellen), ctx['tabellen']))
@test('kern', 'Migration legt keinen zweiten Administrator an')
def _(ctx):
"""Der alte Stand setzte bei jedem Aufruf das Admin-Passwort zurueck."""
r = ctx.client.get('/install_db.php', token=ctx.cfg['keys']['shared_key'])
erwarte_erfolg(r)
admin = r.feld('admin', default={})
erwarte('password' not in admin,
"Die Migration hat ein Passwort ausgegeben: %s" % admin)
# ---------------------------------------------------------------------------
# Token
# ---------------------------------------------------------------------------
@test('token', 'Master-Token wird akzeptiert')
def _(ctx):
r = ctx.client.get('/api/bugtracker/v1/projects')
erwarte_erfolg(r)
@test('token', 'Ohne Token abgewiesen')
def _(ctx):
r = ctx.client.get('/api/bugtracker/v1/projects', token=None)
erwarte_fehler(r, http=401, code='unauthorized')
@test('token', 'Erfundenes Token abgewiesen')
def _(ctx):
r = ctx.client.get('/api/bugtracker/v1/projects', token='dc_master_' + 'f' * 48)
erwarte_fehler(r, http=401, code='unauthorized')
@test('token', 'Sub-Token laesst sich provisionieren')
def _(ctx):
r = ctx.client.post('/api/tokens/v1/provision', body={
'client_name': 'e2e-testlauf',
'instance_id': 'testrunner',
'scopes': ['bugtracker:report'],
'environment': 'testing',
})
erwarte_erfolg(r)
erwarte(r.feld('sub_token'), "Kein sub_token in der Antwort: %s" % r.kurz(300))
erwarte(r.feld('type') == 'sub', "type ist %r" % r.feld('type'))
ctx['sub_token'] = r.feld('sub_token')
ctx['sub_scopes'] = r.feld('scopes', default=[])
@test('token', 'Sub-Token bekommt nur die erbetenen Rechte')
def _(ctx):
"""Provisionierung darf einschraenken, niemals erweitern."""
scopes = ctx['sub_scopes']
erwarte('*' not in scopes, "Sub-Token hat den Vollzugriff geerbt: %s" % scopes)
unerwartet = [s for s in scopes if not s.startswith('bugtracker:')]
erwarte(not unerwartet, "Nicht erbetene Rechte: %s" % unerwartet)
@test('token', 'Sub-Token darf melden')
def _(ctx):
r = ctx.client.post('/api/bugtracker/v1/report', token=ctx['sub_token'], body={
'project_slug': 'e2e-selftest',
'title': 'Meldung ueber Sub-Token',
'description': 'Erzeugt vom automatisierten Testlauf.',
'severity': 'low',
'environment': 'testing',
})
erwarte_erfolg(r)
@test('token', 'Sub-Token darf nicht verwalten')
def _(ctx):
"""bugtracker:report schliesst bugtracker:manage nicht ein."""
r = ctx.client.post('/api/bugtracker/v1/manage?action=claim&id=1',
token=ctx['sub_token'], body={})
erwarte_fehler(r, http=(401, 403))
@test('token', 'whoami nennt Bindung und Rechte des eigenen Tokens')
def _(ctx):
"""
Ohne diese Auskunft liesse sich ein Recht nur durch Ausprobieren
feststellen - und das hat Nebenwirkungen: eine Probe auf
bugtracker:manage uebernaehme ein Item, eine auf license:dev stellte
einen Schluessel aus.
"""
r = ctx.client.get('/api/tokens/v1/whoami', token=ctx['sub_token'])
erwarte_erfolg(r)
erwarte(r.feld('type') == 'sub', "type ist %r" % r.feld('type'))
erwarte(sorted(r.feld('scopes', default=[])) == sorted(ctx['sub_scopes']),
"whoami meldet %s, provisioniert wurde %s"
% (r.feld('scopes'), ctx['sub_scopes']))
erwarte(isinstance(r.feld('is_workbench'), bool),
"is_workbench fehlt oder ist kein Wahrheitswert: %s" % r.kurz(300))
@test('token', 'whoami liefert den Klartextwert nicht aus')
def _(ctx):
"""Eine Auskunft ueber ein Token darf das Token nicht weiterreichen."""
r = ctx.client.get('/api/tokens/v1/whoami', token=ctx['sub_token'])
erwarte(ctx['sub_token'] not in (r.body or ''),
"Der Tokenwert steht in der Antwort: %s" % r.kurz(300))
@test('token', 'whoami ohne Token abgewiesen')
def _(ctx):
r = ctx.client.get('/api/tokens/v1/whoami', token=None)
erwarte_fehler(r, http=401)
@test('token', 'Provisionierung ohne Master-Token abgewiesen')
def _(ctx):
r = ctx.client.post('/api/tokens/v1/provision', token=None,
body={'scopes': ['bugtracker:report']})
erwarte_fehler(r, http=401, code='missing_master_token')
@test('token', 'Provisionierung mit erfundenem Master-Token abgewiesen')
def _(ctx):
r = ctx.client.post('/api/tokens/v1/provision', token='dc_master_' + '0' * 48,
body={'scopes': ['bugtracker:report']})
erwarte_fehler(r, http=(401, 403))
@test('token', 'Sub-Token kann sich nicht selbst vermehren')
def _(ctx):
"""
Ein Sub-Token ist kein Master-Token. Koennte es weitere Tokens ausstellen,
liesse sich aus einem ausgelieferten Client eine Token-Fabrik machen.
"""
r = ctx.client.post('/api/tokens/v1/provision', token=ctx['sub_token'],
body={'scopes': ['bugtracker:report']})
erwarte_fehler(r, http=(401, 403))
# ---------------------------------------------------------------------------
# Token nachtraeglich aendern
#
# Bis 2.7.0 ging das nicht: Ein Schreibfehler im Namen oder ein nachtraeglich
# gebrauchtes Recht erzwang ein komplett neues Token - mit der Folge, dass der
# neue Wert auf alle Maschinen verteilt werden muss und am Ende das alte
# mitlaeuft, weil eine vergessen wurde.
# ---------------------------------------------------------------------------
@test('token', 'Aenderbares Token laesst sich anlegen')
def _(ctx):
r = ctx.testctl('workbench_token', project_slug='myapp', name='vorher')
erwarte_erfolg(r)
ctx['edit_token_id'] = r.feld('token_id')
ctx['edit_token_raw'] = r.feld('raw_token')
@test('token', 'Name und Rechte lassen sich nachtraeglich aendern')
def _(ctx):
r = ctx.testctl('update_master_token',
token_id=ctx['edit_token_id'],
name='nachher',
owner_type='dev_agent',
project_slug='myapp',
environment='development',
scopes=['bugtracker:read', 'supervisor:read'])
erwarte_erfolg(r)
erwarte(r.feld('token', 'name') == 'nachher',
'Name nicht uebernommen: %r' % r.feld('token', 'name'))
erwarte(r.feld('token', 'environment') == 'development',
'Umgebung nicht uebernommen: %r' % r.feld('token', 'environment'))
erwarte('supervisor:read' in (r.feld('token', 'scopes') or ''),
'Rechte nicht uebernommen: %r' % r.feld('token', 'scopes'))
@test('token', 'Der Tokenwert bleibt bei einer Aenderung derselbe')
def _(ctx):
"""
Die eigentliche Zusage der Funktion. Waere der Wert danach ein anderer,
muessten alle Maschinen neu bestueckt werden - dann koennte man auch
gleich ein neues Token ausstellen, und die Bearbeitung waere sinnlos.
"""
r = ctx.client.get('/api/bugtracker/v1/projects', token=ctx['edit_token_raw'])
erwarte_erfolg(r)
@test('token', 'Entzogene Rechte greifen sofort')
def _(ctx):
"""
Die Gegenprobe: Der Wert bleibt, die Rechte nicht. Das Token hatte
bugtracker:manage und hat es nach der Aenderung nicht mehr.
"""
r = ctx.client.post('/api/bugtracker/v1/manage',
query={'action': 'next'},
token=ctx['edit_token_raw'],
body={'limit': 1})
erwarte_fehler(r, http=(401, 403))
@test('token', 'Aenderung ohne Rechte wird abgewiesen')
def _(ctx):
"""
Ein Token ohne jedes Recht sieht gueltig aus und kann nichts - der
Aufrufer bekaeme "unauthorized" ohne erkennbaren Grund.
"""
r = ctx.testctl('update_master_token', token_id=ctx['edit_token_id'],
name='ohne rechte', scopes=[])
erwarte_fehler(r, http=400, code='invalid_update')
@test('token', 'Aenderung ohne Namen wird abgewiesen')
def _(ctx):
r = ctx.testctl('update_master_token', token_id=ctx['edit_token_id'],
name=' ', scopes=['bugtracker:read'])
erwarte_fehler(r, http=400, code='invalid_update')
@test('token', 'Aenderung an einem fehlenden Token antwortet 404')
def _(ctx):
r = ctx.testctl('update_master_token', token_id='tok_m_gibtsnicht',
name='egal', scopes=['bugtracker:read'])
erwarte_fehler(r, http=404, code='not_found')
# ---------------------------------------------------------------------------
# Allgemeines Verhalten der Schnittstelle
# ---------------------------------------------------------------------------
@test('kern', 'Unbekannter Endpunkt antwortet 404')
def _(ctx):
r = ctx.client.get('/api/gibtesnicht/v1/nix')
erwarte_status(r, 404)
@test('kern', 'Falsche Methode antwortet 405')
def _(ctx):
r = ctx.client.get('/api/bugtracker/v1/report')
erwarte_fehler(r, http=405, code='method_not_allowed')
@test('kern', 'Kaputtes JSON antwortet 400')
def _(ctx):
"""Ein stiller Rueckfall auf einen leeren Body wuerde Fehler verschleiern."""
r = ctx.client.request('POST', '/api/bugtracker/v1/report',
roh_body='{"title": "unvollstaendig"')
erwarte_fehler(r, http=400)
@test('kern', 'Leerer Body wird abgewiesen')
def _(ctx):
r = ctx.client.request('POST', '/api/bugtracker/v1/report', roh_body='')
erwarte_fehler(r, http=400, code='empty_body')
@test('kern', 'CORS-Vorabanfrage wird beantwortet')
def _(ctx):
"""Die Maschinenschnittstellen erlauben Browser-Clients ausdruecklich."""
r = ctx.client.request('OPTIONS', '/api/bugtracker/v1/report')
erwarte_status(r, 204, 200)