I have these 2 classes to encrypt and decrypt an email address in the postgres database using pgcrypto. They both work. One uses a postgres bytea column while the other uses a postgres varchar column.
The problem though I face is though is this:
>>> CustomUser.objects.filter(email="[email protected]")
<QuerySet []>
CustomUser.objects.filter(username="jtwlucas").values_list("email", flat=True)
<QuerySet ['[email protected]']>
>>> CustomUser.objects.filter(username="jtwlucas").values_list("email", flat=True)
<QuerySet [None]>
select PGP_SYM_DECRYPT(email::bytea, 'SECRET_KEY') from authentication_customuser;
[email protected]
'[email protected]' in CustomUser.objects.values_list('email', flat=True)
True`
Even though that email address exists in the database.
The database shows this for the varchar column:
xc30d04090302b95c598cd3d427836dd24c01eb524973d9a2742acb4e24091faf0bb6507c338d97dcfac6fbfda038c34fc6bbd59ccad723acf4235c9825fe14981b7ae2e63fc79277cf442480f30249adefaa964b9f19be3592c9f0c219
The database shows this for the bytea column:
[binary data]
If possible I want the be able to perform the lookups as I am doing them in this example automatically without needing to change my entire code base's ORM querysets.
I do not want to use any third party github libraries.
#postgres bytea column
class SecureEmail(models.EmailField):
"""Custom Encrypted Field"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.secret_key = str(settings.SECRET_KEY)
def from_db_value(self, value, expression, connection):
# Decrypt value when reading from database
if value is not None:
cursor = connection.cursor()
cursor.execute("SELECT pgp_sym_decrypt(%s::bytea, %s)::text", [value, self.secret_key])
return cursor.fetchone()[0]
def get_prep_value(self, value):
# Encrypt value when writing to database
cursor = connection.cursor()
cursor.execute("SELECT pgp_sym_encrypt(%s::text, %s)", [value, self.secret_key])
return cursor.fetchone()[0]
#postgres varchar column
class SecureEmail(models.EmailField):
"""Custom Encrypted Field"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.secret_key = str(settings.SECRET_KEY)
def from_db_value(self, value, expression, connection):
# Decrypt value when reading from database
if value is not None:
cursor = connection.cursor()
cursor.execute("SELECT pgp_sym_decrypt(%s::bytea, %s)::text", [value,
self.secret_key])
return cursor.fetchone()[0]
def get_prep_value(self, value):
cursor = connection.cursor()
cursor.execute("SELECT pgp_sym_encrypt(%s::text, %s)::character varying", [value, self.secret_key])
return cursor.fetchone()[0]
class CustomUser(AbstractBaseUser, PermissionsMixin):
email = SecureEmail()
I tried using lookups but have not had any success:
https://django.readthedocs.io/en/latest/howto/custom-lookups.html