Migrate Cognito user pool with AWS Lambda and CDK for IaC

During the last week me and my teammates had the objective to change the sign in flow of the mobile application to allow a sign in with preferred username. One little obstacle was, that Cognito attributes were immutable and we will need to create a new Cognito to fulfill the requirement. This led to the need for migration of our user-base.

There are two possible types of user pool migrations:

1. Migrate user one at a time, only when the user sign in. This option is slower and requires all users from the user-base to sign in. The advantage is that the password hash is also migrated. It is a valuable option for disaster recovery, because it allows you to have Cognito in another region with same user base.

2. Batch user migration has the downside, that users are required to perform password reset after the migration. The advantage is, that the batch migration is faster and is easy to perform

This article is primarily for the second option, where we perform the batch user migration. Below is the diagram of the solution:

A lambda function will perform listUsers api call to get users attributes. After that the function will save the output in S3 as csv. To import users there is a requirement, that the Cognito headers are present in the csv. To find out the required headers, you can use following command:

	cloudshell-user@ip-10-0-93-38 ~]$ aws cognito-idp get-csv-header --user-pool-id "eu-west-1_XXXXXXX"
	{
	"UserPoolId": "eu-west-1_XXXXXXX",
	"CSVHeader": [
	"name",
	"given_name",
	"family_name",
	"middle_name",
	"nickname",
	"preferred_username",
	"profile",
	"picture",
	"website",
	"email",
	"email_verified",
	"gender",
	"birthdate",
	"zoneinfo",
	"locale",
	"phone_number",
	"phone_number_verified",
	"address",
	"updated_at",
	"custom:joinedOn",
	"cognito:mfa_enabled",
	"cognito:username"
	]
	}

view raw getCSVHeaders.sh hosted with ❤ by GitHub

All the headers are required to be present as column in the csv. Some of the values in the csv could be empty. The bare minimum values, that should be filled for a user are:

• name

• email

• email_verified — should be set to true. After the imports user should reset their password. The import in the new Cognito will fail if this field is empty or false

• phone_number_verfied — should be set to true or false, even when it is not in use

• cognito:mfa_enabled — should be set to false before the export

• cognito:username — In our case this was required field to preserve as unique sequence, because it was used in the MongoDB to save user specific values.

I would recommend you to start with one user import to see how it is working. Below is a sample:

	name	given_name	family_name	middle_name	nickname	preferred_username	profile	picture	website	email	email_verified	gender	birthdate	zoneinfo	locale	phone_number	phone_number_verified	address	updated_at	custom:joinedOn	cognito:mfa_enabled	cognito:username
	probko									[e-mail]	TRUE						FALSE				FALSE	probko.testov

view raw csvMinimumRequirements.csv hosted with ❤ by GitHub

The full script, that i used to migrate was developed in Python. I have added some comments in each function:

	import os
	import time
	import sys
	import boto3
	REGION = "eu-central-1"
	S3_CLIENT = boto3.client("s3", REGION)
	COGNITO_CLIENT = boto3.client("cognito-idp", REGION)
	USER_POOL_ID = os.environ['USER_POOL_ID']
	EXPORTED_RECORDS_COUNT = 0
	BUCKET_NAME = os.environ["BUCKET_NAME"]
	PAGINATION_COUNTER = 0
	CSV_FILE_NAME = "CognitoUsers.csv"
	REQUIRED_ATTRIBUTES = [
	"name", "given_name", "family_name", "middle_name", "nickname",
	"preferred_username", "profile", "picture", "website", "email",
	"email_verified", "gender", "birthdate", "zoneinfo", "locale",
	"phone_number", "phone_number_verified", "address", "updated_at",
	"custom:joinedOn", "cognito:mfa_enabled", "cognito:username"
	]
	MAX_NUMBER_RECORDS = 0
	LIMIT = 60
	FILE_PATH = os.path.join(os.path.dirname(__file__),
	f"../../tmp/{CSV_FILE_NAME}")
	def get_list_cognito_users(cognito_idp_client,
	next_pagination_token="",
	limit=LIMIT):
	"""
	First we get a list of all users in the pull.
	The Limit per page is 60. If there is next page we will move to it
	"""
	return cognito_idp_client.list_users(
	UserPoolId=USER_POOL_ID,
	limit=limit,
	PaginationToken=next_pagination_token
	) if next_pagination_token else cognito_idp_client.list_users(
	UserPoolId=USER_POOL_ID, limit=limit)
	def check_next_pagination_token_existence(user_records):
	"""
	We are checking there is a next page
	"""
	if set(["PaginationToken", "NextToken"]).intersection(set(user_records)):
	pagination_token = user_records[
	"PaginationToken"] if "PaginationToken" in user_records else user_records[
	"NextToken"]
	return pagination_token
	def write_to_csv_file(user_records, csv_file, csv_new_line):
	"""
	We will save each user attribute as csv
	"""
	csv_lines = []
	for user in user_records["Users"]:
	csv_line = csv_new_line.copy()
	for required_attribute in REQUIRED_ATTRIBUTES:
	csv_line[required_attribute] = ""
	if required_attribute in user.keys():
	csv_line[required_attribute] = str(user[required_attribute])
	continue
	if required_attribute in ("phone_number_verified",
	"cognito:mfa_enabled"):
	csv_line[required_attribute] = "false"
	if required_attribute == "email_verified" and required_attribute not in user.keys(
	):
	csv_line[required_attribute] = "false"
	for user_attribute in user["Attributes"]:
	if user_attribute["Name"] == required_attribute:
	csv_line[required_attribute] = str(user_attribute["Value"])
	csv_line["cognito:username"] = user["Username"]
	if not csv_line["email_verified"] == "false":
	csv_lines.append(",".join(csv_line.values()) + "\n")
	print(csv_lines)
	csv_file.writelines(csv_lines)
	global EXPORTED_RECORDS_COUNT
	EXPORTED_RECORDS_COUNT += len(csv_lines)
	global PAGINATION_COUNTER
	print("Page: #{} \n Total Exported Records: #{} \n".format(
	str(PAGINATION_COUNTER), str(EXPORTED_RECORDS_COUNT)))
	def open_csv_file(csv_new_line):
	"""
	Create a csv file in the tmp directory of the lambda function
	"""
	try:
	csv_file = open(FILE_PATH, "w")
	csv_file.write(",".join(csv_new_line.keys()) + "\n")
	except Exception as err:
	error_message = repr(err)
	print("\nERROR: Can not create file: " + FILE_PATH)
	print("\tError Reason: " + error_message)
	sys.exit()
	return csv_file
	def cooldown_before_next_batch():
	"""
	Wait before next batch
	"""
	time.sleep(0.15)
	def save_file():
	"""
	Upload file to S3
	"""
	with open(FILE_PATH, "rb") as file:
	S3_CLIENT.upload_fileobj(file, BUCKET_NAME, CSV_FILE_NAME)
	def lambda_handler(event, context):
	"""
	lambda_handler is called, when the lambda function was invoked
	"""
	pagination_token = ""
	csv_new_line = {
	REQUIRED_ATTRIBUTES[i]: ""
	for i in range(len(REQUIRED_ATTRIBUTES))
	}
	csv_file = open_csv_file(csv_new_line)
	global PAGINATION_COUNTER
	while pagination_token is not None:
	try:
	user_records = get_list_cognito_users(
	cognito_idp_client=COGNITO_CLIENT,
	next_pagination_token=pagination_token,
	limit=LIMIT
	if LIMIT < MAX_NUMBER_RECORDS else MAX_NUMBER_RECORDS)
	except COGNITO_CLIENT.exceptions.ClientError as client_error:
	error_message = client_error.response["Error"]["Message"]
	print("PLEASE CHECK YOUR COGNITO CONFIGS")
	print("Error Reason: " + error_message)
	csv_file.close()
	sys.exit()
	except Exception as unknown_exception:
	print("Error Reason: " + str(unknown_exception))
	csv_file.close()
	sys.exit()
	pagination_token = check_next_pagination_token_existence(user_records)
	write_to_csv_file(user_records, csv_file, csv_new_line)
	PAGINATION_COUNTER += 1
	cooldown_before_next_batch()
	csv_file.close()
	save_file()

view raw cognitoUserMigration.py hosted with ❤ by GitHub

To deploy it I used cdk as infrastructure as code. One of the benefits of cdk is the usage of high level constructs, which make the work with it easy and fast:

	import { Construct, Duration, Stack } from '@aws-cdk/core';
	import { Bucket, BucketEncryption } from '@aws-cdk/aws-s3';
	import { Code, Function, IFunction, Runtime } from '@aws-cdk/aws-lambda';
	import * as path from 'path';
	import { Effect, PolicyStatement } from '@aws-cdk/aws-iam';
	export interface CognitoUserMigrationLambdaProps {
	readonly userPoolId: string;
	readonly bucketName: string;
	}
	export class CognitoUserMigrationLambdaStack extends Stack {
	public readonly function: IFunction;
	constructor(scope: Construct, id: string, props: CognitoUserMigrationLambdaProps) {
	super(scope, id);
	// Create a bucket, where to store the user list
	const bucket = new Bucket(this, props.bucketName, {
	encryption: BucketEncryption.S3_MANAGED,
	bucketName: props.bucketName,
	});
	// create aws lambda function
	this.function = new Function(this, `${id}Function`, {
	code: Code.fromAsset(path.join(__dirname, './cognito-user-migration')),
	handler: 'cognitoUserMigration.lambda_handler',
	runtime: Runtime.PYTHON_3_8,
	timeout: Duration.seconds(60),
	environment: {
	USER_POOL_ID: props.userPoolId,
	BUCKET_NAME: props.bucketName,
	},
	});
	// add to policy allow get and put objects in the bucket
	this.function.addToRolePolicy(
	new PolicyStatement({
	actions: ['s3:*Object*', 's3:ListBucket'],
	resources: [bucket.arnForObjects('*'), bucket.bucketArn],
	effect: Effect.ALLOW,
	}),
	);
	// add to policy the possibility to list userpool users
	this.function.addToRolePolicy(
	new PolicyStatement({
	actions: ['cognito-idp:ListUsers'],
	resources: [`arn:aws:cognito-idp:${this.region}:${this.account}:userpool/${props.userPoolId}`],
	effect: Effect.ALLOW,
	}),
	);
	// add to the bucket policy read and write permissions for the lambda
	bucket.grantReadWrite(this.function);
	}
	}

view raw CognitoUserMigrationLambdaStack.ts hosted with ❤ by GitHub

After the lambda was invoked, you an object CognitoUsers.csv will be present in the bucket. To import the users in the new Cognito follow following procedure:

1. Go to Cognito in the management console

2. Select “Manage User Pools” and select the name of the new userpool

3. Go to “Users and groups” , which is below the “General settings”

4. Select the “Import users” button

5. “Create import job”, it is required, that you created a role or the role will be created during the process with permissions to write to Cloudwatch logs — arn:aws:iam::XXX:role/service-role/Cognito-UserImport-Role. You will select the csv file, that was downloaded from S3

6. After the job is created you should click on the start.

Summary: Until this task, I was convinced, that the export and migration was hard task, simply, because all transfers out of the cloud were not so easy and there is always the vendor lock-in and the specifics of some of the services. I was pleasantly surprised, that the Cognito user migration was easy,at least between two Cognitos, and I finished it for a couple of hours, even with the change of the attributes. My next task is improve the whole process by automating also the import with the help of a Step Function
Configuring User Pool Attributes
既存のログイン認証基盤からCognitoに移行するために調べてわかったこと - Qiita

https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/cognito-idp.html
Creating the User Import .csv File