add SpamChecker callback for silently dropping inbound federated events

changelog.d/12744.feature

@@ -0,0 +1 @@
+Add a `drop_federated_event` callback to `SpamChecker` to disregard inbound federated events before they take up much processing power, in an emergency.

docs/modules/spam_checker_callbacks.md

@@ -249,6 +249,24 @@ callback returns `False`, Synapse falls through to the next one. The value of th
 callback that does not return `False` will be used. If this happens, Synapse will not call
 any of the subsequent implementations of this callback.
 
+### `drop_federated_event`
+
+_First introduced in Synapse v1.60.0_
+
+```python
+async def drop_federated_event(event: "synapse.events.EventBase") -> bool
+```
+
+Called when checking whether a remote server can federate an event with us. **Returning
+`True` from this function will silently drop a federated event and split-brain our view
+of a room's DAG, and thus you shouldn't use this callback unless you know what you are
+doing.**
+
+If multiple modules implement this callback, they will be considered in order. If a
+callback returns `False`, Synapse falls through to the next one. The value of the first
+callback that does not return `False` will be used. If this happens, Synapse will not call
+any of the subsequent implementations of this callback.
+
 ## Example
 
 The example below is a module that implements the spam checker callback

docs/spam_checker.md

@@ -32,6 +32,7 @@ well as some specific methods:
 * `check_username_for_spam`
 * `check_registration_for_spam`
 * `check_media_file_for_spam`
+* `drop_federated_event`
 
 The details of each of these methods (as well as their inputs and outputs)
 are documented in the `synapse.events.spamcheck.SpamChecker` class.
@@ -86,6 +87,10 @@ class ExampleSpamChecker:
 
     async def check_media_file_for_spam(self, file_wrapper, file_info):
         return False  # allow all media
+
+
+    async def drop_federated_event(self, foo):
+        return False  # don't silently drop any inbound federated events
 ```
 
 ## Configuration

synapse/events/spamcheck.py

@@ -44,6 +44,10 @@
     ["synapse.events.EventBase"],
     Awaitable[Union[bool, str]],
 ]
+DROP_FEDERATED_EVENT_CALLBACK = Callable[
+    ["synapse.events.EventBase"],
+    Awaitable[Union[bool, str]],
+]
 USER_MAY_JOIN_ROOM_CALLBACK = Callable[[str, str, bool], Awaitable[bool]]
 USER_MAY_INVITE_CALLBACK = Callable[[str, str, str], Awaitable[bool]]
 USER_MAY_SEND_3PID_INVITE_CALLBACK = Callable[[str, str, str, str], Awaitable[bool]]
@@ -93,6 +97,7 @@ def load_legacy_spam_checkers(hs: "synapse.server.HomeServer") -> None:
     # which name appears in this set, we'll want to register it.
     spam_checker_methods = {
         "check_event_for_spam",
+        "drop_federated_event",
         "user_may_invite",
         "user_may_create_room",
         "user_may_create_room_alias",
@@ -168,6 +173,7 @@ def __init__(self, hs: "synapse.server.HomeServer") -> None:
         self.clock = hs.get_clock()
 
         self._check_event_for_spam_callbacks: List[CHECK_EVENT_FOR_SPAM_CALLBACK] = []
+        self._drop_federated_event_callbacks: List[DROP_FEDERATED_EVENT_CALLBACK] = []
         self._user_may_join_room_callbacks: List[USER_MAY_JOIN_ROOM_CALLBACK] = []
         self._user_may_invite_callbacks: List[USER_MAY_INVITE_CALLBACK] = []
         self._user_may_send_3pid_invite_callbacks: List[
@@ -191,6 +197,7 @@ def __init__(self, hs: "synapse.server.HomeServer") -> None:
     def register_callbacks(
         self,
         check_event_for_spam: Optional[CHECK_EVENT_FOR_SPAM_CALLBACK] = None,
+        drop_federated_event: Optional[DROP_FEDERATED_EVENT_CALLBACK] = None,
         user_may_join_room: Optional[USER_MAY_JOIN_ROOM_CALLBACK] = None,
         user_may_invite: Optional[USER_MAY_INVITE_CALLBACK] = None,
         user_may_send_3pid_invite: Optional[USER_MAY_SEND_3PID_INVITE_CALLBACK] = None,
@@ -209,6 +216,9 @@ def register_callbacks(
         if check_event_for_spam is not None:
             self._check_event_for_spam_callbacks.append(check_event_for_spam)
 
+        if drop_federated_event is not None:
+            self._drop_federated_event_callbacks.append(drop_federated_event)
+
         if user_may_join_room is not None:
             self._user_may_join_room_callbacks.append(user_may_join_room)
 
@@ -268,6 +278,31 @@ async def check_event_for_spam(
 
         return False
 
+    async def drop_federated_event(
+        self, event: "synapse.events.EventBase"
+    ) -> Union[bool, str]:
+        """Checks if a given federated event is considered "spammy" by this
+        server.
+
+        If the server considers an event spammy, it will be silently dropped,
+        and in doing so will split-brain our view of the room's DAG.
+
+        Args:
+            event: the event to be checked
+
+        Returns:
+            True if the event should be silently dropped
+        """
+        for callback in self._drop_federated_event_callbacks:
+            with Measure(
+                self.clock, "{}.{}".format(callback.__module__, callback.__qualname__)
+            ):
+                res: Union[bool, str] = await delay_cancellation(callback(event))
+            if res:
+                return res
+
+        return False
+
     async def user_may_join_room(
         self, user_id: str, room_id: str, is_invited: bool
     ) -> bool:

synapse/federation/federation_server.py

@@ -110,6 +110,7 @@ def __init__(self, hs: "HomeServer"):
 
         self.handler = hs.get_federation_handler()
         self.storage = hs.get_storage()
+        self._spam_checker = hs.get_spam_checker()
         self._federation_event_handler = hs.get_federation_event_handler()
         self.state = hs.get_state_handler()
         self._event_auth_handler = hs.get_event_auth_handler()
@@ -1019,6 +1020,12 @@ async def _handle_received_pdu(self, origin: str, pdu: EventBase) -> None:
         except SynapseError as e:
             raise FederationError("ERROR", e.code, e.msg, affected=pdu.event_id)
 
+        if await self._spam_checker.drop_federated_event(pdu):
+            logger.warning(
+                "Unstaged federated event contains spam, dropping %s", pdu.event_id
+            )
+            return
+
         # Add the event to our staging area
         await self.store.insert_received_event_to_staging(origin, pdu)
 
@@ -1109,16 +1116,30 @@ async def _process_incoming_pdus_in_room_inner(
                         (self._clock.time_msec() - received_ts) / 1000
                     )
 
-            # We need to do this check outside the lock to avoid a race between
-            # a new event being inserted by another instance and it attempting
-            # to acquire the lock.
-            next = await self.store.get_next_staged_event_for_room(
-                room_id, room_version
-            )
-            if not next:
+            while True:
+                # We need to do this check outside the lock to avoid a race between
+                # a new event being inserted by another instance and it attempting
+                # to acquire the lock.
+                next = await self.store.get_next_staged_event_for_room(
+                    room_id, room_version
+                )
+
+                if next is None:
+                    break
+
+                origin, event = next
+
+                if await self._spam_checker.drop_federated_event(event):
+                    logger.warning(
+                        "Staged federated event contains spam, dropping %s",
+                        event.event_id,
+                    )
+                    continue
+
                 break
 
-            origin, event = next
+            if not next:
+                break
 
             # Prune the event queue if it's getting large.
             #

synapse/module_api/__init__.py

@@ -47,6 +47,7 @@
     CHECK_MEDIA_FILE_FOR_SPAM_CALLBACK,
     CHECK_REGISTRATION_FOR_SPAM_CALLBACK,
     CHECK_USERNAME_FOR_SPAM_CALLBACK,
+    DROP_FEDERATED_EVENT_CALLBACK,
     USER_MAY_CREATE_ROOM_ALIAS_CALLBACK,
     USER_MAY_CREATE_ROOM_CALLBACK,
     USER_MAY_INVITE_CALLBACK,
@@ -234,6 +235,7 @@ def register_spam_checker_callbacks(
         self,
         *,
         check_event_for_spam: Optional[CHECK_EVENT_FOR_SPAM_CALLBACK] = None,
+        drop_federated_event: Optional[DROP_FEDERATED_EVENT_CALLBACK] = None,
         user_may_join_room: Optional[USER_MAY_JOIN_ROOM_CALLBACK] = None,
         user_may_invite: Optional[USER_MAY_INVITE_CALLBACK] = None,
         user_may_send_3pid_invite: Optional[USER_MAY_SEND_3PID_INVITE_CALLBACK] = None,
@@ -254,6 +256,7 @@ def register_spam_checker_callbacks(
         """
         return self._spam_checker.register_callbacks(
             check_event_for_spam=check_event_for_spam,
+            drop_federated_event=drop_federated_event,
             user_may_join_room=user_may_join_room,
             user_may_invite=user_may_invite,
             user_may_send_3pid_invite=user_may_send_3pid_invite,