Merge branch 'master' into github-actions

Author: mhindery

CHANGES

@@ -19,6 +19,10 @@ Thanks to plumdog
 
 Thanks to plumdog
 
+UNRELEASED
+----------
+- Allowed creating Users with multiple required fields.
+
 0.17.1 (2018-07-16)
 ----------
 - A 403 (permission denied) is now raised if a SAMLResponse is replayed, instead of 500.

README.rst

@@ -315,6 +315,24 @@ setting::
   SAML_CONFIG_LOADER = 'python.path.to.your.callable'
 
 
+Custom error handler
+....................
+
+When an error occurs during the authentication flow, djangosaml2 will render
+a simple error page with an error message and status code. You can customize
+this behaviour by specifying the path to your own error handler in the settings:
+
+  SAML_ACS_FAILURE_RESPONSE_FUNCTION = 'python.path.to.your.view'
+
+This should be a view which takes a request, optional exception which occured
+and status code, and returns a response to serve the user. E.g. The default
+implementation looks like this::
+
+  def template_failure(request, exception=None, **kwargs):
+      """ Renders a simple template with an error message. """
+      return render(request, 'djangosaml2/login_error.html', {'exception': exception}, status=kwargs.get('status', 403))
+
+
 User attributes
 ---------------
 

djangosaml2/acs_failures.py

@@ -4,18 +4,9 @@
 # produce an output suitable for end user in case of SAML failure.
 #
 
-from django.core.exceptions import PermissionDenied
 from django.shortcuts import render
 
 
-def template_failure(request, status=403, **kwargs):
-    """ Renders a SAML-specific template with general authentication error description. """
-    return render(request, 'djangosaml2/login_error.html', status=status)
-
-
-def exception_failure(request, exc_class=PermissionDenied, **kwargs):
-    """ Rather than using a custom SAML specific template that is rendered on failure,
-    this makes use of a standard exception handling machinery present in Django
-    and thus ends up rendering a project-wide error page for Permission Denied exceptions.
-    """
-    raise exc_class
+def template_failure(request, exception=None, status=403, **kwargs):
+    """ Renders a simple template with an error message. """
+    return render(request, 'djangosaml2/login_error.html', {'exception': exception}, status=status)

djangosaml2/backends.py

@@ -147,19 +147,21 @@ def _get_or_create_saml2_user(self, main_attribute, attributes, attribute_mappin
                      main_attribute)
         django_user_main_attribute = self.get_django_user_main_attribute()
         user_query_args = self.get_user_query_args(main_attribute)
-        user_create_defaults = {django_user_main_attribute: main_attribute}
 
         User = get_saml_user_model()
+        built = False
         try:
-            user, created = User.objects.get_or_create(
-                defaults=user_create_defaults, **user_query_args)
+            user = User.objects.get(**user_query_args)
+        except User.DoesNotExist:
+            user = User(**{django_user_main_attribute: main_attribute})
+            built = True
         except MultipleObjectsReturned:
             logger.error("There are more than one user with %s = %s",
                          django_user_main_attribute, main_attribute)
             return None
 
-        if created:
-            logger.debug('New user created')
+        if built:
+            logger.debug('Configuring new user "%s"', main_attribute)
             user = self.configure_user(user, attributes, attribute_mapping)
         else:
             logger.debug('User updated')

djangosaml2/tests/__init__.py

@@ -18,6 +18,7 @@
 import datetime
 import re
 import sys
+
 from unittest import skip
 
 from django.conf import settings
@@ -519,6 +520,43 @@ def test_idplist_templatetag(self):
 
         self.assertEqual(rendered, expected)
 
+    def test_sigalg_not_passed_when_not_signing_request(self):
+        # monkey patch SAML configuration
+        settings.SAML_CONFIG = conf.create_conf(
+            sp_host='sp.example.com',
+            idp_hosts=['idp.example.com'],
+            metadata_file='remote_metadata_one_idp.xml',
+        )
+
+        with mock.patch(
+            'djangosaml2.views.Saml2Client.prepare_for_authenticate',
+            return_value=('session_id', {'url': 'fake'}),
+
+        ) as prepare_for_auth_mock:
+            self.client.get(reverse('saml2_login'))
+        prepare_for_auth_mock.assert_called_once()
+        _args, kwargs = prepare_for_auth_mock.call_args
+        self.assertNotIn('sigalg', kwargs)
+
+    def test_sigalg_passed_when_signing_request(self):
+        # monkey patch SAML configuration
+        settings.SAML_CONFIG = conf.create_conf(
+            sp_host='sp.example.com',
+            idp_hosts=['idp.example.com'],
+            metadata_file='remote_metadata_one_idp.xml',
+        )
+
+        settings.SAML_CONFIG['service']['sp']['authn_requests_signed'] = True
+        with mock.patch(
+            'djangosaml2.views.Saml2Client.prepare_for_authenticate',
+            return_value=('session_id', {'url': 'fake'}),
+
+        ) as prepare_for_auth_mock:
+            self.client.get(reverse('saml2_login'))
+        prepare_for_auth_mock.assert_called_once()
+        _args, kwargs = prepare_for_auth_mock.call_args
+        self.assertIn('sigalg', kwargs)
+
 
 def test_config_loader(request):
     config = SPConfig()

djangosaml2/tests/attribute-maps/django_saml_uri.py

@@ -0,0 +1,19 @@
+X500ATTR_OID = 'urn:oid:2.5.4.'
+PKCS_9 = 'urn:oid:1.2.840.113549.1.9.1.'
+UCL_DIR_PILOT = 'urn:oid:0.9.2342.19200300.100.1.'
+
+MAP = {
+    'identifier': 'urn:oasis:names:tc:SAML:2.0:attrname-format:uri',
+    'fro': {
+        X500ATTR_OID+'3': 'first_name', # cn
+        X500ATTR_OID+'4': 'last_name', # sn
+        PKCS_9+'1': 'email',
+        UCL_DIR_PILOT+'1': 'uid',
+    },
+    'to': {
+        'first_name': X500ATTR_OID+'3',
+        'last_name': X500ATTR_OID+'4',
+        'email' : PKCS_9+'1',
+        'uid': UCL_DIR_PILOT+'1',
+    }
+}

djangosaml2/views.py

@@ -29,13 +29,20 @@
 from django.utils.http import is_safe_url
 from django.views.decorators.csrf import csrf_exempt
 from django.views.decorators.http import require_POST
-from saml2 import BINDING_HTTP_POST, BINDING_HTTP_REDIRECT
+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,
+    UnsolicitedResponse, StatusNoAuthnContext,
+)
+from saml2.mdstore import SourceNotFound
 from saml2.sigver import MissingKey
 from saml2.validate import ResponseLifetimeExceed, ToEarly
 from saml2.xmldsig import (  # support for SHA1 is required by spec
@@ -122,7 +129,15 @@ def login(request,
                     })
 
     selected_idp = request.GET.get('idp', None)
-    conf = get_config(config_loader_path, request)
+    try:
+        conf = get_config(config_loader_path, request)
+    except SourceNotFound as excp:
+        msg = ('Error, IdP EntityID was not found '
+               'in metadata: {}')
+        logger.exception(msg.format(excp))
+        return HttpResponse(msg.format(('Please contact '
+                                        'technical support.')),
+                            status=500)
 
     kwargs = {}
     # pysaml needs a string otherwise: "cannot serialize True (type bool)"
@@ -176,17 +191,18 @@ def login(request,
     logger.debug('Redirecting user to the IdP via %s binding.', binding)
     if binding == BINDING_HTTP_REDIRECT:
         try:
-            # do not sign the xml itself, instead use the sigalg to
-            # generate the signature as a URL param
-            sig_alg_option_map = {'sha1': SIG_RSA_SHA1,
-                                  'sha256': SIG_RSA_SHA256}
-            sig_alg_option = getattr(conf, '_sp_authn_requests_signed_alg', 'sha1')
-            sigalg = sig_alg_option_map[sig_alg_option] if sign_requests else None
             nsprefix = get_namespace_prefixes()
+            if sign_requests:
+                # do not sign the xml itself, instead use the sigalg to
+                # generate the signature as a URL param
+                sig_alg_option_map = {'sha1': SIG_RSA_SHA1,
+                                      'sha256': SIG_RSA_SHA256}
+                sig_alg_option = getattr(conf, '_sp_authn_requests_signed_alg', 'sha1')
+                kwargs["sigalg"] = sig_alg_option_map[sig_alg_option]
             session_id, result = client.prepare_for_authenticate(
                 entityid=selected_idp, relay_state=came_from,
-                binding=binding, sign=False, sigalg=sigalg,
-                nsprefix=nsprefix, **kwargs)
+                binding=binding, sign=False, nsprefix=nsprefix,
+                **kwargs)
         except TypeError as e:
             logger.error('Unable to know which IdP to use')
             return HttpResponse(str(e))
@@ -269,34 +285,34 @@ def assertion_consumer_service(request,
 
     try:
         response = client.parse_authn_request_response(xmlstr, BINDING_HTTP_POST, outstanding_queries)
-    except (StatusError, ToEarly):
+    except (StatusError, ToEarly) as e:
         logger.exception("Error processing SAML Assertion.")
-        return fail_acs_response(request)
-    except ResponseLifetimeExceed:
+        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)
-    except SignatureError:
+        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)
-    except StatusAuthnFailed:
+        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)
-    except StatusRequestDenied:
+        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)
-    except StatusNoAuthnContext:
+        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)
-    except MissingKey:
+        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)
-    except UnsolicitedResponse:
+        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)
+        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, exc_class=SuspiciousOperation)
+        return fail_acs_response(request, status=400, exception=SuspiciousOperation('Unknown SAML2 error'))
 
     session_id = response.session_id()
     oq_cache.delete(session_id)
@@ -316,7 +332,7 @@ def assertion_consumer_service(request,
                              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)
-        raise PermissionDenied
+        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'])
@@ -376,7 +392,13 @@ def logout(request, config_loader_path=None):
             'The session does not contain the subject id for user %s',
             request.user)
 
-    result = client.global_logout(subject_id)
+    try:
+        result = client.global_logout(subject_id)
+    except LogoutError as exp:
+        logger.exception('Error Handled - SLO not supported by IDP: {}'.format(exp))
+        auth.logout(request)
+        state.sync()
+        return HttpResponseRedirect('/')
 
     state.sync()
 

setup.py

@@ -15,7 +15,6 @@
 
 import codecs
 import os
-import sys
 from setuptools import setup, find_packages
 
 
@@ -63,4 +62,8 @@ def read(*rnames):
         'Django>=2.2',
         'pysaml2>=4.6.0',
         ],
+    tests_require=[
+        # Provides assert_called_once.
+        'mock;python_version < "3.6"',
+    ]
     )

tests/testprofiles/models.py

@@ -30,3 +30,13 @@ class StandaloneUserModel(models.Model):
     USERNAME_FIELD.
     """
     username = models.CharField(max_length=30, unique=True)
+
+
+class RequiredFieldUser(models.Model):
+    email = models.EmailField(unique=True)
+    email_verified = models.BooleanField()
+
+    USERNAME_FIELD = 'email'
+
+    def set_unusable_password(self):
+        pass

tests/testprofiles/tests.py

@@ -122,6 +122,26 @@ def test_invalid_model_attribute_log(self):
             logs.output,
         )
 
+    @override_settings(AUTH_USER_MODEL='testprofiles.RequiredFieldUser')
+    def test_create_user_with_required_fields(self):
+        backend = Saml2Backend()
+        attribute_mapping = {
+            'mail': ['email'],
+            'mail_verified': ['email_verified']
+        }
+        attributes = {
+            'mail': ['[email protected]'],
+            'mail_verified': [True],
+        }
+        # User creation does not fail if several fields are required.
+        user = backend._get_or_create_saml2_user(
+            '[email protected]',
+            attributes,
+            attribute_mapping,
+        )
+        self.assertEquals(user.email, '[email protected]')
+        self.assertIs(user.email_verified, True)
+
     def test_django_user_main_attribute(self):
         backend = Saml2Backend()