Coverage for open_webui/utils/access_control/files.py: 13%
80 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 05:07 +0000
1import logging
3from open_webui.models.access_grants import AccessGrants
4from open_webui.models.channels import Channels
5from open_webui.models.chats import Chats
6from open_webui.models.files import Files
7from open_webui.models.folders import FolderModel
8from open_webui.models.groups import Groups
9from open_webui.models.knowledge import Knowledges
10from open_webui.models.models import Models
11from open_webui.models.users import UserModel, Users
12from sqlalchemy.ext.asyncio import AsyncSession
14log = logging.getLogger(__name__)
16FOLDER_FILE_TYPES = {'file', 'collection', 'note'}
19async def has_access_to_file(
20 file_id: str | None,
21 access_type: str,
22 user: UserModel,
23 db: AsyncSession | None = None,
24 user_group_ids: set[str] | None = None,
25) -> bool:
26 """
27 Check if a user has the specified access to a file through any of:
28 - Knowledge bases (ownership or access grants)
29 - Shared workspace models that attach the file directly
30 - Channels the user is a member of
31 - Shared chats
33 NOTE: This does NOT check direct file ownership — callers should check
34 file.user_id == user.id separately before calling this.
35 """
36 file = await Files.get_file_by_id(file_id, db=db)
37 log.debug('Checking if user has %s access to file', access_type)
38 if not file:
39 return False
41 # Direct ownership
42 if file.user_id == user.id:
43 return True
45 # Check if the file is associated with any knowledge bases the user has access to.
46 # An object (knowledge base or workspace model) confers write/delete on a file only when
47 # the object's OWNER owns that file; otherwise a read-only file laundered into an object
48 # the user controls would gain write/delete on it (CWE-863). Read access is unaffected.
49 knowledge_bases = await Knowledges.get_knowledges_by_file_id(file_id, db=db)
50 if user_group_ids is None:
51 user_group_ids = {group.id for group in await Groups.get_groups_by_member_id(user.id, db=db)}
52 for knowledge_base in knowledge_bases:
53 if (
54 knowledge_base.user_id == user.id
55 or await AccessGrants.has_access(
56 user_id=user.id,
57 resource_type='knowledge',
58 resource_id=knowledge_base.id,
59 permission=access_type,
60 user_group_ids=user_group_ids,
61 db=db,
62 )
63 ) and (access_type == 'read' or knowledge_base.user_id == file.user_id):
64 return True
66 # Check if the file is associated with any channels the user has access to
67 channels = await Channels.get_channels_by_file_id_and_user_id(file_id, user.id, db=db)
68 if access_type == 'read' and channels:
69 return True
71 # Check if the file is associated with any chats the user has access to
72 shared_chat_ids = await Chats.get_shared_chat_ids_by_file_id(file_id, db=db)
73 if access_type == 'read' and shared_chat_ids:
74 accessible_ids = await AccessGrants.get_accessible_resource_ids(
75 user_id=user.id,
76 resource_type='shared_chat',
77 resource_ids=shared_chat_ids,
78 permission='read',
79 user_group_ids=user_group_ids,
80 db=db,
81 )
82 if accessible_ids:
83 return True
85 # Check if the file is directly attached to a shared workspace model (per the ownership
86 # note above, model write is conferred only for files the model owner owns).
87 model_owners = await Models.get_model_owner_ids_by_file_id(file.id, db=db, include_background=access_type == 'read')
88 if access_type != 'read':
89 model_owners = {model_id: owner_id for model_id, owner_id in model_owners.items() if owner_id == file.user_id}
90 if user.id in model_owners.values():
91 return True
93 return bool(
94 await AccessGrants.get_accessible_resource_ids(
95 user_id=user.id,
96 resource_type='model',
97 resource_ids=list(model_owners),
98 permission=access_type,
99 user_group_ids=user_group_ids,
100 db=db,
101 )
102 )
105async def get_accessible_folder_files(
106 entries: list[dict] | None,
107 user: UserModel,
108 db: AsyncSession | None = None,
109 user_group_ids: set[str] | None = None,
110) -> list[dict]:
111 """Filter folder.data['files'] entries to those the caller can read.
113 Entries carry a 'type' ('file', 'collection' or 'note') and 'id'. Entries of any other
114 shape are dropped because they cannot be access-checked.
115 """
116 if not isinstance(entries, list):
117 return []
118 entries = [
119 entry
120 for entry in entries
121 if isinstance(entry, dict) and entry.get('type') in FOLDER_FILE_TYPES and entry.get('id')
122 ]
123 if user.role == 'admin':
124 return entries
126 if user_group_ids is None:
127 user_group_ids = {group.id for group in await Groups.get_groups_by_member_id(user.id, db=db)}
129 accessible: list[dict] = []
130 for entry in entries:
131 entry_type = entry.get('type')
132 entry_id = entry.get('id')
133 if entry_type == 'file':
134 if await has_access_to_file(entry_id, 'read', user, db=db, user_group_ids=user_group_ids):
135 accessible.append(entry)
136 elif entry_type == 'collection':
137 if await Knowledges.check_access_by_user_id(
138 entry_id, user.id, 'read', db=db, user_group_ids=user_group_ids
139 ):
140 accessible.append(entry)
141 elif entry_type == 'note':
142 # Owner has no self-grant (notes are private by default), so check ownership too.
143 from open_webui.models.notes import Notes
145 note = await Notes.get_note_by_id(entry_id, db=db)
146 if note and (
147 note.user_id == user.id
148 or await AccessGrants.has_access(
149 user_id=user.id,
150 resource_type='note',
151 resource_id=entry_id,
152 permission='read',
153 user_group_ids=user_group_ids,
154 db=db,
155 )
156 ):
157 accessible.append(entry)
158 return accessible
161async def can_read_all_folder_files(
162 entries: list[dict] | None,
163 user: UserModel,
164 db: AsyncSession | None = None,
165) -> bool:
166 if entries is None:
167 return True
168 if not isinstance(entries, list):
169 return False
170 if not entries:
171 return True
173 return len(await get_accessible_folder_files(entries, user, db=db)) == len(entries)
176async def get_owner_accessible_folder_files(folder: FolderModel, db: AsyncSession | None = None) -> list[dict]:
177 """Return the folder entries its owner can still delegate."""
178 files = (folder.data or {}).get('files') or []
179 if not files:
180 return []
182 owner = await Users.get_user_by_id(folder.user_id, db=db)
183 if not owner:
184 return []
186 return await get_accessible_folder_files(files, owner, db=db)