Spaces:
Runtime error
Runtime error
| import logging | |
| from rexpro_ai.models.access_grants import AccessGrants | |
| from rexpro_ai.models.channels import Channels | |
| from rexpro_ai.models.chats import Chats | |
| from rexpro_ai.models.files import Files | |
| from rexpro_ai.models.groups import Groups | |
| from rexpro_ai.models.knowledge import Knowledges | |
| from rexpro_ai.models.models import Models | |
| from rexpro_ai.models.users import UserModel | |
| from sqlalchemy.ext.asyncio import AsyncSession | |
| log = logging.getLogger(__name__) | |
| async def has_access_to_file( | |
| file_id: str | None, | |
| access_type: str, | |
| user: UserModel, | |
| db: AsyncSession | None = None, | |
| ) -> bool: | |
| """ | |
| Check if a user has the specified access to a file through any of: | |
| - Knowledge bases (ownership or access grants) | |
| - Shared workspace models that attach the file directly | |
| - Channels the user is a member of | |
| - Shared chats | |
| NOTE: This does NOT check direct file ownership — callers should check | |
| file.user_id == user.id separately before calling this. | |
| """ | |
| file = await Files.get_file_by_id(file_id, db=db) | |
| log.debug(f'Checking if user has {access_type} access to file') | |
| if not file: | |
| return False | |
| # Direct ownership | |
| if file.user_id == user.id: | |
| return True | |
| # Check if the file is associated with any knowledge bases the user has access to | |
| knowledge_bases = await Knowledges.get_knowledges_by_file_id(file_id, db=db) | |
| user_group_ids = {group.id for group in await Groups.get_groups_by_member_id(user.id, db=db)} | |
| for knowledge_base in knowledge_bases: | |
| if knowledge_base.user_id == user.id or await AccessGrants.has_access( | |
| user_id=user.id, | |
| resource_type='knowledge', | |
| resource_id=knowledge_base.id, | |
| permission=access_type, | |
| user_group_ids=user_group_ids, | |
| db=db, | |
| ): | |
| return True | |
| knowledge_base_id = file.meta.get('collection_name') if file.meta else None | |
| if knowledge_base_id: | |
| knowledge_bases = await Knowledges.get_knowledge_bases_by_user_id(user.id, access_type, db=db) | |
| for knowledge_base in knowledge_bases: | |
| if knowledge_base.id == knowledge_base_id: | |
| return True | |
| # Check if the file is associated with any channels the user has access to | |
| channels = await Channels.get_channels_by_file_id_and_user_id(file_id, user.id, db=db) | |
| if access_type == 'read' and channels: | |
| return True | |
| # Check if the file is associated with any chats the user has access to | |
| shared_chat_ids = await Chats.get_shared_chat_ids_by_file_id(file_id, db=db) | |
| if access_type == 'read' and shared_chat_ids: | |
| accessible_ids = await AccessGrants.get_accessible_resource_ids( | |
| user_id=user.id, | |
| resource_type='shared_chat', | |
| resource_ids=shared_chat_ids, | |
| permission='read', | |
| user_group_ids=user_group_ids, | |
| db=db, | |
| ) | |
| if accessible_ids: | |
| return True | |
| # Check if the file is directly attached to a shared workspace model | |
| for model in await Models.get_models_by_user_id(user.id, permission=access_type, db=db): | |
| knowledge_items = getattr(model.meta, 'knowledge', None) or [] | |
| for item in knowledge_items: | |
| if isinstance(item, dict) and item.get('type') == 'file' and item.get('id') == file.id: | |
| return True | |
| return False | |
| async def get_accessible_folder_files( | |
| entries: list[dict] | None, | |
| user: UserModel, | |
| db: AsyncSession | None = None, | |
| ) -> list[dict]: | |
| """Filter folder.data['files'] entries to those the caller can read. | |
| Each entry is expected to have 'type' ('file' or 'collection') and 'id'. | |
| Admins bypass all checks. Unknown types are kept as-is. | |
| """ | |
| if not entries: | |
| return [] | |
| if user.role == 'admin': | |
| return list(entries) | |
| accessible: list[dict] = [] | |
| for entry in entries: | |
| if not isinstance(entry, dict): | |
| continue | |
| entry_type = entry.get('type') | |
| entry_id = entry.get('id') | |
| if not entry_id: | |
| accessible.append(entry) | |
| continue | |
| if entry_type == 'file': | |
| if await has_access_to_file(entry_id, 'read', user, db=db): | |
| accessible.append(entry) | |
| elif entry_type == 'collection': | |
| if await Knowledges.check_access_by_user_id(entry_id, user.id, 'read', db=db): | |
| accessible.append(entry) | |
| else: | |
| accessible.append(entry) | |
| return accessible | |