"""Automate end-to-end OAuth2/OIDC SSO integration between Authentik and Coder. Performs: 1. Authentik OAuth2 Provider creation with redirect URI: https://coder.snarfnet.net/api/v2/users/oidc/callback 2. Authentik Application creation ('coder') linked to the provider. 3. Patching Coder Deployment with CODER_OIDC_* environment variables. 4. Rollout restart of 'coder' deployment. 5. Live verification of Coder login and OIDC endpoints. """ 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 Coder") 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("--coder-url", default="https://coder.snarfnet.net", help="Coder Public Base URL") args = parser.parse_args() print("=========================================================") print("=== STEP 1: AUTHENTIK OIDC PROVIDER FOR CODER ===") 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.coder_url.rstrip('/')}/api/v2/users/oidc/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") == "Coder"), 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 Coder OAuth2 Provider (ID {provider_id})...") provider_payload = { "name": "Coder", "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": "regex", "url": r"https://coder\.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 Coder OAuth2 Provider...") client_id = f"coder-{secrets.token_hex(4)}" client_secret = secrets.token_urlsafe(40) provider_payload = { "name": "Coder", "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": "regex", "url": r"https://coder\.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") == "coder"), None) app_payload = { "name": "Coder", "slug": "coder", "provider": provider_id, "meta_launch_url": args.coder_url, "open_in_new_tab": True, } if existing_app: authentik_request(args.authentik_url, args.authentik_token, "core/applications/coder/", method="PATCH", data=app_payload) print("Authentik Application 'coder' updated.") else: authentik_request(args.authentik_url, args.authentik_token, "core/applications/", method="POST", data=app_payload) print("Authentik Application 'coder' created.") issuer_url = f"{args.authentik_url.rstrip('/')}/application/o/coder/" print("\n=========================================================") print("=== STEP 2: CONFIGURE CODER DEPLOYMENT ENVIRONMENT ===") print("=========================================================") # 3. Read Coder Deployment dep = k8s_request(args.k8s_url, args.k8s_token, "apis/apps/v1/namespaces/infrastructure/deployments/coder") container = dep["spec"]["template"]["spec"]["containers"][0] existing_env = container.get("env", []) # Filter out any old OIDC envs new_env = [e for e in existing_env if not e["name"].startswith("CODER_OIDC_")] oidc_envs = [ {"name": "CODER_OIDC_ISSUER_URL", "value": issuer_url}, {"name": "CODER_OIDC_CLIENT_ID", "value": client_id}, {"name": "CODER_OIDC_CLIENT_SECRET", "value": client_secret}, {"name": "CODER_OIDC_SCOPES", "value": "openid,profile,email,groups"}, {"name": "CODER_OIDC_SIGN_IN_TEXT", "value": "Sign in with Authentik"}, {"name": "CODER_OIDC_ALLOW_SIGNUPS", "value": "true"}, ] new_env.extend(oidc_envs) container_name = container["name"] dep_patch = { "spec": { "template": { "metadata": { "annotations": { "kubectl.kubernetes.io/restartedAt": datetime.datetime.now(datetime.timezone.utc).isoformat() } }, "spec": { "containers": [ { "name": container_name, "env": new_env } ] } } } } print("Patching deployment 'coder' with OIDC environment variables...") k8s_request( args.k8s_url, args.k8s_token, "apis/apps/v1/namespaces/infrastructure/deployments/coder", method="PATCH", data=dep_patch, content_type="application/strategic-merge-patch+json" ) print("Deployment patched. Waiting for Coder pod to reach Ready status...") 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=coder").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("Coder pod is Ready!") break print("\n=========================================================") print("=== STEP 3: LIVE VERIFICATION ===") print("=========================================================") login_url = f"{args.coder_url.rstrip('/')}/login" print(f"Probing Coder 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 "oidc" in content.lower() or "login" in content.lower(): print("SUCCESS: Coder login page loaded with Authentik SSO active!") except Exception as e: print(f"Note: Could not probe public page directly ({e}).") print("\n=========================================================") print("CODER AUTHENTIK SSO INTEGRATION COMPLETE!") print(f"Client ID: {client_id}") print(f"Issuer URL: {issuer_url}") print(f"Redirect URI: {redirect_uri}") print(f"Login URL: {args.coder_url}") print("=========================================================") if __name__ == "__main__": main()