August 22, 2021

CDK Tricks

There are a few tricks I’ve learnt recently that I thought I would share with you, as I’ve found them really useful.

Constructs

Constructs are objects that can contain a set of other objects to define a standard set of components. You can use constructs inside of constructs or you can user lower level Cfn primitives.

You can find the AWS Documentation here:

To get started you inherit the Construct class, creating your new class. You can initiate the class with any variables you wish to pass in to make decisions on your components.

As part of the initiation, you call super() to initiate the parent class.

from aws_cdk import core as cdk

class MyConstruct(cdk.Construct):

    def __init__(self, scope: cdk.Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)
        
        # TODO: add your stack components here.

As with any other pythonic class (as I’m using python in my examples) you can have various method types and attributes. Remember to use appropriate decorators (@staticmethod, @classmethod, @property) when defining methods and properties.

Example:


	 @property
    def my_service_arn(self):
        """return the arn of the service"""
        return self.my_service.attr_arn

Using in your stack

Now that you have your new construct, that might be a reusable component, you can go ahead and use it in your main stack:

from mystack.my_construct import MyConstruct
...
    new_service = MyConstruct(self)
    
...
    my_var = new_service.my_service_arn

This is really interesting and I expect that a library such as https://constructs.dev/ will become a place to get a lot of standard constructs from ready to go, and your code will become less and less just tying these neat packages together.

At the moment it’s a bit of the wild-west as everyone matures and finds their feet, but with many similar repositories over the years, I’ve seen them mature and become places of high-quality modules.

AWS SDK (Call) Custom Resource

This one is pretty neat. It’s a custom resource, but you only need to supply enough to do the customer AWS call without all the boiler plate.

Let’s say you want to be able to get a SSM Parameter from another region to be able to use as an input into your stack.

You might create something like this:

ssm_param = AwsSdkCall(
    service='SSM',
    action='getParameter',
    parameters={'Name': parameter_name},
    region=region,
    physical_resource_id=PhysicalResourceId.of(parameter_name+'-'+region+'v5')
)

Where parameter___name is the name of the parameter that you wish to get and region is the region you wish to get the parameter from.

You can then get access the results as follows:

ssm_param.get_response_field("Parameter.Value")

This makes writing custom resources for small things that only require the AWS SDK really easy.

CDK Managed Custom Resources

Lastly, there is creating CDK Managed Custom Resources. Similar to the AWS SDK (Call) Custom Resource, CDK helps you create the custom resource with a lot of ready to go boiler plate, so that you don’t need know everything.

A good example is storing cognito secrets in secrets manager so that your application can retrieve them (using the reference to secrets manager as an environment variable in your application, maybe running on ECS).

This is setting up the role, policy, and lambda function:

        cr_role = iam.Role(self, 'customresourcerole',
            assumed_by=iam.ServicePrincipal('lambda.amazonaws.com')
        )

        cr_role.add_managed_policy(
            iam.ManagedPolicy.from_aws_managed_policy_name(
                'service-role/AWSLambdaBasicExecutionRole'
            )
        )

        cr_policy = iam.Policy(self, 'cr_policy',
            statements = [iam.PolicyStatement(
                actions=[
                    'secretsmanager:CreateSecret',
                    'secretsmanager:TagResource',
                    'secretsmanager:UpdateSecret',
                    'secretsmanager:DeleteSecret'
                ],
                resources=['*']
            ),
            iam.PolicyStatement(
                actions=[
                    'cognito-idp:DescribeUserPoolClient'
                ],
                resources=[userpool.user_pool_arn]
            )]
        )

        cr_role.attach_inline_policy(cr_policy)

        on_event = lambda_.Function(self, 'customresourcelambda',
            handler = 'index.on_event',
            runtime = lambda_.Runtime.PYTHON_3_8,
            code=lambda_.Code.from_inline(lambda_code.custom_resource),
            role=cr_role
        )

        cr_depends = cdk.ConcreteDependable()
        cr_depends.add(cr_policy)
        on_event.node.add_dependency(cr_depends)

        secret_provider = cr.Provider(self, "customresourceprovider",
            on_event_handler=on_event,
            log_retention=logs.RetentionDays.ONE_DAY
        )

Now we can use the custom resource to store references from our Cognito User Pool and App Client:


        secret = CustomResource(self, "secret",
            service_token=secret_provider.service_token,
            properties={
                'UserPoolId': userpool.user_pool_id,
                'AppClientId': app_client.user_pool_client_id
            }
        )

Then we can reference the secret name for the environment variable as follows:

secret.ref

To make sure you can read the secret from your app, you’ll need the ARN, and you would retrieve it as follows:

secret.get_att_string('ARN')

That’s pretty much it. Of course, you’ll need some code and you can find that in the next section.

Lambda Code

import os
import json
import hashlib
import boto3
from botocore.exceptions import ClientError

cogidp = boto3.client('cognito-idp')
secman = boto3.client('secretsmanager')

def get_client_secret(user_pool_id, app_client_id):
    try:
        pool_client_description = cogidp.describe_user_pool_client(
            UserPoolId=user_pool_id,
            ClientId=app_client_id
        )
    except ClientError as error:
        raise error
    else:
        return pool_client_description['UserPoolClient']['ClientSecret']

def get_secret(secret_id):
    try:
        get_secret_value_response = secman.get_secret_value(
            SecretId=secret_id
        )
    except ClientError as error:
        raise error
    else:
        if 'SecretString' in get_secret_value_response:
            secret = get_secret_value_response['SecretString']
            return json.loads(secret)


def on_event(event, context):
    print(event)
    request_type = event['RequestType']
    if request_type == 'Create':
        return on_create(event)
    if request_type == 'Update':
        return on_update(event)
    if request_type == 'Delete':
        return on_delete(event)
    raise Exception("Invalid request type: %s" % request_type)

def on_create(event):
    props = event["ResourceProperties"]
    print("create new resource with props %s" % props)

    physical_id = event['StackId'].split(':')[-1].split('/')[1] + "-" + event['LogicalResourceId']

    secret_message = json.dumps({
        'COGNITO_CLIENT_SECRET': get_client_secret(props['UserPoolId'], props['AppClientId']),
        'COGNITO_STATE': hashlib.sha512(os.urandom(128)).hexdigest(),
        'SECRET_KEY': hashlib.sha512(os.urandom(128)).hexdigest()
    })

    try:
        response = secman.create_secret(
            Name=physical_id,
            SecretString=secret_message
        )
    except ClientError as error:
        raise error
    else:
        print(response)
        return { 'PhysicalResourceId': physical_id, 'Data': response }

def on_update(event):
    physical_id = event["PhysicalResourceId"]
    props = event["ResourceProperties"]
    print("update resource %s with props %s" % (physical_id, props))

    original_secret_message = get_secret(physical_id)

    original_secret_message['COGNITO_CLIENT_SECRET'] = get_client_secret(props['UserPoolId'], props['AppClientId'])

    secret_message = json.dumps(original_secret_message)

    try:
        response = secman.update_secret(
            SecretId=physical_id,
            SecretString=secret_message
        )
    except ClientError as error:
        raise error
    else:
        print(response)
        return { 'PhysicalResourceId': physical_id, 'Data': response }

def on_delete(event):
    physical_id = event["PhysicalResourceId"]
    print("delete resource %s" % physical_id)
    response = secman.delete_secret(
        SecretId=physical_id,
        ForceDeleteWithoutRecovery=True
    )
    print(response)

Conclusion

This is just a couple of examples of some of the interesting tricks I’ve learnt with CDK recently. Let me know of your tricks with CDK.

© Greg Cockburn

Powered by Hugo & Kiss.