"""Automate end-to-end OAuth2/OIDC SSO integration between Authentik and Argo CD. Performs: 1. Authentik OAuth2 Provider creation with redirect URI: https://argo.snarfnet.net/auth/callback 2. Authentik Application creation ('argo-cd') linked to the provider. 3. Update ConfigMap 'argocd-cm' with native OIDC configuration. 4. Update ConfigMap 'argocd-rbac-cm' with group policy mapping 'authentik Admins' -> role:admin. 5. Rollout restart of 'argocd-server' deployment. 6. Live verification of the Argo CD 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 Argo CD") 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("--argo-url", default="https://argo.snarfnet.net", help="Argo CD Public Base URL") args = parser.parse_args() print("=========================================================") print("=== STEP 1: AUTHENTIK OIDC PROVIDER FOR ARGO CD ===") 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.argo_url.rstrip('/')}/auth/callback" 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") == "Argo CD"), 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 Argo CD OAuth2 Provider (ID {provider_id})...") provider_payload = { "name": "Argo CD", "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 Argo CD OAuth2 Provider...") client_id = f"argocd-{secrets.token_hex(4)}" client_secret = secrets.token_urlsafe(40) provider_payload = { "name": "Argo CD", "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 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") == "argo-cd"), None) app_payload = { "name": "Argo CD", "slug": "argo-cd", "provider": provider_id, "meta_launch_url": args.argo_url, "open_in_new_tab": True, } if existing_app: authentik_request(args.authentik_url, args.authentik_token, "core/applications/argo-cd/", method="PATCH", data=app_payload) print("Authentik Application 'argo-cd' updated.") else: authentik_request(args.authentik_url, args.authentik_token, "core/applications/", method="POST", data=app_payload) print("Authentik Application 'argo-cd' created.") issuer_url = f"{args.authentik_url.rstrip('/')}/application/o/argo-cd/" print("\n=========================================================") print("=== STEP 2: CONFIGURE ARGO CD KUBERNETES CONFIGMAPS ===") print("=========================================================") # 3. Patch argocd-cm with oidc.config print("Updating ConfigMap 'argocd-cm'...") oidc_config = f"""name: Authentik issuer: {issuer_url} clientID: {client_id} clientSecret: {client_secret} requestedScopes: - openid - profile - email - groups """ cm_patch = { "data": { "url": args.argo_url, "oidc.config": oidc_config } } k8s_request( args.k8s_url, args.k8s_token, "api/v1/namespaces/infrastructure/configmaps/argocd-cm", method="PATCH", data=cm_patch, content_type="application/strategic-merge-patch+json" ) print("argocd-cm updated.") # 4. Patch argocd-rbac-cm with policy mapping print("Updating ConfigMap 'argocd-rbac-cm'...") rbac_patch = { "data": { "policy.csv": "g, authentik Admins, role:admin\ng, *, role:readonly\n", "policy.default": "role:readonly", "scopes": "[groups]" } } k8s_request( args.k8s_url, args.k8s_token, "api/v1/namespaces/infrastructure/configmaps/argocd-rbac-cm", method="PATCH", data=rbac_patch, content_type="application/strategic-merge-patch+json" ) print("argocd-rbac-cm updated.") # 5. Rollout restart argocd-server print("\nRestarting deployment 'argocd-server'...") 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/infrastructure/deployments/argocd-server", method="PATCH", data=restart_patch, content_type="application/strategic-merge-patch+json" ) print("Rollout restart triggered. Waiting for pod to become ready...") for _ in range(15): time.sleep(3) pods = k8s_request(args.k8s_url, args.k8s_token, "api/v1/namespaces/infrastructure/pods?labelSelector=app.kubernetes.io/name=argocd-server").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" * Pod: {p['metadata']['name']} (Phase: {phase}, Ready: {ready})") if phase == "Running" and ready: print("argocd-server pod is Ready!") break print("\n=========================================================") print("=== STEP 3: LIVE VERIFICATION ===") print("=========================================================") login_url = f"{args.argo_url.rstrip('/')}/auth/login" print(f"Probing Argo CD 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 "authentik" in content.lower() or "login" in content.lower(): print("SUCCESS: Argo CD login page loaded with SSO active!") except Exception as e: print(f"Note: Could not probe public page directly ({e}).") print("\n=========================================================") print("ARGO CD AUTHENTIK SSO INTEGRATION COMPLETE!") print(f"Client ID: {client_id}") print(f"Issuer URL: {issuer_url}") print(f"Redirect URI: {redirect_uri}") print(f"Role Mapping: 'authentik Admins' -> role:admin | Other Users -> role:readonly") print(f"Login URL: {args.argo_url}") print("=========================================================") if __name__ == "__main__": main()