Fix open redirect in legacy SSO flow (idp)

changelog.d/18909.bugfix

@@ -0,0 +1 @@
+Fix open redirect in legacy SSO flow with the `idp` query parameter.

synapse/api/urls.py

@@ -22,6 +22,7 @@
 """Contains the URL paths to prefix various aspects of the server with."""
 
 import hmac
+import urllib.parse
 from hashlib import sha256
 from typing import Optional
 from urllib.parse import urlencode, urljoin
@@ -96,11 +97,21 @@ def build_login_sso_redirect_uri(
         serialized_query_parameters = urlencode({"redirectUrl": client_redirect_url})
 
         if idp_id:
+            # Since this is a user-controlled string, make it safe to include in a URL path.
+            url_encoded_idp_id = urllib.parse.quote(
+                idp_id,
+                # Since this defaults to `safe="/"`, we have to override it. We're
+                # working with an individual URL path parameter so there shouldn't be
+                # any slashes in it which could change the request path.
+                safe="",
+                encoding="utf8",
+            )
+
             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}",
+                f"{url_encoded_idp_id}?{serialized_query_parameters}",
             )
         else:
             resultant_url = f"{base_url}?{serialized_query_parameters}"

synapse/rest/synapse/client/pick_idp.py

@@ -63,6 +63,22 @@ async def _async_render_GET(self, request: SynapseRequest) -> None:
         if not idp:
             return await self._serve_id_picker(request, client_redirect_url)
 
+        # Validate the `idp` query parameter. We should only be working with known IdPs.
+        # No need waste further effort if we don't know about it.
+        #
+        # Although, we primarily prevent open redirect attacks by URL encoding all of
+        # the parameters we use in the redirect URL below, this validation also helps
+        # prevent Synapse from crafting arbitrary URLs and being used in open redirect
+        # attacks (defense in depth).
+        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"
+            )
+            return
+
         # 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).
         #

tests/api/test_urls.py

@@ -53,3 +53,29 @@ def test_tricky_redirect_uri(self) -> None:
             ),
             "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",
         )
+
+    def test_idp_id_with_slash_is_escaped(self) -> None:
+        """
+        Test to make sure that we properly URL encode the IdP ID.
+        """
+        self.assertEqual(
+            self.login_sso_redirect_url_builder.build_login_sso_redirect_uri(
+                idp_id="foo/bar",
+                client_redirect_url="http://example.com/redirect",
+            ),
+            "https://test/_matrix/client/v3/login/sso/redirect/foo%2Fbar?redirectUrl=http%3A%2F%2Fexample.com%2Fredirect",
+        )
+
+    def test_url_as_idp_id_is_escaped(self) -> None:
+        """
+        Test to make sure that we properly URL encode the IdP ID.
+
+        The IdP ID shouldn't be a URL.
+        """
+        self.assertEqual(
+            self.login_sso_redirect_url_builder.build_login_sso_redirect_uri(
+                idp_id="http://should-not-be-url.com/",
+                client_redirect_url="http://example.com/redirect",
+            ),
+            "https://test/_matrix/client/v3/login/sso/redirect/http%3A%2F%2Fshould-not-be-url.com%2F?redirectUrl=http%3A%2F%2Fexample.com%2Fredirect",
+        )

tests/rest/client/test_login.py

@@ -939,39 +939,32 @@ def test_login_via_oidc(self) -> None:
         self.assertEqual(chan.code, 200, chan.result)
         self.assertEqual(chan.json_body["user_id"], "@user1:test")
 
-    def test_multi_sso_redirect_to_unknown(self) -> None:
-        """An unknown IdP should cause a 404"""
+    def test_multi_sso_redirect_unknown_idp(self) -> None:
+        """An unknown IdP should cause a 400 bad request error"""
         channel = self.make_request(
             "GET",
             "/_synapse/client/pick_idp?redirectUrl=http://x&idp=xyz",
         )
-        self.assertEqual(channel.code, 302, channel.result)
-        location_headers = channel.headers.getRawHeaders("Location")
-        assert location_headers
-        sso_login_redirect_uri = location_headers[0]
-
-        # it should redirect us to the standard login SSO redirect flow
-        self.assertEqual(
-            sso_login_redirect_uri,
-            self.login_sso_redirect_url_builder.build_login_sso_redirect_uri(
-                idp_id="xyz", client_redirect_url="http://x"
-            ),
-        )
+        self.assertEqual(channel.code, 400, channel.result)
 
-        # follow the redirect
+    def test_multi_sso_redirect_unknown_idp_as_url(self) -> None:
+        """
+        An unknown IdP that looks like a URL should cause a 400 bad request error (to
+        avoid open redirects).
+
+        Ideally, we'd have another test for a known IdP with a URL as the `idp_id`, but
+        we can't configure that in our tests because the config validation on
+        `oidc_providers` only allows a subset of characters. If we could configure
+        `oidc_providers` with a URL as the `idp_id`, it should still be URL-encoded
+        properly to avoid open redirections. We do have `test_url_as_idp_id_is_escaped`
+        in the URL building tests to cover this case but is only a unit test vs
+        something at the REST layer here that covers things end-to-end.
+        """
         channel = self.make_request(
             "GET",
-            # We have to make this relative to be compatible with `make_request(...)`
-            get_relative_uri_from_absolute_uri(sso_login_redirect_uri),
-            # We have to set the Host header to match the `public_baseurl` to avoid
-            # the extra redirect in the `SsoRedirectServlet` in order for the
-            # cookies to be visible.
-            custom_headers=[
-                ("Host", SYNAPSE_SERVER_PUBLIC_HOSTNAME),
-            ],
+            "/_synapse/client/pick_idp?redirectUrl=something&idp=https://element.io/",
         )
-
-        self.assertEqual(channel.code, 404, channel.result)
+        self.assertEqual(channel.code, 400, channel.result)
 
     def test_client_idp_redirect_to_unknown(self) -> None:
         """If the client tries to pick an unknown IdP, return a 404"""