JSONType 更改的字符串加密类型未保存到数据库

Question

背景故事

我有一份问卷，其中询问敏感问题，其中大部分是真/假。大多数情况下，这些值是错误的，这对保持静态数据的私密性构成了挑战。将每个问题加密到单独的列中时，通过一些猜测很容易分辨出哪个值是真的，哪个是假的。为了解决这个问题，问题和答案被放入带有一些盐（随机变化的废话）的字典对象中，然后加密。没有钥匙就不可能知道答案是什么。

方法

下面是一个模型示例，该模型用于使用静态盐对数据进行加密，从而无法查看数据并了解其内容。

import sqlalchemy as sa
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy_utils.types import JSONType
from sqlalchemy_utils.types.encrypted.encrypted_type import StringEncryptedType, AesEngine


Base = declarative_base()

class SensitiveQuestionnaire(Base):
    user_id = sa.Column(sa.Integer, primary_key=True, autoincrement=True)
    _data = data: dict = sa.Column(StringEncryptedType(JSONType, 'secret', AesEngine, 'pkcs5'),
        nullable=False, default=lambda: {'_salt': salt_shaker()})

    # values are viewed using a python property to look into the `_data` dict
    @property
    def sensitive_question(self) -> Optional[float]:
        return self._data.get('sensitive_question')

    # values are set into the `_data` dict
    @sensitive_question.setter
    def sensitive_question(self, value: bool) -> None:
        self._data['sensitive_question'] = value

    # in a real example there would be 20+ properties that map to questions

    def __init__(self, **kwargs):
        # Sqlalchemy does not use the __init__ method so we are free to set object defaults here
        self._data = {'_salt': salt_shaker()}
        for key in kwargs:
            setattr(self, key, kwargs[key])

    @property
    def _salt(self) -> str:
        return self._data['_salt']


def salt_shaker():
    return ''.join([random.choice('hldjs..' for i in range(50)])

问题

SensitiveQuestionnaire 对象初始化后，所有更改都不会保存在数据库中。

# GIVEN a questionnaire 
questionnaire = model.SensitiveQuestionnaire(user_id=1)
db.session.add()
db.session.commit()

# WHEN updating the questionnaire and saving it to the database
questionnaire.sensitive_question= True
db.session.commit()

# THEN we get the questionnaire from the database
db_questionnaire = model.SensitiveQuestionnaire.query\
                   .filter(model.SensitiveQuestionnaire.user_id == 1).first()

# THEN the sensitive_question value is persisted
assert db_questionnaire.sensitive_question is True

db_questionnaire.sensitive_question 中的值是 None，而它应该是 True。

Answer 1

在花了一天的大部分时间来解决这个问题后，问题的原因是 Sqlalchemy 如何知道何时发生了变化。简短的版本是 sqlalchemy 使用 python 的 __setitem__ 挂钩 sqlalchemy 的 change() 方法，让它知道发生了变化。更多信息可以在 sqlalchemy 的文档中找到。

答案是将StringEncryptedType 包装在MultableDict 类型中

突变跟踪

支持跟踪对标量值的就地更改，这些更改会传播到拥有父对象的 ORM 更改事件中。 来自 SqlAlchemy 的文档：https://docs.sqlalchemy.org/en/13/orm/extensions/mutable.html

解决方案

精简版...将 StringEncryptedType 包装在 MutableDict 中

_data = data: dict = sa.Column(
        MutableDict.as_mutable(StringEncryptedType(JSONType, 'secret', AesEngine, 'pkcs5')),
        nullable=False, default=lambda: {'_salt': salt_shaker()})

以上问题的完整版

import sqlalchemy as sa
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.mutable import MutableDict
from sqlalchemy_utils.types import JSONType
from sqlalchemy_utils.types.encrypted.encrypted_type import StringEncryptedType, AesEngine


Base = declarative_base()

class SensitiveQuestionnaire(Base):
    user_id: int = sa.Column(sa.Integer, primary_key=True, autoincrement=True)

    # The MutableDict.as_mutable below is what changed!
    _data = data: dict = sa.Column(
        MutableDict.as_mutable(StringEncryptedType(JSONType, 'secret', AesEngine, 'pkcs5')),
        nullable=False, default=lambda: {'_salt': salt_shaker()})

    @property
    def sensitive_question(self) -> Optional[float]:
        return self._data.get('sensitive_question')

    # values are set into the `_data` dict
    @sensitive_question.setter
    def sensitive_question(self, value: bool) -> None:
        self._data['sensitive_question'] = value

    # in a real example there would be 20+ properties that map to questions

    def __init__(self, **kwargs):
        self._data = {'_salt': salt_shaker()}
        for key in kwargs:
            setattr(self, key, kwargs[key])

    @property
    def _salt(self) -> str:
        return self._data['_salt']


def salt_shaker():
    return ''.join([random.choice('hldjs..' for i in range(50)])