"""Automate end-to-end OAuth2/OIDC SSO integration between Authentik and Gitea. Performs: 1. Authentik OAuth2 Provider creation with redirect URI: https://git.snarfnet.net/user/oauth2/authentik/callback 2. Authentik Application creation ('gitea') linked to the provider. 3. Execution of 'gitea admin auth add-oauth' via Kubernetes Job to register the OIDC provider with group mapping. 4. Verification of the live Gitea 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 Gitea") 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("--gitea-url", default="https://git.snarfnet.net", help="Gitea Public Base URL") args = parser.parse_args() print("=========================================================") print("=== STEP 1: AUTHENTIK OIDC PROVIDER FOR GITEA ===") print("=========================================================") # 1. Resolve Flows & Keys 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.gitea_url.rstrip('/')}/user/oauth2/authentik/callback" # 2. Check or Create OAuth2 Provider 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") == "Gitea"), 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 Gitea OAuth2 Provider (ID {provider_id})...") provider_payload = { "name": "Gitea", "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 Gitea OAuth2 Provider...") client_id = f"gitea-{secrets.token_hex(4)}" client_secret = secrets.token_urlsafe(40) provider_payload = { "name": "Gitea", "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}") # 3. 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") == "gitea"), None) app_payload = { "name": "Gitea", "slug": "gitea", "provider": provider_id, "meta_launch_url": args.gitea_url, "open_in_new_tab": True, } if existing_app: authentik_request(args.authentik_url, args.authentik_token, "core/applications/gitea/", method="PATCH", data=app_payload) print("Authentik Application 'gitea' updated.") else: authentik_request(args.authentik_url, args.authentik_token, "core/applications/", method="POST", data=app_payload) print("Authentik Application 'gitea' created.") auto_discover_url = f"{args.authentik_url.rstrip('/')}/application/o/gitea/.well-known/openid-configuration" print("\n=========================================================") print("=== STEP 2: REGISTER AUTHENTIK IN GITEA ===") print("=========================================================") # 4. Run Job to execute gitea admin auth add-oauth job_name = f"gitea-add-oauth-{int(time.time())}" print(f"Launching K8s Job '{job_name}' to register OAuth2 source in Gitea...") job_manifest = { "apiVersion": "batch/v1", "kind": "Job", "metadata": {"name": job_name, "namespace": "infrastructure"}, "spec": { "template": { "spec": { "restartPolicy": "Never", "securityContext": { "fsGroup": 1000, "runAsUser": 1000, "runAsGroup": 1000 }, "nodeSelector": {"kubernetes.io/hostname": "sninfr0lnxwrk01"}, "containers": [{ "name": "gitea-cli", "image": "docker.gitea.com/gitea:1.25.5-rootless", "command": ["/bin/sh", "-c"], "args": [ f""" /usr/local/bin/gitea admin auth delete --name authentik --config /data/gitea/conf/app.ini --work-path /data 2>/dev/null || true /usr/local/bin/gitea admin auth add-oauth \ --name authentik \ --provider openidConnect \ --key '{client_id}' \ --secret '{client_secret}' \ --auto-discover-url '{auto_discover_url}' \ --scopes openid --scopes email --scopes profile --scopes groups \ --group-claim-name groups \ --admin-group 'authentik Admins' \ --config /data/gitea/conf/app.ini \ --work-path /data echo "Listing authentication sources:" /usr/local/bin/gitea admin auth list --config /data/gitea/conf/app.ini --work-path /data """ ], "volumeMounts": [ {"name": "data", "mountPath": "/data"} ] }], "volumes": [ {"name": "data", "persistentVolumeClaim": {"claimName": "gitea-shared-storage"}} ] } } } } k8s_request(args.k8s_url, args.k8s_token, "apis/batch/v1/namespaces/infrastructure/jobs", method="POST", data=job_manifest) # Poll Job completion for _ in range(12): time.sleep(3) pods = k8s_request(args.k8s_url, args.k8s_token, f"api/v1/namespaces/infrastructure/pods?labelSelector=job-name={job_name}").get("items", []) if pods: phase = pods[0]["status"].get("phase", "") if phase in ["Succeeded", "Failed"]: pname = pods[0]["metadata"]["name"] try: logs = urllib.request.urlopen( urllib.request.Request(f"{args.k8s_url}/api/v1/namespaces/infrastructure/pods/{pname}/log", headers={"Authorization": f"Bearer {args.k8s_token}"}), timeout=5, context=ctx ).read().decode("utf-8") print(logs) except Exception as e: print("Log read note:", e) break # Clean up job try: k8s_request(args.k8s_url, args.k8s_token, f"apis/batch/v1/namespaces/infrastructure/jobs/{job_name}?propagationPolicy=Foreground", method="DELETE") except: pass print("\n=========================================================") print("=== STEP 3: LIVE VERIFICATION ===") print("=========================================================") login_url = f"{args.gitea_url.rstrip('/')}/user/login" print(f"Probing Gitea 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 "oauth2" in content or "authentik" in content.lower(): print("SUCCESS: 'Sign in with authentik' is active on Gitea login screen!") else: print("Gitea login page loaded successfully. HTTP 200 OK.") except Exception as e: print(f"Note: Could not probe public page directly ({e}).") print("\n=========================================================") print("GITEA AUTHENTIK SSO INTEGRATION COMPLETE!") print(f"Client ID: {client_id}") print(f"Redirect URI: {redirect_uri}") print(f"Admin Group: 'authentik Admins'") print(f"Login URL: {args.gitea_url}") print("=========================================================") if __name__ == "__main__": main()