"""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()