ai-workspace-services/litellm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #26825 from stuxf/fix/oauth2-proxy-header-forgery

Browse Source 
chore(auth): require trusted proxy for header identity auth
This commit is contained in:
yuneng-jiang 2026-05-01 18:47:58 -07:00 committed by GitHub
commit 5614469f22
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 542 additions and 87 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

54
enterprise/litellm_enterprise/proxy/auth/custom_sso_handler.py
View File

@ -10,28 +10,21 @@ has already authenticated the user) and you need to extract user information fro
custom headers or other request attributes.
"""
from typing import TYPE_CHECKING, Dict, Optional, Union, cast
from typing import cast
from fastapi import Request
from fastapi.responses import RedirectResponse
if TYPE_CHECKING:
from fastapi_sso.sso.base import OpenID
else:
from typing import Any as OpenID
from litellm.proxy.management_endpoints.types import CustomOpenID
class EnterpriseCustomSSOHandler:
"""
Enterprise Custom SSO Handler for LiteLLM Proxy
This class provides methods for handling custom SSO authentication flows
where users can implement their own authentication logic by processing
request headers and returning user information in OpenID format.
"""
@staticmethod
async def handle_custom_ui_sso_sign_in(
request: Request,
@ -40,16 +33,16 @@ class EnterpriseCustomSSOHandler:
Allow a user to execute their custom code to parse incoming request headers and return a OpenID object
Use this when you have an OAuth proxy in front of LiteLLM (where the OAuth proxy has already authenticated the user)
Args:
request: The FastAPI request object containing headers and other request data
Returns:
RedirectResponse: Redirect response that sends the user to the LiteLLM UI with authentication token
Raises:
ValueError: If custom_ui_sso_sign_in_handler is not configured
Example:
This method is typically called when a user has already been authenticated by an
external OAuth proxy and the proxy has added custom headers containing user information.
@ -60,27 +53,44 @@ class EnterpriseCustomSSOHandler:
from litellm.integrations.custom_sso_handler import CustomSSOLoginHandler
from litellm.proxy.proxy_server import (
CommonProxyErrors,
general_settings,
premium_user,
user_custom_ui_sso_sign_in_handler,
)
from litellm.proxy.auth.trusted_proxy_utils import (
require_trusted_proxy_request,
)
if premium_user is not True:
raise ValueError(CommonProxyErrors.not_premium_user.value)
if user_custom_ui_sso_sign_in_handler is None:
raise ValueError("custom_ui_sso_sign_in_handler is not configured. Please set it in general_settings.")
custom_sso_login_handler = cast(CustomSSOLoginHandler, user_custom_ui_sso_sign_in_handler)
openid_response: OpenID = await custom_sso_login_handler.handle_custom_ui_sso_sign_in(
raise ValueError(
"custom_ui_sso_sign_in_handler is not configured. Please set it in general_settings."
)
require_trusted_proxy_request(
request=request,
general_settings=general_settings,
feature_name="Custom UI SSO",
)
custom_sso_login_handler = cast(
CustomSSOLoginHandler, user_custom_ui_sso_sign_in_handler
)
openid_response: OpenID = (
await custom_sso_login_handler.handle_custom_ui_sso_sign_in(
request=request,
)
)
# Import here to avoid circular imports
from litellm.proxy.management_endpoints.ui_sso import SSOAuthenticationHandler
return await SSOAuthenticationHandler.get_redirect_response_from_openid(
result=openid_response,
request=request,
received_response=None,
generic_client_id=None,
ui_access_mode=None,
)
)

11
litellm/integrations/custom_sso_handler.py
View File

@ -18,6 +18,17 @@ class CustomSSOLoginHandler(CustomLogger):
self,
request: Request,
) -> OpenID:
from litellm.proxy.auth.trusted_proxy_utils import (
require_trusted_proxy_request,
)
from litellm.proxy.proxy_server import general_settings
require_trusted_proxy_request(
request=request,
general_settings=general_settings,
feature_name="Custom UI SSO",
)
request_headers_dict = dict(request.headers)
return OpenID(
id=request_headers_dict.get("x-litellm-user-id"),

4
litellm/proxy/_types.py
View File

@ -2438,6 +2438,10 @@ class ConfigGeneralSettings(LiteLLMPydanticObjectBase):
None,
description="CIDR ranges of trusted reverse proxies. When set, X-Forwarded-For headers are only trusted from these IPs.",
)
trusted_proxy_ranges: Optional[List[str]] = Field(
None,
description="CIDR ranges of trusted reverse proxies allowed to provide identity headers for header-based auth paths such as enable_oauth2_proxy_auth and custom_ui_sso_sign_in_handler.",
)
store_model_in_db: Optional[bool] = Field(
None,
description="If True, models and config are stored in and loaded from the database. Default is False.",

92
litellm/proxy/auth/oauth2_proxy_hook.py
View File

@ -1,19 +1,69 @@
from typing import Any, Dict
from typing import Any, Dict, FrozenSet
from fastapi import Request
from litellm._logging import verbose_proxy_logger
from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy.auth.trusted_proxy_utils import require_trusted_proxy_request
# OAuth2-proxy header trust is for **identity assertion** from a trusted
# upstream auth proxy (oauth2-proxy, Authelia, etc.). The allowlist below
# is the only safe surface — anything else (``user_role``, ``api_key``,
# ``permissions``, ``max_budget``, ``user_max_budget``,
# ``team_tpm_limit``, ``end_user_max_budget``, ``allowed_model_region``,
# and dozens of similar policy fields scattered across the
# ``LiteLLM_VerificationTokenView`` hierarchy) is a privilege grant that
# would let a caller forge their own enforcement parameters by sending
# the matching header.
#
# A denylist of "privileged fields" is unmaintainable in this codebase:
# the auth model has ~50 budget/spend/limit/permission fields and gains
# more with each release. An allowlist scoped to identity assertion is
# default-secure — new fields are blocked automatically.
#
# Operators who need a trusted upstream to assert anything beyond
# identity should switch to JWT authentication, which validates a
# signature on the assertion rather than blindly trusting headers.
ALLOWED_OAUTH2_PROXY_FIELDS: FrozenSet[str] = frozenset(
{
"user_id",
"user_email",
"team_id",
"team_alias",
"org_id",
"models",
}
)
async def handle_oauth2_proxy_request(request: Request) -> UserAPIKeyAuth:
"""
Handle request from oauth2 proxy.
Resolve a ``UserAPIKeyAuth`` from request headers per the admin-set
``oauth2_config_mappings``.
The auth model assumes the proxy is deployed behind a trusted OAuth2
reverse proxy that injects authenticated identity headers (e.g.
oauth2-proxy, Authelia).
**Identity-only allowlist.** ``oauth2_config_mappings`` maps header
names to ``UserAPIKeyAuth`` fields. Without an allowlist, an admin
who maps the wrong header to ``user_role`` lets any caller send
``X-User-Role: proxy_admin`` and gain full admin privileges
(Pydantic coerces the string into the enum). Only fields in
``ALLOWED_OAUTH2_PROXY_FIELDS`` (identity assertion only — see the
constant's comment) may be mapped; any other mapping is rejected at
request time so the misconfiguration surfaces loudly rather than as
a silent privesc.
"""
from litellm.proxy.proxy_server import general_settings
verbose_proxy_logger.debug("Handling oauth2 proxy request")
# Define the OAuth2 config mappings
require_trusted_proxy_request(
request=request,
general_settings=general_settings,
feature_name="OAuth2 proxy auth",
)
oauth2_config_mappings: Dict[str, str] = (
general_settings.get("oauth2_config_mappings") or {}
)
@ -21,21 +71,32 @@ async def handle_oauth2_proxy_request(request: Request) -> UserAPIKeyAuth:
if not oauth2_config_mappings:
raise ValueError("Oauth2 config mappings not found in general_settings")
# Initialize a dictionary to store the mapped values
auth_data: Dict[str, Any] = {}
# Extract values from headers based on the mappings
disallowed = sorted(
set(oauth2_config_mappings.keys()) - ALLOWED_OAUTH2_PROXY_FIELDS
)
if disallowed:
raise ValueError(
"Oauth2 proxy auth refuses to map non-identity UserAPIKeyAuth "
f"fields from request headers: {disallowed}. Only identity "
f"fields are accepted ({sorted(ALLOWED_OAUTH2_PROXY_FIELDS)}); "
"anything else (privileges, budgets, rate limits, metadata) "
"would let a caller forge enforcement parameters by spoofing "
"the matching header. If you need a trusted upstream to "
"assert anything beyond identity, use JWT auth "
"(signature-validated) instead of header-trust."
)
auth_data: Dict[str, Any] = {}
for key, header in oauth2_config_mappings.items():
value = request.headers.get(header)
if value:
# Convert max_budget to float if present
if key == "max_budget":
auth_data[key] = float(value)
# Convert models to list if present
elif key == "models":
auth_data[key] = [model.strip() for model in value.split(",")]
else:
auth_data[key] = value
if not value:
continue
if key == "models":
auth_data[key] = [model.strip() for model in value.split(",")]
else:
auth_data[key] = value
verbose_proxy_logger.debug(
"Auth data before creating UserAPIKeyAuth object: keys=%s",
list(auth_data.keys()),
@ -45,5 +106,4 @@ async def handle_oauth2_proxy_request(request: Request) -> UserAPIKeyAuth:
"UserAPIKeyAuth object created with keys: %s",
list(user_api_key_auth.__fields_set__),
)
# Create and return UserAPIKeyAuth object
return user_api_key_auth

118
litellm/proxy/auth/trusted_proxy_utils.py Normal file
View File

@ -0,0 +1,118 @@
import ipaddress
from typing import Any, Dict, List, Optional, Union
from fastapi import Request
from litellm._logging import verbose_proxy_logger
TRUSTED_PROXY_RANGES_KEY = "trusted_proxy_ranges"
TrustedProxyNetwork = Union[ipaddress.IPv4Network, ipaddress.IPv6Network]
def _get_proxy_general_settings() -> Dict[str, Any]:
try:
from litellm.proxy.proxy_server import general_settings
return general_settings or {}
except ImportError:
return {}
def _normalize_cidr_ranges(configured_ranges: Any, *, setting_name: str) -> List[str]:
if not configured_ranges:
return []
if isinstance(configured_ranges, str):
return [
raw_range.strip()
for raw_range in configured_ranges.split(",")
if raw_range.strip()
]
if isinstance(configured_ranges, (list, tuple, set)):
return [
str(raw_range).strip()
for raw_range in configured_ranges
if str(raw_range).strip()
]
verbose_proxy_logger.warning(
"Invalid %s value: expected a list of CIDR ranges, got %s",
setting_name,
type(configured_ranges).__name__,
)
return []
def parse_trusted_proxy_ranges(
configured_ranges: Any,
*,
setting_name: str = TRUSTED_PROXY_RANGES_KEY,
) -> List[TrustedProxyNetwork]:
networks: List[TrustedProxyNetwork] = []
for cidr in _normalize_cidr_ranges(configured_ranges, setting_name=setting_name):
try:
networks.append(ipaddress.ip_network(cidr, strict=False))
except ValueError:
verbose_proxy_logger.warning(
"Invalid CIDR in %s: %s, skipping", setting_name, cidr
)
return networks
def _get_direct_client_ip(request: Request) -> Optional[str]:
client = getattr(request, "client", None)
client_host = getattr(client, "host", None)
if isinstance(client_host, str):
return client_host
return None
def _is_ip_in_networks(
client_ip: Optional[str], networks: List[TrustedProxyNetwork]
) -> bool:
if not client_ip or not networks:
return False
try:
addr = ipaddress.ip_address(client_ip.strip())
except ValueError:
return False
return any(addr in network for network in networks)
def require_trusted_proxy_request(
*,
request: Request,
general_settings: Optional[Dict[str, Any]] = None,
feature_name: str,
setting_name: str = TRUSTED_PROXY_RANGES_KEY,
) -> None:
"""
Fail closed unless the direct TCP peer is one of the configured
trusted reverse proxies.
Header-based auth paths must validate the direct peer, not
X-Forwarded-For, because the direct peer is the actor supplying the
identity headers.
"""
if general_settings is None:
general_settings = _get_proxy_general_settings()
trusted_networks = parse_trusted_proxy_ranges(
general_settings.get(setting_name), setting_name=setting_name
)
if not trusted_networks:
raise ValueError(
f"{feature_name} requires general_settings.{setting_name} before "
"trusting identity headers from an upstream proxy."
)
direct_client_ip = _get_direct_client_ip(request)
if not _is_ip_in_networks(direct_client_ip, trusted_networks):
verbose_proxy_logger.warning(
"%s rejected identity headers from untrusted direct client IP %r",
feature_name,
direct_client_ip,
)
raise ValueError(
f"{feature_name} only accepts identity headers from configured "
f"trusted proxy ranges. Direct client IP {direct_client_ip!r} "
"is not trusted."
)

211
tests/test_litellm/proxy/auth/test_oauth2_proxy_hook.py Normal file
View File

@ -0,0 +1,211 @@
"""
Regression tests for the OAuth2-proxy header-forgery fix
(GHSA-5c3m-qffq-4r9m).
The hook reads HTTP request headers per ``oauth2_config_mappings`` and
constructs a ``UserAPIKeyAuth`` from them. The fix has two parts:
1. Only requests from configured trusted proxy CIDR ranges may provide
identity headers.
2. Only identity fields may be mapped from those headers. Without the
identity-only allowlist any field could be mapped — including
``user_role``, which Pydantic coerces from the string
``"proxy_admin"`` into ``LitellmUserRoles.PROXY_ADMIN``.
"""
import os
import sys
import pytest
from fastapi import Request
from starlette.datastructures import Headers
sys.path.insert(0, os.path.abspath("../../../.."))
from litellm.proxy._types import LitellmUserRoles
from litellm.proxy.auth.oauth2_proxy_hook import (
ALLOWED_OAUTH2_PROXY_FIELDS,
handle_oauth2_proxy_request,
)
def _request_with_headers(headers: dict, *, client_host: str = "127.0.0.1") -> Request:
scope = {
"type": "http",
"client": (client_host, 12345),
"headers": [(k.lower().encode(), v.encode()) for k, v in headers.items()],
}
request = Request(scope=scope)
request._headers = Headers(headers)
return request
@pytest.fixture
def configure_proxy(monkeypatch):
"""
Yields a callable that sets ``oauth2_config_mappings`` and
``trusted_proxy_ranges`` on the proxy_server module for the duration
of one test. Defaults to a single identity mapping and localhost as
a trusted proxy.
"""
import litellm.proxy.proxy_server as proxy_server
def _configure(*, mappings=None, trusted_proxy_ranges=("127.0.0.1/32",)):
if mappings is None:
mappings = {"user_id": "x-user-id"}
settings = {
"oauth2_config_mappings": mappings,
"trusted_proxy_ranges": trusted_proxy_ranges,
}
monkeypatch.setattr(
proxy_server,
"general_settings",
settings,
raising=False,
)
return _configure
@pytest.mark.asyncio
async def test_returns_auth_for_simple_user_id_mapping(configure_proxy):
configure_proxy()
request = _request_with_headers({"x-user-id": "alice"})
auth = await handle_oauth2_proxy_request(request)
assert auth.user_id == "alice"
assert auth.user_role is None
@pytest.mark.asyncio
async def test_rejects_identity_headers_without_trusted_proxy_ranges(configure_proxy):
configure_proxy(trusted_proxy_ranges=None)
request = _request_with_headers({"x-user-id": "alice"})
with pytest.raises(ValueError, match="trusted_proxy_ranges"):
await handle_oauth2_proxy_request(request)
@pytest.mark.asyncio
async def test_rejects_identity_headers_from_untrusted_direct_client(configure_proxy):
configure_proxy(trusted_proxy_ranges=["10.0.0.0/24"])
request = _request_with_headers({"x-user-id": "alice"}, client_host="203.0.113.10")
with pytest.raises(ValueError, match="not trusted"):
await handle_oauth2_proxy_request(request)
@pytest.mark.parametrize(
"privileged_field",
[
# The GHSA-5c3m-qffq-4r9m primary privesc field.
"user_role",
# Key-level enforcement bypass shapes.
"api_key",
"token",
"permissions",
"allowed_routes",
"max_budget",
"spend",
"tpm_limit",
"rpm_limit",
"model_max_budget",
"metadata",
# User-level enforcement bypass — flagged by Greptile as a denylist gap.
"user_max_budget",
"user_tpm_limit",
"user_rpm_limit",
"user_spend",
# Team / org / end-user / region — same class, all denied by the
# identity-only allowlist.
"team_max_budget",
"team_spend",
"team_member_tpm_limit",
"organization_max_budget",
"organization_tpm_limit",
"end_user_max_budget",
"allowed_model_region",
# Anything not on ALLOWED_OAUTH2_PROXY_FIELDS is blocked, even
# fabricated field names admins might try.
"definitely_not_a_real_field",
],
)
@pytest.mark.asyncio
async def test_refuses_to_map_non_identity_fields(configure_proxy, privileged_field):
# GHSA-5c3m-qffq-4r9m attack shape: admin maps a privileged field
# to a header and a caller forges the value. The allowlist rejects
# any non-identity mapping at request time, regardless of whether
# the field ever appeared on a denylist — which is the whole reason
# we use an allowlist instead.
configure_proxy(mappings={privileged_field: f"x-{privileged_field}"})
request = _request_with_headers({f"x-{privileged_field}": "proxy_admin"})
with pytest.raises(ValueError) as exc:
await handle_oauth2_proxy_request(request)
assert privileged_field in str(exc.value)
@pytest.mark.parametrize("identity_field", sorted(ALLOWED_OAUTH2_PROXY_FIELDS))
def test_allowlist_is_identity_only(identity_field):
# Lock in the allowlist's intent: only identity-assertion fields are
# safe to populate from a header. If anyone proposes adding budget /
# spend / role / permission to ``ALLOWED_OAUTH2_PROXY_FIELDS``, this
# assertion forces them to update the test deliberately.
assert identity_field in {
"user_id",
"user_email",
"team_id",
"team_alias",
"org_id",
"models",
}
@pytest.mark.asyncio
async def test_user_role_header_forgery_attack_is_blocked(configure_proxy):
# End-to-end form of the privesc: with ``user_role`` mapped, the
# forged ``X-User-Role: proxy_admin`` header would have produced
# a ``UserAPIKeyAuth(user_role=PROXY_ADMIN)``. Now rejected before
# any auth object is constructed.
configure_proxy(
mappings={"user_id": "x-user-id", "user_role": "x-user-role"},
)
request = _request_with_headers(
{
"x-user-id": "attacker",
"x-user-role": LitellmUserRoles.PROXY_ADMIN.value,
}
)
with pytest.raises(ValueError, match="user_role"):
await handle_oauth2_proxy_request(request)
@pytest.mark.asyncio
async def test_safe_fields_still_pass_through(configure_proxy):
# The documented use case for OAuth2 proxy auth: identity assertion
# from a trusted upstream. Must remain unaffected by the denylist.
configure_proxy(
mappings={
"user_id": "x-user-id",
"user_email": "x-user-email",
"team_id": "x-team-id",
"models": "x-models",
},
)
request = _request_with_headers(
{
"x-user-id": "alice",
"x-user-email": "alice@example.com",
"x-team-id": "team-corp",
"x-models": "gpt-4, gpt-3.5-turbo",
}
)
auth = await handle_oauth2_proxy_request(request)
assert auth.user_id == "alice"
assert auth.user_email == "alice@example.com"
assert auth.team_id == "team-corp"
assert auth.models == ["gpt-4", "gpt-3.5-turbo"]

139
tests/test_litellm/proxy/management_endpoints/test_ui_sso.py
View File

@ -1841,6 +1841,7 @@ class TestCustomUISSO:
"x-forwarded-for": "192.168.1.1",
}
mock_request.base_url = "https://test.litellm.ai/"
mock_request.client.host = "10.0.0.10"
# Mock the custom handler
mock_custom_handler = MagicMock(spec=CustomSSOLoginHandler)
@ -1866,36 +1867,73 @@ class TestCustomUISSO:
"litellm.proxy.proxy_server.user_custom_ui_sso_sign_in_handler",
mock_custom_handler,
):
with patch.object(
SSOAuthenticationHandler,
"get_redirect_response_from_openid",
return_value=mock_redirect_response,
) as mock_get_redirect:
# Act
result = (
with patch(
"litellm.proxy.proxy_server.general_settings",
{"trusted_proxy_ranges": ["10.0.0.0/24"]},
):
with patch.object(
SSOAuthenticationHandler,
"get_redirect_response_from_openid",
return_value=mock_redirect_response,
) as mock_get_redirect:
# Act
result = await EnterpriseCustomSSOHandler.handle_custom_ui_sso_sign_in(
request=mock_request
)
# Assert
# Verify the custom handler was called with the request
mock_custom_handler.handle_custom_ui_sso_sign_in.assert_called_once_with(
request=mock_request
)
# Verify the redirect response was generated with correct OpenID
mock_get_redirect.assert_called_once_with(
result=expected_openid,
request=mock_request,
received_response=None,
generic_client_id=None,
ui_access_mode=None,
)
# Verify the result is the redirect response
assert result == mock_redirect_response
assert result.status_code == 303
@pytest.mark.asyncio
async def test_handle_custom_ui_sso_sign_in_rejects_untrusted_proxy(self):
"""Custom UI SSO rejects spoofed identity headers from direct clients."""
from enterprise.litellm_enterprise.proxy.auth.custom_sso_handler import (
EnterpriseCustomSSOHandler,
)
from litellm.integrations.custom_sso_handler import CustomSSOLoginHandler
mock_request = MagicMock(spec=Request)
mock_request.headers = {
"x-litellm-user-id": "admin",
"x-litellm-user-email": "admin@example.com",
}
mock_request.base_url = "https://test.litellm.ai/"
mock_request.client.host = "203.0.113.10"
mock_custom_handler = MagicMock(spec=CustomSSOLoginHandler)
mock_custom_handler.handle_custom_ui_sso_sign_in = AsyncMock()
with patch("litellm.proxy.proxy_server.premium_user", True):
with patch(
"litellm.proxy.proxy_server.user_custom_ui_sso_sign_in_handler",
mock_custom_handler,
):
with patch(
"litellm.proxy.proxy_server.general_settings",
{"trusted_proxy_ranges": ["10.0.0.0/24"]},
):
with pytest.raises(ValueError, match="not trusted"):
await EnterpriseCustomSSOHandler.handle_custom_ui_sso_sign_in(
request=mock_request
)
)
# Assert
# Verify the custom handler was called with the request
mock_custom_handler.handle_custom_ui_sso_sign_in.assert_called_once_with(
request=mock_request
)
# Verify the redirect response was generated with correct OpenID
mock_get_redirect.assert_called_once_with(
result=expected_openid,
request=mock_request,
received_response=None,
generic_client_id=None,
ui_access_mode=None,
)
# Verify the result is the redirect response
assert result == mock_redirect_response
assert result.status_code == 303
mock_custom_handler.handle_custom_ui_sso_sign_in.assert_not_called()
@pytest.mark.asyncio
async def test_custom_ui_sso_handler_execution_with_real_class(self):
@ -1946,6 +1984,7 @@ class TestCustomUISSO:
"x-forwarded-for": "10.0.0.1",
}
mock_request.base_url = "https://custom.litellm.ai/"
mock_request.client.host = "10.0.0.20"
# Mock the redirect response method
mock_redirect_response = MagicMock()
@ -1956,34 +1995,36 @@ class TestCustomUISSO:
"litellm.proxy.proxy_server.user_custom_ui_sso_sign_in_handler",
test_handler_instance,
):
with patch.object(
SSOAuthenticationHandler,
"get_redirect_response_from_openid",
return_value=mock_redirect_response,
) as mock_get_redirect:
# Act
result = (
await EnterpriseCustomSSOHandler.handle_custom_ui_sso_sign_in(
with patch(
"litellm.proxy.proxy_server.general_settings",
{"trusted_proxy_ranges": ["10.0.0.0/24"]},
):
with patch.object(
SSOAuthenticationHandler,
"get_redirect_response_from_openid",
return_value=mock_redirect_response,
) as mock_get_redirect:
# Act
result = await EnterpriseCustomSSOHandler.handle_custom_ui_sso_sign_in(
request=mock_request
)
)
# Assert that our custom handler was executed
assert test_handler_instance.method_called is True
assert test_handler_instance.received_request == mock_request
# Assert that our custom handler was executed
assert test_handler_instance.method_called is True
assert test_handler_instance.received_request == mock_request
# Verify the redirect response was called with the OpenID from our custom handler
mock_get_redirect.assert_called_once()
call_args = mock_get_redirect.call_args.kwargs
# Verify the redirect response was called with the OpenID from our custom handler
mock_get_redirect.assert_called_once()
call_args = mock_get_redirect.call_args.kwargs
# Verify the OpenID object has the expected values from our custom handler
openid_result = call_args["result"]
assert openid_result.id == "custom_test_user_456"
assert openid_result.email == "custom@example.com"
assert openid_result.first_name == "Custom"
assert openid_result.last_name == "Handler"
assert openid_result.display_name == "Custom Handler Test"
assert openid_result.provider == "custom"
# Verify the OpenID object has the expected values from our custom handler
openid_result = call_args["result"]
assert openid_result.id == "custom_test_user_456"
assert openid_result.email == "custom@example.com"
assert openid_result.first_name == "Custom"
assert openid_result.last_name == "Handler"
assert openid_result.display_name == "Custom Handler Test"
assert openid_result.provider == "custom"
# Verify the request and other parameters were passed correctly
assert call_args["request"] == mock_request