fix(agents): deny authoritative policy reads on storage failure

litellm/proxy/auth/auth_checks.py

@@ -3983,6 +3983,8 @@ async def get_object_permission(
         ).find_unique(where={"object_permission_id": object_permission_id})
 
         if response is None:
+            if check_db_only:
+                raise HTTPException(status_code=403, detail="Referenced object permission does not exist")
             return None
 
         _perm_obj: Final = LiteLLM_ObjectPermissionTable.model_validate(response.dict())
@@ -3995,6 +3997,8 @@ async def get_object_permission(
 
         return _perm_obj
     except Exception:
+        if check_db_only:
+            raise
         return None
 
 

tests/test_litellm/proxy/_experimental/mcp_server/auth/test_user_api_key_auth_mcp.py

@@ -10113,3 +10113,25 @@ async def test_fresh_mcp_user_permission_link_ignores_cached_and_replica_grants(
     with pytest.raises(HTTPException) as denied:
         await MCPRequestHandler._user_object_permission_id("fresh-human", database, check_db_only=True)
     assert denied.value.status_code == 503
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("operation", ["servers", "tools"])
+async def test_managed_agent_permission_resolution_outage_is_not_an_unrestricted_grant(monkeypatch, operation):
+    from litellm.proxy._experimental.mcp_server import mcp_server_manager
+    from litellm.types.agents import AgentResponse
+
+    auth = UserAPIKeyAuth(agent_id="managed")
+    auth.managed_agent_policy = AgentResponse(agent_id="managed", agent_name="Managed", agent_card_params={})
+    permission = LiteLLM_ObjectPermissionTable(object_permission_id="policy", mcp_toolsets=["unavailable"])
+    manager = MagicMock()
+    manager.expand_permission_list.return_value = []
+    manager.resolve_toolset_tool_permissions = AsyncMock(side_effect=RuntimeError("policy unavailable"))
+    monkeypatch.setattr(mcp_server_manager, "global_mcp_server_manager", manager)
+    resolution = (
+        MCPRequestHandler.get_allowed_mcp_servers_for_agent(auth, permission)
+        if operation == "servers"
+        else MCPRequestHandler.get_agent_tool_permissions_for_server("slack", auth, permission)
+    )
+    with pytest.raises(RuntimeError, match="policy unavailable"):
+        await resolution

tests/test_litellm/proxy/auth/test_auth_checks.py

@@ -10047,6 +10047,48 @@ async def test_authoritative_access_group_outage_does_not_use_cached_grants() ->
     cache.async_get_cache.assert_not_awaited()
 
 
+@pytest.mark.asyncio
+async def test_authoritative_team_permission_outage_cannot_drop_the_teams_restrictions() -> None:
+    from fastapi import HTTPException
+
+    from litellm.proxy.auth.auth_checks import get_team_object
+
+    row: Final = LiteLLM_TeamTable(team_id="team-policy-outage", object_permission_id="team-permission")
+    client: Final = MagicMock()
+    client.writer_db.litellm_teamtable.find_unique = AsyncMock(return_value=row)
+    client.writer_db.litellm_objectpermissiontable.find_unique = AsyncMock(side_effect=RuntimeError("unavailable"))
+    cache: Final = MagicMock()
+    cache.async_get_cache = AsyncMock()
+    cache.async_set_cache = AsyncMock()
+    with pytest.raises(HTTPException) as failure:
+        await get_team_object(row.team_id, client, cache, check_db_only=True)
+    assert failure.value.status_code == 404
+    client.writer_db.litellm_objectpermissiontable.find_unique.assert_awaited_once()
+    cache.async_set_cache.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("strict", [True, False])
+@pytest.mark.parametrize("missing", [True, False])
+async def test_referenced_permission_failures_preserve_legacy_behavior_and_deny_strict_reads(strict, missing):
+    from fastapi import HTTPException
+
+    from litellm.proxy.auth.auth_checks import get_object_permission
+
+    client = MagicMock()
+    lookup = AsyncMock(return_value=None, side_effect=None if missing else RuntimeError("unavailable"))
+    client.writer_db.litellm_objectpermissiontable.find_unique = lookup
+    client.db.litellm_objectpermissiontable.find_unique = lookup
+    cache = MagicMock()
+    cache.async_get_cache = AsyncMock(return_value=None)
+    if strict:
+        with pytest.raises(HTTPException if missing else RuntimeError):
+            await get_object_permission("referenced", client, cache, check_db_only=True)
+        cache.async_get_cache.assert_not_awaited()
+    else:
+        assert await get_object_permission("referenced", client, cache) is None
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
     "models,key_aliases,team_aliases,allowed",

tests/test_litellm/proxy/management_endpoints/test_mcp_management_endpoints.py

@@ -7338,7 +7338,7 @@ class TestConnectedAppViewAnnotation:
 
         flags = {server.server_id: server.connected_app_reachable for server in result}
         assert flags == {"server-1": True, "server-2": False}
-        reload_mock.assert_awaited_once_with("test_user_id")
+        reload_mock.assert_awaited_once_with("test_user_id", requires_fresh_policy=False)
         mock_manager.get_allowed_mcp_servers.assert_awaited_once_with(admitted_auth)
 
     @pytest.mark.asyncio

tests/test_litellm/proxy/management_endpoints/test_team_endpoints.py

@@ -4355,7 +4355,7 @@ async def test_list_team_v2_org_admin_own_query_keeps_memberships_in_other_orgs(
     prisma_client.db.litellm_teamtable.find_many = AsyncMock(side_effect=find_many)
     prisma_client.db.litellm_teamtable.count = AsyncMock(side_effect=count)
     prisma_client.db.litellm_verificationtoken.group_by = AsyncMock(return_value=[])
-    prisma_client.db.litellm_usertable.find_unique = AsyncMock(
+    prisma_client.writer_db.litellm_usertable.find_unique = AsyncMock(
         return_value=LiteLLM_UserTable(
             user_id="org_admin_user",
             teams=["team_in_org_A", "team_in_org_B"],
@@ -4394,11 +4394,11 @@ async def test_list_team_v2_org_admin_own_query_keeps_memberships_in_other_orgs(
     assert await list_teams(None) == own_view
     assert await list_teams("org_admin_user", search="team_in_org_B") == ["team_in_org_B"]
     assert await list_teams("other_user") == ["other_team_in_org_A"]
-    prisma_client.db.litellm_usertable.find_unique.assert_awaited_with(
+    prisma_client.writer_db.litellm_usertable.find_unique.assert_awaited_with(
         where={"user_id": "org_admin_user"}, include={"organization_memberships": True}
     )
 
-    prisma_client.db.litellm_usertable.find_unique.side_effect = RuntimeError("db down")
+    prisma_client.writer_db.litellm_usertable.find_unique.side_effect = RuntimeError("db down")
     with pytest.raises(ValueError, match="db down"):
         await list_teams("org_admin_user")
 
@@ -16025,7 +16025,7 @@ async def test_get_team_spend_by_user_team_admin_sees_every_member(mock_db_clien
     alpha = _team_spend_by_user_team("team-alpha", "Team Alpha", Member(user_id="alice", role="admin"), [])
     mock_db_client.db.litellm_teamtable.find_many = AsyncMock(return_value=[alpha])
     mock_db_client.db.query_raw = AsyncMock(return_value=[])
-    mock_db_client.db.litellm_usertable.find_unique = AsyncMock(
+    mock_db_client.writer_db.litellm_usertable.find_unique = AsyncMock(
         return_value=_team_spend_by_user_caller("alice", ["team-alpha"])
     )
 
@@ -16047,7 +16047,7 @@ async def test_get_team_spend_by_user_plain_member_only_sees_own_row(mock_db_cli
     mock_db_client.db.litellm_teamtable.find_many = AsyncMock(return_value=[alpha])
     mock_db_client.db.litellm_verificationtoken.find_many = AsyncMock(return_value=[])
     mock_db_client.db.query_raw = AsyncMock(return_value=[_team_spend_by_user_db_row("team-alpha", "bob", 0.25, 2)])
-    mock_db_client.db.litellm_usertable.find_unique = AsyncMock(
+    mock_db_client.writer_db.litellm_usertable.find_unique = AsyncMock(
         return_value=_team_spend_by_user_caller("bob", ["team-alpha"])
     )
 
@@ -16068,7 +16068,7 @@ async def test_get_team_spend_by_user_member_of_other_team_gets_404(mock_db_clie
 
     caller = UserAPIKeyAuth(user_id="bob", user_role=LitellmUserRoles.INTERNAL_USER)
     mock_db_client.db.query_raw = AsyncMock(return_value=[])
-    mock_db_client.db.litellm_usertable.find_unique = AsyncMock(
+    mock_db_client.writer_db.litellm_usertable.find_unique = AsyncMock(
         return_value=_team_spend_by_user_caller("bob", ["team-alpha"])
     )