Merge pull request #2594 from mwaaas/fix/fetch_secret_manage_via_arn

fixing fetch secret manager via arn
This commit is contained in:
Mike Grima 2019-12-09 14:03:37 -08:00 committed by GitHub
commit e00c12a1e6
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 53 additions and 3 deletions

View file

@ -17,7 +17,7 @@ from .exceptions import (
InvalidRequestException,
ClientError,
)
from .utils import random_password, secret_arn
from .utils import random_password, secret_arn, get_secret_name_from_arn
class SecretsManager(BaseModel):
@ -25,11 +25,25 @@ class SecretsManager(BaseModel):
self.region = region_name
class SecretsStore(dict):
def __setitem__(self, key, value):
new_key = get_secret_name_from_arn(key)
super(SecretsStore, self).__setitem__(new_key, value)
def __getitem__(self, key):
new_key = get_secret_name_from_arn(key)
return super(SecretsStore, self).__getitem__(new_key)
def __contains__(self, key):
new_key = get_secret_name_from_arn(key)
return dict.__contains__(self, new_key)
class SecretsManagerBackend(BaseBackend):
def __init__(self, region_name=None, **kwargs):
super(SecretsManagerBackend, self).__init__()
self.region = region_name
self.secrets = {}
self.secrets = SecretsStore()
def reset(self):
region_name = self.region
@ -44,7 +58,6 @@ class SecretsManagerBackend(BaseBackend):
return (dt - epoch).total_seconds()
def get_secret_value(self, secret_id, version_id, version_stage):
if not self._is_valid_identifier(secret_id):
raise SecretNotFoundException()

View file

@ -72,6 +72,19 @@ def secret_arn(region, secret_id):
)
def get_secret_name_from_arn(secret_id):
# can fetch by both arn and by name
# but we are storing via name
# so we need to change the arn to name
# if it starts with arn then the secret id is arn
if secret_id.startswith("arn:aws:secretsmanager:"):
# split the arn by colon
# then get the last value which is the name appended with a random string
# then remove the random string
secret_id = "-".join(secret_id.split(":")[-1].split("-")[:-1])
return secret_id
def _exclude_characters(password, exclude_characters):
for c in exclude_characters:
if c in string.punctuation: