### Python 中使用数据库连接池并实现密码加密 为了确保安全性，在实际应用中不建议直接存储或传输明文密码。对于数据库连接池中的密码管理，可以采用以下方法来处理： #### 使用 DruidEncryptDecrypt 进行加解密操作 通过自定义的 `DruidEncryptDecrypt` 类来进行密码的加密和解密工作。当应用程序启动时，先加载已加密过的配置项，并在建立数据库连接之前完成对敏感字段（如密码）的解密过程。 ```java // Java代码片段展示如何获取经过解密后的密码字符串 public String getPassword() { // 假设这里是从配置文件读取到已经过AES/GCM/PKCS5Padding方式加密后的base64编码串 byte[] encryptedPwdBytes = Base64.getDecoder().decode("encrypted_password_base64_string"); // 创建用于解密的对象实例 Cipher cipher = Cipher.getInstance("AES/GCM/NoPadding"); SecretKeySpec keySpec = new SecretKeySpec(key.getBytes(), "AES"); GCMParameterSpec gcmSpec = new GCMParameterSpec(TAG_LENGTH_BIT, iv); // 初始化cipher对象为解密模式 cipher.init(Cipher.DECRYPT_MODE, keySpec, gcmSpec); try { // 执行解密得到原始byte数组再转换成String返回给调用者 return new String(cipher.doFinal(encryptedPwdBytes), StandardCharsets.UTF_8); } catch (Exception e) { throw new RuntimeException(e.getMessage()); } }[^1] ``` 由于上述例子是以Java语言编写，而在Python环境下则推荐利用第三方库PyCryptodome来代替原生API实现相同功能。下面给出一个完整的解决方案示例: ```python from Crypto.Cipher import AES import base64 import os class PasswordManager: def __init__(self, secret_key=b'your-secret-key'): self.key = secret_key[:32] @staticmethod def pad(s): block_size = AES.block_size padding_length = block_size - len(s) % block_size return s + chr(padding_length) * padding_length @staticmethod def unpad(s): return s[:-ord(s[-1:])] def encrypt(self, raw): raw_padded = PasswordManager.pad(raw).encode() iv = os.urandom(AES.block_size) cipher = AES.new(self.key, AES.MODE_CBC, iv=iv) ciphertext_bytes = cipher.encrypt(raw_padded) combined_iv_ciphertext = iv + ciphertext_bytes encoded_cipher_text = base64.b64encode(combined_iv_ciphertext).decode('utf-8') return encoded_cipher_text def decrypt(self, enc): decoded_encrypted_data = base64.b64decode(enc.encode()) iv = decoded_encrypted_data[:AES.block_size] actual_encrypted_message = decoded_encrypted_data[AES.block_size:] cipher = AES.new(self.key, AES.MODE_CBC, iv=iv) decrypted_padded_msg = cipher.decrypt(actual_encrypted_message) original_message = PasswordManager.unpad(decrypted_padded_msg.decode()) return original_message def create_db_connection_pool(password_manager_instance, config_dict): password = password_manager_instance.decrypt(config_dict['password']) conn_str = f"mysql+pymysql://{config_dict['username']}:{password}@{config_dict['host']}/{config_dict['database']}" engine = create_engine(conn_str, pool_recycle=3600, echo=False) return engine.pool() # Example usage of the above functions with a hypothetical configuration dictionary. if __name__ == "__main__": pm = PasswordManager(secret_key='a'*32) db_config = {'host': 'localhost', 'port': 3306, 'username': 'root', 'password': '<Encrypted_Password>', 'database': 'test'} connection_pool = create_db_connection_pool(pm, db_config)[^2][^3] ``` 在这个案例里，`PasswordManager` 类负责提供安全可靠的加解密服务；而函数 `create_db_connection_pool()` 则展示了怎样结合 SQLAlchemy ORM 和 PyMySQL 来构建带有自动重连特性的持久化连接池。注意这里的 `<Encrypted_Password>` 应该替换为你自己事先保存好的经由 `encrypt()` 函数产生的密文版本的真实 MySQL root 用户账户对应的密码值。