Ratelimit cross user room key share requests.

changelog.d/8957.feature

@@ -0,0 +1 @@
+Add rate limiters to key sharing requests.

docs/sample_config.yaml

@@ -819,6 +819,9 @@ log_config: "CONFDIR/SERVERNAME.log.config"
 #     users are joining rooms the server is already in (this is cheap) vs
 #     "remote" for when users are trying to join rooms not on the server (which
 #     can be more expensive)
+#   - one for ratelimiting how frequently to-device messages are sent
+#   - one that reatelimits EDUs received over federation based on the origin
+#     and type
 #
 # The defaults are as shown below.
 #
@@ -852,6 +855,15 @@ log_config: "CONFDIR/SERVERNAME.log.config"
 #  remote:
 #    per_second: 0.01
 #    burst_count: 3
+#
+#rc_send_to_device:
+#  per_second: 0.1
+#  burst_count: 3
+#
+#rc_federation:
+#  edu:
+#    per_second: 0.1
+#    burst_count: 3
 
 
 # Ratelimiting settings for incoming federation

synapse/api/ratelimiting.py

@@ -14,7 +14,7 @@
 # limitations under the License.
 
 from collections import OrderedDict
-from typing import Any, Optional, Tuple
+from typing import Hashable, Optional, Tuple
 
 from synapse.api.errors import LimitExceededError
 from synapse.types import Requester
@@ -42,7 +42,9 @@ def __init__(self, clock: Clock, rate_hz: float, burst_count: int):
         #   * How many times an action has occurred since a point in time
         #   * The point in time
         #   * The rate_hz of this particular entry. This can vary per request
-        self.actions = OrderedDict()  # type: OrderedDict[Any, Tuple[float, int, float]]
+        self.actions = (
+            OrderedDict()
+        )  # type: OrderedDict[Hashable, Tuple[float, int, float]]
 
     def can_requester_do_action(
         self,
@@ -82,7 +84,7 @@ def can_requester_do_action(
 
     def can_do_action(
         self,
-        key: Any,
+        key: Hashable,
         rate_hz: Optional[float] = None,
         burst_count: Optional[int] = None,
         update: bool = True,
@@ -175,7 +177,7 @@ def _prune_message_counts(self, time_now_s: int):
 
     def ratelimit(
         self,
-        key: Any,
+        key: Hashable,
         rate_hz: Optional[float] = None,
         burst_count: Optional[int] = None,
         update: bool = True,

synapse/config/ratelimiting.py

@@ -102,6 +102,16 @@ def read_config(self, config, **kwargs):
             defaults={"per_second": 0.01, "burst_count": 3},
         )
 
+        self.rc_send_to_device = RateLimitConfig(
+            config.get("rc_send_to_device", {}),
+            defaults={"per_second": 0.1, "burst_count": 3},
+        )
+
+        self.rc_federation_edu = RateLimitConfig(
+            config.get("rc_federation", {}).get("edu", {}),
+            defaults={"per_second": 0.1, "burst_count": 3},
+        )
+
     def generate_config_section(self, **kwargs):
         return """\
         ## Ratelimiting ##
@@ -131,6 +141,9 @@ def generate_config_section(self, **kwargs):
         #     users are joining rooms the server is already in (this is cheap) vs
         #     "remote" for when users are trying to join rooms not on the server (which
         #     can be more expensive)
+        #   - one for ratelimiting how frequently to-device messages are sent
+        #   - one that reatelimits EDUs received over federation based on the origin
+        #     and type
         #
         # The defaults are as shown below.
         #
@@ -164,6 +177,15 @@ def generate_config_section(self, **kwargs):
         #  remote:
         #    per_second: 0.01
         #    burst_count: 3
+        #
+        #rc_send_to_device:
+        #  per_second: 0.1
+        #  burst_count: 3
+        #
+        #rc_federation:
+        #  edu:
+        #    per_second: 0.1
+        #    burst_count: 3
 
 
         # Ratelimiting settings for incoming federation

synapse/federation/federation_server.py

@@ -43,6 +43,7 @@
     SynapseError,
     UnsupportedRoomVersionError,
 )
+from synapse.api.ratelimiting import Ratelimiter
 from synapse.api.room_versions import KNOWN_ROOM_VERSIONS
 from synapse.events import EventBase
 from synapse.federation.federation_base import FederationBase, event_from_pdu_json
@@ -863,6 +864,13 @@ def __init__(self, hs: "HomeServer"):
         # Map from type to instance name that we should route EDU handling to.
         self._edu_type_to_instance = {}  # type: Dict[str, str]
 
+        # A rate limiter for incoming EDUs per type/origin.
+        self._edu_rate_limiter = Ratelimiter(
+            clock=self.clock,
+            rate_hz=self.config.rc_federation_edu.per_second,
+            burst_count=self.config.rc_federation_edu.burst_count,
+        )
+
     def register_edu_handler(
         self, edu_type: str, handler: Callable[[str, JsonDict], Awaitable[None]]
     ):
@@ -911,6 +919,10 @@ async def on_edu(self, edu_type: str, origin: str, content: dict):
         if not self.config.use_presence and edu_type == "m.presence":
             return
 
+        # If the incoming EDUs from this origin/type are over the limit, drop them.
+        if not self._edu_rate_limiter.can_do_action(key=(edu_type, origin)):
+            return
+
         # Check if we have a handler on this instance
         handler = self.edu_handlers.get(edu_type)
         if handler:

synapse/rest/client/v2_alpha/sendtodevice.py

@@ -16,6 +16,7 @@
 import logging
 from typing import Tuple
 
+from synapse.api.ratelimiting import Ratelimiter
 from synapse.http import servlet
 from synapse.http.servlet import parse_json_object_from_request
 from synapse.logging.opentracing import set_tag, trace
@@ -42,6 +43,12 @@ def __init__(self, hs):
         self.txns = HttpTransactionCache(hs)
         self.device_message_handler = hs.get_device_message_handler()
 
+        self._ratelimiter = Ratelimiter(
+            clock=hs.get_clock(),
+            rate_hz=self.hs.config.rc_send_to_device.per_second,
+            burst_count=self.hs.config.rc_send_to_device.burst_count,
+        )
+
     @trace(opname="sendToDevice")
     def on_PUT(self, request, message_type, txn_id):
         set_tag("message_type", message_type)
@@ -52,10 +59,11 @@ def on_PUT(self, request, message_type, txn_id):
 
     async def _put(self, request, message_type, txn_id):
         requester = await self.auth.get_user_by_req(request, allow_guest=True)
+        sender_user_id = requester.user.to_string()
 
-        content = parse_json_object_from_request(request)
+        self._ratelimiter.ratelimit(sender_user_id)
 
-        sender_user_id = requester.user.to_string()
+        content = parse_json_object_from_request(request)
 
         await self.device_message_handler.send_device_message(
             sender_user_id, message_type, content["messages"]