Produits Entr'ouvert » Authentic 2

          'django-ratelimit',

    @property

    def ACCESS_TOKEN_DURATION(self):

        return self._setting('ACCESS_TOKEN_DURATION', 3600 * 8)

    @property

    def PASSWORD_GRANT_RATELIMIT(self):

        return self._setting('PASSWORD_GRANT_RATELIMIT', '100/m')

                ('session_key', models.CharField(blank=True, max_length=128, verbose_name='session key')),

                ('authorization_flow', models.PositiveIntegerField(choices=[(1, 'authorization code'), (2, 'implicit/native'), (3, 'resource owner password credentials')], default=1, verbose_name='authorization flow')),

# -*- coding: utf-8 -*-

# Generated by Django 1.11.20 on 2020-01-22 21:58

from __future__ import unicode_literals

from django.db import migrations, models

class Migration(migrations.Migration):

    dependencies = [

        ('authentic2_idp_oidc', '0011_auto_20180808_1546'),

    ]

    operations = [

        migrations.AddField(

            model_name='oidcclient',

            name='access_token_duration',

            field=models.DurationField(blank=True, default=None, null=True, verbose_name='time during which the access token is valid'),

        ),

        migrations.AddField(

            model_name='oidcclient',

            name='scope',

            field=models.TextField(blank=True, default=b'', verbose_name='resource owner credentials grant scope'),

        ),

    ]

    FLOW_RESOURCE_OWNER_CRED = 3

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

    access_token_duration = models.DurationField(

        verbose_name=_('time during which the access token is valid'),

        blank=True,

        null=True,

        default=None)

    scope = models.TextField(

        verbose_name=_('resource owner credentials grant scope'),

        help_text=_('Permitted or default scopes (for credentials grant)'),

        default='',

        blank=True)

    def scope_set(self):

        return utils.scope_set(self.scope)

        max_length=128,

        blank=True)

    '''Create user info dictionary'''

    if 'openid' in scope_set:

        user_info['sub'] = make_sub(client, user)

import math

from django.http import (HttpResponse, HttpResponseBadRequest,

                         HttpResponseNotAllowed, JsonResponse)

from django.contrib.auth import authenticate

from ratelimit.utils import is_ratelimited

from authentic2 import app_settings as a2_app_settings

from authentic2.exponential_retry_timeout import ExponentialRetryTimeout

    return JsonResponse(metadata)

    return client.idtoken_duration or datetime.timedelta(seconds=app_settings.IDTOKEN_DURATION)

def access_token_duration(client):

    return client.access_token_duration or datetime.timedelta(seconds=app_settings.IDTOKEN_DURATION)

def allowed_scopes(client):

    return client.scope_set() or app_settings.SCOPES or ['openid', 'email', 'profile']

def is_scopes_allowed(scopes, client):

    return scopes <= set(allowed_scopes(client))

    if client.authorization_flow == client.FLOW_RESOURCE_OWNER_CRED:

        messages.warning(request, _('Client is configured for resource owner password crendetial grant type'))

        return authorization_error(request, 'auth_homepage',

                                   'unauthorized_client',

                                   error_description='authz endpoint is configured '

                                   'for resource owner password credential grant type')

    if not is_scopes_allowed(scopes, client):

            ', '.join(allowed_scopes(client)), ', '.join(scopes))

        expires_in = access_token_duration(client)

                expired=start + expires_in)

                'expires_in': expires_in.total_seconds(),

def error_response(error, error_description=None, status=400):

        'error': error,

    if error_description:

        content['error_description'] = error_description

    return JsonResponse(content, status=status)

def invalid_request_response(error_description=None):

    return error_response('invalid_request', error_description=error_description)

def access_denied_response(error_description=None):

    return error_response('access_denied', error_description=error_description)

def unauthorized_client_response(error_description=None):

    return error_response('unauthorized_client', error_description=error_description)

def invalid_client_response(error_description=None):

    return error_response('invalid_client', error_description=error_description)

def credential_grant_ratelimit_key(group, request):

    client = authenticate_client(request, client=None)

    if client:

        return client.client_id

    # return remote address when no valid client credentials have been provided

    return request.META['REMOTE_ADDR']

def idtoken_from_user_credential(request):

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

        return invalid_request_response(

            'wrong content type. request content type must be \'application/x-www-form-urlencoded\'')

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

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

    # scope is ignored, we used the configured scope

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

        return invalid_request_response(

            'request must bear both username and password as '

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

            'media type')

    if is_ratelimited(

            request, group='ro-cred-grant', increment=True,

            key=credential_grant_ratelimit_key,

            rate=app_settings.PASSWORD_GRANT_RATELIMIT):

        return invalid_request_response(

            'reached rate limitation, too many erroneous requests')

    client = authenticate_client(request, client=None)

    if not client:

        return invalid_client_response('client authentication failed')

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

        return unauthorized_client_response(

            '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)

    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_response(

            '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_response('invalid resource owner credentials')

    # limit requested scopes

    if scope is not None:

        scopes = utils.scope_set(scope) & client.scope_set()

    else:

        scopes = client.scope_set()

    exponential_backoff.success(*backoff_keys)

    start = now()

    # make access_token

    expires_in = access_token_duration(client)

    access_token = models.OIDCAccessToken.objects.create(

        client=client,

        user=user,

        scopes=' '.join(scopes),

        session_key='',

        expired=start + expires_in)

    # make id_token

    id_token = utils.create_user_info(

        request,

        client,

        user,

        scopes,

        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',

    })

    return JsonResponse({

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

        'token_type': 'Bearer',

        'expires_in': expires_in.total_seconds(),

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

    })

def tokens_from_authz_code(request):

        return invalid_request_response('missing code')

        return invalid_request_response('invalid code')

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

        return invalid_request_response('invalid redirect_uri')

    expires_in = access_token_duration(client)

        expired=oidc_code.created + expires_in)

    return JsonResponse({

        'expires_in': expires_in.total_seconds(),

    })

@setting_enabled('ENABLE', settings=app_settings)

@csrf_exempt

def token(request, *args, **kwargs):

    if request.method != 'POST':

        return HttpResponseNotAllowed(['POST'])

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

    if grant_type == 'password':

        response = idtoken_from_user_credential(request)

    elif grant_type == 'authorization_code':

        response = tokens_from_authz_code(request)

    else:

        return invalid_request_response('grant_type must be either authorization_code or password')

    return JsonResponse(user_info)

from django.core.cache import cache

from django.test.client import RequestFactory

from ratelimit.utils import is_ratelimited

from authentic2_idp_oidc.utils import (make_sub, get_first_rsa_sig_key,

        base64url)

    settings.A2_IDP_OIDC_PASSWORD_GRANT_RATELIMIT = '100/m'

        assert response.json['error_description'] == 'code has expired or user is disconnected'

        assert response.json['error_description'] == 'code has expired or user is disconnected'

def test_credentials_grant(app, oidc_client, admin, simple_user):

    cache.clear()

    oidc_client.authorization_flow = OIDCClient.FLOW_RESOURCE_OWNER_CRED

    oidc_client.scope = 'openid'

    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': simple_user.username,

        'password': simple_user.username,

    }

    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 set(claims) == set(['acr', 'aud', 'auth_time', 'exp', 'iat', 'iss', 'sub'])

    assert all(claims.values())

    # 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 set(claims) == set(['acr', 'aud', 'auth_time', 'exp', 'iat', 'iss', 'sub'])

    assert all(claims.values())

def test_credentials_grant_ratelimitation_invalid_client(

        app, oidc_client, admin, simple_user, oidc_settings, freezer):

    freezer.move_to('2020-01-01')

    cache.clear()

    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': 'notgood',

        'grant_type': 'password',

        'username': simple_user.username,

        'password': simple_user.username,

    }

    for i in range(int(oidc_settings.A2_IDP_OIDC_PASSWORD_GRANT_RATELIMIT.split('/')[0])):

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

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

        assert 'client authentication failed' in response.json['error_description']

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

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

    assert 'reached rate limitation' in response.json['error_description']

def test_credentials_grant_ratelimitation_valid_client(

        app, oidc_client, admin, simple_user, oidc_settings, freezer):

    freezer.move_to('2020-01-01')

    cache.clear()

    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': simple_user.username,

        'password': simple_user.username,

    }

    for i in range(int(oidc_settings.A2_IDP_OIDC_PASSWORD_GRANT_RATELIMIT.split('/')[0])):

        app.post(token_url, params=params)

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

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

    assert 'reached rate limitation' in response.json['error_description']

def test_credentials_grant_retrytimout(

        app, oidc_client, admin, simple_user, settings, freezer):

    freezer.move_to('2020-01-01')

    cache.clear()

    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': simple_user.username,

        'password': u'SurelyNotTheRightPassword',

    }

    attempts = 0

    while attempts < 100:

        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['error_description']

    # freeze some time after backoff delay expiration

    freezer.move_to(datetime.timedelta(days=2))

    # obtain a successful login

    params['password'] = simple_user.username

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

    assert 'id_token' in response.json

def test_credentials_grant_invalid_flow(

        app, oidc_client, admin, simple_user, settings):

    cache.clear()

    params = {

        'client_id': oidc_client.client_id,

        'client_secret': oidc_client.client_secret,

        'grant_type': 'password',

        'username': simple_user.username,

        'password': u'SurelyNotTheRightPassword',

    }

    token_url = make_url('oidc-token')

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

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

    assert 'is not configured' in response.json['error_description']

def test_credentials_grant_invalid_client(

        app, oidc_client, admin, simple_user, settings):

    cache.clear()

    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': simple_user.username,

        'password': simple_user.username,

    }

    token_url = make_url('oidc-token')

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

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

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

def test_credentials_grant_unauthz_client(

        app, oidc_client, admin, simple_user, settings):

    cache.clear()

    params = {

        'client_id': oidc_client.client_id,

        'client_secret': oidc_client.client_secret,

        'grant_type': 'password',

        'username': simple_user.username,

        'password': simple_user.username,

    }

    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['error_description']

def test_credentials_grant_invalid_content_type(

        app, oidc_client, admin, simple_user, settings):

    cache.clear()

    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': simple_user.username,

        'password': simple_user.username,

    }

    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['error_description']