Produits Entr'ouvert » Authentic 2

    return nonce

    '''Record an authentication event in the session and in the database, in

       later version the database persistence can be removed'''

    from . import models

        'who': unicode(request.user)[:80],

        'how': how,

    }

    if nonce:

        kwargs['nonce'] = nonce

        event['nonce'] = nonce

    return None

    '''Login a user model, record the authentication event and redirect to next

       URL or settings.LOGIN_REDIRECT_URL.'''

    from . import hooks

    if constants.LAST_LOGIN_SESSION_KEY not in request.session:

        request.session[constants.LAST_LOGIN_SESSION_KEY] = \

            localize(to_current_timezone(last_login), True)

    hooks.call_hooks('event', name='login', user=user, how=how, service=service_slug)

    return continue_to_next_url(request, **kwargs)

from jwcrypto.jwt import JWT

from jwcrypto.jwk import JWK

from django.utils.timezone import now

from django.contrib.auth import get_user_model

from django.contrib.auth.backends import ModelBackend

class OIDCBackend(ModelBackend):

        logger = logging.getLogger(__name__)

        if id_token is None:

            return

                               duration)

                return None

        User = get_user_model()

        user = None

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

            return redirect(request, settings.LOGIN_REDIRECT_URL)

        try:

            issuer = oidc_state.get('issuer')

            provider = get_provider_by_issuer(issuer)

        except models.OIDCProvider.DoesNotExist:

            messages.warning(request, _('Unknown OpenID connect issuer'))

            return self.continue_to_next_url()

        logger.info(u'got token response %s', result)

        access_token = result.get('access_token')

        if user:

            # remember last tokens for logout

            tokens = request.session.setdefault('auth_oidc', {}).setdefault('tokens', [])

                'provider_pk': provider.pk,

            })

            request.session.modified = True

        else:

            messages.warning(request, _('No user found'))

        return self.continue_to_next_url()

                                        has_providers)

from authentic2_auth_oidc.models import OIDCProvider, OIDCClaimMapping

from authentic2.models import AttributeValue

from authentic2.a2_rbac.utils import get_default_ou

from authentic2.crypto import base64url_encode

def oidc_provider_mock(oidc_provider, oidc_provider_jwkset, code, extra_id_token=None,

    token_endpoint = urlparse.urlparse(oidc_provider.token_endpoint)

    userinfo_endpoint = urlparse.urlparse(oidc_provider.userinfo_endpoint)

    token_revocation_endpoint = urlparse.urlparse(oidc_provider.token_revocation_endpoint)

                'aud': str(oidc_provider.client_id),

                'exp': timestamp_from_datetime(now() + datetime.timedelta(seconds=10)),

            }

            if extra_id_token:

                id_token.update(extra_id_token)

    assert query['client_id'] == str(oidc_provider.client_id)

    assert query['scope'] == 'openid'

    assert query['redirect_uri'] == 'http://testserver' + reverse('oidc-login-callback')

    if oidc_provider.claims_parameter_supported:

        claims = json.loads(query['claims'])

        with oidc_provider_mock(oidc_provider, oidc_provider_jwkset, code,

                                extra_id_token={'aud': 'zz'}):

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

    assert not hooks.auth_oidc_backend_modify_user

    with utils.check_log(caplog, 'created user'):

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

    assert len(hooks.auth_oidc_backend_modify_user) == 1

    assert set(hooks.auth_oidc_backend_modify_user[0]['kwargs']) >= set(['user', 'provider', 'user_info', 'id_token', 'access_token'])

    assert user.attributes.last_name == 'Doe'

    assert AttributeValue.objects.filter(content='John', verified=True).count() == 1

    assert AttributeValue.objects.filter(content='Doe', verified=False).count() == 1

    with oidc_provider_mock(oidc_provider, oidc_provider_jwkset, code,

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

    assert AttributeValue.objects.filter(content='Doe', verified=False).count() == 0

    assert AttributeValue.objects.filter(content='Doe', verified=True).count() == 1

    with oidc_provider_mock(oidc_provider, oidc_provider_jwkset, code,

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

    assert User.objects.count() == 1

    user = User.objects.get()

    assert user.ou == cassis

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

    assert User.objects.count() == 1

    user = User.objects.get()

    response = response.click(oidc_provider.name)

    location = urlparse.urlparse(response.location)

    query = check_simple_qs(urlparse.parse_qs(location.query))

    # sub=john.doe, MUST not work

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

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

    # sub=simple_user.uuid MUST work

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

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

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

    response = app.get(reverse('account_management'))

    with utils.check_log(caplog, 'revoked token from OIDC'):

            response = response.click(href='logout')

    assert 'https://idp.example.com/logout' in response.content