Check that the response has all of the AuthnContexts that we provided in the request.

Author: jeffFranklin

README.md

@@ -396,11 +396,13 @@ In addition to the required settings data (idp, sp), extra settings can be defin
 
         // Authentication context.
         // Set to false and no AuthContext will be sent in the AuthNRequest,
-        // Set true or don't present thi parameter and you will get an AuthContext 'exact' 'urn:oasis:names:tc:SAML:2.0:ac:classes:PasswordProtectedTransport'
+        // Set true or don't present this parameter and you will get an AuthContext 'exact' 'urn:oasis:names:tc:SAML:2.0:ac:classes:PasswordProtectedTransport'
         // Set an array with the possible auth context values: array ('urn:oasis:names:tc:SAML:2.0:ac:classes:Password', 'urn:oasis:names:tc:SAML:2.0:ac:classes:X509'),
         "requestedAuthnContext": true,
 	// Allows the authn comparison parameter to be set, defaults to 'exact' if the setting is not present.
         "requestedAuthnContextComparison": "exact",
+        // Set to true to check that the AuthnContext received matches the one requested.
+        "failOnAuthnContextMismatch": false,
 
         // In some environment you will need to set how long the published metadata of the Service Provider gonna be valid.
         // is possible to not set the 2 following parameters (or set to null) and default values will be set (2 days, 1 week)

src/onelogin/saml2/auth.py

@@ -63,6 +63,7 @@ def __init__(self, request_data, old_settings=None, custom_base_path=None):
         self.__last_request_id = None
         self.__last_message_id = None
         self.__last_assertion_id = None
+        self.__last_authn_contexts = []
         self.__last_request = None
         self.__last_response = None
         self.__last_assertion_not_on_or_after = None
@@ -110,6 +111,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.__authenticated = True
                 self.__last_assertion_not_on_or_after = response.get_assertion_not_on_or_after()
 
@@ -318,6 +320,13 @@ def get_last_assertion_id(self):
         """
         return self.__last_assertion_id
 
+    def get_last_authn_contexts(self):
+        """
+        :returns: The list of authentication contexts sent in the last SAML resposne.
+        :rtype: list
+        """
+        return self.__last_authn_contexts
+
     def login(self, return_to=None, force_authn=False, is_passive=False, set_nameid_policy=True):
         """
         Initiates the SSO process.

src/onelogin/saml2/errors.py

@@ -109,6 +109,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

@@ -161,6 +161,18 @@ 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['requestedAuthnContext']
+                if security['failOnAuthnContextMismatch'] 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:
@@ -361,6 +373,16 @@ def get_audiences(self):
         audience_nodes = self.__query_assertion('/saml:Conditions/saml:AudienceRestriction/saml:Audience')
         return [OneLogin_Saml2_XML.element_text(node) for node in audience_nodes if OneLogin_Saml2_XML.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_XML.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

@@ -310,6 +310,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/auth_test.py

@@ -1231,6 +1231,20 @@ def testGetLastAuthnRequest(self):
         )
         self.assertIn(expectedFragment, auth.get_last_request_xml())
 
+    def testGetLastAuthnContexts(self):
+        settings = self.loadSettingsJSON()
+        request_data = self.get_request()
+        message = self.file_contents(
+            join(self.data_path, 'responses', 'valid_response.xml.base64'))
+        del request_data['get_data']
+        request_data['post_data'] = {
+            'SAMLResponse': message
+        }
+        auth = OneLogin_Saml2_Auth(request_data, old_settings=settings)
+
+        auth.process_response()
+        self.assertEqual(auth.get_last_authn_contexts(), ['urn:oasis:names:tc:SAML:2.0:ac:classes:Password'])
+
     def testGetLastLogoutRequest(self):
         settings = self.loadSettingsJSON()
         auth = OneLogin_Saml2_Auth({'http_host': 'localhost', 'script_name': 'thing'}, old_settings=settings)

tests/src/OneLogin/saml2_tests/response_test.py

@@ -932,6 +932,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 = compat.to_string(OneLogin_Saml2_Utils.b64decode(message))
+        two_factor_message = original_message.replace(password_context, two_factor_context)
+        two_factor_message = OneLogin_Saml2_Utils.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