Introduce spam_checker_spammy internal event metadata. (#19453)

Follows: #19365

Part of: MSC4354 Sticky Events (experimental feature #19409)

This PR introduces a `spam_checker_spammy` flag, analogous to
`policy_server_spammy`, as an explicit flag
that an event was decided to be spammy by a spam-checker module.

The original Sticky Events PR (#18968) just reused
`policy_server_spammy`, but it didn't sit right with me
because we (at least appear to be experimenting with features that)
allow users to opt-in to seeing
`policy_server_spammy` events (presumably for moderation purposes).

Keeping these flags separate felt best, therefore.

As for why we need this flag: soon soft-failed status won't be
permanent, at least for sticky events.
The spam checker modules currently work by making events soft-failed.
We want to prevent spammy events from getting
reconsidered/un-soft-failed, so it seems like we need
a flag to track spam-checker spamminess *separately* from soft-failed.

Should be commit-by-commit friendly, but is also small.

---------

Signed-off-by: Olivier 'reivilibre <oliverw@matrix.org>

Author: reivilibre

changelog.d/19453.misc

@@ -0,0 +1 @@
+Introduce `spam_checker_spammy` internal event metadata.

rust/src/events/internal_metadata.rs

@@ -55,6 +55,7 @@ enum EventInternalMetadataData {
     SoftFailed(bool),
     ProactivelySend(bool),
     PolicyServerSpammy(bool),
+    SpamCheckerSpammy(bool),
     Redacted(bool),
     TxnId(Box<str>),
     DelayId(Box<str>),
@@ -105,6 +106,13 @@ impl EventInternalMetadataData {
                     .to_owned()
                     .into_any(),
             ),
+            EventInternalMetadataData::SpamCheckerSpammy(o) => (
+                pyo3::intern!(py, "spam_checker_spammy"),
+                o.into_pyobject(py)
+                    .unwrap_infallible()
+                    .to_owned()
+                    .into_any(),
+            ),
             EventInternalMetadataData::Redacted(o) => (
                 pyo3::intern!(py, "redacted"),
                 o.into_pyobject(py)
@@ -173,6 +181,11 @@ impl EventInternalMetadataData {
                     .extract()
                     .with_context(|| format!("'{key_str}' has invalid type"))?,
             ),
+            "spam_checker_spammy" => EventInternalMetadataData::SpamCheckerSpammy(
+                value
+                    .extract()
+                    .with_context(|| format!("'{key_str}' has invalid type"))?,
+            ),
             "redacted" => EventInternalMetadataData::Redacted(
                 value
                     .extract()
@@ -468,6 +481,17 @@ impl EventInternalMetadata {
         set_property!(self, PolicyServerSpammy, obj);
     }
 
+    #[getter]
+    fn get_spam_checker_spammy(&self) -> PyResult<bool> {
+        Ok(get_property_opt!(self, SpamCheckerSpammy)
+            .copied()
+            .unwrap_or(false))
+    }
+    #[setter]
+    fn set_spam_checker_spammy(&mut self, obj: bool) {
+        set_property!(self, SpamCheckerSpammy, obj);
+    }
+
     #[getter]
     fn get_redacted(&self) -> PyResult<bool> {
         let bool = get_property!(self, Redacted)?;

synapse/federation/federation_base.py

@@ -177,7 +177,9 @@ async def _check_sigs_and_hash(
             # Note: we don't redact the event so admins can inspect the event after the
             # fact. Other processes may redact the event, but that won't be applied to
             # the database copy of the event until the server's config requires it.
-            return pdu
+            #
+            # We also *don't* return early here as we would still like to evaluate
+            # `spam_checker_spammy`, for completeness.
 
         spam_check = await self._spam_checker_module_callbacks.check_event_for_spam(pdu)
 
@@ -194,6 +196,8 @@ async def _check_sigs_and_hash(
             # using the event in prev_events).
             redacted_event = prune_event(pdu)
             redacted_event.internal_metadata.soft_failed = True
+            # Mark this as spam so we don't re-evaluate soft-failure status.
+            redacted_event.internal_metadata.spam_checker_spammy = True
             return redacted_event
 
         return pdu

synapse/storage/databases/main/sticky_events.py

@@ -302,10 +302,14 @@ def insert_sticky_events_txn(
         Skips inserting events:
             - if they are considered spammy by the policy server;
               (unsure if correct, track: https://github.com/matrix-org/matrix-spec-proposals/pull/4354#discussion_r2727593350)
+            - if they are considered spammy by a Synapse spam checker module;
             - if they are rejected;
             - if they are outliers (they should be reconsidered for insertion when de-outliered); or
             - if they are not sticky (e.g. if the stickiness expired).
 
+        Note: Soft-failed sticky events ARE inserted, as their soft-failed status
+            could be re-evaluated later.
+
         Skipping the insertion of these types of 'invalid' events is useful for performance reasons because
         they would fill up the table yet we wouldn't show them to clients anyway.
 
@@ -321,7 +325,12 @@ def insert_sticky_events_txn(
         sticky_events: list[tuple[EventBase, int]] = []
         for ev in events:
             # MSC: Note: policy servers and other similar antispam techniques still apply to these events.
-            if ev.internal_metadata.policy_server_spammy:
+            # We don't filter out soft-failed events altogether (in case they get re-evaluated later),
+            # so filter out `spam_checker_spammy` events specifically as we don't want to re-evaluate _those_ later.
+            if (
+                ev.internal_metadata.policy_server_spammy
+                or ev.internal_metadata.spam_checker_spammy
+            ):
                 continue
             # We shouldn't be passed rejected events, but if we do, we filter them out too.
             if ev.rejected_reason is not None:
@@ -332,7 +341,7 @@ def insert_sticky_events_txn(
             sticky_duration = ev.sticky_duration()
             if sticky_duration is None:
                 continue
-            # Calculate the end time as start_time + effecitve sticky duration
+            # Calculate the end time as start_time + effective sticky duration
             expires_at = min(ev.origin_server_ts, now_ms) + sticky_duration.as_millis()
             # Filter out already expired sticky events
             if expires_at <= now_ms:

synapse/synapse_rust/events.pyi

@@ -38,6 +38,19 @@ class EventInternalMetadata:
     policy_server_spammy: bool
     """whether the policy server indicated that this event is spammy"""
 
+    spam_checker_spammy: bool
+    """Whether a spam checker module indicated that this event is spammy
+
+    Note that spam checkers also cause the event to be marked as soft-failed.
+
+    This flags exists for two reasons:
+        1. as debugging information
+        2. to prevent the soft-failed re-evaluation of spammy events
+           (the re-evaluation behaviour originates from MSC4354 Sticky Events)
+
+    Note that historical spammy events won't have this flag.
+    """
+
     txn_id: str
     """The transaction ID, if it was set when the event was created."""
     delay_id: str

tests/module_api/test_spamchecker.py

@@ -12,22 +12,33 @@
 # <https://www.gnu.org/licenses/agpl-3.0.html>.
 #
 #
+from http import HTTPStatus
 from typing import Literal
 
 from twisted.internet.testing import MemoryReactor
 
-from synapse.api.constants import EventContentFields, EventTypes
+from synapse.api.constants import (
+    EventContentFields,
+    EventTypes,
+    Membership,
+)
+from synapse.api.room_versions import RoomVersions
 from synapse.config.server import DEFAULT_ROOM_VERSION
+from synapse.events import make_event_from_dict
+from synapse.module_api import EventBase
 from synapse.rest import admin, login, room, room_upgrade_rest_servlet
 from synapse.server import HomeServer
 from synapse.types import Codes, JsonDict
 from synapse.util.clock import Clock
 
+from tests import unittest
 from tests.server import FakeChannel
 from tests.unittest import HomeserverTestCase
 
 
 class SpamCheckerTestCase(HomeserverTestCase):
+    """Tests for the spam checker module API."""
+
     servlets = [
         room.register_servlets,
         admin.register_servlets,
@@ -284,3 +295,178 @@ async def user_may_send_state_event(
 
         self.assertEqual(channel.code, 403)
         self.assertEqual(channel.json_body["errcode"], Codes.FORBIDDEN)
+
+
+class FederatedEventSpamCheckMetadataTestCase(unittest.FederatingHomeserverTestCase):
+    servlets = [
+        admin.register_servlets,
+        login.register_servlets,
+        room.register_servlets,
+    ]
+
+    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
+        super().prepare(reactor, clock, hs)
+        self._module_api = hs.get_module_api()
+        self._store = hs.get_datastores().main
+        self._storage_controllers = hs.get_storage_controllers()
+        self._federation_event_handler = hs.get_federation_event_handler()
+        self._federation_server = hs.get_federation_server()
+        self._state_handler = hs.get_state_handler()
+        self._persistence_controller = hs.get_storage_controllers().persistence
+
+        # Create a room
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass")
+        self.room_id = self.helper.create_room_as(
+            user1_id,
+            tok=user1_tok,
+            is_public=True,
+            room_version=RoomVersions.V10.identifier,
+        )
+
+        # Prepare a join for the 'remote' user
+        state_map = self.get_success(
+            self._storage_controllers.state.get_current_state(self.room_id)
+        )
+        forward_extremity_event_ids = self.get_success(
+            self.hs.get_datastores().main.get_latest_event_ids_in_room(self.room_id)
+        )
+        self.remote_user_id = f"@remoteuser:{self.OTHER_SERVER_NAME}"
+        self.remote_user_join_event = make_event_from_dict(
+            self.add_hashes_and_signatures_from_other_server(
+                {
+                    "room_id": self.room_id,
+                    "sender": self.remote_user_id,
+                    "state_key": self.remote_user_id,
+                    "depth": 1000,
+                    "origin_server_ts": 1,
+                    "type": EventTypes.Member,
+                    "content": {"membership": Membership.JOIN},
+                    "auth_events": [
+                        state_map[(EventTypes.Create, "")].event_id,
+                        state_map[(EventTypes.JoinRules, "")].event_id,
+                    ],
+                    "prev_events": list(forward_extremity_event_ids),
+                }
+            ),
+            room_version=RoomVersions.V10,
+        )
+
+        # Send the join
+        self.get_success(
+            self._federation_event_handler.on_receive_pdu(
+                self.OTHER_SERVER_NAME, self.remote_user_join_event
+            )
+        )
+
+        # Check the join made it to the 'local' view of the room
+        self.helper.get_event(
+            room_id=self.room_id,
+            event_id=self.remote_user_join_event.event_id,
+            tok=user1_tok,
+            expect_code=HTTPStatus.OK,
+        )
+
+    def test_federated_events_with_spam_checker_metadata(self) -> None:
+        """
+        Simulates receiving spammy and non-spammy events over federation,
+        then checks their `spam_checker_spammy` flag is set properly.
+        """
+
+        async def check_event_for_spam(event: EventBase) -> Literal["NOT_SPAM"] | Codes:
+            if event.type == EventTypes.Message:
+                if "ham" not in event.content["body"]:
+                    return Codes.FORBIDDEN
+            return "NOT_SPAM"
+
+        # Register a spam checker callback that only allows messages with 'ham'
+        self._module_api.register_spam_checker_callbacks(
+            check_event_for_spam=check_event_for_spam
+        )
+
+        # Prepare a spammy and a non-spammy event.
+        forward_extremity_event_ids = self.get_success(
+            self._store.get_latest_event_ids_in_room(self.room_id)
+        )
+        state_map = self.get_success(
+            self._storage_controllers.state.get_current_state(self.room_id)
+        )
+        spammy_event = make_event_from_dict(
+            self.add_hashes_and_signatures_from_other_server(
+                {
+                    "room_id": self.room_id,
+                    "sender": self.remote_user_id,
+                    "depth": 2000,
+                    "origin_server_ts": 2,
+                    "type": EventTypes.Message,
+                    "content": {"body": "this is spam", "msgtype": "m.text"},
+                    "auth_events": [
+                        state_map[(EventTypes.Create, "")].event_id,
+                        state_map[(EventTypes.JoinRules, "")].event_id,
+                        state_map[(EventTypes.Member, self.remote_user_id)].event_id,
+                    ],
+                    "prev_events": list(forward_extremity_event_ids),
+                }
+            ),
+            room_version=RoomVersions.V10,
+        )
+        non_spammy_event = make_event_from_dict(
+            self.add_hashes_and_signatures_from_other_server(
+                {
+                    "room_id": self.room_id,
+                    "sender": self.remote_user_id,
+                    "depth": 2000,
+                    "origin_server_ts": 2,
+                    "type": EventTypes.Message,
+                    "content": {"body": "delicious ham", "msgtype": "m.text"},
+                    "auth_events": [
+                        state_map[(EventTypes.Create, "")].event_id,
+                        state_map[(EventTypes.JoinRules, "")].event_id,
+                        state_map[(EventTypes.Member, self.remote_user_id)].event_id,
+                    ],
+                    "prev_events": list(forward_extremity_event_ids),
+                }
+            ),
+            room_version=RoomVersions.V10,
+        )
+
+        # Receive these events over federation
+        # We need to let the federation server have them because it will
+        # invoke `_check_sigs_and_hash` which invokes the spam checker.
+        self.get_success(
+            self._federation_server._handle_received_pdu(
+                self.OTHER_SERVER_NAME, spammy_event
+            )
+        )
+        self.get_success(
+            self._federation_server._handle_received_pdu(
+                self.OTHER_SERVER_NAME, non_spammy_event
+            )
+        )
+
+        # Retrieve the events from the database
+        retrieved_spammy_event = self.get_success(
+            self._store.get_event(spammy_event.event_id, allow_rejected=True)
+        )
+        retrieved_non_spammy_event = self.get_success(
+            self._store.get_event(non_spammy_event.event_id, allow_rejected=True)
+        )
+
+        # Assert the spammy flags (and soft-failed flags, for good measure) are set properly
+        self.assertTrue(
+            retrieved_spammy_event.internal_metadata.spam_checker_spammy,
+            "Spammy inbound event should be marked as spam_checker_spammy!",
+        )
+        self.assertTrue(
+            retrieved_spammy_event.internal_metadata.is_soft_failed(),
+            "Spammy inbound event should be soft-failed.",
+        )
+
+        self.assertFalse(
+            retrieved_non_spammy_event.internal_metadata.spam_checker_spammy,
+            "Non-spammy inbound event should not be marked as spam_checker_spammy!",
+        )
+        self.assertFalse(
+            retrieved_non_spammy_event.internal_metadata.is_soft_failed(),
+            "Non-spammy inbound event should not be soft-failed.",
+        )

tests/rest/client/sliding_sync/test_rooms_meta.py

@@ -12,6 +12,7 @@
 # <https://www.gnu.org/licenses/agpl-3.0.html>.
 #
 import logging
+from typing import Any
 
 from parameterized import parameterized, parameterized_class
 
@@ -966,7 +967,7 @@ def test_rooms_bump_stamp_backfill(self) -> None:
         creator = "@user:other"
         room_id = "!foo:other"
         room_version = RoomVersions.V10
-        shared_kwargs = {
+        shared_kwargs: dict[str, Any] = {
             "room_id": room_id,
             "room_version": room_version.identifier,
         }

tests/storage/test_roommember.py

@@ -20,7 +20,7 @@
 #
 #
 import logging
-from typing import cast
+from typing import Any, cast
 
 from twisted.internet.testing import MemoryReactor
 
@@ -238,7 +238,7 @@ def test_join_locally_forgotten_room(self) -> None:
         creator = "@user:other"
         room_id = "!foo:other"
         room_version = RoomVersions.V10
-        shared_kwargs = {
+        shared_kwargs: dict[str, Any] = {
             "room_id": room_id,
             "room_version": room_version.identifier,
         }

tests/storage/test_sliding_sync_tables.py

@@ -18,7 +18,7 @@
 #
 #
 import logging
-from typing import cast
+from typing import Any, cast
 
 import attr
 from parameterized import parameterized
@@ -873,7 +873,7 @@ def test_joined_room_bump_stamp_backfill(self) -> None:
         creator = "@user:other"
         room_id = "!foo:other"
         room_version = RoomVersions.V10
-        shared_kwargs = {
+        shared_kwargs: dict[str, Any] = {
             "room_id": room_id,
             "room_version": room_version.identifier,
         }