diff --git a/src/palace/manager/api/authenticator.py b/src/palace/manager/api/authenticator.py index 7cee0a19ba..154ec8f201 100644 --- a/src/palace/manager/api/authenticator.py +++ b/src/palace/manager/api/authenticator.py @@ -3,9 +3,9 @@ import enum import json import logging -from abc import ABC +from abc import ABC, abstractmethod from collections.abc import Iterable -from typing import Any, Self +from typing import TYPE_CHECKING, Any, Self import flask import jwt @@ -64,6 +64,16 @@ from palace.manager.util.opds_writer import OPDSFeed from palace.manager.util.problem_detail import ProblemDetail, ProblemDetailException +if TYPE_CHECKING: + from palace.manager.api.authentication.base import PatronData + from palace.manager.integration.patron_auth.oidc.auth import ( + OIDCAuthenticationManager, + ) + from palace.manager.integration.patron_auth.oidc.credential import ( + OIDCCredentialManager, + ) + from palace.manager.sqlalchemy.model.credential import Credential + class CirculationPatronProfileStorage(PatronProfileStorage): """A patron profile storage that can also provide short client tokens""" @@ -970,8 +980,49 @@ class BaseOIDCAuthenticationProvider[ OPDSAuthenticationFlow, ABC, ): - """Base class for OIDC authentication providers.""" + """Base class for OIDC authentication providers. + + The abstract members below are the contract OIDCController relies on. + Every provider registered through the OIDC dispatch path must supply + them. Keeping the contract here lets the controller work with any + OIDC-family provider without reaching into implementation details. + """ @property def flow_type(self) -> str: return "http://palaceproject.io/authtype/OpenIDConnect" + + @abstractmethod + def extract_patron_identifier(self, id_token_claims: dict[str, Any]) -> str | None: + """Extract the patron identifier from validated token claims. + + Must apply the same extraction used at login, so that logout + lookups match the stored authorization identifier. + + :param id_token_claims: Validated token claims + :return: Patron identifier, or None if it cannot be determined + """ + + @property + @abstractmethod + def credential_manager(self) -> OIDCCredentialManager: + """Credential manager storing this provider's tokens.""" + + @abstractmethod + def get_authentication_manager(self) -> OIDCAuthenticationManager: + """Return the authentication manager driving the OIDC protocol flow.""" + + @abstractmethod + def oidc_callback( + self, + db: Session, + id_token_claims: dict[str, Any], + access_token: str, + refresh_token: str | None = None, + expires_in: int | None = None, + id_token: str | None = None, + ) -> tuple[Credential, Patron, PatronData]: + """Authorize and persist a patron session after ID token validation. + + :return: 3-tuple (Credential, Patron, PatronData) + """ diff --git a/src/palace/manager/integration/patron_auth/oidc/auth.py b/src/palace/manager/integration/patron_auth/oidc/auth.py index 4cf632ebb0..c8d0315bfd 100644 --- a/src/palace/manager/integration/patron_auth/oidc/auth.py +++ b/src/palace/manager/integration/patron_auth/oidc/auth.py @@ -12,7 +12,8 @@ from __future__ import annotations import json -from typing import Any, cast +from collections.abc import Sequence +from typing import Any, Protocol, cast from urllib.parse import urlencode from pydantic import HttpUrl @@ -21,9 +22,6 @@ from palace.util.exceptions import BasePalaceException from palace.util.log import LoggerMixin -from palace.manager.integration.patron_auth.oidc.configuration.model import ( - OIDCAuthSettings, -) from palace.manager.integration.patron_auth.oidc.util import ( OIDCDiscoveryError, OIDCUtility, @@ -49,6 +47,62 @@ class OIDCRefreshTokenError(OIDCAuthenticationError): """Raised when token refresh fails.""" +class OIDCConnectionSettings(Protocol): + """The connection attributes OIDCAuthenticationManager reads from its settings. + + Read-only properties keep the contract structural. It is satisfied both by + the admin-configured OIDCAuthSettings model and by any plain object with + these attributes (for example, a provider-specific configuration with + fixed endpoints). + + Either ``issuer_url`` (discovery mode) or the individual endpoint fields + (manual mode) must be populated, matching the validation performed by + OIDCAuthSettings. + """ + + @property + def issuer_url(self) -> HttpUrl | None: ... + + @property + def issuer(self) -> str | None: ... + + @property + def authorization_endpoint(self) -> HttpUrl | None: ... + + @property + def token_endpoint(self) -> HttpUrl | None: ... + + @property + def jwks_uri(self) -> HttpUrl | None: ... + + @property + def userinfo_endpoint(self) -> HttpUrl | None: ... + + @property + def end_session_endpoint(self) -> HttpUrl | None: ... + + @property + def revocation_endpoint(self) -> HttpUrl | None: ... + + @property + def client_id(self) -> str: ... + + @property + def client_secret(self) -> str: ... + + @property + def scopes(self) -> Sequence[str]: ... + + @property + def use_pkce(self) -> bool: ... + + @property + def token_endpoint_auth_method(self) -> str: ... + + @property + def access_type(self) -> str: ... + + class OIDCAuthenticationManager(LoggerMixin): """Manages OIDC authentication flow. @@ -62,13 +116,13 @@ class OIDCAuthenticationManager(LoggerMixin): def __init__( self, - settings: OIDCAuthSettings, + settings: OIDCConnectionSettings, redis_client: Redis | None = None, secret_key: str | None = None, ): """Initialize OIDC authentication manager. - :param settings: OIDC authentication settings + :param settings: Connection settings for the OIDC provider :param redis_client: Optional Redis client for caching :param secret_key: Secret key for state generation (required for auth flow) """ @@ -79,6 +133,11 @@ def __init__( self._validator = OIDCTokenValidator() self._metadata: dict[str, Any] | None = None + @property + def use_pkce(self) -> bool: + """Whether the authorization flow should use PKCE.""" + return self._settings.use_pkce + def _extract_error_detail_from_response( self, exception: RequestNetworkException ) -> str: diff --git a/src/palace/manager/integration/patron_auth/oidc/configuration/model.py b/src/palace/manager/integration/patron_auth/oidc/configuration/model.py index fd9b17c553..e294d7a682 100644 --- a/src/palace/manager/integration/patron_auth/oidc/configuration/model.py +++ b/src/palace/manager/integration/patron_auth/oidc/configuration/model.py @@ -2,11 +2,13 @@ from __future__ import annotations +import logging from re import Pattern from typing import Annotated, Any from flask_babel import lazy_gettext as _ from pydantic import ( + AfterValidator, HttpUrl, PositiveInt, TypeAdapter, @@ -16,8 +18,6 @@ ) from pydantic_core.core_schema import ValidationInfo -from palace.util.log import LoggerMixin - from palace.manager.api.admin.problem_details import ( INCOMPLETE_CONFIGURATION, INVALID_CONFIGURATION_OPTION, @@ -27,7 +27,6 @@ AuthProviderSettings, ) from palace.manager.integration.settings import ( - BaseSettings, FormFieldType, FormMetadata, SettingsValidationError, @@ -50,8 +49,11 @@ ) -def _validate_filter_expression(cls: type[BaseSettings], v: str | None) -> str | None: - """Shared validator for OIDC filter expression fields on settings models. +def _check_filter_expression_syntax(v: str | None) -> str | None: + """Validate the syntax of a filter expression setting value. + + Attached to filter expression fields with ``AfterValidator`` so the + check travels with the field definition. :raises SettingsValidationError: if the expression is syntactically invalid. """ @@ -59,7 +61,7 @@ def _validate_filter_expression(cls: type[BaseSettings], v: str | None) -> str | try: FilterExpression(v).check_syntax() except FilterExpressionError as exception: - cls.logger().warning( + logging.getLogger(__name__).warning( f"Validation of the filter expression failed: {exception}" ) raise SettingsValidationError( @@ -70,7 +72,180 @@ def _validate_filter_expression(cls: type[BaseSettings], v: str | None) -> str | return v -class OIDCAuthSettings(AuthProviderSettings, LoggerMixin): +# Field definitions shared with other providers built on the OIDC machinery. +# Settings classes declare fields using these aliases, so field order (and +# therefore admin form order) is still determined by where each field is +# declared in its class. +ClientIdSetting = Annotated[ + str, + FormMetadata( + label=_("Client ID"), + description=_( + "OAuth 2.0 Client ID assigned by the OIDC provider during registration. " + "This is a public identifier for your application." + ), + ), +] + +ClientSecretSetting = Annotated[ + str, + FormMetadata( + label=_("Client Secret"), + description=_( + "OAuth 2.0 Client Secret assigned by the OIDC provider. " + "This is a confidential credential - keep it secure. " + "Used for authenticating token exchange requests." + ), + type=FormFieldType.TEXT, + ), +] + +SessionLifetimeSetting = Annotated[ + PositiveInt | None, + FormMetadata( + label=_("Session Lifetime (Days)"), + description=_( + "Override the OIDC provider's token lifetime with a custom session duration in days. " + "Leave empty to use the provider's token expiry. " + "Note: This only affects the Circulation Manager's session. " + "Protected content access is still governed by the OIDC provider's tokens." + ), + patron_auth_filter_context=True, + ), +] + +IntegrationFilterExpressionSetting = Annotated[ + str | None, + FormMetadata( + label="Filter Expression", + description=( + "A Python expression that may restrict a patron's access based on their" + " ID token claims and other settings." + " When present, it must evaluate to True in order for the patron to gain access." + " If a per-library Filter Expression is also configured, both must evaluate to True." + "
" + "
" + "The expression has access to: 'claims' (dict of ID token claims)," + " 'integration' (selected integration settings)," + " and 'library' (id, name, short_name)." + "
" + "
" + "Example — restrict to a specific email domain:" + "
" + "
claims['email'].endswith('@example.edu')
" + ), + type=FormFieldType.TEXTAREA, + use_monospace_font=True, + ), + AfterValidator(_check_filter_expression_syntax), +] + +LibraryFilterExpressionSetting = Annotated[ + str | None, + FormMetadata( + label="Filter Expression", + description=( + "A Python expression that may restrict a patron's access to this library" + " based on their ID token claims and other settings." + " When present, it must evaluate to True in order for the patron to gain" + " access to this library." + "
" + "
" + "Refer to the integration-level Filter Expression setting for expression" + " syntax and available context variables." + ), + type=FormFieldType.TEXTAREA, + use_monospace_font=True, + ), + AfterValidator(_check_filter_expression_syntax), +] + +# Note: the key-ordering caveat in the description below is a JSONB storage +# side effect. PostgreSQL sorts all object keys when storing/retrieving JSONB. +ExtraDataSetting = Annotated[ + dict[str, Any] | list[Any] | int | float | bool | str | None, + FormMetadata( + label="Extra Data", + description=( + "Extra data available to patron filter expressions via the" + " 'integration.extra_data' context variable." + " Accepts any JSON value (object, array, string, number, boolean, or null)." + "
" + "Note: Dictionary/object key order may not be preserved after saving." + ), + type=FormFieldType.JSON, + patron_auth_filter_context=True, + use_monospace_font=True, + ), +] + +AuthLinkDisplayNameSetting = Annotated[ + str | None, + FormMetadata( + label=_("Authorization Link: Display Name (Optional)"), + description=_( + "Human-readable name for this authentication provider shown to patrons. " + "If not provided, the integration name will be used. " + "Example: 'University Single Sign-On' or 'Library Login'" + ), + weight=1000, + ), +] + +AuthLinkDescriptionSetting = Annotated[ + str | None, + FormMetadata( + label=_("Authorization Link: Description (Optional)"), + description=_( + "Brief description of this authentication method shown to patrons. " + "If not provided, the display name will be used. " + "Example: 'Log in with your university credentials'" + ), + type=FormFieldType.TEXTAREA, + weight=1001, + ), +] + +AuthLinkLogoUrlSetting = Annotated[ + HttpUrl | None, + FormMetadata( + label=_("Authorization Link: Logo URL (Optional)"), + description=_( + "URL to a logo image representing this authentication provider. " + "Displayed in authentication selection screens. " + "Should be a publicly accessible HTTPS URL. " + "Recommended size: 64x64 pixels or larger." + ), + weight=1002, + ), +] + +AuthLinkInformationUrlSetting = Annotated[ + HttpUrl | None, + FormMetadata( + label=_("Authorization Link: Information URL (Optional)"), + description=_( + "URL to a page with more information about this authentication method. " + "Example: Help page, registration instructions, or provider information." + ), + weight=1003, + ), +] + +AuthLinkPrivacyStatementUrlSetting = Annotated[ + HttpUrl | None, + FormMetadata( + label=_("Authorization Link: Privacy Statement URL (Optional)"), + description=_( + "URL to the authentication provider's privacy policy or statement. " + "Helps patrons understand how their data is handled." + ), + weight=1004, + ), +] + + +class OIDCAuthSettings(AuthProviderSettings): """OIDC Authentication Provider Settings. Configures OpenID Connect (OIDC) authentication for patron authentication. @@ -212,29 +387,9 @@ class OIDCAuthSettings(AuthProviderSettings, LoggerMixin): ] = None # Client Configuration - client_id: Annotated[ - str, - FormMetadata( - label=_("Client ID"), - description=_( - "OAuth 2.0 Client ID assigned by the OIDC provider during registration. " - "This is a public identifier for your application." - ), - ), - ] + client_id: ClientIdSetting - client_secret: Annotated[ - str, - FormMetadata( - label=_("Client Secret"), - description=_( - "OAuth 2.0 Client Secret assigned by the OIDC provider. " - "This is a confidential credential - keep it secure. " - "Used for authenticating token exchange requests." - ), - type=FormFieldType.TEXT, - ), - ] + client_secret: ClientSecretSetting # Scopes & Claims scopes: Annotated[ @@ -282,63 +437,11 @@ class OIDCAuthSettings(AuthProviderSettings, LoggerMixin): ] = None # Session Configuration - session_lifetime: Annotated[ - PositiveInt | None, - FormMetadata( - label=_("Session Lifetime (Days)"), - description=_( - "Override the OIDC provider's token lifetime with a custom session duration in days. " - "Leave empty to use the provider's token expiry. " - "Note: This only affects the Circulation Manager's session. " - "Protected content access is still governed by the OIDC provider's tokens." - ), - patron_auth_filter_context=True, - ), - ] = None + session_lifetime: SessionLifetimeSetting = None - filter_expression: Annotated[ - str | None, - FormMetadata( - label="Filter Expression", - description=( - "A Python expression that may restrict a patron's access based on their" - " ID token claims and other settings." - " When present, it must evaluate to True in order for the patron to gain access." - " If a per-library Filter Expression is also configured, both must evaluate to True." - "
" - "
" - "The expression has access to: 'claims' (dict of ID token claims)," - " 'integration' (selected integration settings)," - " and 'library' (id, name, short_name)." - "
" - "
" - "Example — restrict to a specific email domain:" - "
" - "
claims['email'].endswith('@example.edu')
" - ), - type=FormFieldType.TEXTAREA, - use_monospace_font=True, - ), - ] = None + filter_expression: IntegrationFilterExpressionSetting = None - # Note: the key-ordering caveat in the description below is a JSONB storage - # side effect — PostgreSQL sorts all object keys when storing/retrieving JSONB. - extra_data: Annotated[ - dict[str, Any] | list[Any] | int | float | bool | str | None, - FormMetadata( - label="Extra Data", - description=( - "Extra data available to patron filter expressions via the" - " 'integration.extra_data' context variable." - " Accepts any JSON value (object, array, string, number, boolean, or null)." - "
" - "Note: Dictionary/object key order may not be preserved after saving." - ), - type=FormFieldType.JSON, - patron_auth_filter_context=True, - use_monospace_font=True, - ), - ] = None + extra_data: ExtraDataSetting = None # Advanced Options use_pkce: Annotated[ @@ -395,70 +498,15 @@ class OIDCAuthSettings(AuthProviderSettings, LoggerMixin): ] = "offline" # Authentication Link Settings - auth_link_display_name: Annotated[ - str | None, - FormMetadata( - label=_("Authorization Link: Display Name (Optional)"), - description=_( - "Human-readable name for this authentication provider shown to patrons. " - "If not provided, the integration name will be used. " - "Example: 'University Single Sign-On' or 'Library Login'" - ), - weight=1000, - ), - ] = None + auth_link_display_name: AuthLinkDisplayNameSetting = None - auth_link_description: Annotated[ - str | None, - FormMetadata( - label=_("Authorization Link: Description (Optional)"), - description=_( - "Brief description of this authentication method shown to patrons. " - "If not provided, the display name will be used. " - "Example: 'Log in with your university credentials'" - ), - type=FormFieldType.TEXTAREA, - weight=1001, - ), - ] = None + auth_link_description: AuthLinkDescriptionSetting = None - auth_link_logo_url: Annotated[ - HttpUrl | None, - FormMetadata( - label=_("Authorization Link: Logo URL (Optional)"), - description=_( - "URL to a logo image representing this authentication provider. " - "Displayed in authentication selection screens. " - "Should be a publicly accessible HTTPS URL. " - "Recommended size: 64x64 pixels or larger." - ), - weight=1002, - ), - ] = None + auth_link_logo_url: AuthLinkLogoUrlSetting = None - auth_link_information_url: Annotated[ - HttpUrl | None, - FormMetadata( - label=_("Authorization Link: Information URL (Optional)"), - description=_( - "URL to a page with more information about this authentication method. " - "Example: Help page, registration instructions, or provider information." - ), - weight=1003, - ), - ] = None + auth_link_information_url: AuthLinkInformationUrlSetting = None - auth_link_privacy_statement_url: Annotated[ - HttpUrl | None, - FormMetadata( - label=_("Authorization Link: Privacy Statement URL (Optional)"), - description=_( - "URL to the authentication provider's privacy policy or statement. " - "Helps patrons understand how their data is handled." - ), - weight=1004, - ), - ] = None + auth_link_privacy_statement_url: AuthLinkPrivacyStatementUrlSetting = None @model_validator(mode="after") def validate_configuration_mode(self) -> OIDCAuthSettings: @@ -549,11 +597,6 @@ def validate_scopes(cls, v: list[str]) -> list[str]: ) return v - @field_validator("filter_expression") - @classmethod - def validate_filter_expression(cls, v: str | None) -> str | None: - return _validate_filter_expression(cls, v) - @field_validator("patron_id_regular_expression") @classmethod def validate_patron_id_regex(cls, v: Pattern[str] | None) -> Pattern[str] | None: @@ -572,26 +615,4 @@ def validate_patron_id_regex(cls, v: Pattern[str] | None) -> Pattern[str] | None class OIDCAuthLibrarySettings(AuthProviderLibrarySettings): """Per-library OIDC authentication settings.""" - filter_expression: Annotated[ - str | None, - FormMetadata( - label="Filter Expression", - description=( - "A Python expression that may restrict a patron's access to this library" - " based on their ID token claims and other settings." - " When present, it must evaluate to True in order for the patron to gain" - " access to this library." - "
" - "
" - "Refer to the integration-level Filter Expression setting for expression" - " syntax and available context variables." - ), - type=FormFieldType.TEXTAREA, - use_monospace_font=True, - ), - ] = None - - @field_validator("filter_expression") - @classmethod - def validate_filter_expression(cls, v: str | None) -> str | None: - return _validate_filter_expression(cls, v) + filter_expression: LibraryFilterExpressionSetting = None diff --git a/src/palace/manager/integration/patron_auth/oidc/controller.py b/src/palace/manager/integration/patron_auth/oidc/controller.py index a194f4caa7..6164770544 100644 --- a/src/palace/manager/integration/patron_auth/oidc/controller.py +++ b/src/palace/manager/integration/patron_auth/oidc/controller.py @@ -3,7 +3,7 @@ from __future__ import annotations import json -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any from urllib.parse import SplitResult, parse_qs, urlencode, urlsplit import jwt @@ -247,7 +247,7 @@ def oidc_authentication_redirect( } code_challenge = None - if provider._settings.use_pkce: + if authentication_manager.use_pkce: code_verifier, code_challenge = utility.generate_pkce() state_data["code_verifier"] = code_verifier @@ -436,7 +436,7 @@ def oidc_logout_initiate( if isinstance(provider, ProblemDetail): return provider - auth_manager = provider.get_authentication_manager() # type: ignore[attr-defined] + auth_manager = provider.get_authentication_manager() # The bearer token's payload IS the credential JSON — parse it directly. # This avoids a DB lookup that would fail after token refresh (the credential @@ -448,7 +448,7 @@ def oidc_logout_initiate( return OIDC_INVALID_REQUEST.detailed(_("Invalid credential data")) id_token_claims = token_data["id_token_claims"] - patron_identifier = id_token_claims.get(provider._settings.patron_id_claim) # type: ignore[attr-defined] + patron_identifier = provider.extract_patron_identifier(id_token_claims) # Best-effort: revoke access and refresh tokens to prevent silent re-authentication # at the IdP after local CM logout. revoke_token suppresses all errors internally. @@ -464,7 +464,7 @@ def oidc_logout_initiate( # Invalidate our local patron credentials. try: - credential_manager = provider._credential_manager # type: ignore[attr-defined] + credential_manager = provider.credential_manager patron = credential_manager.lookup_patron_by_identifier( db, patron_identifier, library.id ) @@ -474,7 +474,7 @@ def oidc_logout_initiate( self.log.info(f"Invalidated credentials for patron {patron_identifier}") else: self.log.warning(f"Patron not found for identifier {patron_identifier}") - except (SQLAlchemyError, AttributeError): + except SQLAlchemyError: self.log.exception("Failed to invalidate credentials") # If the provider does not support RP-Initiated Logout (only token revocation), @@ -604,7 +604,18 @@ def oidc_backchannel_logout( return "", 400 # We need to determine which provider sent this logout token. - # Try all configured OIDC providers until we find one that can validate the token. + # Every provider that validates it gets to invalidate its own + # library's sessions: an integration associated with several + # libraries appears as a sibling provider in each library's + # authenticator, and patron lookup is library-scoped, so stopping + # at the first match would log the patron out of only one library. + # Sibling providers share one integration's settings, so once one + # of them validates the token the claims are reused. Only + # successes are cached: a transient failure for one library's + # provider must not prevent a sibling from validating. + claims_by_integration: dict[int, dict[str, Any]] = {} + processed = False + failed = False for ( library_authenticator ) in self._authenticator.library_authenticators.values(): @@ -614,47 +625,63 @@ def oidc_backchannel_logout( if not isinstance(provider, BaseOIDCAuthenticationProvider): continue - try: - auth_manager = provider._authentication_manager_factory.create( # type: ignore[attr-defined] - provider._settings # type: ignore[attr-defined] - ) - - # Try to validate the logout token with this provider - claims = auth_manager.validate_logout_token(logout_token) + claims = claims_by_integration.get(provider.integration_id) + if claims is None: + try: + auth_manager = provider.get_authentication_manager() + claims = auth_manager.validate_logout_token(logout_token) + except Exception as e: + # This provider couldn't validate the token; try the next one. + self.log.debug( + f"Provider {provider.label()} could not validate logout token: {e}" + ) + continue + claims_by_integration[provider.integration_id] = claims - # Successfully validated - get patron identifier - patron_identifier = claims.get( - provider._settings.patron_id_claim # type: ignore[attr-defined] - ) - if not patron_identifier: - self.log.warning("Logout token missing patron identifier claim") - return "", 400 - - # Invalidate patron credentials - credential_manager = provider._credential_manager # type: ignore[attr-defined] - patron = credential_manager.lookup_patron_by_identifier( - db, patron_identifier, provider.library_id - ) + patron_identifier = provider.extract_patron_identifier(claims) + if not patron_identifier: + self.log.warning("Logout token missing patron identifier claim") + continue - if patron: - credential_manager.invalidate_patron_credentials(db, patron.id) - self.log.info( - f"Back-channel logout: invalidated credentials for patron {patron_identifier}" - ) - else: - self.log.warning( - f"Back-channel logout: patron not found for identifier {patron_identifier}" + # Invalidate patron credentials. Failures here must not be + # mistaken for validation failures: they mark the whole + # request failed so the IdP gets a 400 it can retry. The + # savepoint scopes the rollback to the failing provider, + # so completed invalidations for other libraries survive. + try: + with db.begin_nested(): + credential_manager = provider.credential_manager + patron = credential_manager.lookup_patron_by_identifier( + db, patron_identifier, provider.library_id ) - return "", 200 - - except Exception as e: - # This provider couldn't validate the token, try the next one - self.log.debug( - f"Provider {provider.label()} could not validate logout token: {e}" + if patron: + credential_manager.invalidate_patron_credentials( + db, patron.id + ) + self.log.info( + "Back-channel logout: invalidated credentials for patron " + f"{patron_identifier} in library {provider.library_id}" + ) + else: + self.log.warning( + "Back-channel logout: patron not found for identifier " + f"{patron_identifier} in library {provider.library_id}" + ) + + processed = True + except SQLAlchemyError: + self.log.exception( + f"Back-channel logout failed for provider {provider.label()} " + f"in library {provider.library_id}" ) - continue + failed = True + + if processed and not failed: + return "", 200 - # No provider could validate the logout token - self.log.error("No OIDC provider could validate the logout token") + if not claims_by_integration: + self.log.error("No OIDC provider could validate the logout token") + else: + self.log.error("Back-channel logout did not complete successfully") return "", 400 diff --git a/src/palace/manager/integration/patron_auth/oidc/credential.py b/src/palace/manager/integration/patron_auth/oidc/credential.py index 7d6e20fd4a..b1405fedf3 100644 --- a/src/palace/manager/integration/patron_auth/oidc/credential.py +++ b/src/palace/manager/integration/patron_auth/oidc/credential.py @@ -28,7 +28,7 @@ from palace.manager.sqlalchemy.model.credential import Credential from palace.manager.sqlalchemy.model.datasource import DataSource from palace.manager.sqlalchemy.model.patron import Patron -from palace.manager.sqlalchemy.util import get_one_or_create +from palace.manager.sqlalchemy.util import get_one, get_one_or_create class OIDCCredentialManager(LoggerMixin): @@ -43,15 +43,37 @@ class OIDCCredentialManager(LoggerMixin): TOKEN_TYPE = "OIDC token" TOKEN_DATA_SOURCE_NAME = "OIDC" + def __init__( + self, + token_type: str | None = None, + data_source_name: str | None = None, + ) -> None: + """Initialize the credential manager. + + The defaults suit the generic OIDC provider. Other providers built on + the same machinery can pass distinct values, or subclass and override + the class constants, so their credentials do not collide with OIDC + ones. + + :param token_type: Value stored as ``Credential.type``. + Defaults to the class's ``TOKEN_TYPE``. + :param data_source_name: Name of the ``DataSource`` owning the + credentials. Defaults to the class's ``TOKEN_DATA_SOURCE_NAME``. + """ + self._token_type = token_type if token_type is not None else self.TOKEN_TYPE + self._data_source_name = ( + data_source_name + if data_source_name is not None + else self.TOKEN_DATA_SOURCE_NAME + ) + def _get_token_data_source(self, db: Session) -> DataSource: """Get or create the data source for OIDC credentials. :param db: Database session :return: DataSource for OIDC credentials """ - datasource, _ = get_one_or_create( - db, DataSource, name=self.TOKEN_DATA_SOURCE_NAME - ) + datasource, _ = get_one_or_create(db, DataSource, name=self._data_source_name) return datasource @staticmethod @@ -171,7 +193,7 @@ def create_oidc_token( ) oidc_credential, is_new = Credential.temporary_token_create( - db, data_source, self.TOKEN_TYPE, patron, session_lifetime, token_value + db, data_source, self._token_type, patron, session_lifetime, token_value ) return oidc_credential @@ -189,8 +211,8 @@ def lookup_oidc_token_by_patron( credential = Credential.lookup_by_patron( db, - self.TOKEN_DATA_SOURCE_NAME, - self.TOKEN_TYPE, + self._data_source_name, + self._token_type, patron, allow_persistent_token=False, auto_create_datasource=True, @@ -227,7 +249,7 @@ def lookup_oidc_token_by_value( credential = Credential.lookup_by_token( db, self._get_token_data_source(db), - self.TOKEN_TYPE, + self._token_type, token_value, constraint=credential_constraint, ) @@ -385,17 +407,28 @@ def invalidate_credential(self, db: Session, credential_id: int) -> None: self.log.info(f"Invalidated credential {credential_id}") def invalidate_patron_credentials(self, db: Session, patron_id: int) -> int: - """Invalidate all OIDC credentials for a patron. + """Invalidate all of a patron's credentials managed by this manager. + + Only credentials matching this manager's data source and token type + are affected, so providers with distinct credential namespaces can + be invalidated independently. :param db: Database session :param patron_id: Patron ID :return: Number of credentials invalidated """ + # Plain lookup rather than _get_token_data_source: invalidation must + # not create the data source when nothing was ever stored under it. + data_source = get_one(db, DataSource, name=self._data_source_name) + if data_source is None: + return 0 + credentials = ( db.query(Credential) .filter( Credential.patron_id == patron_id, - Credential.type == self.TOKEN_TYPE, + Credential.data_source == data_source, + Credential.type == self._token_type, ) .all() ) diff --git a/src/palace/manager/integration/patron_auth/oidc/provider.py b/src/palace/manager/integration/patron_auth/oidc/provider.py index 767cc09710..d23530fe64 100644 --- a/src/palace/manager/integration/patron_auth/oidc/provider.py +++ b/src/palace/manager/integration/patron_auth/oidc/provider.py @@ -5,11 +5,12 @@ from __future__ import annotations -from collections.abc import Generator -from typing import TYPE_CHECKING, Any +from collections.abc import Generator, Sequence +from typing import TYPE_CHECKING, Any, Protocol from flask import url_for from flask_babel import lazy_gettext as _ +from pydantic import HttpUrl from sqlalchemy.orm import Session from werkzeug.datastructures import Authorization @@ -17,6 +18,7 @@ from palace.opds.authentication.palace import LocalizedValue from palace.opds.rwpm import Link from palace.util.exceptions import PalaceValueError +from palace.util.log import LoggerType from palace.manager.api.authentication.base import PatronData, PatronLookupNotSupported from palace.manager.api.authenticator import BaseOIDCAuthenticationProvider @@ -37,6 +39,7 @@ ) from palace.manager.service.analytics.analytics import Analytics from palace.manager.sqlalchemy.model.credential import Credential +from palace.manager.sqlalchemy.model.library import Library from palace.manager.sqlalchemy.model.patron import Patron from palace.manager.util.filter import FilterExpression, FilterExpressionError from palace.manager.util.problem_detail import ( @@ -88,6 +91,184 @@ ) +def evaluate_patron_filters( + expressions: Sequence[tuple[str, str]], + context: dict[str, Any], + *, + library: Library, + claim_names: Sequence[str], + log: LoggerType, +) -> None: + """Evaluate labeled filter expressions as an authorization check. + + Every expression must evaluate to True for access to be granted. All + expressions are always evaluated so that a single log entry can report + every failing level at once. Callers assemble the evaluation context, + which lets providers expose whatever data their filters operate on. + + :param expressions: (label, expression) pairs to evaluate + :param context: Evaluation context passed to each expression + :param library: Library the patron is authenticating against (for logging) + :param claim_names: Claim names included in denial log entries + :param log: Logger to record evaluation results with + :raises ProblemDetailException: if the patron is denied access or an expression errors + """ + failing: list[str] = [] + eval_errors: list[FilterExpressionError] = [] + for label, expression in expressions: + try: + result = FilterExpression(expression).evaluate(context) + except FilterExpressionError as exc: + log.error( + "Filter [%s] evaluation error for library %s (%s): %s", + label, + library.name, + library.short_name, + exc, + ) + failing.append(label) + eval_errors.append(exc) + continue + log.info( + "Filter [%s] %s for library %s (%s)", + label, + "passed" if result else "failed", + library.name, + library.short_name, + ) + if not result: + failing.append(label) + + if failing: + log.warning( + "Access denied for library %s (%s): filter(s) failed: %s; claim names=%s", + library.name, + library.short_name, + ", ".join(failing), + claim_names, + ) + if eval_errors: + raise ProblemDetailException( + problem_detail=OIDC_FILTER_EVALUATION_ERROR.detailed( + "; ".join(str(e) for e in eval_errors) + ) + ) from eval_errors[0] + raise ProblemDetailException(problem_detail=OIDC_NO_ACCESS_ERROR) + + +class OIDCAuthLinkSettings(Protocol): + """Display attributes for the authenticate link in an authentication document.""" + + @property + def auth_link_display_name(self) -> str | None: ... + + @property + def auth_link_description(self) -> str | None: ... + + @property + def auth_link_logo_url(self) -> HttpUrl | None: ... + + @property + def auth_link_information_url(self) -> HttpUrl | None: ... + + @property + def auth_link_privacy_statement_url(self) -> HttpUrl | None: ... + + +def build_authenticate_link( + settings: OIDCAuthLinkSettings, label: str, authenticate_url: str +) -> AuthenticateLink: + """Build an authentication link for an authentication entry. + + :param settings: Display attributes for the link + :param label: Provider label, used when no display name is configured + :param authenticate_url: URL the client follows to start authentication + """ + display_name = settings.auth_link_display_name or label + description = settings.auth_link_description or display_name + + information_urls: list[LocalizedValue] = [] + if settings.auth_link_information_url: + information_urls = [ + LocalizedValue(value=str(settings.auth_link_information_url), language="en") + ] + privacy_statement_urls: list[LocalizedValue] = [] + if settings.auth_link_privacy_statement_url: + privacy_statement_urls = [ + LocalizedValue( + value=str(settings.auth_link_privacy_statement_url), + language="en", + ) + ] + logo_urls: list[LocalizedValue] = [] + if settings.auth_link_logo_url: + logo_urls = [ + LocalizedValue(value=str(settings.auth_link_logo_url), language="en") + ] + + return AuthenticateLink( + rel="authenticate", + href=authenticate_url, + display_names=[LocalizedValue(value=display_name, language="en")], + descriptions=[LocalizedValue(value=description, language="en")], + information_urls=information_urls, + privacy_statement_urls=privacy_statement_urls, + logo_urls=logo_urls, + ) + + +def build_oidc_authentication_document( + settings: OIDCAuthLinkSettings, + *, + flow_type: str, + label: str, + library_short_name: str, + supports_logout: bool, +) -> PalaceAuthentication: + """Build an `authentication` entry suitable for an authentication document. + + The authenticate and logout links point at the shared OIDC routes, which + dispatch to the right provider by its label. Any provider registered on + the OIDC path can therefore use this builder, passing its own flow type + and label and omitting the logout link when logout is unsupported. + + :param settings: Display attributes for the authenticate link + :param flow_type: Authentication flow type URI advertised to clients + :param label: Provider label, used for route dispatch and as the description + :param library_short_name: Library the entry is generated for + :param supports_logout: Whether to include a templated logout link + :return: Authentication entry + """ + authenticate_url = url_for( + "oidc_authenticate", + _external=True, + library_short_name=library_short_name, + provider=label, + ) + links: list[Link] = [build_authenticate_link(settings, label, authenticate_url)] + + if supports_logout: + logout_url = url_for( + "oidc_logout", + _external=True, + library_short_name=library_short_name, + provider=label, + ) + links.append( + Link( + rel="logout", + href=f"{logout_url}{{&{LOGOUT_REDIRECT_QUERY_PARAM}}}", + templated=True, + ) + ) + + return PalaceAuthentication( + type=flow_type, + description=label, + links=links, + ) + + class OIDCAuthenticationProvider( BaseOIDCAuthenticationProvider[OIDCAuthSettings, OIDCAuthLibrarySettings] ): @@ -136,6 +317,16 @@ def identifies_individuals(self) -> bool: """Indicate whether this provider identifies individual patrons.""" return True + @property + def patron_id_claim(self) -> str: + """Name of the ID token claim used to identify the patron.""" + return self._settings.patron_id_claim + + @property + def credential_manager(self) -> OIDCCredentialManager: + """Credential manager storing this provider's tokens.""" + return self._credential_manager + @classmethod def settings_class(cls) -> type[OIDCAuthSettings]: """Return the settings class for this provider.""" @@ -158,44 +349,6 @@ def get_credential_from_header(self, auth: Authorization) -> str | None: return auth.token return None - def _create_authentication_link(self, authenticate_url: str) -> AuthenticateLink: - """Build an authentication link for an authentication entry.""" - display_name = self._settings.auth_link_display_name or self.label() - description = self._settings.auth_link_description or display_name - - information_urls: list[LocalizedValue] = [] - if self._settings.auth_link_information_url: - information_urls = [ - LocalizedValue( - value=str(self._settings.auth_link_information_url), language="en" - ) - ] - privacy_statement_urls: list[LocalizedValue] = [] - if self._settings.auth_link_privacy_statement_url: - privacy_statement_urls = [ - LocalizedValue( - value=str(self._settings.auth_link_privacy_statement_url), - language="en", - ) - ] - logo_urls: list[LocalizedValue] = [] - if self._settings.auth_link_logo_url: - logo_urls = [ - LocalizedValue( - value=str(self._settings.auth_link_logo_url), language="en" - ) - ] - - return AuthenticateLink( - rel="authenticate", - href=authenticate_url, - display_names=[LocalizedValue(value=display_name, language="en")], - descriptions=[LocalizedValue(value=description, language="en")], - information_urls=information_urls, - privacy_statement_urls=privacy_statement_urls, - logo_urls=logo_urls, - ) - def _authentication_flow_document(self, db: Session) -> PalaceAuthentication: """Build an `authentication` entry suitable for an authentication document. @@ -206,34 +359,12 @@ def _authentication_flow_document(self, db: Session) -> PalaceAuthentication: if not library: raise PalaceValueError("Library not found") - authenticate_url = url_for( - "oidc_authenticate", - _external=True, + return build_oidc_authentication_document( + self._settings, + flow_type=self.flow_type, + label=self.label(), library_short_name=library.short_name, - provider=self.label(), - ) - links: list[Link] = [self._create_authentication_link(authenticate_url)] - - auth_manager = self.get_authentication_manager() - if auth_manager.supports_logout(): - logout_url = url_for( - "oidc_logout", - _external=True, - library_short_name=library.short_name, - provider=self.label(), - ) - links.append( - Link( - rel="logout", - href=f"{logout_url}{{&{LOGOUT_REDIRECT_QUERY_PARAM}}}", - templated=True, - ) - ) - - return PalaceAuthentication( - type=self.flow_type, - description=self.label(), - links=links, + supports_logout=self.get_authentication_manager().supports_logout(), ) def _run_self_tests(self, db: Session) -> Generator[SelfTestResult]: @@ -301,16 +432,17 @@ def get_authentication_manager(self) -> OIDCAuthenticationManager: self._auth_manager = manager return self._auth_manager - def remote_patron_lookup_from_oidc_claims( - self, id_token_claims: dict[str, Any] - ) -> PatronData: - """Create PatronData from ID token claims. + def extract_patron_identifier(self, id_token_claims: dict[str, Any]) -> str | None: + """Extract the patron identifier from validated token claims. - :param id_token_claims: Validated ID token claims - :return: PatronData object - :raises: ProblemDetailException if patron cannot be determined + Applies the configured patron ID claim and optional extraction + regular expression. Logout paths use this so their patron lookups + match the identifier stored at login. + + :param id_token_claims: Validated token claims + :return: Patron identifier, or None if it cannot be determined """ - patron_id_claim = self._settings.patron_id_claim + patron_id_claim = self.patron_id_claim id_token_claim_names = list(id_token_claims.keys()) raw_patron_id = id_token_claims.get(patron_id_claim) @@ -320,7 +452,7 @@ def remote_patron_lookup_from_oidc_claims( patron_id_claim, id_token_claim_names, ) - raise ProblemDetailException(problem_detail=OIDC_CANNOT_DETERMINE_PATRON) + return None if self._settings.patron_id_regular_expression: match = self._settings.patron_id_regular_expression.match( @@ -333,9 +465,7 @@ def remote_patron_lookup_from_oidc_claims( raw_patron_id, patron_id_claim, ) - raise ProblemDetailException( - problem_detail=OIDC_CANNOT_DETERMINE_PATRON - ) + return None patron_id = match.group("patron_id") else: patron_id = str(raw_patron_id) @@ -346,6 +476,21 @@ def remote_patron_lookup_from_oidc_claims( patron_id_claim, id_token_claim_names, ) + return patron_id + + def remote_patron_lookup_from_oidc_claims( + self, id_token_claims: dict[str, Any] + ) -> PatronData: + """Create PatronData from ID token claims. + + :param id_token_claims: Validated ID token claims + :return: PatronData object + :raises: ProblemDetailException if patron cannot be determined + """ + patron_id = self.extract_patron_identifier(id_token_claims) + if patron_id is None: + raise ProblemDetailException(problem_detail=OIDC_CANNOT_DETERMINE_PATRON) + return PatronData( permanent_id=patron_id, authorization_identifier=patron_id, @@ -415,47 +560,13 @@ def _filter_claims(self, db: Session, id_token_claims: dict[str, Any]) -> None: }, } - failing: list[str] = [] - eval_errors: list[FilterExpressionError] = [] - for label, expression in labeled_expressions: - try: - result = FilterExpression(expression).evaluate(context) - except FilterExpressionError as exc: - self.log.error( - "Filter [%s] evaluation error for library %s (%s): %s", - label, - library.name, - library.short_name, - exc, - ) - failing.append(label) - eval_errors.append(exc) - continue - self.log.info( - "Filter [%s] %s for library %s (%s)", - label, - "passed" if result else "failed", - library.name, - library.short_name, - ) - if not result: - failing.append(label) - - if failing: - self.log.warning( - "Access denied for library %s (%s): filter(s) failed: %s; claim names=%s", - library.name, - library.short_name, - ", ".join(failing), - id_token_claim_names, - ) - if eval_errors: - raise ProblemDetailException( - problem_detail=OIDC_FILTER_EVALUATION_ERROR.detailed( - "; ".join(str(e) for e in eval_errors) - ) - ) from eval_errors[0] - raise ProblemDetailException(problem_detail=OIDC_NO_ACCESS_ERROR) + evaluate_patron_filters( + labeled_expressions, + context, + library=library, + claim_names=id_token_claim_names, + log=self.log, + ) def oidc_callback( self, diff --git a/src/palace/manager/integration/patron_auth/oidc/util.py b/src/palace/manager/integration/patron_auth/oidc/util.py index 7de5a532a9..7f34152019 100644 --- a/src/palace/manager/integration/patron_auth/oidc/util.py +++ b/src/palace/manager/integration/patron_auth/oidc/util.py @@ -47,7 +47,6 @@ class OIDCUtility(LoggerMixin): # Cache TTLs DISCOVERY_CACHE_TTL = 24 * 60 * 60 # 24 hours JWKS_CACHE_TTL = 24 * 60 * 60 # 24 hours - PKCE_CACHE_TTL = 10 * 60 # 10 minutes STATE_MAX_AGE = 10 * 60 # 10 minutes LOGOUT_STATE_CACHE_TTL = 10 * 60 # 10 minutes LOGOUT_STATE_MAX_AGE = 10 * 60 # 10 minutes @@ -55,7 +54,6 @@ class OIDCUtility(LoggerMixin): # Cache key prefixes DISCOVERY_KEY_PREFIX = "oidc:discovery:" JWKS_KEY_PREFIX = "oidc:jwks:" - PKCE_KEY_PREFIX = "oidc:pkce:" LOGOUT_STATE_KEY_PREFIX = "oidc:logout_state:" def __init__(self, redis_client: Redis | None = None): @@ -107,11 +105,11 @@ def _retrieve_token_data( ) -> dict[str, Any] | None: """Retrieve token-related data from Redis cache. - Generic method for retrieving PKCE, logout state, or other token data. + Generic method for retrieving logout state or other token data. - :param key_prefix: Cache key prefix (e.g., PKCE_KEY_PREFIX) + :param key_prefix: Cache key prefix (e.g., LOGOUT_STATE_KEY_PREFIX) :param state_token: State token used as cache key - :param data_type: Description for logging (e.g., "PKCE", "logout state") + :param data_type: Description for logging (e.g., "logout state") :param delete: Whether to delete the entry after retrieval (one-time use) :return: Dictionary with token data, or None if not found :raises OIDCUtilityError: If Redis client is not available @@ -399,45 +397,6 @@ def fetch_jwks(self, jwks_uri: HttpUrl, use_cache: bool = True) -> dict[str, Any return jwks - def store_pkce( - self, - state_token: str, - code_verifier: str, - metadata: dict[str, Any] | None = None, - ) -> None: - """Store PKCE code_verifier in Redis cache. - - :param state_token: State token to use as cache key - :param code_verifier: PKCE code verifier to store - :param metadata: Optional additional metadata to store - """ - if not self._redis: - raise OIDCUtilityError("Redis client required for PKCE storage") - - data = { - "code_verifier": code_verifier, - "timestamp": int(time.time()), - } - if metadata: - data.update(metadata) - - cache_key = self._redis.get_key(self.PKCE_KEY_PREFIX + state_token) - self._redis.set(cache_key, json.dumps(data), ex=self.PKCE_CACHE_TTL) - self.log.debug(f"Stored PKCE for state: {state_token[:16]}...") - - def retrieve_pkce( - self, state_token: str, delete: bool = True - ) -> dict[str, Any] | None: - """Retrieve PKCE code_verifier from Redis cache. - - :param state_token: State token used as cache key - :param delete: Whether to delete the entry after retrieval (one-time use) - :return: Dictionary with code_verifier and metadata, or None if not found - """ - return self._retrieve_token_data( - self.PKCE_KEY_PREFIX, state_token, "PKCE", delete - ) - def store_logout_state( self, state_token: str, diff --git a/src/palace/manager/integration/patron_auth/oidc/validator.py b/src/palace/manager/integration/patron_auth/oidc/validator.py index e6942af554..505e315e38 100644 --- a/src/palace/manager/integration/patron_auth/oidc/validator.py +++ b/src/palace/manager/integration/patron_auth/oidc/validator.py @@ -3,13 +3,11 @@ This module provides functionality for validating OIDC ID tokens including: - Signature verification using JWKS - Claims validation (issuer, audience, expiry, etc.) -- Patron ID extraction from claims """ from __future__ import annotations import time -from re import Pattern from typing import Any, cast from joserfc import jwt @@ -32,10 +30,6 @@ class OIDCTokenClaimsError(OIDCTokenValidationError): """Raised when ID token claims validation fails.""" -class OIDCPatronIDExtractionError(OIDCTokenValidationError): - """Raised when patron ID cannot be extracted from claims.""" - - class OIDCTokenValidator(LoggerMixin): """Validator for OIDC ID tokens.""" @@ -184,111 +178,3 @@ def validate_claims( raise OIDCTokenClaimsError(f"Invalid ID token claims: {error_msg}") self.log.debug("ID token claims validated successfully") - - def extract_patron_id( - self, - claims: dict[str, Any], - claim_name: str, - regex_pattern: Pattern[str] | None = None, - ) -> str: - """Extract patron ID from ID token claims. - - :param claims: Decoded ID token claims - :param claim_name: Name of claim containing patron ID - :param regex_pattern: Optional regex pattern to extract ID from claim value. - Must contain a named group 'patron_id'. - :raises OIDCPatronIDExtractionError: If patron ID cannot be extracted - :return: Extracted patron ID - """ - # Get claim value - claim_value = claims.get(claim_name) - if claim_value is None: - self.log.error(f"Claim '{claim_name}' not found in ID token") - available_claims = ", ".join(claims.keys()) - raise OIDCPatronIDExtractionError( - f"Patron ID claim '{claim_name}' not found in ID token. " - f"Available claims: {available_claims}" - ) - - # Convert to string if needed - if not isinstance(claim_value, str): - try: - claim_value = str(claim_value) - except Exception as e: - raise OIDCPatronIDExtractionError( - f"Cannot convert claim '{claim_name}' value to string: {claim_value}" - ) from e - - # Apply regex if provided - if regex_pattern: - match = regex_pattern.search(claim_value) - if not match: - self.log.error( - f"Regex pattern did not match claim value: '{claim_value}'" - ) - raise OIDCPatronIDExtractionError( - f"Patron ID regex pattern did not match claim '{claim_name}' value: '{claim_value}'" - ) - - try: - patron_id = match.group("patron_id") - except IndexError: - raise OIDCPatronIDExtractionError( - "Regex pattern must contain a named group 'patron_id'" - ) - - if not patron_id: - raise OIDCPatronIDExtractionError( - f"Regex pattern matched but 'patron_id' group is empty" - ) - - self.log.debug( - f"Extracted patron ID '{patron_id}' from claim '{claim_name}' " - "using regex pattern" - ) - return patron_id - - # No regex - use full claim value - if not claim_value: - raise OIDCPatronIDExtractionError( - f"Claim '{claim_name}' is empty or whitespace-only" - ) - - self.log.debug(f"Extracted patron ID '{claim_value}' from claim '{claim_name}'") - return claim_value - - def validate_and_extract( - self, - id_token: str, - jwks: dict[str, Any], - expected_issuer: str, - expected_audience: str, - patron_id_claim: str, - nonce: str | None = None, - patron_id_regex: Pattern[str] | None = None, - ) -> tuple[dict[str, Any], str]: - """Validate ID token and extract patron ID in one operation. - - Convenience method that combines signature validation, claims validation, - and patron ID extraction. - - :param id_token: Raw ID token (JWT) - :param jwks: JSON Web Key Set from provider - :param expected_issuer: Expected issuer URL - :param expected_audience: Expected audience (client_id) - :param patron_id_claim: Name of claim containing patron ID - :param nonce: Expected nonce value (if used) - :param patron_id_regex: Optional regex pattern for patron ID extraction - :raises OIDCTokenValidationError: If validation fails - :return: Tuple of (claims dict, patron_id) - """ - # Validate signature - claims = self.validate_signature(id_token, jwks) - - # Validate claims - self.validate_claims(claims, expected_issuer, expected_audience, nonce) - - # Extract patron ID - patron_id = self.extract_patron_id(claims, patron_id_claim, patron_id_regex) - - return claims, patron_id diff --git a/tests/manager/integration/patron_auth/oidc/configuration/test_model.py b/tests/manager/integration/patron_auth/oidc/configuration/test_model.py index 4ce185c63b..d12f615931 100644 --- a/tests/manager/integration/patron_auth/oidc/configuration/test_model.py +++ b/tests/manager/integration/patron_auth/oidc/configuration/test_model.py @@ -17,6 +17,7 @@ OIDCAuthSettings, ) from palace.manager.util.problem_detail import ProblemDetailException +from tests.fixtures.database import DatabaseTransactionFixture class TestOIDCAuthSettings: @@ -603,6 +604,46 @@ def test_extra_data_invalid_json(self) -> None: exc_info.value.problem_detail.detail or "" ) + def test_configuration_form_field_order( + self, db: DatabaseTransactionFixture + ) -> None: + """The admin form must present fields in this exact order. + + Field order is determined by declaration order within the settings + class, with the weighted authentication link fields sorted last. + This guards against reorderings from refactors of the settings + model, including moves of field definitions to module level. + """ + form = OIDCAuthSettings.configuration_form(db.session) + + assert [entry["key"] for entry in form] == [ + "test_mode", + "issuer_url", + "issuer", + "authorization_endpoint", + "token_endpoint", + "jwks_uri", + "userinfo_endpoint", + "end_session_endpoint", + "revocation_endpoint", + "client_id", + "client_secret", + "scopes", + "patron_id_claim", + "patron_id_regular_expression", + "session_lifetime", + "filter_expression", + "extra_data", + "use_pkce", + "token_endpoint_auth_method", + "access_type", + "auth_link_display_name", + "auth_link_description", + "auth_link_logo_url", + "auth_link_information_url", + "auth_link_privacy_statement_url", + ] + class TestOIDCAuthLibrarySettings: """Tests for OIDCAuthLibrarySettings.""" diff --git a/tests/manager/integration/patron_auth/oidc/test_auth.py b/tests/manager/integration/patron_auth/oidc/test_auth.py index 799609c40a..f21604767a 100644 --- a/tests/manager/integration/patron_auth/oidc/test_auth.py +++ b/tests/manager/integration/patron_auth/oidc/test_auth.py @@ -81,6 +81,20 @@ def test_init_without_secret_key(self, oidc_settings_with_discovery, redis_fixtu assert manager._secret_key is None + @pytest.mark.parametrize( + "use_pkce", + [ + pytest.param(True, id="pkce-enabled"), + pytest.param(False, id="pkce-disabled"), + ], + ) + def test_use_pkce_property(self, create_oidc_settings, use_pkce: bool) -> None: + """Test that use_pkce reflects the configured settings value.""" + settings = create_oidc_settings(use_pkce=use_pkce) + manager = OIDCAuthenticationManager(settings=settings) + + assert manager.use_pkce is use_pkce + class TestOIDCAuthenticationManagerMetadata: """Tests for provider metadata loading.""" diff --git a/tests/manager/integration/patron_auth/oidc/test_controller.py b/tests/manager/integration/patron_auth/oidc/test_controller.py index 437f3d9458..9a0995a735 100644 --- a/tests/manager/integration/patron_auth/oidc/test_controller.py +++ b/tests/manager/integration/patron_auth/oidc/test_controller.py @@ -2,16 +2,22 @@ import json import logging +from typing import Any from unittest.mock import ANY, MagicMock, Mock, patch import jwt import pytest from sqlalchemy.exc import SQLAlchemyError +from palace.util.datetime_helpers import utc_now + from palace.manager.api.authentication.base import PatronData from palace.manager.api.authenticator import BaseOIDCAuthenticationProvider from palace.manager.api.problem_details import LIBRARY_NOT_FOUND, UNKNOWN_OIDC_PROVIDER -from palace.manager.integration.patron_auth.oidc.auth import OIDCAuthenticationError +from palace.manager.integration.patron_auth.oidc.auth import ( + OIDCAuthenticationError, + OIDCAuthenticationManager, +) from palace.manager.integration.patron_auth.oidc.configuration.model import ( OIDCAuthLibrarySettings, OIDCAuthSettings, @@ -22,6 +28,9 @@ OIDC_INVALID_STATE, OIDCController, ) +from palace.manager.integration.patron_auth.oidc.credential import ( + OIDCCredentialManager, +) from palace.manager.integration.patron_auth.oidc.provider import ( OIDC_CANNOT_DETERMINE_PATRON, OIDCAuthenticationProvider, @@ -32,6 +41,47 @@ from tests.fixtures.database import DatabaseTransactionFixture +def create_mock_oidc_provider( + *, + patron_id_claim: str = "sub", + label: str | None = None, + integration_id: int = 1, + **attrs: Any, +) -> tuple[Mock, Mock]: + """Create a mock OIDC provider wired to a mock authentication manager. + + The provider is specced against BaseOIDCAuthenticationProvider, so + isinstance checks pass and reads of attributes outside the provider + contract fail. It satisfies the public contract the controller consumes: + ``extract_patron_identifier``, ``credential_manager``, and + ``get_authentication_manager``. Tests configure the returned + authentication manager and the provider's credential manager as needed. + + :param patron_id_claim: Name of the token claim the provider's + ``extract_patron_identifier`` reads the identifier from + :param label: Return value of the provider's label(), when given + :param integration_id: Integration the provider belongs to. Providers + sharing an integration share one logout token validation, so pass + distinct values when simulating distinct integrations. + :param attrs: Additional attributes to set on the provider + (e.g., library_id) + :return: 2-tuple (provider, authentication manager) + """ + provider = Mock(spec=BaseOIDCAuthenticationProvider) + provider.extract_patron_identifier.side_effect = lambda claims: claims.get( + patron_id_claim + ) + provider.integration_id = integration_id + provider.credential_manager = Mock(spec=OIDCCredentialManager) + if label is not None: + provider.label.return_value = label + for name, value in attrs.items(): + setattr(provider, name, value) + auth_manager = Mock(spec=OIDCAuthenticationManager) + provider.get_authentication_manager.return_value = auth_manager + return provider, auth_manager + + class TestOIDCController: @pytest.fixture def mock_circulation_manager(self): @@ -223,6 +273,7 @@ def test_oidc_authentication_redirect_success_without_pkce( } mock_auth_manager = MagicMock() + mock_auth_manager.use_pkce = False mock_auth_manager.build_authorization_url.return_value = "https://idp.example.com/authorize?client_id=test-client-id&response_type=code&state=test&nonce=test" controller._authenticator.oidc_provider_lookup.return_value = oidc_provider @@ -249,6 +300,10 @@ def test_oidc_authentication_redirect_success_without_pkce( assert "client_id=test-client-id" in result.location assert "response_type=code" in result.location + # The non-PKCE branch must not send a code challenge. + call_kwargs = mock_auth_manager.build_authorization_url.call_args.kwargs + assert call_kwargs["code_challenge"] is None + def test_oidc_authentication_redirect_success_with_pkce( self, db: DatabaseTransactionFixture, controller ): @@ -273,6 +328,7 @@ def test_oidc_authentication_redirect_success_with_pkce( } mock_auth_manager = MagicMock() + mock_auth_manager.use_pkce = True mock_auth_manager.build_authorization_url.return_value = "https://idp.example.com/authorize?code_challenge=test&code_challenge_method=S256" controller._authenticator.oidc_provider_lookup.return_value = oidc_provider @@ -298,6 +354,10 @@ def test_oidc_authentication_redirect_success_with_pkce( assert "code_challenge=" in result.location assert "code_challenge_method=S256" in result.location + # The PKCE branch must send a generated code challenge. + call_kwargs = mock_auth_manager.build_authorization_url.call_args.kwargs + assert call_kwargs["code_challenge"] is not None + @pytest.mark.parametrize( "prompt", [ @@ -793,13 +853,7 @@ def test_oidc_logout_initiate_success(self, logout_controller, db): mock_library_auth = Mock() mock_library_auth.bearer_token_signing_secret = "test-secret" - mock_provider = Mock() - mock_provider._settings = Mock() - mock_provider._settings.patron_id_claim = "sub" - mock_provider._credential_manager = Mock() - mock_provider.integration_id = 1 - - mock_auth_manager = Mock() + mock_provider, mock_auth_manager = create_mock_oidc_provider(integration_id=1) mock_auth_manager.validate_id_token_hint.return_value = { "sub": "user123@example.com" } @@ -809,14 +863,11 @@ def test_oidc_logout_initiate_success(self, logout_controller, db): "?id_token_hint=test.token" "&post_logout_redirect_uri=https://cm.test/oidc_logout_callback&state=test-state" ) - mock_provider.get_authentication_manager.return_value = mock_auth_manager - mock_patron = Mock() - mock_patron.id = patron.id - mock_provider._credential_manager.lookup_patron_by_identifier.return_value = ( - mock_patron + mock_provider.credential_manager.lookup_patron_by_identifier.return_value = ( + Mock(id=patron.id) ) - mock_provider._credential_manager.invalidate_patron_credentials.return_value = 1 + mock_provider.credential_manager.invalidate_patron_credentials.return_value = 1 mock_library_auth.oidc_provider_lookup.return_value = mock_provider mock_library_auth.decode_bearer_token.return_value = ( @@ -859,7 +910,7 @@ def test_oidc_logout_initiate_success(self, logout_controller, db): assert "https://oidc.provider.test/logout" in result.location # Verify credentials were invalidated - mock_provider._credential_manager.invalidate_patron_credentials.assert_called_once_with( + mock_provider.credential_manager.invalidate_patron_credentials.assert_called_once_with( db.session, patron.id ) # Verify both tokens were revoked @@ -895,23 +946,14 @@ def test_oidc_logout_initiate_revocation_only( mock_library_auth = Mock() mock_library_auth.bearer_token_signing_secret = "test-secret" - mock_provider = Mock() - mock_provider._settings = Mock() - mock_provider._settings.patron_id_claim = "sub" - mock_provider._credential_manager = Mock() - mock_provider.integration_id = 1 - - mock_auth_manager = Mock() + mock_provider, mock_auth_manager = create_mock_oidc_provider(integration_id=1) # Provider has revocation but NOT RP-Initiated Logout mock_auth_manager.supports_rp_initiated_logout.return_value = False - mock_provider.get_authentication_manager.return_value = mock_auth_manager - mock_patron = Mock() - mock_patron.id = patron.id - mock_provider._credential_manager.lookup_patron_by_identifier.return_value = ( - mock_patron + mock_provider.credential_manager.lookup_patron_by_identifier.return_value = ( + Mock(id=patron.id) ) - mock_provider._credential_manager.invalidate_patron_credentials.return_value = 1 + mock_provider.credential_manager.invalidate_patron_credentials.return_value = 1 mock_library_auth.oidc_provider_lookup.return_value = mock_provider mock_library_auth.decode_bearer_token.return_value = ( @@ -957,7 +999,7 @@ def test_oidc_logout_initiate_revocation_only( mock_auth_manager.revoke_token.assert_any_call("refresh-token") # Verify CM credentials were invalidated - mock_provider._credential_manager.invalidate_patron_credentials.assert_called_once_with( + mock_provider.credential_manager.invalidate_patron_credentials.assert_called_once_with( db.session, patron.id ) @@ -981,21 +1023,13 @@ def test_oidc_logout_initiate_no_stored_id_token( mock_library_auth = Mock() mock_library_auth.bearer_token_signing_secret = "test-secret" - mock_provider = Mock() - mock_provider._settings = Mock() - mock_provider._settings.patron_id_claim = "sub" - mock_provider._credential_manager = Mock() - - mock_auth_manager = Mock() + mock_provider, mock_auth_manager = create_mock_oidc_provider() mock_auth_manager.supports_rp_initiated_logout.return_value = True - mock_provider.get_authentication_manager.return_value = mock_auth_manager - mock_patron = Mock() - mock_patron.id = patron.id - mock_provider._credential_manager.lookup_patron_by_identifier.return_value = ( - mock_patron + mock_provider.credential_manager.lookup_patron_by_identifier.return_value = ( + Mock(id=patron.id) ) - mock_provider._credential_manager.invalidate_patron_credentials.return_value = 1 + mock_provider.credential_manager.invalidate_patron_credentials.return_value = 1 mock_library_auth.oidc_provider_lookup.return_value = mock_provider mock_library_auth.decode_bearer_token.return_value = ( @@ -1073,24 +1107,19 @@ def test_oidc_logout_initiate_uses_correct_library( mock_library_auth = Mock() mock_library_auth.bearer_token_signing_secret = f"secret{i+1}" - mock_provider = Mock() - mock_provider._settings = Mock() - mock_provider._settings.patron_id_claim = "sub" - mock_provider._credential_manager = Mock() - mock_provider.integration_id = i + 1 - - mock_auth_manager = Mock() + mock_provider, mock_auth_manager = create_mock_oidc_provider( + integration_id=i + 1 + ) mock_auth_manager.supports_rp_initiated_logout.return_value = True mock_auth_manager.build_logout_url.return_value = ( f"https://provider{i+1}.test/logout" ) mock_auth_managers.append(mock_auth_manager) - mock_provider.get_authentication_manager.return_value = mock_auth_manager - mock_provider._credential_manager.lookup_patron_by_identifier.return_value = Mock( + mock_provider.credential_manager.lookup_patron_by_identifier.return_value = Mock( id=patron.id ) - mock_provider._credential_manager.invalidate_patron_credentials.return_value = ( + mock_provider.credential_manager.invalidate_patron_credentials.return_value = ( 1 ) mock_library_auth.decode_bearer_token.return_value = ( @@ -1377,13 +1406,7 @@ def test_oidc_logout_initiate_credential_data_errors( "Test OIDC", provider_token, ) - mock_provider = Mock() - mock_provider._settings = Mock() - mock_provider._settings.patron_id_claim = "sub" - mock_provider._credential_manager = Mock() - - mock_auth_manager = Mock() - mock_provider.get_authentication_manager.return_value = mock_auth_manager + mock_provider, _ = create_mock_oidc_provider() mock_library_auth.oidc_provider_lookup.return_value = mock_provider logout_controller._authenticator.library_authenticators[library.short_name] = ( @@ -1425,12 +1448,7 @@ def test_oidc_logout_initiate_missing_patron_claim_revokes_token( "Test OIDC", json.dumps(token_data), ) - mock_provider = Mock() - mock_provider._settings = Mock() - mock_provider._settings.patron_id_claim = "sub" - - mock_auth_manager = Mock() - mock_provider.get_authentication_manager.return_value = mock_auth_manager + mock_provider, mock_auth_manager = create_mock_oidc_provider() mock_library_auth.oidc_provider_lookup.return_value = mock_provider logout_controller._authenticator.library_authenticators[library.short_name] = ( @@ -1515,29 +1533,25 @@ def test_oidc_logout_initiate_exceptions( "Test OIDC", json.dumps(token_data), ) - mock_provider = Mock() - mock_provider._settings = Mock() - mock_provider._settings.patron_id_claim = "sub" - mock_provider._credential_manager = Mock() + mock_provider, mock_auth_manager = create_mock_oidc_provider() if patron_found: - mock_provider._credential_manager.lookup_patron_by_identifier.return_value = Mock( + mock_provider.credential_manager.lookup_patron_by_identifier.return_value = Mock( id=patron.id ) if invalidation_error: - mock_provider._credential_manager.invalidate_patron_credentials.side_effect = ( + mock_provider.credential_manager.invalidate_patron_credentials.side_effect = ( invalidation_error ) else: - mock_provider._credential_manager.invalidate_patron_credentials.return_value = ( + mock_provider.credential_manager.invalidate_patron_credentials.return_value = ( 1 ) else: - mock_provider._credential_manager.lookup_patron_by_identifier.return_value = ( + mock_provider.credential_manager.lookup_patron_by_identifier.return_value = ( None ) - mock_auth_manager = Mock() mock_auth_manager.supports_rp_initiated_logout.return_value = bool( build_url_error ) @@ -1547,7 +1561,6 @@ def test_oidc_logout_initiate_exceptions( mock_auth_manager.build_logout_url.return_value = ( "https://oidc.provider.test/logout" ) - mock_provider.get_authentication_manager.return_value = mock_auth_manager mock_library_auth.oidc_provider_lookup.return_value = mock_provider logout_controller._authenticator.library_authenticators[library.short_name] = ( @@ -1579,7 +1592,7 @@ def test_oidc_logout_initiate_exceptions( if expected_status: assert result.status_code == expected_status if not patron_found: - mock_provider._credential_manager.invalidate_patron_credentials.assert_not_called() + mock_provider.credential_manager.invalidate_patron_credentials.assert_not_called() else: assert result.uri == expected_uri if build_url_error: @@ -1612,20 +1625,14 @@ def test_oidc_logout_initiate_credential_invalidation_is_nonfatal( } ), ) - mock_provider = Mock() - mock_provider._settings = Mock() - mock_provider._settings.patron_id_claim = "sub" - mock_provider._credential_manager = Mock() - mock_provider._credential_manager.lookup_patron_by_identifier.return_value = ( + mock_provider, mock_auth_manager = create_mock_oidc_provider() + mock_provider.credential_manager.lookup_patron_by_identifier.return_value = ( Mock(id=patron.id) ) - mock_provider._credential_manager.invalidate_patron_credentials.side_effect = ( + mock_provider.credential_manager.invalidate_patron_credentials.side_effect = ( SQLAlchemyError("DB error") ) - - mock_auth_manager = Mock() mock_auth_manager.supports_rp_initiated_logout.return_value = False - mock_provider.get_authentication_manager.return_value = mock_auth_manager mock_library_auth.oidc_provider_lookup.return_value = mock_provider logout_controller._authenticator.library_authenticators[library.short_name] = ( mock_library_auth @@ -1865,15 +1872,11 @@ def test_oidc_backchannel_logout_success(self, backchannel_controller, db): library = db.default_library() # Create mock provider with spec so isinstance checks work - mock_provider = Mock(spec=BaseOIDCAuthenticationProvider) - mock_provider.library_id = library.id - mock_provider._authentication_manager_factory = Mock() - mock_provider._settings = Mock() - mock_provider._settings.patron_id_claim = "sub" - mock_provider._credential_manager = Mock() - - # Mock auth manager that validates the logout token - mock_auth_manager = Mock() + mock_provider, mock_auth_manager = create_mock_oidc_provider( + label="Test OIDC", library_id=library.id + ) + + # Auth manager validates the logout token mock_auth_manager.validate_logout_token.return_value = { "sub": "user123@example.com", "iss": "https://oidc.provider.test", @@ -1882,18 +1885,12 @@ def test_oidc_backchannel_logout_success(self, backchannel_controller, db): "jti": "unique-token-id", "events": {"http://schemas.openid.net/event/backchannel-logout": {}}, } - mock_provider._authentication_manager_factory.create.return_value = ( - mock_auth_manager - ) - # Mock credential manager - mock_patron = Mock() - mock_patron.id = patron.id - mock_provider._credential_manager.lookup_patron_by_identifier.return_value = ( - mock_patron + # Credential manager finds the patron + mock_provider.credential_manager.lookup_patron_by_identifier.return_value = ( + Mock(id=patron.id) ) - mock_provider._credential_manager.invalidate_patron_credentials.return_value = 1 - mock_provider.label.return_value = "Test OIDC" + mock_provider.credential_manager.invalidate_patron_credentials.return_value = 1 # Set up library authenticator with the provider mock_library_auth = Mock() @@ -1914,7 +1911,7 @@ def test_oidc_backchannel_logout_success(self, backchannel_controller, db): mock_auth_manager.validate_logout_token.assert_called_once_with( "test.logout.token" ) - mock_provider._credential_manager.invalidate_patron_credentials.assert_called_once() + mock_provider.credential_manager.invalidate_patron_credentials.assert_called_once() def test_oidc_backchannel_logout_missing_token(self, backchannel_controller, db): """Test back-channel logout with missing logout token.""" @@ -1932,15 +1929,8 @@ def test_oidc_backchannel_logout_invalid_token(self, backchannel_controller, db) library = db.default_library() # Create mock provider that rejects the token - mock_provider = Mock() - mock_provider._authentication_manager_factory = Mock() - - mock_auth_manager = Mock() + mock_provider, mock_auth_manager = create_mock_oidc_provider(label="Test OIDC") mock_auth_manager.validate_logout_token.side_effect = Exception("Invalid token") - mock_provider._authentication_manager_factory.create.return_value = ( - mock_auth_manager - ) - mock_provider.label.return_value = "Test OIDC" # Set up library authenticator mock_library_auth = Mock() @@ -1963,14 +1953,9 @@ def test_oidc_backchannel_logout_patron_not_found(self, backchannel_controller, library = db.default_library() # Create mock provider with spec so isinstance checks work - mock_provider = Mock(spec=BaseOIDCAuthenticationProvider) - mock_provider.library_id = library.id - mock_provider._authentication_manager_factory = Mock() - mock_provider._settings = Mock() - mock_provider._settings.patron_id_claim = "sub" - mock_provider._credential_manager = Mock() - - mock_auth_manager = Mock() + mock_provider, mock_auth_manager = create_mock_oidc_provider( + label="Test OIDC", library_id=library.id + ) mock_auth_manager.validate_logout_token.return_value = { "sub": "nonexistent@example.com", "iss": "https://oidc.provider.test", @@ -1979,15 +1964,9 @@ def test_oidc_backchannel_logout_patron_not_found(self, backchannel_controller, "jti": "unique-token-id", "events": {"http://schemas.openid.net/event/backchannel-logout": {}}, } - mock_provider._authentication_manager_factory.create.return_value = ( - mock_auth_manager - ) # Patron not found - mock_provider._credential_manager.lookup_patron_by_identifier.return_value = ( - None - ) - mock_provider.label.return_value = "Test OIDC" + mock_provider.credential_manager.lookup_patron_by_identifier.return_value = None # Set up library authenticator mock_library_auth = Mock() @@ -2012,13 +1991,9 @@ def test_oidc_backchannel_logout_missing_patron_identifier_claim( """Test back-channel logout when logout token is missing patron identifier claim.""" library = db.default_library() - mock_provider = Mock(spec=BaseOIDCAuthenticationProvider) - mock_provider.library_id = library.id - mock_provider._authentication_manager_factory = Mock() - mock_provider._settings = Mock() - mock_provider._settings.patron_id_claim = "sub" - - mock_auth_manager = Mock() + mock_provider, mock_auth_manager = create_mock_oidc_provider( + label="Test OIDC", library_id=library.id + ) mock_auth_manager.validate_logout_token.return_value = { "iss": "https://oidc.provider.test", "aud": "test-client-id", @@ -2026,10 +2001,6 @@ def test_oidc_backchannel_logout_missing_patron_identifier_claim( "jti": "unique-token-id", "events": {"http://schemas.openid.net/event/backchannel-logout": {}}, } - mock_provider._authentication_manager_factory.create.return_value = ( - mock_auth_manager - ) - mock_provider.label.return_value = "Test OIDC" mock_library_auth = Mock() mock_library_auth.providers = [mock_provider] @@ -2052,28 +2023,21 @@ def test_oidc_backchannel_logout_no_provider_validates_token( """Test back-channel logout when no provider can validate the token.""" library = db.default_library() - # Create multiple OIDC providers that all reject the token - mock_provider1 = Mock(spec=BaseOIDCAuthenticationProvider) - mock_provider1._authentication_manager_factory = Mock() - mock_auth_manager1 = Mock() + # Create multiple OIDC providers, from distinct integrations, that + # all reject the token. + mock_provider1, mock_auth_manager1 = create_mock_oidc_provider( + label="Test OIDC 1", integration_id=1 + ) mock_auth_manager1.validate_logout_token.side_effect = Exception( "Provider 1 cannot validate" ) - mock_provider1._authentication_manager_factory.create.return_value = ( - mock_auth_manager1 - ) - mock_provider1.label.return_value = "Test OIDC 1" - mock_provider2 = Mock(spec=BaseOIDCAuthenticationProvider) - mock_provider2._authentication_manager_factory = Mock() - mock_auth_manager2 = Mock() + mock_provider2, mock_auth_manager2 = create_mock_oidc_provider( + label="Test OIDC 2", integration_id=2 + ) mock_auth_manager2.validate_logout_token.side_effect = Exception( "Provider 2 cannot validate" ) - mock_provider2._authentication_manager_factory.create.return_value = ( - mock_auth_manager2 - ) - mock_provider2.label.return_value = "Test OIDC 2" mock_library_auth = Mock() mock_library_auth.providers = [mock_provider1, mock_provider2] @@ -2089,3 +2053,237 @@ def test_oidc_backchannel_logout_no_provider_validates_token( assert status == 400 assert body == "" + + def test_oidc_backchannel_logout_invalidates_all_libraries( + self, backchannel_controller, db + ): + """A logout token must end the patron's sessions in every library. + + One integration associated with several libraries surfaces as a + provider in each library's authenticator, and each validates the + same token, so the sweep must not stop at the first match. + """ + claims = { + "sub": "user123@example.com", + "iss": "https://oidc.provider.test", + "aud": "test-client-id", + "iat": 1234567890, + "jti": "unique-token-id", + "events": {"http://schemas.openid.net/event/backchannel-logout": {}}, + } + + providers = [] + auth_managers = [] + for index, library_key in enumerate(["lib-a", "lib-b"]): + mock_provider, mock_auth_manager = create_mock_oidc_provider( + label="Test OIDC", library_id=index + 1 + ) + mock_auth_manager.validate_logout_token.return_value = claims + mock_provider.credential_manager.lookup_patron_by_identifier.return_value = Mock( + id=index + 1 + ) + providers.append(mock_provider) + auth_managers.append(mock_auth_manager) + + mock_library_auth = Mock() + mock_library_auth.providers = [mock_provider] + backchannel_controller._authenticator.library_authenticators[ + library_key + ] = mock_library_auth + + form_data = {"logout_token": "test.logout.token"} + + body, status = backchannel_controller.oidc_backchannel_logout( + form_data, db.session + ) + + assert status == 200 + assert body == "" + for mock_provider in providers: + mock_provider.credential_manager.invalidate_patron_credentials.assert_called_once() + + # Both providers share one integration, so the token is validated + # only once and the claims reused for the sibling. + auth_managers[0].validate_logout_token.assert_called_once() + auth_managers[1].validate_logout_token.assert_not_called() + + def test_oidc_backchannel_logout_partial_failure_returns_400( + self, backchannel_controller, db + ): + """A failure during any library's invalidation must produce a 400. + + The IdP treats 200 as final, so reporting success while one + library's sessions survive would leave them stranded. A 400 lets + the IdP retry the logout. Library A uses a real credential so the + test also proves the failing library's rollback does not discard + library A's completed invalidation. + """ + claims = { + "sub": "user123@example.com", + "iss": "https://oidc.provider.test", + "aud": "test-client-id", + "iat": 1234567890, + "jti": "unique-token-id", + "events": {"http://schemas.openid.net/event/backchannel-logout": {}}, + } + + # Library A: a real patron and credential behind a real manager. + patron = db.patron() + patron.authorization_identifier = "user123@example.com" + credential_manager = OIDCCredentialManager() + credential = credential_manager.create_oidc_token( + db.session, + patron, + claims, + "access-token", + expires_in=3600, + ) + db.session.flush() + assert credential.expires is not None + assert credential.expires > utc_now() + + provider_a, auth_manager_a = create_mock_oidc_provider( + label="Test OIDC", library_id=patron.library_id + ) + auth_manager_a.validate_logout_token.return_value = claims + provider_a.credential_manager = credential_manager + + provider_b, auth_manager_b = create_mock_oidc_provider( + label="Test OIDC", library_id=2 + ) + provider_b.credential_manager.lookup_patron_by_identifier.side_effect = ( + SQLAlchemyError("connection blip") + ) + + for library_key, mock_provider in [ + ("lib-a", provider_a), + ("lib-b", provider_b), + ]: + mock_library_auth = Mock() + mock_library_auth.providers = [mock_provider] + backchannel_controller._authenticator.library_authenticators[ + library_key + ] = mock_library_auth + + form_data = {"logout_token": "test.logout.token"} + + body, status = backchannel_controller.oidc_backchannel_logout( + form_data, db.session + ) + + assert status == 400 + assert body == "" + + # Library A's invalidation survived library B's failure. + assert credential.expires <= utc_now() + provider_b.credential_manager.invalidate_patron_credentials.assert_not_called() + # Provider B shares provider A's integration, so its claims come + # from the cache and its auth manager never validates the token. + auth_manager_b.validate_logout_token.assert_not_called() + + def test_oidc_backchannel_logout_sibling_recovers_after_transient_failure( + self, backchannel_controller, db + ): + """A transient validation failure must not poison sibling providers. + + Sibling providers of one integration validate independently, so + when the first library's provider fails transiently, the second + library's provider must still get its chance to validate and + complete its logout. + """ + claims = { + "sub": "user123@example.com", + "iss": "https://oidc.provider.test", + "aud": "test-client-id", + "iat": 1234567890, + "jti": "unique-token-id", + "events": {"http://schemas.openid.net/event/backchannel-logout": {}}, + } + + provider_a, auth_manager_a = create_mock_oidc_provider( + label="Test OIDC", library_id=1 + ) + auth_manager_a.validate_logout_token.side_effect = Exception( + "transient IdP blip" + ) + + provider_b, auth_manager_b = create_mock_oidc_provider( + label="Test OIDC", library_id=2 + ) + auth_manager_b.validate_logout_token.return_value = claims + provider_b.credential_manager.lookup_patron_by_identifier.return_value = Mock( + id=2 + ) + + for library_key, mock_provider in [ + ("lib-a", provider_a), + ("lib-b", provider_b), + ]: + mock_library_auth = Mock() + mock_library_auth.providers = [mock_provider] + backchannel_controller._authenticator.library_authenticators[ + library_key + ] = mock_library_auth + + form_data = {"logout_token": "test.logout.token"} + + body, status = backchannel_controller.oidc_backchannel_logout( + form_data, db.session + ) + + assert status == 200 + assert body == "" + auth_manager_a.validate_logout_token.assert_called_once() + auth_manager_b.validate_logout_token.assert_called_once() + provider_a.credential_manager.invalidate_patron_credentials.assert_not_called() + provider_b.credential_manager.invalidate_patron_credentials.assert_called_once() + + def test_oidc_backchannel_logout_second_integration_validates( + self, backchannel_controller, db + ): + """The sweep keeps trying distinct integrations after a rejection. + + The first integration rejects the token; the second validates it + and must still complete the logout. + """ + claims = { + "sub": "user123@example.com", + "iss": "https://oidc.provider.test", + "aud": "test-client-id", + "iat": 1234567890, + "jti": "unique-token-id", + "events": {"http://schemas.openid.net/event/backchannel-logout": {}}, + } + + provider_a, auth_manager_a = create_mock_oidc_provider( + label="Test OIDC 1", integration_id=1, library_id=1 + ) + auth_manager_a.validate_logout_token.side_effect = Exception( + "Provider 1 cannot validate" + ) + + provider_b, auth_manager_b = create_mock_oidc_provider( + label="Test OIDC 2", integration_id=2, library_id=2 + ) + auth_manager_b.validate_logout_token.return_value = claims + provider_b.credential_manager.lookup_patron_by_identifier.return_value = Mock( + id=1 + ) + + mock_library_auth = Mock() + mock_library_auth.providers = [provider_a, provider_b] + backchannel_controller._authenticator.library_authenticators["test-library"] = ( + mock_library_auth + ) + + form_data = {"logout_token": "test.logout.token"} + + body, status = backchannel_controller.oidc_backchannel_logout( + form_data, db.session + ) + + assert status == 200 + assert body == "" + auth_manager_a.validate_logout_token.assert_called_once() + provider_a.credential_manager.invalidate_patron_credentials.assert_not_called() + provider_b.credential_manager.invalidate_patron_credentials.assert_called_once() diff --git a/tests/manager/integration/patron_auth/oidc/test_credential.py b/tests/manager/integration/patron_auth/oidc/test_credential.py index d6648313cc..6ee4fbd0a2 100644 --- a/tests/manager/integration/patron_auth/oidc/test_credential.py +++ b/tests/manager/integration/patron_auth/oidc/test_credential.py @@ -15,6 +15,7 @@ from palace.manager.integration.patron_auth.oidc.credential import OIDCCredentialManager from palace.manager.sqlalchemy.model.credential import Credential from palace.manager.sqlalchemy.model.datasource import DataSource +from palace.manager.sqlalchemy.util import get_one from tests.fixtures.database import DatabaseTransactionFixture @@ -285,6 +286,76 @@ def test_create_oidc_token_with_expiry_strategy( else: assert "refresh_token" not in token_data + @pytest.mark.parametrize( + "token_type,data_source_name", + [ + pytest.param( + OIDCCredentialManager.TOKEN_TYPE, + OIDCCredentialManager.TOKEN_DATA_SOURCE_NAME, + id="default-names", + ), + pytest.param("Custom token", "CustomProvider", id="custom-names"), + ], + ) + def test_create_oidc_token_with_configured_names( + self, + db: DatabaseTransactionFixture, + mock_patron, + sample_id_token_claims, + token_type: str, + data_source_name: str, + ): + """Test that configured token type and data source flow through create and lookup.""" + manager = OIDCCredentialManager( + token_type=token_type, data_source_name=data_source_name + ) + + credential = manager.create_oidc_token( + db.session, + mock_patron, + sample_id_token_claims, + "test-access-token", + ) + db.session.commit() + + assert credential.type == token_type + assert credential.data_source.name == data_source_name + + found_by_patron = manager.lookup_oidc_token_by_patron(db.session, mock_patron) + assert found_by_patron is not None + assert found_by_patron.id == credential.id + + assert credential.credential is not None + found_by_value = manager.lookup_oidc_token_by_value( + db.session, credential.credential, mock_patron.library_id + ) + assert found_by_value is not None + assert found_by_value.id == credential.id + + def test_create_oidc_token_with_subclassed_names( + self, + db: DatabaseTransactionFixture, + mock_patron, + sample_id_token_claims, + ): + """Class-constant overrides in a subclass take effect without constructor arguments.""" + + class SubclassedCredentialManager(OIDCCredentialManager): + TOKEN_TYPE = "Subclassed token" + TOKEN_DATA_SOURCE_NAME = "SubclassedProvider" + + manager = SubclassedCredentialManager() + + credential = manager.create_oidc_token( + db.session, + mock_patron, + sample_id_token_claims, + "test-access-token", + ) + + assert credential.type == "Subclassed token" + assert credential.data_source.name == "SubclassedProvider" + def test_lookup_oidc_token_by_patron_found( self, manager, @@ -832,3 +903,38 @@ def test_invalidate_patron_credentials_no_credentials(self, db, manager): count = manager.invalidate_patron_credentials(db.session, patron.id) assert count == 0 + + def test_invalidate_patron_credentials_scoped_to_data_source(self, db, manager): + """Invalidation only affects credentials in the manager's own namespace. + + Two managers sharing a token type but using different data sources + must be able to invalidate a patron's credentials independently. + """ + patron = db.patron() + other_manager = OIDCCredentialManager( + token_type=OIDCCredentialManager.TOKEN_TYPE, + data_source_name="OtherProvider", + ) + + oidc_credential = manager.create_oidc_token( + db.session, patron, {"sub": "user123"}, "access-token-oidc" + ) + other_credential = other_manager.create_oidc_token( + db.session, patron, {"sub": "user123"}, "access-token-other" + ) + + count = other_manager.invalidate_patron_credentials(db.session, patron.id) + + assert count == 1 + assert other_credential.expires <= utc_now() + assert oidc_credential.expires > utc_now() + + def test_invalidate_patron_credentials_does_not_create_data_source(self, db): + """Invalidation with nothing stored must not create the data source row.""" + patron = db.patron() + manager = OIDCCredentialManager(data_source_name="NeverStored") + + count = manager.invalidate_patron_credentials(db.session, patron.id) + + assert count == 0 + assert get_one(db.session, DataSource, name="NeverStored") is None diff --git a/tests/manager/integration/patron_auth/oidc/test_provider.py b/tests/manager/integration/patron_auth/oidc/test_provider.py index beeed1fb04..906f26a756 100644 --- a/tests/manager/integration/patron_auth/oidc/test_provider.py +++ b/tests/manager/integration/patron_auth/oidc/test_provider.py @@ -21,6 +21,9 @@ OIDCAuthLibrarySettings, OIDCAuthSettings, ) +from palace.manager.integration.patron_auth.oidc.credential import ( + OIDCCredentialManager, +) from palace.manager.integration.patron_auth.oidc.provider import ( OIDC_CANNOT_DETERMINE_PATRON, OIDC_FILTER_EVALUATION_ERROR, @@ -28,6 +31,7 @@ OIDC_NO_ACCESS_ERROR, OIDC_TOKEN_EXPIRED, OIDCAuthenticationProvider, + evaluate_patron_filters, ) from palace.manager.integration.patron_auth.oidc.util import ( OIDCDiscoveryError, @@ -60,6 +64,52 @@ def test_library_settings_class(self): def test_identifies_individuals(self, oidc_provider): assert oidc_provider.identifies_individuals is True + def test_patron_id_claim(self, oidc_provider: OIDCAuthenticationProvider) -> None: + assert oidc_provider.patron_id_claim == oidc_provider._settings.patron_id_claim + + @pytest.mark.parametrize( + "claims,patron_id_regex,expected", + [ + pytest.param({"sub": "user123"}, None, "user123", id="plain-claim"), + pytest.param( + {"sub": "jdoe@example.edu"}, + r"(?P[^@]+)@example\.edu", + "jdoe", + id="regex-extraction", + ), + pytest.param( + {"sub": "jdoe@other.edu"}, + r"(?P[^@]+)@example\.edu", + None, + id="regex-no-match", + ), + pytest.param({"email": "x@y.z"}, None, None, id="missing-claim"), + ], + ) + def test_extract_patron_identifier( + self, + create_oidc_settings: Callable[..., OIDCAuthSettings], + create_oidc_provider: Callable[..., OIDCAuthenticationProvider], + claims: dict[str, Any], + patron_id_regex: str | None, + expected: str | None, + ) -> None: + """The extraction applies the same claim and optional regex used at login.""" + settings_kwargs: dict[str, Any] = {} + if patron_id_regex is not None: + settings_kwargs["patron_id_regular_expression"] = patron_id_regex + provider = create_oidc_provider( + settings=create_oidc_settings(**settings_kwargs) + ) + + assert provider.extract_patron_identifier(claims) == expected + + def test_credential_manager( + self, oidc_provider: OIDCAuthenticationProvider + ) -> None: + assert isinstance(oidc_provider.credential_manager, OIDCCredentialManager) + assert oidc_provider.credential_manager is oidc_provider._credential_manager + def test_get_credential_from_header_with_bearer_token(self, oidc_provider): auth = MagicMock() auth.type = "Bearer" @@ -780,6 +830,58 @@ def test_filter_claims_both_levels_eval_error( assert "; " in detail assert len([r for r in caplog.records if r.levelno == logging.ERROR]) == 2 + @pytest.mark.parametrize( + "expressions,expected_uri", + [ + pytest.param( + [("integration", "claims['role'] == 'student'")], + None, + id="passing-expression", + ), + pytest.param( + [ + ("integration", "claims['role'] == 'student'"), + ("library", "claims['role'] == 'teacher'"), + ], + OIDC_NO_ACCESS_ERROR.uri, + id="failing-expression", + ), + pytest.param( + [("integration", "claims.no_such_attr == 'x'")], + OIDC_FILTER_EVALUATION_ERROR.uri, + id="evaluation-error", + ), + ], + ) + def test_evaluate_patron_filters( + self, + db: DatabaseTransactionFixture, + expressions: list[tuple[str, str]], + expected_uri: str | None, + ) -> None: + """evaluate_patron_filters grants access only when every expression passes.""" + claims = {"role": "student"} + context = {"claims": claims} + library = db.default_library() + + context_manager = ( + pytest.raises(ProblemDetailException) + if expected_uri is not None + else nullcontext() + ) + with context_manager as exc_info: + evaluate_patron_filters( + expressions, + context, + library=library, + claim_names=list(claims.keys()), + log=logging.getLogger(__name__), + ) + + if expected_uri is not None: + assert exc_info is not None + assert exc_info.value.problem_detail.uri == expected_uri + def test_filter_claims_library_not_found( self, db: DatabaseTransactionFixture, diff --git a/tests/manager/integration/patron_auth/oidc/test_util.py b/tests/manager/integration/patron_auth/oidc/test_util.py index f66ea51e4b..ec7ba5f2ba 100644 --- a/tests/manager/integration/patron_auth/oidc/test_util.py +++ b/tests/manager/integration/patron_auth/oidc/test_util.py @@ -352,6 +352,16 @@ def test_discover_oidc_configuration_invalid_json( assert result == mock_discovery_document mock_get.assert_called_once() + def test_store_in_cache_without_redis_is_noop(self): + """Storing is silently skipped when no Redis client is configured. + + The public callers never reach this branch (they only pass a cache + key obtained with Redis present), so exercise the guard directly. + """ + utility = OIDCUtility(redis_client=None) + + utility._store_in_cache("some-key", {"a": 1}, ttl=60) + class TestOIDCUtilityJWKS: """Tests for JWKS fetching.""" @@ -486,101 +496,6 @@ def test_fetch_jwks_invalid_json( mock_get.assert_called_once() -class TestOIDCUtilityPKCEStorage: - """Tests for PKCE storage and retrieval in Redis.""" - - def test_store_pkce_success(self, redis_fixture): - state_token = "test-state-token" - code_verifier = "test-code-verifier-abc123" - - utility = OIDCUtility(redis_client=redis_fixture.client) - utility.store_pkce(state_token, code_verifier) - - cache_key = redis_fixture.client.get_key( - f"{OIDCUtility.PKCE_KEY_PREFIX}{state_token}" - ) - cached = redis_fixture.client.get(cache_key) - - assert cached is not None - data = json.loads(cached) - assert data["code_verifier"] == code_verifier - assert "timestamp" in data - - def test_store_pkce_with_metadata(self, redis_fixture): - state_token = "test-state-token" - code_verifier = "test-code-verifier" - metadata = {"extra_field": "extra_value"} - - utility = OIDCUtility(redis_client=redis_fixture.client) - utility.store_pkce(state_token, code_verifier, metadata=metadata) - - cache_key = redis_fixture.client.get_key( - f"{OIDCUtility.PKCE_KEY_PREFIX}{state_token}" - ) - cached = redis_fixture.client.get(cache_key) - - data = json.loads(cached) - assert data["code_verifier"] == code_verifier - assert data["extra_field"] == "extra_value" - - def test_store_pkce_without_redis_raises_error(self): - utility = OIDCUtility(redis_client=None) - - with pytest.raises(OIDCUtilityError, match="Redis client required"): - utility.store_pkce("state", "verifier") - - def test_retrieve_pkce_success(self, redis_fixture): - state_token = "test-state-token" - code_verifier = "test-code-verifier" - - utility = OIDCUtility(redis_client=redis_fixture.client) - utility.store_pkce(state_token, code_verifier) - - retrieved = utility.retrieve_pkce(state_token, delete=False) - - assert retrieved is not None - assert retrieved["code_verifier"] == code_verifier - - def test_retrieve_pkce_with_delete(self, redis_fixture): - state_token = "test-state-token" - code_verifier = "test-code-verifier" - - utility = OIDCUtility(redis_client=redis_fixture.client) - utility.store_pkce(state_token, code_verifier) - - retrieved1 = utility.retrieve_pkce(state_token, delete=True) - assert retrieved1 is not None - - retrieved2 = utility.retrieve_pkce(state_token, delete=False) - assert retrieved2 is None - - def test_retrieve_pkce_not_found(self, redis_fixture): - utility = OIDCUtility(redis_client=redis_fixture.client) - - retrieved = utility.retrieve_pkce("nonexistent-state") - - assert retrieved is None - - def test_retrieve_pkce_without_redis_raises_error(self): - utility = OIDCUtility(redis_client=None) - - with pytest.raises(OIDCUtilityError, match="Redis client required"): - utility.retrieve_pkce("state") - - def test_retrieve_pkce_corrupted_json(self, redis_fixture): - state_token = "test-state-token" - - cache_key = redis_fixture.client.get_key( - f"{OIDCUtility.PKCE_KEY_PREFIX}{state_token}" - ) - redis_fixture.client.set(cache_key, "invalid-json{{{", ex=600) - - utility = OIDCUtility(redis_client=redis_fixture.client) - retrieved = utility.retrieve_pkce(state_token) - - assert retrieved is None - - class TestOIDCUtilityLogoutState: """Tests for logout state storage and retrieval.""" diff --git a/tests/manager/integration/patron_auth/oidc/test_validator.py b/tests/manager/integration/patron_auth/oidc/test_validator.py index 3af8500546..b82d0e77c8 100644 --- a/tests/manager/integration/patron_auth/oidc/test_validator.py +++ b/tests/manager/integration/patron_auth/oidc/test_validator.py @@ -2,13 +2,11 @@ from __future__ import annotations -import re import time import pytest from palace.manager.integration.patron_auth.oidc.validator import ( - OIDCPatronIDExtractionError, OIDCTokenClaimsError, OIDCTokenSignatureError, OIDCTokenValidator, @@ -263,206 +261,6 @@ def test_validate_claims_multiple_errors(self): assert "Missing required claim: 'sub'" in error_message -class TestOIDCTokenValidatorPatronIDExtraction: - """Tests for patron ID extraction from claims.""" - - @pytest.mark.parametrize( - "claim_name,expected_patron_id", - [ - pytest.param("sub", "user123", id="from-sub"), - pytest.param("email", "testuser@example.com", id="from-email"), - pytest.param( - "preferred_username", "testuser", id="from-preferred-username" - ), - ], - ) - def test_extract_patron_id_from_claim( - self, claim_name, expected_patron_id, mock_id_token_claims - ): - """Test extracting patron ID from different claim names.""" - validator = OIDCTokenValidator() - - patron_id = validator.extract_patron_id( - mock_id_token_claims, claim_name=claim_name - ) - - assert patron_id == expected_patron_id - - def test_extract_patron_id_missing_claim(self): - validator = OIDCTokenValidator() - claims = {"sub": "user123", "email": "test@example.com"} - - with pytest.raises(OIDCPatronIDExtractionError, match="not found in ID token"): - validator.extract_patron_id(claims, claim_name="nonexistent_claim") - - def test_extract_patron_id_empty_claim(self): - validator = OIDCTokenValidator() - claims = {"sub": "user123", "email": ""} - - with pytest.raises( - OIDCPatronIDExtractionError, match="is empty or whitespace-only" - ): - validator.extract_patron_id(claims, claim_name="email") - - def test_extract_patron_id_with_regex(self, mock_id_token_claims): - validator = OIDCTokenValidator() - regex_pattern = re.compile(r"(?P[^@]+)@") - - patron_id = validator.extract_patron_id( - mock_id_token_claims, claim_name="email", regex_pattern=regex_pattern - ) - - assert patron_id == "testuser" - - def test_extract_patron_id_regex_no_match(self, mock_id_token_claims): - validator = OIDCTokenValidator() - regex_pattern = re.compile(r"(?PNOMATCH)") - - with pytest.raises( - OIDCPatronIDExtractionError, match="regex pattern did not match" - ): - validator.extract_patron_id( - mock_id_token_claims, claim_name="email", regex_pattern=regex_pattern - ) - - def test_extract_patron_id_regex_missing_named_group(self, mock_id_token_claims): - validator = OIDCTokenValidator() - regex_pattern = re.compile(r"([^@]+)@") - - with pytest.raises( - OIDCPatronIDExtractionError, match="must contain a named group 'patron_id'" - ): - validator.extract_patron_id( - mock_id_token_claims, claim_name="email", regex_pattern=regex_pattern - ) - - def test_extract_patron_id_regex_empty_match(self): - validator = OIDCTokenValidator() - claims = {"email": "@example.com"} - regex_pattern = re.compile(r"(?P[^@]*)@") - - with pytest.raises( - OIDCPatronIDExtractionError, match="'patron_id' group is empty" - ): - validator.extract_patron_id( - claims, claim_name="email", regex_pattern=regex_pattern - ) - - def test_extract_patron_id_non_string_claim(self): - validator = OIDCTokenValidator() - claims = {"sub": "user123", "age": 42} - - patron_id = validator.extract_patron_id(claims, claim_name="age") - - assert patron_id == "42" - - def test_extract_patron_id_complex_regex(self): - validator = OIDCTokenValidator() - claims = {"eduPersonPrincipalName": "jsmith123@university.edu"} - regex_pattern = re.compile(r"(?P[a-z]+\d+)@") - - patron_id = validator.extract_patron_id( - claims, claim_name="eduPersonPrincipalName", regex_pattern=regex_pattern - ) - - assert patron_id == "jsmith123" - - -class TestOIDCTokenValidatorCombined: - """Tests for combined validation and extraction.""" - - def test_validate_and_extract_success( - self, mock_id_token, mock_jwks, mock_id_token_claims - ): - validator = OIDCTokenValidator() - - claims, patron_id = validator.validate_and_extract( - id_token=mock_id_token, - jwks=mock_jwks, - expected_issuer=TEST_ISSUER, - expected_audience=TEST_CLIENT_ID, - patron_id_claim="sub", - nonce=TEST_NONCE, - ) - - assert claims["sub"] == "user123" - assert patron_id == "user123" - - def test_validate_and_extract_with_regex( - self, mock_id_token, mock_jwks, mock_id_token_claims - ): - validator = OIDCTokenValidator() - regex_pattern = re.compile(r"(?P[^@]+)@") - - claims, patron_id = validator.validate_and_extract( - id_token=mock_id_token, - jwks=mock_jwks, - expected_issuer=TEST_ISSUER, - expected_audience=TEST_CLIENT_ID, - patron_id_claim="email", - nonce=TEST_NONCE, - patron_id_regex=regex_pattern, - ) - - assert claims["email"] == "testuser@example.com" - assert patron_id == "testuser" - - def test_validate_and_extract_signature_failure(self, mock_jwks): - validator = OIDCTokenValidator() - invalid_token = "invalid.jwt.token" - - with pytest.raises(OIDCTokenSignatureError): - validator.validate_and_extract( - id_token=invalid_token, - jwks=mock_jwks, - expected_issuer=TEST_ISSUER, - expected_audience=TEST_CLIENT_ID, - patron_id_claim="sub", - ) - - def test_validate_and_extract_claims_failure(self, oidc_test_keys, mock_jwks): - validator = OIDCTokenValidator() - - claims = { - "iss": "https://wrong.issuer.com", - "aud": TEST_CLIENT_ID, - "sub": "user123", - "exp": int(time.time()) + 3600, - "iat": int(time.time()), - } - invalid_token = oidc_test_keys.sign_jwt(claims) - - with pytest.raises(OIDCTokenClaimsError, match="Issuer mismatch"): - validator.validate_and_extract( - id_token=invalid_token, - jwks=mock_jwks, - expected_issuer=TEST_ISSUER, - expected_audience=TEST_CLIENT_ID, - patron_id_claim="sub", - ) - - def test_validate_and_extract_patron_id_failure(self, oidc_test_keys, mock_jwks): - validator = OIDCTokenValidator() - - claims = { - "iss": TEST_ISSUER, - "aud": TEST_CLIENT_ID, - "sub": "user123", - "exp": int(time.time()) + 3600, - "iat": int(time.time()), - } - token = oidc_test_keys.sign_jwt(claims) - - with pytest.raises(OIDCPatronIDExtractionError, match="not found in ID token"): - validator.validate_and_extract( - id_token=token, - jwks=mock_jwks, - expected_issuer=TEST_ISSUER, - expected_audience=TEST_CLIENT_ID, - patron_id_claim="nonexistent_claim", - ) - - class TestOIDCTokenValidatorClockSkew: """Tests for clock skew tolerance in token validation."""