patrickandClaude Sonnet 5
403ceb6b61
CI / Web-Barrierefreiheitspruefung (axe, Tastatur, Reflow, Textabstaende) (push) Canceled after 0s
CI / Core-Modul (Build/Vet/Test) (push) Canceled after 0s
CI / DMS Security-Test-Gate (AC5, Merge-Gate) (push) Canceled after 0s
CI / DMS-Modul (Lint/Build/Test/Coverage/Artefakt) (push) Canceled after 0s
CI / DMS-Modul Supply-Chain-Scan (govulncheck) (push) Canceled after 0s
CI / Web (dms-browser) (push) Canceled after 0s
CI / Web (pdf-viewer) (push) Canceled after 0s
CI / Web (shl) (push) Canceled after 0s
Supply-Chain-Scan (Go) / govulncheck (push) Canceled after 0s
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
94 lines
4.1 KiB
Python
Executable File
94 lines
4.1 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Zusatzpruefungen fuer den Token-Modus-Smoke mit ECHTEM Core-Kontodienst (scripts/gesamt-token-smoke.sh).
|
|
|
|
gesamt_token_smoke.py roles Rollen: user vs. tenant_admin an Verwaltungsendpunkten
|
|
gesamt_token_smoke.py save-token <datei> Token von alice holen und ablegen (vor Neustart des Kontodienstes)
|
|
gesamt_token_smoke.py after-restart <datei> altes Token weiter gueltig, JWKS gleiche KID, Neu-Login moeglich
|
|
gesamt_token_smoke.py flag <200|403> Zugriff mit frischem Token erwartet Status (Modul aktiv/abgewaehlt)
|
|
|
|
Umgebung wie fdn19_smoke.py plus FDN19_CORE_ACCOUNT, FDN19_CORE_PASSWORD, FDN19_CORE_USERS.
|
|
"""
|
|
import json
|
|
import sys
|
|
import time
|
|
import urllib.request
|
|
|
|
import fdn19_smoke as f
|
|
|
|
ALICE, BOB = "alice@e2e.test", "bob@e2e.test"
|
|
|
|
|
|
def status(path, token, method="GET"):
|
|
st, _ = f.http(method, f.API + path, headers={"Authorization": "Bearer " + token})
|
|
return st
|
|
|
|
|
|
def jwks_kids():
|
|
with urllib.request.urlopen(f.CORE_ACCOUNT + "/auth/module-jwks", timeout=10) as r:
|
|
return sorted(k["kid"] for k in json.loads(r.read())["keys"])
|
|
|
|
|
|
def roles():
|
|
user_tok = f.core_module_token(f.CORE_ACCOUNT, ALICE)
|
|
admin_tok = f.core_module_token(f.CORE_ACCOUNT, BOB)
|
|
for path in ("/tags", "/search?q=x", "/saved-searches"):
|
|
f.check("R1 user darf %s (nicht 401/403)" % path.split("?")[0], status(path, user_tok) not in (401, 403), "status %s" % status(path, user_tok))
|
|
for path in ("/audit-log", "/auto-rules", "/ocr/review"):
|
|
su, sa = status(path, user_tok), status(path, admin_tok)
|
|
f.check("R2 %s: user 403, tenant_admin erlaubt" % path, su == 403 and sa not in (401, 403), "user %s, admin %s" % (su, sa))
|
|
st_u = status("/trash/purge", user_tok, "POST")
|
|
f.check("R3 POST /trash/purge: user 403", st_u == 403, "status %s" % st_u)
|
|
uid = f.psql("SELECT id FROM users WHERE email = '%s'" % ALICE).split("\n")[0]
|
|
f.CORE_EMAIL[uid] = ALICE
|
|
u = f.Actor("core", uid)
|
|
doc = u.upload("user-rolle.txt", "text/plain", ("Rollenbeleg %d" % time.time_ns()).encode())
|
|
st, _ = u.call("GET", "/documents/%s" % doc)
|
|
f.check("R5 user (nur Standardrolle) laedt hoch und liest eigenes Dokument", st == 200, "status %s" % st)
|
|
st_none = f.http("GET", f.API + "/tags")[0]
|
|
f.check("R4 ohne Token 401", st_none == 401, "status %s" % st_none)
|
|
|
|
|
|
def save_token(path):
|
|
tok = f.core_module_token(f.CORE_ACCOUNT, ALICE)
|
|
with open(path, "w") as fh:
|
|
json.dump({"token": tok, "kids": jwks_kids()}, fh)
|
|
f.check("K1 Token vor Neustart gueltig", status("/tags", tok) == 200)
|
|
|
|
|
|
def after_restart(path):
|
|
saved = json.load(open(path))
|
|
end = time.time() + 30
|
|
while time.time() < end:
|
|
try:
|
|
kids = jwks_kids()
|
|
break
|
|
except Exception: # noqa: BLE001 - Dienst startet noch
|
|
time.sleep(1)
|
|
else:
|
|
f.check("K2 Kontodienst nach Neustart erreichbar", False)
|
|
return
|
|
f.check("K2 JWKS nach Neustart mit derselben KID (persistente Schluessel, API-16)", kids == saved["kids"], "vorher %s nachher %s" % (saved["kids"], kids))
|
|
f.check("K3 Token von vor dem Neustart weiter gueltig", status("/tags", saved["token"]) == 200)
|
|
fresh = f.core_module_token(f.CORE_ACCOUNT, ALICE)
|
|
f.check("K4 Neu-Login und neues Token nach Neustart", status("/tags", fresh) == 200)
|
|
|
|
|
|
def flag(expect):
|
|
tok = f.core_module_token(f.CORE_ACCOUNT, ALICE)
|
|
end, got = time.time() + 15, None
|
|
while time.time() < end:
|
|
got = status("/tags", tok)
|
|
if got == int(expect):
|
|
break
|
|
time.sleep(1)
|
|
f.check("M1 Modul dms fuer den Mandanten: Zugriff liefert %s" % expect, got == int(expect), "status %s" % got)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
cmd = sys.argv[1]
|
|
{"roles": lambda: roles(), "save-token": lambda: save_token(sys.argv[2]),
|
|
"after-restart": lambda: after_restart(sys.argv[2]), "flag": lambda: flag(sys.argv[2])}[cmd]()
|
|
failed = [r for r in f.results if not r[1]]
|
|
print("\nGESAMT: %d Pruefungen, %d bestanden, %d fehlgeschlagen" % (len(f.results), len(f.results) - len(failed), len(failed)))
|
|
sys.exit(1 if failed else 0)
|