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."""