使用AWS Lambda轮换密钥时如何保留数据库其他凭证？

原模板的核心问题是create_secret函数生成新密码后，直接将SecretString设为纯密码，覆盖了原有的用户名、数据库名等字段。要保留这些信息，需要读取当前版本的完整Secret内容，仅替换密码字段，再存入AWSPENDING版本。

修改方案 #

1. 新增JSON处理依赖 #

在代码开头添加import json，用于解析和生成Secret的JSON格式内容。

2. 重写create_secret函数 #

调整该函数逻辑，实现：

• 获取当前AWSCURRENT版本的完整Secret（JSON格式）
• 生成新密码后，仅替换JSON中的密码字段（需匹配你的Secret实际字段名）
• 将包含所有字段的完整JSON作为SecretString存入AWSPENDING版本

修改后的完整代码 #

# Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: MIT-0

import boto3
import logging
import os
import json  # 新增：处理JSON格式的Secret内容

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


def lambda_handler(event, context):
    """Secrets Manager Rotation Template

    This is a template for creating an AWS Secrets Manager rotation lambda

    Args:
        event (dict): Lambda dictionary of event parameters. These keys must include the following:
            - SecretId: The secret ARN or identifier
            - ClientRequestToken: The ClientRequestToken of the secret version
            - Step: The rotation step (one of createSecret, setSecret, testSecret, or finishSecret)

        context (LambdaContext): The Lambda runtime information

    Raises:
        ResourceNotFoundException: If the secret with the specified arn and stage does not exist

        ValueError: If the secret is not properly configured for rotation

        KeyError: If the event parameters do not contain the expected keys

    """
    print(event)
    arn = event['SecretId']
    token = event['ClientRequestToken']
    step = event['Step']

    # Setup the client
    service_client = boto3.client('secretsmanager')#, endpoint_url=os.environ['SECRETS_MANAGER_ENDPOINT'])

    # Make sure the version is staged correctly
    metadata = service_client.describe_secret(SecretId=arn)
    if not metadata['RotationEnabled']:
        logger.error("Secret %s is not enabled for rotation" % arn)
        raise ValueError("Secret %s is not enabled for rotation" % arn)
    versions = metadata['VersionIdsToStages']
    if token not in versions:
        logger.error("Secret version %s has no stage for rotation of secret %s." % (token, arn))
        raise ValueError("Secret version %s has no stage for rotation of secret %s." % (token, arn))
    if "AWSCURRENT" in versions[token]:
        logger.info("Secret version %s already set as AWSCURRENT for secret %s." % (token, arn))
        return
    elif "AWSPENDING" not in versions[token]:
        logger.error("Secret version %s not set as AWSPENDING for rotation of secret %s." % (token, arn))
        raise ValueError("Secret version %s not set as AWSPENDING for rotation of secret %s." % (token, arn))

    if step == "createSecret":
        create_secret(service_client, arn, token)

    elif step == "setSecret":
        set_secret(service_client, arn, token)

    elif step == "testSecret":
        test_secret(service_client, arn, token)

    elif step == "finishSecret":
        finish_secret(service_client, arn, token)

    else:
        raise ValueError("Invalid step parameter")


def create_secret(service_client, arn, token):
    """Create the secret

    This method first checks for the existence of a secret for the passed in token. If one does not exist, it will generate a
    new secret and put it with the passed in token.

    Args:
        service_client (client): The secrets manager service client

        arn (string): The secret ARN or other identifier

        token (string): The ClientRequestToken associated with the secret version

    Raises:
        ResourceNotFoundException: If the secret with the specified arn and stage does not exist

    """
    # 获取当前AWSCURRENT版本的完整Secret内容
    current_secret = service_client.get_secret_value(SecretId=arn, VersionStage="AWSCURRENT")
    current_secret_dict = json.loads(current_secret['SecretString'])

    # 尝试获取AWSPENDING版本，不存在则生成新密码并创建
    try:
        service_client.get_secret_value(SecretId=arn, VersionId=token, VersionStage="AWSPENDING")
        logger.info("createSecret: Successfully retrieved secret for %s." % arn)
    except service_client.exceptions.ResourceNotFoundException:
        # 获取排除字符
        exclude_characters = os.environ['EXCLUDE_CHARACTERS'] if 'EXCLUDE_CHARACTERS' in os.environ else '/@"\'\\'
        # 生成随机密码
        passwd = service_client.get_random_password(ExcludeCharacters=exclude_characters)

        # 替换密码字段（根据你的Secret结构调整字段名，比如'password'）
        current_secret_dict['password'] = passwd['RandomPassword']
        # 序列化回JSON字符串
        new_secret_string = json.dumps(current_secret_dict)

        # 存入AWSPENDING版本
        service_client.put_secret_value(
            SecretId=arn,
            ClientRequestToken=token,
            SecretString=new_secret_string,
            VersionStages=['AWSPENDING']
        )
        logger.info("createSecret: Successfully put secret for ARN %s and version %s." % (arn, token))


def set_secret(service_client, arn, token):
    secret = service_client.get_secret_value(SecretId=arn, VersionId=token, VersionStage="AWSPENDING")    
    logger.info(secret)
    """Set the secret

    This method should set the AWSPENDING secret in the service that the secret belongs to. For example, if the secret is a database
    credential, this method should take the value of the AWSPENDING secret and set the user's password to this value in the database.

    Args:
        service_client (client): The secrets manager service client

        arn (string): The secret ARN or other identifier

        token (string): The ClientRequestToken associated with the secret version

    """
    # 此处需添加代码，将AWSPENDING中的密码更新到目标服务（例如数据库ALTER USER语句）
    # raise NotImplementedError


def test_secret(service_client, arn, token):
    """Test the secret

    This method should validate that the AWSPENDING secret works in the service that the secret belongs to. For example, if the secret
    is a database credential, this method should validate that the user can login with the password in AWSPENDING and that the user has
    all of the expected permissions against the database.

    Args:
        service_client (client): The secrets manager service client

        arn (string): The secret ARN or other identifier

        token (string): The ClientRequestToken associated with the secret version

    """
    # 此处需添加代码，用AWSPENDING的凭证测试目标服务连接（例如数据库登录验证）
    # raise NotImplementedError


def finish_secret(service_client, arn, token):
    """Finish the secret

    This method finalizes the rotation process by marking the secret version passed in as the AWSCURRENT secret.

    Args:
        service_client (client): The secrets manager service client

        arn (string): The secret ARN or other identifier

        token (string): The ClientRequestToken associated with the secret version

    Raises:
        ResourceNotFoundException: If the secret with the specified arn does not exist

    """
    # First describe the secret to get the current version
    metadata = service_client.describe_secret(SecretId=arn)
    current_version = None
    for version in metadata["VersionIdsToStages"]:
        if "AWSCURRENT" in metadata["VersionIdsToStages"][version]:
            if version == token:
                # The correct version is already marked as current, return
                logger.info("finishSecret: Version %s already marked as AWSCURRENT for %s" % (version, arn))
                return
            current_version = version
            break

    # Finalize by staging the secret version current
    service_client.update_secret_version_stage(SecretId=arn, VersionStage="AWSCURRENT", MoveToVersionId=token, RemoveFromVersionId=current_version)
    logger.info("finishSecret: Successfully set AWSCURRENT stage to version %s for secret %s." % (token, arn))

关键注意事项 #

• 确保你的Secret是以JSON字符串存储，格式示例：{"username":"admin","password":"oldpass","dbname":"mydb"}
• 如果密码字段名不是password，请修改current_secret_dict['password']中的键名以匹配实际结构
• set_secret和test_secret函数需自行实现对应服务的密码更新和验证逻辑

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