Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix: aws warnings regarding ssl connection #98

Closed
wants to merge 3 commits into from
Closed

fix: aws warnings regarding ssl connection #98

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
93f24f6
fix: aws warnings regarding ssl connection
BlackDark Jul 25, 2023
5b28cd2
Merge branch 'develop' into fix/aws-warnings
glennmatthews Dec 8, 2023
4547bd4
Update aws.py for black
glennmatthews Dec 8, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
92 changes: 48 additions & 44 deletions nautobot_secrets_providers/providers/aws.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,12 +9,13 @@
except (ImportError, ModuleNotFoundError):
boto3 = None

from contextlib import closing

from django import forms

from nautobot.core.forms import BootstrapMixin
from nautobot.extras.secrets import exceptions, SecretsProvider


__all__ = ("AWSSecretsManagerSecretsProvider", "AWSSystemsManagerParameterStore")


Expand Down Expand Up @@ -55,42 +56,43 @@ def get_value_for_secret(cls, secret, obj=None, **kwargs):

# Create a Secrets Manager client.
session = boto3.session.Session()
client = session.client(service_name="secretsmanager", region_name=region_name)

# This is based on sample code to 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 err:
if err.response["Error"]["Code"] == "DecryptionFailureException": # pylint: disable=no-else-raise
# 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 exceptions.SecretProviderError(secret, cls, str(err))
elif err.response["Error"]["Code"] == "InternalServiceErrorException":
# An error occurred on the server side.
# Deal with the exception here, and/or rethrow at your discretion.
raise exceptions.SecretProviderError(secret, cls, str(err))
elif err.response["Error"]["Code"] == "InvalidParameterException":
# You provided an invalid value for a parameter.
# Deal with the exception here, and/or rethrow at your discretion.
raise exceptions.SecretParametersError(secret, cls, str(err))
elif err.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 exceptions.SecretProviderError(secret, cls, str(err))
elif err.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 exceptions.SecretValueNotFoundError(secret, cls, str(err))
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_value = get_secret_value_response["SecretString"]
# https://github.com/boto/boto3/issues/454
with closing(session.client(service_name="secretsmanager", region_name=region_name)) as client:
# This is based on sample code to 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 err:
if err.response["Error"]["Code"] == "DecryptionFailureException": # pylint: disable=no-else-raise
# 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 exceptions.SecretProviderError(secret, cls, str(err))
elif err.response["Error"]["Code"] == "InternalServiceErrorException":
# An error occurred on the server side.
# Deal with the exception here, and/or rethrow at your discretion.
raise exceptions.SecretProviderError(secret, cls, str(err))
elif err.response["Error"]["Code"] == "InvalidParameterException":
# You provided an invalid value for a parameter.
# Deal with the exception here, and/or rethrow at your discretion.
raise exceptions.SecretParametersError(secret, cls, str(err))
elif err.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 exceptions.SecretProviderError(secret, cls, str(err))
elif err.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 exceptions.SecretValueNotFoundError(secret, cls, str(err))
else:
# TODO(jathan): Do we care about this? Let's figure out what to do about a binary value?
secret_value = base64.b64decode(get_secret_value_response["SecretBinary"]) # noqa
# 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_value = get_secret_value_response["SecretString"]
else:
# TODO(jathan): Do we care about this? Let's figure out what to do about a binary value?
secret_value = base64.b64decode(get_secret_value_response["SecretBinary"]) # noqa

# If we get this far it should be valid JSON.
data = json.loads(secret_value)
Expand Down Expand Up @@ -140,17 +142,19 @@ def get_value_for_secret(cls, secret, obj=None, **kwargs):

# Create a SSM client.
session = boto3.session.Session()
client = session.client(service_name="ssm", region_name=parameters.get("region"))
try:
get_secret_value_response = client.get_parameter(Name=parameters.get("name"), WithDecryption=True)
except ClientError as err:
if err.response["Error"]["Code"] == "ParameterNotFound":
raise exceptions.SecretParametersError(secret, cls, str(err))

if err.response["Error"]["Code"] == "ParameterVersionNotFound":
raise exceptions.SecretValueNotFoundError(secret, cls, str(err))
# https://github.com/boto/boto3/issues/454
with closing(session.client(service_name="ssm", region_name=parameters.get("region"))) as client:
try:
get_secret_value_response = client.get_parameter(Name=parameters.get("name"), WithDecryption=True)
except ClientError as err:
if err.response["Error"]["Code"] == "ParameterNotFound":
raise exceptions.SecretParametersError(secret, cls, str(err))

raise exceptions.SecretProviderError(secret, cls, str(err))
if err.response["Error"]["Code"] == "ParameterVersionNotFound":
raise exceptions.SecretValueNotFoundError(secret, cls, str(err))

raise exceptions.SecretProviderError(secret, cls, str(err))

try:
# Fetch the Value field from the parameter which must be a json field.
Expand Down
Loading