mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-19 00:01:29 +00:00
6c76f5f9c6
Every ruff-strict rule that sat above its budget limit (FURB188, RUF022, SIM118, UP007, UP032, UP037) is now at zero, LIT001 and LIT006 are back under their ceilings, and the freed headroom is ratcheted out of ruff-strict-budget.json, type-discipline-budget.json, and basedpyright-code-budget.json so the gates take the fast path again
93 lines
3.2 KiB
Python
93 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
import ipaddress
|
|
from typing import Any, Final
|
|
|
|
from fastapi import Request
|
|
from pydantic import BaseModel, Field
|
|
|
|
from litellm._logging import verbose_proxy_logger
|
|
|
|
TrustedProxyNetwork = ipaddress.IPv4Network | ipaddress.IPv6Network
|
|
|
|
|
|
class NetworkContext(BaseModel):
|
|
client_ip: str | None = None
|
|
host: str | None = None
|
|
via_trusted_proxy: bool = False
|
|
|
|
|
|
class TrustedProxyConfig(BaseModel):
|
|
use_forwarded_for: bool = False
|
|
trusted_proxy_cidrs: list[str] = Field(default_factory=list)
|
|
|
|
|
|
def normalize_cidr_ranges(configured_ranges: Any, *, setting_name: str = "trusted_proxy_cidrs") -> list[str]:
|
|
if not configured_ranges:
|
|
return []
|
|
if isinstance(configured_ranges, str):
|
|
return [r.strip() for r in configured_ranges.split(",") if r.strip()]
|
|
if isinstance(configured_ranges, (list, tuple, set)):
|
|
return [str(r).strip() for r in configured_ranges if str(r).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_cidrs"
|
|
) -> list[TrustedProxyNetwork]:
|
|
networks: Final[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 ip_in_networks(client_ip: str | None, networks: list[TrustedProxyNetwork]) -> bool:
|
|
if not client_ip or not networks:
|
|
return False
|
|
try:
|
|
addr: Final = ipaddress.ip_address(client_ip.strip())
|
|
except ValueError:
|
|
return False
|
|
return any(addr in network 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)
|
|
return NetworkContext(
|
|
client_ip=ip,
|
|
host=request.headers.get("host"),
|
|
via_trusted_proxy=via_proxy,
|
|
)
|