秘密值无法转换为键名和值对
The secret value can't be converted to key name and value pairs
我已经有一个由 cloudformation 模板创建的秘密,格式如下 key/value:
UserPassword:
Type: AWS::SecretsManager::Secret
Properties:
Name: mysecret
GenerateSecretString:
SecretStringTemplate: '{"username": "test-user"}'
GenerateStringKey: "password"
PasswordLength: 16
ExcludeCharacters: '"@/\'
我已经创建了一个旋转 lambda,但是当我旋转秘密时,我只得到没有 key/value 对的纯文本格式,如下错误:
密钥值无法转换为密钥名称和值对。
这是我的 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))
您将需要更新默认轮换模板,以便它以您在 cloudformation 模板中指定的相同格式生成新密码。默认情况下,旋转 lambda 只是创建一个新的秘密字符串并仅填充该值。如果您将以下行添加到旋转 lambda 函数,您新旋转的秘密将采用 JSON 格式,允许 Web UI 将其解析为 name/value 对。
您需要在脚本顶部添加 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))
我已经有一个由 cloudformation 模板创建的秘密,格式如下 key/value:
UserPassword:
Type: AWS::SecretsManager::Secret
Properties:
Name: mysecret
GenerateSecretString:
SecretStringTemplate: '{"username": "test-user"}'
GenerateStringKey: "password"
PasswordLength: 16
ExcludeCharacters: '"@/\'
我已经创建了一个旋转 lambda,但是当我旋转秘密时,我只得到没有 key/value 对的纯文本格式,如下错误: 密钥值无法转换为密钥名称和值对。
这是我的 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))
您将需要更新默认轮换模板,以便它以您在 cloudformation 模板中指定的相同格式生成新密码。默认情况下,旋转 lambda 只是创建一个新的秘密字符串并仅填充该值。如果您将以下行添加到旋转 lambda 函数,您新旋转的秘密将采用 JSON 格式,允许 Web UI 将其解析为 name/value 对。
您需要在脚本顶部添加 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))