Customizabl acs error handler, more test coverage (#219)

Author: mhindery

djangosaml2/acs_failures.py

@@ -39,7 +39,8 @@
                                get_session_id_from_saml2,
                                get_subject_id_from_saml2,
                                saml2_from_httpredirect_request)
-from djangosaml2.views import finish_logout
+from djangosaml2.views import (EchoAttributesView, _set_subject_id,
+                               finish_logout)
 from saml2.config import SPConfig
 from saml2.s_utils import decode_base64_and_inflate, deflate_and_base64_encode
 
@@ -95,9 +96,6 @@ def add_outstanding_query(self, session_id, came_from):
         self.saml_session.save()
         self.client.cookies[settings.SESSION_COOKIE_NAME] = self.saml_session.session_key
 
-    def render_template(self, text):
-        return Template(text).render(Context())
-
     def b64_for_post(self, xml_text, encoding='utf-8'):
         return base64.b64encode(xml_text.encode(encoding)).decode('ascii')
 
@@ -406,6 +404,47 @@ def do_login(self):
         self.assertEqual(response.status_code, 302)
         return subject_id
 
+    def test_echo_view_no_saml_session(self):
+        settings.SAML_CONFIG = conf.create_conf(
+            sp_host='sp.example.com',
+            idp_hosts=['idp.example.com'],
+            metadata_file='remote_metadata_one_idp.xml',
+        )
+        self.do_login()
+
+        request = RequestFactory().get('/bar/foo')
+        request.COOKIES = self.client.cookies
+        request.user = User.objects.last()
+
+        middleware = SamlSessionMiddleware()
+        middleware.process_request(request)
+
+        response = EchoAttributesView.as_view()(request)
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.content.decode(), 'No active SAML identity found. Are you sure you have logged in via SAML?')
+
+    def test_echo_view_success(self):
+        settings.SAML_CONFIG = conf.create_conf(
+            sp_host='sp.example.com',
+            idp_hosts=['idp.example.com'],
+            metadata_file='remote_metadata_one_idp.xml',
+        )
+        self.do_login()
+
+        request = RequestFactory().get('/')
+        request.user = User.objects.last()
+
+        middleware = SamlSessionMiddleware()
+        middleware.process_request(request)
+
+        saml_session_name = getattr(settings, 'SAML_SESSION_COOKIE_NAME', 'saml_session')
+        getattr(request, saml_session_name)['_saml2_subject_id'] = '1f87035b4c1325b296a53d92097e6b3fa36d7e30ee82e3fcb0680d60243c1f03'
+        getattr(request, saml_session_name).save()
+        
+        response = EchoAttributesView.as_view()(request)
+        self.assertEqual(response.status_code, 200)
+        self.assertIn('<h1>SAML attributes</h1>', response.content.decode(), 'Echo page not rendered')
+
     def test_logout(self):
         settings.SAML_CONFIG = conf.create_conf(
             sp_host='sp.example.com',
@@ -428,8 +467,7 @@ def test_logout(self):
 
         saml_request = params['SAMLRequest'][0]
 
-        if 'LogoutRequest xmlns' not in decode_base64_and_inflate(saml_request).decode('utf-8'):
-            raise Exception('Not a valid LogoutRequest')
+        self.assertIn('LogoutRequest xmlns', decode_base64_and_inflate(saml_request).decode('utf-8'), 'Not a valid LogoutRequest')
 
     def test_logout_service_local(self):
         settings.SAML_CONFIG = conf.create_conf(
@@ -453,8 +491,8 @@ def test_logout_service_local(self):
         self.assertIn('SAMLRequest', params)
 
         saml_request = params['SAMLRequest'][0]
-        if 'LogoutRequest xmlns' not in decode_base64_and_inflate(saml_request).decode('utf-8'):
-            raise Exception('Not a valid LogoutRequest')
+
+        self.assertIn('LogoutRequest xmlns', decode_base64_and_inflate(saml_request).decode('utf-8'), 'Not a valid LogoutRequest')
 
         # now simulate a logout response sent by the idp
         expected_request = """<samlp:LogoutRequest xmlns:saml="urn:oasis:names:tc:SAML:2.0:assertion" xmlns:samlp="urn:oasis:names:tc:SAML:2.0:protocol" ID="XXXXXXXXXXXXXXXXXXXXXX" Version="2.0" Destination="https://idp.example.com/simplesaml/saml2/idp/SingleLogoutService.php" Reason=""><saml:Issuer Format="urn:oasis:names:tc:SAML:2.0:nameid-format:entity">http://sp.example.com/saml2/metadata/</saml:Issuer><saml:NameID SPNameQualifier="http://sp.example.com/saml2/metadata/" Format="urn:oasis:names:tc:SAML:2.0:nameid-format:transient">1f87035b4c1325b296a53d92097e6b3fa36d7e30ee82e3fcb0680d60243c1f03</saml:NameID><samlp:SessionIndex>a0123456789abcdef0123456789abcdef</samlp:SessionIndex></samlp:LogoutRequest>"""
@@ -501,8 +539,7 @@ def test_logout_service_global(self):
         self.assertIn('SAMLResponse', params)
         saml_response = params['SAMLResponse'][0]
 
-        if 'Response xmlns' not in decode_base64_and_inflate(saml_response).decode('utf-8'):
-            raise Exception('Not a valid Response')
+        self.assertIn('Response xmlns', decode_base64_and_inflate(saml_response).decode('utf-8'), 'Not a valid Response')
 
     def test_incomplete_logout(self):
         settings.SAML_CONFIG = conf.create_conf(sp_host='sp.example.com',
@@ -620,11 +657,8 @@ def test_custom_conf_loader_from_view(self):
 
 class SessionEnabledTestCase(TestCase):
     def get_session(self):
-        if self.client.session:
-            session = self.client.session
-        else:
-            engine = import_module(settings.SESSION_ENGINE)
-            session = engine.SessionStore()
+        engine = import_module(settings.SESSION_ENGINE)
+        session = self.client.session or engine.SessionStore()
         return session
 
     def set_session_cookies(self, session):

djangosaml2/tests/__init__.py

@@ -76,19 +76,6 @@ def get_location(http_info):
         return http_info['url']
 
 
-def fail_acs_response(request, *args, **kwargs):
-    """ Serves as a common mechanism for ending ACS in case of any SAML related failure.
-    Handling can be configured by setting the SAML_ACS_FAILURE_RESPONSE_FUNCTION as
-    suitable for the project.
-
-    The default behavior uses SAML specific template that is rendered on any ACS error,
-    but this can be simply changed so that PermissionDenied exception is raised instead.
-    """
-    failure_function = import_string(get_custom_setting('SAML_ACS_FAILURE_RESPONSE_FUNCTION',
-                                                        'djangosaml2.acs_failures.template_failure'))
-    return failure_function(request, *args, **kwargs)
-
-
 def validate_referral_url(request, url):
     # Ensure the user-originating redirection url is safe.
     # By setting SAML_ALLOWED_HOSTS in settings.py the user may provide a list of "allowed"

djangosaml2/utils.py

@@ -48,7 +48,7 @@
 from .conf import get_config
 from .exceptions import IdPConfigurationMissing
 from .overrides import Saml2Client
-from .utils import (available_idps, fail_acs_response, get_custom_setting,
+from .utils import (available_idps, get_custom_setting,
                     get_idp_sso_supported_bindings, get_location,
                     validate_referral_url)
 
@@ -269,6 +269,18 @@ class AssertionConsumerServiceView(SPConfigMixin, View):
         though some implementations may instead register their own subclasses of Saml2Backend.
     """
 
+    def handle_acs_failure(self, request, exception=None, status=403, **kwargs):
+        """ Error handler if the login attempt fails. Override this to customize the error response.
+        """
+
+        # Backwards compatibility: if a custom setting was defined, use that one
+        custom_failure_function = get_custom_setting('SAML_ACS_FAILURE_RESPONSE_FUNCTION')
+        if custom_failure_function:
+            failure_function = custom_failure_function if callable(custom_failure_function) else import_string(custom_failure_function)
+            return failure_function(request, exception, status, **kwargs)
+
+        return render(request, 'djangosaml2/login_error.html', {'exception': exception}, status=status)
+
     def post(self, request, attribute_mapping=None, create_unknown_user=None):
         """ SAML Authorization Response endpoint
         """
@@ -317,10 +329,10 @@ def post(self, request, attribute_mapping=None, create_unknown_user=None):
             logger.exception("Received SAMLResponse when no request has been made.")
 
         if _exception:
-            return fail_acs_response(request, exception=_exception)
+            return self.handle_acs_failure(request, exception=_exception)
         elif response is None:
             logger.warning("Invalid SAML Assertion received (unknown error).")
-            return fail_acs_response(request, status=400, exception=SuspiciousOperation('Unknown SAML2 error'))
+            return self.handle_acs_failure(request, status=400, exception=SuspiciousOperation('Unknown SAML2 error'))
 
         session_id = response.session_id()
         oq_cache.delete(session_id)
@@ -340,7 +352,7 @@ def post(self, request, attribute_mapping=None, create_unknown_user=None):
                                  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.'))
+            return self.handle_acs_failure(request, exception=PermissionDenied('No user could be authenticated.'))
 
         auth.login(self.request, user)
         _set_subject_id(request.saml_session, session_info['name_id'])