mirror of
https://github.com/element-hq/synapse.git
synced 2026-10-23 06:32:54 +00:00
MSC4140: authenticate the delayed event management endpoints and allow guests (#20257)
This commit is contained in:
1 parent
636c6a3812
commit
d773067277
7 files changed
+301
-75
No files matched your search
@@ -0,0 +1 @@
|
||||
Require authentication for the [MSC4140](https://github.com/matrix-org/matrix-spec-proposals/pull/4140) delayed event management endpoints, restrict them to the requesting user's own delayed events, and allow guests to use them.
|
||||
@@ -2023,7 +2023,7 @@ rc_presence:
|
||||
|
||||
*(object)* Ratelimiting settings for delayed event management.
|
||||
|
||||
This is a ratelimiting option that ratelimits attempts to restart, cancel, or view delayed events based on the sending client's account, or its source IP when requests are unauthenticated.
|
||||
This is a ratelimiting option that ratelimits attempts to restart, cancel, or view delayed events based on the sending client's account.
|
||||
|
||||
Attempts to create or send delayed events are ratelimited not by this setting, but by `rc_message`.
|
||||
|
||||
|
||||
@@ -2344,8 +2344,7 @@ properties:
|
||||
|
||||
|
||||
This is a ratelimiting option that ratelimits attempts to restart, cancel,
|
||||
or view delayed events based on the sending client's account,
|
||||
or its source IP when requests are unauthenticated.
|
||||
or view delayed events based on the sending client's account.
|
||||
|
||||
|
||||
Attempts to create or send delayed events are ratelimited not by this
|
||||
|
||||
@@ -22,7 +22,6 @@ from synapse.api.constants import EventTypes, StickyEvent, StickyEventField
|
||||
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
|
||||
from synapse.logging.context import make_deferred_yieldable
|
||||
from synapse.logging.opentracing import set_tag
|
||||
from synapse.replication.http.delayed_events import (
|
||||
@@ -211,7 +210,7 @@ class DelayedEventsHandler:
|
||||
if self._next_send_ts_changed(next_send_ts):
|
||||
self._schedule_next_at(next_send_ts)
|
||||
|
||||
async def cancel(self, request: SynapseRequest, delay_id: str) -> None:
|
||||
async def cancel(self, requester: Requester, delay_id: str) -> None:
|
||||
"""
|
||||
Cancels the scheduled delivery of the matching delayed event.
|
||||
|
||||
@@ -219,10 +218,13 @@ class DelayedEventsHandler:
|
||||
NotFoundError: if no matching delayed event could be found.
|
||||
"""
|
||||
assert self._is_master
|
||||
await self._mgmt_ratelimit(request)
|
||||
await self._delayed_event_mgmt_ratelimiter.ratelimit(requester)
|
||||
await make_deferred_yieldable(self._initialized_from_db)
|
||||
|
||||
next_send_ts = await self._store.cancel_delayed_event(delay_id)
|
||||
next_send_ts = await self._store.cancel_delayed_event(
|
||||
delay_id=delay_id,
|
||||
user_localpart=requester.user.localpart,
|
||||
)
|
||||
|
||||
if self._next_send_ts_changed(next_send_ts):
|
||||
self._schedule_next_at_or_none(next_send_ts)
|
||||
@@ -251,14 +253,14 @@ class DelayedEventsHandler:
|
||||
if self._next_send_ts_changed(next_send_ts):
|
||||
self._schedule_next_at_or_none(next_send_ts)
|
||||
|
||||
async def restart(self, request: SynapseRequest, delay_id: str) -> None:
|
||||
async def restart(self, requester: Requester, delay_id: str) -> None:
|
||||
"""
|
||||
Restarts the scheduled delivery of the matching delayed event.
|
||||
|
||||
Raises:
|
||||
NotFoundError: if no matching delayed event could be found.
|
||||
"""
|
||||
await self._mgmt_ratelimit(request)
|
||||
await self._delayed_event_mgmt_ratelimiter.ratelimit(requester)
|
||||
|
||||
# Note: We don't need to wait on `self._initialized_from_db` here as the
|
||||
# events that deals with are already marked as processed.
|
||||
@@ -266,7 +268,9 @@ class DelayedEventsHandler:
|
||||
# `restart_delayed_events` will skip over such events entirely.
|
||||
|
||||
next_send_ts = await self._store.restart_delayed_event(
|
||||
delay_id, self._get_current_ts()
|
||||
delay_id=delay_id,
|
||||
user_localpart=requester.user.localpart,
|
||||
current_ts=self._get_current_ts(),
|
||||
)
|
||||
|
||||
# Only the main process handles sending delayed events.
|
||||
@@ -274,7 +278,7 @@ class DelayedEventsHandler:
|
||||
if self._next_send_ts_changed(next_send_ts):
|
||||
self._schedule_next_at(next_send_ts)
|
||||
|
||||
async def send(self, request: SynapseRequest, delay_id: str) -> None:
|
||||
async def send(self, requester: Requester, delay_id: str) -> None:
|
||||
"""
|
||||
Immediately sends the matching delayed event, instead of waiting for its scheduled delivery.
|
||||
|
||||
@@ -282,29 +286,19 @@ class DelayedEventsHandler:
|
||||
NotFoundError: if no matching delayed event could be found.
|
||||
"""
|
||||
assert self._is_master
|
||||
await self._mgmt_ratelimit(request)
|
||||
await self._delayed_event_mgmt_ratelimiter.ratelimit(requester)
|
||||
await make_deferred_yieldable(self._initialized_from_db)
|
||||
|
||||
event, next_send_ts = await self._store.process_target_delayed_event(delay_id)
|
||||
event, next_send_ts = await self._store.process_target_delayed_event(
|
||||
delay_id=delay_id,
|
||||
user_localpart=requester.user.localpart,
|
||||
)
|
||||
|
||||
if self._next_send_ts_changed(next_send_ts):
|
||||
self._schedule_next_at_or_none(next_send_ts)
|
||||
|
||||
await self._send_event(event)
|
||||
|
||||
async def _mgmt_ratelimit(self, request: SynapseRequest) -> None:
|
||||
"""
|
||||
Ratelimit requests with the `_delayed_event_mgmt_ratelimiter` keyed on the
|
||||
user making the request, or the request's IP address if unauthed.
|
||||
"""
|
||||
if self._auth.has_access_token(request):
|
||||
requester = await self._auth.get_user_by_req(request)
|
||||
key = None
|
||||
else:
|
||||
requester = None
|
||||
key = request.getClientAddress().host
|
||||
await self._delayed_event_mgmt_ratelimiter.ratelimit(requester, key)
|
||||
|
||||
async def _send_on_timeout(self) -> None:
|
||||
self._next_delayed_event_call = None
|
||||
|
||||
|
||||
@@ -47,11 +47,13 @@ class UpdateDelayedEventServlet(RestServlet):
|
||||
|
||||
def __init__(self, hs: "HomeServer"):
|
||||
super().__init__()
|
||||
self.auth = hs.get_auth()
|
||||
self.delayed_events_handler = hs.get_delayed_events_handler()
|
||||
|
||||
async def on_POST(
|
||||
self, request: SynapseRequest, delay_id: str
|
||||
) -> tuple[int, JsonDict]:
|
||||
requester = await self.auth.get_user_by_req(request, allow_guest=True)
|
||||
body = parse_json_object_from_request(request)
|
||||
try:
|
||||
action = str(body["action"])
|
||||
@@ -72,11 +74,11 @@ class UpdateDelayedEventServlet(RestServlet):
|
||||
)
|
||||
|
||||
if enum_action == _UpdateDelayedEventAction.CANCEL:
|
||||
await self.delayed_events_handler.cancel(request, delay_id)
|
||||
await self.delayed_events_handler.cancel(requester, delay_id)
|
||||
elif enum_action == _UpdateDelayedEventAction.RESTART:
|
||||
await self.delayed_events_handler.restart(request, delay_id)
|
||||
await self.delayed_events_handler.restart(requester, delay_id)
|
||||
elif enum_action == _UpdateDelayedEventAction.SEND:
|
||||
await self.delayed_events_handler.send(request, delay_id)
|
||||
await self.delayed_events_handler.send(requester, delay_id)
|
||||
return 200, {}
|
||||
|
||||
|
||||
@@ -89,12 +91,14 @@ class CancelDelayedEventServlet(RestServlet):
|
||||
|
||||
def __init__(self, hs: "HomeServer"):
|
||||
super().__init__()
|
||||
self.auth = hs.get_auth()
|
||||
self.delayed_events_handler = hs.get_delayed_events_handler()
|
||||
|
||||
async def on_POST(
|
||||
self, request: SynapseRequest, delay_id: str
|
||||
) -> tuple[int, JsonDict]:
|
||||
await self.delayed_events_handler.cancel(request, delay_id)
|
||||
requester = await self.auth.get_user_by_req(request, allow_guest=True)
|
||||
await self.delayed_events_handler.cancel(requester, delay_id)
|
||||
return 200, {}
|
||||
|
||||
|
||||
@@ -107,12 +111,14 @@ class RestartDelayedEventServlet(RestServlet):
|
||||
|
||||
def __init__(self, hs: "HomeServer"):
|
||||
super().__init__()
|
||||
self.auth = hs.get_auth()
|
||||
self.delayed_events_handler = hs.get_delayed_events_handler()
|
||||
|
||||
async def on_POST(
|
||||
self, request: SynapseRequest, delay_id: str
|
||||
) -> tuple[int, JsonDict]:
|
||||
await self.delayed_events_handler.restart(request, delay_id)
|
||||
requester = await self.auth.get_user_by_req(request, allow_guest=True)
|
||||
await self.delayed_events_handler.restart(requester, delay_id)
|
||||
return 200, {}
|
||||
|
||||
|
||||
@@ -125,12 +131,14 @@ class SendDelayedEventServlet(RestServlet):
|
||||
|
||||
def __init__(self, hs: "HomeServer"):
|
||||
super().__init__()
|
||||
self.auth = hs.get_auth()
|
||||
self.delayed_events_handler = hs.get_delayed_events_handler()
|
||||
|
||||
async def on_POST(
|
||||
self, request: SynapseRequest, delay_id: str
|
||||
) -> tuple[int, JsonDict]:
|
||||
await self.delayed_events_handler.send(request, delay_id)
|
||||
requester = await self.auth.get_user_by_req(request, allow_guest=True)
|
||||
await self.delayed_events_handler.send(requester, delay_id)
|
||||
return 200, {}
|
||||
|
||||
|
||||
@@ -149,7 +157,7 @@ class DelayedEventServlet(RestServlet):
|
||||
async def on_GET(
|
||||
self, request: SynapseRequest, delay_id: str
|
||||
) -> tuple[int, JsonDict]:
|
||||
requester = await self.auth.get_user_by_req(request)
|
||||
requester = await self.auth.get_user_by_req(request, allow_guest=True)
|
||||
delayed_event = await self.delayed_events_handler.get_for_user(
|
||||
requester, delay_id
|
||||
)
|
||||
@@ -169,7 +177,7 @@ class DelayedEventsServlet(RestServlet):
|
||||
self.delayed_events_handler = hs.get_delayed_events_handler()
|
||||
|
||||
async def on_GET(self, request: SynapseRequest) -> tuple[int, JsonDict]:
|
||||
requester = await self.auth.get_user_by_req(request)
|
||||
requester = await self.auth.get_user_by_req(request, allow_guest=True)
|
||||
# TODO: Support Pagination stream API ("from" query parameter)
|
||||
delayed_events = await self.delayed_events_handler.get_all_for_user(requester)
|
||||
return 200, {
|
||||
|
||||
@@ -233,14 +233,16 @@ class DelayedEventsStore(SQLBaseStore):
|
||||
async def restart_delayed_event(
|
||||
self,
|
||||
delay_id: str,
|
||||
user_localpart: str,
|
||||
current_ts: Timestamp,
|
||||
) -> Timestamp:
|
||||
"""
|
||||
Restarts the send time of the matching delayed event,
|
||||
Restarts the send time of the matching delayed event owned by the given user,
|
||||
as long as it hasn't already been marked for processing.
|
||||
|
||||
Args:
|
||||
delay_id: The ID of the delayed event to restart.
|
||||
user_localpart: The localpart of the user who owns the delayed event.
|
||||
current_ts: The current time, which will be used to calculate the new send time.
|
||||
|
||||
Returns: The send time of the next delayed event to be sent,
|
||||
@@ -258,11 +260,12 @@ class DelayedEventsStore(SQLBaseStore):
|
||||
"""
|
||||
UPDATE delayed_events
|
||||
SET send_ts = ? + delay
|
||||
WHERE delay_id = ? AND NOT is_processed
|
||||
WHERE delay_id = ? AND user_localpart = ? AND NOT is_processed
|
||||
""",
|
||||
(
|
||||
current_ts,
|
||||
delay_id,
|
||||
user_localpart,
|
||||
),
|
||||
)
|
||||
if txn.rowcount == 0:
|
||||
@@ -442,13 +445,18 @@ class DelayedEventsStore(SQLBaseStore):
|
||||
async def process_target_delayed_event(
|
||||
self,
|
||||
delay_id: str,
|
||||
user_localpart: str,
|
||||
) -> tuple[
|
||||
DelayedEventDetails,
|
||||
Timestamp | None,
|
||||
]:
|
||||
"""
|
||||
Marks for processing the matching delayed event, regardless of its timeout time,
|
||||
as long as it has not already been marked as such.
|
||||
Marks for processing the matching delayed event owned by the given user,
|
||||
regardless of its timeout time, as long as it has not already been marked as such.
|
||||
|
||||
Args:
|
||||
delay_id: The ID of the delayed event to process.
|
||||
user_localpart: The localpart of the user who owns the delayed event.
|
||||
|
||||
Returns: The details of the matching delayed event,
|
||||
and the send time of the next delayed event to be sent, if any.
|
||||
@@ -467,7 +475,7 @@ class DelayedEventsStore(SQLBaseStore):
|
||||
"""
|
||||
UPDATE delayed_events
|
||||
SET is_processed = TRUE
|
||||
WHERE delay_id = ? AND NOT is_processed
|
||||
WHERE delay_id = ? AND user_localpart = ? AND NOT is_processed
|
||||
RETURNING
|
||||
room_id,
|
||||
event_type,
|
||||
@@ -478,7 +486,7 @@ class DelayedEventsStore(SQLBaseStore):
|
||||
sticky_duration_ms,
|
||||
user_localpart
|
||||
""",
|
||||
(delay_id,),
|
||||
(delay_id, user_localpart),
|
||||
)
|
||||
row = txn.fetchone()
|
||||
if row is None:
|
||||
@@ -502,9 +510,18 @@ class DelayedEventsStore(SQLBaseStore):
|
||||
"process_target_delayed_event", process_target_delayed_event_txn
|
||||
)
|
||||
|
||||
async def cancel_delayed_event(self, delay_id: str) -> Timestamp | None:
|
||||
async def cancel_delayed_event(
|
||||
self,
|
||||
delay_id: str,
|
||||
user_localpart: str,
|
||||
) -> Timestamp | None:
|
||||
"""
|
||||
Cancels the matching delayed event, i.e. remove it as long as it hasn't been processed.
|
||||
Cancels the matching delayed event owned by the given user,
|
||||
i.e. remove it as long as it hasn't been processed.
|
||||
|
||||
Args:
|
||||
delay_id: The ID of the delayed event to cancel.
|
||||
user_localpart: The localpart of the user who owns the delayed event.
|
||||
|
||||
Returns: The send time of the next delayed event to be sent, if any.
|
||||
|
||||
@@ -521,6 +538,7 @@ class DelayedEventsStore(SQLBaseStore):
|
||||
table="delayed_events",
|
||||
keyvalues={
|
||||
"delay_id": delay_id,
|
||||
"user_localpart": user_localpart,
|
||||
"is_processed": False,
|
||||
},
|
||||
)
|
||||
|
||||
@@ -17,6 +17,8 @@
|
||||
import sqlite3
|
||||
from http import HTTPStatus
|
||||
from typing import Literal, overload
|
||||
from unittest.mock import Mock, patch
|
||||
from urllib.parse import urlencode
|
||||
|
||||
from parameterized import parameterized
|
||||
|
||||
@@ -30,6 +32,7 @@ from synapse.api.constants import (
|
||||
StickyEvent,
|
||||
)
|
||||
from synapse.api.errors import Codes, NotFoundError
|
||||
from synapse.appservice import ApplicationService
|
||||
from synapse.rest import admin
|
||||
from synapse.rest.client import (
|
||||
account,
|
||||
@@ -42,7 +45,7 @@ from synapse.rest.client import (
|
||||
)
|
||||
from synapse.rest.client.delayed_events import _UpdateDelayedEventAction
|
||||
from synapse.server import HomeServer
|
||||
from synapse.types import JsonDict, create_requester
|
||||
from synapse.types import JsonDict, UserID, create_requester
|
||||
from synapse.util.clock import Clock
|
||||
from synapse.util.duration import Duration
|
||||
|
||||
@@ -242,12 +245,27 @@ class DelayedEventsTestCaseBase(DelayedEventsHelperMixin):
|
||||
|
||||
return content
|
||||
|
||||
def _schedule_delayed_event(self, access_token: str) -> str:
|
||||
"""Schedule a delayed message event with a short delay, returning its delay_id."""
|
||||
channel = self._make_delayed_event_request(
|
||||
room_id=self.room_id,
|
||||
delay=Duration(milliseconds=900),
|
||||
event_type=_EVENT_TYPE,
|
||||
content={},
|
||||
access_token=access_token,
|
||||
)
|
||||
self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
|
||||
delay_id = channel.json_body.get("delay_id")
|
||||
assert delay_id is not None
|
||||
return delay_id
|
||||
|
||||
def _update_delayed_event(
|
||||
self,
|
||||
delay_id: str,
|
||||
action: _UpdateDelayedEventAction,
|
||||
action_in_path: bool,
|
||||
access_token: str | None = None,
|
||||
appservice_user_id: str | None = None,
|
||||
) -> FakeChannel:
|
||||
"""
|
||||
Call the endpoint to cancel/restart/send a scheduled delayed event.
|
||||
@@ -258,6 +276,8 @@ class DelayedEventsTestCaseBase(DelayedEventsHelperMixin):
|
||||
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.
|
||||
appservice_user_id: The user an appservice asserts with `?user_id=`,
|
||||
when `access_token` is an appservice token.
|
||||
"""
|
||||
path = f"{_MANAGEMENT_PATH_PREFIX}/{delay_id}"
|
||||
body = {}
|
||||
@@ -265,12 +285,30 @@ class DelayedEventsTestCaseBase(DelayedEventsHelperMixin):
|
||||
path += f"/{action.value}"
|
||||
else:
|
||||
body["action"] = action.value
|
||||
if appservice_user_id is not None:
|
||||
path += "?" + urlencode({"user_id": appservice_user_id})
|
||||
return self.make_request("POST", path, body, access_token)
|
||||
|
||||
|
||||
class DelayedEventsTestCase(DelayedEventsTestCaseBase):
|
||||
"""Tests getting and managing delayed events."""
|
||||
|
||||
def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer:
|
||||
# An appservice that may act on behalf of any user of the homeserver.
|
||||
self.appservice = ApplicationService(
|
||||
token="i_am_an_app_service",
|
||||
id="1234",
|
||||
namespaces={"users": [{"regex": "@.*", "exclusive": False}]},
|
||||
sender=UserID.from_string("@as_main:test"),
|
||||
)
|
||||
|
||||
mock_load_appservices = Mock(return_value=[self.appservice])
|
||||
with patch(
|
||||
"synapse.storage.databases.main.appservice.load_appservices",
|
||||
mock_load_appservices,
|
||||
):
|
||||
return self.setup_test_homeserver()
|
||||
|
||||
def test_delayed_events_empty_on_startup(self) -> None:
|
||||
self.assertListEqual([], self._get_delayed_events())
|
||||
|
||||
@@ -516,6 +554,7 @@ class DelayedEventsTestCase(DelayedEventsTestCaseBase):
|
||||
def test_get_delayed_events_auth(self) -> None:
|
||||
channel = self.make_request("GET", _MANAGEMENT_PATH_PREFIX)
|
||||
self.assertEqual(HTTPStatus.UNAUTHORIZED, channel.code, channel.result)
|
||||
self.assertEqual(Codes.MISSING_TOKEN, channel.json_body["errcode"])
|
||||
|
||||
@unittest.override_config(
|
||||
{"rc_delayed_event_mgmt": {"per_second": 0.5, "burst_count": 1}}
|
||||
@@ -551,6 +590,7 @@ class DelayedEventsTestCase(DelayedEventsTestCaseBase):
|
||||
channel = self.make_request(
|
||||
"POST",
|
||||
f"{_MANAGEMENT_PATH_PREFIX}/abc",
|
||||
access_token=self.user1_access_token,
|
||||
)
|
||||
self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.result)
|
||||
self.assertEqual(
|
||||
@@ -563,6 +603,7 @@ class DelayedEventsTestCase(DelayedEventsTestCaseBase):
|
||||
"POST",
|
||||
f"{_MANAGEMENT_PATH_PREFIX}/abc",
|
||||
{},
|
||||
access_token=self.user1_access_token,
|
||||
)
|
||||
self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.result)
|
||||
self.assertEqual(
|
||||
@@ -575,6 +616,7 @@ class DelayedEventsTestCase(DelayedEventsTestCaseBase):
|
||||
"POST",
|
||||
f"{_MANAGEMENT_PATH_PREFIX}/abc",
|
||||
{"action": "oops"},
|
||||
access_token=self.user1_access_token,
|
||||
)
|
||||
self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.result)
|
||||
self.assertEqual(
|
||||
@@ -594,8 +636,181 @@ class DelayedEventsTestCase(DelayedEventsTestCaseBase):
|
||||
action: _UpdateDelayedEventAction,
|
||||
action_in_path: bool,
|
||||
) -> None:
|
||||
channel = self._update_delayed_event("abc", action, action_in_path)
|
||||
channel = self._update_delayed_event(
|
||||
"abc", action, action_in_path, self.user1_access_token
|
||||
)
|
||||
self.assertEqual(HTTPStatus.NOT_FOUND, channel.code, channel.result)
|
||||
self.assertEqual(Codes.NOT_FOUND, channel.json_body["errcode"])
|
||||
|
||||
@parameterized.expand(
|
||||
(
|
||||
(action, action_in_path)
|
||||
for action in _UpdateDelayedEventAction
|
||||
for action_in_path in (True, False)
|
||||
)
|
||||
)
|
||||
def test_update_delayed_event_without_auth(
|
||||
self,
|
||||
action: _UpdateDelayedEventAction,
|
||||
action_in_path: bool,
|
||||
) -> None:
|
||||
"""Delayed events cannot be managed without an access token."""
|
||||
delay_id = self._schedule_delayed_event(self.user1_access_token)
|
||||
|
||||
channel = self._update_delayed_event(delay_id, action, action_in_path)
|
||||
self.assertEqual(HTTPStatus.UNAUTHORIZED, channel.code, channel.result)
|
||||
self.assertEqual(Codes.MISSING_TOKEN, channel.json_body["errcode"])
|
||||
|
||||
# The event is still scheduled, and is sent on its original timeout.
|
||||
self.assertEqual(1, len(self._get_delayed_events()))
|
||||
self._check_for_delayed_event_in_sync(self.user1_access_token, delay_id, False)
|
||||
self.reactor.advance(1)
|
||||
self._check_for_delayed_event_in_sync(self.user1_access_token, delay_id, True)
|
||||
|
||||
@parameterized.expand(
|
||||
(
|
||||
(action, action_in_path)
|
||||
for action in _UpdateDelayedEventAction
|
||||
for action_in_path in (True, False)
|
||||
)
|
||||
)
|
||||
def test_update_delayed_event_of_other_user(
|
||||
self,
|
||||
action: _UpdateDelayedEventAction,
|
||||
action_in_path: bool,
|
||||
) -> None:
|
||||
"""Delayed events can only be managed by the user who scheduled them."""
|
||||
delay_id = self._schedule_delayed_event(self.user1_access_token)
|
||||
|
||||
channel = self._update_delayed_event(
|
||||
delay_id, action, action_in_path, self.user2_access_token
|
||||
)
|
||||
self.assertEqual(HTTPStatus.NOT_FOUND, channel.code, channel.result)
|
||||
self.assertEqual(Codes.NOT_FOUND, channel.json_body["errcode"])
|
||||
|
||||
# The event is still scheduled, and is sent on its original timeout.
|
||||
self.assertEqual(1, len(self._get_delayed_events()))
|
||||
self._check_for_delayed_event_in_sync(self.user1_access_token, delay_id, False)
|
||||
self.reactor.advance(1)
|
||||
self._check_for_delayed_event_in_sync(self.user1_access_token, delay_id, True)
|
||||
|
||||
@parameterized.expand(
|
||||
(
|
||||
(action, action_in_path)
|
||||
for action in _UpdateDelayedEventAction
|
||||
for action_in_path in (True, False)
|
||||
)
|
||||
)
|
||||
def test_update_delayed_event_as_appservice(
|
||||
self,
|
||||
action: _UpdateDelayedEventAction,
|
||||
action_in_path: bool,
|
||||
) -> None:
|
||||
"""An appservice can manage the delayed events of the user it asserts."""
|
||||
delay_id = self._schedule_delayed_event(self.user1_access_token)
|
||||
|
||||
# Asserting a different user gives no access to user1's delayed event.
|
||||
channel = self._update_delayed_event(
|
||||
delay_id,
|
||||
action,
|
||||
action_in_path,
|
||||
self.appservice.token,
|
||||
appservice_user_id=self.user2_user_id,
|
||||
)
|
||||
self.assertEqual(HTTPStatus.NOT_FOUND, channel.code, channel.result)
|
||||
self.assertEqual(Codes.NOT_FOUND, channel.json_body["errcode"])
|
||||
self.assertEqual(1, len(self._get_delayed_events()))
|
||||
|
||||
channel = self._update_delayed_event(
|
||||
delay_id,
|
||||
action,
|
||||
action_in_path,
|
||||
self.appservice.token,
|
||||
appservice_user_id=self.user1_user_id,
|
||||
)
|
||||
self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
|
||||
|
||||
if action == _UpdateDelayedEventAction.CANCEL:
|
||||
self.assertListEqual([], self._get_delayed_events())
|
||||
self.reactor.advance(1)
|
||||
self._check_for_delayed_event_in_sync(
|
||||
self.user1_access_token, delay_id, False
|
||||
)
|
||||
elif action == _UpdateDelayedEventAction.RESTART:
|
||||
self.assertEqual(1, len(self._get_delayed_events()))
|
||||
self.reactor.advance(1)
|
||||
self._check_for_delayed_event_in_sync(
|
||||
self.user1_access_token, delay_id, True
|
||||
)
|
||||
else:
|
||||
self.assertListEqual([], self._get_delayed_events())
|
||||
self._check_for_delayed_event_in_sync(
|
||||
self.user1_access_token, delay_id, True
|
||||
)
|
||||
|
||||
@unittest.override_config(
|
||||
{
|
||||
"allow_guest_access": True,
|
||||
# This test makes more management requests than the default burst allows.
|
||||
"rc_delayed_event_mgmt": {"per_second": 100, "burst_count": 100},
|
||||
}
|
||||
)
|
||||
def test_guest_can_manage_delayed_events(self) -> None:
|
||||
"""All delayed event endpoints are available to guest accounts."""
|
||||
self.helper.send_state(
|
||||
self.room_id,
|
||||
EventTypes.GuestAccess,
|
||||
{EventContentFields.GUEST_ACCESS: GuestAccess.CAN_JOIN},
|
||||
tok=self.user1_access_token,
|
||||
)
|
||||
|
||||
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.join(self.room_id, user=guest_user_id, tok=guest_access_token)
|
||||
|
||||
delay_id = self._schedule_delayed_event(guest_access_token)
|
||||
|
||||
channel = self.make_request(
|
||||
"GET", _MANAGEMENT_PATH_PREFIX, access_token=guest_access_token
|
||||
)
|
||||
self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
|
||||
self.assertEqual(
|
||||
[delay_id],
|
||||
[event["delay_id"] for event in channel.json_body["delayed_events"]],
|
||||
)
|
||||
|
||||
channel = self.make_request(
|
||||
"GET",
|
||||
f"{_MANAGEMENT_PATH_PREFIX}/{delay_id}",
|
||||
access_token=guest_access_token,
|
||||
)
|
||||
self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
|
||||
|
||||
for action in (
|
||||
_UpdateDelayedEventAction.RESTART,
|
||||
_UpdateDelayedEventAction.CANCEL,
|
||||
):
|
||||
channel = self._update_delayed_event(
|
||||
delay_id, action, True, guest_access_token
|
||||
)
|
||||
self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
|
||||
|
||||
channel = self.make_request(
|
||||
"GET", _MANAGEMENT_PATH_PREFIX, access_token=guest_access_token
|
||||
)
|
||||
self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
|
||||
self.assertListEqual([], channel.json_body["delayed_events"])
|
||||
|
||||
delay_id = self._schedule_delayed_event(guest_access_token)
|
||||
|
||||
channel = self._update_delayed_event(
|
||||
delay_id, _UpdateDelayedEventAction.SEND, True, guest_access_token
|
||||
)
|
||||
self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
|
||||
self._check_for_delayed_event_in_sync(guest_access_token, delay_id, True)
|
||||
|
||||
@parameterized.expand((True, False))
|
||||
def test_cancel_delayed_state_event(self, action_in_path: bool) -> None:
|
||||
@@ -632,7 +847,10 @@ class DelayedEventsTestCase(DelayedEventsTestCaseBase):
|
||||
)
|
||||
|
||||
channel = self._update_delayed_event(
|
||||
delay_id, _UpdateDelayedEventAction.CANCEL, action_in_path
|
||||
delay_id,
|
||||
_UpdateDelayedEventAction.CANCEL,
|
||||
action_in_path,
|
||||
self.user1_access_token,
|
||||
)
|
||||
self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
|
||||
self.assertListEqual([], self._get_delayed_events())
|
||||
@@ -656,7 +874,7 @@ class DelayedEventsTestCase(DelayedEventsTestCaseBase):
|
||||
)
|
||||
def test_cancel_delayed_event_ratelimit(self, action_in_path: bool) -> None:
|
||||
delay_ids = []
|
||||
for _ in range(3):
|
||||
for _ in range(2):
|
||||
channel = self._make_delayed_event_request(
|
||||
room_id=self.room_id,
|
||||
delay=Duration(milliseconds=100000),
|
||||
@@ -670,18 +888,6 @@ class DelayedEventsTestCase(DelayedEventsTestCaseBase):
|
||||
delay_ids.append(delay_id)
|
||||
|
||||
delay_id = delay_ids.pop(0)
|
||||
channel = self._update_delayed_event(
|
||||
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, _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,
|
||||
@@ -757,7 +963,10 @@ class DelayedEventsTestCase(DelayedEventsTestCaseBase):
|
||||
)
|
||||
|
||||
channel = self._update_delayed_event(
|
||||
delay_id, _UpdateDelayedEventAction.SEND, action_in_path
|
||||
delay_id,
|
||||
_UpdateDelayedEventAction.SEND,
|
||||
action_in_path,
|
||||
self.user1_access_token,
|
||||
)
|
||||
self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
|
||||
self.assertListEqual([], self._get_delayed_events())
|
||||
@@ -790,12 +999,18 @@ class DelayedEventsTestCase(DelayedEventsTestCaseBase):
|
||||
delay_ids.append(delay_id)
|
||||
|
||||
channel = self._update_delayed_event(
|
||||
delay_ids.pop(0), _UpdateDelayedEventAction.SEND, action_in_path
|
||||
delay_ids.pop(0),
|
||||
_UpdateDelayedEventAction.SEND,
|
||||
action_in_path,
|
||||
self.user1_access_token,
|
||||
)
|
||||
self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
|
||||
|
||||
channel = self._update_delayed_event(
|
||||
delay_ids.pop(0), _UpdateDelayedEventAction.SEND, action_in_path
|
||||
delay_ids.pop(0),
|
||||
_UpdateDelayedEventAction.SEND,
|
||||
action_in_path,
|
||||
self.user1_access_token,
|
||||
)
|
||||
self.assertEqual(HTTPStatus.TOO_MANY_REQUESTS, channel.code, channel.result)
|
||||
|
||||
@@ -834,7 +1049,10 @@ class DelayedEventsTestCase(DelayedEventsTestCaseBase):
|
||||
)
|
||||
|
||||
channel = self._update_delayed_event(
|
||||
delay_id, _UpdateDelayedEventAction.RESTART, action_in_path
|
||||
delay_id,
|
||||
_UpdateDelayedEventAction.RESTART,
|
||||
action_in_path,
|
||||
self.user1_access_token,
|
||||
)
|
||||
self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
|
||||
|
||||
@@ -872,7 +1090,7 @@ class DelayedEventsTestCase(DelayedEventsTestCaseBase):
|
||||
)
|
||||
def test_restart_delayed_event_ratelimit(self, action_in_path: bool) -> None:
|
||||
delay_ids = []
|
||||
for _ in range(3):
|
||||
for _ in range(2):
|
||||
channel = self._make_delayed_event_request(
|
||||
room_id=self.room_id,
|
||||
delay=Duration(milliseconds=100000),
|
||||
@@ -886,18 +1104,6 @@ class DelayedEventsTestCase(DelayedEventsTestCaseBase):
|
||||
delay_ids.append(delay_id)
|
||||
|
||||
delay_id = delay_ids.pop(0)
|
||||
channel = self._update_delayed_event(
|
||||
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, _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,
|
||||
|
||||
Reference in new issue
Block a user