fix(mcp): security/correctness fixes for user-fields header injection and BYOK annotation

- resolve_user_field_headers: switch from str.format to str.replace so
  admin-supplied templates cannot use Python format-string attribute or
  item access (e.g. {value.__class__}) to leak object internals into
  outbound HTTP headers. Update fallback test to assert literal token
  preservation, and add a regression test for the attribute-access case.
- _annotate_user_credential_flags: skip OAuth2 rows when populating
  has_user_credential for is_byok=True servers. Otherwise a server
  reconfigured from OAuth2 to BYOK shows a misleading 'Connected' badge
  while the actual tool call would fail (get_user_credential filters
  OAuth2 rows out at execution time). Re-uses the already-decrypted
  plaintext, so no extra crypto cost.

Co-authored-by: Yassin Kortam <yassin@berri.ai>

litellm/proxy/_experimental/mcp_server/user_fields.py

@@ -168,10 +168,16 @@ def resolve_user_field_headers(
         if not value:
             continue
         template = entry.get("header_value_template") or "{value}"
+        # Use plain substitution rather than ``str.format``: the latter
+        # exposes attribute / item access (``{value.__class__}``, ``{value[0]}``)
+        # which an admin-supplied template could use — intentionally or by
+        # accident — to leak Python object internals into outbound HTTP
+        # headers. ``str.replace`` only matches the literal ``{value}`` token.
         try:
-            headers[header_name] = template.format(value=value)
-        except (KeyError, IndexError, ValueError):
-            # Malformed template — fall back to raw value rather than crashing.
+            headers[header_name] = template.replace("{value}", value)
+        except (TypeError, AttributeError):
+            # Non-string template (e.g. corrupt JSONB row) — fall back to
+            # raw value rather than crashing the request.
             verbose_logger.warning(
                 "MCP user_fields: invalid header_value_template %r for field %r "
                 "on server %s; falling back to raw value.",

litellm/proxy/management_endpoints/mcp_management_endpoints.py

@@ -859,6 +859,8 @@ if MCP_AVAILABLE:
         """Populate ``has_user_credential`` and ``missing_user_field_keys``
         for the calling user across BYOK / user-fields servers in a single
         batched query."""
+        import json as _json
+
         from litellm.proxy._experimental.mcp_server.db import (
             _decode_user_credential,
             _parse_user_fields_plaintext,
@@ -886,12 +888,24 @@ if MCP_AVAILABLE:
             payload = _parse_user_fields_plaintext(decoded) if decoded else None
             if payload is not None:
                 user_fields_by_server[row.server_id] = payload
-            else:
-                # Either a BYOK credential or an undecryptable row (e.g.
-                # after a salt-key rotation). Either way, a credential row
-                # exists for this (user, server), so surface it as present
-                # instead of silently telling the user to reconnect.
-                byok_set.add(row.server_id)
+                continue
+            # Skip OAuth2 rows so a server that's been reconfigured from
+            # OAuth2 to BYOK doesn't show "Connected" while the actual
+            # tool call would fail (``get_user_credential`` filters OAuth2
+            # rows out at execution time). Re-uses the already-decrypted
+            # plaintext, so no extra crypto cost.
+            if decoded is not None:
+                try:
+                    parsed = _json.loads(decoded)
+                except (ValueError, TypeError):
+                    parsed = None
+                if isinstance(parsed, dict) and parsed.get("type") == "oauth2":
+                    continue
+            # Either a BYOK credential or an undecryptable row (e.g.
+            # after a salt-key rotation). Either way, a credential row
+            # exists for this (user, server), so surface it as present
+            # instead of silently telling the user to reconnect.
+            byok_set.add(row.server_id)
         for server in servers:
             if getattr(server, "is_byok", False):
                 server.has_user_credential = server.server_id in byok_set

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

@@ -293,8 +293,11 @@ def test_resolve_user_field_headers_skips_unset_values():
     assert headers == {"X-Workspace": "ws1"}
 
 
-def test_resolve_user_field_headers_falls_back_on_bad_template():
-    """A broken template must not crash the request path."""
+def test_resolve_user_field_headers_passes_unknown_placeholders_through():
+    """Templates with non-``{value}`` placeholders are preserved literally
+    instead of triggering format-string evaluation. ``str.replace`` only
+    matches the literal ``{value}`` token, so unrelated braces flow through
+    untouched and never crash the request path."""
     srv = _gmail_server(
         user_fields=[
             {
@@ -306,8 +309,25 @@ def test_resolve_user_field_headers_falls_back_on_bad_template():
         ]
     )
     headers = resolve_user_field_headers(srv, {"TOKEN": "raw"})
-    # Falls back to the raw value rather than raising.
-    assert headers == {"Authorization": "raw"}
+    assert headers == {"Authorization": "Bearer {unknown_placeholder}"}
+
+
+def test_resolve_user_field_headers_does_not_evaluate_attribute_access():
+    """A template containing ``{value.__class__}`` must NOT evaluate the
+    attribute traversal — that would leak Python internals into outbound
+    HTTP headers. With ``str.replace`` it is preserved as a literal token."""
+    srv = _gmail_server(
+        user_fields=[
+            {
+                "field_key": "TOKEN",
+                "header_name": "Authorization",
+                "header_value_template": "Bearer {value.__class__}",
+                "required": True,
+            }
+        ]
+    )
+    headers = resolve_user_field_headers(srv, {"TOKEN": "raw"})
+    assert headers == {"Authorization": "Bearer {value.__class__}"}
 
 
 def test_resolve_user_field_headers_skips_entries_missing_header_name():