Check phone_number_verified when post /token API

Author: y-temp4

.envrc.sample

@@ -6,7 +6,8 @@ export PYTHONPATH=./ # 変更必要なし
 export ALIS_APP_ID=xxxxxxxx # serverless-applicationで指定した値を指定してください。environmentsリポジトリのsecretパラメータを判定するのに必要です
 export TEST_COGNITO_USER_POOL_ID=xxxxx # インテグレーションテストで使用されるCognito User Pool IDを指定してください
 export TEST_COGNITO_CLIENT_ID=xxxxx # インテグレーションテストで使用されるCognito User Pool へアクセスできるアプリクライアントのIDを指定してください。このアプリはADMIN_NO_SRP_AUTHをサポートしている必要があります
-export TEST_COGNITO_USER_ID=xxxxx # インテグレーションテストで使用されるCognitoのUserIDを指定してください。environmentリポジトリで指定しているUserpoolにそのユーザが存在している必要があります
+export TEST_COGNITO_USER_ID=xxxxx # インテグレーションテストで使用されるCognitoのUserIDを指定してください。environmentリポジトリで指定しているUserpoolにそのユーザが存在している必要があります。また、このユーザーはphone_number_verifiedをtrueの状態にしてください。
+export TEST_COGNITO_PHONE_NUMBER_NOT_VERIFIED_USER_ID=xxxxx # TEST_COGNITO_USER_IDと同様なユーザー。ただし、このユーザーはphone_number_verifiedをfalseの状態にしてください。
 export TEST_AUTHLETE_SERVER_APP_CLIENT_ID=xxxxx # インテグレーションテストで使用するサーバーサイドアプリケーションのAuthleteのclient_idを指定してください
 export TEST_AUTHLETE_SERVER_APP_CLIENT_SECRET=xxxxx　# インテグレーションテストで使用するサーバーサイドアプリケーションのAuthleteのclient_secretを指定してください
 export TEST_AUTHLETE_CLIENT_APP_CLIENT_ID=xxxxx # インテグレーションテストで使用するクライアントアプリケーションのAuthleteのclient_idを指定してください

lambdas/http/token.py

@@ -1,12 +1,14 @@
 import os
 from lib.authlete_sdk import AuthleteSdk
 from lib.exceptions import ValidationError, AuthleteApiError
+from lib.cognito_user_pool import CognitoUserPool
 from lib.utils import response_builder, logger, verify_supported_media_type
 
 
 def handler(event, context):
+    token = {}
+
     try:
-        logger.info(event)
         if verify_supported_media_type(event['headers']) is False:
             return response_builder(415, {
                 'error_message': "This API only support 'content-type: application/x-www-form-urlencoded' media type"
@@ -26,6 +28,7 @@ def handler(event, context):
             body=event['body']
         )
 
+        # トークン取得処理
         if grant_type == 'authorization_code':
             if data.get('client_secret') is None:
                 token = authlete.get_access_token_from_code(
@@ -38,7 +41,6 @@ def handler(event, context):
                     client_id=data['client_id'],
                     client_secret=data['client_secret']
                 )
-            return response_builder(200, token)
         elif grant_type == 'refresh_token':
             if data.get('client_secret') is None:
                 token = authlete.get_access_token_from_refresh_token(
@@ -51,7 +53,6 @@ def handler(event, context):
                     client_id=data['client_id'],
                     client_secret=data['client_secret']
                 )
-            return response_builder(200, token)
         else:
             return response_builder(400, {
                 'error_message': 'invalid grant_type'
@@ -70,3 +71,35 @@ def handler(event, context):
         return response_builder(500, {
             'error_message': 'Internal Server Error'
         })
+
+    try:
+        congito_user_pool = CognitoUserPool(
+            user_pool_id=os.environ['COGNITO_USER_POOL_ID']
+        )
+
+        authlete = AuthleteSdk(
+            api_key=os.environ['AUTHLETE_API_KEY'],
+            api_secret=os.environ['AUTHLETE_API_SECRET']
+        )
+
+        access_token = token.get('access_token')
+        response_content = authlete.get_user_info(access_token=access_token)
+        attributes = congito_user_pool.get_user_attributes(username=response_content['sub'])
+        phone_number_verified = 'false'
+
+        for attribute in attributes:
+            if attribute['Name'] == 'phone_number_verified':
+                phone_number_verified = attribute['Value']
+
+        if phone_number_verified == 'true':
+            return response_builder(200, token)
+        else:
+            return response_builder(403, {
+                'error_message': 'phone_number must be verified'
+            })
+
+    except Exception as e:
+        logger.error(e)
+        return response_builder(500, {
+            'error_message': 'Internal Server Error'
+        })

lib/cognito_user_pool.py

@@ -18,3 +18,14 @@ def get_user_sub_value(self, username):
 
         except ClientError as e:
             raise e
+
+    def get_user_attributes(self, username):
+        try:
+            response = self.client.admin_get_user(
+                UserPoolId=self.user_pool_id,
+                Username=username
+            )
+            return response['UserAttributes']
+
+        except ClientError as e:
+            raise e

tests/integration/common.py

@@ -97,7 +97,7 @@ def get_refresh_token(self, code_verifier, code_challenge):
         data = response.json()
         return data['refreshToken']
 
-    def get_authorization_code(self, code_challenge):
+    def get_authorization_code(self, code_challenge, is_phone_number_verified_user=True):
         response = requests.post(
             url='https://api.authlete.com/api/auth/authorization',
             auth=(
@@ -109,14 +109,16 @@ def get_authorization_code(self, code_challenge):
         )
         data = response.json()
 
+        cognito_user_id = os.environ['TEST_COGNITO_USER_ID'] if is_phone_number_verified_user else os.environ['TEST_COGNITO_PHONE_NUMBER_NOT_VERIFIED_USER_ID']
+
         response = requests.post(
             url='https://api.authlete.com/api/auth/authorization/issue',
             auth=(
                 self.authlete_api_key,
                 self.authlete_api_secret
             ),
             headers={'content-type': 'application/json'},
-            data=json.dumps({'ticket': data['ticket'], 'subject': os.environ['TEST_COGNITO_USER_ID']})
+            data=json.dumps({'ticket': data['ticket'], 'subject': cognito_user_id})
         )
 
         data = response.json()

tests/integration/test_token_client_app.py

@@ -12,6 +12,7 @@ def setup(self):
         self.code_verifier = authlete_test_sdk.get_code_verifier()
         self.code_challenge = authlete_test_sdk.get_code_challenge(self.code_verifier)
         self.authorization_code = authlete_test_sdk.get_authorization_code(self.code_challenge)
+        self.authorization_code_by_phone_number_not_verified_user = authlete_test_sdk.get_authorization_code(self.code_challenge, False)
 
     def test_return_200_with_authorization_code(self, endpoint):
         response = requests.post(
@@ -133,3 +134,16 @@ def test_return_400_with_missing_client_id(self, endpoint):
         assert response.status_code == 400
         error = response.json()
         assert error['error_message'] == 'Missing client_id'
+
+    def test_return_403_with_phone_number_not_verified_user(self, endpoint):
+        response = requests.post(
+            url=endpoint + '/token',
+            headers={
+                'Content-Type': 'application/x-www-form-urlencoded'
+            },
+            data='grant_type=authorization_code&code='+self.authorization_code_by_phone_number_not_verified_user+'&redirect_uri=http://localhost&code_verifier='+self.code_verifier+'&client_id='+os.environ['TEST_AUTHLETE_CLIENT_APP_CLIENT_ID']
+        )
+
+        assert response.status_code == 403
+        error = response.json()
+        assert error['error_message'] == 'phone_number must be verified'

tests/integration/test_token_server_app.py

@@ -13,6 +13,7 @@ def setup(self):
         self.code_verifier = authlete_test_sdk.get_code_verifier()
         self.code_challenge = authlete_test_sdk.get_code_challenge(self.code_verifier)
         self.authorization_code = authlete_test_sdk.get_authorization_code(self.code_challenge)
+        self.authorization_code_by_phone_number_not_verified_user = authlete_test_sdk.get_authorization_code(self.code_challenge, False)
 
     def test_return_200_with_authorization_code(self, endpoint):
         response = requests.post(
@@ -158,3 +159,17 @@ def test_return_400_with_invalid_client_secret(self, endpoint):
         assert response.status_code == 400
         error = response.json()
         assert error['error_message'] == 'The client credentials contained in the token request are invalid.'
+
+    def test_return_403_with_phone_number_not_verified_user(self, endpoint):
+        response = requests.post(
+            url=endpoint + '/token',
+            headers={
+                'Content-Type': 'application/x-www-form-urlencoded',
+                'Authorization': 'Basic '+authlete_test_sdk.get_token_for_client_authentication()
+            },
+            data='grant_type=authorization_code&code='+self.authorization_code_by_phone_number_not_verified_user+'&redirect_uri=http://localhost&code_verifier='+self.code_verifier
+        )
+
+        assert response.status_code == 403
+        error = response.json()
+        assert error['error_message'] == 'phone_number must be verified'