Migrating Users and Groups from Cognito

Migrating your users and groups from one auth source to another can seem scary. Fortunately, we're here to help! See our Migrating Users to PropelAuth guide for more general information or stay here for an advanced guide, including code snippets, on how to migrate your users and groups from AWS Cognito to PropelAuth.

Exporting Users from Cognito

You can export your users from Cognito using the AWS SDK (boto3). First, let's install the required libraries:

pip install boto3

You'll need your AWS credentials configured (via environment variables, AWS CLI, or IAM role) and your Cognito User Pool ID. Run this script to create a cognito_users.json file containing your user data:

import boto3
import json

# Replace with your AWS region and User Pool ID
AWS_REGION = "us-east-1"
USER_POOL_ID = "us-east-1_xxxxxxxxx"

client = boto3.client('cognito-idp', region_name=AWS_REGION)

def get_attribute(attributes, name):
    for attr in attributes:
        if attr['Name'] == name:
            return attr['Value']
    return None

def export_users():
    all_users = []
    pagination_token = None

    while True:
        if pagination_token:
            response = client.list_users(
                UserPoolId=USER_POOL_ID,
                PaginationToken=pagination_token
            )
        else:
            response = client.list_users(
                UserPoolId=USER_POOL_ID
            )

        for user in response['Users']:
            user_data = {
                'id': get_attribute(user['Attributes'], 'sub'),
                'email': get_attribute(user['Attributes'], 'email'),
                'email_confirmed': get_attribute(user['Attributes'], 'email_verified') == 'true',
                'first_name': get_attribute(user['Attributes'], 'given_name'),
                'last_name': get_attribute(user['Attributes'], 'family_name'),
                'username': user['Username'],
                'enabled': user['Enabled'],
            }
            all_users.append(user_data)

        pagination_token = response.get('PaginationToken')
        if not pagination_token:
            break

    with open("cognito_users.json", "w") as outfile:
        json.dump(all_users, outfile, indent=2)

    print(f"Exported {len(all_users)} users to cognito_users.json")

if __name__ == "__main__":
    export_users()

Importing Users into PropelAuth

Unfortunately, Cognito does not allow you to export password hashes. This means a migration away from Cognito (assuming your users login via password) requires your users to reset their password. There are some other options, but they would depend on your exact Cognito setup - if you are interested, please reach out to support@propelauth.com.

We can then use cognito_users.json to import your user data to PropelAuth. PropelAuth offers an API endpoint to make migrating users as simple as possible. Before we run the script, let's install PropelAuth's Python library:

pip install propelauth_py

This script will attempt to import each of the users found in your cognito_users.json file into PropelAuth. If any of the imports fail, it will automatically generate a ./unmigrated_users.json file for you.

To run the script we need to get your PROPELAUTH_AUTH_URL and PROPELAUTH_AUTH_API_KEY, both of which can be found in your Backend Integration page in PropelAuth.

from propelauth_py import init_base_auth
import json

# Replace with your PropelAuth Auth URL and API key
PROPELAUTH_AUTH_URL = "https://auth.yourdomain.com"
PROPELAUTH_AUTH_API_KEY = "b1988511..."

# replace with your JSON path
cognito_json_filename = "./cognito_users.json"

# If we fail to migrate any users, we'll output them here
unmigrated_users_filename = "./unmigrated_users.json"

auth = init_base_auth(PROPELAUTH_AUTH_URL, PROPELAUTH_AUTH_API_KEY)

def create_propelauth_user(user_data):
    auth.migrate_user_from_external_source(
        email = user_data.get('email'),
        email_confirmed = user_data.get('email_confirmed'),
        existing_user_id = user_data.get('id'),
        enabled = user_data.get('enabled', True),
        username = user_data.get('username'),
        first_name = user_data.get('first_name'),
        last_name = user_data.get('last_name'),
        properties = user_data.get('properties'),
    )
    print(f"API call successful for user: {user_data.get('id')}")

def process_users():
    users_with_errors = []

    with open(cognito_json_filename, "r") as cognito_json_file:
        user_data = json.load(cognito_json_file)
        # Iterate through each user
        for user in user_data:
            try:
                create_propelauth_user({
                    'id': user.get('id'),
                    'first_name': user.get('first_name'),
                    'last_name': user.get('last_name'),
                    'email': user.get('email'),
                    'email_confirmed': user.get('email_confirmed'),
                    'enabled': user.get('enabled'),

                    # Cognito does not allow you to export password hashes.
                    # Users with passwords will need to reset their password.
                    # password_hash = None

                    # Fields below this are dependent on your setup, you may need to modify
                    #'username': user.get('username'),
                    # Any custom user properties that aren't first_name, last_name, or username go here
                    #'properties': {
                    #    'sample_custom_property': user.get('custom:sample')
                    #}
                })
            except Exception as e:
                users_with_errors.append(user)
                print(f"Error during processing user: {user.get('id')}")
                print(f"Error: {e}")

    if len(users_with_errors) > 0:
        with open(unmigrated_users_filename, "w") as unmigrated_users_file:
            for failed_user in users_with_errors:
                unmigrated_users_file.write(json.dumps(failed_user) + "\n")

        print(f"We couldn't migrate all users - the users we failed to migrate are here {unmigrated_users_filename}")

# Run the script
if __name__ == "__main__":
    process_users()

Exporting Groups from Cognito

Cognito Groups are commonly used as tenants. PropelAuth's equivalent concept is Organizations. In this section, we'll learn how to export your Cognito groups and convert them into PropelAuth organizations.

This script will create a file called group_data.json containing all of your groups and their members.

import boto3
import json
from propelauth_py import init_base_auth

# Replace with your AWS region and User Pool ID
AWS_REGION = "us-east-1"
USER_POOL_ID = "us-east-1_xxxxxxxxx"

# Replace with your PropelAuth Auth URL and API key
PROPELAUTH_AUTH_URL = "https://auth.yourdomain.com"
PROPELAUTH_AUTH_API_KEY = "b1988511..."

client = boto3.client('cognito-idp', region_name=AWS_REGION)
auth = init_base_auth(PROPELAUTH_AUTH_URL, PROPELAUTH_AUTH_API_KEY)

def get_attribute(attributes, name):
    for attr in attributes:
        if attr['Name'] == name:
            return attr['Value']
    return None

def get_propelauth_user(email):
    try:
        user = auth.fetch_user_metadata_by_email(email)
        return user['user_id']
    except Exception as e:
        print(f"Error getting PropelAuth user for email: {email}")
        print(f"Error: {e}")
        return None

def get_group_members(group_name):
    members = []
    pagination_token = None

    while True:
        if pagination_token:
            response = client.list_users_in_group(
                UserPoolId=USER_POOL_ID,
                GroupName=group_name,
                NextToken=pagination_token
            )
        else:
            response = client.list_users_in_group(
                UserPoolId=USER_POOL_ID,
                GroupName=group_name
            )

        for user in response['Users']:
            email = get_attribute(user['Attributes'], 'email')
            propelauth_user_id = get_propelauth_user(email)
            members.append({
                'email': email,
                'propelauth_user_id': propelauth_user_id
            })

        pagination_token = response.get('NextToken')
        if not pagination_token:
            break

    return members

def export_groups():
    all_groups = []
    pagination_token = None

    while True:
        if pagination_token:
            response = client.list_groups(
                UserPoolId=USER_POOL_ID,
                NextToken=pagination_token
            )
        else:
            response = client.list_groups(
                UserPoolId=USER_POOL_ID
            )

        for group in response['Groups']:
            group_data = {
                'name': group['GroupName'],
                'description': group.get('Description', ''),
                'members': get_group_members(group['GroupName'])
            }
            all_groups.append(group_data)

        pagination_token = response.get('NextToken')
        if not pagination_token:
            break

    with open("group_data.json", "w") as outfile:
        json.dump(all_groups, outfile, indent=2)

    print(f"Exported {len(all_groups)} groups to group_data.json")

if __name__ == "__main__":
    export_groups()

Importing Groups into PropelAuth

You should now have a file called group_data.json that contains all of your groups and who belongs to each group. Let's import this data using PropelAuth's API.

Cognito groups don't have built-in roles like some other auth providers. By default, this script will assign all members the "Member" role. You can customize this by updating the get_role_for_member function.

import json
import requests
from propelauth_py import init_base_auth

# Replace with your PropelAuth Auth URL and API key
PROPELAUTH_AUTH_URL = "https://auth.yourdomain.com"
PROPELAUTH_AUTH_API_KEY = "b1988511..."

# replace with your JSON path
GROUP_JSON = "./group_data.json"

auth = init_base_auth(PROPELAUTH_AUTH_URL, PROPELAUTH_AUTH_API_KEY)

def get_role_for_member(member):
    # Customize this function to assign roles based on your needs
    # By default, all members get the "Member" role
    return "Member"

def create_propelauth_org(org_data):
    try:
        response = auth.create_org(
            name = org_data['name']
        )
        print(f"API call successful to create org: {org_data['name']}")
        return response
    except requests.exceptions.RequestException as e:
        print(f"Error making API call to create org: {org_data['name']}")
        print(f"Error: {e}")

def add_user_to_org(org_id, org_user):
    if not org_user.get('propelauth_user_id'):
        print(f"Skipping user {org_user.get('email')} - no PropelAuth user ID found")
        return

    try:
        auth.add_user_to_org(
            user_id=org_user['propelauth_user_id'],
            org_id=org_id,
            role=get_role_for_member(org_user),
        )
        print(f"API call successful to add user {org_user.get('email')} to org {org_id}")
    except requests.exceptions.RequestException as e:
        print(f"Error making API call to add user {org_user.get('email')} to org {org_id}")
        print(f"Error: {e}")

def process_groups():
    with open(GROUP_JSON, 'r') as infile:
        group_data = json.load(infile)

        for group in group_data:
            create_org_response = create_propelauth_org({
                'name': group['name']
            })
            org_id = create_org_response['org_id']
            for member in group['members']:
                add_user_to_org(org_id, member)

# Run the script
if __name__ == "__main__":
    process_groups()

All of your groups and users are now imported! If you run into any problems, do not hesitate to reach out to support@propelauth.com.