315 lines
13 KiB
Python
315 lines
13 KiB
Python
"""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()
|