Produits Entr'ouvert » Authentic 2

        # map claims to attributes or user fields

        # mapping is done before eventual creation of user as EMAIL_IS_UNIQUE needs to know if the

        # mapping will provide some mail to us

        ou_map = {ou.slug: ou for ou in OrganizationalUnit.cached()}

        user_ou = provider.ou

        user_info = None

        save_user = False

        mappings = []

        context = id_token_content.copy()

                    context.update(user_info or {})

        # check for required claims

        for claim_mapping in provider.claim_mappings.all():

            claim = claim_mapping.claim

            if claim_mapping.required:

                if '{{' in claim or '{%' in claim:

                    logger.warning('claim \'%r\' is templated, it cannot be set as required')

                elif claim_mapping.idtoken_claim and claim not in id_token:

                    logger.warning(

                        'auth_oidc: cannot create user missing required claim %r in id_token (%r)',

                        claim,

                        id_token,

                    )

                    return None

                elif not user_info or claim not in user_info:

                    logger.warning(

                        'auth_oidc: cannot create user missing required claim %r in user_info (%r)',

                        claim,

                        user_info,

                    )

                    return None

        User = get_user_model()

        user = None

        if provider.strategy == models.OIDCProvider.STRATEGY_FIND_UUID:

            # use the OP sub to find an user by UUUID

            # it means OP and RP share the same account base and OP is passing its UUID as sub

            try:

                user = User.objects.get(uuid=id_token.sub, is_active=True)

            except User.DoesNotExist:

                pass

            else:

                logger.info('auth_oidc: found user using UUID (=sub) "%s": %s', id_token.sub, user)

        elif provider.strategy == models.OIDCProvider.STRATEGY_FIND_USERNAME:

            users = User.objects.filter(username=id_token.sub, is_active=True).order_by('pk')

            if not users:

                logger.warning('auth_oidc: user with username (=sub) "%s" not found', id_token.sub)

            else:

                user = users[0]

                logger.info('auth_oidc: found user using username (=sub) "%s": %s', id_token.sub, user)

        elif provider.strategy == models.OIDCProvider.STRATEGY_FIND_EMAIL:

            if email and id_token.sub != email:

                logger.warning(

                    'auth_oidc: email claim (%s) does not match subject identifier (%s) yet STRATEGY_FIND_EMAIL, fallback on email claim.',

                    email,

                    id_token.sub,

                )

            elif not email:

                logger.warning(

                    'auth_oidc: email claim absent yet STRATEGY_FIND_EMAIL is set, using subject identifier (%s) instead',

                    id_token.sub,

                )

                email = id_token.sub

            if not email:

                logger.error(

                    'auth_oidc: email lookup activated for provider "%s" yet no email received', provider

                )

            users = User.objects.filter(email__iexact=email, is_active=True)

            if not app_settings.A2_EMAIL_IS_UNIQUE and provider.ou:

                users = users.filter(ou=provider.ou)

            Lock.lock_email(email)

            try:

                user = users.get()

            except User.DoesNotExist:

                logger.warning('auth_oidc: user with email "%s" not found', email)

            else:

                logger.info('auth_oidc: found user using email "%s": %s', email, user)

        else:

            try:

                user = User.objects.get(

                    oidc_account__provider=provider, oidc_account__sub=id_token.sub, is_active=True

                )

            except User.DoesNotExist:

                pass

            else:

                logger.info('auth_oidc: found user using with sub "%s": %s', id_token.sub, user)

                    (

                        'find-email',

                        'use email claim (or sub if claim is absent) to find existing user through email',

                    ),

    STRATEGY_FIND_EMAIL = 'find-email'

        (

            STRATEGY_FIND_EMAIL,

            _('use email claim (or sub if claim is absent) to find existing user through email'),

        ),

def test_strategy_find_email(app, caplog, code, oidc_provider, oidc_provider_jwkset, simple_user):

    OIDCClaimMapping.objects.all().delete()

    OIDCClaimMapping.objects.create(

        provider=oidc_provider,

        claim='email',

        attribute='email',

        idtoken_claim=False,  # served by user_info endpoint

    )

    oidc_provider.strategy = oidc_provider.STRATEGY_FIND_EMAIL

    oidc_provider.save()

    oidc_provider.ou.email_is_unique = True

    oidc_provider.ou.save()

    assert User.objects.count() == 1

    response = app.get('/').maybe_follow()

    assert oidc_provider.name in response.text

    response = response.click(oidc_provider.name)

    location = urllib.parse.urlparse(response.location)

    query = QueryDict(location.query)

    state = query['state']

    nonce = query['nonce']

    with utils.check_log(caplog, 'cannot create user'):

        with oidc_provider_mock(oidc_provider, oidc_provider_jwkset, code, nonce=nonce):

            response = app.get(login_callback_url(oidc_provider), params={'code': code, 'state': state})

    simple_user.email = 'sub@example.com'

    simple_user.save()

    with utils.check_log(caplog, 'cannot create user'):

        with oidc_provider_mock(

            oidc_provider, oidc_provider_jwkset, code, sub='sub@example.com', nonce=nonce

        ):

            response = app.get(login_callback_url(oidc_provider), params={'code': code, 'state': state})

    simple_user.email = 'john.doe@example.com'

    simple_user.save()

    with utils.check_log(caplog, 'found user using email'):

        with oidc_provider_mock(oidc_provider, oidc_provider_jwkset, code, nonce=nonce):

            response = app.get(login_callback_url(oidc_provider), params={'code': code, 'state': state})

    assert urllib.parse.urlparse(response['Location']).path == '/'

    assert User.objects.count() == 1

    user = User.objects.get()

    # verify user was not modified

    assert user.username == 'user'

    assert user.first_name == 'Jôhn'

    assert user.last_name == 'Dôe'

    assert user.email == 'john.doe@example.com'

    assert user.attributes.first_name == 'Jôhn'

    assert user.attributes.last_name == 'Dôe'