Skip to content

Instantly share code, notes, and snippets.

@bendu
Created November 7, 2022 21:28
Show Gist options
  • Save bendu/05396a39d47b2646ff43752a07767ab9 to your computer and use it in GitHub Desktop.
Save bendu/05396a39d47b2646ff43752a07767ab9 to your computer and use it in GitHub Desktop.
How I get around CDK limitations for stack policy, termination protection, and rollback alarms
There are two files here: a python file to call CFN APIs and a TypeScript file to define a construct
========================================
import { CustomResource, Duration } from 'aws-cdk-lib';
import { PolicyStatement } from 'aws-cdk-lib/aws-iam';
import { Code, Runtime, SingletonFunction } from 'aws-cdk-lib/aws-lambda';
import { Construct } from 'constructs';
import * as fs from 'fs';
export interface StackConstructProps {
stackArn: string,
terminationProtection?: boolean,
stackPolicy?: string,
rollbackAlarmArn?: string,
monitoringMinutes?: number
}
export class StackConstruct extends Construct {
constructor(parent: Construct, id: string, props: StackConstructProps) {
super(parent, id);
const code = fs.readFileSync('custom_resource_lambdas/stack_construct.py', 'utf-8');
const lambda = new SingletonFunction(this, 'Provider', {
code: Code.fromInline(code),
runtime: Runtime.PYTHON_3_9,
handler: 'index.handler',
timeout: Duration.seconds(120),
uuid: '536642a2-1418-41c1-ab6c-4b181eb51188', // Magic guid to make this a singleton. Do not modify. Do not copy.
lambdaPurpose: 'Cfn stack configuration' // Do not modify. Do not copy.
});
lambda.addToRolePolicy(new PolicyStatement({
actions: [
'cloudformation:DescribeStacks',
'cloudformation:SetStackPolicy',
'cloudformation:UpdateStack',
'cloudformation:UpdateTerminationProtection',
],
resources: [props.stackArn + '/*']
}));
const customResourceToTriggerLambda = new CustomResource(this, 'SetStackConfig', {
serviceToken: lambda.functionArn,
resourceType: "Custom::StackConfiguration",
properties: props
});
}
}
========================================
import boto3
import botocore
import cfnresponse
import time
# Sets termination protection, stack policy, rollback config
def create_or_update(event, context):
props = event['ResourceProperties']
stack_arn = props['stackArn']
stack_name = stack_arn.split('/')[1]
termination_protection = props.get('terminationProtection')
stack_policy = props.get('stackPolicy')
rollback_alarm = props.get('rollbackAlarmArn')
monitoring_minutes = int(props.get('monitoringMinutes'))
if monitoring_minutes is None:
monitoring_minutes = 0
client = boto3.client('cloudformation')
# Only allow termination protection be True
if termination_protection and termination_protection.lower() == 'true':
client.update_termination_protection(EnableTerminationProtection=True, StackName=stack_name)
if stack_policy:
client.set_stack_policy(StackName=stack_name, StackPolicyBody=stack_policy)
if rollback_alarm:
stack_info = client.describe_stacks(StackName=stack_name)['Stacks'][0]
rollback_config = {
"RollbackTriggers": [{ "Arn": rollback_alarm, "Type": "AWS::CloudWatch::Alarm" }],
"MonitoringTimeInMinutes": monitoring_minutes
}
try:
client.update_stack(StackName=stack_name, UsePreviousTemplate=True, Parameters=stack_info['Parameters'],
Capabilities=stack_info['Capabilities'], RollbackConfiguration=rollback_config)
update_success = False
for i in range(3):
time.sleep(20)
status = client.describe_stacks(StackName=stack_name)['Stacks'][0]['StackStatus']
if status == 'UPDATE_COMPLETE':
update_success = True
break
if not update_success:
raise RuntimeError('Status did not reach update complete state')
except botocore.exceptions.ClientError as error:
# CFN throws an exception if there are no changes, such as when stack policy is updated but not rollback configuration
if error.response['Error']['Code'] == 'ValidationError' and 'No updates are to be performed' in error.response['Error']['Message']:
print('Response from CFN update stack indicates no stack updates to make')
else:
raise error
# Lambda entry
def handler(event, context):
print(event)
# we need to maintain the same physical resource id or CF thinks we've replaces
# tries to delete the old one, and our states get out of sync. If we don't
# return the value then it's defaulted to the current log stream, which rotates
resource_id = event.get('PhysicalResourceId')
try:
if event['RequestType'] != 'Delete':
create_or_update(event, context)
cfnresponse.send(event, context, cfnresponse.SUCCESS, {}, resource_id)
except Exception as e:
print(e)
cfnresponse.send(event, context, cfnresponse.FAILED, {}, resource_id)
========================================
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment