326 lines
13 KiB
Python
326 lines
13 KiB
Python
"""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()
|