IdP metadata parser/merger adjustments

Author: pitbulk

src/onelogin/saml2/idp_metadata_parser.py

@@ -10,6 +10,8 @@
 """
 
 import urllib2
+
+from copy import deepcopy
 from defusedxml.lxml import fromstring
 
 from onelogin.saml2.constants import OneLogin_Saml2_Constants
@@ -51,7 +53,7 @@ def get_metadata(url):
         return xml
 
     @staticmethod
-    def parse_remote(url):
+    def parse_remote(url, **kwargs):
         """
         Get the metadata XML from the provided URL and parse it, returning a dict with extracted data
 
@@ -62,22 +64,41 @@ def parse_remote(url):
         :rtype: dict
         """
         idp_metadata = OneLogin_Saml2_IdPMetadataParser.get_metadata(url)
-        return OneLogin_Saml2_IdPMetadataParser.parse(idp_metadata)
+        return OneLogin_Saml2_IdPMetadataParser.parse(idp_metadata, **kwargs)
 
     @staticmethod
-    def parse(idp_metadata):
+    def parse(
+            idp_metadata,
+            required_sso_binding=OneLogin_Saml2_Constants.BINDING_HTTP_REDIRECT,
+            required_slo_binding=OneLogin_Saml2_Constants.BINDING_HTTP_REDIRECT):
         """
-        Parse the Identity Provider metadata and returns a dict with extracted data
-        If there are multiple IDPSSODescriptor it will only parse the first
+        Parse the Identity Provider metadata and return a dict with extracted data.
+
+        If there are multiple <IDPSSODescriptor> tags, parse only the first.
+
+        Parse only those SSO endpoints with the same binding as given by
+        the `required_sso_binding` parameter.
+
+        Parse only those SLO endpoints with the same binding as given by
+        the `required_slo_binding` parameter.
+
+        If the metadata specifies multiple SSO endpoints with the required
+        binding, extract only the first (the same holds true for SLO
+        endpoints).
 
         :param idp_metadata: XML of the Identity Provider Metadata.
         :type idp_metadata: string
 
-        :param url: If true and the URL is HTTPs, the cert of the domain is checked.
-        :type url: bool
+        :param required_sso_binding: Parse only POST or REDIRECT SSO endpoints.
+        :type required_sso_binding: one of OneLogin_Saml2_Constants.BINDING_HTTP_REDIRECT
+            or OneLogin_Saml2_Constants.BINDING_HTTP_POST
+
+        :param required_slo_binding: Parse only POST or REDIRECT SLO endpoints.
+        :type required_slo_binding: one of OneLogin_Saml2_Constants.BINDING_HTTP_REDIRECT
+            or OneLogin_Saml2_Constants.BINDING_HTTP_POST
 
         :returns: settings dict with extracted data
-        :rtype: string
+        :rtype: dict
         """
         data = {}
 
@@ -100,11 +121,18 @@ def parse(idp_metadata):
                     if len(name_id_format_nodes) > 0:
                         idp_name_id_format = name_id_format_nodes[0].text
 
-                    sso_nodes = OneLogin_Saml2_Utils.query(idp_descriptor_node, "./md:SingleSignOnService[@Binding='%s']" % OneLogin_Saml2_Constants.BINDING_HTTP_REDIRECT)
+                    sso_nodes = OneLogin_Saml2_Utils.query(
+                        idp_descriptor_node,
+                        "./md:SingleSignOnService[@Binding='%s']" % required_sso_binding
+                    )
+
                     if len(sso_nodes) > 0:
                         idp_sso_url = sso_nodes[0].get('Location', None)
 
-                    slo_nodes = OneLogin_Saml2_Utils.query(idp_descriptor_node, "./md:SingleLogoutService[@Binding='%s']" % OneLogin_Saml2_Constants.BINDING_HTTP_REDIRECT)
+                    slo_nodes = OneLogin_Saml2_Utils.query(
+                        idp_descriptor_node,
+                        "./md:SingleLogoutService[@Binding='%s']" % required_slo_binding
+                    )
                     if len(slo_nodes) > 0:
                         idp_slo_url = slo_nodes[0].get('Location', None)
 
@@ -116,12 +144,17 @@ def parse(idp_metadata):
 
                     if idp_entity_id is not None:
                         data['idp']['entityId'] = idp_entity_id
+
                     if idp_sso_url is not None:
                         data['idp']['singleSignOnService'] = {}
                         data['idp']['singleSignOnService']['url'] = idp_sso_url
+                        data['idp']['singleSignOnService']['binding'] = required_sso_binding
+
                     if idp_slo_url is not None:
                         data['idp']['singleLogoutService'] = {}
                         data['idp']['singleLogoutService']['url'] = idp_slo_url
+                        data['idp']['singleLogoutService']['binding'] = required_slo_binding
+
                     if idp_x509_cert is not None:
                         data['idp']['x509cert'] = idp_x509_cert
 
@@ -150,6 +183,34 @@ def merge_settings(settings, new_metadata_settings):
         :returns: merged settings
         :rtype: dict
         """
-        result_settings = settings.copy()
-        result_settings.update(new_metadata_settings)
+        for d in (settings, new_metadata_settings):
+            if not isinstance(d, dict):
+                raise TypeError('Both arguments must be dictionaries.')
+
+        # Guarantee to not modify original data (`settings.copy()` would not
+        # be sufficient, as it's just a shallow copy).
+        result_settings = deepcopy(settings)
+        # Merge `new_metadata_settings` into `result_settings`.
+        dict_deep_merge(result_settings, new_metadata_settings)
         return result_settings
+
+
+def dict_deep_merge(a, b, path=None):
+    """Deep-merge dictionary `b` into dictionary `a`.
+    Kudos to http://stackoverflow.com/a/7205107/145400
+    """
+    if path is None:
+        path = []
+    for key in b:
+        if key in a:
+            if isinstance(a[key], dict) and isinstance(b[key], dict):
+                dict_deep_merge(a[key], b[key], path + [str(key)])
+            elif a[key] == b[key]:
+                # Key conflict, but equal value.
+                pass
+            else:
+                # Key/value conflict. Prioritize b over a.
+                a[key] = b[key]
+        else:
+            a[key] = b[key]
+    return a