From 58e78d50e32a13d71a1465f2052e54b84eb06f5e Mon Sep 17 00:00:00 2001
From: Nicolas ROCHE <nroche@entrouvert.com>
Date: Tue, 28 Jul 2020 17:58:20 +0200
Subject: [PATCH] auth_fc: do not update non-uniq email returned by FC (#45199)

---
 src/authentic2_auth_fc/views.py | 37 ++++++++++++--
 tests/auth_fc/test_auth_fc.py   | 85 +++++++++++++++++++++++++++++++++
 2 files changed, 119 insertions(+), 3 deletions(-)

diff --git a/src/authentic2_auth_fc/views.py b/src/authentic2_auth_fc/views.py
index e64a11ce..6630b72f 100644
--- a/src/authentic2_auth_fc/views.py
+++ b/src/authentic2_auth_fc/views.py
@@ -340,20 +340,51 @@ 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'])
+
+        # do not update email if it is not unique
+        default_ou = get_default_ou()
+        email_is_unique = a2_app_settings.A2_EMAIL_IS_UNIQUE or default_ou.email_is_unique
+        if email_is_unique:
+            mappings = app_settings.user_info_mappings
+            mapping = {'ref': mappings['email']}
+            email = ''
+            try:
+                email = utils.mapping_to_value(mapping, self.user_info)
+            except (ValueError, KeyError, NotImplementedError):
+                pass
+            if email:
+                User = get_user_model()
+                qs = User.objects.filter(email__iexact=email)
+                if not a2_app_settings.A2_EMAIL_IS_UNIQUE and default_ou.email_is_unique:
+                    qs = qs.filter(ou=default_ou)
+
+                qs = qs.exclude(uuid=self.fc_account.user.uuid)
+                if qs.exists():
+                # there should not be other accounts with the same mail
+                    if self.logger:
+                        self.logger.error(u'other account already use the same mail %s, %s',
+                                          email, list(qs))
+                    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.'
+                    ) % email)
+                    del self.user_info[mapping['ref']]
+
         utils.apply_user_info_mappings(self.fc_account.user, self.user_info)
         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
@@ -385,17 +416,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)
 
         default_ou = get_default_ou()
         email_is_unique = a2_app_settings.A2_EMAIL_IS_UNIQUE or default_ou.email_is_unique
         user = a2_utils.authenticate(
             request,
             sub=self.sub,
             user_info=self.user_info,
@@ -447,17 +478,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..0be7e116 100644
--- a/tests/auth_fc/test_auth_fc.py
+++ b/tests/auth_fc/test_auth_fc.py
@@ -650,8 +650,93 @@ 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_email_from_fc(app, fc_settings, caplog):
+    # 1. we have 2 accounts havinf one linked to FC
+    # 2. mail is updated if FC return a new one,
+    # 3. but only if the mail is not already used
+
+    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 = 'foo@example.com'
+    EMAIL3 = '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='{}')
+    user2 = User.objects.create(email=EMAIL3, 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': email,
+        })
+
+    fc_settings.A2_EMAIL_IS_UNIQUE = True
+    email = EMAIL2
+    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 == 'foo@example.com'
+    assert app.session['_auth_user_id']
+    response = response.follow()
+    assert not response.html.find('li', {'class': 'warning'})
+
+    email = EMAIL3
+    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 == 'foo@example.com'
+    assert app.session['_auth_user_id']
+    response = response.follow()
+    assert 'already used by another account' in response.html.find(
+        'li', {'class': 'warning'}).text
+
+    fc_settings.A2_EMAIL_IS_UNIQUE = False
+    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 == 'john.doe@example.com'
+    assert app.session['_auth_user_id']
-- 
2.27.0

