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

本文目录导读：

1. 使用加密库对数据进行透明加密
2. 使用 SQLAlchemy 实现透明加密
3. 使用数据库原生加密功能
4. 使用 SQLCipher 实现 SQLite 加密
5. 实现自动加密/解密装饰器
6. 关键注意事项

我来介绍Python操作数据库透明加密的几种方法：

使用加密库对数据进行透明加密

使用 cryptography 库实现列级加密

from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
import base64
import mysql.connector
import sqlite3
class DatabaseTransparentEncryption:
    def __init__(self, encryption_key):
        """初始化加密器"""
        self.cipher = Fernet(encryption_key)
    @staticmethod
    def generate_key(password: str, salt: bytes = 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, salt
    def encrypt_data(self, data: str) -> bytes:
        """加密数据"""
        if isinstance(data, str):
            data = data.encode()
        return self.cipher.encrypt(data)
    def decrypt_data(self, encrypted_data: bytes) -> str:
        """解密数据"""
        return self.cipher.decrypt(encrypted_data).decode()
    def save_encrypted_user(self, db_connection, user_data):
        """保存加密的用户数据"""
        encrypted_name = self.encrypt_data(user_data['name'])
        encrypted_email = self.encrypt_data(user_data['email'])
        cursor = db_connection.cursor()
        sql = """
        INSERT INTO users_encrypted (id, encrypted_name, encrypted_email, created_at)
        VALUES (%s, %s, %s, NOW())
        """
        cursor.execute(sql, (user_data['id'], encrypted_name, encrypted_email))
        db_connection.commit()
    def get_decrypted_user(self, db_connection, user_id):
        """获取解密后的用户数据"""
        cursor = db_connection.cursor(dictionary=True)
        sql = "SELECT * FROM users_encrypted WHERE id = %s"
        cursor.execute(sql, (user_id,))
        encrypted_user = cursor.fetchone()
        if encrypted_user:
            return {
                'id': encrypted_user['id'],
                'name': self.decrypt_data(encrypted_user['encrypted_name']),
                'email': self.decrypt_data(encrypted_user['encrypted_email']),
                'created_at': encrypted_user['created_at']
            }
        return None
# 使用示例
def example_usage():
    # 生成密钥
    from cryptography.fernet import Fernet
    key = Fernet.generate_key()
    # 或者从密码生成
    password = "my_secure_password"
    key, salt = DatabaseTransparentEncryption.generate_key(password)
    # 初始化加密器
    encryptor = DatabaseTransparentEncryption(key)
    # 连接数据库
    db_config = {
        'host': 'localhost',
        'user': 'root',
        'password': 'password',
        'database': 'test_db'
    }
    connection = mysql.connector.connect(**db_config)
    # 保存加密数据
    user_data = {
        'id': 1,
        'name': '张三',
        'email': 'zhangsan@example.com'
    }
    encryptor.save_encrypted_user(connection, user_data)
    # 读取解密数据
    decrypted_user = encryptor.get_decrypted_user(connection, 1)
    print(f"解密后的用户: {decrypted_user}")

使用 SQLAlchemy 实现透明加密

from sqlalchemy import create_engine, Column, Integer, String, DateTime, LargeBinary
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from cryptography.fernet import Fernet
from datetime import datetime
import json
Base = declarative_base()
class EncryptedField:
    """自定义加密字段类型"""
    def __init__(self, cipher):
        self.cipher = cipher
    def encrypt(self, value):
        if value is None:
            return None
        if isinstance(value, dict):
            value = json.dumps(value)
        return self.cipher.encrypt(value.encode())
    def decrypt(self, encrypted_value):
        if encrypted_value is None:
            return None
        decrypted = self.cipher.decrypt(encrypted_value).decode()
        try:
            return json.loads(decrypted)
        except json.JSONDecodeError:
            return decrypted
class EncryptedUser(Base):
    """加密的用户模型"""
    __tablename__ = 'encrypted_users'
    id = Column(Integer, primary_key=True)
    encrypted_name = Column(LargeBinary, nullable=False)
    encrypted_email = Column(LargeBinary, nullable=False)
    encrypted_phone = Column(LargeBinary, nullable=True)
    created_at = Column(DateTime, default=datetime.utcnow)
class UserService:
    def __init__(self, database_url: str, encryption_key: bytes):
        self.engine = create_engine(database_url)
        self.Session = sessionmaker(bind=self.engine)
        self.cipher = Fernet(encryption_key)
        self.encryptor = EncryptedField(self.cipher)
        # 创建表
        Base.metadata.create_all(self.engine)
    def create_user(self, name: str, email: str, phone: str = None):
        """创建加密用户"""
        session = self.Session()
        try:
            user = EncryptedUser(
                encrypted_name=self.encryptor.encrypt(name),
                encrypted_email=self.encryptor.encrypt(email),
                encrypted_phone=self.encryptor.encrypt(phone) if phone else None
            )
            session.add(user)
            session.commit()
            return user.id
        finally:
            session.close()
    def get_user(self, user_id: int):
        """获取解密用户"""
        session = self.Session()
        try:
            user = session.query(EncryptedUser).filter_by(id=user_id).first()
            if user:
                return {
                    'id': user.id,
                    'name': self.encryptor.decrypt(user.encrypted_name),
                    'email': self.encryptor.decrypt(user.encrypted_email),
                    'phone': self.encryptor.decrypt(user.encrypted_phone) if user.encrypted_phone else None,
                    'created_at': user.created_at
                }
            return None
        finally:
            session.close()
    def search_users_by_email(self, email_prefix: str):
        """通过邮箱前缀搜索（注意：无法直接搜索加密字段）"""
        session = self.Session()
        try:
            users = session.query(EncryptedUser).all()
            result = []
            for user in users:
                decrypted_email = self.encryptor.decrypt(user.encrypted_email)
                if decrypted_email and decrypted_email.startswith(email_prefix):
                    result.append({
                        'id': user.id,
                        'name': self.encryptor.decrypt(user.encrypted_name),
                        'email': decrypted_email
                    })
            return result
        finally:
            session.close()
# 使用示例
def sqlalchemy_example():
    # 配置
    DATABASE_URL = "mysql://root:password@localhost/test_db"
    KEY = Fernet.generate_key()
    # 初始化服务
    service = UserService(DATABASE_URL, KEY)
    # 创建用户
    user_id = service.create_user(
        name="李四",
        email="lisi@example.com",
        phone="13800138000"
    )
    print(f"创建用户ID: {user_id}")
    # 获取用户
    user = service.get_user(user_id)
    print(f"解密用户: {user}")

使用数据库原生加密功能

import mysql.connector
from mysql.connector import errorcode
class MySQLEncryptionManager:
    """MySQL数据库透明加密管理"""
    def __init__(self, connection_config):
        self.config = connection_config
    def enable_tablespace_encryption(self):
        """启用表空间加密（MySQL 8.0+）"""
        connection = mysql.connector.connect(**self.config)
        cursor = connection.cursor()
        try:
            # 创建加密表空间
            cursor.execute("""
                CREATE TABLESPACE `encrypted_ts`
                ADD DATAFILE 'encrypted_ts.ibd'
                ENCRYPTION = 'Y'
                DEFAULT ENCRYPTION = 'Y'
            """)
            # 在加密表空间中创建表
            cursor.execute("""
                CREATE TABLE if not exists `sensitive_data` (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    credit_card_number VARCHAR(255),
                    ssn VARCHAR(255),
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                ) TABLESPACE = `encrypted_ts`
            """)
            connection.commit()
            print("表空间加密已启用")
        except mysql.connector.Error as err:
            print(f"错误: {err}")
            connection.rollback()
        finally:
            cursor.close()
            connection.close()
    def use_transparent_data_encryption(self):
        """使用MySQL透明数据加密（TDE）"""
        connection = mysql.connector.connect(**self.config)
        cursor = connection.cursor()
        try:
            # 检查是否支持TDE
            cursor.execute("SELECT @@innodb_redo_log_encrypt")
            redo_encrypt = cursor.fetchone()[0]
            if redo_encrypt == 1:
                print("MySQL TDE已启用")
            # 创建加密表
            cursor.execute("""
                CREATE TABLE if not exists `encrypted_transactions` (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    amount DECIMAL(10,2),
                    account_number VARCHAR(255),
                    transaction_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                ) ENCRYPTION='Y'
            """)
            connection.commit()
            print("加密表已创建")
        except mysql.connector.Error as err:
            print(f"错误: {err}")
        finally:
            cursor.close()
            connection.close()
# 使用示例
def mysql_encryption_example():
    config = {
        'user': 'root',
        'password': 'password',
        'host': 'localhost',
        'database': 'test_db',
        'raise_on_warnings': True
    }
    # 需要先配置MySQL密钥环插件
    # 在MySQL中执行:
    # INSTALL PLUGIN keyring_file SONAME 'keyring_file.so';
    # SET GLOBAL keyring_file_data = '/var/lib/mysql-keyring/keyring';
    manager = MySQLEncryptionManager(config)
    manager.enable_tablespace_encryption()

使用 SQLCipher 实现 SQLite 加密

import sqlite3
from pysqlcipher3 import dbapi2 as sqlcipher
class SQLiteEncryptionManager:
    """使用SQLCipher加密SQLite数据库"""
    def __init__(self, db_path: str, password: str):
        self.db_path = db_path
        self.password = password
    def create_encrypted_database(self):
        """创建加密的SQLite数据库"""
        connection = sqlcipher.connect(self.db_path)
        cursor = connection.cursor()
        # 设置加密密钥
        cursor.execute(f"PRAGMA key = '{self.password}'")
        # 创建表（数据会自动加密存储）
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS secure_data (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                username TEXT NOT NULL,
                password_hash TEXT NOT NULL,
                personal_info BLOB,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        connection.commit()
        connection.close()
        print(f"加密数据库已创建: {self.db_path}")
    def insert_secure_data(self, data: dict):
        """插入加密数据"""
        connection = sqlcipher.connect(self.db_path)
        cursor = connection.cursor()
        # 输入密码
        cursor.execute(f"PRAGMA key = '{self.password}'")
        cursor.execute("""
            INSERT INTO secure_data (username, password_hash, personal_info)
            VALUES (?, ?, ?)
        """, (data['username'], data['password_hash'], data.get('personal_info')))
        connection.commit()
        connection.close()
    def query_secure_data(self, username: str):
        """查询加密数据"""
        connection = sqlcipher.connect(self.db_path)
        cursor = connection.cursor()
        cursor.execute(f"PRAGMA key = '{self.password}'")
        cursor.execute("SELECT * FROM secure_data WHERE username = ?", (username,))
        result = cursor.fetchall()
        connection.close()
        return result
# 使用示例
def sqlcipher_example():
    manager = SQLiteEncryptionManager("encrypted.db", "my_strong_password")
    # 创建加密数据库
    manager.create_encrypted_database()
    # 插入数据
    manager.insert_secure_data({
        'username': 'admin',
        'password_hash': 'hashed_password_here',
        'personal_info': b'{"email": "admin@example.com", "phone": "123456789"}'
    })
    # 查询数据
    result = manager.query_secure_data('admin')
    print(f"查询结果: {result}")

实现自动加密/解密装饰器

from functools import wraps
from cryptography.fernet import Fernet
import sqlite3
import json
class AutoEncryptDecrypt:
    """自动加密/解密装饰器"""
    def __init__(self, cipher, encrypted_fields):
        self.cipher = cipher
        self.encrypted_fields = encrypted_fields
    def encrypt_record(self, func):
        """装饰器：自动加密插入/更新操作的敏感字段"""
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 获取要操作的记录
            if 'data' in kwargs:
                data = kwargs['data'].copy()
                # 加密敏感字段
                for field in self.encrypted_fields:
                    if field in data and data[field]:
                        data[field] = self.cipher.encrypt(
                            str(data[field]).encode()
                        )
                kwargs['data'] = data
            return func(*args, **kwargs)
        return wrapper
    def decrypt_record(self, func):
        """装饰器：自动解密查询结果的敏感字段"""
        @wraps(func)
        def wrapper(*args, **kwargs):
            result = func(*args, **kwargs)
            if isinstance(result, list):
                # 处理多条记录
                for record in result:
                    if isinstance(record, dict):
                        for field in self.encrypted_fields:
                            if field in record and record[field]:
                                try:
                                    record[field] = self.cipher.decrypt(
                                        record[field]
                                    ).decode()
                                except:
                                    pass
                return result
            elif isinstance(result, dict):
                # 处理单条记录
                for field in self.encrypted_fields:
                    if field in result and result[field]:
                        try:
                            result[field] = self.cipher.decrypt(
                                result[field]
                            ).decode()
                        except:
                            pass
                return result
            return result
        return wrapper
# 使用示例
class EncryptedDatabase:
    def __init__(self, db_path: str, encryption_key: bytes):
        self.connection = sqlite3.connect(db_path)
        self.cipher = Fernet(encryption_key)
        # 创建自动加密/解密处理器
        self.encryptor = AutoEncryptDecrypt(
            self.cipher,
            ['name', 'email', 'phone']  # 要加密的字段
        )
        self._create_tables()
    def _create_tables(self):
        cursor = self.connection.cursor()
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS users (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name BLOB,
                email BLOB,
                phone BLOB,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        self.connection.commit()
    @AutoEncryptDecrypt(None, ['name', 'email', 'phone']).encrypt_record
    def insert_user(self, data: dict):
        cursor = self.connection.cursor()
        cursor.execute(
            "INSERT INTO users (name, email, phone) VALUES (?, ?, ?)",
            (data['name'], data['email'], data['phone'])
        )
        self.connection.commit()
        return cursor.lastrowid
    @AutoEncryptDecrypt(None, ['name', 'email', 'phone']).decrypt_record
    def get_user(self, user_id: int):
        cursor = self.connection.cursor()
        cursor.execute(
            "SELECT id, name, email, phone FROM users WHERE id = ?",
            (user_id,)
        )
        row = cursor.fetchone()
        if row:
            return {
                'id': row[0],
                'name': row[1],
                'email': row[2],
                'phone': row[3],
            }
        return None
# 主程序
def main():
    # 生成密钥
    from cryptography.fernet import Fernet
    key = Fernet.generate_key()
    # 使用加密数据库
    db = EncryptedDatabase("secure_data.db", key)
    # 插入加密数据
    user_id = db.insert_user({
        'name': '王五',
        'email': 'wangwu@example.com',
        'phone': '13900139000'
    })
    print(f"用户ID: {user_id}")
    # 读取解密数据
    user = db.get_user(user_id)
    print(f"解密用户: {user}")
if __name__ == "__main__":
    main()

关键注意事项

1. 密钥管理：密钥必须安全存储，可以使用密钥管理服务（KMS）
2. 性能影响：加密/解密操作会影响数据库性能
3. 搜索限制：加密字段无法直接搜索，需要特殊处理
4. 字段大小：加密后的数据会变大，需要调整字段大小
5. 索引限制：加密字段无法建立索引
6. 备份恢复：确保备份也能保持加密状态

选择合适的加密方案取决于你的具体需求：

• 需要搜索功能：考虑使用确定性加密或SQL Server的Always Encrypted
• 需要性能优：使用数据库原生TDE
• 需要兼容性：使用应用层加密

建议使用专业的加密密钥管理服务,如AWS KMS、Azure Key Vault或HashiCorp Vault。