From 6baeda9ab7df685ebd2d6f4bb40a8a0af4dfea34 Mon Sep 17 00:00:00 2001 From: Nicolas Werner Date: Wed, 17 Aug 2022 23:29:36 +0200 Subject: [PATCH 1/5] Move the older dehydrated devices behind a config flag --- synapse/config/experimental.py | 4 ++++ synapse/rest/client/devices.py | 5 +++-- 2 files changed, 7 insertions(+), 2 deletions(-) diff --git a/synapse/config/experimental.py b/synapse/config/experimental.py index c1ff41753994..fc5d959432e8 100644 --- a/synapse/config/experimental.py +++ b/synapse/config/experimental.py @@ -35,6 +35,10 @@ def read_config(self, config: JsonDict, **kwargs: Any) -> None: # MSC2285 (unstable private read receipts) self.msc2285_enabled: bool = experimental.get("msc2285_enabled", False) + # MSC2697 (device dehydration) + # Enabled by default since this option was added after adding the feature. + self.msc2697_enabled: bool = experimental.get("msc2697_enabled", True) + # MSC3244 (room version capabilities) self.msc3244_enabled: bool = experimental.get("msc3244_enabled", True) diff --git a/synapse/rest/client/devices.py b/synapse/rest/client/devices.py index ed6ce78d4771..ab6149f0c6c4 100644 --- a/synapse/rest/client/devices.py +++ b/synapse/rest/client/devices.py @@ -331,5 +331,6 @@ def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None: DeleteDevicesRestServlet(hs).register(http_server) DevicesRestServlet(hs).register(http_server) DeviceRestServlet(hs).register(http_server) - DehydratedDeviceServlet(hs).register(http_server) - ClaimDehydratedDeviceServlet(hs).register(http_server) + if hs.config.experimental.msc2697_enabled: + DehydratedDeviceServlet(hs).register(http_server) + ClaimDehydratedDeviceServlet(hs).register(http_server) From ace4f49aeb9ec8344059b378bc15c957215eff76 Mon Sep 17 00:00:00 2001 From: Nicolas Werner Date: Thu, 18 Aug 2022 19:23:53 +0200 Subject: [PATCH 2/5] Implement MSC3814 Some differences: - We use GET instead of POST for the events endpoint. - We don't delete the device implicitly when fetching events. - We allow the fetch device messages endpoint for both dehydrated and the current requesters device. Signed-off-by: Nicolas Werner --- synapse/config/experimental.py | 4 ++ synapse/handlers/devicemessage.py | 69 +++++++++++++++++++- synapse/rest/client/devices.py | 53 +++++++++++++-- tests/handlers/test_device.py | 104 ++++++++++++++++++++++++++++++ 4 files changed, 223 insertions(+), 7 deletions(-) diff --git a/synapse/config/experimental.py b/synapse/config/experimental.py index fc5d959432e8..e1ea65b69592 100644 --- a/synapse/config/experimental.py +++ b/synapse/config/experimental.py @@ -39,6 +39,10 @@ def read_config(self, config: JsonDict, **kwargs: Any) -> None: # Enabled by default since this option was added after adding the feature. self.msc2697_enabled: bool = experimental.get("msc2697_enabled", True) + # MSC3814 (dehydrated devices with SSSS) + # This is an alternative method to achieve the same goals as MSC2697. + self.msc3814_enabled: bool = experimental.get("msc3814_enabled", False) + # MSC3244 (room version capabilities) self.msc3244_enabled: bool = experimental.get("msc3244_enabled", True) diff --git a/synapse/handlers/devicemessage.py b/synapse/handlers/devicemessage.py index 444c08bc2eef..39cd0084b630 100644 --- a/synapse/handlers/devicemessage.py +++ b/synapse/handlers/devicemessage.py @@ -13,10 +13,10 @@ # limitations under the License. import logging -from typing import TYPE_CHECKING, Any, Dict +from typing import TYPE_CHECKING, Any, Dict, Optional from synapse.api.constants import EduTypes, ToDeviceEventTypes -from synapse.api.errors import SynapseError +from synapse.api.errors import Codes, SynapseError from synapse.api.ratelimiting import Ratelimiter from synapse.logging.context import run_in_background from synapse.logging.opentracing import ( @@ -46,6 +46,9 @@ def __init__(self, hs: "HomeServer"): self.store = hs.get_datastores().main self.notifier = hs.get_notifier() self.is_mine = hs.is_mine + if hs.config.experimental.msc3814_enabled: + self.event_sources = hs.get_event_sources() + self.device_handler = hs.get_device_handler() # We only need to poke the federation sender explicitly if its on the # same instance. Other federation sender instances will get notified by @@ -293,3 +296,65 @@ async def send_device_message( # Enqueue a new federation transaction to send the new # device messages to each remote destination. self.federation_sender.send_device_messages(destination) + + async def get_events_for_dehydrated_device( + self, + requester: Requester, + device_id: str, + since_token: Optional[str], + limit: int, + ) -> JsonDict: + """Fetches up to `limit` events sent to `device_id` starting from `since_token` and returns the new since token.""" + + user_id = requester.user.to_string() + + # TODO(Nico): Figure out who should be allowed to use that endpoint. + # For now we just allow it for yourself and for the dehydrated device. + if device_id != requester.device_id: + dehydrated_device = await self.device_handler.get_dehydrated_device(user_id) + if dehydrated_device is not None and device_id != dehydrated_device[0]: + raise SynapseError( + 403, + "Can only fetch messages for own device or dehydrated devices", + Codes.UNAUTHORIZED, + ) + + since_stream_id = 0 + if since_token and len(since_token) > 1 and since_token[0] == "d": + since_stream_id = int(since_token[1:]) + + # if we have a since token, delete any to-device messages before that token + # (since we now know that the device has received them) + deleted = await self.store.delete_messages_for_device( + user_id, device_id, since_stream_id + ) + logger.debug( + "Deleted %d to-device messages up to %d", deleted, since_stream_id + ) + + to_token = self.event_sources.get_current_token().to_device_key + + messages, stream_id = await self.store.get_messages_for_device( + user_id, device_id, since_stream_id, to_token, limit + ) + + for message in messages: + # We pop here as we shouldn't be sending the message ID down + # `/sync` + message_id = message.pop("message_id", None) + if message_id: + set_tag(SynapseTags.TO_DEVICE_MESSAGE_ID, message_id) + + logger.debug( + "Returning %d to-device messages between %d and %d (current token: %d) for dehydrated device %s", + len(messages), + since_stream_id, + stream_id, + to_token, + device_id, + ) + + return { + "events": messages, + "next_batch": f"d{stream_id}", + } diff --git a/synapse/rest/client/devices.py b/synapse/rest/client/devices.py index ab6149f0c6c4..8600bf5718d1 100644 --- a/synapse/rest/client/devices.py +++ b/synapse/rest/client/devices.py @@ -18,11 +18,13 @@ from synapse.api import errors from synapse.api.errors import NotFoundError -from synapse.http.server import HttpServer +from synapse.http.server import HttpServer, cancellable from synapse.http.servlet import ( RestServlet, assert_params_in_dict, + parse_integer, parse_json_object_from_request, + parse_string, ) from synapse.http.site import SynapseRequest from synapse.rest.client._base import client_patterns, interactive_auth_handler @@ -194,6 +196,8 @@ async def on_PUT( class DehydratedDeviceServlet(RestServlet): """Retrieve or store a dehydrated device. + Implements both MSC2697 and MSC3814. + GET /org.matrix.msc2697.v2/dehydrated_device HTTP/1.1 200 OK @@ -226,14 +230,19 @@ class DehydratedDeviceServlet(RestServlet): """ - PATTERNS = client_patterns("/org.matrix.msc2697.v2/dehydrated_device", releases=()) - - def __init__(self, hs: "HomeServer"): + def __init__(self, hs: "HomeServer", msc2697: bool = True): super().__init__() self.hs = hs self.auth = hs.get_auth() self.device_handler = hs.get_device_handler() + self.PATTERNS = client_patterns( + "/org.matrix.msc2697.v2/dehydrated_device" + if msc2697 + else "/org.matrix.msc3814.v1/dehydrated_device$", + releases=(), + ) + async def on_GET(self, request: SynapseRequest) -> Tuple[int, JsonDict]: requester = await self.auth.get_user_by_req(request) dehydrated_device = await self.device_handler.get_dehydrated_device( @@ -327,10 +336,44 @@ async def on_POST(self, request: SynapseRequest) -> Tuple[int, JsonDict]: return 200, result +class DehydratedDeviceEventsServlet(RestServlet): + PATTERNS = client_patterns( + "/org.matrix.msc3814.v1/dehydrated_device/(?P[^/]*)/events$", + releases=(), + ) + + def __init__(self, hs: "HomeServer"): + super().__init__() + self.message_handler = hs.get_device_message_handler() + self.auth = hs.get_auth() + self.store = hs.get_datastores().main + + @cancellable + async def on_GET( + self, request: SynapseRequest, device_id: str + ) -> Tuple[int, JsonDict]: + requester = await self.auth.get_user_by_req(request) + + from_tok = parse_string(request, "from") + limit = parse_integer(request, "limit", 100) + + msgs = await self.message_handler.get_events_for_dehydrated_device( + requester=requester, + device_id=device_id, + since_token=from_tok, + limit=limit, + ) + + return 200, msgs + + def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None: DeleteDevicesRestServlet(hs).register(http_server) DevicesRestServlet(hs).register(http_server) DeviceRestServlet(hs).register(http_server) if hs.config.experimental.msc2697_enabled: - DehydratedDeviceServlet(hs).register(http_server) + DehydratedDeviceServlet(hs, msc2697=True).register(http_server) ClaimDehydratedDeviceServlet(hs).register(http_server) + if hs.config.experimental.msc3814_enabled: + DehydratedDeviceServlet(hs, msc2697=False).register(http_server) + DehydratedDeviceEventsServlet(hs).register(http_server) diff --git a/tests/handlers/test_device.py b/tests/handlers/test_device.py index b8b465d35b8f..de3cccb5d6d5 100644 --- a/tests/handlers/test_device.py +++ b/tests/handlers/test_device.py @@ -16,11 +16,13 @@ from typing import Optional +from twisted.internet.defer import ensureDeferred from twisted.test.proto_helpers import MemoryReactor from synapse.api.errors import NotFoundError, SynapseError from synapse.handlers.device import MAX_DEVICE_DISPLAY_NAME_LEN from synapse.server import HomeServer +from synapse.types import create_requester from synapse.util import Clock from tests import unittest @@ -265,6 +267,7 @@ class DehydrationTestCase(unittest.HomeserverTestCase): def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer: hs = self.setup_test_homeserver("server", federation_http_client=None) self.handler = hs.get_device_handler() + self.message_handler = hs.get_device_message_handler() self.registration = hs.get_registration_handler() self.auth = hs.get_auth() self.store = hs.get_datastores().main @@ -342,3 +345,104 @@ def test_dehydrate_and_rehydrate_device(self) -> None: ret = self.get_success(self.handler.get_dehydrated_device(user_id=user_id)) self.assertIsNone(ret) + + @unittest.override_config( + {"experimental_features": {"msc2697_enabled": False, "msc3814_enabled": True}} + ) + def test_dehydrate_v2_and_fetch_events(self) -> None: + user_id = "@boris:server" + + self.get_success(self.store.register_user(user_id, "foobar")) + + # First check if we can store and fetch a dehydrated device + stored_dehydrated_device_id = self.get_success( + self.handler.store_dehydrated_device( + user_id=user_id, + device_data={"device_data": {"foo": "bar"}}, + initial_device_display_name="dehydrated device", + ) + ) + + retrieved_device_id, device_data = self.get_success( + self.handler.get_dehydrated_device(user_id=user_id) + ) + + self.assertEqual(retrieved_device_id, stored_dehydrated_device_id) + self.assertEqual(device_data, {"device_data": {"foo": "bar"}}) + + # Create a new login for the user + device_id, access_token, _expiration_time, _refresh_token = self.get_success( + self.registration.register_device( + user_id=user_id, + device_id=None, + initial_display_name="new device", + ) + ) + + requester = create_requester(user_id, device_id=device_id) + + # Fetching messages for a non existing device should return an error + self.get_failure( + self.message_handler.get_events_for_dehydrated_device( + requester=requester, + device_id="not the right device ID", + since_token=None, + limit=10, + ), + SynapseError, + ) + + # Send a message to the dehydrated device + ensureDeferred( + self.message_handler.send_device_message( + requester=requester, + message_type="test.message", + messages={user_id: {stored_dehydrated_device_id: {"body": "foo"}}}, + ) + ) + self.pump() + + # Fetch the message of the dehydrated device + res = self.get_success( + self.message_handler.get_events_for_dehydrated_device( + requester=requester, + device_id=stored_dehydrated_device_id, + since_token=None, + limit=10, + ) + ) + + self.assertTrue(len(res["next_batch"]) > 1) + self.assertEqual(len(res["events"]), 1) + self.assertEqual(res["events"][0]["content"]["body"], "foo") + + # Fetch the message of the dehydrated device again, which should return nothing and delete the old messages + res = self.get_success( + self.message_handler.get_events_for_dehydrated_device( + requester=requester, + device_id=stored_dehydrated_device_id, + since_token=res["next_batch"], + limit=10, + ) + ) + self.assertTrue(len(res["next_batch"]) > 1) + self.assertEqual(len(res["events"]), 0) + + # Fetching messages without since should return nothing, since the messages got deleted + res = self.get_success( + self.message_handler.get_events_for_dehydrated_device( + requester=requester, + device_id=stored_dehydrated_device_id, + since_token=None, + limit=10, + ) + ) + self.assertTrue(len(res["next_batch"]) > 1) + self.assertEqual(len(res["events"]), 0) + + # We don't delete the device when fetch messages for now. + # # make sure that the device ID that we were initially assigned no longer exists + # self.get_failure( + # self.handler.get_device(user_id, device_id), + # NotFoundError, + # ) From 49f892db8d10b9bee5a94d257b430eed4ebbce53 Mon Sep 17 00:00:00 2001 From: Nicolas Werner Date: Mon, 22 Aug 2022 18:04:22 +0200 Subject: [PATCH 3/5] Add changelog --- changelog.d/13581.feature | 1 + 1 file changed, 1 insertion(+) create mode 100644 changelog.d/13581.feature diff --git a/changelog.d/13581.feature b/changelog.d/13581.feature new file mode 100644 index 000000000000..1db4ab71d361 --- /dev/null +++ b/changelog.d/13581.feature @@ -0,0 +1 @@ +Implement [MSC3814](https://github.com/matrix-org/matrix-spec-proposals/pull/3814), dehydrated devices v2/shrivelled sessions, with a few changes (as proposed on the MSC) and move [MSC2697](https://github.com/matrix-org/matrix-spec-proposals/pull/2697) behind a config flag. Contributed by Nico from Famedly. From 6c183c53f27f2780f4e63a7f4c9729aaa313db29 Mon Sep 17 00:00:00 2001 From: Nicolas Werner Date: Thu, 25 Aug 2022 15:10:08 +0200 Subject: [PATCH 4/5] Address review feedback Return errors on invalid token formats and terminate regex in both cases --- synapse/handlers/devicemessage.py | 21 ++++++++++++++++++--- synapse/rest/client/devices.py | 2 +- 2 files changed, 19 insertions(+), 4 deletions(-) diff --git a/synapse/handlers/devicemessage.py b/synapse/handlers/devicemessage.py index 39cd0084b630..4eecaba926f7 100644 --- a/synapse/handlers/devicemessage.py +++ b/synapse/handlers/devicemessage.py @@ -13,6 +13,7 @@ # limitations under the License. import logging +from http import HTTPStatus from typing import TYPE_CHECKING, Any, Dict, Optional from synapse.api.constants import EduTypes, ToDeviceEventTypes @@ -314,14 +315,28 @@ async def get_events_for_dehydrated_device( dehydrated_device = await self.device_handler.get_dehydrated_device(user_id) if dehydrated_device is not None and device_id != dehydrated_device[0]: raise SynapseError( - 403, + HTTPStatus.FORBIDDEN, "Can only fetch messages for own device or dehydrated devices", Codes.UNAUTHORIZED, ) since_stream_id = 0 - if since_token and len(since_token) > 1 and since_token[0] == "d": - since_stream_id = int(since_token[1:]) + if since_token: + if not since_token.startswith("d"): + raise SynapseError( + HTTPStatus.BAD_REQUEST, + "from parameter %r has an invalid format" % (since_token,), + errcode=Codes.INVALID_PARAM, + ) + + try: + since_stream_id = int(since_token[1:]) + except Exception: + raise SynapseError( + HTTPStatus.BAD_REQUEST, + "from parameter %r has an invalid format" % (since_token,), + errcode=Codes.INVALID_PARAM, + ) # if we have a since token, delete any to-device messages before that token # (since we now know that the device has received them) diff --git a/synapse/rest/client/devices.py b/synapse/rest/client/devices.py index 8600bf5718d1..c5260aff28a8 100644 --- a/synapse/rest/client/devices.py +++ b/synapse/rest/client/devices.py @@ -237,7 +237,7 @@ def __init__(self, hs: "HomeServer", msc2697: bool = True): self.device_handler = hs.get_device_handler() self.PATTERNS = client_patterns( - "/org.matrix.msc2697.v2/dehydrated_device" + "/org.matrix.msc2697.v2/dehydrated_device$" if msc2697 else "/org.matrix.msc3814.v1/dehydrated_device$", releases=(), From 6d0ce6f9fd9e6129a74f99f3781ae45fb4c46540 Mon Sep 17 00:00:00 2001 From: Nicolas Werner <89468146+nico-famedly@users.noreply.github.com> Date: Tue, 6 Sep 2022 10:59:07 +0200 Subject: [PATCH 5/5] Update synapse/handlers/devicemessage.py Co-authored-by: Hubert Chathi --- synapse/handlers/devicemessage.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/synapse/handlers/devicemessage.py b/synapse/handlers/devicemessage.py index 4eecaba926f7..7419e7e62a07 100644 --- a/synapse/handlers/devicemessage.py +++ b/synapse/handlers/devicemessage.py @@ -317,7 +317,7 @@ async def get_events_for_dehydrated_device( raise SynapseError( HTTPStatus.FORBIDDEN, "Can only fetch messages for own device or dehydrated devices", - Codes.UNAUTHORIZED, + Codes.FORBIDDEN, ) since_stream_id = 0