Return the ResponseLocation before falling back to Location

ResponseLocation [Optional]

    Optionally specifies a different location to which response messages sent as part of
    the protocol or profile should be sent. The allowable syntax of this URI depends on
    the protocol binding.

The ResponseLocation attribute is used to enable different endpoints to be specified for
receiving request and response messages associated with a protocol or profile, not as a
means of load-balancing or redundancy (multiple elements of this type can be included
for this purpose). When a role contains an element of this type pertaining to a protocol
or profile for which only a single type of message (request or response) is applicable,
then the ResponseLocation attribute is unused. [E41]If the ResponseLocation attribute is
omitted, any response messages associated with a protocol or profile may be assumed to
be handled at the URI indicated by the Location attribute.

ArtifactResolutionService, SingleSignOnService and NameIDMappingService MUST omit the
ResponseLocation attribute. This is enforced here, but metadata with such service
declarations and such attributes should not have been part of the metadata store in the
first place.

Signed-off-by: Ivan Kanakarakis <[email protected]>

Author: c00kiemon5ter

src/saml2/client.py

@@ -29,7 +29,7 @@
 from saml2.client_base import SignOnError
 from saml2.client_base import LogoutError
 from saml2.client_base import NoServiceDefined
-from saml2.mdstore import destinations
+from saml2.mdstore import locations
 
 import logging
 
@@ -209,7 +209,7 @@ def do_logout(self, name_id, entity_ids, reason, expire, sign=None,
                     logger.debug("No SLO '%s' service", binding)
                     continue
 
-                destination = destinations(srvs)[0]
+                destination = next(locations(srvs), None)
                 logger.info("destination to provider: %s", destination)
                 try:
                     session_info = self.users.get_info_from(name_id,
@@ -374,7 +374,7 @@ def do_authz_decision_query(self, entity_id, action,
                                 name_qualifier=name_qualifier))
 
         srvs = self.metadata.authz_service(entity_id, BINDING_SOAP)
-        for dest in destinations(srvs):
+        for dest in locations(srvs):
             resp = self._use_soap(dest, "authz_decision_query",
                                   action=action, evidence=evidence,
                                   resource=resource, subject=subject)
@@ -397,7 +397,7 @@ def do_assertion_id_request(self, assertion_ids, entity_id,
 
         _id_refs = [AssertionIDRef(_id) for _id in assertion_ids]
 
-        for destination in destinations(srvs):
+        for destination in locations(srvs):
             res = self._use_soap(destination, "assertion_id_request",
                                  assertion_id_refs=_id_refs, consent=consent,
                                  extensions=extensions, sign=sign)
@@ -411,7 +411,7 @@ def do_authn_query(self, entity_id,
 
         srvs = self.metadata.authn_request_service(entity_id, BINDING_SOAP)
 
-        for destination in destinations(srvs):
+        for destination in locations(srvs):
             resp = self._use_soap(destination, "authn_query", consent=consent,
                                   extensions=extensions, sign=sign)
             if resp:
@@ -461,7 +461,7 @@ def do_attribute_query(self, entityid, subject_id,
             if srvs is []:
                 raise SAMLError("No attribute service support at entity")
 
-            destination = destinations(srvs)[0]
+            destination = next(locations(srvs), None)
 
         if binding == BINDING_SOAP:
             return self._use_soap(destination, "attribute_query",

src/saml2/client_base.py

@@ -12,7 +12,7 @@
 
 from saml2.entity import Entity
 
-from saml2.mdstore import destinations
+from saml2.mdstore import locations
 from saml2.profile import paos, ecp
 from saml2.saml import NAMEID_FORMAT_PERSISTENT
 from saml2.saml import NAMEID_FORMAT_TRANSIENT
@@ -212,7 +212,7 @@ def _sso_location(self, entityid=None, binding=BINDING_HTTP_REDIRECT):
             # verify that it's in the metadata
             srvs = self.metadata.single_sign_on_service(entityid, binding)
             if srvs:
-                return destinations(srvs)[0]
+                return next(locations(srvs), None)
             else:
                 logger.info("_sso_location: %s, %s", entityid, binding)
                 raise IdpUnspecified("No IdP to send to given the premises")
@@ -224,9 +224,8 @@ def _sso_location(self, entityid=None, binding=BINDING_HTTP_REDIRECT):
             raise IdpUnspecified("Too many IdPs to choose from: %s" % eids)
 
         try:
-            srvs = self.metadata.single_sign_on_service(list(eids.keys())[0],
-                                                        binding)
-            return destinations(srvs)[0]
+            srvs = self.metadata.single_sign_on_service(list(eids.keys())[0], binding)
+            return next(locations(srvs), None)
         except IndexError:
             raise IdpUnspecified("No IdP to send to given the premises")
 

src/saml2/entity.py

@@ -53,7 +53,7 @@
 from saml2.samlp import Artifact
 from saml2.samlp import LogoutRequest
 from saml2.samlp import AttributeQuery
-from saml2.mdstore import destinations, response_destinations
+from saml2.mdstore import all_locations
 from saml2 import BINDING_HTTP_POST
 from saml2 import BINDING_HTTP_REDIRECT
 from saml2 import BINDING_SOAP
@@ -249,8 +249,9 @@ def apply_binding(self, binding, msg_str, destination="", relay_state="",
 
         return info
 
-    def pick_binding(self, service, bindings=None, descr_type="", request=None,
-                     entity_id="", response=False):
+    def pick_binding(
+        self, service, bindings=None, descr_type="", request=None, entity_id=""
+    ):
         if request and not entity_id:
             entity_id = request.issuer.text.strip()
 
@@ -284,10 +285,8 @@ def pick_binding(self, service, bindings=None, descr_type="", request=None,
                             if srv["index"] == _index:
                                 return binding, srv["location"]
                     else:
-                        if response:
-                            return binding, response_destinations(srvs)[0]
-                        else:
-                            return binding, destinations(srvs)[0]
+                        destination = next(all_locations(srvs), None)
+                        return binding, destination
             except UnsupportedBinding:
                 pass
 
@@ -352,10 +351,9 @@ def response_args(self, message, bindings=None, descr_type=""):
                 else:
                     descr_type = "spsso"
 
-            binding, destination = self.pick_binding(rsrv, bindings,
-                                                     descr_type=descr_type,
-                                                     request=message,
-                                                     response=True)
+            binding, destination = self.pick_binding(
+                rsrv, bindings, descr_type=descr_type, request=message
+            )
             info["binding"] = binding
             info["destination"] = destination
 

src/saml2/mdstore.py

@@ -5,6 +5,8 @@
 import logging
 import os
 import sys
+from itertools import chain
+from warnings import warn as _warn
 
 from hashlib import sha1
 from os.path import isfile
@@ -26,7 +28,11 @@
 from saml2.httpbase import HTTPBase
 from saml2.extension.idpdisc import BINDING_DISCO
 from saml2.extension.idpdisc import DiscoveryResponse
+from saml2.md import NAMESPACE as NS_MD
 from saml2.md import EntitiesDescriptor
+from saml2.md import ArtifactResolutionService
+from saml2.md import NameIDMappingService
+from saml2.md import SingleSignOnService
 from saml2.mdie import to_dict
 from saml2.s_utils import UnsupportedBinding
 from saml2.s_utils import UnknownSystemEntity
@@ -70,6 +76,9 @@
         ns=NS_MDUI, tag=PrivacyStatementURL.c_tag
     ),
     "mdui_uiinfo_logo": "{ns}&{tag}".format(ns=NS_MDUI, tag=Logo.c_tag),
+    "service_artifact_resolution": "{ns}&{tag}".format(ns=NS_MD, tag=ArtifactResolutionService.c_tag),
+    "service_single_sign_on": "{ns}&{tag}".format(ns=NS_MD, tag=SingleSignOnService.c_tag),
+    "service_nameid_mapping": "{ns}&{tag}".format(ns=NS_MD, tag=NameIDMappingService.c_tag),
 }
 
 ENTITY_CATEGORY = "http://macedir.org/entity-category"
@@ -79,8 +88,6 @@
 SAML_METADATA_CONTENT_TYPE = "application/samlmetadata+xml"
 DEFAULT_FRESHNESS_PERIOD = "P0Y0M0DT12H0M0S"
 
-
-
 REQ2SRV = {
     # IDP
     "authn_request": "single_sign_on_service",
@@ -149,17 +156,54 @@ def metadata_modules():
     return _res
 
 
-def response_destinations(srvs):
-    _res = []
-    for s in srvs:
-        if "response_location" in s:
-            _res.append(s["response_location"])
-        else:
-            _res.append(s["location"])
-    return _res
+def response_locations(srvs):
+    """
+    Return the ResponseLocation attributes mapped to the services.
+
+    ArtifactResolutionService, SingleSignOnService and NameIDMappingService MUST omit
+    the ResponseLocation attribute. This is enforced here, but metadata with such
+    service declarations and such attributes should not have been part of the metadata
+    store in the first place.
+    """
+    values = (
+        s["response_location"]
+        for s in srvs
+        if "response_location" in s
+        if s["__class__"] not in [
+            classnames["service_artifact_resolution"],
+            classnames["service_single_sign_on"],
+            classnames["service_nameid_mapping"],
+        ]
+    )
+    return values
+
+
+def locations(srvs):
+    values = (
+        s["location"]
+        for s in srvs
+        if "location" in s
+    )
+    return values
+
 
 def destinations(srvs):
-    return [s["location"] for s in srvs]
+    warn_msg = (
+        "`saml2.mdstore.destinations` function is deprecated; "
+        "instead, use `saml2.mdstore.locations` or `saml2.mdstore.all_locations`."
+    )
+    logger.warning(warn_msg)
+    _warn(warn_msg)
+    values = list(locations(srvs))
+    return values
+
+
+def all_locations(srvs):
+    values = chain(
+        response_locations(srvs),
+        locations(srvs),
+    )
+    return values
 
 
 def attribute_requirement(entity, index=None):

tests/test_30_mdstore.py

@@ -15,7 +15,7 @@
 from saml2.mdstore import MetadataStore, MetaDataExtern
 from saml2.mdstore import MetaDataMDX
 from saml2.mdstore import SAML_METADATA_CONTENT_TYPE
-from saml2.mdstore import destinations
+from saml2.mdstore import locations
 from saml2.mdstore import name
 from saml2 import sigver
 from saml2.httpbase import HTTPBase
@@ -177,8 +177,9 @@ def test_swami_1():
     assert idps.keys()
     idpsso = mds.single_sign_on_service(UMU_IDP)
     assert len(idpsso) == 1
-    assert destinations(idpsso) == [
-        'https://idp.umu.se/saml2/idp/SSOService.php']
+    assert list(locations(idpsso)) == [
+        'https://idp.umu.se/saml2/idp/SSOService.php'
+    ]
 
     _name = name(mds[UMU_IDP])
     assert _name == u'Umeå University (SAML2)'
@@ -219,8 +220,9 @@ def test_incommon_1():
     idpsso = mds.single_sign_on_service('urn:mace:incommon:alaska.edu')
     assert len(idpsso) == 1
     print(idpsso)
-    assert destinations(idpsso) == [
-        'https://idp.alaska.edu/idp/profile/SAML2/Redirect/SSO']
+    assert list(locations(idpsso)) == [
+        'https://idp.alaska.edu/idp/profile/SAML2/Redirect/SSO'
+    ]
 
     sps = mds.with_descriptor("spsso")
 
@@ -279,8 +281,9 @@ def test_switch_1():
         'https://aai-demo-idp.switch.ch/idp/shibboleth')
     assert len(idpsso) == 1
     print(idpsso)
-    assert destinations(idpsso) == [
-        'https://aai-demo-idp.switch.ch/idp/profile/SAML2/Redirect/SSO']
+    assert list(locations(idpsso)) == [
+        'https://aai-demo-idp.switch.ch/idp/profile/SAML2/Redirect/SSO'
+    ]
     assert len(idps) > 30
     aas = mds.with_descriptor("attribute_authority")
     print(aas.keys())

tests/test_30_mdstore_old.py

@@ -6,7 +6,7 @@
 from unittest.mock import patch
 
 from saml2.mdstore import MetadataStore, MetaDataMDX
-from saml2.mdstore import destinations
+from saml2.mdstore import locations
 from saml2.mdstore import name
 
 from saml2 import md
@@ -145,8 +145,9 @@ def test_swami_1():
     assert idps.keys()
     idpsso = mds.single_sign_on_service(UMU_IDP)
     assert len(idpsso) == 1
-    assert destinations(idpsso) == [
-        'https://idp.umu.se/saml2/idp/SSOService.php']
+    assert list(locations(idpsso)) == [
+        'https://idp.umu.se/saml2/idp/SSOService.php'
+    ]
 
     _name = name(mds[UMU_IDP])
     assert _name == u'Umeå University (SAML2)'
@@ -187,8 +188,9 @@ def test_incommon_1():
     idpsso = mds.single_sign_on_service('urn:mace:incommon:alaska.edu')
     assert len(idpsso) == 1
     print(idpsso)
-    assert destinations(idpsso) == [
-        'https://idp.alaska.edu/idp/profile/SAML2/Redirect/SSO']
+    assert list(locations(idpsso)) == [
+        'https://idp.alaska.edu/idp/profile/SAML2/Redirect/SSO'
+    ]
 
     sps = mds.with_descriptor("spsso")
 
@@ -247,8 +249,9 @@ def test_switch_1():
         'https://aai-demo-idp.switch.ch/idp/shibboleth')
     assert len(idpsso) == 1
     print(idpsso)
-    assert destinations(idpsso) == [
-        'https://aai-demo-idp.switch.ch/idp/profile/SAML2/Redirect/SSO']
+    assert list(locations(idpsso)) == [
+        'https://aai-demo-idp.switch.ch/idp/profile/SAML2/Redirect/SSO'
+    ]
     assert len(idps) > 30
     aas = mds.with_descriptor("attribute_authority")
     print(aas.keys())

tests/test_50_server.py

@@ -2297,19 +2297,19 @@ def _logout_request(conf_file):
 
 
 class TestServerLogout():
-
     def test_1(self):
         with closing(Server("idp_slo_redirect_conf")) as server:
             req_id, request = _logout_request("sp_slo_redirect_conf")
             print(request)
             bindings = [BINDING_HTTP_REDIRECT]
             response = server.create_logout_response(request, bindings)
-            binding, destination = server.pick_binding("single_logout_service",
-                                                       bindings, "spsso",
-                                                       request, response=True)
 
-            http_args = server.apply_binding(binding, "%s" % response, destination,
-                                             "relay_state", response=True)
+            binding, destination = server.pick_binding(
+                "single_logout_service", bindings, "spsso", request
+            )
+            http_args = server.apply_binding(
+                binding, "%s" % response, destination, "relay_state", response=True
+            )
 
             assert len(http_args) == 4
             assert http_args["headers"][0][0] == "Location"
@@ -2322,18 +2322,20 @@ def test_2(self):
             print(request)
             bindings = [BINDING_HTTP_POST]
             response = server.create_logout_response(request, bindings)
-            binding, destination = server.pick_binding("single_logout_service",
-                                                       bindings, "spsso",
-                                                       request, response=True)
 
-            http_args = server.apply_binding(binding, "%s" % response, destination,
-                                             "relay_state", response=True)
+            binding, destination = server.pick_binding(
+                "single_logout_service", bindings, "spsso", request
+            )
+            http_args = server.apply_binding(
+                binding, "%s" % response, destination, "relay_state", response=True
+            )
 
             assert len(http_args) == 4
             assert len(http_args["data"]) > 0
             assert http_args["method"] == "POST"
             assert http_args['url'] == 'http://lingon.catalogix.se:8087/slo'
 
+
 if __name__ == "__main__":
     ts = TestServer1()
     ts.setup_class()

tests/test_76_metadata_in_mdb.py

@@ -5,7 +5,7 @@
 from saml2.mongo_store import export_mdstore_to_mongo_db
 from saml2.mongo_store import MetadataMDB
 from saml2.mdstore import MetadataStore
-from saml2.mdstore import destinations
+from saml2.mdstore import locations
 from saml2.mdstore import name
 from saml2 import config
 from pathutils import full_path
@@ -46,7 +46,7 @@ def test_metadata():
         assert idps.keys()
         idpsso = mds.single_sign_on_service(umu_idp)
         assert len(idpsso) == 1
-        assert destinations(idpsso) == [
+        assert list(locations(idpsso)) == [
             'https://idp.umu.se/saml2/idp/SSOService.php']
 
         _name = name(mds[umu_idp])