Spaces:
Runtime error
Runtime error
File size: 4,547 Bytes
9e93b10 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 | 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
|