Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/management_endpoints/team_admin_field_permissions.py: 32%
162 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1"""Proxy-wide allow-list of what a team admin may do on the teams they administer: team-settings fields on
2/team/update, the ``projects`` permission for /project/new and /project/update, and the
3``member_key_budgets`` permission for budget fields on other members' keys via /key/update."""
5from collections.abc import Mapping
6from dataclasses import dataclass
7from types import MappingProxyType
8from typing import Final, Literal, TypeAlias
10from fastapi import HTTPException
11from pydantic import TypeAdapter, ValidationError
12from typing_extensions import assert_never
14from litellm._logging import verbose_proxy_logger
15from litellm.models.team import LiteLLM_TeamTable
16from litellm.models.verification_token import LiteLLM_VerificationToken
17from litellm.proxy._types import (
18 LiteLLM_ManagementEndpoint_MetadataFields,
19 LiteLLM_ManagementEndpoint_MetadataFields_Premium,
20 UpdateKeyRequest,
21 UpdateTeamRequest,
22)
24TEAM_ADMIN_EDITABLE_TEAM_FIELDS_SETTING: Final = "team_admin_editable_team_fields"
26# TODO(LIT-5722): add the remaining team settings one per PR, each with its value-diff tests and dashboard field
27SUPPORTED_TEAM_ADMIN_EDITABLE_TEAM_FIELDS: Final[frozenset[str]] = frozenset({"tpm_limit", "rpm_limit", "max_budget"})
28TEAM_ADMIN_PROJECTS_PERMISSION: Final = "projects"
29TEAM_ADMIN_MEMBER_KEY_BUDGETS_PERMISSION: Final = "member_key_budgets"
30SUPPORTED_TEAM_ADMIN_PERMISSIONS: Final[frozenset[str]] = SUPPORTED_TEAM_ADMIN_EDITABLE_TEAM_FIELDS | {
31 TEAM_ADMIN_PROJECTS_PERMISSION,
32 TEAM_ADMIN_MEMBER_KEY_BUDGETS_PERMISSION,
33}
35# spend is deliberately excluded: the stored row lags the live cross-pod counter, so a value-diff gate
36# would let a team admin overwrite real usage.
37KEY_BUDGET_FIELDS: Final[frozenset[str]] = frozenset({"max_budget", "budget_duration", "soft_budget", "budget_limits"})
38_KEY_REQUEST_IDENTITY: Final[frozenset[str]] = frozenset({"key", "token", "metadata"})
40_FIELD_LIST: Final = TypeAdapter(list[str])
41_JSON_OBJECT: Final = TypeAdapter(dict[str, object])
42_WINDOW_LIST: Final = TypeAdapter(list[dict[str, object]])
43_EMPTY: Final[Mapping[str, object]] = MappingProxyType({})
44_METADATA_FOLDED_FIELDS: Final[frozenset[str]] = frozenset(
45 (*LiteLLM_ManagementEndpoint_MetadataFields, *LiteLLM_ManagementEndpoint_MetadataFields_Premium)
46)
47_SYSTEM_MANAGED_METADATA_KEYS: Final[frozenset[str]] = frozenset({"team_member_budget_id"})
48_NOT_COLUMNS: Final[frozenset[str]] = frozenset({"team_id", "metadata"})
49_SETTINGS_LOCATION: Final = "Settings > UI > Team admin editable fields"
52@dataclass(frozen=True, slots=True)
53class TeamAdminEditAllowed:
54 request: UpdateTeamRequest
55 kind: Literal["allowed"] = "allowed"
58@dataclass(frozen=True, slots=True)
59class TeamAdminEditingDisabled:
60 kind: Literal["disabled"] = "disabled"
63@dataclass(frozen=True, slots=True)
64class TeamAdminFieldNotPermitted:
65 field: str
66 kind: Literal["field_not_permitted"] = "field_not_permitted"
69TeamAdminEditVerdict: TypeAlias = TeamAdminEditAllowed | TeamAdminEditingDisabled | TeamAdminFieldNotPermitted
72def resolve_team_admin_editable_fields(
73 general_settings: Mapping[str, object],
74 supported: frozenset[str],
75) -> frozenset[str]:
76 raw: Final = general_settings.get(TEAM_ADMIN_EDITABLE_TEAM_FIELDS_SETTING)
77 if raw is None:
78 return frozenset()
79 try:
80 configured: Final = frozenset(_FIELD_LIST.validate_python(raw))
81 except ValidationError:
82 verbose_proxy_logger.warning(
83 "%s must be a list of field names; ignoring %r", TEAM_ADMIN_EDITABLE_TEAM_FIELDS_SETTING, raw
84 )
85 return frozenset()
86 unsupported: Final = configured - supported - SUPPORTED_TEAM_ADMIN_PERMISSIONS
87 if unsupported:
88 verbose_proxy_logger.warning(
89 "%s ignores unsupported field(s) %s; supported: %s",
90 TEAM_ADMIN_EDITABLE_TEAM_FIELDS_SETTING,
91 sorted(unsupported),
92 sorted(supported | SUPPORTED_TEAM_ADMIN_PERMISSIONS),
93 )
94 return configured & supported
97def team_admin_may_manage_projects(general_settings: Mapping[str, object]) -> bool:
98 return TEAM_ADMIN_PROJECTS_PERMISSION in resolve_team_admin_editable_fields(
99 general_settings, frozenset({TEAM_ADMIN_PROJECTS_PERMISSION})
100 )
103def team_admin_may_edit_member_key_budgets(general_settings: Mapping[str, object]) -> bool:
104 return TEAM_ADMIN_MEMBER_KEY_BUDGETS_PERMISSION in resolve_team_admin_editable_fields(
105 general_settings, frozenset({TEAM_ADMIN_MEMBER_KEY_BUDGETS_PERMISSION})
106 )
109def _as_object(value: object) -> Mapping[str, object]:
110 try:
111 return _JSON_OBJECT.validate_json(value) if isinstance(value, str) else _JSON_OBJECT.validate_python(value)
112 except ValidationError:
113 return _EMPTY
116def _stored_metadata(existing: Mapping[str, object]) -> Mapping[str, object]:
117 return _as_object(existing.get("metadata"))
120def _submitted_metadata(
121 data: UpdateTeamRequest | UpdateKeyRequest, submitted: Mapping[str, object], existing: Mapping[str, object]
122) -> Mapping[str, object]:
123 """Metadata as it would be stored: the caller's dict (or the stored one) with top-level folded fields laid over."""
124 base: Final = (
125 _as_object(submitted.get("metadata")) if "metadata" in data.model_fields_set else _stored_metadata(existing)
126 )
127 folded: Final = data.model_fields_set & _METADATA_FOLDED_FIELDS
128 return MappingProxyType({key: submitted[key] if key in folded else base[key] for key in base.keys() | folded})
131def _metadata_changes(
132 data: UpdateTeamRequest | UpdateKeyRequest, submitted: Mapping[str, object], existing: Mapping[str, object]
133) -> frozenset[str]:
134 merged: Final = _submitted_metadata(data, submitted, existing)
135 stored: Final = _stored_metadata(existing)
136 return frozenset(
137 key if key in _METADATA_FOLDED_FIELDS else "metadata"
138 for key in (merged.keys() | stored.keys()) - _SYSTEM_MANAGED_METADATA_KEYS
139 if merged.get(key) != stored.get(key)
140 )
143def _stored_model_aliases(existing_row: LiteLLM_TeamTable) -> Mapping[str, object]:
144 table: Final = existing_row.litellm_model_table
145 return _as_object(_JSON_OBJECT.validate_json(table.model_dump_json()).get("model_aliases")) if table else _EMPTY
148def _column_changed(
149 field: str, submitted: Mapping[str, object], existing: Mapping[str, object], existing_row: LiteLLM_TeamTable
150) -> bool:
151 if field == "model_aliases":
152 return _as_object(submitted.get(field)) != _stored_model_aliases(existing_row)
153 if field in LiteLLM_TeamTable.model_fields:
154 return submitted.get(field) != existing.get(field)
155 return True
158def changed_team_fields(data: UpdateTeamRequest, existing_row: LiteLLM_TeamTable) -> frozenset[str]:
159 """Logical field names whose stored value the request would change.
161 Request and stored row are compared as JSON values so both sides share one representation. Fields the
162 server folds into metadata are attributed to their own name whether they arrive top-level or inside
163 ``metadata``; anything else in ``metadata`` is attributed to ``metadata``. Fields with no stored
164 counterpart on the team row count as changed whenever they are sent.
165 """
166 submitted: Final = _JSON_OBJECT.validate_json(data.model_dump_json(exclude_unset=True))
167 existing: Final = _JSON_OBJECT.validate_json(existing_row.model_dump_json())
168 column_fields: Final = frozenset(data.model_fields_set) - _NOT_COLUMNS - _METADATA_FOLDED_FIELDS
169 column_changes: Final = frozenset(
170 field for field in column_fields if _column_changed(field, submitted, existing, existing_row)
171 )
172 return column_changes | _metadata_changes(data, submitted, existing)
175def _only_changes(data: UpdateTeamRequest, changed: frozenset[str]) -> UpdateTeamRequest:
176 """The request without the values it resends unchanged, which would otherwise still trigger derived writes
177 such as a resent budget_duration pushing budget_reset_at back."""
178 sent: Final = frozenset(data.model_fields_set)
179 via_metadata: Final = frozenset({"metadata"}) if changed - sent else frozenset[str]()
180 kept: Final = frozenset({"team_id"}) | (changed & sent) | via_metadata
181 return UpdateTeamRequest.model_validate(data.model_dump(include=MappingProxyType({field: True for field in kept})))
184def team_admin_edit_verdict(
185 data: UpdateTeamRequest,
186 existing: LiteLLM_TeamTable,
187 permitted: frozenset[str],
188) -> TeamAdminEditVerdict:
189 if not permitted:
190 return TeamAdminEditingDisabled()
191 changed: Final = changed_team_fields(data, existing)
192 blocked: Final = sorted(changed - permitted)
193 if blocked:
194 return TeamAdminFieldNotPermitted(field=blocked[0])
195 return TeamAdminEditAllowed(request=_only_changes(data, changed))
198def team_admin_request_or_raise(verdict: TeamAdminEditVerdict) -> UpdateTeamRequest:
199 match verdict:
200 case TeamAdminEditAllowed():
201 return verdict.request
202 case TeamAdminEditingDisabled():
203 raise HTTPException(
204 status_code=403,
205 detail=(
206 "Team admins on this proxy cannot edit team settings. "
207 f"Ask a proxy admin to enable fields under {_SETTINGS_LOCATION}."
208 ),
209 )
210 case TeamAdminFieldNotPermitted(field=field):
211 raise HTTPException(
212 status_code=403,
213 detail=(
214 f"Team admins on this proxy do not have permission to update '{field}'. "
215 f"Ask a proxy admin to add it under {_SETTINGS_LOCATION}."
216 ),
217 )
218 case _:
219 assert_never(verdict)
222def _budget_windows(value: object) -> frozenset[tuple[object, object]] | None:
223 """(budget_duration, max_budget) pairs for a stored or submitted budget_limits value.
225 Stored windows carry server-added keys like ``reset_at``; only the caller-owned pair matters.
226 ``None`` means the value is not a list of windows and needs a plain comparison.
227 """
228 if value is None:
229 return frozenset()
230 if not isinstance(value, list):
231 return None
232 try:
233 windows_input: Final = _WINDOW_LIST.validate_python(value)
234 except ValidationError:
235 return None
236 windows: Final = frozenset((window.get("budget_duration"), window.get("max_budget")) for window in windows_input)
237 if len(windows) != len(windows_input):
238 return None
239 return windows
242def _key_column_changed(field: str, submitted: Mapping[str, object], existing: Mapping[str, object]) -> bool:
243 if field == "budget_limits":
244 sent: Final = _budget_windows(submitted.get(field))
245 stored: Final = _budget_windows(existing.get(field))
246 if sent is not None and stored is not None:
247 return sent != stored
248 if field in LiteLLM_VerificationToken.model_fields:
249 return submitted.get(field) != existing.get(field)
250 return True
253def changed_key_fields(data: UpdateKeyRequest, existing_row: LiteLLM_VerificationToken) -> frozenset[str]:
254 """Logical field names whose stored value the key-update request would change.
256 Same JSON-value comparison as :func:`changed_team_fields`: columns compare against the stored row,
257 fields the key endpoint folds into ``metadata`` compare against ``existing_row.metadata``, other
258 ``metadata`` keys are attributed to ``metadata``, and fields with no stored counterpart count as
259 changed whenever they are sent. ``budget_limits`` compares (budget_duration, max_budget) pairs so
260 order and server-computed ``reset_at`` values do not read as edits.
261 """
262 submitted: Final = _JSON_OBJECT.validate_json(data.model_dump_json(exclude_unset=True))
263 existing: Final = _JSON_OBJECT.validate_json(existing_row.model_dump_json())
264 column_fields: Final = frozenset(data.model_fields_set) - _KEY_REQUEST_IDENTITY - _METADATA_FOLDED_FIELDS
265 column_changes: Final = frozenset(
266 field for field in column_fields if _key_column_changed(field, submitted, existing)
267 )
268 return column_changes | _metadata_changes(data, submitted, existing)
271@dataclass(frozen=True, slots=True)
272class TeamAdminKeyEditAllowed:
273 changed: frozenset[str]
274 kind: Literal["allowed"] = "allowed"
277@dataclass(frozen=True, slots=True)
278class TeamAdminMemberKeyEditingDisabled:
279 kind: Literal["disabled"] = "disabled"
282TeamAdminKeyEditVerdict: TypeAlias = (
283 TeamAdminKeyEditAllowed | TeamAdminMemberKeyEditingDisabled | TeamAdminFieldNotPermitted
284)
287def team_admin_key_edit_verdict(
288 data: UpdateKeyRequest,
289 existing: LiteLLM_VerificationToken,
290 enabled: bool,
291) -> TeamAdminKeyEditVerdict:
292 if not enabled:
293 return TeamAdminMemberKeyEditingDisabled()
294 changed: Final = changed_key_fields(data, existing)
295 blocked: Final = sorted(
296 (changed | (frozenset({"spend"}) if "spend" in data.model_fields_set else frozenset())) - KEY_BUDGET_FIELDS
297 )
298 if blocked:
299 return TeamAdminFieldNotPermitted(field=blocked[0])
300 return TeamAdminKeyEditAllowed(changed=changed)
303def team_admin_key_request_or_raise(verdict: TeamAdminKeyEditVerdict) -> None:
304 match verdict:
305 case TeamAdminKeyEditAllowed():
306 return
307 case TeamAdminMemberKeyEditingDisabled():
308 raise HTTPException(
309 status_code=403,
310 detail=(
311 "Team admins on this proxy cannot update budgets on other members' keys. "
312 f"Ask a proxy admin to enable '{TEAM_ADMIN_MEMBER_KEY_BUDGETS_PERMISSION}' "
313 f"under {_SETTINGS_LOCATION}."
314 ),
315 )
316 case TeamAdminFieldNotPermitted(field=field):
317 raise HTTPException(
318 status_code=403,
319 detail=(
320 "Team admins on this proxy may only update budget fields on other members' keys, "
321 f"not '{field}'. Ask a proxy admin to add it under {_SETTINGS_LOCATION}."
322 ),
323 )
324 case _:
325 assert_never(verdict)