Add support for Organization provisioning

[srekkas]

Is your feature request related to a problem? Please describe. We cant provision Keycloak organisation automticaly

Describe the solution you'd like this keycloak-operator provision Keycloak organisation

[jliokaitis]

yes it would be great to be able to manage keycloak organisations https://www.keycloak.org/docs/latest/release_notes/index.html#organizations-supported via operator

[zmotso]

Thank you for your suggestion. We’ve added this functionality to our backlog and aim to implement it in the near future. If you have specific requirements or use cases, feel free to share them with us.

[srekkas]

We want to use organization to redirect users by domain to specific IDP, organization feature seems much better than plugin which we considered before.

I think this can be enough to working solution.

• Enable organization feature for realm
• Create organization
• Add domains to organization
• Assig idp to organization, add domain and enable redirect, hide on login page (operator already can do it?).

Off course it goes in other way, to remove resources :)

I with help of AI tried to write poc, seems to work, but long way to working solution and not very good one :)

import os
import requests
import json


# Configuration
KEYCLOAK_URL = "https://idp.test.my.domain.com"
KEYCLOAK_REALM = "master"
KEYCLOAK_CLIENT_ID = "temp-admin"
KEYCLOAK_CLIENT_SECRET = "password"
ORGANIZATION = "OrgName"
DOMAINS = ["example2.com", "example3.com"]
PROVIDERS = ["oidc"]
PROVIDER_DOMAIN = "example3.com"

# Fetch Access Token
def get_tokens():
    """
    Fetch access token, id token, and refresh token from Keycloak.
    """
    url = f"{KEYCLOAK_URL}/realms/{KEYCLOAK_REALM}/protocol/openid-connect/token"
    payload = {
        "username": KEYCLOAK_CLIENT_ID,
        "password": KEYCLOAK_CLIENT_SECRET,
        "grant_type": "password",
        "client_id": "admin-cli",
    }
    headers = {"Content-Type": "application/x-www-form-urlencoded"}
    
    # Make the request
    response = requests.post(url, data=payload, headers=headers)
    response.raise_for_status()
    
    # Parse the tokens
    tokens = response.json()
    return {
        "access_token": tokens.get("access_token"),
        "id_token": tokens.get("id_token"),
        "refresh_token": tokens.get("refresh_token"),
    }

# Check if organizations are enabled for the realm
def is_organizations_enabled(token):
    url = f"{KEYCLOAK_URL}/admin/realms/{KEYCLOAK_REALM}"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {token}"
    }
    response = requests.get(url, headers=headers)
    response.raise_for_status()
    realm_config = response.json()
    return realm_config.get("organizationsEnabled", False)

# Enable organizations for the realm
def enable_organizations(token):
    url = f"{KEYCLOAK_URL}/admin/realms/{KEYCLOAK_REALM}"
    payload = {"organizationsEnabled": True}
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {token}"
    }
    response = requests.put(url, json=payload, headers=headers)
    response.raise_for_status()

# Get organizations
def get_organizations(token):
    url = f"{KEYCLOAK_URL}/admin/realms/{KEYCLOAK_REALM}/organizations"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {token}"
    }
    response = requests.get(url, headers=headers)
    response.raise_for_status()
    return response.json()

# Create organization
def create_organization(token, organization, domains):
    url = f"{KEYCLOAK_URL}/admin/realms/{KEYCLOAK_REALM}/organizations"
    payload = {
        "name": organization,
        "alias": organization,
        "enabled": True,
        "domains": domains
    }
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {token}"
    }
    response = requests.post(url, json=payload, headers=headers)
    response.raise_for_status()
    try:
        return response.json().get("id")
    except json.JSONDecodeError:
        print(f"Failed to decode response: {response.text}")
        raise

# Update organization's domains
def update_organization_domains(token, org_id, organization, domains):
    url = f"{KEYCLOAK_URL}/admin/realms/{KEYCLOAK_REALM}/organizations/{org_id}"
    payload = {
        "name": organization,
        "alias": organization,
        "domains": domains
    }
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {token}"
    }
    response = requests.put(url, json=payload, headers=headers)
    response.raise_for_status()

# Check if provider is already associated
def is_provider_associated(token, org_id, provider):
    url = f"{KEYCLOAK_URL}/admin/realms/{KEYCLOAK_REALM}/organizations/{org_id}/identity-providers/{provider}"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {token}"
    }
    response = requests.get(url, headers=headers)
    if response.status_code == 200:
        return True
    elif response.status_code == 404:
        return False
    response.raise_for_status()

# Disassociate provider from organization
def disassociate_provider(token, org_id, provider):
    url = f"{KEYCLOAK_URL}/admin/realms/{KEYCLOAK_REALM}/organizations/{org_id}/identity-providers/{provider}"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {token}"
    }
    response = requests.delete(url, headers=headers)
    if response.status_code == 204:
        print(f"Provider {provider} successfully disassociated from the organization.")
    else:
        response.raise_for_status()

# Check if provider exists in realm
def does_provider_exist_in_realm(token, provider):
    url = f"{KEYCLOAK_URL}/admin/realms/{KEYCLOAK_REALM}/identity-provider/instances/{provider}"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {token}"
    }
    response = requests.get(url, headers=headers)
    if response.status_code == 200:
        return True
    elif response.status_code == 404:
        return False
    response.raise_for_status()

# Associate identity provider with organization
def associate_provider(token, org_id, provider):
    if not does_provider_exist_in_realm(token, provider):
        print(f"Provider {provider} does not exist in the realm.")
        return

    if is_provider_associated(token, org_id, provider):
        print(f"Provider {provider} is already associated with the organization.")
        return

    url = f"{KEYCLOAK_URL}/admin/realms/{KEYCLOAK_REALM}/organizations/{org_id}/identity-providers"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {token}"
    }
    response = requests.post(url, data=provider, headers=headers)
    if response.status_code == 409:
        print(f"Provider {provider} is already associated with the organization.")
    response.raise_for_status()

def fetch_idp(token, idp_alias):

    url = f"{KEYCLOAK_URL}/admin/realms/{KEYCLOAK_REALM}/identity-provider/instances/{idp_alias}"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {token}"
    }
    response = requests.get(url, headers=headers)
    response.raise_for_status()
    return response.json()

# Update identity provider
def update_idp(token, idp_alias, payload):

    url = f"{KEYCLOAK_URL}/admin/realms/{KEYCLOAK_REALM}/identity-provider/instances/{idp_alias}"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {token}"
    }
    response = requests.put(url, json=payload, headers=headers)
    response.raise_for_status()

# Configure identity provider
def configure_provider(token, provider, provider_domain):
    idp_config = fetch_idp(token, provider)
    idp_config["config"].update({
        "kc.org.broker.redirect.mode.email-matches": "true",
        "kc.org.domain": provider_domain
    })
    update_idp(token, provider, idp_config)

# Logout Function
def logout(refresh_token):
    """
    Logs out the user session using the refresh token.
    """
    url = f"{KEYCLOAK_URL}/realms/{KEYCLOAK_REALM}/protocol/openid-connect/logout"
    payload = {
        "client_id": "admin-cli",
        "refresh_token": refresh_token,
    }
    headers = {"Content-Type": "application/x-www-form-urlencoded"}
    
    response = requests.post(url, data=payload, headers=headers)
    if response.status_code == 204:
        print("Logout successful.")
    else:
        print(f"Failed to logout: {response.status_code} - {response.text}")


# Main workflow
def main():
    tokens = get_tokens()
    token = tokens["access_token"]
    id_token = tokens["id_token"]
    refresh_token = tokens["refresh_token"]

    # Debug tokens
    # print(f"Access Token: {token}")
    # print(f"ID Token: {id_token}")
    # print(f"Refresh Token: {refresh_token}")

    # Check if organizations are enabled for the realm
    if not is_organizations_enabled(token):
        print("Organizations are not enabled for the realm. Enabling...")
        enable_organizations(token)

        # Verify if enabling was successful
        if not is_organizations_enabled(token):
            print("Failed to enable organizations for the realm.")
            return
        print("Organizations have been successfully enabled.")

    # Check if the organization exists
    organizations = get_organizations(token)
    org_id = next((org["id"] for org in organizations if org["name"] == ORGANIZATION), None)

    if not org_id:
        # Organization does not exist, create it
        print(f"Creating organization '{ORGANIZATION}'...")
        org_id = create_organization(token, ORGANIZATION, DOMAINS)
        print(f"Organization created: Name = {ORGANIZATION}, ID = {org_id}")
    else:
        print(f"Organization exists: Name = {ORGANIZATION}, ID = {org_id}")

    # Update organization's domains
    print(f"Updating domains for organization '{ORGANIZATION}'...")
    update_organization_domains(token, org_id,ORGANIZATION, DOMAINS)
    print(f"Domains updated for organization '{ORGANIZATION}'")

    # Manage providers
    current_providers = [p for p in PROVIDERS if does_provider_exist_in_realm(token, p)]
    associated_providers = [p for p in current_providers if is_provider_associated(token, org_id, p)]

    # Debug providers
    print(current_providers)
    print(associated_providers)

    # Disassociate providers not in PROVIDERS list
    for provider in associated_providers:
        if provider not in PROVIDERS:
            print(f"Disassociating provider '{provider}' from organization '{ORGANIZATION}'...")
            disassociate_provider(token, org_id, provider)

    # Associate missing providers
    for provider in PROVIDERS:
        print(f"Ensuring provider '{provider}' is associated with organization '{ORGANIZATION}'...")
        associate_provider(token, org_id, provider)

    # Configure providers
    for provider in PROVIDERS:
        print(f"Configuring provider '{provider}'...")
        configure_provider(token, provider, PROVIDER_DOMAIN)

    # Logout
    logout(refresh_token)

if __name__ == "__main__":
    main()

[srekkas]

Found out that from Organization linked provider is configured same as Identity provider, so this part is already covered :)

kind: KeycloakRealmIdentityProvider config: kc.org.broker.redirect.mode.email-matches: "true" kc.org.domain: example.com

[zmotso]

The Organization feature has been implemented and will be available in the upcoming release. Configuration example:

apiVersion: v1.edp.epam.com/v1alpha1
kind: KeycloakOrganization
metadata:
  name: test-keycloak-organization
  namespace: default
spec:
  name: "Test Organization"
  alias: "test-org"
  domains:
    - "example.com"
    - "test.com"
  redirectUrl: "https://example.com/redirect"
  description: "Test organization"
  attributes:
    department:
      - "engineering"
      - "qa"
    location:
      - "us-east"
  identityProviders:
    - alias: "test-org-idp"
  realmRef:
    kind: KeycloakRealm
    name: test-org-realm

---

apiVersion: v1.edp.epam.com/v1
kind: KeycloakRealmIdentityProvider
metadata:
  name: test-org-idp
  namespace: default
spec:
  alias: "test-org-idp"
  enabled: true
  providerId: "github"
  realmRef:
    kind: KeycloakRealm
    name: test-org-realm
  config:
    clientId: "test-org-client-id"
    clientSecret: "test-org-client-secret"
    kc.org.domain: "example.com"
    kc.org.broker.redirect.mode.email-matches: "true"