Add experimental support for delayed events management scopes

changelog.d/19974.feature

@@ -0,0 +1 @@
+Add experimental support for MSC4502: Targeted and unrestricted room member queries.

changelog.d/19975.feature

@@ -0,0 +1 @@
+Add experimental support for delayed events management scopes.

rust/src/config/mod.rs

@@ -73,4 +73,5 @@ pub struct ExperimentalConfig {
     pub msc4491_enabled: bool,
     pub msc4143_enabled: bool,
     pub msc4446_enabled: bool,
+    pub msc4502_enabled: bool,
 }

rust/src/handlers/versions.rs

@@ -269,6 +269,9 @@ pub struct UnstableFeatureMap {
     /// MSC4446: Allow moving the fully read marker backwards.
     #[serde(rename = "com.beeper.msc4446")]
     msc4446_enabled: bool,
+    /// MSC4502: Targeted and unrestricted room member queries
+    #[serde(rename = "io.element.msc4502")]
+    msc4502: bool,
 
     // Whether new rooms will be set to encrypted or not (based on presets).
     #[serde(rename = "io.element.e2ee_forced.public")]
@@ -320,6 +323,7 @@ pub fn synapse_config_to_global_unstable_feature_map(
         msc4491_enabled: config.experimental.msc4491_enabled,
         msc4143_enabled: config.experimental.msc4143_enabled,
         msc4446_enabled: config.experimental.msc4446_enabled,
+        msc4502: config.experimental.msc4502_enabled,
         e2ee_forced_public: config
             .room
             .encryption_enabled_by_default_for_room_presets

synapse/appservice/__init__.py

@@ -62,6 +62,21 @@
 #   user ID -> {device ID -> [algorithm]}
 TransactionUnusedFallbackKeys = dict[str, dict[str, list[str]]]
 
+# Scopes assignable to application services for extended privileges.
+SCOPE_QUERY_ROOM_MEMBERSHIP = "urn:matrix:client:io.element.msc4502:rooms:is_joined"
+SCOPE_RESTART_DELAYED_EVENT = (
+    "urn:matrix:client:io.element.mscXXXX:delayed_events:restart"
+)
+SCOPE_SEND_DELAYED_EVENT = "urn:matrix:client:io.element.mscXXXX:delayed_events:send"
+
+KNOWN_SCOPES = frozenset(
+    {
+        SCOPE_QUERY_ROOM_MEMBERSHIP,
+        SCOPE_RESTART_DELAYED_EVENT,
+        SCOPE_SEND_DELAYED_EVENT,
+    }
+)
+
 
 class ApplicationServiceState(Enum):
     DOWN = "down"
@@ -104,6 +119,7 @@ def __init__(
         supports_unstable_ephemeral: bool = False,
         msc3202_transaction_extensions: bool = False,
         msc4190_device_management: bool = False,
+        scopes: Iterable[str] | None = None,
     ):
         self.token = token
         self.url = (
@@ -140,6 +156,11 @@ def __init__(
         else:
             self.protocols = set()
 
+        self.scopes = set(scopes) if scopes else set()
+        unknown_scopes = self.scopes - KNOWN_SCOPES
+        if unknown_scopes:
+            raise ValueError(f"Unknown application service scope(s): {unknown_scopes}")
+
         self.rate_limited = rate_limited
 
     def _check_namespaces(
@@ -379,6 +400,9 @@ def is_exclusive_user(self, user_id: str) -> bool:
     def is_interested_in_protocol(self, protocol: str) -> bool:
         return protocol in self.protocols
 
+    def has_scope(self, scope: str) -> bool:
+        return scope in self.scopes
+
     def is_exclusive_alias(self, alias: str) -> bool:
         return self._is_exclusive(ApplicationService.NS_ALIASES, alias)
 

synapse/config/appservice.py

@@ -199,6 +199,14 @@ def _load_appservice(
             "The `io.element.msc4190` option should be true or false if specified."
         )
 
+    # Opt-in list of scopes granted to this appservice for restricted C-S API
+    # functionality.
+    scopes = as_info.get("io.element.msc4502.scopes", [])
+    if not isinstance(scopes, list) or not all(isinstance(s, str) for s in scopes):
+        raise ValueError(
+            "The `io.element.msc4502.scopes` option should be a list of strings if specified."
+        )
+
     return ApplicationService(
         token=as_info["as_token"],
         url=as_info["url"],
@@ -213,4 +221,5 @@ def _load_appservice(
         supports_ephemeral=supports_ephemeral,
         msc3202_transaction_extensions=msc3202_transaction_extensions,
         msc4190_device_management=msc4190_enabled,
+        scopes=scopes,
     )

synapse/config/experimental.py

@@ -203,6 +203,12 @@ def read_config(
         # See https://github.com/element-hq/synapse/issues/19524
         self.msc4370_enabled = experimental.get("msc4370_enabled", False)
 
+        # MSC4502: Targeted and unrestricted room member queries
+        self.msc4502_enabled: bool = experimental.get("msc4502_enabled", False)
+
+        # MSCXXXX: ...
+        self.mscXXXX_enabled: bool = experimental.get("mscXXXX_enabled", False)
+
         auth_delegated = (config.get("matrix_authentication_service") or {}).get(
             "enabled", False
         )

synapse/handlers/delayed_events.py

@@ -19,8 +19,15 @@
 from twisted.internet.interfaces import IDelayedCall
 
 from synapse.api.constants import EventTypes, StickyEvent, StickyEventField
-from synapse.api.errors import Codes, ShadowBanError, SynapseError
+from synapse.api.errors import (
+    AuthError,
+    Codes,
+    NotFoundError,
+    ShadowBanError,
+    SynapseError,
+)
 from synapse.api.ratelimiting import Ratelimiter
+from synapse.appservice import SCOPE_RESTART_DELAYED_EVENT, SCOPE_SEND_DELAYED_EVENT
 from synapse.config.workers import MAIN_PROCESS_INSTANCE_NAME
 from synapse.http.site import SynapseRequest
 from synapse.logging.context import make_deferred_yieldable
@@ -430,7 +437,7 @@ async def cancel(self, request: SynapseRequest, delay_id: str) -> None:
             NotFoundError: if no matching delayed event could be found.
         """
         assert self._is_master
-        await self._mgmt_ratelimit(request)
+        await self._authorise_and_ratelimit(request, delay_id, bypass_scope=None)
         await make_deferred_yieldable(self._initialized_from_db)
 
         next_send_ts = await self._store.cancel_delayed_event(delay_id)
@@ -445,7 +452,9 @@ async def restart(self, request: SynapseRequest, delay_id: str) -> None:
         Raises:
             NotFoundError: if no matching delayed event could be found.
         """
-        await self._mgmt_ratelimit(request)
+        await self._authorise_and_ratelimit(
+            request, delay_id, bypass_scope=SCOPE_RESTART_DELAYED_EVENT
+        )
 
         # Note: We don't need to wait on `self._initialized_from_db` here as the
         # events that deals with are already marked as processed.
@@ -469,7 +478,9 @@ async def send(self, request: SynapseRequest, delay_id: str) -> None:
             NotFoundError: if no matching delayed event could be found.
         """
         assert self._is_master
-        await self._mgmt_ratelimit(request)
+        await self._authorise_and_ratelimit(
+            request, delay_id, bypass_scope=SCOPE_SEND_DELAYED_EVENT
+        )
         await make_deferred_yieldable(self._initialized_from_db)
 
         event, next_send_ts = await self._store.process_target_delayed_event(delay_id)
@@ -492,6 +503,53 @@ async def _mgmt_ratelimit(self, request: SynapseRequest) -> None:
             key = request.getClientAddress().host
         await self._delayed_event_mgmt_ratelimiter.ratelimit(requester, key)
 
+    async def _authorise_and_ratelimit(
+        self,
+        request: SynapseRequest,
+        delay_id: str,
+        bypass_scope: Optional[str],
+    ) -> None:
+        """
+        Verify request authorization and perform rate limiting accordingly.
+        """
+        if not self._config.experimental.mscXXXX_enabled:
+            await self._mgmt_ratelimit(request)
+            return
+
+        requester = await self._auth.get_user_by_req(request)
+        await self._delayed_event_mgmt_ratelimiter.ratelimit(requester)
+
+        creator_localpart = await self._store.get_delayed_event_creator(delay_id)
+        if creator_localpart is None:
+            raise NotFoundError("Delayed event not found")
+
+        # The creator of the delayed event is always allowed to manage it.
+        if creator_localpart == requester.user.localpart:
+            return
+
+        # Other users (or appservices) can manage the delayed event if they
+        # have the required scopes.
+        if bypass_scope is not None and self._requester_has_scope(
+            requester, bypass_scope
+        ):
+            return
+
+        # Otherwise access is denied.
+        raise AuthError(
+            HTTPStatus.FORBIDDEN,
+            "You do not have permission to act on this delayed event",
+        )
+
+    def _requester_has_scope(self, requester: Requester, scope: str) -> bool:
+        app_service = (
+            self._store.get_app_service_by_id(requester.app_service_id)
+            if requester.app_service_id
+            else None
+        )
+        return (app_service is not None and app_service.has_scope(scope)) or (
+            scope in requester.scope
+        )
+
     async def _send_on_timeout(self) -> None:
         self._next_delayed_event_call = None
 

synapse/rest/__init__.py

@@ -60,6 +60,7 @@
     retention,
     room,
     room_keys,
+    room_membership,
     room_upgrade_rest_servlet,
     sendtodevice,
     sync,
@@ -128,6 +129,7 @@
     rendezvous.register_servlets,
     auth_metadata.register_servlets,
     thread_subscriptions.register_servlets,
+    room_membership.register_servlets,
 )
 
 SERVLET_GROUPS: dict[str, Iterable[RegisterServletsFunc]] = {

synapse/rest/client/room_membership.py

@@ -0,0 +1,133 @@
+#
+# This file is licensed under the Affero General Public License (AGPL) version 3.
+#
+# Copyright (C) 2026 Element Creations 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
+import re
+from http import HTTPStatus
+from typing import TYPE_CHECKING
+
+from synapse.api.constants import EventTypes, Membership
+from synapse.api.errors import Codes, SynapseError
+from synapse.appservice import SCOPE_QUERY_ROOM_MEMBERSHIP
+from synapse.http.server import HttpServer
+from synapse.http.servlet import RestServlet, parse_string
+from synapse.http.site import SynapseRequest
+from synapse.types import JsonDict, RoomID, UserID
+from synapse.util.stringutils import parse_and_validate_server_name
+
+if TYPE_CHECKING:
+    from synapse.server import HomeServer
+
+logger = logging.getLogger(__name__)
+
+
+class AppserviceRoomMembershipRestServlet(RestServlet):
+    PATTERNS = [
+        re.compile(
+            r"^/_matrix/client/unstable/io\.element\.msc4502/rooms/(?P<room_id>[^/]*)/is_joined$"
+        )
+    ]
+    CATEGORY = "Client API requests"
+
+    def __init__(self, hs: "HomeServer"):
+        super().__init__()
+        self.auth = hs.get_auth()
+        self.store = hs.get_datastores().main
+        self.storage_controllers = hs.get_storage_controllers()
+        self.is_mine_id = hs.is_mine_id
+        self.is_mine_server_name = hs.is_mine_server_name
+
+    async def on_GET(
+        self, request: SynapseRequest, room_id: str
+    ) -> tuple[int, JsonDict]:
+        requester = await self.auth.get_user_by_req(request, allow_guest=False)
+
+        app_service = (
+            self.store.get_app_service_by_id(requester.app_service_id)
+            if requester.app_service_id
+            else None
+        )
+
+        # Users and appservices can call this endpoint if they have the scope.
+        has_scope = (
+            app_service is not None
+            and app_service.has_scope(SCOPE_QUERY_ROOM_MEMBERSHIP)
+        ) or SCOPE_QUERY_ROOM_MEMBERSHIP in requester.scope
+
+        if not has_scope:
+            raise SynapseError(
+                HTTPStatus.FORBIDDEN,
+                f"Missing {SCOPE_QUERY_ROOM_MEMBERSHIP} scope",
+                Codes.FORBIDDEN,
+            )
+
+        if not RoomID.is_valid(room_id):
+            raise SynapseError(
+                HTTPStatus.BAD_REQUEST, "Invalid room ID", Codes.INVALID_PARAM
+            )
+
+        mxid = parse_string(request, "mxid")
+        server_name = parse_string(request, "server_name")
+
+        if (mxid is None) == (server_name is None):
+            raise SynapseError(
+                HTTPStatus.BAD_REQUEST,
+                "Exactly one of 'mxid' or 'server_name' query parameters must be given",
+                Codes.MISSING_PARAM,
+            )
+
+        if mxid is not None:
+            if not UserID.is_valid(mxid):
+                raise SynapseError(
+                    HTTPStatus.BAD_REQUEST,
+                    "Invalid MXID: %s" % (mxid,),
+                    Codes.INVALID_PARAM,
+                )
+            if self.is_mine_id(mxid):
+                (
+                    membership,
+                    _,
+                ) = await self.store.get_local_current_membership_for_user_in_room(
+                    mxid, room_id
+                )
+                joined = membership == Membership.JOIN
+            else:
+                event = await self.storage_controllers.state.get_current_state_event(
+                    room_id, EventTypes.Member, mxid
+                )
+                joined = (
+                    event is not None
+                    and event.content.get("membership") == Membership.JOIN
+                )
+        else:
+            assert server_name is not None
+            try:
+                parse_and_validate_server_name(server_name)
+            except ValueError:
+                raise SynapseError(
+                    HTTPStatus.BAD_REQUEST,
+                    "Invalid server name: %s" % (server_name,),
+                    Codes.INVALID_PARAM,
+                )
+            if self.is_mine_server_name(server_name):
+                joined = await self.store.is_locally_joined(room_id)
+            else:
+                joined = await self.store.is_host_joined(room_id, server_name)
+
+        return HTTPStatus.OK, {"joined": joined}
+
+
+def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None:
+    if hs.config.experimental.msc4502_enabled:
+        AppserviceRoomMembershipRestServlet(hs).register(http_server)

synapse/storage/databases/main/delayed_events.py

@@ -229,6 +229,21 @@ def add_delayed_event_txn(txn: LoggingTransaction) -> Timestamp:
 
         return delay_id, next_send_ts
 
+    async def get_delayed_event_creator(self, delay_id: str) -> str | None:
+        """
+        Retrieves the localpart of the user who created the matching delayed event.
+
+        Returns: The localpart of the creating user, or None if there is no matching
+            delayed event.
+        """
+        return await self.db_pool.simple_select_one_onecol(
+            table="delayed_events",
+            keyvalues={"delay_id": delay_id},
+            retcol="user_localpart",
+            allow_none=True,
+            desc="get_delayed_event_owner",
+        )
+
     async def restart_delayed_event(
         self,
         delay_id: str,