auth_oidc: show a warning message if target user is already linked to another provider (#65692)
This commit is contained in:
parent
855bb179b0
commit
2480687f3f
|
@ -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,
|
||||
|
|
|
@ -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 = {
|
||||
|
|
|
@ -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()
|
||||
|
|
Loading…
Reference in New Issue