chore(mcp): address mypy + Greptile feedback on PR 27575

- Cast client_ip to str at the bottom of _is_server_accessible_from_ip so
  mypy sees the narrowing the early returns already perform. The runtime
  shape is unchanged — INTERNAL_REQUEST and None are both handled in the
  early branches above.
- Distinguish "client_ip_unknown" from "ip_filtering" in the
  rest_endpoints 403 response. When IP extraction fails on a public
  server the previous error told the operator to set
  available_on_public_internet=True, which was already set; the new
  message points them at use_x_forwarded_for / mcp_trusted_proxy_ranges.
- Preserve the explicit port in the Host header when
  litellm.user_url_validation is False. The opt-out path previously
  dropped the port, sending a Host that didn't match the connection
  target for non-default-port URLs.
This commit is contained in:
user 2026-05-10 02:18:07 +00:00
parent 9baa07b51e
commit fc018a9fca
No known key found for this signature in database
4 changed files with 42 additions and 4 deletions

View file

@ -39,7 +39,10 @@ def _validate_mcp_oauth_outbound_url(
operator who points an MCP server at an internal IdP doesn't unintentionally
expose it as an SSRF probe via the proxy."""
if not getattr(litellm, "user_url_validation", True):
return url, urlparse(url).hostname or ""
parsed = urlparse(url)
host = parsed.hostname or ""
host_header = f"{host}:{parsed.port}" if parsed.port else host
return url, host_header
try:
return validate_url(url)
except SSRFError as exc:

View file

@ -3109,12 +3109,14 @@ class MCPServerManager:
public_ids = set(litellm.public_mcp_servers or [])
if server.server_id in public_ids:
return True
# Non-public server: only accessible from internal IPs
# Non-public server: only accessible from internal IPs. The two
# early returns above narrow client_ip to ``str`` for the type
# checker; cast keeps mypy happy without a runtime assert.
general_settings = self._get_general_settings()
internal_networks = IPAddressUtils.parse_internal_networks(
general_settings.get("mcp_internal_ip_ranges")
)
return IPAddressUtils.is_internal_ip(client_ip, internal_networks)
return IPAddressUtils.is_internal_ip(cast(str, client_ip), internal_networks)
def get_mcp_server_by_id(self, server_id: str) -> Optional[MCPServer]:
"""

View file

@ -397,6 +397,19 @@ if MCP_AVAILABLE:
_server, rest_client_ip
)
):
if rest_client_ip is None:
raise HTTPException(
status_code=403,
detail={
"error": "client_ip_unknown",
"message": (
"Cannot determine client IP for IP-based access "
"control. If the proxy is behind a load balancer "
"or reverse proxy, configure use_x_forwarded_for "
"and mcp_trusted_proxy_ranges in general_settings."
),
},
)
raise HTTPException(
status_code=403,
detail={
@ -486,6 +499,19 @@ if MCP_AVAILABLE:
_server, rest_client_ip
)
):
if rest_client_ip is None:
raise HTTPException(
status_code=403,
detail={
"error": "client_ip_unknown",
"message": (
"Cannot determine client IP for IP-based access "
"control. If the proxy is behind a load balancer "
"or reverse proxy, configure use_x_forwarded_for "
"and mcp_trusted_proxy_ranges in general_settings."
),
},
)
raise HTTPException(
status_code=403,
detail={

View file

@ -2530,7 +2530,14 @@ class TestOutboundOAuthURLValidation:
"http://10.0.0.1:5000/token", role="token"
)
assert url == "http://10.0.0.1:5000/token"
assert host == "10.0.0.1"
# Host header includes the explicit port from the URL.
assert host == "10.0.0.1:5000"
# When the URL omits the port, the Host header omits it too.
url2, host2 = _validate_mcp_oauth_outbound_url(
"http://10.0.0.1/token", role="token"
)
assert host2 == "10.0.0.1"
finally:
litellm.user_url_validation = original