This is an automated email from the ASF dual-hosted git repository. asf-gitbox-commits pushed a commit to branch cc/8610-part2 in repository https://gitbox.apache.org/repos/asf/allura.git
commit f85be70b6931b324f7a2ccb35f4bc44a2e944c18 Author: Carlos Cruz <[email protected]> AuthorDate: Tue Jun 16 12:56:24 2026 -0600 [#8610] Add encrypted field for Users email addresses --- Allura/allura/model/auth.py | 18 +++++++++++++++++ Allura/allura/tests/model/test_auth.py | 35 ++++++++++++++++++++++++++++++++++ 2 files changed, 53 insertions(+) diff --git a/Allura/allura/model/auth.py b/Allura/allura/model/auth.py index a4abc45d7..d9d26baf1 100644 --- a/Allura/allura/model/auth.py +++ b/Allura/allura/model/auth.py @@ -266,6 +266,15 @@ def get_decrypted(self, instance): return super().__get__(instance, type(instance)) +class UserEmailAddressesMapperExtension(MapperExtension): + + def before_insert(self, obj, state, sess): + obj.sync_email_addresses_encrypted() + + def before_update(self, obj, state, sess): + obj.sync_email_addresses_encrypted() + + class User(MappedClass, ActivityNode, ActivityObject, SearchIndexable): SALT_LEN = 8 @@ -274,6 +283,7 @@ class __mongometa__: session = main_orm_session indexes = ['tool_data.AuthPasswordReset.hash'] unique_indexes = ['username'] + extensions = [UserEmailAddressesMapperExtension] custom_indexes = [ dict(fields=('tool_data.phone_verification.number_hash',), sparse=True), ] @@ -286,6 +296,7 @@ class __mongometa__: sfx_userid = FieldProperty(S.Deprecated) username = FieldProperty(str) email_addresses = FieldProperty([str]) + email_addresses_encrypted = FieldProperty([S.Binary], if_missing=[]) password = FieldProperty(str) # hashed last_password_updated = FieldProperty(datetime) # to access, use AuthProvider's get_last_password_updated password_algorithm = FieldProperty(str) @@ -787,6 +798,13 @@ def set_tool_data(self, tool, **kw): d.update(kw) state(self).soil() + @classmethod + def encrypt_email_addresses(cls, email_addresses): + return [cls.encr(addr) if addr is not None else None for addr in email_addresses or []] + + def sync_email_addresses_encrypted(self): + self.email_addresses_encrypted = type(self).encrypt_email_addresses(self.email_addresses) + def address_object(self, addr): return EmailAddress.get(email=addr, claimed_by_user_id=self._id) diff --git a/Allura/allura/tests/model/test_auth.py b/Allura/allura/tests/model/test_auth.py index d1dd70771..014cf0021 100644 --- a/Allura/allura/tests/model/test_auth.py +++ b/Allura/allura/tests/model/test_auth.py @@ -432,6 +432,41 @@ def test_display_name_is_stored_encrypted(self): display_name_encrypted=M.User.encr('Encrypted Display Name')) assert user.username == 'encrypted-display-name-test' + def test_email_addresses_encrypted_is_populated_on_creation(self): + user = M.User( + username='email-addresses-encrypted-create-test', + email_addresses=['[email protected]', None, '[email protected]']) + ThreadLocalODMSession.flush_all() + + expected_encrypted = [ + M.User.encr('[email protected]'), + None, + M.User.encr('[email protected]'), + ] + assert user.email_addresses == ['[email protected]', None, '[email protected]'] + assert user.email_addresses_encrypted == expected_encrypted + assert state(user).document['email_addresses'] == ['[email protected]', None, '[email protected]'] + assert state(user).document['email_addresses_encrypted'] == expected_encrypted + + def test_email_addresses_encrypted_is_synced_on_update(self): + user = M.User( + username='email-addresses-encrypted-update-test', + email_addresses=['[email protected]', '[email protected]']) + ThreadLocalODMSession.flush_all() + + user.email_addresses.append('[email protected]') + user.email_addresses.remove('[email protected]') + ThreadLocalODMSession.flush_all() + + expected_encrypted = [ + M.User.encr('[email protected]'), + M.User.encr('[email protected]'), + ] + assert user.email_addresses == ['[email protected]', '[email protected]'] + assert user.email_addresses_encrypted == expected_encrypted + assert state(user).document['email_addresses'] == ['[email protected]', '[email protected]'] + assert state(user).document['email_addresses_encrypted'] == expected_encrypted + def test_set_display_name_pref_updates_encrypted_field_and_cache(self): user = M.User(username='display-name-cache-test', display_name='Original Display Name')
