Python Flask下用pycrypto解密数据实现SQLAlchemy查询过滤

当前PostgreSQL库中同时存在两类数据：未加密的存量历史数据、后续新写入的加密数据，全量存量数据回溯加密完成前，无法对整列做统一加密处理，加解密逻辑只能在业务侧实现。
目前用pycrypto实现的入库加密、读取后解密流程运行正常，但做入库前去重查询时遇到障碍：非加密场景下通过Item.item_code == new_item_code的等值查询即可判断数据是否存在，但当前加密方案每次生成的密文均不相同，无法直接通过密文匹配完成校验。
已尝试的方案均存在缺陷：

• 尝试用SQLAlchemy的@hybrid_property实现字段解密，因执行阶段逻辑不匹配，操作对象为字段属性而非实际存储值，触发解码错误
• 尝试全表拉取item_code字段到内存批量解密后做匹配，性能极差，数据量增长后存在明显瓶颈，不具备生产可用性
  核心诉求为找到可落地的实现方式，支持解密表中字段值后直接用于SQLAlchemy的查询过滤逻辑，兼顾性能和过渡阶段的存量数据兼容性。

核心矛盾是当前使用的随机IV加密方案天生不支持密文直接等值匹配，以下两个方案按推荐优先级排序：

方案1：确定性加密盲索引（性能最优，生产环境首选） #

当前加密后密文每次不一致，本质是加密时用了随机生成的IV，这种模式安全性高但没法直接做等值匹配。针对需要查询的字段单独加一个盲索引列即可解决，不需要改动现有加密存储逻辑：

• 给表新增item_code_bidx列，字段类型和原item_code字段保持一致
• 所有新写入数据时，除了按原有逻辑生成随机IV的高安全性密文存入item_code，额外用固定密钥、固定IV的确定性加密算法对同一个明文生成密文，存入item_code_bidx列。确定性加密的特性是同一个明文永远生成相同密文，支持直接等值匹配
• 过渡阶段兼容存量数据：存量未加密数据的item_code_bidx列留空，查询时用OR条件同时覆盖两类数据：
  • 新数据：把待查询的明文用相同的确定性加密规则生成匹配密文，直接和item_code_bidx做等值匹配
  • 存量数据：item_code_bidx为空的行，直接用原item_code明文和待查询值匹配
• 给item_code_bidx加上普通B树索引，查询性能和明文查询几乎没有差异。等后续全量存量数据完成加密回填、所有行的item_code_bidx都有值之后，就可以去掉存量明文匹配的分支，逻辑更简洁
• 注意不要用item_code_bidx存储的数据做业务返回，这个列只用来做查询匹配，业务侧读取数据还是解密原item_code列的随机IV密文，整体安全性不会有明显损失

对应SQLAlchemy查询代码示例：

from sqlalchemy import or_

new_item_code = "待校验的编码值"
# 用和写入时一致的确定性加密逻辑生成匹配密文
match_ciphertext = deterministic_aes_encrypt(new_item_code, fixed_key, fixed_iv)

exist_item = Item.query.filter(
    or_(
        Item.item_code_bidx == match_ciphertext,
        Item.item_code_bidx.is_(None),
        Item.item_code == new_item_code
    )
).first()

方案2：PostgreSQL服务端解密计算（无需改表结构，适合小数据量场景） #

如果不想新增列，可以在数据库端用pgcrypto扩展实现和pycrypto逻辑完全一致的解密函数，查询时直接在SQL层面逐行解密后做匹配：

• 先给PostgreSQL安装启用pgcrypto扩展，自定义解密函数，保证函数的解密逻辑、使用的密钥和Python侧pycrypto的逻辑完全一致，解密结果和Python侧解密结果完全相同
• 查询时调用SQL函数逐行解密字段值，和传入的明文做匹配，同时兼容存量明文数据
• 这个方案的缺点是无法利用索引，数据量超过十万级之后查询延迟会明显升高，只适合数据规模不大的场景临时过渡使用

对应SQLAlchemy查询代码示例：

from sqlalchemy import func

new_item_code = "待校验的编码值"
# pydb_decrypt是在PostgreSQL中自定义的、和Python侧加解密逻辑对齐的解密函数
exist_item = Item.query.filter(
    or_(
        func.pydb_decrypt(Item.item_code, "你的加密密钥") == new_item_code,
        Item.item_code == new_item_code
    )
).first()

用@hybrid_property失败是因为搞混了执行阶段：@hybrid_property的Python侧逻辑是在数据库返回结果、实例化模型对象的时候才执行的，生成查询SQL的阶段拿到的是字段属性对象，不是实际存储的值，自然没法解密做SQL层面的过滤。全表拉取到内存解密的方案没有扩展性，数据量上来之后一定会出性能问题，不要在生产环境使用。

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