0001-idp_oidc-support-oauth2-resource-owner-password-cred.patch

Paul Marillonnet, 09 janvier 2020 18:22

Voir les différences: en ligne côte à côte

Subject: [PATCH] idp_oidc: support oauth2 resource owner password credential
 grant (#35205)

 setup.py                                |   1 +
 src/authentic2_idp_oidc/app_settings.py |   4 +
 src/authentic2_idp_oidc/models.py       |   2 +
 src/authentic2_idp_oidc/utils.py        |   2 +-
 src/authentic2_idp_oidc/views.py        | 148 ++++++++++++++++--
 tests/test_idp_oidc.py                  | 196 +++++++++++++++++++++++-
 6 files changed, 340 insertions(+), 13 deletions(-)

setup.py
122	122	'dnspython>=1.10',
123	123	'Django-Select2>5,<6',
124	124	'django-tables2>=1.0,<2.0',
	125	'django-ratelimit',
125	126	'gadjo>=0.53',
126	127	'django-import-export>=0.2.7,<=0.4.5',
127	128	'djangorestframework>=3.3,<3.5',

         def IDTOKEN_DURATION(self):
             return self._setting('IDTOKEN_DURATION', 30)
         @property
         def PASSWORD_GRANT_RATELIMIT(self):
             return self._setting('PASSWORD_GRANT_RATELIMIT', '100/m')
     app_settings = AppSettings('A2_IDP_OIDC_')
     app_settings.__name__ = __name__
     sys.modules[__name__] = app_settings

+        ]
         FLOW_AUTHORIZATION_CODE = 1
         FLOW_IMPLICIT = 2
         FLOW_RESOURCE_OWNER_CRED = 3
         FLOW_CHOICES = [
             (FLOW_AUTHORIZATION_CODE, _('authorization code')),
             (FLOW_IMPLICIT, _('implicit/native')),
             (FLOW_RESOURCE_OWNER_CRED, _('resource owner password credentials')),
+        ]
         AUTHORIZATION_MODE_BY_SERVICE = 1

     def create_user_info(request, client, user, scope_set, id_token=False):
         '''Create user info dictionnary'''
         '''Create user info dictionary'''
         user_info = {
             'sub': make_sub(client, user)
+        }

     # along with this program.  If not, see <http://www.gnu.org/licenses/>.
     import logging
     import math
     import datetime
     import json
     import base64
     import time
     from django.http import HttpResponse, HttpResponseBadRequest, HttpResponseNotAllowed
     from django.http import (HttpResponse, HttpResponseBadRequest,
             HttpResponseNotAllowed, JsonResponse)
     from django.utils import six
     from django.utils.timezone import now, utc
     from django.utils.http import urlencode
-...
     from django.views.decorators.csrf import csrf_exempt
     from django.core.urlresolvers import reverse
     from django.contrib import messages
     from django.contrib.auth import authenticate
     from django.conf import settings
     from django.utils.translation import ugettext as _
     from ratelimit.utils import is_ratelimited
     from authentic2 import app_settings as a2_app_settings
     from authentic2.decorators import setting_enabled
     from authentic2.exponential_retry_timeout import ExponentialRetryTimeout
     from authentic2.utils import (login_require, redirect, timestamp_from_datetime,
                                   last_authentication_event, make_url)
     from authentic2.views import logout as a2_logout
-...
                            redirect_uri, client_id)
             return redirect(request, 'auth_homepage')
         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')
         if not client.is_valid_redirect_uri(redirect_uri):
             messages.warning(request, _('Authorization request is invalid'))
             logger.warning(u'idp_oidc: authorization request error, unknown redirect_uri redirect_uri=%r client_id=%r',
-...
         return HttpResponseBadRequest(json.dumps(content), content_type='application/json')
     @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 != 'authorization_code':
             return invalid_request('grant_type is not authorization_code')
     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')
     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(
                         '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')
             if is_ratelimited(
                     request, group='ro-cred-grant', increment=True,
                     key=credential_grant_ratelimit_key,
                     rate=app_settings.PASSWORD_GRANT_RATELIMIT):
                 return invalid_request(
                         'reached rate limitation, too many erroneous requests')
             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)
             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',
             })
             return JsonResponse({'id_token': utils.make_idtoken(client, id_token)})
     def tokens_from_authz_code(request):
         code = request.POST.get('code')
         if code is None:
             return invalid_request('missing code')
-...
         })
         if oidc_code.nonce is not None:
             id_token['nonce'] = oidc_code.nonce
         response = HttpResponse(json.dumps({
         return JsonResponse({
             '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')
         })
     @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(
                     'grant_type must be either authorization_code or password')
         response['Cache-Control'] = 'no-store'
         response['Pragma'] = 'no-cache'
         return response

     from django.db import connection
     from django.db.migrations.executor import MigrationExecutor
     from django.utils.timezone import now
     from django.test.client import RequestFactory
     from django.contrib.auth import get_user_model
     from django.utils.six.moves.urllib import parse as urlparse
     from ratelimit.utils import is_ratelimited
     User = get_user_model()
     from authentic2.models import Attribute, AuthorizedRole
     from authentic2_idp_oidc.models import OIDCClient, OIDCAuthorization, OIDCCode, OIDCAccessToken, OIDCClaim
     from authentic2_idp_oidc.utils import make_sub
     from authentic2_idp_oidc.utils import (make_sub, get_first_rsa_sig_key,
             base64url)
     from authentic2.a2_rbac.utils import get_default_ou
     from authentic2.utils import make_url
     from authentic2_auth_oidc.utils import parse_timestamp
-...
     @pytest.fixture
     def oidc_settings(settings):
         settings.A2_IDP_OIDC_JWKSET = JWKSET
         settings.A2_IDP_OIDC_PASSWORD_GRANT_RATELIMIT = '100/m'
         return settings
-...
         response = app.get('/api/users/')
         assert len(response.json['results']) == count
     def test_resource_owner_password_credential_grant(app, oidc_client, admin, simple_user):
         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': 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 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_ratelimitation_invalid_client(app, oidc_client, admin, simple_user, oidc_settings):
         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,
+        }
         attempts = 0
         dummy_post = RequestFactory().post('/dummy')
         while attempts < 1000:
             before = now()
             attempts += 1
             ratelimited = is_ratelimited(
                     request=dummy_post, group='test-ro-cred-grant', increment=True,
                     key=lambda x, y: '127.0.0.1',
                     rate=oidc_settings.A2_IDP_OIDC_PASSWORD_GRANT_RATELIMIT)
             response = app.post(token_url, params=params, status=400)
             if not ratelimited:
                 assert response.json['error'] == 'invalid_client'
                 assert 'client authentication failed' in response.json['desc']
                 continue
             else:
                 assert response.json['error'] == 'invalid_request'
                 assert 'reached rate limitation' in response.json['desc']
                 break
         if not ratelimited:
             assert 0
     def test_resource_owner_password_credential_grant_ratelimitation_valid_client(app, oidc_client, admin, simple_user, oidc_settings):
         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,
+        }
         attempts = 0
         dummy_post = RequestFactory().post('/dummy')
         while attempts < 1000:
             before = now()
             attempts += 1
             ratelimited = is_ratelimited(
                     request=dummy_post, group='test-ro-cred-grant', increment=True,
                     key=lambda x, y: oidc_client.client_id,
                     rate=oidc_settings.A2_IDP_OIDC_PASSWORD_GRANT_RATELIMIT)
             if ratelimited:
                 response = app.post(token_url, params=params, status=400)
                 assert response.json['error'] == 'invalid_request'
                 assert 'reached rate limitation' in response.json['desc']
                 break
             else:
                 response = app.post(token_url, params=params)
         if not ratelimited:
             assert 0
     def test_resource_owner_password_credential_grant_retrytimout(app, oidc_client, admin, simple_user, 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': simple_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']
         # 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'] = simple_user.username
         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, simple_user, 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': 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['desc'] == 'client authentication failed'
     def test_resource_owner_password_credential_grant_unauthz_client(app, oidc_client, admin, simple_user, settings):
         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['desc']
     def test_resource_owner_password_credential_grant_invalid_content_type(app, oidc_client, admin, simple_user, 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': 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['desc']
+    -

Projet

Général

Profil

Produits Entr'ouvert » Authentic 2

0001-idp_oidc-support-oauth2-resource-owner-password-cred.patch