こんにちは。ROBOT PAYMENT （以下、ロボペイ）でエンジニアをしているtakamoriです。 私が所属しているチームでは、請求先マイページ機能を開発しており、その中でユーザー認証基盤をAuth0からCognitoへと移行させました。そこで今回は、Auth0からCognitoへのユーザー移行手順を書いていきたいと思います。 ※ 本記事ではAuth0やCognitoの環境構築は対象外で、それぞれの環境が構築済み前提となります。
移行手順
- Auth0からユーザーをエクスポート
- Auth0ユーザー情報をCognitoユーザー情報へマッピング
- Cognitoへユーザーをインポート
Auth0からユーザーをエクスポート
Auth0からのユーザーをエクスポートするには、ExportUsersJob APIを利用します。GetUsers APIを利用して取得することも可能ですが1,000件の取得制限があるので、制限のないExportUsersJob APIを利用しました。
ExportUsersJob APIは非同期APIなので、進行状況をGetJob APIで確認したり、結果のダウンロードが必要となります。 またExportUsersJob APIの実行結果はcsv or json形式で取得可能です。今回はCognitoへのインポートがcsv形式のみサポートしているのでcsv形式で実装しています。
Auth0からユーザーをエクスポートする流れ
- ExportUsersJob APIの呼び出し
- ExportUsersJobのステータスを確認し、完了まで待つ
- 結果をダウンロード
実装したスクリプト
CSV_HEADER_MAPPING = {
"cognito:username": "email",
"email": "email",
"name": "name",
"family_name": "family_name",
"given_name": "given_name",
"phone_number": "phone_number",
"nickname": "nickname",
"updated_at": "updated_at",
"email_verified": "email_verified",
"phone_number_verified": "phone_verified",
}
OUTPUT_PATH = "./"
def main(token, connection_id, domain):
# 1. ExportUsersJob APIの呼び出し
job_id = _post_users_exports_job(token, connection_id, domain)["id"]
# 2. ExportUsersJobのステータスが完了するまで繰り返し確認し、完了したら結果をダウンロード
while True:
# 指定Jobの状況取得
json_response = _get_job(token, job_id)
status = json_response["status"]
if status == "completed":
# 3. 結果のダウンロード
_download_csv(json_response["location"], OUTPUT_PATH)
break
time.sleep(1)
def _post_users_exports_job(token, connection_id, domain):
url = "https://" + domain + "/api/v2/jobs/users-exports"
headers = {
"Authorization": "Bearer " + token,
"content-type": "application/json",
}
fields = []
for value in CSV_HEADER_MAPPING.values():
d = { "name": value }
fields.append(d)
params = {
"connection_id": connection_id,
"format": "csv",
"fields": fields,
}
# ExportUsersJob API呼び出し
result_users_exports = requests.post(
url=url,
headers=headers,
data=json.dumps(params),
)
return result_users_exports.json()
def _get_job(token, job_id):
url = "https://" + domain + "/api/v2/jobs/" + job_id
headers = {
"Authorization": "Bearer " + token,
"content-type": "application/json",
}
# ExportUsersJobステータス取得
result_get_job = requests.get(
url=url,
headers=headers,
)
if result_get_job.status_code != 200:
raise Exception("status :" + result_get_job.status_code)
return result_get_job.json()
def _download_csv(url, output_path):
tmp_path = "./tmp"
urllib.request.urlretrieve(url, tmp_path)
with gzip.open(tmp_path, mode="rb") as gzip_file:
with open(output_path, mode="wb") as file:
shutil.copyfileobj(gzip_file, file)
os.remove(tmp_path)
Auth0ユーザー情報をCognitoユーザー情報へマッピング
ExportUsersJob APIで取得したAuth0のユーザー情報をCognitoのユーザー情報へマッピングします。今回は移行スクリプトをPythonで作成したので、データ処理で有名なpandasを利用してマッピングしました。 Cognitoのユーザー情報用csvヘッダーはCognito管理画面から取得できます。
Cognitoユーザー用csvヘッダーの取得
取得したcsvヘッダー
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,cognito:mfa_enabled,cognito:username
実装したスクリプト
CSV_HEADER_MAPPING = {
"cognito:username": "email",
"email": "email",
"name": "name",
"family_name": "family_name",
"given_name": "given_name",
"phone_number": "phone_number",
"nickname": "nickname",
"updated_at": "updated_at",
"email_verified": "email_verified",
"phone_number_verified": "phone_verified",
}
COGNITO_CSV_HEADER = "./template/header.csv"
COGNITO_CSV_PATH = "cognito.csv"
def _create_mapped_csv(csv_for_auth0):
with open(CSV_HEADER_MAPPING, "r") as json_open:
mapping_json = json.load(json_open)
try:
df_auth0 = pd.read_csv(csv_for_auth0, encoding="shift_jis")
except EmptyDataError:
print("ERROR: auth0 users is empty")
# auth0ユーザーがいない場合はメッセージ出力後、Exceptionで実行を停止する
raise EmptyDataError
# csv情報のマッピング
df_cognito_headers = pd.read_csv(COGNITO_CSV_HEADER, encoding="shift_jis")
df_cognito = pd.DataFrame()
for column in df_cognito_headers:
if column in mapping_json.keys() and mapping_json[column] in df_auth0:
df_cognito.insert(
loc=len(df_cognito.columns),
column=column,
value=df_auth0[mapping_json[column]],
)
else:
df_cognito.insert(loc=len(df_cognito.columns), column=column, value="")
# csvファイル出力
df_cognito.to_csv(COGNITO_CSV_PATH, index=False, encoding="shift_jis")
Cognitoへユーザー情報をインポート
Cognitoへのユーザーインポートは、aws cognito-idpコマンドを利用して実施します。 今回のユーザー移行に利用したコマンドを簡単に説明します。 - create-user-import-job：UserImportJobを作成。ユーザーインポート実行にはstart-user-import-jobコマンド呼び出しが必要。 - start-user-import-job：UserImportJobの実行。Jobは非同期で実行される。 - describe-user-import-job：UserImportJobのステータス確認に利用。
Cognitoへユーザーをインポートする流れ
- UserImportJobの作成
- UserImportJob用のCSVをアップロード
- UserImportJobの実行
- UserImportJobのステータス確認
実装したスクリプト
POOL_ID = "ap-northeast-1_USERPOOLID" REGION = "ap-northeast-1" ROLE_ARN = "cognito-import-role-arn" COGNITO_CSV_PATH = "cognito.csv" def main(env): # 1. UserImportJobの作成 response = _create_user_import_job(POOL_ID, ROLE_ARN)["UserImportJob"] pre_signed_url = response["PreSignedUrl"] job_id = response["JobId"] # 2. UserImportJob用のCSVをアップロード _csv_upload(COGNITO_CSV_PATH, pre_signed_url) # 3. UserImportJobの実行 _start_user_import_job(pool_id, job_id) # 4. UserImportJobのステータス確認 while True: res_descrobe = _describe_user_import_job(pool_id, job_id) status = res_descrobe["UserImportJob"]["Status"] if status == "Succeeded": print("completed import job") break elif status in [ "Expired", "Failed", "Stopped"]: raise Exception("%s import job. please check cloud-watch-log" % status) else: print("job [%s] processing..." % job_id) def _create_user_import_job(pool_id, role_arn): cmd = "aws cognito-idp create-user-import-job --job-name %s --user-pool-id %s --cloud-watch-logs-role-arn %s" % ( "migrate-from-auth0", pool_id, role_arn ) return _subprocess_call_common(cmd) def _csv_upload(input_file_path, pre_signed_url): content_deposition = "attachment;filename=" + os.path.basename(input_file_path) headers_dict = { "x-amz-server-side-encryption": "aws:kms", "Content-Disposition": content_deposition, } with open(input_file_path, "rb") as f: file_upload_response = requests.put(pre_signed_url, data=f, headers=headers_dict) return file_upload_response def _start_user_import_job(pool_id, job_id): cmd = "aws cognito-idp start-user-import-job --user-pool-id %s --job-id %s" % ( pool_id, job_id, ) _subprocess_call_common(cmd) def _describe_user_import_job(pool_id, job_id): cmd = "aws cognito-idp describe-user-import-job --user-pool-id %s --job-id %s" % ( pool_id, job_id, ) return _subprocess_call_common(cmd) # コマンド実行共通処理 def _subprocess_call_common(cmd): result_run = subprocess.run(cmd, capture_output=True, text=True, shell=True) error = result_run.stderr response = result_run.stdout if _is_json(response): result = json.loads(response) print("json response =======>" + str(result)) else: print("not json response =======>" + str(result)) if error: raise Exception("!!!response error!!! error is " + str(error)) return result
終わりに
Auth0からCognitoへユーザー移行を行う手順や実装例でした。今回はPythonで実装を統一しましたが、Cognito呼び出し部分はCLIコマンドを呼びだす部分がほとんどなので、シェルスクリプトで実装しても良いと思いました。またユーザー移行に関しては、既に稼働しているプロダクトだとユーザー差分を確認するスクリプトも必要になるかと思います。移行に際しては余裕のある計画と厳密な手順が必要だとしみじみ感じました。
We are hiring！！
ROBOT PAYMENTでは一緒に働く仲間を募集しています!!!
speakerdeck.com
www.robotpayment.co.jp
