django-oidc-provider/oidc_provider/views.py

239 lines
8.2 KiB
Python
Raw Normal View History

import logging
2016-10-03 15:54:54 +00:00
from Cryptodome.PublicKey import RSA
from django.contrib.auth.views import redirect_to_login, logout
2015-07-31 17:19:53 +00:00
from django.core.urlresolvers import reverse
from django.http import JsonResponse
2014-12-19 15:27:43 +00:00
from django.shortcuts import render
from django.template.loader import render_to_string
2014-12-19 15:27:43 +00:00
from django.views.decorators.http import require_http_methods
from django.views.generic import View
2015-07-13 20:34:43 +00:00
from jwkest import long_to_base64
2016-02-15 20:21:46 +00:00
from oidc_provider.lib.claims import StandardScopeClaims
from oidc_provider.lib.endpoints.authorize import AuthorizeEndpoint
from oidc_provider.lib.endpoints.token import TokenEndpoint
from oidc_provider.lib.errors import (
AuthorizeError,
ClientIdError,
RedirectUriError,
TokenError,
)
2016-05-25 21:58:58 +00:00
from oidc_provider.lib.utils.common import redirect, get_site_url, get_issuer
from oidc_provider.lib.utils.oauth2 import protected_resource_view
2016-04-25 20:33:52 +00:00
from oidc_provider.models import RESPONSE_TYPE_CHOICES, RSAKey
from oidc_provider import settings
2014-12-19 15:27:43 +00:00
2015-06-19 20:46:00 +00:00
logger = logging.getLogger(__name__)
2014-12-19 15:27:43 +00:00
class AuthorizeView(View):
def get(self, request, *args, **kwargs):
authorize = AuthorizeEndpoint(request)
try:
authorize.validate_params()
if request.user.is_authenticated():
2015-03-19 17:04:32 +00:00
# Check if there's a hook setted.
hook_resp = settings.get('OIDC_AFTER_USERLOGIN_HOOK', import_str=True)(
2015-03-19 17:04:32 +00:00
request=request, user=request.user,
client=authorize.client)
if hook_resp:
return hook_resp
if settings.get('OIDC_SKIP_CONSENT_ALWAYS') and not (authorize.client.client_type == 'public') \
and not (authorize.params['prompt'] == 'consent'):
2016-02-01 17:34:39 +00:00
return redirect(authorize.create_response_uri())
2015-06-24 15:40:00 +00:00
if settings.get('OIDC_SKIP_CONSENT_ENABLE'):
# Check if user previously give consent.
if authorize.client_has_user_consent() and not (authorize.client.client_type == 'public') \
and not (authorize.params['prompt'] == 'consent'):
2016-02-01 17:34:39 +00:00
return redirect(authorize.create_response_uri())
if authorize.params['prompt'] == 'none':
raise AuthorizeError(authorize.params['redirect_uri'], 'interaction_required', authorize.grant_type)
2016-06-01 15:09:40 +00:00
if authorize.params['prompt'] == 'login':
2016-06-01 15:09:40 +00:00
return redirect_to_login(request.get_full_path())
if authorize.params['prompt'] == 'select_account':
2016-06-01 15:09:40 +00:00
# TODO: see how we can support multiple accounts for the end-user.
raise AuthorizeError(authorize.params['redirect_uri'], 'account_selection_required', authorize.grant_type)
2016-06-01 15:09:40 +00:00
# Generate hidden inputs for the form.
context = {
2014-12-19 15:27:43 +00:00
'params': authorize.params,
}
2016-06-01 15:09:40 +00:00
hidden_inputs = render_to_string('oidc_provider/hidden_inputs.html', context)
2015-03-19 17:04:32 +00:00
# Remove `openid` from scope list
# since we don't need to print it.
if 'openid' in authorize.params['scope']:
authorize.params['scope'].remove('openid')
context = {
2014-12-19 15:27:43 +00:00
'client': authorize.client,
'hidden_inputs': hidden_inputs,
'params': authorize.params,
'scopes': authorize.get_scopes_information(),
2014-12-19 15:27:43 +00:00
}
2015-02-18 18:07:22 +00:00
return render(request, 'oidc_provider/authorize.html', context)
2014-12-19 15:27:43 +00:00
else:
if authorize.params['prompt'] == 'none':
raise AuthorizeError(authorize.params['redirect_uri'], 'login_required', authorize.grant_type)
2016-04-13 20:19:37 +00:00
return redirect_to_login(request.get_full_path())
2014-12-19 15:27:43 +00:00
except (ClientIdError, RedirectUriError) as error:
context = {
'error': error.error,
'description': error.description,
}
2015-02-18 18:07:22 +00:00
return render(request, 'oidc_provider/error.html', context)
2014-12-19 15:27:43 +00:00
except (AuthorizeError) as error:
uri = error.create_uri(
authorize.params['redirect_uri'],
authorize.params['state'])
2014-12-19 15:27:43 +00:00
return redirect(uri)
2014-12-19 15:27:43 +00:00
def post(self, request, *args, **kwargs):
authorize = AuthorizeEndpoint(request)
2015-06-15 19:04:44 +00:00
try:
authorize.validate_params()
2016-06-01 15:09:40 +00:00
2016-04-13 20:19:37 +00:00
if not request.POST.get('allow'):
raise AuthorizeError(authorize.params['redirect_uri'],
2015-06-15 19:04:44 +00:00
'access_denied',
authorize.grant_type)
# Save the user consent given to the client.
authorize.set_client_user_consent()
2014-12-19 15:27:43 +00:00
uri = authorize.create_response_uri()
return redirect(uri)
2014-12-19 15:27:43 +00:00
except (AuthorizeError) as error:
2015-01-30 20:20:36 +00:00
uri = error.create_uri(
authorize.params['redirect_uri'],
authorize.params['state'])
2014-12-19 15:27:43 +00:00
return redirect(uri)
2014-12-19 15:27:43 +00:00
2014-12-19 15:27:43 +00:00
class TokenView(View):
def post(self, request, *args, **kwargs):
token = TokenEndpoint(request)
try:
token.validate_params()
dic = token.create_response_dic()
return TokenEndpoint.response(dic)
except (TokenError) as error:
return TokenEndpoint.response(error.create_dict(), status=400)
2014-12-19 15:27:43 +00:00
@require_http_methods(['GET', 'POST'])
@protected_resource_view(['openid'])
def userinfo(request, *args, **kwargs):
"""
Create a diccionary with all the requested claims about the End-User.
See: http://openid.net/specs/openid-connect-core-1_0.html#UserInfoResponse
2014-12-19 15:27:43 +00:00
Return a diccionary.
"""
token = kwargs['token']
2014-12-19 15:27:43 +00:00
dic = {
'sub': token.id_token.get('sub'),
}
2014-12-19 15:27:43 +00:00
standard_claims = StandardScopeClaims(token.user, token.scope)
dic.update(standard_claims.create_response_dic())
2016-07-07 15:50:27 +00:00
if settings.get('OIDC_EXTRA_SCOPE_CLAIMS'):
extra_claims = settings.get('OIDC_EXTRA_SCOPE_CLAIMS', import_str=True)(token.user, token.scope)
dic.update(extra_claims.create_response_dic())
response = JsonResponse(dic, status=200)
2016-09-06 18:37:23 +00:00
response['Access-Control-Allow-Origin'] = '*'
response['Cache-Control'] = 'no-store'
response['Pragma'] = 'no-cache'
2016-05-30 16:28:07 +00:00
return response
class ProviderInfoView(View):
def get(self, request, *args, **kwargs):
2015-07-31 17:19:53 +00:00
dic = dict()
2016-05-25 21:58:58 +00:00
site_url = get_site_url(request=request)
dic['issuer'] = get_issuer(site_url=site_url, request=request)
2015-07-31 17:19:53 +00:00
2016-05-25 21:58:58 +00:00
dic['authorization_endpoint'] = site_url + reverse('oidc_provider:authorize')
dic['token_endpoint'] = site_url + reverse('oidc_provider:token')
dic['userinfo_endpoint'] = site_url + reverse('oidc_provider:userinfo')
dic['end_session_endpoint'] = site_url + reverse('oidc_provider:logout')
2015-07-31 17:19:53 +00:00
2016-04-25 20:33:52 +00:00
types_supported = [x[0] for x in RESPONSE_TYPE_CHOICES]
2015-07-31 17:19:53 +00:00
dic['response_types_supported'] = types_supported
2016-05-25 21:58:58 +00:00
dic['jwks_uri'] = site_url + reverse('oidc_provider:jwks')
2015-07-31 17:19:53 +00:00
dic['id_token_signing_alg_values_supported'] = ['HS256', 'RS256']
2015-07-31 17:19:53 +00:00
# See: http://openid.net/specs/openid-connect-core-1_0.html#SubjectIDTypes
dic['subject_types_supported'] = ['public']
2016-08-08 17:54:40 +00:00
dic['token_endpoint_auth_methods_supported'] = ['client_secret_post',
'client_secret_basic']
response = JsonResponse(dic)
response['Access-Control-Allow-Origin'] = '*'
return response
2015-07-13 20:34:43 +00:00
class JwksView(View):
def get(self, request, *args, **kwargs):
dic = dict(keys=[])
for rsakey in RSAKey.objects.all():
2016-08-08 17:54:40 +00:00
public_key = RSA.importKey(rsakey.key).publickey()
dic['keys'].append({
'kty': 'RSA',
'alg': 'RS256',
'use': 'sig',
'kid': rsakey.kid,
'n': long_to_base64(public_key.n),
'e': long_to_base64(public_key.e),
})
2015-07-13 20:34:43 +00:00
response = JsonResponse(dic)
response['Access-Control-Allow-Origin'] = '*'
return response
class LogoutView(View):
2015-07-31 17:19:53 +00:00
def get(self, request, *args, **kwargs):
# We should actually verify if the requested redirect URI is safe
return logout(request, next_page=request.GET.get('post_logout_redirect_uri'))