We believe empowering engineers drives innovation.
We believe empowering engineers drives innovation.
Onboarding a new application to any customer identity as a service platform, such as Auth0 or AWS Cognito is simple and straightforward; however, existing customer identities in a brownfield development scenario (for example an existing application with active users where we want or need to replace the existing identity platform) can pose a challenge - how do you migrate these customers and applications to a new identity platform with minimal impact to the user experience?
With Auth0, we can leverage the Bulk User Import feature to import these existing customer identities while maintaining their existing password. In brownfield development this allows for an optimal user experience, minimal additional development effort, and automatic conversion of the hashed password to Auth0’s standard hashing algorithm.
To accomplish this, Auth0 provides an endpoint on their management API to create user import jobs. This endpoint takes a JSON file in a format specified by Auth0, the connection_id (destination) for the import, and a few other optional parameters. Once these users have been imported and our application has been updated to use Auth0 for authentication, Auth0 will take the password provided by the user on the login page, hash it according to the algorithm we specified in the import, and, if the hashes match, rehash the password to the hash they use for secure password storage. This allows for existing customers to log in with their existing username and password and have that password be converted to Auth0’s algorithm as part of the login process. We don’t have to force them to reset their passwords or write custom code to validate the existing password as we would with AWS Cognito.
Let’s say we’re a developer and this sprint we have been tasked with migrating users from our bespoke identity system to Auth0. Let’s assume we have a JSON formatted export of our users and our Auth0 tenant is already setup. We’ll need to do the following:
For demonstration purposes, I have created a set of 100 usernames and passwords; the usernames were randomly generated and the passwords taken from Have I Been Pwned’s Pwned Passwords list. The example identities used in this article can be found in “Example Users” file in the resources section at the end of the article.
The JSON schema provided by Auth0 supports quite a few properties, but we’re going to keep our example basic. We’ll be converting to JSON that resembles the following. The script used to convert our example users as well as the example import file can be found later in this section.
[
{
"email": "jdoe@example.com",
"email_verified": false,
"custom_password_hash": {
"algorithm": "sha1",
"hash": {
"value": "MDAwMDAwQTc4NkVDNTY3NjBGMzUwRTlBREUxMUFFOEE3QjcwNUU2Qg==",
"encoding": "base64"
}
}
}
]
Our input JSON resembles the following:
[
{
"username": "jdoe@example.com",
"hash": "000000A786EC56760F350E9ADE11AE8A7B705E6B"
}
]
We can do this with a simple Python script:
import json
import base64
def convert_example_user_to_auth0(example_user):
return {
'email': example_user['username'],
'email_verified': False,
'custom_password_hash': {
'algorithm': "sha1",
'hash': {
'value': base64.b64encode(example_user['hash'].encode('utf-8')).decode('utf-8'),
'encoding': "base64"
}
}
}
example_users_file = "./pwned_passwords/example_users.json"
auth0_import_file = "./example_users_import.json"
with open(example_users_file, "r") as f_in:
example_users = json.load(f_in)
auth0_users = list(map(convert_example_user_to_auth0, example_users))
with open(auth0_import_file, "w") as f_out:
f_out.write(json.dumps(auth0_users, indent=2))
The resulting file (“Example Import File”) can be found in the Resources section below.
The JSON schema supports several other properties that can be useful to your application. In particular, app_metadata can be used to track data, such as a tenant ID, that affect your application’s core functionality. As you’re planning your migration, be sure to review this schema thoroughly to identify which properties are critical to your application.
To perform a test import with this file, we don’t even need to write any code. Instead, we can go to Auth0’s API docs, set our API token, and invoke the endpoint via our browser: https://auth0.com/docs/api/management/v2#!/Jobs/post_users_imports.
We can see the response here:
Since we set send_completion_email to true, once the import has been completed we’ll receive an email similar to the one in the screenshot below:
Manually sending the request via the browser is fine for a test migration, but we’ll want to make this repeatable. Let’s start down this path by writing a script which allows us to provide the import file, our API URL, API token, and other parameters. This could, of course, be expanded to allow for larger files to be split into 500kb chunks, partitioning of users based on tenant, and other nice-to-have features.
import argparse
import requests
import os
import sys
parser = argparse.ArgumentParser(
prog="auth0_user_importer",
description="""Submit batches of users to be imported to Auth0 via the create user import
job endpoint.""",
)
parser.add_argument(
"-f",
"--filename",
required=True,
help="The name of the file containing the users to import.",
)
parser.add_argument(
"-t",
"--auth-token",
required=True,
help="The authorization token to use when calling the Auth0 Management API.",
)
parser.add_argument(
"--api-url", required=True, help="The URL for the Auth0 Management API."
)
parser.add_argument(
"--connection-id",
required=True,
help="The connection ID to which the users should be imported.",
)
parser.add_argument(
"--upsert",
required=False,
action="store_true",
help="Whether to update users if they already exist (true) or to ignore them (false).",
)
parser.add_argument(
"--send-completion-email",
required=False,
action="store_true",
help="""Whether to send a completion email to all tenant owners when the job is
finished (true) or not (false).""",
)
parser.add_argument(
"--wait",
required=False,
action="store_true",
help="""Whether the script should wait to exit until the import job has completed.
You may hit rate limit issues when using this.""",
)
args = parser.parse_args()
api_url = args.api_url
headers = {"authorization": f"Bearer {args.auth_token}"}
if not (api_url.endswith("/")):
api_url += "/"
file_size = os.path.getsize(args.filename)
if file_size >= 500000:
print("ERROR: file is greater than 500kb")
sys.exit(1)
s = requests.Session()
req = requests.Request(
"POST",
f"{args.api_url}jobs/users-imports",
headers=headers,
files=[
("users", ("USERS_IMPORT_FILE.json", open(args.filename, "rb"), "text/json")),
("connection_id", (None, args.connection_id)),
("upsert", (None, "true" if args.upsert else "false")),
(
"send_completion_email",
(None, "true" if args.send_completion_email else "false"),
),
],
)
prepped_req = req.prepare()
resp = s.send(prepped_req)
if not (resp.ok):
print(f"ERROR: {resp.json()}")
sys.exit(1)
print(f'Submitted job {resp.json()["id"]}')
if not (args.wait):
sys.exit(0)
job_id = resp.json()["id"]
resp = requests.get(f"{api_url}jobs/{job_id}", headers=headers)
while resp.ok and resp.json()["status"] != "completed":
resp = requests.get(f"{api_url}jobs/{job_id}", headers=headers)
print(resp.json())
resp = requests.get(f"{api_url}jobs/{job_id}/errors", headers=headers)
print(resp.json())
By leveraging Auth0’s Bulk User Import feature, we can quickly and easily import customer identities with no negative impact to the user experience and minimal code. Unlike some other platforms, Auth0 allows us to bring our existing password hashes and has support for a broad range of hashing algorithms. This reduces onboarding complexity and provides a great story for migrating applications in a brownfield development scenario.