scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-14 23:21:35 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 102

Merge pull request #40814 from BerriAI/litellm_gate_health_services_alert_tests

Browse source 
fix(proxy): gate the webhook test alert on proxy admins
This commit is contained in:
ryan-crabbe-berri 2026-09-12 21:13:30 -07:00 • committed by GitHub
commit 760119681c
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 60 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
litellm/proxy/health_endpoints/_health_endpoints.py
View file

@ -443,6 +443,11 @@ async def health_services_endpoint(
}
return pointfive_health
if service == "webhook":
if not _is_proxy_admin(user_api_key_dict):
webhook_non_admin_detail: Final[_ServiceTestErrorDetail] = {
"error": "Only proxy admins can trigger the webhook test alert."
}
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=webhook_non_admin_detail)
user_info: Final = CallInfo(
token=user_api_key_dict.token or "",
spend=1,

55
tests/test_litellm/proxy/health_endpoints/test_health_endpoints.py
View file

@ -1195,6 +1195,61 @@ async def test_health_services_endpoint_newrelic_allows_proxy_admin(admin_role):
mock_instance.async_health_check.assert_awaited_once()
@pytest.mark.asyncio
@pytest.mark.parametrize(
"role",
[
None,
LitellmUserRoles.INTERNAL_USER,
LitellmUserRoles.INTERNAL_USER_VIEW_ONLY,
LitellmUserRoles.TEAM,
LitellmUserRoles.CUSTOMER,
],
)
async def test_health_services_endpoint_webhook_blocks_non_admin(role):
"""
/health/services?service=webhook fires a real budget_crossed alert for the
caller's user_id and writes the same dedup cache entry the auth-time user
budget alert uses, so a non-admin could suppress their own real alert for
the cache TTL. Only proxy admins may trigger it.
"""
mock_proxy_logging = MagicMock()
mock_proxy_logging.budget_alerts = AsyncMock()
user_api_key_dict = UserAPIKeyAuth(token="non-admin-token", user_id="non-admin-user", user_role=role)
with patch( # test-quality-ok: endpoint reads proxy_server module globals, same pattern as sibling tests
"litellm.proxy.proxy_server.proxy_logging_obj",
mock_proxy_logging,
):
with pytest.raises(ProxyException) as exc_info:
await health_services_endpoint(user_api_key_dict=user_api_key_dict, service="webhook")
assert str(exc_info.value.code) == "403"
mock_proxy_logging.budget_alerts.assert_not_awaited()
@pytest.mark.asyncio
@pytest.mark.parametrize(
"admin_role",
[LitellmUserRoles.PROXY_ADMIN, LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY],
)
async def test_health_services_endpoint_webhook_allows_proxy_admin(admin_role):
mock_proxy_logging = MagicMock()
mock_proxy_logging.budget_alerts = AsyncMock()
user_api_key_dict = UserAPIKeyAuth(token="admin-token", user_id="admin-user", user_role=admin_role)
with patch( # test-quality-ok: endpoint reads proxy_server module globals, same pattern as sibling tests
"litellm.proxy.proxy_server.proxy_logging_obj",
mock_proxy_logging,
):
await health_services_endpoint(user_api_key_dict=user_api_key_dict, service="webhook")
mock_proxy_logging.budget_alerts.assert_awaited_once()
sent = mock_proxy_logging.budget_alerts.await_args.kwargs
assert sent["type"] == "user_budget"
assert sent["user_info"].user_id == "admin-user"
@pytest.fixture(scope="function")
def proxy_client(monkeypatch):
"""