This is an automated email from the ASF dual-hosted git repository. asf-gitbox-commits pushed a commit to branch cc/11633-part2 in repository https://gitbox.apache.org/repos/asf/allura.git
commit 0c3a6d834300eae1e44e4567077539f809de8c25 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/controllers/auth.py | 29 ++++++--- Allura/allura/lib/mail_util.py | 6 +- Allura/allura/model/auth.py | 72 +++++++++++------------ Allura/allura/tests/functional/test_auth.py | 59 +++++++++++++------ Allura/allura/tests/functional/test_root.py | 8 ++- Allura/allura/tests/functional/test_site_admin.py | 6 +- Allura/allura/tests/model/test_auth.py | 32 +++++----- Allura/allura/tests/test_mail_util.py | 16 ++--- Allura/allura/websetup/bootstrap.py | 5 +- ForgeUserStats/forgeuserstats/tests/test_model.py | 3 +- ForgeUserStats/forgeuserstats/tests/test_stats.py | 3 +- 11 files changed, 145 insertions(+), 94 deletions(-) diff --git a/Allura/allura/controllers/auth.py b/Allura/allura/controllers/auth.py index 669117200..66630e5a1 100644 --- a/Allura/allura/controllers/auth.py +++ b/Allura/allura/controllers/auth.py @@ -263,18 +263,24 @@ def password_recovery_hash(self, email=None, **kw): if user_record and user_record.pending: message = 'If the given email address is on record, '\ 'an email has been sent to the account\'s primary email address.' - email_record = M.EmailAddress.get(email=provider.get_primary_email_address(user_record=user_record), - confirmed=False) + primary_email_encrypted = M.EmailAddress.encrypted_email( + provider.get_primary_email_address(user_record=user_record)) + email_record = M.EmailAddress.get(email_encrypted=primary_email_encrypted, + confirmed=False) if primary_email_encrypted else None provider.resend_verification_link(user_record, email_record) elif not allow_non_primary_email_reset: message = 'If the given email address is on record, '\ 'a password reset email has been sent to the account\'s primary email address.' - email_record = M.EmailAddress.get(email=provider.get_primary_email_address(user_record=user_record), - confirmed=True) + primary_email_encrypted = M.EmailAddress.encrypted_email( + provider.get_primary_email_address(user_record=user_record)) + email_record = M.EmailAddress.get(email_encrypted=primary_email_encrypted, + confirmed=True) if primary_email_encrypted else None else: message = 'A password reset email has been sent, if the given email address is on record in our system.' - email_record = M.EmailAddress.get(email=email, confirmed=True) + email_encrypted = M.EmailAddress.encrypted_email(email) + email_record = M.EmailAddress.get(email_encrypted=email_encrypted, + confirmed=True) if email_encrypted else None if user_record and email_record: if email_record.confirmed: @@ -317,8 +323,11 @@ def save_new(self, display_name=None, username=None, pw=None, email=None, **kw): @expose() def send_verification_link(self, a): - addr = M.EmailAddress.get(email=a, claimed_by_user_id=c.user._id) - confirmed_emails = M.EmailAddress.find(dict(email=a, confirmed=True)).all() + email_encrypted = M.EmailAddress.encrypted_email(a) + addr = M.EmailAddress.get(email_encrypted=email_encrypted, + claimed_by_user_id=c.user._id) if email_encrypted else None + confirmed_emails = M.EmailAddress.find(dict(email_encrypted=email_encrypted, + confirmed=True)).all() if email_encrypted else [] confirmed_emails = [item for item in confirmed_emails if item != addr] if addr: @@ -332,7 +341,8 @@ def send_verification_link(self, a): redirect(six.ensure_text(request.referer or '/')) def _verify_addr(self, addr, do_auth_check=True): - confirmed_by_other = M.EmailAddress.find(dict(email=addr.email, confirmed=True)).all() if addr else [] + confirmed_by_other = M.EmailAddress.find(dict(email_encrypted=addr.email_encrypted, + confirmed=True)).all() if addr else [] confirmed_by_other = [item for item in confirmed_by_other if item != addr] if addr and not confirmed_by_other: @@ -731,7 +741,8 @@ def _update_emails(self, user, admin=False, form_params={}): flash('You must provide your current password to claim new email', 'error') return - claimed_emails = M.EmailAddress.find({'email': new_addr['addr']}).all() + email_encrypted = M.EmailAddress.encrypted_email(new_addr['addr']) + claimed_emails = M.EmailAddress.find({'email_encrypted': email_encrypted}).all() if email_encrypted else [] if any(email.claimed_by_user_id == user._id for email in claimed_emails): flash('Email address already claimed', 'error') diff --git a/Allura/allura/lib/mail_util.py b/Allura/allura/lib/mail_util.py index d086dc9f4..ed7ff690c 100644 --- a/Allura/allura/lib/mail_util.py +++ b/Allura/allura/lib/mail_util.py @@ -179,13 +179,15 @@ def parse_message(data): def identify_sender(peer, email_address, headers, msg): from allura import model as M # Dumb ID -- just look for email address claimed by a particular user - addr = M.EmailAddress.get(email=email_address, confirmed=True) + email_encrypted = M.EmailAddress.encrypted_email(email_address) + addr = M.EmailAddress.get(email_encrypted=email_encrypted, confirmed=True) if addr and addr.claimed_by_user_id: return addr.claimed_by_user() or M.User.anonymous() from_address = headers.get('From', '').strip() if not from_address: return M.User.anonymous() - addr = M.EmailAddress.get(email=from_address) + email_encrypted = M.EmailAddress.encrypted_email(from_address) + addr = M.EmailAddress.get(email_encrypted=email_encrypted) if addr and addr.claimed_by_user_id: return addr.claimed_by_user() or M.User.anonymous() return M.User.anonymous() diff --git a/Allura/allura/model/auth.py b/Allura/allura/model/auth.py index 75e3e32fd..f302208d7 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, @@ -168,7 +161,7 @@ def canonical(cls, addr): return None def send_claim_attempt(self): - confirmed_email = self.find(dict(email=self.email, confirmed=True)).all() + confirmed_email = self.find(dict(email_encrypted=self.email_encrypted, confirmed=True)).all() if confirmed_email: log.info('Sending claim attempt email to %s', self.email) @@ -751,7 +744,10 @@ def upsert(cls, username): @classmethod def by_email_address(cls, addr, only_confirmed=True): - q = dict(email=addr) + email_encrypted = EmailAddress.encrypted_email(addr) + if email_encrypted is None: + return None + q = dict(email_encrypted=email_encrypted) if only_confirmed: q['confirmed'] = True addrs = EmailAddress.find(q) @@ -780,7 +776,11 @@ def set_tool_data(self, tool, **kw): state(self).soil() def address_object(self, addr): - return EmailAddress.get(email=addr, claimed_by_user_id=self._id) + email_encrypted = EmailAddress.encrypted_email(addr) + if email_encrypted is None: + return None + return EmailAddress.get(email_encrypted=email_encrypted, + claimed_by_user_id=self._id) def claim_address(self, email_address): addr = EmailAddress.canonical(email_address) diff --git a/Allura/allura/tests/functional/test_auth.py b/Allura/allura/tests/functional/test_auth.py index 5db3a7f36..350af1056 100644 --- a/Allura/allura/tests/functional/test_auth.py +++ b/Allura/allura/tests/functional/test_auth.py @@ -72,7 +72,8 @@ def test_login(self): ThreadLocalODMSession.flush_all() r = self.app.get('/auth/verify_addr', params=dict(a='foo')) assert json.loads(self.webflash(r))['status'] == 'error', self.webflash(r) - ea = M.EmailAddress.find({'email': email}).first() + ea = M.EmailAddress.find({ + 'email_encrypted': M.EmailAddress.encrypted_email(email)}).first() r = self.app.get('/auth/verify_addr', params=dict(a=ea.nonce)) assert json.loads(self.webflash(r))['status'] == 'ok', self.webflash(r) @@ -383,7 +384,9 @@ def test_user_can_not_claim_duplicate_emails(self): }, extra_environ=dict(username='test-admin')) - assert M.EmailAddress.find(dict(email=email_address, claimed_by_user_id=user._id)).count() == 1 + assert M.EmailAddress.find(dict( + email_encrypted=M.EmailAddress.encrypted_email(email_address), + claimed_by_user_id=user._id)).count() == 1 r = self.app.post('/auth/preferences/update_emails', params={ 'new_addr.addr': email_address, @@ -396,7 +399,9 @@ def test_user_can_not_claim_duplicate_emails(self): extra_environ=dict(username='test-admin')) assert json.loads(self.webflash(r))['status'] == 'error', self.webflash(r) - assert M.EmailAddress.find(dict(email=email_address, claimed_by_user_id=user._id)).count() == 1 + assert M.EmailAddress.find(dict( + email_encrypted=M.EmailAddress.encrypted_email(email_address), + claimed_by_user_id=user._id)).count() == 1 assert len(M.User.query.get(username='test-admin').email_addresses) == addresses_number + 1 @td.with_user_project('test-admin') @@ -409,7 +414,8 @@ def test_user_added_claimed_address_by_other_user_confirmed(self, gen_message_id # test-user claimed & confirmed email address user = M.User.query.get(username='test-user') user.claim_address(email_address) - email = M.EmailAddress.find(dict(email=email_address)).first() + email = M.EmailAddress.find(dict( + email_encrypted=M.EmailAddress.encrypted_email(email_address))).first() email.confirmed = True ThreadLocalODMSession.flush_all() @@ -441,7 +447,8 @@ def test_user_added_claimed_address_by_other_user_confirmed(self, gen_message_id "but it is already claimed by your %s account." % (email_address, user.username) in kwargs['text'] assert len(M.User.query.get(username='test-admin').email_addresses) == addresses_number + 1 - assert len(M.EmailAddress.find(dict(email=email_address)).all()) == 2 + assert len(M.EmailAddress.find(dict( + email_encrypted=M.EmailAddress.encrypted_email(email_address))).all()) == 2 @td.with_user_project('test-admin') @patch('allura.tasks.mail_tasks.sendsimplemail') @@ -452,7 +459,8 @@ def test_user_added_claimed_address_by_other_user_not_confirmed(self, gen_messag # test-user claimed email address user = M.User.query.get(username='test-user') user.claim_address(email_address) - email = M.EmailAddress.find(dict(email=email_address)).first() + email = M.EmailAddress.find(dict( + email_encrypted=M.EmailAddress.encrypted_email(email_address))).first() email.confirmed = False ThreadLocalODMSession.flush_all() # Claiming the same email address by test-admin @@ -477,7 +485,8 @@ def test_user_added_claimed_address_by_other_user_not_confirmed(self, gen_messag 'Please check your email and click to confirm.' assert sendsimplemail.post.called assert len(M.User.query.get(username='test-user-1').email_addresses) == addresses_number + 1 - assert len(M.EmailAddress.find(dict(email=email_address)).all()) == 2 + assert len(M.EmailAddress.find(dict( + email_encrypted=M.EmailAddress.encrypted_email(email_address))).all()) == 2 @td.with_user_project('test-admin') @patch('allura.tasks.mail_tasks.sendsimplemail') @@ -520,12 +529,16 @@ def test_verification_link_for_confirmed_email(self, gen_message_id, sendsimplem # test-user claimed email address user = M.User.query.get(username='test-user') user.claim_address(email_address) - email = M.EmailAddress.find(dict(email=email_address, claimed_by_user_id=user._id)).first() + email = M.EmailAddress.find(dict( + email_encrypted=M.EmailAddress.encrypted_email(email_address), + claimed_by_user_id=user._id)).first() email.confirmed = True user1 = M.User.query.get(username='test-user-1') user1.claim_address(email_address) - email = M.EmailAddress.find(dict(email=email_address, claimed_by_user_id=user1._id)).first() + email = M.EmailAddress.find(dict( + email_encrypted=M.EmailAddress.encrypted_email(email_address), + claimed_by_user_id=user1._id)).first() email.confirmed = False ThreadLocalODMSession.flush_all() @@ -551,7 +564,9 @@ def test_invalidate_verification_link_if_email_was_confirmed(self): # test-user claimed email address user = M.User.query.get(username='test-user') user.claim_address(email_address) - email = M.EmailAddress.find(dict(email=email_address, claimed_by_user_id=user._id)).first() + email = M.EmailAddress.find(dict( + email_encrypted=M.EmailAddress.encrypted_email(email_address), + claimed_by_user_id=user._id)).first() email.confirmed = False ThreadLocalODMSession.flush_all() @@ -562,7 +577,9 @@ def test_invalidate_verification_link_if_email_was_confirmed(self): user1 = M.User.query.get(username='test-user-1') user1.claim_address(email_address) - email1 = M.EmailAddress.find(dict(email=email_address, claimed_by_user_id=user1._id)).first() + email1 = M.EmailAddress.find(dict( + email_encrypted=M.EmailAddress.encrypted_email(email_address), + claimed_by_user_id=user1._id)).first() email1.confirmed = True ThreadLocalODMSession.flush_all() # Verify first email with the verification link @@ -570,7 +587,9 @@ def test_invalidate_verification_link_if_email_was_confirmed(self): extra_environ=dict(username='test-user')) assert json.loads(self.webflash(r))['status'] == 'error' - email = M.EmailAddress.find(dict(email=email_address, claimed_by_user_id=user._id)).first() + email = M.EmailAddress.find(dict( + email_encrypted=M.EmailAddress.encrypted_email(email_address), + claimed_by_user_id=user._id)).first() assert not email.confirmed @patch('allura.tasks.mail_tasks.sendsimplemail') @@ -582,7 +601,9 @@ def test_verify_addr_correct_session(self, gen_message_id, sendsimplemail): # test-user claimed email address user = M.User.query.get(username='test-user') user.claim_address(email_address) - email = M.EmailAddress.find(dict(email=email_address, claimed_by_user_id=user._id)).first() + email = M.EmailAddress.find(dict( + email_encrypted=M.EmailAddress.encrypted_email(email_address), + claimed_by_user_id=user._id)).first() email.confirmed = False ThreadLocalODMSession.flush_all() @@ -1081,11 +1102,13 @@ def test_verify_email(self): )) r = r.follow() user = M.User.query.get(username='aaa') - em = M.EmailAddress.get(email='[email protected]') + em = M.EmailAddress.get( + email_encrypted=M.EmailAddress.encrypted_email('[email protected]')) assert user._id == em.claimed_by_user_id r = self.app.get('/auth/verify_addr', params=dict(a=em.nonce)) user = M.User.query.get(username='aaa') - em = M.EmailAddress.get(email='[email protected]') + em = M.EmailAddress.get( + email_encrypted=M.EmailAddress.encrypted_email('[email protected]')) assert not user.pending assert em.confirmed assert user.get_pref('email_address') @@ -1722,7 +1745,8 @@ def test_only_primary_email_reset_allowed(self, gen_message_id, sendmail): user.claim_address(self.test_primary_email) user.set_pref('email_address', self.test_primary_email) - email = M.EmailAddress.find({'email': self.test_primary_email}).first() + email = M.EmailAddress.find({ + 'email_encrypted': M.EmailAddress.encrypted_email(self.test_primary_email)}).first() email.confirmed = True ThreadLocalODMSession.flush_all() @@ -1743,7 +1767,8 @@ def test_non_primary_email_reset_allowed(self, gen_message_id, sendmail): email1 = M.EmailAddress.find({'claimed_by_user_id': user._id}).first() user.claim_address(self.test_primary_email) user.set_pref('email_address', self.test_primary_email) - email = M.EmailAddress.find({'email': self.test_primary_email}).first() + email = M.EmailAddress.find({ + 'email_encrypted': M.EmailAddress.encrypted_email(self.test_primary_email)}).first() email.confirmed = True ThreadLocalODMSession.flush_all() with h.push_config(config, **{'auth.allow_non_primary_email_password_reset': 'true'}): diff --git a/Allura/allura/tests/functional/test_root.py b/Allura/allura/tests/functional/test_root.py index 3fadfa247..29faa8ebd 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/functional/test_site_admin.py b/Allura/allura/tests/functional/test_site_admin.py index ae3b358f4..22fe2c63f 100644 --- a/Allura/allura/tests/functional/test_site_admin.py +++ b/Allura/allura/tests/functional/test_site_admin.py @@ -659,7 +659,8 @@ def test_emails(self): extra_environ=dict(username='test-admin')) r = self.app.get('/nf/admin/user/test-user') assert '[email protected]' in r - em = M.EmailAddress.get(email='[email protected]') + em = M.EmailAddress.get( + email_encrypted=M.EmailAddress.encrypted_email('[email protected]')) assert em.confirmed is True user = M.User.query.get(username='test-user') assert user.get_pref('email_address') == '[email protected]' @@ -674,7 +675,8 @@ def test_emails(self): extra_environ=dict(username='test-admin')) r = self.app.get('/nf/admin/user/test-user') assert '[email protected]' in r - em = M.EmailAddress.get(email='[email protected]') + em = M.EmailAddress.get( + email_encrypted=M.EmailAddress.encrypted_email('[email protected]')) assert em.confirmed is True user = M.User.query.get(username='test-user') assert user.get_pref('email_address') == '[email protected]' diff --git a/Allura/allura/tests/model/test_auth.py b/Allura/allura/tests/model/test_auth.py index 839b9912e..8fbd2a451 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]') @@ -78,21 +81,22 @@ def selftest_email_address_lookup_helpers(): ThreadLocalODMSession.flush_all() assert addr.email == '[email protected]' - assert M.EmailAddress.get(email='[email protected]') == addr - assert M.EmailAddress.get(email='[email protected]') == addr - assert M.EmailAddress.get(email='[email protected]') is None - assert M.EmailAddress.get(email=None) is None - assert M.EmailAddress.get(email='[email protected]') == nobody + assert M.EmailAddress.get(email_encrypted=M.EmailAddress.encrypted_email('[email protected]')) == addr + assert M.EmailAddress.get(email_encrypted=M.EmailAddress.encrypted_email('[email protected]')) == addr + assert M.EmailAddress.get(email_encrypted=M.EmailAddress.encrypted_email('[email protected]')) is None + assert M.EmailAddress.get(email_encrypted=M.EmailAddress.encrypted_email('[email protected]')) == nobody # invalid email returns None, but not [email protected] as before - assert M.EmailAddress.get(email='invalid') is None - - assert M.EmailAddress.find(dict(email='[email protected]')).all() == [addr] - assert M.EmailAddress.find(dict(email='[email protected]')).all() == [addr] - assert M.EmailAddress.find(dict(email='[email protected]')).all() == [] - assert M.EmailAddress.find(dict(email=None)).all() == [] - assert M.EmailAddress.find(dict(email='[email protected]')).all() == [nobody] + assert M.EmailAddress.encrypted_email('invalid') is None + + assert M.EmailAddress.find(dict( + email_encrypted=M.EmailAddress.encrypted_email('[email protected]'))).all() == [addr] + assert M.EmailAddress.find(dict( + email_encrypted=M.EmailAddress.encrypted_email('[email protected]'))).all() == [addr] + assert M.EmailAddress.find(dict(email_encrypted=M.EmailAddress.encrypted_email('[email protected]'))).all() == [] + assert M.EmailAddress.find(dict( + email_encrypted=M.EmailAddress.encrypted_email('[email protected]'))).all() == [nobody] # invalid email returns empty query, but not [email protected] as before - assert M.EmailAddress.find(dict(email='invalid')).all() == [] + assert M.EmailAddress.encrypted_email('invalid') is None def test_email_address_canonical(self): assert M.EmailAddress.canonical('[email protected]') == \ diff --git a/Allura/allura/tests/test_mail_util.py b/Allura/allura/tests/test_mail_util.py index be44647a2..e659f788c 100644 --- a/Allura/allura/tests/test_mail_util.py +++ b/Allura/allura/tests/test_mail_util.py @@ -283,42 +283,42 @@ class TestIdentifySender: @mock.patch('allura.model.EmailAddress') def test_arg(self, EA): - EA.canonical = lambda e: e + EA.encrypted_email.side_effect = lambda e: e EA.get.side_effect = [ mock.Mock(claimed_by_user_id=True, claimed_by_user=lambda: 'user')] assert identify_sender(None, 'arg', None, None) == 'user' - EA.get.assert_called_once_with(email='arg', confirmed=True) + EA.get.assert_called_once_with(email_encrypted='arg', confirmed=True) @mock.patch('allura.model.EmailAddress') def test_header(self, EA): - EA.canonical = lambda e: e + EA.encrypted_email.side_effect = lambda e: e EA.get.side_effect = [ None, mock.Mock(claimed_by_user_id=True, claimed_by_user=lambda: 'user')] assert ( identify_sender(None, 'arg', {'From': 'from'}, None) == 'user') assert (EA.get.call_args_list == - [mock.call(email='arg', confirmed=True), mock.call(email='from')]) + [mock.call(email_encrypted='arg', confirmed=True), mock.call(email_encrypted='from')]) @mock.patch('allura.model.User') @mock.patch('allura.model.EmailAddress') def test_no_header(self, EA, User): anon = User.anonymous() - EA.canonical = lambda e: e + EA.encrypted_email.side_effect = lambda e: e EA.get.side_effect = [ None, mock.Mock(claimed_by_user_id=True, claimed_by_user=lambda: 'user')] assert identify_sender(None, 'arg', {}, None) == anon - assert EA.get.call_args_list == [mock.call(email='arg', confirmed=True)] + assert EA.get.call_args_list == [mock.call(email_encrypted='arg', confirmed=True)] @mock.patch('allura.model.User') @mock.patch('allura.model.EmailAddress') def test_no_match(self, EA, User): anon = User.anonymous() - EA.canonical = lambda e: e + EA.encrypted_email.side_effect = lambda e: e EA.get.side_effect = [None, None] assert ( identify_sender(None, 'arg', {'From': 'from'}, None) == anon) assert (EA.get.call_args_list == - [mock.call(email='arg', confirmed=True), mock.call(email='from')]) + [mock.call(email_encrypted='arg', confirmed=True), mock.call(email_encrypted='from')]) def test_parse_message_id(): diff --git a/Allura/allura/websetup/bootstrap.py b/Allura/allura/websetup/bootstrap.py index c7a903f9e..98d0df38e 100644 --- a/Allura/allura/websetup/bootstrap.py +++ b/Allura/allura/websetup/bootstrap.py @@ -207,7 +207,8 @@ def set_nbhd_wiki_content(nbhd_proj, content): u_admin.claim_address('[email protected]') ThreadLocalODMSession.flush_all() - admin_email = M.EmailAddress.get(email='[email protected]') + admin_email = M.EmailAddress.get( + email_encrypted=M.EmailAddress.encrypted_email('[email protected]')) admin_email.confirmed = True admin_email.confirmed_date = datetime.utcnow() else: @@ -328,7 +329,7 @@ def create_user(display_name, username=None, password='foo', make_project=False) email = username+"@allura.local" user.claim_address(email) from allura.model.auth import EmailAddress - kw = {"email": email} + kw = {"email_encrypted": EmailAddress.encrypted_email(email)} em = EmailAddress.get(**kw) em.confirmed = True em.confirmed_date = datetime.utcnow() diff --git a/ForgeUserStats/forgeuserstats/tests/test_model.py b/ForgeUserStats/forgeuserstats/tests/test_model.py index 6fa47db9e..ab0ad8af1 100644 --- a/ForgeUserStats/forgeuserstats/tests/test_model.py +++ b/ForgeUserStats/forgeuserstats/tests/test_model.py @@ -370,7 +370,8 @@ def test_commit_stats(self): self.user.set_password('testpassword') self.user.claim_address('[email protected]') - addr = M.EmailAddress.get(email='[email protected]') + addr = M.EmailAddress.get( + email_encrypted=M.EmailAddress.encrypted_email('[email protected]')) addr.confirmed = True repo_dir = pkg_file('forgeuserstats', 'tests/data') diff --git a/ForgeUserStats/forgeuserstats/tests/test_stats.py b/ForgeUserStats/forgeuserstats/tests/test_stats.py index 321a98337..59f4904ae 100644 --- a/ForgeUserStats/forgeuserstats/tests/test_stats.py +++ b/ForgeUserStats/forgeuserstats/tests/test_stats.py @@ -190,7 +190,8 @@ def setup_method(self, method): user = User.by_username('test-admin') user.set_password('testpassword') user.claim_address('[email protected]') - addr = M.EmailAddress.get(email='[email protected]') + addr = M.EmailAddress.get( + email_encrypted=M.EmailAddress.encrypted_email('[email protected]')) addr.confirmed = True self.setup_with_tools()
