FastAPI中PostgreSQL加密列搜索时的排序异常问题

问题根源 #

你当前的排序逻辑是直接对数据库中存储的加密后字段进行排序，而StringEncryptedType默认使用的Fernet加密是带随机向量的，加密后的字符串和明文的字符顺序完全无关，自然得不到预期的排序结果。另外，直接在SQLAlchemy查询中调用本地fernet.decrypt会报错，因为Customer.first_name是ORM列对象，不是实际的字符串/字节数据，无法被本地解密函数处理。

解决方案 #

方案一：新增确定性加密的排序辅助列（推荐，性能最优）

由于随机加密的字段无法直接用于排序，我们可以新增一个使用确定性加密算法的辅助列，保证同一明文加密后结果固定，这样排序逻辑就和明文一致。

1. 自定义确定性加密引擎（基于AES-CBC固定IV）：

from sqlalchemy_utils import StringEncryptedType
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend

class DeterministicAESEngine:
    def __init__(self, key):
        # 截取密钥前32字节用于AES-256
        self.key = key.encode('utf-8')[:32].ljust(32, b'\x00')
        # 固定IV（仅用于排序辅助列，不要用于敏感主数据存储）
        self.fixed_iv = b'your-fixed-16-byte-iv'  # 必须是16字节长度

    def encrypt(self, value):
        if value is None:
            return None
        cipher = Cipher(algorithms.AES(self.key), modes.CBC(self.fixed_iv), backend=default_backend())
        encryptor = cipher.encryptor()
        # PKCS7填充，确保数据长度为16的倍数
        padding_len = 16 - len(value.encode('utf-8')) % 16
        padded_value = value.encode('utf-8') + bytes([padding_len]) * padding_len
        return encryptor.update(padded_value) + encryptor.finalize()

    def decrypt(self, value):
        if value is None:
            return None
        cipher = Cipher(algorithms.AES(self.key), modes.CBC(self.fixed_iv), backend=default_backend())
        decryptor = cipher.decryptor()
        decrypted = decryptor.update(value) + decryptor.finalize()
        # 去除PKCS7填充
        padding_len = decrypted[-1]
        return decrypted[:-padding_len].decode('utf-8')

2. 在模型中添加排序辅助列：

first_name = sa.Column(
    StringEncryptedType(sa.Unicode, settings.encryption_key), nullable=True
)
# 新增排序辅助列
first_name_sort = sa.Column(
    StringEncryptedType(sa.Unicode, settings.encryption_key, engine=DeterministicAESEngine),
    nullable=True
)

3. 保存数据时同步更新辅助列：

# 创建/更新Customer对象时
customer = Customer(first_name="John Doe")
customer.first_name_sort = customer.first_name  # 同步明文值到辅助列
db.session.add(customer)
db.session.commit()

4. 排序时使用辅助列：

elif sort_by == SortByEnum.first_name._value_:
    if sort == SortEnum.desc._value_:
        query = query.order_by(Customer.first_name_sort.desc())
    else:
        query = query.order_by(Customer.first_name_sort.asc())

方案二：查询后在Python层排序（适合小数据量）

如果不想修改数据库结构，可以先将符合条件的数据全部查询到内存，利用SQLAlchemy自动解密的特性，直接对明文进行排序：

# 先获取所有符合条件的记录
customers = query.all()

# 在Python层面按解密后的明文排序
if sort_by == SortByEnum.first_name._value_:
    if sort == SortEnum.desc._value_:
        customers.sort(key=lambda x: x.first_name or "", reverse=True)
    else:
        customers.sort(key=lambda x: x.first_name or "")

注意：此方法仅适合数据量较小的场景，数据量大时会占用大量内存且性能低下。

方案三：数据库层面创建解密UDF（不推荐，有安全/性能风险）

仅适用于支持用户自定义函数（UDF）的数据库（如PostgreSQL），通过在数据库中创建解密函数实现排序，但会存在密钥泄露、性能低下的问题，不建议生产环境使用。

以PostgreSQL为例：

1. 创建PL/Python解密函数：

CREATE OR REPLACE FUNCTION decrypt_fernet(token text, key text) RETURNS text AS $$
import base64
from cryptography.fernet import Fernet
# 转换密钥为Fernet要求的格式
fernet_key = base64.urlsafe_b64encode(key.encode('utf-8').ljust(32)[:32])
f = Fernet(fernet_key)
return f.decrypt(token.encode('utf-8')).decode('utf-8')
$$ LANGUAGE plpython3u;

2. 在SQLAlchemy中调用函数排序：

from sqlalchemy import func

elif sort_by == SortByEnum.first_name._value_:
    decrypt_func = func.decrypt_fernet(Customer.first_name, settings.encryption_key)
    query = query.order_by(decrypt_func.desc()) if sort == SortEnum.desc._value_ else query.order_by(decrypt_func.asc())

内容的提问来源于stack exchange，提问作者kah