"""Automate end-to-end OAuth2/OIDC SSO integration between Authentik and Grafana 11. Performs: 1. Authentik OAuth2 Provider creation with redirect URIs and scope mappings (openid, email, profile, groups). 2. Authentik Application creation ('grafana') linked to the provider. 3. Kubernetes ConfigMap 'grafana' update in namespace 'default' with [auth.generic_oauth] settings and role mapping. 4. Grafana Deployment rollout restart to reload configuration. 5. Live verification of the Grafana SSO endpoint. """ import argparse import datetime import json import secrets import ssl import sys import time import urllib.error import urllib.request ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE def authentik_request(base_url: str, token: str, endpoint: str, method: str = "GET", data: dict = None): url = f"{base_url.rstrip('/')}/api/v3/{endpoint.lstrip('/')}" req = urllib.request.Request( url, data=json.dumps(data).encode("utf-8") if data else None, headers={ "Authorization": f"Bearer {token}", "Content-Type": "application/json", "Accept": "application/json", }, method=method, ) try: with urllib.request.urlopen(req, timeout=10, context=ctx) as resp: content = resp.read().decode("utf-8") return json.loads(content) if content else {} except urllib.error.HTTPError as e: err_msg = e.read().decode("utf-8", errors="ignore") raise RuntimeError(f"Authentik API Error [{e.code}]: {err_msg}") def k8s_request(base_url: str, token: str, path: str, method: str = "GET", data: dict = None, content_type: str = "application/json"): url = f"{base_url.rstrip('/')}/{path.lstrip('/')}" req = urllib.request.Request( url, data=json.dumps(data).encode("utf-8") if data else None, headers={ "Authorization": f"Bearer {token}", "Content-Type": content_type, "Accept": "application/json", }, method=method, ) with urllib.request.urlopen(req, timeout=10, context=ctx) as resp: content = resp.read().decode("utf-8") return json.loads(content) if content else {} def main(): parser = argparse.ArgumentParser(description="Configure Authentik OIDC SSO for Grafana") parser.add_argument("--authentik-url", default="https://authentik.snarfnet.net", help="Authentik Base URL") parser.add_argument("--authentik-token", default="oRq4mTt7i5TtQ4IGV9BWjheywOZaStqaJyU0m7J14pJnzJiVC6r9xFjkDYDp", help="Authentik API Token") parser.add_argument("--k8s-url", default="https://aerodrome.snarfnet.net/k8s/clusters/local", help="Aerodrome K8s API Base URL") parser.add_argument("--k8s-token", default="kubeconfig-u-aqzkufonckwqpfv:4nqwf2z7cqkbnt9jfvzpbh5hc6bkfghkjtgnrpqkx7zv7lmxbxg5z8", help="Aerodrome K8s Bearer Token") parser.add_argument("--grafana-url", default="https://grafana.snarfnet.net", help="Grafana Public Base URL") args = parser.parse_args() print("=========================================================") print("=== STEP 1: AUTHENTIK OIDC PROVIDER & APP CONFIGURATION ===") print("=========================================================") # 1. Discover Flows print("Resolving Authentik Authorization & Invalidation flows...") flows = authentik_request(args.authentik_url, args.authentik_token, "flows/instances/").get("results", []) auth_flow_pk = None inval_flow_pk = None for f in flows: slug = f.get("slug", "") desig = f.get("designation", "") if "implicit-consent" in slug or (desig == "authorization" and not auth_flow_pk): auth_flow_pk = f["pk"] if "default-provider-invalidation" in slug or (desig == "invalidation" and not inval_flow_pk): inval_flow_pk = f["pk"] print(f" * Authorization Flow: {auth_flow_pk}") print(f" * Invalidation Flow: {inval_flow_pk}") # 2. Discover Signing Key print("Resolving Authentik Signing Certificate...") certs = authentik_request(args.authentik_url, args.authentik_token, "crypto/certificatekeypairs/").get("results", []) cert_pk = certs[0]["pk"] if certs else None for c in certs: if "authentik" in c.get("name", "").lower() or "snarfnet" in c.get("name", "").lower(): cert_pk = c["pk"] break print(f" * Signing Key: {cert_pk}") # 3. Discover Scope Mappings (openid, email, profile, groups) print("Resolving Scope Mappings...") scope_mappings = authentik_request(args.authentik_url, args.authentik_token, "propertymappings/provider/scope/").get("results", []) selected_scopes = [] for sm in scope_mappings: sname = sm.get("scope_name", "") if sname in ["openid", "email", "profile", "groups"]: selected_scopes.append(sm["pk"]) print(f" * Found scope mapping '{sname}': {sm['pk']}") # 4. Check or Create OAuth2 Provider print("\nChecking existing OAuth2 Providers in Authentik...") providers = authentik_request(args.authentik_url, args.authentik_token, "providers/oauth2/").get("results", []) existing_provider = None for p in providers: if p.get("name") == "Grafana": existing_provider = p break redirect_uri = f"{args.grafana_url.rstrip('/')}/login/generic_oauth" if existing_provider: provider_id = existing_provider["pk"] client_id = existing_provider["client_id"] client_secret = existing_provider["client_secret"] print(f"Updating existing Grafana OAuth2 Provider (ID {provider_id})...") provider_payload = { "name": "Grafana", "authorization_flow": auth_flow_pk, "invalidation_flow": inval_flow_pk, "property_mappings": selected_scopes, "client_type": "confidential", "client_id": client_id, "client_secret": client_secret, "include_claims_in_id_token": True, "signing_key": cert_pk, "redirect_uris": [{"matching_mode": "strict", "url": redirect_uri}], "sub_mode": "user_id", "issuer_mode": "per_provider", } authentik_request(args.authentik_url, args.authentik_token, f"providers/oauth2/{provider_id}/", method="PATCH", data=provider_payload) print("Provider updated.") else: print("Creating new Grafana OAuth2 Provider...") client_id = f"grafana-{secrets.token_hex(4)}" client_secret = secrets.token_urlsafe(40) provider_payload = { "name": "Grafana", "authorization_flow": auth_flow_pk, "invalidation_flow": inval_flow_pk, "property_mappings": selected_scopes, "client_type": "confidential", "client_id": client_id, "client_secret": client_secret, "include_claims_in_id_token": True, "signing_key": cert_pk, "redirect_uris": [{"matching_mode": "strict", "url": redirect_uri}], "sub_mode": "user_id", "issuer_mode": "per_provider", } new_provider = authentik_request(args.authentik_url, args.authentik_token, "providers/oauth2/", method="POST", data=provider_payload) provider_id = new_provider["pk"] print(f"Created Provider with ID: {provider_id}, Client ID: {client_id}") # 5. Check or Create Application in Authentik print("\nChecking Authentik Applications...") apps = authentik_request(args.authentik_url, args.authentik_token, "core/applications/").get("results", []) existing_app = None for a in apps: if a.get("slug") == "grafana": existing_app = a break app_payload = { "name": "Grafana", "slug": "grafana", "provider": provider_id, "meta_launch_url": args.grafana_url, "open_in_new_tab": True, } if existing_app: print("Updating existing Grafana Application...") authentik_request(args.authentik_url, args.authentik_token, f"core/applications/{existing_app['slug']}/", method="PATCH", data=app_payload) print("Application updated.") else: print("Creating new Grafana Application...") authentik_request(args.authentik_url, args.authentik_token, "core/applications/", method="POST", data=app_payload) print("Created Application 'Grafana'.") print("\n=========================================================") print("=== STEP 2: UPDATE GRAFANA KUBERNETES CONFIGMAP ===") print("=========================================================") # 6. Read existing configmap print("Fetching ConfigMap 'grafana' in namespace 'default'...") cm = k8s_request(args.k8s_url, args.k8s_token, "api/v1/namespaces/default/configmaps/grafana") new_grafana_ini = f"""[analytics] check_for_updates = true [grafana_net] url = {args.grafana_url} [log] mode = console [paths] data = /var/lib/grafana/ logs = /var/log/grafana plugins = /var/lib/grafana/plugins provisioning = /etc/grafana/provisioning [server] root_url = {args.grafana_url} domain = grafana.snarfnet.net enforce_domain = false [auth.generic_oauth] enabled = true name = Authentik allow_sign_up = true auto_login = false client_id = {client_id} client_secret = {client_secret} scopes = openid email profile groups empty_scopes = false auth_url = {args.authentik_url.rstrip('/')}/application/o/authorize/ token_url = {args.authentik_url.rstrip('/')}/application/o/token/ api_url = {args.authentik_url.rstrip('/')}/application/o/userinfo/ login_attribute_path = preferred_username name_attribute_path = name email_attribute_path = email role_attribute_path = contains(groups[*], 'authentik Admins') && 'Admin' || 'Viewer' role_attribute_strict = false allow_assign_grafana_admin = true """ cm_patch = { "data": { "grafana.ini": new_grafana_ini } } print("Patching ConfigMap with updated grafana.ini containing [auth.generic_oauth]...") k8s_request( args.k8s_url, args.k8s_token, "api/v1/namespaces/default/configmaps/grafana", method="PATCH", data=cm_patch, content_type="application/strategic-merge-patch+json", ) print("ConfigMap patched successfully.") # 7. Restart Grafana Deployment print("\nTriggering Grafana Deployment rollout restart...") restart_patch = { "spec": { "template": { "metadata": { "annotations": { "kubectl.kubernetes.io/restartedAt": datetime.datetime.now(datetime.timezone.utc).isoformat() } } } } } k8s_request( args.k8s_url, args.k8s_token, "apis/apps/v1/namespaces/default/deployments/grafana", method="PATCH", data=restart_patch, content_type="application/strategic-merge-patch+json", ) print("Deployment patched. Waiting for new pod to reach Ready status...") # Poll pod status for i in range(15): time.sleep(3) pods = k8s_request(args.k8s_url, args.k8s_token, "api/v1/namespaces/default/pods?labelSelector=app.kubernetes.io/name=grafana") items = pods.get("items", []) if items: p = items[0] status = p.get("status", {}) phase = status.get("phase", "") ready = any(c.get("type") == "Ready" and c.get("status") == "True" for c in status.get("conditions", [])) print(f" * Pod: {p['metadata']['name']} (Phase: {phase}, Ready: {ready})") if phase == "Running" and ready: print("Grafana pod is Ready!") break print("\n=========================================================") print("=== STEP 3: LIVE VERIFICATION ===") print("=========================================================") time.sleep(2) login_url = f"{args.grafana_url.rstrip('/')}/login" print(f"Probing Grafana login endpoint: {login_url}...") try: req = urllib.request.Request(login_url, headers={"User-Agent": "Mozilla/5.0"}) with urllib.request.urlopen(req, timeout=8, context=ctx) as resp: content = resp.read().decode("utf-8", errors="ignore") if "generic_oauth" in content or "Authentik" in content or "Sign in with" in content: print("SUCCESS: Authentik SSO button ('Sign in with Authentik') is live on Grafana login page!") else: print("Grafana login page loaded successfully. HTTP 200 OK.") except Exception as e: print(f"Note: Could not probe public page directly ({e}).") print("\n=========================================================") print("GRAFANA AUTHENTIK SSO INTEGRATION COMPLETE!") print(f"Client ID: {client_id}") print(f"Redirect URI: {redirect_uri}") print(f"Role Mapping: 'authentik Admins' -> Admin | Other Users -> Viewer") print(f"Login URL: {args.grafana_url}") print("=========================================================") if __name__ == "__main__": main()