Improve validation for threads.

Author: clokep

synapse/events/utils.py

@@ -506,7 +506,7 @@ async def _injected_bundled_aggregations(
             (
                 thread_count,
                 latest_thread_event,
-            ) = await self.store.get_thread_summary(event_id)
+            ) = await self.store.get_thread_summary(event_id, room_id)
             if latest_thread_event:
                 aggregations[RelationTypes.THREAD] = {
                     # Don't bundle aggregations as this could recurse forever.

synapse/storage/databases/main/events.py

@@ -1783,7 +1783,9 @@ def _handle_event_relations(
             txn.call_after(self.store.get_applicable_edit.invalidate, (parent_id,))
 
         if rel_type == RelationTypes.THREAD:
-            txn.call_after(self.store.get_thread_summary.invalidate, (parent_id,))
+            txn.call_after(
+                self.store.get_thread_summary.invalidate, (parent_id, event.room_id)
+            )
 
     def _handle_insertion_event(self, txn: LoggingTransaction, event: EventBase):
         """Handles keeping track of insertion events and edges/connections.

synapse/storage/databases/main/relations.py

@@ -348,13 +348,14 @@ def _get_applicable_edit_txn(txn: LoggingTransaction) -> Optional[str]:
 
     @cached()
     async def get_thread_summary(
-        self, event_id: str
+        self, event_id: str, room_id: str
     ) -> Tuple[int, Optional[EventBase]]:
         """Get the number of threaded replies, the senders of those replies, and
         the latest reply (if any) for the given event.
 
         Args:
-            event_id: The original event ID
+            event_id: Summarize the thread related to this event ID.
+            room_id: The room the event belongs to.
 
         Returns:
             The number of items in the thread and the most recent response, if any.
@@ -371,12 +372,13 @@ def _get_thread_summary_txn(
                 INNER JOIN events USING (event_id)
                 WHERE
                     relates_to_id = ?
+                    AND room_id = ?
                     AND relation_type = ?
                 ORDER BY topological_ordering DESC, stream_ordering DESC
                 LIMIT 1
             """
 
-            txn.execute(sql, (event_id, RelationTypes.THREAD))
+            txn.execute(sql, (event_id, room_id, RelationTypes.THREAD))
             row = txn.fetchone()
             if row is None:
                 return 0, None
@@ -386,11 +388,13 @@ def _get_thread_summary_txn(
             sql = """
                 SELECT COALESCE(COUNT(event_id), 0)
                 FROM event_relations
+                INNER JOIN events USING (event_id)
                 WHERE
                     relates_to_id = ?
+                    AND room_id = ?
                     AND relation_type = ?
             """
-            txn.execute(sql, (event_id, RelationTypes.THREAD))
+            txn.execute(sql, (event_id, room_id, RelationTypes.THREAD))
             count = txn.fetchone()[0]  # type: ignore[index]
 
             return count, latest_event_id

tests/rest/client/test_relations.py

@@ -653,6 +653,7 @@ def test_aggregation_get_event_for_thread(self):
             },
         )
 
+    @unittest.override_config({"experimental_features": {"msc3440_enabled": True}})
     def test_ignore_invalid_room(self):
         """Test that we ignore invalid relations over federation."""
         # Create another room and send a message in it.
@@ -664,7 +665,8 @@ def test_ignore_invalid_room(self):
         with patch(
             "synapse.handlers.message.EventCreationHandler._validate_event_relation"
         ):
-            # Generate a reaction and reference relations from a different room.
+            # Generate a reaction, reference, and thread relations from a
+            # different room.
             self.get_success(
                 inject_event(
                     self.hs,
@@ -698,6 +700,23 @@ def test_ignore_invalid_room(self):
                 )
             )
 
+            self.get_success(
+                inject_event(
+                    self.hs,
+                    room_id=self.room,
+                    type="m.room.message",
+                    sender=self.user_id,
+                    content={
+                        "body": "foo",
+                        "msgtype": "m.text",
+                        "m.relates_to": {
+                            "rel_type": RelationTypes.THREAD,
+                            "event_id": parent_id,
+                        },
+                    },
+                )
+            )
+
         # They should be ignored when fetching relations.
         channel = self.make_request(
             "GET",