feat: add homelab topology documentation, docker deployments, and authentik sso configuration scripts

This commit is contained in:
2026-10-30 12:56:53 -04:00
parent e07e2204e3
commit 6b12646a0f
24 changed files with 3902 additions and 1 deletions

View File

@@ -0,0 +1,21 @@
import json
import ssl
import urllib.request
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
req = urllib.request.Request(
'https://authentik.snarfnet.net/api/v3/events/events/?page_size=25',
headers={'Authorization': 'Bearer SnarfNetPermanentAutomationToken2026'}
)
res = json.loads(urllib.request.urlopen(req, context=ctx).read())
print(f"Total events found: {len(res['results'])}")
for e in res['results']:
action = e.get('action')
app = e.get('app')
user = e.get('user', {}).get('username')
created = e.get('created')
context = json.dumps(e.get('context', {}))
print(f"{created} | {action:25} | app: {app:20} | user: {user} | {context[:120]}")

View File

@@ -0,0 +1,450 @@
r"""Customize Authentik application dashboard with high-resolution icons, groups, and launch URLs.
Full homelab catalog covering all active services across:
- CI/CD & Automation
- Developer Tools
- Infrastructure & Hypervisors
- Storage & Media
- Smart Home & IoT
- Observability & Operations
- Networking & Edge
- Hardware & Server BMC
- Portals & Dashboards
"""
import json
import ssl
import sys
import urllib.error
import urllib.request
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
AK_BASE = "https://authentik.snarfnet.net"
AK_TOKEN = "SnarfNetPermanentAutomationToken2026"
APPLICATIONS = [
# CI/CD & Automation
{
"slug": "argo-cd",
"name": "Argo CD",
"group": "CI/CD & Automation",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/argo-cd.png",
"meta_description": "Declarative GitOps Continuous Delivery",
"meta_launch_url": "https://argo.snarfnet.net",
"open_in_new_tab": True,
},
{
"slug": "awx",
"name": "Ansible AWX",
"group": "CI/CD & Automation",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/ansible.png",
"meta_description": "Ansible Automation Controller & Playbook Execution",
"meta_launch_url": "https://awx.snarfnet.net",
"open_in_new_tab": True,
},
{
"slug": "jenkins",
"name": "Jenkins",
"group": "CI/CD & Automation",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/jenkins.png",
"meta_description": "Continuous Integration & Build Server",
"meta_launch_url": "https://jenkins.snarfnet.net",
"open_in_new_tab": True,
},
{
"slug": "semaphore",
"name": "Semaphore UI",
"group": "CI/CD & Automation",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/semaphore.png",
"meta_description": "Modern Web UI for Ansible & Task Automation",
"meta_launch_url": "https://semaphore.snarfnet.net",
"open_in_new_tab": True,
},
# Developer Tools
{
"slug": "coder",
"name": "Coder",
"group": "Developer Tools",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/coder.png",
"meta_description": "Self-Hosted Cloud Development Environments",
"meta_launch_url": "https://coder.snarfnet.net",
"open_in_new_tab": True,
},
{
"slug": "gitea",
"name": "Gitea",
"group": "Developer Tools",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/gitea.png",
"meta_description": "Git Version Control & Repository Hosting",
"meta_launch_url": "https://git.snarfnet.net",
"open_in_new_tab": True,
},
{
"slug": "vscode",
"name": "VS Code Server",
"group": "Developer Tools",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/vscode.png",
"meta_description": "Browser-Based Code Editor & Remote Terminal",
"meta_launch_url": "http://code.snarfnet.net",
"open_in_new_tab": True,
},
# Infrastructure & Hypervisors
{
"slug": "rancher",
"name": "Rancher",
"group": "Infrastructure & Hypervisors",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/rancher.png",
"meta_description": "Enterprise Multi-Cluster Kubernetes Management",
"meta_launch_url": "https://aerodrome.snarfnet.net",
"open_in_new_tab": True,
},
{
"slug": "harvester",
"name": "Harvester HCI",
"group": "Infrastructure & Hypervisors",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/harvester.png",
"meta_description": "Hyperconverged Infrastructure & Virtualization",
"meta_launch_url": "https://harvester.spacedock.snarfnet.net",
"open_in_new_tab": True,
},
{
"slug": "portainer",
"name": "Portainer EE",
"group": "Infrastructure & Hypervisors",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/portainer.png",
"meta_description": "Docker Swarm & Container Management (OAuth Active)",
"meta_launch_url": "https://shipyard.snarfnet.net",
"open_in_new_tab": True,
},
{
"slug": "meshcentral",
"name": "MeshCentral",
"group": "Infrastructure & Hypervisors",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/meshcentral.png",
"meta_description": "Remote Device Management & Web KVM",
"meta_launch_url": "https://mesh.snarfnet.net",
"open_in_new_tab": True,
},
{
"slug": "longhorn-aerodrome",
"name": "Longhorn (Aerodrome)",
"group": "Infrastructure & Hypervisors",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/longhorn.png",
"meta_description": "Distributed Block Storage (Aerodrome Cluster)",
"meta_launch_url": "https://longhorn.aerodrome.snarfnet.net",
"open_in_new_tab": True,
},
{
"slug": "longhorn-spaceport",
"name": "Longhorn (Spaceport)",
"group": "Infrastructure & Hypervisors",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/longhorn.png",
"meta_description": "Distributed Block Storage (Spaceport Cluster)",
"meta_launch_url": "https://longhorn.spaceport.snarfnet.net",
"open_in_new_tab": True,
},
# Storage & Media
{
"slug": "snarfnas",
"name": "Synology GA (SnarfNAS)",
"group": "Storage & Media",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/synology.png",
"meta_description": "Primary Network Attached Storage (DSM 7)",
"meta_launch_url": "https://snarfnas.snarfnet.net",
"open_in_new_tab": True,
},
{
"slug": "sn-nas",
"name": "Synology SC (SN-NAS)",
"group": "Storage & Media",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/synology.png",
"meta_description": "Secondary Storage & Backup NAS (DSM 7)",
"meta_launch_url": "https://sn-nas.snarfnet.net",
"open_in_new_tab": True,
},
{
"slug": "plex",
"name": "Plex Media Server",
"group": "Storage & Media",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/plex.png",
"meta_description": "Home Streaming Media, Movies & TV Shows",
"meta_launch_url": "https://plex.snarfnet.net",
"open_in_new_tab": True,
},
{
"slug": "kavita",
"name": "Kavita",
"group": "Storage & Media",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/kavita.png",
"meta_description": "Digital Comic, Manga & Book Library",
"meta_launch_url": "https://kavita.snarfnet.net",
"open_in_new_tab": True,
},
# Smart Home & IoT
{
"slug": "home-assistant",
"name": "Home Assistant",
"group": "Smart Home & IoT",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/home-assistant.png",
"meta_description": "Smart Home Automation Hub & Sensors",
"meta_launch_url": "https://snarfhome.snarfnet.net",
"open_in_new_tab": True,
},
{
"slug": "wled",
"name": "WLED Lights",
"group": "Smart Home & IoT",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/wled.png",
"meta_description": "Office Addressable LED Lighting Controller",
"meta_launch_url": "https://lights.snarfnet.net",
"open_in_new_tab": True,
},
# Observability & Operations
{
"slug": "grafana",
"name": "Grafana",
"group": "Observability & Operations",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/grafana.png",
"meta_description": "Dashboards, Metrics & Visualizations",
"meta_launch_url": "https://grafana.snarfnet.net",
"open_in_new_tab": True,
},
{
"slug": "uptime-kuma",
"name": "Uptime Kuma",
"group": "Observability & Operations",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/uptime-kuma.png",
"meta_description": "Service Uptime & Status Monitoring",
"meta_launch_url": "https://uptime.snarfnet.net",
"open_in_new_tab": True,
},
{
"slug": "zabbix",
"name": "Zabbix",
"group": "Observability & Operations",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/zabbix.png",
"meta_description": "Enterprise Network & Infrastructure Monitoring",
"meta_launch_url": "https://zabbix.snarfnet.net",
"open_in_new_tab": True,
},
{
"slug": "peanut",
"name": "PeaNUT",
"group": "Observability & Operations",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/nut.png",
"meta_description": "UPS Power & Server Closet Battery Monitoring",
"meta_launch_url": "https://peanut.snarfnet.net/device/server-closet",
"open_in_new_tab": True,
},
# Networking & Edge
{
"slug": "pihole",
"name": "Pi-hole Web",
"group": "Networking & Edge",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/pi-hole.png",
"meta_description": "Network DNS Sinkhole & Ad Blocker",
"meta_launch_url": "https://pihole3.snarfnet.net",
"open_in_new_tab": True,
},
{
"slug": "nginx-proxy",
"name": "Nginx Proxy Manager",
"group": "Networking & Edge",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/nginx-proxy-manager.png",
"meta_description": "Edge Reverse Proxy & SSL Certificate Admin",
"meta_launch_url": "https://nginx.snarfnet.net",
"open_in_new_tab": True,
},
{
"slug": "gotify",
"name": "Gotify",
"group": "Networking & Edge",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/gotify.png",
"meta_description": "Self-Hosted Push Notification Server",
"meta_launch_url": "https://gotify.snarfnet.net",
"open_in_new_tab": True,
},
{
"slug": "unifi",
"name": "UniFi Cloud",
"group": "Networking & Edge",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/unifi.png",
"meta_description": "Ubiquiti UniFi Network & Gateway Controller",
"meta_launch_url": "https://unifi.ui.com",
"open_in_new_tab": True,
},
# Hardware & Server BMC
{
"slug": "idrac-1",
"name": "iDRAC 1 (Spacedock 1)",
"group": "Hardware & Server BMC",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/dell.png",
"meta_description": "Dell PowerEdge Server 1 Out-of-Band BMC",
"meta_launch_url": "https://spacedock1.snarfnet.net",
"open_in_new_tab": True,
},
{
"slug": "idrac-2",
"name": "iDRAC 2 (Spacedock 2)",
"group": "Hardware & Server BMC",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/dell.png",
"meta_description": "Dell PowerEdge Server 2 Out-of-Band BMC",
"meta_launch_url": "https://spacedock2.snarfnet.net",
"open_in_new_tab": True,
},
{
"slug": "idrac-3",
"name": "iDRAC 3 (Spacedock 3)",
"group": "Hardware & Server BMC",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/dell.png",
"meta_description": "Dell PowerEdge Server 3 Out-of-Band BMC",
"meta_launch_url": "https://spacedock3.snarfnet.net",
"open_in_new_tab": True,
},
{
"slug": "idrac-4",
"name": "iDRAC 4 (Spacedock 4)",
"group": "Hardware & Server BMC",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/dell.png",
"meta_description": "Dell PowerEdge Server 4 Out-of-Band BMC",
"meta_launch_url": "https://spacedock4.snarfnet.net",
"open_in_new_tab": True,
},
# Portals & Dashboards
{
"slug": "snarfnet-homelab",
"name": "Homepage",
"group": "Portals & Dashboards",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/homepage.png",
"meta_description": "SnarfNet Homelab Central Portal",
"meta_launch_url": "https://homepage.snarfnet.net",
"open_in_new_tab": True,
},
{
"slug": "homebox",
"name": "Homebox",
"group": "Portals & Dashboards",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/homebox.png",
"meta_description": "Physical Asset & Inventory Management",
"meta_launch_url": "https://homebox.snarfnet.net",
"open_in_new_tab": True,
},
{
"slug": "gaia",
"name": "Gaia Web App",
"group": "Portals & Dashboards",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/open-webui.png",
"meta_description": "Swarm Services & AI Stack Portal",
"meta_launch_url": "https://gaia.snarfnet.net",
"open_in_new_tab": True,
},
{
"slug": "dakboard",
"name": "Dakboard Display",
"group": "Portals & Dashboards",
"meta_icon": "https://dakboard.com/favicon.ico",
"meta_description": "Custom Digital Wall Calendar & Display",
"meta_launch_url": "https://dakboard.com/site",
"open_in_new_tab": True,
},
]
def ak_request(endpoint, method="GET", data=None):
url = f"{AK_BASE.rstrip('/')}/{endpoint.lstrip('/')}"
body = json.dumps(data).encode("utf-8") if data else None
headers = {
"Authorization": f"Bearer {AK_TOKEN}",
"Content-Type": "application/json",
"Accept": "application/json",
}
req = urllib.request.Request(url, data=body, headers=headers, method=method)
try:
with urllib.request.urlopen(req, timeout=10, context=ctx) as r:
res_body = r.read().decode("utf-8")
return r.getcode(), json.loads(res_body) if res_body else {}
except urllib.error.HTTPError as e:
err_msg = e.read().decode("utf-8")
return e.code, err_msg
def fetch_all_apps():
apps = []
page = 1
while True:
code, res = ak_request(f"/api/v3/core/applications/?page={page}")
if code != 200 or not res.get("results"):
break
apps.extend(res["results"])
if not res.get("pagination", {}).get("next"):
break
page += 1
return {a["slug"]: a for a in apps}
def main():
print("=" * 60, flush=True)
print("SnarfNet Authentik Application Customizer (Full Catalog)", flush=True)
print("=" * 60, flush=True)
# 1. Fetch current applications
existing_apps = fetch_all_apps()
print(f"Found {len(existing_apps)} existing applications in Authentik.\n", flush=True)
for app in APPLICATIONS:
slug = app["slug"]
if slug in existing_apps:
# Preserve existing provider if assigned
patch_data = {
"name": app["name"],
"group": app["group"],
"meta_icon": app["meta_icon"],
"meta_description": app["meta_description"],
"meta_launch_url": app["meta_launch_url"],
"open_in_new_tab": app["open_in_new_tab"],
}
code, r = ak_request(f"/api/v3/core/applications/{slug}/", method="PATCH", data=patch_data)
if code == 200:
print(f"[UPDATED] {slug:20} -> {app['name']:28} ({app['group']})", flush=True)
else:
print(f"[FAIL] {slug:20} -> Error {code}: {r}", flush=True)
else:
# POST new application without provider (standalone launcher)
post_data = {
"name": app["name"],
"slug": app["slug"],
"group": app["group"],
"meta_icon": app["meta_icon"],
"meta_description": app["meta_description"],
"meta_launch_url": app["meta_launch_url"],
"open_in_new_tab": app["open_in_new_tab"],
}
code, r = ak_request("/api/v3/core/applications/", method="POST", data=post_data)
if code == 201:
print(f"[CREATED] {slug:20} -> {app['name']:28} ({app['group']})", flush=True)
else:
print(f"[FAIL] {slug:20} -> Error {code}: {r}", flush=True)
print("\n" + "=" * 60, flush=True)
print("Verifying updated applications catalog:", flush=True)
print("=" * 60, flush=True)
all_apps = fetch_all_apps()
for slug, a in sorted(all_apps.items(), key=lambda x: (x[1].get("group") or "", x[1].get("name") or "")):
prov_name = a.get("provider_obj", {}).get("name") if a.get("provider_obj") else "Direct"
print(f"[{a.get('group', 'Default'):28}] {a['name']:28} (Auth: {prov_name:18}) -> {a.get('launch_url')}", flush=True)
print("\nFull catalog update complete!", flush=True)
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,266 @@
"""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()

View File

@@ -0,0 +1,67 @@
import urllib.request
import json
import ssl
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
AK_BASE = "https://authentik.snarfnet.net"
AK_TOKEN = "SnarfNetPermanentAutomationToken2026"
def ak_req(endpoint, method="GET", data=None):
url = f"{AK_BASE}/api/v3/{endpoint.lstrip('/')}"
body = json.dumps(data).encode("utf-8") if data else None
headers = {
"Authorization": f"Bearer {AK_TOKEN}",
"Content-Type": "application/json",
"Accept": "application/json"
}
req = urllib.request.Request(url, data=body, headers=headers, method=method)
with urllib.request.urlopen(req, timeout=10, context=ctx) as r:
raw = r.read().decode("utf-8")
return r.getcode(), json.loads(raw) if raw else {}
# 1. Check or Create Flow
code, flows = ak_req("flows/instances/")
flow = next((f for f in flows.get("results", []) if f["slug"] == "ldap-bind-flow"), None)
if not flow:
payload = {
"name": "LDAP Bind Flow",
"slug": "ldap-bind-flow",
"title": "LDAP Bind Flow",
"designation": "authentication",
}
code, flow = ak_req("flows/instances/", method="POST", data=payload)
print("Created flow:", flow["pk"])
else:
print("Flow exists:", flow["pk"])
flow_pk = flow["pk"]
# 2. Check or Create Bindings
code, bindings = ak_req(f"flows/bindings/?target={flow_pk}")
existing_stage_pks = [b["stage"] for b in bindings.get("results", [])]
stages = [
(10, "f46eee4d-5bf7-415a-9eea-ba589536f90a"), # Identification
(20, "8f68c5d8-52f8-49a2-a578-5f43846d5edb"), # Password
(30, "84c2290f-eed3-464d-a997-5b70bb435b7e"), # Login
]
for order, stage_pk in stages:
if stage_pk not in existing_stage_pks:
b_payload = {
"target": flow_pk,
"stage": stage_pk,
"order": order,
}
code, new_b = ak_req("flows/bindings/", method="POST", data=b_payload)
print(f"Bound stage {stage_pk} at order {order}")
else:
print(f"Stage {stage_pk} already bound")
# 3. Update Provider 22
code, prov = ak_req("providers/ldap/22/", method="PATCH", data={"authentication_flow": flow_pk})
print("Updated Provider 22 authentication_flow to:", prov.get("authentication_flow"))

View File

@@ -0,0 +1,249 @@
"""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()

View File

@@ -0,0 +1,263 @@
"""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()

View File

@@ -0,0 +1,325 @@
"""Automate end-to-end OAuth2/OIDC SSO integration between Authentik and Grafana 11.
Performs:
1. Authentik OAuth2 Provider creation with redirect URIs and scope mappings (openid, email, profile, groups).
2. Authentik Application creation ('grafana') linked to the provider.
3. Kubernetes ConfigMap 'grafana' update in namespace 'default' with [auth.generic_oauth] settings and role mapping.
4. Grafana Deployment rollout restart to reload configuration.
5. Live verification of the Grafana SSO 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,
)
try:
with urllib.request.urlopen(req, timeout=10, context=ctx) as resp:
content = resp.read().decode("utf-8")
return json.loads(content) if content else {}
except urllib.error.HTTPError as e:
err_msg = e.read().decode("utf-8", errors="ignore")
raise RuntimeError(f"Authentik API Error [{e.code}]: {err_msg}")
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 Grafana")
parser.add_argument("--authentik-url", default="https://authentik.snarfnet.net", help="Authentik Base URL")
parser.add_argument("--authentik-token", default="oRq4mTt7i5TtQ4IGV9BWjheywOZaStqaJyU0m7J14pJnzJiVC6r9xFjkDYDp", help="Authentik API Token")
parser.add_argument("--k8s-url", default="https://aerodrome.snarfnet.net/k8s/clusters/local", help="Aerodrome K8s API Base URL")
parser.add_argument("--k8s-token", default="kubeconfig-u-aqzkufonckwqpfv:4nqwf2z7cqkbnt9jfvzpbh5hc6bkfghkjtgnrpqkx7zv7lmxbxg5z8", help="Aerodrome K8s Bearer Token")
parser.add_argument("--grafana-url", default="https://grafana.snarfnet.net", help="Grafana Public Base URL")
args = parser.parse_args()
print("=========================================================")
print("=== STEP 1: AUTHENTIK OIDC PROVIDER & APP CONFIGURATION ===")
print("=========================================================")
# 1. Discover Flows
print("Resolving Authentik Authorization & Invalidation flows...")
flows = authentik_request(args.authentik_url, args.authentik_token, "flows/instances/").get("results", [])
auth_flow_pk = None
inval_flow_pk = None
for f in flows:
slug = f.get("slug", "")
desig = f.get("designation", "")
if "implicit-consent" in slug or (desig == "authorization" and not auth_flow_pk):
auth_flow_pk = f["pk"]
if "default-provider-invalidation" in slug or (desig == "invalidation" and not inval_flow_pk):
inval_flow_pk = f["pk"]
print(f" * Authorization Flow: {auth_flow_pk}")
print(f" * Invalidation Flow: {inval_flow_pk}")
# 2. Discover Signing Key
print("Resolving Authentik Signing Certificate...")
certs = authentik_request(args.authentik_url, args.authentik_token, "crypto/certificatekeypairs/").get("results", [])
cert_pk = certs[0]["pk"] if certs else None
for c in certs:
if "authentik" in c.get("name", "").lower() or "snarfnet" in c.get("name", "").lower():
cert_pk = c["pk"]
break
print(f" * Signing Key: {cert_pk}")
# 3. Discover Scope Mappings (openid, email, profile, groups)
print("Resolving Scope Mappings...")
scope_mappings = authentik_request(args.authentik_url, args.authentik_token, "propertymappings/provider/scope/").get("results", [])
selected_scopes = []
for sm in scope_mappings:
sname = sm.get("scope_name", "")
if sname in ["openid", "email", "profile", "groups"]:
selected_scopes.append(sm["pk"])
print(f" * Found scope mapping '{sname}': {sm['pk']}")
# 4. Check or Create OAuth2 Provider
print("\nChecking existing OAuth2 Providers in Authentik...")
providers = authentik_request(args.authentik_url, args.authentik_token, "providers/oauth2/").get("results", [])
existing_provider = None
for p in providers:
if p.get("name") == "Grafana":
existing_provider = p
break
redirect_uri = f"{args.grafana_url.rstrip('/')}/login/generic_oauth"
if existing_provider:
provider_id = existing_provider["pk"]
client_id = existing_provider["client_id"]
client_secret = existing_provider["client_secret"]
print(f"Updating existing Grafana OAuth2 Provider (ID {provider_id})...")
provider_payload = {
"name": "Grafana",
"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 Grafana OAuth2 Provider...")
client_id = f"grafana-{secrets.token_hex(4)}"
client_secret = secrets.token_urlsafe(40)
provider_payload = {
"name": "Grafana",
"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 with ID: {provider_id}, Client ID: {client_id}")
# 5. Check or Create Application in Authentik
print("\nChecking Authentik Applications...")
apps = authentik_request(args.authentik_url, args.authentik_token, "core/applications/").get("results", [])
existing_app = None
for a in apps:
if a.get("slug") == "grafana":
existing_app = a
break
app_payload = {
"name": "Grafana",
"slug": "grafana",
"provider": provider_id,
"meta_launch_url": args.grafana_url,
"open_in_new_tab": True,
}
if existing_app:
print("Updating existing Grafana Application...")
authentik_request(args.authentik_url, args.authentik_token, f"core/applications/{existing_app['slug']}/", method="PATCH", data=app_payload)
print("Application updated.")
else:
print("Creating new Grafana Application...")
authentik_request(args.authentik_url, args.authentik_token, "core/applications/", method="POST", data=app_payload)
print("Created Application 'Grafana'.")
print("\n=========================================================")
print("=== STEP 2: UPDATE GRAFANA KUBERNETES CONFIGMAP ===")
print("=========================================================")
# 6. Read existing configmap
print("Fetching ConfigMap 'grafana' in namespace 'default'...")
cm = k8s_request(args.k8s_url, args.k8s_token, "api/v1/namespaces/default/configmaps/grafana")
new_grafana_ini = f"""[analytics]
check_for_updates = true
[grafana_net]
url = {args.grafana_url}
[log]
mode = console
[paths]
data = /var/lib/grafana/
logs = /var/log/grafana
plugins = /var/lib/grafana/plugins
provisioning = /etc/grafana/provisioning
[server]
root_url = {args.grafana_url}
domain = grafana.snarfnet.net
enforce_domain = false
[auth.generic_oauth]
enabled = true
name = Authentik
allow_sign_up = true
auto_login = false
client_id = {client_id}
client_secret = {client_secret}
scopes = openid email profile groups
empty_scopes = false
auth_url = {args.authentik_url.rstrip('/')}/application/o/authorize/
token_url = {args.authentik_url.rstrip('/')}/application/o/token/
api_url = {args.authentik_url.rstrip('/')}/application/o/userinfo/
login_attribute_path = preferred_username
name_attribute_path = name
email_attribute_path = email
role_attribute_path = contains(groups[*], 'authentik Admins') && 'Admin' || 'Viewer'
role_attribute_strict = false
allow_assign_grafana_admin = true
"""
cm_patch = {
"data": {
"grafana.ini": new_grafana_ini
}
}
print("Patching ConfigMap with updated grafana.ini containing [auth.generic_oauth]...")
k8s_request(
args.k8s_url,
args.k8s_token,
"api/v1/namespaces/default/configmaps/grafana",
method="PATCH",
data=cm_patch,
content_type="application/strategic-merge-patch+json",
)
print("ConfigMap patched successfully.")
# 7. Restart Grafana Deployment
print("\nTriggering Grafana Deployment rollout restart...")
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/default/deployments/grafana",
method="PATCH",
data=restart_patch,
content_type="application/strategic-merge-patch+json",
)
print("Deployment patched. Waiting for new pod to reach Ready status...")
# Poll pod status
for i in range(15):
time.sleep(3)
pods = k8s_request(args.k8s_url, args.k8s_token, "api/v1/namespaces/default/pods?labelSelector=app.kubernetes.io/name=grafana")
items = pods.get("items", [])
if items:
p = items[0]
status = p.get("status", {})
phase = status.get("phase", "")
ready = any(c.get("type") == "Ready" and c.get("status") == "True" for c in status.get("conditions", []))
print(f" * Pod: {p['metadata']['name']} (Phase: {phase}, Ready: {ready})")
if phase == "Running" and ready:
print("Grafana pod is Ready!")
break
print("\n=========================================================")
print("=== STEP 3: LIVE VERIFICATION ===")
print("=========================================================")
time.sleep(2)
login_url = f"{args.grafana_url.rstrip('/')}/login"
print(f"Probing Grafana 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 "generic_oauth" in content or "Authentik" in content or "Sign in with" in content:
print("SUCCESS: Authentik SSO button ('Sign in with Authentik') is live on Grafana login page!")
else:
print("Grafana login page loaded successfully. HTTP 200 OK.")
except Exception as e:
print(f"Note: Could not probe public page directly ({e}).")
print("\n=========================================================")
print("GRAFANA AUTHENTIK SSO INTEGRATION COMPLETE!")
print(f"Client ID: {client_id}")
print(f"Redirect URI: {redirect_uri}")
print(f"Role Mapping: 'authentik Admins' -> Admin | Other Users -> Viewer")
print(f"Login URL: {args.grafana_url}")
print("=========================================================")
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,120 @@
r"""Automate provisioning of Authentik OAuth2/OIDC Provider & Application for Harvester HCI.
Pre-configures:
1. Authentik OAuth2 Provider with endpoints and Harvester redirect URIs:
- https://harvester.spacedock.snarfnet.net/verify-auth
- Regex: https://harvester\.spacedock\.snarfnet\.net.*
2. Authentik Application ('harvester') with brand icon and grouping.
3. Outputs configuration details for Harvester Auth Settings.
"""
import json
import secrets
import ssl
import sys
import urllib.request
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
AK_BASE = "https://authentik.snarfnet.net"
AK_TOKEN = "SnarfNetPermanentAutomationToken2026"
AUTH_FLOW = "38a86ef3-f43a-4756-8f54-ca008a8eef6e"
INVAL_FLOW = "7a8145b8-2476-4636-9901-825f826df805"
CERT_PK = "6d48ba88-4b51-428f-903c-93334988a941"
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
]
def ak_request(endpoint, method="GET", data=None):
url = f"{AK_BASE.rstrip('/')}/api/v3/{endpoint.lstrip('/')}"
body = json.dumps(data).encode("utf-8") if data else None
headers = {
"Authorization": f"Bearer {AK_TOKEN}",
"Content-Type": "application/json",
"Accept": "application/json",
}
req = urllib.request.Request(url, data=body, headers=headers, method=method)
with urllib.request.urlopen(req, timeout=10, context=ctx) as r:
raw = r.read().decode("utf-8")
return r.getcode(), json.loads(raw) if raw else {}
def main():
print("=" * 60)
print("Harvester HCI <-> Authentik OIDC Provisioner")
print("=" * 60)
code, res = ak_request("providers/oauth2/")
existing_provider = next((p for p in res.get("results", []) if p.get("name") == "Harvester"), None)
if existing_provider:
provider_id = existing_provider["pk"]
client_id = existing_provider["client_id"]
client_secret = existing_provider["client_secret"]
print(f"Existing Harvester Provider found (ID {provider_id})")
else:
client_id = f"harvester-{secrets.token_hex(4)}"
client_secret = secrets.token_urlsafe(40)
provider_payload = {
"name": "Harvester",
"authorization_flow": AUTH_FLOW,
"invalidation_flow": INVAL_FLOW,
"property_mappings": 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": "https://harvester.spacedock.snarfnet.net/verify-auth"},
{"matching_mode": "regex", "url": r"https://harvester\.spacedock\.snarfnet\.net.*"},
],
"sub_mode": "user_username",
"issuer_mode": "per_provider",
}
code, new_prov = ak_request("providers/oauth2/", method="POST", data=provider_payload)
provider_id = new_prov["pk"]
print(f"Created Harvester Provider (ID {provider_id}, client_id: {client_id})")
code, res = ak_request("core/applications/")
existing_app = next((a for a in res.get("results", []) if a.get("slug") == "harvester"), None)
app_payload = {
"name": "Harvester HCI",
"slug": "harvester",
"provider": provider_id,
"group": "Infrastructure & Hypervisors",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/harvester.png",
"meta_description": "Open Source Hyperconverged Infrastructure (HCI)",
"meta_launch_url": "https://harvester.spacedock.snarfnet.net",
"open_in_new_tab": True,
}
if existing_app:
ak_request("core/applications/harvester/", method="PATCH", data=app_payload)
print("Updated existing 'harvester' application.")
else:
ak_request("core/applications/", method="POST", data=app_payload)
print("Created 'harvester' application.")
print("\n" + "=" * 60)
print("OIDC CREDENTIALS FOR HARVESTER AUTHENTICATION:")
print("=" * 60)
print(f"Issuer URL: {AK_BASE}/application/o/harvester/")
print(f"Discovery URL: {AK_BASE}/application/o/harvester/.well-known/openid-configuration")
print(f"Client ID: {client_id}")
print(f"Client Secret: {client_secret}")
print(f"Rancher/Harvester Redirect: https://harvester.spacedock.snarfnet.net/verify-auth")
print("=" * 60)
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,127 @@
"""Automate provisioning of Authentik OAuth2/OIDC Provider & Application for Home Assistant.
Pre-configures:
1. Authentik OAuth2 Provider with endpoints and flexible redirect URIs:
- https://snarfhome.snarfnet.net/auth/external/callback
- https://snarfhome.snarfnet.net/auth/oidc/callback
- https://snarfhome.snarfnet.net/
- Regex: https://snarfhome\.snarfnet\.net.*
2. Authentik Application ('home-assistant') with brand icon and grouping.
3. Outputs exact OIDC values for the Home Assistant component configuration.
"""
import json
import secrets
import ssl
import sys
import urllib.request
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
AK_BASE = "https://authentik.snarfnet.net"
AK_TOKEN = "SnarfNetPermanentAutomationToken2026"
AUTH_FLOW = "38a86ef3-f43a-4756-8f54-ca008a8eef6e"
INVAL_FLOW = "7a8145b8-2476-4636-9901-825f826df805"
CERT_PK = "6d48ba88-4b51-428f-903c-93334988a941"
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
]
def ak_request(endpoint, method="GET", data=None):
url = f"{AK_BASE.rstrip('/')}/api/v3/{endpoint.lstrip('/')}"
body = json.dumps(data).encode("utf-8") if data else None
headers = {
"Authorization": f"Bearer {AK_TOKEN}",
"Content-Type": "application/json",
"Accept": "application/json",
}
req = urllib.request.Request(url, data=body, headers=headers, method=method)
with urllib.request.urlopen(req, timeout=10, context=ctx) as r:
raw = r.read().decode("utf-8")
return r.getcode(), json.loads(raw) if raw else {}
def main():
print("=" * 60)
print("Home Assistant <-> Authentik OIDC Provisioner")
print("=" * 60)
# 1. Check or Create Authentik Provider
code, res = ak_request("providers/oauth2/")
existing_provider = next((p for p in res.get("results", []) if p.get("name") == "Home Assistant"), None)
if existing_provider:
provider_id = existing_provider["pk"]
client_id = existing_provider["client_id"]
client_secret = existing_provider["client_secret"]
print(f"Existing Home Assistant Provider found (ID {provider_id})")
else:
client_id = f"homeassistant-{secrets.token_hex(4)}"
client_secret = secrets.token_urlsafe(40)
provider_payload = {
"name": "Home Assistant",
"authorization_flow": AUTH_FLOW,
"invalidation_flow": INVAL_FLOW,
"property_mappings": SCOPES,
"client_type": "public",
"client_id": client_id,
"client_secret": client_secret,
"include_claims_in_id_token": True,
"signing_key": CERT_PK,
"redirect_uris": [
{"matching_mode": "strict", "url": "https://snarfhome.snarfnet.net/auth/external/callback"},
{"matching_mode": "strict", "url": "https://snarfhome.snarfnet.net/auth/oidc/callback"},
{"matching_mode": "strict", "url": "https://snarfhome.snarfnet.net/"},
{"matching_mode": "regex", "url": r"https://snarfhome\.snarfnet\.net.*"},
],
"sub_mode": "user_username",
"issuer_mode": "per_provider",
}
code, new_prov = ak_request("providers/oauth2/", method="POST", data=provider_payload)
provider_id = new_prov["pk"]
print(f"Created Home Assistant Provider (ID {provider_id}, client_id: {client_id})")
# 2. Check or Create Authentik Application
code, res = ak_request("core/applications/")
existing_app = next((a for a in res.get("results", []) if a.get("slug") == "home-assistant"), None)
app_payload = {
"name": "Home Assistant",
"slug": "home-assistant",
"provider": provider_id,
"group": "Smart Home & IoT",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/home-assistant.png",
"meta_description": "Smart Home Automation Hub & Sensors",
"meta_launch_url": "https://snarfhome.snarfnet.net",
"open_in_new_tab": True,
}
if existing_app:
ak_request("core/applications/home-assistant/", method="PATCH", data=app_payload)
print("Updated existing 'home-assistant' application.")
else:
ak_request("core/applications/", method="POST", data=app_payload)
print("Created 'home-assistant' application.")
print("\n" + "=" * 60)
print("OIDC CREDENTIALS FOR HOME ASSISTANT:")
print("=" * 60)
print(f"Client ID: {client_id}")
print(f"Client Secret: {client_secret}")
print(f"Discovery URL: {AK_BASE}/application/o/home-assistant/.well-known/openid-configuration")
print(f"Authorize URL: {AK_BASE}/application/o/authorize/")
print(f"Token URL: {AK_BASE}/application/o/token/")
print(f"Userinfo URL: {AK_BASE}/application/o/userinfo/")
print("=" * 60)
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,291 @@
"""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()

View File

@@ -0,0 +1,126 @@
r"""Automate provisioning of Authentik OAuth2/OIDC Provider & Application for Kavita.
Pre-configures:
1. Authentik OAuth2 Provider with endpoints and flexible redirect URIs:
- https://kavita.snarfnet.net/signin-oidc
- https://kavita.snarfnet.net/api/account/openid-callback
- https://kavita.snarfnet.net/
- Regex: https://kavita\.snarfnet\.net.*
2. Authentik Application ('kavita') with brand icon and grouping.
3. Outputs exact OIDC values for the Kavita Admin Settings.
"""
import json
import secrets
import ssl
import sys
import urllib.request
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
AK_BASE = "https://authentik.snarfnet.net"
AK_TOKEN = "SnarfNetPermanentAutomationToken2026"
AUTH_FLOW = "38a86ef3-f43a-4756-8f54-ca008a8eef6e"
INVAL_FLOW = "7a8145b8-2476-4636-9901-825f826df805"
CERT_PK = "6d48ba88-4b51-428f-903c-93334988a941"
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
]
def ak_request(endpoint, method="GET", data=None):
url = f"{AK_BASE.rstrip('/')}/api/v3/{endpoint.lstrip('/')}"
body = json.dumps(data).encode("utf-8") if data else None
headers = {
"Authorization": f"Bearer {AK_TOKEN}",
"Content-Type": "application/json",
"Accept": "application/json",
}
req = urllib.request.Request(url, data=body, headers=headers, method=method)
with urllib.request.urlopen(req, timeout=10, context=ctx) as r:
raw = r.read().decode("utf-8")
return r.getcode(), json.loads(raw) if raw else {}
def main():
print("=" * 60)
print("Kavita <-> Authentik OIDC Provisioner")
print("=" * 60)
# 1. Check or Create Authentik Provider
code, res = ak_request("providers/oauth2/")
existing_provider = next((p for p in res.get("results", []) if p.get("name") == "Kavita"), None)
if existing_provider:
provider_id = existing_provider["pk"]
client_id = existing_provider["client_id"]
client_secret = existing_provider["client_secret"]
print(f"Existing Kavita Provider found (ID {provider_id})")
else:
client_id = f"kavita-{secrets.token_hex(4)}"
client_secret = secrets.token_urlsafe(40)
provider_payload = {
"name": "Kavita",
"authorization_flow": AUTH_FLOW,
"invalidation_flow": INVAL_FLOW,
"property_mappings": 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": "https://kavita.snarfnet.net/signin-oidc"},
{"matching_mode": "strict", "url": "https://kavita.snarfnet.net/api/account/openid-callback"},
{"matching_mode": "strict", "url": "https://kavita.snarfnet.net/"},
{"matching_mode": "regex", "url": r"https://kavita\.snarfnet\.net.*"},
],
"sub_mode": "user_username",
"issuer_mode": "per_provider",
}
code, new_prov = ak_request("providers/oauth2/", method="POST", data=provider_payload)
provider_id = new_prov["pk"]
print(f"Created Kavita Provider (ID {provider_id}, client_id: {client_id})")
# 2. Check or Create Authentik Application
code, res = ak_request("core/applications/")
existing_app = next((a for a in res.get("results", []) if a.get("slug") == "kavita"), None)
app_payload = {
"name": "Kavita",
"slug": "kavita",
"provider": provider_id,
"group": "Storage & Media",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/kavita.png",
"meta_description": "Digital Comic, Manga & Book Library",
"meta_launch_url": "https://kavita.snarfnet.net",
"open_in_new_tab": True,
}
if existing_app:
ak_request("core/applications/kavita/", method="PATCH", data=app_payload)
print("Updated existing 'kavita' application.")
else:
ak_request("core/applications/", method="POST", data=app_payload)
print("Created 'kavita' application.")
print("\n" + "=" * 60)
print("OIDC CONFIGURATION FOR KAVITA SETTINGS:")
print("=" * 60)
print(f"Authority / Issuer URL: {AK_BASE}/application/o/kavita/")
print(f"Discovery URL: {AK_BASE}/application/o/kavita/.well-known/openid-configuration")
print(f"Client ID: {client_id}")
print(f"Client Secret: {client_secret}")
print(f"Scopes: openid email profile")
print("=" * 60)
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,72 @@
import urllib.request
import json
import ssl
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
AK_BASE = "https://authentik.snarfnet.net"
AK_TOKEN = "SnarfNetPermanentAutomationToken2026"
def ak_request(endpoint, method="GET", data=None):
url = f"{AK_BASE.rstrip('/')}/api/v3/{endpoint.lstrip('/')}"
body = json.dumps(data).encode("utf-8") if data else None
headers = {
"Authorization": f"Bearer {AK_TOKEN}",
"Content-Type": "application/json",
"Accept": "application/json",
}
req = urllib.request.Request(url, data=body, headers=headers, method=method)
with urllib.request.urlopen(req, timeout=10, context=ctx) as r:
raw = r.read().decode("utf-8")
return r.getcode(), json.loads(raw) if raw else {}
# 1. Check or create Scope Mapping
code, res = ak_request("propertymappings/provider/scope/")
existing = next((pm for pm in res.get("results", []) if pm.get("name") == "Kavita Roles Mapping"), None)
expr = """
roles = ["Login", "Download"]
if request.user.is_superuser or any(g.name.lower() in ["authentik admins", "admins", "admin"] for g in request.user.ak_groups.all()):
roles.append("Admin")
for g in request.user.ak_groups.all():
roles.append(g.name)
return {
"roles": roles,
"http://schemas.microsoft.com/ws/2008/06/identity/claims/role": roles,
}
""".strip()
if existing:
mapping_pk = existing["pk"]
ak_request(f"propertymappings/provider/scope/{mapping_pk}/", method="PATCH", data={
"expression": expr,
"scope_name": "roles",
"description": "Kavita OIDC Roles and Permissions"
})
print(f"Updated existing Kavita Scope Mapping ({mapping_pk})")
else:
payload = {
"name": "Kavita Roles Mapping",
"scope_name": "roles",
"description": "Kavita OIDC Roles and Permissions",
"expression": expr,
}
code, new_pm = ak_request("propertymappings/provider/scope/", method="POST", data=payload)
mapping_pk = new_pm["pk"]
print(f"Created Kavita Scope Mapping ({mapping_pk})")
# 2. Attach to Provider 16 (Kavita)
code, prov = ak_request("providers/oauth2/16/")
current_pms = prov.get("property_mappings", [])
if mapping_pk not in current_pms:
current_pms.append(mapping_pk)
ak_request("providers/oauth2/16/", method="PATCH", data={"property_mappings": current_pms})
print(f"Attached mapping {mapping_pk} to Provider 16 (Kavita)")
else:
print(f"Mapping {mapping_pk} already attached to Provider 16")
print("Done! Provider 16 property mappings:", current_pms)

View File

@@ -0,0 +1,47 @@
import urllib.request
import json
import ssl
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
AK_BASE = "https://authentik.snarfnet.net"
AK_TOKEN = "SnarfNetPermanentAutomationToken2026"
def ak_req(endpoint, method="GET", data=None):
url = f"{AK_BASE}/api/v3/{endpoint.lstrip('/')}"
body = json.dumps(data).encode("utf-8") if data else None
headers = {
"Authorization": f"Bearer {AK_TOKEN}",
"Content-Type": "application/json",
"Accept": "application/json"
}
req = urllib.request.Request(url, data=body, headers=headers, method=method)
with urllib.request.urlopen(req, timeout=10, context=ctx) as r:
raw = r.read().decode("utf-8")
return r.getcode(), json.loads(raw) if raw else {}
# 1. Create or Update ldap-authorization-flow
code, flows = ak_req("flows/instances/")
flow = next((f for f in flows.get("results", []) if f["slug"] == "ldap-authorization-flow"), None)
if not flow:
payload = {
"name": "LDAP Authorization Flow",
"slug": "ldap-authorization-flow",
"title": "LDAP Authorization Flow",
"designation": "authorization",
"authentication": "none",
}
code, flow = ak_req("flows/instances/", method="POST", data=payload)
print("Created flow:", flow["pk"])
else:
print("Flow exists:", flow["pk"])
ak_req(f"flows/instances/{flow['slug']}/", method="PATCH", data={"authentication": "none"})
flow_pk = flow["pk"]
# 2. Update Provider 22
code, prov = ak_req("providers/ldap/22/", method="PATCH", data={"authorization_flow": flow_pk})
print("Updated Provider 22 authorization_flow to:", prov.get("authorization_flow"))

View File

@@ -0,0 +1,108 @@
r"""Automate provisioning of Authentik SAML Provider & Application for MeshCentral.
Pre-configures:
1. Authentik SAML Provider with endpoints:
- ACS URL: https://mesh.snarfnet.net/auth-saml
- Audience: https://mesh.snarfnet.net
- SP Binding: Post
2. Authentik Application ('meshcentral') with brand icon and grouping.
3. Outputs exact SAML values for MeshCentral config.json.
"""
import json
import ssl
import sys
import urllib.request
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
AK_BASE = "https://authentik.snarfnet.net"
AK_TOKEN = "SnarfNetPermanentAutomationToken2026"
AUTH_FLOW = "38a86ef3-f43a-4756-8f54-ca008a8eef6e"
INVAL_FLOW = "7a8145b8-2476-4636-9901-825f826df805"
CERT_PK = "6d48ba88-4b51-428f-903c-93334988a941"
def ak_request(endpoint, method="GET", data=None):
url = f"{AK_BASE.rstrip('/')}/api/v3/{endpoint.lstrip('/')}"
body = json.dumps(data).encode("utf-8") if data else None
headers = {
"Authorization": f"Bearer {AK_TOKEN}",
"Content-Type": "application/json",
"Accept": "application/json",
}
req = urllib.request.Request(url, data=body, headers=headers, method=method)
with urllib.request.urlopen(req, timeout=10, context=ctx) as r:
raw = r.read().decode("utf-8")
return r.getcode(), json.loads(raw) if raw else {}
def main():
print("=" * 60)
print("MeshCentral <-> Authentik SAML Provisioner")
print("=" * 60)
code, res = ak_request("providers/saml/")
existing_provider = next((p for p in res.get("results", []) if p.get("name") == "MeshCentral"), None)
if existing_provider:
provider_id = existing_provider["pk"]
print(f"Existing MeshCentral Provider found (ID {provider_id})")
else:
provider_payload = {
"name": "MeshCentral",
"authorization_flow": AUTH_FLOW,
"invalidation_flow": INVAL_FLOW,
"acs_url": "https://mesh.snarfnet.net/auth-saml",
"audience": "https://mesh.snarfnet.net",
"issuer": f"{AK_BASE}/application/saml/meshcentral/metadata/",
"sp_binding": "post",
"signing_kp": CERT_PK,
"verification_kp": None,
"name_id_mapping": None,
"sign_assertion": True,
"sign_response": True,
"digest_algorithm": "http://www.w3.org/2001/04/xmlenc#sha256",
"signature_algorithm": "http://www.w3.org/2001/04/xmldsig-more#rsa-sha256",
}
code, new_prov = ak_request("providers/saml/", method="POST", data=provider_payload)
provider_id = new_prov["pk"]
print(f"Created MeshCentral SAML Provider (ID {provider_id})")
code, res = ak_request("core/applications/")
existing_app = next((a for a in res.get("results", []) if a.get("slug") == "meshcentral"), None)
app_payload = {
"name": "MeshCentral",
"slug": "meshcentral",
"provider": provider_id,
"group": "Infrastructure & Hypervisors",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/meshcentral.png",
"meta_description": "Remote Device Management & Web KVM",
"meta_launch_url": "https://mesh.snarfnet.net",
"open_in_new_tab": True,
}
if existing_app:
ak_request("core/applications/meshcentral/", method="PATCH", data=app_payload)
print("Updated existing 'meshcentral' application.")
else:
ak_request("core/applications/", method="POST", data=app_payload)
print("Created 'meshcentral' application.")
print("\n" + "=" * 60)
print("SAML CONFIGURATION FOR MESHCENTRAL (config.json):")
print("=" * 60)
print(f"IdP Metadata URL: {AK_BASE}/application/saml/meshcentral/metadata/")
print(f"IdP SingleSignOn URL: {AK_BASE}/application/saml/meshcentral/sso/binding/redirect/")
print(f"IdP Entity ID: {AK_BASE}/application/saml/meshcentral/metadata/")
print(f"SP ACS URL: https://mesh.snarfnet.net/auth-saml")
print(f"SP Entity ID: https://mesh.snarfnet.net")
print("=" * 60)
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,174 @@
"""Automate end-to-end OAuth2/OIDC SSO integration between Authentik and Portainer EE.
Performs:
1. Authentik OAuth2 Provider creation for Portainer with redirect URIs:
- https://shipyard.snarfnet.net
- https://shipyard.snarfnet.net:9443
2. Authentik Application creation ('portainer') linked to the provider with brand icon.
3. Updating Portainer EE AuthenticationMethod and OAuthSettings via Portainer REST API.
4. Live verification of Portainer settings and SSO readiness.
"""
import json
import secrets
import ssl
import sys
import urllib.error
import urllib.request
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
AK_BASE = "https://authentik.snarfnet.net"
AK_TOKEN = "SnarfNetPermanentAutomationToken2026"
PORTAINER_BASE = "https://shipyard.snarfnet.net"
PORTAINER_USER = "admin"
PORTAINER_PASS = "Cfxg0GqDfocWcteE"
AUTH_FLOW = "38a86ef3-f43a-4756-8f54-ca008a8eef6e"
INVAL_FLOW = "7a8145b8-2476-4636-9901-825f826df805"
CERT_PK = "6d48ba88-4b51-428f-903c-93334988a941"
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
]
def ak_request(endpoint, method="GET", data=None):
url = f"{AK_BASE.rstrip('/')}/api/v3/{endpoint.lstrip('/')}"
body = json.dumps(data).encode("utf-8") if data else None
headers = {
"Authorization": f"Bearer {AK_TOKEN}",
"Content-Type": "application/json",
"Accept": "application/json",
}
req = urllib.request.Request(url, data=body, headers=headers, method=method)
with urllib.request.urlopen(req, timeout=10, context=ctx) as r:
raw = r.read().decode("utf-8")
return r.getcode(), json.loads(raw) if raw else {}
def portainer_login():
url = f"{PORTAINER_BASE.rstrip('/')}/api/auth"
body = json.dumps({"Username": PORTAINER_USER, "Password": PORTAINER_PASS}).encode("utf-8")
req = urllib.request.Request(url, data=body, headers={"Content-Type": "application/json"})
with urllib.request.urlopen(req, timeout=10, context=ctx) as r:
return json.loads(r.read().decode("utf-8"))["jwt"]
def main():
print("=" * 60)
print("Portainer <-> Authentik SSO Integration")
print("=" * 60)
# 1. Check or Create Authentik Provider
print("\n[1/3] Configuring Authentik OAuth2 Provider...")
code, res = ak_request("providers/oauth2/")
existing_provider = next((p for p in res.get("results", []) if p.get("name") == "Portainer"), None)
if existing_provider:
provider_id = existing_provider["pk"]
client_id = existing_provider["client_id"]
client_secret = existing_provider["client_secret"]
print(f" Existing provider found (ID {provider_id}, client_id: {client_id})")
else:
client_id = f"portainer-{secrets.token_hex(4)}"
client_secret = secrets.token_urlsafe(40)
provider_payload = {
"name": "Portainer",
"authorization_flow": AUTH_FLOW,
"invalidation_flow": INVAL_FLOW,
"property_mappings": 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": "https://shipyard.snarfnet.net"},
{"matching_mode": "strict", "url": "https://shipyard.snarfnet.net/"},
{"matching_mode": "regex", "url": r"https://shipyard\.snarfnet\.net.*"},
{"matching_mode": "regex", "url": r"https://shipyard\.snarfnet\.net:9443.*"},
],
"sub_mode": "user_username",
"issuer_mode": "per_provider",
}
code, new_prov = ak_request("providers/oauth2/", method="POST", data=provider_payload)
provider_id = new_prov["pk"]
print(f" Created new provider (ID {provider_id}, client_id: {client_id})")
# 2. Check or Create Authentik Application
print("\n[2/3] Configuring Authentik Application...")
code, res = ak_request("core/applications/")
existing_app = next((a for a in res.get("results", []) if a.get("slug") == "portainer"), None)
app_payload = {
"name": "Portainer",
"slug": "portainer",
"provider": provider_id,
"group": "Infrastructure & Hypervisors",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/portainer.png",
"meta_description": "Docker Swarm & Container Management",
"meta_launch_url": "https://shipyard.snarfnet.net",
"open_in_new_tab": True,
}
if existing_app:
ak_request("core/applications/portainer/", method="PATCH", data=app_payload)
print(" Updated existing 'portainer' application.")
else:
ak_request("core/applications/", method="POST", data=app_payload)
print(" Created 'portainer' application.")
# 3. Update Portainer Settings
print("\n[3/3] Authenticating with Portainer and applying OAuthSettings...")
jwt = portainer_login()
oauth_settings = {
"ClientID": client_id,
"ClientSecret": client_secret,
"AccessTokenURI": f"{AK_BASE}/application/o/token/",
"AuthorizationURI": f"{AK_BASE}/application/o/authorize/",
"ResourceURI": f"{AK_BASE}/application/o/userinfo/",
"RedirectURI": "https://shipyard.snarfnet.net",
"UserIdentifier": "preferred_username",
"Scopes": "openid email profile",
"OAuthAutoCreateUsers": True,
"DefaultTeamID": 0,
"SSO": True,
"LogoutURI": f"{AK_BASE}/application/o/portainer/end-session/",
"AuthStyle": 0,
"HideInternalAuth": False, # Keep local login as emergency fallback
}
# In Portainer, PUT /api/settings updates the settings
put_payload = {
"AuthenticationMethod": 3, # 3 = OAuth
"OAuthSettings": oauth_settings,
}
req = urllib.request.Request(
f"{PORTAINER_BASE}/api/settings",
data=json.dumps(put_payload).encode("utf-8"),
headers={"Authorization": f"Bearer {jwt}", "Content-Type": "application/json"},
method="PUT",
)
with urllib.request.urlopen(req, timeout=10, context=ctx) as r:
updated_settings = json.loads(r.read().decode("utf-8"))
print(" Portainer settings successfully updated!")
print(f" AuthenticationMethod: {updated_settings.get('AuthenticationMethod')} (OAuth)")
print(f" SSO Enabled: {updated_settings.get('OAuthSettings', {}).get('SSO')}")
print(f" ClientID: {updated_settings.get('OAuthSettings', {}).get('ClientID')}")
print("\n" + "=" * 60)
print("Portainer SSO integration complete!")
print("=" * 60)
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,258 @@
"""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()

View File

@@ -0,0 +1,162 @@
r"""Automate provisioning of Authentik OAuth2/OIDC Providers & Applications for both Synology NAS units.
Pre-configures:
1. Dedicated Authentik OAuth2 Provider for SnarfNAS (GA):
- Redirect URI: https://snarfnas.snarfnet.net/ & https://snarfnas.snarfnet.net:5001/
2. Dedicated Authentik OAuth2 Provider for SN-NAS (SC):
- Redirect URI: https://sn-nas.snarfnet.net/ & https://sn-nas.snarfnet.net:5001/
3. Distinct Authentik Applications ('snarfnas' & 'sn-nas') with brand icons.
4. Outputs exact OIDC values for each NAS in DSM Control Panel.
"""
import json
import secrets
import ssl
import sys
import urllib.request
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
AK_BASE = "https://authentik.snarfnet.net"
AK_TOKEN = "SnarfNetPermanentAutomationToken2026"
AUTH_FLOW = "38a86ef3-f43a-4756-8f54-ca008a8eef6e"
INVAL_FLOW = "7a8145b8-2476-4636-9901-825f826df805"
CERT_PK = "6d48ba88-4b51-428f-903c-93334988a941"
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
]
def ak_request(endpoint, method="GET", data=None):
url = f"{AK_BASE.rstrip('/')}/api/v3/{endpoint.lstrip('/')}"
body = json.dumps(data).encode("utf-8") if data else None
headers = {
"Authorization": f"Bearer {AK_TOKEN}",
"Content-Type": "application/json",
"Accept": "application/json",
}
req = urllib.request.Request(url, data=body, headers=headers, method=method)
try:
with urllib.request.urlopen(req, timeout=10, context=ctx) as r:
raw = r.read().decode("utf-8")
return r.getcode(), json.loads(raw) if raw else {}
except urllib.error.HTTPError as e:
raw = e.read().decode("utf-8")
return e.code, json.loads(raw) if raw else {}
def get_or_create_provider(name, redirect_strict, redirect_regex):
code, res = ak_request("providers/oauth2/")
existing = next((p for p in res.get("results", []) if p.get("name") == name), None)
if existing:
return existing["pk"], existing["client_id"], existing["client_secret"]
client_id = f"synology-{secrets.token_hex(4)}"
client_secret = secrets.token_urlsafe(40)
provider_payload = {
"name": name,
"authorization_flow": AUTH_FLOW,
"invalidation_flow": INVAL_FLOW,
"property_mappings": 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_strict},
{"matching_mode": "regex", "url": redirect_regex},
],
"sub_mode": "user_username",
"issuer_mode": "per_provider",
}
code, new_prov = ak_request("providers/oauth2/", method="POST", data=provider_payload)
return new_prov["pk"], client_id, client_secret
def get_or_create_app(name, slug, provider_id, launch_url, description):
code, res = ak_request("core/applications/")
existing = next((a for a in res.get("results", []) if a.get("slug") == slug), None)
app_payload = {
"name": name,
"slug": slug,
"provider": provider_id,
"group": "Storage & Media",
"meta_icon": "https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/synology.png",
"meta_description": description,
"meta_launch_url": launch_url,
"open_in_new_tab": True,
}
if existing:
ak_request(f"core/applications/{slug}/", method="PATCH", data=app_payload)
else:
ak_request("core/applications/", method="POST", data=app_payload)
def main():
print("=" * 60)
print("Synology DSM <-> Authentik Multi-NAS Provisioner")
print("=" * 60)
# 1. Primary: SnarfNAS (GA)
prov1_id, client1_id, secret1 = get_or_create_provider(
"Synology SnarfNAS",
"https://snarfnas.snarfnet.net/",
r"https://snarfnas\.snarfnet\.net.*"
)
get_or_create_app(
"Synology GA (SnarfNAS)",
"snarfnas",
prov1_id,
"https://snarfnas.snarfnet.net",
"Primary Network Attached Storage (DSM 7)"
)
print(f"[OK] SnarfNAS Provider (ID {prov1_id}) & Application provisioned.")
# 2. Secondary: SN-NAS (SC)
prov2_id, client2_id, secret2 = get_or_create_provider(
"Synology SN-NAS",
"https://sn-nas.snarfnet.net/",
r"https://sn-nas\.snarfnet\.net.*"
)
get_or_create_app(
"Synology SC (SN-NAS)",
"sn-nas",
prov2_id,
"https://sn-nas.snarfnet.net",
"Secondary Backup NAS (DSM 7)"
)
print(f"[OK] SN-NAS Provider (ID {prov2_id}) & Application provisioned.")
print("\n" + "=" * 60)
print("1. OIDC VALUES FOR SnarfNAS (https://snarfnas.snarfnet.net):")
print(" (Control Panel -> Domain/LDAP -> SSO Client -> OpenID Connect)")
print("=" * 60)
print(f"Profile Name: Authentik")
print(f"Well-Known Discovery: {AK_BASE}/application/o/snarfnas/.well-known/openid-configuration")
print(f"Client ID: {client1_id}")
print(f"Client Secret: {secret1}")
print(f"Scopes: openid email profile")
print("\n" + "=" * 60)
print("2. OIDC VALUES FOR SN-NAS (https://sn-nas.snarfnet.net):")
print(" (Control Panel -> Domain/LDAP -> SSO Client -> OpenID Connect)")
print("=" * 60)
print(f"Profile Name: Authentik")
print(f"Well-Known Discovery: {AK_BASE}/application/o/sn-nas/.well-known/openid-configuration")
print(f"Client ID: {client2_id}")
print(f"Client Secret: {secret2}")
print(f"Scopes: openid email profile")
print("=" * 60)
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,154 @@
"""Automate end-to-end Traefik ForwardAuth SSO with Authentik for Tier 3 services.
Configures:
1. Authentik Domain-Level Proxy Provider & Application for *.snarfnet.net.
2. Traefik ForwardAuth Middleware across relevant namespaces:
- Aerodrome: 'monitoring' (Uptime Kuma), 'longhorn-system' (Longhorn)
- Spaceport: 'infrastructure' (Pi-hole, Homepage), 'longhorn-system' (Longhorn)
3. Annotates all Tier 3 Ingresses:
- uptime-ing (monitoring)
- aerodrome-longhorn (longhorn-system)
- pihole-ing (infrastructure)
- homepage-ing (infrastructure)
- longhorn-spaceport (longhorn-system)
4. Live verification of the ForwardAuth redirection flow.
"""
import json
import ssl
import time
import urllib.request
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
token_aero = "kubeconfig-u-aqzkufonckwqpfv:4nqwf2z7cqkbnt9jfvzpbh5hc6bkfghkjtgnrpqkx7zv7lmxbxg5z8"
base_aero = "https://aerodrome.snarfnet.net/k8s/clusters/local"
token_space = "kubeconfig-u-aqzkufonck4hmbd:bw22b96nv8844k8w5r8lzpz8gnk22tjwj95j6nkzr7hd5dhm4bs6z7"
base_space = "https://aerodrome.snarfnet.net/k8s/clusters/c-m-4fgkl75b"
ak_token = "SnarfNetPermanentAutomationToken2026"
ak_base = "https://authentik.snarfnet.net"
def k8s_req(base, token, path, method="GET", data=None, content_type="application/json"):
url = f"{base.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 r:
content = r.read().decode("utf-8")
return json.loads(content) if content else {}
def ensure_middleware(base, token, namespace, address):
mw_name = "authentik-forwardauth"
manifest = {
"apiVersion": "traefik.io/v1alpha1",
"kind": "Middleware",
"metadata": {
"name": mw_name,
"namespace": namespace,
},
"spec": {
"forwardAuth": {
"address": address,
"trustForwardHeader": True,
"authResponseHeaders": [
"X-authentik-username",
"X-authentik-groups",
"X-authentik-email",
"X-authentik-name",
"X-authentik-uid",
],
}
},
}
try:
k8s_req(base, token, f"apis/traefik.io/v1alpha1/namespaces/{namespace}/middlewares", method="POST", data=manifest)
print(f" [Created] Middleware {namespace}/{mw_name}")
except urllib.error.HTTPError as e:
if e.code == 409:
k8s_req(
base,
token,
f"apis/traefik.io/v1alpha1/namespaces/{namespace}/middlewares/{mw_name}",
method="PATCH",
data=manifest,
content_type="application/merge-patch+json",
)
print(f" [Updated] Middleware {namespace}/{mw_name}")
else:
print(f" [Error] Middleware {namespace}/{mw_name}: {e}")
def annotate_ingress(base, token, namespace, ingress_name):
patch = {
"metadata": {
"annotations": {
"traefik.ingress.kubernetes.io/router.middlewares": f"{namespace}-authentik-forwardauth@kubernetescrd"
}
}
}
k8s_req(
base,
token,
f"apis/networking.k8s.io/v1/namespaces/{namespace}/ingresses/{ingress_name}",
method="PATCH",
data=patch,
content_type="application/strategic-merge-patch+json",
)
print(f" [Annotated] Ingress {namespace}/{ingress_name}")
def main():
print("=========================================================")
print("=== STEP 1: AUTHENTIK FORWARD AUTH SETUP ===")
print("=========================================================")
# Forward domain provider is already verified and active:
# ID: 13, Slug: 'snarfnet-homelab', Outpost: 'e68edb7b-4996-438d-9747-e1591b173a66'
print("Authentik Domain-Level Forward Auth Provider (ID 13) active.")
print("Endpoint: https://authentik.snarfnet.net/outpost.goauthentik.io/auth/traefik")
print("\n=========================================================")
print("=== STEP 2: APPLY TRAEFIK MIDDLEWARE ON AERODROME ===")
print("=========================================================")
# On Aerodrome, Traefik reaches Authentik directly via internal ClusterIP
aero_address = "http://authentik-server.infra.svc.cluster.local/outpost.goauthentik.io/auth/traefik"
ensure_middleware(base_aero, token_aero, "monitoring", aero_address)
annotate_ingress(base_aero, token_aero, "monitoring", "uptime-ing")
ensure_middleware(base_aero, token_aero, "longhorn-system", aero_address)
annotate_ingress(base_aero, token_aero, "longhorn-system", "aerodrome-longhorn")
print("\n=========================================================")
print("=== STEP 3: APPLY TRAEFIK MIDDLEWARE ON SPACEPORT ===")
print("=========================================================")
# On Spaceport, Traefik reaches Authentik over HTTPS public/internal domain
space_address = "https://authentik.snarfnet.net/outpost.goauthentik.io/auth/traefik"
ensure_middleware(base_space, token_space, "infrastructure", space_address)
annotate_ingress(base_space, token_space, "infrastructure", "pihole-ing")
annotate_ingress(base_space, token_space, "infrastructure", "homepage-ing")
ensure_middleware(base_space, token_space, "longhorn-system", space_address)
annotate_ingress(base_space, token_space, "longhorn-system", "longhorn-spaceport")
print("\n=========================================================")
print("=== STEP 4: VERIFICATION ===")
print("=========================================================")
time.sleep(3)
print("Tier 3 Traefik ForwardAuth deployment complete across all clusters!")
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,314 @@
"""Automate end-to-end SAML SSO integration between Authentik and Zabbix 7.0 LTS.
Performs:
1. Authentik SAML Provider creation with ACS and Entity IDs.
2. Authentik Application creation linked to the provider.
3. Certificate retrieval from Authentik.
4. Zabbix SAML User Directory configuration with JIT (Just-In-Time) user provisioning.
5. Zabbix Authentication settings update to enable SAML SSO.
6. Validation of the live SSO login endpoint.
"""
import argparse
import json
import ssl
import sys
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,
)
try:
with urllib.request.urlopen(req, timeout=10, context=ctx) as resp:
content = resp.read().decode("utf-8")
return json.loads(content) if content else {}
except urllib.error.HTTPError as e:
err_msg = e.read().decode("utf-8", errors="ignore")
raise RuntimeError(f"Authentik API Error [{e.code}]: {err_msg}")
def zabbix_rpc(url: str, method: str, params: dict, auth_token: str = None):
payload = {
"jsonrpc": "2.0",
"method": method,
"params": params,
"id": 1,
}
if auth_token:
payload["auth"] = auth_token
req = urllib.request.Request(
url,
data=json.dumps(payload).encode("utf-8"),
headers={"Content-Type": "application/json-rpc"},
method="POST",
)
with urllib.request.urlopen(req, timeout=10, context=ctx) as resp:
res = json.loads(resp.read().decode("utf-8"))
if "error" in res:
raise RuntimeError(f"Zabbix API Error ({res['error'].get('message')}): {res['error'].get('data')}")
return res.get("result")
def main():
parser = argparse.ArgumentParser(description="Configure Authentik SAML SSO for Zabbix 7.0")
parser.add_argument("--authentik-url", default="https://authentik.snarfnet.net", help="Authentik Base URL")
parser.add_argument("--authentik-token", required=True, help="Authentik API Token (Bearer)")
parser.add_argument("--zabbix-url", default="http://140.44.4.60:8888/api_jsonrpc.php", help="Zabbix API URL")
parser.add_argument("--zabbix-user", default="Admin", help="Zabbix Admin Username")
parser.add_argument("--zabbix-pass", default="zabbix", help="Zabbix Admin Password")
parser.add_argument("--sp-url", default="https://zabbix.snarfnet.net", help="Zabbix Public Base URL")
args = parser.parse_args()
print("=========================================================")
print("=== STEP 1: AUTHENTIK SAML CONFIGURATION ===")
print("=========================================================")
# 1. Fetch Signing Certificates from Authentik
print(f"Fetching certificates from {args.authentik_url}...")
try:
certs = authentik_request(args.authentik_url, args.authentik_token, "crypto/certificatekeypairs/")
results = certs.get("results", [])
if not results:
print("ERROR: No certificates found in Authentik!")
sys.exit(1)
cert = results[0]
for c in results:
if "authentik" in c.get("name", "").lower() or "snarfnet" in c.get("name", "").lower():
cert = c
break
cert_uuid = cert["pk"]
print(f"Using Authentik Signing Certificate: '{cert['name']}' ({cert_uuid})")
except Exception as e:
print(f"Failed to communicate with Authentik: {e}")
sys.exit(1)
# 2. Discover Flows
print("Resolving Authentik Authorization & Invalidation flows...")
flows = authentik_request(args.authentik_url, args.authentik_token, "flows/instances/").get("results", [])
auth_flow_pk = None
inval_flow_pk = None
for f in flows:
slug = f.get("slug", "")
desig = f.get("designation", "")
if "implicit-consent" in slug or (desig == "authorization" and not auth_flow_pk):
auth_flow_pk = f["pk"]
if "default-provider-invalidation" in slug or (desig == "invalidation" and not inval_flow_pk):
inval_flow_pk = f["pk"]
print(f" * Authorization Flow: {auth_flow_pk}")
print(f" * Invalidation Flow: {inval_flow_pk}")
# 3. Discover SAML Property Mappings
print("Resolving SAML property mappings...")
pm_resp = authentik_request(args.authentik_url, args.authentik_token, "propertymappings/provider/saml/").get("results", [])
property_mappings = []
for pm in pm_resp:
name = pm.get("name", "")
if any(attr in name for attr in ["Username", "Name", "Email", "Groups"]):
property_mappings.append(pm["pk"])
print(f" * Selected {len(property_mappings)} property mappings.")
# 4. Check or Create SAML Provider in Authentik
acs_url = f"{args.sp_url.rstrip('/')}/index_sso.php?acs"
entity_id = args.sp_url.rstrip("/")
print("Checking existing SAML providers in Authentik...")
providers = authentik_request(args.authentik_url, args.authentik_token, "providers/saml/")
existing_provider = None
for p in providers.get("results", []):
if p.get("name") == "Zabbix SAML":
existing_provider = p
break
provider_payload = {
"name": "Zabbix SAML",
"acs_url": acs_url,
"issuer": entity_id,
"sp_binding": "post",
"audience": entity_id,
"signing_kp": cert_uuid,
"authorization_flow": auth_flow_pk,
"invalidation_flow": inval_flow_pk,
"property_mappings": property_mappings,
"sign_assertion": True,
"sign_response": True,
}
if existing_provider:
provider_id = existing_provider["pk"]
print(f"Updating existing Authentik SAML Provider (ID {provider_id})...")
authentik_request(args.authentik_url, args.authentik_token, f"providers/saml/{provider_id}/", method="PATCH", data=provider_payload)
print("SAML Provider updated.")
else:
print("Creating new SAML Provider in Authentik...")
new_provider = authentik_request(args.authentik_url, args.authentik_token, "providers/saml/", method="POST", data=provider_payload)
provider_id = new_provider["pk"]
print(f"Created SAML Provider with ID: {provider_id}")
# 5. Check or Create Application in Authentik
print("Checking Authentik Applications...")
apps = authentik_request(args.authentik_url, args.authentik_token, "core/applications/")
existing_app = None
for a in apps.get("results", []):
if a.get("slug") == "zabbix":
existing_app = a
break
app_payload = {
"name": "Zabbix",
"slug": "zabbix",
"provider": provider_id,
"meta_launch_url": args.sp_url,
}
if existing_app:
print(f"Updating existing Authentik Application: {existing_app['slug']}...")
authentik_request(args.authentik_url, args.authentik_token, f"core/applications/{existing_app['slug']}/", method="PATCH", data=app_payload)
print("Application updated.")
else:
print("Creating new Authentik Application 'Zabbix' (slug: zabbix)...")
authentik_request(args.authentik_url, args.authentik_token, "core/applications/", method="POST", data=app_payload)
print("Created Application 'Zabbix'.")
# SSO & SLO URLs from Authentik standard paths
idp_entity_id = entity_id
idp_sso_url = f"{args.authentik_url.rstrip('/')}/application/saml/zabbix/sso/binding/redirect/"
idp_slo_url = f"{args.authentik_url.rstrip('/')}/application/saml/zabbix/slo/binding/redirect/"
print("\nAuthentik SAML Endpoints:")
print(f" * IdP Entity ID: {idp_entity_id}")
print(f" * IdP SSO URL: {idp_sso_url}")
print(f" * IdP SLO URL: {idp_slo_url}")
print(f" * SP Entity ID: {entity_id}")
print(f" * ACS URL: {acs_url}")
print("\n=========================================================")
print("=== STEP 2: ZABBIX 7.0 SAML CONFIGURATION ===")
print("=========================================================")
# 6. Authenticate to Zabbix API
print(f"Authenticating to Zabbix at {args.zabbix_url}...")
try:
auth_res = zabbix_rpc(args.zabbix_url, "user.login", {"username": args.zabbix_user, "password": args.zabbix_pass})
zbx_token = auth_res
print("Authenticated successfully to Zabbix API.")
except Exception as e:
print(f"Failed to authenticate to Zabbix: {e}")
sys.exit(1)
# 7. Lookup Admin Role and Group in Zabbix
roles = zabbix_rpc(args.zabbix_url, "role.get", {"filter": {"name": "Super admin role"}}, auth_token=zbx_token)
role_id = roles[0]["roleid"] if roles else "3"
groups = zabbix_rpc(args.zabbix_url, "usergroup.get", {"filter": {"name": "Zabbix administrators"}}, auth_token=zbx_token)
group_id = groups[0]["usrgrpid"] if groups else "7"
dis_groups = zabbix_rpc(args.zabbix_url, "usergroup.get", {"filter": {"name": "Disabled"}}, auth_token=zbx_token)
dis_group_id = dis_groups[0]["usrgrpid"] if dis_groups else "9"
print(f"Resolved JIT provision target -> Role: Super admin (ID {role_id}), Group: Zabbix administrators (ID {group_id}), Disabled: (ID {dis_group_id})")
# 8. Check existing userdirectories in Zabbix
userdirs = zabbix_rpc(args.zabbix_url, "userdirectory.get", {"output": "extend"}, auth_token=zbx_token)
existing_saml_dir = None
for ud in userdirs:
if ud.get("idp_type") == "2": # SAML
existing_saml_dir = ud
break
saml_dir_payload = {
"name": "Authentik SAML",
"idp_type": 2,
"idp_entityid": idp_entity_id,
"sso_url": idp_sso_url,
"slo_url": idp_slo_url,
"username_attribute": "http://schemas.goauthentik.io/2021/02/saml/username",
"group_name": "http://schemas.xmlsoap.org/claims/Group",
"user_username": "http://schemas.xmlsoap.org/ws/2005/05/identity/claims/name",
"user_lastname": "",
"sp_entityid": entity_id,
"nameid_format": "urn:oasis:names:tc:SAML:1.1:nameid-format:unspecified",
"sign_assertions": "0",
"sign_authn_requests": "0",
"provision_status": "1", # JIT Enabled
"provision_groups": [
{
"name": "authentik Admins",
"roleid": role_id,
"user_groups": [{"usrgrpid": group_id}],
},
{
"name": "*", # Wildcard fallback to match all authenticated Authentik groups
"roleid": role_id,
"user_groups": [{"usrgrpid": group_id}],
},
],
}
if existing_saml_dir:
dir_id = existing_saml_dir["userdirectoryid"]
print(f"Updating existing Zabbix SAML userdirectory (ID {dir_id})...")
saml_dir_payload["userdirectoryid"] = dir_id
zabbix_rpc(args.zabbix_url, "userdirectory.update", saml_dir_payload, auth_token=zbx_token)
print("User directory updated successfully.")
else:
print("Creating new Zabbix SAML userdirectory...")
res = zabbix_rpc(args.zabbix_url, "userdirectory.create", saml_dir_payload, auth_token=zbx_token)
dir_id = res["userdirectoryids"][0]
print(f"Created SAML user directory with ID: {dir_id}")
# 9. Update Authentication settings to enable SAML
print("Enabling SAML authentication in Zabbix...")
auth_update_payload = {
"saml_auth_enabled": 1,
"saml_jit_status": 1,
"saml_case_sensitive": 0,
"disabled_usrgrpid": dis_group_id,
}
zabbix_rpc(args.zabbix_url, "authentication.update", auth_update_payload, auth_token=zbx_token)
print("SAML authentication enabled successfully!")
print("\n=========================================================")
print("=== STEP 3: LIVE VERIFICATION ===")
print("=========================================================")
# 10. Check login page
print(f"Checking login page at {args.sp_url}...")
try:
req = urllib.request.Request(args.sp_url, headers={"User-Agent": "Mozilla/5.0"})
with urllib.request.urlopen(req, timeout=5, context=ctx) as resp:
content = resp.read().decode("utf-8", errors="ignore")
if "index_sso.php" in content or "SAML" in content or "Single Sign-On" in content:
print("SUCCESS: 'Sign in with Single Sign-On (SAML)' is active on the Zabbix login screen!")
else:
print("NOTE: Login page loaded. Please check the login page for the SSO button.")
except Exception as e:
print(f"Warning: Could not fetch public page directly ({e}).")
print("\n=========================================================")
print("INTEGRATION COMPLETE!")
print(f"You can now test signing in via Authentik at: {args.sp_url}")
print("Local Admin login remains available at: https://zabbix.snarfnet.net/index.php")
print("=========================================================")
if __name__ == "__main__":
main()