From 3e7eaceb9ca3db34c85d3f04f0e61834fccb5757 Mon Sep 17 00:00:00 2001 From: zerone0x Date: Tue, 1 Sep 2026 03:53:33 +0800 Subject: [PATCH 1/5] fix(passthrough): enforce default_on and request-level post-call guardrails A mode: post_call guardrail (e.g. tool_permission with on_disallowed_action: block) attached to a pass-through route was consulted but never enforced unless the endpoint's own guardrails config listed it: the response-side hook in pass_through_request was gated solely on the endpoint-level opt-in collection, so default_on: true guardrails and per-request `guardrails` body params - both honored by pre_call on the same routes - silently degraded into a plain relay of the upstream response. On the provider-native passthrough routes (/anthropic/*, /gemini/*, ...) there is no endpoint-level config at all, so post-call enforcement never ran. Run the post-call hook whenever any registered guardrail would fire for post_call: keep the endpoint-level opt-in fast path, and otherwise probe should_run_guardrail per registered CustomGuardrail (PassthroughGuardrailHandler.has_applicable_post_call_guardrail). Restore request-level `guardrails` params - stripped from the forwarded body by _init_kwargs_for_pass_through_endpoint - onto the hook data so post_call sees the same attach pre_call saw. Plain pass-through traffic with no applicable guardrail keeps skipping the hook entirely, so non-guardrail callbacks do not start firing. Key/team-attached guardrails without endpoint-level opt-in are unchanged: their inheritance is documented as gated on the endpoint enabling passthrough guardrails. Streaming responses return before this section and stay out of scope. Fixes #32201 Co-Authored-By: Claude --- .../pass_through_endpoints.py | 61 +++-- .../passthrough_guardrails.py | 44 ++++ .../test_passthrough_guardrails.py | 76 ++++++ .../test_passthrough_post_call_guardrails.py | 243 ++++++++++++++++++ 4 files changed, 403 insertions(+), 21 deletions(-) diff --git a/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py b/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py index 09d3dedaafa..ada5b281c0c 100644 --- a/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py +++ b/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py @@ -1370,33 +1370,52 @@ async def pass_through_request( request_payload=failure_request_payload, ) - if response.status_code < 400 and response_body is not None and guardrails_to_run: + if response.status_code < 400 and response_body is not None: # Build an enriched data dict: _parsed_body has been stripped of # `metadata` by both pre_call_hook and _init_kwargs_for_pass_through_endpoint, # so we re-attach the configured guardrails here so should_run_guardrail # sees them. hook_data: Final = dict(_parsed_body or {}) - existing_metadata = hook_data.get("metadata") - if not isinstance(existing_metadata, dict): - existing_metadata = {} - hook_data["metadata"] = { - **existing_metadata, - "guardrails": guardrails_to_run, - } - post_call_guardrail_data = hook_data - response_body = await proxy_logging_obj.post_call_success_hook( - data=hook_data, - user_api_key_dict=user_api_key_dict, - response=response_body, - ) - if isinstance(response_body, dict): - content = json.dumps(response_body).encode("utf-8") - _content_modified = True - else: - verbose_proxy_logger.debug( - "pass_through_endpoint: post_call_success_hook returned %s, expected dict — using original response", - type(response_body).__name__, + if guardrails_to_run: + existing_metadata = hook_data.get("metadata") + if not isinstance(existing_metadata, dict): + existing_metadata = {} + hook_data["metadata"] = { + **existing_metadata, + "guardrails": guardrails_to_run, + } + # Per-request `guardrails` from the request body were popped out of + # `_parsed_body` by _init_kwargs_for_pass_through_endpoint (so they + # are not forwarded upstream) and parked in kwargs["litellm_params"]. + # pre_call_hook ran before that strip and honored them; restore them + # here so post-call enforcement sees the same request-level attach. + # Endpoint-level guardrails keep precedence: a top-level "guardrails" + # key would shadow the metadata entry in get_guardrail_from_metadata. + request_level_guardrails: Final = (kwargs.get("litellm_params") or {}).get("guardrails") if kwargs else None + if request_level_guardrails and not guardrails_to_run and "guardrails" not in hook_data: + hook_data["guardrails"] = request_level_guardrails + # Endpoint-level opt-in is not the only way a post-call guardrail + # is attached: `default_on: true` guardrails and per-request + # `guardrails` body params apply here too (issue #32201). Gating + # solely on `guardrails_to_run` silently relayed responses those + # guardrails were configured to block. Only invoke the hook when + # a guardrail would actually run, so plain pass-through traffic + # does not start triggering non-guardrail callback hooks. + if guardrails_to_run or PassthroughGuardrailHandler.has_applicable_post_call_guardrail(hook_data): + post_call_guardrail_data = hook_data + response_body = await proxy_logging_obj.post_call_success_hook( + data=hook_data, + user_api_key_dict=user_api_key_dict, + response=response_body, ) + if isinstance(response_body, dict): + content = json.dumps(response_body).encode("utf-8") + _content_modified = True + else: + verbose_proxy_logger.debug( + "pass_through_endpoint: post_call_success_hook returned %s, expected dict — using original response", + type(response_body).__name__, + ) elif response_body is None: verbose_proxy_logger.debug( "pass_through_endpoint: response body not JSON-parseable, skipping post-call guardrails" diff --git a/litellm/proxy/pass_through_endpoints/passthrough_guardrails.py b/litellm/proxy/pass_through_endpoints/passthrough_guardrails.py index de9b0acf081..e1c3f1b4406 100644 --- a/litellm/proxy/pass_through_endpoints/passthrough_guardrails.py +++ b/litellm/proxy/pass_through_endpoints/passthrough_guardrails.py @@ -273,6 +273,50 @@ class PassthroughGuardrailHandler: return guardrails_to_run if guardrails_to_run else None + @staticmethod + def has_applicable_post_call_guardrail(data: dict) -> bool: + """ + Check whether any registered guardrail would run post_call for ``data``. + + Endpoint-level opt-in (``collect_guardrails``) is not the only way a + post-call guardrail can be attached to a pass-through request: + guardrails configured with ``default_on: true`` and per-request + ``guardrails`` body params must enforce too — pre-call enforcement on + pass-through already honors both, and so do the normal completion + routes. Gating the response-side hook on endpoint opt-in alone made a + ``mode: post_call`` guardrail with ``on_disallowed_action: block`` + silently degrade into a plain relay of the upstream response + (issue #32201). + + Errors from ``should_run_guardrail`` count as applicable so the real + ``post_call_success_hook`` invocation surfaces them the same way the + completion routes would, instead of silently skipping enforcement + (fail closed). + """ + import litellm + from litellm.integrations.custom_guardrail import CustomGuardrail + from litellm.litellm_core_utils.litellm_logging import ( + get_custom_logger_compatible_class, + ) + from litellm.types.guardrails import GuardrailEventHooks + + for callback in litellm.callbacks: + resolved = get_custom_logger_compatible_class(callback) if isinstance(callback, str) else callback + if not isinstance(resolved, CustomGuardrail): + continue + try: + if resolved.should_run_guardrail(data=data, event_type=GuardrailEventHooks.post_call): + return True + except Exception: + verbose_proxy_logger.debug( + "Passthrough post-call guardrail applicability check errored for %s; " + "running the post-call hook so the error surfaces (fail closed)", + getattr(resolved, "guardrail_name", type(resolved).__name__), + exc_info=True, + ) + return True + return False + @staticmethod def get_field_targeted_text( data: dict, diff --git a/tests/test_litellm/proxy/pass_through_endpoints/test_passthrough_guardrails.py b/tests/test_litellm/proxy/pass_through_endpoints/test_passthrough_guardrails.py index 40403de846f..038f0bb292a 100644 --- a/tests/test_litellm/proxy/pass_through_endpoints/test_passthrough_guardrails.py +++ b/tests/test_litellm/proxy/pass_through_endpoints/test_passthrough_guardrails.py @@ -260,3 +260,79 @@ class TestPassthroughGuardrailHandlerPrepareOutput: assert "targeted1" in result assert "targeted2" in result assert "ignored" not in result + + +class TestHasApplicablePostCallGuardrail: + """Tests for PassthroughGuardrailHandler.has_applicable_post_call_guardrail. + + The post-call hook in pass_through_request is gated on this check when no + endpoint-level guardrails are configured, so that default_on / request-level + guardrails still enforce (issue #32201) while plain pass-through traffic + keeps skipping the hook entirely. + """ + + def _register_guardrail(self, monkeypatch, guardrail): + import litellm + + monkeypatch.setattr(litellm, "callbacks", [guardrail]) + + def _make_guardrail(self, event_hook: str, default_on: bool): + from litellm.integrations.custom_guardrail import CustomGuardrail + + return CustomGuardrail( + guardrail_name="test-guard", + event_hook=event_hook, + default_on=default_on, + ) + + def test_returns_false_with_no_callbacks(self, monkeypatch): + import litellm + + monkeypatch.setattr(litellm, "callbacks", []) + assert PassthroughGuardrailHandler.has_applicable_post_call_guardrail({}) is False + + def test_returns_true_for_default_on_post_call_guardrail(self, monkeypatch): + guardrail = self._make_guardrail(event_hook="post_call", default_on=True) + self._register_guardrail(monkeypatch, guardrail) + assert PassthroughGuardrailHandler.has_applicable_post_call_guardrail({}) is True + + def test_returns_false_for_default_on_pre_call_only_guardrail(self, monkeypatch): + """A pre_call-only guardrail never runs post_call, so the hook stays skipped.""" + guardrail = self._make_guardrail(event_hook="pre_call", default_on=True) + self._register_guardrail(monkeypatch, guardrail) + assert PassthroughGuardrailHandler.has_applicable_post_call_guardrail({}) is False + + def test_returns_false_for_unattached_non_default_on_guardrail(self, monkeypatch): + """Opt-in stays opt-in: a registered but unrequested guardrail does not + trigger the post-call hook.""" + guardrail = self._make_guardrail(event_hook="post_call", default_on=False) + self._register_guardrail(monkeypatch, guardrail) + assert PassthroughGuardrailHandler.has_applicable_post_call_guardrail({}) is False + + def test_returns_true_for_request_attached_guardrail(self, monkeypatch): + guardrail = self._make_guardrail(event_hook="post_call", default_on=False) + self._register_guardrail(monkeypatch, guardrail) + data = {"guardrails": ["test-guard"]} + assert PassthroughGuardrailHandler.has_applicable_post_call_guardrail(data) is True + + def test_returns_false_for_non_guardrail_callbacks(self, monkeypatch): + """Plain CustomLogger callbacks (loggers, budget hooks) must not flip + the check — otherwise every pass-through response would start invoking + post_call_success_hook.""" + import litellm + from litellm.integrations.custom_logger import CustomLogger + + monkeypatch.setattr(litellm, "callbacks", [CustomLogger()]) + assert PassthroughGuardrailHandler.has_applicable_post_call_guardrail({}) is False + + def test_fails_closed_when_should_run_guardrail_errors(self, monkeypatch): + """If applicability cannot be determined, run the hook so the error + surfaces there instead of silently skipping enforcement.""" + guardrail = self._make_guardrail(event_hook="post_call", default_on=True) + + def _boom(**kwargs): + raise RuntimeError("boom") + + monkeypatch.setattr(guardrail, "should_run_guardrail", _boom) + self._register_guardrail(monkeypatch, guardrail) + assert PassthroughGuardrailHandler.has_applicable_post_call_guardrail({}) is True diff --git a/tests/test_litellm/proxy/pass_through_endpoints/test_passthrough_post_call_guardrails.py b/tests/test_litellm/proxy/pass_through_endpoints/test_passthrough_post_call_guardrails.py index 9d1975513a1..6023e301830 100644 --- a/tests/test_litellm/proxy/pass_through_endpoints/test_passthrough_post_call_guardrails.py +++ b/tests/test_litellm/proxy/pass_through_endpoints/test_passthrough_post_call_guardrails.py @@ -3,6 +3,10 @@ Tests for post-call guardrail invocation on pass-through endpoints. Verifies that apply_guardrail(input_type="response") is called for non-streaming pass-through responses. Addresses issue #20270. + +Also verifies that post-call guardrails enforce (block / rewrite) no matter +how they are attached — endpoint-level config, ``default_on: true``, or a +per-request ``guardrails`` body param. Addresses issue #32201. """ import json @@ -13,6 +17,7 @@ import httpx import pytest from fastapi import HTTPException +import litellm from litellm.integrations.custom_guardrail import ( CustomGuardrail, ModifyResponseException, @@ -312,3 +317,241 @@ def test_modify_response_exception_importable_from_both_paths(): ) assert FromExceptions is FromGuardrail + + +# --------------------------------------------------------------------------- +# Issue #32201: post_call guardrails attached to a pass-through endpoint are +# consulted but never enforce. +# +# These tests run the REAL dispatch (real ProxyLogging + real +# ToolPermissionGuardrail registered in litellm.callbacks) so a regression in +# any layer — the gate in pass_through_request, should_run_guardrail, or the +# guardrail's dict-response handling — fails them. +# --------------------------------------------------------------------------- + +_ANTHROPIC_DENIED_TOOL_USE_RESPONSE = { + "id": "msg_1", + "type": "message", + "role": "assistant", + "model": "claude-x", + "stop_reason": "tool_use", + "content": [ + { + "type": "tool_use", + "id": "t1", + "name": "Bash", + "input": {"command": "rm -rf /"}, + } + ], +} + +_ANTHROPIC_SAFE_RESPONSE = { + "id": "msg_2", + "type": "message", + "role": "assistant", + "model": "claude-x", + "stop_reason": "tool_use", + "content": [ + { + "type": "tool_use", + "id": "t2", + "name": "Bash", + "input": {"command": "ls -la"}, + } + ], +} + +_TOOL_FIREWALL_RULES = [ + { + "id": "allow_safe_bash", + "tool_name": "Bash", + "decision": "allow", + "allowed_param_patterns": { + "command": r"^(?!.*(rm\s+-rf|terraform\s+destroy|kubectl\s+delete)).*$" + }, + } +] + + +def _make_tool_firewall_guardrail(default_on: bool): + from litellm.proxy.guardrails.guardrail_hooks.tool_permission import ( + ToolPermissionGuardrail, + ) + + return ToolPermissionGuardrail( + guardrail_name="tool-firewall", + event_hook="post_call", + rules=_TOOL_FIREWALL_RULES, + default_action="deny", + on_disallowed_action="block", + default_on=default_on, + ) + + +def _make_real_dispatch_user_api_key_dict(): + d = _make_user_api_key_dict(request_route="/anthropic/v1/messages") + # collect_guardrails reads these when an endpoint-level config is present + d.metadata = {} + d.team_metadata = {} + return d + + +def _real_dispatch_patches(upstream_body: dict, request_body: dict): + """Patches for end-to-end pass_through_request runs with a REAL ProxyLogging.""" + from litellm.caching.dual_cache import DualCache + from litellm.proxy.utils import ProxyLogging + + proxy_logging_obj = ProxyLogging(user_api_key_cache=DualCache()) + + mock_response = _make_httpx_response(upstream_body) + mock_async_client_obj = MagicMock() + mock_async_client_obj.client = AsyncMock() + mock_pt_logging = MagicMock() + mock_pt_logging.pass_through_async_success_handler = AsyncMock() + + patches = [ + patch( + f"{_PT_MOD}.HttpPassThroughEndpointHelpers.non_streaming_http_request_handler", + new_callable=AsyncMock, + return_value=mock_response, + ), + patch(f"{_PT_MOD}._is_streaming_response", return_value=False), + patch("litellm.proxy.proxy_server.proxy_logging_obj", proxy_logging_obj), + patch(f"{_PT_MOD}.pass_through_endpoint_logging", mock_pt_logging), + patch(f"{_PT_MOD}.get_async_httpx_client", return_value=mock_async_client_obj), + patch( + f"{_PT_MOD}._read_request_body", + new_callable=AsyncMock, + return_value=request_body, + ), + patch(f"{_PT_MOD}._safe_get_request_headers", return_value={}), + ] + + stack = ExitStack() + for p in patches: + stack.enter_context(p) + return stack + + +_PASSTHROUGH_REQUEST_BODY = { + "model": "claude-x", + "max_tokens": 64, + "messages": [{"role": "user", "content": "go"}], +} + + +@pytest.mark.asyncio +class TestPostCallGuardrailEnforcement: + """A post_call guardrail attached to a pass-through endpoint must enforce.""" + + async def test_endpoint_attached_guardrail_blocks_denied_tool_use(self): + """Issue #32201 primary repro: guardrail attached via the endpoint's + `guardrails` config blocks an upstream response with a denied tool_use + instead of relaying it with HTTP 200.""" + guardrail = _make_tool_firewall_guardrail(default_on=False) + litellm.logging_callback_manager.add_litellm_callback(guardrail) + + with _real_dispatch_patches( + _ANTHROPIC_DENIED_TOOL_USE_RESPONSE, dict(_PASSTHROUGH_REQUEST_BODY) + ): + with pytest.raises(ProxyException) as exc_info: + await pass_through_request( + request=_make_mock_request(), + target="https://example.com/v1/messages", + custom_headers={"Content-Type": "application/json"}, + user_api_key_dict=_make_real_dispatch_user_api_key_dict(), + stream=False, + guardrails_config={"tool-firewall": None}, + ) + + assert str(exc_info.value.code) == "400" + assert "tool-firewall" in str(exc_info.value.message) + + async def test_default_on_guardrail_blocks_without_endpoint_config(self): + """A `default_on: true` post_call guardrail must enforce on + pass-through routes that have no endpoint-level guardrails config + (e.g. the provider-native /anthropic/* passthrough).""" + guardrail = _make_tool_firewall_guardrail(default_on=True) + litellm.logging_callback_manager.add_litellm_callback(guardrail) + + with _real_dispatch_patches( + _ANTHROPIC_DENIED_TOOL_USE_RESPONSE, dict(_PASSTHROUGH_REQUEST_BODY) + ): + with pytest.raises(ProxyException) as exc_info: + await pass_through_request( + request=_make_mock_request(), + target="https://example.com/v1/messages", + custom_headers={"Content-Type": "application/json"}, + user_api_key_dict=_make_real_dispatch_user_api_key_dict(), + stream=False, + ) + + assert str(exc_info.value.code) == "400" + + async def test_request_body_guardrails_param_enforced_post_call(self): + """A per-request `guardrails` body param (honored by pre_call today) + must also attach the guardrail for post-call enforcement.""" + guardrail = _make_tool_firewall_guardrail(default_on=False) + litellm.logging_callback_manager.add_litellm_callback(guardrail) + + body = dict(_PASSTHROUGH_REQUEST_BODY) + body["guardrails"] = ["tool-firewall"] + + with _real_dispatch_patches(_ANTHROPIC_DENIED_TOOL_USE_RESPONSE, body): + with pytest.raises(ProxyException) as exc_info: + await pass_through_request( + request=_make_mock_request(), + target="https://example.com/v1/messages", + custom_headers={"Content-Type": "application/json"}, + user_api_key_dict=_make_real_dispatch_user_api_key_dict(), + stream=False, + ) + + assert str(exc_info.value.code) == "400" + + async def test_passing_response_is_relayed_unchanged(self): + """An upstream response whose tool_use passes the rules must be + relayed unchanged with the upstream status code.""" + guardrail = _make_tool_firewall_guardrail(default_on=True) + litellm.logging_callback_manager.add_litellm_callback(guardrail) + + with _real_dispatch_patches( + _ANTHROPIC_SAFE_RESPONSE, dict(_PASSTHROUGH_REQUEST_BODY) + ): + result = await pass_through_request( + request=_make_mock_request(), + target="https://example.com/v1/messages", + custom_headers={"Content-Type": "application/json"}, + user_api_key_dict=_make_real_dispatch_user_api_key_dict(), + stream=False, + ) + + assert result.status_code == 200 + assert json.loads(bytes(result.body)) == _ANTHROPIC_SAFE_RESPONSE + + async def test_unattached_guardrail_skips_post_call_hook(self): + """With a registered but unattached (non-default_on) guardrail, the + post-call hook must not run: pass-through stays opt-in for guardrails + and non-guardrail callbacks must not start firing on plain traffic.""" + guardrail = _make_tool_firewall_guardrail(default_on=False) + litellm.logging_callback_manager.add_litellm_callback(guardrail) + + mock_proxy_logging = MagicMock() + mock_proxy_logging.pre_call_hook = AsyncMock( + return_value=dict(_PASSTHROUGH_REQUEST_BODY) + ) + mock_proxy_logging.post_call_success_hook = AsyncMock() + mock_proxy_logging.post_call_response_headers_hook = AsyncMock(return_value={}) + + mock_response = _make_httpx_response(_ANTHROPIC_DENIED_TOOL_USE_RESPONSE) + with _common_patches(mock_proxy_logging, mock_response): + result = await pass_through_request( + request=_make_mock_request(), + target="https://example.com/v1/messages", + custom_headers={"Content-Type": "application/json"}, + user_api_key_dict=_make_real_dispatch_user_api_key_dict(), + stream=False, + ) + + mock_proxy_logging.post_call_success_hook.assert_not_awaited() + assert result.status_code == 200 From e3f6d85d78e2d87400de6023e6e124807163f723 Mon Sep 17 00:00:00 2001 From: zerone0x Date: Tue, 1 Sep 2026 03:56:02 +0800 Subject: [PATCH 2/5] fix(passthrough): cast string callbacks for the resolver's literal type Mirror the cast post_call_success_hook uses so the applicability probe type-checks the same way the dispatch loop it mirrors does. Co-Authored-By: Claude --- .../pass_through_endpoints/passthrough_guardrails.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/litellm/proxy/pass_through_endpoints/passthrough_guardrails.py b/litellm/proxy/pass_through_endpoints/passthrough_guardrails.py index e1c3f1b4406..3c4879966a7 100644 --- a/litellm/proxy/pass_through_endpoints/passthrough_guardrails.py +++ b/litellm/proxy/pass_through_endpoints/passthrough_guardrails.py @@ -293,7 +293,10 @@ class PassthroughGuardrailHandler: completion routes would, instead of silently skipping enforcement (fail closed). """ + from typing import cast + import litellm + from litellm import _custom_logger_compatible_callbacks_literal from litellm.integrations.custom_guardrail import CustomGuardrail from litellm.litellm_core_utils.litellm_logging import ( get_custom_logger_compatible_class, @@ -301,7 +304,11 @@ class PassthroughGuardrailHandler: from litellm.types.guardrails import GuardrailEventHooks for callback in litellm.callbacks: - resolved = get_custom_logger_compatible_class(callback) if isinstance(callback, str) else callback + resolved = ( + get_custom_logger_compatible_class(cast(_custom_logger_compatible_callbacks_literal, callback)) + if isinstance(callback, str) + else callback + ) if not isinstance(resolved, CustomGuardrail): continue try: From 48d4d33621aab86e33d58dbe880f0905e19deba7 Mon Sep 17 00:00:00 2001 From: zerone0x Date: Tue, 1 Sep 2026 04:00:10 +0800 Subject: [PATCH 3/5] fix(passthrough): resolve string callbacks without a banned cast import The strict-rule budget (TID251) bans new typing.cast imports; resolve string callbacks the same way llm_http_handler does - plain call plus isinstance narrowing - which needs no cast at all. Co-Authored-By: Claude --- .../pass_through_endpoints/passthrough_guardrails.py | 9 +-------- 1 file changed, 1 insertion(+), 8 deletions(-) diff --git a/litellm/proxy/pass_through_endpoints/passthrough_guardrails.py b/litellm/proxy/pass_through_endpoints/passthrough_guardrails.py index 3c4879966a7..e1c3f1b4406 100644 --- a/litellm/proxy/pass_through_endpoints/passthrough_guardrails.py +++ b/litellm/proxy/pass_through_endpoints/passthrough_guardrails.py @@ -293,10 +293,7 @@ class PassthroughGuardrailHandler: completion routes would, instead of silently skipping enforcement (fail closed). """ - from typing import cast - import litellm - from litellm import _custom_logger_compatible_callbacks_literal from litellm.integrations.custom_guardrail import CustomGuardrail from litellm.litellm_core_utils.litellm_logging import ( get_custom_logger_compatible_class, @@ -304,11 +301,7 @@ class PassthroughGuardrailHandler: from litellm.types.guardrails import GuardrailEventHooks for callback in litellm.callbacks: - resolved = ( - get_custom_logger_compatible_class(cast(_custom_logger_compatible_callbacks_literal, callback)) - if isinstance(callback, str) - else callback - ) + resolved = get_custom_logger_compatible_class(callback) if isinstance(callback, str) else callback if not isinstance(resolved, CustomGuardrail): continue try: From a0bd99b79fe1714993fd9e98de747a877378ae79 Mon Sep 17 00:00:00 2001 From: zerone0x Date: Tue, 1 Sep 2026 04:19:50 +0800 Subject: [PATCH 4/5] chore(passthrough): satisfy the type-discipline and test-quality gates - annotate the probe's data param as Mapping (LIT001: no bare dict annotations) and drop the dict-literal fallback when reading the request-level guardrails out of kwargs (LIT002) - declare the kwargs-derived locals as object and carry reasoned pyright ignores for the Unknown chain an untyped kwargs dict forces, so no basedpyright rule count grows over base - give the proxy_logging_obj patch in the new tests its test-quality-ok reason: the real ProxyLogging must be injected where the code under test imports it at call time Co-Authored-By: Claude --- .../pass_through_endpoints/pass_through_endpoints.py | 11 +++++++++-- .../pass_through_endpoints/passthrough_guardrails.py | 3 ++- .../test_passthrough_post_call_guardrails.py | 5 ++++- 3 files changed, 15 insertions(+), 4 deletions(-) diff --git a/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py b/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py index ada5b281c0c..b6f4de34f4c 100644 --- a/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py +++ b/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py @@ -1391,7 +1391,12 @@ async def pass_through_request( # here so post-call enforcement sees the same request-level attach. # Endpoint-level guardrails keep precedence: a top-level "guardrails" # key would shadow the metadata entry in get_guardrail_from_metadata. - request_level_guardrails: Final = (kwargs.get("litellm_params") or {}).get("guardrails") if kwargs else None + _request_litellm_params: Final[object] = ( # pyright: ignore[reportUnknownVariableType, reportUnknownMemberType] # kwargs is an untyped dict + kwargs.get("litellm_params") if kwargs else None + ) + request_level_guardrails: Final[object] = ( # pyright: ignore[reportUnknownVariableType] # value read out of the untyped kwargs dict + _request_litellm_params.get("guardrails") if isinstance(_request_litellm_params, dict) else None # pyright: ignore[reportUnknownMemberType] # isinstance narrows to an unparameterized dict + ) if request_level_guardrails and not guardrails_to_run and "guardrails" not in hook_data: hook_data["guardrails"] = request_level_guardrails # Endpoint-level opt-in is not the only way a post-call guardrail @@ -1401,7 +1406,9 @@ async def pass_through_request( # guardrails were configured to block. Only invoke the hook when # a guardrail would actually run, so plain pass-through traffic # does not start triggering non-guardrail callback hooks. - if guardrails_to_run or PassthroughGuardrailHandler.has_applicable_post_call_guardrail(hook_data): + if guardrails_to_run or PassthroughGuardrailHandler.has_applicable_post_call_guardrail( + hook_data # pyright: ignore[reportUnknownArgumentType] # hook_data mirrors the untyped _parsed_body dict + ): post_call_guardrail_data = hook_data response_body = await proxy_logging_obj.post_call_success_hook( data=hook_data, diff --git a/litellm/proxy/pass_through_endpoints/passthrough_guardrails.py b/litellm/proxy/pass_through_endpoints/passthrough_guardrails.py index e1c3f1b4406..19315d25866 100644 --- a/litellm/proxy/pass_through_endpoints/passthrough_guardrails.py +++ b/litellm/proxy/pass_through_endpoints/passthrough_guardrails.py @@ -7,6 +7,7 @@ Handles guardrail execution for passthrough endpoints with: - Automatic inheritance from org/team/key levels when enabled """ +from collections.abc import Mapping from typing import Any, Final from litellm._logging import verbose_proxy_logger @@ -274,7 +275,7 @@ class PassthroughGuardrailHandler: return guardrails_to_run if guardrails_to_run else None @staticmethod - def has_applicable_post_call_guardrail(data: dict) -> bool: + def has_applicable_post_call_guardrail(data: Mapping[str, object]) -> bool: """ Check whether any registered guardrail would run post_call for ``data``. diff --git a/tests/test_litellm/proxy/pass_through_endpoints/test_passthrough_post_call_guardrails.py b/tests/test_litellm/proxy/pass_through_endpoints/test_passthrough_post_call_guardrails.py index 6023e301830..027e429ab01 100644 --- a/tests/test_litellm/proxy/pass_through_endpoints/test_passthrough_post_call_guardrails.py +++ b/tests/test_litellm/proxy/pass_through_endpoints/test_passthrough_post_call_guardrails.py @@ -416,7 +416,10 @@ def _real_dispatch_patches(upstream_body: dict, request_body: dict): return_value=mock_response, ), patch(f"{_PT_MOD}._is_streaming_response", return_value=False), - patch("litellm.proxy.proxy_server.proxy_logging_obj", proxy_logging_obj), + patch( # test-quality-ok: inject the real ProxyLogging that pass_through_request imports at call time + "litellm.proxy.proxy_server.proxy_logging_obj", + proxy_logging_obj, + ), patch(f"{_PT_MOD}.pass_through_endpoint_logging", mock_pt_logging), patch(f"{_PT_MOD}.get_async_httpx_client", return_value=mock_async_client_obj), patch( From 828ca35062e7c458fe378bf3a19ff839993f0d94 Mon Sep 17 00:00:00 2001 From: zerone0x Date: Tue, 1 Sep 2026 04:36:23 +0800 Subject: [PATCH 5/5] fix(passthrough): annotate custom-logger lookup for the basedpyright budget --- litellm/proxy/pass_through_endpoints/passthrough_guardrails.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/litellm/proxy/pass_through_endpoints/passthrough_guardrails.py b/litellm/proxy/pass_through_endpoints/passthrough_guardrails.py index 19315d25866..f59976d49e8 100644 --- a/litellm/proxy/pass_through_endpoints/passthrough_guardrails.py +++ b/litellm/proxy/pass_through_endpoints/passthrough_guardrails.py @@ -302,7 +302,7 @@ class PassthroughGuardrailHandler: from litellm.types.guardrails import GuardrailEventHooks for callback in litellm.callbacks: - resolved = get_custom_logger_compatible_class(callback) if isinstance(callback, str) else callback + resolved = get_custom_logger_compatible_class(callback) if isinstance(callback, str) else callback # pyright: ignore[reportArgumentType] # resolver wants its callback-name literal; str does not narrow to it, and unknown names return None if not isinstance(resolved, CustomGuardrail): continue try: