Rate limit joins per-room (#13276)

6 files changed, 358 insertions, 6 deletions

diff --git a/tests/federation/test_federation_server.py b/tests/federation/test_federation_server.py
index 8ea13ceb93..3a6ef221ae 100644
--- a/tests/federation/test_federation_server.py
+++ b/tests/federation/test_federation_server.py
@@ -148,7 +148,7 @@ class SendJoinFederationTests(unittest.FederatingHomeserverTestCase):
         tok2 = self.login("fozzie", "bear")
         self.helper.join(self._room_id, second_member_user_id, tok=tok2)
 
-    def _make_join(self, user_id) -> JsonDict:
+    def _make_join(self, user_id: str) -> JsonDict:
         channel = self.make_signed_federation_request(
             "GET",
             f"/_matrix/federation/v1/make_join/{self._room_id}/{user_id}"
@@ -260,6 +260,67 @@ class SendJoinFederationTests(unittest.FederatingHomeserverTestCase):
         )
         self.assertEqual(r[("m.room.member", joining_user)].membership, "join")
 
+    @override_config({"rc_joins_per_room": {"per_second": 0, "burst_count": 3}})
+    def test_make_join_respects_room_join_rate_limit(self) -> None:
+        # In the test setup, two users join the room. Since the rate limiter burst
+        # count is 3, a new make_join request to the room should be accepted.
+
+        joining_user = "@ronniecorbett:" + self.OTHER_SERVER_NAME
+        self._make_join(joining_user)
+
+        # Now have a new local user join the room. This saturates the rate limiter
+        # bucket, so the next make_join should be denied.
+        new_local_user = self.register_user("animal", "animal")
+        token = self.login("animal", "animal")
+        self.helper.join(self._room_id, new_local_user, tok=token)
+
+        joining_user = "@ronniebarker:" + self.OTHER_SERVER_NAME
+        channel = self.make_signed_federation_request(
+            "GET",
+            f"/_matrix/federation/v1/make_join/{self._room_id}/{joining_user}"
+            f"?ver={DEFAULT_ROOM_VERSION}",
+        )
+        self.assertEqual(channel.code, HTTPStatus.TOO_MANY_REQUESTS, channel.json_body)
+
+    @override_config({"rc_joins_per_room": {"per_second": 0, "burst_count": 3}})
+    def test_send_join_contributes_to_room_join_rate_limit_and_is_limited(self) -> None:
+        # Make two make_join requests up front. (These are rate limited, but do not
+        # contribute to the rate limit.)
+        join_event_dicts = []
+        for i in range(2):
+            joining_user = f"@misspiggy{i}:{self.OTHER_SERVER_NAME}"
+            join_result = self._make_join(joining_user)
+            join_event_dict = join_result["event"]
+            self.add_hashes_and_signatures_from_other_server(
+                join_event_dict,
+                KNOWN_ROOM_VERSIONS[DEFAULT_ROOM_VERSION],
+            )
+            join_event_dicts.append(join_event_dict)
+
+        # In the test setup, two users join the room. Since the rate limiter burst
+        # count is 3, the first send_join should be accepted...
+        channel = self.make_signed_federation_request(
+            "PUT",
+            f"/_matrix/federation/v2/send_join/{self._room_id}/join0",
+            content=join_event_dicts[0],
+        )
+        self.assertEqual(channel.code, 200, channel.json_body)
+
+        # ... but the second should be denied.
+        channel = self.make_signed_federation_request(
+            "PUT",
+            f"/_matrix/federation/v2/send_join/{self._room_id}/join1",
+            content=join_event_dicts[1],
+        )
+        self.assertEqual(channel.code, HTTPStatus.TOO_MANY_REQUESTS, channel.json_body)
+
+    # NB: we could write a test which checks that the send_join event is seen
+    #   by other workers over replication, and that they update their rate limit
+    #   buckets accordingly. I'm going to assume that the join event gets sent over
+    #   replication, at which point the tests.handlers.room_member test
+    #       test_local_users_joining_on_another_worker_contribute_to_rate_limit
+    #   is probably sufficient to reassure that the bucket is updated.
+
 
 def _create_acl_event(content):
     return make_event_from_dict(
diff --git a/tests/handlers/test_room_member.py b/tests/handlers/test_room_member.py
new file mode 100644
index 0000000000..254e7e4b80
--- /dev/null
+++ b/tests/handlers/test_room_member.py
@@ -0,0 +1,290 @@
+from http import HTTPStatus
+from unittest.mock import Mock, patch
+
+from twisted.test.proto_helpers import MemoryReactor
+
+import synapse.rest.admin
+import synapse.rest.client.login
+import synapse.rest.client.room
+from synapse.api.constants import EventTypes, Membership
+from synapse.api.errors import LimitExceededError
+from synapse.crypto.event_signing import add_hashes_and_signatures
+from synapse.events import FrozenEventV3
+from synapse.federation.federation_client import SendJoinResult
+from synapse.server import HomeServer
+from synapse.types import UserID, create_requester
+from synapse.util import Clock
+
+from tests.replication._base import RedisMultiWorkerStreamTestCase
+from tests.server import make_request
+from tests.test_utils import make_awaitable
+from tests.unittest import FederatingHomeserverTestCase, override_config
+
+
+class TestJoinsLimitedByPerRoomRateLimiter(FederatingHomeserverTestCase):
+    servlets = [
+        synapse.rest.admin.register_servlets,
+        synapse.rest.client.login.register_servlets,
+        synapse.rest.client.room.register_servlets,
+    ]
+
+    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
+        self.handler = hs.get_room_member_handler()
+
+        # Create three users.
+        self.alice = self.register_user("alice", "pass")
+        self.alice_token = self.login("alice", "pass")
+        self.bob = self.register_user("bob", "pass")
+        self.bob_token = self.login("bob", "pass")
+        self.chris = self.register_user("chris", "pass")
+        self.chris_token = self.login("chris", "pass")
+
+        # Create a room on this homeserver. Note that this counts as a join: it
+        # contributes to the rate limter's count of actions
+        self.room_id = self.helper.create_room_as(self.alice, tok=self.alice_token)
+
+        self.intially_unjoined_room_id = f"!example:{self.OTHER_SERVER_NAME}"
+
+    @override_config({"rc_joins_per_room": {"per_second": 0, "burst_count": 2}})
+    def test_local_user_local_joins_contribute_to_limit_and_are_limited(self) -> None:
+        # The rate limiter has accumulated one token from Alice's join after the create
+        # event.
+        # Try joining the room as Bob.
+        self.get_success(
+            self.handler.update_membership(
+                requester=create_requester(self.bob),
+                target=UserID.from_string(self.bob),
+                room_id=self.room_id,
+                action=Membership.JOIN,
+            )
+        )
+
+        # The rate limiter bucket is full. A second join should be denied.
+        self.get_failure(
+            self.handler.update_membership(
+                requester=create_requester(self.chris),
+                target=UserID.from_string(self.chris),
+                room_id=self.room_id,
+                action=Membership.JOIN,
+            ),
+            LimitExceededError,
+        )
+
+    @override_config({"rc_joins_per_room": {"per_second": 0, "burst_count": 2}})
+    def test_local_user_profile_edits_dont_contribute_to_limit(self) -> None:
+        # The rate limiter has accumulated one token from Alice's join after the create
+        # event. Alice should still be able to change her displayname.
+        self.get_success(
+            self.handler.update_membership(
+                requester=create_requester(self.alice),
+                target=UserID.from_string(self.alice),
+                room_id=self.room_id,
+                action=Membership.JOIN,
+                content={"displayname": "Alice Cooper"},
+            )
+        )
+
+        # Still room in the limiter bucket. Chris's join should be accepted.
+        self.get_success(
+            self.handler.update_membership(
+                requester=create_requester(self.chris),
+                target=UserID.from_string(self.chris),
+                room_id=self.room_id,
+                action=Membership.JOIN,
+            )
+        )
+
+    @override_config({"rc_joins_per_room": {"per_second": 0, "burst_count": 1}})
+    def test_remote_joins_contribute_to_rate_limit(self) -> None:
+        # Join once, to fill the rate limiter bucket.
+        #
+        # To do this we have to mock the responses from the remote homeserver.
+        # We also patch out a bunch of event checks on our end. All we're really
+        # trying to check here is that remote joins will bump the rate limter when
+        # they are persisted.
+        create_event_source = {
+            "auth_events": [],
+            "content": {
+                "creator": f"@creator:{self.OTHER_SERVER_NAME}",
+                "room_version": self.hs.config.server.default_room_version.identifier,
+            },
+            "depth": 0,
+            "origin_server_ts": 0,
+            "prev_events": [],
+            "room_id": self.intially_unjoined_room_id,
+            "sender": f"@creator:{self.OTHER_SERVER_NAME}",
+            "state_key": "",
+            "type": EventTypes.Create,
+        }
+        self.add_hashes_and_signatures_from_other_server(
+            create_event_source,
+            self.hs.config.server.default_room_version,
+        )
+        create_event = FrozenEventV3(
+            create_event_source,
+            self.hs.config.server.default_room_version,
+            {},
+            None,
+        )
+
+        join_event_source = {
+            "auth_events": [create_event.event_id],
+            "content": {"membership": "join"},
+            "depth": 1,
+            "origin_server_ts": 100,
+            "prev_events": [create_event.event_id],
+            "sender": self.bob,
+            "state_key": self.bob,
+            "room_id": self.intially_unjoined_room_id,
+            "type": EventTypes.Member,
+        }
+        add_hashes_and_signatures(
+            self.hs.config.server.default_room_version,
+            join_event_source,
+            self.hs.hostname,
+            self.hs.signing_key,
+        )
+        join_event = FrozenEventV3(
+            join_event_source,
+            self.hs.config.server.default_room_version,
+            {},
+            None,
+        )
+
+        mock_make_membership_event = Mock(
+            return_value=make_awaitable(
+                (
+                    self.OTHER_SERVER_NAME,
+                    join_event,
+                    self.hs.config.server.default_room_version,
+                )
+            )
+        )
+        mock_send_join = Mock(
+            return_value=make_awaitable(
+                SendJoinResult(
+                    join_event,
+                    self.OTHER_SERVER_NAME,
+                    state=[create_event],
+                    auth_chain=[create_event],
+                    partial_state=False,
+                    servers_in_room=[],
+                )
+            )
+        )
+
+        with patch.object(
+            self.handler.federation_handler.federation_client,
+            "make_membership_event",
+            mock_make_membership_event,
+        ), patch.object(
+            self.handler.federation_handler.federation_client,
+            "send_join",
+            mock_send_join,
+        ), patch(
+            "synapse.event_auth._is_membership_change_allowed",
+            return_value=None,
+        ), patch(
+            "synapse.handlers.federation_event.check_state_dependent_auth_rules",
+            return_value=None,
+        ):
+            self.get_success(
+                self.handler.update_membership(
+                    requester=create_requester(self.bob),
+                    target=UserID.from_string(self.bob),
+                    room_id=self.intially_unjoined_room_id,
+                    action=Membership.JOIN,
+                    remote_room_hosts=[self.OTHER_SERVER_NAME],
+                )
+            )
+
+            # Try to join as Chris. Should get denied.
+            self.get_failure(
+                self.handler.update_membership(
+                    requester=create_requester(self.chris),
+                    target=UserID.from_string(self.chris),
+                    room_id=self.intially_unjoined_room_id,
+                    action=Membership.JOIN,
+                    remote_room_hosts=[self.OTHER_SERVER_NAME],
+                ),
+                LimitExceededError,
+            )
+
+    # TODO: test that remote joins to a room are rate limited.
+    #   Could do this by setting the burst count to 1, then:
+    #   - remote-joining a room
+    #   - immediately leaving
+    #   - trying to remote-join again.
+
+
+class TestReplicatedJoinsLimitedByPerRoomRateLimiter(RedisMultiWorkerStreamTestCase):
+    servlets = [
+        synapse.rest.admin.register_servlets,
+        synapse.rest.client.login.register_servlets,
+        synapse.rest.client.room.register_servlets,
+    ]
+
+    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
+        self.handler = hs.get_room_member_handler()
+
+        # Create three users.
+        self.alice = self.register_user("alice", "pass")
+        self.alice_token = self.login("alice", "pass")
+        self.bob = self.register_user("bob", "pass")
+        self.bob_token = self.login("bob", "pass")
+        self.chris = self.register_user("chris", "pass")
+        self.chris_token = self.login("chris", "pass")
+
+        # Create a room on this homeserver.
+        # Note that this counts as a
+        self.room_id = self.helper.create_room_as(self.alice, tok=self.alice_token)
+        self.intially_unjoined_room_id = "!example:otherhs"
+
+    @override_config({"rc_joins_per_room": {"per_second": 0, "burst_count": 2}})
+    def test_local_users_joining_on_another_worker_contribute_to_rate_limit(
+        self,
+    ) -> None:
+        # The rate limiter has accumulated one token from Alice's join after the create
+        # event.
+        self.replicate()
+
+        # Spawn another worker and have bob join via it.
+        worker_app = self.make_worker_hs(
+            "synapse.app.generic_worker", extra_config={"worker_name": "other worker"}
+        )
+        worker_site = self._hs_to_site[worker_app]
+        channel = make_request(
+            self.reactor,
+            worker_site,
+            "POST",
+            f"/_matrix/client/v3/rooms/{self.room_id}/join",
+            access_token=self.bob_token,
+        )
+        self.assertEqual(channel.code, HTTPStatus.OK, channel.json_body)
+
+        # wait for join to arrive over replication
+        self.replicate()
+
+        # Try to join as Chris on the worker. Should get denied because Alice
+        # and Bob have both joined the room.
+        self.get_failure(
+            worker_app.get_room_member_handler().update_membership(
+                requester=create_requester(self.chris),
+                target=UserID.from_string(self.chris),
+                room_id=self.room_id,
+                action=Membership.JOIN,
+            ),
+            LimitExceededError,
+        )
+
+        # Try to join as Chris on the original worker. Should get denied because Alice
+        # and Bob have both joined the room.
+        self.get_failure(
+            self.handler.update_membership(
+                requester=create_requester(self.chris),
+                target=UserID.from_string(self.chris),
+                room_id=self.room_id,
+                action=Membership.JOIN,
+            ),
+            LimitExceededError,
+        )
diff --git a/tests/rest/client/test_rooms.py b/tests/rest/client/test_rooms.py
index 17571b2d33..c45cb32090 100644
--- a/tests/rest/client/test_rooms.py
+++ b/tests/rest/client/test_rooms.py
@@ -710,7 +710,7 @@ class RoomsCreateTestCase(RoomBase):
         self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
         self.assertTrue("room_id" in channel.json_body)
         assert channel.resource_usage is not None
-        self.assertEqual(43, channel.resource_usage.db_txn_count)
+        self.assertEqual(44, channel.resource_usage.db_txn_count)
 
     def test_post_room_initial_state(self) -> None:
         # POST with initial_state config key, expect new room id
@@ -723,7 +723,7 @@ class RoomsCreateTestCase(RoomBase):
         self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
         self.assertTrue("room_id" in channel.json_body)
         assert channel.resource_usage is not None
-        self.assertEqual(49, channel.resource_usage.db_txn_count)
+        self.assertEqual(50, channel.resource_usage.db_txn_count)
 
     def test_post_room_visibility_key(self) -> None:
         # POST with visibility config key, expect new room id
diff --git a/tests/test_server.py b/tests/test_server.py
index fc4bce899c..2fe4411401 100644
--- a/tests/test_server.py
+++ b/tests/test_server.py
@@ -231,7 +231,7 @@ class OptionsResourceTests(unittest.TestCase):
             parse_listener_def({"type": "http", "port": 0}),
             self.resource,
             "1.0",
-            max_request_body_size=1234,
+            max_request_body_size=4096,
             reactor=self.reactor,
         )
 
diff --git a/tests/unittest.py b/tests/unittest.py
index 9f1ff774a8..66ce92f4a6 100644
--- a/tests/unittest.py
+++ b/tests/unittest.py
@@ -284,7 +284,7 @@ class HomeserverTestCase(TestCase):
             config=self.hs.config.server.listeners[0],
             resource=self.resource,
             server_version_string="1",
-            max_request_body_size=1234,
+            max_request_body_size=4096,
             reactor=self.reactor,
         )
 
@@ -773,7 +773,7 @@ class FederatingHomeserverTestCase(HomeserverTestCase):
                         verify_key_id,
                         FetchKeyResult(
                             verify_key=verify_key,
-                            valid_until_ts=clock.time_msec() + 1000,
+                            valid_until_ts=clock.time_msec() + 10000,
                         ),
                     )
                 ],
diff --git a/tests/utils.py b/tests/utils.py
index 424cc4c2a0..d2c6d1e852 100644
--- a/tests/utils.py
+++ b/tests/utils.py
@@ -167,6 +167,7 @@ def default_config(
             "local": {"per_second": 10000, "burst_count": 10000},
             "remote": {"per_second": 10000, "burst_count": 10000},
         },
+        "rc_joins_per_room": {"per_second": 10000, "burst_count": 10000},
         "rc_invites": {
             "per_room": {"per_second": 10000, "burst_count": 10000},
             "per_user": {"per_second": 10000, "burst_count": 10000},