Created
July 29, 2021 02:52
-
-
Save brevityinmotion/69bd2a7ac2d645b1829fb3aadff25192 to your computer and use it in GitHub Desktop.
Code samples for utilizing AWS Secrets Manager
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import boto3 | |
import logging | |
import ClientError | |
import json | |
from botocore.exceptions import ClientError | |
# update a secret value within AWS Secrets Manager | |
def put_secret(secretName,secretValue,regionName): | |
# Create a Secrets Manager client | |
session = boto3.session.Session() | |
client = session.client( | |
service_name='secretsmanager', | |
region_name=regionName | |
) | |
response = client.put_secret_value( | |
SecretId=secretName, | |
SecretString=secretValue | |
) | |
return response | |
def create_secret(secretName,secretValue,regionName,secretDesc): | |
session = boto3.session.Session() | |
client = session.client( | |
service_name='secretsmanager', | |
region_name=regionName | |
) | |
response = client.create_secret( | |
Name=secretName, | |
Description=secretDesc, | |
SecretString=secretValue | |
) | |
return response | |
def get_secret(secret_name, region_name): | |
# Create a Secrets Manager client | |
session = boto3.session.Session() | |
client = session.client( | |
service_name='secretsmanager', | |
region_name=region_name | |
) | |
# In this sample we only handle the specific exceptions for the 'GetSecretValue' API. | |
# See https://docs.aws.amazon.com/secretsmanager/latest/apireference/API_GetSecretValue.html | |
# We rethrow the exception by default. | |
try: | |
get_secret_value_response = client.get_secret_value( | |
SecretId=secret_name | |
) | |
except ClientError as e: | |
if e.response['Error']['Code'] == 'DecryptionFailureException': | |
# Secrets Manager can't decrypt the protected secret text using the provided KMS key. | |
# Deal with the exception here, and/or rethrow at your discretion. | |
raise e | |
elif e.response['Error']['Code'] == 'InternalServiceErrorException': | |
# An error occurred on the server side. | |
# Deal with the exception here, and/or rethrow at your discretion. | |
raise e | |
elif e.response['Error']['Code'] == 'InvalidParameterException': | |
# You provided an invalid value for a parameter. | |
# Deal with the exception here, and/or rethrow at your discretion. | |
raise e | |
elif e.response['Error']['Code'] == 'InvalidRequestException': | |
# You provided a parameter value that is not valid for the current state of the resource. | |
# Deal with the exception here, and/or rethrow at your discretion. | |
raise e | |
elif e.response['Error']['Code'] == 'ResourceNotFoundException': | |
# We can't find the resource that you asked for. | |
# Deal with the exception here, and/or rethrow at your discretion. | |
raise e | |
else: | |
# Decrypts secret using the associated KMS CMK. | |
# Depending on whether the secret is a string or binary, one of these fields will be populated. | |
if 'SecretString' in get_secret_value_response: | |
secret = get_secret_value_response['SecretString'] | |
return secret | |
else: | |
decoded_binary_secret = base64.b64decode(get_secret_value_response['SecretBinary']) | |
return json.loads(secret) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment