diff --git a/src/palace/manager/integration/patron_auth/saml/auth.py b/src/palace/manager/integration/patron_auth/saml/auth.py index 558ccc49b2..a694d7691e 100644 --- a/src/palace/manager/integration/patron_auth/saml/auth.py +++ b/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 diff --git a/src/palace/manager/integration/patron_auth/saml/controller.py b/src/palace/manager/integration/patron_auth/saml/controller.py index ad020d1502..163a85465c 100644 --- a/src/palace/manager/integration/patron_auth/saml/controller.py +++ b/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. diff --git a/tests/manager/integration/patron_auth/saml/test_auth.py b/tests/manager/integration/patron_auth/saml/test_auth.py index ab58e7b534..51dc29d739 100644 --- a/tests/manager/integration/patron_auth/saml/test_auth.py +++ b/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" ) diff --git a/tests/manager/integration/patron_auth/saml/test_controller.py b/tests/manager/integration/patron_auth/saml/test_controller.py index 6c44ccf7de..4d4a2a1126 100644 --- a/tests/manager/integration/patron_auth/saml/test_controller.py +++ b/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(