scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-17 04:34:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 172

feat(proxy): add agent_access_default_deny for least-privilege agent access (#43979)

Browse source 
* feat(proxy): add require_key_agent_access_defined for least-privilege agent access

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(proxy): align A2A model visibility assertions with supported routes

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(proxy): keep A2A model entries on their existing routes

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(proxy): point the listing model at the harness upstream

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(proxy): coerce require_key_agent_access_defined like the settings validator

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(proxy): inject agent access lookups instead of patching the handler

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* refactor(proxy): rename require_key_agent_access_defined to agent_access_default_deny

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

---------

Co-authored-by: ryan <ryan@berri.ai>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
devin-ai-integration[bot] 2026-10-15 16:49:26 -07:00 • committed by GitHub
parent b9079755bb
commit 77c22a4c01
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 576 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
litellm/proxy/_types.py
View file

@ -3027,6 +3027,13 @@ class ConfigGeneralSettings(LiteLLMPydanticObjectBase):
"(see GitHub issue #12905)."
),
)
agent_access_default_deny: bool = Field(
False,
description=(
"When true, a key with no agent grants on itself or its team can reach no agents instead of all agents. "
"Proxy admins are exempt."
),
)
user_url_validation: bool | None = Field(
None,
description=(

51
litellm/proxy/agent_endpoints/auth/agent_permission_handler.py
View file

@ -9,9 +9,10 @@ import asyncio
from collections.abc import Awaitable, Callable, Sequence
from dataclasses import dataclass
from types import MappingProxyType
from typing import Final, TypeAlias
from typing import Final, Protocol, TypeAlias
from fastapi import HTTPException
from pydantic import TypeAdapter, ValidationError
from litellm._logging import verbose_logger
from litellm.proxy._experimental.mcp_server.ui_session_utils import build_effective_auth_contexts
@ -47,6 +48,29 @@ class RestrictedAgentAccess:
AgentAccess: TypeAlias = UnrestrictedAgentAccess | RestrictedAgentAccess
class AgentAccessLookup(Protocol):
def __call__(self, user_api_key_auth: UserAPIKeyAuth | None, /, *, strict: bool) -> Awaitable[AgentAccess]: ...
_DEFAULT_DENY_FLAG: Final = TypeAdapter(bool)
def agent_access_default_deny() -> bool:
from litellm.proxy.proxy_server import general_settings_view
try:
return _DEFAULT_DENY_FLAG.validate_python(general_settings_view().get("agent_access_default_deny", False))
except ValidationError:
return True
def _is_proxy_admin(user_api_key_auth: UserAPIKeyAuth | None) -> bool:
return user_api_key_auth is not None and user_api_key_auth.user_role in (
LitellmUserRoles.PROXY_ADMIN,
LitellmUserRoles.PROXY_ADMIN.value,
)
def _to_stable_ids(agent_ids: frozenset[str]) -> frozenset[str]:
from litellm.proxy.agent_endpoints.agent_registry import global_agent_registry
@ -80,7 +104,7 @@ class AgentRequestHandler:
- If team has restrictions and key has restrictions: use intersection
- If team has restrictions and key has none: inherit from team
- If team has no restrictions: use key restrictions
- If no restrictions: allow all agents
- If no restrictions: allow all agents unless agent_access_default_deny is enabled for a non-admin key
"""
@staticmethod
@ -89,15 +113,22 @@ class AgentRequestHandler:
resolve_ceiling: CeilingResolver = resolve_agent_access_group_ceiling,
*,
strict: bool = False,
default_deny: Callable[[], bool] = agent_access_default_deny,
lookup_key_access: AgentAccessLookup | None = None,
lookup_team_access: AgentAccessLookup | None = None,
) -> AgentAccess:
"""Agents the key may reach: key and team grants, intersected with the agent's access group ceiling
and, for an agent key acting on behalf of an invoking user, with that user's team grants."""
if managed_agent_policy(user_api_key_auth) is not None:
return await _managed_actor_agent_access(user_api_key_auth)
key_team_access: Final = await AgentRequestHandler.resolve_key_team_agent_access(
user_api_key_auth, strict=strict
user_api_key_auth,
strict=strict,
lookup_key_access=lookup_key_access,
lookup_team_access=lookup_team_access,
)
if strict and isinstance(key_team_access, UnrestrictedAgentAccess):
deny_undefined: Final = strict or (not _is_proxy_admin(user_api_key_auth) and default_deny())
if deny_undefined and isinstance(key_team_access, UnrestrictedAgentAccess):
return RestrictedAgentAccess(frozenset())
caller_access: Final = await AgentRequestHandler.agent_caller_access(user_api_key_auth, strict=strict)
own_access: Final = _intersect_agent_access(key_team_access, caller_access)
@ -122,12 +153,18 @@ class AgentRequestHandler:
user_api_key_auth: UserAPIKeyAuth | None,
*,
strict: bool = False,
lookup_key_access: AgentAccessLookup | None = None,
lookup_team_access: AgentAccessLookup | None = None,
) -> AgentAccess:
try:
key_access: Final = await AgentRequestHandler.get_allowed_agents_for_key(user_api_key_auth, strict=strict)
team_access: Final = await AgentRequestHandler._get_allowed_agents_for_team(
user_api_key_auth, strict=strict
key_lookup: Final = (
AgentRequestHandler.get_allowed_agents_for_key if lookup_key_access is None else lookup_key_access
)
team_lookup: Final = (
AgentRequestHandler._get_allowed_agents_for_team if lookup_team_access is None else lookup_team_access
)
key_access: Final = await key_lookup(user_api_key_auth, strict=strict)
team_access: Final = await team_lookup(user_api_key_auth, strict=strict)
except Exception as e:
if strict:
raise HTTPException(503, "Agent invocation policy is unavailable") from e

1
litellm/proxy/proxy_server.py
View file

@ -18824,6 +18824,7 @@ _GENERAL_SETTINGS_CONFIG_LIST_FIELD_TYPES: Final[Mapping[str, str]] = MappingPro
"cancel_on_disconnect": "Boolean",
"disable_auto_add_proxy_admin_to_teams": "Boolean",
"apply_user_budget_to_team_keys": "Boolean",
"agent_access_default_deny": "Boolean",
"user_api_key_cache_max_size": "Integer",
"transcribe_media_buckets": "List",
}

338
tests/integration/authorization/test_agent_least_privilege.py Normal file
View file

@ -0,0 +1,338 @@
import json
import uuid
from contextlib import ExitStack
from pathlib import Path
from typing import Final
import httpx
import pytest
import yaml
from integration._support.client import Gateway, eventually, object_value, string_value
from integration._support.database import read_rows
from integration._support.process import owned_proxy
from integration._support.wire import Reply, Request, Wire, wire_server
def _write_access_config(directory: Path, *, default_deny: bool, model_name: str, api_base: str) -> Path:
source: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text())
config: Final = {
**source,
"model_list": [
{
"model_name": model_name,
"litellm_params": {
"model": "openai/gpt-4o-mini",
"api_key": "integration-provider-key",
"api_base": api_base,
},
}
],
"general_settings": {
**source["general_settings"],
"agent_access_default_deny": default_deny,
},
}
path: Final = directory / f"agent-access-{uuid.uuid4().hex}.yaml"
path.write_text(yaml.safe_dump(config))
return path
def _peer_response(request: Request, wire: Wire, agent_name: str) -> Reply:
if request.method == "GET":
assert request.target in ("/.well-known/agent-card.json", "/.well-known/agent.json")
return Reply(
body=json.dumps(
{
"protocolVersion": "0.3",
"name": agent_name,
"description": "Synthetic permission peer",
"version": "1.0.0",
"url": wire.url + "/",
"capabilities": {"streaming": False},
"defaultInputModes": ["text"],
"defaultOutputModes": ["text"],
"skills": [],
}
).encode()
)
assert request.method == "POST" and request.target == "/"
body: Final = json.loads(request.body)
assert body["jsonrpc"] == "2.0" and body["method"] == "message/send"
message: Final = body["params"]["message"]
assert message["role"] == "user"
assert message["kind"] == "message"
assert message["parts"] in (
[{"kind": "text", "text": "synthetic ping"}],
[{"kind": "text", "text": "user: synthetic ping"}],
)
return Reply(
body=json.dumps(
{
"jsonrpc": "2.0",
"id": body["id"],
"result": {
"kind": "message",
"role": "agent",
"messageId": message["messageId"] + "-out",
"parts": [{"kind": "text", "text": "synthetic pong"}],
},
}
).encode()
)
def _create_agent(candidate: Gateway, wire: Wire, agent_name: str, cleanups: ExitStack) -> str:
created: Final = candidate.post(
"/v1/agents",
{
"agent_name": agent_name,
"agent_card_params": {
"protocolVersion": "0.3",
"name": agent_name,
"description": "Synthetic permission peer",
"version": "1.0.0",
"url": wire.url + "/",
"capabilities": {"streaming": False},
"defaultInputModes": ["text"],
"defaultOutputModes": ["text"],
"skills": [],
},
"litellm_params": {"make_public": False},
},
)
agent_id: Final = string_value(created["agent_id"])
cleanups.callback(_delete_agent, candidate, agent_id)
return agent_id
def _delete_agent(candidate: Gateway, agent_id: str) -> None:
deleted: Final = candidate.request("DELETE", f"/v1/agents/{agent_id}")
assert deleted.status_code == 200, deleted.text
assert read_rows('SELECT agent_id FROM "LiteLLM_AgentsTable" WHERE agent_id=%s', (agent_id,)) == []
def _card(candidate: Gateway, agent_id: str, key: str) -> httpx.Response:
return candidate.request(
"GET",
f"/a2a/{agent_id}/.well-known/agent-card.json",
key=key,
headers={"Connection": "close"},
)
def _send(candidate: Gateway, agent_id: str, key: str, marker: str) -> httpx.Response:
return candidate.request(
"POST",
f"/a2a/{agent_id}",
{
"jsonrpc": "2.0",
"id": marker,
"method": "message/send",
"params": {
"message": {
"kind": "message",
"role": "user",
"messageId": marker + "-in",
"parts": [{"kind": "text", "text": "synthetic ping"}],
}
},
},
key=key,
headers={"Connection": "close"},
)
def _chat_completion(candidate: Gateway, agent_name: str, key: str) -> httpx.Response:
return candidate.request(
"POST",
"/v1/chat/completions",
{
"model": f"a2a/{agent_name}",
"messages": [{"role": "user", "content": "synthetic ping"}],
},
key=key,
headers={"Connection": "close"},
)
def _model_names(candidate: Gateway, path: str, key: str) -> frozenset[str]:
response: Final = candidate.request("GET", path, key=key, headers={"Connection": "close"})
assert response.status_code == 200, response.text
payload: Final = response.json()
raw_entries: Final = payload if isinstance(payload, list) else payload["data"]
entries: Final = tuple(object_value(entry) for entry in raw_entries)
if path == "/model_group/info":
return frozenset(string_value(entry["model_group"]) for entry in entries)
return frozenset(string_value(entry["model_name"]) for entry in entries)
def _listed_agent_ids(candidate: Gateway, key: str) -> frozenset[str]:
response: Final = candidate.request("GET", "/v1/agents", key=key, headers={"Connection": "close"})
assert response.status_code == 200, response.text
return frozenset(string_value(object_value(agent)["agent_id"]) for agent in response.json())
def _assert_successful_send(response: httpx.Response, marker: str) -> None:
assert response.status_code == 200, response.text
body: Final = response.json()
assert body["jsonrpc"] == "2.0" and body["id"] == marker and "error" not in body, response.text
assert body["result"]["messageId"] == marker + "-in-out", response.text
assert body["result"]["parts"] == [{"kind": "text", "text": "synthetic pong"}], response.text
def _assert_denial(response: httpx.Response, agent_id: str) -> None:
assert response.status_code == 403, response.text
assert response.json() == {
"detail": f"Agent '{agent_id}' is not allowed for your key/team. Contact proxy admin for access."
}, response.text
@pytest.mark.timeout(180)
def test_agent_access_default_deny_denies_ungranted_key_but_keeps_explicit_grants(
gateway: Gateway, tmp_path: Path
) -> None:
agent_name: Final = "least-privilege-" + uuid.uuid4().hex
v2_model: Final = "listing-" + agent_name
config: Final = _write_access_config(
tmp_path, default_deny=True, model_name=v2_model, api_base=gateway.upstream_url + "/v1"
)
def upstream(request: Request) -> Reply:
return _peer_response(request, wire, agent_name)
with (
wire_server(upstream) as wire,
owned_proxy(gateway, tmp_path, {}, config=config, workers=2) as candidate,
candidate.scenario() as scenario,
):
agent_id: Final = _create_agent(candidate, wire, agent_name, scenario.cleanups)
ungranted_team: Final = scenario.team()
key_a: Final = scenario.key(team_id=ungranted_team)
key_b: Final = scenario.key(object_permission={"agents": [agent_id]})
granted_team: Final = scenario.team(object_permission={"agents": [agent_id]})
key_c: Final = scenario.key(team_id=granted_team)
wire.drain()
denied_cards: Final = tuple(_card(candidate, agent_id, key_a) for _ in range(8))
_assert_denial(denied_cards[0], agent_id)
assert all(response.json() == denied_cards[0].json() for response in denied_cards), [
response.text for response in denied_cards
]
denied_sends: Final = tuple(
_send(candidate, agent_id, key_a, f"denied-{index}-{uuid.uuid4().hex}") for index in range(8)
)
_assert_denial(denied_sends[0], agent_id)
assert all(response.json() == denied_sends[0].json() for response in denied_sends), [
response.text for response in denied_sends
]
denied_completion: Final = _chat_completion(candidate, agent_name, key_a)
assert denied_completion.status_code == 403, denied_completion.text
assert denied_completion.json() == {
"error": {
"message": f"Agent '{agent_name}' is not allowed for your key/team. Contact proxy admin for access.",
"type": "permission_error",
"param": None,
"code": "403",
}
}, denied_completion.text
denied_agents: Final = tuple(
candidate.request("GET", f"/v1/agents/{agent_id}", key=key_a, headers={"Connection": "close"})
for _ in range(8)
)
_assert_denial(denied_agents[0], agent_id)
assert all(response.json() == denied_agents[0].json() for response in denied_agents), [
response.text for response in denied_agents
]
assert v2_model in _model_names(candidate, "/v2/model/info", candidate.key)
assert _listed_agent_ids(candidate, key_a) == frozenset()
assert f"a2a/{agent_name}" not in _model_names(candidate, "/v2/model/info", key_a)
assert f"a2a/{agent_name}" not in _model_names(candidate, "/model_group/info", key_a)
assert tuple(item for item in wire.drain() if item.method == "POST") == ()
for key in (key_b, key_c, candidate.key):
cards: Final = tuple(_card(candidate, agent_id, key) for _ in range(8))
assert all(response.status_code == 200 for response in cards), [response.text for response in cards]
completion: Final = _chat_completion(candidate, agent_name, key)
assert completion.status_code == 200, completion.text
assert completion.json()["choices"][0]["message"]["content"] == "synthetic pong", completion.text
wire.drain()
for index in range(8):
marker: Final = f"granted-{index}-{uuid.uuid4().hex}"
_assert_successful_send(_send(candidate, agent_id, key, marker), marker)
posts: Final = tuple(item for item in wire.drain() if item.method == "POST")
assert len(posts) == 1, posts
if key != candidate.key:
assert agent_id in _listed_agent_ids(candidate, key)
assert f"a2a/{agent_name}" in _model_names(candidate, "/v2/model/info", key)
@pytest.mark.timeout(180)
@pytest.mark.parametrize("workers", [2, 4])
def test_agent_access_default_deny_toggles_at_runtime_through_config_api(
gateway: Gateway, tmp_path: Path, workers: int
) -> None:
agent_name: Final = "runtime-agent-" + uuid.uuid4().hex
def upstream(request: Request) -> Reply:
return _peer_response(request, wire, agent_name)
with (
wire_server(upstream) as wire,
owned_proxy(
gateway,
tmp_path,
{},
config=Path("tests/integration/proxy_config.yaml"),
workers=workers,
) as candidate,
candidate.scenario() as scenario,
):
agent_id: Final = _create_agent(candidate, wire, agent_name, scenario.cleanups)
ungranted_key: Final = scenario.key()
wire.drain()
assert _card(candidate, agent_id, ungranted_key).status_code == 200
enabled: Final = candidate.request(
"POST",
"/config/field/update",
{
"config_type": "general_settings",
"field_name": "agent_access_default_deny",
"field_value": "true",
},
)
assert enabled.status_code == 200, enabled.text
config_fields: Final = candidate.request(
"GET",
"/config/list",
params={"config_type": "general_settings"},
)
assert config_fields.status_code == 200, config_fields.text
setting: Final = next(
object_value(field)
for field in config_fields.json()
if object_value(field)["field_name"] == "agent_access_default_deny"
)
assert setting["field_type"] == "Boolean" and setting["field_value"] == "true", setting
eventually(
lambda: tuple(_card(candidate, agent_id, ungranted_key) for _ in range(8)),
lambda responses: all(response.status_code == 403 for response in responses),
seconds=40,
)
disabled: Final = candidate.request(
"POST",
"/config/field/update",
{
"config_type": "general_settings",
"field_name": "agent_access_default_deny",
"field_value": False,
},
)
assert disabled.status_code == 200, disabled.text
eventually(
lambda: tuple(_card(candidate, agent_id, ungranted_key) for _ in range(8)),
lambda responses: all(response.status_code == 200 for response in responses),
seconds=40,
)

180
tests/unit/proxy/agent_endpoints/auth/test_agent_permission_handler.py
View file

@ -15,6 +15,7 @@ from litellm.proxy.agent_endpoints.agent_registry import AgentRegistry
from litellm.proxy.agent_endpoints.auth.agent_access_groups import AgentAccessGroupCeiling, CeilingResolver
from litellm.proxy.agent_endpoints.auth.agent_permission_handler import (
AgentAccess,
AgentAccessLookup,
AgentRequestHandler,
RestrictedAgentAccess,
UnrestrictedAgentAccess,
@ -43,6 +44,15 @@ def _agent_id(registry: AgentRegistry, agent_name: str) -> str:
return agent.agent_id
def _lookup(result: AgentAccess | Exception) -> AgentAccessLookup:
async def lookup(_user_api_key_auth: UserAPIKeyAuth | None, /, *, strict: bool) -> AgentAccess:
if isinstance(result, Exception):
raise result
return result
return lookup
async def _single_context(user_api_key_auth: UserAPIKeyAuth) -> list[UserAPIKeyAuth]:
return [user_api_key_auth]
@ -133,6 +143,176 @@ class TestAgentRequestHandler:
)
assert result == UnrestrictedAgentAccess()
async def test_agent_access_default_deny_is_opt_in(self):
auth: Final = UserAPIKeyAuth(api_key="test-key", user_id="test-user", team_id="test-team")
key_lookup: Final = _lookup(UnrestrictedAgentAccess())
team_lookup: Final = _lookup(UnrestrictedAgentAccess())
assert (
await AgentRequestHandler.resolve_agent_access(
auth,
default_deny=lambda: False,
lookup_key_access=key_lookup,
lookup_team_access=team_lookup,
)
== UnrestrictedAgentAccess()
)
assert await AgentRequestHandler.resolve_agent_access(
auth,
default_deny=lambda: True,
lookup_key_access=key_lookup,
lookup_team_access=team_lookup,
) == RestrictedAgentAccess(frozenset())
assert await AgentRequestHandler.resolve_agent_access(
None,
default_deny=lambda: True,
lookup_key_access=key_lookup,
lookup_team_access=team_lookup,
) == RestrictedAgentAccess(frozenset())
async def test_explicit_key_and_team_grants_work_with_key_agent_access_required(self):
auth: Final = UserAPIKeyAuth(api_key="test-key", user_id="test-user", team_id="test-team")
unrestricted_lookup: Final = _lookup(UnrestrictedAgentAccess())
assert await AgentRequestHandler.resolve_agent_access(
auth,
default_deny=lambda: True,
lookup_key_access=unrestricted_lookup,
lookup_team_access=_lookup(RestrictedAgentAccess(frozenset({"team-agent"}))),
) == RestrictedAgentAccess(frozenset({"team-agent"}))
assert await AgentRequestHandler.resolve_agent_access(
auth,
default_deny=lambda: True,
lookup_key_access=_lookup(RestrictedAgentAccess(frozenset({"key-agent"}))),
lookup_team_access=unrestricted_lookup,
) == RestrictedAgentAccess(frozenset({"key-agent"}))
async def test_proxy_admin_is_exempt_from_key_agent_access_requirement(self):
admin: Final = UserAPIKeyAuth(
api_key="admin-key",
user_id="admin",
user_role=LitellmUserRoles.PROXY_ADMIN,
)
string_admin: Final = admin.model_copy(update={"user_role": LitellmUserRoles.PROXY_ADMIN.value})
key_lookup: Final = _lookup(UnrestrictedAgentAccess())
team_lookup: Final = _lookup(UnrestrictedAgentAccess())
assert (
await AgentRequestHandler.resolve_agent_access(
admin,
default_deny=lambda: True,
lookup_key_access=key_lookup,
lookup_team_access=team_lookup,
)
== UnrestrictedAgentAccess()
)
assert (
await AgentRequestHandler.resolve_agent_access(
string_admin,
default_deny=lambda: True,
lookup_key_access=key_lookup,
lookup_team_access=team_lookup,
)
== UnrestrictedAgentAccess()
)
async def test_key_agent_access_requirement_precedes_agent_access_group_ceiling(self):
agent_key: Final = UserAPIKeyAuth(
api_key="test-key",
user_id="test-user",
agent_id="caller-agent",
)
resolve, _ = self._ceiling_resolver(frozenset({"agent-beta"}))
unrestricted_lookup: Final = _lookup(UnrestrictedAgentAccess())
assert await AgentRequestHandler.resolve_agent_access(
agent_key,
resolve,
default_deny=lambda: True,
lookup_key_access=unrestricted_lookup,
lookup_team_access=unrestricted_lookup,
) == RestrictedAgentAccess(frozenset())
async def test_key_agent_access_requirement_fails_closed_on_lookup_errors(self):
auth: Final = UserAPIKeyAuth(api_key="test-key", user_id="test-user", team_id="test-team")
unrestricted_lookup: Final = _lookup(UnrestrictedAgentAccess())
key_error_lookup: Final = _lookup(Exception("DB Error"))
assert await AgentRequestHandler.resolve_agent_access(
auth,
default_deny=lambda: True,
lookup_key_access=key_error_lookup,
lookup_team_access=unrestricted_lookup,
) == RestrictedAgentAccess(frozenset())
assert (
await AgentRequestHandler.resolve_agent_access(
auth,
default_deny=lambda: False,
lookup_key_access=key_error_lookup,
lookup_team_access=unrestricted_lookup,
)
== UnrestrictedAgentAccess()
)
team_error_lookup: Final = _lookup(Exception("DB Error"))
key_grant_lookup: Final = _lookup(RestrictedAgentAccess(frozenset({"key-agent"})))
assert await AgentRequestHandler.resolve_agent_access(
auth,
default_deny=lambda: True,
lookup_key_access=key_grant_lookup,
lookup_team_access=team_error_lookup,
) == RestrictedAgentAccess(frozenset())
assert (
await AgentRequestHandler.resolve_agent_access(
auth,
default_deny=lambda: False,
lookup_key_access=key_grant_lookup,
lookup_team_access=team_error_lookup,
)
== UnrestrictedAgentAccess()
)
async def test_agent_access_default_deny_reads_general_settings(self, monkeypatch: pytest.MonkeyPatch):
from litellm.proxy import proxy_server
auth: Final = UserAPIKeyAuth(api_key="test-key", user_id="test-user")
monkeypatch.setattr(proxy_server, "prisma_client", None)
with patch.object(proxy_server, "general_settings", {"agent_access_default_deny": True}):
assert await AgentRequestHandler.is_agent_allowed("agent-alpha", auth) is False
with patch.object(proxy_server, "general_settings", {}):
assert await AgentRequestHandler.is_agent_allowed("agent-alpha", auth) is True
@pytest.mark.parametrize(
("settings", "expected_allowed"),
[
pytest.param({"agent_access_default_deny": True}, False, id="true-bool"),
pytest.param({"agent_access_default_deny": "true"}, False, id="true-lower-string"),
pytest.param({"agent_access_default_deny": "True"}, False, id="true-capitalized-string"),
pytest.param({"agent_access_default_deny": 1}, False, id="one"),
pytest.param({"agent_access_default_deny": False}, True, id="false-bool"),
pytest.param({"agent_access_default_deny": "false"}, True, id="false-string"),
pytest.param({"agent_access_default_deny": 0}, True, id="zero"),
pytest.param({}, True, id="missing"),
pytest.param({"agent_access_default_deny": "not-a-bool"}, False, id="invalid-string"),
],
)
async def test_agent_access_default_deny_coerces_general_settings(
self,
monkeypatch: pytest.MonkeyPatch,
settings: dict[str, object],
expected_allowed: bool,
):
from litellm.proxy import proxy_server
auth: Final = UserAPIKeyAuth(api_key="test-key", user_id="test-user")
monkeypatch.setattr(proxy_server, "prisma_client", None)
with patch.object(proxy_server, "general_settings", settings):
assert await AgentRequestHandler.is_agent_allowed("agent-alpha", auth) is expected_allowed
async def test_disjoint_key_and_team_grants_deny_every_agent(self):
"""LIT-5143: a key restricted to one agent inside a team restricted to another
must reach nothing. The empty intersection used to read as "no restrictions",

6
ui/litellm-dashboard/src/lib/http/schema.d.ts generated vendored
View file

@ -29649,6 +29649,12 @@ export interface components {
* @default 1
*/
admission_queue_timeout_seconds: number;
/**
* Agent Access Default Deny
* @description When true, a key with no agent grants on itself or its team can reach no agents instead of all agents. Proxy admins are exempt.
* @default false
*/
agent_access_default_deny: boolean;
/**
* Alert To Webhook Url
* @description Mapping of alert type to webhook url. e.g. `alert_to_webhook_url: {'budget_alerts': 'https://nothooks.slack.com/services/T00000000/B00000000/XXXXXXXXXXXXXXXXXXXXXXXX'}`