From 0c9fda8c1eb4093efb3a76c817e4d5d386729a74 Mon Sep 17 00:00:00 2001
From: ryan-crabbe-berri <ryan@berri.ai>
Date: Fri, 11 Sep 2026 18:00:13 -0700
Subject: [PATCH] fix(proxy): gate the webhook test alert on proxy admins

/health/services?service=webhook fired a budget_crossed alert for the
caller's own user_id with any authenticated key. That alert writes the
same dedup cache entry the auth-time user budget alert uses, so a
non-admin could pre-populate it and suppress their real budget alert
for the cache TTL. Match the newrelic and pointfive branches and reject
non-admin callers with a 403 before the alert fires.
---
 .../health_endpoints/_health_endpoints.py     |  5 ++
 .../health_endpoints/test_health_endpoints.py | 55 +++++++++++++++++++
 2 files changed, 60 insertions(+)

diff --git a/litellm/proxy/health_endpoints/_health_endpoints.py b/litellm/proxy/health_endpoints/_health_endpoints.py
index bf527e1e868..175dc1b3172 100644
--- a/litellm/proxy/health_endpoints/_health_endpoints.py
+++ b/litellm/proxy/health_endpoints/_health_endpoints.py
@@ -437,6 +437,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,
diff --git a/tests/test_litellm/proxy/health_endpoints/test_health_endpoints.py b/tests/test_litellm/proxy/health_endpoints/test_health_endpoints.py
index 527d46931fe..577baaef895 100644
--- a/tests/test_litellm/proxy/health_endpoints/test_health_endpoints.py
+++ b/tests/test_litellm/proxy/health_endpoints/test_health_endpoints.py
@@ -1191,6 +1191,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):
     """