This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/allura.git
The following commit(s) were added to refs/heads/master by this push:
new f84206aa8 [#8606] Implement field level encryption for EmailAddress
email field
f84206aa8 is described below
commit f84206aa8f8f3956b53864b965c97c12846eeca3
Author: Carlos Cruz <[email protected]>
AuthorDate: Wed May 13 16:05:28 2026 -0600
[#8606] Implement field level encryption for EmailAddress email field
---
Allura/allura/model/auth.py | 59 +++++++++++++----------------
Allura/allura/tests/functional/test_root.py | 8 +++-
Allura/allura/tests/model/test_auth.py | 5 ++-
3 files changed, 36 insertions(+), 36 deletions(-)
diff --git a/Allura/allura/model/auth.py b/Allura/allura/model/auth.py
index 22eee8722..89df3aa9a 100644
--- a/Allura/allura/model/auth.py
+++ b/Allura/allura/model/auth.py
@@ -38,8 +38,8 @@
from tg import tmpl_context as c, app_globals as g
from tg import request
from ming import schema as S
-from ming.odm import session, state, MapperExtension
-from ming.odm import FieldProperty, RelationProperty, ForeignIdProperty
+from ming.odm import session, state
+from ming.odm import FieldProperty, RelationProperty, ForeignIdProperty,
DecryptedProperty
from ming.odm.declarative import MappedClass
from ming.odm.odmsession import ThreadLocalODMSession
from ming.utils import LazyProperty
@@ -74,20 +74,6 @@ def __init__(self, **kwargs):
super().__init__(User, allow_none=True, **kwargs)
-class EmailAddressMapperExtension(MapperExtension):
-
- def before_insert(self, instance, state, sess):
- self._set_email_encrypted(instance, state)
-
- def before_update(self, instance, state, sess):
- self._set_email_encrypted(instance, state)
-
- @staticmethod
- def _set_email_encrypted(instance, state):
- email = state.document.get('email')
- state.document['email_encrypted'] = instance.encr(email)
-
-
class EmailAddress(MappedClass):
re_format = re.compile(r'^.*\s+<(.*)>\s*$')
@@ -95,13 +81,12 @@ class __mongometa__:
name = 'email_address'
session = main_orm_session
indexes = ['nonce', ]
- unique_indexes = [('email', 'claimed_by_user_id'), ]
- extensions = [EmailAddressMapperExtension]
+ unique_indexes = [('email_encrypted', 'claimed_by_user_id'), ]
query: Query[EmailAddress]
_id = FieldProperty(S.ObjectId)
- email = FieldProperty(str)
+ email = DecryptedProperty(str, 'email_encrypted')
email_encrypted = FieldProperty(S.Binary)
claimed_by_user_id = FieldProperty(S.ObjectId, if_missing=None)
confirmed = FieldProperty(bool, if_missing=False)
@@ -113,30 +98,38 @@ class __mongometa__:
@classmethod
def get(cls, **kw):
- '''Equivalent to Ming's query.get but calls self.canonical on address
- before lookup. You should always use this instead of query.get'''
- if kw.get('email'):
- email = cls.canonical(kw['email'])
- if email is not None:
- kw['email'] = email
- else:
+ '''Equivalent to Ming's query.get but translates email lookups to
+ canonicalized email_encrypted queries. You should always use this
+ instead of query.get'''
+ if 'email' in kw:
+ email_encrypted = cls.encrypted_email(kw.pop('email'))
+ if email_encrypted is None:
return None
+ kw['email_encrypted'] = email_encrypted
return cls.query.get(**kw)
@classmethod
def find(cls, q=None):
- '''Equivalent to Ming's query.find but calls self.canonical on address
- before lookup. You should always use this instead of query.find'''
+ '''Equivalent to Ming's query.find but translates email lookups to
+ canonicalized email_encrypted queries. You should always use this
+ instead of query.find'''
if q:
- if q.get('email'):
- email = cls.canonical(q['email'])
- if email is not None:
- q['email'] = email
- else:
+ q = q.copy()
+ if 'email' in q:
+ email_encrypted = cls.encrypted_email(q.pop('email'))
+ if email_encrypted is None:
return utils.EmptyCursor()
+ q['email_encrypted'] = email_encrypted
return cls.query.find(q)
return cls.query.find()
+ @classmethod
+ def encrypted_email(cls, addr):
+ email = cls.canonical(addr) if isinstance(addr, str) and addr else None
+ if email is None:
+ return None
+ return cls.encr(email)
+
def claimed_by_user(self, include_pending=False, include_disabled=False):
q = {'_id': self.claimed_by_user_id,
'disabled': False,
diff --git a/Allura/allura/tests/functional/test_root.py
b/Allura/allura/tests/functional/test_root.py
index b3ca1139b..ac7862854 100644
--- a/Allura/allura/tests/functional/test_root.py
+++ b/Allura/allura/tests/functional/test_root.py
@@ -319,7 +319,9 @@ def setup_method(self, method):
def teardown_method(self, method):
u = M.User.query.get(username='test-admin')
- email = M.EmailAddress.query.get(claimed_by_user_id=u._id,
email='[email protected]')
+ email = M.EmailAddress.query.get(
+ claimed_by_user_id=u._id,
+ email_encrypted=M.EmailAddress.encrypted_email('[email protected]'))
email.delete()
ThreadLocalODMSession.flush_all()
@@ -330,7 +332,9 @@ def test_unconfirmed_message(self):
def test_confirmed_message(self):
u = M.User.query.get(username='test-admin')
- email = M.EmailAddress.query.get(claimed_by_user_id=u._id,
email=self.unconfirmed_email)
+ email = M.EmailAddress.query.get(
+ claimed_by_user_id=u._id,
+
email_encrypted=M.EmailAddress.encrypted_email(self.unconfirmed_email))
email.confirmed = True
ThreadLocalODMSession.flush_all()
login(self.app, username='test-admin')
diff --git a/Allura/allura/tests/model/test_auth.py
b/Allura/allura/tests/model/test_auth.py
index 839b9912e..54d25b828 100644
--- a/Allura/allura/tests/model/test_auth.py
+++ b/Allura/allura/tests/model/test_auth.py
@@ -28,7 +28,7 @@
from mock import patch, Mock
from ming.odm.odmsession import ThreadLocalODMSession
-from ming.odm import session
+from ming.odm import session, state
from allura import model as M
from allura.lib import helpers as h
@@ -69,8 +69,11 @@ def test_email_address_stores_encrypted_email(self):
direct_addr = M.EmailAddress(email='[email protected]')
ThreadLocalODMSession.flush_all()
+ assert addr.email == '[email protected]'
assert addr.email_encrypted == M.EmailAddress.encr('[email protected]')
assert direct_addr.email_encrypted ==
M.EmailAddress.encr('[email protected]')
+ assert 'email' not in state(addr).document
+ assert 'email' not in state(direct_addr).document
def selftest_email_address_lookup_helpers():
addr = M.EmailAddress.create('[email protected]')
