Fix #239 .Check that the response has all of the AuthnContexts that we provided

Author: pitbulk

src/onelogin/saml2/auth.py

@@ -62,6 +62,7 @@ def __init__(self, request_data, old_settings=None, custom_base_path=None):
         self.__last_message_id = None
         self.__last_assertion_id = None
         self.__last_assertion_not_on_or_after = None
+        self.__last_authn_contexts = []
         self.__last_request = None
         self.__last_response = None
 
@@ -107,6 +108,7 @@ def process_response(self, request_id=None):
                 self.__session_expiration = response.get_session_not_on_or_after()
                 self.__last_message_id = response.get_id()
                 self.__last_assertion_id = response.get_assertion_id()
+                self.__last_authn_contexts = response.get_authn_contexts()
                 self.__last_assertion_not_on_or_after = response.get_assertion_not_on_or_after()
                 self.__authenticated = True
 

src/onelogin/saml2/errors.py

@@ -112,6 +112,7 @@ class OneLogin_Saml2_ValidationError(Exception):
     INVALID_SIGNATURE = 42
     WRONG_NUMBER_OF_SIGNATURES = 43
     RESPONSE_EXPIRED = 44
+    AUTHN_CONTEXT_MISMATCH = 45
 
     def __init__(self, message, code=0, errors=None):
         """

src/onelogin/saml2/response.py

@@ -178,6 +178,19 @@ def is_valid(self, request_data, request_id=None, raise_exceptions=False):
                         OneLogin_Saml2_ValidationError.WRONG_NUMBER_OF_AUTHSTATEMENTS
                     )
 
+                # Checks that the response has all of the AuthnContexts that we provided in the request.
+                # Only check if failOnAuthnContextMismatch is true and requestedAuthnContext is set to a list.
+                requested_authn_contexts = security.get('requestedAuthnContext', True)
+
+                if security.get('failOnAuthnContextMismatch', False) and requested_authn_contexts and requested_authn_contexts is not True:
+                    authn_contexts = self.get_authn_contexts()
+                    unmatched_contexts = set(requested_authn_contexts).difference(authn_contexts)
+                    if unmatched_contexts:
+                        raise OneLogin_Saml2_ValidationError(
+                            'The AuthnContext "%s" didn\'t include requested context "%s"' % (', '.join(authn_contexts), ', '.join(unmatched_contexts)),
+                            OneLogin_Saml2_ValidationError.AUTHN_CONTEXT_MISMATCH
+                        )
+
                 # Checks that there is at least one AttributeStatement if required
                 attribute_statement_nodes = self.__query_assertion('/saml:AttributeStatement')
                 if security.get('wantAttributeStatement', True) and not attribute_statement_nodes:
@@ -383,6 +396,15 @@ def get_audiences(self):
         audience_nodes = self.__query_assertion('/saml:Conditions/saml:AudienceRestriction/saml:Audience')
         return [OneLogin_Saml2_Utils.element_text(node) for node in audience_nodes if OneLogin_Saml2_Utils.element_text(node) is not None]
 
+    def get_authn_contexts(self):
+        """
+        Gets the authentication contexts
+         :returns: The authentication classes for the SAML Response
+        :rtype: list
+        """
+        authn_context_nodes = self.__query_assertion('/saml:AuthnStatement/saml:AuthnContext/saml:AuthnContextClassRef')
+        return [OneLogin_Saml2_Utils.element_text(node) for node in authn_context_nodes]
+
     def get_issuers(self):
         """
         Gets the issuers (from message and from assertion)

src/onelogin/saml2/settings.py

@@ -304,6 +304,7 @@ def __add_default_values(self):
         self.__sp.setdefault('privateKey', '')
 
         self.__security.setdefault('requestedAuthnContext', True)
+        self.__security.setdefault('failOnAuthnContextMismatch', False)
 
     def check_settings(self, settings):
         """

tests/src/OneLogin/saml2_tests/response_test.py

@@ -980,6 +980,44 @@ def testIsInValidAudience(self):
         self.assertFalse(response_2.is_valid(request_data))
         self.assertIn('is not a valid audience for this Response', response_2.get_error())
 
+    def testIsInValidAuthenticationContext(self):
+        """
+        Tests that requestedAuthnContext, when set, is compared against the
+        response AuthnContext, which is what you use for two-factor
+        authentication. Without this check you can get back a valid response
+        that didn't complete the two-factor step.
+        """
+        request_data = self.get_request_data()
+        message = self.file_contents(join(self.data_path, 'responses', 'valid_response.xml.base64'))
+        two_factor_context = 'urn:oasis:names:tc:SAML:2.0:ac:classes:TimeSyncToken'
+        password_context = 'urn:oasis:names:tc:SAML:2.0:ac:classes:Password'
+        settings_dict = self.loadSettingsJSON()
+        settings_dict['security']['requestedAuthnContext'] = [two_factor_context]
+        settings_dict['security']['failOnAuthnContextMismatch'] = True
+        settings_dict['strict'] = True
+        settings = OneLogin_Saml2_Settings(settings_dict)
+
+        # check that we catch when the contexts don't match
+        response = OneLogin_Saml2_Response(settings, message)
+        self.assertFalse(response.is_valid(request_data))
+        self.assertIn('The AuthnContext "%s" didn\'t include requested context "%s"' % (password_context, two_factor_context), response.get_error())
+
+        # now drop in the expected AuthnContextClassRef and see that it passes
+        original_message = b64decode(message)
+        two_factor_message = original_message.replace(password_context, two_factor_context)
+        two_factor_message = b64encode(two_factor_message)
+        response = OneLogin_Saml2_Response(settings, two_factor_message)
+        response.is_valid(request_data)
+        # check that we got as far as destination validation, which comes later
+        self.assertIn('The response was received at', response.get_error())
+
+        # with the default setting, check that we succeed with our original context
+        settings_dict['security']['requestedAuthnContext'] = True
+        settings = OneLogin_Saml2_Settings(settings_dict)
+        response = OneLogin_Saml2_Response(settings, message)
+        response.is_valid(request_data)
+        self.assertIn('The response was received at', response.get_error())
+
     def testIsInValidIssuer(self):
         """
         Tests the is_valid method of the OneLogin_Saml2_Response class