From 05f5ce231b81c645d88a52fabf30e90e40f47aab Mon Sep 17 00:00:00 2001 From: Kegan Dougal <7190048+kegsay@users.noreply.github.com> Date: Wed, 22 Apr 2026 16:58:37 +0100 Subject: [PATCH] fedclient: add support for MSC4242 fields Pending unit tests --- synapse/federation/federation_client.py | 187 +++++++++++++++--------- synapse/federation/transport/client.py | 32 +++- synapse/handlers/federation.py | 7 + tests/handlers/test_device.py | 1 + tests/handlers/test_federation.py | 1 + tests/handlers/test_room_member.py | 1 + tests/storage/test_stream.py | 1 + 7 files changed, 155 insertions(+), 75 deletions(-) diff --git a/synapse/federation/federation_client.py b/synapse/federation/federation_client.py index 78a1900c73..1b444c3179 100644 --- a/synapse/federation/federation_client.py +++ b/synapse/federation/federation_client.py @@ -126,6 +126,9 @@ class SendJoinResult: # Always contains the server we joined off. servers_in_room: AbstractSet[str] + # Only valid for state DAG rooms (MSC4242) + state_dag: list[EventBase] | None + class FederationClient(FederationBase): def __init__(self, hs: "HomeServer"): @@ -1108,11 +1111,12 @@ class FederationClient(FederationBase): SynapseError: if the chosen remote server returns a 300/400 code, or no servers successfully handle the request. """ - # See related restriction in /createRoom requests in handlers/room.py - if room_version.msc4242_state_dags: - raise UnsupportedRoomVersionError( - "Homeserver does not support this room version over federation" - ) + + def find_create_event(events: list[EventBase]) -> EventBase | None: + for e in events: + if (e.type, e.state_key) == (EventTypes.Create, ""): + return e + return None async def send_request(destination: str) -> SendJoinResult: response = await self._do_send_join( @@ -1142,13 +1146,16 @@ class FederationClient(FederationBase): state = response.state auth_chain = response.auth_events + state_dag: list[EventBase] = [] + if room_version.msc4242_state_dags: + if not response.state_dag: + raise InvalidResponseError("No state_dag returned") + state_dag = response.state_dag - create_event = None - for e in state: - if (e.type, e.state_key) == (EventTypes.Create, ""): - create_event = e - break - + # Validate the create event and room version are what we expect to see. + create_event = find_create_event( + state_dag if room_version.msc4242_state_dags else state + ) if create_event is None: # If the state doesn't have a create event then the room is # invalid, and it would fail auth checks anyway. @@ -1166,62 +1173,7 @@ class FederationClient(FederationBase): % (create_room_version,) ) - logger.info( - "Processing from send_join %d events", len(state) + len(auth_chain) - ) - - # We now go and check the signatures and hashes for the event. Note - # that we limit how many events we process at a time to keep the - # memory overhead from exploding. - valid_pdus_map: dict[str, EventBase] = {} - - async def _execute(pdu: EventBase) -> None: - valid_pdu = await self._check_sigs_and_hash_and_fetch_one( - pdu=pdu, - origin=destination, - room_version=room_version, - ) - - if valid_pdu: - valid_pdus_map[valid_pdu.event_id] = valid_pdu - - await concurrently_execute( - _execute, itertools.chain(state, auth_chain), 10000 - ) - - # NB: We *need* to copy to ensure that we don't have multiple - # references being passed on, as that causes... issues. - signed_state = [ - copy.copy(valid_pdus_map[p.event_id]) - for p in state - if p.event_id in valid_pdus_map - ] - - signed_auth = [ - valid_pdus_map[p.event_id] - for p in auth_chain - if p.event_id in valid_pdus_map - ] - - # NB: We *need* to copy to ensure that we don't have multiple - # references being passed on, as that causes... issues. - for s in signed_state: - s.internal_metadata = s.internal_metadata.copy() - - # double-check that the auth chain doesn't include a different create event - auth_chain_create_events = [ - e.event_id - for e in signed_auth - if (e.type, e.state_key) == (EventTypes.Create, "") - ] - if auth_chain_create_events and auth_chain_create_events != [ - create_event.event_id - ]: - raise InvalidResponseError( - "Unexpected create event(s) in auth chain: %s" - % (auth_chain_create_events,) - ) - + # Validate and set faster room joins fields servers_in_room = None if response.servers_in_room is not None: servers_in_room = set(response.servers_in_room) @@ -1241,6 +1193,103 @@ class FederationClient(FederationBase): # Fix things up in case the remote homeserver is badly behaved. servers_in_room.add(destination) + logger.info( + "Processing from send_join %d events", + len(state_dag) + if room_version.msc4242_state_dags + else (len(state) + len(auth_chain)), + ) + + # We now go and check the signatures and hashes for the event. Note + # that we limit how many events we process at a time to keep the + # memory overhead from exploding. + valid_pdus_map: dict[str, EventBase] = {} + + async def _execute(pdu: EventBase) -> None: + valid_pdu = await self._check_sigs_and_hash_and_fetch_one( + pdu=pdu, + origin=destination, + room_version=room_version, + ) + + if valid_pdu: + valid_pdus_map[valid_pdu.event_id] = valid_pdu + + # Verify signatures/hashes on events, and make sure they all refer to the same room. + if room_version.msc4242_state_dags: + if state or auth_chain: + raise InvalidResponseError( + "State DAG rooms must not set state or auth_chain fields" + ) + await concurrently_execute(_execute, itertools.chain(state_dag), 10000) + # Copy valid PDUs along with internal metadata. + # It's unclear why this is needed but the code seems to expect it. + signed_state_dag = [ + copy.copy(valid_pdus_map[p.event_id]) + for p in state_dag + if p.event_id in valid_pdus_map + ] + for s in signed_state_dag: + s.internal_metadata = s.internal_metadata.copy() + + # Verify each event is for this room (and thus has the same create event as it is v12+) + for state_event in signed_state_dag: + if state_event.room_id != pdu.room_id: + raise InvalidResponseError( + "%s in state_dag belongs to room %s, not %s which we are joining" + % (state_event.event_id, state_event.room_id, pdu.room_id) + ) + return SendJoinResult( + event=event, + state=[], + auth_chain=[], + state_dag=signed_state_dag, + origin=destination, + partial_state=response.members_omitted, + servers_in_room=servers_in_room or frozenset(), + ) + else: + if state_dag: + raise InvalidResponseError( + "Room does not support state DAGs but set state_dag field" + ) + await concurrently_execute( + _execute, itertools.chain(state, auth_chain), 10000 + ) + + # NB: We *need* to copy to ensure that we don't have multiple + # references being passed on, as that causes... issues. + signed_state = [ + copy.copy(valid_pdus_map[p.event_id]) + for p in state + if p.event_id in valid_pdus_map + ] + + signed_auth = [ + valid_pdus_map[p.event_id] + for p in auth_chain + if p.event_id in valid_pdus_map + ] + + # NB: We *need* to copy to ensure that we don't have multiple + # references being passed on, as that causes... issues. + for s in signed_state: + s.internal_metadata = s.internal_metadata.copy() + + # double-check that the auth chain doesn't include a different create event + auth_chain_create_events = [ + e.event_id + for e in signed_auth + if (e.type, e.state_key) == (EventTypes.Create, "") + ] + if auth_chain_create_events and auth_chain_create_events != [ + create_event.event_id + ]: + raise InvalidResponseError( + "Unexpected create event(s) in auth chain: %s" + % (auth_chain_create_events,) + ) + return SendJoinResult( event=event, state=signed_state, @@ -1248,6 +1297,7 @@ class FederationClient(FederationBase): origin=destination, partial_state=response.members_omitted, servers_in_room=servers_in_room or frozenset(), + state_dag=None, ) # MSC3083 defines additional error codes for room joins. @@ -1548,6 +1598,7 @@ class FederationClient(FederationBase): limit: int, min_depth: int, timeout: int, + state_dag: bool = False, ) -> list[EventBase]: """Tries to fetch events we are missing. This is called when we receive an event without having received all of its ancestors. @@ -1563,6 +1614,7 @@ class FederationClient(FederationBase): limit: Maximum number of events to return. min_depth: Minimum depth of events to return. timeout: Max time to wait in ms + state_dag: True to walk the state DAG (MSC4242 rooms) """ try: content = await self.transport_layer.get_missing_events( @@ -1573,6 +1625,7 @@ class FederationClient(FederationBase): limit=limit, min_depth=min_depth, timeout=timeout, + state_dag=state_dag, ) room_version = await self.store.get_room_version(room_id) diff --git a/synapse/federation/transport/client.py b/synapse/federation/transport/client.py index 5d5212ef96..e4747016c4 100644 --- a/synapse/federation/transport/client.py +++ b/synapse/federation/transport/client.py @@ -776,18 +776,21 @@ class TransportLayerClient: limit: int, min_depth: int, timeout: int, + state_dag: bool, ) -> JsonDict: path = _create_v1_path("/get_missing_events/%s", room_id) - + request_body = { + "limit": int(limit), + "min_depth": int(min_depth), + "earliest_events": earliest_events, + "latest_events": latest_events, + } + if state_dag: + request_body["org.matrix.msc4242.state_dag"] = True return await self.client.post_json( destination=destination, path=path, - data={ - "limit": int(limit), - "min_depth": int(min_depth), - "earliest_events": earliest_events, - "latest_events": latest_events, - }, + data=request_body, timeout=timeout, ) @@ -986,6 +989,10 @@ class SendJoinResponse: # "event" is not included in the response. event: EventBase | None = None + # MSC4242: State DAGs. Always included for state dag rooms, else None. + # Replaces auth_events. + state_dag: list[EventBase] | None = None + # The room state is incomplete members_omitted: bool = False @@ -1068,7 +1075,7 @@ class SendJoinParser(ByteParser[SendJoinResponse]): MAX_RESPONSE_SIZE = 500 * 1024 * 1024 def __init__(self, room_version: RoomVersion, v1_api: bool): - self._response = SendJoinResponse([], [], event_dict={}) + self._response = SendJoinResponse([], [], event_dict={}, state_dag=[]) self._room_version = room_version self._coros: list[Generator[None, bytes, None]] = [] @@ -1112,6 +1119,15 @@ class SendJoinParser(ByteParser[SendJoinResponse]): ) ) + if room_version.msc4242_state_dags: + self._coros.append( + ijson.items_coro( + _event_list_parser(room_version, self._response.state_dag), + prefix + "state_dag.item", + use_float=True, + ) + ) + def write(self, data: bytes) -> int: for c in self._coros: c.send(data) diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py index b3444dd2ef..c7f12310a7 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -53,6 +53,7 @@ from synapse.api.errors import ( PartialStateConflictError, RequestSendFailed, SynapseError, + UnsupportedRoomVersionError, ) from synapse.api.room_versions import KNOWN_ROOM_VERSIONS, RoomVersion from synapse.crypto.event_signing import compute_event_signature @@ -646,6 +647,12 @@ class FederationHandler: room_id ) + # See related restriction in /createRoom requests in handlers/room.py + if room_version_obj.msc4242_state_dags: + raise UnsupportedRoomVersionError( + "Homeserver does not support this room version over federation" + ) + ret = await self.federation_client.send_join( host_list, event, diff --git a/tests/handlers/test_device.py b/tests/handlers/test_device.py index 9e44b1dc1e..9026e1697f 100644 --- a/tests/handlers/test_device.py +++ b/tests/handlers/test_device.py @@ -822,6 +822,7 @@ class DeviceUnPartialStateTestCase(unittest.HomeserverTestCase): partial_state=True, # Only REMOTE1_SERVER_NAME is known at join time. servers_in_room={self.REMOTE1_SERVER_NAME}, + state_dag=None, ) ) diff --git a/tests/handlers/test_federation.py b/tests/handlers/test_federation.py index e4a41cf1ae..efb800ba06 100644 --- a/tests/handlers/test_federation.py +++ b/tests/handlers/test_federation.py @@ -652,6 +652,7 @@ class PartialJoinTestCase(unittest.FederatingHomeserverTestCase): ], partial_state=True, servers_in_room={"example.com"}, + state_dag=None, ) ) diff --git a/tests/handlers/test_room_member.py b/tests/handlers/test_room_member.py index 3890abdbc8..c84ad5e9e7 100644 --- a/tests/handlers/test_room_member.py +++ b/tests/handlers/test_room_member.py @@ -172,6 +172,7 @@ class TestJoinsLimitedByPerRoomRateLimiter(FederatingHomeserverTestCase): auth_chain=[create_event], partial_state=False, servers_in_room=frozenset(), + state_dag=None, ) ) diff --git a/tests/storage/test_stream.py b/tests/storage/test_stream.py index d51fa1f8ba..8fce7c03e9 100644 --- a/tests/storage/test_stream.py +++ b/tests/storage/test_stream.py @@ -1455,6 +1455,7 @@ class GetCurrentStateDeltaMembershipChangesForUserFederationTestCase( auth_chain=[create_event, creator_join_event], partial_state=False, servers_in_room=frozenset(), + state_dag=None, ) )