基本的な API と機能の理解

Cognito ユーザープールの低レベル API に対応する boto3 のインターフェースを直接操作し以下のようなことを実行することにより、Cognito ユーザープールにおける認証の流れや利用法を理解してみる

サインアップの基本的な流れ

import env
import boto3
from boto3.session import Session
from getpass import getpass

session = Session(profile_name=env.profile)
client = session.client('cognito-idp', region_name=env.region)

print('=== SIGN UP ===')
email = input('Enter Your Email Address: ')
password = getpass('Enter Your Password: ')
response = client.sign_up(
    ClientId=env.client_id,
    Username=email,
    Password=password,
    UserAttributes=[
        {
            'Name': 'email',
            'Value': email
        },
    ]
)
print('=== SIGN UP RESULT ===')
print(response)


print('=== CONFIRM SIGN UP ===')
code = input('Enter the Verification Code: ')
response = client.confirm_sign_up(
    ClientId=env.client_id,
    Username=email,
    ConfirmationCode=code,
)
print('=== CONFIRM SIGN UP RESULT ===')
print(response)

サインインの基本的な流れ

基本的には各フローに共通して以下のようなやりとりを行う

認証に成功すると IDToken, AccessToken, RefreshToken が返却される

上記のような Cognito トークンに関連して、以下のような関連事項がある

続いて各認証フローの流れを boto3 (AWS SDK for Python) を使って確認していく

USER_SRP_AUTH: クライアント側認証フロー

import env
from boto3.session import Session
from getpass import getpass
from warrant.aws_srp import AWSSRP

session = Session(profile_name=env.profile)
client = session.client('cognito-idp', region_name=env.region)

print('=== SIGN IN ===')
email = input('Enter Your Email Address: ')
password = getpass('Enter Your Password: ')

srp = AWSSRP(username=email, password=password, pool_id=env.user_pool_id, client_id=env.client_side_id, client=client)
srp_a = srp.get_auth_params()['SRP_A']
response = client.initiate_auth(
    AuthFlow='USER_SRP_AUTH',
    AuthParameters={
        'USERNAME': email,
        'SRP_A': srp_a,
    },
    ClientId=env.client_side_id,
)
print(response)

assert(response['ChallengeName'] == 'PASSWORD_VERIFIER')
challenge_response = srp.process_challenge(response['ChallengeParameters'])
response = client.respond_to_auth_challenge(
    ClientId=env.client_side_id,
    ChallengeName='PASSWORD_VERIFIER',
    ChallengeResponses=challenge_response
)
print('=== SIGN IN RESULT ===')
print(response)

ADMIN_USER_PASSWORD_AUTH: 管理 API によるユーザー名パスワードベース認証

import env
import json
from boto3.session import Session
from getpass import getpass

session = Session(profile_name=env.profile)
client = session.client('cognito-idp', region_name=env.region)

print('=== SIGN IN ===')
email = input('Enter Your Email Address: ')
password = getpass('Enter Your Password: ')
response = client.admin_initiate_auth(
    UserPoolId=env.user_pool_id,
    ClientId=env.client_side_id,
    AuthFlow='ADMIN_USER_PASSWORD_AUTH',
    AuthParameters={
        'USERNAME': email,
        'PASSWORD': password,
    },
)
print(json.dumps(response, ensure_ascii=False))

USER_PASSWORD_AUTH: ユーザー名パスワードベース認証

import env
import json
import boto3
import base64
from boto3.session import Session
from getpass import getpass

session = Session(profile_name=env.profile)
client = session.client('cognito-idp', region_name=env.region)

print('=== InitiateAuth ===')
email = input('Enter Your Email Address: ')
password = getpass('Enter Your Password: ')
response = client.initiate_auth(
    AuthFlow='USER_PASSWORD_AUTH',
    AuthParameters={
        'USERNAME': email,
        'PASSWORD': password,
    },
    ClientId=env.client_side_id,
)
print(json.dumps(response, ensure_ascii=False))

jwt = response['AuthenticationResult']['IdToken']
tmp = jwt.split('.')
tmp[0] += "=" * ((4 - len(tmp[0]) % 4) % 4)
tmp[1] += "=" * ((4 - len(tmp[1]) % 4) % 4)
header = json.loads(base64.b64decode(tmp[0]).decode())
payload = json.loads(base64.b64decode(tmp[1]).decode())
print('IDToken Header:')
print(header)
print('IDToken Payload')
print(payload)

# IDToken Header:
# {'kid': '5zQK+U...=', 'alg': 'RS256'}
# IDToken Payload
# {'sub': '18021af4-...', 'cognito:groups': ['admin'], 'email_verified': True, 'cognito:preferred_role': 'arn:aws:iam::...:role/CmdCognitoIdPEC2ReadOnlyRole', 'iss': 'https://cognito-idp.ap-northeast-1.amazonaws.com/ap-northeast-1_....', 'cognito:username': '....', 'custom:oshi': 'Yoshino Aoyama', 'cognito:roles': ['arn:aws:iam::...:role/CmdCognitoIdPEC2ReadOnlyRole'], 'aud': '50t17eat...', 'event_id': '93f6e6ce-...', 'token_use': 'id', 'auth_time': 1579300572, 'exp': 1579304172, 'iat': 1579300572, 'email': '...'}

トークンの更新　

import env
from time import sleep
from boto3.session import Session
from getpass import getpass
from warrant.aws_srp import AWSSRP

session = Session(profile_name=env.profile)
client = session.client('cognito-idp', region_name=env.region)

print('=== Refresh Token Flow ===')
email = input('Enter Your Email Address: ')
password = getpass('Enter Your Password: ')
srp = AWSSRP(username=email, password=password, pool_id=env.user_pool_id, client_id=env.client_side_id, client=client)
tokens = srp.authenticate_user()
refresh_token = tokens['AuthenticationResult']['RefreshToken']
print('got a refresh token: ' + refresh_token)
sleep(2)

response = client.initiate_auth(
    AuthFlow='REFRESH_TOKEN_AUTH',
    AuthParameters={
        'REFRESH_TOKEN': refresh_token
    },
    ClientId=env.client_side_id,
)
print(response)

MFA 必須時のサインアップ・サインイン

ソーシャル IdP を介したサインイン

Login with Amazon (LWA)

import env
import json
import requests

# ブラウザで以下の URL にアクセスしてログインする
# 'https://' + env.domain + /login?client_id=' + env.client_side_id + '&response_type=code&scope=openid+profile&redirect_uri=' + env.redirect_url
print('===  POST /oauth2/token ===')
code = input('Enter Code: ')
url = 'https://' + env.domain + '/oauth2/token'
headers = {'content-type': 'application/x-www-form-urlencoded'}
payload = [
    ('grant_type', 'authorization_code'),
    ('client_id', env.client_side_id),
    ('code', code),
    ('redirect_uri', env.redirect_url)
]
response = requests.post(url=url, headers=headers, params=payload)
print(response.text)
# ===  GET /oauth2/token ===
# return {"id_token":"eyJ...","access_token":"eyJ...","refresh_token":"eyJ...","expires_in":3600,"token_type":"Bearer"}

tokens = json.loads(response.text)
print(tokens)
access_token = tokens['access_token']

print('===  GET /oauth2/userInfo ===')
url = 'https://' + env.domain + '/oauth2/userInfo'
headers = {'Authorization': 'Bearer ' + access_token }
response = requests.get(url=url, headers=headers)
print(response.text)

# ===  GET /oauth2/userInfo ===
# {"sub":"f9969dec-...","email":"...","username":"Facebook_296..."}

Facebook ログイン

Google サインイン

Sign in with Apple