diff --git a/changelog.d/19539.bugfix b/changelog.d/19539.bugfix new file mode 100644 index 00000000000..41beb3a179b --- /dev/null +++ b/changelog.d/19539.bugfix @@ -0,0 +1 @@ +[MSC4140: Cancellable delayed events](https://github.com/matrix-org/matrix-spec-proposals/pull/4140): Update error responses to match their format in the current draft of the MSC. diff --git a/changelog.d/19539.feature b/changelog.d/19539.feature new file mode 100644 index 00000000000..93eb4cc1b97 --- /dev/null +++ b/changelog.d/19539.feature @@ -0,0 +1 @@ +[MSC4140: Cancellable delayed events](https://github.com/matrix-org/matrix-spec-proposals/pull/4140): Limit how many delayed events a user may have scheduled at once. diff --git a/synapse/config/server.py b/synapse/config/server.py index ca94c224ea5..70248e66a50 100644 --- a/synapse/config/server.py +++ b/synapse/config/server.py @@ -914,10 +914,27 @@ def read_config(self, config: JsonDict, **kwargs: Any) -> None: max_event_delay_duration ) if self.max_event_delay_ms <= 0: - raise ConfigError("max_event_delay_duration must be a positive value") + raise ConfigError( + "'max_event_delay_duration' must be a positive value if set", + ("max_event_delay_duration",), + ) else: self.max_event_delay_ms = None + # The maximum number of delayed events a user may have scheduled at a time. + # (Defined here despite being experimental to be near the other MSC4140 config) + self.max_delayed_events_per_user: int = config.get( + "experimental_features", {} + ).get("msc4140_max_delayed_events_per_user", 100) + if ( + not isinstance(self.max_delayed_events_per_user, int) + or self.max_delayed_events_per_user < 0 + ): + raise ConfigError( + "'msc4140_max_delayed_events_per_user' must be a non-negative integer", + ("experimental", "msc4140_max_delayed_events_per_user"), + ) + def has_tls_listener(self) -> bool: return any(listener.is_tls() for listener in self.listeners) diff --git a/synapse/handlers/delayed_events.py b/synapse/handlers/delayed_events.py index 4a9f646d4db..b119ca62519 100644 --- a/synapse/handlers/delayed_events.py +++ b/synapse/handlers/delayed_events.py @@ -13,12 +13,13 @@ # import logging +from http import HTTPStatus from typing import TYPE_CHECKING, Optional from twisted.internet.interfaces import IDelayedCall from synapse.api.constants import EventTypes, StickyEvent, StickyEventField -from synapse.api.errors import ShadowBanError, SynapseError +from synapse.api.errors import Codes, ShadowBanError, SynapseError from synapse.api.ratelimiting import Ratelimiter from synapse.config.workers import MAIN_PROCESS_INSTANCE_NAME from synapse.http.site import SynapseRequest @@ -353,13 +354,33 @@ async def add( Returns: The ID of the added delayed event. Raises: - SynapseError: if the delayed event fails validation checks. + 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. """ # 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) + max_delay = self._config.server.max_event_delay_ms + if ( + max_delay is None + or max_delay <= 0 + or (limit := self._config.server.max_delayed_events_per_user) <= 0 + ): + raise SynapseError( + HTTPStatus.FORBIDDEN, + "Sending delayed events has been disallowed", + Codes.FORBIDDEN, + ) + if delay > max_delay: + raise SynapseError( + HTTPStatus.FORBIDDEN, + "The requested delay exceeds the allowed maximum", + Codes.FORBIDDEN, + ) + self._event_creation_handler.validator.validate_builder( self._event_creation_handler.event_builder_factory.for_room_version( await self._store.get_room_version(room_id), @@ -386,6 +407,7 @@ async def add( content=content, delay=delay, sticky_duration_ms=sticky_duration_ms, + limit=limit, ) if self._repl_client is not None: diff --git a/synapse/rest/client/capabilities.py b/synapse/rest/client/capabilities.py index 2be5f5849d7..450a6e35814 100644 --- a/synapse/rest/client/capabilities.py +++ b/synapse/rest/client/capabilities.py @@ -109,6 +109,11 @@ async def on_GET(self, request: SynapseRequest) -> tuple[int, JsonDict]: "capabilities" ]["m.profile_fields"] + response["capabilities"]["org.matrix.msc4140.delayed_events"] = { + "max_delay_ms": self.config.server.max_event_delay_ms or 0, + "max_scheduled": self.config.server.max_delayed_events_per_user, + } + if self.config.experimental.msc4267_enabled: response["capabilities"]["org.matrix.msc4267.forget_forced_upon_leave"] = { "enabled": self.config.room.forget_on_leave, diff --git a/synapse/rest/client/room.py b/synapse/rest/client/room.py index 48ef42c7d6f..51793cd338f 100644 --- a/synapse/rest/client/room.py +++ b/synapse/rest/client/room.py @@ -215,7 +215,6 @@ def __init__(self, hs: "HomeServer"): self.auth = hs.get_auth() self.clock = hs.get_clock() self._event_serializer = hs.get_event_client_serializer() - self._max_event_delay_ms = hs.config.server.max_event_delay_ms self._spam_checker_module_callbacks = hs.get_module_api_callbacks().spam_checker self._msc4354_enabled = hs.config.experimental.msc4354_enabled @@ -343,7 +342,7 @@ async def on_PUT( if self._msc4354_enabled: sticky_duration_ms = parse_integer(request, StickyEvent.QUERY_PARAM_NAME) - delay = _parse_request_delay(request, self._max_event_delay_ms) + delay = _parse_request_delay(request) if delay is not None: delay_id = await self.delayed_events_handler.add( requester, @@ -416,7 +415,6 @@ def __init__(self, hs: "HomeServer"): self.event_creation_handler = hs.get_event_creation_handler() self.delayed_events_handler = hs.get_delayed_events_handler() self.auth = hs.get_auth() - self._max_event_delay_ms = hs.config.server.max_event_delay_ms self._msc4354_enabled = hs.config.experimental.msc4354_enabled def register(self, http_server: HttpServer) -> None: @@ -442,7 +440,7 @@ async def _do( if self._msc4354_enabled: sticky_duration_ms = parse_integer(request, StickyEvent.QUERY_PARAM_NAME) - delay = _parse_request_delay(request, self._max_event_delay_ms) + delay = _parse_request_delay(request) if delay is not None: delay_id = await self.delayed_events_handler.add( requester, @@ -515,47 +513,19 @@ async def on_PUT( ) -def _parse_request_delay( - request: SynapseRequest, - max_delay: int | None, -) -> int | None: +def _parse_request_delay(request: SynapseRequest) -> int | None: """Parses from the request string the delay parameter for delayed event requests, and checks it for correctness. Args: request: the twisted HTTP request. - max_delay: the maximum allowed value of the delay parameter, - or None if no delay parameter is allowed. Returns: The value of the requested delay, or None if it was absent. Raises: - SynapseError: if the delay parameter is present and forbidden, - or if it exceeds the maximum allowed value. + SynapseError: if the delay parameter is present and invalid. """ - delay = parse_integer(request, "org.matrix.msc4140.delay") - if delay is None: - return None - if max_delay is None: - raise SynapseError( - HTTPStatus.BAD_REQUEST, - "Delayed events are not supported on this server", - Codes.UNKNOWN, - { - "org.matrix.msc4140.errcode": "M_MAX_DELAY_UNSUPPORTED", - }, - ) - if delay > max_delay: - raise SynapseError( - HTTPStatus.BAD_REQUEST, - "The requested delay exceeds the allowed maximum.", - Codes.UNKNOWN, - { - "org.matrix.msc4140.errcode": "M_MAX_DELAY_EXCEEDED", - "org.matrix.msc4140.max_delay": max_delay, - }, - ) - return delay + return parse_integer(request, "org.matrix.msc4140.delay") # TODO: Needs unit testing for room ID + alias joins diff --git a/synapse/storage/databases/main/delayed_events.py b/synapse/storage/databases/main/delayed_events.py index 1727f589e2a..47987eec01c 100644 --- a/synapse/storage/databases/main/delayed_events.py +++ b/synapse/storage/databases/main/delayed_events.py @@ -17,7 +17,7 @@ import attr -from synapse.api.errors import NotFoundError +from synapse.api.errors import LimitExceededError, NotFoundError from synapse.storage._base import SQLBaseStore, db_to_json from synapse.storage.database import ( DatabasePool, @@ -124,18 +124,79 @@ async def add_delayed_event( content: JsonDict, delay: int, sticky_duration_ms: int | None, + limit: int, ) -> tuple[DelayID, Timestamp]: """ Inserts a new delayed event in the DB. + Args: + user_localpart: The localpart of the requester of the delayed event, who will be its owner. + device_id: The device ID of the requester. + creation_ts: The timestamp of when the request to add the delayed event was made. + room_id: The ID of the room where the event should be sent to. + event_type: The type of event to be sent. + state_key: The state key of the event to be sent, or None if it is not a state event. + origin_server_ts: The custom timestamp to send the event with. + If None, the timestamp will be the actual time when the event is sent. + content: The content of the event to be sent. + delay: How long (in milliseconds) to wait before automatically sending the event. + sticky_duration_ms: If an MSC4354 sticky event: the sticky duration (in milliseconds). + The event will be attempted to be reliably delivered to clients and remote servers + during its sticky period. + limit: The maximum number of delayed events the DB may store for the given requester. + Must be greater than 0. Returns: The generated ID assigned to the added delayed event, and the send time of the next delayed event to be sent, which is either the event just added or one added earlier. + + Raises: + LimitExceededError: if the DB has reached the limit of + how many delayed events it may store for the given requester. + ValueError: if the limit is not greater than 0. """ + if limit <= 0: + raise ValueError("limit must be greater than 0") + delay_id = _generate_delay_id() send_ts = Timestamp(creation_ts + delay) def add_delayed_event_txn(txn: LoggingTransaction) -> Timestamp: + num_existing: int = self.db_pool.simple_select_one_onecol_txn( + txn, + table="delayed_events", + keyvalues={"user_localpart": user_localpart}, + retcol="COUNT(*)", + ) + if num_existing >= limit: + # Cover case of num_existing > limit, which can happen by reducing + # the configured limit after delayed events have already been added + # (or by calling this method with a limit lower than the configured one) + # + # FIXME: Remove "AS subquery" after dropping support for PostgreSQL <16 + txn.execute( + """ + SELECT MAX(send_ts) FROM ( + SELECT * FROM delayed_events + WHERE user_localpart = ? + ORDER BY send_ts ASC + LIMIT ? + ) AS subquery + """, + ( + user_localpart, + num_existing - limit + 1, + ), + ) + row = txn.fetchone() + assert row + retry_after_ms = int(row[0]) - creation_ts + err = LimitExceededError( + limiter_name="add_delayed_event", + retry_after_ms=retry_after_ms if retry_after_ms > 0 else None, + ) + err.msg = "The maximum number of delayed events has been reached." + raise err + self.db_pool.simple_insert_txn( txn, table="delayed_events", diff --git a/tests/config/test_server.py b/tests/config/test_server.py index d3c59ae14ca..41ea8fb5b1b 100644 --- a/tests/config/test_server.py +++ b/tests/config/test_server.py @@ -18,11 +18,15 @@ # # + +from typing import Any + import yaml from synapse.config._base import ConfigError, RootConfig from synapse.config.homeserver import HomeServerConfig from synapse.config.server import ServerConfig, generate_ip_set, is_threepid_reserved +from synapse.types import JsonDict from tests import unittest @@ -189,6 +193,54 @@ def test_listeners_set_correctly_open_private_ports_true(self) -> None: self.assertEqual(conf["listeners"], expected_listeners) + def test_max_delayed_events_enforces_positive(self) -> None: + """ + Test that the configured maximum allowed delay must be a positive value if set, + as per documentation + """ + + def generate_config(value: int) -> JsonDict: + return {"max_event_delay_duration": value} + + _read_config(generate_config(1)) + + with self.assertRaises(ConfigError): + _read_config(generate_config(0)) + + with self.assertRaises(ConfigError): + _read_config(generate_config(-1)) + + def test_max_delayed_events_per_user_enforces_non_negative_int(self) -> None: + """ + Test that the configured maximum number of delayed events must be a non-negative value if set, + as a negative limit can never be satisfied + """ + + def generate_config(value: Any) -> JsonDict: + return { + "experimental_features": {"msc4140_max_delayed_events_per_user": value} + } + + for allowed_value in (0, 1): + _read_config(generate_config(allowed_value)) + + for disallowed_value in (-1, 0.5): + with self.assertRaises(ConfigError): + _read_config(generate_config(disallowed_value)) + + +def _read_config(config_values: JsonDict) -> None: + ServerConfig(RootConfig()).read_config( + yaml.safe_load( + HomeServerConfig().generate_config( + config_dir_path="CONFDIR", + data_dir_path="/data_dir_path", + server_name="che.org", + ) + ) + | config_values + ) + class GenerateIpSetTestCase(unittest.TestCase): def test_empty(self) -> None: diff --git a/tests/rest/client/test_capabilities.py b/tests/rest/client/test_capabilities.py index c28e0605b54..e8099fa4a2b 100644 --- a/tests/rest/client/test_capabilities.py +++ b/tests/rest/client/test_capabilities.py @@ -203,6 +203,42 @@ def test_get_set_avatar_url_capabilities_avatar_url_disabled_msc4133(self) -> No ["avatar_url"], ) + def test_get_delayed_events_capabilities_default_config_msc4140(self) -> None: + access_token = self.login(self.localpart, self.password) + + channel = self.make_request("GET", self.url, access_token=access_token) + capabilities = channel.json_body["capabilities"] + + self.assertEqual(channel.code, HTTPStatus.OK) + self.assertEqual( + capabilities["org.matrix.msc4140.delayed_events"]["max_delay_ms"], 0 + ) + self.assertEqual( + capabilities["org.matrix.msc4140.delayed_events"]["max_scheduled"], 100 + ) + + @override_config( + { + "max_event_delay_duration": "24h", + "experimental_features": { + "msc4140_max_delayed_events_per_user": 50, + }, + } + ) + def test_get_delayed_events_capabilities_custom_config_msc4140(self) -> None: + access_token = self.login(self.localpart, self.password) + + channel = self.make_request("GET", self.url, access_token=access_token) + capabilities = channel.json_body["capabilities"] + + self.assertEqual(channel.code, HTTPStatus.OK) + self.assertEqual( + capabilities["org.matrix.msc4140.delayed_events"]["max_delay_ms"], 86400000 + ) + self.assertEqual( + capabilities["org.matrix.msc4140.delayed_events"]["max_scheduled"], 50 + ) + @override_config({"enable_3pid_changes": False}) def test_get_change_3pid_capabilities_3pid_disabled(self) -> None: """Test if change 3pid is disabled that the server responds it.""" diff --git a/tests/rest/client/test_rooms.py b/tests/rest/client/test_rooms.py index 10325c536a0..07a5136cc5e 100644 --- a/tests/rest/client/test_rooms.py +++ b/tests/rest/client/test_rooms.py @@ -61,6 +61,7 @@ from synapse.server import HomeServer from synapse.types import JsonDict, JsonMapping, RoomAlias, UserID, create_requester from synapse.util.clock import Clock +from synapse.util.duration import Duration from synapse.util.stringutils import random_string from tests import unittest @@ -2503,7 +2504,10 @@ def test_send_delayed_invalid_event(self) -> None: {}, ) self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.result) - self.assertNotIn("org.matrix.msc4140.errcode", channel.json_body) + self.assertTrue( + channel.json_body.get("errcode", "").startswith("M_"), + channel.json_body, + ) def test_delayed_event_unsupported_by_default(self) -> None: """Test that sending a delayed event is unsupported with the default config.""" @@ -2515,10 +2519,35 @@ def test_delayed_event_unsupported_by_default(self) -> None: ).encode("ascii"), {"body": "test", "msgtype": "m.text"}, ) - self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.result) + self.assertEqual(HTTPStatus.FORBIDDEN, channel.code, channel.result) + self.assertEqual( + Codes.FORBIDDEN, + channel.json_body.get("errcode"), + channel.json_body, + ) + + @unittest.override_config( + { + "max_event_delay_duration": "24h", + "experimental_features": { + "msc4140_max_delayed_events_per_user": 0, + }, + } + ) + 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"}, + ) + self.assertEqual(HTTPStatus.FORBIDDEN, channel.code, channel.result) self.assertEqual( - "M_MAX_DELAY_UNSUPPORTED", - channel.json_body.get("org.matrix.msc4140.errcode"), + Codes.FORBIDDEN, + channel.json_body.get("errcode"), channel.json_body, ) @@ -2533,12 +2562,162 @@ def test_delayed_event_exceeds_max_delay(self) -> None: ).encode("ascii"), {"body": "test", "msgtype": "m.text"}, ) - self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.result) + self.assertEqual(HTTPStatus.FORBIDDEN, channel.code, channel.result) + self.assertEqual( + Codes.FORBIDDEN, + channel.json_body.get("errcode"), + channel.json_body, + ) + + @unittest.override_config( + { + "max_event_delay_duration": "24h", + "experimental_features": { + "msc4140_max_delayed_events_per_user": 1, + }, + } + ) + def test_delayed_event_user_limit_exceeded(self) -> None: + """Test that users cannot have more delayed events scheduled at once than allowed.""" + # Confirm that lifting temporal ratelimits doesn't lift this storage-based limit + self.get_success( + self.hs.get_datastores().main.set_ratelimit_for_user(self.user_id, 0, 0) + ) + + make_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"}, + ) + channel = make_request() + self.assertEqual(HTTPStatus.OK, channel.code, channel.result) + + channel = make_request() + self.assertEqual(HTTPStatus.TOO_MANY_REQUESTS, channel.code, channel.result) + self.assertEqual( + Codes.LIMIT_EXCEEDED, + channel.json_body["errcode"], + channel.json_body, + ) + # Confirm that the response includes the time remaining until the next of the user's + # delayed events to be sent, at which point another delayed event may be scheduled + # without exceeding the limit + retry_after_headers = channel.headers.getRawHeaders("Retry-After") + assert retry_after_headers + retry_after_sec = int(retry_after_headers[0]) + self.assertGreater(retry_after_sec, 0) + # Confirm that there is only a single value to the Retry-After header, as per RFC9110 + self.assertEqual(1, len(retry_after_headers)) + + self.reactor.advance(retry_after_sec) + channel = make_request() + self.assertEqual(HTTPStatus.OK, channel.code, channel.result) + + @unittest.override_config( + { + "max_event_delay_duration": "24h", + "experimental_features": { + "msc4140_max_delayed_events_per_user": 1, + }, + } + ) + def test_delayed_event_processed_user_limit_exceeded(self) -> None: + """ + Test that delayed events in the midst of being sent still count towards the limit of + how many delayed events a user may have scheduled at once. + """ + send_after_ms = 1000 + make_request = lambda: self.make_request( + "POST", + ( + f"rooms/%s/send/m.room.message?org.matrix.msc4140.delay={send_after_ms}" + % self.room_id + ).encode("ascii"), + {"body": "test", "msgtype": "m.text"}, + ) + channel = make_request() + self.assertEqual(HTTPStatus.OK, channel.code, channel.result) + + # Simulate the server taking a long time to persist delayed events + simulated_send_lag_ms = 5000 + event_creation_handler = self.hs.get_event_creation_handler() + orig_send_fn = event_creation_handler.create_and_send_nonmember_event + + async def slow_send_fn(*args: Any, **kwargs: Any) -> Any: + await self.clock.sleep(Duration(milliseconds=simulated_send_lag_ms)) + return await orig_send_fn(*args, **kwargs) + + with patch.object(event_creation_handler, orig_send_fn.__name__, slow_send_fn): + self.reactor.advance(send_after_ms / 1000) + channel = make_request() + self.assertEqual(HTTPStatus.TOO_MANY_REQUESTS, channel.code, channel.result) + self.assertEqual( + Codes.LIMIT_EXCEEDED, + channel.json_body["errcode"], + channel.json_body, + ) + # Confirm that the response lacks a Retry-After header, because the reason for this limit + # is the server taking an indeterminitely long time to process a delayed event, and the + # server doesn't know how much longer the client should wait before sending more requests + retry_after_headers = channel.headers.getRawHeaders("Retry-After") + assert not retry_after_headers + + self.reactor.advance(simulated_send_lag_ms / 1000) + channel = make_request() + self.assertEqual(HTTPStatus.OK, channel.code, channel.result) + + @unittest.override_config( + { + "max_event_delay_duration": "24h", + "experimental_features": { + "msc4140_max_delayed_events_per_user": 5, + }, + } + ) + def test_delayed_event_custom_user_limit_exceeded(self) -> None: + """ + Test that delayed event limits work properly when + the number of already scheduled events exceeds the configured limit. + + This can be invoked by the server admin lowering the configured limit & restarting the server + while a user has fewer scheduled delayed events than the old limit, but more than the new limit. + """ + send_after_ms: int + make_request = lambda: self.make_request( + "POST", + ( + f"rooms/%s/send/m.room.message?org.matrix.msc4140.delay={send_after_ms}" + % self.room_id + ).encode("ascii"), + {"body": f"test (send after {send_after_ms})", "msgtype": "m.text"}, + ) + + for i in range(3): + send_after_ms = 1000 * i + channel = make_request() + self.assertEqual(HTTPStatus.OK, channel.code, channel.result) + + # Simulate lowering the configured limit & applying it with a server restart + self.hs.config.server.max_delayed_events_per_user = 2 + + channel = make_request() + self.assertEqual(HTTPStatus.TOO_MANY_REQUESTS, channel.code, channel.result) self.assertEqual( - "M_MAX_DELAY_EXCEEDED", - channel.json_body.get("org.matrix.msc4140.errcode"), + Codes.LIMIT_EXCEEDED, + channel.json_body["errcode"], channel.json_body, ) + retry_after_header = channel.headers.getRawHeaders("Retry-After") + assert retry_after_header + retry_after_sec = int(retry_after_header[0]) + assert retry_after_sec > 0 + + self.reactor.advance(retry_after_sec) + channel = make_request() + self.assertEqual(HTTPStatus.OK, channel.code, channel.result) @unittest.override_config({"max_event_delay_duration": "24h"}) def test_delayed_event_with_negative_delay(self) -> None: @@ -2595,7 +2774,7 @@ def test_add_delayed_event_ratelimit(self) -> None: """ # Test that new delayed events are correctly ratelimited. - args = ( + make_request = lambda: self.make_request( "POST", ( "rooms/%s/send/m.room.message?org.matrix.msc4140.delay=2000" @@ -2603,9 +2782,9 @@ def test_add_delayed_event_ratelimit(self) -> None: ).encode("ascii"), {"body": "test", "msgtype": "m.text"}, ) - channel = self.make_request(*args) + channel = make_request() self.assertEqual(HTTPStatus.OK, channel.code, channel.result) - channel = self.make_request(*args) + channel = make_request() self.assertEqual(HTTPStatus.TOO_MANY_REQUESTS, channel.code, channel.result) # Add the current user to the ratelimit overrides, allowing them no ratelimiting. @@ -2614,7 +2793,7 @@ def test_add_delayed_event_ratelimit(self) -> None: ) # Test that the new delayed events aren't ratelimited anymore. - channel = self.make_request(*args) + channel = make_request() self.assertEqual(HTTPStatus.OK, channel.code, channel.result)