SecretsManagerRDSSQLServerRotationMultiUser/lambda_function.py [220:450]: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - finally: conn.close() def test_secret(service_client, arn, token): """Test the pending secret against the database This method tries to log into the database with the secrets staged with AWSPENDING and runs a permissions check to ensure the user has the correct permissions. 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 pending credentials could not be used to login to the database KeyError: If the secret json does not contain the expected keys """ # Try to login with the pending secret, if it succeeds, return conn = get_connection(get_secret_dict(service_client, arn, "AWSPENDING", token)) if conn: # This is where the lambda will validate the user's permissions. Uncomment/modify the below lines to # tailor these validations to your needs try: with conn.cursor() as cur: cur.execute("SELECT @@VERSION AS version") finally: conn.close() logger.info("testSecret: Successfully signed into SQL Server DB with AWSPENDING secret in %s." % arn) return else: logger.error("testSecret: Unable to log into database with pending secret of secret ARN %s" % arn) raise ValueError("Unable to log into database with pending secret of secret ARN %s" % arn) def finish_secret(service_client, arn, token): """Finish the rotation by marking the pending secret as current This method moves the secret from the AWSPENDING stage to the AWSCURRENT stage. 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 does not exist """ # First describe the secret to get the current version metadata = service_client.describe_secret(SecretId=arn) current_version = None for version in metadata["VersionIdsToStages"]: if "AWSCURRENT" in metadata["VersionIdsToStages"][version]: if version == token: # The correct version is already marked as current, return logger.info("finishSecret: Version %s already marked as AWSCURRENT for %s" % (version, arn)) return current_version = version break # Finalize by staging the secret version current service_client.update_secret_version_stage(SecretId=arn, VersionStage="AWSCURRENT", MoveToVersionId=token, RemoveFromVersionId=current_version) logger.info("finishSecret: Successfully set AWSCURRENT stage to version %s for secret %s." % (token, arn)) def get_connection(secret_dict): """Gets a connection to a SQL Server DB from a secret dictionary This helper function uses connectivity information from the secret dictionary to initiate connection attempt(s) to the database. Will attempt a fallback, non-SSL connection when initial connection fails using SSL and fall_back is True. Args: secret_dict (dict): The Secret Dictionary Returns: Connection: The pymssql.Connection object if successful. None otherwise Raises: KeyError: If the secret json does not contain the expected keys """ # Parse and validate the secret JSON string port = str(secret_dict['port']) if 'port' in secret_dict else '1433' dbname = secret_dict['dbname'] if 'dbname' in secret_dict else 'master' # Get SSL connectivity configuration use_ssl, fall_back = get_ssl_config(secret_dict) # if an 'ssl' key is not found or does not contain a valid value, attempt an SSL connection and fall back to non-SSL on failure conn = connect_and_authenticate(secret_dict, port, dbname, use_ssl) if conn or not fall_back: return conn else: return connect_and_authenticate(secret_dict, port, dbname, False) def get_ssl_config(secret_dict): """Gets the desired SSL and fall back behavior using a secret dictionary This helper function uses the existance and value the 'ssl' key in a secret dictionary to determine desired SSL connectivity configuration. Its behavior is as follows: - 'ssl' key DNE or invalid type/value: return True, True - 'ssl' key is bool: return secret_dict['ssl'], False - 'ssl' key equals "true" ignoring case: return True, False - 'ssl' key equals "false" ignoring case: return False, False Args: secret_dict (dict): The Secret Dictionary Returns: Tuple(use_ssl, fall_back): SSL configuration - use_ssl (bool): Flag indicating if an SSL connection should be attempted - fall_back (bool): Flag indicating if non-SSL connection should be attempted if SSL connection fails """ # Default to True for SSL and fall_back mode if 'ssl' key DNE if 'ssl' not in secret_dict: return True, True # Handle type bool if isinstance(secret_dict['ssl'], bool): return secret_dict['ssl'], False # Handle type string if isinstance(secret_dict['ssl'], str): ssl = secret_dict['ssl'].lower() if ssl == "true": return True, False elif ssl == "false": return False, False else: # Invalid string value, default to True for both SSL and fall_back mode return True, True # Invalid type, default to True for both SSL and fall_back mode return True, True def connect_and_authenticate(secret_dict, port, dbname, use_ssl): """Attempt to connect and authenticate to a SQL Server DB This helper function tries to connect to the database using connectivity info passed in. If successful, it returns the connection, else None Args: - secret_dict (dict): The Secret Dictionary - port (int): The databse port to connect to - dbname (str): Name of the database - use_ssl (bool): Flag indicating whether connection should use SSL/TLS Returns: Connection: The pymssql.Connection object if successful. None otherwise Raises: KeyError: If the secret json does not contain the expected keys """ # Dynamically set tds configuration based on ssl flag os.environ['FREETDSCONF'] = '/var/task/%s' % 'freetds_ssl.conf' if use_ssl else 'freetds.conf' # Try to obtain a connection to the db try: conn = pymssql.connect(server=secret_dict['host'], user=secret_dict['username'], password=secret_dict['password'], database=dbname, port=port, login_timeout=5, as_dict=True) logger.info("Successfully established %s connection as user '%s' with host: '%s'" % ("SSL/TLS" if use_ssl else "non SSL/TLS", secret_dict['username'], secret_dict['host'])) return conn except pymssql.OperationalError: return None def get_secret_dict(service_client, arn, stage, token=None): """Gets the secret dictionary corresponding for the secret arn, stage, and token This helper function gets credentials for the arn and stage passed in and returns the dictionary by parsing the JSON string 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, or None if no validation is desired stage (string): The stage identifying the secret version Returns: SecretDictionary: Secret dictionary Raises: ResourceNotFoundException: If the secret with the specified arn and stage does not exist ValueError: If the secret is not valid JSON KeyError: If the secret json does not contain the expected keys """ required_fields = ['host', 'username', 'password'] # Only do VersionId validation against the stage if a token is passed in if token: secret = service_client.get_secret_value(SecretId=arn, VersionId=token, VersionStage=stage) else: secret = service_client.get_secret_value(SecretId=arn, VersionStage=stage) plaintext = secret['SecretString'] secret_dict = json.loads(plaintext) # Run validations against the secret if 'engine' not in secret_dict or secret_dict['engine'] != 'sqlserver': raise KeyError("Database engine must be set to 'sqlserver' in order to use this rotation lambda") for field in required_fields: if field not in secret_dict: raise KeyError("%s key is missing from secret JSON" % field) # Parse and return the secret JSON string return secret_dict - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - SecretsManagerRDSSQLServerRotationSingleUser/lambda_function.py [226:451]: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - finally: conn.close() def test_secret(service_client, arn, token): """Test the pending secret against the database This method tries to log into the database with the secrets staged with AWSPENDING and runs a permissions check to ensure the user has the corrrect permissions. 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 valid credentials are found to login to the database KeyError: If the secret json does not contain the expected keys """ # Try to login with the pending secret, if it succeeds, return conn = get_connection(get_secret_dict(service_client, arn, "AWSPENDING", token)) if conn: # This is where the lambda will validate the user's permissions. Uncomment/modify the below lines to # tailor these validations to your needs try: with conn.cursor() as cur: cur.execute("SELECT @@VERSION AS version") finally: conn.close() logger.info("testSecret: Successfully signed into SQL Server DB with AWSPENDING secret in %s." % arn) return else: logger.error("testSecret: Unable to log into database with pending secret of secret ARN %s" % arn) raise ValueError("Unable to log into database with pending secret of secret ARN %s" % arn) def finish_secret(service_client, arn, token): """Finish the rotation by marking the pending secret as current This method finishes the secret rotation by staging the secret staged AWSPENDING with the AWSCURRENT stage. 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 """ # First describe the secret to get the current version metadata = service_client.describe_secret(SecretId=arn) current_version = None for version in metadata["VersionIdsToStages"]: if "AWSCURRENT" in metadata["VersionIdsToStages"][version]: if version == token: # The correct version is already marked as current, return logger.info("finishSecret: Version %s already marked as AWSCURRENT for %s" % (version, arn)) return current_version = version break # Finalize by staging the secret version current service_client.update_secret_version_stage(SecretId=arn, VersionStage="AWSCURRENT", MoveToVersionId=token, RemoveFromVersionId=current_version) logger.info("finishSecret: Successfully set AWSCURRENT stage to version %s for secret %s." % (token, arn)) def get_connection(secret_dict): """Gets a connection to a SQL Server DB from a secret dictionary This helper function uses connectivity information from the secret dictionary to initiate connection attempt(s) to the database. Will attempt a fallback, non-SSL connection when initial connection fails using SSL and fall_back is True. Args: secret_dict (dict): The Secret Dictionary Returns: Connection: The pymssql.Connection object if successful. None otherwise Raises: KeyError: If the secret json does not contain the expected keys """ # Parse and validate the secret JSON string port = str(secret_dict['port']) if 'port' in secret_dict else '1433' dbname = secret_dict['dbname'] if 'dbname' in secret_dict else 'master' # Get SSL connectivity configuration use_ssl, fall_back = get_ssl_config(secret_dict) # if an 'ssl' key is not found or does not contain a valid value, attempt an SSL connection and fall back to non-SSL on failure conn = connect_and_authenticate(secret_dict, port, dbname, use_ssl) if conn or not fall_back: return conn else: return connect_and_authenticate(secret_dict, port, dbname, False) def get_ssl_config(secret_dict): """Gets the desired SSL and fall back behavior using a secret dictionary This helper function uses the existance and value the 'ssl' key in a secret dictionary to determine desired SSL connectivity configuration. Its behavior is as follows: - 'ssl' key DNE or invalid type/value: return True, True - 'ssl' key is bool: return secret_dict['ssl'], False - 'ssl' key equals "true" ignoring case: return True, False - 'ssl' key equals "false" ignoring case: return False, False Args: secret_dict (dict): The Secret Dictionary Returns: Tuple(use_ssl, fall_back): SSL configuration - use_ssl (bool): Flag indicating if an SSL connection should be attempted - fall_back (bool): Flag indicating if non-SSL connection should be attempted if SSL connection fails """ # Default to True for SSL and fall_back mode if 'ssl' key DNE if 'ssl' not in secret_dict: return True, True # Handle type bool if isinstance(secret_dict['ssl'], bool): return secret_dict['ssl'], False # Handle type string if isinstance(secret_dict['ssl'], str): ssl = secret_dict['ssl'].lower() if ssl == "true": return True, False elif ssl == "false": return False, False else: # Invalid string value, default to True for both SSL and fall_back mode return True, True # Invalid type, default to True for both SSL and fall_back mode return True, True def connect_and_authenticate(secret_dict, port, dbname, use_ssl): """Attempt to connect and authenticate to a SQL Server DB This helper function tries to connect to the database using connectivity info passed in. If successful, it returns the connection, else None Args: - secret_dict (dict): The Secret Dictionary - port (int): The databse port to connect to - dbname (str): Name of the database - use_ssl (bool): Flag indicating whether connection should use SSL/TLS Returns: Connection: The pymssql.Connection object if successful. None otherwise Raises: KeyError: If the secret json does not contain the expected keys """ # Dynamically set tds configuration based on ssl flag os.environ['FREETDSCONF'] = '/var/task/%s' % 'freetds_ssl.conf' if use_ssl else 'freetds.conf' # Try to obtain a connection to the db try: conn = pymssql.connect(server=secret_dict['host'], user=secret_dict['username'], password=secret_dict['password'], database=dbname, port=port, login_timeout=5, as_dict=True) logger.info("Successfully established %s connection as user '%s' with host: '%s'" % ("SSL/TLS" if use_ssl else "non SSL/TLS", secret_dict['username'], secret_dict['host'])) return conn except pymssql.OperationalError: return None def get_secret_dict(service_client, arn, stage, token=None): """Gets the secret dictionary corresponding for the secret arn, stage, and token This helper function gets credentials for the arn and stage passed in and returns the dictionary by parsing the JSON string 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, or None if no validation is desired stage (string): The stage identifying the secret version Returns: SecretDictionary: Secret dictionary Raises: ResourceNotFoundException: If the secret with the specified arn and stage does not exist ValueError: If the secret is not valid JSON """ required_fields = ['host', 'username', 'password'] # Only do VersionId validation against the stage if a token is passed in if token: secret = service_client.get_secret_value(SecretId=arn, VersionId=token, VersionStage=stage) else: secret = service_client.get_secret_value(SecretId=arn, VersionStage=stage) plaintext = secret['SecretString'] secret_dict = json.loads(plaintext) # Run validations against the secret if 'engine' not in secret_dict or secret_dict['engine'] != 'sqlserver': raise KeyError("Database engine must be set to 'sqlserver' in order to use this rotation lambda") for field in required_fields: if field not in secret_dict: raise KeyError("%s key is missing from secret JSON" % field) # Parse and return the secret JSON string return secret_dict - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -