Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/auth/ip_address_utils.py: 35%
157 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1"""
2IP address utilities for MCP public/private access control.
4Internal callers (private IPs) see all MCP servers.
5External callers (public IPs) only see servers with available_on_public_internet=True.
6"""
8import ipaddress
9import os
10from collections.abc import Mapping
11from dataclasses import dataclass
12from typing import Any, Final
13from urllib.parse import urlparse
15from fastapi import Request
16from pydantic import TypeAdapter, ValidationError
18from litellm._logging import verbose_proxy_logger
19from litellm.proxy.auth.auth_utils import _get_request_ip_address
21# One-shot warning so operators upgrading from the prior "always trust X-Forwarded-*"
22# behaviour see an actionable message in their logs the first time it triggers.
23_warned_xff_without_trusted_ranges = False
25# Error for the inverse footgun: requests arrive with an X-Forwarded-For header
26# but use_x_forwarded_for is off, so the real client IP is silently dropped and
27# "internal network only" access control trusts the load balancer's IP instead.
28# Logged once per misconfiguration window (not per-request) so a flood of crafted
29# XFF headers can't spam the logs; re-arms whenever use_x_forwarded_for is observed
30# enabled, so a later rollback to disabled warns again.
31_warned_xff_present_but_disabled = False
33_NUM_TRUSTED_HOPS_ADAPTER: Final = TypeAdapter(int)
36@dataclass(frozen=True, slots=True)
37class _HopCountUnset:
38 """mcp_xff_num_trusted_hops is absent: keep the legacy leftmost-XFF path."""
41@dataclass(frozen=True, slots=True)
42class _HopCountInvalid:
43 """mcp_xff_num_trusted_hops is present but unusable: fail closed, never legacy."""
46@dataclass(frozen=True, slots=True)
47class _HopCount:
48 value: int
51_HopCountSetting = _HopCountUnset | _HopCountInvalid | _HopCount
54class IPAddressUtils:
55 """Static utilities for IP-based MCP access control."""
57 _DEFAULT_INTERNAL_NETWORKS = [
58 ipaddress.ip_network("10.0.0.0/8"),
59 ipaddress.ip_network("172.16.0.0/12"),
60 ipaddress.ip_network("192.168.0.0/16"),
61 ipaddress.ip_network("127.0.0.0/8"),
62 ipaddress.ip_network("::1/128"),
63 ipaddress.ip_network("fc00::/7"),
64 ]
66 @staticmethod
67 def parse_internal_networks(
68 configured_ranges: list[str] | None,
69 ) -> list[ipaddress.IPv4Network | ipaddress.IPv6Network]:
70 """Parse configured CIDR ranges into network objects, falling back to defaults."""
71 if not configured_ranges: 71 ↛ 73line 71 didn't jump to line 73 because the condition on line 71 was always true
72 return IPAddressUtils._DEFAULT_INTERNAL_NETWORKS
73 networks: Final[list[ipaddress.IPv4Network | ipaddress.IPv6Network]] = []
74 for cidr in configured_ranges:
75 try:
76 networks.append(ipaddress.ip_network(cidr, strict=False))
77 except ValueError:
78 verbose_proxy_logger.warning("Invalid CIDR in mcp_internal_ip_ranges: %s, skipping", cidr)
79 return networks if networks else IPAddressUtils._DEFAULT_INTERNAL_NETWORKS
81 @staticmethod
82 def parse_trusted_proxy_networks(
83 configured_ranges: list[str] | None,
84 ) -> list[ipaddress.IPv4Network | ipaddress.IPv6Network]:
85 """
86 Parse trusted proxy CIDR ranges for XFF validation.
87 Returns empty list if not configured (XFF will not be trusted).
88 """
89 if not configured_ranges:
90 return []
91 networks: Final[list[ipaddress.IPv4Network | ipaddress.IPv6Network]] = []
92 for cidr in configured_ranges:
93 try:
94 networks.append(ipaddress.ip_network(cidr, strict=False))
95 except ValueError:
96 verbose_proxy_logger.warning("Invalid CIDR in mcp_trusted_proxy_ranges: %s, skipping", cidr)
97 return networks
99 @staticmethod
100 def is_trusted_proxy(
101 proxy_ip: str | None,
102 trusted_networks: list[ipaddress.IPv4Network | ipaddress.IPv6Network],
103 ) -> bool:
104 """Check if the direct connection IP is from a trusted proxy."""
105 if not proxy_ip or not trusted_networks:
106 return False
107 try:
108 addr: Final = ipaddress.ip_address(proxy_ip.strip())
109 return any(addr in network for network in trusted_networks)
110 except ValueError:
111 return False
113 @staticmethod
114 def is_internal_ip(
115 client_ip: str | None,
116 internal_networks: list[ipaddress.IPv4Network | ipaddress.IPv6Network] | None = None,
117 ) -> bool:
118 """
119 Check if a client IP is from an internal/private network.
121 Handles X-Forwarded-For comma chains (takes leftmost = original client).
122 Fails closed: empty/invalid IPs are treated as external.
123 """
124 if not client_ip: 124 ↛ 125line 124 didn't jump to line 125 because the condition on line 124 was never true
125 return False
127 # X-Forwarded-For may contain comma-separated chain; leftmost is original client
128 if "," in client_ip: 128 ↛ 129line 128 didn't jump to line 129 because the condition on line 128 was never true
129 client_ip = client_ip.split(",")[0].strip()
131 networks: Final = internal_networks or IPAddressUtils._DEFAULT_INTERNAL_NETWORKS
133 try:
134 addr: Final = ipaddress.ip_address(client_ip.strip())
135 except ValueError:
136 return False
138 return any(addr in network for network in networks)
140 @staticmethod
141 def is_request_from_trusted_proxy(
142 request: Request,
143 general_settings: Mapping[str, Any] | None = None,
144 ) -> bool:
145 """
146 Return True if X-Forwarded-* headers on this request should be trusted.
148 Trusts the headers iff both:
149 1. ``use_x_forwarded_for`` is enabled in proxy settings, AND
150 2. ``mcp_trusted_proxy_ranges`` is configured AND the direct
151 connection IP (``request.client.host``) falls inside one of
152 those CIDRs.
154 When ``use_x_forwarded_for`` is enabled but ``mcp_trusted_proxy_ranges``
155 is missing, the headers are NOT trusted: there is no way to
156 distinguish a trusted reverse proxy from a direct attacker, so callers
157 that build URLs (OAuth issuer / redirect_uri / etc.) must fall back
158 to the request's literal base URL instead of risking a poisoned host.
159 """
160 if general_settings is None: 160 ↛ 170line 160 didn't jump to line 170 because the condition on line 160 was always true
161 try:
162 from litellm.proxy.proxy_server import (
163 general_settings as proxy_general_settings,
164 )
166 general_settings = proxy_general_settings
167 except ImportError:
168 general_settings = {}
170 if general_settings is None: 170 ↛ 171line 170 didn't jump to line 171 because the condition on line 170 was never true
171 general_settings = {}
173 if not general_settings.get("use_x_forwarded_for", False): 173 ↛ 176line 173 didn't jump to line 176 because the condition on line 173 was always true
174 return False
176 trusted_ranges: Final = general_settings.get("mcp_trusted_proxy_ranges")
177 if not trusted_ranges:
178 global _warned_xff_without_trusted_ranges
179 if not _warned_xff_without_trusted_ranges:
180 verbose_proxy_logger.warning(
181 "use_x_forwarded_for is enabled but mcp_trusted_proxy_ranges "
182 "is not configured, so X-Forwarded-* headers will NOT be trusted. "
183 "MCP OAuth discovery URLs fall back to the proxy's literal request "
184 "URL, and MCP access-control client-IP resolution fails closed "
185 "(callers are treated as external). Set mcp_trusted_proxy_ranges in "
186 "general_settings to your reverse-proxy CIDR(s) to trust "
187 "X-Forwarded-*."
188 )
189 _warned_xff_without_trusted_ranges = True
190 return False
192 direct_ip: Final = request.client.host if request.client else None
193 trusted_networks: Final = IPAddressUtils.parse_trusted_proxy_networks(trusted_ranges)
194 return IPAddressUtils.is_trusted_proxy(direct_ip, trusted_networks)
196 @staticmethod
197 def is_request_https(
198 request: Request,
199 general_settings: Mapping[str, Any] | None = None,
200 ) -> bool:
201 """
202 Whether this request's PUBLIC-facing origin is HTTPS, for deciding
203 whether a cookie set on the response should be marked ``Secure``.
205 litellm only sees a plain-HTTP hop whenever TLS terminates at a
206 reverse proxy, so ``request.url.scheme`` alone cannot answer this in
207 that deployment shape. Resolved from the first trusted signal:
208 1. ``PROXY_BASE_URL`` (operator-declared public origin).
209 2. ``X-Forwarded-Proto``, only when the request's direct peer is a
210 configured trusted proxy -- see ``is_request_from_trusted_proxy``.
211 An untrusted caller cannot spoof this header to strip Secure.
212 3. The request's own literal scheme (direct TLS termination, or no
213 reverse proxy in front of litellm).
214 """
215 configured_base_url: Final = os.environ.get("PROXY_BASE_URL", "").strip()
216 if configured_base_url:
217 return urlparse(configured_base_url).scheme == "https"
219 if IPAddressUtils.is_request_from_trusted_proxy(request, general_settings=general_settings):
220 forwarded_proto: Final = request.headers.get("X-Forwarded-Proto")
221 if forwarded_proto:
222 return forwarded_proto.split(",")[0].strip().lower() == "https"
224 return request.url.scheme == "https"
226 @staticmethod
227 def extract_client_ip_from_xff_hops(
228 xff_header: str,
229 num_trusted_hops: int,
230 ) -> str | None:
231 """
232 Resolve the originating client IP from an X-Forwarded-For chain by
233 counting ``num_trusted_hops`` entries from the right.
235 Each trusted proxy appends the address it received the connection from,
236 so the right end of the chain is written by infrastructure while the
237 left end is attacker-controllable. Selecting the Nth entry from the
238 right, where N is the number of trusted appending proxies in front of
239 the gateway, yields the real client IP and discards any values a client
240 prepended to spoof an allowed address.
242 Returns None when the chain has fewer than ``num_trusted_hops`` entries
243 or the selected entry is not a valid IP, so callers can fail closed.
244 """
245 entries: Final = tuple(part.strip() for part in xff_header.split(",") if part.strip())
246 if num_trusted_hops < 1 or len(entries) < num_trusted_hops:
247 return None
248 candidate: Final = entries[-num_trusted_hops]
249 try:
250 ipaddress.ip_address(candidate)
251 except ValueError:
252 return None
253 return candidate
255 @staticmethod
256 def _resolve_num_trusted_hops(raw_num_trusted_hops: object) -> _HopCountSetting:
257 if raw_num_trusted_hops is None:
258 return _HopCountUnset()
259 try:
260 num_hops: Final = _NUM_TRUSTED_HOPS_ADAPTER.validate_python(raw_num_trusted_hops)
261 except ValidationError:
262 verbose_proxy_logger.warning(
263 "Invalid mcp_xff_num_trusted_hops value %r; failing closed for "
264 "MCP client IP resolution. Set it to a positive integer, or "
265 "remove the setting to restore the legacy X-Forwarded-For path",
266 raw_num_trusted_hops,
267 )
268 return _HopCountInvalid()
269 if num_hops < 1:
270 verbose_proxy_logger.warning(
271 "mcp_xff_num_trusted_hops=%s is below the minimum of 1; failing "
272 "closed for MCP client IP resolution. Set it to a positive "
273 "integer, or remove the setting to restore the legacy "
274 "X-Forwarded-For path",
275 num_hops,
276 )
277 return _HopCountInvalid()
278 return _HopCount(num_hops)
280 @staticmethod
281 def get_mcp_client_ip(
282 request: Request,
283 general_settings: dict[str, Any] | None = None,
284 ) -> str | None:
285 """
286 Extract client IP from a FastAPI request for MCP access control.
288 Security: Only trusts X-Forwarded-For if:
289 1. use_x_forwarded_for is enabled in settings
290 2. The direct connection is from a trusted proxy (if mcp_trusted_proxy_ranges configured)
292 When ``mcp_xff_num_trusted_hops`` is set, the client IP is read that many
293 entries from the right of the chain instead of the spoofable leftmost
294 value, defeating append-style X-Forwarded-For forgery. A present-but-invalid
295 value (non-integer or below 1) fails closed rather than silently reverting
296 to the legacy path, so a config typo cannot quietly weaken access control.
298 Args:
299 request: FastAPI request object
300 general_settings: Optional settings dict. If not provided, imports from proxy_server.
301 """
302 if general_settings is None: 302 ↛ 313line 302 didn't jump to line 313 because the condition on line 302 was always true
303 try:
304 from litellm.proxy.proxy_server import (
305 general_settings as proxy_general_settings,
306 )
308 general_settings = proxy_general_settings
309 except ImportError:
310 general_settings = {}
312 # Handle case where general_settings is still None after import
313 if general_settings is None: 313 ↛ 314line 313 didn't jump to line 314 because the condition on line 313 was never true
314 general_settings = {}
316 use_xff: Final = general_settings.get("use_x_forwarded_for", False)
318 global _warned_xff_present_but_disabled
319 if use_xff: 319 ↛ 320line 319 didn't jump to line 320 because the condition on line 319 was never true
320 _warned_xff_present_but_disabled = False
321 elif "x-forwarded-for" in request.headers: 321 ↛ 322line 321 didn't jump to line 322 because the condition on line 321 was never true
322 if not _warned_xff_present_but_disabled:
323 verbose_proxy_logger.error(
324 "Received a request with an X-Forwarded-For header but "
325 "use_x_forwarded_for is not enabled. The real client IP is "
326 "being ignored and the direct peer's IP (typically your load "
327 "balancer / reverse proxy) is used for MCP access control. "
328 "Because that peer almost always falls within "
329 "general_settings.mcp_internal_ip_ranges, every external caller "
330 "is treated as internal and 'available_on_public_internet: "
331 "false' MCP servers are effectively exposed. Set "
332 "use_x_forwarded_for: true (and mcp_trusted_proxy_ranges to "
333 "your proxy CIDRs) in general_settings to honor the real "
334 "client IP. Not failing the request: if there is no load "
335 "balancer, a crafted X-Forwarded-For header must not be able "
336 "to take down the service."
337 )
338 _warned_xff_present_but_disabled = True
340 # If XFF is enabled, validate the request comes from a trusted proxy
341 if use_xff and "x-forwarded-for" in request.headers: 341 ↛ 342line 341 didn't jump to line 342 because the condition on line 341 was never true
342 if not IPAddressUtils.is_request_from_trusted_proxy(request, general_settings=general_settings):
343 direct_ip: Final = request.client.host if request.client else None
344 if general_settings.get("mcp_trusted_proxy_ranges"):
345 # Direct connection isn't in any configured trusted CIDR.
346 verbose_proxy_logger.warning("XFF header from untrusted IP %s, ignoring", direct_ip)
347 return direct_ip
348 # XFF enabled but no trusted proxy ranges configured: the direct
349 # peer is typically the reverse proxy's own (private) IP, so
350 # returning it would mis-classify external callers as internal.
351 # Fail closed for access control.
352 return ""
353 match IPAddressUtils._resolve_num_trusted_hops(general_settings.get("mcp_xff_num_trusted_hops")):
354 case _HopCountInvalid():
355 return ""
356 case _HopCount(value=num_trusted_hops):
357 client_ip: Final = IPAddressUtils.extract_client_ip_from_xff_hops(
358 request.headers["x-forwarded-for"], num_trusted_hops
359 )
360 if client_ip is None:
361 verbose_proxy_logger.warning(
362 "X-Forwarded-For chain has fewer than "
363 "mcp_xff_num_trusted_hops=%s entries or an invalid "
364 "address; failing closed",
365 num_trusted_hops,
366 )
367 return ""
368 return client_ip
369 case _HopCountUnset():
370 pass
371 return _get_request_ip_address(request, use_x_forwarded_for=use_xff)