kerimatilganandClaude Opus 5.5
cfe59e3652
Ohne grant_types lehnt Authentik jede Anmeldung mit "invalid_request / The request is otherwise malformed" ab. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
278 lines
11 KiB
Python
278 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Richtet Authentik für das Handy-Insel Portal ein (idempotent, mehrfach ausführbar).
|
||
|
||
AK_TOKEN=<admin-api-token> python3 scripts/authentik-setup.py [--apply] [--env /etc/hi-portal/portal.env]
|
||
|
||
Ohne --apply wird nur angezeigt, was passieren würde (Probelauf).
|
||
|
||
Schritte:
|
||
1. Apps an Gruppen binden (wer sieht/öffnet welches Tool)
|
||
2. OIDC-Provider + App "Portal" anlegen (für alle HI-Mitarbeiter)
|
||
3. Service-Account "hi-portal" mit Leserecht auf "Apps pro Benutzer" + API-Token
|
||
4. Branding (Titel "Handy-Insel", Logo/Favicon vom Portal)
|
||
5. Client-ID/Secret/Token in die portal.env schreiben (falls --env angegeben)
|
||
"""
|
||
import argparse
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
import urllib.error
|
||
import urllib.request
|
||
|
||
AK_URL = os.environ.get("AK_URL", "https://login.handy-insel.de").rstrip("/")
|
||
PORTAL_URL = os.environ.get("PORTAL_URL", "https://portal.handy-insel.de").rstrip("/")
|
||
|
||
# Wer darf welche App öffnen (Slug -> Gruppen). Mitglieder einer der Gruppen haben Zugriff.
|
||
APP_ACCESS = {
|
||
"preisliste": ["HI-Mitarbeiter"],
|
||
"vvl": ["HI-Mitarbeiter"],
|
||
"zeiterfassung": ["HI-Mitarbeiter"],
|
||
"timer": ["HI-Mitarbeiter"],
|
||
"kundenkarte": ["HI-Mitarbeiter"],
|
||
"datenuebertragungen": ["HI-Mitarbeiter"],
|
||
"service": ["HI-Mitarbeiter"],
|
||
"portal": ["HI-Mitarbeiter"],
|
||
"paperless": ["HI-Leitung", "HI-Admin"],
|
||
"patch-mon": ["HI-Admin"],
|
||
"preisliste-dev": ["HI-Admin"],
|
||
"zeit-dev": ["HI-Admin"],
|
||
}
|
||
|
||
SERVICE_ACCOUNT = "hi-portal"
|
||
SERVICE_GROUP = "hi-portal-service"
|
||
SERVICE_ROLE = "Portal: Apps pro Benutzer lesen"
|
||
SERVICE_PERMISSIONS = ["authentik_core.view_user", "authentik_core.view_user_applications"]
|
||
|
||
APPLY = False
|
||
|
||
|
||
def call(method, path, body=None):
|
||
if method != "GET" and not APPLY:
|
||
print(f" [Probelauf] {method} {path}")
|
||
return {}
|
||
req = urllib.request.Request(
|
||
f"{AK_URL}/api/v3{path}",
|
||
method=method,
|
||
data=json.dumps(body).encode() if body is not None else None,
|
||
headers={
|
||
"Authorization": f"Bearer {os.environ['AK_TOKEN']}",
|
||
"Content-Type": "application/json",
|
||
"Accept": "application/json",
|
||
},
|
||
)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=20) as r:
|
||
raw = r.read()
|
||
return json.loads(raw) if raw else {}
|
||
except urllib.error.HTTPError as e:
|
||
sys.exit(f"FEHLER {method} {path}: HTTP {e.code} {e.read().decode()[:400]}")
|
||
|
||
|
||
def results(path):
|
||
sep = "&" if "?" in path else "?"
|
||
return call("GET", f"{path}{sep}page_size=200")["results"]
|
||
|
||
|
||
def step(title):
|
||
print(f"\n== {title}")
|
||
|
||
|
||
def bind_apps(groups, apps):
|
||
step("1. Apps an Gruppen binden")
|
||
for slug, group_names in APP_ACCESS.items():
|
||
app = apps.get(slug)
|
||
if not app:
|
||
print(f" - {slug}: App existiert (noch) nicht, übersprungen")
|
||
continue
|
||
existing = results(f"/policies/bindings/?target={app['pk']}")
|
||
bound = {b.get("group") for b in existing}
|
||
orders = {b["order"] for b in existing}
|
||
for name in group_names:
|
||
if groups[name] in bound:
|
||
print(f" = {slug} -> {name}")
|
||
continue
|
||
order = 10
|
||
while order in orders:
|
||
order += 1
|
||
orders.add(order)
|
||
call("POST", "/policies/bindings/", {
|
||
"target": app["pk"], "group": groups[name], "order": order,
|
||
"enabled": True, "negate": False, "timeout": 30, "failure_result": False,
|
||
})
|
||
print(f" + {slug} -> {name}")
|
||
if app.get("policy_engine_mode") != "any":
|
||
call("PATCH", f"/core/applications/{slug}/", {"policy_engine_mode": "any"})
|
||
|
||
|
||
def ensure_portal_app(apps):
|
||
step("2. OIDC-Provider und App 'Portal'")
|
||
template = next(p for p in results("/providers/oauth2/") if p["name"] == "Preisliste")
|
||
provider = next((p for p in results("/providers/oauth2/") if p["name"] == "Portal"), None)
|
||
if provider:
|
||
print(" = Provider 'Portal' vorhanden")
|
||
else:
|
||
provider = call("POST", "/providers/oauth2/", {
|
||
"name": "Portal",
|
||
"authorization_flow": template["authorization_flow"],
|
||
"invalidation_flow": template["invalidation_flow"],
|
||
"client_type": "confidential",
|
||
# Neuere Authentik-Versionen setzen per API keine Standard-Grant-Types;
|
||
# ohne diese Liste lehnt Authentik jede Anmeldung als "malformed" ab.
|
||
"grant_types": ["authorization_code", "refresh_token"],
|
||
"redirect_uris": [
|
||
{"matching_mode": "strict", "url": f"{PORTAL_URL}/auth/callback", "redirect_uri_type": "authorization"},
|
||
# Rücksprung nach dem Abmelden (post_logout_redirect_uri)
|
||
{"matching_mode": "strict", "url": f"{PORTAL_URL}/", "redirect_uri_type": "logout"},
|
||
],
|
||
"property_mappings": template["property_mappings"],
|
||
"signing_key": template.get("signing_key"),
|
||
"sub_mode": template.get("sub_mode", "hashed_user_id"),
|
||
"include_claims_in_id_token": True,
|
||
})
|
||
print(" + Provider 'Portal' angelegt")
|
||
if "portal" in apps:
|
||
print(" = App 'portal' vorhanden")
|
||
elif APPLY:
|
||
apps["portal"] = call("POST", "/core/applications/", {
|
||
"name": "Portal", "slug": "portal", "provider": provider["pk"],
|
||
"meta_launch_url": f"{PORTAL_URL}/", "policy_engine_mode": "any",
|
||
})
|
||
print(" + App 'portal' angelegt")
|
||
return provider
|
||
|
||
|
||
def ensure_service_account(groups):
|
||
step("3. Service-Account für die App-Liste")
|
||
user = next((u for u in results(f"/core/users/?username={SERVICE_ACCOUNT}") if u["username"] == SERVICE_ACCOUNT), None)
|
||
token_key = None
|
||
if user:
|
||
print(f" = Service-Account '{SERVICE_ACCOUNT}' vorhanden")
|
||
else:
|
||
res = call("POST", "/core/users/service_account/", {"name": SERVICE_ACCOUNT, "create_group": False, "expiring": False})
|
||
user = {"pk": res.get("user_pk")}
|
||
print(f" + Service-Account '{SERVICE_ACCOUNT}' angelegt")
|
||
|
||
role = next((r for r in results("/rbac/roles/") if r["name"] == SERVICE_ROLE), None)
|
||
if not role:
|
||
role = call("POST", "/rbac/roles/", {"name": SERVICE_ROLE})
|
||
print(f" + Rolle '{SERVICE_ROLE}' angelegt")
|
||
if role.get("pk"):
|
||
call("POST", f"/rbac/permissions/assigned_by_roles/{role['pk']}/assign/", {"permissions": SERVICE_PERMISSIONS})
|
||
print(" + Rechte: " + ", ".join(SERVICE_PERMISSIONS))
|
||
|
||
if SERVICE_GROUP not in groups:
|
||
call("POST", "/core/groups/", {
|
||
"name": SERVICE_GROUP, "is_superuser": False,
|
||
"users": [user["pk"]] if user.get("pk") else [],
|
||
"roles": [role["pk"]] if role.get("pk") else [],
|
||
})
|
||
print(f" + Gruppe '{SERVICE_GROUP}' (Rolle + Service-Account)")
|
||
else:
|
||
print(f" = Gruppe '{SERVICE_GROUP}' vorhanden")
|
||
|
||
# Der Service-Account bekommt von Authentik nur ein App-Passwort (intent=app_password),
|
||
# das die API nicht als Bearer-Token akzeptiert. Deshalb eigener API-Token.
|
||
if APPLY and user.get("pk"):
|
||
ident = f"{SERVICE_ACCOUNT}-api"
|
||
if not results(f"/core/tokens/?identifier={ident}"):
|
||
call("POST", "/core/tokens/", {
|
||
"identifier": ident, "user": user["pk"], "intent": "api", "expiring": False,
|
||
"description": "Portal: App-Liste pro Benutzer lesen",
|
||
})
|
||
print(f" + API-Token '{ident}' angelegt")
|
||
token_key = call("GET", f"/core/tokens/{ident}/view_key/")["key"]
|
||
return user, token_key
|
||
|
||
|
||
def branding():
|
||
step("4. Branding")
|
||
try:
|
||
urllib.request.urlopen(f"{PORTAL_URL}/icon-192.png", timeout=10)
|
||
except Exception:
|
||
print(f" - übersprungen: {PORTAL_URL} ist noch nicht erreichbar (sonst kaputtes Logo auf der Login-Seite)")
|
||
return
|
||
brand = next((b for b in results("/core/brands/") if b.get("default")), None) or results("/core/brands/")[0]
|
||
call("PATCH", f"/core/brands/{brand['brand_uuid']}/", {
|
||
"branding_title": "Handy-Insel",
|
||
"branding_logo": f"{PORTAL_URL}/icon-192.png",
|
||
"branding_favicon": f"{PORTAL_URL}/favicon.png",
|
||
})
|
||
print(" + Titel 'Handy-Insel', Logo und Favicon vom Portal")
|
||
|
||
|
||
def write_env(path, values):
|
||
step(f"5. {path} aktualisieren")
|
||
text = open(path).read() if os.path.exists(path) else ""
|
||
for key, value in values.items():
|
||
if not value:
|
||
continue
|
||
line = f"{key}={value}"
|
||
text = re.sub(rf"^{key}=.*$", line, text, flags=re.M) if re.search(rf"^{key}=", text, flags=re.M) else text + f"\n{line}\n"
|
||
if APPLY:
|
||
with open(path, "w") as f:
|
||
f.write(text)
|
||
print(" + " + ", ".join(k for k, v in values.items() if v))
|
||
|
||
|
||
def verify(token, user_pk):
|
||
step("Prüfung: Kann der Service-Account die Apps eines Benutzers lesen?")
|
||
kerim = next((u for u in results("/core/users/?username=kerim")), None)
|
||
if not (token and kerim):
|
||
print(" (übersprungen)")
|
||
return
|
||
req = urllib.request.Request(
|
||
f"{AK_URL}/api/v3/core/applications/?for_user={kerim['pk']}&page_size=200",
|
||
headers={"Authorization": f"Bearer {token}", "Accept": "application/json"},
|
||
)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=20) as r:
|
||
slugs = [a["slug"] for a in json.load(r)["results"]]
|
||
print(f" OK – Apps für kerim: {', '.join(slugs)}")
|
||
except urllib.error.HTTPError as e:
|
||
print(f" FEHLER HTTP {e.code}: {e.read().decode()[:300]}")
|
||
|
||
|
||
def main():
|
||
global APPLY
|
||
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
||
parser.add_argument("--apply", action="store_true", help="Änderungen wirklich durchführen")
|
||
parser.add_argument("--env", help="Pfad zur portal.env, in die Client-ID/Secret/Token geschrieben werden")
|
||
parser.add_argument("--skip-branding", action="store_true")
|
||
args = parser.parse_args()
|
||
APPLY = args.apply
|
||
if "AK_TOKEN" not in os.environ:
|
||
sys.exit("Bitte AK_TOKEN (API-Token eines Authentik-Admins) setzen.")
|
||
|
||
print(f"Authentik: {AK_URL} | Portal: {PORTAL_URL} | {'AUSFÜHREN' if APPLY else 'PROBELAUF'}")
|
||
groups = {g["name"]: g["pk"] for g in results("/core/groups/")}
|
||
for needed in ("HI-Mitarbeiter", "HI-Leitung", "HI-Admin"):
|
||
if needed not in groups:
|
||
sys.exit(f"Gruppe {needed} fehlt in Authentik.")
|
||
apps = {a["slug"]: a for a in results("/core/applications/?superuser_full_list=true")}
|
||
|
||
provider = ensure_portal_app(apps)
|
||
bind_apps(groups, apps)
|
||
user, token = ensure_service_account(groups)
|
||
if not args.skip_branding:
|
||
branding()
|
||
if args.env:
|
||
write_env(args.env, {
|
||
"OIDC_CLIENT_ID": provider.get("client_id"),
|
||
"OIDC_CLIENT_SECRET": provider.get("client_secret"),
|
||
"AUTHENTIK_TOKEN": token,
|
||
})
|
||
elif APPLY:
|
||
print("\nFür die portal.env:")
|
||
print(f" OIDC_CLIENT_ID={provider.get('client_id')}")
|
||
print(f" OIDC_CLIENT_SECRET={provider.get('client_secret')}")
|
||
print(f" AUTHENTIK_TOKEN={token}")
|
||
if APPLY:
|
||
verify(token, user.get("pk"))
|
||
print("\nFertig." if APPLY else "\nProbelauf beendet – mit --apply ausführen.")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|