scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-13 23:11:40 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 101

fix(anthropic): require caller api_key and SSRF-validate api_base in advisor tool (#32093)

Browse source 
* fix(anthropic): require caller api_key and SSRF-validate api_base in advisor tool

The advisor_20260301 interceptor honored a caller-supplied api_base once
allow_client_side_credentials was enabled, even without a caller-supplied
api_key. AnthropicModelInfo.get_auth_header() then fell back to the proxy's
own ANTHROPIC_API_KEY/ANTHROPIC_AUTH_TOKEN, so the server's real credentials
plus the conversation history got sent to a caller-chosen destination

_resolve_advisor_credentials() now only honors api_base alongside a
non-empty caller-supplied api_key, requires the https scheme, and validates
api_base via validate_url() before use, mirroring check_complete_credentials
in auth_utils.py. https is required because validate_url only DNS-pins the
connection for http; for https with TLS verification on it returns the URL
unchanged and relies on certificate validation to block DNS rebinding

* fix(anthropic): also reject advisor api_base when ssl_verify is disabled

validate_url only DNS-pins the connection for http, or for https with
litellm.ssl_verify disabled; the previous https-only check missed the
ssl_verify=False case, where validate_url's rewritten URL was still being
discarded, per Greptile's review of this PR. Reject api_base outright when
ssl_verify is False so the discarded rewrite can no longer matter

(cherry picked from commit 07b9ea8c3b)
This commit is contained in:
yucheng-berri 2026-07-04 12:06:09 -07:00 • committed by Yuneng Jiang
parent 6f4f4d3afe
commit ebb8783a58
No known key found for this signature in database
2 changed files with 242 additions and 13 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

60
litellm/llms/anthropic/experimental_pass_through/messages/interceptors/advisor.py
View file

@ -17,7 +17,9 @@ How it works:
import uuid
from typing import Any, AsyncIterator, Dict, List, Optional, Union
import litellm
import litellm.constants as _c
from litellm.litellm_core_utils.url_utils import validate_url
from litellm.llms.anthropic.common_utils import strip_advisor_blocks_from_messages
from litellm.types.llms.anthropic_messages.anthropic_response import (
AnthropicMessagesResponse,
@ -82,16 +84,7 @@ class AdvisorOrchestrationHandler(MessagesInterceptor):
max_uses: int = (
ADVISOR_MAX_USES if _raw_max_uses is None else int(_raw_max_uses)
)
# Optional routing overrides for the advisor sub-call (e.g. proxy routing).
# If not set in the tool definition, litellm resolves from env vars.
# The advisor tool is caller-controlled; only honor a client-supplied
# api_base/api_key when the proxy has enabled clientside credentials,
# otherwise let litellm resolve from server config.
advisor_api_key: Optional[str] = None
advisor_api_base: Optional[str] = None
if _allow_client_side_advisor_credentials():
advisor_api_key = advisor_tool.get("api_key")
advisor_api_base = advisor_tool.get("api_base")
advisor_api_key, advisor_api_base = _resolve_advisor_credentials(advisor_tool)
# Build the synthetic tool definition the provider will receive.
synthetic_advisor_tool = _make_synthetic_advisor_tool()
@ -201,6 +194,53 @@ def _allow_client_side_advisor_credentials() -> bool:
return general_settings.get("allow_client_side_credentials") is True
def _resolve_advisor_credentials(
advisor_tool: dict,
) -> tuple[Optional[str], Optional[str]]:
"""Resolve the (api_key, api_base) override for the advisor sub-call.
A caller-supplied ``api_base`` is only honored alongside a caller-supplied
``api_key``: without one, ``AnthropicModelInfo.get_auth_header()`` falls
back to the proxy's own Anthropic credentials, which would then be sent to
the caller-chosen ``api_base``. A caller-supplied ``api_base`` is also
required to be https with TLS verification on, and SSRF-validated so it
can't target a private/internal/cloud-metadata address, mirroring
``proxy.auth.auth_utils.check_complete_credentials``. https with TLS
verification is required because ``validate_url`` only rewrites the
connection to a DNS-pinned IP for http, or for https with
``litellm.ssl_verify`` disabled; otherwise it returns the URL unchanged
and relies on certificate validation to block DNS rebinding, so this
closes the same gap without threading the pinned URL through the whole
``anthropic_messages()`` call chain.
"""
if not _allow_client_side_advisor_credentials():
return None, None
api_key: Optional[str] = advisor_tool.get("api_key")
api_base: Optional[str] = advisor_tool.get("api_base")
if api_base is None:
return api_key, None
if not api_key:
raise ValueError(
"advisor tool definition sets 'api_base' without 'api_key'. A "
"caller-supplied api_base is only honored alongside a "
"caller-supplied api_key, so the proxy's own credentials are "
"never sent to a caller-chosen destination."
)
if not api_base.startswith("https://"):
raise ValueError(
f"advisor tool definition sets 'api_base'={api_base!r}, which must use the https scheme."
)
if getattr(litellm, "ssl_verify", True) is False:
raise ValueError(
"advisor tool definition sets 'api_base' but the proxy has TLS verification "
"disabled (litellm.ssl_verify=False), so a caller-supplied api_base can't be "
"safely validated against DNS rebinding."
)
if getattr(litellm, "user_url_validation", True):
validate_url(api_base)
return api_key, api_base
def _make_synthetic_advisor_tool() -> Dict:
"""Build a regular tool definition the executor provider can understand."""
return {

195
tests/test_litellm/llms/anthropic/messages/test_advisor_orchestration.py
View file

@ -558,9 +558,14 @@ async def _run_advisor_and_capture_subcall_kwargs():
return advisor_advice_resp
return final_resp
with patch(
"litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor._call_messages_handler",
side_effect=mock_call,
with (
patch(
"litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor._call_messages_handler",
side_effect=mock_call,
),
patch(
"litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor.validate_url",
),
):
h = AdvisorOrchestrationHandler()
await h.handle(
@ -730,3 +735,187 @@ async def test_advisor_uses_tool_credentials_when_clientside_enabled():
captured = await _run_advisor_and_capture_subcall_kwargs()
assert captured["api_key"] == "sk-other"
assert captured["api_base"] == "https://other.example"
# ---------------------------------------------------------------------------
# 14. _resolve_advisor_credentials: api_base is only honored alongside a
# caller-supplied api_key, and is SSRF-validated before use.
# ---------------------------------------------------------------------------
def test_resolve_advisor_credentials_returns_none_when_gate_closed():
from litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor import (
_resolve_advisor_credentials,
)
with patch(
"litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor._allow_client_side_advisor_credentials",
return_value=False,
):
result = _resolve_advisor_credentials(ADVISOR_TOOL_WITH_CREDS)
assert result == (None, None)
def test_resolve_advisor_credentials_allows_api_key_without_api_base():
from litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor import (
_resolve_advisor_credentials,
)
tool = {**ADVISOR_TOOL, "api_key": "sk-other"}
with (
patch(
"litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor._allow_client_side_advisor_credentials",
return_value=True,
),
patch(
"litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor.validate_url",
side_effect=AssertionError("validate_url must not run without an api_base"),
),
):
result = _resolve_advisor_credentials(tool)
assert result == ("sk-other", None)
def test_resolve_advisor_credentials_rejects_api_base_without_api_key():
from litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor import (
_resolve_advisor_credentials,
)
tool = {**ADVISOR_TOOL, "api_base": "https://other.example"}
with patch(
"litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor._allow_client_side_advisor_credentials",
return_value=True,
):
with pytest.raises(ValueError, match="api_base"):
_resolve_advisor_credentials(tool)
def test_resolve_advisor_credentials_validates_api_base_before_use():
from litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor import (
_resolve_advisor_credentials,
)
with (
patch(
"litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor._allow_client_side_advisor_credentials",
return_value=True,
),
patch(
"litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor.validate_url"
) as mock_validate,
):
result = _resolve_advisor_credentials(ADVISOR_TOOL_WITH_CREDS)
mock_validate.assert_called_once_with("https://other.example")
assert result == ("sk-other", "https://other.example")
def test_resolve_advisor_credentials_propagates_ssrf_error():
from litellm.litellm_core_utils.url_utils import SSRFError
from litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor import (
_resolve_advisor_credentials,
)
with (
patch(
"litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor._allow_client_side_advisor_credentials",
return_value=True,
),
patch(
"litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor.validate_url",
side_effect=SSRFError("URL targets a blocked address"),
),
):
with pytest.raises(SSRFError):
_resolve_advisor_credentials(ADVISOR_TOOL_WITH_CREDS)
def test_resolve_advisor_credentials_skips_validation_when_url_validation_disabled():
import litellm
from litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor import (
_resolve_advisor_credentials,
)
with (
patch(
"litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor._allow_client_side_advisor_credentials",
return_value=True,
),
patch.object(litellm, "user_url_validation", False),
patch(
"litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor.validate_url",
side_effect=AssertionError("validate_url must not run when user_url_validation is disabled"),
),
):
result = _resolve_advisor_credentials(ADVISOR_TOOL_WITH_CREDS)
assert result == ("sk-other", "https://other.example")
def test_resolve_advisor_credentials_blocks_real_cloud_metadata_address():
"""End-to-end (no mocked validate_url): a caller can't redirect the
advisor sub-call to the cloud-metadata address even with an api_key."""
from litellm.litellm_core_utils.url_utils import SSRFError
from litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor import (
_resolve_advisor_credentials,
)
tool = {
**ADVISOR_TOOL,
"api_key": "sk-other",
"api_base": "https://169.254.169.254/latest/meta-data/",
}
with patch(
"litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor._allow_client_side_advisor_credentials",
return_value=True,
):
with pytest.raises(SSRFError):
_resolve_advisor_credentials(tool)
def test_resolve_advisor_credentials_rejects_non_https_api_base():
from litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor import (
_resolve_advisor_credentials,
)
tool = {**ADVISOR_TOOL, "api_key": "sk-other", "api_base": "http://8.8.8.8"}
with patch(
"litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor._allow_client_side_advisor_credentials",
return_value=True,
):
with pytest.raises(ValueError, match="https"):
_resolve_advisor_credentials(tool)
def test_resolve_advisor_credentials_rejects_api_base_when_ssl_verify_disabled():
import litellm
from litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor import (
_resolve_advisor_credentials,
)
tool = {**ADVISOR_TOOL, "api_key": "sk-other", "api_base": "https://8.8.8.8"}
with (
patch(
"litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor._allow_client_side_advisor_credentials",
return_value=True,
),
patch.object(litellm, "ssl_verify", False),
):
with pytest.raises(ValueError, match="ssl_verify"):
_resolve_advisor_credentials(tool)
def test_resolve_advisor_credentials_allows_real_public_ip_address():
"""End-to-end (no mocked validate_url): a globally-routable literal IP
api_base is honored when paired with an api_key."""
from litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor import (
_resolve_advisor_credentials,
)
tool = {**ADVISOR_TOOL, "api_key": "sk-other", "api_base": "https://8.8.8.8"}
with patch(
"litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor._allow_client_side_advisor_credentials",
return_value=True,
):
result = _resolve_advisor_credentials(tool)
assert result == ("sk-other", "https://8.8.8.8")