diff --git a/backend/open_webui/routers/channels.py b/backend/open_webui/routers/channels.py index 5c2ab9dcac..83247e24f3 100644 --- a/backend/open_webui/routers/channels.py +++ b/backend/open_webui/routers/channels.py @@ -13,6 +13,7 @@ from pydantic import field_validator from open_webui.socket.main import ( emit_to_users, enter_room_for_users, + leave_rooms_for_users, sio, get_user_ids_from_room, ) @@ -544,6 +545,10 @@ async def update_is_active_member_by_id_and_user_id( raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND) await Channels.update_member_active_status(channel.id, user.id, form_data.is_active, db=db) + if not form_data.is_active: + # Deactivating membership revokes live subscription; evict the user's + # open sessions from the channel room so they stop receiving events. + await leave_rooms_for_users(f'channel:{channel.id}', [user.id]) return True @@ -613,6 +618,9 @@ async def remove_members_by_id( try: deleted = await Channels.remove_members_from_channel(channel.id, form_data.user_ids, db=db) + if form_data.user_ids: + await leave_rooms_for_users(f'channel:{channel.id}', form_data.user_ids) + return deleted except Exception as e: log.exception(e) @@ -651,6 +659,18 @@ async def update_channel_by_id( try: channel = await Channels.update_channel_by_id(id, form_data, db=db) + + # Access grants may have changed. Evict live subscribers who no longer + # belong — checked against the current membership/grant resolution. + room_name = f'channel:{id}' + for participant_user_id in get_user_ids_from_room(room_name): + try: + still_member = await Channels.is_user_channel_member(id, participant_user_id, db=db) + except Exception: + still_member = True # Fail-open on transient errors; next mutation retries. + if not still_member: + await leave_rooms_for_users(room_name, [participant_user_id]) + return ChannelModel(**channel.model_dump()) except Exception as e: log.exception(e) @@ -680,6 +700,12 @@ async def delete_channel_by_id( try: await Channels.delete_channel_by_id(id, db=db) + # Tear down every open subscription to this channel room — no one has + # access after deletion. + try: + await sio.close_room(f'channel:{id}') + except Exception as e: + log.debug(f'Failed to close channel room {id}: {e}') return True except Exception as e: log.exception(e) diff --git a/backend/open_webui/routers/groups.py b/backend/open_webui/routers/groups.py index c45690fc3a..98e87edca5 100755 --- a/backend/open_webui/routers/groups.py +++ b/backend/open_webui/routers/groups.py @@ -21,6 +21,7 @@ from open_webui.internal.db import get_async_session from sqlalchemy.ext.asyncio import AsyncSession from open_webui.utils.auth import get_admin_user, get_verified_user +from open_webui.socket.main import resync_channel_rooms_for_users log = logging.getLogger(__name__) @@ -235,8 +236,13 @@ async def remove_users_from_group( db: AsyncSession = Depends(get_async_session), ): try: + removed_user_ids = list(form_data.user_ids or []) group = await Groups.remove_users_from_group(id, form_data.user_ids, db=db) if group: + # Group membership backs channel access grants; evict the removed + # users from any channel:* rooms they no longer qualify for. + if removed_user_ids: + await resync_channel_rooms_for_users(removed_user_ids) return GroupResponse( **group.model_dump(), member_count=await Groups.get_group_member_count_by_id(group.id, db=db), @@ -246,6 +252,8 @@ async def remove_users_from_group( status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT('Error removing users from group'), ) + except HTTPException: + raise except Exception as e: log.exception(f'Error removing users from group {id}: {e}') raise HTTPException( @@ -262,14 +270,26 @@ async def remove_users_from_group( @router.delete('/id/{id}/delete', response_model=bool) async def delete_group_by_id(id: str, user=Depends(get_admin_user), db: AsyncSession = Depends(get_async_session)): try: + # Capture members before deletion so we can resync their live rooms + # afterwards — the group is the access basis, not the members. + try: + members_before = await Users.get_users_by_group_id(id, db=db) + affected_user_ids = [m.id for m in members_before] + except Exception: + affected_user_ids = [] + result = await Groups.delete_group_by_id(id, db=db) if result: + if affected_user_ids: + await resync_channel_rooms_for_users(affected_user_ids) return result else: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT('Error deleting group'), ) + except HTTPException: + raise except Exception as e: log.exception(f'Error deleting group {id}: {e}') raise HTTPException( diff --git a/backend/open_webui/routers/notes.py b/backend/open_webui/routers/notes.py index 61c9fb7d95..8fca059eca 100644 --- a/backend/open_webui/routers/notes.py +++ b/backend/open_webui/routers/notes.py @@ -6,7 +6,7 @@ from typing import Optional from fastapi import APIRouter, Depends, HTTPException, Request, status, BackgroundTasks from pydantic import BaseModel -from open_webui.socket.main import sio +from open_webui.socket.main import sio, evict_users_from_note, get_user_ids_from_room from open_webui.models.groups import Groups from open_webui.models.users import Users, UserResponse @@ -361,6 +361,27 @@ async def update_note_access_by_id( await AccessGrants.set_access_grants('note', id, form_data.access_grants, db=db) + # Access grants just changed. Sweep every user currently subscribed to + # either note room and drop the ones whose access no longer stands. + # The note owner and admins are never evicted by has_access — the check + # below only returns False for users who truly lost the grant. + participant_user_ids = set(get_user_ids_from_room(f'note:{id}')) | set(get_user_ids_from_room(f'doc_{id}')) + for participant_user_id in participant_user_ids: + if participant_user_id == note.user_id: + continue + participant = await Users.get_user_by_id(participant_user_id, db=db) + if participant is None or participant.role == 'admin': + continue + still_has_access = await AccessGrants.has_access( + user_id=participant_user_id, + resource_type='note', + resource_id=id, + permission='read', + db=db, + ) + if not still_has_access: + await evict_users_from_note(id, [participant_user_id]) + return await Notes.get_note_by_id(id, db=db) @@ -402,6 +423,12 @@ async def delete_note_by_id( try: note = await Notes.delete_note_by_id(id, db=db) + # Close both note rooms — with the note gone, no one retains access. + for room_name in (f'note:{id}', f'doc_{id}'): + try: + await sio.close_room(room_name) + except Exception as e: + log.debug(f'Failed to close room {room_name}: {e}') return True except Exception as e: log.exception(e) diff --git a/backend/open_webui/socket/main.py b/backend/open_webui/socket/main.py index 2c44eb25c5..26c2ebd739 100644 --- a/backend/open_webui/socket/main.py +++ b/backend/open_webui/socket/main.py @@ -312,6 +312,132 @@ async def enter_room_for_users(room: str, user_ids: list[str]): log.debug(f'Failed to make users {user_ids} join room {room}: {e}') +async def leave_rooms_for_users(room: str, user_ids: list[str]): + """ + Make all active sessions of each user leave a specific Socket.IO room. + + Call this when an access basis (group membership, channel membership, + access grant) is revoked — the user's open sessions still hold the + room subscriptions established at join time and would otherwise keep + receiving real-time events from a resource they no longer access. + + leave_room is a no-op when the sid is not actually a member, so it is + safe to call optimistically. + """ + try: + for user_id in user_ids: + for sid in get_session_ids_from_room(f'user:{user_id}'): + await sio.leave_room(sid, room) + except Exception as e: + log.debug(f'Failed to leave room {room} for users {user_ids}: {e}') + + +async def resync_channel_rooms_for_users(user_ids: list[str]): + """ + Recompute each user's channel access and evict their active sessions + from channel:* rooms they no longer have access to. + + Used by access-control mutations that do not name a specific channel + but can still revoke channel access indirectly — removing a user from + a group, deleting a group that held grants, or channel access-grant + updates where the exact delta is cheaper to re-derive than to diff. + + Uses the manager's local room map (available in both the in-memory + and Redis-backed managers) to enumerate candidate rooms. Degrades + gracefully if the attribute is not present in a future library + version — in that case no eviction happens and callers should rely + on targeted leave_rooms_for_users calls instead. + """ + if not user_ids: + return + + try: + ns_rooms = getattr(sio.manager, 'rooms', {}).get('/', {}) or {} + except Exception as e: + log.debug(f'resync_channel_rooms_for_users: could not read manager rooms: {e}') + return + + channel_rooms = [r for r in list(ns_rooms.keys()) if isinstance(r, str) and r.startswith('channel:')] + if not channel_rooms: + return + + for user_id in user_ids: + session_ids = get_session_ids_from_room(f'user:{user_id}') + if not session_ids: + continue + try: + user = await Users.get_user_by_id(user_id) + except Exception: + user = None + allowed_ids: set[str] = set() + if user is not None and (user.role == 'admin' or await has_permission(user.id, 'features.channels')): + try: + allowed_ids = {c.id for c in await Channels.get_channels_by_user_id(user.id)} + except Exception as e: + log.debug(f'resync_channel_rooms_for_users: channel lookup failed for {user_id}: {e}') + continue + for room in channel_rooms: + channel_id = room[len('channel:'):] + if channel_id in allowed_ids: + continue + for sid in session_ids: + try: + await sio.leave_room(sid, room) + except Exception as e: + log.debug(f'Failed to leave {room} for sid {sid}: {e}') + + +async def evict_users_from_note(note_id: str, user_ids: list[str]): + """ + Evict users from both note rooms for a given note: the events-bus + room (note:{id}) and the ydoc collaboration room (doc_{id}). Call + this when a note's access grants change or the note is deleted. + """ + await leave_rooms_for_users(f'note:{note_id}', user_ids) + await leave_rooms_for_users(f'doc_{note_id}', user_ids) + + +async def resync_note_rooms_for_users(user_ids: list[str], note_ids: list[str]): + """ + For each (user, note) pair, leave the note's rooms if the user no + longer has access. Use this after access-grant changes that affect + specific notes. + """ + if not user_ids or not note_ids: + return + for user_id in user_ids: + session_ids = get_session_ids_from_room(f'user:{user_id}') + if not session_ids: + continue + try: + user = await Users.get_user_by_id(user_id) + except Exception: + user = None + if user is None: + continue + for note_id in note_ids: + try: + note = await Notes.get_note_by_id(note_id) + except Exception: + note = None + has_access = False + if note is not None: + if user.role == 'admin' or user.id == note.user_id: + has_access = True + else: + try: + has_access = await AccessGrants.has_access( + user_id=user.id, + resource_type='note', + resource_id=note.id, + permission='read', + ) + except Exception: + has_access = False + if not has_access: + await evict_users_from_note(note_id, [user_id]) + + async def disconnect_user_sessions(user_id: str): """Disconnect all Socket.IO sessions belonging to a user.