Manual Approval in AWS CodePipeline with Lambda and SNS

Arya P B

5 min read

Manual Approval in AWS CodePipeline with Lambda and SNS

Generating audio, please wait...

Introduction

Welcome to our guide on implementing a manual approval process in AWS CodePipeline using the powerful combination of Lambda functions and Simple Notification Service (SNS). In this blog post, we’ll walk you through the setup, allowing you to add a human touch to your automated deployment process.

Prerequisites

Before we dive in, make sure you have the following prerequisites:

AWS Account
AWS CodePipeline configured with at least one pipeline
Basic understanding of AWS Lambda and SNS

Overview

Let’s break down the high-level overview of the setup:

Detecting CodePipeline State Changes using EventBridge:

Utilize EventBridge to trigger actions based on changes in your CodePipeline’s execution state, specifically when it reaches a “SUCCEEDED” state.

Creating an SNS Topic for Sending Email Notifications:

Set up an SNS topic to streamline the process of sending email notifications.

Sending Email Notifications for Manual Approval using Lambda and SNS:

Walk through a Lambda function responsible for extracting pipeline names, constructing approval and rejection URLs, and seamlessly sending notifications through SNS.

Accepting or Rejecting the Manual Approval using Another Lambda Function:

Dive into a second Lambda function designed to take action based on the approval or rejection URLs. Learn how to extract pipeline names, action types, and trigger CodePipeline accordingly.

Create an SNS Topic

Let’s create an SNS topic to facilitate email notifications. Follow these steps:

Navigate to the SNS service in the AWS Management Console.
Create a new topic and note the ARN.

EventBridge Rule

Set up an EventBridge rule to trigger the Lambda function when the CodePipeline execution state changes to “SUCCEEDED.”

{
"source": ["aws.codepipeline"],
"detail-type": ["CodePipeline Pipeline Execution State Change"],
"detail": {
"state": ["SUCCEEDED"],
"pipeline": ["YourPipelineName"]
}
}

Lambda for Sending Email Notification

Explore the Lambda function responsible for sending email notifications. This includes extracting pipeline names, constructing URLs, and utilizing SNS for smooth communication.

import boto3
import json

def lambda_handler(event, context):
print(event)
sns_client = boto3.client('sns')
sns_topic_arn = " " # add arn of sns
pipeline_name = event['detail']['pipeline']

print("Pipeline Name:", pipeline_name)

# Include pipeline name in the URLs
url_base = " " //replae with acion lambda function url
approval_url = f"{url_base}?pipeline={pipeline_name}&action=approve"
rejection_url = f"{url_base}?pipeline={pipeline_name}&action=reject"

print("Approval URL:", approval_url)
print("Rejection URL:", rejection_url)

# Message content with approval and rejection URLs
message = f"CodePipeline name: {pipeline_name}\n\n"
message += f"Click the following URL to approve: {approval_url}\n"
message += f"Click the following URL to reject: {rejection_url}\n"

# Publish the message to SNS
sns_client.publish(
TopicArn=sns_topic_arn,
Message=message,
Subject=f"CodePipeline Approval/Rejection Options"
)

return {
'statusCode': 200,
'body': f'Successfully listed CodePipelines.',
}

Lambda for Taking Action

Delve into the second Lambda function designed to take action based on the approval or rejection URLs. Extract pipeline names, action types, and trigger CodePipeline accordingly.

import json
import boto3

def lambda_handler(event, context):
print("Event:", event)

client = boto3.client('codepipeline')
# Extract pipeline name from URL parameters
parameters = event['queryStringParameters']
pipeline_name = parameters.get('pipeline', None)
action = parameters.get('action', None)
pipelineName = pipeline_name
stage_name, action_name = get_manual_approval_details(pipelineName)
print(stage_name)
print(action_name)
stage_name = stage_name
action_name = action_name

token = get_approval_token(pipelineName, stage_name, action_name)
print(f"Approval Token: {token}")

approval_token = token

if pipeline_name:
# Trigger the CodePipeline
if action == 'approve':
# Provide approval result (Approved in this example)
response = client.put_approval_result(
pipelineName=pipeline_name,
stageName=stage_name,
actionName=action_name,
result={
'summary': 'Approval granted programmatically.',
'status': 'Approved'
},
token=approval_token
)

# Print the response for logging or debugging
print(response)

return {
'statusCode': 200,
'body': 'Approval result sent successfully.'
}

if action == 'reject':
# Provide rejection result
response = client.put_approval_result(
pipelineName=pipeline_name,
stageName=stage_name,
actionName=action_name,
result={
'summary': 'Rejection submitted programmatically.',
'status': 'Rejected'
},
token=approval_token
)

# Print the response for logging or debugging
print(response)

return {
'statusCode': 200,
'body': 'Rejection result sent successfully.'
}

return {
'statusCode': 200,
'body': json.dumps(f'Successfully triggered CodePipeline: {pipeline_name}'),
}
else:
return {
'statusCode': 400,
'body': json.dumps('Pipeline name not provided in URL parameters.'),
}


def get_approval_token(pipeline_name, stage_name, action_name):
# AWS CodePipeline client
client = boto3.client('codepipeline')

try:
# Get the current state of the pipeline
response = client.get_pipeline_state(name=pipeline_name)
print(response)
# Find the manual approval action in the response
for stage in response['stageStates']:
if stage['stageName'] == stage_name:
for action in stage['actionStates']:
if action['actionName'] == action_name:
# Extract and return the token for the manual approval action
return action['latestExecution']['token']
except Exception as e:
print(f"Error: {e}")

return None

def get_manual_approval_details(pipeline_name):
# AWS CodePipeline client
client = boto3.client('codepipeline')

try:
# Get details about the specified pipeline
response = client.get_pipeline(name=pipeline_name)
print(response)
manual_approval_details = response['pipeline']['stages'][1]
print(manual_approval_details)

stage_name = manual_approval_details['name']
action_name = manual_approval_details['actions'][0]['name']
return stage_name, action_name

except Exception as e:
print(f"Error: {e}")
return None

1_wkwdUSWcZRZ0jGpDI0_A5w (1).webp

Conclusion

In conclusion, by combining the power of AWS CodePipeline, Lambda functions, and SNS, you’ve successfully implemented a robust manual approval process in your deployment pipeline. This setup not only enhances the security and reliability of your deployment but also provides the flexibility to include human reviews where necessary.

Key Takeaways:

Seamless Manual Approval Setup: Easily incorporate manual approval steps into your CodePipeline for crucial decision-making points.
Lambda for Notification and Action: Leverage Lambda functions to send email notifications and take dynamic actions based on human decisions.
Efficient SNS Integration: Enhance communication with stakeholders through the use of SNS, ensuring timely and effective notifications.

Implement a manual approval process today using Lambda and SNS. Add crucial oversight to your automated deployments and empower your team with timely decision-making