本文目录导读：
我来详细介绍Python操作数据库加密列的几种常见方法：
应用层加密（推荐方式）
使用Python加密库
import pymysql
from cryptography.fernet import Fernet
import base64
import hashlib
class DatabaseEncryptor:
def __init__(self, encryption_key):
# 生成或使用预设密钥
if isinstance(encryption_key, str):
# 将字符串密钥转换为Fernet格式
key = base64.urlsafe_b64encode(
hashlib.sha256(encryption_key.encode()).digest()
)
else:
key = encryption_key
self.cipher = Fernet(key)
def encrypt_data(self, plaintext):
"""加密数据"""
if plaintext is None:
return None
return self.cipher.encrypt(plaintext.encode()).decode()
def decrypt_data(self, ciphertext):
"""解密数据"""
if ciphertext is None:
return None
return self.cipher.decrypt(ciphertext.encode()).decode()
# 使用示例
class UserDatabase:
def __init__(self, connection, encryptor):
self.conn = connection
self.encryptor = encryptor
def insert_encrypted_user(self, user_data):
cursor = self.conn.cursor()
# 加密敏感字段
encrypted_email = self.encryptor.encrypt_data(user_data['email'])
encrypted_phone = self.encryptor.encrypt_data(user_data['phone'])
sql = """INSERT INTO users (username, email_encrypted, phone_encrypted)
VALUES (%s, %s, %s)"""
cursor.execute(sql, (
user_data['username'],
encrypted_email,
encrypted_phone
))
self.conn.commit()
def get_decrypted_user(self, user_id):
cursor = self.conn.cursor()
cursor.execute("SELECT username, email_encrypted, phone_encrypted FROM users WHERE id = %s", (user_id,))
row = cursor.fetchone()
if row:
return {
'username': row[0],
'email': self.encryptor.decrypt_data(row[1]),
'phone': self.encryptor.decrypt_data(row[2])
}
return None
使用SQLAlchemy ORM + 自定义类型
from sqlalchemy import create_engine, Column, String, Integer
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from cryptography.fernet import Fernet
import json
Base = declarative_base()
class EncryptedString:
"""自定义加密字段类型"""
impl = String
def __init__(self, encryption_key):
self.cipher = Fernet(encryption_key)
def process_bind_param(self, value, dialect):
"""存储前加密"""
if value is not None:
return self.cipher.encrypt(value.encode()).decode()
return None
def process_result_value(self, value, dialect):
"""读取时解密"""
if value is not None:
return self.cipher.decrypt(value.encode()).decode()
return None
# 定义模型
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
username = Column(String(50))
# 使用自定义加密类型
@property
def email(self):
return self._email_decrypted
@email.setter
def email(self, value):
self._email_encrypted = self.encrypt(value)
self._email_decrypted = value
# 或者使用混合属性
from sqlalchemy.ext.hybrid import hybrid_property
@hybrid_property
def encrypted_email(self):
return self._email_encrypted
# 完整实现示例
from sqlalchemy import event
from sqlalchemy.orm import mapper
class EncryptedModel:
"""加密模型基类"""
def __init__(self, encryption_key):
self.encryptor = Fernet(encryption_key)
def encrypt(self, value):
if value is None:
return None
return self.encryptor.encrypt(value.encode()).decode()
def decrypt(self, value):
if value is None:
return None
return self.encryptor.decrypt(value.encode()).decode()
# 使用示例
def create_user_with_encryption():
engine = create_engine('mysql+pymysql://user:pass@localhost/db')
Session = sessionmaker(bind=engine)
session = Session()
# 初始化加密器
encryptor = EncryptedModel(b'your-encryption-key-here')
# 创建用户
new_user = User()
new_user.username = 'test_user'
new_user._email_encrypted = encryptor.encrypt('test@example.com')
session.add(new_user)
session.commit()
使用MySQL内置加密函数
import pymysql
from pymysql.cursors import DictCursor
class MySQLEncryptionExample:
def __init__(self, connection_config):
self.connection = pymysql.connect(**connection_config)
def insert_with_aes_encrypt(self, data):
cursor = self.connection.cursor()
sql = """INSERT INTO encrypted_table
(encrypted_data)
VALUES (AES_ENCRYPT(%s, 'your-secret-key'))"""
cursor.execute(sql, (data['sensitive_info'],))
self.connection.commit()
def read_with_aes_decrypt(self, record_id):
cursor = self.connection.cursor(DictCursor)
sql = """SELECT id,
AES_DECRYPT(encrypted_data, 'your-secret-key') as decrypted_data
FROM encrypted_table
WHERE id = %s"""
cursor.execute(sql, (record_id,))
result = cursor.fetchone()
if result:
# 注意：AES_DECRYPT返回bytes类型
result['decrypted_data'] = result['decrypted_data'].decode('utf-8')
return result
最佳实践示例
import os
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
import base64
import pymysql
from pymysql.cursors import DictCursor
class SecureDatabaseHandler:
def __init__(self, db_config, encryption_key_env_var='DB_ENCRYPTION_KEY'):
# 从环境变量获取密钥
encryption_key = os.environ.get(encryption_key_env_var)
if not encryption_key:
raise ValueError(f"Environment variable {encryption_key_env_var} not set")
# 生成密钥
self.encryption_key = self.derive_key(encryption_key)
self.cipher = Fernet(self.encryption_key)
# 数据库连接
self.connection = pymysql.connect(**db_config)
def derive_key(self, password, salt=None):
"""从密码派生加密密钥"""
if salt is None:
salt = os.urandom(16)
kdf = PBKDF2HMAC(
algorithm=hashes.SHA256(),
length=32,
salt=salt,
iterations=100000,
)
key = base64.urlsafe_b64encode(kdf.derive(password.encode()))
return key
def rotate_encryption_key(self, new_key):
"""轮换加密密钥"""
cursor = self.connection.cursor(DictCursor)
# 获取所有加密数据
cursor.execute("SELECT id, encrypted_data FROM sensitive_table")
older_records = cursor.fetchall()
# 创建新加密器
old_cipher = self.cipher
new_cipher = Fernet(new_key)
# 重新加密所有数据
for record in older_records:
decrypted_data = old_cipher.decrypt(record['encrypted_data'].encode())
re_encrypted_data = new_cipher.encrypt(decrypted_data)
cursor.execute(
"UPDATE sensitive_table SET encrypted_data = %s WHERE id = %s",
(re_encrypted_data.decode(), record['id'])
)
self.connection.commit()
self.cipher = new_cipher
def search_encrypted_field(self, search_value):
"""搜索加密字段（只能精确匹配）"""
cursor = self.connection.cursor(DictCursor)
# 加密搜索值
encrypted_search = self.cipher.encrypt(search_value.encode()).decode()
cursor.execute(
"SELECT * FROM sensitive_table WHERE encrypted_data = %s",
(encrypted_search,)
)
results = cursor.fetchall()
# 解密结果
for result in results:
result['decrypted_data'] = self.cipher.decrypt(
result['encrypted_data'].encode()
).decode()
return results
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if self.connection:
self.connection.close()
安全注意事项
# 密钥管理示例
class KeyManagement:
def __init__(self):
self.key_store = {}
def store_key(self, key_id, key):
"""安全存储密钥"""
# 使用密钥管理服务（如AWS KMS, Azure Key Vault）
# 或加密存储到配置文件
encrypted_key = self.encrypt_with_master_key(key)
self.key_store[key_id] = encrypted_key
def get_key(self, key_id):
"""获取解密后的密钥"""
encrypted_key = self.key_store.get(key_id)
if encrypted_key:
return self.decrypt_with_master_key(encrypted_key)
return None
def rotate_keys(self):
"""定期轮换密钥"""
# 实现密钥轮换逻辑
pass
# 使用环境变量管理密钥
import os
from dotenv import load_dotenv
load_dotenv()
# 数据库配置
db_config = {
'host': os.getenv('DB_HOST'),
'user': os.getenv('DB_USER'),
'password': os.getenv('DB_PASSWORD'),
'database': os.getenv('DB_NAME'),
}
# 加密密钥（从环境变量获取）
encryption_key = os.getenv('ENCRYPTION_KEY')
- 应用层加密：最灵活，推荐使用
cryptography库 - ORM加密：使用SQLAlchemy自定义类型
- 数据库函数：使用MySQL的AES_ENCRYPT/AES_DECRYPT
- 密钥管理：从环境变量或密钥管理服务获取
- 性能考虑：加密会增加查询和存储开销
选择哪种方法取决于你的具体需求、安全要求和性能考虑。
