scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-01 02:02:20 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 91

feat(scim): authenticate native agent users with directory policy

Browse source
This commit is contained in:
Joshua Valluru 2026-09-26 12:18:01 -07:00
parent 2a48418bb4
commit 419b4208b9
14 changed files with 629 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

60
litellm/proxy/_lazy_openapi_snapshot.json
View file

@ -2662,6 +2662,17 @@
"title": "Provider",
"type": "string"
},
"provisioning_source_id": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"title": "Provisioning Source Id"
},
"required_roles": {
"default": [],
"items": {
@ -3173,6 +3184,25 @@
],
"title": "Created By"
},
"directory_access_group_ids": {
"anyOf": [
{
"items": {
"type": "string"
},
"type": "array"
},
{
"type": "null"
}
],
"title": "Directory Access Group Ids"
},
"directory_active": {
"default": true,
"title": "Directory Active",
"type": "boolean"
},
"enabled": {
"default": true,
"title": "Enabled",
@ -3699,6 +3729,17 @@
"title": "Provider",
"type": "string"
},
"provisioning_source_id": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"title": "Provisioning Source Id"
},
"required_roles": {
"default": [],
"items": {
@ -3893,6 +3934,25 @@
},
"ManagedAgentIdentityStatus": {
"properties": {
"directory_access_group_ids": {
"anyOf": [
{
"items": {
"type": "string"
},
"type": "array"
},
{
"type": "null"
}
],
"title": "Directory Access Group Ids"
},
"directory_active": {
"default": true,
"title": "Directory Active",
"type": "boolean"
},
"enabled": {
"default": true,
"title": "Enabled",

14
litellm/proxy/agent_endpoints/auth/agent_access_groups.py
View file

@ -86,7 +86,19 @@ async def resolve_managed_agent_ceilings(agent: "AgentResponse") -> tuple[AgentA
async def manual_ids(_agent_id: str) -> AccessGroupIds:
return tuple(agent.access_group_ids or ())
async def directory_ids(_agent_id: str) -> AccessGroupIds:
return agent.directory_access_group_ids or ()
manual: Final = await resolve_agent_access_group_ceiling(
agent.agent_id, load_access_group_ids=manual_ids, load_access_group=authoritative_group
)
return (manual,) if manual is not None else ()
directory: Final = (
await resolve_agent_access_group_ceiling(
agent.agent_id, load_access_group_ids=directory_ids, load_access_group=authoritative_group
)
if agent.directory_access_group_ids
else AgentAccessGroupCeiling((), frozenset(), frozenset(), frozenset())
if agent.directory_access_group_ids is not None
else None
)
return tuple(ceiling for ceiling in (manual, directory) if ceiling is not None)

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

@ -170,6 +170,8 @@ class AgentRequestHandler:
if target is not None and target.identity_managed:
if (
not target.enabled
or not target.directory_active
or target.directory_access_group_ids == ()
or target.identity is None
or not target.identity.active
or user_api_key_auth is None

8
litellm/proxy/agent_endpoints/auth/managed_authorization.py
View file

@ -172,7 +172,13 @@ def actor_admission_failure(
agent: AgentResponse,
context: ManagedAgentContext | None,
) -> AgentIdentityFailure | None:
if not agent.enabled or agent.identity is None or not agent.identity.active:
if (
not agent.enabled
or not agent.directory_active
or agent.directory_access_group_ids == ()
or agent.identity is None
or not agent.identity.active
):
return AgentIdentityFailure(message="Agent execution is disabled")
if context is None:
return AgentIdentityFailure(message="This agent requires its bound identity provider token")

2
litellm/proxy/agent_endpoints/endpoints.py
View file

@ -505,6 +505,8 @@ async def get_agent_identity_status(
return ManagedAgentIdentityStatus(
identity=agent.identity,
identity_managed=agent.identity_managed,
directory_active=agent.directory_active,
directory_access_group_ids=agent.directory_access_group_ids,
enabled=agent.enabled,
execution_mode=agent.execution_mode,
last_authenticated_at=agent.identity.last_authenticated_at if agent.identity else None,

109
litellm/proxy/agent_endpoints/identity_store.py
View file

@ -1,5 +1,7 @@
from collections.abc import Mapping
from datetime import datetime, timezone
from itertools import chain
from types import MappingProxyType
from typing import TYPE_CHECKING, Final
from litellm.proxy.agent_endpoints.managed_identity import classify_agent_subject
@ -8,6 +10,8 @@ from litellm.repositories.table_repositories import (
AgentsRepository,
RetiredAgentIdentityRepository,
RetiredAgentRepository,
SCIMResourceRepository,
SCIMSourceRepository,
VerifiedSubjectRepository,
)
from litellm.types.agents import AgentResponse
@ -15,6 +19,7 @@ from litellm.types.proxy.agent_identity import (
AgentIdentityFailure,
ManagedAgentContext,
MicrosoftInteractiveSubject,
VerifiedAgentSubject,
VerifiedHumanSubject,
)
@ -40,6 +45,8 @@ class AgentIdentityStore:
AgentIdentityRepository(client, use_writer=True),
VerifiedSubjectRepository(client, use_writer=True),
RetiredAgentIdentityRepository(client, use_writer=True),
SCIMSourceRepository(client, use_writer=True),
SCIMResourceRepository(client, use_writer=True),
RetiredAgentRepository(client, use_writer=True),
)
@ -49,12 +56,16 @@ class AgentIdentityStore:
identities: AgentIdentityRepository,
humans: VerifiedSubjectRepository,
retired: RetiredAgentIdentityRepository | None = None,
sources: SCIMSourceRepository | None = None,
resources: SCIMResourceRepository | None = None,
retired_agents: RetiredAgentRepository | None = None,
) -> None:
self.agents = agents
self.identities = identities
self.humans = humans
self.retired = retired
self.sources = sources
self.resources = resources
self.retired_agents = retired_agents
async def agent(self, agent_id: str) -> AgentResponse | AgentIdentityFailure | None:
@ -68,7 +79,10 @@ class AgentIdentityStore:
row: Final = await self.agents.table.find_unique(where=where, include=include)
if row is None:
return None
return AgentResponse.model_validate(row.model_dump())
agent: Final = AgentResponse.model_validate(row.model_dump())
if agent.identity is None or agent.identity.provisioning_source_id is None:
return agent
return await self.directory_policy(agent)
except Exception:
return AgentIdentityFailure(code="policy_unavailable", message="Agent policy could not be loaded")
@ -99,6 +113,19 @@ class AgentIdentityStore:
row: Final = await self.identities.table.find_unique(where=where)
except Exception:
return AgentIdentityFailure(code="policy_unavailable", message="Agent identity could not be loaded")
proven: Final = (
await self.subject(issuer, tenant, claims.get("oid"))
if row is None or row.provisioning_source_id is not None or isinstance(claims.get("scp"), str)
else None
)
if isinstance(proven, AgentIdentityFailure):
return proven
if (
proven is not None
and proven.kind == "agent_user"
and (row is None or proven.agent_id != row.agent_id or proven.parent_client_id != client)
):
return AgentIdentityFailure(message="Provisioned subject does not match an active agent binding")
if row is None:
return await self.unbound_client(where)
agent: Final = await self.agent(row.agent_id)
@ -112,19 +139,27 @@ class AgentIdentityStore:
or not agent.identity.active
):
return AgentIdentityFailure(message="Agent is disabled or no longer bound to an identity")
subject: Final = classify_agent_subject(agent.identity, claims, agent.execution_mode)
native: Final = (
VerifiedAgentSubject.model_validate(proven.model_dump())
if proven is not None and proven.kind == "agent_user" and proven.agent_id is not None
else None
)
if native is not None and (
agent.identity.provisioning_source_id is None
or not agent.directory_active
or agent.directory_access_group_ids == ()
):
return AgentIdentityFailure(message="Provisioned agent is inactive or has no mapped directory entitlement")
subject: Final = classify_agent_subject(agent.identity, claims, agent.execution_mode, native_subject=native)
if isinstance(subject, AgentIdentityFailure):
return subject
if subject.kind == "application":
if subject.kind in ("application", "agent_user"):
return ManagedAgentContext(
agent_id=agent.agent_id,
binding_revision=agent.identity.revision,
mode=subject.mode,
subject_oid=subject.oid,
)
proven: Final = await self.subject(issuer, tenant, claims.get("oid"))
if isinstance(proven, AgentIdentityFailure):
return proven
human: Final = (
VerifiedHumanSubject.model_validate(proven.model_dump())
if proven is not None
@ -156,6 +191,68 @@ class AgentIdentityStore:
except Exception:
return AgentIdentityFailure(code="policy_unavailable", message="Subject classification is unavailable")
async def directory_policy(self, agent: AgentResponse) -> AgentResponse | AgentIdentityFailure:
from pydantic import TypeAdapter
from litellm.types.proxy.management_endpoints.scim_agent_provisioning import (
SCIMGroupMapping,
canonical_directory_id,
)
if self.sources is None or self.resources is None or agent.identity is None:
return AgentIdentityFailure(code="policy_unavailable", message="Provisioning state is unavailable")
from prisma.types import LiteLLM_SCIMResourceWhereInput, LiteLLM_SCIMSourceWhereUniqueInput
source_where: Final[LiteLLM_SCIMSourceWhereUniqueInput] = {"source_id": agent.identity.provisioning_source_id}
resource_where: Final[LiteLLM_SCIMResourceWhereInput] = {
"source_id": agent.identity.provisioning_source_id,
"kind": "Users",
"local_id": agent.agent_id,
}
source: Final = await self.sources.table.find_unique(where=source_where)
subjects: Final = await self.resources.table.find_many(where=resource_where)
live: Final = tuple(subject for subject in subjects if subject.active and not subject.deleted)
if source is None or not source.enabled or len(live) != 1:
return agent.model_copy(
update=MappingProxyType({"directory_active": False, "directory_access_group_ids": ()})
)
from litellm.proxy.management_endpoints.scim.agent_provisioning import user_document
directory_user: Final = user_document(live[0])
if (
source.tenant_id != agent.identity.tenant_id
or directory_user.agent_user is None
or str(directory_user.agent_user.identityParentId) != agent.identity.client_id
):
return AgentIdentityFailure(message="Directory identity does not match the registered binding")
proven: Final = await self.subject(agent.identity.issuer, source.tenant_id, live[0].external_id)
if isinstance(proven, AgentIdentityFailure):
return proven
if (
proven is None
or proven.kind != "agent_user"
or proven.verified_via != "scim"
or proven.agent_id != agent.agent_id
or proven.parent_client_id != agent.identity.client_id
or proven.scim_resource_id != live[0].id
):
return AgentIdentityFailure(message="Directory subject does not match the provisioned resource")
groups_where: Final[LiteLLM_SCIMResourceWhereInput] = {
"source_id": source.source_id,
"kind": "Groups",
"deleted": False,
"active": True,
"member_ids": {"has": live[0].id},
}
groups: Final = await self.resources.table.find_many(where=groups_where)
mappings: Final = TypeAdapter(tuple[SCIMGroupMapping, ...]).validate_python(source.group_mappings)
external_ids: Final = frozenset(canonical_directory_id(group.external_id) for group in groups)
mapped: Final = tuple(
mapping.access_group_ids for mapping in mappings if str(mapping.external_group_id) in external_ids
)
ids: Final = tuple(sorted(frozenset(chain.from_iterable(mapped))))
return agent.model_copy(update=MappingProxyType({"directory_active": True, "directory_access_group_ids": ids}))
async def retired_agent(self, agent_id: str) -> bool | AgentIdentityFailure:
if self.retired_agents is None:
return AgentIdentityFailure(code="policy_unavailable", message="Agent history is unavailable")

51
litellm/proxy/agent_endpoints/managed_identity.py
View file

@ -16,6 +16,7 @@ from litellm.types.proxy.agent_identity import (
AgentIdentityFailure,
AgentSubject,
EntraIdentityConfig,
VerifiedAgentSubject,
)
_MODE: Final = TypeAdapter(AgentExecutionMode)
@ -27,6 +28,7 @@ class IdentityFields(TypedDict, total=False):
client_id: ReadOnly[str]
issuer: ReadOnly[str]
service_principal_id: ReadOnly[str | None]
provisioning_source_id: ReadOnly[str | None]
required_roles: ReadOnly[tuple[str, ...]]
required_scopes: ReadOnly[tuple[str, ...]]
active: ReadOnly[bool]
@ -102,7 +104,15 @@ def _configuration_failure(
mode: AgentExecutionMode,
enabling_without_binding: bool,
) -> AgentIdentityFailure | None:
if identity is not None and mode != "delegated" and not identity.service_principal_id:
if identity is not None and identity.provisioning_source_id is not None:
if mode != "autonomous" or not identity.required_scopes:
return AgentIdentityFailure(message="Provisioned native agent-users require autonomous mode and scopes")
if (
identity is not None
and mode != "delegated"
and not identity.service_principal_id
and not identity.provisioning_source_id
):
return AgentIdentityFailure(
message="Autonomous mode requires the Enterprise application service-principal object ID"
)
@ -137,6 +147,8 @@ def managed_write_fields(
return failure
empty: Final[ManagedWriteFields] = {}
identity_fields: Final = _identity_write(identity, existing) if "identity" in incoming else empty
if isinstance(identity_fields, AgentIdentityFailure):
return identity_fields
budget_fields: Final = (
_budget_write(incoming["budget"], existing, updated_by) if "budget" in incoming else empty
)
@ -151,7 +163,19 @@ def managed_write_fields(
return AgentIdentityFailure(message=f"Invalid agent identity or budget configuration: {exc}")
def _identity_write(identity: EntraIdentityConfig | None, existing: AgentResponse | None) -> ManagedWriteFields:
def _identity_write(
identity: EntraIdentityConfig | None, existing: AgentResponse | None
) -> ManagedWriteFields | AgentIdentityFailure:
source_id: Final = existing.identity.provisioning_source_id if existing and existing.identity else None
if existing is not None and existing.identity is not None and source_id is not None:
if identity is None or (identity.tenant_id, identity.client_id, identity.provisioning_source_id) != (
existing.identity.tenant_id,
existing.identity.client_id,
source_id,
):
return AgentIdentityFailure(message="A directory-owned identity cannot be rebound through agent settings")
elif identity is not None and identity.provisioning_source_id is not None:
return AgentIdentityFailure(message="Only SCIM provisioning can create a directory-owned identity")
if identity is None:
unbind: Final[ManagedWriteFields] = {
**(
@ -175,6 +199,7 @@ def _identity_write(identity: EntraIdentityConfig | None, existing: AgentRespons
"tenant_id": identity.tenant_id,
"client_id": identity.client_id,
"service_principal_id": identity.service_principal_id,
"provisioning_source_id": identity.provisioning_source_id,
"required_roles": identity.required_roles,
"required_scopes": identity.required_scopes,
"issuer": identity.issuer,
@ -257,6 +282,8 @@ def classify_agent_subject(
binding: AgentIdentityBinding,
claims: Mapping[str, object],
allowed_mode: AgentExecutionMode,
*,
native_subject: VerifiedAgentSubject | None = None,
) -> AgentSubject | AgentIdentityFailure:
if (claims.get("iss"), claims.get("tid"), claims.get("azp")) != (
binding.issuer,
@ -268,6 +295,26 @@ def classify_agent_subject(
if not isinstance(oid, str) or not oid:
return AgentIdentityFailure(message="Entra token must identify its object subject")
scope: Final = claims.get("scp")
if native_subject is not None:
if (
(
native_subject.issuer,
native_subject.tenant_id,
native_subject.parent_client_id,
native_subject.agent_id,
native_subject.oid,
)
!= (binding.issuer, binding.tenant_id, binding.client_id, binding.agent_id, oid)
or allowed_mode == "delegated"
or claims.get("idtyp") == "app"
or not isinstance(scope, str)
or not binding.required_scopes
or not frozenset(binding.required_scopes).issubset(scope.split())
):
return AgentIdentityFailure(message="Token does not match the provisioned native agent-user")
return AgentSubject(kind="agent_user", oid=oid, mode="autonomous")
if binding.provisioning_source_id is not None:
return AgentIdentityFailure(message="A verified provisioned agent-user token is required")
facets: Final = claims.get("xms_sub_fct")
if facets is not None and (not isinstance(facets, str) or "13" in facets.split()):
return AgentIdentityFailure(message="Native agent-user authentication is not supported by this binding")

4
litellm/types/agents.py
View file

@ -340,6 +340,10 @@ def agent_spend_filter(counter_key: str) -> "LiteLLM_AgentsTableWhereInput":
class AgentResponse(BaseModel):
budget_id: str | None = None
litellm_budget_table: AgentBudgetState | None = None
directory_active: bool = True
directory_access_group_ids: tuple[str, ...] | None = None
identity: AgentIdentityBinding | None = None
identity_managed: bool = False
enabled: bool = True

17
litellm/types/proxy/agent_identity.py
View file

@ -13,6 +13,7 @@ class EntraIdentityConfig(BaseModel):
provider: Literal["microsoft_entra"]
tenant_id: str
client_id: str
provisioning_source_id: str | None = None
service_principal_id: str | None = None
required_roles: tuple[str, ...] = ()
required_scopes: tuple[str, ...] = Field(
@ -38,6 +39,7 @@ class AgentIdentityBinding(BaseModel):
provider: Literal["microsoft_entra"]
tenant_id: str
client_id: str
provisioning_source_id: str | None = None
service_principal_id: str | None = None
issuer: str
required_roles: tuple[str, ...] = ()
@ -65,7 +67,7 @@ class AgentBudgetState(BaseModel):
class AgentSubject(BaseModel):
model_config = ConfigDict(frozen=True)
kind: Literal["application", "delegated_subject"]
kind: Literal["application", "delegated_subject", "agent_user"]
oid: str
mode: Literal["autonomous", "delegated"]
@ -105,8 +107,21 @@ class MicrosoftInteractiveSubject(BaseModel):
class ManagedAgentIdentityStatus(BaseModel):
directory_active: bool = True
directory_access_group_ids: tuple[str, ...] | None = None
identity: AgentIdentityBinding | None = None
identity_managed: bool = False
enabled: bool = True
execution_mode: AgentExecutionMode = "autonomous"
last_authenticated_at: datetime | None = None
class VerifiedAgentSubject(BaseModel):
model_config = ConfigDict(frozen=True)
issuer: str
tenant_id: str
oid: str
agent_id: str
parent_client_id: str
scim_resource_id: str

2
tests/test_litellm/proxy/agent_endpoints/auth/test_agent_permission_handler.py
View file

@ -639,6 +639,8 @@ class TestAgentRequestHandler:
[
({}, True),
({"enabled": False}, False),
({"directory_active": False}, False),
({"directory_access_group_ids": ()}, False),
],
)
async def test_managed_invocation_requires_local_and_directory_admission(

10
tests/test_litellm/proxy/agent_endpoints/auth/test_managed_authorization.py
View file

@ -320,6 +320,16 @@ async def test_invocation_cannot_bypass_missing_policy_permission_or_invalid_pri
assert auth.agent_invocation_cost is None
def test_native_directory_agent_cannot_authenticate_with_a_virtual_key() -> None:
policy: Final = agent(
identity=BINDING.model_copy(update={"provisioning_source_id": "source"}), execution_mode="autonomous"
)
failure: Final = actor_admission_failure(policy, None)
assert isinstance(failure, AgentIdentityFailure)
assert failure.code == "identity_denied"
assert "bound identity provider token" in failure.message
@pytest.mark.asyncio
async def test_legacy_jwt_cannot_adopt_an_agent_bound_on_another_worker() -> None:
database: Final = MagicMock()

235
tests/test_litellm/proxy/agent_endpoints/test_identity_store.py
View file

@ -1,3 +1,4 @@
import json
from datetime import datetime, timezone
from types import SimpleNamespace
from typing import Final
@ -233,6 +234,209 @@ async def test_retired_client_cannot_fall_back_to_ordinary_user_authentication()
assert "retired" in result.message
def native_store():
from prisma.models import LiteLLM_SCIMResource, LiteLLM_SCIMSource
from litellm.repositories.table_repositories import SCIMResourceRepository, SCIMSourceRepository
from litellm.types.proxy.management_endpoints.scim_agent_provisioning import SCIM_AGENT_USER_SCHEMA
now: Final = datetime.now(timezone.utc)
native: Final = LiteLLM_VerifiedSubject(
subject_id="native-subject",
kind="agent_user",
issuer=ISSUER,
tenant_id=TENANT,
oid=HUMAN,
agent_id="agent-one",
parent_client_id=CLIENT,
scim_resource_id="directory-user",
verified_via="scim",
verified_at=now,
)
store, agents, identities, humans = setup_store(human=native)
binding: Final = BINDING.model_copy(update={"provisioning_source_id": "source", "service_principal_id": None})
agents.find_unique.return_value = stored_agent(identity=binding, execution_mode="autonomous")
identities.find_unique.return_value = binding
source: Final = LiteLLM_SCIMSource(
source_id="source",
display_name="Directory",
tenant_id=TENANT,
key_hash="hash",
enabled=True,
group_mappings=json.dumps([{"external_group_id": PRINCIPAL, "access_group_ids": ["read"]}]),
created_at=now,
updated_at=now,
)
resource: Final = LiteLLM_SCIMResource(
id="directory-user",
source_id="source",
kind="Users",
external_id=HUMAN,
local_id="agent-one",
display_name="Native",
member_ids=[],
document=json.dumps(
{
"schemas": [],
"userName": "native@example.com",
"externalId": HUMAN,
SCIM_AGENT_USER_SCHEMA: {"identityParentId": CLIENT},
}
),
active=True,
deleted=False,
created_at=now,
updated_at=now,
)
group: Final = resource.model_copy(update={"id": "directory-group", "kind": "Groups", "external_id": PRINCIPAL})
sources: Final = AsyncMock()
resources: Final = AsyncMock()
sources.find_unique.return_value = source
resources.find_many.side_effect = lambda **query: [resource] if query["where"]["kind"] == "Users" else [group]
client: Final = SimpleNamespace(db=SimpleNamespace(litellm_scimsource=sources, litellm_scimresource=resources))
return (
AgentIdentityStore(
store.agents,
store.identities,
store.humans,
sources=SCIMSourceRepository(client),
resources=SCIMResourceRepository(client),
),
sources,
resources,
humans,
native,
resource,
)
@pytest.mark.asyncio
async def test_native_subject_is_autonomous_and_directory_grants_are_separate() -> None:
store, _, _, _, _, _ = native_store()
context: Final = await store.resolve_verified_claims({**CLAIMS, "oid": HUMAN, "scp": "user_impersonation"})
assert isinstance(context, ManagedAgentContext)
assert context.mode == "autonomous"
assert context.user_id is None
assert context.subject_oid == HUMAN
agent: Final = await store.agent("agent-one")
assert isinstance(agent, AgentResponse)
assert agent.directory_active is True
assert agent.directory_access_group_ids == ("read",)
assert agent.access_group_ids is None
@pytest.mark.asyncio
@pytest.mark.parametrize(
"field,value",
[
("kind", "human"),
("verified_via", "sso_interactive"),
("agent_id", "foreign-agent"),
("parent_client_id", PRINCIPAL),
("scim_resource_id", "foreign-resource"),
],
)
async def test_directory_policy_rejects_mismatched_subject_ownership(field: str, value: str) -> None:
store, _, _, humans, native, _ = native_store()
humans.find_unique.return_value = native.model_copy(update={field: value})
result: Final = await store.agent("agent-one")
assert isinstance(result, AgentIdentityFailure)
assert "subject" in result.message
@pytest.mark.asyncio
@pytest.mark.parametrize("state", ["source-disabled", "resource-disabled", "resource-deleted", "no-groups"])
async def test_native_revocation_cannot_fall_back_to_human_or_unrestricted_agent(state: str) -> None:
store, sources, resources, _, _, resource = native_store()
if state == "source-disabled":
sources.find_unique.return_value = sources.find_unique.return_value.model_copy(update={"enabled": False})
elif state == "no-groups":
resources.find_many.side_effect = lambda **query: [resource] if query["where"]["kind"] == "Users" else []
else:
resources.find_many.side_effect = None
resources.find_many.return_value = [
resource.model_copy(
update={
"active": state != "resource-disabled",
"deleted": state == "resource-deleted",
}
)
]
result: Final = await store.resolve_verified_claims({**CLAIMS, "oid": HUMAN, "scp": "user_impersonation"})
assert isinstance(result, AgentIdentityFailure)
assert result.code == "identity_denied"
@pytest.mark.asyncio
@pytest.mark.parametrize("groups", ["both", "second-only"])
async def test_multiple_directory_groups_union_and_removal_preserves_remaining_grants(groups: str) -> None:
store, sources, resources, _, _, resource = native_store()
sources.find_unique.return_value = sources.find_unique.return_value.model_copy(
update={
"group_mappings": [
{"external_group_id": PRINCIPAL, "access_group_ids": ["read"]},
{"external_group_id": TENANT, "access_group_ids": ["write"]},
]
}
)
first: Final = resource.model_copy(update={"id": "group-one", "kind": "Groups", "external_id": PRINCIPAL})
second: Final = resource.model_copy(update={"id": "group-two", "kind": "Groups", "external_id": TENANT})
rows: Final = [first, second] if groups == "both" else [second]
resources.find_many.side_effect = lambda **query: [resource] if query["where"]["kind"] == "Users" else rows
result: Final = await store.agent("agent-one")
assert isinstance(result, AgentResponse)
assert result.directory_access_group_ids == (("read", "write") if groups == "both" else ("write",))
assert result.access_group_ids is None
@pytest.mark.asyncio
@pytest.mark.parametrize("failure", ["source-unavailable", "wrong-tenant", "missing-subject", "subject-unavailable"])
async def test_native_directory_policy_fails_closed_when_correspondence_cannot_be_proven(failure: str) -> None:
store, sources, _, humans, _, _ = native_store()
if failure == "source-unavailable":
sources.find_unique.side_effect = RuntimeError("writer unavailable")
elif failure == "wrong-tenant":
sources.find_unique.return_value = sources.find_unique.return_value.model_copy(update={"tenant_id": HUMAN})
elif failure == "missing-subject":
humans.find_unique.return_value = None
else:
humans.find_unique.side_effect = RuntimeError("writer unavailable")
result: Final = await store.agent("agent-one")
assert isinstance(result, AgentIdentityFailure)
@pytest.mark.asyncio
async def test_verified_subject_cannot_switch_to_an_unrelated_registered_parent() -> None:
store, _, _, humans, native, _ = native_store()
humans.find_unique.return_value = native.model_copy(update={"parent_client_id": PRINCIPAL})
result: Final = await store.resolve_verified_claims({**CLAIMS, "oid": HUMAN, "scp": "user_impersonation"})
assert isinstance(result, AgentIdentityFailure)
assert "does not match" in result.message
@pytest.mark.asyncio
@pytest.mark.parametrize("kind", ["human", "agent_user", "untrusted", "missing"])
async def test_human_lookup_requires_trusted_interactive_enrollment(kind: str) -> None:
store, _, _, humans = setup_store()
native: Final = native_store()[4]
humans.find_unique.return_value = (
None
if kind == "missing"
else native.model_copy(update={"kind": "human", "verified_via": "sso_interactive", "user_id": "local-human"})
if kind == "human"
else native.model_copy(update={"kind": "human", "verified_via": "untrusted"})
if kind == "untrusted"
else native
)
result: Final = await store.resolve_verified_claims({**CLAIMS, "oid": HUMAN, "scp": "user_impersonation"})
if kind == "human":
assert isinstance(result, ManagedAgentContext)
assert result.mode == "delegated"
assert result.user_id == "local-human"
else:
assert isinstance(result, AgentIdentityFailure)
@pytest.mark.asyncio
async def test_missing_revision_cannot_create_entra_authentication_evidence() -> None:
store, _, identities, _ = setup_store()
@ -271,6 +475,15 @@ async def test_retired_binding_denies_and_history_outage_cannot_become_legacy_fa
)
@pytest.mark.asyncio
async def test_missing_directory_repositories_cannot_admit_a_provisioned_agent() -> None:
store, _, _, _ = setup_store()
native: Final = stored_agent(identity=BINDING.model_copy(update={"provisioning_source_id": "source"}))
result: Final = await store.directory_policy(native)
assert isinstance(result, AgentIdentityFailure)
assert result.code == "policy_unavailable"
@pytest.mark.asyncio
async def test_new_binding_is_enforced_after_another_worker_commits_it() -> None:
_, agents, identities, humans = setup_store()
@ -388,7 +601,7 @@ async def test_resolver_preserves_unconfigured_and_unrelated_authentication() ->
@pytest.mark.asyncio
@pytest.mark.parametrize("registered", [True, False])
async def test_application_and_unregistered_clients_do_not_depend_on_human_subject_storage(registered: bool) -> None:
async def test_subject_outage_only_blocks_clients_that_need_directory_classification(registered: bool) -> None:
store, _, identities, humans = setup_store()
identities.find_unique.return_value = BINDING if registered else None
humans.find_unique.side_effect = RuntimeError("subject database unavailable")
@ -397,6 +610,22 @@ async def test_application_and_unregistered_clients_do_not_depend_on_human_subje
assert isinstance(result, ManagedAgentContext)
assert result.mode == "autonomous"
assert result.user_id is None
humans.find_unique.assert_not_awaited()
else:
assert result is None
humans.find_unique.assert_not_awaited()
assert isinstance(result, AgentIdentityFailure)
assert result.code == "policy_unavailable"
humans.find_unique.assert_awaited_once()
@pytest.mark.asyncio
async def test_directory_group_guid_case_does_not_remove_agent_grants() -> None:
store, sources, resources, _, _, resource = native_store()
guid: Final = "abcdefab-abcd-4abc-8abc-abcdefabcdef"
sources.find_unique.return_value = sources.find_unique.return_value.model_copy(
update={"group_mappings": [{"external_group_id": guid, "access_group_ids": ["read"]}]}
)
group: Final = resource.model_copy(update={"id": "group", "kind": "Groups", "external_id": guid.upper()})
resources.find_many.side_effect = lambda **query: [resource] if query["where"]["kind"] == "Users" else [group]
result: Final = await store.agent("agent-one")
assert isinstance(result, AgentResponse)
assert result.directory_access_group_ids == ("read",)

111
tests/test_litellm/proxy/agent_endpoints/test_managed_identity.py
View file

@ -170,6 +170,101 @@ def test_each_application_binding_records_its_history_atomically() -> None:
assert replacement["retired_identities"]["connectOrCreate"]["create"]["client_id"] == HUMAN
def test_native_agent_user_uses_proven_subject_without_optional_facets() -> None:
from litellm.types.proxy.agent_identity import VerifiedAgentSubject
subject: Final = VerifiedAgentSubject(
issuer=ISSUER,
tenant_id=TENANT,
oid=HUMAN,
agent_id=BINDING.agent_id,
parent_client_id=CLIENT,
scim_resource_id="scim-subject",
)
result: Final = classify_agent_subject(
BINDING,
claims(oid=HUMAN, scp="user_impersonation"),
"autonomous",
native_subject=subject,
)
assert result == AgentSubject(kind="agent_user", oid=HUMAN, mode="autonomous")
@pytest.mark.parametrize(
"overrides",
[
{"oid": PRINCIPAL},
{"azp": HUMAN},
{"tid": HUMAN},
{"scp": "unrelated"},
{"scp": None},
{"idtyp": "app"},
],
)
def test_native_subject_binding_rejects_other_subjects_parents_and_scopes(overrides: dict[str, object]) -> None:
from litellm.types.proxy.agent_identity import VerifiedAgentSubject
subject: Final = VerifiedAgentSubject(
issuer=ISSUER,
tenant_id=TENANT,
oid=HUMAN,
agent_id=BINDING.agent_id,
parent_client_id=CLIENT,
scim_resource_id="scim-subject",
)
result: Final = classify_agent_subject(
BINDING,
claims(**{"oid": HUMAN, "scp": "user_impersonation", **overrides}),
"both",
native_subject=subject,
)
assert isinstance(result, AgentIdentityFailure)
@pytest.mark.parametrize("change", [None, {"client_id": HUMAN}, {"provisioning_source_id": "another-source"}])
def test_directory_owned_identity_cannot_be_unbound_or_reassigned(change: dict[str, str] | None) -> None:
native_binding: Final = BINDING.model_copy(update={"provisioning_source_id": "source"})
existing: Final = AgentResponse(
agent_id="agent-one",
agent_name="Native",
agent_card_params={},
identity=native_binding,
identity_managed=True,
execution_mode="autonomous",
)
identity: Final = (
None
if change is None
else {
"provider": "microsoft_entra",
"tenant_id": TENANT,
"client_id": CLIENT,
"provisioning_source_id": "source",
**change,
}
)
result: Final = managed_write_fields({"identity": identity}, existing, "admin")
assert isinstance(result, AgentIdentityFailure)
assert "directory-owned" in result.message
def test_manual_registration_cannot_claim_directory_ownership() -> None:
result: Final = managed_write_fields(
{
"identity": {
"provider": "microsoft_entra",
"tenant_id": TENANT,
"client_id": CLIENT,
"provisioning_source_id": "source",
}
},
None,
"admin",
)
assert isinstance(result, AgentIdentityFailure)
assert "Only SCIM" in result.message
def test_unchanged_binding_preserves_revision_and_authentication_evidence() -> None:
configuration: Final = BINDING.model_dump(
exclude={"agent_id", "issuer", "revision", "last_authenticated_at", "active"}
@ -206,6 +301,15 @@ def test_explicit_empty_scope_requirements_can_be_registered_and_preserved(mode:
assert not isinstance(updated, AgentIdentityFailure)
assert updated["execution_mode"] == mode
@pytest.mark.parametrize("mode", ["delegated", "both"])
def test_native_directory_identity_cannot_switch_to_delegated_execution(mode: str) -> None:
agent: Final = managed_agent().model_copy(
update={"identity": BINDING.model_copy(update={"provisioning_source_id": "source"})}
)
result: Final = managed_write_fields({"execution_mode": mode}, agent, "admin")
assert isinstance(result, AgentIdentityFailure)
assert "autonomous mode" in result.message
@pytest.mark.parametrize(
"incoming",
@ -311,3 +415,10 @@ def test_budget_write_stamps_the_same_window_on_the_agent_row() -> None:
assert not isinstance(result, AgentIdentityFailure)
assert result["spend_window"] == result["litellm_budget_table"]["create"]["budget_reset_at"]
assert result["spend"] == 0.0
def test_directory_binding_cannot_fall_back_to_an_application_token() -> None:
binding: Final = BINDING.model_copy(update={"provisioning_source_id": "source"})
result: Final = classify_agent_subject(binding, claims(), "autonomous")
assert isinstance(result, AgentIdentityFailure)
assert "verified provisioned agent-user" in result.message

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

@ -24378,6 +24378,8 @@ export interface components {
* @constant
*/
provider: "microsoft_entra";
/** Provisioning Source Id */
provisioning_source_id?: string | null;
/**
* Required Roles
* @default []
@ -24554,6 +24556,13 @@ export interface components {
created_at?: string | null;
/** Created By */
created_by?: string | null;
/** Directory Access Group Ids */
directory_access_group_ids?: string[] | null;
/**
* Directory Active
* @default true
*/
directory_active: boolean;
/**
* Enabled
* @default true
@ -30266,6 +30275,8 @@ export interface components {
* @constant
*/
provider: "microsoft_entra";
/** Provisioning Source Id */
provisioning_source_id?: string | null;
/**
* Required Roles
* @default []
@ -35693,6 +35704,13 @@ export interface components {
};
/** ManagedAgentIdentityStatus */
ManagedAgentIdentityStatus: {
/** Directory Access Group Ids */
directory_access_group_ids?: string[] | null;
/**
* Directory Active
* @default true
*/
directory_active: boolean;
/**
* Enabled
* @default true