scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-15 09:12:18 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 311

Merge 06827e398a into baddc67041

Browse source
This commit is contained in:
yujonglee 2026-11-03 13:09:44 +00:00 • committed by GitHub
commit c706188580
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
19 changed files with 522 additions and 252 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
litellm/proxy/_types.py
View file

@ -3019,7 +3019,7 @@ class ConfigGeneralSettings(LiteLLMPydanticObjectBase):
)
trusted_proxy_ranges: list[str] | None = 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, and whose X-Forwarded-For is used to attribute Admin UI sign-in attempts to a source address. Set it to an empty list when clients connect directly, so the peer address is the source. Left unset, or containing an entry that is not an address or CIDR range, the per-source sign-in limit is off.",
description="CIDR ranges of trusted reverse proxies. Their X-Forwarded-For and X-Forwarded-Proto are resolved once by ASGI middleware for auth, allowed_ips, sign-in throttling and requester IP logging. Identity-header auth still checks the original peer. Set [] for direct connections. When unset, use_x_forwarded_for retains deprecated raw-header behavior for legacy IP consumers and the per-source sign-in limit is off. Invalid ranges are ignored, and disable the per-source sign-in limit. Disable server-level proxy-header rewriting when launching an external ASGI server. MCP keeps its separate trust settings.",
)
store_model_in_db: bool | None = Field(
None,

17
litellm/proxy/auth/auth_utils.py
View file

@ -95,15 +95,16 @@ def mark_invalid_virtual_key_error(exception: ProxyException, is_invalid_virtual
def get_request_ip_address(request: Request, use_x_forwarded_for: bool | None = False) -> str | None:
client_ip = None
if use_x_forwarded_for is True and "x-forwarded-for" in request.headers:
client_ip = request.headers["x-forwarded-for"]
elif request.client is not None:
client_ip = request.client.host
else:
client_ip = ""
from litellm.proxy.auth.network import proxy_header_context
return client_ip
context: Final = proxy_header_context(request)
if (
(context is None or not context.configured)
and use_x_forwarded_for is True
and "x-forwarded-for" in request.headers
):
return request.headers["x-forwarded-for"]
return request.client.host if request.client else ""
_get_request_ip_address: Final = get_request_ip_address

18
litellm/proxy/auth/ip_address_utils.py
View file

@ -20,6 +20,7 @@ from litellm.proxy.auth.auth_utils import ( # noqa: F401 # legacy module expor
_get_request_ip_address, # pyright: ignore[reportPrivateUsage,reportUnusedImport] # backwards-compatible package export
get_request_ip_address,
)
from litellm.proxy.auth.network import get_direct_client_ip, get_direct_request, proxy_header_context
# One-shot warning so operators upgrading from the prior "always trust X-Forwarded-*"
# behaviour see an actionable message in their logs the first time it triggers.
@ -192,7 +193,7 @@ class IPAddressUtils:
_warned_xff_without_trusted_ranges = True
return False
direct_ip: Final = request.client.host if request.client else None
direct_ip: Final = get_direct_client_ip(request)
trusted_networks: Final = IPAddressUtils.parse_trusted_proxy_networks(trusted_ranges)
return IPAddressUtils.is_trusted_proxy(direct_ip, trusted_networks)
@ -209,22 +210,27 @@ class IPAddressUtils:
reverse proxy, so ``request.url.scheme`` alone cannot answer this in
that deployment shape. Resolved from the first trusted signal:
1. ``PROXY_BASE_URL`` (operator-declared public origin).
2. ``X-Forwarded-Proto``, only when the request's direct peer is a
2. Middleware-resolved scheme when ``trusted_proxy_ranges`` is set.
3. Legacy ``X-Forwarded-Proto``, only when the request's direct peer is a
configured trusted proxy -- see ``is_request_from_trusted_proxy``.
An untrusted caller cannot spoof this header to strip Secure.
3. The request's own literal scheme (direct TLS termination, or no
4. The request's own literal scheme (direct TLS termination, or no
reverse proxy in front of litellm).
"""
configured_base_url: Final = os.environ.get("PROXY_BASE_URL", "").strip()
if configured_base_url:
return urlparse(configured_base_url).scheme == "https"
context: Final = proxy_header_context(request)
if context is not None and context.configured:
return request.url.scheme == "https"
if IPAddressUtils.is_request_from_trusted_proxy(request, general_settings=general_settings):
forwarded_proto: Final = request.headers.get("X-Forwarded-Proto")
if forwarded_proto:
return forwarded_proto.split(",")[0].strip().lower() == "https"
return request.url.scheme == "https"
return get_direct_request(request).url.scheme == "https"
@staticmethod
def extract_client_ip_from_xff_hops(
@ -343,7 +349,7 @@ class IPAddressUtils:
# If XFF is enabled, validate the request comes from a trusted proxy
if use_xff and "x-forwarded-for" in request.headers:
if not IPAddressUtils.is_request_from_trusted_proxy(request, general_settings=general_settings):
direct_ip: Final = request.client.host if request.client else None
direct_ip: Final = get_direct_client_ip(request)
if general_settings.get("mcp_trusted_proxy_ranges"):
# Direct connection isn't in any configured trusted CIDR.
verbose_proxy_logger.warning("XFF header from untrusted IP %s, ignoring", direct_ip)
@ -371,4 +377,4 @@ class IPAddressUtils:
return client_ip
case _HopCountUnset():
pass
return get_request_ip_address(request, use_x_forwarded_for=use_xff is True)
return get_request_ip_address(get_direct_request(request), use_x_forwarded_for=use_xff is True)

5
litellm/proxy/auth/login_throttle.py
View file

@ -36,7 +36,6 @@ from litellm.constants import (
LOGIN_THROTTLE_UNKNOWN_SOURCE,
)
from litellm.proxy._types import ProxyErrorTypes, ProxyException
from litellm.proxy.auth.network import TrustedProxyConfig, resolve_client_ip
from litellm.secret_managers.main import get_secret_bool
_LOGIN_THROTTLE_TARGET: Final = "login_throttle"
@ -295,9 +294,7 @@ class LoginThrottle:
) -> LoginThrottle:
settings: Final[Mapping[str, object]] = general_settings if general_settings is not None else EMPTY_MAPPING
proxies: Final = declared_proxy_ranges(settings)
resolved, _ = resolve_client_ip(
request, TrustedProxyConfig(use_forwarded_for=bool(proxies), trusted_proxy_cidrs=proxies or ())
)
resolved: Final = request.client.host if request.client else None
source_limit: Final = _source_limit(settings, resolved or LOGIN_THROTTLE_UNKNOWN_SOURCE)
exempt: Final = source_limit == EXEMPT
return cls(

68
litellm/proxy/auth/network.py
View file

@ -1,11 +1,10 @@
from __future__ import annotations
import ipaddress
from collections.abc import Sequence
from dataclasses import dataclass
from typing import Final
from fastapi import Request
from pydantic import Field
from litellm._logging import verbose_proxy_logger
from litellm.types.llms.base import LiteLLMBaseModel
@ -19,9 +18,35 @@ class NetworkContext(LiteLLMBaseModel):
via_trusted_proxy: bool = False
class TrustedProxyConfig(LiteLLMBaseModel):
use_forwarded_for: bool = False
trusted_proxy_cidrs: Sequence[str] = Field(default_factory=tuple)
PROXY_HEADER_CONTEXT: Final = "litellm.proxy_header_context"
@dataclass(frozen=True, slots=True)
class ProxyHeaderContext:
direct_client: tuple[str, int] | None
direct_scheme: str
configured: bool
via_trusted_proxy: bool
def proxy_header_context(request: Request) -> ProxyHeaderContext | None:
context: Final[object] = request.get(PROXY_HEADER_CONTEXT)
return context if isinstance(context, ProxyHeaderContext) else None
def get_direct_client_ip(request: Request) -> str | None:
context: Final = proxy_header_context(request)
if context is not None:
return context.direct_client[0] if context.direct_client else None
return request.client.host if request.client else None
def get_direct_request(request: Request) -> Request:
context: Final = proxy_header_context(request)
if context is None:
return request
scope: Final = {key: value for key, value in request.scope.items() if key != PROXY_HEADER_CONTEXT}
return Request({**scope, "client": context.direct_client, "scheme": context.direct_scheme})
def normalize_cidr_ranges(configured_ranges: object, *, setting_name: str = "trusted_proxy_cidrs") -> list[str]:
@ -68,35 +93,10 @@ def ip_in_networks(client_ip: str | None, networks: list[TrustedProxyNetwork]) -
return any(candidate in network for candidate in candidates for network in networks)
def _is_valid_ip(value: str) -> bool:
try:
ipaddress.ip_address(value)
return True
except ValueError:
return False
def resolve_client_ip(request: Request, config: TrustedProxyConfig) -> tuple[str | None, bool]:
"""Resolve the real client IP, trusting X-Forwarded-For only when the direct
peer is itself a configured trusted proxy. Walks the header right-to-left and
returns the first hop that is not a trusted proxy, so a forged left-most entry
cannot spoof the client."""
peer: Final = request.client.host if request.client else None
networks: Final = parse_trusted_proxy_ranges(config.trusted_proxy_cidrs)
if not config.use_forwarded_for or not ip_in_networks(peer, networks):
return peer, False
forwarded: Final = request.headers.get("x-forwarded-for", "")
hops: Final = [h.strip() for h in forwarded.split(",") if h.strip()]
for hop in reversed(hops):
if _is_valid_ip(hop) and not ip_in_networks(hop, networks):
return hop, True
return peer, True
def resolve_network_context(request: Request, config: TrustedProxyConfig) -> NetworkContext:
ip, via_proxy = resolve_client_ip(request, config)
def resolve_network_context(request: Request) -> NetworkContext:
context: Final = proxy_header_context(request)
return NetworkContext(
client_ip=ip,
client_ip=request.client.host if request.client else None,
host=request.headers.get("host"),
via_trusted_proxy=via_proxy,
via_trusted_proxy=context.via_trusted_proxy if context else False,
)

11
litellm/proxy/auth/trusted_proxy_utils.py
View file

@ -4,6 +4,7 @@ from fastapi import Request
from litellm._logging import verbose_proxy_logger
from litellm.proxy.auth.network import (
get_direct_client_ip,
ip_in_networks,
normalize_cidr_ranges,
parse_trusted_proxy_ranges,
@ -37,14 +38,6 @@ def get_trusted_proxy_cidrs(
)
def _get_direct_client_ip(request: Request) -> str | None:
client: Final = getattr(request, "client", None)
client_host: Final = getattr(client, "host", None)
if isinstance(client_host, str):
return client_host
return None
def require_trusted_proxy_request(
*,
request: Request,
@ -70,7 +63,7 @@ def require_trusted_proxy_request(
"trusting identity headers from an upstream proxy."
)
direct_client_ip: Final = _get_direct_client_ip(request)
direct_client_ip: Final = get_direct_client_ip(request)
if not ip_in_networks(direct_client_ip, trusted_networks):
verbose_proxy_logger.warning(
"%s rejected identity headers from untrusted direct client IP %r",

9
litellm/proxy/auth/user_api_key_auth.py
View file

@ -122,7 +122,7 @@ from litellm.proxy.auth.auth_utils import (
route_in_additonal_public_routes,
)
from litellm.proxy.auth.handle_jwt import JWTAuthManager, JWTHandler
from litellm.proxy.auth.network import TrustedProxyConfig, resolve_network_context
from litellm.proxy.auth.network import resolve_network_context
from litellm.proxy.auth.oauth2_check import Oauth2Handler
from litellm.proxy.auth.oauth2_proxy_hook import handle_oauth2_proxy_request
from litellm.proxy.auth.resolvers import CredentialRef, Principal
@ -137,7 +137,6 @@ from litellm.proxy.auth.resolvers.grants import (
from litellm.proxy.auth.resolvers.store import IdentityStore
from litellm.proxy.auth.route_checks import RouteChecks
from litellm.proxy.auth.team_grants import team_grants
from litellm.proxy.auth.trusted_proxy_utils import get_trusted_proxy_cidrs
from litellm.proxy.common_utils.cache_coordinator import EventDrivenCacheCoordinator
from litellm.proxy.common_utils.http_parsing_utils import ( # noqa: F401 # legacy module exports
_read_request_body, # pyright: ignore[reportPrivateUsage,reportUnusedImport] # backwards-compatible package export
@ -4099,11 +4098,7 @@ def _resolve_request_principal(request: Request, valid_token: UserAPIKeyAuth) ->
credential_ref and a stable subject fallback are always set off the token so
the Principal can never be anonymous, even for a keyless service-account key
with no user or alias."""
cidrs: Final = get_trusted_proxy_cidrs()
network: Final = resolve_network_context(
request,
TrustedProxyConfig(use_forwarded_for=bool(cidrs), trusted_proxy_cidrs=cidrs),
)
network: Final = resolve_network_context(request)
auth_method: Final = AuthMethod.BEARER_JWT if valid_token.jwt_claims else AuthMethod.API_KEY
return IdentityStore.principal_from_key(
valid_token,

22
litellm/proxy/litellm_pre_call_utils.py
View file

@ -75,7 +75,7 @@ from litellm.proxy._types import (
TeamCallbackMetadata,
UserAPIKeyAuth,
)
from litellm.proxy.auth.auth_utils import get_request_route
from litellm.proxy.auth.auth_utils import get_request_ip_address, get_request_route
from litellm.proxy.auth.route_checks import RouteChecks
from litellm.proxy.common_utils.callback_utils import (
decrypt_callback_vars,
@ -3020,22 +3020,10 @@ async def add_litellm_data_to_request(
data["allowed_model_region"] = user_api_key_dict.allowed_model_region
start_time: Final = time.time()
## [Enterprise Only]
# Add User-IP Address
requester_ip_address = ""
if True: # Always set the IP Address if available
# logic for tracking IP Address
# logic for tracking IP Address
if (
general_settings is not None
and general_settings.get("use_x_forwarded_for") is True
and hasattr(request, "headers")
and "x-forwarded-for" in request.headers
):
requester_ip_address = request.headers["x-forwarded-for"]
elif hasattr(request, "client") and hasattr(request.client, "host") and request.client is not None:
requester_ip_address = request.client.host
data[_metadata_variable_name]["requester_ip_address"] = requester_ip_address
data[_metadata_variable_name]["requester_ip_address"] = (
get_request_ip_address(request, use_x_forwarded_for=(general_settings or {}).get("use_x_forwarded_for") is True)
or ""
)
# Add User-Agent
user_agent = ""

4
litellm/proxy/mcp_server/oauth_utils.py
View file

@ -235,7 +235,9 @@ def get_request_base_url(request: Request) -> str:
if configured:
return configured
base_url: Final = str(request.base_url).rstrip("/")
from litellm.proxy.auth.network import get_direct_request
base_url: Final = str(get_direct_request(request).base_url).rstrip("/")
parsed: Final = urlparse(base_url)
if not IPAddressUtils.is_request_from_trusted_proxy(request):

81
litellm/proxy/middleware/proxy_headers_middleware.py Normal file
View file

@ -0,0 +1,81 @@
from collections.abc import Callable, Mapping
from typing import (
Final,
cast, # noqa: TID251 # uvicorn and Starlette have incompatible ASGI protocol stubs
)
from pydantic import TypeAdapter
from starlette.datastructures import Headers
from starlette.types import ASGIApp, Receive, Scope, Send
from uvicorn._types import ASGI3Application
from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware
from litellm._logging import verbose_proxy_logger
from litellm.proxy.auth.network import (
PROXY_HEADER_CONTEXT,
ProxyHeaderContext,
normalize_cidr_ranges,
parse_trusted_proxy_ranges,
)
_CLIENT: Final = TypeAdapter[tuple[str, int] | None](tuple[str, int] | None)
_SCHEME: Final = TypeAdapter[str](str)
class TrustedProxyMiddleware:
def __init__(self, app: ASGIApp, settings: Callable[[], Mapping[str, object]]) -> None:
self.app = app
self.settings = settings
self._ranges: tuple[str, ...] = ()
self._middleware: ProxyHeadersMiddleware | None = None
self._warned_legacy = False
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
if scope["type"] not in {"http", "websocket"}:
await self.app(scope, receive, send)
return
settings: Final = self.settings()
configured: Final = settings.get("trusted_proxy_ranges") is not None
ranges: Final = tuple(
normalize_cidr_ranges(settings.get("trusted_proxy_ranges"), setting_name="trusted_proxy_ranges")
)
if self._middleware is None or ranges != self._ranges:
hosts: Final = [
str(network) for network in parse_trusted_proxy_ranges(ranges, setting_name="trusted_proxy_ranges")
]
app: Final = cast(ASGI3Application, self.app) # cast-ok: equivalent ASGI protocols use different stubs
self._middleware = ProxyHeadersMiddleware(app, trusted_hosts=hosts)
self._ranges = ranges
if settings.get("use_x_forwarded_for") is True and not self._warned_legacy:
verbose_proxy_logger.warning(
"general_settings.use_x_forwarded_for is deprecated for client-IP resolution. "
"Set trusted_proxy_ranges to your reverse-proxy CIDRs, or [] for direct connections. "
"When trusted_proxy_ranges is unset, legacy raw X-Forwarded-For handling remains enabled. "
"MCP continues to use its separate proxy settings."
)
self._warned_legacy = True
client: Final = _CLIENT.validate_python(scope.get("client"))
peer: Final = client[0] if client else None
context: Final = ProxyHeaderContext(
direct_client=client,
direct_scheme=_SCHEME.validate_python(scope.get("scheme", "http" if scope["type"] == "http" else "ws")),
configured=configured,
via_trusted_proxy=peer in self._middleware.trusted_hosts,
)
forwarded_proto: Final = (
Headers(raw=scope.get("headers", [])).get("x-forwarded-proto", "").split(",", 1)[0].strip().lower()
)
scheme: Final = (
forwarded_proto
if context.via_trusted_proxy and forwarded_proto in {"http", "https"}
else context.direct_scheme
)
resolved_scope: Final = {
**scope,
"scheme": scheme.replace("http", "ws") if scope["type"] == "websocket" else scheme,
PROXY_HEADER_CONTEXT: context,
}
handler: Final = cast(ASGIApp, self._middleware) # cast-ok: equivalent ASGI protocol stubs
await handler(resolved_scope, receive, send)

1
litellm/proxy/proxy_cli.py
View file

@ -426,6 +426,7 @@ class ProxyInitializationHelpers:
"host": host,
"port": port,
"server_header": False,
"proxy_headers": False,
"timeout_graceful_shutdown": GracefulShutdownManager.get_timeout(),
}
if log_config is not None:

2
litellm/proxy/proxy_server.py
View file

@ -918,6 +918,7 @@ from litellm.proxy.middleware.per_request_root_path_middleware import (
get_server_root_paths,
)
from litellm.proxy.middleware.prometheus_auth_middleware import PrometheusAuthMiddleware
from litellm.proxy.middleware.proxy_headers_middleware import TrustedProxyMiddleware
from litellm.proxy.middleware.request_body_decompression_middleware import RequestBodyDecompressionMiddleware
from litellm.proxy.middleware.request_size_limit_middleware import (
RequestSizeLimitMiddleware,
@ -2993,6 +2994,7 @@ app.add_middleware(LensInternalMiddleware)
app.add_middleware(InFlightRequestsMiddleware)
app.add_middleware(SecurityHeadersMiddleware)
app.add_middleware(GZipBufferedResponseMiddleware)
app.add_middleware(TrustedProxyMiddleware, settings=lambda: general_settings_view())
def mount_swagger_ui():

6
litellm/proxy/shutdown/graceful_uvicorn_worker.py
View file

@ -4,7 +4,11 @@ from uvicorn.workers import UvicornWorker
def build_graceful_shutdown_uvicorn_worker(timeout_graceful_shutdown: float) -> type[UvicornWorker]:
class GracefulShutdownUvicornWorker(UvicornWorker):
def __init__(self, *args: object) -> None:
self.CONFIG_KWARGS = {**UvicornWorker.CONFIG_KWARGS, "timeout_graceful_shutdown": timeout_graceful_shutdown}
self.CONFIG_KWARGS = {
**UvicornWorker.CONFIG_KWARGS,
"timeout_graceful_shutdown": timeout_graceful_shutdown,
"proxy_headers": False,
}
super().__init__(*args)
return GracefulShutdownUvicornWorker

55
tests/unit/proxy/auth/test_login_utils.py
View file

@ -124,7 +124,7 @@ async def test_authenticate_user_admin_login_with_ui_credentials():
"litellm.proxy.auth.login_utils.user_update",
new_callable=AsyncMock,
return_value=None,
) as mock_user_update:
):
with patch(
"litellm.proxy.auth.login_utils.get_secret_bool",
return_value=False,
@ -182,7 +182,7 @@ async def test_authenticate_user_admin_login_with_master_key_as_password(monkeyp
"litellm.proxy.auth.login_utils.user_update",
new_callable=AsyncMock,
return_value=None,
) as mock_user_update:
):
with patch(
"litellm.proxy.auth.login_utils.get_secret_bool",
return_value=False,
@ -430,7 +430,7 @@ async def test_authenticate_user_admin_login_with_non_ascii_characters():
"litellm.proxy.auth.login_utils.user_update",
new_callable=AsyncMock,
return_value=None,
) as mock_user_update:
):
with patch(
"litellm.proxy.auth.login_utils.get_secret_bool",
return_value=False,
@ -465,7 +465,7 @@ def test_authenticate_user_non_ascii_direct_comparison():
assert result is True
# And correctly returns False for different passwords
result = secrets.compare_digest(password.encode("utf-8"), "different£pass".encode("utf-8"))
result = secrets.compare_digest(password.encode("utf-8"), "different£pass".encode())
assert result is False
@ -1018,18 +1018,12 @@ def test_declared_proxy_ranges_distinguishes_none_from_empty_from_configured():
@pytest.mark.asyncio
async def test_with_trusted_proxy_ranges_the_source_is_the_forwarded_client(monkeypatch):
"""The header is walked right to left past the trusted hops, so a forged left-most entry cannot pick the bucket."""
from litellm.proxy.auth.login_throttle import LoginThrottle
monkeypatch.setenv("UI_USERNAME", "admin")
monkeypatch.setenv("UI_PASSWORD", "right")
settings = {"trusted_proxy_ranges": ["10.0.0.0/8"], "max_failed_login_attempts_per_source": 2}
def _from(peer: str, forwarded: str):
request = MagicMock()
request.headers = {"x-forwarded-for": forwarded}
request.client = MagicMock()
request.client.host = peer
return LoginThrottle.from_request(request, general_settings=settings, redis_cache=None)
return _throttle_behind_trusted_proxy(forwarded, settings, peer=peer)
via_proxy = _from("10.0.0.1", "1.1.1.1, 203.0.113.9, 10.0.0.2")
assert via_proxy.client_ip == "203.0.113.9"
@ -1041,8 +1035,6 @@ async def test_with_trusted_proxy_ranges_the_source_is_the_forwarded_client(monk
def test_source_overrides_pick_the_most_specific_matching_range():
"""An exact address beats a /16 beats a /8; an address in none of them keeps the default."""
from litellm.proxy.auth.login_throttle import LoginThrottle
settings = {
"trusted_proxy_ranges": ["10.0.0.0/8"],
"max_failed_login_attempts_per_source": 7,
@ -1056,11 +1048,7 @@ def test_source_overrides_pick_the_most_specific_matching_range():
}
def _limit(client: str) -> int | None:
request = MagicMock()
request.headers = {"x-forwarded-for": client}
request.client = MagicMock()
request.client.host = "10.0.0.1"
return LoginThrottle.from_request(request, general_settings=settings, redis_cache=None).source_limit
return _throttle_behind_trusted_proxy(client, settings).source_limit
assert _limit("203.0.113.9") == 300
assert _limit("203.0.113.10") == 200
@ -1588,14 +1576,31 @@ def test_the_per_username_allowance_is_half_the_address_allowance_rounded_down_a
assert user_limit_for(source_limit) == expected_user_limit
def _throttle_behind_trusted_proxy(client_ip: str, settings: Mapping[str, object]) -> "LoginThrottle":
from litellm.proxy.auth.login_throttle import LoginThrottle
def _throttle_behind_trusted_proxy(
client_ip: str, settings: Mapping[str, object], *, peer: str = "10.0.0.1"
) -> "LoginThrottle":
from queue import SimpleQueue
request = MagicMock()
request.headers = {"x-forwarded-for": client_ip}
request.client = MagicMock()
request.client.host = "10.0.0.1"
return LoginThrottle.from_request(request, general_settings=settings, redis_cache=None)
from fastapi import Request
from starlette.applications import Starlette
from starlette.responses import Response
from starlette.routing import Route
from starlette.testclient import TestClient
from litellm.proxy.auth.login_throttle import LoginThrottle
from litellm.proxy.middleware.proxy_headers_middleware import TrustedProxyMiddleware
throttles: Final[SimpleQueue[LoginThrottle]] = SimpleQueue()
async def endpoint(request: Request) -> Response:
throttles.put(LoginThrottle.from_request(request, general_settings=settings, redis_cache=None))
return Response()
app: Final = Starlette(routes=[Route("/", endpoint)])
app.add_middleware(TrustedProxyMiddleware, settings=lambda: settings)
with TestClient(app, client=(peer, 5555)) as client:
client.get("/", headers={"x-forwarded-for": client_ip})
return throttles.get_nowait()
def test_a_per_address_override_also_raises_that_address_per_username_allowance():

157
tests/unit/proxy/auth/test_network.py
View file

@ -1,119 +1,64 @@
from __future__ import annotations
from typing import Any, Dict, List, Optional, Tuple
from typing import Final
from fastapi import Request
from litellm.proxy.auth.network import (
TrustedProxyConfig,
resolve_client_ip,
PROXY_HEADER_CONTEXT,
ProxyHeaderContext,
get_direct_client_ip,
get_direct_request,
resolve_network_context,
)
TRUSTED = TrustedProxyConfig(use_forwarded_for=True, trusted_proxy_cidrs=["10.0.0.0/8"])
def make_request(
*,
headers: Optional[Dict[str, str]] = None,
client: Optional[Tuple[str, int]] = ("203.0.113.7", 5555),
) -> Request:
raw_headers: List[Tuple[bytes, bytes]] = [
(key.lower().encode(), value.encode()) for key, value in (headers or {}).items()
]
scope: Dict[str, Any] = {
"type": "http",
"http_version": "1.1",
"method": "GET",
"path": "/",
"raw_path": b"/",
"query_string": b"",
"headers": raw_headers,
"client": client,
"server": ("testserver", 80),
"scheme": "http",
def test_network_context_reads_the_resolved_scope_without_reparsing_headers() -> None:
request: Final = Request(
{
"type": "http",
"client": ("203.0.113.9", 0),
"scheme": "https",
"headers": [(b"host", b"proxy.example"), (b"x-forwarded-for", b"198.51.100.4")],
PROXY_HEADER_CONTEXT: ProxyHeaderContext(("10.0.0.1", 5555), "http", True, True),
}
)
assert resolve_network_context(request).model_dump() == {
"client_ip": "203.0.113.9",
"host": "proxy.example",
"via_trusted_proxy": True,
}
return Request(scope)
assert get_direct_client_ip(request) == "10.0.0.1"
direct: Final = get_direct_request(request)
assert direct.client == ("10.0.0.1", 5555)
assert direct.scope["scheme"] == "http"
assert request.client == ("203.0.113.9", 0)
assert request.scope["scheme"] == "https"
def test_xff_ignored_when_forwarding_disabled():
config = TrustedProxyConfig(
use_forwarded_for=False, trusted_proxy_cidrs=["10.0.0.0/8"]
def test_missing_peer_stays_missing_after_forwarding() -> None:
request: Final = Request(
{
"type": "http",
"client": None,
"headers": [],
"scheme": "http",
PROXY_HEADER_CONTEXT: ProxyHeaderContext(None, "http", True, False),
}
)
request = make_request(
headers={"x-forwarded-for": "203.0.113.9"}, client=("10.0.0.1", 1)
assert get_direct_client_ip(request) is None
assert resolve_network_context(request).client_ip is None
def test_request_without_middleware_uses_the_peer() -> None:
request: Final = Request(
{
"type": "http",
"client": ("203.0.113.7", 5555),
"headers": [(b"x-forwarded-for", b"198.51.100.4")],
}
)
ip, via_proxy = resolve_client_ip(request, config)
assert ip == "10.0.0.1"
assert via_proxy is False
def test_xff_honored_from_trusted_peer():
request = make_request(
headers={"x-forwarded-for": "203.0.113.9, 10.0.0.5"}, client=("10.0.0.1", 1)
)
ip, via_proxy = resolve_client_ip(request, TRUSTED)
assert ip == "203.0.113.9"
assert via_proxy is True
def test_ipv4_mapped_peer_and_hop_match_ipv4_trusted_ranges():
request = make_request(headers={"x-forwarded-for": "203.0.113.9, ::ffff:10.0.0.5"}, client=("::ffff:10.0.0.1", 1))
ip, via_proxy = resolve_client_ip(request, TRUSTED)
assert ip == "203.0.113.9"
assert via_proxy is True
def test_ipv4_mapped_peer_still_matches_mapped_notation_trusted_range():
config = TrustedProxyConfig(use_forwarded_for=True, trusted_proxy_cidrs=["::ffff:10.0.0.0/104"])
request = make_request(headers={"x-forwarded-for": "203.0.113.9"}, client=("::ffff:10.0.0.1", 1))
ip, via_proxy = resolve_client_ip(request, config)
assert ip == "203.0.113.9"
assert via_proxy is True
def test_spoofed_xff_from_untrusted_peer_is_ignored():
request = make_request(
headers={"x-forwarded-for": "203.0.113.9"}, client=("8.8.8.8", 1)
)
ip, via_proxy = resolve_client_ip(request, TRUSTED)
assert ip == "8.8.8.8"
assert via_proxy is False
def test_right_to_left_parse_skips_chained_trusted_proxies():
request = make_request(
headers={"x-forwarded-for": "198.51.100.4, 10.1.1.1, 10.0.0.9"},
client=("10.0.0.1", 1),
)
ip, via_proxy = resolve_client_ip(request, TRUSTED)
assert ip == "198.51.100.4"
assert via_proxy is True
def test_all_trusted_hops_fall_back_to_peer():
request = make_request(
headers={"x-forwarded-for": "10.1.1.1, 10.0.0.9"}, client=("10.0.0.1", 1)
)
ip, via_proxy = resolve_client_ip(request, TRUSTED)
assert ip == "10.0.0.1"
assert via_proxy is True
def test_invalid_xff_token_is_skipped():
request = make_request(
headers={"x-forwarded-for": "not-an-ip, 203.0.113.50"}, client=("10.0.0.1", 1)
)
ip, _ = resolve_client_ip(request, TRUSTED)
assert ip == "203.0.113.50"
def test_network_context_captures_host_and_proxy_flag():
request = make_request(
headers={"x-forwarded-for": "203.0.113.9", "host": "proxy.litellm.ai"},
client=("10.0.0.1", 1),
)
ctx = resolve_network_context(request, TRUSTED)
assert ctx.client_ip == "203.0.113.9"
assert ctx.host == "proxy.litellm.ai"
assert ctx.via_trusted_proxy is True
assert resolve_network_context(request).model_dump() == {
"client_ip": "203.0.113.7",
"host": None,
"via_trusted_proxy": False,
}
assert get_direct_request(request) is request

27
tests/unit/proxy/auth/test_user_api_key_auth.py
View file

@ -5,7 +5,7 @@
import litellm.proxy
import litellm.proxy.proxy_server
from typing import Dict, Final, List, Mapping, Optional
from typing import Final, List, Mapping, Optional
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
@ -20,19 +20,12 @@ from litellm.proxy.auth.user_api_key_auth import (
UserAPIKeyAuth,
get_api_key_from_custom_header,
)
from fastapi import WebSocket, HTTPException, status
from fastapi import Request, WebSocket, HTTPException, status
from litellm.proxy._types import LiteLLM_UserTable, LitellmUserRoles
from tests._master_key import MASTER_KEY
class Request:
def __init__(self, client_ip: Optional[str] = None, headers: Optional[dict] = None):
self.client = MagicMock()
self.client.host = client_ip
self.headers: Dict[str, str] = {}
@pytest.mark.parametrize(
"allowed_ips, client_ip, expected_result",
[
@ -51,9 +44,11 @@ class Request:
def test_check_valid_ip(allowed_ips: Optional[List[str]], client_ip: Optional[str], expected_result: bool):
from litellm.proxy.auth.auth_utils import _check_valid_ip
request = Request(client_ip)
request: Final = Request(
{"type": "http", "client": (client_ip, 1234) if client_ip else None, "headers": []}
)
assert _check_valid_ip(allowed_ips, request)[0] == expected_result # type: ignore
assert _check_valid_ip(allowed_ips, request)[0] == expected_result
# test x-forwarder for is used when user has opted in
@ -79,9 +74,15 @@ def test_check_valid_ip_sent_with_x_forwarded_for(
):
from litellm.proxy.auth.auth_utils import _check_valid_ip
request = Request(client_ip, headers={"X-Forwarded-For": client_ip})
request: Final = Request(
{
"type": "http",
"client": ("192.0.2.1", 1234),
"headers": [(b"x-forwarded-for", client_ip.encode())] if client_ip else [],
}
)
assert _check_valid_ip(allowed_ips, request, use_x_forwarded_for=True)[0] == expected_result # type: ignore
assert _check_valid_ip(allowed_ips, request, use_x_forwarded_for=True)[0] == expected_result
@pytest.mark.asyncio

38
tests/unit/proxy/management_endpoints/test_saml_sso.py
View file

@ -771,22 +771,22 @@ async def test_read_acs_post_data_rejects_oversized_stream_without_content_lengt
assert exc.value.status_code == 413
def _fake_request_with_scheme(scheme, headers=None, client_host="203.0.113.5"):
"""A fuller fake Request than ``_fake_request``: adds ``url``, ``headers`` and
``client``, which ``IPAddressUtils.is_request_https`` reads directly instead of
going through ``PROXY_BASE_URL``."""
return type(
"Req",
(),
def _request_with_scheme(
scheme: str, headers: dict[str, str] | None = None, client_host: str = "203.0.113.5"
) -> Request:
return Request(
{
"base_url": URL(f"{scheme}://proxy.example.com/"),
"url": URL(f"{scheme}://proxy.example.com/sso/saml/login"),
"query_params": {},
"cookies": {},
"headers": headers or {},
"client": type("Client", (), {"host": client_host})(),
},
)()
"type": "http",
"scheme": scheme,
"path": "/sso/saml/login",
"query_string": b"",
"headers": [
(key.lower().encode(), value.encode())
for key, value in {"host": "proxy.example.com", **(headers or {})}.items()
],
"client": (client_host, 1234),
}
)
class TestSAMLAuthnCookieSecureFlag:
@ -798,7 +798,7 @@ class TestSAMLAuthnCookieSecureFlag:
async def test_secure_over_direct_https(self, saml_env, monkeypatch):
monkeypatch.delenv("PROXY_BASE_URL", raising=False)
cache = DualCache()
request = _fake_request_with_scheme("https")
request = _request_with_scheme("https")
redirect = await SAMLAuthHandler.build_login_redirect(request, cache)
cookie = redirect.headers["set-cookie"]
assert "Secure" in cookie
@ -808,7 +808,7 @@ class TestSAMLAuthnCookieSecureFlag:
async def test_not_secure_over_direct_http(self, saml_env, monkeypatch):
monkeypatch.delenv("PROXY_BASE_URL", raising=False)
cache = DualCache()
request = _fake_request_with_scheme("http")
request = _request_with_scheme("http")
redirect = await SAMLAuthHandler.build_login_redirect(request, cache)
cookie = redirect.headers["set-cookie"]
assert "Secure" not in cookie
@ -825,7 +825,7 @@ class TestSAMLAuthnCookieSecureFlag:
{"use_x_forwarded_for": True, "mcp_trusted_proxy_ranges": ["10.0.0.0/8"]},
)
cache = DualCache()
request = _fake_request_with_scheme("http", headers={"X-Forwarded-Proto": "https"}, client_host="10.0.0.5")
request = _request_with_scheme("http", headers={"X-Forwarded-Proto": "https"}, client_host="10.0.0.5")
redirect = await SAMLAuthHandler.build_login_redirect(request, cache)
cookie = redirect.headers["set-cookie"]
assert "Secure" in cookie
@ -835,7 +835,7 @@ class TestSAMLAuthnCookieSecureFlag:
monkeypatch.delenv("PROXY_BASE_URL", raising=False)
monkeypatch.setattr("litellm.proxy.proxy_server.general_settings", {})
cache = DualCache()
request = _fake_request_with_scheme("http", headers={"X-Forwarded-Proto": "https"}, client_host="203.0.113.5")
request = _request_with_scheme("http", headers={"X-Forwarded-Proto": "https"}, client_host="203.0.113.5")
redirect = await SAMLAuthHandler.build_login_redirect(request, cache)
cookie = redirect.headers["set-cookie"]
assert "Secure" not in cookie

249
tests/unit/proxy/middleware/test_proxy_headers_middleware.py Normal file
View file

@ -0,0 +1,249 @@
from collections.abc import AsyncIterator, Callable, Mapping
from contextlib import asynccontextmanager
from queue import SimpleQueue
from typing import Final
import pytest
from fastapi import Request
from starlette.applications import Starlette
from starlette.datastructures import URL
from starlette.responses import JSONResponse
from starlette.routing import Route, WebSocketRoute
from starlette.testclient import TestClient
from starlette.types import Message, Receive, Scope, Send
from starlette.websockets import WebSocket
from litellm.proxy.auth.auth_utils import _check_valid_ip, get_request_ip_address
from litellm.proxy.auth.ip_address_utils import IPAddressUtils
from litellm.proxy.auth.login_throttle import LoginThrottle
from litellm.proxy.auth.network import get_direct_client_ip, resolve_network_context
from litellm.proxy.auth.trusted_proxy_utils import require_trusted_proxy_request
from litellm.proxy.middleware.proxy_headers_middleware import TrustedProxyMiddleware
@pytest.mark.asyncio
@pytest.mark.parametrize("scope_type,expected", [("http", "http://testserver/"), ("websocket", "ws://testserver/")])
async def test_omitted_scheme_reaches_application_with_protocol_default(scope_type: str, expected: str) -> None:
urls: Final[SimpleQueue[str]] = SimpleQueue()
async def endpoint(scope: Scope, receive: Receive, send: Send) -> None:
urls.put(str(URL(scope=scope)))
async def receive() -> Message:
raise AssertionError("Application does not read messages")
async def send(message: Message) -> None:
raise AssertionError("Application does not send messages")
middleware: Final = TrustedProxyMiddleware(endpoint, settings=lambda: {})
await middleware({"type": scope_type, "path": "/", "headers": [(b"host", b"testserver")]}, receive, send)
assert urls.get_nowait() == expected
def client_for(settings: Callable[[], Mapping[str, object]], peer: str = "10.0.0.1") -> TestClient:
async def endpoint(request: Request) -> JSONResponse:
config: Final = dict(settings())
try:
require_trusted_proxy_request(request=request, general_settings=config, feature_name="test auth")
identity_trusted = True
except ValueError:
identity_trusted = False
throttle: Final = LoginThrottle.from_request(request, config, None)
return JSONResponse(
{
"network": resolve_network_context(request).model_dump(),
"ip": get_request_ip_address(request, use_x_forwarded_for=config.get("use_x_forwarded_for") is True),
"allowed": _check_valid_ip(["203.0.113.9"], request, use_x_forwarded_for=True)[0],
"peer": get_direct_client_ip(request),
"scheme": request.url.scheme,
"forwarded_proto": request.headers.get("x-forwarded-proto"),
"identity_trusted": identity_trusted,
"login_ip": throttle.client_ip,
"source_limit": throttle.source_limit,
"mcp_ip": IPAddressUtils.get_mcp_client_ip(request, config),
"secure_cookie": IPAddressUtils.is_request_https(request, config),
}
)
app: Final = Starlette(routes=[Route("/", endpoint)])
app.add_middleware(TrustedProxyMiddleware, settings=settings)
return TestClient(app, client=(peer, 5555))
@pytest.mark.parametrize(
"ranges,peer,xff,expected,trusted",
[
(None, "10.0.0.1", "203.0.113.9", "10.0.0.1", False),
([], "10.0.0.1", "203.0.113.9", "10.0.0.1", False),
(["10.0.0.0/8"], "192.0.2.1", "203.0.113.9", "192.0.2.1", False),
(["10.0.0.0/8"], "10.0.0.1", "198.51.100.4, 203.0.113.9, 10.0.0.5", "203.0.113.9", True),
(["10.0.0.1/8"], "10.0.0.1", "203.0.113.9", "203.0.113.9", True),
(["10.0.0.0/8"], "10.0.0.1", "10.1.1.1, 10.0.0.9", "10.1.1.1", True),
(["10.0.0.0/8"], "10.0.0.1", "", "10.0.0.1", True),
(["10.0.0.0/8"], "10.0.0.1", "203.0.113.9, invalid, 10.0.0.5", "invalid", True),
(["invalid", "*"], "10.0.0.1", "203.0.113.9", "10.0.0.1", False),
(["2001:db8::/32"], "2001:db8::1", "203.0.113.9, 2001:db8::2", "203.0.113.9", True),
(["10.0.0.0/8"], "::ffff:10.0.0.1", "203.0.113.9", "::ffff:10.0.0.1", False),
(["::ffff:10.0.0.0/104"], "::ffff:10.0.0.1", "203.0.113.9", "203.0.113.9", True),
],
)
def test_trust_chain_resolves_once_for_ip_consumers(
ranges: list[str] | None,
peer: str,
xff: str,
expected: str,
trusted: bool,
) -> None:
client: Final = client_for(lambda: {"trusted_proxy_ranges": ranges}, peer)
result: Final = client.get("/", headers={"x-forwarded-for": xff, "x-forwarded-proto": "https"}).json()
assert result["network"] == {"client_ip": expected, "host": "testserver", "via_trusted_proxy": trusted}
assert result["ip"] == expected
assert result["login_ip"] == expected
assert result["peer"] == peer
assert result["scheme"] == ("https" if trusted else "http")
def test_identity_headers_check_the_original_peer_not_the_resolved_client() -> None:
client: Final = client_for(lambda: {"trusted_proxy_ranges": ["10.0.0.0/8"], "use_x_forwarded_for": True})
result: Final = client.get("/", headers={"x-forwarded-for": "198.51.100.4, 203.0.113.9, 10.0.0.5"}).json()
assert result["identity_trusted"] is True
assert result["allowed"] is True
assert result["ip"] == "203.0.113.9"
def test_untrusted_peer_cannot_supply_identity_or_an_allowlisted_ip() -> None:
client: Final = client_for(
lambda: {"trusted_proxy_ranges": ["10.0.0.0/8"], "use_x_forwarded_for": True},
"192.0.2.1",
)
result: Final = client.get(
"/", headers={"x-forwarded-for": "203.0.113.9", "x-litellm-rust-peer": "10.0.0.1"}
).json()
assert result["identity_trusted"] is False
assert result["allowed"] is False
assert result["ip"] == "192.0.2.1"
@pytest.mark.parametrize("ranges,expected", [(None, "203.0.113.9"), ([], "10.0.0.1")])
def test_legacy_flag_only_trusts_raw_headers_when_ranges_are_unset(
ranges: list[str] | None,
expected: str,
caplog: pytest.LogCaptureFixture,
) -> None:
client: Final = client_for(lambda: {"trusted_proxy_ranges": ranges, "use_x_forwarded_for": True})
assert client.get("/", headers={"x-forwarded-for": "203.0.113.9"}).json()["ip"] == expected
assert client.get("/", headers={"x-forwarded-for": "203.0.113.9"}).json()["ip"] == expected
assert caplog.text.count("deprecated for client-IP resolution") == 1
@pytest.mark.parametrize("mcp_ranges,expected", [(["10.0.0.0/8"], "198.51.100.4"), (["192.0.2.0/24"], "10.0.0.1")])
def test_mcp_keeps_its_own_trust_and_hop_policy(mcp_ranges: list[str], expected: str) -> None:
client: Final = client_for(
lambda: {
"trusted_proxy_ranges": ["10.0.0.0/8"],
"use_x_forwarded_for": True,
"mcp_trusted_proxy_ranges": mcp_ranges,
"mcp_xff_num_trusted_hops": 2,
}
)
result: Final = client.get(
"/", headers={"x-forwarded-for": "198.51.100.4, 203.0.113.9", "x-forwarded-proto": "https"}
).json()
assert result["ip"] == "203.0.113.9"
assert result["mcp_ip"] == expected
assert result["secure_cookie"] is True
def test_mcp_without_forwarding_uses_original_peer_but_ui_cookies_use_resolved_scheme() -> None:
client: Final = client_for(lambda: {"trusted_proxy_ranges": ["10.0.0.0/8"]})
result: Final = client.get("/", headers={"x-forwarded-for": "203.0.113.9", "x-forwarded-proto": "https"}).json()
assert result["ip"] == "203.0.113.9"
assert result["mcp_ip"] == "10.0.0.1"
assert result["secure_cookie"] is True
@pytest.mark.parametrize(
"peer,proto,direct_scheme,expected",
[
("10.0.0.1", "https, http", "http", "https"),
("10.0.0.1", " HTTPS , http", "http", "https"),
("10.0.0.1", "http, https", "http", "http"),
("10.0.0.1", "invalid, https", "http", "http"),
("192.0.2.1", "https, http", "http", "http"),
("192.0.2.1", "http, https", "https", "https"),
],
)
def test_forwarded_protocol_chains_preserve_cookie_security(
peer: str, proto: str, direct_scheme: str, expected: str, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.delenv("PROXY_BASE_URL", raising=False)
client: Final = client_for(lambda: {"trusted_proxy_ranges": ["10.0.0.0/8"]}, peer)
result: Final = client.get(f"{direct_scheme}://testserver/", headers={"x-forwarded-proto": proto}).json()
assert result["scheme"] == expected
assert result["secure_cookie"] is (expected == "https")
assert result["forwarded_proto"] == proto
@pytest.mark.parametrize(
"mcp_ranges,expected", [(["10.0.0.0/8"], "https://public.example"), (["192.0.2.0/24"], "http://testserver")]
)
def test_mcp_discovery_origin_uses_its_own_trust_policy(
mcp_ranges: list[str],
expected: str,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from litellm.proxy import proxy_server
from litellm.proxy.mcp_server.oauth_utils import get_request_base_url
settings: Final = {
"trusted_proxy_ranges": ["10.0.0.0/8"],
"use_x_forwarded_for": True,
"mcp_trusted_proxy_ranges": mcp_ranges,
}
monkeypatch.setattr(proxy_server, "general_settings", settings)
monkeypatch.delenv("PROXY_BASE_URL", raising=False)
async def endpoint(request: Request) -> JSONResponse:
return JSONResponse({"origin": get_request_base_url(request), "scheme": request.url.scheme})
app: Final = Starlette(routes=[Route("/", endpoint)])
app.add_middleware(TrustedProxyMiddleware, settings=lambda: settings)
client: Final = TestClient(app, client=("10.0.0.1", 5555))
assert client.get(
"/",
headers={
"x-forwarded-for": "203.0.113.9",
"x-forwarded-proto": "https",
"x-forwarded-host": "public.example",
},
).json() == {"origin": expected, "scheme": "https"}
def test_config_refresh_does_not_reuse_stale_trust() -> None:
configs: Final = iter(({"trusted_proxy_ranges": ["10.0.0.0/8"]}, {"trusted_proxy_ranges": []}))
app: Final = Starlette(routes=[Route("/", lambda request: JSONResponse({"ip": request.client.host}))])
app.add_middleware(TrustedProxyMiddleware, settings=lambda: next(configs))
client: Final = TestClient(app, client=("10.0.0.1", 5555))
assert client.get("/", headers={"x-forwarded-for": "203.0.113.9"}).json() == {"ip": "203.0.113.9"}
assert client.get("/", headers={"x-forwarded-for": "203.0.113.9"}).json() == {"ip": "10.0.0.1"}
@pytest.mark.parametrize("proto", ["https", "https, http"])
def test_websockets_resolve_client_and_scheme_after_lifespan_configuration(proto: str) -> None:
@asynccontextmanager
async def lifespan(app: Starlette) -> AsyncIterator[None]:
app.state.proxy_ranges = ["10.0.0.0/8"]
yield
async def endpoint(websocket: WebSocket) -> None:
await websocket.accept()
await websocket.send_json({"client": websocket.scope["client"], "scheme": websocket.scope["scheme"]})
await websocket.close()
app: Final = Starlette(routes=[WebSocketRoute("/", endpoint)], lifespan=lifespan)
app.add_middleware(TrustedProxyMiddleware, settings=lambda: {"trusted_proxy_ranges": app.state.proxy_ranges})
with TestClient(app, client=("10.0.0.1", 5555)) as client:
with client.websocket_connect(
"/", headers={"x-forwarded-for": "203.0.113.9", "x-forwarded-proto": proto}
) as ws:
assert ws.receive_json() == {"client": ["203.0.113.9", 0], "scheme": "wss"}

2
ui/litellm-dashboard/src/lib/http/schema.d.ts generated vendored
View file

@ -30809,7 +30809,7 @@ export interface components {
transcribe_media_buckets?: string[] | null;
/**
* Trusted Proxy Ranges
* @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, and whose X-Forwarded-For is used to attribute Admin UI sign-in attempts to a source address. Set it to an empty list when clients connect directly, so the peer address is the source. Left unset, or containing an entry that is not an address or CIDR range, the per-source sign-in limit is off.
* @description CIDR ranges of trusted reverse proxies. Their X-Forwarded-For and X-Forwarded-Proto are resolved once by ASGI middleware for auth, allowed_ips, sign-in throttling and requester IP logging. Identity-header auth still checks the original peer. Set [] for direct connections. When unset, use_x_forwarded_for retains deprecated raw-header behavior for legacy IP consumers and the per-source sign-in limit is off. Invalid ranges are ignored, and disable the per-source sign-in limit. Disable server-level proxy-header rewriting when launching an external ASGI server. MCP keeps its separate trust settings.
*/
trusted_proxy_ranges?: string[] | null;
/**