Manage encryption on arbitrary column type

[Hameer Abbasi]

Hello. I've created an encryption wrapper that marshals things in and out of encryption, except that for the TypeDecorator the input and output type are different. I'm having trouble wrapping float and Enum types while still keeping them queryable.

Can someone take a look at the attached reproducer and let me know what I can do to fix it up/work around the issue?

[Mike Bayer]

On Wed, Dec 4, 2019, at 7:57 AM, Hameer Abbasi wrote:

Hello. I've created an encryption wrapper that marshals things in and out of encryption, except that for the TypeDecorator the input and output type are different. I'm having trouble wrapping float and Enum types while still keeping them queryable.

Can someone take a look at the attached reproducer and let me know what I can do to fix it up/work around the issue?

hi -

this is pretty well written, which is good to see.   I got your test to run and the error is "TypeError: can't escape Type to binary" and that's being raised by psycopg2 for me.

your type is actually wrapping a nested type and there are issues in how that is working.  first off the float needs to be a type instance, using self.inner = sqlalchemy.types.to_instance(inner), or otherwise pass a Float() explicitly.

then if we look what's coming in, the values are still seen as Binary, not as a simple string / float:

(Pdb) parameters

{'user_type': <psycopg2.extensions.Binary object at 0x7fb602b9d810>, 'pgp_sym_encrypt_1': 'PASSPHRASE_USER', 'reputation': <psycopg2.extensions.Binary object at 0x7fb602bad6c0>, 'pgp_sym_encrypt_2': 'PASSPHRASE_USER'}

next, the type handlers for Enum and Float, of which the former is necessary, aren't invoked at all because your "inner" type is not known to SQLAlchemy.  so we have to add handlers for these as:

    def process_bind_param(self, value, dialect):

        proc = self.inner.bind_processor(dialect)

        return proc(value)

    def process_result_value(self, value, dialect):

        proc = self.inner.result_processor(dialect, None)

        return proc(value)

but that is still not enough because this TypeDecorator is running the BYTEA value processors on top of those in any case, putting the value inside of a psycopg2.Binary() which still fails, with the same message but different type, "TypeError: can't escape str to binary".    so we have to also ensure we set the correct type for that nested bind parameter:

    def bind_expression(self, bindvalue):

        bindvalue.type = self.inner

        bindvalue = sqlalchemy.cast(bindvalue, sqlalchemy.String)

        return sqlalchemy.func.pgp_sym_encrypt(bindvalue, self.passphrase)

but still wrong because String is not instantiated, leading to the wrong CAST:

    def bind_expression(self, bindvalue):

        bindvalue.type = self.inner

        bindvalue = sqlalchemy.cast(bindvalue, sqlalchemy.String())

        return sqlalchemy.func.pgp_sym_encrypt(bindvalue, self.passphrase)

INSERT now succeeds.

SELECT fails with

sqlalchemy.exc.ProgrammingError: (psycopg2.errors.UndefinedObject) type "type" does not exist

LINE 1: ...ym_decrypt(users.user_type, 'PASSPHRASE_USER') AS type) AS u...

                                                             ^

[SQL: SELECT users.id AS users_id, CAST(pgp_sym_decrypt(users.user_type, %(pgp_sym_decrypt_1)s) AS type) AS users_user_type, CAST(pgp_sym_decrypt(users.reputation, %(pgp_sym_decrypt_2)s) AS FLOAT) AS users_reputation

FROM users]

still a "type" getting in there , what's happening is that your Enum is a "native" enum, that is, it expects Postgresql to have a "CREATE TYPE" happening in there, but because your TypeDecorator is holding onto that Enum, none of that is happening so there is no "type".   Hoping you don't want to use a native enum in any case here as they are more trouble than they're worth; setting the Enum as native=False removes the cumbsersome requirement for a "CREATE TYPE":

    user_type = sqlalchemy.Column(

        Encrypted(sqlalchemy.types.Enum(Type, native_enum=False), PASSPHRASE_USER)

    )

and that brings it all up:

class Encrypted(sqlalchemy.types.TypeDecorator):

    impl = postgresql.BYTEA

    def __init__(self, inner, passphrase):

        super().__init__()

        self.inner = types.to_instance(inner)

        self.passphrase = passphrase

    def process_bind_param(self, value, dialect):

        proc = self.inner.bind_processor(dialect)

        return proc(value)

    def process_result_value(self, value, dialect):

        proc = self.inner.result_processor(dialect, None)

        return proc(value)

    def bind_expression(self, bindvalue):

        bindvalue.type = self.inner

        bindvalue = sqlalchemy.cast(bindvalue, sqlalchemy.String())

        return sqlalchemy.func.pgp_sym_encrypt(bindvalue, self.passphrase)

    def column_expression(self, col):

        value = sqlalchemy.func.pgp_sym_decrypt(col, self.passphrase)

        value = sqlalchemy.cast(value, self.inner)

        return value

class User(Base):

    __tablename__ = "users"

    id = sqlalchemy.Column(sqlalchemy.types.BigInteger, primary_key=True)

    user_type = sqlalchemy.Column(

        Encrypted(sqlalchemy.types.Enum(Type, native_enum=False), PASSPHRASE_USER)

    )

    reputation = sqlalchemy.Column(

        Encrypted(sqlalchemy.types.Float, PASSPHRASE_USER)

    )

--

SQLAlchemy -

The Python SQL Toolkit and Object Relational Mapper

 

http://www.sqlalchemy.org/

 

To post example code, please provide an MCVE: Minimal, Complete, and Verifiable Example. See http://stackoverflow.com/help/mcve for a full description.

---

You received this message because you are subscribed to the Google Groups "sqlalchemy" group.

To unsubscribe from this group and stop receiving emails from it, send an email to sqlalchemy+...@googlegroups.com.

To view this discussion on the web visit https://groups.google.com/d/msgid/sqlalchemy/c44677c2-343e-49de-8bfe-0b5cfab70b41%40googlegroups.com.

Attachments:

• repro.py
  

[Hameer Abbasi]

Thanks. Yes, that's enough for my use-case, more than enough, to be honest.

Out of curiosity, how would you do it for a native ENUM?

Thanks a lot!

To unsubscribe from this group and stop receiving emails from it, send an email to sqlal...@googlegroups.com.