fix auth check for team member permissions

This commit is contained in:
Ishaan Jaff 2025-04-12 18:19:02 -07:00
parent 2bf5fd1988
commit f3f6140b55
2 changed files with 79 additions and 19 deletions

View file

@ -458,6 +458,8 @@ class LiteLLMRoutes(enum.Enum):
self_managed_routes = [ self_managed_routes = [
"/team/member_add", "/team/member_add",
"/team/member_delete", "/team/member_delete",
"/team/permissions_list",
"/team/permissions_update",
"/model/new", "/model/new",
"/model/update", "/model/update",
"/model/delete", "/model/delete",

View file

@ -1952,30 +1952,62 @@ async def team_member_permissions(
""" """
Get the team member permissions for a team Get the team member permissions for a team
""" """
from litellm.proxy.proxy_server import prisma_client from litellm.proxy.proxy_server import (
prisma_client,
proxy_logging_obj,
user_api_key_cache,
)
if prisma_client is None: if prisma_client is None:
raise HTTPException(status_code=500, detail={"error": "No db connected"}) raise HTTPException(status_code=500, detail={"error": "No db connected"})
team_row = await prisma_client.db.litellm_teamtable.find_unique( ## CHECK IF USER IS PROXY ADMIN OR TEAM ADMIN
where={"team_id": team_id} existing_team_row = await get_team_object(
team_id=team_id,
prisma_client=prisma_client,
user_api_key_cache=user_api_key_cache,
parent_otel_span=None,
proxy_logging_obj=proxy_logging_obj,
check_cache_only=False,
check_db_only=True,
) )
if existing_team_row is None:
if team_row is None:
raise HTTPException( raise HTTPException(
status_code=404, status_code=404,
detail={"error": f"Team not found, passed team_id={team_id}"}, detail={"error": f"Team not found for team_id={team_id}"},
) )
team_obj = LiteLLM_TeamTable(**team_row.model_dump()) complete_team_data = LiteLLM_TeamTable(**existing_team_row.model_dump())
if team_obj.team_member_permissions is None:
team_obj.team_member_permissions = ( if (
hasattr(user_api_key_dict, "user_role")
and user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN.value
and not _is_user_team_admin(
user_api_key_dict=user_api_key_dict, team_obj=complete_team_data
)
and not _is_available_team(
team_id=complete_team_data.team_id,
user_api_key_dict=user_api_key_dict,
)
):
raise HTTPException(
status_code=403,
detail={
"error": "Call not allowed. User not proxy admin OR team admin. route={}, team_id={}".format(
"/team/member_add",
complete_team_data.team_id,
)
},
)
if existing_team_row.team_member_permissions is None:
existing_team_row.team_member_permissions = (
TeamMemberPermissionChecks.default_team_member_permissions() TeamMemberPermissionChecks.default_team_member_permissions()
) )
return GetTeamMemberPermissionsResponse( return GetTeamMemberPermissionsResponse(
team_id=team_id, team_id=team_id,
team_member_permissions=team_obj.team_member_permissions, team_member_permissions=existing_team_row.team_member_permissions,
all_available_permissions=TeamMemberPermissionChecks.get_all_available_team_member_permissions(), all_available_permissions=TeamMemberPermissionChecks.get_all_available_team_member_permissions(),
) )
@ -1993,27 +2025,53 @@ async def update_team_member_permissions(
""" """
Update the team member permissions for a team Update the team member permissions for a team
""" """
from litellm.proxy.proxy_server import prisma_client from litellm.proxy.proxy_server import (
prisma_client,
proxy_logging_obj,
user_api_key_cache,
)
if prisma_client is None: if prisma_client is None:
raise HTTPException(status_code=500, detail={"error": "No db connected"}) raise HTTPException(status_code=500, detail={"error": "No db connected"})
team_row = await prisma_client.db.litellm_teamtable.find_unique( ## CHECK IF USER IS PROXY ADMIN OR TEAM ADMIN
where={"team_id": data.team_id} existing_team_row = await get_team_object(
team_id=data.team_id,
prisma_client=prisma_client,
user_api_key_cache=user_api_key_cache,
parent_otel_span=None,
proxy_logging_obj=proxy_logging_obj,
check_cache_only=False,
check_db_only=True,
) )
if existing_team_row is None:
if team_row is None:
raise HTTPException( raise HTTPException(
status_code=404, status_code=404,
detail={"error": f"Team not found, passed team_id={data.team_id}"}, detail={"error": f"Team not found for team_id={data.team_id}"},
) )
if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN.value: complete_team_data = LiteLLM_TeamTable(**existing_team_row.model_dump())
if (
hasattr(user_api_key_dict, "user_role")
and user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN.value
and not _is_user_team_admin(
user_api_key_dict=user_api_key_dict, team_obj=complete_team_data
)
and not _is_available_team(
team_id=complete_team_data.team_id,
user_api_key_dict=user_api_key_dict,
)
):
raise HTTPException( raise HTTPException(
status_code=403, status_code=403,
detail={"error": "Only proxy admin can update team member permissions"}, detail={
"error": "Call not allowed. User not proxy admin OR team admin. route={}, team_id={}".format(
"/team/member_add",
complete_team_data.team_id,
)
},
) )
# Update the team member permissions # Update the team member permissions
updated_team = await prisma_client.db.litellm_teamtable.update( updated_team = await prisma_client.db.litellm_teamtable.update(
where={"team_id": data.team_id}, where={"team_id": data.team_id},