dandri/synapse
1
0
Fork 0
mirror of https://github.com/element-hq/synapse.git synced 2026-09-17 01:34:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

Handle user leave/ban rooms to prevent leaking data

Browse Source
This commit is contained in:
Devon Hudson
2025-11-09 08:45:52 -07:00
parent 0c0ece9612
commit cb82a4a687
2 changed files with 119 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files
synapse/handlers/sliding_sync/extensions.py
+91 -88
View File
@@ -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
tests/rest/client/sliding_sync/test_extension_threads.py
+28 -10
View File
@@ -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: