本文目录导读：
我来介绍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()
关键注意事项
- 密钥管理：密钥必须安全存储，可以使用密钥管理服务（KMS）
- 性能影响：加密/解密操作会影响数据库性能
- 搜索限制：加密字段无法直接搜索，需要特殊处理
- 字段大小：加密后的数据会变大，需要调整字段大小
- 索引限制：加密字段无法建立索引
- 备份恢复：确保备份也能保持加密状态
选择合适的加密方案取决于你的具体需求：
- 需要搜索功能：考虑使用确定性加密或SQL Server的Always Encrypted
- 需要性能优：使用数据库原生TDE
- 需要兼容性：使用应用层加密
建议使用专业的加密密钥管理服务,如AWS KMS、Azure Key Vault或HashiCorp Vault。
