feat(mcp): add mcp_xff_num_trusted_hops to harden X-Forwarded-For client IP resolution (#31257)

* feat(mcp): add mcp_xff_num_trusted_hops to harden XFF client IP resolution

MCP per-server IP access control reads the client IP from X-Forwarded-For
and trusts the leftmost entry. Behind an append-style proxy or load
balancer (AWS ALB, nginx with $proxy_add_x_forwarded_for, HAProxy, Envoy,
Cloudflare), a client can prepend an arbitrary value to the header, so the
leftmost entry is attacker-controllable even when the direct peer is a
trusted proxy. An attacker can therefore spoof an internal IP and reach
servers marked available_on_public_internet=false.

This adds an optional mcp_xff_num_trusted_hops general setting modelled on
Envoy's xff_num_trusted_hops. When set to N, the client IP is read N entries
from the right of the chain (where N is the number of trusted appending
proxies in front of the gateway) instead of the leftmost value, so any
entries a client prepends are ignored. It composes with mcp_trusted_proxy_ranges,
which still validates the direct peer, and only takes effect once that check
passes; without a validated direct peer the gateway keeps failing closed, so
hop counting cannot be abused by a direct-to-pod attacker. The chain must
contain at least N valid entries or resolution fails closed.

Default is unset, preserving existing behaviour.

* chore(ui): regenerate dashboard schema for mcp_xff_num_trusted_hops

* fix(mcp): warn when mcp_xff_num_trusted_hops is below the minimum

A 0 or negative value is silently treated as disabled, which could leave
an operator believing they enabled append-style X-Forwarded-For hardening
while client IP resolution stays on the spoofable leftmost value. Emit a
warning, consistent with how the module already surfaces invalid CIDR
config, so the misconfiguration is visible in logs.

* fix(mcp): reject mcp_xff_num_trusted_hops < 1 at config-parse time

Add a ge=1 bound to the ConfigGeneralSettings field so the
update_config_general_settings path rejects 0 and negative values with a
clear validation error instead of accepting them, and self-documents the
valid range. The runtime warning stays as defense-in-depth for raw-dict
config that bypasses model validation.

* style(mcp): black-format ip_address_utils.py

* fix(mcp): fail closed when mcp_xff_num_trusted_hops is set but invalid

A present-but-invalid mcp_xff_num_trusted_hops (non-integer, or below 1)
previously made _resolve_num_trusted_hops return None, which the caller
treated identically to "unset" and silently fell back to the legacy
leftmost X-Forwarded-For value. An operator who set the value to harden
client IP resolution but typo'd it would get weaker security than before,
with no fail-closed signal.

Model the setting as a tagged union (_HopCountUnset, _HopCountInvalid,
_HopCount) so the three states are distinct: unset keeps the legacy path,
a valid count drives hop-counting, and an invalid value fails closed
(returns "") instead of reverting to the spoofable leftmost address. The
caller matches on the union exhaustively.

Add a parametrized regression test asserting get_mcp_client_ip returns ""
for 0, -1, "abc", and 1.5 even with a spoofed internal leftmost entry,
and update the resolver unit tests for the new return type.

litellm/proxy/_types.py

@@ -2361,6 +2361,11 @@ class ConfigGeneralSettings(LiteLLMPydanticObjectBase):
         None,
         description="CIDR ranges of trusted reverse proxies. When set, X-Forwarded-For and X-Forwarded-* origin headers are only trusted from these IPs.",
     )
+    mcp_xff_num_trusted_hops: Optional[int] = Field(
+        None,
+        ge=1,
+        description="Number of trusted reverse proxies/load balancers in front of the gateway that append to X-Forwarded-For. When set (and mcp_trusted_proxy_ranges validates the direct peer), the client IP for MCP access control is read this many entries from the right of the chain instead of the spoofable leftmost value, defeating append-style X-Forwarded-For forgery.",
+    )
     trusted_proxy_ranges: Optional[List[str]] = 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.",

litellm/proxy/auth/ip_address_utils.py

@@ -6,9 +6,11 @@ External callers (public IPs) only see servers with available_on_public_internet
 """
 
 import ipaddress
+from dataclasses import dataclass
 from typing import Any, Dict, List, Optional, Union
 
 from fastapi import Request
+from pydantic import TypeAdapter, ValidationError
 
 from litellm._logging import verbose_proxy_logger
 from litellm.proxy.auth.auth_utils import _get_request_ip_address
@@ -25,6 +27,26 @@ _warned_xff_without_trusted_ranges = False
 # enabled, so a later rollback to disabled warns again.
 _warned_xff_present_but_disabled = False
 
+_NUM_TRUSTED_HOPS_ADAPTER = TypeAdapter(int)
+
+
+@dataclass(frozen=True, slots=True)
+class _HopCountUnset:
+    """mcp_xff_num_trusted_hops is absent: keep the legacy leftmost-XFF path."""
+
+
+@dataclass(frozen=True, slots=True)
+class _HopCountInvalid:
+    """mcp_xff_num_trusted_hops is present but unusable: fail closed, never legacy."""
+
+
+@dataclass(frozen=True, slots=True)
+class _HopCount:
+    value: int
+
+
+_HopCountSetting = Union[_HopCountUnset, _HopCountInvalid, _HopCount]
+
 
 class IPAddressUtils:
     """Static utilities for IP-based MCP access control."""
@@ -174,6 +196,60 @@ class IPAddressUtils:
         trusted_networks = IPAddressUtils.parse_trusted_proxy_networks(trusted_ranges)
         return IPAddressUtils.is_trusted_proxy(direct_ip, trusted_networks)
 
+    @staticmethod
+    def extract_client_ip_from_xff_hops(
+        xff_header: str,
+        num_trusted_hops: int,
+    ) -> Optional[str]:
+        """
+        Resolve the originating client IP from an X-Forwarded-For chain by
+        counting ``num_trusted_hops`` entries from the right.
+
+        Each trusted proxy appends the address it received the connection from,
+        so the right end of the chain is written by infrastructure while the
+        left end is attacker-controllable. Selecting the Nth entry from the
+        right, where N is the number of trusted appending proxies in front of
+        the gateway, yields the real client IP and discards any values a client
+        prepended to spoof an allowed address.
+
+        Returns None when the chain has fewer than ``num_trusted_hops`` entries
+        or the selected entry is not a valid IP, so callers can fail closed.
+        """
+        entries = tuple(part.strip() for part in xff_header.split(",") if part.strip())
+        if num_trusted_hops < 1 or len(entries) < num_trusted_hops:
+            return None
+        candidate = entries[-num_trusted_hops]
+        try:
+            ipaddress.ip_address(candidate)
+        except ValueError:
+            return None
+        return candidate
+
+    @staticmethod
+    def _resolve_num_trusted_hops(raw_num_trusted_hops: object) -> _HopCountSetting:
+        if raw_num_trusted_hops is None:
+            return _HopCountUnset()
+        try:
+            num_hops = _NUM_TRUSTED_HOPS_ADAPTER.validate_python(raw_num_trusted_hops)
+        except ValidationError:
+            verbose_proxy_logger.warning(
+                "Invalid mcp_xff_num_trusted_hops value %r; failing closed for "
+                "MCP client IP resolution. Set it to a positive integer, or "
+                "remove the setting to restore the legacy X-Forwarded-For path",
+                raw_num_trusted_hops,
+            )
+            return _HopCountInvalid()
+        if num_hops < 1:
+            verbose_proxy_logger.warning(
+                "mcp_xff_num_trusted_hops=%s is below the minimum of 1; failing "
+                "closed for MCP client IP resolution. Set it to a positive "
+                "integer, or remove the setting to restore the legacy "
+                "X-Forwarded-For path",
+                num_hops,
+            )
+            return _HopCountInvalid()
+        return _HopCount(num_hops)
+
     @staticmethod
     def get_mcp_client_ip(
         request: Request,
@@ -186,6 +262,12 @@ class IPAddressUtils:
         1. use_x_forwarded_for is enabled in settings
         2. The direct connection is from a trusted proxy (if mcp_trusted_proxy_ranges configured)
 
+        When ``mcp_xff_num_trusted_hops`` is set, the client IP is read that many
+        entries from the right of the chain instead of the spoofable leftmost
+        value, defeating append-style X-Forwarded-For forgery. A present-but-invalid
+        value (non-integer or below 1) fails closed rather than silently reverting
+        to the legacy path, so a config typo cannot quietly weaken access control.
+
         Args:
             request: FastAPI request object
             general_settings: Optional settings dict. If not provided, imports from proxy_server.
@@ -245,4 +327,24 @@ class IPAddressUtils:
                 # returning it would mis-classify external callers as internal.
                 # Fail closed for access control.
                 return ""
+            match IPAddressUtils._resolve_num_trusted_hops(
+                general_settings.get("mcp_xff_num_trusted_hops")
+            ):
+                case _HopCountInvalid():
+                    return ""
+                case _HopCount(value=num_trusted_hops):
+                    client_ip = IPAddressUtils.extract_client_ip_from_xff_hops(
+                        request.headers["x-forwarded-for"], num_trusted_hops
+                    )
+                    if client_ip is None:
+                        verbose_proxy_logger.warning(
+                            "X-Forwarded-For chain has fewer than "
+                            "mcp_xff_num_trusted_hops=%s entries or an invalid "
+                            "address; failing closed",
+                            num_trusted_hops,
+                        )
+                        return ""
+                    return client_ip
+                case _HopCountUnset():
+                    pass
         return _get_request_ip_address(request, use_x_forwarded_for=use_xff)

litellm/proxy/proxy_server.py

@@ -15385,6 +15385,7 @@ async def get_config_list(
         "maximum_spend_logs_retention_period": {"type": "String"},
         "mcp_internal_ip_ranges": {"type": "List"},
         "mcp_trusted_proxy_ranges": {"type": "List"},
+        "mcp_xff_num_trusted_hops": {"type": "Integer"},
         "always_include_stream_usage": {"type": "Boolean"},
         "forward_client_headers_to_llm_api": {"type": "Boolean"},
         "mcp_required_fields": {"type": "List"},

tests/test_litellm/proxy/auth/test_mcp_ip_filtering.py

@@ -8,11 +8,19 @@ external callers only see servers with available_on_public_internet=True.
 import logging
 from unittest.mock import MagicMock, patch
 
+import pytest
 from fastapi import Request
+from pydantic import ValidationError
 
 import litellm.proxy.auth.ip_address_utils as ip_mod
 from litellm._logging import verbose_proxy_logger
-from litellm.proxy.auth.ip_address_utils import IPAddressUtils
+from litellm.proxy._types import ConfigGeneralSettings
+from litellm.proxy.auth.ip_address_utils import (
+    IPAddressUtils,
+    _HopCount,
+    _HopCountInvalid,
+    _HopCountUnset,
+)
 from litellm.types.mcp_server.mcp_server_manager import MCPServer
 
 
@@ -169,6 +177,222 @@ class TestMCPClientIPExtraction:
         assert result == "203.0.113.5"
 
 
+def _make_request(client_host, xff):
+    request = MagicMock(spec=Request)
+    request.client = MagicMock()
+    request.client.host = client_host
+    request.headers = {"x-forwarded-for": xff}
+    return request
+
+
+class TestExtractClientIpFromXffHops:
+    def test_single_hop_picks_rightmost(self):
+        assert (
+            IPAddressUtils.extract_client_ip_from_xff_hops(
+                "10.0.0.99, 203.0.113.9", num_trusted_hops=1
+            )
+            == "203.0.113.9"
+        )
+
+    def test_two_hops_picks_second_from_right(self):
+        assert (
+            IPAddressUtils.extract_client_ip_from_xff_hops(
+                "10.0.0.99, 203.0.113.9, 172.16.0.1", num_trusted_hops=2
+            )
+            == "203.0.113.9"
+        )
+
+    def test_whitespace_and_empty_entries_are_ignored(self):
+        assert (
+            IPAddressUtils.extract_client_ip_from_xff_hops(
+                " 1.1.1.1 ,  , 203.0.113.9 ", num_trusted_hops=1
+            )
+            == "203.0.113.9"
+        )
+
+    def test_chain_shorter_than_hops_returns_none(self):
+        assert (
+            IPAddressUtils.extract_client_ip_from_xff_hops(
+                "203.0.113.9", num_trusted_hops=2
+            )
+            is None
+        )
+
+    def test_zero_or_negative_hops_returns_none(self):
+        assert (
+            IPAddressUtils.extract_client_ip_from_xff_hops(
+                "203.0.113.9", num_trusted_hops=0
+            )
+            is None
+        )
+
+    def test_invalid_selected_entry_returns_none(self):
+        assert (
+            IPAddressUtils.extract_client_ip_from_xff_hops(
+                "not-an-ip, 10.0.0.1", num_trusted_hops=2
+            )
+            is None
+        )
+
+
+class TestResolveNumTrustedHops:
+    def test_unset_is_unset(self):
+        assert IPAddressUtils._resolve_num_trusted_hops(None) == _HopCountUnset()
+
+    def test_int_value(self):
+        assert IPAddressUtils._resolve_num_trusted_hops(2) == _HopCount(2)
+
+    def test_numeric_string_is_coerced(self):
+        assert IPAddressUtils._resolve_num_trusted_hops("3") == _HopCount(3)
+
+    def test_zero_is_invalid_not_disabled(self):
+        assert IPAddressUtils._resolve_num_trusted_hops(0) == _HopCountInvalid()
+
+    def test_non_numeric_value_is_invalid(self):
+        assert IPAddressUtils._resolve_num_trusted_hops("abc") == _HopCountInvalid()
+
+    def test_below_minimum_warns_so_misconfig_is_visible(self):
+        with patch(
+            "litellm.proxy.auth.ip_address_utils.verbose_proxy_logger"
+        ) as mock_logger:
+            assert IPAddressUtils._resolve_num_trusted_hops(0) == _HopCountInvalid()
+            assert IPAddressUtils._resolve_num_trusted_hops(-3) == _HopCountInvalid()
+        assert mock_logger.warning.call_count == 2
+
+    def test_unset_does_not_warn(self):
+        with patch(
+            "litellm.proxy.auth.ip_address_utils.verbose_proxy_logger"
+        ) as mock_logger:
+            assert IPAddressUtils._resolve_num_trusted_hops(None) == _HopCountUnset()
+        mock_logger.warning.assert_not_called()
+
+    def test_valid_value_does_not_warn(self):
+        with patch(
+            "litellm.proxy.auth.ip_address_utils.verbose_proxy_logger"
+        ) as mock_logger:
+            assert IPAddressUtils._resolve_num_trusted_hops(2) == _HopCount(2)
+        mock_logger.warning.assert_not_called()
+
+
+class TestConfigGeneralSettingsHopsValidation:
+    """mcp_xff_num_trusted_hops must reject sub-minimum values at config-parse time."""
+
+    @pytest.mark.parametrize("bad_value", [0, -1])
+    def test_below_minimum_is_rejected(self, bad_value):
+        with pytest.raises(ValidationError):
+            ConfigGeneralSettings(mcp_xff_num_trusted_hops=bad_value)
+
+    def test_valid_value_and_unset_are_accepted(self):
+        assert (
+            ConfigGeneralSettings(mcp_xff_num_trusted_hops=1).mcp_xff_num_trusted_hops
+            == 1
+        )
+        assert ConfigGeneralSettings().mcp_xff_num_trusted_hops is None
+
+
+class TestXffTrustedHopsAccessControl:
+    def test_spoofed_internal_leftmost_is_defeated(self):
+        request = _make_request("10.0.0.5", "10.0.0.99, 203.0.113.9")
+
+        result = IPAddressUtils.get_mcp_client_ip(
+            request,
+            general_settings={
+                "use_x_forwarded_for": True,
+                "mcp_trusted_proxy_ranges": ["10.0.0.0/8"],
+                "mcp_xff_num_trusted_hops": 1,
+            },
+        )
+
+        assert result == "203.0.113.9"
+        assert IPAddressUtils.is_internal_ip(result) is False
+
+    def test_genuine_internal_client_is_preserved(self):
+        request = _make_request("10.0.0.5", "10.0.0.50")
+
+        result = IPAddressUtils.get_mcp_client_ip(
+            request,
+            general_settings={
+                "use_x_forwarded_for": True,
+                "mcp_trusted_proxy_ranges": ["10.0.0.0/8"],
+                "mcp_xff_num_trusted_hops": 1,
+            },
+        )
+
+        assert result == "10.0.0.50"
+        assert IPAddressUtils.is_internal_ip(result) is True
+
+    def test_two_hops_skips_proxy_appended_addresses(self):
+        request = _make_request("10.0.0.5", "10.0.0.99, 203.0.113.9, 172.16.0.1")
+
+        result = IPAddressUtils.get_mcp_client_ip(
+            request,
+            general_settings={
+                "use_x_forwarded_for": True,
+                "mcp_trusted_proxy_ranges": ["10.0.0.0/8"],
+                "mcp_xff_num_trusted_hops": 2,
+            },
+        )
+
+        assert result == "203.0.113.9"
+
+    def test_short_chain_fails_closed(self):
+        request = _make_request("10.0.0.5", "203.0.113.9")
+
+        result = IPAddressUtils.get_mcp_client_ip(
+            request,
+            general_settings={
+                "use_x_forwarded_for": True,
+                "mcp_trusted_proxy_ranges": ["10.0.0.0/8"],
+                "mcp_xff_num_trusted_hops": 2,
+            },
+        )
+
+        assert result == ""
+        assert IPAddressUtils.is_internal_ip(result) is False
+
+    def test_hops_without_trusted_ranges_still_fails_closed(self):
+        request = _make_request("203.0.113.5", "10.0.0.99, 8.8.8.8")
+
+        result = IPAddressUtils.get_mcp_client_ip(
+            request,
+            general_settings={
+                "use_x_forwarded_for": True,
+                "mcp_xff_num_trusted_hops": 1,
+            },
+        )
+
+        assert result == ""
+
+    def test_unset_hops_keeps_legacy_leftmost_behavior(self):
+        request = _make_request("10.0.0.5", "10.0.0.99, 203.0.113.9")
+
+        result = IPAddressUtils.get_mcp_client_ip(
+            request,
+            general_settings={
+                "use_x_forwarded_for": True,
+                "mcp_trusted_proxy_ranges": ["10.0.0.0/8"],
+            },
+        )
+
+        assert result == "10.0.0.99, 203.0.113.9"
+
+    @pytest.mark.parametrize("bad_value", [0, -1, "abc", 1.5])
+    def test_invalid_hops_config_fails_closed_not_legacy(self, bad_value):
+        request = _make_request("10.0.0.5", "10.0.0.99, 203.0.113.9")
+
+        result = IPAddressUtils.get_mcp_client_ip(
+            request,
+            general_settings={
+                "use_x_forwarded_for": True,
+                "mcp_trusted_proxy_ranges": ["10.0.0.0/8"],
+                "mcp_xff_num_trusted_hops": bad_value,
+            },
+        )
+
+        assert result == ""
+        assert IPAddressUtils.is_internal_ip(result) is False
+
+
 class TestXffPresentButDisabledWarning:
     """When an XFF header arrives but use_x_forwarded_for is off, the proxy must
     loudly warn (the internal-only check is silently trusting the load balancer's

ui/litellm-dashboard/src/lib/http/schema.d.ts

@@ -22426,6 +22426,11 @@ export interface components {
              * @description CIDR ranges of trusted reverse proxies. When set, X-Forwarded-For and X-Forwarded-* origin headers are only trusted from these IPs.
              */
             mcp_trusted_proxy_ranges?: string[] | null;
+            /**
+             * Mcp Xff Num Trusted Hops
+             * @description Number of trusted reverse proxies/load balancers in front of the gateway that append to X-Forwarded-For. When set (and mcp_trusted_proxy_ranges validates the direct peer), the client IP for MCP access control is read this many entries from the right of the chain instead of the spoofable leftmost value, defeating append-style X-Forwarded-For forgery.
+             */
+            mcp_xff_num_trusted_hops?: number | null;
             /**
              * Otel
              * @description [BETA] OpenTelemetry support - this might change, use with caution.