From 7286c4c6f337521b385d5ba4908699b732c7b6f7 Mon Sep 17 00:00:00 2001
From: Nicolas ROCHE <nroche@entrouvert.com>
Date: Thu, 30 Jul 2020 11:55:25 +0200
Subject: [PATCH 2/2] auth_fc: do not update non-uniq email returned by FC
 (#45199)

---
 src/authentic2_auth_fc/utils.py | 11 ++++
 src/authentic2_auth_fc/views.py | 15 +++--
 tests/auth_fc/test_auth_fc.py   | 99 +++++++++++++++++++++++++++++++++
 3 files changed, 121 insertions(+), 4 deletions(-)

diff --git a/src/authentic2_auth_fc/utils.py b/src/authentic2_auth_fc/utils.py
index 3d33b1b8..f10b14e8 100644
--- a/src/authentic2_auth_fc/utils.py
+++ b/src/authentic2_auth_fc/utils.py
@@ -160,31 +160,41 @@ def users_having_email(email):
 
 
 def apply_user_info_mappings(user, user_info):
     assert user
     assert user_info
 
     logger = logging.getLogger(__name__)
     mappings = app_settings.user_info_mappings
+    status = {'err': 0}
 
     save_user = False
     tags = set()
     for attribute, mapping in mappings.items():
         # normalize mapping to dictionaries: if string, convert into a simple reference
         if hasattr(mapping, 'format'):
             mapping = {'ref': mapping}
         try:
             value = mapping_to_value(mapping, user_info)
         except (ValueError, KeyError, NotImplementedError) as e:
             logger.warning(u'auth_fc: cannot apply mapping %s <- %r: %s', attribute, mapping, e)
             continue
         if mapping.get('if-tag') and mapping['if-tag'] not in tags:
             continue
 
+        # do not update email if it is not unique
+        if attribute == 'email' and email_is_unique():
+            qs = users_having_email(value)
+            qs = qs.exclude(uuid=user.uuid)
+            if qs.exists():
+                logger.error(u'other accounts already use the same mail %s, %s', value, list(qs))
+                status = {'err': 1, 'email': value}
+                continue
+
         if attribute == 'password':
             if mapping.get('if-empty') and user.has_usable_password():
                 continue
             save_user = True
             user.set_password(value)
         elif hasattr(user.attributes, attribute):
             if mapping.get('if-empty') and getattr(user.attributes, attribute):
                 continue
@@ -199,16 +209,17 @@ def apply_user_info_mappings(user, user_info):
             setattr(user, attribute, value)
         else:
             logger.warning(u'auth_fc: unknown attribute in user_info mapping: %s', attribute)
             continue
         if mapping.get('tag'):
             tags.add(mapping['tag'])
     if save_user:
         user.save()
+    return status
 
 
 def requests_retry_session(
     retries=3,
     backoff_factor=0.5,
     status_forcelist=(500, 502, 504),
     session=None,
 ):
diff --git a/src/authentic2_auth_fc/views.py b/src/authentic2_auth_fc/views.py
index 38d7df9e..778bbc21 100644
--- a/src/authentic2_auth_fc/views.py
+++ b/src/authentic2_auth_fc/views.py
@@ -339,21 +339,28 @@ class PopupViewMixin(object):
 
 
 class LoginOrLinkView(PopupViewMixin, FcOAuthSessionViewMixin, View):
     '''Login with FC, if the FC account is already linked, connect this user,
        if a user is logged link the user to this account, otherwise display an
        error message.
     '''
 
-    def update_user_info(self):
+    def update_user_info(self, request):
         self.fc_account.token = json.dumps(self.token)
         self.fc_account.user_info = json.dumps(self.user_info)
         self.fc_account.save(update_fields=['token', 'user_info'])
-        utils.apply_user_info_mappings(self.fc_account.user, self.user_info)
+
+        status = utils.apply_user_info_mappings(self.fc_account.user, self.user_info)
+        if status['err'] and status.get('email'):
+            messages.warning(request, _(
+                'Your FranceConnect email address \'%s\' is already used by another '
+                'account, so we cannot update your account for you. Please ensure your '
+                'account email is still up to date using your account management page.'
+            ) % status.get('email'))
         self.logger.debug('updating user_info %s', self.fc_account.user_info)
 
     def uniqueness_check_failed(self, request):
         if request.user.is_authenticated():
             # currently logged :
             if models.FcAccount.objects.filter(user=request.user, order=0).count():
                 # cannot link because we are already linked to another FC account
                 messages.error(request,
@@ -384,17 +391,17 @@ class LoginOrLinkView(PopupViewMixin, FcOAuthSessionViewMixin, View):
 
             if created:
                 self.logger.info('fc link created sub %s', self.sub)
                 messages.info(request,
                               _('Your FranceConnect account {} has been linked.').format(self.fc_display_name))
                 hooks.call_hooks('event', name='fc-link', user=request.user, sub=self.sub, request=request)
             else:
                 messages.info(request, _('Your local account has been updated.'))
-            self.update_user_info()
+            self.update_user_info(request)
             return self.redirect(request)
 
         user = a2_utils.authenticate(
             request,
             sub=self.sub,
             user_info=self.user_info,
             token=self.token)
         if user:
@@ -440,17 +447,17 @@ class LoginOrLinkView(PopupViewMixin, FcOAuthSessionViewMixin, View):
         if user:
             views_utils.check_cookie_works(request)
             a2_utils.login(request, user, 'france-connect', service_slug=self.service_slug)
             # set session expiration policy to EXPIRE_AT_BROWSER_CLOSE
             request.session.set_expiry(0)
             self.fc_account = models.FcAccount.objects.get(sub=self.sub, user=user)
             self.fc_account.token = json.dumps(self.token)
             self.fc_account.save(update_fields=['token'])
-            self.update_user_info()
+            self.update_user_info(request)
             self.logger.info('logged in using fc sub %s', self.sub)
             return self.redirect(request)
         else:
             params = {}
             if self.service_slug:
                 params[constants.SERVICE_FIELD_NAME] = self.service_slug
             if registration:
                 return self.redirect_and_come_back(request,
diff --git a/tests/auth_fc/test_auth_fc.py b/tests/auth_fc/test_auth_fc.py
index a8d64d17..e9620911 100644
--- a/tests/auth_fc/test_auth_fc.py
+++ b/tests/auth_fc/test_auth_fc.py
@@ -650,8 +650,107 @@ def test_can_change_password(app, fc_settings, caplog, hooks):
     response = app.get(reverse('fc-logout') + '?state=' + state)
     assert path(response['Location']) == '/accounts/'
     response = response.follow()
     assert len(response.pyquery('[href*="password/change"]')) > 0
 
 
 def test_invalid_next_url(app, fc_settings, caplog, hooks):
     assert app.get('/fc/callback/?code=coin&next=JJJ72QQQ').location == 'JJJ72QQQ'
+
+
+def test_update_user_info_having_non_uniq_new_email_from_fc(app, fc_settings, caplog):
+    # We have 2 accounts, account1 is linked to FC.
+    # Account1's email is not updated if FC return the account2's email
+
+    callback = reverse('fc-login-or-link')
+    response = app.get(callback, status=302)
+    location = response['Location']
+    state = check_authorization_url(location)
+
+    EMAIL1 = 'fred@example.com'
+    EMAIL2 = 'john.doe@example.com'
+    SUB = '1234'
+    user1 = User.objects.create(email=EMAIL1, first_name='Frédérique', last_name='Ÿuñe')
+    user1.save()
+    models.FcAccount.objects.create(user=user1, sub='1234', token='xxx', user_info='{}').save()
+    user2 = User.objects.create(email=EMAIL2, first_name='John', last_name='Doe')
+    user2.save()
+
+    @httmock.urlmatch(path=r'.*/token$')
+    def access_token_response(url, request):
+        parsed = {x: y[0] for x, y in urlparse.parse_qs(request.body).items()}
+        assert set(parsed.keys()) == set(['code', 'client_id', 'client_secret', 'redirect_uri',
+                                          'grant_type'])
+        assert parsed['code'] == 'zzz'
+        assert parsed['client_id'] == 'xxx'
+        assert parsed['client_secret'] == 'yyy'
+        assert parsed['grant_type'] == 'authorization_code'
+        parsed_redirect = urlparse.urlparse(parsed['redirect_uri'])
+        parsed_callback = urlparse.urlparse(callback)
+        assert parsed_redirect.path == parsed_callback.path
+        for cb_key, cb_value in urlparse.parse_qs(parsed_callback.query).items():
+            urlparse.parse_qs(parsed_redirect.query)[cb_key] == cb_value
+        exp = now() + datetime.timedelta(seconds=1000)
+        id_token = {
+            'sub': SUB,
+            'aud': 'xxx',
+            'nonce': state,
+            'exp': int(exp.timestamp()),
+            'iss': 'https://fcp.integ01.dev-franceconnect.fr/',
+        }
+        return json.dumps({
+            'access_token': 'uuu',
+            'id_token': hmac_jwt(id_token, 'yyy')
+        })
+
+    @httmock.urlmatch(path=r'.*userinfo$')
+    def user_info_response(url, request):
+        assert request.headers['Authorization'] == 'Bearer uuu'
+        return json.dumps({
+            'sub': SUB,
+            'family_name': u'Frédérique',
+            'given_name': u'Ÿuñe',
+            'email': EMAIL2,  # get a new mail from FC
+        })
+
+    fc_settings.A2_EMAIL_IS_UNIQUE = True
+
+    # user not already authenticated
+    with httmock.HTTMock(access_token_response, user_info_response):
+        response = app.get(callback + '?code=zzz&state=%s' % state, status=302)
+    assert User.objects.count() == 2
+    assert User.objects.get(last_name='Frédérique').email == EMAIL1
+    response = response.follow()
+    assert 'already used by another account' in response.html.find(
+        'li', {'class': 'warning'}).text
+
+    # user is authenticated
+    assert app.session['_auth_user_id']
+    with httmock.HTTMock(access_token_response, user_info_response):
+        response = app.get(callback + '?code=zzz&state=%s' % state, status=302)
+    assert User.objects.count() == 2
+    assert User.objects.get(last_name='Frédérique').email == EMAIL1
+    response = response.follow()
+    assert 'already used by another account' in response.html.find(
+        'li', {'class': 'warning'}).text
+
+    fc_settings.A2_EMAIL_IS_UNIQUE = False
+    app.session.flush()
+    callback = reverse('fc-login-or-link')
+    response = app.get(callback, status=302)
+    location = response['Location']
+    state = check_authorization_url(location)
+
+    # user not already authenticated
+    with httmock.HTTMock(access_token_response, user_info_response):
+        response = app.get(callback + '?code=zzz&state=%s' % state, status=302)
+    assert User.objects.get(last_name='Frédérique').email == EMAIL2
+
+    # user is authenticated
+    user1 = User.objects.get(id=user1.id)
+    user1.email = EMAIL1
+    user1.save()
+    assert User.objects.get(last_name='Frédérique').email == EMAIL1
+    assert app.session['_auth_user_id']
+    with httmock.HTTMock(access_token_response, user_info_response):
+        response = app.get(callback + '?code=zzz&state=%s' % state, status=302)
+    assert User.objects.get(last_name='Frédérique').email == EMAIL2
-- 
2.27.0

