Be able to get at the auth object the last processed ID (response/assertion) and the last generated ID. Reset errorReason attribute of the auth object after each Process method

Author: pitbulk

README.md

@@ -889,6 +889,13 @@ The 'x509certMulti' is an array with 2 keys:
 - 'encryption' An array with one unique cert that will be used to encrypt data to be sent to the IdP
 
 
+### Replay attacks ###
+ 
+ In order to avoid reply attacks, you can store the ID of the SAML messages already processed, to avoid processing them twice. Since the Messages expires and will be invalidated due that fact, you don't need to store those IDs longer than the time frame that you currently accepting.
+ 
+ Get the ID of the last processed message/assertion with the get_last_message_id/get_last_assertion_id method of the Auth object.
+
+
 ### Main classes and methods ###
 
 Described below are the main classes and methods that can be invoked from the SAML2 library.
@@ -920,6 +927,9 @@ Main class of OneLogin Python Toolkit
 * ***set_strict*** Set the strict mode active/disable.
 * ***get_last_request_xml*** Returns the most recently-constructed/processed XML SAML request (AuthNRequest, LogoutRequest)
 * ***get_last_response_xml*** Returns the most recently-constructed/processed XML SAML response (SAMLResponse, LogoutResponse). If the SAMLResponse had an encrypted assertion, decrypts it.
+* ***get_last_message_id*** The ID of the last Response SAML message processed.
+* ***get_last_assertion_id*** The ID of the last assertion processed.
+* ***get_last_assertion_not_on_or_after*** The NotOnOrAfter value of the valid SubjectConfirmationData node (if any) of the last assertion processed (is only calculated with strict = true)
 
 #### OneLogin_Saml2_Auth - authn_request.py ####
 
@@ -948,6 +958,9 @@ SAML 2 Authentication Response class
 * ***validate_timestamps*** Verifies that the document is valid according to Conditions Element
 * ***get_error*** After execute a validation process, if fails this method returns the cause
 * ***get_xml_document*** Returns the SAML Response document (If contains an encrypted assertion, decrypts it).
+* ***get_id*** the ID of the response
+* ***get_assertion_id*** the ID of the assertion in the response
+* ***get_assertion_not_on_or_after*** the NotOnOrAfter value of the valid SubjectConfirmationData if any
 
 #### OneLogin_Saml2_LogoutRequest - logout_request.py ####
 

src/onelogin/saml2/auth.py

@@ -59,6 +59,9 @@ def __init__(self, request_data, old_settings=None, custom_base_path=None):
         self.__errors = []
         self.__error_reason = None
         self.__last_request_id = None
+        self.__last_message_id = None
+        self.__last_assertion_id = None
+        self.__last_assertion_not_on_or_after = None
         self.__last_request = None
         self.__last_response = None
 
@@ -90,6 +93,7 @@ def process_response(self, request_id=None):
         :raises: OneLogin_Saml2_Error.SAML_RESPONSE_NOT_FOUND, when a POST with a SAMLResponse is not found
         """
         self.__errors = []
+        self.__error_reason = None
 
         if 'post_data' in self.__request_data and 'SAMLResponse' in self.__request_data['post_data']:
             # AuthnResponse -- HTTP_POST Binding
@@ -101,6 +105,9 @@ def process_response(self, request_id=None):
                 self.__nameid_format = response.get_nameid_format()
                 self.__session_index = response.get_session_index()
                 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_assertion_not_on_or_after = response.get_assertion_not_on_or_after()
                 self.__authenticated = True
 
             else:
@@ -127,6 +134,7 @@ def process_slo(self, keep_local_session=False, request_id=None, delete_session_
         :returns: Redirection URL
         """
         self.__errors = []
+        self.__error_reason = None
 
         if 'get_data' in self.__request_data and 'SAMLResponse' in self.__request_data['get_data']:
             logout_response = OneLogin_Saml2_Logout_Response(self.__settings, self.__request_data['get_data']['SAMLResponse'])
@@ -136,8 +144,10 @@ def process_slo(self, keep_local_session=False, request_id=None, delete_session_
                 self.__error_reason = logout_response.get_error()
             elif logout_response.get_status() != OneLogin_Saml2_Constants.STATUS_SUCCESS:
                 self.__errors.append('logout_not_success')
-            elif not keep_local_session:
-                OneLogin_Saml2_Utils.delete_local_session(delete_session_cb)
+            else:
+                self.__last_message_id = logout_response.id
+                if not keep_local_session:
+                    OneLogin_Saml2_Utils.delete_local_session(delete_session_cb)
 
         elif 'get_data' in self.__request_data and 'SAMLRequest' in self.__request_data['get_data']:
             logout_request = OneLogin_Saml2_Logout_Request(self.__settings, self.__request_data['get_data']['SAMLRequest'])
@@ -150,6 +160,7 @@ def process_slo(self, keep_local_session=False, request_id=None, delete_session_
                     OneLogin_Saml2_Utils.delete_local_session(delete_session_cb)
 
                 in_response_to = logout_request.id
+                self.__last_message_id = logout_request.id
                 response_builder = OneLogin_Saml2_Logout_Response(self.__settings)
                 response_builder.build(in_response_to)
                 self.__last_response = response_builder.get_xml()
@@ -241,6 +252,13 @@ def get_session_expiration(self):
         """
         return self.__session_expiration
 
+    def get_last_assertion_not_on_or_after(self):
+        """
+        The NotOnOrAfter value of the valid SubjectConfirmationData node
+        (if any) of the last assertion processed
+        """
+        return self.__last_assertion_not_on_or_after
+
     def get_errors(self):
         """
         Returns a list with code errors if something went wrong
@@ -282,6 +300,20 @@ def get_last_request_id(self):
         """
         return self.__last_request_id
 
+    def get_last_message_id(self):
+        """
+        :returns: The ID of the last Response SAML message processed.
+        :rtype: string
+        """
+        return self.__last_message_id
+
+    def get_last_assertion_id(self):
+        """
+        :returns: The ID of the last assertion processed.
+        :rtype: string
+        """
+        return self.__last_assertion_id
+
     def login(self, return_to=None, force_authn=False, is_passive=False, set_nameid_policy=True):
         """
         Initiates the SSO process.

src/onelogin/saml2/logout_response.py

@@ -40,10 +40,12 @@ def __init__(self, settings, response=None):
         """
         self.__settings = settings
         self.__error = None
+        self.id = None
 
         if response is not None:
             self.__logout_response = OneLogin_Saml2_Utils.decode_base64_and_inflate(response)
             self.document = parseString(self.__logout_response)
+            self.id = self.document.documentElement.getAttribute('ID')
 
     def get_issuer(self):
         """

src/onelogin/saml2/response.py

@@ -43,6 +43,7 @@ def __init__(self, settings, response):
         self.document = fromstring(self.response)
         self.decrypted_document = None
         self.encrypted = None
+        self.valid_scd_not_on_or_after = None
 
         # Quick check for the presence of EncryptedAssertion
         encrypted_assertion_nodes = self.__query('/samlp:Response/saml:EncryptedAssertion')
@@ -258,6 +259,10 @@ def is_valid(self, request_data, request_id=None, raise_exceptions=False):
                             parsed_nb = OneLogin_Saml2_Utils.parse_SAML_to_time(nb)
                             if parsed_nb > OneLogin_Saml2_Utils.now():
                                 continue
+
+                        if nooa:
+                            self.valid_scd_not_on_or_after = OneLogin_Saml2_Utils.parse_SAML_to_time(nooa)
+
                         any_subject_confirmation = True
                         break
 
@@ -487,6 +492,12 @@ def get_session_not_on_or_after(self):
             not_on_or_after = OneLogin_Saml2_Utils.parse_SAML_to_time(authn_statement_nodes[0].get('SessionNotOnOrAfter'))
         return not_on_or_after
 
+    def get_assertion_not_on_or_after(self):
+        """
+        Returns the NotOnOrAfter value of the valid SubjectConfirmationData node if any
+        """
+        return self.valid_scd_not_on_or_after
+
     def get_session_index(self):
         """
         Gets the SessionIndex from the AuthnStatement
@@ -820,3 +831,22 @@ def get_xml_document(self):
             return self.decrypted_document
         else:
             return self.document
+
+    def get_id(self):
+        """
+        :returns: the ID of the response
+        :rtype: string
+        """
+        return self.document.get('ID', None)
+
+    def get_assertion_id(self):
+        """
+        :returns: the ID of the assertion in the response
+        :rtype: string
+        """
+        if not self.validate_num_assertions():
+            raise OneLogin_Saml2_ValidationError(
+                'SAML Response must contain 1 assertion',
+                OneLogin_Saml2_ValidationError.WRONG_NUMBER_OF_ASSERTIONS
+            )
+        return self.__query_assertion('')[0].get('ID', None)

tests/data/responses/valid_response.xml.base64

@@ -118,7 +118,7 @@ def testGetSessionExpiration(self):
         self.assertIsNone(auth2.get_session_expiration())
 
         auth2.process_response()
-        self.assertEqual(1392802621, auth2.get_session_expiration())
+        self.assertEqual(2655106621, auth2.get_session_expiration())
 
     def testGetLastErrorReason(self):
         """
@@ -1002,6 +1002,21 @@ def testBuildResponseSignature(self):
         with self.assertRaisesRegexp(OneLogin_Saml2_Error, "Trying to sign the SAMLResponse but can't load the SP private key"):
             auth2.build_response_signature(message, relay_state)
 
+    def testGetLastRequestID(self):
+        settings_info = self.loadSettingsJSON()
+        request_data = self.get_request()
+        auth = OneLogin_Saml2_Auth(request_data, old_settings=settings_info)
+
+        auth.login()
+        id1 = auth.get_last_request_id()
+        self.assertNotEqual(id1, None)
+
+        auth.logout()
+        id2 = auth.get_last_request_id()
+        self.assertNotEqual(id2, None)
+
+        self.assertNotEqual(id1, id2)
+
     def testGetLastSAMLResponse(self):
         settings = self.loadSettingsJSON()
         message = self.file_contents(join(self.data_path, 'responses', 'signed_message_response.xml.base64'))
@@ -1084,6 +1099,71 @@ def testGetLastLogoutResponse(self):
         auth.process_slo()
         self.assertEqual(response, auth.get_last_response_xml())
 
+    def testGetInfoFromLastResponseReceived(self):
+        """
+        Tests the get_last_message_id, get_last_assertion_id and get_last_assertion_not_on_or_after
+        of the OneLogin_Saml2_Auth class
+        """
+        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_message_id(), 'pfx42be40bf-39c3-77f0-c6ae-8bf2e23a1a2e')
+        self.assertEqual(auth.get_last_assertion_id(), 'pfx57dfda60-b211-4cda-0f63-6d5deb69e5bb')
+        self.assertIsNone(auth.get_last_assertion_not_on_or_after())
+
+        # NotOnOrAfter is only calculated with strict = true
+        # If invalid, response id and assertion id are not obtained
+
+        settings['strict'] = True
+        auth = OneLogin_Saml2_Auth(request_data, old_settings=settings)
+        auth.process_response()
+        self.assertNotEqual(len(auth.get_errors()), 0)
+        self.assertIsNone(auth.get_last_message_id())
+        self.assertIsNone(auth.get_last_assertion_id())
+        self.assertIsNone(auth.get_last_assertion_not_on_or_after())
+
+        request_data['https'] = 'on'
+        request_data['http_host'] = 'pitbulk.no-ip.org'
+        request_data['script_name'] = '/newonelogin/demo1/index.php?acs'
+        auth = OneLogin_Saml2_Auth(request_data, old_settings=settings)
+        auth.process_response()
+        self.assertEqual(len(auth.get_errors()), 0)
+        self.assertEqual(auth.get_last_message_id(), 'pfx42be40bf-39c3-77f0-c6ae-8bf2e23a1a2e')
+        self.assertEqual(auth.get_last_assertion_id(), 'pfx57dfda60-b211-4cda-0f63-6d5deb69e5bb')
+        self.assertEqual(auth.get_last_assertion_not_on_or_after(), 2671081021)
+
+    def testGetIdFromLogoutRequest(self):
+        """
+        Tests the get_last_message_id of the OneLogin_Saml2_Auth class
+        Case Valid Logout request
+        """
+        settings = self.loadSettingsJSON()
+        request = self.file_contents(join(self.data_path, 'logout_requests', 'logout_request.xml'))
+        message = OneLogin_Saml2_Utils.deflate_and_base64_encode(request)
+        message_wrapper = {'get_data': {'SAMLRequest': message}}
+        auth = OneLogin_Saml2_Auth(message_wrapper, old_settings=settings)
+        auth.process_slo()
+        self.assertIn(auth.get_last_message_id(), 'ONELOGIN_21584ccdfaca36a145ae990442dcd96bfe60151e')
+
+    def testGetIdFromLogoutResponse(self):
+        """
+        Tests the get_last_message_id of the OneLogin_Saml2_Auth class
+        Case Valid Logout response
+        """
+        settings = self.loadSettingsJSON()
+        response = self.file_contents(join(self.data_path, 'logout_responses', 'logout_response.xml'))
+        message = OneLogin_Saml2_Utils.deflate_and_base64_encode(response)
+        message_wrapper = {'get_data': {'SAMLResponse': message}}
+        auth = OneLogin_Saml2_Auth(message_wrapper, old_settings=settings)
+        auth.process_slo()
+        self.assertIn(auth.get_last_message_id(), '_f9ee61bd9dbf63606faa9ae3b10548d5b3656fb859')
 
 if __name__ == '__main__':
     if is_running_under_teamcity():

tests/src/OneLogin/saml2_tests/auth_test.py

@@ -1408,6 +1408,55 @@ def testStatusCheckBeforeAssertionCheck(self):
         with self.assertRaisesRegexp(OneLogin_Saml2_ValidationError, 'The status code of the Response was not Success, was Responder'):
             response.is_valid(self.get_request_data(), raise_exceptions=True)
 
+    def testGetId(self):
+        """
+        Tests that we can retrieve the ID of the Response
+        """
+        settings = OneLogin_Saml2_Settings(self.loadSettingsJSON())
+        xml = self.file_contents(join(self.data_path, 'responses', 'signed_message_response.xml.base64'))
+        response = OneLogin_Saml2_Response(settings, xml)
+        self.assertEqual(response.get_id(), 'pfxc3d2b542-0f7e-8767-8e87-5b0dc6913375')
+
+    def testGetAssertionId(self):
+        """
+        Tests that we can retrieve the ID of the Assertion
+        """
+        settings = OneLogin_Saml2_Settings(self.loadSettingsJSON())
+        xml = self.file_contents(join(self.data_path, 'responses', 'signed_message_response.xml.base64'))
+        response = OneLogin_Saml2_Response(settings, xml)
+        self.assertEqual(response.get_assertion_id(), '_cccd6024116641fe48e0ae2c51220d02755f96c98d')
+
+    def testGetAssertionNotOnOrAfter(self):
+        """
+        Tests that we can retrieve the NotOnOrAfter value of
+        the valid SubjectConfirmationData
+        """
+        settings_data = self.loadSettingsJSON()
+        request_data = self.get_request_data()
+        settings = OneLogin_Saml2_Settings(settings_data)
+        message = self.file_contents(join(self.data_path, 'responses', 'valid_response.xml.base64'))
+        response = OneLogin_Saml2_Response(settings, message)
+        self.assertIsNone(response.get_assertion_not_on_or_after())
+
+        response.is_valid(request_data)
+        self.assertIsNone(response.get_error())
+        self.assertIsNone(response.get_assertion_not_on_or_after())
+
+        settings_data['strict'] = True
+        settings = OneLogin_Saml2_Settings(settings_data)
+        response = OneLogin_Saml2_Response(settings, message)
+
+        response.is_valid(request_data)
+        self.assertNotEqual(response.get_error(), None)
+        self.assertIsNone(response.get_assertion_not_on_or_after())
+
+        request_data['https'] = 'on'
+        request_data['http_host'] = 'pitbulk.no-ip.org'
+        request_data['script_name'] = '/newonelogin/demo1/index.php?acs'
+        response.is_valid(request_data)
+        self.assertIsNone(response.get_error())
+        self.assertEqual(response.get_assertion_not_on_or_after(), 2671081021)
+
 
 if __name__ == '__main__':
     if is_running_under_teamcity():