dandri/synapse
1
0
Fork 0
mirror of https://github.com/element-hq/synapse.git synced 2026-08-28 13:44:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

fedclient: add support for MSC4242 fields

Browse Source 
Pending unit tests
This commit is contained in:
Kegan Dougal
2026-04-22 16:58:37 +01:00
parent 613cb4df1c
commit 05f5ce231b
7 changed files with 155 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
synapse/federation/federation_client.py
+120 -67
View File
@@ -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)
synapse/federation/transport/client.py
+24 -8
View File
@@ -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)
synapse/handlers/federation.py
+7
View File
@@ -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,
tests/handlers/test_device.py
+1
View File
@@ -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,
)
)
tests/handlers/test_federation.py
+1
View File
@@ -652,6 +652,7 @@ class PartialJoinTestCase(unittest.FederatingHomeserverTestCase):
],
partial_state=True,
servers_in_room={"example.com"},
state_dag=None,
)
)
tests/handlers/test_room_member.py
+1
View File
@@ -172,6 +172,7 @@ class TestJoinsLimitedByPerRoomRateLimiter(FederatingHomeserverTestCase):
auth_chain=[create_event],
partial_state=False,
servers_in_room=frozenset(),
state_dag=None,
)
)
tests/storage/test_stream.py
+1
View File
@@ -1455,6 +1455,7 @@ class GetCurrentStateDeltaMembershipChangesForUserFederationTestCase(
auth_chain=[create_event, creator_join_event],
partial_state=False,
servers_in_room=frozenset(),
state_dag=None,
)
)