Properly verify and error on power-level mismatch in /batch_send endpoint

changelog.d/13229.bugfix

@@ -0,0 +1 @@
+Added better verification and tests for the `/batch_send` endpoint related to which users are allowed to send historical events. Specifically, when the room version does not support the `historical` power level, the `/batch_send` endpoint will now fail if the user performing the request is not the room creator. Contributed by @sumnerevans @ Beeper.

synapse/rest/client/room_batch.py

@@ -20,7 +20,7 @@
 from twisted.web.server import Request
 
 from synapse.api.constants import EventContentFields
-from synapse.api.errors import AuthError, Codes, SynapseError
+from synapse.api.errors import AuthError, Codes, NotFoundError, SynapseError
 from synapse.http.server import HttpServer
 from synapse.http.servlet import (
     RestServlet,
@@ -89,7 +89,33 @@ async def on_POST(
         if not requester.app_service:
             raise AuthError(
                 HTTPStatus.FORBIDDEN,
-                "Only application services can use the /batchsend endpoint",
+                "Only application services can use the /batch_send endpoint",
+                errcode=Codes.FORBIDDEN,
+            )
+
+        # Verify that the user actually has permission to batch send in this
+        # room.
+        #
+        # We only have to check when the room version does not support
+        # msc2716_historical because if it does, the event auth code verifies
+        # whether the user has permission to send historical events.
+        #
+        # For room versions that don't support msc2716_historical, only the
+        # room creator is allowed to batch send.
+        room = await self.store.get_room(room_id)
+        if not room:
+            raise NotFoundError("Room not found")
+
+        room_version = await self.store.get_room_version(room_id)
+        if (
+            not room_version.msc2716_historical
+            and requester.user.to_string() != room["creator"]
+        ):
+            raise SynapseError(
+                HTTPStatus.FORBIDDEN,
+                f"In room version {room_version.identifier}, only the room "
+                "creator can use the /batch_send endpoint",
+                errcode=Codes.FORBIDDEN,
             )
 
         body = parse_json_object_from_request(request)

tests/rest/client/test_room_batch.py

@@ -1,10 +1,12 @@
 import logging
-from typing import List, Tuple
+from typing import List, Optional, Tuple
 from unittest.mock import Mock, patch
+from urllib.parse import urlencode
 
 from twisted.test.proto_helpers import MemoryReactor
 
 from synapse.api.constants import EventContentFields, EventTypes
+from synapse.api.room_versions import RoomVersion, RoomVersions
 from synapse.appservice import ApplicationService
 from synapse.rest import admin
 from synapse.rest.client import login, register, room, room_batch, sync
@@ -89,50 +91,65 @@ def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
         self.clock = clock
         self._storage_controllers = hs.get_storage_controllers()
 
+        self.room_creator_user_id, _ = self.register_appservice_user(
+            "as_user_creator", self.appservice.token
+        )
         self.virtual_user_id, _ = self.register_appservice_user(
             "as_user_potato", self.appservice.token
         )
 
-    def _create_test_room(self) -> Tuple[str, str, str, str]:
+    def _create_test_room(
+        self, room_version: RoomVersion = RoomVersions.V9
+    ) -> Tuple[str, str, str, str]:
         room_id = self.helper.create_room_as(
-            self.appservice.sender, tok=self.appservice.token
+            self.appservice.sender,
+            room_version=room_version.identifier,
+            tok=self.appservice.token,
+            appservice_user_id=self.room_creator_user_id,
         )
 
         res_a = self.helper.send_event(
             room_id=room_id,
             type=EventTypes.Message,
-            content={
-                "msgtype": "m.text",
-                "body": "A",
-            },
+            content={"msgtype": "m.text", "body": "A"},
             tok=self.appservice.token,
+            appservice_user_id=self.room_creator_user_id,
         )
         event_id_a = res_a["event_id"]
 
         res_b = self.helper.send_event(
             room_id=room_id,
             type=EventTypes.Message,
-            content={
-                "msgtype": "m.text",
-                "body": "B",
-            },
+            content={"msgtype": "m.text", "body": "B"},
             tok=self.appservice.token,
+            appservice_user_id=self.room_creator_user_id,
         )
         event_id_b = res_b["event_id"]
 
         res_c = self.helper.send_event(
             room_id=room_id,
             type=EventTypes.Message,
-            content={
-                "msgtype": "m.text",
-                "body": "C",
-            },
+            content={"msgtype": "m.text", "body": "C"},
             tok=self.appservice.token,
+            appservice_user_id=self.room_creator_user_id,
         )
         event_id_c = res_c["event_id"]
 
         return room_id, event_id_a, event_id_b, event_id_c
 
+    def _create_batch_send_url(
+        self, room_id: str, prev_event_id: str, user_id: Optional[str] = None
+    ) -> str:
+        params = {
+            "prev_event_id": prev_event_id,
+            "user_id": user_id or self.room_creator_user_id,
+        }
+        return (
+            "/_matrix/client/unstable/org.matrix.msc2716/rooms/{}/batch_send?{}".format(
+                room_id, urlencode(params)
+            )
+        )
+
     @unittest.override_config({"experimental_features": {"msc2716_enabled": True}})
     def test_same_state_groups_for_whole_historical_batch(self) -> None:
         """Make sure that when using the `/batch_send` endpoint to import a
@@ -147,8 +164,7 @@ def test_same_state_groups_for_whole_historical_batch(self) -> None:
 
         channel = self.make_request(
             "POST",
-            "/_matrix/client/unstable/org.matrix.msc2716/rooms/%s/batch_send?prev_event_id=%s"
-            % (room_id, event_id_a),
+            self._create_batch_send_url(room_id, event_id_a),
             content={
                 "events": _create_message_events_for_batch_send_request(
                     self.virtual_user_id, time_before_room, 3
@@ -198,19 +214,26 @@ def test_sync_while_batch_importing(self) -> None:
         room_id, _, _, _ = self._create_test_room()
         # Invite the user
         self.helper.invite(
-            room_id, src=self.appservice.sender, tok=self.appservice.token, targ=user_id
+            room_id,
+            src=self.appservice.sender,
+            targ=user_id,
+            tok=self.appservice.token,
+            appservice_user_id=self.room_creator_user_id,
         )
 
         # Create another room, send a bunch of events to advance the stream token
         other_room_id = self.helper.create_room_as(
-            self.appservice.sender, tok=self.appservice.token
+            self.appservice.sender,
+            tok=self.appservice.token,
+            appservice_user_id=self.room_creator_user_id,
         )
         for _ in range(5):
             self.helper.send_event(
                 room_id=other_room_id,
                 type=EventTypes.Message,
                 content={"msgtype": "m.text", "body": "C"},
                 tok=self.appservice.token,
+                appservice_user_id=self.room_creator_user_id,
             )
 
         # Join the room as the normal user
@@ -231,18 +254,15 @@ def test_sync_while_batch_importing(self) -> None:
         response = self.helper.send_event(
             room_id=room_id,
             type=EventTypes.Message,
-            content={
-                "msgtype": "m.text",
-                "body": "C",
-            },
+            content={"msgtype": "m.text", "body": "C"},
             tok=self.appservice.token,
+            appservice_user_id=self.room_creator_user_id,
         )
         event_to_hang_id = response["event_id"]
 
         channel = self.make_request(
             "POST",
-            "/_matrix/client/unstable/org.matrix.msc2716/rooms/%s/batch_send?prev_event_id=%s"
-            % (room_id, event_to_hang_id),
+            self._create_batch_send_url(room_id, event_to_hang_id),
             content={
                 "events": _create_message_events_for_batch_send_request(
                     self.virtual_user_id, time_before_room, 3
@@ -300,3 +320,149 @@ def test_sync_while_batch_importing(self) -> None:
         assert (
             "m.room.create" in state_event_types
         ), "Missing room full state in sync response"
+
+    @unittest.override_config({"experimental_features": {"msc2716_enabled": True}})
+    def test_room_batch_id_persistence(self) -> None:
+        """Make sure that the next_batch_id sent back to the user is persisted
+        in the database.
+        """
+
+        time_before_room = int(self.clock.time_msec())
+        room_id, event_id_a, _, _ = self._create_test_room()
+
+        channel = self.make_request(
+            "POST",
+            self._create_batch_send_url(room_id, event_id_a),
+            content={
+                "events": _create_message_events_for_batch_send_request(
+                    self.virtual_user_id, time_before_room, 3
+                ),
+                "state_events_at_start": _create_join_state_events_for_batch_send_request(
+                    [self.virtual_user_id], time_before_room
+                ),
+            },
+            access_token=self.appservice.token,
+        )
+        self.assertEqual(channel.code, 200, channel.result)
+
+        corresponding_insertion_event_id = self.get_success(
+            self._storage_controllers.main.get_insertion_event_id_by_batch_id(
+                room_id, channel.json_body["next_batch_id"]
+            )
+        )
+
+        # The insertion event ID from the database should be the same as the
+        # insertion event ID from the batch response.
+        self.assertEqual(
+            corresponding_insertion_event_id,
+            channel.json_body["insertion_event_id"],
+        )
+
+    @unittest.override_config({"experimental_features": {"msc2716_enabled": True}})
+    def test_batch_send_as_non_creator_v9(self) -> None:
+        """Make sure that the batch send fails with 403 FORBIDDEN when the user
+        making the /batch_send request is not the creator of the room.
+        """
+
+        time_before_room = int(self.clock.time_msec())
+        room_id, event_id_a, _, _ = self._create_test_room()
+
+        channel = self.make_request(
+            "POST",
+            self._create_batch_send_url(
+                room_id, event_id_a, user_id=self.virtual_user_id
+            ),
+            content={
+                "events": _create_message_events_for_batch_send_request(
+                    self.virtual_user_id, time_before_room, 3
+                ),
+                "state_events_at_start": _create_join_state_events_for_batch_send_request(
+                    [self.virtual_user_id], time_before_room
+                ),
+            },
+            access_token=self.appservice.token,
+        )
+        self.assertEqual(channel.code, 403, channel.result)
+        self.assertEqual(
+            channel.json_body["error"],
+            "In room version 9, only the room creator can use the /batch_send endpoint",
+            channel.json_body,
+        )
+
+    @unittest.override_config({"experimental_features": {"msc2716_enabled": True}})
+    def test_batch_send_without_permission_msc2716_room_version(self) -> None:
+        """Make sure that the batch send fails when the user does not have the
+        ``historical`` power level in a room version with MSC2716 enabled.
+        """
+
+        time_before_room = int(self.clock.time_msec())
+        room_id, event_id_a, _, _ = self._create_test_room(
+            room_version=RoomVersions.MSC2716v3
+        )
+
+        channel = self.make_request(
+            "POST",
+            self._create_batch_send_url(
+                room_id, event_id_a, user_id=self.virtual_user_id
+            ),
+            content={
+                "events": _create_message_events_for_batch_send_request(
+                    self.virtual_user_id, time_before_room, 3
+                ),
+                "state_events_at_start": _create_join_state_events_for_batch_send_request(
+                    [self.virtual_user_id], time_before_room
+                ),
+            },
+            access_token=self.appservice.token,
+        )
+        self.assertEqual(channel.code, 403, channel.result)
+        self.assertEqual(
+            channel.json_body["error"],
+            "You don't have permission to send send historical related events "
+            '("insertion", "batch", and "marker")',
+            channel.json_body,
+        )
+
+    @unittest.override_config({"experimental_features": {"msc2716_enabled": True}})
+    def test_batch_send_with_permission_msc2716_room_version(self) -> None:
+        """Make sure that the batch send succeeds when the user does have the
+        ``historical`` power level in a room version with MSC2716 enabled, even
+        if they are not the creator of the room.
+        """
+
+        time_before_room = int(self.clock.time_msec())
+        room_id, event_id_a, _, _ = self._create_test_room(
+            room_version=RoomVersions.MSC2716v3
+        )
+
+        power_levels = self.helper.get_state(
+            room_id,
+            EventTypes.PowerLevels,
+            tok=self.appservice.token,
+            appservice_user_id=self.room_creator_user_id,
+        )
+        power_levels["historical"] = 0
+        self.helper.send_state(
+            room_id,
+            EventTypes.PowerLevels,
+            power_levels,
+            tok=self.appservice.token,
+            appservice_user_id=self.room_creator_user_id,
+        )
+
+        channel = self.make_request(
+            "POST",
+            self._create_batch_send_url(
+                room_id, event_id_a, user_id=self.virtual_user_id
+            ),
+            content={
+                "events": _create_message_events_for_batch_send_request(
+                    self.virtual_user_id, time_before_room, 3
+                ),
+                "state_events_at_start": _create_join_state_events_for_batch_send_request(
+                    [self.virtual_user_id], time_before_room
+                ),
+            },
+            access_token=self.appservice.token,
+        )
+        self.assertEqual(channel.code, 200, channel.result)