fix(mcp): escape user identities in catalog-related logs

litellm/proxy/_experimental/mcp_server/db.py

@@ -1227,7 +1227,7 @@ def _warn_undecryptable_credential(user_id: str, server_id: str) -> None:
     verbose_proxy_logger.warning(
         "MCP user credential for user=%s server=%s could not be decrypted (likely written under a "
         "previous LITELLM_SALT_KEY); the user is treated as not connected and must re-authorize.",
-        user_id,
+        user_id.replace("\r", "\\r").replace("\n", "\\n"),
         server_id,
     )
 
@@ -1451,7 +1451,7 @@ async def store_user_oauth_credential(
                 "store_user_oauth_credential: existing credential for user=%s server=%s could not be "
                 "decrypted (likely written under a previous LITELLM_SALT_KEY); replacing it with the "
                 "newly authorized OAuth2 token.",
-                user_id,
+                user_id.replace("\r", "\\r").replace("\n", "\\n"),
                 server_id,
             )
 
@@ -1685,7 +1685,7 @@ async def refresh_user_oauth_token(
     if not refresh_token:
         verbose_proxy_logger.debug(
             "refresh_user_oauth_token: no refresh_token stored for user=%s server=%s",
-            user_id,
+            user_id.replace("\r", "\\r").replace("\n", "\\n"),
             server_id,
         )
         return None
@@ -1719,7 +1719,7 @@ async def refresh_user_oauth_token(
     except Exception as exc:
         verbose_proxy_logger.warning(
             "refresh_user_oauth_token: refresh request failed for user=%s server=%s: %s",
-            user_id,
+            user_id.replace("\r", "\\r").replace("\n", "\\n"),
             server_id,
             exc,
         )
@@ -1742,7 +1742,7 @@ async def refresh_user_oauth_token(
     if not access_token:
         verbose_proxy_logger.warning(
             "refresh_user_oauth_token: token response missing access_token for user=%s server=%s",
-            user_id,
+            user_id.replace("\r", "\\r").replace("\n", "\\n"),
             server_id,
         )
         return None
@@ -1776,7 +1776,7 @@ async def refresh_user_oauth_token(
 
     verbose_proxy_logger.info(
         "refresh_user_oauth_token: refreshed token for user=%s server=%s",
-        user_id,
+        user_id.replace("\r", "\\r").replace("\n", "\\n"),
         server_id,
     )
     return await get_user_oauth_credential(prisma_client, user_id, server_id)

litellm/proxy/_experimental/mcp_server/oauth_identity_binding.py

@@ -405,7 +405,7 @@ async def enforce_oauth_identity_binding(
         verbose_logger.warning(
             "oauth_identity_binding audit: server=%s user=%s grant=%s rejected=%s (%s)",
             server.server_id,
-            litellm_user_id,
+            str(litellm_user_id).replace("\r", "\\r").replace("\n", "\\n"),
             grant_type,
             rejection.code,
             rejection.description,

litellm/proxy/_experimental/mcp_server/rest_endpoints.py

@@ -502,7 +502,7 @@ if MCP_AVAILABLE:
         except Exception as e:
             verbose_logger.warning(
                 "_get_user_oauth_extra_headers: failed to retrieve credential for user=%s server=%s: %s",
-                user_id,
+                user_id.replace("\r", "\\r").replace("\n", "\\n"),
                 server_id,
                 e,
             )

litellm/proxy/_experimental/mcp_server/ui_session_utils.py

@@ -94,7 +94,11 @@ async def admitted_user_context(user_api_key_auth: UserAPIKeyAuth) -> UserAPIKey
     try:
         admitted: Final = await MCPRequestHandler.reload_admitted_user(user_id)
     except HTTPException as e:
-        verbose_logger.warning("MCP dashboard session: admitted-subject reload failed for %s: %s", user_id, e.detail)
+        verbose_logger.warning(
+            "MCP dashboard session: admitted-subject reload failed for %s: %s",
+            user_id.replace("\r", "\\r").replace("\n", "\\n"),
+            e.detail,
+        )
         return None
     return admitted.model_copy(update={"parent_otel_span": user_api_key_auth.parent_otel_span})
 

litellm/proxy/management_endpoints/internal_user_endpoints.py

@@ -360,7 +360,7 @@ async def _add_user_to_team(
             verbose_proxy_logger.error(
                 "litellm.proxy.management_endpoints.internal_user_endpoints._add_user_to_team(): "
                 "failed to add user %s to team %s - %s",
-                user_id,
+                user_id.replace("\r", "\\r").replace("\n", "\\n"),
                 team_id,
                 str(e),
             )
@@ -379,7 +379,7 @@ async def _add_user_to_team(
             verbose_proxy_logger.error(
                 "litellm.proxy.management_endpoints.internal_user_endpoints._add_user_to_team(): "
                 "failed to add user %s to team %s - %s",
-                user_id,
+                user_id.replace("\r", "\\r").replace("\n", "\\n"),
                 team_id,
                 str(e),
             )

tests/test_litellm/proxy/_experimental/mcp_server/test_db_credentials.py

@@ -17,6 +17,7 @@ from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 from prisma.models import LiteLLM_MCPServerTable as PrismaMCPServer
+from respx import MockRouter
 
 from litellm.proxy._experimental.mcp_server.db import (
     _decode_user_credential,
@@ -1751,3 +1752,44 @@ async def test_unverified_legacy_cache_cannot_bypass_enforcement(monkeypatch):
     await mcp_per_user_token_cache.set("alice", "srv", "bob", 60)
     assert await module.resolve_user_oauth_access_token("alice", server) is None
     assert await mcp_per_user_token_cache.get("alice", "srv") is None
+
+
+@pytest.mark.parametrize("user_id", ["normal-user", "user\r\nFORGED"])
+def test_undecryptable_credential_logs_escape_user_id(user_id: str, caplog: pytest.LogCaptureFixture) -> None:
+    from litellm.proxy._experimental.mcp_server.db import _warn_undecryptable_credential
+
+    _warn_undecryptable_credential(user_id, "srv-1")
+
+    message: Final = caplog.records[-1].getMessage()
+    assert len(message.splitlines()) == 1
+    assert user_id.replace("\r", "\\r").replace("\n", "\\n") in message
+    assert "could not be decrypted" in message
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("response_status", [None, 503, 200])
+@pytest.mark.respx(using="httpx", assert_all_called=False)
+async def test_refresh_failure_logs_escape_user_id(
+    response_status: int | None, respx_mock: MockRouter, caplog: pytest.LogCaptureFixture
+) -> None:
+    from litellm._logging import verbose_proxy_logger
+    from litellm.proxy._experimental.mcp_server.db import refresh_user_oauth_token
+
+    caplog.set_level(10, logger=verbose_proxy_logger.name)
+    route: Final = respx_mock.post("https://idp.example.com/token").respond(response_status or 200, json={})
+
+    result: Final = await refresh_user_oauth_token(
+        prisma_client=MagicMock(),
+        user_id="user\r\nFORGED",
+        server=_refresh_server(),
+        cred={} if response_status is None else {"refresh_token": "rt"},
+    )
+
+    messages: Final = tuple(
+        record.getMessage() for record in caplog.records if record.funcName == "refresh_user_oauth_token"
+    )
+    assert result is None
+    assert route.called is (response_status is not None)
+    assert len(messages) == 1
+    assert "user\\r\\nFORGED" in messages[0].splitlines()[0]
+    assert "\r\nFORGED" not in messages[0]

tests/test_litellm/proxy/_experimental/mcp_server/test_oauth_identity_binding.py

@@ -471,12 +471,12 @@ async def test_refresh_with_mismatched_id_token_rejected():
 
 
 @pytest.mark.asyncio
-async def test_audit_mode_logs_but_does_not_reject(caplog):
+async def test_audit_mode_logs_but_does_not_reject(caplog: pytest.LogCaptureFixture) -> None:
     token: Final = _sign_id_token({"email": "mallory@example.com", "email_verified": True})
     result: Final = await enforce_oauth_identity_binding(
         server=_server(mode="audit"),
         token_response={"access_token": "at", "id_token": token},
-        litellm_user_id="user-a",
+        litellm_user_id="user-a\r\nFORGED",
         grant_type="authorization_code",
         refresh_ownership=None,
         jwks_fetcher=_jwks_fetcher,
@@ -486,6 +486,7 @@ async def test_audit_mode_logs_but_does_not_reject(caplog):
     assert "oauth_principal_mismatch" in caplog.text
     assert "nonce" not in caplog.text
 
+    assert all("\n" not in record.getMessage() and "\r" not in record.getMessage() for record in caplog.records)
 
 @pytest.mark.asyncio
 async def test_unverified_email_rejected():

tests/test_litellm/proxy/_experimental/mcp_server/test_rest_endpoints.py

@@ -4701,3 +4701,25 @@ class TestClientAllowlistOnRestRoutes:
         )
 
         assert result.model_dump() in ({"prompts": []}, {"resources": [], "resource_templates": []})
+
+
+@pytest.mark.asyncio
+async def test_oauth_header_failure_escapes_logged_identity(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    from litellm.proxy._experimental.mcp_server import db as mcp_db
+
+    server: Final = MCPServer(
+        server_id="oauth2-srv", name="oauth2-srv", url="https://upstream.example.com/mcp",
+        transport=MCPTransport.http, auth_type=MCPAuth.oauth2, delegate_auth_to_upstream=True,
+    )
+    auth: Final = UserAPIKeyAuth(user_id="user\r\nFORGED")
+    resolver: Final = AsyncMock(side_effect=RuntimeError("lookup failed"))
+    monkeypatch.setattr(mcp_db, "resolve_valid_user_oauth_token", resolver)
+
+    result: Final = await rest_endpoints._get_user_oauth_extra_headers(server, auth, prefetched_creds={})
+
+    assert result is None
+    assert resolver.await_args.kwargs["user_id"] == "user\r\nFORGED"
+    assert "failed to retrieve credential" in caplog.text
+    assert all("\n" not in record.getMessage() and "\r" not in record.getMessage() for record in caplog.records)

tests/test_litellm/proxy/_experimental/mcp_server/test_ui_session_utils.py

@@ -164,8 +164,10 @@ async def test_build_effective_auth_contexts_never_widens_caller_passed_keys(mon
 
 
 @pytest.mark.asyncio
-async def test_build_effective_auth_contexts_survives_admitted_reload_failure(monkeypatch):
-    user_auth = UserAPIKeyAuth(team_id=UI_SESSION_TOKEN_TEAM_ID, user_id="user-9")
+async def test_build_effective_auth_contexts_survives_admitted_reload_failure(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    user_auth = UserAPIKeyAuth(team_id=UI_SESSION_TOKEN_TEAM_ID, user_id="user-9\r\nFORGED")
 
     monkeypatch.setattr(
         "litellm.proxy._experimental.mcp_server.ui_session_utils.resolve_ui_session_team_ids",
@@ -179,6 +181,9 @@ async def test_build_effective_auth_contexts_survives_admitted_reload_failure(mo
     contexts = await build_effective_auth_contexts(user_auth)
 
     assert [ctx.team_id for ctx in contexts] == ["team-a"]
+    assert contexts[0].user_id == "user-9\r\nFORGED"
+    assert all("\n" not in record.getMessage() and "\r" not in record.getMessage() for record in caplog.records)
+    assert "admitted-subject reload failed" in caplog.text
 
 
 @pytest.mark.asyncio

tests/test_litellm/proxy/management_endpoints/test_internal_user_endpoints.py

@@ -4603,3 +4603,24 @@ async def test_user_update_hashes_and_persists_strong_password(_admin_prisma, mo
     written_data = mock_prisma_client.update_data.call_args.kwargs["data"]
     assert written_data.get("password") is not None
     assert written_data["password"] != strong_password
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("http_error", [True, False])
+async def test_team_add_failure_escapes_logged_identity(
+    http_error: bool, mocker: MockerFixture, caplog: pytest.LogCaptureFixture
+) -> None:
+    from litellm.proxy.management_endpoints.internal_user_endpoints import _add_user_to_team
+
+    failure: Final = HTTPException(status_code=503, detail="unavailable") if http_error else RuntimeError("unavailable")
+    mocker.patch(  # test-quality-ok: inject a failing downstream team operation to exercise both wrapper error handlers
+        "litellm.proxy.management_endpoints.team_endpoints.team_member_add", side_effect=failure
+    )
+    if http_error:
+        await _add_user_to_team("user\r\nFORGED", "team", UserAPIKeyAuth())
+    else:
+        with pytest.raises(RuntimeError, match="unavailable"):
+            await _add_user_to_team("user\r\nFORGED", "team", UserAPIKeyAuth())
+
+    assert "failed to add user" in caplog.text
+    assert all("\n" not in record.getMessage() and "\r" not in record.getMessage() for record in caplog.records)