Django + AWS Secret Manager Password Rotation

Question

I have a Django app that fetches DB secret from AWS Secret Manager. It contains all the DB parameters like username, password, host, port, etc. When I start the Django application on EC2, it successfully retrieves the secret from the Secret Manager and establishes a DB connection.

Now the problem is that I have a password rotation policy set for 30 days. To test the flow, at present, I have set it to 1 day. Every time the password rotates, my Django app loses DB connectivity. So, I have to manually restart the application to allow the app to fetch the new DB credentials from the Secret Manager.

Is there a way that secret fetching can happen automatically and without a manual restart of the server.? Once way possibly is to trigger an AWS CodeDeploy or similar service that will restart the server automatically. However, there will be some downtime if I take this approach.

Any other approach that can seamlessly work without any downtime.

Answer 1

You need to alter one function in the default DatabaseWrapper class that Django uses to handle DB connections. The following example is for Postgres. Other databases will be similar.

In your settings.py change the DATABASES variable to allow for custom engine:

DATABASES = {
    'default': {
        'ENGINE': 'custom_postgresql_engine',
        'NAME': 'DEFAULT_NAME',
        'USER': 'DEFAULT_USER',
        'PASSWORD': 'UNROTATED_DEFAULT_PASSWORD',  # The unrotated default password goes here.
        'HOST': 'DEFAULT_HOST',
        'PORT': 'DEFAULT_PORT',
    }
}

Then create a Python package in the root directory of your Django project named custom_postgresql_engine (or whatever you used for the ENGINE variable in your settings.py). Within that directory, there must be two files. An empty __init__.py and a base.py.

root_django_directory:
--project_dir
--app1
--app2
--app3
--custom_postgresql_engine:
-- -- __init__.py
-- -- base.py

Then inside the base.py, you can use the following code.

from django.db.backends.postgresql import base
from django.core.exceptions import ImproperlyConfigured
import json
import boto3


class DatabaseWrapper(base.DatabaseWrapper):
    # This method returns the latest password stored in the secret
    def get_most_recent_password(self):
        try:
            secret_name = 'SECRET_NAME_HERE'
            region_name = 'AWS_REGION_HERE'
            session = boto3.session.Session()
            client = session.client(
                service_name='secretsmanager',
                region_name=region_name
            )
            secrets = client.get_secret_value(SecretId=secret_name)
            password = json.loads(secrets['SecretString'])['password']
        except Exception as e:
            raise Exception('Failed to retrieve credentials from Secrets Manager', e)

        return password

    """
    This method is overriden from the base DatabaseWrapper class
    in a way to allow the usage of dynamic, rotating passwords.
    Note that the if statement that sets the password is commented out
    and a call to get_most_recent_password() is used to fetch the
    latest password from Secrets Manager.
    Everything else remains unchanged from the original code.
    """
    def get_connection_params(self):
        settings_dict = self.settings_dict
        # None may be used to connect to the default 'postgres' db
        if settings_dict['NAME'] == '':
            raise ImproperlyConfigured(
                "settings.DATABASES is improperly configured. "
                "Please supply the NAME value.")
        if len(settings_dict['NAME'] or '') > self.ops.max_name_length():
            raise ImproperlyConfigured(
                "The database name '%s' (%d characters) is longer than "
                "PostgreSQL's limit of %d characters. Supply a shorter NAME "
                "in settings.DATABASES." % (
                    settings_dict['NAME'],
                    len(settings_dict['NAME']),
                    self.ops.max_name_length(),
                )
            )
        conn_params = {
            'database': settings_dict['NAME'] or 'postgres',
            **settings_dict['OPTIONS'],
        }
        conn_params.pop('isolation_level', None)
        if settings_dict['USER']:
            conn_params['user'] = settings_dict['USER']
        # if settings_dict['PASSWORD']:
        #     conn_params['password'] = settings_dict['PASSWORD']
        if settings_dict['HOST']:
            conn_params['host'] = settings_dict['HOST']
        if settings_dict['PORT']:
            conn_params['port'] = settings_dict['PORT']

        conn_params['password'] = self.get_most_recent_password()
        return conn_params