fix(proxy): tighten role gating on /get/config/callbacks response

litellm/proxy/common_utils/callback_utils.py

@@ -611,18 +611,25 @@ def _transform_callback_vars(metadata: Any, transform: Callable[[str, Any], Any]
     return out
 
 
-def _is_sensitive_callback_var(key: str) -> bool:
-    """Match codebase precedent: only credential-bearing fields get encrypted;
-    routing/identifier fields (host, base_url, project, region) stay plain."""
-    if key in _EXTRA_SENSITIVE_CALLBACK_KEYS:
+def is_sensitive_callback_key(
+    key: str,
+    extra: Optional[set[str]] = None,
+) -> bool:
+    """Return ``True`` if ``key`` is present in ``extra`` (checked as-is), or
+    if its lowercase form is in ``_EXTRA_SENSITIVE_CALLBACK_KEYS``, or if
+    ``_CALLBACK_VAR_MASKER.is_sensitive_key`` matches it.
+    """
+    if extra and key in extra:
+        return True
+    if key.lower() in _EXTRA_SENSITIVE_CALLBACK_KEYS:
         return True
     return _CALLBACK_VAR_MASKER.is_sensitive_key(key)
 
 
 def _encrypt_if_plaintext(key: str, value: Any) -> Any:
     if not isinstance(value, str) or not value:
         return value
-    if not _is_sensitive_callback_var(key):
+    if not is_sensitive_callback_key(key):
         return value
     if value.startswith(_CALLBACK_VAR_ENCRYPTED_PREFIX):
         # Already encrypted — round-tripping ciphertext (e.g. UI Edit Settings

litellm/proxy/proxy_server.py

@@ -102,6 +102,7 @@
 )
 from litellm.proxy.common_utils.cache_pydantic_utils import CacheCodec
 from litellm.proxy.common_utils.callback_utils import (
+    is_sensitive_callback_key,
     normalize_callback_names,
     process_callback,
 )
@@ -14311,6 +14312,50 @@ async def create_config_audit_log(
     )
 
 
+_EXTRA_SECRET_CALLBACK_ENV_VARS = frozenset(
+    {
+        "GALILEO_USERNAME",
+        "GENERIC_LOGGER_HEADERS",
+        "OTEL_HEADERS",
+        "SLACK_WEBHOOK_URL",
+        "SMTP_USERNAME",
+    }
+)
+
+
+def _redact_callback_env_vars(env_vars: dict[str, Optional[str]]) -> dict[str, Optional[str]]:
+    """Return a copy of ``env_vars`` with values for keys classified as
+    sensitive by ``is_sensitive_callback_key`` replaced with ``"REDACTED"``.
+    ``None`` values pass through unchanged.
+    """
+    return {
+        key: (
+            "REDACTED"
+            if value is not None and is_sensitive_callback_key(key, extra=_EXTRA_SECRET_CALLBACK_ENV_VARS)
+            else value
+        )
+        for key, value in env_vars.items()
+    }
+
+
+def _apply_callback_role_gate(entries: list, is_full_admin: bool) -> list:
+    if is_full_admin:
+        return entries
+    return [{**entry, "variables": _redact_callback_env_vars(entry.get("variables") or {})} for entry in entries]
+
+
+def _apply_alerting_env_role_gate(env_vars: dict, is_full_admin: bool) -> dict:
+    if is_full_admin:
+        return mask_sensitive_keys(env_vars, _ALERTING_SENSITIVE_VARS)
+    return _redact_callback_env_vars(env_vars)
+
+
+def _apply_webhook_role_gate(webhook_map, is_full_admin: bool):
+    if is_full_admin or not isinstance(webhook_map, dict):
+        return webhook_map
+    return {alert_type: "REDACTED" for alert_type in webhook_map}
+
+
 @router.get(
     "/config/field/info",
     tags=["config.yaml"],
@@ -14721,7 +14766,9 @@ async def delete_callback(
     include_in_schema=False,
     dependencies=[Depends(user_api_key_auth)],
 )
-async def get_config():
+async def get_config(
+    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
+):
     """
     For Admin UI - allows admin to view config via UI
     # return the callbacks and the env variables for the callback
@@ -14736,6 +14783,8 @@ async def get_config():
         _general_settings = config_data.get("general_settings", {})
         environment_variables = config_data.get("environment_variables", {})
 
+        is_full_admin = user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN
+
         _success_callbacks = _litellm_settings.get("success_callback", [])
         _failure_callbacks = _litellm_settings.get("failure_callback", [])
         _success_and_failure_callbacks = _litellm_settings.get("callbacks", [])
@@ -14777,6 +14826,8 @@ def normalize_callback(callback):
         for _callback in _success_and_failure_callbacks:
             _data_to_return.append(process_callback(_callback, "success_and_failure", environment_variables))
 
+        _data_to_return = _apply_callback_role_gate(_data_to_return, is_full_admin)
+
         # Check if slack alerting is on
         _alerting = _general_settings.get("alerting", [])
         alerting_data = []
@@ -14788,11 +14839,13 @@ def normalize_callback(callback):
                 _var: (value if (value := environment_variables.get(_var)) is not None else os.getenv(_var))
                 for _var in _slack_vars
             }
-            _slack_env_vars = mask_sensitive_keys(_slack_env_vars, _ALERTING_SENSITIVE_VARS)
+            _slack_env_vars = _apply_alerting_env_role_gate(_slack_env_vars, is_full_admin)
 
             _alerting_types = proxy_logging_obj.slack_alerting_instance.alert_types
             _all_alert_types = proxy_logging_obj.slack_alerting_instance._all_possible_alert_types()
-            _alerts_to_webhook = proxy_logging_obj.slack_alerting_instance.alert_to_webhook_url
+            _alerts_to_webhook = _apply_webhook_role_gate(
+                proxy_logging_obj.slack_alerting_instance.alert_to_webhook_url, is_full_admin
+            )
             alerting_data.append(
                 {
                     "name": "slack",
@@ -14812,8 +14865,9 @@ def normalize_callback(callback):
             "EMAIL_LOGO_URL",
             "EMAIL_SUPPORT_CONTACT",
         ]
-        _email_env_vars = {_var: environment_variables.get(_var) for _var in _email_vars}
-        _email_env_vars = mask_sensitive_keys(_email_env_vars, _ALERTING_SENSITIVE_VARS)
+        _email_env_vars = _apply_alerting_env_role_gate(
+            {_var: environment_variables.get(_var) for _var in _email_vars}, is_full_admin
+        )
 
         alerting_data.append(
             {

tests/proxy_unit_tests/test_proxy_server.py

@@ -2844,7 +2844,9 @@ async def test_get_config_callbacks_with_all_types(client_no_auth):
 async def test_get_config_callbacks_environment_variables(client_no_auth):
     """
     Test that /get/config/callbacks correctly includes environment variables
-    for each callback type. Values are returned as-is from the config (no decryption).
+    for each callback type. Under ``client_no_auth`` the resolved role is
+    not ``PROXY_ADMIN``, so values matched by the redaction helper come back
+    as ``"REDACTED"`` and other values pass through verbatim.
     """
     from litellm.proxy.proxy_server import ProxyConfig
 
@@ -2886,12 +2888,11 @@ async def test_get_config_callbacks_environment_variables(client_no_auth):
         assert langfuse_callback["type"] == "success"
         assert "variables" in langfuse_callback
 
-        # Verify langfuse env vars are present (values returned as-is, no decryption)
         langfuse_vars = langfuse_callback["variables"]
         assert "LANGFUSE_PUBLIC_KEY" in langfuse_vars
-        assert langfuse_vars["LANGFUSE_PUBLIC_KEY"] == "test-public-key"
+        assert langfuse_vars["LANGFUSE_PUBLIC_KEY"] == "REDACTED"
         assert "LANGFUSE_SECRET_KEY" in langfuse_vars
-        assert langfuse_vars["LANGFUSE_SECRET_KEY"] == "test-secret-key"
+        assert langfuse_vars["LANGFUSE_SECRET_KEY"] == "REDACTED"
         assert "LANGFUSE_HOST" in langfuse_vars
         assert langfuse_vars["LANGFUSE_HOST"] == "https://cloud.langfuse.com"
 
@@ -2901,14 +2902,13 @@ async def test_get_config_callbacks_environment_variables(client_no_auth):
         assert otel_callback["type"] == "success_and_failure"
         assert "variables" in otel_callback
 
-        # Verify otel env vars are present
         otel_vars = otel_callback["variables"]
         assert "OTEL_EXPORTER" in otel_vars
         assert otel_vars["OTEL_EXPORTER"] == "otlp"
         assert "OTEL_ENDPOINT" in otel_vars
         assert otel_vars["OTEL_ENDPOINT"] == "http://localhost:4317"
         assert "OTEL_HEADERS" in otel_vars
-        assert otel_vars["OTEL_HEADERS"] == "key=value"
+        assert otel_vars["OTEL_HEADERS"] == "REDACTED"
 
 
 @pytest.mark.asyncio