"""Automate end-to-end OAuth2/OIDC SSO integration between Authentik and Semaphore UI. Performs: 1. Authentik OAuth2 Provider creation with redirect URI: https://semaphore.snarfnet.net/api/auth/oidc/authentik/redirect/ 2. Authentik Application creation ('semaphore') linked to the provider. 3. Patching Semaphore Deployment environment with SEMAPHORE_OIDC_PROVIDERS. 4. Rollout restart of 'semaphore' deployment. 5. Live verification of Semaphore 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 Semaphore UI") 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("--semaphore-url", default="https://semaphore.snarfnet.net", help="Semaphore Public Base URL") args = parser.parse_args() print("=========================================================") print("=== STEP 1: AUTHENTIK OIDC PROVIDER FOR SEMAPHORE ===") 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.semaphore_url.rstrip('/')}/api/auth/oidc/authentik/redirect/" 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") == "Semaphore"), 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 Semaphore OAuth2 Provider (ID {provider_id})...") provider_payload = { "name": "Semaphore", "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://semaphore\.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 Semaphore OAuth2 Provider...") client_id = f"semaphore-{secrets.token_hex(4)}" client_secret = secrets.token_urlsafe(40) provider_payload = { "name": "Semaphore", "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://semaphore\.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") == "semaphore"), None) app_payload = { "name": "Semaphore", "slug": "semaphore", "provider": provider_id, "meta_launch_url": args.semaphore_url, "open_in_new_tab": True, } if existing_app: authentik_request(args.authentik_url, args.authentik_token, "core/applications/semaphore/", method="PATCH", data=app_payload) print("Authentik Application 'semaphore' updated.") else: authentik_request(args.authentik_url, args.authentik_token, "core/applications/", method="POST", data=app_payload) print("Authentik Application 'semaphore' created.") provider_url = f"{args.authentik_url.rstrip('/')}/application/o/semaphore/" print("\n=========================================================") print("=== STEP 2: CONFIGURE SEMAPHORE DEPLOYMENT ENVIRONMENT ===") print("=========================================================") # 3. Build SEMAPHORE_OIDC_PROVIDERS json string oidc_config = { "authentik": { "display_name": "Sign in with Authentik", "provider_url": provider_url, "client_id": client_id, "client_secret": client_secret, "redirect_url": redirect_uri, "username_claim": "preferred_username", "name_claim": "name", "email_claim": "email", "scopes": ["openid", "profile", "email", "groups"] } } # 4. Read Semaphore Deployment dep = k8s_request(args.k8s_url, args.k8s_token, "apis/apps/v1/namespaces/infrastructure/deployments/semaphore") container = dep["spec"]["template"]["spec"]["containers"][0] existing_env = container.get("env", []) # Filter out old SEMAPHORE_OIDC_PROVIDERS if present new_env = [e for e in existing_env if e["name"] != "SEMAPHORE_OIDC_PROVIDERS"] new_env.append({ "name": "SEMAPHORE_OIDC_PROVIDERS", "value": json.dumps(oidc_config) }) 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 'semaphore' with SEMAPHORE_OIDC_PROVIDERS...") k8s_request( args.k8s_url, args.k8s_token, "apis/apps/v1/namespaces/infrastructure/deployments/semaphore", method="PATCH", data=dep_patch, content_type="application/strategic-merge-patch+json" ) print("Deployment patched. Waiting for Semaphore 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=semaphore").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("Semaphore pod is Ready!") break print("\n=========================================================") print("=== STEP 3: LIVE VERIFICATION ===") print("=========================================================") login_url = f"{args.semaphore_url.rstrip('/')}/auth/login" print(f"Probing Semaphore 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 "oidc" in content.lower() or "authentik" in content.lower() or "login" in content.lower(): print("SUCCESS: Semaphore login page loaded with Authentik SSO active!") except Exception as e: print(f"Note: Could not probe public page directly ({e}).") print("\n=========================================================") print("SEMAPHORE AUTHENTIK SSO INTEGRATION COMPLETE!") print(f"Client ID: {client_id}") print(f"Provider URL: {provider_url}") print(f"Redirect URI: {redirect_uri}") print(f"Login URL: {args.semaphore_url}") print("=========================================================") if __name__ == "__main__": main()