1. In your IDE, open the handlers.py file, located in the src folder.
2. Replace the entire contents of the handlers.py file with the following code.
Example handlers.py
import logging
from typing import Any, MutableMapping, Optional import botocore
from cloudformation_cli_python_lib import ( BaseHookHandlerRequest,
HandlerErrorCode, Hook,
HookInvocationPoint, OperationStatus, ProgressEvent, SessionProxy, exceptions, )
from .models import HookHandlerRequest, TypeConfigurationModel
# Use this logger to forward log messages to CloudWatch Logs.
LOG = logging.getLogger(__name__)
TYPE_NAME = "MyCompany::Testing::MyTestHook"
LOG.setLevel(logging.INFO)
hook = Hook(TYPE_NAME, TypeConfigurationModel) test_entrypoint = hook.test_entrypoint
def _validate_s3_bucket_encryption(
bucket: MutableMapping[str, Any], required_encryption_algorithm: str ) -> ProgressEvent:
status = None
Walkthrough: Python message = ""
error_code = None if bucket:
bucket_name = bucket.get("BucketName")
bucket_encryption = bucket.get("BucketEncryption") if bucket_encryption:
server_side_encryption_rules = bucket_encryption.get(
"ServerSideEncryptionConfiguration" for AWS::S3::Bucket with name: {bucket_name}"
else:
status = OperationStatus.FAILED
message = f"SSE Encryption Algorithm is incorrect for bucket with name: {bucket_name}"
else:
if status == OperationStatus.FAILED:
error_code = HandlerErrorCode.NonCompliant
return ProgressEvent(status=status, message=message, errorCode=error_code)
Walkthrough: Python
def _validate_sqs_queue_encryption(queue: MutableMapping[str, Any]) -> ProgressEvent:
if not queue:
queue_name = queue.get("QueueName") kms_key_id = queue.get(
"KmsMasterKeyId"
) # "KmsMasterKeyId" is name of the property for an AWS::SQS::Queue if not kms_key_id:
message=f"Successfully invoked PreCreateHookHandler for targetAWS::SQS::Queue with name: {queue_name}",
)
@hook.handler(HookInvocationPoint.CREATE_PRE_PROVISION) def pre_create_handler(
session: Optional[SessionProxy], request: HookHandlerRequest,
callback_context: MutableMapping[str, Any], type_configuration: TypeConfigurationModel, ) -> ProgressEvent:
target_name = request.hookContext.targetName if "AWS::S3::Bucket" == target_name:
return _validate_s3_bucket_encryption(
request.hookContext.targetModel.get("resourceProperties"), type_configuration.encryptionAlgorithm,
)
elif "AWS::SQS::Queue" == target_name:
return _validate_sqs_queue_encryption(
request.hookContext.targetModel.get("resourceProperties") )
else:
raise exceptions.InvalidRequest(f"Unknown target type: {target_name}")
def _validate_bucket_encryption_rules_not_updated(
resource_properties, previous_resource_properties ) -> ProgressEvent:
bucket_encryption_configs = resource_properties.get("BucketEncryption", {}).get(
"ServerSideEncryptionConfiguration", []
)
previous_bucket_encryption_configs = previous_resource_properties.get(
"BucketEncryption", {}
).get("ServerSideEncryptionConfiguration", [])
if len(bucket_encryption_configs) != len(previous_bucket_encryption_configs):
return ProgressEvent(
status=OperationStatus.FAILED,
message=f"Current number of bucket encryption configs does not match previous. Current has {str(len(bucket_encryption_configs))} configs while previously there were {str(len(previous_bucket_encryption_configs))} configs",
errorCode=HandlerErrorCode.NonCompliant, )
Walkthrough: Python
for i in range(len(bucket_encryption_configs)):
current_encryption_algorithm = (
if current_encryption_algorithm != previous_encryption_algorithm:
return ProgressEvent(
status=OperationStatus.FAILED,
message=f"Bucket Encryption algorithm can not be changed once set. The encryption algorithm was changed to {current_encryption_algorithm} from {previous_encryption_algorithm}.",
errorCode=HandlerErrorCode.NonCompliant, )
return ProgressEvent(
status=OperationStatus.SUCCESS,
message="Successfully invoked PreUpdateHookHandler for target:
AWS::SQS::Queue", )
def _validate_queue_encryption_not_disabled(
resource_properties, previous_resource_properties ) -> ProgressEvent:
if previous_resource_properties.get(
"KmsMasterKeyId"
) and not resource_properties.get("KmsMasterKeyId"):
return ProgressEvent(
return ProgressEvent(status=OperationStatus.SUCCESS)
@hook.handler(HookInvocationPoint.UPDATE_PRE_PROVISION) def pre_update_handler(
session: Optional[SessionProxy], request: BaseHookHandlerRequest,
callback_context: MutableMapping[str, Any], type_configuration: MutableMapping[str, Any], ) -> ProgressEvent:
target_name = request.hookContext.targetName if "AWS::S3::Bucket" == target_name:
resource_properties =
request.hookContext.targetModel.get("resourceProperties")
previous_resource_properties = request.hookContext.targetModel.get(
"previousResourceProperties"
)
return _validate_bucket_encryption_rules_not_updated(
resource_properties, previous_resource_properties )
elif "AWS::SQS::Queue" == target_name:
resource_properties =
request.hookContext.targetModel.get("resourceProperties")
previous_resource_properties = request.hookContext.targetModel.get(
Walkthrough: Python "previousResourceProperties"
)
return _validate_queue_encryption_not_disabled(
resource_properties, previous_resource_properties )
else:
raise exceptions.InvalidRequest(f"Unknown target type: {target_name}")
Registering hooks
Next, you must package and register your hook for use in your AWS account.
Packaging a hook
Build and package the hook to deploy it to the registry.
Registering a hook
Register your hook with CloudFormation, so it's available for use in the AWS CloudFormation registry.
To register a hook
1. (Optional) Configure your default AWS Region name to us-west-2, by submitting the configure operation.
aws configure
AWS Access Key ID [None]: <Your Access Key ID>
AWS Secret Access Key [None]: <Your Secret Key>
Default region name [None]: us-west-2 Default output format [None]: json
2. (Optional) The following command builds and packages your hook project without registering it.
$ cfn submit --dry-run
3. Register your hook by using the CloudFormation CLI the section called “submit” (p. 170) operation.
cfn submit --set-default
The command returns the following command.
{‘ProgressStatus’: ‘COMPLETE’}
Results: You've successfully registered your hook.
Verifying hooks are accessible in your account
Verify that you are subscribed to the hook in your AWS account
1. To verify your hook, use the list-types command to list your newly registered hook and return a summary description of it.
aws cloudformation list-types
Walkthrough: Python
The command returns the following output and will also show you publicly available hooks you can subscribe to.
{
"TypeSummaries": [ {
"Type": "HOOK",
"TypeName": "MyCompany::Testing::MyTestHook", "DefaultVersionId": "00000001",
"TypeArn": "arn:aws:cloudformation:us-west-2:ACCOUNT_ID/type/hook/
MyCompany-Testing-MyTestHook",
"LastUpdated": "2021-08-04T23:00:03.058000+00:00",
"Description": "Verifies S3 bucket and SQS queues properties before creating or updating"
} ] }
2. Retrieve the TypeArn from the list-type output for your hook and save it.
export HOOK_TYPE_ARN=arn:aws:cloudformation:us-west-2:ACCOUNT_ID/type/hook/MyCompany-Testing-MyTestHook
To learn how to publish hooks for public use, see Publishing hooks for public use.
Activate hooks
After you've developed and registered your hook, you can activate your hook in your AWS account by publishing it to the registry.
• To activate a hook in your account, use the SetTypeConfiguration operation. This operation enables the hook’s properties that are defined in the hook’s schema properties section. In the following example, the limitSize property is set to 1 in the configuration.
Note
By enabling hooks in your account, you are authorizing a hook to use defined permissions from your AWS account. CloudFormation removes non-required permissions before passing your permissions to the hook. CloudFormation recommends customers or hook users to review the hook permissions and be aware of what permissions the hooks are allowed to before enabling hooks in your account.Specify the configuration data for your registered hook extension in the same account and AWS Region.
aws cloudformation --region us-west-2 set-type-configuration \
--configuration "{\"CloudFormationConfiguration\":{\"HookConfiguration\":
{\"TargetStacks\":\"ALL\",\"FailureMode\":\"FAIL\",\"Properties\":{\"limitSize\":
\"1\",\"encryptionAlgorithm\": \"aws:kms\"}}}}" \ --type-arn $HOOK_TYPE_ARN
Important
To enable your hook to proactively inspect the configuration of your stack, you must set the TargetStacks to ALL in the HookConfiguration section, after the hook has been registered and activated in your account.
Walkthrough: Python
Accessing AWS APIs in handlers
If your hooks uses an AWS API in any of its handlers, create an IAM execution role that includes the necessary permissions and provision that execution role in your AWS account. CloudFormation assumes that role to provide your hook with the specified credentials during a hook invocation. For more information, see AWS Lambda execution role.
The the section called “generate” (p. 167) operation automatically generates an execution role template, hook-role.yaml. The hook-role.yaml template is based on the permissions specified for each handler in the handlers' section of the hook schema. The following template uses the AWS Identity and Access Management (IAM) permissions s3:ListAllMyBuckets and sqs:ListQueues. For more information about IAM user policies, see Using IAM user policies.
For more information, see Accessing AWS APIs from a resource type.
hook-role.yaml template
AWSTemplateFormatVersion: 2010-09-09 Description: >
This CloudFormation template creates a role assumed by CloudFormation during Hook operations on behalf of the customer.
Resources:
ExecutionRole:
Type: 'AWS::IAM::Role' Properties:
MaxSessionDuration: 8400 AssumeRolePolicyDocument:
Version: 2012-10-17
Walkthrough: Python
Testing hooks in your AWS account
Now that you've coded your handler functions that correspond to an invocation point, it's time to test your hook on a CloudFormation stack.
In Implementing hook handlers, the hook failure mode was set to FAIL if the CloudFormation template didn't provision an S3 bucket with the following:
• The Amazon S3 bucket encryption is set.
• The Amazon S3 bucket key is enabled for the bucket.
• The encryption algorithm set for the Amazon S3 bucket is the correct algorithm required.
• The AWS Key Management Service key ID is set.
In the following example, create a template called my-failed-bucket-stack with a stack name of my-hook-stack that fails the stack configuration and stops before the resource provisions.
Testing hooks by provisioning a stack
To provision a stack (example 1) Provision a non-compliant stack
1. Author a template that specifies an S3 bucket. For example, my-failed-bucket-stack.yml.
AWSTemplateFormatVersion: 2010-09-09 Resources:
S3Bucket:
Type: 'AWS::S3::Bucket' Properties: {}
2. Create a stack, and specify your template in the AWS Command Line Interface (AWS CLI). In the following example, specify the stack name as hook-stack and the template name as my-failed-bucket-stack.yml.
aws cloudformation create-stack \ --stack-name my-hook-stack \
--template-body file://my-failed-bucket-stack.yml
3. (Optional) View your stack progress by specifying your stack name. In the following example, specify the stack name my-hook-stack.
aws cloudformation describe-stack-events \ --stack-name my-hook-stack
Use the describe-stack-events operation to see the hook failure while creating the bucket. The following is an example output of the command.
{
"StackEvents": [ ...
{
"StackId": "arn:aws:cloudformation:us-west-2:ACCOUNT_ID:stack/my-hook-stack/2c693970-f57e-11eb-a0fb-061a2a83f0b9",
"EventId": "S3Bucket-CREATE_FAILED-2021-08-04T23:47:03.305Z", "StackName": "my-hook-stack",
"LogicalResourceId": "S3Bucket", "PhysicalResourceId": "",
Walkthrough: Python "ResourceType": "AWS::S3::Bucket",
"Timestamp": "2021-08-04T23:47:03.305000+00:00", "ResourceStatus": "CREATE_FAILED",
"ResourceStatusReason": "The following hook(s) failed:
[MyCompany::Testing::MyTestHook]", "ResourceProperties": "{}",
"ClientRequestToken": "Console-CreateStack-abe71ac2-ade4-a762-0499-8d34d91d6a92"
}, ...
] }
Results: The hook invocation failed the stack configuration and stopped the resource from provisioning.
Use a CloudFormation template to pass hook validation
1. To create a stack and pass the hook validation, update the template so that your resource uses an encrypted S3 bucket. This example uses the template my-encrypted-bucket-stack.yml.
AWSTemplateFormatVersion: 2010-09-09 Description: |
This CloudFormation template provisions an encrypted S3 Bucket Resources:
EncryptedS3Bucket:
Type: 'AWS::S3::Bucket' Properties:
BucketName: !Sub 'encryptedbucket-${AWS::Region}-${AWS::AccountId}' BucketEncryption:
ServerSideEncryptionConfiguration:
- ServerSideEncryptionByDefault:
SSEAlgorithm: 'aws:kms'
KMSMasterKeyID: !Ref EncryptionKey BucketKeyEnabled: true
EncryptionKey:
Type: 'AWS::KMS::Key' DeletionPolicy: Retain Properties:
Description: KMS key used to encrypt the resource type artifacts EnableKeyRotation: true
KeyPolicy:
Version: 2012-10-17 Statement:
- Sid: Enable full access for owning account Effect: Allow
Principal:
AWS: !Ref 'AWS::AccountId' Action: 'kms:*'
Resource: '*' Outputs:
EncryptedBucketName:
Value: !Ref EncryptedS3Bucket
Note
Hook won't be invoked for skipped resources.2. Create a stack and specify your template. In this example, the stack name is my-encrypted-bucket-stack.
aws cloudformation create-stack \
--stack-name my-encrypted-bucket-stack \
Walkthrough: Python
--template-body file://my-encrypted-bucket-stack.yml \ 3. (Optional) View your stack progress by specifying the stack name.
aws cloudformation describe-stack-events \ --stack-name my-encrypted-bucket-stack
Use the describe-stack-events command to view the response. The following is an example of the describe-stack-events command.
{
"StackEvents": [ ...
{
"StackId": "arn:aws:cloudformation:us-west-2:ACCOUNT_ID:stack/my-encrypted-bucket-stack/82a97150-f57a-11eb-8eb2-06a6bdcc7779",
"EventId": "EncryptedS3Bucket-CREATE_COMPLETE-2021-08-04T23:23:20.973Z", "StackName": "my-encrypted-bucket-stack",
"LogicalResourceId": "EncryptedS3Bucket",
"PhysicalResourceId": "encryptedbucket-us-west-2-ACCOUNT_ID", "ResourceType": "AWS::S3::Bucket",
"Timestamp": "2021-08-04T23:23:20.973000+00:00", "ResourceStatus": "CREATE_COMPLETE",
"ResourceProperties":
"{\"BucketName\":\"encryptedbucket-us-west-2-071617338693\",\"BucketEncryption\":{\"ServerSideEncryptionConfiguration\":
[{\"BucketKeyEnabled\":\"true\",\"ServerSideEncryptionByDefault\":{\"SSEAlgorithm\":
\"aws:kms\",\"KMSMasterKeyID\":\"ENCRYPTION_KEY_ARN\"}}]}}",
"ClientRequestToken": "Console-CreateStack-39df35ac-ca00-b7f6-5661-4e917478d075"
}, {
"StackId": "arn:aws:cloudformation:us-west-2:ACCOUNT_ID:stack/my-encrypted-bucket-stack/82a97150-f57a-11eb-8eb2-06a6bdcc7779",
"EventId": "EncryptedS3Bucket-CREATE_IN_PROGRESS-2021-08-04T23:22:59.410Z", "StackName": "my-encrypted-bucket-stack",
"LogicalResourceId": "EncryptedS3Bucket",
"PhysicalResourceId": "encryptedbucket-us-west-2-ACCOUNT_ID", "ResourceType": "AWS::S3::Bucket",
"Timestamp": "2021-08-04T23:22:59.410000+00:00", "ResourceStatus": "CREATE_IN_PROGRESS",
"ResourceStatusReason": "Resource creation Initiated", "ResourceProperties":
"{\"BucketName\":\"encryptedbucket-us-west-2-071617338693\",\"BucketEncryption\":{\"ServerSideEncryptionConfiguration\":
[{\"BucketKeyEnabled\":\"true\",\"ServerSideEncryptionByDefault\":{\"SSEAlgorithm\":
\"aws:kms\",\"KMSMasterKeyID\":\"ENCRYPTION_KEY_ARN\"}}]}}",
"ClientRequestToken": "Console-CreateStack-39df35ac-ca00-b7f6-5661-4e917478d075"
}, {
"StackId": "arn:aws:cloudformation:us-west-2:ACCOUNT_ID:stack/my-encrypted-bucket-stack/82a97150-f57a-11eb-8eb2-06a6bdcc7779",
"EventId": "EncryptedS3Bucket-6516081f-c1f2-4bfe-a0f0-cefa28679994", "StackName": "my-encrypted-bucket-stack",
"LogicalResourceId": "EncryptedS3Bucket", "PhysicalResourceId": "",
"ResourceType": "AWS::S3::Bucket",
"Timestamp": "2021-08-04T23:22:58.349000+00:00", "ResourceStatus": "CREATE_IN_PROGRESS",
"ResourceStatusReason": "Hook invocations complete. Resource creation initiated",
"ClientRequestToken": "Console-CreateStack-39df35ac-ca00-b7f6-5661-4e917478d075"
},
Walkthrough: Python ...
] }
Results: CloudFormation successfully created the stack by verifying that the AWS::S3::Bucket resource contained server-side encryption before provisioning the resource.