Coverage for open_webui/utils/access_control/__init__.py: 24%
140 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 05:07 +0000
1import logging
2from typing import Any
4from open_webui.config import DEFAULT_USER_PERMISSIONS
5from open_webui.models.access_grants import (
6 has_anyone_read_access_grant,
7 has_public_read_access_grant,
8 has_public_write_access_grant,
9 has_user_access_grant,
10 strip_anyone_access_grants,
11 strip_user_access_grants,
12)
13from open_webui.models.groups import Groups
14from open_webui.models.users import UserModel
15from open_webui.utils.json_codec import JSONCodec
16from sqlalchemy.ext.asyncio import AsyncSession
18log = logging.getLogger(__name__)
21def fill_missing_permissions(permissions: dict[str, Any], default_permissions: dict[str, Any]) -> dict[str, Any]:
22 """
23 Recursively fills in missing properties in the permissions dictionary
24 using the default permissions as a template.
25 """
26 for key, value in default_permissions.items():
27 if key not in permissions: 27 ↛ 28line 27 didn't jump to line 28 because the condition on line 27 was never true
28 permissions[key] = value
29 elif isinstance(value, dict) and isinstance(permissions[key], dict): # Both are nested dictionaries
30 permissions[key] = fill_missing_permissions(permissions[key], value)
32 return permissions
35async def get_permissions(
36 user_id: str,
37 default_permissions: dict[str, Any],
38 db: AsyncSession | None = None,
39) -> dict[str, Any]:
40 """
41 Get all permissions for a user by combining the permissions of all groups the user is a member of.
42 If a permission is defined in multiple groups, the most permissive value is used (True > False).
43 Permissions are nested in a dict with the permission key as the key and a boolean as the value.
44 """
46 def combine_permissions(permissions: dict[str, Any], group_permissions: dict[str, Any]) -> dict[str, Any]:
47 """Combine permissions from multiple groups by taking the most permissive value."""
48 for key, value in group_permissions.items():
49 if isinstance(value, dict):
50 if key not in permissions:
51 permissions[key] = {}
52 permissions[key] = combine_permissions(permissions[key], value)
53 else:
54 if key not in permissions:
55 permissions[key] = value
56 else:
57 permissions[key] = permissions[key] or value # Use the most permissive value (True > False)
58 return permissions
60 user_groups = await Groups.get_groups_by_member_id(user_id, db=db)
62 # Deep copy default permissions to avoid modifying the original dict
63 permissions = JSONCodec.loads(JSONCodec.dumps(default_permissions))
65 # Combine permissions from all user groups
66 for group in user_groups: 66 ↛ 67line 66 didn't jump to line 67 because the loop on line 66 never started
67 permissions = combine_permissions(permissions, group.permissions or {})
69 # Ensure all fields from default_permissions are present and filled in
70 permissions = fill_missing_permissions(permissions, default_permissions)
72 return permissions
75async def has_permission(
76 user_id: str,
77 permission_key: str,
78 default_permissions: dict[str, Any] = {},
79 db: AsyncSession | None = None,
80) -> bool:
81 """
82 Check if a user has a specific permission by checking the group permissions
83 and fall back to default permissions if not found in any group.
85 Permission keys can be hierarchical and separated by dots ('.').
86 """
88 def get_permission(permissions: dict[str, Any], keys: list[str]) -> bool:
89 """Traverse permissions dict using a list of keys (from dot-split permission_key)."""
90 for key in keys:
91 if key not in permissions:
92 return False # If any part of the hierarchy is missing, deny access
93 permissions = permissions[key] # Traverse one level deeper
95 return bool(permissions) # Return the boolean at the final level
97 permission_hierarchy = permission_key.split('.')
99 # Retrieve user group permissions
100 user_groups = await Groups.get_groups_by_member_id(user_id, db=db)
102 for group in user_groups:
103 if get_permission(group.permissions or {}, permission_hierarchy):
104 return True
106 # Check default permissions afterward if the group permissions don't allow it
107 default_permissions = fill_missing_permissions(default_permissions, DEFAULT_USER_PERMISSIONS)
108 return get_permission(default_permissions, permission_hierarchy)
111async def has_access(
112 user_id: str,
113 permission: str = 'read',
114 access_grants: list | None = None,
115 user_group_ids: set[str] | None = None,
116 db: AsyncSession | None = None,
117) -> bool:
118 """
119 Check if a user has the specified permission using an in-memory access_grants list.
121 Used for config-driven resources (arena models, tool servers) that store
122 access control as JSON config rather than in the access_grant DB table.
124 Semantics:
125 - None or [] → private (owner-only, deny all)
126 - [{"principal_type": "user", "principal_id": "*", "permission": "read"}] → public read
127 - Specific grants → check user/group membership
128 """
129 if not access_grants:
130 return False
132 if user_group_ids is None:
133 user_groups = await Groups.get_groups_by_member_id(user_id, db=db)
134 user_group_ids = {group.id for group in user_groups}
136 for grant in access_grants:
137 if not isinstance(grant, dict):
138 continue
139 if grant.get('permission') != permission:
140 continue
141 principal_type = grant.get('principal_type')
142 principal_id = grant.get('principal_id')
143 if principal_type == 'user' and (principal_id == '*' or principal_id == user_id):
144 return True
145 if principal_type == 'group' and user_group_ids and principal_id in user_group_ids:
146 return True
148 return False
151async def has_connection_access(
152 user: UserModel,
153 connection: dict,
154 user_group_ids: set[str] | None = None,
155) -> bool:
156 """
157 Check if a user can access a server connection (tool server, terminal, etc.)
158 based on ``config.access_grants`` within the connection dict.
160 - Admin with BYPASS_ADMIN_ACCESS_CONTROL → always allowed
161 - Missing, None, or empty access_grants → private, admin-only
162 - access_grants has entries → delegates to ``has_access``
163 """
164 from open_webui.config import BYPASS_ADMIN_ACCESS_CONTROL
166 if user.role == 'admin' and BYPASS_ADMIN_ACCESS_CONTROL: 166 ↛ 169line 166 didn't jump to line 169 because the condition on line 166 was always true
167 return True
169 access_grants = (connection.get('config') or {}).get('access_grants', [])
170 if not access_grants:
171 # No grants configured → private, admin-only: admins must keep access
172 # to connections only they can configure, even when they do not bypass
173 # access control globally.
174 return user.role == 'admin'
176 if user_group_ids is None:
177 user_group_ids = {group.id for group in await Groups.get_groups_by_member_id(user.id)}
179 return await has_access(user.id, 'read', access_grants, user_group_ids)
182def migrate_access_control(data: dict, ac_key: str = 'access_control', grants_key: str = 'access_grants') -> None:
183 """
184 Auto-migrate a config dict in-place from legacy access_control dict to access_grants list.
186 If `grants_key` already exists, does nothing.
187 If `ac_key` exists (old format), converts it and stores as `grants_key`, then removes `ac_key`.
188 """
189 if grants_key in data:
190 return
192 access_control = data.get(ac_key)
193 if access_control is None and ac_key not in data:
194 return
196 grants: list[dict[str, str]] = []
197 if access_control and isinstance(access_control, dict):
198 for perm in ['read', 'write']:
199 perm_data = access_control.get(perm, {})
200 if not perm_data:
201 continue
202 for group_id in perm_data.get('group_ids', []):
203 grants.append(
204 {
205 'principal_type': 'group',
206 'principal_id': group_id,
207 'permission': perm,
208 }
209 )
210 for uid in perm_data.get('user_ids', []):
211 grants.append(
212 {
213 'principal_type': 'user',
214 'principal_id': uid,
215 'permission': perm,
216 }
217 )
219 data[grants_key] = grants
220 data.pop(ac_key, None)
223async def filter_allowed_access_grants(
224 default_permissions: dict[str, Any],
225 user_id: str,
226 user_role: str,
227 access_grants: list,
228 public_permission_key: str,
229 anyone_permission_key: str | None = None,
230 db: AsyncSession | None = None,
231) -> list:
232 """
233 Checks if the user has the required permissions to grant access to a resource.
234 Returns the filtered list of access grants if permissions are missing.
235 """
236 if not access_grants:
237 return access_grants
239 if has_anyone_read_access_grant(access_grants) and ( 239 ↛ 251line 239 didn't jump to line 251 because the condition on line 239 was never true
240 not anyone_permission_key
241 or (
242 user_role != 'admin'
243 and not await has_permission(
244 user_id,
245 anyone_permission_key,
246 default_permissions,
247 db=db,
248 )
249 )
250 ):
251 access_grants = strip_anyone_access_grants(access_grants)
253 if user_role == 'admin': 253 ↛ 257line 253 didn't jump to line 257 because the condition on line 253 was always true
254 return access_grants
256 # Check if user can share publicly
257 if (
258 has_public_read_access_grant(access_grants) or has_public_write_access_grant(access_grants)
259 ) and not await has_permission(
260 user_id,
261 public_permission_key,
262 default_permissions,
263 db=db,
264 ):
265 access_grants = [
266 grant
267 for grant in access_grants
268 if not (
269 (grant.get('principal_type') if isinstance(grant, dict) else getattr(grant, 'principal_type', None))
270 == 'user'
271 and (grant.get('principal_id') if isinstance(grant, dict) else getattr(grant, 'principal_id', None))
272 == '*'
273 )
274 ]
276 # Strip individual user sharing if user lacks permission
277 if has_user_access_grant(access_grants) and not await has_permission(
278 user_id,
279 'access_grants.allow_users',
280 default_permissions,
281 db=db,
282 ):
283 access_grants = strip_user_access_grants(access_grants)
285 if any(
286 (grant.get('principal_type') if isinstance(grant, dict) else getattr(grant, 'principal_type', None)) == 'group'
287 for grant in access_grants
288 ) and not await has_permission(
289 user_id,
290 'access_grants.allow_groups',
291 default_permissions,
292 db=db,
293 ):
294 access_grants = [
295 grant
296 for grant in access_grants
297 if (grant.get('principal_type') if isinstance(grant, dict) else getattr(grant, 'principal_type', None))
298 != 'group'
299 ]
301 return access_grants
304async def has_base_model_access(
305 user_id: str,
306 model_info,
307 *,
308 user_role: str | None = None,
309 user_group_ids: set[str] | None = None,
310 db=None,
311) -> bool:
312 """
313 Walk the ``base_model_id`` chain and verify the caller has read access
314 at every hop.
316 A base model without a ``model`` table row is admin-only, matching how
317 unregistered models are treated for direct use (``get_filtered_models``
318 hides them from non-admins and ``check_model_access`` rejects them), so
319 a shared preset cannot be used to reach a base model the caller could
320 not use directly. Returns ``False`` the moment any hop denies access.
321 """
322 from open_webui.models.access_grants import AccessGrants
323 from open_webui.models.models import Models
325 base_model_id = getattr(model_info, 'base_model_id', None)
326 seen = {model_info.id}
327 while base_model_id and base_model_id not in seen:
328 seen.add(base_model_id)
329 base_model_info = await Models.get_model_by_id(base_model_id, db=db)
330 if base_model_info is None:
331 if user_role != 'admin':
332 log.warning(
333 'Model access denied: user_id=%r model_id=%r base_model_id=%r reason=base_model_unregistered',
334 user_id,
335 model_info.id,
336 base_model_id,
337 )
338 return user_role == 'admin'
339 if not (
340 user_id == base_model_info.user_id
341 or await AccessGrants.has_access(
342 user_id=user_id,
343 resource_type='model',
344 resource_id=base_model_info.id,
345 permission='read',
346 user_group_ids=user_group_ids,
347 db=db,
348 )
349 ):
350 log.warning(
351 'Model access denied: user_id=%r model_id=%r base_model_id=%r reason=base_model_read_denied',
352 user_id,
353 model_info.id,
354 base_model_id,
355 )
356 return False
357 base_model_id = getattr(base_model_info, 'base_model_id', None)
358 return True
361async def check_model_access(
362 user: UserModel,
363 model_info,
364 bypass_filter: bool = False,
365) -> None:
366 """
367 Enforce per-model read access for the given user.
369 Raises HTTPException(403) if the user is not authorized.
370 Does nothing if bypass_filter is True.
372 Args:
373 user: The authenticated user.
374 model_info: The model record from await Models.get_model_by_id(),
375 or None if the model is not registered.
376 bypass_filter: If True, skip all access checks (used by
377 internal callers and BYPASS_MODEL_ACCESS_CONTROL).
378 """
379 from fastapi import HTTPException
381 if bypass_filter: 381 ↛ 382line 381 didn't jump to line 382 because the condition on line 381 was never true
382 return
384 if model_info: 384 ↛ 386line 384 didn't jump to line 386 because the condition on line 384 was never true
385 # Enforce for every non-admin role (including pending); never fail open.
386 if user.role != 'admin':
387 from open_webui.models.access_grants import AccessGrants
389 user_group_ids = {group.id for group in await Groups.get_groups_by_member_id(user.id)}
390 if not (
391 user.id == model_info.user_id
392 or await AccessGrants.has_access(
393 user_id=user.id,
394 resource_type='model',
395 resource_id=model_info.id,
396 permission='read',
397 user_group_ids=user_group_ids,
398 )
399 ):
400 log.warning(
401 'Model access denied: user_id=%r model_id=%r reason=model_read_denied',
402 user.id,
403 model_info.id,
404 )
405 raise HTTPException(status_code=403, detail='Model not found')
407 # Enforce access on chained base models
408 if not await has_base_model_access(user.id, model_info, user_role=user.role, user_group_ids=user_group_ids):
409 raise HTTPException(status_code=403, detail='Model not found')
410 else:
411 if user.role != 'admin': 411 ↛ 412line 411 didn't jump to line 412 because the condition on line 411 was never true
412 log.warning('Model access denied: user_id=%r reason=model_unregistered', user.id)
413 raise HTTPException(status_code=403, detail='Model not found')