Produits Entr'ouvert » Authentic 2

    FLOW_RESOURCE_OWNER_CRED = 3

        (FLOW_RESOURCE_OWNER_CRED, _('resource owner password credentials')),

    '''Create user info dictionary'''

import math

from django.contrib.auth import authenticate

from authentic2 import app_settings as a2_app_settings

from authentic2.exponential_retry_timeout import ExponentialRetryTimeout

    if client.authorization_flow == client.FLOW_RESOURCE_OWNER_CRED:

        return authorization_error(request, 'auth_homepage',

                                   'unauthorized_client',

                                   error_description='authz endpoint is not '

                                   'part of resource owner password credential '

                                   'grant type')

def access_denied(desc=None):

    content = {

        'error': 'access_denied',

    }

    if desc:

        content['desc'] = desc

    return HttpResponseBadRequest(json.dumps(content), content_type='application/json')

def unauthorized_client(desc=None):

    content = {

        'error': 'unauthorized_client',

    }

    if desc:

        content['desc'] = desc

    return HttpResponseBadRequest(json.dumps(content), content_type='application/json')

def invalid_client(desc=None):

    content = {

        'error': 'invalid_client',

    }

    if desc:

        content['desc'] = desc

    return HttpResponseBadRequest(json.dumps(content), content_type='application/json')

    if grant_type == 'authorization_code':

        code = request.POST.get('code')

        if code is None:

            return invalid_request('missing code')

        try:

            oidc_code = models.OIDCCode.objects.select_related().get(uuid=code)

        except models.OIDCCode.DoesNotExist:

            return invalid_request('invalid code')

        if not oidc_code.is_valid():

            return invalid_request('code has expired or user is disconnected')

        client = authenticate_client(request, client=oidc_code.client)

        if client is None:

            return HttpResponse('unauthenticated', status=401)

        # delete immediately

        models.OIDCCode.objects.filter(uuid=code).delete()

        redirect_uri = request.POST.get('redirect_uri')

        if oidc_code.redirect_uri != redirect_uri:

            return invalid_request('invalid redirect_uri')

        expires_in = 3600 * 8

        access_token = models.OIDCAccessToken.objects.create(

            client=client,

            user=oidc_code.user,

            scopes=oidc_code.scopes,

            session_key=oidc_code.session_key,

            expired=oidc_code.created + datetime.timedelta(seconds=expires_in))

        start = now()

        acr = '0'

        if (oidc_code.nonce is not None

                and last_authentication_event(session=oidc_code.session).get('nonce') == oidc_code.nonce):

            acr = '1'

        # prefill id_token with user info

        id_token = utils.create_user_info(

            request,

            client,

            oidc_code.user,

            oidc_code.scope_set(),

            id_token=True)

        id_token.update({

            'iss': utils.get_issuer(request),

            'sub': utils.make_sub(client, oidc_code.user),

            'aud': client.client_id,

            'exp': timestamp_from_datetime(start + idtoken_duration(client)),

            'iat': timestamp_from_datetime(start),

            'auth_time': timestamp_from_datetime(oidc_code.auth_time),

            'acr': acr,

        })

        if oidc_code.nonce is not None:

            id_token['nonce'] = oidc_code.nonce

        response = HttpResponse(json.dumps({

            'access_token': six.text_type(access_token.uuid),

            'token_type': 'Bearer',

            'expires_in': expires_in,

            'id_token': utils.make_idtoken(client, id_token),

        }), content_type='application/json')

    elif grant_type == 'password':

        if request.META.get('CONTENT_TYPE') != 'application/x-www-form-urlencoded':

            return invalid_request(

                    'wrong content type \'%s\'.  request content type must be '

                    '\'application/x-www-form-urlencoded\'')

        username = request.POST.get('username')

        scope = request.POST.get('scope', '')

        if not all((username, request.POST.get('password'))):

            return invalid_request(

                    'request must bear both username and password as '

                    'parameters using the "application/x-www-form-urlencoded" '

                    'media type')

        client = authenticate_client(request, client=None)

        if not client:

            return invalid_client(

                'client authentication failed')

        if client.authorization_flow != models.OIDCClient.FLOW_RESOURCE_OWNER_CRED:

            return unauthorized_client(

                'client is not configured for resource owner password '

                'credential grant')

        exponential_backoff = ExponentialRetryTimeout(

            key_prefix='idp-oidc-ro-cred-grant',

            duration=a2_app_settings.A2_LOGIN_EXPONENTIAL_RETRY_TIMEOUT_DURATION,

            factor=a2_app_settings.A2_LOGIN_EXPONENTIAL_RETRY_TIMEOUT_FACTOR)

        backoff_keys = (username, client.client_id, request.META['REMOTE_ADDR'])

        seconds_to_wait = exponential_backoff.seconds_to_wait(*backoff_keys)

        if seconds_to_wait > a2_app_settings.A2_LOGIN_EXPONENTIAL_RETRY_TIMEOUT_MAX_DURATION:

            seconds_to_wait = a2_app_settings.A2_LOGIN_EXPONENTIAL_RETRY_TIMEOUT_MAX_DURATION

        if seconds_to_wait:

            return invalid_request(

                'too many attempts with erroneous RO password, you must wait '

                '%s seconds to try again.' % int(math.ceil(seconds_to_wait)))

        user = authenticate(request, username=username, password=request.POST.get('password'))

        if not user:

            exponential_backoff.failure(*backoff_keys)

            return access_denied(

                    'invalid resource owner credentials')

        exponential_backoff.success(*backoff_keys)

        start = now()

        id_token = utils.create_user_info(

            request,

            client,

            user,

            scope,

            id_token=True)

        id_token.update({

            'iss': utils.get_issuer(request),

            'aud': client.client_id,

            'exp': timestamp_from_datetime(start + idtoken_duration(client)),

            'iat': timestamp_from_datetime(start),

            'auth_time': timestamp_from_datetime(start),

            'acr': '0', # XXX Corner cases where the authn context class ref is required?

        })

        response = HttpResponse(json.dumps({

            'id_token': utils.make_idtoken(client, id_token),

        }), content_type='application/json')

    else:

        return invalid_request(

                'grant_type must be either authorization_code or password')

@pytest.fixture

def cleartext_pw_user(db, ou1):

    return create_user(username='user', first_name=u'Jôhn', last_name=u'Dôe',

                       email='user@example.net', ou=get_default_ou(),

                       password='auie1234!')

from authentic2_idp_oidc.utils import (make_sub, get_first_rsa_sig_key,

        base64url)

def test_resource_owner_password_credential_grant(app, oidc_client, admin, cleartext_pw_user, role_random):

    oidc_client.authorization_flow = OIDCClient.FLOW_RESOURCE_OWNER_CRED

    oidc_client.save()

    token_url = make_url('oidc-token')

    if oidc_client.idtoken_algo == OIDCClient.ALGO_HMAC:

        jwk = JWK(kty='oct', k=base64url(oidc_client.client_secret.encode('utf-8')))

    elif oidc_client.idtoken_algo == OIDCClient.ALGO_RSA:

        jwk = get_first_rsa_sig_key()

    # 1. test in-request client credentials

    params = {

        'client_id': oidc_client.client_id,

        'client_secret': oidc_client.client_secret,

        'grant_type': 'password',

        'username': cleartext_pw_user.username,

        'password': u'auie1234!',

    }

    response = app.post(token_url, params=params)

    assert 'id_token' in response.json

    token = response.json['id_token']

    header, payload, signature = token.split('.')

    jwt = JWT()

    jwt.deserialize(token, key=jwk)

    claims = json.loads(jwt.claims)

    # xxx already verified by jwcrypto deserialization?

    assert all(claims.get(key) for key in ('acr', 'aud', 'auth_time', 'exp',

            'iat', 'iss', 'sub'))

    # 2. test basic authz

    params.pop('client_id')

    params.pop('client_secret')

    response = app.post(

            token_url, params=params,

            headers=client_authentication_headers(oidc_client))

    assert 'id_token' in response.json

    token = response.json['id_token']

    header, payload, signature = token.split('.')

    jwt = JWT()

    jwt.deserialize(token, key=jwk)

    claims = json.loads(jwt.claims)

    # xxx already verified by jwcrypto deserialization?

    assert all(claims.get(key) for key in ('acr', 'aud', 'auth_time', 'exp',

            'iat', 'iss', 'sub'))

def test_resource_owner_password_credential_grant_throttling(app, oidc_client, admin, cleartext_pw_user, role_random, settings, freezer):

    settings.A2_LOGIN_EXPONENTIAL_RETRY_TIMEOUT_DURATION = 2

    oidc_client.authorization_flow = OIDCClient.FLOW_RESOURCE_OWNER_CRED

    oidc_client.save()

    token_url = make_url('oidc-token')

    params = {

        'client_id': oidc_client.client_id,

        'client_secret': oidc_client.client_secret,

        'grant_type': 'password',

        'username': cleartext_pw_user.username,

        'password': u'SurelyNotTheRightPassword',

    }

    attempts = 0

    while attempts < 100:

        before = now()

        response = app.post(token_url, params=params, status=400)

        attempts += 1

        if attempts >= 10:

            assert response.json['error'] == 'invalid_request'

            assert 'too many attempts with erroneous RO password' in response.json['desc']

            # XXX parse backoff value announced in request.body

    # freeze some time after backoff delay expiration

    today = datetime.date.today()

    dayafter = today + datetime.timedelta(days=2)

    freezer.move_to(dayafter.strftime('%Y-%m-%d'))

    # obtain a successful login

    params['password'] = u'auie1234!'

    response = app.post(token_url, params=params, status=200)

    assert 'id_token' in response.json

def test_resource_owner_password_credential_grant_invalid_client(app, oidc_client, admin, cleartext_pw_user, role_random, settings):

    oidc_client.authorization_flow = OIDCClient.FLOW_RESOURCE_OWNER_CRED

    oidc_client.save()

    params = {

        'client_id': oidc_client.client_id,

        'client_secret': 'tryingthis', # Nope, wrong secret

        'grant_type': 'password',

        'username': cleartext_pw_user.username,

        'password': u'auie1234!',

    }

    token_url = make_url('oidc-token')

    response = app.post(token_url, params=params, status=400)

    assert response.json['error'] == 'invalid_client'

    assert response.json['desc'] == 'client authentication failed'

def test_resource_owner_password_credential_grant_unauthz_client(app, oidc_client, admin, cleartext_pw_user, role_random, settings):

    params = {

        'client_id': oidc_client.client_id,

        'client_secret': oidc_client.client_secret,

        'grant_type': 'password',

        'username': cleartext_pw_user.username,

        'password': u'auie1234!',

    }

    token_url = make_url('oidc-token')

    response = app.post(token_url, params=params, status=400)

    assert response.json['error'] == 'unauthorized_client'

    assert 'client is not configured for resource owner'in response.json['desc']

def test_resource_owner_password_credential_grant_invalid_content_type(app, oidc_client, admin, cleartext_pw_user, role_random, settings):

    oidc_client.authorization_flow = OIDCClient.FLOW_RESOURCE_OWNER_CRED

    oidc_client.save()

    params = {

        'client_id': oidc_client.client_id,

        'client_secret': oidc_client.client_secret,

        'grant_type': 'password',

        'username': cleartext_pw_user.username,

        'password': u'auie1234!',

    }

    token_url = make_url('oidc-token')

    response = app.post(

            token_url, params=params, content_type='multipart/form-data',

            status=400)

    assert response.json['error'] == 'invalid_request'

    assert 'wrong content type' in response.json['desc']