From 947b9107e0a148c51b305a42bf11edc4feacc64e Mon Sep 17 00:00:00 2001 From: Kegan Dougal <7190048+kegsay@users.noreply.github.com> Date: Thu, 24 Sep 2026 16:50:18 +0100 Subject: [PATCH] Various bug fixes and enable Complement tests for MSC4242 - Enable MSC4242 complement tests - Fix an issue which caused cascade rejections to not be applied to events in the /send_join response becau se we didn't remember in-memory which events were rejected. - Accept rejected events in the /send_join response, it gives us a bit of robustness at no safety cost. - Recalculate room state if backfilled events caused the state DAG fwd extrems to change. - Don't filter out already seen events from event_map, so the statement "the event_map is the complete stat e DAG" still holds. - Add validation to length of prev_state_events - Don't validate auth_events length for state DAG events as it is done too early, before we've had time to calculate the auth events (and validation is moot anyway given it isn't attacker controlled anymore) - Let backfilled events update state dag fwd extrems - Add test for edge case with backfilled state events --- .../conf/workers-shared-extra.yaml.j2 | 1 + scripts-dev/complement.sh | 1 + synapse/event_auth.py | 13 +- synapse/handlers/federation_event.py | 110 ++++++++++++-- synapse/storage/controllers/persist_events.py | 49 ++++++- tests/handlers/test_federation_event.py | 134 ++++++++++++++++++ 6 files changed, 290 insertions(+), 18 deletions(-) diff --git a/docker/complement/conf/workers-shared-extra.yaml.j2 b/docker/complement/conf/workers-shared-extra.yaml.j2 index 251018e1d2..bafcd9f3d1 100644 --- a/docker/complement/conf/workers-shared-extra.yaml.j2 +++ b/docker/complement/conf/workers-shared-extra.yaml.j2 @@ -147,6 +147,7 @@ experimental_features: msc4354_enabled: true # `/sync` `state_after` msc4222_enabled: true + msc4242_enabled: true server_notices: system_mxid_localpart: _server diff --git a/scripts-dev/complement.sh b/scripts-dev/complement.sh index d8e553f446..ec2e26961c 100755 --- a/scripts-dev/complement.sh +++ b/scripts-dev/complement.sh @@ -287,6 +287,7 @@ main() { ./tests/msc4306 ./tests/msc4222 ./tests/msc4429 + ./tests/msc4242 ) # Export the list of test packages as a space-separated environment variable, so other diff --git a/synapse/event_auth.py b/synapse/event_auth.py index 81a454b544..c2a18889a3 100644 --- a/synapse/event_auth.py +++ b/synapse/event_auth.py @@ -164,6 +164,7 @@ async def check_state_independent_auth_rules( store: _EventSourceStore, event: "EventBase", batched_auth_events: Mapping[str, "EventBase"] | None = None, + batched_rejections: Mapping[str, str] | None = None, ) -> None: """Check that an event complies with auth rules that are independent of room state @@ -175,6 +176,11 @@ async def check_state_independent_auth_rules( event: the event being checked. batched_auth_events: if the event being authed is part of a batch, any events from the same batch that may be necessary to auth the current event + batched_rejections: event ID to rejection reason for any events in `batched_auth_events` + which have been rejected. Events in the batch have not been persisted + yet, so their rejection is recorded neither on the event nor in the database, + and the caller has to tell us. Needed for MSC4242 State DAG rooms, where + rejection cascades through `prev_state_events`. Raises: AuthError if the checks fail @@ -244,11 +250,14 @@ async def check_state_independent_auth_rules( ) # 2.3 If there are entries which were themselves rejected under the checks performed on # receipt of a PDU, reject. - if prev_state_event.rejected_reason is not None: + rejected_reason = prev_state_event.rejected_reason + if rejected_reason is None and batched_rejections is not None: + rejected_reason = batched_rejections.get(prev_state_event.event_id) + if rejected_reason is not None: raise AuthError( 403, f"During auth for event {event.event_id} in room {event.room_id}, event has a " - + f"prev_state_event which is rejected ({prev_state_event.rejected_reason}): " + + f"prev_state_event which is rejected ({rejected_reason}): " + f"{prev_state_event.event_id}", ) diff --git a/synapse/handlers/federation_event.py b/synapse/handlers/federation_event.py index 8b77269a7d..265e2673c1 100644 --- a/synapse/handlers/federation_event.py +++ b/synapse/handlers/federation_event.py @@ -585,10 +585,19 @@ class FederationEventHandler: from_send_join=True, ) if room_version.msc4242_state_dags and has_rejected_events: - # The state DAG must not include rejected events - raise SynapseError( - 502, - "Unable to join because the remote server passed back a state DAG that includes rejected events (invalid)", + # A conformant server excludes rejected events from the state DAG it sends to + # new joiners, so their presence means the remote server is faulty. We log + # that rather than refusing the join: rejected events are excluded from the + # state DAG forward extremities, so they cannot contribute to the room state + # we calculate, and any event referencing one is rejected in turn. Refusing + # would cost availability without buying any safety. + logger.warning( + "Remote server %s sent a state DAG containing rejected events when we " + "joined %s. This is a bug in that server: rejected events must be " + "excluded from /send_join responses. Joining anyway; the rejected events " + "form no part of the room state.", + origin, + room_id, ) # and now persist the join event itself. @@ -899,6 +908,12 @@ class FederationEventHandler: backfilled: True if this is part of a historical batch of events (inhibits notification to clients, and validation of device keys.) """ + if not events: + return + + # Our callers only ever pass events for a single room. + room_id = next(iter(events)).room_id + set_tag( SynapseTags.FUNC_ARG_PREFIX + "event_ids", str([event.event_id for event in events]), @@ -968,10 +983,50 @@ class FederationEventHandler: # ascending) because one backfill event is likely to be the `prev_event` of # the next event we're going to process. sorted_events = sorted(new_events, key=lambda x: x.depth) + + # Backfilled events are persisted without recalculating the current state, + # which is right for the room DAG but not always for the state DAG. Backfill + # can teach us about state events we have never seen, which then belong on + # the state DAG frontier and so do change the current state. That happens + # when an event's `prev_events` reach state events which its + # `prev_state_events` do not, because the state DAG walk in + # /get_missing_events only ever follows `prev_state_events`. + # This is defence-in-depth: prev_events on honest servers is a superset of the state + # events in that event's prev_state_events, and only when they disagree (e.g. refer to + # an event on a fork) will this occur, but we have to handle the adversarial case too. + # + # Remember the frontier so that we can recalculate the current state if it moves. + state_dag_extremities_before: frozenset[str] | None = None + if ( + backfilled # Non-backfilled pulled events will automatically update the current state, this check is only for /backfill + and sorted_events + and supports_msc4242_state_dag(sorted_events[0]) + ): + state_dag_extremities_before = ( + await self._store.get_state_dag_extremities(room_id) + ) + for ev in sorted_events: with nested_logging_context(ev.event_id): await self._process_pulled_event(origin, ev, backfilled=backfilled) + if state_dag_extremities_before is not None: + state_dag_extremities_after = ( + await self._store.get_state_dag_extremities(room_id) + ) + if state_dag_extremities_after != state_dag_extremities_before: + # This is unusual because it implies the graph induced by prev_events vs prev_state_events + # has diverged, so warn about it. + logger.warning( + "Backfill from %s moved the state DAG forward extremities of %s from %s " + "to %s, recalculating the current state", + origin, + room_id, + shortstr(state_dag_extremities_before), + shortstr(state_dag_extremities_after), + ) + await self._state_handler.update_current_state(room_id) + # Check if we've already tried to process these events at some point in the # past. We aren't concerned with the expontntial backoff here, just whether it # has failed to be processed before. @@ -1770,9 +1825,10 @@ class FederationEventHandler: # persister does another round of deduplication. has_rejected_events = False seen_remotes = await self._store.have_seen_events(room_id, event_map.keys()) + unseen_event_map = dict(event_map) if seen_remotes: for s in seen_remotes: - event_map.pop(s, None) + unseen_event_map.pop(s, None) rejected_event_ids = await self._store.has_rejected_event_ids(seen_remotes) if rejected_event_ids: has_rejected_events = True @@ -1789,12 +1845,12 @@ class FederationEventHandler: if supports_msc4242_state_dag(ev) else ev.auth_event_ids() ) - if e_id in event_map + if e_id in unseen_event_map ] - for ev in event_map.values() + for ev in unseen_event_map.values() } - sorted_event_ids = sorted_topologically(event_map.keys(), auth_graph) - sorted_events = [event_map[e_id] for e_id in sorted_event_ids] + sorted_event_ids = sorted_topologically(unseen_event_map.keys(), auth_graph) + sorted_events = [unseen_event_map[e_id] for e_id in sorted_event_ids] logger.info( "Persisting %i remaining outliers: %s", len(sorted_events), @@ -1922,7 +1978,10 @@ class FederationEventHandler: Params: room_id: the room that the events are meant to be in (though this has not yet been checked) - event_map: all of the events being persisted, by event ID + event_map: every event we were given, by event ID. This includes events we + have already persisted and so are not persisting again, because we may + still need them to work out the auth events of an event we have not seen + before. On /send_join this is the complete state DAG. sorted_events: the events being persisted, sorted so that an event's `prev_state_events` come before it from_send_join: True if `event_map` is the complete state DAG from a @@ -1936,6 +1995,12 @@ class FederationEventHandler: state_group_to_state_map: dict[int, StateMap[str]] = {} # Tracks which events have been processed in causal order (prev_state_events first) processed_event_map: dict[str, EventBase] = {} + # The events in this batch which we have rejected, and why. Rejection cascades + # through `prev_state_events`, but nothing in the batch is persisted until we have + # processed all of it, so an event we rejected a moment ago still looks accepted + # both in the database and on the in-memory event. We therefore have to remember + # our own decisions and tell the auth rules about them. + rejections: dict[str, str] = {} async def process(event: EventBase) -> EventPersistencePair: assert supports_msc4242_state_dag(event) @@ -1974,9 +2039,9 @@ class FederationEventHandler: if from_send_join: # event_map is the complete state DAG - # The events in this batch aren't persisted yet, so pull the auth - # events out of the batch, falling back to the database for events we - # had already seen and hence filtered out of the batch. + # The events being persisted aren't in the database yet, so pull the + # auth events out of `event_map`, which holds the whole state DAG, + # including the events we had already persisted before this join. calculated_auth_events = { event_id: event_map[event_id] for event_id in calculated_auth_event_ids @@ -2028,7 +2093,10 @@ class FederationEventHandler: try: validate_event_for_room_version(event) await check_state_independent_auth_rules( - self._store, event, batched_auth_events + self._store, + event, + batched_auth_events, + batched_rejections=rejections, ) check_state_dependent_auth_rules( event, calculated_auth_events.values() @@ -2048,6 +2116,8 @@ class FederationEventHandler: logger.warning("While validating received event %r: %s", event, e) context.rejected = RejectedReason.OVERSIZED_EVENT + if context.rejected is not None: + rejections[event.event_id] = context.rejected processed_event_map[event.event_id] = event return event, context @@ -2827,7 +2897,17 @@ class FederationEventHandler: ) raise SynapseError(HTTPStatus.BAD_REQUEST, "Too many prev_events") - if len(ev.auth_event_ids()) > 10: + if supports_msc4242_state_dag(ev) and len(ev.prev_state_events) > 20: + logger.warning( + "Rejecting event %s which has %i prev_state_events", + ev.event_id, + len(ev.prev_state_events), + ) + raise SynapseError(HTTPStatus.BAD_REQUEST, "Too many prev_state_events") + + # MSC4242 State DAG events don't list their auth events (they're calculated from + # the state DAG), so there's nothing to bound here for them. + if not supports_msc4242_state_dag(ev) and len(ev.auth_event_ids()) > 10: logger.warning( "Rejecting event %s which has %i auth_events", ev.event_id, diff --git a/synapse/storage/controllers/persist_events.py b/synapse/storage/controllers/persist_events.py index 13f958d998..2994d637bc 100644 --- a/synapse/storage/controllers/persist_events.py +++ b/synapse/storage/controllers/persist_events.py @@ -653,7 +653,54 @@ class EventsPersistenceStorageController: state_delta_for_room = None new_state_dag_extrems = None - if not backfilled: + if all_supports_msc4242_state_dag(chunk) and backfilled: + # Backfilled events do not update the room DAG extremities or the + # current state, but a state event which is new to us still has to move + # the state DAG forward extremities. + # + # In practice the only thing which gets here are state events from /send_join. + # Those events are persisted as backfilled so that they do not appear down + # /sync, but the state DAG forwards extremities must be updated otherwise when + # we hit the join event we won't calculate the current state correctly due to having + # stale extremities. + # + # Catching up on a room we are already in does not come through here: the + # state DAG walk in /get_missing_events fetches `prev_state_events` at + # backfilled=False, and we refuse to persist any event whose + # `prev_state_events` we do not already hold, so an event cannot reach us + # ahead of the prev_state_events it references. + # + # There is one edge case here: when prev_events disagrees with prev_state_events. E.g: + # + # prev_events graph We know S1, S2, S3 + # S1 <- S2 <- S3 Sx = state event, M = message + # ^-- S4 <- S5 <- M Backfilling M returns S4, S5 alongside it + # + # Suppose M incorrectly has prev_state_events=[S3] but prev_events=[S5] + # then /get_missing_events will NOT fetch S4, S5 and only when you /backfill + # the prev_events graph will you suddenly become aware of [S4, S5], + # which will hit this code path. In this case, we should be recalculating the current + # state to take S4,S5 into account but this code path does not. + # We rely on the caller (_process_pulled_events) to handle this case by checking if + # _backfilled_ events changed the state DAG forward extremities, and in the extremely + # unlikely case it did, warn in the logs and recalculate the room state. + seen_event_ids = await self.main_store.have_seen_events( + room_id, [event.event_id for event, _ in chunk] + ) + unseen_chunk = [ + (event, context) + for event, context in chunk + if event.event_id not in seen_event_ids + ] + if unseen_chunk: + new_state_dag_extrems = ( + await self._calculate_new_state_dag_extremities( + room_id, + await self.main_store.get_state_dag_extremities(room_id), + unseen_chunk, + ) + ) + elif not backfilled: if all_supports_msc4242_state_dag(chunk): with Measure( self._clock, diff --git a/tests/handlers/test_federation_event.py b/tests/handlers/test_federation_event.py index 0e15c8d744..554de2b3dd 100644 --- a/tests/handlers/test_federation_event.py +++ b/tests/handlers/test_federation_event.py @@ -1240,3 +1240,137 @@ class IsStateDagConnectedTests(unittest.TestCase): third = self._make_event([second.event_id]) self.assertFalse(is_state_dag_connected([create, third])) + + +class BackfilledStateDagEventTests(unittest.FederatingHomeserverTestCase): + servlets = [ + admin.register_servlets, + login.register_servlets, + room.register_servlets, + ] + + @unittest.override_config({"experimental_features": {"msc4242_enabled": True}}) + def test_backfilled_state_events_update_current_state(self) -> None: + """State events we only learn about through backfill must reach the current state. + + An event's `prev_events` can reach state events which its `prev_state_events` do + not. This should not normally happen but the protocol allows it, so we have to handle it. + The state DAG walk in /get_missing_events only follows `prev_state_events`, + so those state events are never fetched that way, and we first see them when the + `prev_events` graph is backfilled. They are still state events which nothing in + our state DAG references, so they belong on the state DAG frontier and therefore + change the current state. + + prev_state_events graph We know MEMBER and TOPIC + Sx = state event + MEMBER <- TOPIC Backfill hands us S4 and S5, forking at MEMBER + `-- S4 <- S5 + + The frontier becomes {TOPIC, S5}, so S5 has to be merged into the current state. + """ + OTHER_USER = f"@user:{self.OTHER_SERVER_NAME}" + main_store = self.hs.get_datastores().main + federation_event_handler = self.hs.get_federation_event_handler() + + user_id = self.register_user("kermit", "test") + tok = self.login("kermit", "test") + room_id = self.helper.create_room_as( + room_creator=user_id, + tok=tok, + room_version=RoomVersions.MSC4242v12.identifier, + ) + room_version = self.get_success(main_store.get_room_version(room_id)) + self.assertTrue(room_version.msc4242_state_dags) + + # allow the remote user to send state events + self.helper.send_state( + room_id, + "m.room.power_levels", + {"events_default": 0, "state_default": 0}, + tok=tok, + ) + + # add the remote user to the room. The backfilled branch forks here. + member_event = self.get_success( + event_injection.inject_member_event(self.hs, room_id, OTHER_USER, "join") + ) + + # Move the state DAG on locally, so that what we backfill below is a genuine fork + # rather than an extension of the only branch we have. + topic_event_id = self.helper.send_state( + room_id, "m.room.topic", {"topic": "local"}, tok=tok + )["event_id"] + + extremities_before = self.get_success( + main_store.get_state_dag_extremities(room_id) + ) + self.assertIncludes({topic_event_id}, extremities_before, exact=True) + + # S4 and S5 are a branch of the state DAG which we have never been told about: + # nothing we hold references them via `prev_state_events`, so the state DAG walk + # would never have fetched them. + s4 = make_test_event( + self.add_hashes_and_signatures_from_other_server( + { + "type": "test_state_type", + "state_key": "", + "room_id": room_id, + "sender": OTHER_USER, + "prev_events": [member_event.event_id], + "prev_state_events": [member_event.event_id], + "origin_server_ts": 1, + "depth": 10, + "content": {"body": "s4"}, + }, + room_version, + ), + room_version, + ) + s5 = make_test_event( + self.add_hashes_and_signatures_from_other_server( + { + "type": "test_state_type", + "state_key": "", + "room_id": room_id, + "sender": OTHER_USER, + "prev_events": [s4.event_id], + "prev_state_events": [s4.event_id], + "origin_server_ts": 2, + "depth": 11, + "content": {"body": "s5"}, + }, + room_version, + ), + room_version, + ) + + self.get_success( + federation_event_handler._process_pulled_events( + self.OTHER_SERVER_NAME, [s4, s5], backfilled=True + ) + ) + + # The tip of the backfilled branch is now a state DAG forward extremity, and the + # branch we already had still is too: the fork is unmerged in the state DAG. + extremities_after = self.get_success( + main_store.get_state_dag_extremities(room_id) + ) + self.assertIncludes( + {topic_event_id, s5.event_id}, + extremities_after, + exact=True, + message="the backfilled state event did not become a state DAG forward extremity", + ) + + # ...and resolving that frontier is what the current state must reflect. + state_map = self.get_success(main_store.get_partial_current_state_ids(room_id)) + self.assertEqual( + state_map.get(("test_state_type", "")), + s5.event_id, + "the state event learned through backfill is not part of the current state", + ) + self.assertEqual( + state_map.get(("m.room.topic", "")), + topic_event_id, + "merging in the backfilled branch lost the state on the branch we had", + )