mirror of
https://github.com/element-hq/synapse.git
synced 2024-12-27 09:58:41 +00:00
eda735e4bb
### Pull Request Checklist <!-- Please read https://element-hq.github.io/synapse/latest/development/contributing_guide.html before submitting your pull request --> * [X] Pull request is based on the develop branch * [X] Pull request includes a [changelog file](https://element-hq.github.io/synapse/latest/development/contributing_guide.html#changelog). The entry should: - Be a short description of your change which makes sense to users. "Fixed a bug that prevented receiving messages from other servers." instead of "Moved X method from `EventStore` to `EventWorkerStore`.". - Use markdown where necessary, mostly for `code blocks`. - End with either a period (.) or an exclamation mark (!). - Start with a capital letter. - Feel free to credit yourself, by adding a sentence "Contributed by @github_username." or "Contributed by [Your Name]." to the end of the entry. * [X] [Code style](https://element-hq.github.io/synapse/latest/code_style.html) is correct (run the [linters](https://element-hq.github.io/synapse/latest/development/contributing_guide.html#run-the-linters)) --------- Co-authored-by: Andrew Morgan <1342360+anoadragon453@users.noreply.github.com>
801 lines
30 KiB
Python
801 lines
30 KiB
Python
#
|
|
# This file is licensed under the Affero General Public License (AGPL) version 3.
|
|
#
|
|
# Copyright 2019 The Matrix.org Foundation C.I.C.
|
|
# Copyright (C) 2023 New Vector, Ltd
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU Affero General Public License as
|
|
# published by the Free Software Foundation, either version 3 of the
|
|
# License, or (at your option) any later version.
|
|
#
|
|
# See the GNU Affero General Public License for more details:
|
|
# <https://www.gnu.org/licenses/agpl-3.0.html>.
|
|
#
|
|
# Originally licensed under the Apache License, Version 2.0:
|
|
# <http://www.apache.org/licenses/LICENSE-2.0>.
|
|
#
|
|
# [This file includes modifications made by New Vector Limited]
|
|
#
|
|
#
|
|
import logging
|
|
from typing import Collection, Optional, cast
|
|
from unittest import TestCase
|
|
from unittest.mock import AsyncMock, Mock, patch
|
|
|
|
from twisted.internet.defer import Deferred
|
|
from twisted.test.proto_helpers import MemoryReactor
|
|
|
|
from synapse.api.constants import EventTypes
|
|
from synapse.api.errors import (
|
|
AuthError,
|
|
Codes,
|
|
LimitExceededError,
|
|
NotFoundError,
|
|
SynapseError,
|
|
)
|
|
from synapse.api.room_versions import RoomVersions
|
|
from synapse.events import EventBase, make_event_from_dict
|
|
from synapse.federation.federation_base import event_from_pdu_json
|
|
from synapse.federation.federation_client import SendJoinResult
|
|
from synapse.logging.context import LoggingContext, run_in_background
|
|
from synapse.rest import admin
|
|
from synapse.rest.client import login, room
|
|
from synapse.server import HomeServer
|
|
from synapse.storage.databases.main.events_worker import EventCacheEntry
|
|
from synapse.util import Clock
|
|
from synapse.util.events import generate_fake_event_id
|
|
|
|
from tests import unittest
|
|
from tests.test_utils import event_injection
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class FederationTestCase(unittest.FederatingHomeserverTestCase):
|
|
servlets = [
|
|
admin.register_servlets,
|
|
login.register_servlets,
|
|
room.register_servlets,
|
|
]
|
|
|
|
def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer:
|
|
hs = self.setup_test_homeserver()
|
|
self.handler = hs.get_federation_handler()
|
|
self.store = hs.get_datastores().main
|
|
return hs
|
|
|
|
def test_exchange_revoked_invite(self) -> None:
|
|
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)
|
|
|
|
# Send a 3PID invite event with an empty body so it's considered as a revoked one.
|
|
invite_token = "sometoken"
|
|
self.helper.send_state(
|
|
room_id=room_id,
|
|
event_type=EventTypes.ThirdPartyInvite,
|
|
state_key=invite_token,
|
|
body={},
|
|
tok=tok,
|
|
)
|
|
|
|
d = self.handler.on_exchange_third_party_invite_request(
|
|
event_dict={
|
|
"type": EventTypes.Member,
|
|
"room_id": room_id,
|
|
"sender": user_id,
|
|
"state_key": "@someone:example.org",
|
|
"content": {
|
|
"membership": "invite",
|
|
"third_party_invite": {
|
|
"display_name": "alice",
|
|
"signed": {
|
|
"mxid": "@alice:localhost",
|
|
"token": invite_token,
|
|
"signatures": {
|
|
"magic.forest": {
|
|
"ed25519:3": "fQpGIW1Snz+pwLZu6sTy2aHy/DYWWTspTJRPyNp0PKkymfIsNffysMl6ObMMFdIJhk6g6pwlIqZ54rxo8SLmAg"
|
|
}
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
)
|
|
|
|
failure = self.get_failure(d, AuthError).value
|
|
|
|
self.assertEqual(failure.code, 403, failure)
|
|
self.assertEqual(failure.errcode, Codes.FORBIDDEN, failure)
|
|
self.assertEqual(failure.msg, "You are not invited to this room.")
|
|
|
|
def test_rejected_message_event_state(self) -> None:
|
|
"""
|
|
Check that we store the state group correctly for rejected non-state events.
|
|
|
|
Regression test for https://github.com/matrix-org/synapse/issues/6289.
|
|
"""
|
|
OTHER_SERVER = "otherserver"
|
|
OTHER_USER = "@otheruser:" + OTHER_SERVER
|
|
|
|
# create the room
|
|
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 = self.get_success(self.store.get_room_version(room_id))
|
|
|
|
# pretend that another server has joined
|
|
join_event = self._build_and_send_join_event(OTHER_SERVER, OTHER_USER, room_id)
|
|
|
|
# check the state group
|
|
sg = self.get_success(
|
|
self.store._get_state_group_for_event(join_event.event_id)
|
|
)
|
|
|
|
# build and send an event which will be rejected
|
|
ev = event_from_pdu_json(
|
|
{
|
|
"type": EventTypes.Message,
|
|
"content": {},
|
|
"room_id": room_id,
|
|
"sender": "@yetanotheruser:" + OTHER_SERVER,
|
|
"depth": cast(int, join_event["depth"]) + 1,
|
|
"prev_events": [join_event.event_id],
|
|
"auth_events": [],
|
|
"origin_server_ts": self.clock.time_msec(),
|
|
},
|
|
room_version,
|
|
)
|
|
|
|
with LoggingContext("send_rejected"):
|
|
d = run_in_background(
|
|
self.hs.get_federation_event_handler().on_receive_pdu, OTHER_SERVER, ev
|
|
)
|
|
self.get_success(d)
|
|
|
|
# that should have been rejected
|
|
e = self.get_success(self.store.get_event(ev.event_id, allow_rejected=True))
|
|
self.assertIsNotNone(e.rejected_reason)
|
|
|
|
# ... and the state group should be the same as before
|
|
sg2 = self.get_success(self.store._get_state_group_for_event(ev.event_id))
|
|
|
|
self.assertEqual(sg, sg2)
|
|
|
|
def test_rejected_state_event_state(self) -> None:
|
|
"""
|
|
Check that we store the state group correctly for rejected state events.
|
|
|
|
Regression test for https://github.com/matrix-org/synapse/issues/6289.
|
|
"""
|
|
OTHER_SERVER = "otherserver"
|
|
OTHER_USER = "@otheruser:" + OTHER_SERVER
|
|
|
|
# create the room
|
|
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 = self.get_success(self.store.get_room_version(room_id))
|
|
|
|
# pretend that another server has joined
|
|
join_event = self._build_and_send_join_event(OTHER_SERVER, OTHER_USER, room_id)
|
|
|
|
# check the state group
|
|
sg = self.get_success(
|
|
self.store._get_state_group_for_event(join_event.event_id)
|
|
)
|
|
|
|
# build and send an event which will be rejected
|
|
ev = event_from_pdu_json(
|
|
{
|
|
"type": "org.matrix.test",
|
|
"state_key": "test_key",
|
|
"content": {},
|
|
"room_id": room_id,
|
|
"sender": "@yetanotheruser:" + OTHER_SERVER,
|
|
"depth": cast(int, join_event["depth"]) + 1,
|
|
"prev_events": [join_event.event_id],
|
|
"auth_events": [],
|
|
"origin_server_ts": self.clock.time_msec(),
|
|
},
|
|
room_version,
|
|
)
|
|
|
|
with LoggingContext("send_rejected"):
|
|
d = run_in_background(
|
|
self.hs.get_federation_event_handler().on_receive_pdu, OTHER_SERVER, ev
|
|
)
|
|
self.get_success(d)
|
|
|
|
# that should have been rejected
|
|
e = self.get_success(self.store.get_event(ev.event_id, allow_rejected=True))
|
|
self.assertIsNotNone(e.rejected_reason)
|
|
|
|
# ... and the state group should be the same as before
|
|
sg2 = self.get_success(self.store._get_state_group_for_event(ev.event_id))
|
|
|
|
self.assertEqual(sg, sg2)
|
|
|
|
def test_backfill_with_many_backward_extremities(self) -> None:
|
|
"""
|
|
Check that we can backfill with many backward extremities.
|
|
The goal is to make sure that when we only use a portion
|
|
of backwards extremities(the magic number is more than 5),
|
|
no errors are thrown.
|
|
|
|
Regression test, see https://github.com/matrix-org/synapse/pull/11027
|
|
"""
|
|
# create the room
|
|
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 = self.get_success(self.store.get_room_version(room_id))
|
|
|
|
# we need a user on the remote server to be a member, so that we can send
|
|
# extremity-causing events.
|
|
remote_server_user_id = f"@user:{self.OTHER_SERVER_NAME}"
|
|
self.get_success(
|
|
event_injection.inject_member_event(
|
|
self.hs, room_id, remote_server_user_id, "join"
|
|
)
|
|
)
|
|
|
|
send_result = self.helper.send(room_id, "first message", tok=tok)
|
|
ev1 = self.get_success(
|
|
self.store.get_event(send_result["event_id"], allow_none=False)
|
|
)
|
|
current_state = self.get_success(
|
|
self.store.get_events_as_list(
|
|
(
|
|
self.get_success(self.store.get_partial_current_state_ids(room_id))
|
|
).values()
|
|
)
|
|
)
|
|
|
|
# Create "many" backward extremities. The magic number we're trying to
|
|
# create more than is 5 which corresponds to the number of backward
|
|
# extremities we slice off in `_maybe_backfill_inner`
|
|
federation_event_handler = self.hs.get_federation_event_handler()
|
|
auth_events = [
|
|
ev
|
|
for ev in current_state
|
|
if (ev.type, ev.state_key)
|
|
in {("m.room.create", ""), ("m.room.member", remote_server_user_id)}
|
|
]
|
|
for _ in range(8):
|
|
event = make_event_from_dict(
|
|
self.add_hashes_and_signatures_from_other_server(
|
|
{
|
|
"origin_server_ts": 1,
|
|
"type": "m.room.message",
|
|
"content": {
|
|
"msgtype": "m.text",
|
|
"body": "message connected to fake event",
|
|
},
|
|
"room_id": room_id,
|
|
"sender": remote_server_user_id,
|
|
"prev_events": [
|
|
ev1.event_id,
|
|
# We're creating an backward extremity each time thanks
|
|
# to this fake event
|
|
generate_fake_event_id(),
|
|
],
|
|
"auth_events": [ev.event_id for ev in auth_events],
|
|
"depth": ev1.depth + 1,
|
|
},
|
|
room_version,
|
|
),
|
|
room_version,
|
|
)
|
|
|
|
# we poke this directly into _process_received_pdu, to avoid the
|
|
# federation handler wanting to backfill the fake event.
|
|
state_handler = self.hs.get_state_handler()
|
|
context = self.get_success(
|
|
state_handler.compute_event_context(
|
|
event,
|
|
state_ids_before_event={
|
|
(e.type, e.state_key): e.event_id for e in current_state
|
|
},
|
|
partial_state=False,
|
|
)
|
|
)
|
|
self.get_success(
|
|
federation_event_handler._process_received_pdu(
|
|
self.OTHER_SERVER_NAME,
|
|
event,
|
|
context,
|
|
)
|
|
)
|
|
|
|
# we should now have 8 backwards extremities.
|
|
backwards_extremities = self.get_success(
|
|
self.store.db_pool.simple_select_list(
|
|
"event_backward_extremities",
|
|
keyvalues={"room_id": room_id},
|
|
retcols=["event_id"],
|
|
)
|
|
)
|
|
self.assertEqual(len(backwards_extremities), 8)
|
|
|
|
current_depth = 1
|
|
limit = 100
|
|
with LoggingContext("receive_pdu"):
|
|
# Make sure backfill still works
|
|
d = run_in_background(
|
|
self.hs.get_federation_handler().maybe_backfill,
|
|
room_id,
|
|
current_depth,
|
|
limit,
|
|
)
|
|
self.get_success(d)
|
|
|
|
def test_backfill_ignores_known_events(self) -> None:
|
|
"""
|
|
Tests that events that we already know about are ignored when backfilling.
|
|
"""
|
|
# Set up users
|
|
user_id = self.register_user("kermit", "test")
|
|
tok = self.login("kermit", "test")
|
|
|
|
other_server = "otherserver"
|
|
other_user = "@otheruser:" + other_server
|
|
|
|
# Create a room to backfill events into
|
|
room_id = self.helper.create_room_as(room_creator=user_id, tok=tok)
|
|
room_version = self.get_success(self.store.get_room_version(room_id))
|
|
|
|
# Build an event to backfill
|
|
event = event_from_pdu_json(
|
|
{
|
|
"type": EventTypes.Message,
|
|
"content": {"body": "hello world", "msgtype": "m.text"},
|
|
"room_id": room_id,
|
|
"sender": other_user,
|
|
"depth": 32,
|
|
"prev_events": [],
|
|
"auth_events": [],
|
|
"origin_server_ts": self.clock.time_msec(),
|
|
},
|
|
room_version,
|
|
)
|
|
|
|
# Ensure the event is not already in the DB
|
|
self.get_failure(
|
|
self.store.get_event(event.event_id),
|
|
NotFoundError,
|
|
)
|
|
|
|
# Backfill the event and check that it has entered the DB.
|
|
|
|
# We mock out the FederationClient.backfill method, to pretend that a remote
|
|
# server has returned our fake event.
|
|
federation_client_backfill_mock = AsyncMock(return_value=[event])
|
|
self.hs.get_federation_client().backfill = federation_client_backfill_mock # type: ignore[method-assign]
|
|
|
|
# We also mock the persist method with a side effect of itself. This allows us
|
|
# to track when it has been called while preserving its function.
|
|
persist_events_and_notify_mock = Mock(
|
|
side_effect=self.hs.get_federation_event_handler().persist_events_and_notify
|
|
)
|
|
self.hs.get_federation_event_handler().persist_events_and_notify = ( # type: ignore[method-assign]
|
|
persist_events_and_notify_mock
|
|
)
|
|
|
|
# Small side-tangent. We populate the event cache with the event, even though
|
|
# it is not yet in the DB. This is an invalid scenario that can currently occur
|
|
# due to not properly invalidating the event cache.
|
|
# See https://github.com/matrix-org/synapse/issues/13476.
|
|
#
|
|
# As a result, backfill should not rely on the event cache to check whether
|
|
# we already have an event in the DB.
|
|
# TODO: Remove this bit when the event cache is properly invalidated.
|
|
cache_entry = EventCacheEntry(
|
|
event=event,
|
|
redacted_event=None,
|
|
)
|
|
self.store._get_event_cache.set_local((event.event_id,), cache_entry)
|
|
|
|
# We now call FederationEventHandler.backfill (a separate method) to trigger
|
|
# a backfill request. It should receive the fake event.
|
|
self.get_success(
|
|
self.hs.get_federation_event_handler().backfill(
|
|
other_user,
|
|
room_id,
|
|
limit=10,
|
|
extremities=[],
|
|
)
|
|
)
|
|
|
|
# Check that our fake event was persisted.
|
|
persist_events_and_notify_mock.assert_called_once()
|
|
persist_events_and_notify_mock.reset_mock()
|
|
|
|
# Now we repeat the backfill, having the homeserver receive the fake event
|
|
# again.
|
|
self.get_success(
|
|
self.hs.get_federation_event_handler().backfill(
|
|
other_user,
|
|
room_id,
|
|
limit=10,
|
|
extremities=[],
|
|
),
|
|
)
|
|
|
|
# This time, we expect no event persistence to have occurred, as we already
|
|
# have this event.
|
|
persist_events_and_notify_mock.assert_not_called()
|
|
|
|
@unittest.override_config(
|
|
{"rc_invites": {"per_user": {"per_second": 0.5, "burst_count": 3}}}
|
|
)
|
|
def test_invite_by_user_ratelimit(self) -> None:
|
|
"""Tests that invites from federation to a particular user are
|
|
actually rate-limited.
|
|
"""
|
|
other_server = "otherserver"
|
|
other_user = "@otheruser:" + other_server
|
|
|
|
# create the room
|
|
user_id = self.register_user("kermit", "test")
|
|
tok = self.login("kermit", "test")
|
|
|
|
def create_invite() -> EventBase:
|
|
room_id = self.helper.create_room_as(room_creator=user_id, tok=tok)
|
|
room_version = self.get_success(self.store.get_room_version(room_id))
|
|
return event_from_pdu_json(
|
|
{
|
|
"type": EventTypes.Member,
|
|
"content": {"membership": "invite"},
|
|
"room_id": room_id,
|
|
"sender": other_user,
|
|
"state_key": "@user:test",
|
|
"depth": 32,
|
|
"prev_events": [],
|
|
"auth_events": [],
|
|
"origin_server_ts": self.clock.time_msec(),
|
|
},
|
|
room_version,
|
|
)
|
|
|
|
for _ in range(3):
|
|
event = create_invite()
|
|
self.get_success(
|
|
self.handler.on_invite_request(
|
|
other_server,
|
|
event,
|
|
event.room_version,
|
|
)
|
|
)
|
|
|
|
event = create_invite()
|
|
self.get_failure(
|
|
self.handler.on_invite_request(
|
|
other_server,
|
|
event,
|
|
event.room_version,
|
|
),
|
|
exc=LimitExceededError,
|
|
by=0.5,
|
|
)
|
|
|
|
def _build_and_send_join_event(
|
|
self, other_server: str, other_user: str, room_id: str
|
|
) -> EventBase:
|
|
join_event = self.get_success(
|
|
self.handler.on_make_join_request(other_server, room_id, other_user)
|
|
)
|
|
# the auth code requires that a signature exists, but doesn't check that
|
|
# signature... go figure.
|
|
join_event.signatures[other_server] = {"x": "y"}
|
|
with LoggingContext("send_join"):
|
|
d = run_in_background(
|
|
self.hs.get_federation_event_handler().on_send_membership_event,
|
|
other_server,
|
|
join_event,
|
|
)
|
|
self.get_success(d)
|
|
|
|
# sanity-check: the room should show that the new user is a member
|
|
r = self.get_success(self.store.get_partial_current_state_ids(room_id))
|
|
self.assertEqual(r[(EventTypes.Member, other_user)], join_event.event_id)
|
|
|
|
return join_event
|
|
|
|
|
|
class EventFromPduTestCase(TestCase):
|
|
def test_valid_json(self) -> None:
|
|
"""Valid JSON should be turned into an event."""
|
|
ev = event_from_pdu_json(
|
|
{
|
|
"type": EventTypes.Message,
|
|
"content": {"bool": True, "null": None, "int": 1, "str": "foobar"},
|
|
"room_id": "!room:test",
|
|
"sender": "@user:test",
|
|
"depth": 1,
|
|
"prev_events": [],
|
|
"auth_events": [],
|
|
"origin_server_ts": 1234,
|
|
},
|
|
RoomVersions.V6,
|
|
)
|
|
|
|
self.assertIsInstance(ev, EventBase)
|
|
|
|
def test_invalid_numbers(self) -> None:
|
|
"""Invalid values for an integer should be rejected, all floats should be rejected."""
|
|
for value in [
|
|
-(2**53),
|
|
2**53,
|
|
1.0,
|
|
float("inf"),
|
|
float("-inf"),
|
|
float("nan"),
|
|
]:
|
|
with self.assertRaises(SynapseError):
|
|
event_from_pdu_json(
|
|
{
|
|
"type": EventTypes.Message,
|
|
"content": {"foo": value},
|
|
"room_id": "!room:test",
|
|
"sender": "@user:test",
|
|
"depth": 1,
|
|
"prev_events": [],
|
|
"auth_events": [],
|
|
"origin_server_ts": 1234,
|
|
},
|
|
RoomVersions.V6,
|
|
)
|
|
|
|
def test_invalid_nested(self) -> None:
|
|
"""List and dictionaries are recursively searched."""
|
|
with self.assertRaises(SynapseError):
|
|
event_from_pdu_json(
|
|
{
|
|
"type": EventTypes.Message,
|
|
"content": {"foo": [{"bar": 2**56}]},
|
|
"room_id": "!room:test",
|
|
"sender": "@user:test",
|
|
"depth": 1,
|
|
"prev_events": [],
|
|
"auth_events": [],
|
|
"origin_server_ts": 1234,
|
|
},
|
|
RoomVersions.V6,
|
|
)
|
|
|
|
|
|
class PartialJoinTestCase(unittest.FederatingHomeserverTestCase):
|
|
def test_failed_partial_join_is_clean(self) -> None:
|
|
"""
|
|
Tests that, when failing to partial-join a room, we don't get stuck with
|
|
a partial-state flag on a room.
|
|
"""
|
|
|
|
fed_handler = self.hs.get_federation_handler()
|
|
fed_client = fed_handler.federation_client
|
|
|
|
room_id = "!room:example.com"
|
|
|
|
EVENT_CREATE = make_event_from_dict(
|
|
{
|
|
"room_id": room_id,
|
|
"type": "m.room.create",
|
|
"sender": "@kristina:example.com",
|
|
"state_key": "",
|
|
"depth": 0,
|
|
"content": {"creator": "@kristina:example.com", "room_version": "10"},
|
|
"auth_events": [],
|
|
"origin_server_ts": 1,
|
|
},
|
|
room_version=RoomVersions.V10,
|
|
)
|
|
EVENT_CREATOR_MEMBERSHIP = make_event_from_dict(
|
|
{
|
|
"room_id": room_id,
|
|
"type": "m.room.member",
|
|
"sender": "@kristina:example.com",
|
|
"state_key": "@kristina:example.com",
|
|
"content": {"membership": "join"},
|
|
"depth": 1,
|
|
"prev_events": [EVENT_CREATE.event_id],
|
|
"auth_events": [EVENT_CREATE.event_id],
|
|
"origin_server_ts": 1,
|
|
},
|
|
room_version=RoomVersions.V10,
|
|
)
|
|
EVENT_INVITATION_MEMBERSHIP = make_event_from_dict(
|
|
{
|
|
"room_id": room_id,
|
|
"type": "m.room.member",
|
|
"sender": "@kristina:example.com",
|
|
"state_key": "@alice:test",
|
|
"content": {"membership": "invite"},
|
|
"depth": 2,
|
|
"prev_events": [EVENT_CREATOR_MEMBERSHIP.event_id],
|
|
"auth_events": [
|
|
EVENT_CREATE.event_id,
|
|
EVENT_CREATOR_MEMBERSHIP.event_id,
|
|
],
|
|
"origin_server_ts": 1,
|
|
},
|
|
room_version=RoomVersions.V10,
|
|
)
|
|
membership_event = make_event_from_dict(
|
|
{
|
|
"room_id": room_id,
|
|
"type": "m.room.member",
|
|
"sender": "@alice:test",
|
|
"state_key": "@alice:test",
|
|
"content": {"membership": "join"},
|
|
"prev_events": [EVENT_INVITATION_MEMBERSHIP.event_id],
|
|
},
|
|
RoomVersions.V10,
|
|
)
|
|
mock_make_membership_event = AsyncMock(
|
|
return_value=(
|
|
"example.com",
|
|
membership_event,
|
|
RoomVersions.V10,
|
|
)
|
|
)
|
|
mock_send_join = AsyncMock(
|
|
return_value=SendJoinResult(
|
|
membership_event,
|
|
"example.com",
|
|
state=[
|
|
EVENT_CREATE,
|
|
EVENT_CREATOR_MEMBERSHIP,
|
|
EVENT_INVITATION_MEMBERSHIP,
|
|
],
|
|
auth_chain=[
|
|
EVENT_CREATE,
|
|
EVENT_CREATOR_MEMBERSHIP,
|
|
EVENT_INVITATION_MEMBERSHIP,
|
|
],
|
|
partial_state=True,
|
|
servers_in_room={"example.com"},
|
|
)
|
|
)
|
|
|
|
with (
|
|
patch.object(
|
|
fed_client, "make_membership_event", mock_make_membership_event
|
|
),
|
|
patch.object(fed_client, "send_join", mock_send_join),
|
|
):
|
|
# Join and check that our join event is rejected
|
|
# (The join event is rejected because it doesn't have any signatures)
|
|
join_exc = self.get_failure(
|
|
fed_handler.do_invite_join(["example.com"], room_id, "@alice:test", {}),
|
|
SynapseError,
|
|
)
|
|
self.assertIn("Join event was rejected", str(join_exc))
|
|
|
|
store = self.hs.get_datastores().main
|
|
|
|
# Check that we don't have a left-over partial_state entry.
|
|
self.assertFalse(
|
|
self.get_success(store.is_partial_state_room(room_id)),
|
|
f"Stale partial-stated room flag left over for {room_id} after a"
|
|
f" failed do_invite_join!",
|
|
)
|
|
|
|
def test_duplicate_partial_state_room_syncs(self) -> None:
|
|
"""
|
|
Tests that concurrent partial state syncs are not started for the same room.
|
|
"""
|
|
is_partial_state = True
|
|
end_sync: "Deferred[None]" = Deferred()
|
|
|
|
async def is_partial_state_room(room_id: str) -> bool:
|
|
return is_partial_state
|
|
|
|
async def sync_partial_state_room(
|
|
initial_destination: Optional[str],
|
|
other_destinations: Collection[str],
|
|
room_id: str,
|
|
) -> None:
|
|
nonlocal end_sync
|
|
try:
|
|
await end_sync
|
|
finally:
|
|
end_sync = Deferred()
|
|
|
|
mock_is_partial_state_room = Mock(side_effect=is_partial_state_room)
|
|
mock_sync_partial_state_room = Mock(side_effect=sync_partial_state_room)
|
|
|
|
fed_handler = self.hs.get_federation_handler()
|
|
store = self.hs.get_datastores().main
|
|
|
|
with (
|
|
patch.object(
|
|
fed_handler, "_sync_partial_state_room", mock_sync_partial_state_room
|
|
),
|
|
patch.object(store, "is_partial_state_room", mock_is_partial_state_room),
|
|
):
|
|
# Start the partial state sync.
|
|
fed_handler._start_partial_state_room_sync("hs1", {"hs2"}, "room_id")
|
|
self.assertEqual(mock_sync_partial_state_room.call_count, 1)
|
|
|
|
# Try to start another partial state sync.
|
|
# Nothing should happen.
|
|
fed_handler._start_partial_state_room_sync("hs3", {"hs2"}, "room_id")
|
|
self.assertEqual(mock_sync_partial_state_room.call_count, 1)
|
|
|
|
# End the partial state sync
|
|
is_partial_state = False
|
|
end_sync.callback(None)
|
|
|
|
# The partial state sync should not be restarted.
|
|
self.assertEqual(mock_sync_partial_state_room.call_count, 1)
|
|
|
|
# The next attempt to start the partial state sync should work.
|
|
is_partial_state = True
|
|
fed_handler._start_partial_state_room_sync("hs3", {"hs2"}, "room_id")
|
|
self.assertEqual(mock_sync_partial_state_room.call_count, 2)
|
|
|
|
def test_partial_state_room_sync_restart(self) -> None:
|
|
"""
|
|
Tests that partial state syncs are restarted when a second partial state sync
|
|
was deduplicated and the first partial state sync fails.
|
|
"""
|
|
is_partial_state = True
|
|
end_sync: "Deferred[None]" = Deferred()
|
|
|
|
async def is_partial_state_room(room_id: str) -> bool:
|
|
return is_partial_state
|
|
|
|
async def sync_partial_state_room(
|
|
initial_destination: Optional[str],
|
|
other_destinations: Collection[str],
|
|
room_id: str,
|
|
) -> None:
|
|
nonlocal end_sync
|
|
try:
|
|
await end_sync
|
|
finally:
|
|
end_sync = Deferred()
|
|
|
|
mock_is_partial_state_room = Mock(side_effect=is_partial_state_room)
|
|
mock_sync_partial_state_room = Mock(side_effect=sync_partial_state_room)
|
|
|
|
fed_handler = self.hs.get_federation_handler()
|
|
store = self.hs.get_datastores().main
|
|
|
|
with (
|
|
patch.object(
|
|
fed_handler, "_sync_partial_state_room", mock_sync_partial_state_room
|
|
),
|
|
patch.object(store, "is_partial_state_room", mock_is_partial_state_room),
|
|
):
|
|
# Start the partial state sync.
|
|
fed_handler._start_partial_state_room_sync("hs1", {"hs2"}, "room_id")
|
|
self.assertEqual(mock_sync_partial_state_room.call_count, 1)
|
|
|
|
# Fail the partial state sync.
|
|
# The partial state sync should not be restarted.
|
|
end_sync.errback(Exception("Failed to request /state_ids"))
|
|
self.assertEqual(mock_sync_partial_state_room.call_count, 1)
|
|
|
|
# Start the partial state sync again.
|
|
fed_handler._start_partial_state_room_sync("hs1", {"hs2"}, "room_id")
|
|
self.assertEqual(mock_sync_partial_state_room.call_count, 2)
|
|
|
|
# Deduplicate another partial state sync.
|
|
fed_handler._start_partial_state_room_sync("hs3", {"hs2"}, "room_id")
|
|
self.assertEqual(mock_sync_partial_state_room.call_count, 2)
|
|
|
|
# Fail the partial state sync.
|
|
# It should restart with the latest parameters.
|
|
end_sync.errback(Exception("Failed to request /state_ids"))
|
|
self.assertEqual(mock_sync_partial_state_room.call_count, 3)
|
|
mock_sync_partial_state_room.assert_called_with(
|
|
initial_destination="hs3",
|
|
other_destinations={"hs2"},
|
|
room_id="room_id",
|
|
)
|