ROBOT PAYMENT TECH-BLOG

株式会社ROBOT PAYMENTのテックブログです

Auth0からCognitoへのユーザー移行

こんにちは。ROBOT PAYMENT (以下、ロボペイ)でエンジニアをしているtakamoriです。 私が所属しているチームでは、請求先マイページ機能を開発しており、その中でユーザー認証基盤をAuth0からCognitoへと移行させました。そこで今回は、Auth0からCognitoへのユーザー移行手順を書いていきたいと思います。 ※ 本記事ではAuth0やCognitoの環境構築は対象外で、それぞれの環境が構築済み前提となります。

移行手順

  1. Auth0からユーザーをエクスポート
  2. Auth0ユーザー情報をCognitoユーザー情報へマッピング
  3. 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からユーザーをエクスポートする流れ

  1. ExportUsersJob APIの呼び出し
  2. ExportUsersJobのステータスを確認し、完了まで待つ
  3. 結果をダウンロード

実装したスクリプト

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へユーザーをインポートする流れ

  1. UserImportJobの作成
  2. UserImportJob用のCSVをアップロード
  3. UserImportJobの実行
  4. 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