Support alternate SAML auth flow with ForceAuthn query param (PP-4792)

src/palace/manager/integration/patron_auth/saml/auth.py

@@ -19,7 +19,7 @@
 from palace.manager.integration.patron_auth.saml.metadata.parser import (
     SAMLSubjectParser,
 )
-from palace.manager.util.problem_detail import ProblemDetail as pd
+from palace.manager.util.problem_detail import ProblemDetail, ProblemDetail as pd
 
 if TYPE_CHECKING:
     import sqlalchemy.orm.session
@@ -177,20 +177,26 @@ def configuration(self):
         """
         return self._configuration
 
-    def start_authentication(self, db, idp_entity_id, return_to_url):
+    def start_authentication(
+        self,
+        db: sqlalchemy.orm.session.Session,
+        idp_entity_id: str,
+        return_to_url: str,
+        force_authn: bool = False,
+    ) -> str | ProblemDetail:
         """Start the SAML authentication workflow by sending a AuthnRequest to the IdP.
 
         :param db: Database session
-        :type db: sqlalchemy.orm.session.Session
 
         :param idp_entity_id: IdP's entityID
-        :type idp_entity_id: string
 
         :param return_to_url: URL which will the user agent will be redirected to after authentication
-        :type return_to_url: string
 
-        :return: Redirection URL
-        :rtype: string
+        :param force_authn: Whether to set ForceAuthn="true" on the AuthnRequest,
+            telling the IdP to re-authenticate the user even if they have an
+            existing session (SAML 2.0 Core 3.4.1)
+
+        :return: Redirection URL, or a ProblemDetail on error
         """
         self._logger.info(
             "Started authentication workflow for IdP '{}' (redirection URL = '{}')".format(
@@ -200,7 +206,7 @@ def start_authentication(self, db, idp_entity_id, return_to_url):
 
         try:
             auth = self._get_auth_object(db, idp_entity_id)
-            redirect_url = auth.login(return_to_url)
+            redirect_url = auth.login(return_to_url, force_authn=force_authn)
 
             if self._logger.isEnabledFor(logging.DEBUG):
                 self._logger.debug(f"SAML request: {auth.get_last_request_xml()}")
@@ -281,7 +287,7 @@ def start_logout(
         name_id: SAMLNameID,
         sp_slo_callback_url: str,
         relay_state: str,
-    ) -> str | pd:
+    ) -> str | ProblemDetail:
         """Initiate SP-Initiated SAML SLO by sending a LogoutRequest to the IdP.
 
         :param db: Database session

src/palace/manager/integration/patron_auth/saml/controller.py

@@ -16,7 +16,7 @@
 from flask_babel import lazy_gettext as _
 from lxml import etree
 from werkzeug import Response as wkResponse
-from werkzeug.datastructures import ImmutableMultiDict
+from werkzeug.datastructures import MultiDict
 
 from palace.manager.api.util.flask import get_request_library
 from palace.manager.integration.patron_auth.constants import LOGOUT_REDIRECT_QUERY_PARAM
@@ -69,6 +69,9 @@ class SAMLController:
     ACCESS_TOKEN = "access_token"
     PATRON_INFO = "patron_info"
     LOGOUT_STATUS = "logout_status"
+    FORCE_AUTHN = "force_authn"
+
+    VALID_FORCE_AUTHN_VALUES: ClassVar[frozenset[str]] = frozenset({"true", "false"})
 
     _SITE_WIDE_METADATA_CACHE_KEY: ClassVar[str | None] = None
     _sp_metadata_cache: ClassVar[dict[str | None, str | None]] = {}
@@ -273,21 +276,26 @@ def _redirect_with_error(self, redirect_uri, problem_detail):
         """
         return redirect(self._error_uri(redirect_uri, problem_detail))
 
-    def saml_authentication_redirect(self, params, db):
+    def saml_authentication_redirect(
+        self,
+        params: MultiDict[str, str],
+        db: sqlalchemy.orm.session.Session,
+    ) -> wkResponse | ProblemDetail:
         """Redirects an unauthenticated patron to the authentication URL of the
         appropriate SAML IdP.
         Over on that other site, the patron will authenticate and be
         redirected back to the circulation manager, ending up in
         saml_authentication_callback.
 
-        :param params: Query parameters
-        :type params: Dict
+        :param params: Query parameters. In addition to the required `provider`,
+            `idp_entity_id`, and `redirect_uri` parameters, an optional
+            `force_authn` parameter ("true" or "false") may be supplied to
+            request that the IdP re-authenticate the patron even if the patron
+            has an existing IdP session.
 
         :param db: Database session
-        :type db: sqlalchemy.orm.session.Session
 
-        :return: Redirection response
-        :rtype: Response
+        :return: Redirection response, or a ProblemDetail on error
         """
         provider_name = self._get_request_parameter(params, self.PROVIDER_NAME)
         if isinstance(provider_name, ProblemDetail):
@@ -301,6 +309,22 @@ def saml_authentication_redirect(self, params, db):
         if isinstance(redirect_uri, ProblemDetail):
             return redirect_uri
 
+        # Optional parameter. Clients pass force_authn=true to request that the
+        # IdP re-authenticate the patron even if the patron has an existing IdP
+        # session, e.g. a session for an account from a different tenant.
+        force_authn = params.get(self.FORCE_AUTHN)
+        if force_authn is not None and force_authn not in self.VALID_FORCE_AUTHN_VALUES:
+            return self._redirect_with_error(
+                redirect_uri,
+                SAML_INVALID_REQUEST.detailed(
+                    _(
+                        'Invalid %(name)s value; expected "true" or "false"',
+                        name=self.FORCE_AUTHN,
+                    )
+                ),
+            )
+        force_authn_requested = force_authn == "true"
+
         provider = self._authenticator.saml_provider_lookup(provider_name)
         if isinstance(provider, ProblemDetail):
             return self._redirect_with_error(redirect_uri, provider)
@@ -334,7 +358,7 @@ def saml_authentication_redirect(self, params, db):
             },
         )
         redirect_uri = authentication_manager.start_authentication(
-            db, idp_entity_id, relay_state
+            db, idp_entity_id, relay_state, force_authn=force_authn_requested
         )
         if isinstance(redirect_uri, ProblemDetail):
             return redirect_uri
@@ -425,7 +449,7 @@ def saml_authentication_callback(self, request, db):
 
     def saml_logout_redirect(
         self,
-        params: ImmutableMultiDict[str, str],
+        params: MultiDict[str, str],
         db: sqlalchemy.orm.session.Session,
     ) -> wkResponse | ProblemDetail:
         """Initiate SP-Initiated SAML SLO.

tests/manager/integration/patron_auth/saml/test_auth.py

@@ -94,13 +94,22 @@ class TestSAMLAuthenticationManager:
             ),
         ],
     )
+    @pytest.mark.parametrize(
+        "force_authn, expected_force_authn_attribute",
+        [
+            pytest.param(False, None, id="without-force-authn"),
+            pytest.param(True, "true", id="with-force-authn"),
+        ],
+    )
     def test_start_authentication(
         self,
         controller_fixture: ControllerFixture,
         create_mock_onelogin_configuration: Callable[..., SAMLOneLoginConfiguration],
-        service_provider,
-        identity_providers,
-    ):
+        service_provider: SAMLServiceProviderMetadata,
+        identity_providers: list[SAMLIdentityProviderMetadata],
+        force_authn: bool,
+        expected_force_authn_attribute: str | None,
+    ) -> None:
         onelogin_configuration = create_mock_onelogin_configuration(
             service_provider, identity_providers
         )
@@ -111,9 +120,13 @@ def test_start_authentication(
 
         with controller_fixture.app.test_request_context("/"):
             result = authentication_manager.start_authentication(
-                controller_fixture.db.session, saml_strings.IDP_1_ENTITY_ID, ""
+                controller_fixture.db.session,
+                saml_strings.IDP_1_ENTITY_ID,
+                "",
+                force_authn=force_authn,
             )
 
+            assert isinstance(result, str)
             query_items = parse_qs(urlsplit(result).query)
             saml_request = query_items["SAMLRequest"][0]
             decoded_saml_request = OneLogin_Saml2_Utils.decode_base64_and_inflate(
@@ -139,6 +152,8 @@ def test_start_authentication(
             sso_url = saml_request_dom.get("Destination")
             assert sso_url == IDENTITY_PROVIDERS[0].sso_service.url
 
+            assert saml_request_dom.get("ForceAuthn") == expected_force_authn_attribute
+
             name_id_policy_nodes = OneLogin_Saml2_XML.query(
                 saml_request_dom, "./samlp:NameIDPolicy"
             )

tests/manager/integration/patron_auth/saml/test_controller.py

@@ -7,6 +7,7 @@
 import pytest
 from flask import request
 from lxml import etree
+from werkzeug import Response as wkResponse
 
 from palace.manager.api.authentication.base import PatronData
 from palace.manager.api.authenticator import Authenticator
@@ -191,26 +192,22 @@ class TestSAMLController:
     def test_saml_authentication_redirect(
         self,
         controller_fixture: ControllerFixture,
-        provider_name,
-        idp_entity_id,
-        redirect_uri,
-        expected_problem,
-        expected_relay_state,
-    ):
+        provider_name: str | None,
+        idp_entity_id: str | None,
+        redirect_uri: str | None,
+        expected_problem: ProblemDetail | None,
+        expected_relay_state: str | None,
+    ) -> None:
         """Make sure that SAMLController.saml_authentication_redirect creates a correct RelayState or
         returns a correct ProblemDetail object in the case of any error.
 
         :param provider_name: Name of the authentication provider which should be passed as a request parameter
-        :type provider_name: str
 
         :param idp_entity_id: Identity Provider's ID which should be passed as a request parameter
-        :type idp_entity_id: str
 
         :param expected_problem: (Optional) Expected ProblemDetail object describing the error occurred (if any)
-        :type expected_problem: Optional[ProblemDetail]
 
         :param expected_relay_state: (Optional) String containing the expected RelayState value
-        :type expected_relay_state: Optional[str]
         """
         # Arrange
         expected_authentication_redirect_uri = "https://idp.circulationmanager.org"
@@ -262,6 +259,7 @@ def test_saml_authentication_redirect(
                 assert isinstance(result, ProblemDetail)
                 assert result.response == expected_problem.response
             else:
+                assert isinstance(result, wkResponse)
                 assert 302 == result.status_code
                 assert expected_authentication_redirect_uri == result.headers.get(
                     "Location"
@@ -271,6 +269,101 @@ def test_saml_authentication_redirect(
                     controller_fixture.db.session,
                     idp_entity_id,
                     expected_relay_state,
+                    force_authn=False,
+                )
+
+    @pytest.mark.parametrize(
+        "force_authn, expected_force_authn, expected_invalid",
+        [
+            pytest.param(None, False, False, id="without-force-authn"),
+            pytest.param("true", True, False, id="with-force-authn-true"),
+            pytest.param("false", False, False, id="with-force-authn-false"),
+            pytest.param("TRUE", None, True, id="with-invalid-force-authn-uppercase"),
+            pytest.param("1", None, True, id="with-invalid-force-authn-numeric"),
+            pytest.param("", None, True, id="with-invalid-force-authn-empty"),
+        ],
+    )
+    def test_saml_authentication_redirect_force_authn(
+        self,
+        controller_fixture: ControllerFixture,
+        force_authn: str | None,
+        expected_force_authn: bool | None,
+        expected_invalid: bool,
+    ) -> None:
+        """Make sure that the optional force_authn parameter is validated and
+        passed through to SAMLAuthenticationManager.start_authentication, and
+        that invalid values redirect back to the client with an error.
+        """
+        # Arrange
+        expected_authentication_redirect_uri = "https://idp.circulationmanager.org"
+        client_redirect_uri = "http://localhost"
+        authentication_manager = create_autospec(spec=SAMLAuthenticationManager)
+        authentication_manager.start_authentication = MagicMock(
+            return_value=expected_authentication_redirect_uri
+        )
+        provider = create_autospec(spec=SAMLWebSSOAuthenticationProvider)
+        provider.label = MagicMock(
+            return_value=SAMLWebSSOAuthenticationProvider.label()
+        )
+        provider.get_authentication_manager = MagicMock(
+            return_value=authentication_manager
+        )
+        provider.library = MagicMock(
+            return_value=controller_fixture.db.default_library()
+        )
+        authenticator = Authenticator(
+            controller_fixture.db.session,
+            controller_fixture.db.session.query(Library),
+        )
+
+        authenticator.library_authenticators["default"].register_saml_provider(provider)
+
+        controller = SAMLController(controller_fixture.app.manager, authenticator)
+        params = {
+            SAMLController.PROVIDER_NAME: SAMLWebSSOAuthenticationProvider.label(),
+            SAMLController.IDP_ENTITY_ID: IDENTITY_PROVIDERS[0].entity_id,
+            SAMLController.REDIRECT_URI: client_redirect_uri,
+        }
+
+        if force_authn is not None:
+            params[SAMLController.FORCE_AUTHN] = force_authn
+
+        query = urlencode(params)
+
+        with controller_fixture.app.test_request_context("/saml_authenticate?" + query):
+            request.library = controller_fixture.db.default_library()  # type: ignore[attr-defined]
+
+            # Act
+            result = controller.saml_authentication_redirect(
+                request.args, controller_fixture.db.session
+            )
+
+            # Assert
+            assert isinstance(result, wkResponse)
+            assert 302 == result.status_code
+            location = result.headers.get("Location")
+
+            if expected_invalid:
+                # The patron is redirected back to the client with the error
+                # encoded in the query string.
+                authentication_manager.start_authentication.assert_not_called()
+                assert location is not None
+                location_parse_result = urlsplit(location)
+                location_params = parse_qs(location_parse_result.query)
+                error = json.loads(location_params[SAMLController.ERROR][0])
+                assert error["type"] == SAML_INVALID_REQUEST.uri
+                # The static detail message must not reflect the submitted value.
+                assert (
+                    error["detail"]
+                    == 'Invalid force_authn value; expected "true" or "false"'
+                )
+            else:
+                assert expected_authentication_redirect_uri == location
+                assert (
+                    authentication_manager.start_authentication.call_args.kwargs[
+                        "force_authn"
+                    ]
+                    == expected_force_authn
                 )
 
     @pytest.mark.parametrize(