如何用CloudFormation在AWS SecretsManager现有Secret中增删键并保留原值

CloudFormation 方案（自定义资源实现） #

原生CloudFormation不支持对SecretsManager Secret的SecretString进行部分更新——因为它是声明式的，更新时必须提供完整的JSON字符串，否则会直接覆盖原有值。要实现保留现有键值并修改部分字段的需求，需要通过Lambda-backed 自定义资源来完成，具体逻辑如下：

1. 自定义资源触发Lambda函数
2. Lambda读取目标Secret的当前值
3. 对JSON对象进行修改（添加DBName空值或删除DBPort）
4. 将修改后的JSON写回Secret

完整CFN模板示例 #

AWSTemplateFormatVersion: '2010-09-09'
Resources:
  # 原有Secret资源
  MySecret:
    Type: "AWS::SecretsManager::Secret"
    Properties:
      Name: MyDatabaseSecret
      Description: Secret for Database Connection
      SecretString: !Sub |
        {
          "DBUserName": "",
          "DBPassword": "",
          "DBPort":""
        }

  # 处理Secret更新的Lambda函数
  SecretUpdateFunction:
    Type: AWS::Lambda::Function
    Properties:
      Handler: index.lambda_handler
      Runtime: python3.11
      Role: !GetAtt SecretUpdateFunctionRole.Arn
      Code:
        ZipFile: |
          import json
          import boto3
          import cfnresponse

          secretsmanager = boto3.client('secretsmanager')

          def lambda_handler(event, context):
              try:
                  secret_name = event['ResourceProperties']['SecretName']
                  action = event['ResourceProperties']['Action']

                  # 读取现有Secret值
                  response = secretsmanager.get_secret_value(SecretId=secret_name)
                  secret_dict = json.loads(response['SecretString'])

                  # 执行指定操作
                  if action == 'AddDBName':
                      secret_dict['DBName'] = ""
                  elif action == 'RemoveDBPort':
                      secret_dict.pop('DBPort', None)

                  # 写回修改后的Secret
                  secretsmanager.put_secret_value(
                      SecretId=secret_name,
                      SecretString=json.dumps(secret_dict)
                  )

                  cfnresponse.send(event, context, cfnresponse.SUCCESS, {})
              except Exception as e:
                  print(f"Error: {str(e)}")
                  cfnresponse.send(event, context, cfnresponse.FAILED, {'Error': str(e)})

  # Lambda执行角色
  SecretUpdateFunctionRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
      Policies:
        - PolicyName: SecretUpdatePolicy
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action:
                  - secretsmanager:GetSecretValue
                  - secretsmanager:PutSecretValue
                Resource: !Ref MySecret
              - Effect: Allow
                Action:
                  - logs:CreateLogGroup
                  - logs:CreateLogStream
                  - logs:PutLogEvents
                Resource: arn:aws:logs:*:*:*

  # 自定义资源，触发Lambda执行更新操作
  UpdateSecretResource:
    Type: Custom::UpdateSecret
    Properties:
      ServiceToken: !GetAtt SecretUpdateFunction.Arn
      SecretName: !Ref MySecret
      Action: AddDBName # 可选值：AddDBName / RemoveDBPort

使用说明：

• 首次部署模板会创建初始Secret
• 需要修改UpdateSecretResource的Action参数来切换操作（添加DBName或删除DBPort）
• 每次更新模板时，自定义资源会触发Lambda执行对应的修改逻辑，保留现有Secret的其他键值

AWS CDK Python 方案 #

CDK同样基于CloudFormation，核心逻辑还是通过自定义资源实现，以下是CDK Python代码示例：

from aws_cdk import (
    Stack,
    aws_secretsmanager as secretsmanager,
    aws_lambda as lambda_,
    aws_iam as iam,
    custom_resources as cr,
    CfnOutput,
)
from constructs import Construct
import json

class SecretUpdateStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # 创建初始Secret
        my_secret = secretsmanager.Secret(self, "MySecret",
            secret_name="MyDatabaseSecret",
            description="Secret for Database Connection",
            secret_string_value=secretsmanager.SecretStringValue.from_plain_text(json.dumps({
                "DBUserName": "",
                "DBPassword": "",
                "DBPort": ""
            }))
        )

        # 创建处理Secret更新的Lambda函数
        secret_update_fn = lambda_.Function(self, "SecretUpdateFunction",
            runtime=lambda_.Runtime.PYTHON_3_11,
            handler="index.lambda_handler",
            code=lambda_.Code.from_inline("""
import json
import boto3
import cfnresponse

secretsmanager = boto3.client('secretsmanager')

def lambda_handler(event, context):
    try:
        secret_name = event['ResourceProperties']['SecretName']
        action = event['ResourceProperties']['Action']

        response = secretsmanager.get_secret_value(SecretId=secret_name)
        secret_dict = json.loads(response['SecretString'])

        if action == 'AddDBName':
            secret_dict['DBName'] = ""
        elif action == 'RemoveDBPort':
            secret_dict.pop('DBPort', None)

        secretsmanager.put_secret_value(
            SecretId=secret_name,
            SecretString=json.dumps(secret_dict)
        )

        cfnresponse.send(event, context, cfnresponse.SUCCESS, {})
    except Exception as e:
        print(f"Error: {str(e)}")
        cfnresponse.send(event, context, cfnresponse.FAILED, {'Error': str(e)})
            """),
            role=iam.Role(self, "SecretUpdateRole",
                assumed_by=iam.ServicePrincipal("lambda.amazonaws.com"),
                managed_policies=[
                    iam.ManagedPolicy.from_aws_managed_policy_name("service-role/AWSLambdaBasicExecutionRole")
                ],
                inline_policies={
                    "SecretAccessPolicy": iam.PolicyDocument(
                        statements=[
                            iam.PolicyStatement(
                                actions=["secretsmanager:GetSecretValue", "secretsmanager:PutSecretValue"],
                                resources=[my_secret.secret_arn]
                            )
                        ]
                    )
                }
            )
        )

        # 创建自定义资源触发更新
        cr.AwsCustomResource(self, "UpdateSecretResource",
            on_create=cr.AwsSdkCall(
                service="Lambda",
                action="invoke",
                parameters={
                    "FunctionName": secret_update_fn.function_name,
                    "Payload": json.dumps({
                        "ResourceProperties": {
                            "SecretName": my_secret.secret_name,
                            "Action": "AddDBName" # 切换操作修改此处
                        }
                    })
                },
                physical_resource_id=cr.PhysicalResourceId.from_response("Payload")
            ),
            policy=cr.AwsCustomResourcePolicy.from_statements([
                iam.PolicyStatement(
                    actions=["lambda:InvokeFunction"],
                    resources=[secret_update_fn.function_arn]
                )
            ])
        )

        CfnOutput(self, "SecretArn", value=my_secret.secret_arn)

使用说明：

• 修改Action参数为AddDBName或RemoveDBPort来执行对应操作
• 部署CDK栈时，自定义资源会触发Lambda修改Secret，保留现有其他键值

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