使用CloudFormation模板通过SNS Topic发送邮件时遇验证错误求助

你在CloudFormation模板中使用了AWS::SNS::Publish作为资源类型，但这是错误的——AWS CloudFormation没有原生支持这个资源类型，这是导致验证错误的直接原因。SNS消息发布属于操作类API调用，而非基础设施资源，不能直接在Resources块中声明。

解决方案：用Lambda-backed自定义资源实现消息发布 #

步骤1：创建Lambda函数（用于发布SNS消息） #

编写Python Lambda函数，接收CloudFormation自定义资源事件，调用SNS Publish接口完成消息发送：

import boto3
import json

sns = boto3.client('sns')

def handler(event, context):
    topic_arn = event['ResourceProperties']['TopicArn']
    message = event['ResourceProperties']['Message']
    
    try:
        sns.publish(
            TopicArn=topic_arn,
            Message=message
        )
        return {
            'Status': 'SUCCESS',
            'PhysicalResourceId': context.log_stream_name,
            'Data': {'Message': 'Published successfully'}
        }
    except Exception as e:
        return {
            'Status': 'FAILED',
            'PhysicalResourceId': context.log_stream_name,
            'Reason': str(e)
        }

步骤2：修改CloudFormation模板 #

在模板中添加Lambda函数、IAM权限，以及自定义资源触发消息发布：

AWSTemplateFormatVersion: '2010-09-09'

Resources:
  MySNSTopic:
    Type: AWS::SNS::Topic
    Properties:
      TopicName: my-sns-topic

  MySNSSubscription1:
    Type: AWS::SNS::Subscription
    Properties:
      TopicArn: !Ref MySNSTopic
      Protocol: email
      Endpoint: email-address-1@example.com

  MySNSSubscription2:
    Type: AWS::SNS::Subscription
    Properties:
      TopicArn: !Ref MySNSTopic
      Protocol: email
      Endpoint: email-address-2@example.com

  # Lambda函数：处理SNS消息发布逻辑
  PublishMessageLambda:
    Type: AWS::Lambda::Function
    Properties:
      Runtime: python3.12
      Handler: index.handler
      Role: !GetAtt LambdaExecutionRole.Arn
      Code:
        ZipFile: |
          import boto3
          import json

          sns = boto3.client('sns')

          def handler(event, context):
              topic_arn = event['ResourceProperties']['TopicArn']
              message = event['ResourceProperties']['Message']
              
              try:
                  sns.publish(
                      TopicArn=topic_arn,
                      Message=message
                  )
                  return {
                      'Status': 'SUCCESS',
                      'PhysicalResourceId': context.log_stream_name,
                      'Data': {'Message': 'Published successfully'}
                  }
              except Exception as e:
                  return {
                      'Status': 'FAILED',
                      'PhysicalResourceId': context.log_stream_name,
                      'Reason': str(e)
                  }

  # Lambda执行权限：允许发布SNS消息及日志操作
  LambdaExecutionRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
      Policies:
        - PolicyName: SNSPublishAccess
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action: sns:Publish
                Resource: !Ref MySNSTopic
              - Effect: Allow
                Action:
                  - logs:CreateLogGroup
                  - logs:CreateLogStream
                  - logs:PutLogEvents
                Resource: arn:aws:logs:*:*:*

  # 自定义资源：触发Lambda发布消息
  PublishSNSMessage:
    Type: Custom::PublishSNSMessage
    Properties:
      ServiceToken: !GetAtt PublishMessageLambda.Arn
      TopicArn: !Ref MySNSTopic
      Message: "status check failed"

关键说明 #

• 自定义资源Custom::PublishSNSMessage会在模板部署（创建/更新）时调用关联的Lambda函数，执行消息发布操作。
• Lambda函数必须具备SNS:Publish权限，以及日志权限用于调试排查问题。

内容的提问来源于stack exchange，提问作者sumit malhotra