Security improvements. Use of tagid to prevent XPath injection. Disable DTD on _parse_etree (fromstring defusedxml) method.

Author: pitbulk

src/onelogin/saml2/response.py

@@ -737,38 +737,44 @@ def __query_assertion(self, xpath_expr):
         signature_expr = '/ds:Signature/ds:SignedInfo/ds:Reference'
         signed_assertion_query = '/samlp:Response' + assertion_expr + signature_expr
         assertion_reference_nodes = self.__query(signed_assertion_query)
+        tagid = None
 
         if not assertion_reference_nodes:
             # Check if the message is signed
             signed_message_query = '/samlp:Response' + signature_expr
             message_reference_nodes = self.__query(signed_message_query)
             if message_reference_nodes:
                 message_id = message_reference_nodes[0].get('URI')
-                final_query = "/samlp:Response[@ID='%s']/" % message_id[1:]
+                final_query = "/samlp:Response[@ID=$tagid]/"
+                tagid = message_id[1:]
             else:
                 final_query = "/samlp:Response"
             final_query += assertion_expr
         else:
             assertion_id = assertion_reference_nodes[0].get('URI')
-            final_query = '/samlp:Response' + assertion_expr + "[@ID='%s']" % assertion_id[1:]
+            final_query = '/samlp:Response' + assertion_expr + "[@ID=$tagid]"
+            tagid = assertion_id[1:]
         final_query += xpath_expr
-        return self.__query(final_query)
+        return self.__query(final_query, tagid)
 
-    def __query(self, query):
+    def __query(self, query, tagid=None):
         """
         Extracts nodes that match the query from the Response
 
         :param query: Xpath Expresion
         :type query: String
 
+        :param tagid: Tag ID
+        :type query: String
+
         :returns: The queried nodes
         :rtype: list
         """
         if self.encrypted:
             document = self.decrypted_document
         else:
             document = self.document
-        return OneLogin_Saml2_XML.query(document, query)
+        return OneLogin_Saml2_XML.query(document, query, None, tagid)
 
     def __decrypt_assertion(self, xml):
         """
@@ -817,7 +823,7 @@ def __decrypt_assertion(self, xml):
                         if not uri.startswith('#'):
                             break
                         uri = uri.split('#')[1]
-                        encrypted_key = OneLogin_Saml2_XML.query(encrypted_assertion_nodes[0], './xenc:EncryptedKey[@Id="' + uri + '"]')
+                        encrypted_key = OneLogin_Saml2_XML.query(encrypted_assertion_nodes[0], './xenc:EncryptedKey[@Id=$tagid]', None, uri)
                         if encrypted_key:
                             keyinfo.append(encrypted_key[0])
 

src/onelogin/saml2/utils.py

@@ -13,7 +13,6 @@
 from copy import deepcopy
 import calendar
 from datetime import datetime
-from defusedxml.lxml import fromstring
 from hashlib import sha1, sha256, sha384, sha512
 from isodate import parse_duration as duration_parser
 import re
@@ -684,7 +683,7 @@ def decrypt_element(encrypted_data, key, debug=False, inplace=False):
         """
 
         if isinstance(encrypted_data, Element):
-            encrypted_data = fromstring(str(encrypted_data.toxml()))
+            encrypted_data = OneLogin_Saml2_XML.to_etree(str(encrypted_data.toxml()))
         if not inplace and isinstance(encrypted_data, OneLogin_Saml2_XML._element_class):
             encrypted_data = deepcopy(encrypted_data)
         elif isinstance(encrypted_data, OneLogin_Saml2_XML._text_class):

src/onelogin/saml2/xml_utils.py

@@ -62,7 +62,7 @@ def to_etree(xml):
         if isinstance(xml, OneLogin_Saml2_XML._element_class):
             return xml
         if isinstance(xml, OneLogin_Saml2_XML._text_class):
-            return OneLogin_Saml2_XML._parse_etree(xml)
+            return OneLogin_Saml2_XML._parse_etree(xml, forbid_dtd=True)
 
         raise ValueError('unsupported type %r' % type(xml))
 
@@ -101,7 +101,7 @@ def validate_xml(xml, schema, debug=False):
         return xml
 
     @staticmethod
-    def query(dom, query, context=None):
+    def query(dom, query, context=None, tagid=None):
         """
         Extracts nodes that match the query from the Element
 
@@ -114,13 +114,21 @@ def query(dom, query, context=None):
         :param context: Context Node
         :type: DOMElement
 
+        :param tagid: Tag ID
+        :type query: String
+
         :returns: The queried nodes
         :rtype: list
         """
         if context is None:
-            return dom.xpath(query, namespaces=OneLogin_Saml2_Constants.NSMAP)
+            source = dom
+        else:
+            source = context
+
+        if tagid is None:
+            return source.xpath(query, namespaces=OneLogin_Saml2_Constants.NSMAP)
         else:
-            return context.xpath(query, namespaces=OneLogin_Saml2_Constants.NSMAP)
+            return source.xpath(query, tagid=tagid, namespaces=OneLogin_Saml2_Constants.NSMAP)
 
     @staticmethod
     def cleanup_namespaces(tree_or_element, top_nsmap=None, keep_ns_prefixes=None):