saml2 backend: support using multiple ACS URLs (#409)

* saml2 backend: support using multiple ACS URLs

When Satosa sends out a SAML2 AuthnRequest, it specifies the
AssertionConsumerServiceUrl parameter as well, unless the
`hide_assertion_consumer_service` configuration parameter is set. However,
Satosa might be deployed in an environment where not all interfaces and host
names are accessible for all users. After this change, Satosa tries to
select the ACS URL based on the current request, and falls back to the first
ACS if there is no match.

* squash! saml2 backend: support using multiple ACS URLs

Make ACS selection configurable with the `acs_selection_strategy`
parameter, keeping the default backwards-compatible (`use_first_acs`).
Added the relevant example and documentation.

Additionally, log an error (instead of debug) message if the
authentication request can not be constructed, since most of the time
this is a configuration or environment error.

Author: bajnokk

doc/README.md

@@ -424,6 +424,26 @@ config:
   [...]
 ```
 
+#### Assertion Consumer Service selection
+
+When SATOSA sends the SAML2 authentication request to the IDP, it always
+specifies the AssertionConsumerServiceURL and binding. When
+`acs_selection_strategy` configuration option is set to `use_first_acs` (the
+default), then the first element of the `assertion_consumer_service` list will
+be selected. If `acs_selection_strategy` is `prefer_matching_host`, then SATOSA
+will try to select the `assertion_consumer_service`, which matches the host in
+the HTTP request (in simple words, it tries to select an ACS that matches the
+URL in the user's browser). If there is no match, it will fall back to using the
+first assertion consumer service.
+
+Default value: `use_first_acs`.
+
+```yaml
+config:
+  acs_selection_strategy: prefer_matching_host
+  [...]
+```
+
 ## OpenID Connect plugins
 
 ### OIDC Frontend

example/plugins/backends/saml2_backend.yaml.example

@@ -16,6 +16,7 @@ config:
   use_memorized_idp_when_force_authn: no
   send_requester_id: no
   enable_metadata_reload: no
+  acs_selection_strategy: prefer_matching_host
 
   sp_config:
     name: "SP Name"

src/satosa/backends/saml2.py

@@ -289,18 +289,19 @@ def authn_request(self, context, entity_id):
             kwargs["is_passive"] = "true"
 
         try:
-            acs_endp, response_binding = self.sp.config.getattr("endpoints", "sp")["assertion_consumer_service"][0]
+            acs_endp, response_binding = self._get_acs(context)
             relay_state = util.rndstr()
             req_id, binding, http_info = self.sp.prepare_for_negotiated_authenticate(
                 entityid=entity_id,
+                assertion_consumer_service_url=acs_endp,
                 response_binding=response_binding,
                 relay_state=relay_state,
                 **kwargs,
             )
         except Exception as e:
             msg = "Failed to construct the AuthnRequest for state"
             logline = lu.LOG_FMT.format(id=lu.get_session_id(context.state), message=msg)
-            logger.debug(logline, exc_info=True)
+            logger.error(logline, exc_info=True)
             raise SATOSAAuthenticationError(context.state, "Failed to construct the AuthnRequest") from e
 
         if self.sp.config.getattr('allow_unsolicited', 'sp') is False:
@@ -314,6 +315,58 @@ def authn_request(self, context, entity_id):
         context.state[self.name] = {"relay_state": relay_state}
         return make_saml_response(binding, http_info)
 
+    def _get_acs(self, context):
+        """
+        Select the AssertionConsumerServiceURL and binding.
+
+        :param context: The current context
+        :type context: satosa.context.Context
+        :return: Selected ACS URL and binding
+        :rtype: tuple(str, str)
+        """
+        acs_strategy = self.config.get("acs_selection_strategy", "use_first_acs")
+        if acs_strategy == "use_first_acs":
+            acs_strategy_fn = self._use_first_acs
+        elif acs_strategy == "prefer_matching_host":
+            acs_strategy_fn = self._prefer_matching_host
+        else:
+            msg = "Invalid value for '{}' ({}). Using the first ACS instead".format(
+                "acs_selection_strategy", acs_strategy
+            )
+            logger.error(msg)
+            acs_strategy_fn = self._use_first_acs
+        return acs_strategy_fn(context)
+
+    def _use_first_acs(self, context):
+        return self.sp.config.getattr("endpoints", "sp")["assertion_consumer_service"][
+            0
+        ]
+
+    def _prefer_matching_host(self, context):
+        acs_config = self.sp.config.getattr("endpoints", "sp")[
+            "assertion_consumer_service"
+        ]
+        try:
+            hostname = context.http_headers["HTTP_HOST"]
+            for acs, binding in acs_config:
+                parsed_acs = urlparse(acs)
+                if hostname == parsed_acs.netloc:
+                    msg = "Selected ACS '{}' based on the request".format(acs)
+                    logline = lu.LOG_FMT.format(
+                        id=lu.get_session_id(context.state), message=msg
+                    )
+                    logger.debug(logline)
+                    return acs, binding
+        except (TypeError, KeyError):
+            pass
+
+        msg = "Can't find an ACS URL to this hostname ({}), selecting the first one".format(
+            context.http_headers.get("HTTP_HOST", "") if context.http_headers else ""
+        )
+        logline = lu.LOG_FMT.format(id=lu.get_session_id(context.state), message=msg)
+        logger.debug(logline)
+        return self._use_first_acs(context)
+
     def authn_response(self, context, binding):
         """
         Endpoint for the idp response

tests/satosa/backends/test_saml2.py

@@ -12,9 +12,11 @@
 import pytest
 
 import saml2
-from saml2 import BINDING_HTTP_REDIRECT
+from saml2 import BINDING_HTTP_REDIRECT, BINDING_HTTP_POST
 from saml2.authn_context import PASSWORD
 from saml2.config import IdPConfig, SPConfig
+from saml2.entity import Entity
+from saml2.samlp import authn_request_from_string
 from saml2.s_utils import deflate_and_base64_encode
 
 from satosa.backends.saml2 import SAMLBackend
@@ -179,6 +181,64 @@ def test_authn_request(self, context, idp_conf):
         req_params = dict(parse_qsl(urlparse(resp.message).query))
         assert context.state[self.samlbackend.name]["relay_state"] == req_params["RelayState"]
 
+    @pytest.mark.parametrize("hostname", ["example.com:8443", "example.net"])
+    @pytest.mark.parametrize(
+        "strat",
+        ["", "use_first_acs", "prefer_matching_host", "invalid"],
+    )
+    def test_acs_selection_strategy(self, context, sp_conf, idp_conf, hostname, strat):
+        acs_endpoints = [
+            ("https://example.com/saml2/acs/post", BINDING_HTTP_POST),
+            ("https://example.net/saml2/acs/post", BINDING_HTTP_POST),
+            ("https://example.com:8443/saml2/acs/post", BINDING_HTTP_POST),
+        ]
+        config = {"sp_config": sp_conf}
+        config["sp_config"]["service"]["sp"]["endpoints"][
+            "assertion_consumer_service"
+        ] = acs_endpoints
+        if strat:
+            config["acs_selection_strategy"] = strat
+
+        req = self._make_authn_request(hostname, context, config, idp_conf["entityid"])
+
+        if strat == "prefer_matching_host":
+            expected_acs = hostname
+        else:
+            expected_acs = urlparse(acs_endpoints[0][0]).netloc
+        assert urlparse(req.assertion_consumer_service_url).netloc == expected_acs
+
+    def _make_authn_request(self, http_host, context, config, entity_id):
+        context.http_headers = {"HTTP_HOST": http_host} if http_host else {}
+        self.samlbackend = SAMLBackend(
+            Mock(),
+            INTERNAL_ATTRIBUTES,
+            config,
+            "base_url",
+            "samlbackend",
+        )
+        resp = self.samlbackend.authn_request(context, entity_id)
+        req_params = dict(parse_qsl(urlparse(resp.message).query))
+        req_xml = Entity.unravel(req_params["SAMLRequest"], BINDING_HTTP_REDIRECT)
+        return authn_request_from_string(req_xml)
+
+    @pytest.mark.parametrize("hostname", ["unknown-hostname", None])
+    def test_unknown_or_no_hostname_selects_first_acs(
+        self, context, sp_conf, idp_conf, hostname
+    ):
+        config = {"sp_config": sp_conf}
+        config["sp_config"]["service"]["sp"]["endpoints"][
+            "assertion_consumer_service"
+        ] = (
+            ("https://first-hostname/saml2/acs/post", BINDING_HTTP_POST),
+            ("https://other-hostname/saml2/acs/post", BINDING_HTTP_POST),
+        )
+        config["acs_selection_strategy"] = "prefer_matching_host"
+        req = self._make_authn_request(hostname, context, config, idp_conf["entityid"])
+        assert (
+            req.assertion_consumer_service_url
+            == "https://first-hostname/saml2/acs/post"
+        )
+
     def test_authn_response(self, context, idp_conf, sp_conf):
         response_binding = BINDING_HTTP_REDIRECT
         fakesp = FakeSP(SPConfig().load(sp_conf))