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", + )