Add experimental support for sending federation requests from app services

changelog.d/19972.feature

@@ -0,0 +1 @@
+Add experimental support for letting application services proxy namespaces in the C-S and S-S API.

changelog.d/19977.feature

@@ -0,0 +1 @@
+Add experimental support for sending federation requests from app services.

docs/application_services.md

@@ -34,3 +34,30 @@ namespaces:
 `group_id`: All users of this application service are dynamically joined to this group. This is useful for e.g user organisation or flairs.
 
 See the [spec](https://matrix.org/docs/spec/application_service/unstable.html) for further details on how application services work.
+
+## `io.element.proxy` (experimental)
+
+> **Warning**: This is an experimental configuration option and may change or be removed without notice.
+
+An application service registration may set `io.element.proxy` to request Synapse to reverse-proxy Client-Server
+and Server-Server requests under the specified prefix to the application service. Setting this option also requires
+`url` to be set and the prefix must be unique across all registered application services.
+
+```yaml
+io.element.proxy: <prefix>
+url: <url>
+```
+
+When set, Synapse will reverse-proxy the following routes:
+
+- `/_matrix/client/{prefix}/*` -> `{url}/_matrix/client/{prefix}/*`
+
+  Only authenticated requests can be proxied. The original `Authorization` header is omitted from the forwarded request.
+  Instead, the resolved MXID is passed to the application service in the `X-Matrix-User-Identifier` header.
+
+- `/_matrix/federation/{prefix}/*` -> `{url}/_matrix/federation/<prefix>/*`
+
+  Only signed requests can be proxied. The original `Authorization` header is omitted from the forwarded request.
+  Instead, the verified origin servername is passed to the application service in the `X-Matrix-Origin` header.
+
+For now, the only allowed proxy prefix is `rtc/livekit`.

synapse/api/errors.py

@@ -125,6 +125,11 @@ class Codes(str, Enum):
     AS_PING_CONNECTION_TIMEOUT = "M_CONNECTION_TIMEOUT"
     AS_PING_CONNECTION_FAILED = "M_CONNECTION_FAILED"
 
+    AS_FEDPROXY_NO_PROXY_PREFIX = "IO.ELEMENT.FEDPROXY_NO_PROXY_PREFIX"
+    AS_FEDPROXY_PATH_NOT_ALLOWED = "IO.ELEMENT.FEDPROXY_PATH_NOT_ALLOWED"
+    AS_FEDPROXY_CONNECTION_FAILED = "IO.ELEMENT.FEDPROXY_CONNECTION_FAILED"
+    AS_FEDPROXY_DESTINATION_DENIED = "IO.ELEMENT.FEDPROXY_DESTINATION_DENIED"
+
     # Attempt to send a second annotation with the same event type & annotation key
     # MSC2677
     DUPLICATE_ANNOTATION = "M_DUPLICATE_ANNOTATION"

synapse/appservice/__init__.py

@@ -89,6 +89,8 @@ class ApplicationService:
     # values.
     NS_LIST = [NS_USERS, NS_ALIASES, NS_ROOMS]
 
+    ALLOWED_PROXY_PREFIXES = {"rtc/livekit"}
+
     def __init__(
         self,
         token: str,
@@ -104,6 +106,7 @@ def __init__(
         supports_unstable_ephemeral: bool = False,
         msc3202_transaction_extensions: bool = False,
         msc4190_device_management: bool = False,
+        proxy_prefix: str | None = None,
     ):
         self.token = token
         self.url = (
@@ -130,10 +133,17 @@ def __init__(
         self.supports_ephemeral = supports_ephemeral
         self.msc3202_transaction_extensions = msc3202_transaction_extensions
         self.msc4190_device_management = msc4190_device_management
+        self.proxy_prefix = proxy_prefix
 
         if "|" in self.id:
             raise Exception("application service ID cannot contain '|' character")
 
+        if proxy_prefix is not None:
+            if not self._is_proxy_prefix_allowed(proxy_prefix):
+                raise ValueError(f"cannot claim reserved proxy prefix {proxy_prefix}")
+            if not self.url:
+                raise KeyError("cannot claim proxy prefix without also setting a url")
+
         # .protocols is a publicly visible field
         if protocols:
             self.protocols = set(protocols)
@@ -192,6 +202,12 @@ def _is_exclusive(self, namespace_key: str, test_string: str) -> bool:
             return namespace.exclusive
         return False
 
+    def _is_proxy_prefix_allowed(self, prefix: str) -> bool:
+        return any(
+            prefix == allowed or prefix.startswith(allowed + "/")
+            for allowed in ApplicationService.ALLOWED_PROXY_PREFIXES
+        )
+
     @cached(num_args=1, cache_context=True)
     async def _matches_user_in_member_list(
         self,

synapse/config/appservice.py

@@ -68,6 +68,7 @@ def load_appservices(
     # Dicts of value -> filename
     seen_as_tokens: dict[str, str] = {}
     seen_ids: dict[str, str] = {}
+    seen_proxy_prefixes: dict[str, str] = {}
 
     appservices = []
 
@@ -93,6 +94,18 @@ def load_appservices(
                         )
                     )
                 seen_as_tokens[appservice.token] = config_file
+                if appservice.proxy_prefix is not None:
+                    if appservice.proxy_prefix in seen_proxy_prefixes:
+                        raise ConfigError(
+                            "Cannot reuse io.element.proxy across application services: "
+                            "%s (files: %s, %s)"
+                            % (
+                                appservice.proxy_prefix,
+                                config_file,
+                                seen_proxy_prefixes[appservice.proxy_prefix],
+                            )
+                        )
+                    seen_proxy_prefixes[appservice.proxy_prefix] = config_file
                 logger.info("Loaded application service: %s", appservice)
                 appservices.append(appservice)
         except Exception as e:
@@ -199,6 +212,17 @@ def _load_appservice(
             "The `io.element.msc4190` option should be true or false if specified."
         )
 
+    # Opt-in setting to enable proxying C-S and S-S API endpoints.
+    # When set, Synapse will reverse-proxy requests under the prefix to the appservice:
+    # - /_matrix/client/{prefix}/* -> {url}/_matrix/client/{prefix}/*
+    # - /_matrix/federation/{prefix}/* -> {url}/_matrix/federation/<prefix>/*
+    proxy_prefix = as_info.get("io.element.proxy")
+    if proxy_prefix is not None:
+        if not isinstance(proxy_prefix, str) or not proxy_prefix:
+            raise ValueError(
+                "The `io.element.proxy` option should be a non-empty string."
+            )
+
     return ApplicationService(
         token=as_info["as_token"],
         url=as_info["url"],
@@ -213,4 +237,5 @@ def _load_appservice(
         supports_ephemeral=supports_ephemeral,
         msc3202_transaction_extensions=msc3202_transaction_extensions,
         msc4190_device_management=msc4190_enabled,
+        proxy_prefix=proxy_prefix,
     )

synapse/federation/transport/server/__init__.py

@@ -24,6 +24,7 @@
 from typing import TYPE_CHECKING, Iterable, Literal
 
 from synapse.api.errors import FederationDeniedError, SynapseError
+from synapse.federation.transport.server import appservice_proxy
 from synapse.federation.transport.server._base import (
     Authenticator,
     BaseFederationServlet,
@@ -340,3 +341,6 @@ def register_servlets(
                 ratelimiter=ratelimiter,
                 server_name=hs.hostname,
             ).register(resource)
+
+    if "federation" in servlet_groups:
+        appservice_proxy.register_servlets(hs, resource, authenticator, ratelimiter)

synapse/federation/transport/server/appservice_proxy.py

@@ -0,0 +1,103 @@
+#
+# 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 io import BytesIO
+from typing import TYPE_CHECKING
+
+from synapse.api.errors import Codes, SynapseError
+from synapse.appservice import ApplicationService
+from synapse.federation.transport.server._base import Authenticator
+from synapse.http import QuieterFileBodyProducer
+from synapse.http.appservice_proxy import proxy_request_to_appservice
+from synapse.http.server import HttpServer, ServletCallback
+from synapse.http.site import SynapseRequest
+from synapse.util.json import json_decoder
+from synapse.util.ratelimitutils import FederationRateLimiter
+
+if TYPE_CHECKING:
+    from synapse.server import HomeServer
+
+logger = logging.getLogger(__name__)
+
+
+def _make_proxy_callback(
+    hs: "HomeServer",
+    authenticator: Authenticator,
+    ratelimiter: FederationRateLimiter,
+    appservice: ApplicationService,
+) -> ServletCallback:
+    async def _proxy(request: SynapseRequest, **kwargs: str) -> None:
+        raw_body = request.content.read()  # type: ignore[union-attr]
+
+        content = None
+        if request.method in (b"PUT", b"POST"):
+            try:
+                content = json_decoder.decode(raw_body.decode("utf-8"))
+            except Exception:
+                raise SynapseError(
+                    HTTPStatus.BAD_REQUEST, "Content not JSON.", Codes.NOT_JSON
+                )
+
+        origin = await authenticator.authenticate_request(request, content)
+
+        # Apply the same per-origin rate limiting that every other federation endpoint gets.
+        with ratelimiter.ratelimit(origin) as d:
+            await d
+            if request._disconnected:
+                logger.warning(
+                    "client disconnected before we started processing request"
+                )
+                return
+
+            await proxy_request_to_appservice(
+                request,
+                hs,
+                appservice,
+                QuieterFileBodyProducer(BytesIO(raw_body)),
+                extra_request_headers={b"X-Matrix-Origin": origin.encode("ascii")},
+            )
+
+    return _proxy
+
+
+def register_servlets(
+    hs: "HomeServer",
+    resource: HttpServer,
+    authenticator: Authenticator,
+    ratelimiter: FederationRateLimiter,
+) -> None:
+    """Registers blanket reverse-proxy routes for each application service that has
+    configured a proxy prefix. This forwards requests under /_matrix/federation/<prefix>/*
+    to the same path under the application service's URL after verifying request
+    authentication.
+    """
+    for appservice in hs.get_datastores().main.get_app_services():
+        if appservice.proxy_prefix is None:
+            continue
+
+        pattern = re.compile(
+            "^/_matrix/federation/%s(/.*)?$" % (re.escape(appservice.proxy_prefix),)
+        )
+        callback = _make_proxy_callback(hs, authenticator, ratelimiter, appservice)
+
+        for method in ("GET", "POST", "PUT", "DELETE"):
+            resource.register_paths(
+                method,
+                (pattern,),
+                callback,
+                "ApplicationServiceFederationProxy",
+            )