scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-08-28 05:25:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 36

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

Browse source 
The handler returned decrypted callback environment values and alerting
routing values verbatim to callers who were not full PROXY_ADMIN. Gate
those on full-admin role, matching the posture used on the sibling
config-inspection endpoints. Non-sensitive routing fields (host / base
URL / port style values) stay visible so the UI can still label which
integration is wired up. Full PROXY_ADMIN sees everything unchanged so
the edit form round-trips on save.

Resolves LIT-4115.
This commit is contained in:
yucheng-berri 2026-07-01 17:58:31 -07:00 • committed by GitHub
parent 6e023f7cf2
commit 8ce6b4d712
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 315 additions and 22 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

17
litellm/proxy/common_utils/callback_utils.py
View file

@ -611,10 +611,17 @@ 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)
@ -622,7 +629,7 @@ def _is_sensitive_callback_var(key: str) -> bool:
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

64
litellm/proxy/proxy_server.py
View file

@ -102,6 +102,7 @@ from litellm.proxy._types import (
)
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,
)
@ -14310,6 +14311,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"],
@ -14720,7 +14765,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
@ -14735,6 +14782,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", [])
@ -14776,6 +14825,8 @@ async def get_config():
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 = []
@ -14787,11 +14838,13 @@ async def get_config():
_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",
@ -14811,8 +14864,9 @@ async def get_config():
"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(
{

12
tests/proxy_unit_tests/test_proxy_server.py
View file

@ -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

216
tests/test_litellm/proxy/proxy_server/test_routes_config.py
View file

@ -741,6 +741,222 @@ def test_get_config_callbacks_internal_error(client, auth_as, mock_prisma, monke
)
_CALLBACK_ENV_FIXTURE = {
"LANGFUSE_PUBLIC_KEY": "pk-public-1234567890",
"LANGFUSE_SECRET_KEY": "sk-langfuse-super-secret",
"LANGFUSE_HOST": "https://cloud.langfuse.com",
"DD_API_KEY": "dd-super-secret-api-key",
"DD_SITE": "datadoghq.com",
"OTEL_HEADERS": "Authorization=Bearer otel-super-secret",
"OTEL_ENDPOINT": "https://otlp.example.com",
"SLACK_WEBHOOK_URL": "https://hooks.slack.com/services/T000/B000/SLACK-WEBHOOK-FIXTURE-SECRET",
}
def _install_callbacks_config(monkeypatch, mock_prisma):
from litellm.proxy import proxy_server as ps
_install_litellm_config(mock_prisma)
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
monkeypatch.setattr(ps, "llm_router", None)
fake_proxy_config = MagicMock()
fake_proxy_config.get_config = AsyncMock(
return_value={
"litellm_settings": {"success_callback": ["langfuse", "datadog", "otel"]},
"general_settings": {"alerting": ["slack"]},
"environment_variables": dict(_CALLBACK_ENV_FIXTURE),
}
)
monkeypatch.setattr(ps, "proxy_config", fake_proxy_config)
def _callback_variables(body: dict, name: str) -> dict:
return next(
cb["variables"] for cb in body["callbacks"] if cb["name"] == name
)
def test_get_config_callbacks_redacts_secret_env_vars_for_view_only_admin(
client, auth_as, mock_prisma, monkeypatch
):
from litellm.proxy._types import LitellmUserRoles
_install_callbacks_config(monkeypatch, mock_prisma)
with auth_as(LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY):
response = client.get("/get/config/callbacks")
assert response.status_code == 200
body = response.json()
for secret in (
_CALLBACK_ENV_FIXTURE["LANGFUSE_SECRET_KEY"],
_CALLBACK_ENV_FIXTURE["DD_API_KEY"],
_CALLBACK_ENV_FIXTURE["OTEL_HEADERS"],
_CALLBACK_ENV_FIXTURE["LANGFUSE_PUBLIC_KEY"],
):
assert secret not in response.text
langfuse_vars = _callback_variables(body, "langfuse")
assert langfuse_vars["LANGFUSE_PUBLIC_KEY"] == "REDACTED"
assert langfuse_vars["LANGFUSE_SECRET_KEY"] == "REDACTED"
assert langfuse_vars["LANGFUSE_HOST"] == _CALLBACK_ENV_FIXTURE["LANGFUSE_HOST"]
datadog_vars = _callback_variables(body, "datadog")
assert datadog_vars["DD_API_KEY"] == "REDACTED"
assert datadog_vars["DD_SITE"] == _CALLBACK_ENV_FIXTURE["DD_SITE"]
otel_vars = _callback_variables(body, "otel")
assert otel_vars["OTEL_HEADERS"] == "REDACTED"
assert otel_vars["OTEL_ENDPOINT"] == _CALLBACK_ENV_FIXTURE["OTEL_ENDPOINT"]
def test_get_config_callbacks_full_admin_still_sees_secret_env_vars(
client, auth_as, mock_prisma, monkeypatch
):
from litellm.proxy._types import LitellmUserRoles
_install_callbacks_config(monkeypatch, mock_prisma)
with auth_as(LitellmUserRoles.PROXY_ADMIN):
response = client.get("/get/config/callbacks")
assert response.status_code == 200
body = response.json()
langfuse_vars = _callback_variables(body, "langfuse")
assert langfuse_vars["LANGFUSE_SECRET_KEY"] == _CALLBACK_ENV_FIXTURE["LANGFUSE_SECRET_KEY"]
assert langfuse_vars["LANGFUSE_PUBLIC_KEY"] == _CALLBACK_ENV_FIXTURE["LANGFUSE_PUBLIC_KEY"]
datadog_vars = _callback_variables(body, "datadog")
assert datadog_vars["DD_API_KEY"] == _CALLBACK_ENV_FIXTURE["DD_API_KEY"]
otel_vars = _callback_variables(body, "otel")
assert otel_vars["OTEL_HEADERS"] == _CALLBACK_ENV_FIXTURE["OTEL_HEADERS"]
def test_get_config_callbacks_redacts_slack_webhook_urls_for_view_only_admin(
client, auth_as, mock_prisma, monkeypatch
):
from litellm.proxy import proxy_server as ps
from litellm.proxy._types import LitellmUserRoles
_install_callbacks_config(monkeypatch, mock_prisma)
webhooks = {
"spend_reports": "https://hooks.slack.com/services/T000/B000/SPEND-WEBHOOK-SECRET",
"budget_alerts": "https://hooks.slack.com/services/T000/B111/BUDGET-WEBHOOK-SECRET",
}
monkeypatch.setattr(
ps.proxy_logging_obj.slack_alerting_instance,
"alert_to_webhook_url",
webhooks,
raising=False,
)
def _slack_block(body):
return next(a for a in body["alerts"] if a["name"] == "slack")
with auth_as(LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY):
view_resp = client.get("/get/config/callbacks")
assert view_resp.status_code == 200
for url in webhooks.values():
assert url not in view_resp.text
assert _CALLBACK_ENV_FIXTURE["SLACK_WEBHOOK_URL"] not in view_resp.text
view_slack = _slack_block(view_resp.json())
assert view_slack["alerts_to_webhook"] == {
"spend_reports": "REDACTED",
"budget_alerts": "REDACTED",
}
assert view_slack["variables"]["SLACK_WEBHOOK_URL"] == "REDACTED"
with auth_as(LitellmUserRoles.PROXY_ADMIN):
admin_resp = client.get("/get/config/callbacks")
assert admin_resp.status_code == 200
admin_slack = _slack_block(admin_resp.json())
assert admin_slack["alerts_to_webhook"] == webhooks
assert admin_slack["variables"]["SLACK_WEBHOOK_URL"] != "REDACTED"
def test_redact_callback_env_vars_helper_handles_none_and_non_secret_keys():
from litellm.proxy import proxy_server as ps
out = ps._redact_callback_env_vars(
{
"LANGFUSE_SECRET_KEY": "sk-leak",
"LANGFUSE_HOST": "https://cloud.langfuse.com",
"DD_API_KEY": None,
"GALILEO_USERNAME": "galileo-user-1234",
"GENERIC_LOGGER_HEADERS": "Authorization=Bearer x",
"GCS_PATH_SERVICE_ACCOUNT": "/etc/secrets/gcs.json",
"SLACK_WEBHOOK_URL": "https://hooks.slack.com/services/T/B/token",
"SMTP_USERNAME": "smtp-user-1234",
}
)
assert out == {
"LANGFUSE_SECRET_KEY": "REDACTED",
"LANGFUSE_HOST": "https://cloud.langfuse.com",
"DD_API_KEY": None,
"GALILEO_USERNAME": "REDACTED",
"GENERIC_LOGGER_HEADERS": "REDACTED",
"GCS_PATH_SERVICE_ACCOUNT": "REDACTED",
"SLACK_WEBHOOK_URL": "REDACTED",
"SMTP_USERNAME": "REDACTED",
}
def test_get_config_callbacks_redacts_email_alerting_vars_for_view_only_admin(
client, auth_as, mock_prisma, monkeypatch
):
from litellm.proxy import proxy_server as ps
from litellm.proxy._types import LitellmUserRoles
_install_litellm_config(mock_prisma)
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
monkeypatch.setattr(ps, "llm_router", None)
fake_proxy_config = MagicMock()
fake_proxy_config.get_config = AsyncMock(
return_value={
"litellm_settings": {"success_callback": []},
"general_settings": {"alerting": ["email"]},
"environment_variables": {
"SMTP_HOST": "smtp.resend.com",
"SMTP_PORT": "587",
"SMTP_USERNAME": "smtp-user-fixture-1234",
"SMTP_PASSWORD": "smtp-password-fixture-1234",
"SMTP_SENDER_EMAIL": "alerts@example.com",
"TEST_EMAIL_ADDRESS": "admin@example.com",
"EMAIL_LOGO_URL": "https://example.com/logo.png",
"EMAIL_SUPPORT_CONTACT": "support@example.com",
},
}
)
monkeypatch.setattr(ps, "proxy_config", fake_proxy_config)
def _email_block(body):
return next(a for a in body["alerts"] if a["name"] == "email")
with auth_as(LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY):
view_resp = client.get("/get/config/callbacks")
assert view_resp.status_code == 200
for secret in ("smtp-user-fixture-1234", "smtp-password-fixture-1234"):
assert secret not in view_resp.text
view_email = _email_block(view_resp.json())["variables"]
assert view_email["SMTP_PASSWORD"] == "REDACTED"
assert view_email["SMTP_USERNAME"] == "REDACTED"
assert view_email["SMTP_HOST"] == "smtp.resend.com"
assert view_email["SMTP_PORT"] == "587"
assert view_email["SMTP_SENDER_EMAIL"] == "alerts@example.com"
with auth_as(LitellmUserRoles.PROXY_ADMIN):
admin_resp = client.get("/get/config/callbacks")
assert admin_resp.status_code == 200
admin_email = _email_block(admin_resp.json())["variables"]
assert admin_email["SMTP_USERNAME"] == "smtp-user-fixture-1234"
assert admin_email["SMTP_PASSWORD"] != "REDACTED"
assert admin_email["SMTP_HOST"] == "smtp.resend.com"
# ---------------------------------------------------------------------------
# GET /config/yaml
# ---------------------------------------------------------------------------

20
tests/test_litellm/proxy/test_proxy_server.py
View file

@ -896,7 +896,9 @@ def test_get_config_custom_callback_api_env_vars(monkeypatch):
# Bypass auth dependency
original_overrides = app.dependency_overrides.copy()
app.dependency_overrides[user_api_key_auth] = lambda: MagicMock()
app.dependency_overrides[user_api_key_auth] = lambda: UserAPIKeyAuth(
user_role=LitellmUserRoles.PROXY_ADMIN, api_key="sk-1234"
)
client = TestClient(app)
try:
@ -950,7 +952,9 @@ def test_get_config_returns_email_settings(monkeypatch):
monkeypatch.setattr(proxy_config, "get_config", AsyncMock(return_value=config_data))
original_overrides = app.dependency_overrides.copy()
app.dependency_overrides[user_api_key_auth] = lambda: MagicMock()
app.dependency_overrides[user_api_key_auth] = lambda: UserAPIKeyAuth(
user_role=LitellmUserRoles.PROXY_ADMIN, api_key="sk-1234"
)
client = TestClient(app)
try:
@ -1007,7 +1011,9 @@ def test_get_config_returns_slack_webhook(monkeypatch):
monkeypatch.setattr(proxy_config, "get_config", AsyncMock(return_value=config_data))
original_overrides = app.dependency_overrides.copy()
app.dependency_overrides[user_api_key_auth] = lambda: MagicMock()
app.dependency_overrides[user_api_key_auth] = lambda: UserAPIKeyAuth(
user_role=LitellmUserRoles.PROXY_ADMIN, api_key="sk-1234"
)
client = TestClient(app)
try:
@ -1061,7 +1067,9 @@ def test_get_config_cleared_slack_webhook_not_overridden_by_os_env(monkeypatch):
monkeypatch.setattr(proxy_config, "get_config", AsyncMock(return_value=config_data))
original_overrides = app.dependency_overrides.copy()
app.dependency_overrides[user_api_key_auth] = lambda: MagicMock()
app.dependency_overrides[user_api_key_auth] = lambda: UserAPIKeyAuth(
user_role=LitellmUserRoles.PROXY_ADMIN, api_key="sk-1234"
)
client = TestClient(app)
try:
@ -5205,7 +5213,9 @@ def test_get_config_normalizes_string_callbacks(monkeypatch):
monkeypatch.setattr(proxy_config, "get_config", AsyncMock(return_value=config_data))
original_overrides = app.dependency_overrides.copy()
app.dependency_overrides[user_api_key_auth] = lambda: MagicMock()
app.dependency_overrides[user_api_key_auth] = lambda: UserAPIKeyAuth(
user_role=LitellmUserRoles.PROXY_ADMIN, api_key="sk-1234"
)
client = TestClient(app)
try:

8
tests/test_litellm/test_router_retry_policy_update.py
View file

@ -273,7 +273,13 @@ async def test_config_update_persists_and_reads_back_retry_policy(monkeypatch):
assert isinstance(router.retry_policy, RetryPolicy)
assert router.retry_policy.RateLimitErrorRetries == 7
read_back = (await proxy_server.get_config())["router_settings"]["retry_policy"]
read_back = (
await proxy_server.get_config(
user_api_key_dict=UserAPIKeyAuth(
user_role=LitellmUserRoles.PROXY_ADMIN, api_key="sk-1234"
)
)
)["router_settings"]["retry_policy"]
assert read_back.BadRequestErrorRetries == 5
assert read_back.TimeoutErrorRetries == 3
assert read_back.RateLimitErrorRetries == 7