将 SQLAlchemy hybrid_property 与原生属性构造相结合

Question

我在 SQLAlchemy 中有一个用户类。我希望能够在数据库中加密用户的电子邮件地址属性，但仍然可以通过过滤查询进行搜索。

我的问题是，如果我使用@hybrid_property，我的查询理论上有效，但我的构造无效，如果我使用@property，我的构造有效，但我的查询无效

from cryptography.fernet import Fernet  # <- pip install cryptography
from werkzeug.security import generate_password_hash
class User(db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    email_hash = db.Column(db.String(184), unique=True, nullable=False)
    password_hash = db.Column(db.String(128))

    # @property       # <- Consider this as option 2...
    @hybrid_property  # <- Consider this as option 1...
    def email(self):
        f = Fernet('SOME_ENC_KEY')
        value = f.decrypt(self.email_hash.encode('utf-8'))
        return value
    @email.setter
    def email(self, email):
        f = Fernet('SOME_ENC_KEY')
        self.email_hash = f.encrypt(email.encode('utf-8'))

    @property
    def password(self):
        raise AttributeError('password is not a readable attribute.')
    @password.setter
    def password(self, password):
        self.password_hash = generate_password_hash(password)

    def __init__(self, **kwargs):
        super(User, self).__init__(**kwargs)
        # other checks and modifiers

对于选项 1：当我尝试使用 User(email='a@example.com',password='secret') 构建用户时，我会收到回溯，

~/models.py in __init__(self, **kwargs)
    431     # Established role assignment by default class initiation
    432     def __init__(self, **kwargs):
--> 433         super(User, self).__init__(**kwargs)
    434         if self.role is None:
    435             _default_role = Role.query.filter_by(default=True).first()

~/lib/python3.6/site-packages/sqlalchemy/ext/declarative/base.py in _declarative_constructor(self, **kwargs)
    697             raise TypeError(
    698                 "%r is an invalid keyword argument for %s" %
--> 699                 (k, cls_.__name__))
    700         setattr(self, k, kwargs[k])
    701 _declarative_constructor.__name__ = '__init__'
TypeError: 'email' is an invalid keyword argument for User

对于选项 2：如果我将 @hybrid_property 更改为 @property，则构造很好，但我的查询 User.query.filter_by(email=form.email.data.lower()).first() 失败并返回 None。

我应该进行哪些更改才能使其按要求工作？

==============

注意我应该说我已经尽量避免使用双重属性，因为我不想对底层代码库进行大量编辑。所以我明确地试图避免将创建与User(email_input='a@a.com', password='secret') 和User.query.filter_by(email='a@a.com').first() 的查询分开：

class User(db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    email_hash = db.Column(db.String(184), unique=True, nullable=False)
    password_hash = db.Column(db.String(128))

    @hybrid_property
    def email(self):
        f = Fernet('SOME_ENC_KEY')
        value = f.decrypt(self.email_hash.encode('utf-8'))
        return value
    @property
    def email_input(self):
        raise AttributeError('email_input is not a readable attribute.')
    @email_input.setter
    def email_input(self, email):
        f = Fernet('SOME_ENC_KEY')
        self.email_hash = f.encrypt(email.encode('utf-8'))

    @property
    def password(self):
        raise AttributeError('password is not a readable attribute.')
    @password.setter
    def password(self, password):
        self.password_hash = generate_password_hash(password)

    def __init__(self, **kwargs):
        super(User, self).__init__(**kwargs)
        # other checks and modifiers

Answer 1

在您的hybrid_property、email 中，如果self.email_hash 是str 类型，则self.f.decrypt(self.email_hash.encode('utf-8')) 行很好，但是，因为email 是hybrid_property，当SQLAlchemy 使用它来生成SQL 时self.email_hash 实际上是一个sqlalchemy.orm.attributes.InstrumentedAttribute 类型。

来自docs 关于混合属性：

在许多情况下，Python 内函数的构造和 SQLAlchemy SQL 表达式有足够的差异，两个分开 应该定义 Python 表达式。

因此您可以定义一个 hybrid_property.expression 方法，这是 SQLAlchemy 将用来生成 sql 的方法，允许您在 hybrid_property 方法中保持字符串处理的完整性。

根据您的示例，这是我最终得到的对我有用的代码。为了简单起见，我已经从您的User 模型中删除了很多内容，但所有重要部分都在那里。我还必须为您的代码中调用但未提供的其他函数/类编写实现（请参阅MCVE）：

class Fernet:
    def __init__(self, k):
        self.k = k

    def encrypt(self, s):
        return s

    def decrypt(self, s):
        return s

def get_env_variable(s):
    return s

def generate_password_hash(s):
    return s

class User(db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    email_hash = db.Column(db.String(184), unique=True, nullable=False)

    f = Fernet(get_env_variable('FERNET_KEY'))

    @hybrid_property
    def email(self):
        return self.f.decrypt(self.email_hash.encode('utf-8'))

    @email.expression
    def email(cls):
        return cls.f.decrypt(cls.email_hash)

    @email.setter
    def email(self, email):
        self.email_hash = self.f.encrypt(email.encode('utf-8'))



if __name__ == '__main__':
    db.drop_all()
    db.create_all()
    u = User(email='a@example.com')
    db.session.add(u)
    db.session.commit()
    print(User.query.filter_by(email='a@example.com').first())
    # <User 1> 

不幸的是，上面的代码之所以有效，是因为模拟 Fernet.decrypt 方法返回了传入的确切对象。存储用户电子邮件地址的 Fernet 编码散列的问题是 Fernet.encrypt 不会返回相同的 @ 987654339@ 从一次执行到下一次执行，即使使用相同的密钥。例如：

>>> from cryptography.fernet import Fernet
>>> f = Fernet(Fernet.generate_key())
>>> f.encrypt('a@example.com'.encode('utf-8')) == f.encrypt('a@example.com'.encode('utf-8'))
False

因此，您想查询数据库中的一条记录，但无法知道查询时实际查询的字段的存储值是什么。您可以构建一个classmethod 来查询整个users 表并遍历每条记录，解密其存储的哈希并将其与明文电子邮件进行比较。或者，您可以构建一个始终返回相同值的散列函数，使用该函数散列新用户电子邮件，并使用电子邮件字符串的散列直接查询 email_hash 字段。其中，考虑到大量用户，第一个效率非常低。

Fernet.encrypt 函数是：

def encrypt(self, data):
    current_time = int(time.time())
    iv = os.urandom(16)
    return self._encrypt_from_parts(data, current_time, iv)

因此，您可以定义current_time 和iv 的静态值，然后自己直接调用Fermat._encrypt_from_parts。或者你可以使用python内置的hash和just set a fixed seed，这样它就具有确定性。然后，您可以对要查询的电子邮件字符串进行哈希处理，然后首先直接查询Users.email_hash。只要您没有对密码字段执行上述任何操作！