in functions/source/rotate_sealer_key_secret/lambda_function.py [0:0]
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
"""
metadata = service_client.describe_secret(SecretId=arn)
sorted_versions = sorted(metadata["VersionIdsToStages"].items(),key=sort_by_tag)
logger.info("createSecret: initial versions %s" % pformat(sorted_versions))
# Make sure the current secret exists
try:
service_client.get_secret_value(SecretId=arn, VersionStage="AWSCURRENT")
except service_client.exceptions.ResourceNotFoundException:
logger.info("createSecret: NO CURRENT VERSION FOUND, CREATING ONE")
service_client.put_secret_value(SecretId=arn, ClientRequestToken=token, SecretBinary=secrets.token_bytes(16), VersionStages=['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:
# Put the secret
service_client.put_secret_value(SecretId=arn, ClientRequestToken=token, SecretBinary=secrets.token_bytes(16), VersionStages=['AWSPENDING'])
logger.info("createSecret: Successfully put secret for ARN %s and version %s." % (arn, token))
metadata = service_client.describe_secret(SecretId=arn)
sorted_versions = sorted(metadata["VersionIdsToStages"].items(),key=sort_by_tag)
logger.info("createSecret: final versions %s" % pformat(sorted_versions))