Support 'publish_metadata' in saml backend.

Has the same behavior as the same parameter in the saml frontend.

Author: Rebecka Gulliksson

example/plugins/frontends/saml2_frontend.yaml.example

@@ -32,6 +32,7 @@ config:
 
   state_id: <name>
   base: <base_url>
+  publish_metadata: <base_url>/<name>/metadata
   endpoints:
     single_sign_on_service: {'urn:oasis:names:tc:SAML:2.0:bindings:HTTP-POST': sso/post,
       'urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect': sso/redirect}

src/satosa/backends/saml2.py

@@ -24,29 +24,12 @@
 from satosa.logging_util import satosa_logging
 from satosa.metadata_creation.description import MetadataDescription, OrganizationDesc, \
     ContactPersonDesc, UIInfoDesc
-from satosa.response import SeeOther, Response
+from satosa.response import SeeOther, Response, MetadataResponse
 from satosa.util import rndstr, get_saml_name_id_format
 
 LOGGER = logging.getLogger(__name__)
 
 
-class MetadataResponse(Response):
-    """
-    A response containing metadata for the saml backend
-    """
-
-    def __init__(self, config):
-        """
-        Creates a response containing the metadata generated from the SP config.
-        :type config: dict[str, Any]
-        :param config: The SP config
-        """
-        metadata_string = create_metadata_string(None, config, 4, None, None, None, None,
-                                                 None).decode("utf-8")
-        resp = {"content": "text/xml"}
-        super(MetadataResponse, self).__init__(message=metadata_string, **resp)
-
-
 class SamlBackend(BackendModule):
     """
     A saml2 backend module

src/satosa/frontends/saml2.py

@@ -21,6 +21,7 @@
 from satosa.frontends.base import FrontendModule
 from satosa.internal_data import InternalRequest, DataConverter, UserIdHashType
 from satosa.logging_util import satosa_logging
+from satosa.response import MetadataResponse
 from satosa.util import response, get_saml_name_id_format, saml_name_format_to_hash_type
 
 LOGGER = logging.getLogger(__name__)
@@ -35,7 +36,8 @@ def __init__(self, auth_req_callback_func, internal_attributes, conf):
         self._validate_config(conf)
 
         super(SamlFrontend, self).__init__(auth_req_callback_func, internal_attributes)
-        self.config = conf["idp_config"]
+        self.config = conf
+        self.idp_config = conf["idp_config"]
         self.endpoints = conf["endpoints"]
         self.base = conf["base"]
         self.state_id = conf["state_id"]
@@ -82,9 +84,9 @@ def register_endpoints(self, providers):
         :rtype: list[(str, ((satosa.context.Context, Any) -> satosa.response.Response, Any))]
         """
         self._validate_providers(providers)
-        self.config = self._build_idp_config_endpoints(self.config, providers)
+        self.idp_config = self._build_idp_config_endpoints(self.idp_config, providers)
         # Create the idp
-        idp_config = IdPConfig().load(copy.deepcopy(self.config), metadata_construction=False)
+        idp_config = IdPConfig().load(copy.deepcopy(self.idp_config), metadata_construction=False)
         self.idp = Server(config=idp_config)
         return self._register_endpoints(providers)
 
@@ -446,6 +448,18 @@ def _validate_providers(self, providers):
             LOGGER.error(msg)
             raise TypeError(msg)
 
+    def _metadata(self, context):
+        """
+        Endpoint for retrieving the backend metadata
+        :type context: satosa.context.Context
+        :rtype: satosa.backends.saml2.MetadataResponse
+
+        :param context: The current context
+        :return: response with metadata
+        """
+        satosa_logging(LOGGER, logging.DEBUG, "Sending metadata response", context.state)
+        return MetadataResponse(self.idp.config)
+
     def _register_endpoints(self, providers):
         """
         Register methods to endpoints
@@ -468,6 +482,10 @@ def _register_endpoints(self, providers):
                 url_map.append(("(%s)/%s/(.*)$" % (valid_providers, parsed_endp.path),
                                 (self.handle_authn_request, binding)))
 
+        if "publish_metadata" in self.config:
+            metadata_path = urlparse(self.config["publish_metadata"])
+            url_map.append(("^%s$" % metadata_path.path[1:], self._metadata))
+
         return url_map
 
     def _build_idp_config_endpoints(self, config, providers):
@@ -573,7 +591,7 @@ def _load_idp_dynamic_endpoints(self, context):
         """
         target_entity_id = self._get_target_entity_id(context)
         context.internal_data["mirror.target_entity_id"] = target_entity_id
-        idp_conf_file = self._load_endpoints_to_config(self.config, self.endpoints, self.base,
+        idp_conf_file = self._load_endpoints_to_config(self.idp_config, self.endpoints, self.base,
                                                        context.target_backend, target_entity_id)
         idp_config = IdPConfig().load(idp_conf_file, metadata_construction=False)
         return Server(config=idp_config)
@@ -632,7 +650,7 @@ def handle_backend_error(self, exception):
         :type exception: satosa.exception.SATOSAAuthenticationError
         :rtype: satosa.response.Response
         """
-        idp = self._load_idp_dynamic_entity_id(self.config, exception.state)
+        idp = self._load_idp_dynamic_entity_id(self.idp_config, exception.state)
         return self._handle_backend_error(exception, idp)
 
     def handle_authn_response(self, context, internal_response):
@@ -642,7 +660,7 @@ def handle_authn_response(self, context, internal_response):
         :param internal_response:
         :return:
         """
-        idp = self._load_idp_dynamic_entity_id(self.config, context.state)
+        idp = self._load_idp_dynamic_entity_id(self.idp_config, context.state)
         return self._handle_authn_response(context, internal_response, idp)
 
     def register_endpoints(self, providers):

src/satosa/response.py

@@ -4,6 +4,7 @@
 from urllib.parse import quote
 
 import six
+from saml2.metadata import create_metadata_string
 
 __author__ = 'mathiashedstrom'
 
@@ -151,6 +152,23 @@ def __call__(self, environ, start_response):
         return self.to_list()
 
 
+class MetadataResponse(Response):
+    """
+    A response containing metadata for the saml backend
+    """
+
+    def __init__(self, config):
+        """
+        Creates a response containing the metadata generated from the SP config.
+        :type config: dict[str, Any]
+        :param config: The SP config
+        """
+        metadata_string = create_metadata_string(None, config, 4, None, None, None, None,
+                                                 None).decode("utf-8")
+        resp = {"content": "text/xml"}
+        super(MetadataResponse, self).__init__(message=metadata_string, **resp)
+
+
 def geturl(environ, query=True, path=True, use_server_name=False):
     """Rebuilds a request URL (from PEP 333).
     You may want to chose to use the environment variables

tests/satosa/frontends/test_saml2.py

@@ -51,14 +51,14 @@ def setup_for_authn_req(self, idp_conf, sp_conf, nameid_format):
                                     INTERNAL_ATTRIBUTES, config)
         samlfrontend.register_endpoints(["saml"])
 
-        idp_metadata_str = create_metadata_from_config_dict(samlfrontend.config)
+        idp_metadata_str = create_metadata_from_config_dict(samlfrontend.idp_config)
         sp_conf["metadata"]["inline"].append(idp_metadata_str)
 
         fakesp = FakeSP(None, config=SPConfig().load(sp_conf, metadata_construction=False))
         context = Context()
         context.state = State()
         context.request = parse.parse_qs(
-                urlparse(fakesp.make_auth_req(samlfrontend.config["entityid"], nameid_format)).query)
+                urlparse(fakesp.make_auth_req(samlfrontend.idp_config["entityid"], nameid_format)).query)
         tmp_dict = {}
         for val in context.request:
             if isinstance(context.request[val], list):
@@ -83,23 +83,29 @@ def test_config_error_handling(self, conf):
             SamlFrontend(lambda ctx, req: None, INTERNAL_ATTRIBUTES, conf)
 
     def test_register_endpoints(self, idp_conf):
+        """
+        Tests the method register_endpoints
+        """
+        def get_path_from_url(url):
+            return urlparse(url).path.lstrip("/")
+
+        metadata_url = "http://example.com/SAML2IDP/metadata"
         config = {"idp_config": idp_conf, "endpoints": ENDPOINTS,
                   "base": self.construct_base_url_from_entity_id(idp_conf["entityid"]),
-                  "state_id": "state_id"}
+                  "state_id": "state_id",
+                  "publish_metadata": metadata_url}
+
         samlfrontend = SamlFrontend(lambda context, internal_req: (context, internal_req),
-                                    INTERNAL_ATTRIBUTES, config)
+                                INTERNAL_ATTRIBUTES, config)
 
         providers = ["foo", "bar"]
         url_map = samlfrontend.register_endpoints(providers)
-        for k, v in idp_conf["service"]["idp"]["endpoints"].items():
-            for endp in v:
-                match = False
-                for regex in url_map:
-                    p = re.compile(regex[0])
-                    if p.match(urlparse(endp[0]).path.lstrip("/")):
-                        match = True
-                        break
-                assert match, "Not correct regular expression for endpoint: %s" % endp[0]
+        all_idp_endpoints = [get_path_from_url(v[0][0]) for v in idp_conf["service"]["idp"]["endpoints"].values()]
+        compiled_regex = [re.compile(regex) for regex, _ in url_map]
+        for endp in all_idp_endpoints:
+            assert any(p.match(endp) for p in compiled_regex)
+
+        assert any(p.match(get_path_from_url(metadata_url)) for p in compiled_regex)
 
     def test_handle_authn_request(self, idp_conf, sp_conf):
         """
@@ -200,7 +206,7 @@ def test_acr_mapping_in_authn_response(self, idp_conf, sp_conf):
         samlfrontend = SamlFrontend(None, INTERNAL_ATTRIBUTES, conf)
         samlfrontend.register_endpoints(["foo"])
 
-        idp_metadata_str = create_metadata_from_config_dict(samlfrontend.config)
+        idp_metadata_str = create_metadata_from_config_dict(samlfrontend.idp_config)
         sp_conf["metadata"]["inline"].append(idp_metadata_str)
         fakesp = FakeSP(None, config=SPConfig().load(sp_conf, metadata_construction=False))
 
@@ -241,7 +247,7 @@ def test_acr_mapping_per_idp_in_authn_response(self, idp_conf, sp_conf):
         samlfrontend = SamlFrontend(None, INTERNAL_ATTRIBUTES, conf)
         samlfrontend.register_endpoints(["foo"])
 
-        idp_metadata_str = create_metadata_from_config_dict(samlfrontend.config)
+        idp_metadata_str = create_metadata_from_config_dict(samlfrontend.idp_config)
         sp_conf["metadata"]["inline"].append(idp_metadata_str)
         fakesp = FakeSP(None, config=SPConfig().load(sp_conf, metadata_construction=False))