如何在Django中直接传入Vault获取的SSL证书字符串连接MySQL？

可以实现，核心思路是绕开Django MySQL backend默认仅支持文件路径的限制，直接调用mysqlclient底层的ssl_set()方法传入内存中的证书数据。以下是两种可行方案：

方案一：自定义Django数据库Backend（推荐） #

通过继承官方MySQL Backend，重写连接逻辑，直接处理内存中的证书字符串：

1. 创建自定义Backend文件（如myapp/backends.py）：

import MySQLdb
from django.db.backends.mysql.base import DatabaseWrapper

class VaultMySQLWrapper(DatabaseWrapper):
    def get_new_connection(self, conn_params):
        # 提取SSL配置并从参数中移除
        ssl_config = conn_params.pop('ssl', None)
        if ssl_config:
            # 先建立无SSL的基础连接
            conn = MySQLdb.connect(**conn_params, ssl=None)
            # 从Vault拉取的证书字符串转字节
            ca_bytes = ssl_config.get('ca').encode('utf-8') if ssl_config.get('ca') else None
            cert_bytes = ssl_config.get('cert').encode('utf-8') if ssl_config.get('cert') else None
            key_bytes = ssl_config.get('key').encode('utf-8') if ssl_config.get('key') else None
            
            # 调用底层ssl_set传入内存数据，参数顺序：ca, cert, key, capath, cipher
            conn.ssl_set(ca_bytes, cert_bytes, key_bytes, None, None)
            # 验证连接有效性
            conn.query("SET NAMES utf8mb4")
            return conn
        # 无SSL配置时使用默认逻辑
        return super().get_new_connection(conn_params)

2. 修改DATABASES配置，指定自定义Backend：

import os
from yourapp.utils import get_vault_secret

DATABASES = {
    'default': {
        'ENGINE': 'myapp.backends.VaultMySQLWrapper',
        'HOST': os.environ.get('DATABASE_HOST'),
        'USER': os.environ.get('DATABASE_USER'),
        'PASSWORD': get_vault_secret('DATABASE_PASSWORD'),
        'NAME': os.environ.get('DATABASE_NAME'),
        'OPTIONS': {
            'ssl': {
                'ca': get_vault_secret('MYSQL_CA_CERT'),
                'cert': get_vault_secret('MYSQL_CLIENT_CERT'),
                'key': get_vault_secret('MYSQL_CLIENT_KEY')
            }
        }
    }
}

方案二：猴子补丁快速适配 #

如果不想自定义Backend，可通过猴子补丁修改mysqlclient的连接逻辑：

在项目初始化文件（如yourapp/__init__.py或settings.py开头）添加以下代码：

import MySQLdb
from MySQLdb.connections import Connection

original_connect = Connection.connect

def patched_connect(self, *args, **kwargs):
    # 提取SSL配置
    ssl_config = kwargs.pop('ssl', None)
    # 建立无SSL连接
    original_connect(self, *args, **kwargs, ssl=None)
    # 若有SSL配置则传入内存数据
    if ssl_config:
        ca_bytes = ssl_config.get('ca').encode('utf-8') if ssl_config.get('ca') else None
        cert_bytes = ssl_config.get('cert').encode('utf-8') if ssl_config.get('cert') else None
        key_bytes = ssl_config.get('key').encode('utf-8') if ssl_config.get('key') else None
        self.ssl_set(ca_bytes, cert_bytes, key_bytes, None, None)
    return self

Connection.connect = patched_connect

注意事项 #

• 确保使用的mysqlclient版本≥1.4.0，该版本开始支持ssl_set()方法直接传入内存数据
• 若Vault返回的证书是Base64编码格式，需先解码为原始字符串（如base64.b64decode(vault_data).decode('utf-8')）
• 可通过执行SHOW STATUS LIKE 'Ssl_cipher';查询结果验证SSL连接是否生效

内容的提问来源于stack exchange，提问作者Danielle M.