feat(proxy): synchronize LDAP groups to teams

This commit is contained in:
Mrinal Chanshetty 2026-11-04 15:30:52 -08:00
parent 85fdea0fd1
commit d158eb99d6
12 changed files with 870 additions and 23 deletions

57
docs/ldap_sync.md Normal file
View file

@ -0,0 +1,57 @@
# LDAP team synchronization
LDAP sync periodically reads a selected set of directory groups, creates missing LiteLLM teams, and reconciles their memberships using accounts that already exist in LiteLLM. Login continues to use your existing authentication setup.
Install the proxy dependencies with `pip install 'litellm[proxy]'`. A database and shared Redis are required, including when running multiple workers in one proxy. Each worker uses the existing job lock so only one worker synchronizes at a time.
```yaml
general_settings:
coordination_redis:
host: os.environ/REDIS_HOST
port: os.environ/REDIS_PORT
ldap_sync:
enabled: true
server_url: ldaps://ldap.example.org:636
group_search_base: ou=groups,dc=example,dc=org
group_filter: "(cn=project-*)"
user_search_base: ou=people,dc=example,dc=org
group_id_attribute: cn
group_name_attribute: cn
member_attribute: member
user_match_attribute: mail
litellm_match_field: user_email
interval_minutes: 30
dry_run: true
litellm_settings:
default_team_params:
models: [your-approved-model]
max_budget: 100
budget_duration: 30d
```
The example uses anonymous LDAP reads over TLS. For a service account, add `bind_dn: os.environ/LDAP_BIND_DN` and `bind_password: os.environ/LDAP_BIND_PASSWORD`. Credentials cannot be sent through an unencrypted LDAP connection. With `ldap://`, set `start_tls: true` for a service-account bind. LDAPS and StartTLS verify the server certificate, using the system trust store or the optional `ca_cert_path` CA bundle. Anonymous `ldap://` connections without StartTLS are also supported.
Start with `dry_run: true`, restart the proxy, and review the planned team creation and member additions/removals in its normal logs. Set `dry_run: false` and restart to apply changes. The first run starts shortly after startup and later runs follow `interval_minutes`. Logs also report matched and unmatched users and skipped groups. There is no separate sync UI or run-history table.
## Identity and membership
`group_id_attribute` is matched exactly to LiteLLM's `team_id`. The default `cn` must be a unique, stable string across your selected groups. A group named `project-alpha` therefore creates or updates the team with ID `project-alpha`. A manually created team with that same ID is included in the sync. Matching only the display name does not adopt a different team.
`group_name_attribute` supplies the new team's display name. Team creation inherits `default_team_params`, the same way SCIM-created teams do. Existing team names, model permissions, budgets and other settings remain unchanged.
By default, `member` contains full user DNs. The sync reads each referenced user entry and compares its `mail` attribute to existing LiteLLM `user_email` values, ignoring email case. `litellm_match_field` may instead be `user_id` or `sso_user_id`, matched exactly. Set `user_match_attribute` to the LDAP attribute carrying the corresponding identity. Missing LiteLLM accounts are skipped and counted in the logs, then picked up by a later run after normal onboarding. Duplicate matches skip the affected group rather than choosing an account.
For POSIX groups, set `member_attribute: memberUid`. These values are looked up under `user_search_base` using `member_lookup_attribute`, which defaults to `uid`. Other membership attributes such as `uniqueMember` or `roleOccupant` are treated as full DNs. Only direct users are supported. Binary group identifiers, nested groups and AD ranged membership responses are rejected.
LDAP is authoritative for the membership of every matched team, including manually added members. Retained members keep their team roles. New members receive the regular member role. The shared membership operation updates the team roster, user membership list and member budget/spend association together. Removing a member also deletes their existing keys for that team, following the same behavior as SCIM/team-member removal. User accounts and memberships in other teams are retained. Global user roles and disabled states are unchanged.
## Failure behavior
A group no longer returned by LDAP leaves its team and memberships unchanged. A successfully read group with no members removes all memberships from its corresponding team. No team or user is ever deleted, and no user is provisioned by this sync.
Failed or incomplete group searches, referrals, duplicate group IDs and invalid responses abort the directory snapshot without applying it. Missing, ambiguous, nested or unreadable user entries skip their affected group. A failure for one valid group does not prevent other valid groups from synchronizing. A user disappearing during the membership write causes that group's transaction to make no changes. Repeated successful runs with the same roster do not rewrite it.
The LDAP identity must have permission to read every selected group, its complete membership attributes, and the user identity attributes. LDAP can omit attributes hidden by access controls just as it omits empty attributes, so the sync cannot distinguish every permission restriction from an empty group. Verify the bind account and dry-run output before enabling writes.
Runs have a timeout shorter than their shared Redis lease. A timeout cancels further work and the next interval retries from current state. Cancellation does not undo groups already committed earlier in the run. Stopping the sync (`enabled: false` or removing the block and restarting) leaves current teams and memberships in place.

View file

@ -87,6 +87,7 @@ from litellm.types.proxy.carried_budget_state import (
UserBudgetSnapshot as UserBudgetSnapshot, # noqa: PLC0414 # public re-export
)
from litellm.types.proxy.control_plane_endpoints import WorkerRegistryEntry
from litellm.types.proxy.ldap_sync import LDAPSyncSettings
from litellm.types.proxy.request_tag_mappings import RequestTagMapping
from litellm.types.proxy.spend_capture_rate import SpendCaptureRateCheckSettings
from litellm.types.router import AllowedModelRegion, RouterErrors, UpdateRouterConfig
@ -2480,6 +2481,10 @@ class ConfigGeneralSettings(LiteLLMPydanticObjectBase):
Documents all the fields supported by `general_settings` in config.yaml
"""
ldap_sync: LDAPSyncSettings | None = Field(
None, description="Periodically synchronize LDAP groups to teams and existing users' memberships. Requires Redis."
)
request_tag_mappings: Sequence[RequestTagMapping] | None = Field(
None,
description="Map W3C baggage values to request tags. Exact value/tag rules take precedence over tag_prefix fallbacks.",

View file

@ -0,0 +1,226 @@
import importlib
import ssl
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from typing import Final, Protocol, TypeAlias, cast
from urllib.parse import urlsplit
from pydantic import BaseModel, ConfigDict, StrictStr, TypeAdapter
from litellm.secret_managers.main import get_secret_str
from litellm.types.proxy.ldap_sync import LDAPSyncSettings
_STRINGS: Final = TypeAdapter(tuple[StrictStr, ...])
_GROUP_CLASSES: Final = frozenset({"group", "groupofnames", "groupofuniquenames", "posixgroup"})
@dataclass(frozen=True, slots=True)
class DirectoryGroup:
team_id: str
name: str
identities: tuple[str, ...]
@dataclass(frozen=True, slots=True)
class DirectoryError:
reason: str
group_dn: str | None = None
DirectoryResult: TypeAlias = tuple[DirectoryGroup | DirectoryError, ...] | DirectoryError
class _Entry(BaseModel):
model_config = ConfigDict(frozen=True)
type: str
dn: str = ""
attributes: Mapping[str, object] = {}
_ENTRIES: Final = TypeAdapter(tuple[_Entry, ...])
class _PagedSearch(Protocol):
def paged_search(
self,
*,
search_base: str,
search_filter: str,
search_scope: str,
attributes: Sequence[str],
paged_size: int,
generator: bool,
) -> object: ...
class _Extensions(Protocol):
@property
def standard(self) -> _PagedSearch: ...
class LDAPConnection(Protocol):
@property
def result(self) -> Mapping[str, object]: ...
@property
def response(self) -> object: ...
@property
def extend(self) -> _Extensions: ...
def open(self) -> object: ...
def start_tls(self) -> bool: ...
def bind(self) -> bool: ...
def unbind(self) -> bool: ...
def search(self, *, search_base: str, search_filter: str, search_scope: str, attributes: Sequence[str]) -> bool: ...
class _LDAPLibrary(Protocol):
def Tls(self, *, validate: ssl.VerifyMode, ca_certs_file: str | None) -> object: ...
def Server(self, host: str, *, port: int, use_ssl: bool, tls: object, connect_timeout: int) -> object: ...
def Connection(
self,
server: object,
*,
user: str | None,
password: str | None,
authentication: str,
auto_referrals: bool,
raise_exceptions: bool,
receive_timeout: int,
check_names: bool,
return_empty_attributes: bool,
) -> LDAPConnection: ...
def _secret(value: str | None) -> str | None:
if value is None or not value.startswith("os.environ/"):
return value
resolved: Final = get_secret_str(value)
if resolved is None:
raise ValueError("An LDAP bind credential environment variable is missing")
return resolved
def open_ldap_connection(settings: LDAPSyncSettings) -> LDAPConnection:
library: Final = cast( # cast-ok: ldap3 has no types, this protocol describes its public constructors
_LDAPLibrary, importlib.import_module("ldap3")
)
parsed: Final = urlsplit(settings.server_url)
tls: Final = library.Tls(validate=ssl.CERT_REQUIRED, ca_certs_file=settings.ca_cert_path)
server: Final = library.Server(
parsed.hostname or "",
port=parsed.port or (636 if parsed.scheme == "ldaps" else 389),
use_ssl=parsed.scheme == "ldaps",
tls=tls,
connect_timeout=settings.timeout_seconds,
)
return library.Connection(
server,
user=_secret(settings.bind_dn),
password=_secret(settings.bind_password.get_secret_value()) if settings.bind_password is not None else None,
authentication="SIMPLE" if settings.bind_dn else "ANONYMOUS",
auto_referrals=False,
raise_exceptions=True,
receive_timeout=settings.timeout_seconds,
check_names=False,
return_empty_attributes=True,
)
def _values(entry: _Entry, attribute: str) -> tuple[str, ...]:
value: Final = next((value for key, value in entry.attributes.items() if key.lower() == attribute.lower()), ())
return (value,) if isinstance(value, str) else _STRINGS.validate_python(value)
def _one(entry: _Entry, attribute: str) -> str | None:
values: Final = _values(entry, attribute)
return values[0] if len(values) == 1 and values[0].strip() else None
def _entries(raw: object) -> tuple[_Entry, ...] | DirectoryError:
entries: Final = _ENTRIES.validate_python(raw)
if any(entry.type != "searchResEntry" for entry in entries):
return DirectoryError("LDAP search returned a referral or incomplete entry")
return entries
def _escape_filter(value: str) -> str:
return "".join(
{"\\": r"\5c", "*": r"\2a", "(": r"\28", ")": r"\29", "\x00": r"\00"}.get(char, char) for char in value
)
def _resolve_member(connection: LDAPConnection, settings: LDAPSyncSettings, value: str) -> str | DirectoryError:
usernames: Final = settings.member_attribute.lower() == "memberuid"
_ = connection.search(
search_base=settings.user_search_base if usernames else value,
search_filter=f"({settings.member_lookup_attribute}={_escape_filter(value)})"
if usernames
else "(objectClass=*)",
search_scope="SUBTREE" if usernames else "BASE",
attributes=(settings.user_match_attribute, "objectClass"),
)
if connection.result.get("result") != 0:
return DirectoryError("LDAP member lookup did not complete successfully")
entries: Final = _entries(connection.response)
if isinstance(entries, DirectoryError):
return entries
if len(entries) != 1:
return DirectoryError("LDAP member lookup was missing or ambiguous")
entry: Final = entries[0]
if _GROUP_CLASSES.intersection(value.lower() for value in _values(entry, "objectClass")):
return DirectoryError("Nested LDAP groups are not supported")
identity: Final = _one(entry, settings.user_match_attribute)
return identity if identity is not None else DirectoryError("LDAP member identity is missing or multivalued")
def _read_group(
connection: LDAPConnection, settings: LDAPSyncSettings, entry: _Entry
) -> DirectoryGroup | DirectoryError:
team_id: Final = _one(entry, settings.group_id_attribute)
name: Final = _one(entry, settings.group_name_attribute)
if team_id is None or name is None:
return DirectoryError("LDAP group ID or name is missing or multivalued", entry.dn)
if any(";range=" in key.lower() for key in entry.attributes):
return DirectoryError("LDAP ranged membership responses are not supported", entry.dn)
identities: Final = tuple(
_resolve_member(connection, settings, value) for value in _values(entry, settings.member_attribute)
)
error: Final = next((identity for identity in identities if isinstance(identity, DirectoryError)), None)
if error is not None:
return DirectoryError(error.reason, entry.dn)
return DirectoryGroup(team_id, name, tuple(identity for identity in identities if isinstance(identity, str)))
def read_directory(settings: LDAPSyncSettings, connection: LDAPConnection | None = None) -> DirectoryResult:
try:
client: Final = connection if connection is not None else open_ldap_connection(settings)
try:
_ = client.open()
if settings.start_tls and urlsplit(settings.server_url).scheme == "ldap" and not client.start_tls():
return DirectoryError("LDAP TLS negotiation failed")
if not client.bind():
return DirectoryError("LDAP bind failed")
raw: Final = client.extend.standard.paged_search(
search_base=settings.group_search_base,
search_filter=settings.group_filter,
search_scope="SUBTREE",
attributes=(settings.group_id_attribute, settings.group_name_attribute, settings.member_attribute),
paged_size=500,
generator=False,
)
if client.result.get("result") != 0:
return DirectoryError("LDAP group search did not complete successfully")
entries: Final = _entries(raw)
if isinstance(entries, DirectoryError):
return entries
group_ids: Final = tuple(_one(entry, settings.group_id_attribute) for entry in entries)
if len(group_ids) != len(frozenset(group_ids)):
return DirectoryError("LDAP group IDs are not unique within the selected groups")
return tuple(_read_group(client, settings, entry) for entry in entries)
finally:
_ = client.unbind()
except Exception as exc:
return DirectoryError(f"LDAP read failed ({type(exc).__name__})")

View file

@ -0,0 +1,187 @@
import asyncio
import math
from collections.abc import Mapping
from datetime import datetime, timedelta, timezone
from typing import Final
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from fastapi import Request
from pydantic import TypeAdapter
from litellm._logging import verbose_proxy_logger
from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
from litellm.proxy.common_utils.ldap_directory import DirectoryError, DirectoryGroup, read_directory
from litellm.proxy.common_utils.user_api_key_cache import UserApiKeyCache
from litellm.proxy.db.db_transaction_queue.pod_lock_manager import PodLockManager
from litellm.proxy.management_endpoints.team_endpoints import new_team
from litellm.proxy.management_helpers.team_roster_sync import (
MembersMissing,
RosterSync,
RosterTarget,
new_team_request_with_defaults,
sync_team_roster,
)
from litellm.proxy.utils import PrismaClient, ProxyLogging
from litellm.repositories.team_repository import TeamRepository
from litellm.repositories.user_repository import UserRepository
from litellm.types.proxy.ldap_sync import LDAPSyncSettings
LDAP_SYNC_JOB_ID: Final = "ldap_team_sync"
_SETTINGS: Final = TypeAdapter(Mapping[str, object])
async def resolve_user_ids(
users: UserRepository, group: DirectoryGroup, settings: LDAPSyncSettings
) -> frozenset[str] | DirectoryError:
rows: Final = (
await users.find_by_emails(group.identities)
if settings.litellm_match_field == "user_email"
else await users.find_many_in(settings.litellm_match_field, group.identities)
)
identities: Final = tuple(getattr(row, settings.litellm_match_field) for row in rows)
normalized: Final = tuple(
identity.casefold() if settings.litellm_match_field == "user_email" else identity for identity in identities
)
if len(normalized) != len(frozenset(normalized)):
return DirectoryError("Multiple LiteLLM users match one LDAP identity")
expected: Final = frozenset(
identity.casefold() if settings.litellm_match_field == "user_email" else identity
for identity in group.identities
)
verbose_proxy_logger.info(
"LDAP sync team_id=%s matched_users=%d unmatched_users=%d",
group.team_id,
len(rows),
len(expected - frozenset(normalized)),
)
return frozenset(row.user_id for row in rows)
class LDAPSync:
def __init__(
self,
settings: LDAPSyncSettings,
prisma_client: PrismaClient,
pod_lock_manager: PodLockManager,
user_api_key_cache: UserApiKeyCache,
proxy_logging_obj: ProxyLogging,
) -> None:
self.settings = settings
self.prisma_client = prisma_client
self.pod_lock_manager = pod_lock_manager
self.user_api_key_cache = user_api_key_cache
self.proxy_logging_obj = proxy_logging_obj
self.teams = TeamRepository(prisma_client, use_writer=True)
self.users = UserRepository(prisma_client, use_writer=True)
async def sync_group(self, group: DirectoryGroup) -> None:
member_ids: Final = await resolve_user_ids(self.users, group, self.settings)
if isinstance(member_ids, DirectoryError):
verbose_proxy_logger.warning("LDAP sync skipped team_id=%s: %s", group.team_id, member_ids.reason)
return
team: Final = await self.teams.find_by_id(group.team_id, id_field="team_id")
if self.settings.dry_run:
existing: Final = (
frozenset(member.user_id for member in team.members_with_roles if member.user_id is not None)
if team is not None
else frozenset()
)
verbose_proxy_logger.info(
"LDAP sync dry_run team_id=%s create_team=%s add=%s remove=%s",
group.team_id,
team is None,
sorted(member_ids - existing),
sorted(existing - member_ids),
)
return
actor: Final = UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN)
if team is None:
await new_team(
data=new_team_request_with_defaults(group.team_id, group.name, ()),
http_request=Request(scope={"type": "http", "path": "/team/new"}),
user_api_key_dict=actor,
)
outcome: Final = await sync_team_roster(
self.prisma_client,
group.team_id,
RosterTarget(member_ids),
actor,
"ldap_sync",
self.user_api_key_cache,
self.proxy_logging_obj,
)
if isinstance(outcome, RosterSync):
verbose_proxy_logger.info(
"LDAP sync team_id=%s created=%s added=%d removed=%d",
group.team_id,
team is None,
len(outcome.added),
len(outcome.removed),
)
return
verbose_proxy_logger.warning(
"LDAP sync skipped team_id=%s: %s",
group.team_id,
"user disappeared before membership update" if isinstance(outcome, MembersMissing) else "team disappeared",
)
async def _sync(self) -> None:
snapshot: Final = await asyncio.to_thread(read_directory, self.settings)
if isinstance(snapshot, DirectoryError):
verbose_proxy_logger.warning("LDAP sync aborted: %s", snapshot.reason)
return
for group in snapshot:
if isinstance(group, DirectoryError):
verbose_proxy_logger.warning("LDAP sync skipped group_dn=%s: %s", group.group_dn, group.reason)
continue
try:
await self.sync_group(group)
except Exception as exc:
verbose_proxy_logger.warning(
"LDAP sync failed team_id=%s (%s); retrying next interval", group.team_id, type(exc).__name__
)
async def run(self) -> None:
lease_seconds: Final = math.ceil(self.settings.interval_minutes * 60) - 1
if (
await self.pod_lock_manager.acquire_lock(
cronjob_id=LDAP_SYNC_JOB_ID,
ttl=lease_seconds,
allow_reentrant=False,
)
is not True
):
return
try:
await asyncio.wait_for(self._sync(), timeout=min(600, lease_seconds / 2))
except asyncio.TimeoutError:
verbose_proxy_logger.warning("LDAP sync timed out; retrying next interval")
def initialize_ldap_sync(
scheduler: AsyncIOScheduler,
general_settings: Mapping[str, object],
prisma_client: PrismaClient | None,
pod_lock_manager: PodLockManager,
user_api_key_cache: UserApiKeyCache,
proxy_logging_obj: ProxyLogging,
) -> None:
raw: Final = general_settings.get("ldap_sync")
if raw is None or _SETTINGS.validate_python(raw).get("enabled") is not True:
return
settings: Final = LDAPSyncSettings.model_validate(raw)
if prisma_client is None:
raise ValueError("ldap_sync requires a database")
if pod_lock_manager.redis_cache is None:
raise ValueError("ldap_sync requires shared Redis for cross-worker coordination")
job: Final = LDAPSync(settings, prisma_client, pod_lock_manager, user_api_key_cache, proxy_logging_obj)
scheduler.add_job(
job.run,
"interval",
minutes=settings.interval_minutes,
next_run_time=datetime.now(timezone.utc) + timedelta(seconds=1),
id=LDAP_SYNC_JOB_ID,
replace_existing=True,
max_instances=1,
coalesce=True,
)

View file

@ -6,7 +6,6 @@ This is an enterprise feature and requires a premium license.
import re
from collections.abc import Awaitable, Callable, Iterable, Iterator, Mapping, Sequence
from copy import deepcopy
from dataclasses import dataclass
from functools import partial, reduce
from itertools import chain, groupby
@ -90,6 +89,7 @@ from litellm.proxy.management_helpers.team_roster_sync import (
RosterSyncOutcome,
RosterTarget,
TeamGone,
new_team_request_with_defaults,
sync_team_roster,
team_tx_db,
user_tx_db,
@ -2945,27 +2945,8 @@ def _new_team_request_with_defaults(
) -> NewTeamRequest:
"""Build the SCIM group's team request, applying litellm.default_team_params
(including models) the same way SSO auto-created teams do."""
default_params: Final = litellm.default_team_params
defaults: Final[Mapping[str, object]] = (
deepcopy(default_params)
if isinstance(default_params, dict)
else default_params.model_dump(exclude_none=True)
if default_params is not None
else {}
)
default_metadata: Final = defaults.get("metadata")
metadata: Final = {
**(default_metadata if isinstance(default_metadata, dict) else {}),
SCIM_MANAGED_TEAM_METADATA_KEY: True,
}
return NewTeamRequest.model_validate(
{
**defaults,
"team_id": team_id,
"team_alias": team_alias,
"members_with_roles": members_with_roles,
"metadata": metadata,
}
return new_team_request_with_defaults(
team_id, team_alias, members_with_roles, metadata={SCIM_MANAGED_TEAM_METADATA_KEY: True}
)

View file

@ -1,6 +1,7 @@
import asyncio
import json
from collections.abc import Mapping, Sequence
from copy import deepcopy
from dataclasses import dataclass
from datetime import timedelta
from types import MappingProxyType
@ -9,8 +10,9 @@ from typing import TYPE_CHECKING, Final
from pydantic import TypeAdapter
from typing_extensions import ReadOnly, TypedDict, assert_never
import litellm
from litellm._uuid import uuid
from litellm.proxy._types import LiteLLM_TeamTable, Member, UserAPIKeyAuth
from litellm.proxy._types import LiteLLM_TeamTable, Member, NewTeamRequest, UserAPIKeyAuth
from litellm.proxy.auth.auth_checks import (
await_inflight_team_membership_load,
delete_cache_team_object,
@ -52,6 +54,32 @@ ROSTER_TX_TIMEOUT: Final = timedelta(seconds=60)
_JSON_OBJECT: Final = TypeAdapter(dict[str, object])
def new_team_request_with_defaults(
team_id: str,
team_alias: str | None,
members_with_roles: Sequence[Member],
metadata: Mapping[str, object] | None = None,
) -> NewTeamRequest:
default_params: Final = litellm.default_team_params
defaults: Final[Mapping[str, object]] = (
deepcopy(default_params)
if isinstance(default_params, dict)
else default_params.model_dump(exclude_none=True)
if default_params is not None
else {}
)
default_metadata: Final = defaults.get("metadata")
return NewTeamRequest.model_validate(
{
**defaults,
"team_id": team_id,
"team_alias": team_alias,
"members_with_roles": members_with_roles,
"metadata": {**(default_metadata if isinstance(default_metadata, dict) else {}), **(metadata or {})},
}
)
@dataclass(frozen=True, slots=True)
class RosterTarget:
member_ids: frozenset[str]

View file

@ -1862,6 +1862,10 @@ async def _proxy_lifespan(app: FastAPI, resources: AsyncExitStack) -> AsyncGener
redis_usage_cache=redis_usage_cache,
)
_ldap_sync_settings: Final = general_settings.get("ldap_sync")
if prisma_client is None and isinstance(_ldap_sync_settings, dict) and _ldap_sync_settings.get("enabled") is True:
raise ValueError("ldap_sync requires a database")
### START BATCH WRITING DB + CHECKING NEW MODELS###
worker_heartbeat: Final = (
await ProxyStartupEvent.initialize_scheduled_background_jobs(
@ -11801,6 +11805,19 @@ class ProxyStartupEvent:
await cls._initialize_spend_tracking_background_jobs(scheduler=scheduler)
ldap_settings: Final = general_settings.get("ldap_sync")
if isinstance(ldap_settings, dict) and ldap_settings.get("enabled") is True:
from litellm.proxy.common_utils.ldap_sync import initialize_ldap_sync
initialize_ldap_sync(
scheduler=scheduler,
general_settings=general_settings,
prisma_client=prisma_client,
pod_lock_manager=proxy_logging_obj.db_spend_update_writer.pod_lock_manager,
user_api_key_cache=user_api_key_cache,
proxy_logging_obj=proxy_logging_obj,
)
cls._initialize_spend_capture_rate_check_job(
scheduler=scheduler,
proxy_logging_obj=proxy_logging_obj,

View file

@ -0,0 +1,66 @@
from typing import Final, Literal
from urllib.parse import urlsplit
from pydantic import BaseModel, ConfigDict, Field, SecretStr, field_validator, model_validator
from typing_extensions import Self
class LDAPSyncSettings(BaseModel):
model_config = ConfigDict(frozen=True, extra="forbid")
enabled: bool = False
server_url: str = ""
group_search_base: str = ""
group_filter: str = ""
user_search_base: str = ""
group_id_attribute: str = "cn"
group_name_attribute: str = "cn"
member_attribute: str = "member"
member_lookup_attribute: str = "uid"
user_match_attribute: str = "mail"
litellm_match_field: Literal["user_email", "user_id", "sso_user_id"] = "user_email"
interval_minutes: float = Field(default=30, ge=0.1, allow_inf_nan=False)
dry_run: bool = False
bind_dn: str | None = None
bind_password: SecretStr | None = None
ca_cert_path: str | None = None
start_tls: bool = False
timeout_seconds: int = Field(default=30, ge=1, le=120)
@field_validator("server_url")
@classmethod
def validate_server_url(cls, value: str) -> str:
parsed: Final = urlsplit(value)
if parsed.scheme not in {"ldap", "ldaps"} or not parsed.hostname:
raise ValueError("server_url must be an ldap:// or ldaps:// server URL")
if parsed.username or parsed.password or parsed.query or parsed.fragment or parsed.path not in {"", "/"}:
raise ValueError("server_url must contain only the LDAP host and optional port")
_ = parsed.port
return value
@field_validator(
"group_id_attribute",
"group_name_attribute",
"member_attribute",
"member_lookup_attribute",
"user_match_attribute",
)
@classmethod
def validate_attribute(cls, value: str) -> str:
if not value or not value[0].isalpha() or not all(char.isalnum() or char == "-" for char in value):
raise ValueError("LDAP attributes must be plain attribute names")
return value
@model_validator(mode="after")
def validate_bind(self) -> Self:
if self.enabled and not all(
(self.server_url, self.group_search_base, self.group_filter, self.user_search_base)
):
raise ValueError(
"Enabled LDAP sync requires server_url, group_search_base, group_filter and user_search_base"
)
if bool(self.bind_dn) != bool(self.bind_password):
raise ValueError("bind_dn and bind_password must be configured together")
if self.bind_dn and urlsplit(self.server_url).scheme == "ldap" and not self.start_tls:
raise ValueError("Service-account binds require ldaps:// or start_tls: true")
return self

View file

@ -72,6 +72,7 @@ proxy = [
"hiredis>=3.0.0,<4.0; python_version < '3.15'",
"hiredis>=3.5.0,<4.0; python_version >= '3.15'",
"apscheduler>=3.11.2,<4.0",
"ldap3>=2.9.1,<3.0",
"fastapi-sso>=0.19.0,<1.0",
"PyJWT>=2.13.0,<3.0",
"python-multipart>=0.0.27,<1.0",

View file

@ -0,0 +1,189 @@
from collections.abc import Mapping, Sequence
from typing import Final
import pytest
from pydantic import ValidationError
from litellm.proxy.common_utils.ldap_directory import DirectoryError, DirectoryGroup, read_directory
from litellm.types.proxy.ldap_sync import LDAPSyncSettings
SETTINGS: Final = LDAPSyncSettings(
enabled=True,
server_url="ldap://directory.example.org",
group_search_base="ou=groups,dc=example,dc=org",
group_filter="(cn=project-*)",
user_search_base="ou=people,dc=example,dc=org",
)
ALICE: Final = "uid=alice,ou=people,dc=example,dc=org"
BOB: Final = "uid=bob,ou=people,dc=example,dc=org"
def entry(dn: str, **attributes: object) -> Mapping[str, object]:
return {"type": "searchResEntry", "dn": dn, "attributes": attributes}
def group(name: str = "project-a", **attributes: object) -> Mapping[str, object]:
return entry(f"cn={name},ou=groups,dc=example,dc=org", cn=[name], **attributes)
def user(mail: str, **attributes: object) -> Mapping[str, object]:
return entry(ALICE, mail=[mail], objectClass=["person"], **attributes)
class Connection:
def __init__(
self,
groups: Sequence[Mapping[str, object]],
users: Mapping[str, Sequence[Mapping[str, object]]] | None = None,
result_code: int = 0,
tls_ok: bool = True,
) -> None:
self.groups = groups
self.users = users or {}
self.result_code = result_code
self.result: Mapping[str, object] = {"result": result_code}
self.response: object = ()
self.extend = self
self.standard = self
self.searches: list[tuple[str, str, str]] = []
self.closed = False
self.tls_ok = tls_ok
def open(self) -> None:
pass
def bind(self) -> bool:
return True
def unbind(self) -> bool:
self.closed = True
return True
def start_tls(self) -> bool:
return self.tls_ok
def paged_search(
self,
*,
search_base: str,
search_filter: str,
search_scope: str,
attributes: Sequence[str],
paged_size: int,
generator: bool,
) -> object:
return self.groups
def search(self, *, search_base: str, search_filter: str, search_scope: str, attributes: Sequence[str]) -> bool:
self.searches.append((search_base, search_filter, search_scope))
self.response = self.users.get(search_base if search_scope == "BASE" else search_filter, ())
self.result = {"result": self.result_code}
return self.result_code == 0
def test_direct_members_resolve_to_user_identities_and_connection_is_closed():
connection = Connection(
[group(member=[ALICE, BOB])], {ALICE: [user("alice@example.org")], BOB: [user("bob@example.org")]}
)
result = read_directory(SETTINGS, connection)
assert result == (DirectoryGroup("project-a", "project-a", ("alice@example.org", "bob@example.org")),)
assert connection.closed
@pytest.mark.parametrize("membership", [[], ()])
def test_successfully_read_empty_group_has_empty_roster(membership: Sequence[str]):
assert read_directory(SETTINGS, Connection([group(member=membership)])) == (
DirectoryGroup("project-a", "project-a", ()),
)
def test_omitted_membership_attribute_is_an_empty_roster_with_complete_read_permissions():
assert read_directory(SETTINGS, Connection([group()])) == (DirectoryGroup("project-a", "project-a", ()),)
def test_ldap_attribute_names_are_case_insensitive():
connection = Connection(
[entry("group", CN=["project-a"], MEMBER=[ALICE])], {ALICE: [entry(ALICE, MAIL=["a@example.org"])]}
)
assert read_directory(SETTINGS, connection) == (DirectoryGroup("project-a", "project-a", ("a@example.org",)),)
@pytest.mark.parametrize(
"members",
[
(),
(user("one@example.org"), user("two@example.org")),
(entry(ALICE, mail=["one@example.org"], objectClass=["groupOfNames"]),),
(entry(ALICE, mail=[], objectClass=["person"]),),
],
)
def test_incomplete_or_nested_member_does_not_become_empty_group(members: Sequence[Mapping[str, object]]):
result = read_directory(SETTINGS, Connection([group(member=[ALICE])], {ALICE: members}))
assert isinstance(result, tuple) and isinstance(result[0], DirectoryError)
def test_duplicate_ids_are_rejected_even_when_one_group_cannot_resolve_members():
result = read_directory(SETTINGS, Connection([group(member=[]), group(member=[ALICE])]))
assert result == DirectoryError("LDAP group IDs are not unique within the selected groups")
def test_ranged_membership_is_rejected_instead_of_removing_unread_members():
result = read_directory(SETTINGS, Connection([group(**{"member;range=0-1499": [ALICE]})]))
assert isinstance(result, tuple) and isinstance(result[0], DirectoryError)
assert "ranged" in result[0].reason
@pytest.mark.parametrize("code", [4, 10, 50])
def test_partial_directory_results_are_rejected(code: int):
result = read_directory(SETTINGS, Connection([group(member=[])], result_code=code))
assert result == DirectoryError("LDAP group search did not complete successfully")
def test_referrals_are_not_treated_as_groups():
result = read_directory(SETTINGS, Connection([{"type": "searchResRef", "uri": ["ldap://other.example.org"]}]))
assert isinstance(result, DirectoryError)
def test_missing_group_preserves_absence_in_snapshot():
assert read_directory(SETTINGS, Connection([])) == ()
def test_binary_group_identifiers_are_not_implicitly_decoded():
result = read_directory(SETTINGS, Connection([entry("group", cn=[b"project-a"], member=[])]))
assert isinstance(result, DirectoryError)
def test_member_uid_lookup_escapes_filter_values():
settings = SETTINGS.model_copy(update={"member_attribute": "memberUid"})
expected_filter = r"(uid=alice\2a\29\28uid=\2a)"
connection = Connection([group(memberUid=["alice*)(uid=*"])], {expected_filter: [user("alice@example.org")]})
assert read_directory(settings, connection) == (DirectoryGroup("project-a", "project-a", ("alice@example.org",)),)
assert connection.searches == [(SETTINGS.user_search_base, expected_filter, "SUBTREE")]
def test_start_tls_failure_prevents_search_and_closes_connection():
connection = Connection([group(member=[])], tls_ok=False)
assert read_directory(SETTINGS.model_copy(update={"start_tls": True}), connection) == DirectoryError(
"LDAP TLS negotiation failed"
)
assert connection.closed
def test_disabled_configuration_needs_no_connection_details():
assert not LDAPSyncSettings(enabled=False).enabled
@pytest.mark.parametrize(
"updates",
[
{"server_url": "https://example.org"},
{"server_url": "ldap://user:password@example.org"},
{"bind_dn": "cn=admin", "bind_password": "password"},
{"group_id_attribute": "cn)(uid=*"},
{"litellm_match_field": "user_role"},
{"interval_minutes": 0},
],
)
def test_invalid_configuration_is_rejected(updates: Mapping[str, object]):
with pytest.raises(ValidationError):
LDAPSyncSettings.model_validate({**SETTINGS.model_dump(), **updates})

View file

@ -0,0 +1,75 @@
from collections.abc import Mapping, Sequence
from typing import Final
import pytest
from litellm.models.user import LiteLLM_UserTable
from litellm.proxy.common_utils.ldap_directory import DirectoryError, DirectoryGroup
from litellm.proxy.common_utils.ldap_sync import resolve_user_ids
from litellm.repositories.user_repository import UserRepository
from litellm.types.proxy.ldap_sync import LDAPSyncSettings
SETTINGS: Final = LDAPSyncSettings(
enabled=True,
server_url="ldaps://directory.example.org",
group_search_base="ou=groups,dc=example,dc=org",
group_filter="(cn=project-*)",
user_search_base="ou=people,dc=example,dc=org",
)
class Users:
def __init__(self, rows: Sequence[LiteLLM_UserTable]) -> None:
self.rows = rows
self.db = self
self.writer_db = self
self.litellm_usertable = self
async def find_many(
self,
*,
where: Mapping[str, object] | None = None,
take: int | None = None,
skip: int | None = None,
order: Mapping[str, str] | None = None,
) -> Sequence[LiteLLM_UserTable]:
return self.rows
@pytest.mark.asyncio
async def test_existing_users_match_case_insensitively_without_creating_unknown_users():
users = Users([LiteLLM_UserTable(user_id="alice", user_email="Alice@example.org")])
result = await resolve_user_ids(
UserRepository(users),
DirectoryGroup("project-a", "project-a", ("alice@example.org", "unknown@example.org")),
SETTINGS,
)
assert result == frozenset({"alice"})
assert len(users.rows) == 1
@pytest.mark.asyncio
async def test_duplicate_email_accounts_abort_group_matching():
users = Users(
[
LiteLLM_UserTable(user_id="alice", user_email="Alice@example.org"),
LiteLLM_UserTable(user_id="other", user_email="alice@example.org"),
]
)
result = await resolve_user_ids(
UserRepository(users), DirectoryGroup("project-a", "project-a", ("alice@example.org",)), SETTINGS
)
assert isinstance(result, DirectoryError)
@pytest.mark.asyncio
@pytest.mark.parametrize("field", ["user_id", "sso_user_id"])
async def test_configured_identity_field_matches_existing_user(field: str):
users = Users([LiteLLM_UserTable(user_id="alice", sso_user_id="z1234")])
identity = "alice" if field == "user_id" else "z1234"
result = await resolve_user_ids(
UserRepository(users),
DirectoryGroup("project-a", "project-a", (identity,)),
SETTINGS.model_copy(update={"litellm_match_field": field}),
)
assert result == frozenset({"alice"})

15
uv.lock generated
View file

@ -4820,6 +4820,19 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/03/70/0e0cc80a3b064c8d6c8d697c3125ed86e39d5a7393ec6dc8b07cb1cf13c4/langsmith-0.8.18-py3-none-any.whl", hash = "sha256:3940183349993faef48e6c7d08e4822ee9cefd906b362d0e3c2d650314d2f282", size = 508108, upload-time = "2026-06-19T13:12:15.348Z" },
]
[[package]]
name = "ldap3"
version = "2.9.1"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "pyasn1" },
]
sdist = { url = "https://files.pythonhosted.org/packages/43/ac/96bd5464e3edbc61595d0d69989f5d9969ae411866427b2500a8e5b812c0/ldap3-2.9.1.tar.gz", hash = "sha256:f3e7fc4718e3f09dda568b57100095e0ce58633bcabbed8667ce3f8fbaa4229f", size = 398830, upload-time = "2021-07-18T06:34:21.786Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/4e/f6/71d6ec9f18da0b2201287ce9db6afb1a1f637dedb3f0703409558981c723/ldap3-2.9.1-py2.py3-none-any.whl", hash = "sha256:5869596fc4948797020d3f03b7939da938778a0f9e2009f7a072ccf92b8e8d70", size = 432192, upload-time = "2021-07-18T06:34:12.905Z" },
]
[[package]]
name = "lazy-object-proxy"
version = "1.12.0"
@ -5067,6 +5080,7 @@ mlflow = [
]
proxy = [
{ name = "apscheduler" },
{ name = "ldap3" },
{ name = "azure-identity" },
{ name = "azure-storage-blob" },
{ name = "backoff" },
@ -5265,6 +5279,7 @@ requires-dist = [
{ name = "anthropic", extras = ["vertex"], marker = "extra == 'proxy-runtime'", specifier = ">=0.84.0,<1.0" },
{ name = "anyio", specifier = ">=4.14.0,<5.0" },
{ name = "apscheduler", marker = "extra == 'proxy'", specifier = ">=3.11.2,<4.0" },
{ name = "ldap3", marker = "extra == 'proxy'", specifier = ">=2.9.1,<3.0" },
{ name = "async-timeout", marker = "python_full_version < '3.11'", specifier = ">=4.0.3,<6.0" },
{ name = "audioread", marker = "extra == 'stt-nvidia-riva'", specifier = ">=3.0.1" },
{ name = "aurelio-sdk", marker = "python_full_version < '3.14' and extra == 'semantic-router'", specifier = ">=0.0.19,<1.0" },