秘密值無法轉換為鍵名和值對

我已經有一個由 cloudformation 模板創建的具有以下鍵/值格式的秘密：

  UserPassword:
Type: AWS::SecretsManager::Secret
Properties:
  Name: mysecret
  GenerateSecretString:
    SecretStringTemplate: '{"username": "test-user"}'
    GenerateStringKey: "password"
    PasswordLength: 16
    ExcludeCharacters: '"@/\'

我已經創建了一個旋轉 lambda，但是當我旋轉秘密時，我只得到沒有鍵/值對的純文本格式，因為以下錯誤： 秘密值無法轉換為鍵名和值對。

這是我的 lambda 代碼：

import boto3
import logging
import os

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


def lambda_handler(event, context):
    
    arn = event['SecretId'] 
    token = event['ClientRequestToken']
    step = event['Step']

    # Setup the client
    service_client = boto3.client('secretsmanager')
    

    metadata = service_client.describe_secret(SecretId=arn)
    print(metadata)
    
    
    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):
    
    # Make sure the current secret exists
    service_client.get_secret_value(SecretId=arn, VersionStage="AWSCURRENT")

    # Now try to get the secret version, if that fails, put a new secret
    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:
        # Get exclude characters from environment variable
        exclude_characters = os.environ['EXCLUDE_CHARACTERS'] if 'EXCLUDE_CHARACTERS' in os.environ else '/@"\'\\'
        # Generate a random password
        passwd = service_client.get_random_password(ExcludeCharacters=exclude_characters)

        # Put the secret
        service_client.put_secret_value(SecretId=arn, ClientRequestToken=token, SecretString=passwd['RandomPassword'], VersionStages=['AWSPENDING'])
        logger.info("createSecret: Successfully put secret for ARN %s and version %s." % (arn, token))


def set_secret(service_client, arn, token):
    pass

def test_secret(service_client, arn, token):
    pass


def finish_secret(service_client, arn, token):
    # First describe the secret to get the current version
    metadata = service_client.describe_secret(SecretId=arn)
    current_version = None
    for version in metadata["VersionIdsToStages"]:
        print(version)
        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))

uj5u.com熱心網友回復：

您將需要更新默認輪換模板，以便它以您在 cloudformation 模板中指定的相同格式生成新密鑰。默認情況下，旋轉 lambda 只會創建一個新的秘密字串并僅填充該值。如果您將以下幾行添加到旋轉 lambda 函式中，您新旋轉的密鑰將采用 JSON 格式，允許 Web UI 將其決議為名稱/值對。

您需要在腳本頂部添加import json，然后修改 create_secret 函式，如下所示：

def create_secret(service_client, arn, token):

    # Make sure the current secret exists
    service_client.get_secret_value(SecretId=arn, VersionStage="AWSCURRENT")

    # Now try to get the secret version, if that fails, put a new secret
    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:
        # Get exclude characters from environment variable
        exclude_characters = os.environ['EXCLUDE_CHARACTERS'] if 'EXCLUDE_CHARACTERS' in os.environ else '/@"\'\\'
        # Generate a random password
        passwd = service_client.get_random_password(ExcludeCharacters=exclude_characters)
        # Create a username/password JSON object to pass to secrets manager
        generated_passwd = passwd['RandomPassword']
        secret_template = {"password": generated_passwd,"username": "test-user"}
        # Put the secret
        service_client.put_secret_value(SecretId=arn, ClientRequestToken=token, SecretString=json.dumps(secret_template), VersionStages=['AWSPENDING'])
        logger.info("createSecret: Successfully put secret for ARN %s and version %s." % (arn, token))

標籤：Python 亚马逊网络服务 秘密经理

上一篇：DAX集群-它如何知道要與哪個DynamoDB表交談？

下一篇：將用戶添加到ssh到AwsEc2