背景

自社で検証環境でCognitoを作成し、動作確認を行う場合があります。
今までAWSコンソール上でユーザプールを作成し、ユーザを作成する流れになっていました。検証した後、手動で削除する必要があります。

今回、CloudFormationで作成してみましたので、検証後、スタックを削除することで、リソースが削除されます。

テンプレート

TestApiUserPool_Python.yaml

AWSTemplateFormatVersion: "2010-09-09"
Description: "Example template including Cognito Identity Pool and User Pool."
Parameters:
  ContractId:
    Type: String
    Description: >-
      Your Contract Id. For example:cpi-dev3
    ConstraintDescription: Contract Id is required. 
    Default: ''

Resources:
  # シークレットの作成
  SecretTestUser:
    Type: AWS::SecretsManager::Secret
    Properties:
      Name:
        !Join
          - ''
          - - 'sec-'
            - !Ref ContractId
            - '-CognitoUserForTestUser-1'
      Description: "This secret has a dynamically generated secret password."    
      GenerateSecretString:
        # GenerateStringKey: password
        PasswordLength: 8
        # #$%&'()*+,-./:;<=>?@[\]^_`{|}~
        ExcludeCharacters: '#$%&()*+,-./:;<=>?@[\]^`{|}~"'
        # SecretStringTemplate: '{}'
      Tags:
        -
          Key: AppName
          Value: TestUserApi

  # ユーザープールの作成
  UserPool:
    Type: "AWS::Cognito::UserPool"
    Properties:
      Policies:
        PasswordPolicy:
          MinimumLength: 8
          RequireUppercase: true
          RequireLowercase: true
          RequireNumbers: true
          RequireSymbols: true
      UserPoolName:
        !Join
          - ''
          - - 'cup-'
            - !Ref ContractId
            - '-CognitoUserForTestUser-1'
      MfaConfiguration: 'OFF'
      AdminCreateUserConfig:
        AllowAdminCreateUserOnly: true
        UnusedAccountValidityDays: 7

  # ユーザープールにアプリクライアントを作成
  UserPoolClient:
    Type: "AWS::Cognito::UserPoolClient"
    Properties:
      ExplicitAuthFlows: 
        - ALLOW_ADMIN_USER_PASSWORD_AUTH
        - ALLOW_CUSTOM_AUTH 
        - ALLOW_USER_PASSWORD_AUTH
        - ALLOW_USER_SRP_AUTH
        - ALLOW_REFRESH_TOKEN_AUTH
      UserPoolId:
        Ref: UserPool
      ClientName:
        !Join
          - ''
          - - 'cupc-'
            - !Ref ContractId
            - '-CognitoUserForTestUser-1'
      RefreshTokenValidity: 30

  # ユーザーを追加
  AdminCreateUser:
    Type: Custom::CustomResource
    Properties:
      ServiceToken: !GetAtt CognitoAdminCreateUserFunction.Arn
      UserPoolId:
        Ref: UserPool
      UserName: TestUser
      Password: !Sub "{{resolve:secretsmanager:${SecretTestUser}:SecretString::}}"

  # ユーザープールにユーザーを作成するLambda関数
  CognitoAdminCreateUserFunction:
    Type: AWS::Lambda::Function
    Properties:
      FunctionName:
        !Join
          - ''
          - - 'lmd-'
            - !Ref ContractId
            - '-CognitoUserForTestUser-1'
      Handler: index.handler
      Environment: 
        Variables:
          UserPoolClientId : !Ref UserPoolClient
          SecretName: !Ref SecretTestUser
      Role: !GetAtt CognitoFunctionExecutionRole.Arn
      Code:
        ZipFile: !Sub |
          import cfnresponse
          import boto3
          import os

          client_id = os.getenv('UserPoolClientId')
          print(f'client_id: {client_id}')

          secret_name = os.getenv('SecretName')
          print(f'secret_name: {secret_name}')

          cognito_idp = boto3.client('cognito-idp')

          def handler(event, context):
            print(event['RequestType'])
            # スタック削除時にも実行されるので、処理せずに終了させる
            if event['RequestType'] == 'Delete':
              cfnresponse.send(event, context, cfnresponse.SUCCESS, {})
              return

            # UserPoolIDを取得する
            user_pool_id = event['ResourceProperties']['UserPoolId']
            print(f'user_pool_id: {user_pool_id}')

            # ユーザー名、パスワードを取得する
            username = event['ResourceProperties']['UserName']
            # password = event['ResourceProperties']['Password']
            password = get_secret(secret_name)

            print(f'password: {password}')

            # ユーザーを作成する
            response_data = {}
            try:
              response_data = cognito_idp.admin_create_user(
                UserPoolId=user_pool_id,
                Username=username,
                TemporaryPassword=password,
                MessageAction='SUPPRESS'
              )

              print('admin_create_user')
              print(response_data)

              # ログインを試みる。(パスワードの変更を要求される。)
              response_data = cognito_idp.admin_initiate_auth(
                  UserPoolId=user_pool_id,
                  ClientId=client_id,
                  AuthFlow='ADMIN_NO_SRP_AUTH',
                  AuthParameters={'USERNAME': username, 'PASSWORD': password},
              )

              print('admin_initiate_auth')
              print(response_data)

              session = response_data['Session']

              # パスワードを変更する。
              response_data = cognito_idp.admin_respond_to_auth_challenge(
                  UserPoolId=user_pool_id,
                  ClientId=client_id,
                  ChallengeName='NEW_PASSWORD_REQUIRED',
                  ChallengeResponses={'USERNAME': username, 'NEW_PASSWORD': password},
                  Session=session
              )          

              print('admin_respond_to_auth_challenge')
              print(response_data)

              # ログインを試みる。(パスワードの変更を要求される。)
              response_data = cognito_idp.admin_initiate_auth(
                  UserPoolId=user_pool_id,
                  ClientId=client_id,
                  AuthFlow='ADMIN_NO_SRP_AUTH',
                  AuthParameters={'USERNAME': username, 'PASSWORD': password},
              )
              print('admin_initiate_auth again')
              print(response_data)

            except Exception as e:
              print("error: " + str(e))
              response_data = {'error': str(e)}
              cfnresponse.send(event, context, cfnresponse.FAILED,
                                    {'Response': 'FAILED'})
              return

            # print(response_data)

            cfnresponse.send(event, context, cfnresponse.SUCCESS,
                                    {'Response': 'Success'})

          def get_secret(secret_name):
              """SecretsManagerから情報を取得する"""
              try:
                  rt = None
                  region_name = "ap-northeast-1"
                  # Create a Secrets Manager client
                  client = boto3.client(
                      service_name='secretsmanager',
                      region_name=region_name
                  )
                  get_secret_value_response = client.get_secret_value(
                      SecretId=secret_name
                  )
                  if 'SecretString' in get_secret_value_response:
                      rt = get_secret_value_response['SecretString']
              except Exception:
                  print ('[ConfiRule_Notification] get secret exception error.')
                  rt=None
              return rt

      Runtime: python3.9

  # Lambda関数実行用のロール
  CognitoFunctionExecutionRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: 
        !Join
          - ''
          - - 'irl-'
            - !Ref ContractId
            - '-CognitoFunctionExecutionRole-1'
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
        - Effect: Allow
          Principal:
            Service:
            - lambda.amazonaws.com
          Action:
          - sts:AssumeRole
      Path: "/"
      Policies:
      - PolicyName: root
        PolicyDocument:
          Version: '2012-10-17'
          Statement:
          - Effect: Allow
            Action:
              - logs:CreateLogGroup
              - logs:CreateLogStream
              - logs:PutLogEvents
            Resource: "arn:aws:logs:*:*:*"
          # Cognitoの操作権限を付与する
          - Effect: Allow
            Action:
              - cognito-idp:*
            Resource: "arn:aws:cognito-idp:*:*:userpool/*"
          - Effect: Allow
            Action:
              - secretsmanager:GetSecretValue 
            Resource: "arn:aws:secretsmanager:*:*:*"

ハマったこと

「カスタムリソースは Secrets Manager に登録しているシークレットは動的参照できない」という点を知らないうちに何回やっても成功出来なかった。

解決方法:Lamdba内部で取得することで解決できました。(下記のコードを参照)

          def get_secret(secret_name):
              """SecretsManagerから情報を取得する"""
              try:
                  rt = None
                  region_name = "ap-northeast-1"
                  # Create a Secrets Manager client
                  client = boto3.client(
                      service_name='secretsmanager',
                      region_name=region_name
                  )
                  get_secret_value_response = client.get_secret_value(
                      SecretId=secret_name
                  )
                  if 'SecretString' in get_secret_value_response:
                      rt = get_secret_value_response['SecretString']
              except Exception:
                  print ('[ConfiRule_Notification] get secret exception error.')
                  rt=None
              return rt

注意点

動的参照の制限
CloudFormation テンプレートスタックから Secrets Manager に登録しているシークレットを取得するとき固有の制限ではなくて動的参照に共通している制限だけどいくつか制限があるので書いておく。

スタックテンプレートは動的参照を最大で60個までしか含めない
AWS::Include とか AWS::Serverless などのトランスフォームの場合、AWS CloudFormation はトランスフォームを呼び出す前には動的な参照を解決せず、動的参照のリテラル文字列をトランスフォームに渡す
カスタムリソースは Secrets Manager に登録しているシークレットは動的参照できない
CloudFormation は最終値としてバックスラッシュを含む動的な参照は解決できない

まとめ

今回、カスタムリソースを使ってCognitoのユーザを作成してみましたが、レスポンスの返却やSecretsManagerからパスワードの取得について理解が深まりました。

以上です。
お役に立てれば幸いです。

参照

https://ebc-2in2crc.hatenablog.jp/entry/2020/06/27/204227

https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/cognito-idp.html

Last modified: 2022-06-03

Author