Dedicated API for MAS -> Synapse communication

Author: sandhose

synapse/_pydantic_compat.py

@@ -48,6 +48,7 @@
         conint,
         constr,
         parse_obj_as,
+        root_validator,
         validator,
     )
     from pydantic.v1.error_wrappers import ErrorWrapper
@@ -68,6 +69,7 @@
         conint,
         constr,
         parse_obj_as,
+        root_validator,
         validator,
     )
     from pydantic.error_wrappers import ErrorWrapper
@@ -92,4 +94,5 @@
     "StrictStr",
     "ValidationError",
     "validator",
+    "root_validator",
 )

synapse/api/auth/msc3861_delegated.py

@@ -369,6 +369,12 @@ async def _introspect_token(
     async def is_server_admin(self, requester: Requester) -> bool:
         return "urn:synapse:admin:*" in requester.scope
 
+    def _is_access_token_the_admin_token(self, token: str) -> bool:
+        admin_token = self._admin_token()
+        if admin_token is None:
+            return False
+        return token == admin_token
+
     async def get_user_by_req(
         self,
         request: SynapseRequest,
@@ -434,7 +440,7 @@ async def _wrapped_get_user_by_req(
             requester = await self.get_user_by_access_token(access_token, allow_expired)
 
         # Do not record requests from MAS using the virtual `__oidc_admin` user.
-        if access_token != self._admin_token():
+        if not self._is_access_token_the_admin_token(access_token):
             await self._record_request(request, requester)
 
         if not allow_guest and requester.is_guest:
@@ -470,13 +476,25 @@ async def get_user_by_req_experimental_feature(
 
             raise UnrecognizedRequestError(code=404)
 
+    def is_request_using_the_admin_token(self, request: SynapseRequest) -> bool:
+        """
+        Check if the request is using the admin token.
+
+        Args:
+            request: The request to check.
+
+        Returns:
+            True if the request is using the admin token, False otherwise.
+        """
+        access_token = self.get_access_token_from_request(request)
+        return self._is_access_token_the_admin_token(access_token)
+
     async def get_user_by_access_token(
         self,
         token: str,
         allow_expired: bool = False,
     ) -> Requester:
-        admin_token = self._admin_token()
-        if admin_token is not None and token == admin_token:
+        if self._is_access_token_the_admin_token(token):
             # XXX: This is a temporary solution so that the admin API can be called by
             # the OIDC provider. This will be removed once we have OIDC client
             # credentials grant support in matrix-authentication-service.

synapse/rest/synapse/client/__init__.py

@@ -30,6 +30,7 @@
 from synapse.rest.synapse.client.rendezvous import MSC4108RendezvousSessionResource
 from synapse.rest.synapse.client.sso_register import SsoRegisterResource
 from synapse.rest.synapse.client.unsubscribe import UnsubscribeResource
+from synapse.rest.synapse.mas import MasResource
 
 if TYPE_CHECKING:
     from synapse.server import HomeServer
@@ -60,6 +61,7 @@ def build_synapse_client_resource_tree(hs: "HomeServer") -> Mapping[str, Resourc
         from synapse.rest.synapse.client.jwks import JwksResource
 
         resources["/_synapse/jwks"] = JwksResource(hs)
+        resources["/_synapse/mas"] = MasResource(hs)
 
     # provider-specific SSO bits. Only load these if they are enabled, since they
     # rely on optional dependencies.

synapse/rest/synapse/mas/__init__.py

@@ -0,0 +1,62 @@
+#
+# This file is licensed under the Affero General Public License (AGPL) version 3.
+#
+# Copyright (C) 2025 New Vector, Ltd
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# See the GNU Affero General Public License for more details:
+# <https://www.gnu.org/licenses/agpl_3.0.html>.
+#
+#
+
+
+import logging
+from typing import TYPE_CHECKING
+
+from twisted.web.resource import Resource
+
+from synapse.rest.synapse.mas.devices import (
+    MasCreateDeviceResource,
+    MasSyncDevicesResource,
+    MasUpdateDeviceDisplayNameResource,
+)
+from synapse.rest.synapse.mas.users import (
+    MasAllowCrossSigningResetResource,
+    MasDeleteUserResource,
+    MasIsLocalpartAvailableResource,
+    MasProvisionUserResource,
+    MasQueryUserResource,
+    MasReactivateUserResource,
+    MasSetDisplayNameResource,
+    MasUnsetDisplayNameResource,
+)
+
+if TYPE_CHECKING:
+    from synapse.server import HomeServer
+
+
+logger = logging.getLogger(__name__)
+
+
+class MasResource(Resource):
+    def __init__(self, hs: "HomeServer"):
+        Resource.__init__(self)
+        self.putChild(b"query_user", MasQueryUserResource(hs))
+        self.putChild(b"provision_user", MasProvisionUserResource(hs))
+        self.putChild(b"is_localpart_available", MasIsLocalpartAvailableResource(hs))
+        self.putChild(b"delete_user", MasDeleteUserResource(hs))
+        self.putChild(b"create_device", MasCreateDeviceResource(hs))
+        self.putChild(
+            b"update_device_display_name", MasUpdateDeviceDisplayNameResource(hs)
+        )
+        self.putChild(b"sync_devices", MasSyncDevicesResource(hs))
+        self.putChild(b"reactivate_user", MasReactivateUserResource(hs))
+        self.putChild(b"set_displayname", MasSetDisplayNameResource(hs))
+        self.putChild(b"unset_displayname", MasUnsetDisplayNameResource(hs))
+        self.putChild(
+            b"allow_cross_signing_reset", MasAllowCrossSigningResetResource(hs)
+        )

synapse/rest/synapse/mas/_base.py

@@ -0,0 +1,40 @@
+#
+# This file is licensed under the Affero General Public License (AGPL) version 3.
+#
+# Copyright (C) 2025 New Vector, Ltd
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# See the GNU Affero General Public License for more details:
+# <https://www.gnu.org/licenses/agpl_3.0.html>.
+#
+#
+
+
+from typing import TYPE_CHECKING, cast
+
+from synapse.api.auth.msc3861_delegated import MSC3861DelegatedAuth
+from synapse.api.errors import SynapseError
+from synapse.http.server import DirectServeJsonResource
+
+if TYPE_CHECKING:
+    from synapse.app.generic_worker import GenericWorkerStore
+    from synapse.http.site import SynapseRequest
+    from synapse.server import HomeServer
+
+
+class MasBaseResource(DirectServeJsonResource):
+    def __init__(self, hs: "HomeServer"):
+        DirectServeJsonResource.__init__(self, extract_context=True)
+        auth = hs.get_auth()
+        assert isinstance(auth, MSC3861DelegatedAuth)
+        self.msc3861_auth = auth
+        self.store = cast("GenericWorkerStore", hs.get_datastores().main)
+        self.hostname = hs.hostname
+
+    def assert_mas_request(self, request: "SynapseRequest") -> None:
+        if not self.msc3861_auth.is_request_using_the_admin_token(request):
+            raise SynapseError(403, "This endpoint must only be called by MAS")

synapse/rest/synapse/mas/devices.py

@@ -0,0 +1,150 @@
+#
+# This file is licensed under the Affero General Public License (AGPL) version 3.
+#
+# Copyright (C) 2025 New Vector, Ltd
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# See the GNU Affero General Public License for more details:
+# <https://www.gnu.org/licenses/agpl_3.0.html>.
+#
+#
+
+import logging
+from http import HTTPStatus
+from typing import TYPE_CHECKING, Optional, Tuple
+
+from synapse._pydantic_compat import BaseModel, StrictStr
+from synapse.http.servlet import parse_and_validate_json_object_from_request
+from synapse.types import JsonDict, UserID
+
+if TYPE_CHECKING:
+    from synapse.http.site import SynapseRequest
+    from synapse.server import HomeServer
+
+
+from ._base import MasBaseResource
+
+logger = logging.getLogger(__name__)
+
+
+class MasCreateDeviceResource(MasBaseResource):
+    def __init__(self, hs: "HomeServer"):
+        MasBaseResource.__init__(self, hs)
+
+        self.device_handler = hs.get_device_handler()
+
+    class PostBody(BaseModel):
+        localpart: StrictStr
+        device_id: StrictStr
+        display_name: Optional[StrictStr]
+
+    async def _async_render_POST(
+        self, request: "SynapseRequest"
+    ) -> Tuple[int, JsonDict]:
+        self.assert_mas_request(request)
+
+        body = parse_and_validate_json_object_from_request(request, self.PostBody)
+        user_id = UserID(body.localpart, self.hostname)
+
+        inserted = await self.device_handler.upsert_device(
+            user_id=str(user_id),
+            device_id=body.device_id,
+            display_name=body.display_name,
+        )
+
+        return HTTPStatus.CREATED if inserted else HTTPStatus.OK, {}
+
+
+class MasUpdateDeviceDisplayNameResource(MasBaseResource):
+    def __init__(self, hs: "HomeServer"):
+        MasBaseResource.__init__(self, hs)
+
+        self.device_handler = hs.get_device_handler()
+
+    class PostBody(BaseModel):
+        localpart: StrictStr
+        device_id: StrictStr
+        display_name: StrictStr
+
+    async def _async_render_POST(
+        self, request: "SynapseRequest"
+    ) -> Tuple[int, JsonDict]:
+        self.assert_mas_request(request)
+
+        body = parse_and_validate_json_object_from_request(request, self.PostBody)
+        user_id = UserID(body.localpart, self.hostname)
+
+        await self.device_handler.update_device(
+            user_id=str(user_id),
+            device_id=body.device_id,
+            content={"display_name": body.display_name},
+        )
+
+        return HTTPStatus.OK, {}
+
+
+class MasSyncDevicesResource(MasBaseResource):
+    def __init__(self, hs: "HomeServer"):
+        MasBaseResource.__init__(self, hs)
+
+        self.device_handler = hs.get_device_handler()
+
+    class PostBody(BaseModel):
+        localpart: StrictStr
+        devices: set[StrictStr]
+
+    async def _async_render_GET(
+        self, request: "SynapseRequest"
+    ) -> Tuple[int, JsonDict]:
+        self.assert_mas_request(request)
+
+        body = parse_and_validate_json_object_from_request(request, self.PostBody)
+        user_id = UserID(body.localpart, self.hostname)
+
+        current_devices = await self.store.get_devices_by_user(user_id=str(user_id))
+        current_devices_list = set(current_devices.keys())
+        target_device_list = set(body.devices)
+
+        to_add = target_device_list - current_devices_list
+        to_delete = current_devices_list - target_device_list
+
+        # Log what we're about to do, as this can be an expensive operation
+        if to_add and to_delete:
+            logger.info(
+                "Syncing %d devices for user %s will add %d devices and delete %d devices",
+                len(target_device_list),
+                user_id,
+                len(to_add),
+                len(to_delete),
+            )
+        elif to_add:
+            logger.info(
+                "Syncing %d devices for user %s will add %d devices",
+                len(target_device_list),
+                user_id,
+                len(to_add),
+            )
+        elif to_delete:
+            logger.info(
+                "Syncing %d devices for user %s will delete %d devices",
+                len(target_device_list),
+                user_id,
+                len(to_delete),
+            )
+
+        if to_delete:
+            await self.device_handler.delete_devices(
+                user_id=str(user_id), device_ids=to_delete
+            )
+
+        for device_id in to_add:
+            await self.device_handler.upsert_device(
+                user_id=str(user_id),
+                device_id=device_id,
+            )
+
+        return 200, {}