# SSO orchestration actions

> Define how users are managed during SSO login using orchestration actions

---

Site [llms.txt](/llms.txt)

---

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

1. CMP runs the incoming user through any enabled **SSO User Update** orchestration actions.
2. CMP runs the same user through any enabled **External Users Sync** orchestration actions, as described in [External Users Sync](#external-users-sync).
3. The default SSO User Update action updates the user's fields based on information and any **Required Attributes** defined on the SSO provider.

<div class="alert alert-info" role="alert">


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.
</div>


## 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](/docs/cmp/configure/access-and-security/groups-roles-permissions/permission-sync/#use-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.

## Sync groups on SSO login

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

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

   ![Screenshot of the Orchestration Actions page.](/img/cmp/configure/access-and-security/users/sso-saml/orchestration-actions/orchestration-actions.png)

3. Select the **User Update from SSO Provider** action.
4. Select the **Code** tab.
5. Locate the `sync_groups_for_user` method.
6. Replace the following lines with the appropriate code for your SSO IdP:

   ```python
   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"
   )
   ```

   <div class="alert alert-info" role="alert">


   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.
   </div>


   **Okta**

   ```python
       # 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**

   ```python
       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**

   ```python
       # 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}'.",
       )
   ```

7. In the `run` method, uncomment the group sync code:

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

8. Select **Save Code**.