Spaces:
Build error
Build error
File size: 9,549 Bytes
bf4f454 | 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 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 | from typing import Optional, Set, Union, List, Dict, Any
from open_webui.models.users import Users, UserModel
from open_webui.models.groups import Groups
from open_webui.config import DEFAULT_USER_PERMISSIONS
import json
def fill_missing_permissions(
permissions: Dict[str, Any], default_permissions: Dict[str, Any]
) -> Dict[str, Any]:
"""
Recursively fills in missing properties in the permissions dictionary
using the default permissions as a template.
"""
for key, value in default_permissions.items():
if key not in permissions:
permissions[key] = value
elif isinstance(value, dict) and isinstance(
permissions[key], dict
): # Both are nested dictionaries
permissions[key] = fill_missing_permissions(permissions[key], value)
return permissions
def get_permissions(
user_id: str,
default_permissions: Dict[str, Any],
db: Optional[Any] = None,
) -> Dict[str, Any]:
"""
Get all permissions for a user by combining the permissions of all groups the user is a member of.
If a permission is defined in multiple groups, the most permissive value is used (True > False).
Permissions are nested in a dict with the permission key as the key and a boolean as the value.
"""
def combine_permissions(
permissions: Dict[str, Any], group_permissions: Dict[str, Any]
) -> Dict[str, Any]:
"""Combine permissions from multiple groups by taking the most permissive value."""
for key, value in group_permissions.items():
if isinstance(value, dict):
if key not in permissions:
permissions[key] = {}
permissions[key] = combine_permissions(permissions[key], value)
else:
if key not in permissions:
permissions[key] = value
else:
permissions[key] = (
permissions[key] or value
) # Use the most permissive value (True > False)
return permissions
user_groups = Groups.get_groups_by_member_id(user_id, db=db)
# Deep copy default permissions to avoid modifying the original dict
permissions = json.loads(json.dumps(default_permissions))
# Combine permissions from all user groups
for group in user_groups:
permissions = combine_permissions(permissions, group.permissions or {})
# Ensure all fields from default_permissions are present and filled in
permissions = fill_missing_permissions(permissions, default_permissions)
return permissions
def has_permission(
user_id: str,
permission_key: str,
default_permissions: Dict[str, Any] = {},
db: Optional[Any] = None,
) -> bool:
"""
Check if a user has a specific permission by checking the group permissions
and fall back to default permissions if not found in any group.
Permission keys can be hierarchical and separated by dots ('.').
"""
def get_permission(permissions: Dict[str, Any], keys: List[str]) -> bool:
"""Traverse permissions dict using a list of keys (from dot-split permission_key)."""
for key in keys:
if key not in permissions:
return False # If any part of the hierarchy is missing, deny access
permissions = permissions[key] # Traverse one level deeper
return bool(permissions) # Return the boolean at the final level
permission_hierarchy = permission_key.split(".")
# Retrieve user group permissions
user_groups = Groups.get_groups_by_member_id(user_id, db=db)
for group in user_groups:
if get_permission(group.permissions or {}, permission_hierarchy):
return True
# Check default permissions afterward if the group permissions don't allow it
default_permissions = fill_missing_permissions(
default_permissions, DEFAULT_USER_PERMISSIONS
)
return get_permission(default_permissions, permission_hierarchy)
def has_access(
user_id: str,
permission: str = "read",
access_grants: Optional[list] = None,
user_group_ids: Optional[Set[str]] = None,
db: Optional[Any] = None,
) -> bool:
"""
Check if a user has the specified permission using an in-memory access_grants list.
Used for config-driven resources (arena models, tool servers) that store
access control as JSON in PersistentConfig rather than in the access_grant DB table.
Semantics:
- None or [] → private (owner-only, deny all)
- [{"principal_type": "user", "principal_id": "*", "permission": "read"}] → public read
- Specific grants → check user/group membership
"""
if not access_grants:
return False
if user_group_ids is None:
user_groups = Groups.get_groups_by_member_id(user_id, db=db)
user_group_ids = {group.id for group in user_groups}
for grant in access_grants:
if not isinstance(grant, dict):
continue
if grant.get("permission") != permission:
continue
principal_type = grant.get("principal_type")
principal_id = grant.get("principal_id")
if principal_type == "user" and (
principal_id == "*" or principal_id == user_id
):
return True
if (
principal_type == "group"
and user_group_ids
and principal_id in user_group_ids
):
return True
return False
def has_connection_access(
user: UserModel,
connection: dict,
user_group_ids: Optional[Set[str]] = None,
) -> bool:
"""
Check if a user can access a server connection (tool server, terminal, etc.)
based on ``config.access_grants`` within the connection dict.
- Admin with BYPASS_ADMIN_ACCESS_CONTROL → always allowed
- Missing, None, or empty access_grants → private, admin-only
- access_grants has entries → delegates to ``has_access``
"""
from open_webui.config import BYPASS_ADMIN_ACCESS_CONTROL
if user.role == "admin" and BYPASS_ADMIN_ACCESS_CONTROL:
return True
if user_group_ids is None:
user_group_ids = {group.id for group in Groups.get_groups_by_member_id(user.id)}
access_grants = (connection.get("config") or {}).get("access_grants", [])
return has_access(user.id, "read", access_grants, user_group_ids)
def migrate_access_control(
data: dict, ac_key: str = "access_control", grants_key: str = "access_grants"
) -> None:
"""
Auto-migrate a config dict in-place from legacy access_control dict to access_grants list.
If `grants_key` already exists, does nothing.
If `ac_key` exists (old format), converts it and stores as `grants_key`, then removes `ac_key`.
"""
if grants_key in data:
return
access_control = data.get(ac_key)
if access_control is None and ac_key not in data:
return
grants: List[Dict[str, str]] = []
if access_control and isinstance(access_control, dict):
for perm in ["read", "write"]:
perm_data = access_control.get(perm, {})
if not perm_data:
continue
for group_id in perm_data.get("group_ids", []):
grants.append(
{
"principal_type": "group",
"principal_id": group_id,
"permission": perm,
}
)
for uid in perm_data.get("user_ids", []):
grants.append(
{
"principal_type": "user",
"principal_id": uid,
"permission": perm,
}
)
data[grants_key] = grants
data.pop(ac_key, None)
from open_webui.models.access_grants import (
has_public_read_access_grant,
has_user_access_grant,
strip_user_access_grants,
)
def filter_allowed_access_grants(
default_permissions: Dict[str, Any],
user_id: str,
user_role: str,
access_grants: list,
public_permission_key: str,
db: Optional[Any] = None,
) -> list:
"""
Checks if the user has the required permissions to grant access to a resource.
Returns the filtered list of access grants if permissions are missing.
"""
if user_role == "admin" or not access_grants:
return access_grants
# Check if user can share publicly
if has_public_read_access_grant(access_grants) and not has_permission(
user_id,
public_permission_key,
default_permissions,
db=db,
):
access_grants = [
grant
for grant in access_grants
if not (
(
grant.get("principal_type")
if isinstance(grant, dict)
else getattr(grant, "principal_type", None)
)
== "user"
and (
grant.get("principal_id")
if isinstance(grant, dict)
else getattr(grant, "principal_id", None)
)
== "*"
)
]
# Strip individual user sharing if user lacks permission
if has_user_access_grant(access_grants) and not has_permission(
user_id,
"access_grants.allow_users",
default_permissions,
db=db,
):
access_grants = strip_user_access_grants(access_grants)
return access_grants
|