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

Author: pitbulk

src/onelogin/saml2/idp_metadata_parser.py

@@ -50,7 +50,7 @@ def get_metadata(url, validate_cert=True):
 
         if xml:
             try:
-                dom = fromstring(xml)
+                dom = fromstring(xml, forbid_dtd=True)
                 idp_descriptor_nodes = OneLogin_Saml2_Utils.query(dom, '//md:IDPSSODescriptor')
                 if idp_descriptor_nodes:
                     valid = True
@@ -124,7 +124,7 @@ def parse(
         """
         data = {}
 
-        dom = fromstring(idp_metadata)
+        dom = fromstring(idp_metadata, forbid_dtd=True)
 
         entity_desc_path = '//md:EntityDescriptor'
         if entity_id:

src/onelogin/saml2/logout_request.py

@@ -171,7 +171,7 @@ def get_id(request):
         else:
             if isinstance(request, Document):
                 request = request.toxml()
-            elem = fromstring(request)
+            elem = fromstring(request, forbid_dtd=True)
         return elem.get('ID', None)
 
     @staticmethod
@@ -190,7 +190,7 @@ def get_nameid_data(request, key=None):
         else:
             if isinstance(request, Document):
                 request = request.toxml()
-            elem = fromstring(request)
+            elem = fromstring(request, forbid_dtd=True)
 
         name_id = None
         encrypted_entries = OneLogin_Saml2_Utils.query(elem, '/samlp:LogoutRequest/saml:EncryptedID')
@@ -271,7 +271,7 @@ def get_issuer(request):
         else:
             if isinstance(request, Document):
                 request = request.toxml()
-            elem = fromstring(request)
+            elem = fromstring(request, forbid_dtd=True)
 
         issuer = None
         issuer_nodes = OneLogin_Saml2_Utils.query(elem, '/samlp:LogoutRequest/saml:Issuer')
@@ -293,7 +293,7 @@ def get_session_indexes(request):
         else:
             if isinstance(request, Document):
                 request = request.toxml()
-            elem = fromstring(request)
+            elem = fromstring(request, forbid_dtd=True)
 
         session_indexes = []
         session_index_nodes = OneLogin_Saml2_Utils.query(elem, '/samlp:LogoutRequest/samlp:SessionIndex')
@@ -314,7 +314,7 @@ def is_valid(self, request_data, raise_exceptions=False):
         self.__error = None
         lowercase_urlencoding = False
         try:
-            dom = fromstring(self.__logout_request)
+            dom = fromstring(self.__logout_request, forbid_dtd=True)
 
             idp_data = self.__settings.get_idp_data()
             idp_entity_id = idp_data['entityId']

src/onelogin/saml2/logout_response.py

@@ -44,7 +44,7 @@ def __init__(self, settings, response=None):
 
         if response is not None:
             self.__logout_response = OneLogin_Saml2_Utils.decode_base64_and_inflate(response)
-            self.document = parseString(self.__logout_response)
+            self.document = parseString(self.__logout_response, forbid_dtd=True)
             self.id = self.document.documentElement.getAttribute('ID')
 
     def get_issuer(self):
@@ -200,7 +200,7 @@ def __query(self, query):
         """
         # Switch to lxml for querying
         xml = self.document.toxml()
-        return OneLogin_Saml2_Utils.query(fromstring(xml), query)
+        return OneLogin_Saml2_Utils.query(fromstring(xml, forbid_dtd=True), query)
 
     def build(self, in_response_to):
         """

src/onelogin/saml2/metadata.py

@@ -247,7 +247,7 @@ def add_x509_key_descriptors(metadata, cert=None, add_encryption=True):
         if cert is None or cert == '':
             return metadata
         try:
-            xml = parseString(metadata.encode('utf-8'))
+            xml = parseString(metadata.encode('utf-8'), forbid_dtd=True)
         except Exception as e:
             raise Exception('Error parsing metadata. ' + e.message)
 

src/onelogin/saml2/response.py

@@ -40,7 +40,7 @@ def __init__(self, settings, response):
         self.__settings = settings
         self.__error = None
         self.response = b64decode(response)
-        self.document = fromstring(self.response)
+        self.document = fromstring(self.response, forbid_dtd=True)
         self.decrypted_document = None
         self.encrypted = None
         self.valid_scd_not_on_or_after = None
@@ -735,38 +735,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_Utils.query(document, query)
+        return OneLogin_Saml2_Utils.query(document, query, None, tagid)
 
     def __decrypt_assertion(self, dom):
         """
@@ -817,7 +823,7 @@ def __decrypt_assertion(self, dom):
                         if not uri.startswith('#'):
                             break
                         uri = uri.split('#')[1]
-                        encrypted_key = OneLogin_Saml2_Utils.query(encrypted_assertion_nodes[0], './xenc:EncryptedKey[@Id="' + uri + '"]')
+                        encrypted_key = OneLogin_Saml2_Utils.query(encrypted_assertion_nodes[0], './xenc:EncryptedKey[@Id=$tagid]', None, uri)
                         if encrypted_key:
                             keyinfo.append(encrypted_key[0])
 

src/onelogin/saml2/utils.py

@@ -141,7 +141,7 @@ def validate_xml(xml, schema, debug=False):
 
         # Switch to lxml for schema validation
         try:
-            dom = fromstring(xml.encode('utf-8'))
+            dom = fromstring(xml.encode('utf-8'), forbid_dtd=True)
         except Exception:
             return 'unloaded_xml'
 
@@ -160,7 +160,7 @@ def validate_xml(xml, schema, debug=False):
 
             return 'invalid_xml'
 
-        return parseString(tostring(dom, encoding='unicode').encode('utf-8'))
+        return parseString(tostring(dom, encoding='unicode').encode('utf-8'), forbid_dtd=True)
 
     @staticmethod
     def element_text(node):
@@ -530,7 +530,7 @@ def get_expire_time(cache_duration=None, valid_until=None):
         return None
 
     @staticmethod
-    def query(dom, query, context=None):
+    def query(dom, query, context=None, tagid=None):
         """
         Extracts nodes that match the query from the Element
 
@@ -543,13 +543,21 @@ def query(dom, query, context=None):
         :param context: Context Node
         :type: DOMElement
 
+        :param tagid: Tag ID
+        :type: 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 delete_local_session(callback=None):
@@ -668,7 +676,7 @@ def generate_name_id(value, sp_nq, sp_format=None, cert=None, debug=False, nq=No
 
         if cert is not None:
             xml = name_id_container.toxml()
-            elem = fromstring(xml)
+            elem = fromstring(xml, forbid_dtd=True)
 
             error_callback_method = None
             if debug:
@@ -697,7 +705,7 @@ def generate_name_id(value, sp_nq, sp_format=None, cert=None, debug=False, nq=No
 
             edata = enc_ctx.encryptXml(enc_data, elem[0])
 
-            newdoc = parseString(tostring(edata, encoding='unicode').encode('utf-8'))
+            newdoc = parseString(tostring(edata, encoding='unicode').encode('utf-8'), forbid_dtd=True)
 
             if newdoc.hasChildNodes():
                 child = newdoc.firstChild
@@ -783,9 +791,9 @@ def decrypt_element(encrypted_data, key, debug=False, inplace=False):
         :rtype: lxml.etree.Element
         """
         if isinstance(encrypted_data, Element):
-            encrypted_data = fromstring(str(encrypted_data.toxml()))
+            encrypted_data = fromstring(str(encrypted_data.toxml()), forbid_dtd=True)
         elif isinstance(encrypted_data, basestring):
-            encrypted_data = fromstring(str(encrypted_data))
+            encrypted_data = fromstring(str(encrypted_data), forbid_dtd=True)
         elif not inplace and isinstance(encrypted_data, etree._Element):
             encrypted_data = deepcopy(encrypted_data)
 
@@ -851,7 +859,7 @@ def add_sign(xml, key, cert, debug=False, sign_algorithm=OneLogin_Saml2_Constant
             elem = xml
         elif isinstance(xml, Document):
             xml = xml.toxml()
-            elem = fromstring(xml.encode('utf-8'))
+            elem = fromstring(xml.encode('utf-8'), forbid_dtd=True)
         elif isinstance(xml, Element):
             xml.setAttributeNS(
                 unicode(OneLogin_Saml2_Constants.NS_SAMLP),
@@ -864,9 +872,9 @@ def add_sign(xml, key, cert, debug=False, sign_algorithm=OneLogin_Saml2_Constant
                 unicode(OneLogin_Saml2_Constants.NS_SAML)
             )
             xml = xml.toxml()
-            elem = fromstring(xml.encode('utf-8'))
+            elem = fromstring(xml.encode('utf-8'), forbid_dtd=True)
         elif isinstance(xml, basestring):
-            elem = fromstring(xml.encode('utf-8'))
+            elem = fromstring(xml.encode('utf-8'), forbid_dtd=True)
         else:
             raise Exception('Error parsing xml string')
 
@@ -939,8 +947,6 @@ def add_sign(xml, key, cert, debug=False, sign_algorithm=OneLogin_Saml2_Constant
         dsig_ctx.sign(signature)
 
         return tostring(elem, encoding='unicode').encode('utf-8')
-        newdoc = parseString(tostring(elem, encoding='unicode').encode('utf-8'))
-        return newdoc.saveXML(newdoc.firstChild)
 
     @staticmethod
     @return_false_on_exception
@@ -981,7 +987,7 @@ def validate_sign(xml, cert=None, fingerprint=None, fingerprintalg='sha1', valid
             elem = xml
         elif isinstance(xml, Document):
             xml = xml.toxml()
-            elem = fromstring(str(xml))
+            elem = fromstring(str(xml), forbid_dtd=True)
         elif isinstance(xml, Element):
             xml.setAttributeNS(
                 unicode(OneLogin_Saml2_Constants.NS_SAMLP),
@@ -994,9 +1000,9 @@ def validate_sign(xml, cert=None, fingerprint=None, fingerprintalg='sha1', valid
                 unicode(OneLogin_Saml2_Constants.NS_SAML)
             )
             xml = xml.toxml()
-            elem = fromstring(str(xml))
+            elem = fromstring(str(xml), forbid_dtd=True)
         elif isinstance(xml, basestring):
-            elem = fromstring(str(xml))
+            elem = fromstring(str(xml), forbid_dtd=True)
         else:
             raise Exception('Error parsing xml string')
 
@@ -1065,17 +1071,17 @@ def validate_metadata_sign(xml, cert=None, fingerprint=None, fingerprintalg='sha
             elem = xml
         elif isinstance(xml, Document):
             xml = xml.toxml()
-            elem = fromstring(str(xml))
+            elem = fromstring(str(xml), forbid_dtd=True)
         elif isinstance(xml, Element):
             xml.setAttributeNS(
                 unicode(OneLogin_Saml2_Constants.NS_MD),
                 'xmlns:md',
                 unicode(OneLogin_Saml2_Constants.NS_MD)
             )
             xml = xml.toxml()
-            elem = fromstring(str(xml))
+            elem = fromstring(str(xml), forbid_dtd=True)
         elif isinstance(xml, basestring):
-            elem = fromstring(str(xml))
+            elem = fromstring(str(xml), forbid_dtd=True)
         else:
             raise Exception('Error parsing xml string')