mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-17 23:51:30 +00:00
84 lines
2.7 KiB
Python
84 lines
2.7 KiB
Python
from typing import Any, Final
|
|
|
|
from fastapi import Request
|
|
|
|
from litellm._logging import verbose_proxy_logger
|
|
from litellm.proxy.auth.network import (
|
|
ip_in_networks,
|
|
normalize_cidr_ranges,
|
|
parse_trusted_proxy_ranges,
|
|
)
|
|
|
|
TRUSTED_PROXY_RANGES_KEY: Final = "trusted_proxy_ranges"
|
|
|
|
|
|
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 get_trusted_proxy_cidrs(
|
|
general_settings: dict[str, Any] | None = None,
|
|
) -> list[str]:
|
|
"""Operator-configured trusted reverse-proxy CIDRs, normalized to strings.
|
|
|
|
Empty when none are configured, in which case X-Forwarded-For must not be
|
|
trusted and only the direct peer is authoritative.
|
|
"""
|
|
if general_settings is None:
|
|
general_settings = _get_proxy_general_settings()
|
|
return normalize_cidr_ranges(
|
|
general_settings.get(TRUSTED_PROXY_RANGES_KEY),
|
|
setting_name=TRUSTED_PROXY_RANGES_KEY,
|
|
)
|
|
|
|
|
|
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,
|
|
general_settings: dict[str, Any] | None = 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: Final = 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: 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",
|
|
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."
|
|
)
|