Add configurable rate limiting

synapse/config/experimental.py

@@ -28,6 +28,7 @@ import attr
 from synapse.api.room_versions import KNOWN_ROOM_VERSIONS, RoomVersions
 from synapse.config import ConfigError
 from synapse.config._base import Config, read_file
+from synapse.config.ratelimiting import RatelimitSettings
 from synapse.types import JsonDict, StrSequence
 
 
@@ -247,6 +248,29 @@ class ExperimentalConfig(Config):
                 ("experimental", "msc4388_enabled"),
             )
 
+        # MSC4388 requests are ratelimited by client IP address, as most are
+        # unauthenticated: only discovery and creating a session may carry an
+        # access token, and even then it is not used for ratelimiting.
+        #
+        # Creating a session is limited more strictly than other requests, as
+        # there is a cap on the number of concurrent sessions, and creating new
+        # sessions beyond it evicts the oldest ones. A client using these
+        # defaults can hold at most around 20 live sessions at once, as each
+        # session expires after 2 minutes.
+        self.msc4388_rc_create = RatelimitSettings.parse(
+            experimental,
+            "msc4388_rc_create",
+            defaults={"per_second": 0.1, "burst_count": 10},
+        )
+        # Every other rendezvous request: discovery, and receiving, sending or
+        # cancelling on a session. This needs to allow for both devices polling
+        # for new payloads, possibly from behind the same IP address.
+        self.msc4388_rc_requests = RatelimitSettings.parse(
+            experimental,
+            "msc4388_rc_requests",
+            defaults={"per_second": 5, "burst_count": 20},
+        )
+
         # MSC4133: Custom profile fields
         self.msc4133_enabled: bool = experimental.get("msc4133_enabled", False)
 

synapse/rest/client/rendezvous.py

@@ -25,6 +25,7 @@ from http.client import TEMPORARY_REDIRECT
 from typing import TYPE_CHECKING, Any
 
 from synapse.api.errors import Codes, SynapseError
+from synapse.api.ratelimiting import Ratelimiter
 from synapse.http.server import HttpServer, respond_with_redirect
 from synapse.http.servlet import RestServlet
 from synapse.http.site import SynapseRequest
@@ -75,13 +76,20 @@ class MSC4388CreateRendezvousServlet(RestServlet):
         "/io.element.msc4388/rendezvous$", releases=[], v1=False, unstable=True
     )
 
-    def __init__(self, hs: "HomeServer") -> None:
+    def __init__(
+        self,
+        hs: "HomeServer",
+        create_ratelimiter: Ratelimiter,
+        requests_ratelimiter: Ratelimiter,
+    ) -> None:
         super().__init__()
         self._handler = hs.get_msc4388_rendezvous_handler()
         self.auth = hs.get_auth()
         self.require_authentication = (
             hs.config.experimental.msc4388_requires_authentication
         )
+        self._create_ratelimiter = create_ratelimiter
+        self._requests_ratelimiter = requests_ratelimiter
 
     async def _can_create(self, request: SynapseRequest) -> bool:
         """Whether the requester is permitted by server policy to create a
@@ -102,11 +110,15 @@ class MSC4388CreateRendezvousServlet(RestServlet):
         return not self.require_authentication
 
     async def on_GET(self, request: SynapseRequest) -> tuple[int, Any]:
+        await self._requests_ratelimiter.ratelimit(
+            None, request.getClientAddress().host
+        )
         # A requester which is not permitted to create a session is told so in
         # the response, rather than by an error.
         return HTTPStatus.OK, {"create_available": await self._can_create(request)}
 
     async def on_POST(self, request: SynapseRequest) -> tuple[int, Any]:
+        await self._create_ratelimiter.ratelimit(None, request.getClientAddress().host)
         if not await self._can_create(request):
             raise SynapseError(
                 HTTPStatus.FORBIDDEN,
@@ -124,16 +136,25 @@ class MSC4388UpdateRendezvousServlet(RestServlet):
         unstable=True,
     )
 
-    def __init__(self, hs: "HomeServer") -> None:
+    def __init__(self, hs: "HomeServer", requests_ratelimiter: Ratelimiter) -> None:
         super().__init__()
         self._handler = hs.get_msc4388_rendezvous_handler()
+        self._requests_ratelimiter = requests_ratelimiter
 
-    def on_GET(self, request: SynapseRequest, rendezvous_id: str) -> tuple[int, Any]:
+    async def on_GET(
+        self, request: SynapseRequest, rendezvous_id: str
+    ) -> tuple[int, Any]:
+        await self._requests_ratelimiter.ratelimit(
+            None, request.getClientAddress().host
+        )
         return self._handler.handle_get(rendezvous_id, request)
 
-    def on_DELETE(
-        self, _request: SynapseRequest, rendezvous_id: str
+    async def on_DELETE(
+        self, request: SynapseRequest, rendezvous_id: str
     ) -> tuple[int, Any]:
+        await self._requests_ratelimiter.ratelimit(
+            None, request.getClientAddress().host
+        )
         return self._handler.handle_delete(rendezvous_id)
 
 
@@ -152,13 +173,17 @@ class MSC4388SendRendezvousServlet(RestServlet):
         unstable=True,
     )
 
-    def __init__(self, hs: "HomeServer") -> None:
+    def __init__(self, hs: "HomeServer", requests_ratelimiter: Ratelimiter) -> None:
         super().__init__()
         self._handler = hs.get_msc4388_rendezvous_handler()
+        self._requests_ratelimiter = requests_ratelimiter
 
-    def on_PUT(
+    async def on_PUT(
         self, request: SynapseRequest, rendezvous_id: str, txn_id: str
     ) -> tuple[int, Any]:
+        await self._requests_ratelimiter.ratelimit(
+            None, request.getClientAddress().host
+        )
         return self._handler.handle_put(rendezvous_id, txn_id, request)
 
 
@@ -170,6 +195,20 @@ def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None:
         MSC4108DelegationRendezvousServlet(hs).register(http_server)
 
     if hs.config.experimental.msc4388_enabled:
-        MSC4388CreateRendezvousServlet(hs).register(http_server)
-        MSC4388UpdateRendezvousServlet(hs).register(http_server)
-        MSC4388SendRendezvousServlet(hs).register(http_server)
+        # The rendezvous endpoints are ratelimited by client IP address, with
+        # the ratelimiters shared between the servlets.
+        create_ratelimiter = Ratelimiter(
+            store=hs.get_datastores().main,
+            clock=hs.get_clock(),
+            cfg=hs.config.experimental.msc4388_rc_create,
+        )
+        requests_ratelimiter = Ratelimiter(
+            store=hs.get_datastores().main,
+            clock=hs.get_clock(),
+            cfg=hs.config.experimental.msc4388_rc_requests,
+        )
+        MSC4388CreateRendezvousServlet(
+            hs, create_ratelimiter, requests_ratelimiter
+        ).register(http_server)
+        MSC4388UpdateRendezvousServlet(hs, requests_ratelimiter).register(http_server)
+        MSC4388SendRendezvousServlet(hs, requests_ratelimiter).register(http_server)

tests/rest/client/test_msc4388_rendezvous.py

@@ -81,6 +81,10 @@ class RendezvousServletTestCase(unittest.HomeserverTestCase):
         )
         config["experimental_features"] = {
             "msc4388_mode": "open",
+            # Effectively disable ratelimiting, as some tests make a large
+            # number of requests, unless the test configures it.
+            "msc4388_rc_create": {"per_second": 10000, "burst_count": 10000},
+            "msc4388_rc_requests": {"per_second": 10000, "burst_count": 10000},
             **(config.get("experimental_features") or {}),
         }
 
@@ -919,3 +923,128 @@ class RendezvousServletTestCase(unittest.HomeserverTestCase):
         )
 
         self.assertEqual(channel.code, 200)
+
+    @override_config(
+        {
+            "experimental_features": {
+                "msc4388_rc_create": {"per_second": 0.1, "burst_count": 2},
+            },
+        }
+    )
+    def test_create_ratelimited(self) -> None:
+        """
+        Test that creating a session is ratelimited, separately from other
+        requests.
+        """
+        for _ in range(2):
+            channel = self.make_request(
+                "POST",
+                rz_endpoint,
+                {"data": "foo=bar"},
+                access_token=None,
+            )
+            self.assertEqual(channel.code, 200)
+
+        channel = self.make_request(
+            "POST",
+            rz_endpoint,
+            {"data": "foo=bar"},
+            access_token=None,
+        )
+        self.assertEqual(channel.code, 429)
+        self.assertEqual(channel.json_body["errcode"], "M_LIMIT_EXCEEDED")
+
+        # Other requests are not affected
+        channel = self.make_request("GET", rz_endpoint, {}, access_token=None)
+        self.assertEqual(channel.code, 200)
+
+        # Once enough time has passed, a session can be created again
+        self.reactor.advance(10)
+        channel = self.make_request(
+            "POST",
+            rz_endpoint,
+            {"data": "foo=bar"},
+            access_token=None,
+        )
+        self.assertEqual(channel.code, 200)
+
+    @override_config(
+        {
+            "experimental_features": {
+                "msc4388_rc_requests": {"per_second": 0.1, "burst_count": 4},
+            },
+        }
+    )
+    def test_requests_ratelimited(self) -> None:
+        """
+        Test that discovery, and receiving, sending and cancelling on a session
+        share a ratelimit.
+        """
+        channel = self.make_request(
+            "POST",
+            rz_endpoint,
+            {"data": "foo=bar"},
+            access_token=None,
+        )
+        self.assertEqual(channel.code, 200)
+        rendezvous_id = channel.json_body["id"]
+        sequence_token = channel.json_body["sequence_token"]
+        session_endpoint = rz_endpoint + f"/{rendezvous_id}"
+
+        # Use up the burst with one request to each endpoint
+        channel = self.make_request("GET", rz_endpoint, {}, access_token=None)
+        self.assertEqual(channel.code, 200)
+
+        channel = self.make_request("GET", session_endpoint, access_token=None)
+        self.assertEqual(channel.code, 200)
+
+        channel = self.make_request(
+            "PUT",
+            send_endpoint(rendezvous_id, "txn1"),
+            {"sequence_token": sequence_token, "data": "foo=baz"},
+            access_token=None,
+        )
+        self.assertEqual(channel.code, 200)
+        sequence_token = channel.json_body["sequence_token"]
+
+        channel = self.make_request("GET", session_endpoint, access_token=None)
+        self.assertEqual(channel.code, 200)
+
+        # Now every one of them is ratelimited
+        channel = self.make_request("GET", rz_endpoint, {}, access_token=None)
+        self.assertEqual(channel.code, 429)
+        self.assertEqual(channel.json_body["errcode"], "M_LIMIT_EXCEEDED")
+
+        channel = self.make_request("GET", session_endpoint, access_token=None)
+        self.assertEqual(channel.code, 429)
+        self.assertEqual(channel.json_body["errcode"], "M_LIMIT_EXCEEDED")
+
+        channel = self.make_request(
+            "PUT",
+            send_endpoint(rendezvous_id, "txn2"),
+            {"sequence_token": sequence_token, "data": "foo=qux"},
+            access_token=None,
+        )
+        self.assertEqual(channel.code, 429)
+        self.assertEqual(channel.json_body["errcode"], "M_LIMIT_EXCEEDED")
+
+        channel = self.make_request("DELETE", session_endpoint, access_token=None)
+        self.assertEqual(channel.code, 429)
+        self.assertEqual(channel.json_body["errcode"], "M_LIMIT_EXCEEDED")
+
+        # Creating a session is not affected
+        channel = self.make_request(
+            "POST",
+            rz_endpoint,
+            {"data": "foo=bar"},
+            access_token=None,
+        )
+        self.assertEqual(channel.code, 200)
+
+        # Once enough time has passed, requests are accepted again, and the
+        # ratelimited PUT did not change the payload
+        self.reactor.advance(10)
+        channel = self.make_request("GET", session_endpoint, access_token=None)
+        self.assertEqual(channel.code, 200)
+        self.assertEqual(channel.json_body["data"], "foo=baz")
+        self.assertEqual(channel.json_body["sequence_token"], sequence_token)