A basic cdk template for bootstrapping a serverless app to support an Amplify client.

amplify-base-cdk.py

from pathlib import Path

from aws_cdk import core, aws_s3 as s3, aws_cognito as cognito, aws_iam as iam, aws_lambda as lambda_, aws_apigateway as apigateway, aws_dynamodb as dynamodb




class CloudStack(core.Stack):

    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
        super().__init__(scope, id, **kwargs)

        # A basic bucket with all public acccess blocked.
        bucket = s3.Bucket(self, 
            "Bucket", 
            versioned=True,
            block_public_access=s3.BlockPublicAccess(block_public_acls=True, block_public_policy=True, ignore_public_acls=True, restrict_public_buckets=True))

        # Cognito - The CDK has some higher level constructs that can make life very easy.
        # However, those constructs are still in beta for the Cognito resources so the raw CloudFormation resources are used.

        # A User Pool set up to allow the email to be used as a username which requires the email to be verified.

        account_recovery_setting = cognito.CfnUserPool.AccountRecoverySettingProperty(recovery_mechanisms=[cognito.CfnUserPool.RecoveryOptionProperty(name='verified_email', priority=1)])

        user_pool = cognito.CfnUserPool(self, "UserPool", 
            username_attributes=['email'], 
            auto_verified_attributes=['email'],
            account_recovery_setting=account_recovery_setting)

        # A User Pool Client which represents any end user client such as iOS, Android, or a browser based app.
        # User Pools represents the actual user directory. This is WHO the user is (Authentication).

        user_pool_client = cognito.CfnUserPoolClient(
            self, 
            'UserPoolClient',
            user_pool_id=user_pool.ref,
            generate_secret=True)

        # Identity Pool set up with the userpool as an identity provider.
        # Identity Pools contain identities granted access to AWS resources. This allows access to API Gateway and S3.
        # The access is granted by the role defined below and represents WHAT the user is allowed to do (Authorization).
        user_pool_identity_provider = cognito.CfnIdentityPool.CognitoIdentityProviderProperty(
            client_id=user_pool_client.ref, 
            provider_name=user_pool.attr_provider_name)
        
        identity_pool = cognito.CfnIdentityPool(
            self, 
            "IdentityPool",
            allow_unauthenticated_identities=False,
            cognito_identity_providers=[user_pool_identity_provider])

        # The principal represents who is allowed to assume the rule which grants access to API Gateway and S3.
        # This requires that the user must come from a cognito user pool and must be authenticated.
        # No anonymous/unauthenticated access allowed.
        federated_principal = iam.FederatedPrincipal(
            'cognito-identity.amazonaws.com', 
            {'ForAnyValue:StringLike':{'cognito-identity.amazonaws.com:amr': 'authenticated'}}, 
            'sts:AssumeRoleWithWebIdentity')


        # Here the IAM role is defined but the actual permissions are added later after the resources are defined.
        auth_role = iam.Role(
            self, 
            'AuthRole',
            assumed_by=federated_principal)

        # This tells the Identity Pool that an authenticated user from our User Pool should be given the defined auth role
        role_mapping = cognito.CfnIdentityPoolRoleAttachment.RoleMappingProperty(
            type='Token', 
            ambiguous_role_resolution='AuthenticatedRole', 
            identity_provider=f'{user_pool.attr_provider_name}:{user_pool_client.ref}')

        # This actually assigns the mapping to the Identity Pool
        cognito.CfnIdentityPoolRoleAttachment(
            self, 
            'RoleAttachment', 
            identity_pool_id=identity_pool.ref, 
            role_mappings={'userpool': role_mapping}, 
            roles={'authenticated': auth_role.role_arn})

        # This is a very basic definition of a dynamo table using one of the higher level constructs provided by the CDK.
        # The table also adds a Global Secondary Index.
        table = dynamodb.Table(self, 'Table',
            partition_key=dynamodb.Attribute(name='pk', type=dynamodb.AttributeType.STRING),
            sort_key=dynamodb.Attribute(name='sk', type=dynamodb.AttributeType.STRING))

        table.add_global_secondary_index(partition_key=dynamodb.Attribute(name='sk', type=dynamodb.AttributeType.STRING),
            sort_key=dynamodb.Attribute(name='gsi1sk', type=dynamodb.AttributeType.STRING),
            index_name='gsi1')


        # Define a Lambda function including the details of where the Lambda Handler code is.
        parent_path = Path(__file__).parent.absolute()
        function = lambda_.Function(self, "Function", 
            code=lambda_.Code.from_asset(str(Path(parent_path,'FunctionDirectory'))),
            runtime=lambda_.Runtime.PYTHON_3_8,
            handler='handler.handler',
            environment={'TABLE': table.table_name})


        # Grant the Lambda function CRUD permissions on the table. The end user/client doesn't talk to the table directly, but Lambda does.
        # If Lambda needed to talk to s3, you could also add those permissions here.
        table.grant_read_write_data(function)
        
        # Define an API Gateway Lambda Proxy Integration
        function_integration = apigateway.LambdaIntegration(function)
        
        
        # Define the API
        api = apigateway.RestApi(self, 'API')

        # Define an endpoint on the api. This would be the equivalent of `/resource`
        resource = api.root.add_resource('resource')

        # Adds an HTTP GET method to the resource, adds IAM authorization, and tells API Gateway to route the request to the Lambda Function
        # Seeting IAM auth makes use of the Cognito resources defined above.
        resource.add_method('GET', function_integration, authorization_type=apigateway.AuthorizationType.IAM)

        # Now that all the resources are defined, grant permissions via the auth role. This gives permissions to the end user/client.

        # This gives the role access to all stages, endpoints, and methods on the API Gateway
        auth_role.attach_inline_policy(
            iam.Policy(
                self,
                'AllowAccess',
                statements=[
                    iam.PolicyStatement(
                        actions=['execute-api:Invoke'],
                        effect=iam.Effect.ALLOW,
                        resources=[api.arn_for_execute_api('*', '/*', '*')]
                    )
                ]
            )
        )

        # Grant read/write access to the bucket.
        bucket.grant_read_write(auth_role)