auth_oidc: show a warning message if target user is already linked to another provider (#65692)

src/authentic2_auth_oidc/backends.py

@@ -19,10 +19,13 @@ import logging
 
 import requests
 from django.conf import settings
+from django.contrib import messages
 from django.contrib.auth import get_user_model
 from django.contrib.auth.backends import ModelBackend
-from django.db.transaction import atomic
+from django.db import IntegrityError
+from django.db.transaction import atomic, set_rollback
 from django.utils.timezone import now
+from django.utils.translation import gettext as _
 from jwcrypto.jwk import JWK
 from jwcrypto.jwt import JWT
 
@@ -39,7 +42,7 @@ from . import models, utils
 class OIDCBackend(ModelBackend):
     # pylint: disable=arguments-renamed
     def authenticate(self, request, access_token=None, id_token=None, nonce=None, provider=None):
-        with atomic(savepoint=False):
+        with atomic():
             return self._authenticate(
                 request, access_token=access_token, id_token=id_token, nonce=nonce, provider=provider
             )
@@ -313,9 +316,22 @@ class OIDCBackend(ModelBackend):
                     user = User.objects.create(ou=provider.ou, email=email or '')
                     user.set_unusable_password()
                     created_user = True
-                oidc_account, created = models.OIDCAccount.objects.get_or_create(
-                    provider=provider, user=user, defaults={'sub': id_token.sub}
-                )
+                try:
+                    oidc_account, created = models.OIDCAccount.objects.get_or_create(
+                        provider=provider, user=user, defaults={'sub': id_token.sub}
+                    )
+                except IntegrityError:
+                    set_rollback(True)
+                    logger.warning('auth_oidc: email %s is already linked to another provider.', email)
+                    if request:
+                        messages.warning(
+                            request,
+                            _(
+                                'Your email is already linked to another SSO account, please contact an administrator.'
+                            ),
+                        )
+                        return None
+
                 if not created and oidc_account.sub != id_token.sub:
                     logger.info(
                         'auth_oidc: changed user %s sub from %s to %s (issuer %s)',
@@ -327,6 +343,8 @@ class OIDCBackend(ModelBackend):
                     oidc_account.sub = id_token.sub
                     oidc_account.save()
             else:
+                if request:
+                    messages.warning(request, _('No user found'))
                 logger.warning(
                     'auth_oidc: cannot create user for sub %r as issuer %r does not allow it',
                     id_token.sub,

src/authentic2_auth_oidc/views.py

@@ -295,8 +295,6 @@ class LoginCallback(View):
                     'provider_pk': provider.pk,
                 }
             )
-        else:
-            messages.warning(request, _('No user found'))
         return self.continue_to_next_url(request)
 
     errors = {

tests/test_auth_oidc.py

@@ -1361,3 +1361,36 @@ def test_oidc_unicity_contraint_issuer(db):
     with pytest.raises(IntegrityError):
         with transaction.atomic():
             OIDCProvider.objects.create(issuer='test', slug='d')
+
+
+def test_double_link(app, caplog, code, simple_user, oidc_provider_jwkset):
+    ou = get_default_ou()
+    ou.email_is_unique = True
+    ou.save()
+    provider1 = make_oidc_provider(name='provider1', jwkset=oidc_provider_jwkset)
+    provider2 = make_oidc_provider(name='provider2', jwkset=oidc_provider_jwkset)
+
+    OIDCAccount.objects.create(provider=provider2, sub='1234', user=simple_user)
+
+    response = app.get('/').maybe_follow()
+    response = response.click('provider1')
+    location = urllib.parse.urlparse(response.location)
+    query = QueryDict(location.query)
+    state = query['state']
+    nonce = query['nonce']
+
+    # sub=john.doe
+    with utils.check_log(caplog, 'auth_oidc: email user@example.net is already linked'):
+        with oidc_provider_mock(
+            provider1,
+            oidc_provider_jwkset,
+            code,
+            nonce=nonce,
+            extra_id_token={'email': simple_user.email},
+            extra_user_info={'email': simple_user.email},
+        ):
+            response = app.get(login_callback_url(provider1), params={'code': code, 'state': state})
+        response = response.maybe_follow()
+    warnings = response.pyquery('.warning')
+    assert len(warnings) == 1
+    assert 'Your email is already linked' in warnings.text()