diff --git a/changelog.d/19354.feature b/changelog.d/19354.feature new file mode 100644 index 0000000000..3d4e79e687 --- /dev/null +++ b/changelog.d/19354.feature @@ -0,0 +1 @@ +MSC4140: Add a dedicated endpoint for scheduling delayed events. diff --git a/docker/configure_workers_and_start.py b/docker/configure_workers_and_start.py index 1f97ef12fd..1fea8d1053 100755 --- a/docker/configure_workers_and_start.py +++ b/docker/configure_workers_and_start.py @@ -203,6 +203,7 @@ WORKERS_CONFIG: dict[str, dict[str, Any]] = { "^/_matrix/client/(api/v1|r0|v3|unstable)/keys/device_signing/upload$", "^/_matrix/client/(api/v1|r0|v3|unstable)/keys/signatures/upload$", "^/_matrix/client/unstable/org.matrix.msc4140/delayed_events(/[^/]+(/restart)?)?$", + "^/_matrix/client/unstable/org.matrix.msc4140/rooms/.*/delayed_event/", ], "shared_extra_conf": {}, "worker_extra_conf": "", diff --git a/docs/workers.md b/docs/workers.md index 6984c1337e..67f00e1a9f 100644 --- a/docs/workers.md +++ b/docs/workers.md @@ -288,7 +288,8 @@ information. # User directory search requests ^/_matrix/client/(r0|v3|unstable)/user_directory/search$ - # Unstable MSC4140 support + # Unstable MSC4140 support (delayed event scheduling & management) + ^/_matrix/client/unstable/org.matrix.msc4140/rooms/.*/delayed_event/ ^/_matrix/client/unstable/org.matrix.msc4140/delayed_events(/[^/]+(/restart)?)?$ # Stabilised Delegated Authentication support (`matrix_authentication_service.enabled: true`) diff --git a/synapse/api/constants.py b/synapse/api/constants.py index bfdb3ef926..9d3e86a0d4 100644 --- a/synapse/api/constants.py +++ b/synapse/api/constants.py @@ -477,13 +477,16 @@ class StickyEventField(TypedDict): class StickyEvent: - QUERY_PARAM_NAME: Final = "org.matrix.msc4354.sticky_duration_ms" + REQUEST_PARAM_NAME: Final = "org.matrix.msc4354.sticky_duration_ms" """ - Query parameter used by clients for setting the sticky duration of an event they are sending. + Parameter used by clients for setting the sticky duration of an event they are sending. - Applies to: + Used as a query parameter for: - /rooms/.../send/... - /rooms/.../state/... + + Used as a request body field for: + - /rooms/.../delayed_event/... """ EVENT_FIELD_NAME: Final = "msc4354_sticky" diff --git a/synapse/handlers/delayed_events.py b/synapse/handlers/delayed_events.py index 7ce3bfdd37..f652f9b429 100644 --- a/synapse/handlers/delayed_events.py +++ b/synapse/handlers/delayed_events.py @@ -68,6 +68,7 @@ class DelayedEventsHandler: self._auth = hs.get_auth() self._event_creation_handler = hs.get_event_creation_handler() self._room_member_handler = hs.get_room_member_handler() + self._spam_checker_module_callbacks = hs.get_module_api_callbacks().spam_checker self._request_ratelimiter = hs.get_request_ratelimiter() @@ -369,20 +370,13 @@ class DelayedEventsHandler: Raises: SynapseError: if the delayed event fails validation checks, or - if the requested delay is longer than allowed, or - if sending delayed events has been disallowed entirely. + if the requested delay is longer than allowed. """ # Use standard request limiter for scheduling new delayed events. # TODO: Instead apply ratelimiting based on the scheduled send time. # See https://github.com/element-hq/synapse/issues/18021 await self._request_ratelimiter.ratelimit(requester) - if not self._config.server.msc4140_enabled: - raise SynapseError( - HTTPStatus.FORBIDDEN, - "Sending delayed events has been disallowed", - Codes.FORBIDDEN, - ) if delay > self._config.server.max_event_delay_duration: requested_delay = delay.as_millis() max_delay = self._config.server.max_event_delay_duration.as_millis() @@ -625,6 +619,7 @@ class DelayedEventsHandler: ) -> None: user_id = UserID(event.user_localpart, self._config.server.server_name) user_id_str = user_id.to_string() + room_id_str = event.room_id.to_string() # Create a new requester from what data is currently available requester = create_requester( user_id, @@ -633,13 +628,31 @@ class DelayedEventsHandler: ) try: + if event.state_key is not None: + is_requester_admin = await self._auth.is_server_admin(requester) + if not is_requester_admin: + spam_check = await self._spam_checker_module_callbacks.user_may_send_state_event( + user_id=user_id_str, + room_id=room_id_str, + event_type=event.type, + state_key=event.state_key, + content=event.content, + ) + if spam_check != self._spam_checker_module_callbacks.NOT_SPAM: + raise SynapseError( + 403, + "You are not permitted to send the state event", + errcode=spam_check[0], + additional_fields=spam_check[1], + ) + if event.state_key is not None and event.type == EventTypes.Member: membership = event.content.get("membership") assert membership is not None event_id, _ = await self._room_member_handler.update_membership( requester, target=UserID.from_string(event.state_key), - room_id=event.room_id.to_string(), + room_id=room_id_str, action=membership, content=event.content, origin_server_ts=event.origin_server_ts, @@ -649,7 +662,7 @@ class DelayedEventsHandler: event_dict: JsonDict = { "type": event.type, "content": event.content, - "room_id": event.room_id.to_string(), + "room_id": room_id_str, "sender": user_id_str, } diff --git a/synapse/rest/client/room.py b/synapse/rest/client/room.py index 4ffb6a89b3..160628ef16 100644 --- a/synapse/rest/client/room.py +++ b/synapse/rest/client/room.py @@ -25,11 +25,12 @@ import logging import re from enum import Enum from http import HTTPStatus -from typing import TYPE_CHECKING, Awaitable +from typing import TYPE_CHECKING, Awaitable, NoReturn from urllib import parse as urlparse import attr from prometheus_client.core import Histogram +from pydantic import Field, PositiveInt, StrictStr from twisted.web.server import Request @@ -63,6 +64,7 @@ from synapse.http.servlet import ( ResolveRoomIdMixin, RestServlet, assert_params_in_dict, + parse_and_validate_json_object_from_request, parse_boolean, parse_enum, parse_integer, @@ -80,7 +82,16 @@ from synapse.rest.client.transactions import HttpTransactionCache from synapse.state import CREATE_KEY, POWER_KEY from synapse.storage.databases.main import DataStore from synapse.streams.config import PaginationConfig -from synapse.types import JsonDict, Requester, StreamToken, ThirdPartyInstanceID, UserID +from synapse.types import ( + Absent, + AbsentType, + JsonDict, + Requester, + StreamToken, + ThirdPartyInstanceID, + UserID, +) +from synapse.types.rest import RequestBodyModel from synapse.types.state import StateFilter from synapse.util.cancellation import cancellable from synapse.util.clock import Clock @@ -217,6 +228,7 @@ class RoomStateEventRestServlet(RestServlet): self.clock = hs.get_clock() self._event_serializer = hs.get_event_client_serializer() self._spam_checker_module_callbacks = hs.get_module_api_callbacks().spam_checker + self._msc4140_enabled = hs.config.server.msc4140_enabled def register(self, http_server: HttpServer) -> None: # /rooms/$roomid/state/$eventtype @@ -315,6 +327,28 @@ class RoomStateEventRestServlet(RestServlet): content = parse_json_object_from_request(request) + origin_server_ts = None + if requester.app_service_id: + origin_server_ts = parse_integer(request, "ts") + + # FIXME(MSC4140): remove by 2026-12-31, once RoomDelayedEventRestServlet has been stable for a suitable amount of time. + delay = _parse_request_for_delayed_event_delay(request, self._msc4140_enabled) + if delay is not None: + delay_id = await self.delayed_events_handler.add( + requester, + room_id=room_id, + event_type=event_type, + state_key=state_key, + origin_server_ts=origin_server_ts, + content=content, + delay=delay, + sticky_duration_ms=None, + ) + + set_tag("delay_id", delay_id) + ret = {"delay_id": delay_id} + return 200, ret + is_requester_admin = await self.auth.is_server_admin(requester) if not is_requester_admin: spam_check = ( @@ -334,27 +368,6 @@ class RoomStateEventRestServlet(RestServlet): additional_fields=spam_check[1], ) - origin_server_ts = None - if requester.app_service_id: - origin_server_ts = parse_integer(request, "ts") - - delay = _parse_request_for_delayed_event_delay(request) - if delay is not None: - delay_id = await self.delayed_events_handler.add( - requester, - room_id=room_id, - event_type=event_type, - state_key=state_key, - origin_server_ts=origin_server_ts, - content=content, - delay=delay, - sticky_duration_ms=None, - ) - - set_tag("delay_id", delay_id) - ret = {"delay_id": delay_id} - return 200, ret - try: if event_type == EventTypes.Member: membership = content.get("membership", None) @@ -407,6 +420,7 @@ class RoomSendEventRestServlet(TransactionRestServlet): self.event_creation_handler = hs.get_event_creation_handler() self.delayed_events_handler = hs.get_delayed_events_handler() self.auth = hs.get_auth() + self._msc4140_enabled = hs.config.server.msc4140_enabled self._msc4354_enabled = hs.config.experimental.msc4354_enabled def register(self, http_server: HttpServer) -> None: @@ -430,9 +444,10 @@ class RoomSendEventRestServlet(TransactionRestServlet): sticky_duration_ms: int | None = None if self._msc4354_enabled: - sticky_duration_ms = parse_integer(request, StickyEvent.QUERY_PARAM_NAME) + sticky_duration_ms = parse_integer(request, StickyEvent.REQUEST_PARAM_NAME) - delay = _parse_request_for_delayed_event_delay(request) + # FIXME(MSC4140): remove by 2026-12-31, once RoomDelayedEventRestServlet has been stable for a suitable amount of time. + delay = _parse_request_for_delayed_event_delay(request, self._msc4140_enabled) if delay is not None: delay_id = await self.delayed_events_handler.add( requester, @@ -505,7 +520,118 @@ class RoomSendEventRestServlet(TransactionRestServlet): ) -def _parse_request_for_delayed_event_delay(request: SynapseRequest) -> Duration | None: +class RoomDelayedEventRestServlet(TransactionRestServlet): + CATEGORY = "Delayed event management requests" + PATTERNS = client_patterns( + "/org.matrix.msc4140/rooms/(?P[^/]*)/delayed_event/(?P[^/]*)/(?P[^/]*)$", + releases=(), + unstable=True, + ) + + def __init__(self, hs: "HomeServer"): + super().__init__(hs) + self.delayed_events_handler = hs.get_delayed_events_handler() + self.auth = hs.get_auth() + self._msc4140_enabled = hs.config.server.msc4140_enabled + self._msc4354_enabled = hs.config.experimental.msc4354_enabled + + async def on_PUT( + self, request: SynapseRequest, room_id: str, event_type: str, txn_id: str + ) -> tuple[int, JsonDict]: + requester = await self.auth.get_user_by_req(request, allow_guest=True) + set_tag("txn_id", txn_id) + + return await self.txns.fetch_or_execute_request( + request, + requester, + self._do, + request, + requester, + room_id, + event_type, + ) + + class DelayedEventBodyModel(RequestBodyModel): + delay_ms: PositiveInt + content: JsonDict + state_key: StrictStr | AbsentType = Absent + + class StickyDelayedEventBodyModel(DelayedEventBodyModel): + # MSC4354 allows any integer value in the request, not just positives + sticky_duration_ms: int | AbsentType = Field( + Absent, alias=StickyEvent.REQUEST_PARAM_NAME + ) + + async def _do( + self, + request: SynapseRequest, + requester: Requester, + room_id: str, + event_type: str, + ) -> tuple[int, JsonDict]: + if not self._msc4140_enabled: + _raise_delayed_events_unsupported() + + if not self._msc4354_enabled: + # When the MSC is disabled, skip validating the sticky duration field + request_body = parse_and_validate_json_object_from_request( + request, self.DelayedEventBodyModel + ) + sticky_duration_ms = None + else: + # Support the parameter in both the query & body until the MSC settles on which one to use + request_body = parse_and_validate_json_object_from_request( + request, self.StickyDelayedEventBodyModel + ) + sticky_duration_ms = parse_integer( + request, + StickyEvent.REQUEST_PARAM_NAME, + negative=True, # MSC4354 allows any integer value in the request, not just positives + ) + if request_body.sticky_duration_ms is not Absent: + if sticky_duration_ms is None: + sticky_duration_ms = request_body.sticky_duration_ms + elif request_body.sticky_duration_ms != sticky_duration_ms: + raise SynapseError( + HTTPStatus.BAD_REQUEST, + f"Conflicting values given for {StickyEvent.REQUEST_PARAM_NAME}", + Codes.INVALID_PARAM, + ) + + if request_body.state_key is not Absent: + if sticky_duration_ms is not None: + raise SynapseError( + HTTPStatus.BAD_REQUEST, + "Sticky state events are not allowed. Cannot specify both `state_key` and `sticky_duration_ms`", + Codes.INVALID_PARAM, + ) + state_key = request_body.state_key + else: + state_key = None + + origin_server_ts = None + if requester.app_service_id: + origin_server_ts = parse_integer(request, "ts") + + delay_id = await self.delayed_events_handler.add( + requester, + room_id=room_id, + event_type=event_type, + state_key=state_key, + origin_server_ts=origin_server_ts, + content=request_body.content, + delay=Duration(milliseconds=request_body.delay_ms), + sticky_duration_ms=sticky_duration_ms, + ) + + set_tag("delay_id", delay_id) + ret = {"delay_id": delay_id} + return 200, ret + + +def _parse_request_for_delayed_event_delay( + request: SynapseRequest, msc4140_enabled: bool +) -> Duration | None: """Parses from the request string the delay parameter for delayed event requests, and checks it for correctness. @@ -517,8 +643,29 @@ def _parse_request_for_delayed_event_delay(request: SynapseRequest) -> Duration Raises: SynapseError: if the delay parameter is present and invalid. """ - delay_ms = parse_integer(request, "org.matrix.msc4140.delay") - return Duration(milliseconds=delay_ms) if delay_ms is not None else None + delay_param_name = "org.matrix.msc4140.delay" + # Allow negatives here to validate the delay only if delayed events are enabled, + # and so that any non-positive value is rejected with the same error + delay_ms = parse_integer(request, delay_param_name, negative=True) + if delay_ms is None: + return None + if not msc4140_enabled: + _raise_delayed_events_unsupported() + if delay_ms <= 0: + raise SynapseError( + HTTPStatus.BAD_REQUEST, + f"Query parameter {delay_param_name} must be an integer greater than zero.", + Codes.INVALID_PARAM, + ) + return Duration(milliseconds=delay_ms) + + +def _raise_delayed_events_unsupported() -> NoReturn: + raise SynapseError( + HTTPStatus.FORBIDDEN, + "Sending delayed events is disabled on this homeserver", + Codes.FORBIDDEN, + ) # TODO: Needs unit testing for room ID + alias joins @@ -1750,6 +1897,7 @@ def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None: SearchRestServlet(hs).register(http_server) RoomCreateRestServlet(hs).register(http_server) TimestampLookupRestServlet(hs).register(http_server) + RoomDelayedEventRestServlet(hs).register(http_server) # Some servlets only get registered for the main process. if hs.config.worker.worker_app is None: diff --git a/tests/module_api/test_spamchecker.py b/tests/module_api/test_spamchecker.py index 572f7bdbcf..839d5d1cd5 100644 --- a/tests/module_api/test_spamchecker.py +++ b/tests/module_api/test_spamchecker.py @@ -15,6 +15,8 @@ from http import HTTPStatus from typing import Literal +from parameterized import parameterized + from twisted.internet.testing import MemoryReactor from synapse.api.constants import ( @@ -26,10 +28,11 @@ 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.rest import admin, delayed_events, login, room, room_upgrade_rest_servlet from synapse.server import HomeServer from synapse.types import Codes, JsonDict from synapse.util.clock import Clock +from synapse.util.duration import Duration from tests import unittest from tests.server import FakeChannel @@ -42,6 +45,7 @@ class SpamCheckerTestCase(HomeserverTestCase): servlets = [ room.register_servlets, admin.register_servlets, + delayed_events.register_servlets, login.register_servlets, room_upgrade_rest_servlet.register_servlets, ] @@ -296,6 +300,144 @@ class SpamCheckerTestCase(HomeserverTestCase): self.assertEqual(channel.code, 403) self.assertEqual(channel.json_body["errcode"], Codes.FORBIDDEN) + @parameterized.expand((False, True)) + @unittest.override_config({"max_event_delay_duration": "24h"}) + def test_user_may_send_state_event_with_delay(self, send_manually: bool) -> None: + """Test that the user_may_send_state_event callback is called when a delayed state event + is sent (not when it is scheduled), and that it receives the correct parameters. + """ + + async def user_may_send_state_event( + user_id: str, + room_id: str, + event_type: str, + state_key: str, + content: JsonDict, + ) -> Literal["NOT_SPAM"] | Codes: + self.did_spam_check = True + self.last_user_id = user_id + self.last_room_id = room_id + self.last_event_type = event_type + self.last_state_key = state_key + self.last_content = content + return "NOT_SPAM" + + self._module_api.register_spam_checker_callbacks( + user_may_send_state_event=user_may_send_state_event + ) + self.did_spam_check = False + + channel = self.create_room({}) + + self.assertEqual(channel.code, 200) + + room_id = channel.json_body["room_id"] + + event_type = "test.event.type" + state_key = "test.state.key" + delay = Duration(seconds=10) + channel = self.make_request( + "PUT", + f"/_matrix/client/unstable/org.matrix.msc4140/rooms/{room_id}/delayed_event/{event_type}/mid", + content={ + "delay_ms": delay.as_millis(), + "state_key": state_key, + "content": {"foo": "bar"}, + }, + access_token=self.token, + ) + + self.assertEqual(channel.code, 200) + self.assertFalse(self.did_spam_check) + + delay_id = channel.json_body["delay_id"] + + if send_manually: + channel = self.make_request( + "POST", + f"/_matrix/client/unstable/org.matrix.msc4140/delayed_events/{delay_id}/send", + access_token=self.token, + ) + else: + # Advance time enough so the delayed event is sent + self.reactor.advance(delay.as_secs()) + + self.assertEqual(channel.code, 200) + self.assertTrue(self.did_spam_check) + self.assertEqual(self.last_user_id, self.user_id) + self.assertEqual(self.last_room_id, room_id) + self.assertEqual(self.last_event_type, event_type) + self.assertEqual(self.last_state_key, state_key) + self.assertEqual(self.last_content, {"foo": "bar"}) + + @parameterized.expand((False, True)) + @unittest.override_config({"max_event_delay_duration": "24h"}) + def test_user_may_send_state_event_disallows_with_delay( + self, send_manually: bool + ) -> None: + """Test that the user_may_send_state_event callback is called when a delayed state event + is sent (not when it is scheduled), and that the response is honoured. + """ + + async def user_may_send_state_event( + user_id: str, + room_id: str, + event_type: str, + state_key: str, + content: JsonDict, + ) -> Literal["NOT_SPAM"] | Codes: + return Codes.FORBIDDEN + + self._module_api.register_spam_checker_callbacks( + user_may_send_state_event=user_may_send_state_event + ) + self.did_spam_check = False + + channel = self.create_room({}) + + self.assertEqual(channel.code, 200) + + room_id = channel.json_body["room_id"] + + event_type = "test.event.type" + state_key = "test.state.key" + delay = Duration(seconds=10) + channel = self.make_request( + "PUT", + f"/_matrix/client/unstable/org.matrix.msc4140/rooms/{room_id}/delayed_event/{event_type}/mid", + content={ + "delay_ms": delay.as_millis(), + "state_key": state_key, + "content": {"foo": "bar"}, + }, + access_token=self.token, + ) + + self.assertEqual(channel.code, 200) + self.assertFalse(self.did_spam_check) + + delay_id = channel.json_body["delay_id"] + + if send_manually: + channel = self.make_request( + "POST", + f"/_matrix/client/unstable/org.matrix.msc4140/delayed_events/{delay_id}/send", + access_token=self.token, + ) + self.assertEqual(channel.code, 403) + self.assertEqual(channel.json_body["errcode"], Codes.FORBIDDEN) + else: + # Advance time enough so the delayed event is sent + self.reactor.advance(delay.as_secs()) + + # The state should not have been set by the delayed event + channel = self.make_request( + "GET", + f"/_matrix/client/r0/rooms/{room_id}/state/{event_type}/{state_key}", + access_token=self.token, + ) + self.assertEqual(channel.code, 404) + class FederatedEventSpamCheckMetadataTestCase(unittest.FederatingHomeserverTestCase): servlets = [ diff --git a/tests/rest/client/test_delayed_events.py b/tests/rest/client/test_delayed_events.py index 6877656e7e..8405dedcf9 100644 --- a/tests/rest/client/test_delayed_events.py +++ b/tests/rest/client/test_delayed_events.py @@ -14,23 +14,34 @@ """Tests REST events for /delayed_events paths.""" +import sqlite3 from http import HTTPStatus +from typing import Literal, overload from unittest.mock import AsyncMock from parameterized import parameterized from twisted.internet.testing import MemoryReactor +from synapse.api.constants import ( + EventContentFields, + EventTypes, + EventUnsignedContentFields, + GuestAccess, + StickyEvent, +) from synapse.api.errors import Codes, NotFoundError from synapse.rest import admin from synapse.rest.client import ( account, delayed_events, login, + register, room, sync, versions, ) +from synapse.rest.client.delayed_events import _UpdateDelayedEventAction from synapse.server import HomeServer from synapse.types import JsonDict, create_requester from synapse.util.clock import Clock @@ -38,10 +49,11 @@ from synapse.util.duration import Duration from tests import unittest from tests.replication._base import BaseMultiWorkerStreamTestCase -from tests.server import FakeChannel +from tests.server import FakeChannel, make_request from tests.unittest import HomeserverTestCase +from tests.utils import USE_POSTGRES_FOR_TESTS -PATH_PREFIX = "/_matrix/client/unstable/org.matrix.msc4140/delayed_events" +_MANAGEMENT_PATH_PREFIX = "/_matrix/client/unstable/org.matrix.msc4140/delayed_events" _EVENT_TYPE = "com.example.test" @@ -71,14 +83,111 @@ class DelayedEventsUnstableSupportTestCase(HomeserverTestCase): self.assertTrue(channel.json_body["unstable_features"]["org.matrix.msc4140"]) -class DelayedEventsTestCase(HomeserverTestCase): - """Tests getting and managing delayed events.""" +class DelayedEventsHelperMixin(HomeserverTestCase): + room_id: str + + def _make_delayed_event_request( + self, + *, + room_id: str, + delay: Duration, + event_type: str, + state_key: str | None = None, + content: JsonDict, + txn_id: str | None = None, + access_token: str, + ) -> FakeChannel: + """ + Build and send a request for scheduling a delayed event via the dedicated endpoint. + + Args: + room_id: The room to send the event to. + delay: How long to wait before sending the event. + event_type: The type of the event to send. + state_key: The state key of the event, or None for a message event. + content: The content of the event. + txn_id: The transaction ID for a delayed message event request. + Ignored for a delayed state event request. + Otherwise, generated if not given. + access_token: The access token of the user to make the request for. + + Returns: + The `FakeChannel` object which stores the result of the request. + """ + return self.make_request( + "PUT", + self.helper.build_delayed_event_request_path( + room_id=room_id, + event_type=event_type, + txn_id=txn_id, + ), + self.helper.build_delayed_event_request_body( + delay=delay, + state_key=state_key, + content=content, + ), + access_token, + ) + + @overload + def _check_for_delayed_event_in_sync( + self, access_token: str, delay_id: str, should_find: Literal[True] + ) -> JsonDict: ... + + @overload + def _check_for_delayed_event_in_sync( + self, access_token: str, delay_id: str, should_find: Literal[False] + ) -> None: ... + + def _check_for_delayed_event_in_sync( + self, access_token: str, delay_id: str, should_find: bool + ) -> JsonDict | None: + """Call /sync and look for a synced event with a specified delay_id. + At most one event will ever have a matching delay_id. + + Args: + access_token: The access token of the user to call /sync for. + delay_id: The delay_id to search for in synced events. + should_find: Whether /sync should include an event with a matching delay_id. + + Returns: + The synced event with the matching delay_id, if any. + """ + channel = self.make_request("GET", "/sync", access_token=access_token) + self.assertEqual(HTTPStatus.OK, channel.code) + + rooms = channel.json_body["rooms"] + events = [] + for membership in "join", "leave": + if membership in rooms: + events += rooms[membership][self.room_id]["timeline"]["events"] + + found: JsonDict | None = None + for event in events: + if event["unsigned"].get("org.matrix.msc4140.delay_id") == delay_id: + if not should_find: + self.fail( + f"Found event {event['event_id']} with matching delay_id {delay_id}, but expected to not find one" + ) + if found is not None: + self.fail( + f"Events {found['event_id']} & {event['event_id']} have matching delay_id {delay_id}, but delay_id should be unique" + ) + found = event + if should_find and found is None: + self.fail(f"Did not find any event with matching delay_id {delay_id}") + return found + + +class DelayedEventsTestCaseBase(DelayedEventsHelperMixin): + """Room and user fixtures, and request helpers, for the delayed events tests.""" servlets = [ admin.register_servlets, account.register_servlets, delayed_events.register_servlets, login.register_servlets, + register.register_servlets, room.register_servlets, sync.register_servlets, ] @@ -115,19 +224,68 @@ class DelayedEventsTestCase(HomeserverTestCase): # affect the rate-limits in the test itself self.reactor.advance(Duration(days=1).as_secs()) + def _get_delayed_events(self) -> list[JsonDict]: + channel = self.make_request( + "GET", + _MANAGEMENT_PATH_PREFIX, + access_token=self.user1_access_token, + ) + self.assertEqual(HTTPStatus.OK, channel.code, channel.result) + + events = channel.json_body["delayed_events"] + self.assertIsInstance(events, list) + + return events + + def _get_delayed_event_content(self, event: JsonDict) -> JsonDict: + content = event["content"] + self.assertIsInstance(content, dict) + + return content + + def _update_delayed_event( + self, + delay_id: str, + action: _UpdateDelayedEventAction, + action_in_path: bool, + access_token: str | None = None, + ) -> FakeChannel: + """ + Call the endpoint to cancel/restart/send a scheduled delayed event. + + Args: + delay_id: The ID of the delayed event to manage. + action: The management action to take on the target delayed event. + action_in_path: If False, uses the deprecated version of this endpoint + that expects the action in the request body instead of the path. + access_token: The access token of the user to call the endpoint for. + """ + path = f"{_MANAGEMENT_PATH_PREFIX}/{delay_id}" + body = {} + if action_in_path: + path += f"/{action.value}" + else: + body["action"] = action.value + return self.make_request("POST", path, body, access_token) + + +class DelayedEventsTestCase(DelayedEventsTestCaseBase): + """Tests getting and managing delayed events.""" + def test_delayed_events_empty_on_startup(self) -> None: self.assertListEqual([], self._get_delayed_events()) def test_delayed_event_lookup(self) -> None: # Schedule a message event - delay_ms = 100000 + delay = Duration(milliseconds=100000) content: JsonDict = {"message": "hello"} delayed_since_ts = self.hs.get_clock().time_msec() - channel = self.make_request( - "POST", - _get_path_for_delayed_send(self.room_id, _EVENT_TYPE, delay_ms), - content, - self.user1_access_token, + channel = self._make_delayed_event_request( + room_id=self.room_id, + delay=delay, + event_type=_EVENT_TYPE, + content=content, + access_token=self.user1_access_token, ) self.assertEqual(channel.code, HTTPStatus.OK, channel.result) delay_id = channel.json_body["delay_id"] @@ -135,7 +293,7 @@ class DelayedEventsTestCase(HomeserverTestCase): # Test that the scheduled delayed event can be retrieved channel = self.make_request( "GET", - f"{PATH_PREFIX}/{delay_id}", + f"{_MANAGEMENT_PATH_PREFIX}/{delay_id}", access_token=self.user1_access_token, ) self.assertEqual(channel.code, HTTPStatus.OK, channel.result) @@ -148,7 +306,7 @@ class DelayedEventsTestCase(HomeserverTestCase): "delay_id": delay_id, "room_id": self.room_id, "type": _EVENT_TYPE, - "delay_ms": delay_ms, + "delay_ms": delay.as_millis(), "delayed_since_ts": delayed_since_ts, "content": content, }, @@ -157,7 +315,7 @@ class DelayedEventsTestCase(HomeserverTestCase): # Test that a non-existent delayed event cannot be found channel = self.make_request( "GET", - f"{PATH_PREFIX}/{delay_id}-fake", + f"{_MANAGEMENT_PATH_PREFIX}/{delay_id}-fake", access_token=self.user1_access_token, ) self.assertEqual(channel.code, HTTPStatus.NOT_FOUND, channel.result) @@ -165,7 +323,7 @@ class DelayedEventsTestCase(HomeserverTestCase): # Test that other users cannot access this delayed event channel = self.make_request( "GET", - f"{PATH_PREFIX}/{delay_id}", + f"{_MANAGEMENT_PATH_PREFIX}/{delay_id}", access_token=self.user2_access_token, ) self.assertEqual(channel.code, HTTPStatus.NOT_FOUND, channel.result) @@ -173,18 +331,18 @@ class DelayedEventsTestCase(HomeserverTestCase): # Now schedule a state event. # Do it in this test, as opposed to a new one, to confirm that the correct delayed event # is retrieved when multiple delayed events have been scheduled. - delay_ms += 2000 + delay += Duration(milliseconds=2000) state_key = "" state_event_type = _EVENT_TYPE + "_state" content = {"state_message": "greetings"} delayed_since_ts = self.hs.get_clock().time_msec() - channel = self.make_request( - "PUT", - _get_path_for_delayed_state( - self.room_id, state_event_type, state_key, delay_ms - ), - content, - self.user1_access_token, + channel = self._make_delayed_event_request( + room_id=self.room_id, + delay=delay, + event_type=state_event_type, + state_key=state_key, + content=content, + access_token=self.user1_access_token, ) self.assertEqual(channel.code, HTTPStatus.OK, channel.result) delay_id_2 = channel.json_body["delay_id"] @@ -195,7 +353,7 @@ class DelayedEventsTestCase(HomeserverTestCase): # Test that the scheduled delayed event can be retrieved channel = self.make_request( "GET", - f"{PATH_PREFIX}/{delay_id_2}", + f"{_MANAGEMENT_PATH_PREFIX}/{delay_id_2}", access_token=self.user1_access_token, ) self.assertEqual(channel.code, HTTPStatus.OK, channel.result) @@ -209,7 +367,7 @@ class DelayedEventsTestCase(HomeserverTestCase): "room_id": self.room_id, "type": state_event_type, "state_key": state_key, - "delay_ms": delay_ms, + "delay_ms": delay.as_millis(), "delayed_since_ts": delayed_since_ts, "content": content, }, @@ -237,13 +395,15 @@ class DelayedEventsTestCase(HomeserverTestCase): setter_key = "setter" setter_expected = "on_timeout" - channel = self.make_request( - "PUT", - _get_path_for_delayed_state(self.room_id, _EVENT_TYPE, state_key, 900), - { + channel = self._make_delayed_event_request( + room_id=self.room_id, + delay=Duration(milliseconds=900), + event_type=_EVENT_TYPE, + state_key=state_key, + content={ setter_key: setter_expected, }, - self.user1_access_token, + access_token=self.user1_access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) delay_id = channel.json_body.get("delay_id") @@ -261,7 +421,8 @@ class DelayedEventsTestCase(HomeserverTestCase): expect_code=HTTPStatus.NOT_FOUND, ) - self.reactor.advance(1) + # Advance time enough so the delayed event is sent + self.reactor.advance(Duration(seconds=1).as_secs()) self.assertListEqual([], self._get_delayed_events()) content = self.helper.get_state( self.room_id, @@ -271,23 +432,54 @@ class DelayedEventsTestCase(HomeserverTestCase): ) self.assertEqual(setter_expected, content.get(setter_key), content) - self._find_sent_delayed_event(self.user1_access_token, delay_id, True) - self._find_sent_delayed_event(self.user2_access_token, delay_id, False) + self._check_for_delayed_event_in_sync(self.user1_access_token, delay_id, True) + self._check_for_delayed_event_in_sync(self.user2_access_token, delay_id, False) + + @unittest.override_config({"allow_guest_access": True}) + def test_guest_can_schedule_delayed_event(self) -> None: + """Test that a guest user can schedule a delayed event.""" + channel = self.make_request("POST", "/register?kind=guest", {}) + self.assertEqual(HTTPStatus.OK, channel.code, channel.result) + guest_user_id = channel.json_body["user_id"] + guest_access_token = channel.json_body["access_token"] + + self.helper.send_state( + self.room_id, + EventTypes.GuestAccess, + {EventContentFields.GUEST_ACCESS: GuestAccess.CAN_JOIN}, + tok=self.user1_access_token, + ) + self.helper.join(room=self.room_id, user=guest_user_id, tok=guest_access_token) + + channel = self._make_delayed_event_request( + room_id=self.room_id, + delay=Duration(milliseconds=900), + event_type=_EVENT_TYPE, + content={"body": "from a guest"}, + access_token=guest_access_token, + ) + self.assertEqual(HTTPStatus.OK, channel.code, channel.result) + delay_id = channel.json_body.get("delay_id") + assert delay_id is not None + + # Advance time enough so the delayed event is sent + self.reactor.advance(Duration(seconds=1).as_secs()) + event = self._check_for_delayed_event_in_sync( + guest_access_token, delay_id, True + ) + self.assertEqual(guest_user_id, event["sender"], event) def test_delayed_member_events_are_sent_on_timeout(self) -> None: - channel = self.make_request( - "PUT", - _get_path_for_delayed_state( - self.room_id, - "m.room.member", - self.user2_user_id, - 900, - ), - { + channel = self._make_delayed_event_request( + room_id=self.room_id, + delay=Duration(milliseconds=900), + event_type="m.room.member", + state_key=self.user2_user_id, + content={ "membership": "leave", "reason": "Delayed kick", }, - self.user1_access_token, + access_token=self.user1_access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) delay_id = channel.json_body.get("delay_id") @@ -307,7 +499,8 @@ class DelayedEventsTestCase(HomeserverTestCase): ) self.assertEqual("join", content.get("membership"), content) - self.reactor.advance(1) + # Advance time enough so the delayed event is sent + self.reactor.advance(Duration(seconds=1).as_secs()) self.assertListEqual([], self._get_delayed_events()) content = self.helper.get_state( self.room_id, @@ -318,18 +511,18 @@ class DelayedEventsTestCase(HomeserverTestCase): self.assertEqual("leave", content.get("membership"), content) self.assertEqual("Delayed kick", content.get("reason"), content) - self._find_sent_delayed_event(self.user1_access_token, delay_id, True) - self._find_sent_delayed_event(self.user2_access_token, delay_id, False) + self._check_for_delayed_event_in_sync(self.user1_access_token, delay_id, True) + self._check_for_delayed_event_in_sync(self.user2_access_token, delay_id, False) def test_get_delayed_events_auth(self) -> None: - channel = self.make_request("GET", PATH_PREFIX) + channel = self.make_request("GET", _MANAGEMENT_PATH_PREFIX) self.assertEqual(HTTPStatus.UNAUTHORIZED, channel.code, channel.result) @unittest.override_config( {"rc_delayed_event_mgmt": {"per_second": 0.5, "burst_count": 1}} ) def test_get_delayed_events_ratelimit(self) -> None: - args = ("GET", PATH_PREFIX, b"", self.user1_access_token) + args = ("GET", _MANAGEMENT_PATH_PREFIX, b"", self.user1_access_token) channel = self.make_request(*args) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) @@ -351,14 +544,14 @@ class DelayedEventsTestCase(HomeserverTestCase): def test_update_delayed_event_without_id(self) -> None: channel = self.make_request( "POST", - f"{PATH_PREFIX}/", + f"{_MANAGEMENT_PATH_PREFIX}/", ) self.assertEqual(HTTPStatus.NOT_FOUND, channel.code, channel.result) def test_update_delayed_event_without_body(self) -> None: channel = self.make_request( "POST", - f"{PATH_PREFIX}/abc", + f"{_MANAGEMENT_PATH_PREFIX}/abc", ) self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.result) self.assertEqual( @@ -369,7 +562,7 @@ class DelayedEventsTestCase(HomeserverTestCase): def test_update_delayed_event_without_action(self) -> None: channel = self.make_request( "POST", - f"{PATH_PREFIX}/abc", + f"{_MANAGEMENT_PATH_PREFIX}/abc", {}, ) self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.result) @@ -381,7 +574,7 @@ class DelayedEventsTestCase(HomeserverTestCase): def test_update_delayed_event_with_invalid_action(self) -> None: channel = self.make_request( "POST", - f"{PATH_PREFIX}/abc", + f"{_MANAGEMENT_PATH_PREFIX}/abc", {"action": "oops"}, ) self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.result) @@ -393,12 +586,14 @@ class DelayedEventsTestCase(HomeserverTestCase): @parameterized.expand( ( (action, action_in_path) - for action in ("cancel", "restart", "send") + for action in _UpdateDelayedEventAction for action_in_path in (True, False) ) ) def test_update_delayed_event_without_match( - self, action: str, action_in_path: bool + self, + action: _UpdateDelayedEventAction, + action_in_path: bool, ) -> None: channel = self._update_delayed_event("abc", action, action_in_path) self.assertEqual(HTTPStatus.NOT_FOUND, channel.code, channel.result) @@ -409,19 +604,22 @@ class DelayedEventsTestCase(HomeserverTestCase): setter_key = "setter" setter_expected = "none" - channel = self.make_request( - "PUT", - _get_path_for_delayed_state(self.room_id, _EVENT_TYPE, state_key, 1500), - { + channel = self._make_delayed_event_request( + room_id=self.room_id, + delay=Duration(milliseconds=1500), + event_type=_EVENT_TYPE, + state_key=state_key, + content={ setter_key: setter_expected, }, - self.user1_access_token, + access_token=self.user1_access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) delay_id = channel.json_body.get("delay_id") assert delay_id is not None - self.reactor.advance(1) + # Advance time enough so the delayed event is sent + self.reactor.advance(Duration(seconds=1).as_secs()) events = self._get_delayed_events() self.assertEqual(1, len(events), events) content = self._get_delayed_event_content(events[0]) @@ -434,11 +632,14 @@ class DelayedEventsTestCase(HomeserverTestCase): expect_code=HTTPStatus.NOT_FOUND, ) - channel = self._update_delayed_event(delay_id, "cancel", action_in_path) + channel = self._update_delayed_event( + delay_id, _UpdateDelayedEventAction.CANCEL, action_in_path + ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) self.assertListEqual([], self._get_delayed_events()) - self.reactor.advance(1) + # Advance time enough so the delayed event is sent + self.reactor.advance(Duration(seconds=1).as_secs()) content = self.helper.get_state( self.room_id, _EVENT_TYPE, @@ -447,8 +648,8 @@ class DelayedEventsTestCase(HomeserverTestCase): expect_code=HTTPStatus.NOT_FOUND, ) - self._find_sent_delayed_event(self.user1_access_token, delay_id, False) - self._find_sent_delayed_event(self.user2_access_token, delay_id, False) + self._check_for_delayed_event_in_sync(self.user1_access_token, delay_id, False) + self._check_for_delayed_event_in_sync(self.user2_access_token, delay_id, False) @parameterized.expand((True, False)) @unittest.override_config( @@ -457,11 +658,12 @@ class DelayedEventsTestCase(HomeserverTestCase): def test_cancel_delayed_event_ratelimit(self, action_in_path: bool) -> None: delay_ids = [] for _ in range(3): - channel = self.make_request( - "POST", - _get_path_for_delayed_send(self.room_id, _EVENT_TYPE, 100000), - {}, - self.user1_access_token, + channel = self._make_delayed_event_request( + room_id=self.room_id, + delay=Duration(milliseconds=100000), + event_type=_EVENT_TYPE, + content={}, + access_token=self.user1_access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) delay_id = channel.json_body.get("delay_id") @@ -469,22 +671,32 @@ class DelayedEventsTestCase(HomeserverTestCase): delay_ids.append(delay_id) delay_id = delay_ids.pop(0) - channel = self._update_delayed_event(delay_id, "cancel", action_in_path) - self.assertEqual(HTTPStatus.OK, channel.code, channel.result) - - delay_id = delay_ids.pop(0) - channel = self._update_delayed_event(delay_id, "cancel", action_in_path) - self.assertEqual(HTTPStatus.TOO_MANY_REQUESTS, channel.code, channel.result) - - # Using auth should bypass ratelimit applied against source IP channel = self._update_delayed_event( - delay_id, "cancel", action_in_path, self.user1_access_token + delay_id, _UpdateDelayedEventAction.CANCEL, action_in_path ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) delay_id = delay_ids.pop(0) channel = self._update_delayed_event( - delay_id, "cancel", action_in_path, self.user1_access_token + delay_id, _UpdateDelayedEventAction.CANCEL, action_in_path + ) + self.assertEqual(HTTPStatus.TOO_MANY_REQUESTS, channel.code, channel.result) + + # Using auth should bypass ratelimit applied against source IP + channel = self._update_delayed_event( + delay_id, + _UpdateDelayedEventAction.CANCEL, + action_in_path, + self.user1_access_token, + ) + self.assertEqual(HTTPStatus.OK, channel.code, channel.result) + + delay_id = delay_ids.pop(0) + channel = self._update_delayed_event( + delay_id, + _UpdateDelayedEventAction.CANCEL, + action_in_path, + self.user1_access_token, ) self.assertEqual(HTTPStatus.TOO_MANY_REQUESTS, channel.code, channel.result) @@ -497,7 +709,10 @@ class DelayedEventsTestCase(HomeserverTestCase): # Test that the request isn't ratelimited anymore. channel = self._update_delayed_event( - delay_id, "cancel", action_in_path, self.user1_access_token + delay_id, + _UpdateDelayedEventAction.CANCEL, + action_in_path, + self.user1_access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) @@ -514,19 +729,22 @@ class DelayedEventsTestCase(HomeserverTestCase): state_key = "to_send_on_request" content_property_name = "key" - channel = self.make_request( - "PUT", - _get_path_for_delayed_state(self.room_id, _EVENT_TYPE, state_key, 100000), - { + channel = self._make_delayed_event_request( + room_id=self.room_id, + delay=Duration(milliseconds=100000), + event_type=_EVENT_TYPE, + state_key=state_key, + content={ content_property_name: content_value, }, - self.user1_access_token, + access_token=self.user1_access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) delay_id = channel.json_body.get("delay_id") assert delay_id is not None - self.reactor.advance(1) + # Advance time enough so the delayed event is sent + self.reactor.advance(Duration(seconds=1).as_secs()) events = self._get_delayed_events() self.assertEqual(1, len(events), events) content = self._get_delayed_event_content(events[0]) @@ -539,7 +757,9 @@ class DelayedEventsTestCase(HomeserverTestCase): expect_code=HTTPStatus.NOT_FOUND, ) - channel = self._update_delayed_event(delay_id, "send", action_in_path) + channel = self._update_delayed_event( + delay_id, _UpdateDelayedEventAction.SEND, action_in_path + ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) self.assertListEqual([], self._get_delayed_events()) content = self.helper.get_state( @@ -550,29 +770,34 @@ class DelayedEventsTestCase(HomeserverTestCase): ) self.assertEqual(content_value, content.get(content_property_name), content) - self._find_sent_delayed_event(self.user1_access_token, delay_id, True) - self._find_sent_delayed_event(self.user2_access_token, delay_id, False) + self._check_for_delayed_event_in_sync(self.user1_access_token, delay_id, True) + self._check_for_delayed_event_in_sync(self.user2_access_token, delay_id, False) @parameterized.expand((True, False)) @unittest.override_config({"rc_message": {"per_second": 2.5, "burst_count": 3}}) def test_send_delayed_event_ratelimit(self, action_in_path: bool) -> None: delay_ids = [] for _ in range(2): - channel = self.make_request( - "POST", - _get_path_for_delayed_send(self.room_id, _EVENT_TYPE, 100000), - {}, - self.user1_access_token, + channel = self._make_delayed_event_request( + room_id=self.room_id, + delay=Duration(milliseconds=100000), + event_type=_EVENT_TYPE, + content={}, + access_token=self.user1_access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) delay_id = channel.json_body.get("delay_id") assert delay_id is not None delay_ids.append(delay_id) - channel = self._update_delayed_event(delay_ids.pop(0), "send", action_in_path) + channel = self._update_delayed_event( + delay_ids.pop(0), _UpdateDelayedEventAction.SEND, action_in_path + ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) - channel = self._update_delayed_event(delay_ids.pop(0), "send", action_in_path) + channel = self._update_delayed_event( + delay_ids.pop(0), _UpdateDelayedEventAction.SEND, action_in_path + ) self.assertEqual(HTTPStatus.TOO_MANY_REQUESTS, channel.code, channel.result) @parameterized.expand((True, False)) @@ -581,19 +806,22 @@ class DelayedEventsTestCase(HomeserverTestCase): setter_key = "setter" setter_expected = "on_timeout" - channel = self.make_request( - "PUT", - _get_path_for_delayed_state(self.room_id, _EVENT_TYPE, state_key, 1500), - { + channel = self._make_delayed_event_request( + room_id=self.room_id, + delay=Duration(milliseconds=1500), + event_type=_EVENT_TYPE, + state_key=state_key, + content={ setter_key: setter_expected, }, - self.user1_access_token, + access_token=self.user1_access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) delay_id = channel.json_body.get("delay_id") assert delay_id is not None - self.reactor.advance(1) + # Advance time enough so the delayed event is sent + self.reactor.advance(Duration(seconds=1).as_secs()) events = self._get_delayed_events() self.assertEqual(1, len(events), events) content = self._get_delayed_event_content(events[0]) @@ -606,10 +834,13 @@ class DelayedEventsTestCase(HomeserverTestCase): expect_code=HTTPStatus.NOT_FOUND, ) - channel = self._update_delayed_event(delay_id, "restart", action_in_path) + channel = self._update_delayed_event( + delay_id, _UpdateDelayedEventAction.RESTART, action_in_path + ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) - self.reactor.advance(1) + # Advance time enough so the delayed event is sent + self.reactor.advance(Duration(seconds=1).as_secs()) events = self._get_delayed_events() self.assertEqual(1, len(events), events) content = self._get_delayed_event_content(events[0]) @@ -622,7 +853,8 @@ class DelayedEventsTestCase(HomeserverTestCase): expect_code=HTTPStatus.NOT_FOUND, ) - self.reactor.advance(1) + # Advance time enough so the delayed event is sent + self.reactor.advance(Duration(seconds=1).as_secs()) self.assertListEqual([], self._get_delayed_events()) content = self.helper.get_state( self.room_id, @@ -632,8 +864,8 @@ class DelayedEventsTestCase(HomeserverTestCase): ) self.assertEqual(setter_expected, content.get(setter_key), content) - self._find_sent_delayed_event(self.user1_access_token, delay_id, True) - self._find_sent_delayed_event(self.user2_access_token, delay_id, False) + self._check_for_delayed_event_in_sync(self.user1_access_token, delay_id, True) + self._check_for_delayed_event_in_sync(self.user2_access_token, delay_id, False) @parameterized.expand((True, False)) @unittest.override_config( @@ -642,11 +874,12 @@ class DelayedEventsTestCase(HomeserverTestCase): def test_restart_delayed_event_ratelimit(self, action_in_path: bool) -> None: delay_ids = [] for _ in range(3): - channel = self.make_request( - "POST", - _get_path_for_delayed_send(self.room_id, _EVENT_TYPE, 100000), - {}, - self.user1_access_token, + channel = self._make_delayed_event_request( + room_id=self.room_id, + delay=Duration(milliseconds=100000), + event_type=_EVENT_TYPE, + content={}, + access_token=self.user1_access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) delay_id = channel.json_body.get("delay_id") @@ -654,22 +887,32 @@ class DelayedEventsTestCase(HomeserverTestCase): delay_ids.append(delay_id) delay_id = delay_ids.pop(0) - channel = self._update_delayed_event(delay_id, "restart", action_in_path) - self.assertEqual(HTTPStatus.OK, channel.code, channel.result) - - delay_id = delay_ids.pop(0) - channel = self._update_delayed_event(delay_id, "restart", action_in_path) - self.assertEqual(HTTPStatus.TOO_MANY_REQUESTS, channel.code, channel.result) - - # Using auth should bypass ratelimit applied against source IP channel = self._update_delayed_event( - delay_id, "restart", action_in_path, self.user1_access_token + delay_id, _UpdateDelayedEventAction.RESTART, action_in_path ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) delay_id = delay_ids.pop(0) channel = self._update_delayed_event( - delay_id, "restart", action_in_path, self.user1_access_token + delay_id, _UpdateDelayedEventAction.RESTART, action_in_path + ) + self.assertEqual(HTTPStatus.TOO_MANY_REQUESTS, channel.code, channel.result) + + # Using auth should bypass ratelimit applied against source IP + channel = self._update_delayed_event( + delay_id, + _UpdateDelayedEventAction.RESTART, + action_in_path, + self.user1_access_token, + ) + self.assertEqual(HTTPStatus.OK, channel.code, channel.result) + + delay_id = delay_ids.pop(0) + channel = self._update_delayed_event( + delay_id, + _UpdateDelayedEventAction.RESTART, + action_in_path, + self.user1_access_token, ) self.assertEqual(HTTPStatus.TOO_MANY_REQUESTS, channel.code, channel.result) @@ -682,7 +925,10 @@ class DelayedEventsTestCase(HomeserverTestCase): # Test that the request isn't ratelimited anymore. channel = self._update_delayed_event( - delay_id, "restart", action_in_path, self.user1_access_token + delay_id, + _UpdateDelayedEventAction.RESTART, + action_in_path, + self.user1_access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) @@ -693,13 +939,15 @@ class DelayedEventsTestCase(HomeserverTestCase): setter_key = "setter" setter_expected = "on_timeout" - channel = self.make_request( - "PUT", - _get_path_for_delayed_state(self.room_id, _EVENT_TYPE, state_key, 900), - { + channel = self._make_delayed_event_request( + room_id=self.room_id, + delay=Duration(milliseconds=900), + event_type=_EVENT_TYPE, + state_key=state_key, + content={ setter_key: setter_expected, }, - self.user1_access_token, + access_token=self.user1_access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) delay_id = channel.json_body.get("delay_id") @@ -719,7 +967,8 @@ class DelayedEventsTestCase(HomeserverTestCase): events = self._get_delayed_events() self.assertEqual(1, len(events), events) - self.reactor.advance(1) + # Advance time enough so the delayed event is sent + self.reactor.advance(Duration(seconds=1).as_secs()) content = self.helper.get_state( self.room_id, _EVENT_TYPE, @@ -728,8 +977,8 @@ class DelayedEventsTestCase(HomeserverTestCase): ) self.assertEqual(setter_expected, content.get(setter_key), content) - self._find_sent_delayed_event(self.user1_access_token, delay_id, True) - self._find_sent_delayed_event(self.user2_access_token, delay_id, False) + self._check_for_delayed_event_in_sync(self.user1_access_token, delay_id, True) + self._check_for_delayed_event_in_sync(self.user2_access_token, delay_id, False) def test_delayed_state_is_cancelled_by_new_state_from_other_user( self, @@ -737,13 +986,15 @@ class DelayedEventsTestCase(HomeserverTestCase): state_key = "to_be_cancelled_by_other_user" setter_key = "setter" - channel = self.make_request( - "PUT", - _get_path_for_delayed_state(self.room_id, _EVENT_TYPE, state_key, 900), - { + channel = self._make_delayed_event_request( + room_id=self.room_id, + delay=Duration(milliseconds=900), + event_type=_EVENT_TYPE, + state_key=state_key, + content={ setter_key: "on_timeout", }, - self.user1_access_token, + access_token=self.user1_access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) delay_id = channel.json_body.get("delay_id") @@ -763,7 +1014,8 @@ class DelayedEventsTestCase(HomeserverTestCase): ) self.assertListEqual([], self._get_delayed_events()) - self.reactor.advance(1) + # Advance time enough so the delayed event is sent + self.reactor.advance(Duration(seconds=1).as_secs()) content = self.helper.get_state( self.room_id, _EVENT_TYPE, @@ -772,8 +1024,8 @@ class DelayedEventsTestCase(HomeserverTestCase): ) self.assertEqual(setter_expected, content.get(setter_key), content) - self._find_sent_delayed_event(self.user1_access_token, delay_id, False) - self._find_sent_delayed_event(self.user2_access_token, delay_id, False) + self._check_for_delayed_event_in_sync(self.user1_access_token, delay_id, False) + self._check_for_delayed_event_in_sync(self.user2_access_token, delay_id, False) def test_new_state_processing_skipped_without_new_room_events(self) -> None: """Ensure that we only process state deltas (and hit the database) when @@ -781,11 +1033,13 @@ class DelayedEventsTestCase(HomeserverTestCase): notification.""" state_key = "to_be_cancelled_by_other_user" - channel = self.make_request( - "PUT", - _get_path_for_delayed_state(self.room_id, _EVENT_TYPE, state_key, 900), - {}, - self.user1_access_token, + channel = self._make_delayed_event_request( + room_id=self.room_id, + delay=Duration(milliseconds=900), + event_type=_EVENT_TYPE, + state_key=state_key, + content={}, + access_token=self.user1_access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) self.assertEqual(1, len(self._get_delayed_events())) @@ -819,30 +1073,27 @@ class DelayedEventsTestCase(HomeserverTestCase): handler = self.hs.get_delayed_events_handler() # user1 schedules a delayed message event and a delayed state event. - state_key = "to_cancel_on_deactivation" user1_delay_ids = [] - # TODO: Schedule these through the dedicated delayed events endpoint - # instead of the delay query parameter on `/send` and `/state`, once - # https://github.com/element-hq/synapse/pull/19354 has landed. - for path in ( - _get_path_for_delayed_send(self.room_id, _EVENT_TYPE, 900), - _get_path_for_delayed_state(self.room_id, _EVENT_TYPE, state_key, 900), - ): - channel = self.make_request( - "POST" if "/send/" in path else "PUT", - path, - {"key": "value"}, - self.user1_access_token, + state_key = "to_cancel_on_deactivation" + for state_event in True, False: + channel = self._make_delayed_event_request( + room_id=self.room_id, + delay=Duration(milliseconds=900), + event_type=_EVENT_TYPE, + state_key=state_key if state_event else None, + content={"key": "value"}, + access_token=self.user1_access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) user1_delay_ids.append(channel.json_body["delay_id"]) # user2 schedules a delayed event to be sent later than user1's. - channel = self.make_request( - "POST", - _get_path_for_delayed_send(self.room_id, _EVENT_TYPE, 2000), - {"key": "value"}, - self.user2_access_token, + channel = self._make_delayed_event_request( + room_id=self.room_id, + delay=Duration(milliseconds=2000), + event_type=_EVENT_TYPE, + content={"key": "value"}, + access_token=self.user2_access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) user2_delay_id = channel.json_body["delay_id"] @@ -894,9 +1145,12 @@ class DelayedEventsTestCase(HomeserverTestCase): ) # Nothing gets sent when user1's delayed events would have timed out. - self.reactor.advance(1) + # Advance time enough so the delayed event is sent + self.reactor.advance(Duration(seconds=1).as_secs()) for delay_id in user1_delay_ids: - self._find_sent_delayed_event(self.user2_access_token, delay_id, False) + self._check_for_delayed_event_in_sync( + self.user2_access_token, delay_id, False + ) self.helper.get_state( self.room_id, _EVENT_TYPE, @@ -906,22 +1160,30 @@ class DelayedEventsTestCase(HomeserverTestCase): ) # user2's delayed event is unaffected, and still gets sent. - self._find_sent_delayed_event(self.user2_access_token, user2_delay_id, False) - self.reactor.advance(1) + self._check_for_delayed_event_in_sync( + self.user2_access_token, user2_delay_id, False + ) + # Advance time enough so the delayed event is sent + self.reactor.advance(Duration(seconds=1).as_secs()) for delay_id in user1_delay_ids: - self._find_sent_delayed_event(self.user2_access_token, delay_id, False) - self._find_sent_delayed_event(self.user2_access_token, user2_delay_id, True) + self._check_for_delayed_event_in_sync( + self.user2_access_token, delay_id, False + ) + self._check_for_delayed_event_in_sync( + self.user2_access_token, user2_delay_id, True + ) def test_deactivation_cancels_the_timer_when_no_delayed_events_remain( self, ) -> None: handler = self.hs.get_delayed_events_handler() - channel = self.make_request( - "POST", - _get_path_for_delayed_send(self.room_id, _EVENT_TYPE, 900), - {}, - self.user1_access_token, + channel = self._make_delayed_event_request( + room_id=self.room_id, + delay=Duration(milliseconds=900), + event_type=_EVENT_TYPE, + content={}, + access_token=self.user1_access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) self.assertIsNotNone(handler._next_delayed_event_call) @@ -935,81 +1197,139 @@ class DelayedEventsTestCase(HomeserverTestCase): ) self.assertIsNone(handler._next_delayed_event_call) - def _get_delayed_events(self) -> list[JsonDict]: - channel = self.make_request( - "GET", - PATH_PREFIX, - access_token=self.user1_access_token, + +class DelayedStickyEventsTestCase(DelayedEventsTestCaseBase): + """Tests scheduling delayed sticky events (MSC4354).""" + + if not USE_POSTGRES_FOR_TESTS and sqlite3.sqlite_version_info < (3, 40, 0): + # We need the JSON functionality in SQLite + # (the sticky events store refuses to start without it, so this has to be + # a class-level skip: the homeserver is built before the test body runs) + skip = f"SQLite version is too old to support sticky events: {sqlite3.sqlite_version_info} (See https://github.com/element-hq/synapse/issues/19428)" + + def default_config(self) -> JsonDict: + config = super().default_config() + config["experimental_features"] = {"msc4354_enabled": True} + return config + + @parameterized.expand( + ( + (False, True), + (True, False), + (True, True), ) - self.assertEqual(HTTPStatus.OK, channel.code, channel.result) - - key = "delayed_events" - self.assertIn(key, channel.json_body) - - events = channel.json_body[key] - self.assertIsInstance(events, list) - - return events - - def _get_delayed_event_content(self, event: JsonDict) -> JsonDict: - key = "content" - self.assertIn(key, event) - - content = event[key] - self.assertIsInstance(content, dict) - - return content - - def _update_delayed_event( - self, - delay_id: str, - action: str, - action_in_path: bool, - access_token: str | None = None, - ) -> FakeChannel: - path = f"{PATH_PREFIX}/{delay_id}" - body = {} - if action_in_path: - path += f"/{action}" - else: - body["action"] = action - return self.make_request("POST", path, body, access_token) - - def _find_sent_delayed_event( - self, access_token: str, delay_id: str, should_find: bool + ) + def test_delayed_sticky_event_is_sent_with_sticky_duration( + self, param_in_body: bool, param_in_query: bool ) -> None: - """Call /sync and look for a synced event with a specified delay_id. - At most one event will ever have a matching delay_id. + """Test that the sticky duration given when scheduling a delayed event + is applied to the event once it is sent.""" + path = self.helper.build_delayed_event_request_path( + room_id=self.room_id, + event_type=_EVENT_TYPE, + ) + body = self.helper.build_delayed_event_request_body( + delay=Duration(milliseconds=900), + content={"body": "sticky"}, + ) + sticky_duration = Duration(minutes=1) + if param_in_body: + body[StickyEvent.REQUEST_PARAM_NAME] = sticky_duration.as_millis() + if param_in_query: + path += f"?{StickyEvent.REQUEST_PARAM_NAME}={sticky_duration.as_millis()}" + channel = self.make_request("PUT", path, body, self.user1_access_token) + self.assertEqual(HTTPStatus.OK, channel.code, channel.result) + delay_id = channel.json_body.get("delay_id") + assert delay_id is not None - Args: - access_token: The access token of the user to call /sync for. - delay_id: The delay_id to search for in synced events. - should_find: Whether /sync should include an event with a matching delay_id. - """ - channel = self.make_request("GET", "/sync", access_token=access_token) - self.assertEqual(HTTPStatus.OK, channel.code) + # Advance time enough so the delayed event is sent + self.reactor.advance(Duration(seconds=1).as_secs()) + self.assertListEqual([], self._get_delayed_events()) - rooms = channel.json_body["rooms"] - events = [] - for membership in "join", "leave": - if membership in rooms: - events += rooms[membership][self.room_id]["timeline"]["events"] + event = self._check_for_delayed_event_in_sync( + self.user1_access_token, delay_id, True + ) + self.assertGreater( + event["unsigned"].get(EventUnsignedContentFields.STICKY_TTL, 0), + 0, + event, + ) - found = False - for event in events: - if event["unsigned"].get("org.matrix.msc4140.delay_id") == delay_id: - if not should_find: - self.fail( - "Found event with matching delay_id, but expected to not find one" - ) - if found: - self.fail("Found multiple events with matching delay_id") - found = True - if should_find and not found: - self.fail("Did not find event with matching delay_id") + @parameterized.expand((False, True)) + def test_delayed_sticky_state_event(self, param_in_body: bool) -> None: + """Test that delayed event requests forbid scheduling a sticky state event.""" + path = self.helper.build_delayed_event_request_path( + room_id=self.room_id, + event_type="m.room.topic", + ) + body = self.helper.build_delayed_event_request_body( + delay=Duration(milliseconds=900), + content={"topic": "This is a topic"}, + state_key="", + ) + sticky_duration = Duration(minutes=1) + if param_in_body: + body[StickyEvent.REQUEST_PARAM_NAME] = sticky_duration.as_millis() + else: + path += f"?{StickyEvent.REQUEST_PARAM_NAME}={sticky_duration.as_millis()}" + channel = self.make_request("PUT", path, body, self.user1_access_token) + self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.result) + self.assertEqual(Codes.INVALID_PARAM, channel.json_body["errcode"]) + + @parameterized.expand( + ( + (sticky_duration, param_in_body) + for sticky_duration in ( + Duration(), + Duration(seconds=-2000), + Duration(days=1), + ) + for param_in_body in (False, True) + ) + ) + def test_delayed_sticky_event_with_extreme_duration( + self, sticky_duration: Duration, param_in_body: bool + ) -> None: + """Test that delayed event requests permit a zero, negative, or very long sticky duration. + (The actual sticky duration is clamped at event send time.)""" + path = self.helper.build_delayed_event_request_path( + room_id=self.room_id, + event_type=_EVENT_TYPE, + ) + body = self.helper.build_delayed_event_request_body( + delay=Duration(milliseconds=900), + content={"body": "sticky"}, + ) + if param_in_body: + body[StickyEvent.REQUEST_PARAM_NAME] = sticky_duration.as_millis() + else: + path += f"?{StickyEvent.REQUEST_PARAM_NAME}={sticky_duration.as_millis()}" + channel = self.make_request("PUT", path, body, self.user1_access_token) + self.assertEqual(HTTPStatus.OK, channel.code, channel.result) + delay_id = channel.json_body.get("delay_id") + assert delay_id is not None + + def test_delayed_sticky_event_with_conflicting_sticky_durations(self) -> None: + """Test that delayed event requests reject having a different sticky duration + in the query parameter & request body.""" + path = self.helper.build_delayed_event_request_path( + room_id=self.room_id, + event_type=_EVENT_TYPE, + ) + body = self.helper.build_delayed_event_request_body( + delay=Duration(milliseconds=900), + content={"body": "sticky"}, + ) + path += f"?{StickyEvent.REQUEST_PARAM_NAME}=10000" + body[StickyEvent.REQUEST_PARAM_NAME] = 20000 + channel = self.make_request("PUT", path, body, self.user1_access_token) + self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.result) + self.assertEqual(Codes.INVALID_PARAM, channel.json_body["errcode"]) -class DelayedEventsWorkerTestCase(BaseMultiWorkerStreamTestCase): +class DelayedEventsWorkerTestCase( + BaseMultiWorkerStreamTestCase, DelayedEventsHelperMixin +): """Tests delayed events when some of the work happens off the main process.""" servlets = [ @@ -1025,20 +1345,49 @@ class DelayedEventsWorkerTestCase(BaseMultiWorkerStreamTestCase): config["max_event_delay_duration"] = "24h" return config + def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: + self.user_id = self.register_user("user", "pass") + self.access_token = self.login("user", "pass") + self.room_id = self.helper.create_room_as(self.user_id, tok=self.access_token) + + def test_delayed_event_scheduled_on_worker_is_sent(self) -> None: + """Test that a delayed event scheduled on a worker is sent by the main process.""" + worker_hs = self.make_worker_hs("synapse.app.generic_worker") + worker_site = self._hs_to_site[worker_hs] + + channel = make_request( + self.reactor, + worker_site, + "PUT", + self.helper.build_delayed_event_request_path( + room_id=self.room_id, + event_type=_EVENT_TYPE, + ), + self.helper.build_delayed_event_request_body( + delay=Duration(milliseconds=900), + content={"body": "from a worker"}, + ), + self.access_token, + ) + self.assertEqual(HTTPStatus.OK, channel.code, channel.result) + delay_id = channel.json_body.get("delay_id") + assert delay_id is not None + + # Advance time enough so the delayed event is sent + self.reactor.advance(Duration(seconds=1).as_secs()) + self._check_for_delayed_event_in_sync(self.access_token, delay_id, True) + def test_delayed_events_are_cancelled_on_deactivation_from_worker(self) -> None: worker_hs = self.make_worker_hs("synapse.app.generic_worker") store = self.hs.get_datastores().main handler = self.hs.get_delayed_events_handler() - user_id = self.register_user("user", "pass") - access_token = self.login("user", "pass") - room_id = self.helper.create_room_as(user_id, tok=access_token) - - channel = self.make_request( - "POST", - _get_path_for_delayed_send(room_id, _EVENT_TYPE, 900), - {}, - access_token, + channel = self._make_delayed_event_request( + room_id=self.room_id, + delay=Duration(milliseconds=900), + event_type=_EVENT_TYPE, + content={}, + access_token=self.access_token, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) self.assertIsNotNone(handler._next_delayed_event_call) @@ -1047,9 +1396,9 @@ class DelayedEventsWorkerTestCase(BaseMultiWorkerStreamTestCase): # one to send delayed events, must get to know about it. self.get_success( worker_hs.get_deactivate_account_handler().deactivate_account( - user_id, + self.user_id, erase_data=False, - requester=create_requester(user_id), + requester=create_requester(self.user_id), ) ) @@ -1057,13 +1406,3 @@ class DelayedEventsWorkerTestCase(BaseMultiWorkerStreamTestCase): [], self.get_success(store.get_all_delayed_events_for_user("user")) ) self.assertIsNone(handler._next_delayed_event_call) - - -def _get_path_for_delayed_state( - room_id: str, event_type: str, state_key: str, delay_ms: int -) -> str: - return f"rooms/{room_id}/state/{event_type}/{state_key}?org.matrix.msc4140.delay={delay_ms}" - - -def _get_path_for_delayed_send(room_id: str, event_type: str, delay_ms: int) -> str: - return f"rooms/{room_id}/send/{event_type}?org.matrix.msc4140.delay={delay_ms}" diff --git a/tests/rest/client/test_rooms.py b/tests/rest/client/test_rooms.py index c5a01a9a7b..ccaf97bde6 100644 --- a/tests/rest/client/test_rooms.py +++ b/tests/rest/client/test_rooms.py @@ -24,6 +24,7 @@ """Tests REST events for /rooms paths.""" import json +import time from http import HTTPStatus from typing import Any, Iterable, Literal from unittest.mock import AsyncMock, Mock, call, create_autospec, patch @@ -67,6 +68,7 @@ from synapse.util.stringutils import random_string from tests import unittest from tests.http.server._base import make_request_with_cancellation_test +from tests.server import FakeChannel from tests.storage.test_stream import PaginationTestCase from tests.test_utils import FakeResponse from tests.test_utils.event_injection import ( @@ -1589,6 +1591,37 @@ class RoomAppserviceTsParamTestCase(unittest.HomeserverTestCase): res = self.get_success(self.main_store.get_event(event_id)) self.assertEqual(ts, res.origin_server_ts) + @unittest.override_config({"max_event_delay_duration": "24h"}) + def test_send_delayed_event_ts(self) -> None: + """Test scheduling a delayed event with a custom timestamp.""" + ts = 1 + + url_params = { + "user_id": self.appservice_user, + "ts": ts, + } + channel = self.make_request( + "PUT", + path=f"/_matrix/client/unstable/org.matrix.msc4140/rooms/{self.room}/delayed_event/m.room.message/1234?" + + urlparse.urlencode(url_params), + content={ + "delay_ms": 2000, + "content": {"body": "test", "msgtype": "m.text"}, + }, + access_token=self.appservice.token, + ) + self.assertEqual(channel.code, 200, channel.json_body) + + # Wait for the delayed event to be sent, then ensure it was persisted + # with the custom timestamp. + self.reactor.advance(Duration(seconds=2).as_secs()) + (event_id,) = self.get_success( + self.main_store.get_latest_event_ids_in_room(self.room) + ) + res = self.get_success(self.main_store.get_event(event_id)) + self.assertEqual("m.room.message", res.type, res) + self.assertEqual(ts, res.origin_server_ts) + def test_send_membership_event_ts(self) -> None: """Test sending a membership event with a custom timestamp.""" ts = 1 @@ -2494,6 +2527,42 @@ class RoomDelayedEventTestCase(RoomBase): """Tests delayed events.""" user_id = "@sid1:red" + invalid_delay_error_type = Codes.INVALID_PARAM + + def make_delayed_event_request( + self, + *, + room_id: str, + delay: Duration, + event_type: str, + state_key: str | None = None, + content: JsonDict, + txn_id: str | None = None, + ) -> FakeChannel: + """ + Build and send a request for scheduling a delayed event. + + Args: + room_id: The room to send the event to. + delay: How long to wait before sending the event. + event_type: The type of the event to send. + state_key: The state key of the event, or None for a message event. + content: The content of the event. + txn_id: The transaction ID for a delayed message event request. + Ignored for a delayed state event request. + Otherwise, generated if not given. + + Returns: + The `FakeChannel` object which stores the result of the request. + """ + if state_key is not None: + path = f"rooms/{room_id}/state/{event_type}/{state_key}" + else: + if txn_id is None: + txn_id = "m%s" % (str(time.time())) + path = f"rooms/{room_id}/send/{event_type}/{txn_id}" + path += f"?org.matrix.msc4140.delay={delay.as_millis()}" + return self.make_request("PUT", path, content) def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: self.room_id = self.helper.create_room_as(self.user_id) @@ -2501,13 +2570,11 @@ class RoomDelayedEventTestCase(RoomBase): @unittest.override_config({"max_event_delay_duration": "24h"}) def test_send_delayed_invalid_event(self) -> None: """Test sending a delayed event with invalid content.""" - channel = self.make_request( - "PUT", - ( - "rooms/%s/send/m.room.message/mid1?org.matrix.msc4140.delay=2000" - % self.room_id - ).encode("ascii"), - {}, + channel = self.make_delayed_event_request( + room_id=self.room_id, + delay=Duration(milliseconds=2000), + event_type="m.room.message", + content={}, ) self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.result) # Assert that the standard error response uses a valid errcode. @@ -2519,13 +2586,11 @@ class RoomDelayedEventTestCase(RoomBase): def test_delayed_event_unsupported_by_default(self) -> None: """Test that sending a delayed event is unsupported with the default config.""" - channel = self.make_request( - "PUT", - ( - "rooms/%s/send/m.room.message/mid1?org.matrix.msc4140.delay=2000" - % self.room_id - ).encode("ascii"), - {"body": "test", "msgtype": "m.text"}, + channel = self.make_delayed_event_request( + room_id=self.room_id, + delay=Duration(milliseconds=2000), + event_type="m.room.message", + content={"body": "test", "msgtype": "m.text"}, ) self.assertEqual(HTTPStatus.FORBIDDEN, channel.code, channel.result) self.assertEqual( @@ -2544,13 +2609,11 @@ class RoomDelayedEventTestCase(RoomBase): ) def test_delayed_event_disabled_by_limit(self) -> None: """Test that delayed events are disabled by configuring the per-user limit to 0.""" - channel = self.make_request( - "PUT", - ( - "rooms/%s/send/m.room.message/mid1?org.matrix.msc4140.delay=2000" - % self.room_id - ).encode("ascii"), - {"body": "test", "msgtype": "m.text"}, + channel = self.make_delayed_event_request( + room_id=self.room_id, + delay=Duration(milliseconds=2000), + event_type="m.room.message", + content={"body": "test", "msgtype": "m.text"}, ) self.assertEqual(HTTPStatus.FORBIDDEN, channel.code, channel.result) self.assertEqual( @@ -2562,17 +2625,15 @@ class RoomDelayedEventTestCase(RoomBase): @unittest.override_config({"max_event_delay_duration": "1000"}) def test_delayed_event_exceeds_max_delay(self) -> None: """Test that sending a delayed event fails if its delay is longer than allowed.""" - channel = self.make_request( - "PUT", - ( - "rooms/%s/send/m.room.message/mid1?org.matrix.msc4140.delay=2000" - % self.room_id - ).encode("ascii"), - {"body": "test", "msgtype": "m.text"}, + channel = self.make_delayed_event_request( + room_id=self.room_id, + delay=Duration(milliseconds=2000), + event_type="m.room.message", + content={"body": "test", "msgtype": "m.text"}, ) self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.result) self.assertEqual( - "ORG.MATRIX.MSC4140_DELAY_TOO_LARGE", + Codes.DELAY_TOO_LARGE, channel.json_body.get("errcode"), channel.json_body, ) @@ -2592,13 +2653,11 @@ class RoomDelayedEventTestCase(RoomBase): self.hs.get_datastores().main.set_ratelimit_for_user(self.user_id, 0, 0) ) - make_delayed_event_request = lambda: self.make_request( - "POST", - ( - "rooms/%s/send/m.room.message?org.matrix.msc4140.delay=15000" - % self.room_id - ).encode("ascii"), - {"body": "test", "msgtype": "m.text"}, + make_delayed_event_request = lambda: self.make_delayed_event_request( + room_id=self.room_id, + delay=Duration(milliseconds=15000), + event_type="m.room.message", + content={"body": "test", "msgtype": "m.text"}, ) # Send a delayed event to eat up the limit channel = make_delayed_event_request() @@ -2643,13 +2702,11 @@ class RoomDelayedEventTestCase(RoomBase): how many delayed events a user may have scheduled at once. """ send_after = Duration(seconds=1) - make_delayed_event_request = lambda: self.make_request( - "POST", - ( - f"rooms/%s/send/m.room.message?org.matrix.msc4140.delay={send_after.as_millis()}" - % self.room_id - ).encode("ascii"), - {"body": "test", "msgtype": "m.text"}, + make_delayed_event_request = lambda: self.make_delayed_event_request( + room_id=self.room_id, + delay=send_after, + event_type="m.room.message", + content={"body": "test", "msgtype": "m.text"}, ) channel = make_delayed_event_request() self.assertEqual(HTTPStatus.OK, channel.code, channel.result) @@ -2702,16 +2759,18 @@ class RoomDelayedEventTestCase(RoomBase): while a user has fewer scheduled delayed events than the old limit, but more than the new limit. """ send_after: Duration - make_delayed_event_request = lambda: self.make_request( - "POST", - ( - f"rooms/%s/send/m.room.message?org.matrix.msc4140.delay={send_after.as_millis()}" - % self.room_id - ).encode("ascii"), - {"body": f"test (send after {send_after.as_secs()}s)", "msgtype": "m.text"}, + + make_delayed_event_request = lambda: self.make_delayed_event_request( + room_id=self.room_id, + delay=send_after, + event_type="m.room.message", + content={ + "body": f"test (send after {send_after.as_secs()}s)", + "msgtype": "m.text", + }, ) - for i in range(4): + for i in range(1, 5): send_after = Duration(seconds=i) channel = make_delayed_event_request() self.assertEqual(HTTPStatus.OK, channel.code, channel.result) @@ -2742,45 +2801,100 @@ class RoomDelayedEventTestCase(RoomBase): channel = make_delayed_event_request() self.assertEqual(HTTPStatus.OK, channel.code, channel.result) + @parameterized.expand( + ( + # A negative delay would put the send time in the past. + Duration(milliseconds=-2000), + # A zero delay is no delay at all; the regular send endpoints are for that. + Duration(), + ) + ) @unittest.override_config({"max_event_delay_duration": "24h"}) - def test_delayed_event_with_negative_delay(self) -> None: - """Test that sending a delayed event fails if its delay is negative.""" - channel = self.make_request( - "PUT", - ( - "rooms/%s/send/m.room.message/mid1?org.matrix.msc4140.delay=-2000" - % self.room_id - ).encode("ascii"), - {"body": "test", "msgtype": "m.text"}, + def test_delayed_event_with_invalid_delay(self, invalid_delay: Duration) -> None: + """Test that sending a delayed event fails if its delay is not positive.""" + channel = self.make_delayed_event_request( + room_id=self.room_id, + delay=invalid_delay, + event_type="m.room.message", + content={"body": "test", "msgtype": "m.text"}, ) self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.result) self.assertEqual( - Codes.INVALID_PARAM, channel.json_body["errcode"], channel.json_body + self.invalid_delay_error_type, + channel.json_body["errcode"], + channel.json_body, + ) + + @parameterized.expand( + ( + # A negative delay would put the send time in the past. + Duration(milliseconds=-2000), + # A zero delay is no delay at all; the regular send endpoints are for that. + Duration(), + ) + ) + @unittest.override_config({"max_event_delay_duration": None}) + def test_delayed_event_unsupported_with_invalid_delay( + self, + invalid_delay: Duration, + ) -> None: + """Test that delayed events being unsupported takes precedence over an invalid delay.""" + channel = self.make_delayed_event_request( + room_id=self.room_id, + delay=invalid_delay, + event_type="m.room.message", + content={"body": "test", "msgtype": "m.text"}, + ) + self.assertEqual(HTTPStatus.FORBIDDEN, channel.code, channel.result) + self.assertEqual( + Codes.FORBIDDEN, + channel.json_body["errcode"], + channel.json_body, ) @unittest.override_config({"max_event_delay_duration": "24h"}) def test_send_delayed_message_event(self) -> None: """Test sending a valid delayed message event.""" - channel = self.make_request( - "PUT", - ( - "rooms/%s/send/m.room.message/mid1?org.matrix.msc4140.delay=2000" - % self.room_id - ).encode("ascii"), - {"body": "test", "msgtype": "m.text"}, + channel = self.make_delayed_event_request( + room_id=self.room_id, + delay=Duration(milliseconds=2000), + event_type="m.room.message", + content={"body": "test", "msgtype": "m.text"}, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) + @unittest.override_config({"max_event_delay_duration": "24h"}) + def test_send_delayed_message_event_idempotent(self) -> None: + """ + Test that repeated requests to schedule a delayed message event + with the same transaction ID are idempotent. + """ + make_delayed_event_request = lambda: self.make_delayed_event_request( + room_id=self.room_id, + delay=Duration(milliseconds=2000), + event_type="m.room.message", + content={"body": "test", "msgtype": "m.text"}, + txn_id="txn_id_0", + ) + channel = make_delayed_event_request() + self.assertEqual(HTTPStatus.OK, channel.code, channel.result) + + delay_id = channel.json_body["delay_id"] + + # Repeat the request, which reuses its txn_id + channel = make_delayed_event_request() + self.assertEqual(HTTPStatus.OK, channel.code, channel.result) + self.assertEqual(delay_id, channel.json_body.get("delay_id")) + @unittest.override_config({"max_event_delay_duration": "24h"}) def test_send_delayed_state_event(self) -> None: """Test sending a valid delayed state event.""" - channel = self.make_request( - "PUT", - ( - "rooms/%s/state/m.room.topic/?org.matrix.msc4140.delay=2000" - % self.room_id - ).encode("ascii"), - {"topic": "This is a topic"}, + channel = self.make_delayed_event_request( + room_id=self.room_id, + delay=Duration(milliseconds=2000), + event_type="m.room.topic", + state_key="", + content={"topic": "This is a topic"}, ) self.assertEqual(HTTPStatus.OK, channel.code, channel.result) @@ -2796,15 +2910,13 @@ class RoomDelayedEventTestCase(RoomBase): exempt from ratelimiting. """ - # Test that new delayed events are correctly ratelimited. - make_delayed_event_request = lambda: self.make_request( - "POST", - ( - "rooms/%s/send/m.room.message?org.matrix.msc4140.delay=2000" - % self.room_id - ).encode("ascii"), - {"body": "test", "msgtype": "m.text"}, + make_delayed_event_request = lambda: self.make_delayed_event_request( + room_id=self.room_id, + delay=Duration(milliseconds=2000), + event_type="m.room.message", + content={"body": "test", "msgtype": "m.text"}, ) + # Test that new delayed events are correctly ratelimited. channel = make_delayed_event_request() self.assertEqual(HTTPStatus.OK, channel.code, channel.result) channel = make_delayed_event_request() @@ -2820,6 +2932,152 @@ class RoomDelayedEventTestCase(RoomBase): self.assertEqual(HTTPStatus.OK, channel.code, channel.result) +class RoomDelayedEventDedicatedEndpointTestCase(RoomDelayedEventTestCase): + """Tests the dedicated request endpoint for scheduling delayed events.""" + + invalid_delay_error_type = Codes.BAD_JSON + + def make_delayed_event_request( + self, + *, + room_id: str, + delay: Duration, + event_type: str, + state_key: str | None = None, + content: JsonDict, + txn_id: str | None = None, + ) -> FakeChannel: + return self.make_request( + "PUT", + self.helper.build_delayed_event_request_path( + room_id=room_id, + event_type=event_type, + txn_id=txn_id, + ), + self.helper.build_delayed_event_request_body( + delay=delay, + state_key=state_key, + content=content, + ), + ) + + @unittest.override_config({"max_event_delay_duration": "24h"}) + def test_delayed_event_with_content_at_top(self) -> None: + """Test that the dedicated endpoint fails with event content keys at the top level of the body.""" + channel = self.make_request( + "PUT", + self.helper.build_delayed_event_request_path( + room_id=self.room_id, + event_type="m.room.message", + ), + content={ + "delay_ms": 2000, + "body": "test", + "msgtype": "m.text", + }, + ) + self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.result) + self.assertEqual( + Codes.MISSING_PARAM, + channel.json_body["errcode"], + channel.json_body, + ) + + @parameterized.expand(("delay_ms", "content")) + @unittest.override_config({"max_event_delay_duration": "24h"}) + def test_delayed_event_with_missing_key(self, missing_key: str) -> None: + """Test that the dedicated endpoint fails with a body missing a required key.""" + body = self.helper.build_delayed_event_request_body( + delay=Duration(milliseconds=2000), + content={"body": "test", "msgtype": "m.text"}, + ) + del body[missing_key] + channel = self.make_request( + "PUT", + self.helper.build_delayed_event_request_path( + room_id=self.room_id, + event_type="m.room.message", + ), + content=body, + ) + self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.result) + self.assertEqual( + Codes.MISSING_PARAM, + channel.json_body["errcode"], + channel.json_body, + ) + + @unittest.override_config({"max_event_delay_duration": "24h"}) + def test_delayed_event_with_no_keys(self) -> None: + """Test that the dedicated endpoint fails with a body missing all required keys.""" + channel = self.make_request( + "PUT", + self.helper.build_delayed_event_request_path( + room_id=self.room_id, + event_type="m.room.topic", + ), + {}, + ) + self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.result) + self.assertEqual( + Codes.BAD_JSON, + channel.json_body["errcode"], + channel.json_body, + ) + + @unittest.override_config({"max_event_delay_duration": "24h"}) + def test_delayed_event_with_null_state_key(self) -> None: + """ + Test that the dedicated endpoint fails to schedule a message event + with a "state_key" of `None` in the request body. + """ + body = self.helper.build_delayed_event_request_body( + delay=Duration(milliseconds=2000), + content={"body": "test", "msgtype": "m.text"}, + ) + body["state_key"] = None + channel = self.make_request( + "PUT", + self.helper.build_delayed_event_request_path( + room_id=self.room_id, + event_type="m.room.message", + ), + body, + ) + self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.result) + self.assertEqual( + Codes.BAD_JSON, + channel.json_body["errcode"], + channel.json_body, + ) + + @unittest.override_config({"max_event_delay_duration": "24h"}) + def test_send_delayed_state_event_idempotent(self) -> None: + """ + Test that repeated requests to schedule a delayed state event + with the same transaction ID are idempotent. + Note that only the dedicated endpoint can test this, as the one based on /state + does not support idempotent requests by using a transaction ID. + """ + make_delayed_event_request = lambda: self.make_delayed_event_request( + room_id=self.room_id, + delay=Duration(milliseconds=2000), + event_type="m.room.topic", + state_key="", + content={"topic": "This is a topic"}, + txn_id="txn_id_0", + ) + channel = make_delayed_event_request() + self.assertEqual(HTTPStatus.OK, channel.code, channel.result) + + delay_id = channel.json_body["delay_id"] + + # Repeat the request, which reuses its txn_id + channel = make_delayed_event_request() + self.assertEqual(HTTPStatus.OK, channel.code, channel.result) + self.assertEqual(delay_id, channel.json_body.get("delay_id")) + + class RoomSearchTestCase(unittest.HomeserverTestCase): servlets = [ synapse.rest.admin.register_servlets_for_client_rest_resource, diff --git a/tests/rest/client/utils.py b/tests/rest/client/utils.py index bfa8e6f3d8..fccf8df0a0 100644 --- a/tests/rest/client/utils.py +++ b/tests/rest/client/utils.py @@ -636,6 +636,34 @@ class RestHelper: room_id, event_type, body, tok, expect_code, state_key, method="PUT" ) + @staticmethod + def build_delayed_event_request_path( + *, + room_id: str, + event_type: str, + txn_id: str | None = None, + ) -> str: + """Build the request path for scheduling a delayed event via the dedicated endpoint.""" + if txn_id is None: + txn_id = "m%s" % (str(time.time())) + return f"/_matrix/client/unstable/org.matrix.msc4140/rooms/{room_id}/delayed_event/{event_type}/{txn_id}" + + @staticmethod + def build_delayed_event_request_body( + *, + delay: Duration, + content: JsonDict, + state_key: str | None = None, + ) -> JsonDict: + """Build the request body for scheduling a delayed event via the dedicated endpoint.""" + body = { + "delay_ms": delay.as_millis(), + "content": content, + } + if state_key is not None: + body["state_key"] = state_key + return body + def upload_media( self, image_data: bytes,