SecretsManagerMongoDBRotationMultiUser/lambda_function.py [11:161]: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - logger = logging.getLogger() logger.setLevel(logging.INFO) def lambda_handler(event, context): """Secrets Manager MongoDB Handler This handler uses the master-user rotation scheme to rotate an MongoDB user credential. During the first rotation, this scheme logs into the database as the master user, creates a new user (appending _clone to the username), and grants the new user all of the permissions from the user being rotated. Once the secret is in this state, every subsequent rotation simply creates a new secret with the AWSPREVIOUS user credentials, changes that user's password, and then marks the latest secret as AWSCURRENT. The Secret SecretString is expected to be a JSON string with the following format: { 'engine': , 'host': , 'username': , 'password': , 'dbname': , 'port': , 'masterarn': 'ssl': } Args: event (dict): Lambda dictionary of event parameters. These keys must include the following: - SecretId: The secret ARN or identifier - ClientRequestToken: The ClientRequestToken of the secret version - Step: The rotation step (one of createSecret, setSecret, testSecret, or finishSecret) context (LambdaContext): The Lambda runtime information Raises: ResourceNotFoundException: If the secret with the specified arn and stage does not exist ValueError: If the secret is not properly configured for rotation KeyError: If the secret json does not contain the expected keys """ arn = event['SecretId'] token = event['ClientRequestToken'] step = event['Step'] # Setup the client service_client = boto3.client('secretsmanager', endpoint_url=os.environ['SECRETS_MANAGER_ENDPOINT']) # Make sure the version is staged correctly metadata = service_client.describe_secret(SecretId=arn) if "RotationEnabled" in metadata and not metadata['RotationEnabled']: logger.error("Secret %s is not enabled for rotation" % arn) raise ValueError("Secret %s is not enabled for rotation" % arn) versions = metadata['VersionIdsToStages'] if token not in versions: logger.error("Secret version %s has no stage for rotation of secret %s." % (token, arn)) raise ValueError("Secret version %s has no stage for rotation of secret %s." % (token, arn)) if "AWSCURRENT" in versions[token]: logger.info("Secret version %s already set as AWSCURRENT for secret %s." % (token, arn)) return elif "AWSPENDING" not in versions[token]: logger.error("Secret version %s not set as AWSPENDING for rotation of secret %s." % (token, arn)) raise ValueError("Secret version %s not set as AWSPENDING for rotation of secret %s." % (token, arn)) # Call the appropriate step if step == "createSecret": create_secret(service_client, arn, token) elif step == "setSecret": set_secret(service_client, arn, token) elif step == "testSecret": test_secret(service_client, arn, token) elif step == "finishSecret": finish_secret(service_client, arn, token) else: logger.error("lambda_handler: Invalid step parameter %s for secret %s" % (step, arn)) raise ValueError("Invalid step parameter %s for secret %s" % (step, arn)) def create_secret(service_client, arn, token): """Generate a new secret This method first checks for the existence of a secret for the passed in token. If one does not exist, it will generate a new secret and save it using the passed in token. Args: service_client (client): The secrets manager service client arn (string): The secret ARN or other identifier token (string): The ClientRequestToken associated with the secret version Raises: ValueError: If the current secret is not valid JSON KeyError: If the secret json does not contain the expected keys """ # Make sure the current secret exists current_dict = get_secret_dict(service_client, arn, "AWSCURRENT") # Now try to get the secret version, if that fails, put a new secret try: get_secret_dict(service_client, arn, "AWSPENDING", token) logger.info("createSecret: Successfully retrieved secret for %s." % arn) except service_client.exceptions.ResourceNotFoundException: # Get the alternate username swapping between the original user and the user with _clone appended to it current_dict['username'] = get_alt_username(current_dict['username']) # Get exclude characters from environment variable exclude_characters = os.environ['EXCLUDE_CHARACTERS'] if 'EXCLUDE_CHARACTERS' in os.environ else '/@"\'\\' # Generate a random password passwd = service_client.get_random_password(ExcludeCharacters=exclude_characters) current_dict['password'] = passwd['RandomPassword'] # Put the secret service_client.put_secret_value(SecretId=arn, ClientRequestToken=token, SecretString=json.dumps(current_dict), VersionStages=['AWSPENDING']) logger.info("createSecret: Successfully put secret for ARN %s and version %s." % (arn, token)) def set_secret(service_client, arn, token): """Set the pending secret in the database This method tries to login to the database with the AWSPENDING secret and returns on success. If that fails, it tries to login with the master credentials from the masterarn in the current secret. If this succeeds, it adds all grants for AWSCURRENT user to the AWSPENDING user, creating the user and/or setting the password in the process. Else, it throws a ValueError. Args: service_client (client): The secrets manager service client arn (string): The secret ARN or other identifier token (string): The ClientRequestToken associated with the secret version Raises: ResourceNotFoundException: If the secret with the specified arn and stage does not exist ValueError: If the secret is not valid JSON or master credentials could not be used to login to DB KeyError: If the secret json does not contain the expected keys """ # First try to login with the pending secret, if it succeeds, return current_dict = get_secret_dict(service_client, arn, "AWSCURRENT") pending_dict = get_secret_dict(service_client, arn, "AWSPENDING", token) conn = get_connection(pending_dict) if conn: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - SecretsManagerRDSMySQLRotationMultiUser/lambda_function.py [10:160]: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - logger = logging.getLogger() logger.setLevel(logging.INFO) def lambda_handler(event, context): """Secrets Manager RDS MySQL Handler This handler uses the master-user rotation scheme to rotate an RDS MySQL user credential. During the first rotation, this scheme logs into the database as the master user, creates a new user (appending _clone to the username), and grants the new user all of the permissions from the user being rotated. Once the secret is in this state, every subsequent rotation simply creates a new secret with the AWSPREVIOUS user credentials, changes that user's password, and then marks the latest secret as AWSCURRENT. The Secret SecretString is expected to be a JSON string with the following format: { 'engine': , 'host': , 'username': , 'password': , 'dbname': , 'port': , 'masterarn': } Args: event (dict): Lambda dictionary of event parameters. These keys must include the following: - SecretId: The secret ARN or identifier - ClientRequestToken: The ClientRequestToken of the secret version - Step: The rotation step (one of createSecret, setSecret, testSecret, or finishSecret) context (LambdaContext): The Lambda runtime information Raises: ResourceNotFoundException: If the secret with the specified arn and stage does not exist ValueError: If the secret is not properly configured for rotation KeyError: If the secret json does not contain the expected keys """ arn = event['SecretId'] token = event['ClientRequestToken'] step = event['Step'] # Setup the client service_client = boto3.client('secretsmanager', endpoint_url=os.environ['SECRETS_MANAGER_ENDPOINT']) # Make sure the version is staged correctly metadata = service_client.describe_secret(SecretId=arn) if "RotationEnabled" in metadata and not metadata['RotationEnabled']: logger.error("Secret %s is not enabled for rotation" % arn) raise ValueError("Secret %s is not enabled for rotation" % arn) versions = metadata['VersionIdsToStages'] if token not in versions: logger.error("Secret version %s has no stage for rotation of secret %s." % (token, arn)) raise ValueError("Secret version %s has no stage for rotation of secret %s." % (token, arn)) if "AWSCURRENT" in versions[token]: logger.info("Secret version %s already set as AWSCURRENT for secret %s." % (token, arn)) return elif "AWSPENDING" not in versions[token]: logger.error("Secret version %s not set as AWSPENDING for rotation of secret %s." % (token, arn)) raise ValueError("Secret version %s not set as AWSPENDING for rotation of secret %s." % (token, arn)) # Call the appropriate step if step == "createSecret": create_secret(service_client, arn, token) elif step == "setSecret": set_secret(service_client, arn, token) elif step == "testSecret": test_secret(service_client, arn, token) elif step == "finishSecret": finish_secret(service_client, arn, token) else: logger.error("lambda_handler: Invalid step parameter %s for secret %s" % (step, arn)) raise ValueError("Invalid step parameter %s for secret %s" % (step, arn)) def create_secret(service_client, arn, token): """Generate a new secret This method first checks for the existence of a secret for the passed in token. If one does not exist, it will generate a new secret and save it using the passed in token. Args: service_client (client): The secrets manager service client arn (string): The secret ARN or other identifier token (string): The ClientRequestToken associated with the secret version Raises: ValueError: If the current secret is not valid JSON KeyError: If the secret json does not contain the expected keys """ # Make sure the current secret exists current_dict = get_secret_dict(service_client, arn, "AWSCURRENT") # Now try to get the secret version, if that fails, put a new secret try: get_secret_dict(service_client, arn, "AWSPENDING", token) logger.info("createSecret: Successfully retrieved secret for %s." % arn) except service_client.exceptions.ResourceNotFoundException: # Get the alternate username swapping between the original user and the user with _clone appended to it current_dict['username'] = get_alt_username(current_dict['username']) # Get exclude characters from environment variable exclude_characters = os.environ['EXCLUDE_CHARACTERS'] if 'EXCLUDE_CHARACTERS' in os.environ else '/@"\'\\' # Generate a random password passwd = service_client.get_random_password(ExcludeCharacters=exclude_characters) current_dict['password'] = passwd['RandomPassword'] # Put the secret service_client.put_secret_value(SecretId=arn, ClientRequestToken=token, SecretString=json.dumps(current_dict), VersionStages=['AWSPENDING']) logger.info("createSecret: Successfully put secret for ARN %s and version %s." % (arn, token)) def set_secret(service_client, arn, token): """Set the pending secret in the database This method tries to login to the database with the AWSPENDING secret and returns on success. If that fails, it tries to login with the master credentials from the masterarn in the current secret. If this succeeds, it adds all grants for AWSCURRENT user to the AWSPENDING user, creating the user and/or setting the password in the process. Else, it throws a ValueError. Args: service_client (client): The secrets manager service client arn (string): The secret ARN or other identifier token (string): The ClientRequestToken associated with the secret version Raises: ResourceNotFoundException: If the secret with the specified arn and stage does not exist ValueError: If the secret is not valid JSON or master credentials could not be used to login to DB KeyError: If the secret json does not contain the expected keys """ current_dict = get_secret_dict(service_client, arn, "AWSCURRENT") pending_dict = get_secret_dict(service_client, arn, "AWSPENDING", token) # First try to login with the pending secret, if it succeeds, return conn = get_connection(pending_dict) if conn: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -