MSC4140: Remove auth from delayed event management endpoints (#19152)

As per recent proposals in MSC4140, remove authentication for
restarting/cancelling/sending a delayed event, and give each of those
actions its own endpoint. (The original consolidated endpoint is still
supported for backwards compatibility.)

### Pull Request Checklist

<!-- Please read
https://element-hq.github.io/synapse/latest/development/contributing_guide.html
before submitting your pull request -->

* [x] Pull request is based on the develop branch
* [x] Pull request includes a [changelog
file](https://element-hq.github.io/synapse/latest/development/contributing_guide.html#changelog).
The entry should:
- Be a short description of your change which makes sense to users.
"Fixed a bug that prevented receiving messages from other servers."
instead of "Moved X method from `EventStore` to `EventWorkerStore`.".
  - Use markdown where necessary, mostly for `code blocks`.
  - End with either a period (.) or an exclamation mark (!).
  - Start with a capital letter.
- Feel free to credit yourself, by adding a sentence "Contributed by
@github_username." or "Contributed by [Your Name]." to the end of the
entry.
* [x] [Code
style](https://element-hq.github.io/synapse/latest/code_style.html) is
correct (run the
[linters](https://element-hq.github.io/synapse/latest/development/contributing_guide.html#run-the-linters))

---------

Co-authored-by: Half-Shot <will@half-shot.uk>

Author: AndrewFerr

changelog.d/19152.feature

@@ -0,0 +1 @@
+Remove authentication from `POST /_matrix/client/v1/delayed_events`, and allow calling this endpoint with the update action to take (`send`/`cancel`/`restart`) in the request path instead of the body.

synapse/_scripts/synapse_port_db.py

@@ -58,6 +58,7 @@
 from synapse.storage.databases.main import FilteringWorkerStore
 from synapse.storage.databases.main.account_data import AccountDataWorkerStore
 from synapse.storage.databases.main.client_ips import ClientIpBackgroundUpdateStore
+from synapse.storage.databases.main.delayed_events import DelayedEventsStore
 from synapse.storage.databases.main.deviceinbox import DeviceInboxBackgroundUpdateStore
 from synapse.storage.databases.main.devices import DeviceBackgroundUpdateStore
 from synapse.storage.databases.main.e2e_room_keys import EndToEndRoomKeyBackgroundStore
@@ -273,6 +274,7 @@ class Store(
     RelationsWorkerStore,
     EventFederationWorkerStore,
     SlidingSyncStore,
+    DelayedEventsStore,
 ):
     def execute(self, f: Callable[..., R], *args: Any, **kwargs: Any) -> Awaitable[R]:
         return self.db_pool.runInteraction(f.__name__, f, *args, **kwargs)

synapse/handlers/delayed_events.py

@@ -21,6 +21,7 @@
 from synapse.api.errors import ShadowBanError, SynapseError
 from synapse.api.ratelimiting import Ratelimiter
 from synapse.config.workers import MAIN_PROCESS_INSTANCE_NAME
+from synapse.http.site import SynapseRequest
 from synapse.logging.context import make_deferred_yieldable
 from synapse.logging.opentracing import set_tag
 from synapse.metrics import SERVER_NAME_LABEL, event_processing_positions
@@ -29,11 +30,9 @@
 )
 from synapse.storage.databases.main.delayed_events import (
     DelayedEventDetails,
-    DelayID,
     EventType,
     StateKey,
     Timestamp,
-    UserLocalpart,
 )
 from synapse.storage.databases.main.state_deltas import StateDelta
 from synapse.types import (
@@ -399,96 +398,63 @@ def on_added(self, next_send_ts: int) -> None:
         if self._next_send_ts_changed(next_send_ts):
             self._schedule_next_at(next_send_ts)
 
-    async def cancel(self, requester: Requester, delay_id: str) -> None:
+    async def cancel(self, request: SynapseRequest, delay_id: str) -> None:
         """
         Cancels the scheduled delivery of the matching delayed event.
 
-        Args:
-            requester: The owner of the delayed event to act on.
-            delay_id: The ID of the delayed event to act on.
-
         Raises:
             NotFoundError: if no matching delayed event could be found.
         """
         assert self._is_master
         await self._delayed_event_mgmt_ratelimiter.ratelimit(
-            requester,
-            (requester.user.to_string(), requester.device_id),
+            None, request.getClientAddress().host
         )
         await make_deferred_yieldable(self._initialized_from_db)
 
-        next_send_ts = await self._store.cancel_delayed_event(
-            delay_id=delay_id,
-            user_localpart=requester.user.localpart,
-        )
+        next_send_ts = await self._store.cancel_delayed_event(delay_id)
 
         if self._next_send_ts_changed(next_send_ts):
             self._schedule_next_at_or_none(next_send_ts)
 
-    async def restart(self, requester: Requester, delay_id: str) -> None:
+    async def restart(self, request: SynapseRequest, delay_id: str) -> None:
         """
         Restarts the scheduled delivery of the matching delayed event.
 
-        Args:
-            requester: The owner of the delayed event to act on.
-            delay_id: The ID of the delayed event to act on.
-
         Raises:
             NotFoundError: if no matching delayed event could be found.
         """
         assert self._is_master
         await self._delayed_event_mgmt_ratelimiter.ratelimit(
-            requester,
-            (requester.user.to_string(), requester.device_id),
+            None, request.getClientAddress().host
         )
         await make_deferred_yieldable(self._initialized_from_db)
 
         next_send_ts = await self._store.restart_delayed_event(
-            delay_id=delay_id,
-            user_localpart=requester.user.localpart,
-            current_ts=self._get_current_ts(),
+            delay_id, self._get_current_ts()
         )
 
         if self._next_send_ts_changed(next_send_ts):
             self._schedule_next_at(next_send_ts)
 
-    async def send(self, requester: Requester, delay_id: str) -> None:
+    async def send(self, request: SynapseRequest, delay_id: str) -> None:
         """
         Immediately sends the matching delayed event, instead of waiting for its scheduled delivery.
 
-        Args:
-            requester: The owner of the delayed event to act on.
-            delay_id: The ID of the delayed event to act on.
-
         Raises:
             NotFoundError: if no matching delayed event could be found.
         """
         assert self._is_master
-        # Use standard request limiter for sending delayed events on-demand,
-        # as an on-demand send is similar to sending a regular event.
-        await self._request_ratelimiter.ratelimit(requester)
+        await self._delayed_event_mgmt_ratelimiter.ratelimit(
+            None, request.getClientAddress().host
+        )
         await make_deferred_yieldable(self._initialized_from_db)
 
-        event, next_send_ts = await self._store.process_target_delayed_event(
-            delay_id=delay_id,
-            user_localpart=requester.user.localpart,
-        )
+        event, next_send_ts = await self._store.process_target_delayed_event(delay_id)
 
         if self._next_send_ts_changed(next_send_ts):
             self._schedule_next_at_or_none(next_send_ts)
 
-        await self._send_event(
-            DelayedEventDetails(
-                delay_id=DelayID(delay_id),
-                user_localpart=UserLocalpart(requester.user.localpart),
-                room_id=event.room_id,
-                type=event.type,
-                state_key=event.state_key,
-                origin_server_ts=event.origin_server_ts,
-                content=event.content,
-                device_id=event.device_id,
-            )
-        )
+        await self._send_event(event)
 
     async def _send_on_timeout(self) -> None:
         self._next_delayed_event_call = None
@@ -611,9 +577,7 @@ async def _send_event(
         finally:
             # TODO: If this is a temporary error, retry. Otherwise, consider notifying clients of the failure
             try:
-                await self._store.delete_processed_delayed_event(
-                    event.delay_id, event.user_localpart
-                )
+                await self._store.delete_processed_delayed_event(event.delay_id)
             except Exception:
                 logger.exception("Failed to delete processed delayed event")
 

synapse/rest/client/delayed_events.py

@@ -47,14 +47,11 @@ class UpdateDelayedEventServlet(RestServlet):
 
     def __init__(self, hs: "HomeServer"):
         super().__init__()
-        self.auth = hs.get_auth()
         self.delayed_events_handler = hs.get_delayed_events_handler()
 
     async def on_POST(
         self, request: SynapseRequest, delay_id: str
     ) -> tuple[int, JsonDict]:
-        requester = await self.auth.get_user_by_req(request)
-
         body = parse_json_object_from_request(request)
         try:
             action = str(body["action"])
@@ -75,11 +72,65 @@ async def on_POST(
             )
 
         if enum_action == _UpdateDelayedEventAction.CANCEL:
-            await self.delayed_events_handler.cancel(requester, delay_id)
+            await self.delayed_events_handler.cancel(request, delay_id)
         elif enum_action == _UpdateDelayedEventAction.RESTART:
-            await self.delayed_events_handler.restart(requester, delay_id)
+            await self.delayed_events_handler.restart(request, delay_id)
         elif enum_action == _UpdateDelayedEventAction.SEND:
-            await self.delayed_events_handler.send(requester, delay_id)
+            await self.delayed_events_handler.send(request, delay_id)
+        return 200, {}
+
+
+class CancelDelayedEventServlet(RestServlet):
+    PATTERNS = client_patterns(
+        r"/org\.matrix\.msc4140/delayed_events/(?P<delay_id>[^/]+)/cancel$",
+        releases=(),
+    )
+    CATEGORY = "Delayed event management requests"
+
+    def __init__(self, hs: "HomeServer"):
+        super().__init__()
+        self.delayed_events_handler = hs.get_delayed_events_handler()
+
+    async def on_POST(
+        self, request: SynapseRequest, delay_id: str
+    ) -> tuple[int, JsonDict]:
+        await self.delayed_events_handler.cancel(request, delay_id)
+        return 200, {}
+
+
+class RestartDelayedEventServlet(RestServlet):
+    PATTERNS = client_patterns(
+        r"/org\.matrix\.msc4140/delayed_events/(?P<delay_id>[^/]+)/restart$",
+        releases=(),
+    )
+    CATEGORY = "Delayed event management requests"
+
+    def __init__(self, hs: "HomeServer"):
+        super().__init__()
+        self.delayed_events_handler = hs.get_delayed_events_handler()
+
+    async def on_POST(
+        self, request: SynapseRequest, delay_id: str
+    ) -> tuple[int, JsonDict]:
+        await self.delayed_events_handler.restart(request, delay_id)
+        return 200, {}
+
+
+class SendDelayedEventServlet(RestServlet):
+    PATTERNS = client_patterns(
+        r"/org\.matrix\.msc4140/delayed_events/(?P<delay_id>[^/]+)/send$",
+        releases=(),
+    )
+    CATEGORY = "Delayed event management requests"
+
+    def __init__(self, hs: "HomeServer"):
+        super().__init__()
+        self.delayed_events_handler = hs.get_delayed_events_handler()
+
+    async def on_POST(
+        self, request: SynapseRequest, delay_id: str
+    ) -> tuple[int, JsonDict]:
+        await self.delayed_events_handler.send(request, delay_id)
         return 200, {}
 
 
@@ -108,4 +159,7 @@ def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None:
     # The following can't currently be instantiated on workers.
     if hs.config.worker.worker_app is None:
         UpdateDelayedEventServlet(hs).register(http_server)
+        CancelDelayedEventServlet(hs).register(http_server)
+        RestartDelayedEventServlet(hs).register(http_server)
+        SendDelayedEventServlet(hs).register(http_server)
     DelayedEventsServlet(hs).register(http_server)