Merge pull request #26915 from stuxf/codex/provider-url-destination-guard

chore(providers): guard URL-valued model destinations

litellm/__init__.py

@@ -280,6 +280,7 @@ ssl_security_level: Optional[str] = None
 ssl_certificate: Optional[str] = None
 user_url_validation: bool = True
 user_url_allowed_hosts: List[str] = []
+provider_url_destination_allowed_hosts: List[str] = []
 ssl_ecdh_curve: Optional[str] = (
     None  # Set to 'X25519' to disable PQC and improve performance
 )

litellm/litellm_core_utils/url_utils.py

@@ -21,7 +21,7 @@ Admins can opt out via two ``litellm`` globals (wired from proxy config):
 
 import socket
 from ipaddress import ip_address, ip_network
-from typing import Any, List, Set, Tuple
+from typing import Any, List, Optional, Set, Tuple
 from urllib.parse import quote, urlparse, urlunparse
 
 import httpx
@@ -110,6 +110,85 @@ def _normalize_host(host: str) -> str:
     return host.lower().rstrip(".")
 
 
+def _default_port_for_scheme(scheme: str) -> int:
+    return 443 if scheme == "https" else 80
+
+
+def _parse_url_destination_allowlist_entry(
+    entry: str,
+) -> Optional[Tuple[str, Optional[str], Optional[int]]]:
+    """Parse an admin allowlist entry into host, optional scheme, optional port.
+
+    Entries may be bare hosts (``api.example.com``), host+port
+    (``api.example.com:8443``), or origins (``https://api.example.com``).
+    URL paths are intentionally ignored so admins can paste an api_base value.
+    """
+    entry = entry.strip()
+    if not entry:
+        return None
+
+    has_scheme = "://" in entry
+    parsed = urlparse(entry if has_scheme else f"//{entry}")
+    if has_scheme and parsed.scheme not in _ALLOWED_SCHEMES:
+        return None
+    if parsed.username is not None or parsed.password is not None:
+        return None
+    if not parsed.hostname:
+        return None
+
+    try:
+        port = parsed.port
+    except ValueError:
+        return None
+
+    scheme: Optional[str] = parsed.scheme if has_scheme else None
+    if scheme is not None and port is None:
+        port = _default_port_for_scheme(scheme)
+
+    return _normalize_host(parsed.hostname), scheme, port
+
+
+def is_url_destination_allowed_by_host(url: str, allowed_hosts: List[str]) -> bool:
+    """Return True when a credential-bearing provider URL is admin-allowlisted.
+
+    This does not fetch, resolve, or rewrite URLs. It only answers whether the
+    destination origin is explicitly trusted by configuration. Use ``safe_get``
+    for user-controlled content fetches that require SSRF protection.
+    """
+    parsed = urlparse(url)
+    if parsed.scheme not in _ALLOWED_SCHEMES:
+        return False
+    if parsed.username is not None or parsed.password is not None:
+        return False
+    if not parsed.hostname:
+        return False
+
+    try:
+        effective_port = parsed.port or _default_port_for_scheme(parsed.scheme)
+    except ValueError:
+        return False
+
+    normalized_host = _normalize_host(parsed.hostname)
+    configured_entries = (
+        [allowed_hosts] if isinstance(allowed_hosts, str) else allowed_hosts
+    )
+    for entry in configured_entries or []:
+        if not isinstance(entry, str):
+            continue
+        parsed_entry = _parse_url_destination_allowlist_entry(entry)
+        if parsed_entry is None:
+            continue
+        allowed_host, allowed_scheme, allowed_port = parsed_entry
+        if allowed_host != normalized_host:
+            continue
+        if allowed_scheme is not None and allowed_scheme != parsed.scheme:
+            continue
+        if allowed_port is not None and allowed_port != effective_port:
+            continue
+        return True
+    return False
+
+
 def _format_host_header(hostname: str, port: int, default_port: int) -> str:
     """Build an RFC 7230 Host header value, bracketing IPv6 literals."""
     bracketed = f"[{hostname}]" if ":" in hostname else hostname
@@ -185,7 +264,7 @@ def validate_url(url: str) -> Tuple[str, str]:
         raise SSRFError("URL has no hostname")
 
     port = parsed.port
-    default_port = 443 if parsed.scheme == "https" else 80
+    default_port = _default_port_for_scheme(parsed.scheme)
     effective_port = port if port is not None else default_port
     host_header = _format_host_header(hostname, effective_port, default_port)
 

litellm/proxy/litellm_pre_call_utils.py

@@ -5,7 +5,7 @@ import time
 from collections import OrderedDict
 from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union
 
-from fastapi import Request
+from fastapi import HTTPException, Request
 from pydantic import ValidationError as PydanticValidationError
 from starlette.datastructures import Headers
 
@@ -14,6 +14,7 @@ from litellm._logging import verbose_logger, verbose_proxy_logger
 from litellm._service_logger import ServiceLogging
 from litellm.litellm_core_utils.credential_accessor import CredentialAccessor
 from litellm.litellm_core_utils.safe_json_loads import safe_json_loads
+from litellm.litellm_core_utils.url_utils import is_url_destination_allowed_by_host
 from litellm.proxy._types import (
     AddTeamCallback,
     CommonProxyErrors,
@@ -154,6 +155,45 @@ _ALLOW_CLIENT_MESSAGE_REDACTION_OPT_OUT_METADATA_KEY = (
     "allow_client_message_redaction_opt_out"
 )
 
+# Request fields whose value, when URL-valued, becomes the outbound destination
+# for a provider call. Letting a proxy caller pin the destination is an SSRF
+# primitive (HuggingFace/Oobabooga `model`, Gemini files `file_id`); guard
+# them centrally so SDK users keep working but proxy users default-deny.
+_URL_DESTINATION_REQUEST_FIELDS = ("model", "file_id")
+
+
+def _reject_url_valued_destinations(data: Dict[str, Any]) -> None:
+    """Reject URL-valued ``model``/``file_id`` unless admin-allowlisted.
+
+    Some providers (HuggingFace, Oobabooga, Gemini files) accept a URL in the
+    identifier field and use it as the outbound destination. On the proxy that
+    is an SSRF primitive — a low-privilege caller can point traffic at any
+    host the proxy can reach, including internal services. Reject here at the
+    proxy boundary so SDK users (who legitimately pass URL-valued identifiers)
+    are unaffected, while admins can opt specific hosts back in via
+    ``litellm.provider_url_destination_allowed_hosts``.
+    """
+    allowed_hosts = getattr(litellm, "provider_url_destination_allowed_hosts", []) or []
+    for field in _URL_DESTINATION_REQUEST_FIELDS:
+        value = data.get(field)
+        if not isinstance(value, str) or not value.startswith(("http://", "https://")):
+            continue
+        if is_url_destination_allowed_by_host(value, allowed_hosts):
+            continue
+        raise HTTPException(
+            status_code=400,
+            detail={
+                "error": "invalid_request",
+                "param": field,
+                "message": (
+                    f"URL-valued '{field}' is not allowed. Configure custom "
+                    "endpoints with api_base instead, or add the destination "
+                    "host to `provider_url_destination_allowed_hosts` in "
+                    "litellm_settings."
+                ),
+            },
+        )
+
 
 def _strip_untrusted_request_header_controls(
     headers: Any,
@@ -1109,6 +1149,7 @@ async def add_litellm_data_to_request(  # noqa: PLR0915
         if _allow_client_mock_response and _internal_key in _CLIENT_MOCK_CONTROL_FIELDS:
             continue
         data.pop(_internal_key, None)
+    _reject_url_valued_destinations(data)
     # Strip spoofable auth metadata from user-supplied metadata dict
     _user_metadata = data.get("metadata")
     if isinstance(_user_metadata, dict):

tests/test_litellm/litellm_core_utils/test_url_utils.py

@@ -7,6 +7,7 @@ from litellm.litellm_core_utils import url_utils
 from litellm.litellm_core_utils.url_utils import (
     SSRFError,
     _is_blocked_ip,
+    assert_same_origin,
     encode_url_path_segment,
     encode_url_path_segments,
     validate_url,
@@ -424,12 +425,51 @@ class TestHostAllowlist:
         validate_url("http://internal.corp/")
 
 
+class TestProviderUrlDestinationAllowlist:
+    def test_host_entry_matches_any_scheme_and_port(self):
+        assert url_utils.is_url_destination_allowed_by_host(
+            "https://trusted.example/v1/chat/completions",
+            ["trusted.example"],
+        )
+        assert url_utils.is_url_destination_allowed_by_host(
+            "http://trusted.example:8080/v1/chat/completions",
+            ["trusted.example"],
+        )
+
+    def test_origin_entry_matches_scheme_and_default_port(self):
+        assert url_utils.is_url_destination_allowed_by_host(
+            "https://trusted.example/v1/chat/completions",
+            ["https://trusted.example"],
+        )
+        assert not url_utils.is_url_destination_allowed_by_host(
+            "http://trusted.example/v1/chat/completions",
+            ["https://trusted.example"],
+        )
+
+    def test_port_entry_only_matches_same_effective_port(self):
+        assert url_utils.is_url_destination_allowed_by_host(
+            "https://trusted.example/v1/chat/completions",
+            ["trusted.example:443"],
+        )
+        assert not url_utils.is_url_destination_allowed_by_host(
+            "https://trusted.example:8443/v1/chat/completions",
+            ["trusted.example:443"],
+        )
+
+    def test_rejects_userinfo_and_invalid_port(self):
+        assert not url_utils.is_url_destination_allowed_by_host(
+            "https://user:pass@trusted.example/v1/chat/completions",
+            ["trusted.example"],
+        )
+        assert not url_utils.is_url_destination_allowed_by_host(
+            "https://trusted.example:99999/v1/chat/completions",
+            ["trusted.example"],
+        )
+
+
 # ── assert_same_origin ────────────────────────────────────────────────────────
 
 
-from litellm.litellm_core_utils.url_utils import assert_same_origin
-
-
 def test_assert_same_origin_matches_scheme_host_port():
     """A polling URL on the same scheme + host + port as the api_base
     passes — the upstream is trusted; the URL it returned points back at

tests/test_litellm/proxy/test_provider_url_destination_guard.py

@@ -0,0 +1,139 @@
+"""Proxy-level guard against URL-valued ``model`` / ``file_id`` request fields.
+
+Some providers (HuggingFace, Oobabooga, Gemini files) accept a URL in the
+identifier field and use it as the outbound destination. On the proxy that is
+an SSRF primitive — guarded centrally in ``litellm_pre_call_utils`` so SDK
+users keep working but proxy users default-deny.
+"""
+
+import os
+import sys
+from unittest.mock import MagicMock
+
+import pytest
+from fastapi import HTTPException, Request
+
+import litellm
+from litellm.proxy._types import UserAPIKeyAuth
+from litellm.proxy.litellm_pre_call_utils import (
+    _reject_url_valued_destinations,
+    add_litellm_data_to_request,
+)
+
+sys.path.insert(0, os.path.abspath("../../.."))
+
+
+class TestRejectUrlValuedDestinations:
+    def test_plain_model_passes(self):
+        _reject_url_valued_destinations({"model": "gpt-4"})
+
+    def test_plain_file_id_passes(self):
+        _reject_url_valued_destinations({"file_id": "files/abc123"})
+
+    def test_no_destination_field_passes(self):
+        _reject_url_valued_destinations({"messages": [{"role": "user"}]})
+
+    def test_url_valued_model_rejected(self):
+        with pytest.raises(HTTPException) as exc_info:
+            _reject_url_valued_destinations({"model": "https://attacker.example/v1"})
+        assert exc_info.value.status_code == 400
+        assert exc_info.value.detail["param"] == "model"
+
+    def test_url_valued_file_id_rejected(self):
+        with pytest.raises(HTTPException) as exc_info:
+            _reject_url_valued_destinations(
+                {"file_id": "https://attacker.example/v1beta/files/abc"}
+            )
+        assert exc_info.value.status_code == 400
+        assert exc_info.value.detail["param"] == "file_id"
+
+    def test_http_scheme_also_rejected(self):
+        with pytest.raises(HTTPException):
+            _reject_url_valued_destinations({"model": "http://10.0.0.1:8080/v1"})
+
+    def test_non_string_value_ignored(self):
+        # Defensive: malformed inputs (list, dict, None) shouldn't crash here;
+        # downstream Pydantic validation handles the type error.
+        _reject_url_valued_destinations({"model": None})
+        _reject_url_valued_destinations({"model": 42})
+        _reject_url_valued_destinations({"file_id": ["a", "b"]})
+
+    def test_allowlisted_host_passes(self, monkeypatch):
+        monkeypatch.setattr(
+            litellm,
+            "provider_url_destination_allowed_hosts",
+            ["trusted.example"],
+        )
+        _reject_url_valued_destinations({"model": "https://trusted.example/v1"})
+
+    def test_allowlisted_origin_rejects_mismatched_scheme(self, monkeypatch):
+        monkeypatch.setattr(
+            litellm,
+            "provider_url_destination_allowed_hosts",
+            ["https://trusted.example"],
+        )
+        _reject_url_valued_destinations({"model": "https://trusted.example/v1"})
+        with pytest.raises(HTTPException):
+            _reject_url_valued_destinations({"model": "http://trusted.example/v1"})
+
+    def test_allowlisted_host_port_rejects_other_ports(self, monkeypatch):
+        monkeypatch.setattr(
+            litellm,
+            "provider_url_destination_allowed_hosts",
+            ["trusted.example:8443"],
+        )
+        _reject_url_valued_destinations({"model": "https://trusted.example:8443/v1"})
+        with pytest.raises(HTTPException):
+            _reject_url_valued_destinations({"model": "https://trusted.example/v1"})
+
+    def test_userinfo_in_url_rejected_even_when_host_allowlisted(self, monkeypatch):
+        # Embedded credentials in URL are an exfil channel — must never pass.
+        monkeypatch.setattr(
+            litellm,
+            "provider_url_destination_allowed_hosts",
+            ["trusted.example"],
+        )
+        with pytest.raises(HTTPException):
+            _reject_url_valued_destinations(
+                {"model": "https://user:pass@trusted.example/v1"}
+            )
+
+
+def _make_request_mock() -> Request:
+    request_mock = MagicMock(spec=Request)
+    request_mock.url.path = "/v1/chat/completions"
+    request_mock.url = MagicMock()
+    request_mock.url.__str__.return_value = "http://localhost/v1/chat/completions"
+    request_mock.method = "POST"
+    request_mock.query_params = {}
+    request_mock.headers = {"Content-Type": "application/json"}
+    request_mock.client = MagicMock()
+    request_mock.client.host = "127.0.0.1"
+    return request_mock
+
+
+@pytest.mark.asyncio
+async def test_add_litellm_data_to_request_rejects_url_valued_model():
+    user_api_key_dict = UserAPIKeyAuth(
+        api_key="hashed-key",
+        metadata={},
+        team_metadata={},
+        spend=0.0,
+        max_budget=100.0,
+        model_max_budget={},
+        team_spend=0.0,
+        team_max_budget=200.0,
+    )
+    data = {"model": "https://attacker.example/v1", "messages": []}
+
+    with pytest.raises(HTTPException) as exc_info:
+        await add_litellm_data_to_request(
+            data=data,
+            request=_make_request_mock(),
+            user_api_key_dict=user_api_key_dict,
+            proxy_config=MagicMock(),
+            general_settings={},
+            version="test-version",
+        )
+    assert exc_info.value.status_code == 400
+    assert exc_info.value.detail["param"] == "model"