Skip to content

More extensible view for assertion consumer service #139

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
May 11, 2020
Merged

More extensible view for assertion consumer service #139

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions CHANGES
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,8 @@ Changes
- py38 Test fixes
- CI with Github actions
- Backend restructuring for easier subclassing
- Assertion consumer service now more extensible as a class-based view
with hooks that can be overridden by subclass implementations.

0.18.1 (2020-02-15)
----------
Expand Down
2 changes: 1 addition & 1 deletion djangosaml2/tests/urls.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@

urlpatterns = [
url(r'^login/$', views.login, name='saml2_login'),
url(r'^acs/$', views.assertion_consumer_service, name='saml2_acs'),
url(r'^acs/$', views.AssertionConsumerServiceView.as_view(), name='saml2_acs'),
url(r'^logout/$', views.logout, name='saml2_logout'),
url(r'^ls/$', views.logout_service, name='saml2_ls'),
url(r'^ls/post/$', views.logout_service_post, name='saml2_ls_post'),
Expand Down
2 changes: 1 addition & 1 deletion djangosaml2/urls.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@

urlpatterns = [
path('login/', views.login, name='saml2_login'),
path('acs/', views.assertion_consumer_service, name='saml2_acs'),
path('acs/', views.AssertionConsumerServiceView.as_view(), name='saml2_acs'),
path('logout/', views.logout, name='saml2_logout'),
path('ls/', views.logout_service, name='saml2_ls'),
path('ls/post/', views.logout_service_post, name='saml2_ls_post'),
Expand Down
236 changes: 138 additions & 98 deletions djangosaml2/views.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,15 +27,13 @@
from django.shortcuts import render
from django.template import TemplateDoesNotExist
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST
from django.views.generic import View
from django.utils.decorators import method_decorator

from saml2 import BINDING_HTTP_REDIRECT, BINDING_HTTP_POST
from saml2.client_base import LogoutError
from saml2.metadata import entity_descriptor
from saml2.ident import code, decode
from saml2.metadata import entity_descriptor
from saml2.response import (SignatureError, StatusAuthnFailed, StatusError,
StatusNoAuthnContext, StatusRequestDenied,
UnsolicitedResponse)
from saml2.s_utils import UnsupportedBinding
from saml2.response import (
StatusError, StatusAuthnFailed, SignatureError, StatusRequestDenied,
Expand Down Expand Up @@ -255,103 +253,145 @@ def login(request,
return http_response


@require_POST
@csrf_exempt
def assertion_consumer_service(request,
config_loader_path=None,
attribute_mapping=None,
create_unknown_user=None):
"""SAML Authorization Response endpoint

The IdP will send its response to this view, which
will process it with pysaml2 help and log the user
in using the custom Authorization backend
djangosaml2.backends.Saml2Backend that should be
enabled in the settings.py
class AssertionConsumerServiceView(View):
"""
The IdP will send its response to this view, which will process it using pysaml2 and
log the user in using whatever SAML authentication backend has been enabled in
settings.py. The `djangosaml2.backends.Saml2Backend` can be used for this purpose,
though some implementations may instead register their own subclasses of Saml2Backend.
"""
attribute_mapping = attribute_mapping or get_custom_setting('SAML_ATTRIBUTE_MAPPING', {'uid': ('username', )})
create_unknown_user = create_unknown_user if create_unknown_user is not None else \
get_custom_setting('SAML_CREATE_UNKNOWN_USER', True)
conf = get_config(config_loader_path, request)
xmlstr = request.POST.get('SAMLResponse')
if not xmlstr:
logger.warning('Missing "SAMLResponse" parameter in POST data.')
raise SuspiciousOperation

client = Saml2Client(conf, identity_cache=IdentityCache(request.session))
@method_decorator(csrf_exempt)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You can add the csr_exempt in a more coincise way like this:

from django.utils.decorators import method_decorator

@method_decorator(csrf_exempt, name='dispatch')
class AssertionConsumerServiceView(View):
    ...

That way you don't need override the dispatch solely to add the csrf decorator. Only the dispatch method is enough, there's no need to set this on the post method additionally.

def dispatch(self, request, *args, **kwargs):
"""
This view needs to be CSRF exempt because it is called prior to login.
"""
return super(AssertionConsumerServiceView, self).dispatch(request, *args, **kwargs)

@method_decorator(csrf_exempt)
def post(self,
request,
config_loader_path=None,
attribute_mapping=None,
create_unknown_user=None):
"""
SAML Authorization Response endpoint
"""
attribute_mapping = attribute_mapping or get_custom_setting('SAML_ATTRIBUTE_MAPPING', {'uid': ('username', )})
create_unknown_user = create_unknown_user if create_unknown_user is not None else \
get_custom_setting('SAML_CREATE_UNKNOWN_USER', True)
conf = get_config(config_loader_path, request)
try:
xmlstr = request.POST['SAMLResponse']
except KeyError:
logger.warning('Missing "SAMLResponse" parameter in POST data.')
raise SuspiciousOperation
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can you use fail_acs_response here as well?


oq_cache = OutstandingQueriesCache(request.session)
outstanding_queries = oq_cache.outstanding_queries()
client = Saml2Client(conf, identity_cache=IdentityCache(self.request.session))

oq_cache = OutstandingQueriesCache(self.request.session)
outstanding_queries = oq_cache.outstanding_queries()

try:
response = client.parse_authn_request_response(xmlstr, BINDING_HTTP_POST, outstanding_queries)
except (StatusError, ToEarly) as e:
logger.exception("Error processing SAML Assertion.")
return fail_acs_response(request, exception=e)
except ResponseLifetimeExceed as e:
logger.info("SAML Assertion is no longer valid. Possibly caused by network delay or replay attack.", exc_info=True)
return fail_acs_response(request, exception=e)
except SignatureError as e:
logger.info("Invalid or malformed SAML Assertion.", exc_info=True)
return fail_acs_response(request, exception=e)
except StatusAuthnFailed as e:
logger.info("Authentication denied for user by IdP.", exc_info=True)
return fail_acs_response(request, exception=e)
except StatusRequestDenied as e:
logger.warning("Authentication interrupted at IdP.", exc_info=True)
return fail_acs_response(request, exception=e)
except StatusNoAuthnContext as e:
logger.warning("Missing Authentication Context from IdP.", exc_info=True)
return fail_acs_response(request, exception=e)
except MissingKey as e:
logger.exception("SAML Identity Provider is not configured correctly: certificate key is missing!")
return fail_acs_response(request, exception=e)
except UnsolicitedResponse as e:
logger.exception("Received SAMLResponse when no request has been made.")
return fail_acs_response(request, exception=e)

if response is None:
logger.warning("Invalid SAML Assertion received (unknown error).")
return fail_acs_response(request, status=400, exception=SuspiciousOperation('Unknown SAML2 error'))

session_id = response.session_id()
oq_cache.delete(session_id)

# authenticate the remote user
session_info = response.session_info()

if callable(attribute_mapping):
attribute_mapping = attribute_mapping()
if callable(create_unknown_user):
create_unknown_user = create_unknown_user()

logger.debug('Trying to authenticate the user. Session info: %s', session_info)
user = auth.authenticate(request=request,
session_info=session_info,
attribute_mapping=attribute_mapping,
create_unknown_user=create_unknown_user)
if user is None:
logger.warning("Could not authenticate user received in SAML Assertion. Session info: %s", session_info)
return fail_acs_response(request, exception=PermissionDenied('No user could be authenticated.'))

auth.login(self.request, user)
_set_subject_id(self.request.session, session_info['name_id'])
logger.debug("User %s authenticated via SSO.", user)
logger.debug('Sending the post_authenticated signal')

post_authenticated.send_robust(sender=user, session_info=session_info)
self.customize_session(user, session_info)

relay_state = self.build_relay_state()
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could the build_relay_state, custom_redirect and the is_safe_url be a single method? Which takes the (request, user, session_info) and possibly whatever else is useful (e.g. whether or not to perform the is_safe_check, this might want to be customized entirely), an returns a 'definitive' url to redirect to. And then the default implementation does what is currently done. That way it's a bit clearer due there being a single method to determine the redirect, which users can override, instead of the several pieces which all do a part of it.

custom_redirect_url = self.custom_redirect(user, relay_state, session_info)
if custom_redirect_url:
return HttpResponseRedirect(custom_redirect_url)
relay_state = validate_referral_url(request, relay_state)
logger.debug('Redirecting to the RelayState: %s', relay_state)
return HttpResponseRedirect(relay_state)

def build_relay_state(self):
"""
The relay state is a URL used to redirect the user to the view where they came from.
"""
default_relay_state = get_custom_setting('ACS_DEFAULT_REDIRECT_URL',
settings.LOGIN_REDIRECT_URL)
relay_state = self.request.POST.get('RelayState', '/')
relay_state = self.customize_relay_state(relay_state)
if not relay_state:
logger.warning('The RelayState parameter exists but is empty')
relay_state = default_relay_state
return relay_state

def customize_session(self, user, session_info):
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What kind of customization do you see being applied here? What behaviour will this impact?

"""
Subclasses can use this for customized functionality around user sessions.
"""

def customize_relay_state(self, relay_state):
"""
Subclasses may override this method to implement custom logic for relay state.
"""
return relay_state

def custom_redirect(self, user, relay_state, session_info):
"""
Subclasses may override this method to implement custom logic for redirect.

For example, some sites may require user registration if the user has not
yet been provisioned.
"""
return None

try:
response = client.parse_authn_request_response(xmlstr, BINDING_HTTP_POST, outstanding_queries)
except (StatusError, ToEarly) as e:
logger.exception("Error processing SAML Assertion.")
return fail_acs_response(request, exception=e)
except ResponseLifetimeExceed as e:
logger.info("SAML Assertion is no longer valid. Possibly caused by network delay or replay attack.", exc_info=True)
return fail_acs_response(request, exception=e)
except SignatureError as e:
logger.info("Invalid or malformed SAML Assertion.", exc_info=True)
return fail_acs_response(request, exception=e)
except StatusAuthnFailed as e:
logger.info("Authentication denied for user by IdP.", exc_info=True)
return fail_acs_response(request, exception=e)
except StatusRequestDenied as e:
logger.warning("Authentication interrupted at IdP.", exc_info=True)
return fail_acs_response(request, exception=e)
except StatusNoAuthnContext as e:
logger.warning("Missing Authentication Context from IdP.", exc_info=True)
return fail_acs_response(request, exception=e)
except MissingKey as e:
logger.exception("SAML Identity Provider is not configured correctly: certificate key is missing!")
return fail_acs_response(request, exception=e)
except UnsolicitedResponse as e:
logger.exception("Received SAMLResponse when no request has been made.")
return fail_acs_response(request, exception=e)

if response is None:
logger.warning("Invalid SAML Assertion received (unknown error).")
return fail_acs_response(request, status=400, exception=SuspiciousOperation('Unknown SAML2 error'))

session_id = response.session_id()
oq_cache.delete(session_id)

# authenticate the remote user
session_info = response.session_info()

if callable(attribute_mapping):
attribute_mapping = attribute_mapping()
if callable(create_unknown_user):
create_unknown_user = create_unknown_user()

logger.debug('Trying to authenticate the user. Session info: %s', session_info)
user = auth.authenticate(request=request,
session_info=session_info,
attribute_mapping=attribute_mapping,
create_unknown_user=create_unknown_user)
if user is None:
logger.warning("Could not authenticate user received in SAML Assertion. Session info: %s", session_info)
return fail_acs_response(request, exception=PermissionDenied('No user could be authenticated.'))

auth.login(request, user)
_set_subject_id(request.session, session_info['name_id'])
logger.debug("User %s authenticated via SSO.", user)

logger.debug('Sending the post_authenticated signal')
post_authenticated.send_robust(sender=user, session_info=session_info)

# redirect the user to the view where he came from
default_relay_state = get_custom_setting('ACS_DEFAULT_REDIRECT_URL',
settings.LOGIN_REDIRECT_URL)
relay_state = request.POST.get('RelayState', default_relay_state)
if not relay_state:
logger.warning('The RelayState parameter exists but is empty')
relay_state = default_relay_state
relay_state = validate_referral_url(request, relay_state)

logger.debug('Redirecting to the RelayState: %s', relay_state)
return HttpResponseRedirect(relay_state)


@login_required
Expand Down