CDK Pipeline manual approval step with SNS notification

Background

At the enterprise where I work as a Cloud Consultant at the moment, we have implemented extra security on our pipelines. Our code changes are following the DTAP model. Basically the code is first deployed to DevTest, then to UAT and then to Production. CDK Pipelines is the orchestrator here. Between our UAT and Production accounts, a manual approval is implemented, so all code changes need to be approved before going in to production.

Problem

With CDK Pipelines this is a simple step to add to your Production stage. You can use the pre within StageDeployment functionality to add a manual approval.

# retrieve accounts with variables from cdk.json
accounts= self.node.try_get_context("accounts")

for account in accounts:
    pipeline.add_stage(
        Application(
            self,
            f"{account}".lower(),
            env={"account": accounts[account]["account_id"], "region": accounts[account]["region"]},
            # Pass on account name (test, uat, prod), so we can use it in stacks
            account_name=account,
        ),
        # Add manual approval between uat and prd stage
        pre=None if account != "prd" else [pipelines.ManualApprovalStep("PromoteToProd")],
    )

Solution

But we are missing a piece here.

*So what is better to have than a manual approval within your CDK pipeline? *

A manual approval with SNS Topic for sending notifications on pending approvals of course.

Well, exactly that is a problem with CDK pipelines at the moment. When using CDK version 2 and the new modern API for pipeline, using the CodePipelineEngine, it isn't possible out of the box to have a manual approval with SNS topic configuration.

The ManualApprovalStep only supports adding a comment to the manual approval. This is strange, because the normal CodePipeline Actions construct does support adding a SNS notification topic.

Luckily CDK Pipelines does support an escape hack in the form of arbitrary CodePipeline actions.

Arbitrary CodePipeline Action

To implement an arbitrary CodePipeline action because the CDK Pipeline doesn’t support the manual approval step with SNS notifications you need to define your own step class that extends Step and implements ICodePipelineActionFactory.

Here is the example to for the manual approval class:

@jsii.implements(pipelines.ICodePipelineActionFactory)
class ManualApprovalWithSNSStep(pipelines.Step):
    """
    Create an Arbitrary CodePipeline step to enable SNS with manual approval
    https://docs.aws.amazon.com/cdk/api/v2/python/aws_cdk.pipelines/README.html#arbitrary-codepipeline-actions
    """

    def __init__(self, id_, topic: aws_sns.ITopic):
        super().__init__(id_)

        self.topic = topic

    @jsii.member(jsii_name="produceAction")
    def produce_action(
        self,
        stage: aws_codepipeline.IStage,
        options: pipelines.ProduceActionOptions,
    ) -> pipelines.CodePipelineActionFactoryResult:
        stage.add_action(
            aws_codepipeline_actions.ManualApprovalAction(
                action_name=options.action_name,
                additional_information="please approve",
                run_order=options.run_order,
                notification_topic=self.topic,
            )
        )

        return pipelines.CodePipelineActionFactoryResult(run_orders_consumed=1)

Let's break it down in chunks. So as described we need to implement the ICodePipelineActionFactory from CDK pipelines. It "extends" the pipelines.Step functionality.
In the __init__ function we also expect a SNS topic besides the id. This self.topic is later used in the Action.
Because we create a subclass of the CodePipeline Step, actions are defined in the produce_action function.
The real construct we will be using here, is an aws_codepipeline_actions.ManualApprovalAction. This is the standard CodePipeline construct with the SNS topic configuration available. Here we will link the given topic to the property notification_topic.

In our pipeline, the ARN of the monitoring SNS topic, is available via SSM parameter store. To actually use this new Class which is using the CodePipeline Actions construct, call it in the pre step of the Production stage via:

pre=None
if account != "prd"
else [
    ManualApprovalWithSNSStep(
        "PromoteToProd",
        topic=aws_sns.Topic.from_topic_arn(
            self,
            "MonitoringTopic",
            topic_arn=aws_ssm.StringParameter.from_string_parameter_name(
                self, "SNSMonitoringTopicArn", "/cdp/sns/monitoring_arn"
            ).string_value,
        ),
    )
],

This will result in CloudFormation code:

{
    "ActionTypeId": {
        "Category": "Approval",
        "Owner": "AWS",
        "Provider": "Manual",
        "Version": "1"
    },
    "Configuration": {
        "NotificationArn": {
        "Ref": "SNSMonitoringTopicArnParameter"
        },
        "CustomData": "please approve"
    },
    "Name": "PromoteToProd",
    "RoleArn": {
        "Fn::GetAtt": [
        "PromoteToProdCodePipelineActionRoleEA4779BC",
        "Arn"
        ]
    },
    "RunOrder": 1
},

Conclusion

In this blog I've shown how to use an arbitrary CodePipeline Step with CDK Pipelines (in Python). As the generated example in the CDK documentation took a while to understand and use for ourselves, I've created a real world example for you to use.