SSO orchestration actions

Define how users are managed during SSO login using orchestration actions

8 minute read

In Cloud Management Platform (CMP), SSO orchestration actions control how users, groups, roles, and directory associations are handled when a user logs in through single sign-on (SSO).

You can use:

SSO User Update
External Users Sync
A combination of both

The default SSO User Update action provides basic behavior, such as associating users with an LDAP domain and updating standard user fields. For advanced requirements, such as group mapping or attribute synchronization, you can customize this action or use the External Users Sync action.

How the actions run

CMP runs the incoming user through any enabled SSO User Update orchestration actions.
CMP runs the same user through any enabled External Users Sync orchestration actions, as described in External Users Sync.
The default SSO User Update action updates the user’s fields based on information and any Required Attributes defined on the SSO provider.

The default action updates only fields that are not included in Required Attributes. If a provider defines “email” and “username” as Required Attributes, only an existing user’s first_name and last_name will be updated when the action runs.

Username and LDAP

The default orchestration action has two methods for automatically assigning a user to an LDAP in CMP.

Method 1: username@ldapdomain format

The SSO IdP sends the username attribute in the format <username>@<ldapdomain>.

The action sets the incoming user’s username to <username> and associates the user with the LDAPUtility whose ldap_domain matches <ldapdomain>.

For example, for an SSO IdP configured to send the username herbert@cloudbolt.org the user receives the username “herbert” and is associated with the “cloudbolt.org” LDAPUtility.

Method 2: ldapDomain attribute

Explicitly define the LDAP Utility domain name for a user in the ldapDomain attribute field on your SSO IdP.

Change the global variable LDAP_DOMAIN_SSO_ATTR_NAME in the action to something other than ldapDomain.

For example, if the SSO IdP sends the LDAP domain in a field called UserLDAPIdentifier, update the action’s LDAP_DOMAIN_SSO_ATTR_NAME definition to LDAP_DOMAIN_SSO_ATTR_NAME = “UserLDAPIdentifier”.

Orchestration action arguments

The following arguments are sent to the orchestration action during SSO login:

Argument	Description
job	The `Job` associated with the action. Populated only if the action is being run as part of a `Job`. `None` when called during SSO login.
user	The `User` being updated or created.
attrs_from_sso	A `dict` of all key:value pairs sent from the SSO IdP.
sso_instance	The database model for the SSO provider in CMP.
creating	A boolean flag that indicates if the `User` is being created (`True`) or updated (`False`).

External users sync

External Users Sync orchestration actions run as part of CMP’s Assertion Consumer Service whenever an SSO user logs in. See Using the “User Permission Sync From LDAP” Action.

If enabled, External Users Sync actions run after the SSO User Update actions. If a user logs in via an SSO provider, the External Users Sync action receives the attrs_from_sso kwarg — a Python dict containing all key:value pairs sent from the SSO IdP to CMP.

Configure the SSO User Update Orchestration Action to sync CMP group, role, and permission settings during SSO logins.

Go to Admin > Orchestration Actions.
In the search box, type User Update from SSO Provider.

(opens full-size image in a new tab)
Select the User Update from SSO Provider action.
Select the Code tab.
Locate the sync_groups_for_user method.

Replace the following lines with the appropriate code for your SSO IdP:

logger.info(
    "SSO sync_groups_for_user is not yet implemented. See documentation for setup instructions: "
    "https://docs.cloudbolt.io/articles/#!cloudbolt-latest-docs/sso-user-update-orchestration-action"
)

Select Okta, IdP returns a getMemberOf link, or IdP returns a list of groups, and copy the corresponding code snippet.

The snippet includes variables you must replace with values from your identity provider.

Okta

    # Update these values with your Okta domain and API token
    COMPANY_OKTA_DOMAIN = ""
    API_TOKEN = ""

    import requests
    from accounts.models import GroupRoleMembership
    from authentication.sso.models import SSOMapping

    user_profile = user.userprofile

    # Check if there are any SSO mappings for this SSO instance
    if not SSOMapping.objects.filter(sso_provider=sso_instance).exists():
        return (
            "SUCCESS",
            f"SSO group synchronization complete for User '{user_profile.username}', no group mappings found for '{sso_instance}'.",
        )

    groups = []

    # Setup up Okta API request to get user's groups
    group_url = (
        f"https://{COMPANY_OKTA_DOMAIN}/api/v1/users/{user_profile.user.email}/groups"
    )

    # Make the API request to Okta to get the user's groups
    headers = {
        "Authorization": f"SSWS {API_TOKEN}",
        "Accept": "application/json",
        "Content-Type": "application/json",
    }

    try:
        response = requests.get(group_url, headers=headers, timeout=10)
        response.raise_for_status()
        groups = response.json()
        if not isinstance(groups, list):
            raise ValueError(f"Okta returned non-list response: {groups!r}")
    except (requests.RequestException, ValueError) as e:
        logger.error(f"Okta group fetch failed for {user_profile.user.email}: {e}")
        return (
            "FAILURE",
            f"SSO group sync skipped for '{user_profile.username}': "
            f"Okta request failed ({e}). Existing permissions left unchanged.",
        )

    # Aggregators — initialize ONCE, before the outer loop
    is_super_admin = False
    is_cbadmin = False
    is_devops_admin = False
    is_global_viewer = False
    all_memberships = []
    all_admin_roles = []
    any_mapping_matched = False

    # Process each SSO group the user is a member of
    for group in groups:
        sso_group_identifier = group.get("id", "<no-id>")
        # Resolve all SSO mappings that match the incoming SSO group identifier
        mappings = SSOMapping.get_mappings(sso_instance, sso_group_identifier)

        if not mappings:
            continue

        any_mapping_matched = True
        for m in mappings:
            is_super_admin   |= m.is_super_admin
            is_cbadmin       |= m.is_cbadmin
            is_devops_admin  |= m.is_devops_admin
            is_global_viewer |= m.is_global_viewer
            all_admin_roles.extend(list(m.admin_roles.all()))
            all_memberships.extend(
                m.get_memberships(profile=user_profile, as_dict=True)
            )

    logger.info(f"Aggregated memberships across all groups: {all_memberships}")
    logger.info(f"Aggregated admin_roles across all groups: {all_admin_roles}")

    # Apply flags ONCE
    user_profile.super_admin   = is_super_admin
    user_profile.devops_admin  = is_devops_admin
    user_profile.global_viewer = is_global_viewer
    user_profile.save()

    user_profile.user.is_superuser = is_cbadmin
    user_profile.user.save()

    user_profile.admin_roles.set(all_admin_roles)

    # Reconcile group-role memberships ONCE, against the full aggregated set
    user_membership_ids = []
    descendant_group_ids = []
    role_ids = []
    for membership in all_memberships:
        grm, _ = GroupRoleMembership.objects.get_or_create(**membership)
        user_membership_ids.append(grm.id)
        descendant_group_ids.extend(g.id for g in grm.group.get_descendant_list())
        role_ids.append(grm.role_id)

    invalid_grms = user_profile.grouprolemembership_set.exclude(
        id__in=user_membership_ids
    ).exclude(
        group_inherited_from__in=descendant_group_ids,
        role__in=role_ids,
    )
    deleted = invalid_grms.delete()
    logger.info(
        f"SSO sync for {user_profile.username}: "
        f"kept {len(user_membership_ids)} memberships, "
        f"deleted {deleted[0]} stale memberships."
    )
    return (
        "SUCCESS",
        f"SSO group synchronization complete for User '{user.userprofile.username}'.",
    )

IdP returns a getMemberOf link

    import requests
    from accounts.models import GroupRoleMembership
    from authentication.sso.models import SSOMapping
    from resourcehandlers.azure_arm.models import AzureARMHandler
    from msal import ConfidentialClientApplication

    user_profile = user.userprofile

    # Check if there are any SSO mappings for this SSO instance
    if not SSOMapping.objects.filter(sso_provider=sso_instance).exists():
        return (
            "SUCCESS",
            f"SSO group synchronization complete for User '{user_profile.username}', no group mappings found for '{sso_instance}'.",
        )


    # Update the following items:
    # GROUPS_CLAIM: Update with the dictionary key in attrs_from_sso.
    GROUPS_CLAIM = "http://schemas.microsoft.com/ws/2008/06/identity/claims/groups"

    # Determine which dictionary item in attrs_from_sso has the groups link
    sso_group_ids = attrs_from_sso.get(GROUPS_CLAIM, [])

    if not sso_group_ids:
        logger.info("No Entra groups found in SSO assertion")
        return (
            "SUCCESS",
            f"SSO group synchronization complete for User '{user_profile.username}', no Entra groups in SSO assertion.",
        )

    logger.info(f"SSO Group IDs from IDP:{sso_group_ids}")

    # Use this method if you are returned get member of link instead of a list of groups
    # in the sso_group_ids.

    # RH_ID: Azure Resource Hander ID, must be the Azure subscription where SSO is configured
    RH_ID = 1
    rh = AzureARMHandler.objects.get(id=RH_ID)

    # Get access_token
    app = ConfidentialClientApplication(
        client_id=rh.client_id,
        authority=f"https://login.microsoftonline.com/{rh.azure_tenant_id}",
        client_credential=rh.secret,
    )

    result = app.acquire_token_for_client(
        scopes=["https://graph.microsoft.com/.default"]
    )

    if "access_token" not in result:
        logger.error("Token acquisition failed: %s", result)
        raise Exception(result)

    access_token = result["access_token"]

    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json",
    }

    group_dicts = []

    for sso_group_id in sso_group_ids:
        url = f"https://graph.microsoft.com/v1.0/groups/{sso_group_id}"
        resp = requests.get(url, headers=headers)

        if resp.status_code != 200:
            logger.warning(
                f"Failed to fetch group {sso_group_id}: {resp.status_code} {resp.text}"
            )
            continue

        group = resp.json()
        group_dicts.append(
            {
                "id": group.get("id"),
                "name": group.get("displayName"),
            }
        )

    logger.info(f"Resolved Entra groups: {group_dicts}")

    groups = group_dicts

    # Aggregators — initialize ONCE, before the outer loop
    is_super_admin = False
    is_cbadmin = False
    is_devops_admin = False
    is_global_viewer = False
    all_memberships = []
    all_admin_roles = []
    any_mapping_matched = False

    # Walk each Entra group, accumulate everything
    for group in groups:
        sso_group_identifier = group.get("id", "<no-id>")
        mappings = SSOMapping.get_mappings(sso_instance, sso_group_identifier)
        if not mappings:
            continue

        any_mapping_matched = True
        for m in mappings:
            is_super_admin   |= m.is_super_admin
            is_cbadmin       |= m.is_cbadmin
            is_devops_admin  |= m.is_devops_admin
            is_global_viewer |= m.is_global_viewer
            all_admin_roles.extend(list(m.admin_roles.all()))
            all_memberships.extend(
                m.get_memberships(profile=user_profile, as_dict=True)
            )

    logger.info(f"Aggregated memberships across all groups: {all_memberships}")
    logger.info(f"Aggregated admin_roles across all groups: {all_admin_roles}")

    # Apply flags ONCE
    user_profile.super_admin   = is_super_admin
    user_profile.devops_admin  = is_devops_admin
    user_profile.global_viewer = is_global_viewer
    user_profile.save()

    user_profile.user.is_superuser = is_cbadmin
    user_profile.user.save()

    user_profile.admin_roles.set(all_admin_roles)

    # Reconcile group-role memberships ONCE, against the full aggregated set
    user_membership_ids = []
    descendant_group_ids = []
    role_ids = []
    for membership in all_memberships:
        grm, _ = GroupRoleMembership.objects.get_or_create(**membership)
        user_membership_ids.append(grm.id)
        descendant_group_ids.extend(g.id for g in grm.group.get_descendant_list())
        role_ids.append(grm.role_id)

    invalid_grms = user_profile.grouprolemembership_set.exclude(
        id__in=user_membership_ids
    ).exclude(
        group_inherited_from__in=descendant_group_ids,
        role__in=role_ids,
    )
    deleted = invalid_grms.delete()
    logger.info(
        f"SSO sync for {user_profile.username}: "
        f"kept {len(user_membership_ids)} memberships, "
        f"deleted {deleted[0]} stale memberships."
    )

    return (
        "SUCCESS",
        f"SSO group synchronization complete for User '{user.userprofile.username}'.",
    )

IdP returns a list of groups

    # Update the following items:
    # GROUPS_CLAIM: Update with the dictionary key in attrs_from_sso.
    GROUPS_CLAIM = "http://schemas.microsoft.com/ws/2008/06/identity/claims/groups"

    from accounts.models import GroupRoleMembership
    from authentication.sso.models import SSOMapping

    user_profile = user.userprofile

    # Check if there are any SSO mappings for this SSO instance
    if not SSOMapping.objects.filter(sso_provider=sso_instance).exists():
        return (
            "SUCCESS",
            f"SSO group synchronization complete for User '{user_profile.username}', no group mappings found for '{sso_instance}'.",
        )

    groups = []

    groups = attrs_from_sso.get(GROUPS_CLAIM, [])

    group_dicts = [{"id": group_id} for group_id in groups]

    groups = group_dicts

    # Aggregators — initialize ONCE, before the outer loop
    is_super_admin = False
    is_cbadmin = False
    is_devops_admin = False
    is_global_viewer = False
    all_memberships = []
    all_admin_roles = []
    any_mapping_matched = False

    # Walk each group, accumulate everything
    for group in groups:
        sso_group_identifier = group.get("id", "<no-id>")
        mappings = SSOMapping.get_mappings(sso_instance, sso_group_identifier)
        if not mappings:
            continue

        any_mapping_matched = True
        for m in mappings:
            is_super_admin   |= m.is_super_admin
            is_cbadmin       |= m.is_cbadmin
            is_devops_admin  |= m.is_devops_admin
            is_global_viewer |= m.is_global_viewer
            all_admin_roles.extend(list(m.admin_roles.all()))
            all_memberships.extend(
                m.get_memberships(profile=user_profile, as_dict=True)
            )

    logger.info(f"Aggregated memberships across all groups: {all_memberships}")
    logger.info(f"Aggregated admin_roles across all groups: {all_admin_roles}")

    # Apply flags ONCE
    user_profile.super_admin   = is_super_admin
    user_profile.devops_admin  = is_devops_admin
    user_profile.global_viewer = is_global_viewer
    user_profile.save()

    user_profile.user.is_superuser = is_cbadmin
    user_profile.user.save()

    user_profile.admin_roles.set(all_admin_roles)

    # Reconcile group-role memberships ONCE, against the full aggregated set
    user_membership_ids = []
    descendant_group_ids = []
    role_ids = []
    for membership in all_memberships:
        grm, _ = GroupRoleMembership.objects.get_or_create(**membership)
        user_membership_ids.append(grm.id)
        descendant_group_ids.extend(g.id for g in grm.group.get_descendant_list())
        role_ids.append(grm.role_id)

    invalid_grms = user_profile.grouprolemembership_set.exclude(
        id__in=user_membership_ids
    ).exclude(
        group_inherited_from__in=descendant_group_ids,
        role__in=role_ids,
    )
    deleted = invalid_grms.delete()
    logger.info(
        f"SSO sync for {user_profile.username}: "
        f"kept {len(user_membership_ids)} memberships, "
        f"deleted {deleted[0]} stale memberships."
    )
    return (
        "SUCCESS",
        f"SSO group synchronization complete for User '{user.userprofile.username}'.",
    )

In the run method, uncomment the group sync code:

# status, message = sync_groups_for_user(
#     sso_instance,
#     user,
#     attrs_from_sso,
# )
# msg += f"\n{message}"

Select Save Code.