"""Service for managing organization members.""" from uuid import UUID from server.constants import ROLE_ADMIN, ROLE_OWNER from server.routes.org_models import ( CannotModifySelfError, InsufficientPermissionError, InvalidRoleError, LastOwnerError, MemberUpdateError, MeResponse, OrgMemberNotFoundError, OrgMemberPage, OrgMemberResponse, OrgMemberUpdate, RoleNotFoundError, ) from storage.lite_llm_manager import LiteLlmManager from storage.org_member_store import OrgMemberStore from storage.role_store import RoleStore from storage.user_store import UserStore from openhands.app_server.utils.logger import openhands_logger as logger class OrgMemberService: """Service for organization member operations.""" @staticmethod async def get_me(org_id: UUID, user_id: UUID) -> MeResponse: """Get the current user's membership record for an organization. Retrieves the authenticated user's role, status, email, and LLM override fields (with masked API keys) within the specified organization. Args: org_id: Organization ID (UUID) user_id: User ID (UUID) Returns: MeResponse: The user's membership data with masked API keys Raises: OrgMemberNotFoundError: If user is not a member of the organization RoleNotFoundError: If the role associated with the member is not found """ # Look up the user's membership in this org org_member = await OrgMemberStore.get_org_member(org_id, user_id) if org_member is None: raise OrgMemberNotFoundError(str(org_id), str(user_id)) # Resolve role name from role_id role = await RoleStore.get_role_by_id(org_member.role_id) if role is None: raise RoleNotFoundError(org_member.role_id) # Get user email user = await UserStore.get_user_by_id(str(user_id)) email = user.email if user and user.email else '' return MeResponse.from_org_member(org_member, role, email) @staticmethod async def get_org_members( org_id: UUID, current_user_id: UUID, page_id: str | None = None, limit: int = 10, email_filter: str | None = None, ) -> tuple[bool, str | None, OrgMemberPage | None]: """Get organization members with authorization check. Args: org_id: Organization UUID. current_user_id: Requesting user's UUID. page_id: Offset encoded as string (e.g., "0", "10", "20"). limit: Items per page (default 10). email_filter: Optional case-insensitive partial email match. Returns: Tuple of (success, error_code, data). If success is True, error_code is None. """ # Verify current user is a member of the organization requester_membership = await OrgMemberStore.get_org_member( org_id, current_user_id ) if not requester_membership: return False, 'not_a_member', None # Parse page_id to get offset (page_id is offset encoded as string) offset = 0 if page_id is not None: try: offset = int(page_id) if offset < 0: return False, 'invalid_page_id', None except ValueError: return False, 'invalid_page_id', None # Call store to get paginated members members, _ = await OrgMemberStore.get_org_members_paginated( org_id=org_id, offset=offset, limit=limit, email_filter=email_filter, ) # Transform data to response format items = [] for member in members: # Access user and role relationships (eagerly loaded) user = member.user role = member.role items.append( OrgMemberResponse( user_id=str(member.user_id), email=user.email if user else None, role_id=member.role_id, role=role.name if role else '', role_rank=role.rank if role else 0, status=member.status, ) ) # Calculate current page (1-indexed) current_page = (offset // limit) + 1 return ( True, None, OrgMemberPage( items=items, current_page=current_page, per_page=limit, ), ) @staticmethod async def get_org_members_count( org_id: UUID, current_user_id: UUID, email_filter: str | None = None, ) -> int: """Get count of organization members with authorization check. Args: org_id: Organization UUID. current_user_id: Requesting user's UUID. email_filter: Optional case-insensitive partial email match. Returns: int: Count of organization members matching the filter. Raises: OrgMemberNotFoundError: If requesting user is not a member of the organization. """ # Verify current user is a member of the organization requester_membership = await OrgMemberStore.get_org_member( org_id, current_user_id ) if not requester_membership: raise OrgMemberNotFoundError(str(org_id), str(current_user_id)) return await OrgMemberStore.get_org_members_count( org_id=org_id, email_filter=email_filter, ) @staticmethod async def remove_org_member( org_id: UUID, target_user_id: UUID, current_user_id: UUID, ) -> tuple[bool, str | None]: """Remove a member from an organization. Returns: Tuple of (success, error_message). If success is True, error_message is None. """ # Get current user's membership in the org requester_membership = await OrgMemberStore.get_org_member( org_id, current_user_id ) if not requester_membership: return False, 'not_a_member' # Check if trying to remove self if str(current_user_id) == str(target_user_id): return False, 'cannot_remove_self' # Get target user's membership target_membership = await OrgMemberStore.get_org_member(org_id, target_user_id) if not target_membership: return False, 'member_not_found' requester_role = await RoleStore.get_role_by_id(requester_membership.role_id) target_role = await RoleStore.get_role_by_id(target_membership.role_id) if not requester_role or not target_role: return False, 'role_not_found' # Check permission based on roles if not OrgMemberService._can_remove_member( requester_role.name, target_role.name ): return False, 'insufficient_permission' # Check if removing the last owner if target_role.name == ROLE_OWNER: if await OrgMemberService._is_last_owner(org_id, target_user_id): return False, 'cannot_remove_last_owner' # Perform the removal success = await OrgMemberStore.remove_user_from_org(org_id, target_user_id) if not success: return False, 'removal_failed' # Update user's current_org_id if it points to the org they were removed from user = await UserStore.get_user_by_id(str(target_user_id)) if user and user.current_org_id == org_id: # Set current_org_id to personal workspace (org.id == user.id) await UserStore.update_current_org(str(target_user_id), target_user_id) # If database removal succeeded, also remove from LiteLLM team try: await LiteLlmManager.remove_user_from_team(str(target_user_id), str(org_id)) logger.info( 'Successfully removed user from LiteLLM team', extra={ 'user_id': str(target_user_id), 'org_id': str(org_id), }, ) except Exception as e: # Log but don't fail the operation - database removal already succeeded # LiteLLM state will be eventually consistent logger.warning( 'Failed to remove user from LiteLLM team', extra={ 'user_id': str(target_user_id), 'org_id': str(org_id), 'error': str(e), }, ) return True, None @staticmethod async def update_org_member( org_id: UUID, target_user_id: UUID, current_user_id: UUID, update_data: OrgMemberUpdate, ) -> OrgMemberResponse: """Update a member's role in an organization. Permission rules: - Owners can modify anyone (including other owners), can set any role - Admins can modify other admins and users - Admins can only set admin or user roles (not owner) Args: org_id: Organization ID target_user_id: User ID of the member to update current_user_id: User ID of the requester update_data: Update data containing fields to modify Returns: OrgMemberResponse: The updated member data Raises: OrgMemberNotFoundError: If requester or target is not a member CannotModifySelfError: If trying to modify self RoleNotFoundError: If role configuration is invalid InvalidRoleError: If new_role_name is not a valid role InsufficientPermissionError: If requester lacks permission LastOwnerError: If trying to demote the last owner MemberUpdateError: If update operation fails """ new_role_name = update_data.role # Get current user's membership in the org requester_membership = await OrgMemberStore.get_org_member( org_id, current_user_id ) if not requester_membership: raise OrgMemberNotFoundError(str(org_id), str(current_user_id)) # Check if trying to modify self if str(current_user_id) == str(target_user_id): raise CannotModifySelfError('modify') # Get target user's membership target_membership = await OrgMemberStore.get_org_member(org_id, target_user_id) if not target_membership: raise OrgMemberNotFoundError(str(org_id), str(target_user_id)) # Get roles requester_role = await RoleStore.get_role_by_id(requester_membership.role_id) target_role = await RoleStore.get_role_by_id(target_membership.role_id) if not requester_role: raise RoleNotFoundError(requester_membership.role_id) if not target_role: raise RoleNotFoundError(target_membership.role_id) # If no role change requested, return current state if new_role_name is None: user = await UserStore.get_user_by_id(str(target_user_id)) return OrgMemberResponse( user_id=str(target_membership.user_id), email=user.email if user else None, role_id=target_membership.role_id, role=target_role.name, role_rank=target_role.rank, status=target_membership.status, ) # Validate new role exists new_role = await RoleStore.get_role_by_name(new_role_name.lower()) if not new_role: raise InvalidRoleError(new_role_name) # Check permission to modify target if not OrgMemberService._can_update_member_role( requester_role.name, target_role.name, new_role.name ): raise InsufficientPermissionError( 'You do not have permission to modify this member' ) # Check if demoting the last owner if ( target_role.name == ROLE_OWNER and new_role.name != ROLE_OWNER and await OrgMemberService._is_last_owner(org_id, target_user_id) ): raise LastOwnerError('demote') # Perform the update updated_member = await OrgMemberStore.update_user_role_in_org( org_id, target_user_id, new_role.id ) if not updated_member: raise MemberUpdateError('Failed to update member') # Get user email for response user = await UserStore.get_user_by_id(str(target_user_id)) return OrgMemberResponse( user_id=str(updated_member.user_id), email=user.email if user else None, role_id=updated_member.role_id, role=new_role.name, role_rank=new_role.rank, status=updated_member.status, ) @staticmethod def _can_update_member_role( requester_role_name: str, target_role_name: str, new_role_name: str ) -> bool: """Check if requester can change target's role to new_role. Permission rules: - Owners can modify anyone (including other owners), can set any role - Admins can modify other admins and users - Admins can only set admin or user roles (not owner) """ is_requester_owner = requester_role_name == ROLE_OWNER is_requester_admin = requester_role_name == ROLE_ADMIN is_target_owner = target_role_name == ROLE_OWNER is_new_role_owner = new_role_name == ROLE_OWNER if is_requester_owner: # Owners can modify anyone (including other owners) return True elif is_requester_admin: # Admins cannot modify owners if is_target_owner: return False # Admins can only set admin or user roles (not owner) return not is_new_role_owner return False @staticmethod def _can_remove_member(requester_role_name: str, target_role_name: str) -> bool: """Check if requester can remove target based on roles.""" if requester_role_name == ROLE_OWNER: return True elif requester_role_name == ROLE_ADMIN: # Admins can remove admins and members (not owners) return target_role_name != ROLE_OWNER return False @staticmethod async def _is_last_owner(org_id: UUID, user_id: UUID) -> bool: """Check if user is the last owner of the organization.""" members = await OrgMemberStore.get_org_members(org_id) owners = [] for m in members: # Use role_id (column) instead of role (relationship) to avoid DetachedInstanceError role = await RoleStore.get_role_by_id(m.role_id) if role and role.name == ROLE_OWNER: owners.append(m) return len(owners) == 1 and str(owners[0].user_id) == str(user_id)