chore(proxy): redact all pass-through credential fields, not just headers

A pass-through endpoint also forwards default_query_params (an upstream API
key is often a query param) and can embed credentials in the target URL
userinfo. Mask both for non-admin callers alongside headers, so a read-only
caller cannot read them from /config/pass_through_endpoint.

litellm/proxy/pass_through_endpoints/pass_through_endpoints.py

@@ -38,7 +38,10 @@ from litellm.constants import MAXIMUM_TRACEBACK_LINES_TO_LOG
 from litellm.integrations.custom_logger import CustomLogger
 from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
 from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
-from litellm.litellm_core_utils.sensitive_data_masker import mask_sensitive_keys
+from litellm.litellm_core_utils.sensitive_data_masker import (
+    mask_sensitive_keys,
+    mask_url_credentials,
+)
 from litellm.llms.custom_httpx.http_handler import get_async_httpx_client
 from litellm.passthrough import BasePassthroughUtils
 from litellm.proxy._types import (
@@ -2634,18 +2637,26 @@ async def _filter_endpoints_by_team_allowed_routes(
     return pass_through_endpoints
 
 
-def _mask_pass_through_endpoint_headers(
+def _mask_pass_through_endpoint_secrets(
     endpoint: PassThroughGenericEndpoint,
 ) -> PassThroughGenericEndpoint:
-    """Return a copy of the endpoint with forwarded header values masked.
+    """Return a copy of the endpoint with forwarded credentials masked.
 
-    Header names stay visible so a read-only caller can still see which headers
-    are configured; the values (which carry upstream credentials) are redacted.
+    A pass-through endpoint carries upstream credentials in three places: the
+    forwarded ``headers``, the ``default_query_params`` (an API key is often
+    passed as a query param), and the ``target`` URL userinfo. Names/structure
+    stay visible so a read-only caller can still see what is configured; the
+    values are redacted.
     """
-    if not endpoint.headers:
-        return endpoint
     masked = endpoint.model_copy(deep=True)
-    masked.headers = mask_sensitive_keys(masked.headers, set(masked.headers.keys()))
+    if masked.headers:
+        masked.headers = mask_sensitive_keys(masked.headers, set(masked.headers.keys()))
+    if masked.default_query_params:
+        masked.default_query_params = mask_sensitive_keys(
+            masked.default_query_params, set(masked.default_query_params.keys())
+        )
+    if masked.target:
+        masked.target = mask_url_credentials(masked.target)
     return masked
 
 
@@ -2707,7 +2718,7 @@ async def get_pass_through_endpoints(
     # (read-only admin, team callers) gets the values masked.
     if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN:
         pass_through_endpoints = [
-            _mask_pass_through_endpoint_headers(endpoint)
+            _mask_pass_through_endpoint_secrets(endpoint)
             for endpoint in pass_through_endpoints
         ]
 

tests/test_litellm/proxy/test_readonly_admin_secret_redaction.py

@@ -114,35 +114,58 @@ def test_redact_audit_log_values_masks_short_secrets():
 # --------------------------------------------------------------------------- #
 
 
-def _make_endpoint(headers):
+def _make_endpoint(
+    headers=None,
+    default_query_params=None,
+    target="https://upstream.example.com",
+):
     from litellm.proxy._types import PassThroughGenericEndpoint
 
     return PassThroughGenericEndpoint(
-        path="/foo", target="https://upstream.example.com", headers=headers
+        path="/foo",
+        target=target,
+        headers=headers or {},
+        default_query_params=default_query_params or {},
     )
 
 
-def test_mask_pass_through_endpoint_headers_redacts_values_keeps_names():
+def test_mask_pass_through_endpoint_secrets_redacts_all_credential_fields():
     from litellm.proxy.pass_through_endpoints.pass_through_endpoints import (
-        _mask_pass_through_endpoint_headers,
+        _mask_pass_through_endpoint_secrets,
     )
 
-    ep = _make_endpoint({"Authorization": "Bearer sk-upstream-secret-token"})
-    masked = _mask_pass_through_endpoint_headers(ep)
+    ep = _make_endpoint(
+        headers={"Authorization": "Bearer sk-upstream-secret-token"},
+        default_query_params={"api_key": "qp-secret-key-value"},
+        target="https://user:targetpw@upstream.example.com/v1",
+    )
+    masked = _mask_pass_through_endpoint_secrets(ep)
 
-    assert "Authorization" in masked.headers  # name preserved
-    assert masked.headers["Authorization"] != "Bearer sk-upstream-secret-token"
+    # Header value masked, name preserved.
+    assert "Authorization" in masked.headers
     assert "sk-upstream-secret-token" not in masked.headers["Authorization"]
+    # default_query_params value masked, name preserved.
+    assert "api_key" in masked.default_query_params
+    assert "qp-secret-key-value" not in masked.default_query_params["api_key"]
+    # target URL userinfo password redacted.
+    assert "targetpw" not in masked.target
     # Original object is not mutated.
     assert ep.headers["Authorization"] == "Bearer sk-upstream-secret-token"
+    assert ep.default_query_params["api_key"] == "qp-secret-key-value"
 
 
 @pytest.mark.asyncio
-async def test_pass_through_get_masks_headers_for_non_admin_only():
+async def test_pass_through_get_masks_secrets_for_non_admin_only():
     import litellm.proxy.pass_through_endpoints.pass_through_endpoints as pt
 
-    secret = "Bearer sk-upstream-secret-token"
-    endpoints = [_make_endpoint({"Authorization": secret})]
+    header_secret = "Bearer sk-upstream-secret-token"
+    qp_secret = "qp-secret-key-value"
+    endpoints = [
+        _make_endpoint(
+            headers={"Authorization": header_secret},
+            default_query_params={"api_key": qp_secret},
+        )
+    ]
 
     with (
         patch.object(
@@ -160,9 +183,12 @@ async def test_pass_through_get_masks_headers_for_non_admin_only():
             user_api_key_dict=_user(LitellmUserRoles.PROXY_ADMIN)
         )
 
-    assert secret not in json.dumps(viewer_resp.endpoints[0].headers)
-    # Full admin can still read the plaintext header (it is editable for them).
-    assert admin_resp.endpoints[0].headers["Authorization"] == secret
+    viewer_blob = json.dumps([e.model_dump() for e in viewer_resp.endpoints])
+    assert header_secret not in viewer_blob
+    assert qp_secret not in viewer_blob
+    # Full admin still reads plaintext (the endpoint is editable for them).
+    assert admin_resp.endpoints[0].headers["Authorization"] == header_secret
+    assert admin_resp.endpoints[0].default_query_params["api_key"] == qp_secret
 
 
 # --------------------------------------------------------------------------- #