from typing import List from aws_cdk import core as cdk from cache import elasticache_secret as secret from config.default import default from config import config_util as config def get_user_secrets(scope: cdk.Construct) -> List: """ Create and store username as well as autogenerated password in AWS SecretsManager. Args: scope: the cdk construct. Returns: None """ secret_config = config.get_secret_config() if secret_config is None: return None auth_token_enabled = secret_config.get('auth_token_enabled', default['auth_token_enabled']) if auth_token_enabled is True: return None users = secret_config.get('users', None) if users is None: return None cluster_name = config.get_cluster_name() user_secrets = [] for idx, user in enumerate(users): user_id = config.get_user_id(user) if user_id is None: return None user_name = config.get_user_name(user) if user_name is None: return None user_secret = secret.UserSecret( scope, f"ElasticacheUserSecret-{idx}", secret_name=f"/elasticache/{cluster_name}/{user_name}", user_id=user_id, user_name=user_name, user_acl=config.get_user_acl(user), cluster_name=cluster_name, cmk=config.get_cmk(), ) user_secrets.append(user_secret) return user_secrets def get_auth_token(scope: cdk.Construct) -> str: """ Create and store the auto generated Redis Auth Token/Password in AWS SecretsManager. AuthToken can be specified only on replication groups where TransitEncryptionEnabled is true Args: scope: the cdk construct. Returns: str """ auth_token_enabled = config.get_auth_token_enabled(); if auth_token_enabled is False: return None if config.get_transit_encryption() is False: print('The AUTH token is only supported when encryption-in-transit is enabled') return None cluster_name = config.get_cluster_name() token_secret = secret.TokenSecret( scope, "ElasticacheTokenSecret", secret_name=f"/elasticache/{cluster_name}/auth-token", cluster_name=cluster_name, cmk=config.get_cmk() ) return token_secret.secret.secret_value_from_json("token").to_string()