Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/auth/auth_checks_organization.py: 39%
73 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1"""
2Auth Checks for Organizations
3"""
5from collections.abc import Awaitable, Callable
6from typing import Final
8from fastapi import status
10from litellm.proxy._types import *
13def organization_role_based_access_check(
14 request_body: dict,
15 user_object: LiteLLM_UserTable | None,
16 route: str,
17):
18 """
19 Role based access control checks only run if a user is part of an Organization
21 Organization Checks:
22 ONLY RUN IF user_object.organization_memberships is not None
24 1. Only Proxy Admins can access /organization/new
25 2. IF route is a LiteLLMRoutes.org_admin_only_routes, then check if user is an Org Admin for that organization
27 """
29 if user_object is None:
30 return
32 passed_organization_id: Final[str | None] = request_body.get("organization_id", None)
34 if route == "/organization/new":
35 if user_object.user_role != LitellmUserRoles.PROXY_ADMIN.value: 35 ↛ 36line 35 didn't jump to line 36 because the condition on line 35 was never true
36 raise ProxyException(
37 message=f"Only proxy admins can create new organizations. You are {user_object.user_role}",
38 type=ProxyErrorTypes.auth_error.value,
39 param="user_role",
40 code=status.HTTP_401_UNAUTHORIZED,
41 )
43 if user_object.user_role == LitellmUserRoles.PROXY_ADMIN.value:
44 return
46 # Checks if route is an Org Admin Only Route
47 if route in LiteLLMRoutes.org_admin_only_routes.value: 47 ↛ 48line 47 didn't jump to line 48 because the condition on line 47 was never true
48 (
49 _user_organizations,
50 _user_organization_role_mapping,
51 ) = get_user_organization_info(user_object)
53 if user_object.organization_memberships is None:
54 raise ProxyException(
55 message=f"Tried to access route={route} but you are not a member of any organization. Please contact the proxy admin to request access.",
56 type=ProxyErrorTypes.auth_error.value,
57 param="organization_id",
58 code=status.HTTP_401_UNAUTHORIZED,
59 )
61 if passed_organization_id is None:
62 raise ProxyException(
63 message="Passed organization_id is None, please pass an organization_id in your request",
64 type=ProxyErrorTypes.auth_error.value,
65 param="organization_id",
66 code=status.HTTP_401_UNAUTHORIZED,
67 )
69 user_role: Final[LitellmUserRoles | None] = _user_organization_role_mapping.get(passed_organization_id)
70 if user_role is None:
71 raise ProxyException(
72 message=f"You do not have a role within the selected organization. Passed organization_id: {passed_organization_id}. Please contact the organization admin to request access.",
73 type=ProxyErrorTypes.auth_error.value,
74 param="organization_id",
75 code=status.HTTP_401_UNAUTHORIZED,
76 )
78 if user_role != LitellmUserRoles.ORG_ADMIN.value:
79 raise ProxyException(
80 message=f"You do not have the required role to perform {route} in Organization {passed_organization_id}. Your role is {user_role} in Organization {passed_organization_id}",
81 type=ProxyErrorTypes.auth_error.value,
82 param="user_role",
83 code=status.HTTP_401_UNAUTHORIZED,
84 )
85 elif route == "/team/new": 85 ↛ 87line 85 didn't jump to line 87 because the condition on line 85 was never true
86 # if user is part of multiple teams, then they need to specify the organization_id
87 (
88 _user_organizations,
89 _user_organization_role_mapping,
90 ) = get_user_organization_info(user_object)
91 if user_object.organization_memberships is not None and len(user_object.organization_memberships) > 0:
92 if passed_organization_id is None:
93 raise ProxyException(
94 message=f"Passed organization_id is None, please specify the organization_id in your request. You are part of multiple organizations: {_user_organizations}",
95 type=ProxyErrorTypes.auth_error.value,
96 param="organization_id",
97 code=status.HTTP_401_UNAUTHORIZED,
98 )
100 _user_role_in_passed_org: Final = _user_organization_role_mapping.get(passed_organization_id)
101 if _user_role_in_passed_org != LitellmUserRoles.ORG_ADMIN.value:
102 raise ProxyException(
103 message=f"You do not have the required role to call {route}. Your role is {_user_role_in_passed_org} in Organization {passed_organization_id}",
104 type=ProxyErrorTypes.auth_error.value,
105 param="user_role",
106 code=status.HTTP_401_UNAUTHORIZED,
107 )
110def get_user_organization_info(
111 user_object: LiteLLM_UserTable,
112) -> tuple[list[str], dict[str, LitellmUserRoles | None]]:
113 """
114 Helper function to extract user organization information.
116 Args:
117 user_object (LiteLLM_UserTable): The user object containing organization memberships.
119 Returns:
120 Tuple[List[str], Dict[str, Optional[LitellmUserRoles]]]: A tuple containing:
121 - List of organization IDs the user is a member of
122 - Dictionary mapping organization IDs to user roles
123 """
124 _user_organizations: Final[list[str]] = []
125 _user_organization_role_mapping: Final[dict[str, LitellmUserRoles | None]] = {}
127 if user_object.organization_memberships is not None:
128 for _membership in user_object.organization_memberships:
129 if _membership.organization_id is not None:
130 _user_organizations.append(_membership.organization_id)
131 _user_organization_role_mapping[_membership.organization_id] = _membership.user_role
133 return _user_organizations, _user_organization_role_mapping
136def _user_is_org_admin(
137 request_data: dict,
138 user_object: LiteLLM_UserTable | None = None,
139) -> bool:
140 """
141 Helper function to check if user is an org admin for all of the passed organizations.
143 Checks both:
144 - `organization_id` (singular string) — legacy callers
145 - `organizations` (list of strings) — used by /user/new
146 """
147 if user_object is None:
148 return False
150 if user_object.organization_memberships is None:
151 return False
153 # Collect candidate org IDs from both fields
154 candidate_org_ids: Final[list[str]] = []
155 singular: Final = request_data.get("organization_id", None)
156 if singular is not None:
157 candidate_org_ids.append(singular)
158 orgs_list: Final = request_data.get("organizations", None)
159 if isinstance(orgs_list, list):
160 candidate_org_ids.extend(orgs_list)
162 if not candidate_org_ids:
163 return False
165 # Build set of orgs where user is admin
166 admin_org_ids: Final = {
167 _membership.organization_id
168 for _membership in user_object.organization_memberships
169 if _membership.user_role == LitellmUserRoles.ORG_ADMIN.value and _membership.organization_id is not None
170 }
172 # User must be admin of ALL requested orgs, not just any one
173 return all(org_id in admin_org_ids for org_id in candidate_org_ids)
176TEAM_ORG_CONTEXT_ROUTES: Final = frozenset({"/team/update"})
177# The RESTful update route carries the team id in the path. Match on the route
178# template so the sibling /team/<verb> routes (which share the single-segment
179# shape) are not mistaken for it and don't trigger a team lookup.
180PATCH_TEAM_ROUTE_TEMPLATE: Final = "/team/{team_id}"
183async def add_team_org_context_to_request_body(
184 route: str,
185 request_body: dict,
186 fetch_team_org_id: Callable[[str], Awaitable[str | None]],
187 route_template: str | None = None,
188) -> dict:
189 """
190 Return a copy of request_body with organization_id resolved from the target
191 team when the route identifies the team by team_id and the caller did not
192 pass organization_id. This lets an org admin of the team's own org reach the
193 org-scoped branch of the route gate (which keys off organization_id) without
194 the client having to send it. Returns request_body unchanged when it does
195 not apply, so callers that already pass organization_id and non-team routes
196 are untouched.
198 The team_id is taken from the body for TEAM_ORG_CONTEXT_ROUTES, or from the
199 last path segment when ``route_template`` is the ``/team/{team_id}`` route.
200 """
201 if request_body.get("organization_id"):
202 return request_body
204 if route in TEAM_ORG_CONTEXT_ROUTES:
205 team_id: str | None = request_body.get("team_id")
206 elif route_template == PATCH_TEAM_ROUTE_TEMPLATE:
207 team_id = route.rsplit("/", 1)[-1]
208 else:
209 return request_body
211 if not isinstance(team_id, str) or not team_id:
212 return request_body
213 org_id: Final = await fetch_team_org_id(team_id)
214 if not org_id: 214 ↛ 216line 214 didn't jump to line 216 because the condition on line 214 was always true
215 return request_body
216 return {**request_body, "organization_id": org_id}