saml: Implement SP-initiated Logout.

Closes #20084

This is the flow that this implements:
1. A logged-in user clicks "Logout".
2. If they didn't auth via SAML, just do normal logout. Otherwise:
3. Form a LogoutRequest and redirect the user to
https://idp.example.com/slo-endpoint?SAMLRequest=<LogoutRequest here>
4. The IdP validates the LogoutRequest, terminates its own user session
and redirects the user to
https://thezuliporg.example.com/complete/saml/?SAMLRequest=<LogoutResponse>
with the appropriate LogoutResponse. In case of failure, the
LogoutResponse is expected to express that.
5. Zulip validates the LogoutResponse and if the response is a success
response, it executes the regular Zulip logout and the full flow is
finished.

docs/production/authentication-methods.md

@@ -653,13 +653,15 @@ integration](../production/scim.md).
          importing, only the certificate will be displayed (not the private
          key).
 
-### IdP-initiated SAML Logout
+### SAML Single Logout
 
-Zulip 5.0 introduces beta support for IdP-initiated SAML Logout. The
-implementation has primarily been tested with Keycloak and these
-instructions are for that provider; please [contact
-us](https://zulip.com/help/contact-support) for help using this with
-another IdP.
+Zulip supports both IdP-initiated and SP-initiated SAML Single
+Logout. The implementation has primarily been tested with Keycloak and
+these instructions are for that provider; please [contact
+us](https://zulip.com/help/contact-support) if you need help using
+this with another IdP.
+
+#### IdP-initated Single Logout
 
 1. In the KeyCloak configuration for Zulip, enable `Force Name ID Format`
    and set `Name ID Format` to `email`. Zulip needs to receive
@@ -698,11 +700,28 @@ another IdP.
    /home/zulip/deployments/current/manage.py logout_all_users
    ```
 
+#### SP-initiated Single Logout
+
+After configuring IdP-initiated Logout, you only need to set
+`SAML_ENABLE_SP_INITIATED_SINGLE_LOGOUT = True` in
+`/etc/zulip/settings.py` to also enable SP-initiated Logout. When this
+is active, a user who logged in to Zulip via SAML, upon clicking
+"Logout" in the Zulip web app will be redirected to the IdP's Single
+Logout endpoint with a `LogoutRequest`. If a successful
+`LogoutResponse` is received back, their current Zulip session will be
+terminated.
+
+Note that this doesn't work in the case of desktop and mobile application
+and is reserved to the browser.
+
 #### Caveats
 
-- This beta doesn't support using `SessionIndex` to limit which
-  sessions are affected; it always terminates all logged-in sessions
-  for the user identified in the `NameID`.
+- This implementation doesn't support using `SessionIndex` to limit which
+  sessions are affected; in IdP-initiated Logout it always terminates
+  all logged-in sessions for the user identified in the `NameID`.
+  In SP-initiated Logout this simply means that Zulip does not include
+  `SessionIndex` in the `LogoutRequest` to the IdP - however, that doesn't
+  seem to cause any undesired behavior with Keycloak.
 - SAML Logout in a configuration where your IdP handles authentication
   for multiple organizations is not yet supported.
 

zerver/tests/fixtures/saml/logoutresponse.txt

@@ -0,0 +1 @@
+<samlp:LogoutResponse xmlns:samlp="urn:oasis:names:tc:SAML:2.0:protocol" xmlns="urn:oasis:names:tc:SAML:2.0:assertion" xmlns:saml="urn:oasis:names:tc:SAML:2.0:assertion" Destination="http://zulip.testserver/complete/saml/" ID="ID_d76b3713-ef98-45b2-9ba3-f1c8a5f277f0" InResponseTo="ONELOGIN_0e2c575e2111fe775d776ece5cf7bbad24d1c669" IssueInstant="2022-04-17T20:45:46.273Z" Version="2.0"><Issuer>https://idp.testshib.org/idp/shibboleth</Issuer><samlp:Status><samlp:StatusCode Value="urn:oasis:names:tc:SAML:2.0:status:Success"/></samlp:Status></samlp:LogoutResponse>

zerver/tests/test_auth_backends.py

@@ -46,6 +46,7 @@ from django_auth_ldap.backend import LDAPSearch, _LDAPUser
 from jwt.exceptions import PyJWTError
 from onelogin.saml2.auth import OneLogin_Saml2_Auth
 from onelogin.saml2.logout_request import OneLogin_Saml2_Logout_Request
+from onelogin.saml2.logout_response import OneLogin_Saml2_Logout_Response
 from onelogin.saml2.response import OneLogin_Saml2_Response
 from onelogin.saml2.utils import OneLogin_Saml2_Utils
 from social_core.exceptions import AuthFailed, AuthStateForbidden
@@ -2066,6 +2067,128 @@ class SAMLAuthBackendTest(SocialAuthBase):
     def get_account_data_dict(self, email: str, name: str) -> Dict[str, Any]:
         return dict(email=email, name=name)
 
+    @override_settings(SAML_ENABLE_SP_INITIATED_SINGLE_LOGOUT=True)
+    def test_saml_sp_initiated_logout_success(self) -> None:
+        hamlet = self.example_user("hamlet")
+
+        # Begin by logging in via the IdP (the user needs to log in via SAML for the session
+        # for SP-initiated logout to make sense - and the logout flow uses information
+        # about the IdP that was used for login)
+        account_data_dict = self.get_account_data_dict(email=self.email, name=self.name)
+        with self.assertLogs(self.logger_string, level="INFO"):
+            result = self.social_auth_test(
+                account_data_dict,
+                expect_choose_email_screen=False,
+                subdomain="zulip",
+            )
+        self.client_get(result["Location"])
+
+        self.assert_logged_in_user_id(hamlet.id)
+
+        result = self.client_post("/accounts/logout/")
+        # A redirect to the IdP is returned.
+        self.assertEqual(result.status_code, 302)
+        self.assertIn(
+            settings.SOCIAL_AUTH_SAML_ENABLED_IDPS["test_idp"]["slo_url"], result["Location"]
+        )
+        # This doesn't log the user out yet.
+        self.assert_logged_in_user_id(hamlet.id)
+
+        # Verify the redirect has the correct form - a LogoutRequest for hamlet
+        # is delivered to the IdP in the SAMLRequest param.
+        query_dict = urllib.parse.parse_qs(urllib.parse.urlparse(result["Location"]).query)
+        saml_request_encoded = query_dict["SAMLRequest"][0]
+        saml_request = OneLogin_Saml2_Utils.decode_base64_and_inflate(saml_request_encoded).decode()
+        self.assertIn("<samlp:LogoutRequest", saml_request)
+        self.assertIn(f"saml:NameID>{hamlet.delivery_email}</saml:NameID>", saml_request)
+
+        unencoded_logout_response = self.fixture_data("logoutresponse.txt", type="saml")
+        logout_response: str = base64.b64encode(unencoded_logout_response.encode()).decode()
+        # It's hard to create fully-correct LogoutResponse with signatures in tests,
+        # so we rely on mocking the validating functions instead.
+        with mock.patch.object(
+            OneLogin_Saml2_Logout_Response, "is_valid", return_value=True
+        ), mock.patch.object(
+            OneLogin_Saml2_Auth,
+            "validate_response_signature",
+            return_value=True,
+        ):
+            result = self.client_get(
+                "/complete/saml/",
+                {
+                    "SAMLResponse": logout_response,
+                    "SigAlg": "http://www.w3.org/2001/04/xmldsig-more#rsa-sha256",
+                    "Signature": "foo",
+                },
+            )
+        self.assertEqual(result.status_code, 302)
+        self.assertEqual(result["Location"], "/accounts/login/")
+        self.client_get(result["Location"])
+        self.assert_logged_in_user_id(None)
+
+    @override_settings(SAML_ENABLE_SP_INITIATED_SINGLE_LOGOUT=True)
+    def test_saml_sp_initiated_logout_invalid_logoutresponse(self) -> None:
+        hamlet = self.example_user("hamlet")
+        self.login("hamlet")
+        self.assert_logged_in_user_id(hamlet.id)
+
+        unencoded_logout_response = self.fixture_data("logoutresponse.txt", type="saml")
+        logout_response: str = base64.b64encode(unencoded_logout_response.encode()).decode()
+        result = self.client_get(
+            "/complete/saml/",
+            {
+                "SAMLResponse": logout_response,
+                "SigAlg": "http://www.w3.org/2001/04/xmldsig-more#rsa-sha256",
+                "Signature": "foo",
+            },
+        )
+        self.assert_json_error(
+            result,
+            "LogoutResponse error: ['invalid_logout_response_signature', 'Signature validation failed. Logout Response rejected']",
+        )
+        self.assert_logged_in_user_id(hamlet.id)
+
+    @override_settings(SAML_ENABLE_SP_INITIATED_SINGLE_LOGOUT=True)
+    def test_saml_sp_initiated_logout_endpoint_when_not_logged_in(self) -> None:
+        self.assert_logged_in_user_id(None)
+
+        result = self.client_post("/accounts/logout/")
+        self.assert_json_error(result, "Not logged in.")
+
+    @override_settings(SAML_ENABLE_SP_INITIATED_SINGLE_LOGOUT=True)
+    def test_saml_sp_initiated_logout_logged_in_not_via_saml(self) -> None:
+        """
+        If the user is logged in, but not via SAML, the normal logout flow
+        should be executed instead of the SAML SP-initiated logout flow.
+        """
+        hamlet = self.example_user("hamlet")
+        self.login("hamlet")
+        self.assert_logged_in_user_id(hamlet.id)
+
+        result = self.client_post("/accounts/logout/")
+        self.assertEqual(result.status_code, 302)
+        self.assertEqual(result["Location"], "/accounts/login/")
+        self.client_get(result["Location"])
+        self.assert_logged_in_user_id(None)
+
+    @override_settings(SAML_ENABLE_SP_INITIATED_SINGLE_LOGOUT=True)
+    def test_saml_sp_initiated_logout_when_saml_not_enabled(self) -> None:
+        """
+        If SAML is not enabled, the normal logout flow should be correctly executed.
+        This test verifies that this scenario doesn't end up with some kind of error
+        due to going down the SAML SP-initiated logout codepaths.
+        """
+        hamlet = self.example_user("hamlet")
+        self.login("hamlet")
+        self.assert_logged_in_user_id(hamlet.id)
+
+        with self.settings(AUTHENTICATION_BACKENDS=("zproject.backends.EmailAuthBackend",)):
+            result = self.client_post("/accounts/logout/")
+        self.assertEqual(result.status_code, 302)
+        self.assertEqual(result["Location"], "/accounts/login/")
+        self.client_get(result["Location"])
+        self.assert_logged_in_user_id(None)
+
     def test_saml_idp_initiated_logout_success(self) -> None:
         hamlet = self.example_user("hamlet")
         old_api_key = hamlet.api_key

zerver/views/auth.py

@@ -86,6 +86,7 @@ from zproject.backends import (
     ExternalAuthResult,
     GenericOpenIdConnectBackend,
     SAMLAuthBackend,
+    SAMLSPInitiatedLogout,
     ZulipLDAPAuthBackend,
     ZulipLDAPConfigurationError,
     ZulipRemoteUserBackend,
@@ -1117,6 +1118,28 @@ def json_fetch_api_key(
 logout_then_login = require_post(django_logout_then_login)
 
 
+@require_post
+def logout_view(request: HttpRequest, /, **kwargs: Any) -> HttpResponse:
+    realm = RequestNotes.get_notes(request).realm
+    assert realm is not None
+
+    if not settings.SAML_ENABLE_SP_INITIATED_SINGLE_LOGOUT or not saml_auth_enabled(realm):
+        return logout_then_login(request, **kwargs)
+
+    if not request.user.is_authenticated:
+        raise JsonableError(_("Not logged in."))
+
+    # This will first redirect to the IdP with a LogoutRequest and if successful on the IdP side,
+    # the user will be redirected to our SAMLResponse-handling endpoint with a success LogoutResponse,
+    # where we will finally terminate their session.
+    result = SAMLSPInitiatedLogout.slo_request_to_idp(request, return_to=None)
+    if result is None:
+        # This session wasn't authenticated via SAML, so proceed with normal logout process.
+        return logout_then_login(request, **kwargs)
+
+    return result
+
+
 def password_reset(request: HttpRequest) -> HttpResponse:
     if is_subdomain_root_or_alias(request) and settings.ROOT_DOMAIN_LANDING_PAGE:
         redirect_url = append_url_query_string(

zproject/backends.py

@@ -49,11 +49,15 @@ from django.urls import reverse
 from django.utils.translation import gettext as _
 from django_auth_ldap.backend import LDAPBackend, _LDAPUser, ldap_error
 from lxml.etree import XMLSyntaxError
+from onelogin.saml2 import compat as onelogin_saml2_compat
 from onelogin.saml2.auth import OneLogin_Saml2_Auth
 from onelogin.saml2.errors import OneLogin_Saml2_Error
 from onelogin.saml2.logout_request import OneLogin_Saml2_Logout_Request
+from onelogin.saml2.logout_response import OneLogin_Saml2_Logout_Response
 from onelogin.saml2.response import OneLogin_Saml2_Response
 from onelogin.saml2.settings import OneLogin_Saml2_Settings
+from onelogin.saml2.utils import OneLogin_Saml2_Utils
+from onelogin.saml2.xml_utils import OneLogin_Saml2_XML
 from requests import HTTPError
 from social_core.backends.apple import AppleIdAuth
 from social_core.backends.azuread import AzureADOAuth2
@@ -71,6 +75,7 @@ from social_core.exceptions import (
     SocialAuthBaseException,
 )
 from social_core.pipeline.partial import partial
+from social_django.utils import load_backend, load_strategy
 from zxcvbn import zxcvbn
 
 from zerver.actions.create_user import do_create_user, do_reactivate_user
@@ -1301,6 +1306,10 @@ class ExternalAuthDataDict(TypedDict, total=False):
     desktop_flow_otp: Optional[str]
     multiuse_object_key: str
     full_name_validated: bool
+    # TODO: This currently does not get plumbed through to the final session
+    # in the desktop flow, but it should.
+    # Also, the mobile app doesn't actually use a session, so this
+    # data is not applicable there either.
     params_to_store_in_authenticated_session: Dict[str, str]
 
 
@@ -2258,10 +2267,29 @@ class SAMLDocument:
         self.encoded_saml_message = encoded_saml_message
         self.backend = backend
 
+        self._decoded_saml_message: Optional[str] = None
+
     @property
     def logger(self) -> logging.Logger:
         return self.backend.logger
 
+    @property
+    def decoded_saml_message(self) -> str:
+        """
+        Returns the decoded SAMLRequest/SAMLResponse.
+        """
+        if self._decoded_saml_message is None:
+            # This logic is taken from how
+            # python3-saml handles decoding received SAMLRequest
+            # and SAMLResponse params.
+            self._decoded_saml_message = onelogin_saml2_compat.to_string(
+                OneLogin_Saml2_Utils.decode_base64_and_inflate(
+                    self.encoded_saml_message, ignore_zip=True
+                )
+            )
+
+        return self._decoded_saml_message
+
     def document_type(self) -> str:
         """
         Returns whether the instance is a SAMLRequest or SAMLResponse.
@@ -2318,14 +2346,32 @@ class SAMLResponse(SAMLDocument):
         saml_settings = OneLogin_Saml2_Settings(config, sp_validation_only=True)
 
         try:
+            if not self.is_logout_response():
                 resp = OneLogin_Saml2_Response(
                     settings=saml_settings, response=self.encoded_saml_message
                 )
                 return resp.get_issuers()
+            else:
+                logout_response = OneLogin_Saml2_Logout_Response(
+                    settings=saml_settings, response=self.encoded_saml_message
+                )
+                return logout_response.get_issuer()
         except self.SAML_PARSING_EXCEPTIONS as e:
             self.logger.error("Error parsing SAMLResponse: %s", str(e))
             return []
 
+    def is_logout_response(self) -> bool:
+        """
+        Checks whether the SAMLResponse is a LogoutResponse based on some
+        basic XML parsing.
+        """
+
+        try:
+            parsed_xml = OneLogin_Saml2_XML.to_etree(self.decoded_saml_message)
+            return bool(OneLogin_Saml2_XML.query(parsed_xml, "/samlp:LogoutResponse"))
+        except self.SAML_PARSING_EXCEPTIONS:
+            return False
+
 
 @external_auth_method
 class SAMLAuthBackend(SocialAuthMixin, SAMLAuth):
@@ -2623,7 +2669,14 @@ class SAMLAuthBackend(SocialAuthMixin, SAMLAuth):
             return None
 
         if isinstance(saml_document, SAMLRequest):
+            # We're a Service Provider, so the only SAMLRequest we accept is a LogoutRequest. Thus
+            # we can proceed with process_logout and it's for the lower level libraries to reject it
+            # if it's not a valid LogoutRequest.
             return self.process_logout(subdomain, idp_name)
+        elif isinstance(saml_document, SAMLResponse) and saml_document.is_logout_response():
+            # As a Service Provider, we process SAMLResponse which can be either LogoutResponse
+            # or an authentication response.
+            return SAMLSPInitiatedLogout.process_logout_response(saml_document, idp_name)
 
         result = None
         try:
@@ -2808,6 +2861,75 @@ def validate_otp_params(
         raise JsonableError(_("Can't use both mobile_flow_otp and desktop_flow_otp together."))
 
 
+class SAMLSPInitiatedLogout:
+    @classmethod
+    def get_logged_in_user_idp(cls, request: HttpRequest) -> Optional[str]:
+        """
+        Information about the authentication method which was used for
+        this session is stored in social_auth_backend session attribute.
+        If SAML was used, this extracts the IdP name and returns it.
+        """
+        # Some asserts to ensure this doesn't get called incorrectly:
+        assert hasattr(request, "user")
+        assert isinstance(request.user, UserProfile)
+
+        authentication_method = request.session.get("social_auth_backend", "")
+        if not authentication_method.startswith("saml:"):
+            return None
+
+        return authentication_method.split("saml:")[1]
+
+    @classmethod
+    def slo_request_to_idp(
+        cls, request: HttpRequest, return_to: Optional[str] = None
+    ) -> Optional[HttpResponse]:
+        """
+        Generates the redirect to the IdP's SLO endpoint with
+        the appropriately generated LogoutRequest or None if the session
+        wasn't authenticated via SAML.
+        """
+
+        user_profile = request.user
+        assert isinstance(user_profile, UserProfile)
+
+        realm = user_profile.realm
+        assert saml_auth_enabled(realm)
+
+        complete_url = reverse("social:complete", args=("saml",))
+        saml_backend = load_backend(load_strategy(request), "saml", complete_url)
+
+        idp_name = cls.get_logged_in_user_idp(request)
+        if idp_name is None:
+            return None
+
+        idp = saml_backend.get_idp(idp_name)
+        auth = saml_backend._create_saml_auth(idp)
+        slo_url = auth.logout(name_id=user_profile.delivery_email, return_to=return_to)
+
+        return HttpResponseRedirect(slo_url)
+
+    @classmethod
+    def process_logout_response(cls, logout_response: SAMLResponse, idp_name: str) -> HttpResponse:
+        """
+        Validates the LogoutResponse and logs out the user if successful,
+        finishing the SP-initiated logout flow.
+        """
+        from django.contrib.auth.views import logout_then_login as django_logout_then_login
+
+        idp = logout_response.backend.get_idp(idp_name)
+        auth = logout_response.backend._create_saml_auth(idp)
+        auth.process_slo(keep_local_session=True)
+        errors = auth.get_errors()
+        if errors:
+            # These errors should essentially only happen in case of misconfiguration,
+            # so we give a json error response with the direct error codes from python3-saml.
+            # They're informative but generic enough to not leak any sensitive information.
+            raise JsonableError(f"LogoutResponse error: {errors}")
+
+        # We call Django's version of logout_then_login so that POST isn't required.
+        return django_logout_then_login(logout_response.backend.strategy.request)
+
+
 def get_external_method_dicts(realm: Optional[Realm] = None) -> List[ExternalAuthMethodDictT]:
     """
     Returns a list of dictionaries that represent social backends, sorted

zproject/default_settings.py

@@ -91,6 +91,9 @@ SOCIAL_AUTH_SAML_SECURITY_CONFIG: Dict[str, Any] = {}
 # Set this to True to enforce that any configured IdP needs to specify
 # the limit_to_subdomains setting to be considered valid:
 SAML_REQUIRE_LIMIT_TO_SUBDOMAINS = False
+
+SAML_ENABLE_SP_INITIATED_SINGLE_LOGOUT = False
+
 # Historical name for SOCIAL_AUTH_GITHUB_KEY; still allowed in production.
 GOOGLE_OAUTH2_CLIENT_ID: Optional[str] = None
 

zproject/prod_settings_template.py

@@ -494,6 +494,9 @@ SOCIAL_AUTH_SAML_SUPPORT_CONTACT = {
 #    }
 # }
 
+# This setting allows enabling of SP-initiated logout with SAML.
+# SAML_ENABLE_SP_INITIATED_SINGLE_LOGOUT = True
+
 ########
 ## Apple authentication ("Sign in with Apple").
 ##

zproject/urls.py

@@ -29,7 +29,7 @@ from zerver.views.auth import (
     jwt_fetch_api_key,
     log_into_subdomain,
     login_page,
-    logout_then_login,
+    logout_view,
     password_reset,
     remote_user_jwt,
     remote_user_sso,
@@ -533,7 +533,7 @@ i18n_urls = [
     # return `/accounts/login/`.
     path("accounts/login/", login_page, {"template_name": "zerver/login.html"}, name="login_page"),
     path("accounts/login/", LoginView.as_view(template_name="zerver/login.html"), name="login"),
-    path("accounts/logout/", logout_then_login),
+    path("accounts/logout/", logout_view),
     path("accounts/webathena_kerberos_login/", webathena_kerberos_login),
     path("accounts/password/reset/", password_reset, name="password_reset"),
     path(