Closed 0xjjoyy closed 5 months ago
This is my get_new_connection method for illustration.
def get_new_connection(self, conn_params):
try:
logger.info("get connection")
self.get_conn_params_from_secrets_manager(conn_params)
conn =super(DatabaseWrapper,self).get_new_connection(conn_params)
return conn
except MySQLdb.OperationalError as e:
error_code=e.args[0]
if error_code!=1045:
raise e
logger.info("Authentication error. Going to refresh secret and try again.")
secret_cache_item=self.cache_secrets_manager._get_cached_secret('test/appbeta/mysql')
secret_cache_item._refresh_needed=True
secret_cache_item._execute_refresh()
self.get_conn_params_from_secrets_manager(conn_params)
conn =super(DatabaseWrapper,self).get_new_connection(conn_params)
return conn
This is my get_new_connection method for illustration.
def get_new_connection(self, conn_params): try: logger.info("get connection") self.get_conn_params_from_secrets_manager(conn_params) conn =super(DatabaseWrapper,self).get_new_connection(conn_params) return conn except MySQLdb.OperationalError as e: error_code=e.args[0] if error_code!=1045: raise e logger.info("Authentication error. Going to refresh secret and try again.") secret_cache_item=self.cache_secrets_manager._get_cached_secret('test/appbeta/mysql') secret_cache_item._refresh_needed=True secret_cache_item._execute_refresh() self.get_conn_params_from_secrets_manager(conn_params) conn =super(DatabaseWrapper,self).get_new_connection(conn_params) return conn
@0xjjoyy Thanks for that Does it work for you allow you to get the latest secret after rotate ? I tried to do the same with this code below and it doesn't bring the latest secret after rotate
`
client = botocore.session.get_session().create_client('secretsmanager', region_name=region_name) cache_secrets_manager = SecretCache(config=SecretCacheConfig(), client=client) secret_cache_item=cache_secrets_manager._get_cached_secret(secret_name) secret_cache_item._refresh_needed = True secret_cache_item._execute_refresh() db_secret = json.loads(cache_secrets_manager.get_secret_string(secret_name))
`
Also doing this custom db doesn't do the work
import logging
import botocore
import botocore.session
from aws_secretsmanager_caching import SecretCache, SecretCacheConfig
from django.db.backends.postgresql import base
import json
logger = logging.getLogger(__name__)
class DatabaseCredentials:
def __init__(self):
logger.info("init secrets manager database credentials")
client = botocore.session.get_session().create_client('secretsmanager')
cache_config = SecretCacheConfig()
self.cache_secrets_manager = SecretCache(config=cache_config, client=client)
self.secret_id = "pg-prod-cred"
def get_conn_params_from_secrets_manager(self):
secret_json = self.cache_secrets_manager.get_secret_string(self.secret_id)
secret_dict = json.loads(secret_json)
return secret_dict
def refresh_now(self):
secret_cache_item = self.cache_secrets_manager._get_cached_secret(self.secret_id)
secret_cache_item._refresh_needed = True
secret_cache_item._execute_refresh()
print("refresh_now")
databasecredentials = DatabaseCredentials()
class DatabaseWrapper(base.DatabaseWrapper):
def get_new_connection(self, conn_params):
try:
logger.info("get connection")
databasecredentials.get_conn_params_from_secrets_manager()
conn = super(DatabaseWrapper, self).get_new_connection(conn_params)
return conn
except DatabaseWrapper.Database.OperationalError as e:
error_code = e.args[0]
if error_code != 1045:
raise e
logger.info("Authentication error. Going to refresh secret and try again.")
databasecredentials.refresh_now()
conn_params = databasecredentials.get_conn_params_from_secrets_manager()
conn = super(DatabaseWrapper, self).get_new_connection(conn_params)
logger.info("Successfully refreshed secret and established new database connection.")
return conn
Please consider to add the a public method to allow the ability to refresh a secret on demand. Similar to how the Java cache library has a refreshNow() method.
Currently the python methods are 'internal' class methods (though still accessible).
I can do the following to invoke a refresh, though would be more straightforward if it was supported by the class method.