This is an automated email from the ASF dual-hosted git repository. asf-gitbox-commits pushed a commit to branch cc/11633-part2a in repository https://gitbox.apache.org/repos/asf/allura.git
commit 80d6259165e26a70c7f0788515b98b187dcdcb2c 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 1c0b8c8f7..9add0316b 100644 --- a/Allura/allura/tests/functional/test_root.py +++ b/Allura/allura/tests/functional/test_root.py @@ -318,7 +318,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() @@ -329,7 +331,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]')
