OAuth router helpers¶

def bind_oauth_token_encryption(session: object, oauth_token_encryption: OAuthTokenEncryption) -> None:
    """Bind an explicit OAuth token encryption policy to a SQLAlchemy session path.

    Raises:
        TypeError: When ``oauth_token_encryption`` is not a current-module
            ``OAuthTokenEncryption`` instance.
    """
    if not isinstance(oauth_token_encryption, OAuthTokenEncryption):
        msg = "oauth_token_encryption must be an OAuthTokenEncryption instance from the current module."
        raise TypeError(msg)
    for target in _iter_session_targets(session):
        info = getattr(target, "info", None)
        if isinstance(info, dict):
            info[_OAUTH_TOKEN_ENCRYPTION_INFO_KEY] = oauth_token_encryption

def get_bound_oauth_token_encryption(session: object) -> OAuthTokenEncryption | None:
    """Return the OAuth token encryption policy bound to the given session path."""
    for target in _iter_session_targets(session):
        info = getattr(target, "info", None)
        if isinstance(info, dict):
            policy = info.get(_OAUTH_TOKEN_ENCRYPTION_INFO_KEY)
            if isinstance(policy, OAuthTokenEncryption):
                return policy
    return None

def register_oauth_model_encryption_events(model_base: type[Any]) -> None:
    """Register mapper events that keep OAuth token attributes plaintext in memory."""
    listeners: tuple[tuple[str, object], ...] = (
        ("load", _decrypt_loaded_oauth_tokens),
        ("refresh", _decrypt_refreshed_oauth_tokens),
        ("before_insert", _encrypt_oauth_tokens_before_insert),
        ("before_update", _encrypt_oauth_tokens_before_update),
        ("after_insert", _restore_oauth_tokens_after_write),
        ("after_update", _restore_oauth_tokens_after_write),
    )
    for identifier, listener in listeners:
        if not event.contains(model_base, identifier, listener):
            event.listen(model_base, identifier, listener, propagate=True)

    transaction_listeners: tuple[tuple[str, object], ...] = (
        ("after_rollback", _restore_oauth_token_snapshots_after_rollback),
        ("after_soft_rollback", _restore_oauth_token_snapshots_after_rollback),
    )
    for identifier, listener in transaction_listeners:
        if not event.contains(Session, identifier, listener):
            event.listen(Session, identifier, listener)

def require_oauth_token_encryption(
    oauth_token_encryption: OAuthTokenEncryption | None,
    *,
    context: str = "OAuth token persistence",
) -> OAuthTokenEncryption:
    """Return the explicit policy or fail when persistence would rely on ambient state.

    Raises:
        ConfigurationError: When no explicit policy was supplied, or when a policy without a
            configured key is used while ``unsafe_testing=False``.
    """
    if oauth_token_encryption is None:
        msg = (
            f"{context} requires an explicit oauth_token_encryption policy. "
            "Pass oauth_token_encryption=OAuthTokenEncryption(...) to SQLAlchemyUserDatabase() "
            "or call bind_oauth_token_encryption(session, OAuthTokenEncryption(...))."
        )
        raise ConfigurationError(msg)
    if not isinstance(oauth_token_encryption, OAuthTokenEncryption):
        msg = (
            f"{context} requires an OAuthTokenEncryption instance from the current module. "
            "Create a fresh OAuthTokenEncryption(...) policy before binding or passing it to "
            "SQLAlchemyUserDatabase()."
        )
        raise ConfigurationError(msg)
    oauth_token_encryption.require_configured(context=context)
    return oauth_token_encryption

def create_provider_oauth_controller[UP: UserProtocol[Any], ID](
    *,
    config: ProviderOAuthControllerConfig[UP, ID] | None = None,
    **options: Unpack[ProviderOAuthControllerOptions[UP, ID]],
) -> type[Controller]:
    """Build a provider-specific OAuth controller from a client or lazy factory.

    The authorize endpoint uses only server-configured ``oauth_scopes``. Runtime
    scope-query overrides are rejected. ``redirect_base_url`` must use a
    non-loopback ``https://`` origin; the manual controller API does not expose
    a debug or testing override for insecure callback origins. The generated
    flow encrypts transient state + PKCE verifier material with
    ``oauth_flow_cookie_secret`` and enforces RFC 7636 PKCE S256, so manual
    clients must accept ``code_challenge`` / ``code_challenge_method`` on
    authorization and ``code_verifier`` on token exchange.

    Returns:
        Generated controller class mounted under the provider-specific path.

    Raises:
        ValueError: If ``config`` and keyword options are combined.
    """
    if config is not None and options:
        msg = "Pass either ProviderOAuthControllerConfig or keyword options, not both."
        raise ValueError(msg)
    settings = ProviderOAuthControllerConfig(**options) if config is None else config

    _validate_manual_oauth_redirect_base_url(
        settings.redirect_base_url,
        strict=settings.oauth_redirect_dns_strict,
    )
    oauth_client_adapter = _build_oauth_client_adapter(
        oauth_client=settings.oauth_client,
        oauth_client_factory=settings.oauth_client_factory,
        oauth_client_class=settings.oauth_client_class,
        oauth_client_kwargs=settings.oauth_client_kwargs,
        oauth_client_class_loader=load_httpx_oauth_client,
    )
    resolved_path = settings.path if settings.path is not None else _build_oauth_login_path(settings.auth_path)

    return _create_login_oauth_controller(
        _OAuthLoginControllerSettings(
            provider_name=settings.provider_name,
            backend=settings.backend,
            user_manager=settings.user_manager,
            oauth_client_adapter=oauth_client_adapter,
            redirect_base_url=settings.redirect_base_url,
            oauth_flow_cookie_secret=settings.oauth_flow_cookie_secret,
            path=resolved_path,
            cookie_secure=settings.cookie_secure,
            oauth_scopes=settings.oauth_scopes,
            associate_by_email=settings.associate_by_email,
            trust_provider_email_verified=settings.trust_provider_email_verified,
            oauth_redirect_dns_strict=settings.oauth_redirect_dns_strict,
            validate_redirect_base_url=False,
        ),
    )

def load_httpx_oauth_client(oauth_client_class: str, /, **client_kwargs: object) -> OAuthClientProtocol:
    """Import and instantiate an ``httpx-oauth`` client lazily.

    Returns:
        Instantiated OAuth client.

    Raises:
        ImportError: If the optional `httpx-oauth` dependency is not installed.
        ModuleNotFoundError: If a non-`httpx-oauth` import required by the client cannot be resolved.
        ConfigurationError: If the client path is invalid or the class cannot be imported.
    """
    module_name, _, class_name = oauth_client_class.rpartition(".")
    if not module_name or not class_name:
        msg = "oauth_client_class must be a fully qualified module path."
        raise ConfigurationError(msg)

    try:
        module = import_module(module_name)
    except ModuleNotFoundError as exc:
        if exc.name is not None and exc.name.startswith("httpx_oauth"):
            msg = "Install litestar-auth[oauth] to use OAuth controllers."
            raise ImportError(msg) from exc
        raise

    oauth_client_type: OAuthClientConstructor | None = getattr(module, class_name, None)
    if oauth_client_type is None:
        msg = f"OAuth client class {oauth_client_class!r} could not be imported."
        raise ConfigurationError(msg)

    return oauth_client_type(**client_kwargs)