mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-24 00:52:24 +00:00
9a365d2021
A blocked source, or source and username pair, is now refused with 429 before the database lookup and password check, in place of the soft block that held wrong guesses for 30 seconds and let a correct password through. The env admin credentials and the master key typed into the login form are refused like any other credential while blocked; recovery is the master key as an API bearer token, which never goes through the sign-in path trusted_proxy_ranges: [] now means clients connect directly, so the peer address is the source and the per-source limit stays on. Only an unset or malformed value leaves the topology unknown, warns at startup and turns the per-source limit off Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
101 lines
3.6 KiB
Python
101 lines
3.6 KiB
Python
from __future__ import annotations
|
|
|
|
import ipaddress
|
|
from collections.abc import Sequence
|
|
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: Sequence[str] = Field(default_factory=tuple)
|
|
|
|
|
|
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 _unmapped(addr: ipaddress.IPv4Address | ipaddress.IPv6Address) -> ipaddress.IPv4Address | ipaddress.IPv6Address:
|
|
if isinstance(addr, ipaddress.IPv6Address) and addr.ipv4_mapped is not None:
|
|
return addr.ipv4_mapped
|
|
return addr
|
|
|
|
|
|
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
|
|
candidates: Final = (addr, _unmapped(addr))
|
|
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)
|
|
return NetworkContext(
|
|
client_ip=ip,
|
|
host=request.headers.get("host"),
|
|
via_trusted_proxy=via_proxy,
|
|
)
|