Merge a7503470d7 into b09d90cac9

synapse/handlers/message.py

@@ -1001,13 +1001,45 @@ class EventCreationHandler:
             )
 
         self.validator.validate_new(event, self.config)
+        await self._validate_event_relation(event)
+        logger.debug("Created event %s", event.event_id)
+
+        return event, context
+
+    async def _validate_event_relation(self, event: EventBase) -> None:
+        """
+        Ensure the relation data on a new event is not bogus.
+
+        Args:
+            event: The event being created.
+
+        Raises:
+            SynapseError if the event is invalid.
+        """
+
+        relation = event.content.get("m.relates_to")
+        if not relation:
+            return
+
+        relation_type = relation.get("rel_type")
+        if not relation_type:
+            return
+
+        # Ensure the parent is real.
+        relates_to = relation["event_id"]
+        parent_event = await self.store.get_event(relates_to)
+        if not parent_event:
+            # TODO Attempt to fetch the event over federation.
+            raise SynapseError(400, "Can't send relation to unknown event")
+
+        # And in the same room.
+        if parent_event.room_id != event.room_id:
+            raise SynapseError(400, "Relations must be in the same room")
 
         # If this event is an annotation then we check that that the sender
         # can't annotate the same way twice (e.g. stops users from liking an
         # event multiple times).
-        relation = event.content.get("m.relates_to", {})
-        if relation.get("rel_type") == RelationTypes.ANNOTATION:
-            relates_to = relation["event_id"]
+        if relation_type == RelationTypes.ANNOTATION:
             aggregation_key = relation["key"]
 
             already_exists = await self.store.has_user_annotated_event(
@@ -1016,9 +1048,12 @@ class EventCreationHandler:
             if already_exists:
                 raise SynapseError(400, "Can't send same reaction twice")
 
-        logger.debug("Created event %s", event.event_id)
-
-        return event, context
+        # If this relation is a thread, then ensure thread head is not part of
+        # a thread already.
+        elif relation_type == RelationTypes.THREAD:
+            already_thread = await self.store.get_event_thread(relates_to)
+            if already_thread:
+                raise SynapseError(400, "Can't fork threads")
 
     @measure_func("handle_new_client_event")
     async def handle_new_client_event(

synapse/storage/databases/main/relations.py

@@ -436,6 +436,37 @@ class RelationsWorkerStore(SQLBaseStore):
             "get_if_user_has_annotated_event", _get_if_user_has_annotated_event
         )
 
+    async def get_event_thread(self, event_id: str) -> Optional[str]:
+        """Return an event's thread.
+
+        Args:
+            event_id: The event being used as the start of a new thread.
+
+        Returns:
+            The thread ID of the event.
+        """
+
+        sql = """
+            SELECT relates_to_id FROM event_relations
+            WHERE
+                event_id = ?
+                AND relation_type = ?
+            LIMIT 1;
+        """
+
+        def _get_thread_id(txn) -> Optional[str]:
+            txn.execute(
+                sql,
+                (
+                    event_id,
+                    RelationTypes.THREAD,
+                ),
+            )
+
+            return txn.fetchone()
+
+        return await self.db_pool.runInteraction("get_thread_id", _get_thread_id)
+
 
 class RelationsStore(RelationsWorkerStore):
     pass

tests/rest/client/test_relations.py

@@ -91,6 +91,26 @@ class RelationsTestCase(unittest.HomeserverTestCase):
         channel = self._send_relation(RelationTypes.ANNOTATION, EventTypes.Member)
         self.assertEquals(400, channel.code, channel.json_body)
 
+    def test_deny_invalid_event(self):
+        """Test that we deny relations on non-existant events"""
+        channel = self._send_relation(
+            RelationTypes.ANNOTATION, EventTypes.Member, parent_id="foo"
+        )
+        self.assertEquals(400, channel.code, channel.json_body)
+
+    def test_deny_invalid_room(self):
+        """Test that we deny relations on non-existant events"""
+        # Create another room and send a message in it.
+        room2 = self.helper.create_room_as(self.user_id, tok=self.user_token)
+        res = self.helper.send(room2, body="Hi!", tok=self.user_token)
+        parent_id = res["event_id"]
+
+        # Attempt to send an annotation to that event.
+        channel = self._send_relation(
+            RelationTypes.ANNOTATION, EventTypes.Member, parent_id=parent_id
+        )
+        self.assertEquals(400, channel.code, channel.json_body)
+
     def test_deny_double_react(self):
         """Test that we deny relations on membership events"""
         channel = self._send_relation(RelationTypes.ANNOTATION, "m.reaction", key="a")
@@ -99,6 +119,25 @@ class RelationsTestCase(unittest.HomeserverTestCase):
         channel = self._send_relation(RelationTypes.ANNOTATION, "m.reaction", "a")
         self.assertEquals(400, channel.code, channel.json_body)
 
+    def test_deny_forked_thread(self):
+        """It is invalid to start a thread off a thread."""
+        channel = self._send_relation(
+            RelationTypes.THREAD,
+            "m.room.message",
+            content={"msgtype": "m.text", "body": "foo"},
+            parent_id=self.parent_id,
+        )
+        self.assertEquals(200, channel.code, channel.json_body)
+        parent_id = channel.json_body["event_id"]
+
+        channel = self._send_relation(
+            RelationTypes.THREAD,
+            "m.room.message",
+            content={"msgtype": "m.text", "body": "foo"},
+            parent_id=parent_id,
+        )
+        self.assertEquals(400, channel.code, channel.json_body)
+
     def test_basic_paginate_relations(self):
         """Tests that calling pagination API correctly the latest relations."""
         channel = self._send_relation(RelationTypes.ANNOTATION, "m.reaction", "a")