eliott/open-webui
1
0
Fork 0
mirror of https://github.com/open-webui/open-webui.git synced 2026-05-01 03:47:25 +02:00
Code Issues Packages Projects Releases Wiki Activity

feat: user list in channels

Browse Source
This commit is contained in:
Timothy Jaeryang Baek
2025-11-25 04:38:07 -05:00
parent baa1e07aec
commit c0e1203538
10 changed files with 521 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
backend/open_webui/utils/access_control.py
View File

@@ -105,6 +105,22 @@ def has_permission(
return get_permission(default_permissions, permission_hierarchy)
def get_permitted_group_and_user_ids(
type: str = "write", access_control: Optional[dict] = None
) -> Union[Dict[str, List[str]], None]:
if access_control is None:
return None
permission_access = access_control.get(type, {})
permitted_group_ids = permission_access.get("group_ids", [])
permitted_user_ids = permission_access.get("user_ids", [])
return {
"group_ids": permitted_group_ids,
"user_ids": permitted_user_ids,
}
def has_access(
user_id: str,
type: str = "write",
@@ -122,9 +138,12 @@ def has_access(
user_groups = Groups.get_groups_by_member_id(user_id)
user_group_ids = {group.id for group in user_groups}
permission_access = access_control.get(type, {})
permitted_group_ids = permission_access.get("group_ids", [])
permitted_user_ids = permission_access.get("user_ids", [])
permitted_ids = get_permitted_group_and_user_ids(type, access_control)
if permitted_ids is None:
return False
permitted_group_ids = permitted_ids.get("group_ids", [])
permitted_user_ids = permitted_ids.get("user_ids", [])
return user_id in permitted_user_ids or any(
group_id in permitted_group_ids for group_id in user_group_ids
@@ -136,18 +155,20 @@ def get_users_with_access(
type: str = "write", access_control: Optional[dict] = None
) -> list[UserModel]:
if access_control is None:
result = Users.get_users()
result = Users.get_users(filter={"roles": ["!pending"]})
return result.get("users", [])
permission_access = access_control.get(type, {})
permitted_group_ids = permission_access.get("group_ids", [])
permitted_user_ids = permission_access.get("user_ids", [])
permitted_ids = get_permitted_group_and_user_ids(type, access_control)
if permitted_ids is None:
return []
permitted_group_ids = permitted_ids.get("group_ids", [])
permitted_user_ids = permitted_ids.get("user_ids", [])
user_ids_with_access = set(permitted_user_ids)
for group_id in permitted_group_ids:
group_user_ids = Groups.get_group_user_ids_by_id(group_id)
if group_user_ids:
user_ids_with_access.update(group_user_ids)
group_user_ids_map = Groups.get_group_user_ids_by_ids(permitted_group_ids)
for user_ids in group_user_ids_map.values():
user_ids_with_access.update(user_ids)
return Users.get_users_by_user_ids(list(user_ids_with_access))