Python脚本如何操作数据库加密列

本文目录导读：

1. 应用层加密（推荐方式）
2. 使用SQLAlchemy ORM + 自定义类型
3. 使用MySQL内置加密函数
4. 最佳实践示例
5. 安全注意事项

我来详细介绍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')

1. 应用层加密：最灵活，推荐使用 cryptography 库
2. ORM加密：使用SQLAlchemy自定义类型
3. 数据库函数：使用MySQL的AES_ENCRYPT/AES_DECRYPT
4. 密钥管理：从环境变量或密钥管理服务获取
5. 性能考虑：加密会增加查询和存储开销

选择哪种方法取决于你的具体需求、安全要求和性能考虑。