Improve Signature validation process

Author: pitbulk

src/onelogin/saml2/response.py

@@ -88,6 +88,9 @@ def is_valid(self, request_data, request_id=None):
 
             signed_elements = self.process_signed_elements()
 
+            has_signed_response = '{%s}Response' % OneLogin_Saml2_Constants.NS_SAMLP in signed_elements
+            has_signed_assertion = '{%s}Assertion' % OneLogin_Saml2_Constants.NS_SAML in signed_elements
+
             if self.__settings.is_strict():
                 no_valid_xml_msg = 'Invalid SAML Response. Not match the saml-schema-protocol-2.0.xsd'
                 res = OneLogin_Saml2_Utils.validate_xml(etree.tostring(self.document), 'saml-schema-protocol-2.0.xsd', self.__settings.is_debug_active())
@@ -200,32 +203,26 @@ def is_valid(self, request_data, request_id=None):
                 if not any_subject_confirmation:
                     raise Exception('A valid SubjectConfirmation was not found on this Response')
 
-                if security.get('wantAssertionsSigned', False) and ('{%s}Assertion' % OneLogin_Saml2_Constants.NS_SAML) not in signed_elements:
+                if security.get('wantAssertionsSigned', False) and not has_signed_assertion:
                     raise Exception('The Assertion of the Response is not signed and the SP require it')
 
-                if security.get('wantMessagesSigned', False) and ('{%s}Response' % OneLogin_Saml2_Constants.NS_SAMLP) not in signed_elements:
+                if security.get('wantMessagesSigned', False) and not has_signed_response:
                     raise Exception('The Message of the Response is not signed and the SP require it')
 
-            if len(signed_elements) > 0:
-                if len(signed_elements) > 2:
-                    raise Exception('Too many Signatures found. SAML Response rejected')
+            if not signed_elements or (not has_signed_response and not has_signed_assertion):
+                raise Exception('No Signature found. SAML Response rejected')
+            else:
                 cert = idp_data.get('x509cert', None)
                 fingerprint = idp_data.get('certFingerprint', None)
                 fingerprintalg = idp_data.get('certFingerprintAlgorithm', None)
 
                 # If find a Signature on the Response, validates it checking the original response
-                if '{%s}Response' % OneLogin_Saml2_Constants.NS_SAMLP in signed_elements:
-                    document_to_validate = self.document
-                # Otherwise validates the assertion (decrypted assertion if was encrypted)
-                else:
-                    if self.encrypted:
-                        document_to_validate = self.decrypted_document
-                    else:
-                        document_to_validate = self.document
-                if not OneLogin_Saml2_Utils.validate_sign(document_to_validate, cert, fingerprint, fingerprintalg):
+                if has_signed_response and not OneLogin_Saml2_Utils.validate_sign(self.document, cert, fingerprint, fingerprintalg, xpath=OneLogin_Saml2_Utils.RESPONSE_SIGNATURE_XPATH):
+                    raise Exception('Signature validation failed. SAML Response rejected')
+
+                document_check_assertion = self.decrypted_document if self.encrypted else self.document
+                if has_signed_assertion and not OneLogin_Saml2_Utils.validate_sign(document_check_assertion, cert, fingerprint, fingerprintalg, xpath=OneLogin_Saml2_Utils.ASSERTION_SIGNATURE_XPATH):
                     raise Exception('Signature validation failed. SAML Response rejected')
-            else:
-                raise Exception('No Signature found. SAML Response rejected')
 
             return True
         except Exception as err:
@@ -291,7 +288,7 @@ def get_issuers(self):
         """
         issuers = []
 
-        message_issuer_nodes = self.__query('/samlp:Response/saml:Issuer')
+        message_issuer_nodes = OneLogin_Saml2_Utils.query(self.document, '/samlp:Response/saml:Issuer')
         if len(message_issuer_nodes) == 1:
             issuers.append(message_issuer_nodes[0].text)
         else:
@@ -486,8 +483,41 @@ def process_signed_elements(self):
                     verified_seis.append(sei)
 
             signed_elements.append(signed_element)
+
+        if signed_elements:
+            if not self.validate_signed_elements(signed_elements):
+                raise Exception('Found an unexpected Signature Element. SAML Response rejected')
         return signed_elements
 
+    def validate_signed_elements(self, signed_elements):
+        """
+        Verifies that the document has the expected signed nodes.
+        """
+        if len(signed_elements) > 2:
+            return False
+
+        response_tag = '{%s}Response' % OneLogin_Saml2_Constants.NS_SAMLP
+        assertion_tag = '{%s}Assertion' % OneLogin_Saml2_Constants.NS_SAML
+
+        if (response_tag in signed_elements and signed_elements.count(response_tag) > 1) or \
+           (assertion_tag in signed_elements and signed_elements.count(assertion_tag) > 1) or \
+           (response_tag not in signed_elements and assertion_tag not in signed_elements):
+            return False
+
+        # Check that the signed elements found here, are the ones that will be verified
+        # by OneLogin_Saml2_Utils.validate_sign
+        if response_tag in signed_elements:
+            expected_signature_nodes = OneLogin_Saml2_Utils.query(self.document, OneLogin_Saml2_Utils.RESPONSE_SIGNATURE_XPATH)
+            if len(expected_signature_nodes) != 1:
+                raise Exception('Unexpected number of Response signatures found. SAML Response rejected.')
+
+        if assertion_tag in signed_elements:
+            expected_signature_nodes = self.__query(OneLogin_Saml2_Utils.ASSERTION_SIGNATURE_XPATH)
+            if len(expected_signature_nodes) != 1:
+                raise Exception('Unexpected number of Assertion signatures found. SAML Response rejected.')
+
+        return True
+
     def validate_timestamps(self):
         """
         Verifies that the document is valid according to Conditions Element

src/onelogin/saml2/utils.py

@@ -65,6 +65,9 @@ class OneLogin_Saml2_Utils(object):
 
     """
 
+    RESPONSE_SIGNATURE_XPATH = '/samlp:Response/ds:Signature'
+    ASSERTION_SIGNATURE_XPATH = '/samlp:Response/saml:Assertion/ds:Signature'
+
     @staticmethod
     def decode_base64_and_inflate(value):
         """
@@ -865,7 +868,7 @@ def add_sign(xml, key, cert, debug=False, sign_algorithm=OneLogin_Saml2_Constant
         return newdoc.saveXML(newdoc.firstChild)
 
     @staticmethod
-    def validate_sign(xml, cert=None, fingerprint=None, fingerprintalg='sha1', validatecert=False, debug=False):
+    def validate_sign(xml, cert=None, fingerprint=None, fingerprintalg='sha1', validatecert=False, debug=False, xpath=None):
         """
         Validates a signature (Message or Assertion).
 
@@ -886,6 +889,9 @@ def validate_sign(xml, cert=None, fingerprint=None, fingerprintalg='sha1', valid
 
         :param debug: Activate the xmlsec debug
         :type: bool
+
+        :param xpath: The xpath of the signed element
+        :type: string
         """
         try:
             if xml is None or xml == '':
@@ -918,10 +924,13 @@ def validate_sign(xml, cert=None, fingerprint=None, fingerprintalg='sha1', valid
 
             xmlsec.addIDs(elem, ["ID"])
 
-            signature_nodes = OneLogin_Saml2_Utils.query(elem, '/samlp:Response/ds:Signature')
+            if xpath:
+                signature_nodes = OneLogin_Saml2_Utils.query(elem, xpath)
+            else:
+                signature_nodes = OneLogin_Saml2_Utils.query(elem, OneLogin_Saml2_Utils.RESPONSE_SIGNATURE_XPATH)
 
-            if not len(signature_nodes) > 0:
-                signature_nodes += OneLogin_Saml2_Utils.query(elem, '/samlp:Response/saml:Assertion/ds:Signature')
+                if len(signature_nodes) == 0:
+                    signature_nodes = OneLogin_Saml2_Utils.query(elem, OneLogin_Saml2_Utils.ASSERTION_SIGNATURE_XPATH)
 
             if len(signature_nodes) == 1:
                 signature_node = signature_nodes[0]

tests/src/OneLogin/saml2_tests/signed_response_test.py

@@ -50,7 +50,11 @@ def testResponseAndAssertionSigned(self):
         Tests the getNameId method of the OneLogin_Saml2_Response
         Case valid signed response, signed assertion
         """
-        settings = OneLogin_Saml2_Settings(self.loadSettingsJSON())
+        settings_info = self.loadSettingsJSON()
+        settings_info['idp']['entityId'] = "https://federate.example.net/saml/saml2/idp/metadata.php"
+        settings_info['sp']['entityId'] = "hello.com"
+
+        settings = OneLogin_Saml2_Settings(settings_info)
         message = self.file_contents(join(self.data_path, 'responses', 'simple_saml_php.xml'))
         response = OneLogin_Saml2_Response(settings, b64encode(message))