"""Automate end-to-end OAuth2/OIDC SSO integration between Authentik and Jenkins. Performs: 1. Authentik OAuth2 Provider creation with redirect URIs: - https://jenkins.snarfnet.net/securityRealm/finishLogin - https://jenkins.snarfnet.net/OicLogout 2. Authentik Application creation ('jenkins') linked to the provider. 3. Update ConfigMap 'jenkins' plugins.txt with oic-auth plugin. 4. Update ConfigMap 'jenkins-jenkins-jcasc-config' jcasc-default-config.yaml with oicSecurityRealm. 5. Recreate pod 'jenkins-0' to download plugin and apply JCasC configuration. 6. Live verification of the Jenkins SSO login 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, ) 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 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 Jenkins") parser.add_argument("--authentik-url", default="https://authentik.snarfnet.net", help="Authentik Base URL") parser.add_argument("--authentik-token", default="SnarfNetPermanentAutomationToken2026", help="Authentik API Token") parser.add_argument("--k8s-url", default="https://aerodrome.snarfnet.net/k8s/clusters/c-m-4fgkl75b", help="Spaceport K8s API Base URL") parser.add_argument("--k8s-token", default="kubeconfig-u-aqzkufonck4hmbd:bw22b96nv8844k8w5r8lzpz8gnk22tjwj95j6nkzr7hd5dhm4bs6z7", help="Spaceport K8s Bearer Token") parser.add_argument("--jenkins-url", default="https://jenkins.snarfnet.net", help="Jenkins Public Base URL") args = parser.parse_args() print("=========================================================") print("=== STEP 1: AUTHENTIK OIDC PROVIDER FOR JENKINS ===") print("=========================================================") auth_flow_pk = "38a86ef3-f43a-4756-8f54-ca008a8eef6e" inval_flow_pk = "7a8145b8-2476-4636-9901-825f826df805" cert_pk = "6d48ba88-4b51-428f-903c-93334988a941" selected_scopes = [ "f521356c-080d-4bd6-b19a-68ab159a5aef", # openid "8395b909-8e8d-4fc1-a620-5592e10df8fe", # email "85b89b43-92d8-425c-b9cc-8e00038cad37", # profile "d2248142-6be2-4833-94f8-e02f5d293b02", # groups ] redirect_uri = f"{args.jenkins_url.rstrip('/')}/securityRealm/finishLogin" logout_uri = f"{args.jenkins_url.rstrip('/')}/OicLogout" providers = authentik_request(args.authentik_url, args.authentik_token, "providers/oauth2/").get("results", []) existing_provider = next((p for p in providers if p.get("name") == "Jenkins"), None) if existing_provider: provider_id = existing_provider["pk"] client_id = existing_provider["client_id"] client_secret = existing_provider["client_secret"] print(f"Updating existing Jenkins OAuth2 Provider (ID {provider_id})...") provider_payload = { "name": "Jenkins", "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}, {"matching_mode": "strict", "url": logout_uri}, {"matching_mode": "regex", "url": r"https://jenkins\.snarfnet\.net/.*"} ], "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 Jenkins OAuth2 Provider...") client_id = f"jenkins-{secrets.token_hex(4)}" client_secret = secrets.token_urlsafe(40) provider_payload = { "name": "Jenkins", "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}, {"matching_mode": "strict", "url": logout_uri}, {"matching_mode": "regex", "url": r"https://jenkins\.snarfnet\.net/.*"} ], "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 ID {provider_id}, Client ID: {client_id}") # 2. Check or Create Application in Authentik apps = authentik_request(args.authentik_url, args.authentik_token, "core/applications/").get("results", []) existing_app = next((a for a in apps if a.get("slug") == "jenkins"), None) app_payload = { "name": "Jenkins", "slug": "jenkins", "provider": provider_id, "meta_launch_url": args.jenkins_url, "open_in_new_tab": True, } if existing_app: authentik_request(args.authentik_url, args.authentik_token, "core/applications/jenkins/", method="PATCH", data=app_payload) print("Authentik Application 'jenkins' updated.") else: authentik_request(args.authentik_url, args.authentik_token, "core/applications/", method="POST", data=app_payload) print("Authentik Application 'jenkins' created.") well_known_url = f"{args.authentik_url.rstrip('/')}/application/o/jenkins/.well-known/openid-configuration" print("\n=========================================================") print("=== STEP 2: CONFIGURE JENKINS CONFIGMAPS (PLUGINS & JCASC) ===") print("=========================================================") # 3. Update ConfigMap 'jenkins' plugins.txt with oic-auth print("Checking Jenkins plugins in ConfigMap 'jenkins'...") cm_jenkins = k8s_request(args.k8s_url, args.k8s_token, "api/v1/namespaces/infrastructure/configmaps/jenkins") plugins_txt = cm_jenkins.get("data", {}).get("plugins.txt", "") if "oic-auth" not in plugins_txt: print("Adding oic-auth:4.731.v4932b_506e767 to plugins.txt...") new_plugins_txt = plugins_txt.strip() + "\noic-auth:4.731.v4932b_506e767\n" k8s_request( args.k8s_url, args.k8s_token, "api/v1/namespaces/infrastructure/configmaps/jenkins", method="PATCH", data={"data": {"plugins.txt": new_plugins_txt}}, content_type="application/strategic-merge-patch+json" ) print("plugins.txt updated.") else: print("oic-auth plugin already present in plugins.txt.") # 4. Update JCasC config in ConfigMap 'jenkins-jenkins-jcasc-config' print("Updating Jenkins JCasC configuration in 'jenkins-jenkins-jcasc-config'...") cm_jcasc = k8s_request(args.k8s_url, args.k8s_token, "api/v1/namespaces/infrastructure/configmaps/jenkins-jenkins-jcasc-config") jcasc_yaml = cm_jcasc.get("data", {}).get("jcasc-default-config.yaml", "") # Build new securityRealm section oic_block = f""" securityRealm: oic: clientId: "{client_id}" clientSecret: "{client_secret}" serverConfiguration: wellKnown: wellKnownOpenIDConfigurationUrl: "{well_known_url}" scopesOverride: "openid profile email groups" userNameField: "preferred_username" fullNameFieldName: "name" emailFieldName: "email" groupsFieldName: "groups" logoutFromOpenidProvider: true properties: - escapeHatch: username: "admin" secret: "${{chart-admin-password}}" group: "jenkins-admins" """ # Replace securityRealm: local/oic: ... with oic block import re new_jcasc_yaml = re.sub( r' securityRealm:\s+(?:local|oic):.*?(?=\n disableRememberMe:|\n mode:)', oic_block.rstrip(), jcasc_yaml, flags=re.DOTALL ) # Ensure unclassified.location.url matches public Jenkins URL new_jcasc_yaml = re.sub( r'(\s+location:\s+url:\s+).*', rf'\g<1>{args.jenkins_url.rstrip("/")}', new_jcasc_yaml ) k8s_request( args.k8s_url, args.k8s_token, "api/v1/namespaces/infrastructure/configmaps/jenkins-jenkins-jcasc-config", method="PATCH", data={"data": {"jcasc-default-config.yaml": new_jcasc_yaml}}, content_type="application/strategic-merge-patch+json" ) print("jcasc-default-config.yaml updated with oicSecurityRealm.") # 5. Restart Jenkins Pod jenkins-0 print("\nRestarting Jenkins pod 'jenkins-0' to install plugin and load JCasC config...") try: k8s_request( args.k8s_url, args.k8s_token, "api/v1/namespaces/infrastructure/pods/jenkins-0", method="DELETE" ) print("Pod jenkins-0 deleted. StatefulSet will recreate it.") except Exception as e: print("Note on pod delete:", e) print("Waiting for Jenkins to start and reach Ready status (plugin install + warmup takes ~60-90s)...") for i in range(30): time.sleep(5) pods = k8s_request(args.k8s_url, args.k8s_token, "api/v1/namespaces/infrastructure/pods?labelSelector=app.kubernetes.io/name=jenkins").get("items", []) if pods: p = pods[0] phase = p.get("status", {}).get("phase", "") ready = any(c.get("type") == "Ready" and c.get("status") == "True" for c in p.get("status", {}).get("conditions", [])) print(f" [{i+1}/30] Pod: {p['metadata']['name']} (Phase: {phase}, Ready: {ready})") if phase == "Running" and ready: print("Jenkins pod is Ready!") break print("\n=========================================================") print("=== STEP 3: LIVE VERIFICATION ===") print("=========================================================") login_url = f"{args.jenkins_url.rstrip('/')}/login" print(f"Probing Jenkins 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 "securityrealm" in content.lower() or "login" in content.lower(): print("SUCCESS: Jenkins login endpoint is online!") except Exception as e: print(f"Note: Could not probe public page directly ({e}).") print("\n=========================================================") print("JENKINS AUTHENTIK SSO INTEGRATION COMPLETE!") print(f"Client ID: {client_id}") print(f"Well-Known URL: {well_known_url}") print(f"Redirect URI: {redirect_uri}") print(f"Escape Hatch: user 'admin' (using chart-admin-password)") print(f"Login URL: {args.jenkins_url}") print("=========================================================") if __name__ == "__main__": main()