# # 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: # . # # Originally licensed under the Apache License, Version 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", )