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 ratelimits room key requests received over federation based on
+#     the origin
 #
 # 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:
+#  room_key_request:
+#    per_second: 0.1
+#    burst_count: 3
 
 
 # Ratelimiting settings for incoming federation

synapse/api/constants.py

@@ -94,6 +94,7 @@ class EventTypes:
     Retention = "m.room.retention"
 
     Presence = "m.presence"
+    RoomKeyRequest = "m.room_key_request"
 
 
 class RejectedReason:

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_key_requests = RateLimitConfig(
+            config.get("rc_key_requests", {}),
+            defaults={"per_second": 0.1, "burst_count": 3},
+        )
+
+        self.rc_federation_room_key_request = RateLimitConfig(
+            config.get("rc_federation", {}).get("room_key_request", {}),
+            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 cross-user key requests by user/device
+        #   - one that ratelimits room key requests received over federation based on
+        #     the origin
         #
         # 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_key_requests:
+        #  per_second: 0.1
+        #  burst_count: 3
+        #
+        #rc_federation:
+        #  room_key_request:
+        #    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 room key requests per origin.
+        self._room_key_request_rate_limiter = Ratelimiter(
+            clock=self.clock,
+            rate_hz=self.config.rc_federation_room_key_request.per_second,
+            burst_count=self.config.rc_federation_room_key_request.burst_count,
+        )
+
     def register_edu_handler(
         self, edu_type: str, handler: Callable[[str, JsonDict], Awaitable[None]]
     ):
@@ -908,7 +916,15 @@ def register_instance_for_edu(self, edu_type: str, instance_name: str):
         self._edu_type_to_instance[edu_type] = instance_name
 
     async def on_edu(self, edu_type: str, origin: str, content: dict):
-        if not self.config.use_presence and edu_type == "m.presence":
+        if not self.config.use_presence and edu_type == EventTypes.Presence:
+            return
+
+        # If the incoming room key requests from a particular origin are over
+        # the limit, drop them.
+        if (
+            edu_type == EventTypes.RoomKeyRequest
+            and not self._room_key_request_rate_limiter.can_do_action(origin)
+        ):
             return
 
         # Check if we have a handler on this instance

synapse/handlers/devicemessage.py

@@ -16,7 +16,9 @@
 import logging
 from typing import TYPE_CHECKING, Any, Dict
 
+from synapse.api.constants import EventTypes
 from synapse.api.errors import SynapseError
+from synapse.api.ratelimiting import Ratelimiter
 from synapse.logging.context import run_in_background
 from synapse.logging.opentracing import (
     get_active_span_text_map,
@@ -52,6 +54,12 @@ def __init__(self, hs: "HomeServer"):
 
         self._device_list_updater = hs.get_device_handler().device_list_updater
 
+        self._ratelimiter = Ratelimiter(
+            clock=hs.get_clock(),
+            rate_hz=hs.config.rc_key_requests.per_second,
+            burst_count=hs.config.rc_key_requests.burst_count,
+        )
+
     async def on_direct_to_device_edu(self, origin: str, content: JsonDict) -> None:
         local_messages = {}
         sender_user_id = content["sender"]
@@ -153,6 +161,14 @@ async def send_device_message(
         local_messages = {}
         remote_messages = {}  # type: Dict[str, Dict[str, Dict[str, JsonDict]]]
         for user_id, by_device in messages.items():
+            # Ratelimit local cross-user key requests by user/device.
+            if (
+                message_type == EventTypes.RoomKeyRequest
+                and user_id != sender_user_id
+                and self._ratelimiter.can_do_action((user_id, by_device))
+            ):
+                continue
+
             # we use UserID.from_string to catch invalid user ids
             if self.is_mine(UserID.from_string(user_id)):
                 messages_by_device = {