Raise appropriate error when secret exists but has no value
This commit is contained in:
parent
cc96a5e659
commit
381e7b165f
@ -7,11 +7,11 @@ class SecretsManagerClientError(JsonRESTError):
|
|||||||
|
|
||||||
|
|
||||||
class ResourceNotFoundException(SecretsManagerClientError):
|
class ResourceNotFoundException(SecretsManagerClientError):
|
||||||
def __init__(self):
|
def __init__(self, message):
|
||||||
self.code = 404
|
self.code = 404
|
||||||
super(ResourceNotFoundException, self).__init__(
|
super(ResourceNotFoundException, self).__init__(
|
||||||
"ResourceNotFoundException",
|
"ResourceNotFoundException",
|
||||||
"Secrets Manager can't find the specified secret"
|
message,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@ -46,7 +46,7 @@ class SecretsManagerBackend(BaseBackend):
|
|||||||
def get_secret_value(self, secret_id, version_id, version_stage):
|
def get_secret_value(self, secret_id, version_id, version_stage):
|
||||||
|
|
||||||
if not self._is_valid_identifier(secret_id):
|
if not self._is_valid_identifier(secret_id):
|
||||||
raise ResourceNotFoundException()
|
raise ResourceNotFoundException("Secrets Manager can't find the specified secret")
|
||||||
|
|
||||||
if not version_id and version_stage:
|
if not version_id and version_stage:
|
||||||
# set version_id to match version_stage
|
# set version_id to match version_stage
|
||||||
@ -56,7 +56,7 @@ class SecretsManagerBackend(BaseBackend):
|
|||||||
version_id = ver_id
|
version_id = ver_id
|
||||||
break
|
break
|
||||||
if not version_id:
|
if not version_id:
|
||||||
raise ResourceNotFoundException()
|
raise ResourceNotFoundException("Secrets Manager can't find the specified secret")
|
||||||
|
|
||||||
# TODO check this part
|
# TODO check this part
|
||||||
if 'deleted_date' in self.secrets[secret_id]:
|
if 'deleted_date' in self.secrets[secret_id]:
|
||||||
@ -84,6 +84,12 @@ class SecretsManagerBackend(BaseBackend):
|
|||||||
if 'secret_binary' in secret_version:
|
if 'secret_binary' in secret_version:
|
||||||
response_data["SecretBinary"] = secret_version['secret_binary']
|
response_data["SecretBinary"] = secret_version['secret_binary']
|
||||||
|
|
||||||
|
if 'secret_string' not in secret_version and 'secret_binary' not in secret_version:
|
||||||
|
raise ResourceNotFoundException(
|
||||||
|
"Secrets Manager can’t find the specified secret value for staging label: %s" %
|
||||||
|
(version_stage or "AWSCURRENT")
|
||||||
|
)
|
||||||
|
|
||||||
response = json.dumps(response_data)
|
response = json.dumps(response_data)
|
||||||
|
|
||||||
return response
|
return response
|
||||||
@ -169,7 +175,7 @@ class SecretsManagerBackend(BaseBackend):
|
|||||||
|
|
||||||
def describe_secret(self, secret_id):
|
def describe_secret(self, secret_id):
|
||||||
if not self._is_valid_identifier(secret_id):
|
if not self._is_valid_identifier(secret_id):
|
||||||
raise ResourceNotFoundException
|
raise ResourceNotFoundException("Secrets Manager can't find the specified secret")
|
||||||
|
|
||||||
secret = self.secrets[secret_id]
|
secret = self.secrets[secret_id]
|
||||||
|
|
||||||
@ -198,7 +204,7 @@ class SecretsManagerBackend(BaseBackend):
|
|||||||
rotation_days = 'AutomaticallyAfterDays'
|
rotation_days = 'AutomaticallyAfterDays'
|
||||||
|
|
||||||
if not self._is_valid_identifier(secret_id):
|
if not self._is_valid_identifier(secret_id):
|
||||||
raise ResourceNotFoundException
|
raise ResourceNotFoundException("Secrets Manager can't find the specified secret")
|
||||||
|
|
||||||
if 'deleted_date' in self.secrets[secret_id]:
|
if 'deleted_date' in self.secrets[secret_id]:
|
||||||
raise InvalidRequestException(
|
raise InvalidRequestException(
|
||||||
@ -340,7 +346,7 @@ class SecretsManagerBackend(BaseBackend):
|
|||||||
def delete_secret(self, secret_id, recovery_window_in_days, force_delete_without_recovery):
|
def delete_secret(self, secret_id, recovery_window_in_days, force_delete_without_recovery):
|
||||||
|
|
||||||
if not self._is_valid_identifier(secret_id):
|
if not self._is_valid_identifier(secret_id):
|
||||||
raise ResourceNotFoundException
|
raise ResourceNotFoundException("Secrets Manager can't find the specified secret")
|
||||||
|
|
||||||
if 'deleted_date' in self.secrets[secret_id]:
|
if 'deleted_date' in self.secrets[secret_id]:
|
||||||
raise InvalidRequestException(
|
raise InvalidRequestException(
|
||||||
@ -370,7 +376,7 @@ class SecretsManagerBackend(BaseBackend):
|
|||||||
secret = self.secrets.get(secret_id, None)
|
secret = self.secrets.get(secret_id, None)
|
||||||
|
|
||||||
if not secret:
|
if not secret:
|
||||||
raise ResourceNotFoundException
|
raise ResourceNotFoundException("Secrets Manager can't find the specified secret")
|
||||||
|
|
||||||
arn = secret_arn(self.region, secret['secret_id'])
|
arn = secret_arn(self.region, secret['secret_id'])
|
||||||
name = secret['name']
|
name = secret['name']
|
||||||
@ -380,7 +386,7 @@ class SecretsManagerBackend(BaseBackend):
|
|||||||
def restore_secret(self, secret_id):
|
def restore_secret(self, secret_id):
|
||||||
|
|
||||||
if not self._is_valid_identifier(secret_id):
|
if not self._is_valid_identifier(secret_id):
|
||||||
raise ResourceNotFoundException
|
raise ResourceNotFoundException("Secrets Manager can't find the specified secret")
|
||||||
|
|
||||||
self.secrets[secret_id].pop('deleted_date', None)
|
self.secrets[secret_id].pop('deleted_date', None)
|
||||||
|
|
||||||
|
@ -8,7 +8,7 @@ import string
|
|||||||
import pytz
|
import pytz
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
import sure # noqa
|
import sure # noqa
|
||||||
from nose.tools import assert_raises
|
from nose.tools import assert_raises, assert_raises_regexp
|
||||||
from six import b
|
from six import b
|
||||||
|
|
||||||
DEFAULT_SECRET_NAME = 'test-secret'
|
DEFAULT_SECRET_NAME = 'test-secret'
|
||||||
@ -65,6 +65,21 @@ def test_get_secret_value_that_is_marked_deleted():
|
|||||||
result = conn.get_secret_value(SecretId='test-secret')
|
result = conn.get_secret_value(SecretId='test-secret')
|
||||||
|
|
||||||
|
|
||||||
|
@mock_secretsmanager
|
||||||
|
def test_get_secret_that_has_no_value():
|
||||||
|
conn = boto3.client('secretsmanager', region_name='us-west-2')
|
||||||
|
|
||||||
|
create_secret = conn.create_secret(Name="java-util-test-password")
|
||||||
|
|
||||||
|
with assert_raises_regexp(
|
||||||
|
ClientError,
|
||||||
|
r"An error occurred \(ResourceNotFoundException\) when calling the GetSecretValue "
|
||||||
|
r"operation: Secrets Manager can’t find the specified secret value for staging label: "
|
||||||
|
r"AWSCURRENT"
|
||||||
|
):
|
||||||
|
result = conn.get_secret_value(SecretId='java-util-test-password')
|
||||||
|
|
||||||
|
|
||||||
@mock_secretsmanager
|
@mock_secretsmanager
|
||||||
def test_create_secret():
|
def test_create_secret():
|
||||||
conn = boto3.client('secretsmanager', region_name='us-east-1')
|
conn = boto3.client('secretsmanager', region_name='us-east-1')
|
||||||
|
@ -73,6 +73,26 @@ def test_get_secret_that_does_not_match():
|
|||||||
assert json_data['message'] == "Secrets Manager can't find the specified secret"
|
assert json_data['message'] == "Secrets Manager can't find the specified secret"
|
||||||
assert json_data['__type'] == 'ResourceNotFoundException'
|
assert json_data['__type'] == 'ResourceNotFoundException'
|
||||||
|
|
||||||
|
@mock_secretsmanager
|
||||||
|
def test_get_secret_that_has_no_value():
|
||||||
|
backend = server.create_backend_app('secretsmanager')
|
||||||
|
test_client = backend.test_client()
|
||||||
|
|
||||||
|
create_secret = test_client.post('/',
|
||||||
|
data={"Name": DEFAULT_SECRET_NAME},
|
||||||
|
headers={
|
||||||
|
"X-Amz-Target": "secretsmanager.CreateSecret"},
|
||||||
|
)
|
||||||
|
get_secret = test_client.post('/',
|
||||||
|
data={"SecretId": DEFAULT_SECRET_NAME},
|
||||||
|
headers={
|
||||||
|
"X-Amz-Target": "secretsmanager.GetSecretValue"},
|
||||||
|
)
|
||||||
|
|
||||||
|
json_data = json.loads(get_secret.data.decode("utf-8"))
|
||||||
|
assert json_data['message'] == "Secrets Manager can’t find the specified secret value for staging label: AWSCURRENT"
|
||||||
|
assert json_data['__type'] == 'ResourceNotFoundException'
|
||||||
|
|
||||||
@mock_secretsmanager
|
@mock_secretsmanager
|
||||||
def test_create_secret():
|
def test_create_secret():
|
||||||
|
|
||||||
|
Loading…
Reference in New Issue
Block a user