CloudFormation中SecretsManager密钥传递至Firehose的方法及调试

1. Firehose AccessKey的正确配置代码 #

避免手动拼接Secrets Manager密钥名称（易出错），直接通过CloudFormation的resolve:secretsmanager动态引用密钥逻辑ID，同时确保Firehose的IAM角色具备读取密钥的权限。

正确配置示例 #

StreamNameXyzabc:
  Type: AWS::KinesisFirehose::DeliveryStream
  Properties:
    HttpEndpointDestinationConfiguration:
      RequestConfiguration:
        ContentEncoding: GZIP
        EndpointConfiguration:
          Name: xyz
          Url: 'some url'
          # 直接引用密钥逻辑ID，动态提取LicenseKey字段
          AccessKey: !Sub '{{resolve:secretsmanager:${LicenseKeySecret}:SecretString:LicenseKey}}'

必要的IAM权限配置 #

给Firehose关联的IAM角色添加读取Secrets Manager密钥的权限：

FirehoseExecutionRole:
  Type: AWS::IAM::Role
  Properties:
    AssumeRolePolicyDocument:
      Version: '2012-10-17'
      Statement:
        - Effect: Allow
          Principal:
            Service: firehose.amazonaws.com
          Action: sts:AssumeRole
    Policies:
      - PolicyName: FirehoseSecretAccessPolicy
        PolicyDocument:
          Version: '2012-10-17'
          Statement:
            - Effect: Allow
              Action: secretsmanager:GetSecretValue
              Resource: !Ref LicenseKeySecret

2. AccessKey调试方法 #

AWS的describe_delivery_stream API不会返回敏感的AccessKey字段，需通过以下方式调试：

• 测试投递验证
  在Firehose控制台发送测试数据，查看投递结果。若失败，检查CloudWatch中Firehose的日志，错误信息会直接提示密钥是否无效、权限是否不足等问题。

• Lambda读取密钥验证
  创建Lambda函数直接读取Secrets Manager中的密钥值，确认存储的LicenseKey是否正确（注意：生产环境不要打印敏感信息）：

import boto3
import logging
import json

logger = logging.getLogger()
logger.setLevel(logging.INFO)

def handler(event, context):
    secrets_client = boto3.client('secretsmanager')
    try:
        # 替换为你的密钥逻辑ID或ARN
        secret_response = secrets_client.get_secret_value(SecretId=event['SecretId'])
        secret_data = json.loads(secret_response['SecretString'])
        logger.info(f"存储的LicenseKey: {secret_data['LicenseKey']}")
        return {"status": "success", "license_key": secret_data['LicenseKey']}
    except Exception as e:
        logger.error(f"读取密钥失败: {str(e)}")
        raise e

给Lambda添加secretsmanager:GetSecretValue权限，执行后查看CloudWatch日志确认密钥内容。

• CloudTrail日志排查
  开启CloudTrail后，查看Firehose服务调用secretsmanager:GetSecretValue的日志，确认是否成功获取密钥；同时查看Firehose向Http端点发送请求的日志，检查请求中的密钥是否正确。

内容的提问来源于stack exchange，提问作者Himanshu Rai