From cb82a4a687e78985fb468fb9632f6cd1e039ffb9 Mon Sep 17 00:00:00 2001 From: Devon Hudson Date: Sun, 9 Nov 2025 08:45:52 -0700 Subject: [PATCH] Handle user leave/ban rooms to prevent leaking data --- synapse/handlers/sliding_sync/extensions.py | 179 +++++++++--------- .../sliding_sync/test_extension_threads.py | 38 +++- 2 files changed, 119 insertions(+), 98 deletions(-) diff --git a/synapse/handlers/sliding_sync/extensions.py b/synapse/handlers/sliding_sync/extensions.py index 80453516b1..3bbfc95c46 100644 --- a/synapse/handlers/sliding_sync/extensions.py +++ b/synapse/handlers/sliding_sync/extensions.py @@ -35,6 +35,7 @@ from synapse.api.constants import ( AccountDataTypes, EduTypes, EventContentFields, + Membership, MRelatesToFields, RelationTypes, ) @@ -1074,80 +1075,6 @@ class SlidingSyncExtensionHandler: if not threads_request.enabled: return None - - # if ( - # # No timeline for invite/knock rooms - # room_membership_for_user_at_to_token.membership - # not in (Membership.INVITE, Membership.KNOCK) - # ): - # limited = False - # # We want to start off using the `to_token` (vs `from_token`) because we look - # # backwards from the `to_token` up to the `timeline_limit` and we might not - # # reach the `from_token` before we hit the limit. We will update the room stream - # # position once we've fetched the events to point to the earliest event fetched. - # prev_batch_token = to_token - # - # # We're going to paginate backwards from the `to_token` - # to_bound = to_token.room_key - # # People shouldn't see past their leave/ban event - # if room_membership_for_user_at_to_token.membership in ( - # Membership.LEAVE, - # Membership.BAN, - # ): - # to_bound = room_membership_for_user_at_to_token.event_pos.to_room_stream_token() - # - # # For initial `/sync` (and other historical scenarios mentioned above), we - # # want to view a historical section of the timeline; to fetch events by - # # `topological_ordering` (best representation of the room DAG as others were - # # seeing it at the time). This also aligns with the order that `/messages` - # # returns events in. - # # - # # For incremental `/sync`, we want to get all updates for rooms since - # # the last `/sync` (regardless if those updates arrived late or happened - # # a while ago in the past); to fetch events by `stream_ordering` (in the - # # order they were received by the server). - # # - # # Relevant spec issue: https://github.com/matrix-org/matrix-spec/issues/1917 - # # - # # FIXME: Using workaround for mypy, - # # https://github.com/python/mypy/issues/10740#issuecomment-1997047277 and - # # https://github.com/python/mypy/issues/17479 - # paginate_room_events_by_topological_ordering: PaginateFunction = ( - # self.store.paginate_room_events_by_topological_ordering - # ) - # paginate_room_events_by_stream_ordering: PaginateFunction = ( - # self.store.paginate_room_events_by_stream_ordering - # ) - # pagination_method: PaginateFunction = ( - # # Use `topographical_ordering` for historical events - # paginate_room_events_by_topological_ordering - # if timeline_from_bound is None - # # Use `stream_ordering` for updates - # else paginate_room_events_by_stream_ordering - # ) - # timeline_events, new_room_key, limited = await pagination_method( - # room_id=room_id, - # # The bounds are reversed so we can paginate backwards - # # (from newer to older events) starting at to_bound. - # # This ensures we fill the `limit` with the newest events first, - # from_key=to_bound, - # to_key=timeline_from_bound, - # direction=Direction.BACKWARDS, - # limit=room_sync_config.timeline_limit, - # ) - # - # # Make sure we don't expose any events that the client shouldn't see - # timeline_events = await filter_events_for_client( - # self.storage_controllers, - # user.to_string(), - # timeline_events, - # is_peeking=room_membership_for_user_at_to_token.membership - # != Membership.JOIN, - # filter_send_to_client=True, - # ) - - # TODO: Improve by doing subqueries for rooms where user membership is changed - # Identify which threads already have events in the room timelines. # If include_roots=False, we'll exclude these threads from the DB query # since the client already sees the thread activity in the timeline. @@ -1157,19 +1084,95 @@ class SlidingSyncExtensionHandler: if not threads_request.include_roots: threads_to_exclude = self._find_threads_in_timeline(actual_room_response_map) - # Fetch thread updates globally across all joined rooms. - # The database layer filters out excluded threads and returns a StreamToken - # (exclusive) for prev_batch if there are more results. - ( - all_thread_updates, - prev_batch_token, - ) = await self.store.get_thread_updates_for_rooms( - room_ids=actual_room_ids, - from_token=from_token.stream_token.room_key if from_token else None, - to_token=to_token.room_key, - limit=threads_request.limit, - exclude_thread_ids=threads_to_exclude, - ) + # Separate rooms into groups based on membership status. + # For LEAVE/BAN rooms, we need to bound the to_token to prevent leaking events + # that occurred after the user left/was banned. + leave_ban_rooms: Set[str] = set() + other_rooms: Set[str] = set() + + for room_id in actual_room_ids: + membership_info = room_membership_for_user_at_to_token_map.get(room_id) + if membership_info and membership_info.membership in ( + Membership.LEAVE, + Membership.BAN, + ): + leave_ban_rooms.add(room_id) + else: + other_rooms.add(room_id) + + # Fetch thread updates, handling LEAVE/BAN rooms separately to avoid data leaks. + all_thread_updates: Dict[str, List[ThreadUpdateInfo]] = {} + prev_batch_token: Optional[StreamToken] = None + remaining_limit = threads_request.limit + + # Query for rooms where the user has left or been banned, using their leave/ban + # event position as the upper bound to prevent seeing events after they left. + if leave_ban_rooms: + for room_id in leave_ban_rooms: + if remaining_limit <= 0: + # We've already fetched enough updates, but we still need to set + # prev_batch to indicate there are more results. + prev_batch_token = to_token + break + + membership_info = room_membership_for_user_at_to_token_map[room_id] + bounded_to_token = membership_info.event_pos.to_room_stream_token() + + ( + room_thread_updates, + room_prev_batch, + ) = await self.store.get_thread_updates_for_rooms( + room_ids={room_id}, + from_token=from_token.stream_token.room_key if from_token else None, + to_token=bounded_to_token, + limit=remaining_limit, + exclude_thread_ids=threads_to_exclude, + ) + + # Count how many updates we fetched and reduce the remaining limit + num_updates = sum(len(updates) for updates in room_thread_updates.values()) + remaining_limit -= num_updates + + # Merge results + for thread_id, updates in room_thread_updates.items(): + all_thread_updates.setdefault(thread_id, []).extend(updates) + + # If any room has a prev_batch, we should set the global prev_batch. + # We use the maximum (latest) prev_batch token for backwards pagination. + if room_prev_batch is not None: + if prev_batch_token is None: + prev_batch_token = room_prev_batch + else: + # Take the maximum (latest) prev_batch token for backwards pagination + if room_prev_batch.room_key.stream > prev_batch_token.room_key.stream: + prev_batch_token = room_prev_batch + + # Query for rooms where the user is joined, invited, or knocking, using the + # normal to_token as the upper bound. + if other_rooms and remaining_limit > 0: + ( + other_thread_updates, + other_prev_batch, + ) = await self.store.get_thread_updates_for_rooms( + room_ids=other_rooms, + from_token=from_token.stream_token.room_key if from_token else None, + to_token=to_token.room_key, + limit=remaining_limit, + exclude_thread_ids=threads_to_exclude, + ) + + # Merge results + for thread_id, updates in other_thread_updates.items(): + all_thread_updates.setdefault(thread_id, []).extend(updates) + + # Merge prev_batch tokens + if other_prev_batch is not None: + if prev_batch_token is None: + prev_batch_token = other_prev_batch + else: + # Take the maximum (latest) prev_batch token for backwards pagination + if other_prev_batch.room_key.stream > prev_batch_token.room_key.stream: + prev_batch_token = other_prev_batch # Early return: no thread updates found if len(all_thread_updates) == 0: @@ -1226,7 +1229,7 @@ class SlidingSyncExtensionHandler: # Generate per-thread prev_batch token if this thread has multiple visible updates. # When we hit the global limit, we generate prev_batch tokens for all threads, even if # we only saw 1 update for them. This is to cover the case where we only saw - # a single update for a given thread, but the global limit prevent us from + # a single update for a given thread, but the global limit prevents us from # obtaining other updates which would have otherwise been included in the # range. per_thread_prev_batch = None diff --git a/tests/rest/client/sliding_sync/test_extension_threads.py b/tests/rest/client/sliding_sync/test_extension_threads.py index 4a10313a93..1e84e60046 100644 --- a/tests/rest/client/sliding_sync/test_extension_threads.py +++ b/tests/rest/client/sliding_sync/test_extension_threads.py @@ -295,12 +295,10 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase): def test_threads_not_returned_after_leaving_room(self) -> None: """ - Test that thread updates are not returned after a user leaves the room, - even if the thread was updated while they were joined. + Test that thread updates are properly bounded when a user leaves a room. - This tests the known limitation: if a thread has multiple updates and the - user leaves between them, they won't see any updates (even earlier ones - while joined). + Users should see thread updates that occurred up to the point they left, + but NOT updates that occurred after they left. """ user1_id = self.register_user("user1", "pass") user1_tok = self.login(user1_id, "pass") @@ -369,13 +367,33 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase): # User2 incremental sync response_body, _ = self.do_sync(sync_body, tok=user2_tok, since=sync_pos) - # Assert: User2 should NOT see the thread update (they left before latest update) - # Note: This also demonstrates that only currently joined rooms are returned - user2 - # won't see the thread even though there was an update while they were joined (Reply 1) - self.assertNotIn( + # Assert: User2 SHOULD see Reply 1 (happened while joined) but NOT Reply 2 (after leaving) + self.assertIn( EXT_NAME, response_body["extensions"], - "User2 should not see thread updates after leaving the room", + "User2 should see thread updates up to the point they left", + ) + self.assertIn( + room_id, + response_body["extensions"][EXT_NAME]["updates"], + "Thread updates should include the room user2 left", + ) + self.assertIn( + thread_root, + response_body["extensions"][EXT_NAME]["updates"][room_id], + "Thread root should be in the updates", + ) + + # Verify that only a single update was seen (Reply 1) by checking that there's + # no prev_batch token. If Reply 2 was also included, there would be multiple + # updates and a prev_batch token would be present. + thread_update = response_body["extensions"][EXT_NAME]["updates"][room_id][ + thread_root + ] + self.assertNotIn( + "prev_batch", + thread_update, + "No prev_batch should be present since only one update (Reply 1) is visible", ) def test_threads_with_include_roots_true(self) -> None: