views: automatically link user with existing email if email is unique (fixes #18763)

Email must be unique if settings.A2_EMAIL_IS_UNIQUE is True or
get_default_ou().email_is_unique is True.

src/authentic2_auth_fc/views.py

@@ -10,8 +10,7 @@ from requests_oauthlib import OAuth2Session
 from django.views.generic import View, FormView
 from django.views.generic.detail import SingleObjectMixin
 from django.http import HttpResponseRedirect, Http404
-from django.contrib.auth import authenticate, REDIRECT_FIELD_NAME
-from django.contrib.auth.decorators import user_passes_test
+from django.contrib.auth import authenticate, REDIRECT_FIELD_NAME, get_user_model
 from django.contrib import messages
 from django.shortcuts import resolve_url, render
 from django.utils.translation import ugettext as _
@@ -22,10 +21,10 @@ from django.core.cache import InvalidCacheBackendError, caches
 from django.core.exceptions import PermissionDenied
 from django.core.urlresolvers import reverse
 from django.forms import Form
-from django.conf import settings
 
 from authentic2 import app_settings as a2_app_settings
 from authentic2 import utils as a2_utils
+from authentic2.a2_rbac.utils import get_default_ou
 
 from . import app_settings, models, utils
 
@@ -352,11 +351,32 @@ class LoginOrLinkView(PopupViewMixin, FcOAuthSessionViewMixin, View):
                 messages.info(request, _('Your local account has been updated.'))
             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 = authenticate(sub=self.sub, user_info=self.user_info,
                             token=self.token)
+        if not user and self.user_info.get('email') and email_is_unique:
+            email = self.user_info['email']
+            User = get_user_model()
+            qs = User.objects.filter(email=email)
+            if not a2_app_settings.A2_EMAIL_IS_UNIQUE and default_ou.email_is_unique:
+                qs = qs.filter(ou=default_ou)
+            if qs.exists():
+                # there should not be multiple accounts with the same mail
+                if len(qs) > 1:
+                    self.logger.warning(u'multiple accounts with the same mail %s, %s', email,
+                                        list(qs))
+                # link existing user to received sub
+                models.FcAccount.objects.get_or_create(
+                    defaults={'token': json.dumps(self.token)},
+                    sub=self.sub, user=qs[0])
+                user = authenticate(sub=self.sub, user_info=self.user_info,
+                                    token=self.token)
         if user:
             a2_utils.login(request, user, 'france-connect')
             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.logger.info('logged in using fc sub %s', self.sub)
             return self.redirect(request)

tests/test_auth_fc.py

@@ -99,3 +99,48 @@ def test_login(app, fc_settings, caplog, exp):
         assert User.objects.count() == 0
     else:
         assert User.objects.count() == 1
+
+
+def test_login_email_is_unique(app, fc_settings, caplog):
+    callback = reverse('fc-login-or-link')
+    response = app.get(callback, status=302)
+    location = response['Location']
+    state = check_authorization_url(location)
+
+    @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'
+        assert callback in parsed['redirect_uri']
+        id_token = {
+            'sub': '1234',
+            'aud': 'xxx',
+            'nonce': state,
+            'exp': timestamp_from_datetime(now() + datetime.timedelta(seconds=1000)),
+            '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': '1234',
+            'family_name': u'Frédérique',
+            'given_name': u'Ÿuñe',
+            'email': 'john.doe@example.com',
+        })
+
+    User.objects.create(email='john.doe@example.com', first_name='John', last_name='Doe')
+    fc_settings.A2_EMAIL_IS_UNIQUE = True
+    with httmock.HTTMock(access_token_response, user_info_response):
+        response = app.get(callback + '?code=zzz&state=%s' % state, status=302)
+    assert User.objects.count() == 1