Consolidate SSO redirects through /_matrix/client/v3/login/sso/redirect(/{idpId})

changelog.d/17972.misc

@@ -0,0 +1 @@
+Consolidate SSO redirects through `/_matrix/client/v3/login/sso/redirect(/{idpId})`.

synapse/api/urls.py

@@ -23,7 +23,8 @@
 
 import hmac
 from hashlib import sha256
-from urllib.parse import urlencode
+from typing import Optional
+from urllib.parse import urlencode, urljoin
 
 from synapse.config import ConfigError
 from synapse.config.homeserver import HomeServerConfig
@@ -66,3 +67,42 @@ def build_user_consent_uri(self, user_id: str) -> str:
             urlencode({"u": user_id, "h": mac}),
         )
         return consent_uri
+
+
+class LoginSSORedirectURIBuilder:
+    def __init__(self, hs_config: HomeServerConfig):
+        self._public_baseurl = hs_config.server.public_baseurl
+
+    def build_login_sso_redirect_uri(
+        self, *, idp_id: Optional[str], client_redirect_url: str
+    ) -> str:
+        """Build a `/login/sso/redirect` URI for the given identity provider.
+
+        Builds `/_matrix/client/v3/login/sso/redirect/{idpId}?redirectUrl=xxx` when `idp_id` is specified.
+        Otherwise, builds `/_matrix/client/v3/login/sso/redirect?redirectUrl=xxx` when `idp_id` is `None`.
+
+        Args:
+            idp_id: Optional ID of the identity provider
+            client_redirect_url: URL to redirect the user to after login
+
+        Returns
+            The URI to follow when choosing a specific identity provider.
+        """
+        base_url = urljoin(
+            self._public_baseurl,
+            f"{CLIENT_API_PREFIX}/v3/login/sso/redirect",
+        )
+
+        serialized_query_parameters = urlencode({"redirectUrl": client_redirect_url})
+
+        if idp_id:
+            resultant_url = urljoin(
+                # We have to add a trailing slash to the base URL to ensure that the
+                # last path segment is not stripped away when joining with another path.
+                f"{base_url}/",
+                f"{idp_id}?{serialized_query_parameters}",
+            )
+        else:
+            resultant_url = f"{base_url}?{serialized_query_parameters}"
+
+        return resultant_url

synapse/config/cas.py

@@ -20,7 +20,7 @@
 #
 #
 
-from typing import Any, List
+from typing import Any, List, Optional
 
 from synapse.config.sso import SsoAttributeRequirement
 from synapse.types import JsonDict
@@ -46,7 +46,9 @@ def read_config(self, config: JsonDict, **kwargs: Any) -> None:
 
             # TODO Update this to a _synapse URL.
             public_baseurl = self.root.server.public_baseurl
-            self.cas_service_url = public_baseurl + "_matrix/client/r0/login/cas/ticket"
+            self.cas_service_url: Optional[str] = (
+                public_baseurl + "_matrix/client/r0/login/cas/ticket"
+            )
 
             self.cas_protocol_version = cas_config.get("protocol_version")
             if (

synapse/config/server.py

@@ -332,8 +332,14 @@ def read_config(self, config: JsonDict, **kwargs: Any) -> None:
             logger.info("Using default public_baseurl %s", public_baseurl)
         else:
             self.serve_client_wellknown = True
+            # Ensure that public_baseurl ends with a trailing slash
             if public_baseurl[-1] != "/":
                 public_baseurl += "/"
+
+        # Scrutinize user-provided config
+        if not isinstance(public_baseurl, str):
+            raise ConfigError("Must be a string", ("public_baseurl",))
+
         self.public_baseurl = public_baseurl
 
         # check that public_baseurl is valid

synapse/rest/synapse/client/pick_idp.py

@@ -21,6 +21,7 @@
 import logging
 from typing import TYPE_CHECKING
 
+from synapse.api.urls import LoginSSORedirectURIBuilder
 from synapse.http.server import (
     DirectServeHtmlResource,
     finish_request,
@@ -49,32 +50,32 @@ def __init__(self, hs: "HomeServer"):
             hs.config.sso.sso_login_idp_picker_template
         )
         self._server_name = hs.hostname
+        self._public_baseurl = hs.config.server.public_baseurl
+        self._login_sso_redirect_url_builder = LoginSSORedirectURIBuilder(hs.config)
 
     async def _async_render_GET(self, request: SynapseRequest) -> None:
         client_redirect_url = parse_string(
             request, "redirectUrl", required=True, encoding="utf-8"
         )
         idp = parse_string(request, "idp", required=False)
 
-        # if we need to pick an IdP, do so
+        # If we need to pick an IdP, do so
         if not idp:
             return await self._serve_id_picker(request, client_redirect_url)
 
-        # otherwise, redirect to the IdP's redirect URI
-        providers = self._sso_handler.get_identity_providers()
-        auth_provider = providers.get(idp)
-        if not auth_provider:
-            logger.info("Unknown idp %r", idp)
-            self._sso_handler.render_error(
-                request, "unknown_idp", "Unknown identity provider ID"
+        # Otherwise, redirect to the login SSO redirect endpoint for the given IdP
+        # (which will in turn take us to the the IdP's redirect URI).
+        #
+        # We could go directly to the IdP's redirect URI, but this way we ensure that
+        # the user goes through the same logic as normal flow. Additionally, if a proxy
+        # needs to intercept the request, it only needs to intercept the one endpoint.
+        sso_login_redirect_url = (
+            self._login_sso_redirect_url_builder.build_login_sso_redirect_uri(
+                idp_id=idp, client_redirect_url=client_redirect_url
             )
-            return
-
-        sso_url = await auth_provider.handle_redirect_request(
-            request, client_redirect_url.encode("utf8")
         )
-        logger.info("Redirecting to %s", sso_url)
-        request.redirect(sso_url)
+        logger.info("Redirecting to %s", sso_login_redirect_url)
+        request.redirect(sso_login_redirect_url)
         finish_request(request)
 
     async def _serve_id_picker(

tests/api/test_urls.py

@@ -0,0 +1,55 @@
+#
+# This file is licensed under the Affero General Public License (AGPL) version 3.
+#
+# Copyright (C) 2024 New Vector, Ltd
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# See the GNU Affero General Public License for more details:
+# <https://www.gnu.org/licenses/agpl-3.0.html>.
+#
+
+
+from twisted.test.proto_helpers import MemoryReactor
+
+from synapse.api.urls import LoginSSORedirectURIBuilder
+from synapse.server import HomeServer
+from synapse.util import Clock
+
+from tests.unittest import HomeserverTestCase
+
+# a (valid) url with some annoying characters in.  %3D is =, %26 is &, %2B is +
+TRICKY_TEST_CLIENT_REDIRECT_URL = 'https://x?<ab c>&q"+%3D%2B"="fö%26=o"'
+
+
+class LoginSSORedirectURIBuilderTestCase(HomeserverTestCase):
+    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
+        self.login_sso_redirect_url_builder = LoginSSORedirectURIBuilder(hs.config)
+
+    def test_no_idp_id(self) -> None:
+        self.assertEqual(
+            self.login_sso_redirect_url_builder.build_login_sso_redirect_uri(
+                idp_id=None, client_redirect_url="http://example.com/redirect"
+            ),
+            "https://test/_matrix/client/v3/login/sso/redirect?redirectUrl=http%3A%2F%2Fexample.com%2Fredirect",
+        )
+
+    def test_explicit_idp_id(self) -> None:
+        self.assertEqual(
+            self.login_sso_redirect_url_builder.build_login_sso_redirect_uri(
+                idp_id="oidc-github", client_redirect_url="http://example.com/redirect"
+            ),
+            "https://test/_matrix/client/v3/login/sso/redirect/oidc-github?redirectUrl=http%3A%2F%2Fexample.com%2Fredirect",
+        )
+
+    def test_tricky_redirect_uri(self) -> None:
+        self.assertEqual(
+            self.login_sso_redirect_url_builder.build_login_sso_redirect_uri(
+                idp_id="oidc-github",
+                client_redirect_url=TRICKY_TEST_CLIENT_REDIRECT_URL,
+            ),
+            "https://test/_matrix/client/v3/login/sso/redirect/oidc-github?redirectUrl=https%3A%2F%2Fx%3F%3Cab+c%3E%26q%22%2B%253D%252B%22%3D%22f%C3%B6%2526%3Do%22",
+        )