dandri/synapse
1
0
Fork 0
mirror of https://github.com/element-hq/synapse.git synced 2026-09-17 03:44:29 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add experimental support for sending federation requests from app services as per MSC4512 (#19977)

Browse Source 
This implements the outgoing part of
[MSC4512](https://github.com/matrix-org/matrix-spec-proposals/pull/4512)
and is another stopgap towards
https://github.com/element-hq/voip-internal/issues/641.

This adds the ability for proxying app services
(https://github.com/element-hq/synapse/pull/19972) to trigger federation
requests under their own proxy prefix.

This depends on https://github.com/element-hq/synapse/pull/19972 and
cannot make progress before it lands.
This commit is contained in:
Johannes Marbach
2026-09-08 13:29:03 +01:00
committed by GitHub
parent 07bfe4c2a1
commit f8f7738f5d
10 changed files with 934 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
changelog.d/19977.feature
+1
View File
@@ -0,0 +1 @@
Add experimental support for sending federation requests from app services as per MSC4512.
synapse/api/errors.py
+5
View File
@@ -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.MSC4512.M_FEDPROXY_NO_PROXY_PREFIX"
AS_FEDPROXY_PATH_NOT_ALLOWED = "IO.ELEMENT.MSC4512.M_FEDPROXY_PATH_NOT_ALLOWED"
AS_FEDPROXY_CONNECTION_FAILED = "IO.ELEMENT.MSC4512.M_FEDPROXY_CONNECTION_FAILED"
AS_FEDPROXY_DESTINATION_DENIED = "IO.ELEMENT.MSC4512.M_FEDPROXY_DESTINATION_DENIED"
# Attempt to send a second annotation with the same event type & annotation key
# MSC2677
DUPLICATE_ANNOTATION = "M_DUPLICATE_ANNOTATION"
synapse/federation/transport/server/appservice_proxy.py
+3 -3
View File
@@ -89,9 +89,9 @@ def register_servlets(
) -> None:
"""Registers blanket reverse-proxy routes for each application service that has
configured a proxy prefix. This forwards requests under
/_matrix/federation/<version>/<prefix>/* (where <version> is either "vN" or "unstable")
to the same path under the application service's proxy URL after verifying request
authentication.
/_matrix/federation/<version>/<prefix>/* (where <version> is either "vN" or
"unstable/<namespace>") to the same path under the application service's proxy
URL after verifying request authentication.
"""
if not hs.config.experimental.msc4512_enabled:
return
synapse/http/appservice_proxy.py
+107 -1
View File
@@ -13,6 +13,7 @@
#
import logging
import re
from http import HTTPStatus
from typing import TYPE_CHECKING, Optional, cast
from urllib.parse import parse_qs, unquote_to_bytes, urlencode, urlsplit
@@ -21,7 +22,13 @@ from twisted.python import failure
from twisted.web.http_headers import Headers
from twisted.web.iweb import IBodyProducer, IResponse
from synapse.api.errors import Codes, SynapseError
from synapse.api.errors import (
Codes,
FederationDeniedError,
HttpResponseException,
RequestSendFailed,
SynapseError,
)
from synapse.appservice import ApplicationService
from synapse.http.proxy import (
HOP_BY_HOP_HEADERS_LOWERCASE,
@@ -30,8 +37,11 @@ from synapse.http.proxy import (
)
from synapse.http.server import return_json_error, set_cors_headers
from synapse.http.site import SynapseRequest
from synapse.http.types import QueryParams
from synapse.logging.context import make_deferred_yieldable, run_in_background
from synapse.types import JsonDict
from synapse.util.async_helpers import timeout_deferred
from synapse.util.json import json_decoder
if TYPE_CHECKING:
from synapse.server import HomeServer
@@ -192,3 +202,99 @@ def _send_response(request: SynapseRequest, response: IResponse) -> None:
request.responseHeaders.setRawHeaders(header_name, header_values)
response.deliverBody(_ProxyResponseBody(request))
async def send_federation_request_from_appservice(
hs: "HomeServer",
appservice: ApplicationService,
method: str,
destination: str,
path: str,
data: JsonDict | None,
args: QueryParams | None,
) -> tuple[int, JsonDict | None]:
"""Sign and send a federation request on behalf of an appservice.
Returns:
A `(status, content)` tuple describing the destination's actual HTTP response.
"""
_check_path_allowed_for_appservice(appservice, path)
if hs.is_mine_server_name(destination):
raise SynapseError(
HTTPStatus.FORBIDDEN,
"Cannot target this homeserver itself",
Codes.AS_FEDPROXY_DESTINATION_DENIED,
)
if method not in ("GET", "PUT", "POST", "DELETE"):
raise SynapseError(
HTTPStatus.BAD_REQUEST,
f"Unsupported method {method}",
Codes.INVALID_PARAM,
)
client = hs.get_federation_http_client()
try:
content = await client.send_direct_request(
method, destination, path, args=args, json_body=data
)
return HTTPStatus.OK, content
except HttpResponseException as e:
try:
content = json_decoder.decode(e.response.decode("utf-8"))
except (UnicodeDecodeError, ValueError):
content = None
return e.code, content
except FederationDeniedError as e:
raise SynapseError(
HTTPStatus.FORBIDDEN,
e.msg,
Codes.AS_FEDPROXY_DESTINATION_DENIED,
)
except RequestSendFailed as e:
raise SynapseError(
HTTPStatus.BAD_GATEWAY,
str(e),
Codes.AS_FEDPROXY_CONNECTION_FAILED,
)
def _check_path_allowed_for_appservice(
appservice: ApplicationService, path: str
) -> None:
# Deny relative paths.
if has_dot_segments(path.encode("utf-8")):
raise SynapseError(
HTTPStatus.FORBIDDEN,
"Path must not contain '.' or '..' segments",
Codes.AS_FEDPROXY_PATH_NOT_ALLOWED,
)
# Deny query parameters or fragments smuggled in via the path.
split_path = urlsplit(path)
if split_path.query or split_path.fragment:
raise SynapseError(
HTTPStatus.FORBIDDEN,
"Path must not contain a query string or fragment",
Codes.AS_FEDPROXY_PATH_NOT_ALLOWED,
)
# Ensure the path is under the appservice's own proxy prefix.
if appservice.proxy_prefix is None:
raise SynapseError(
HTTPStatus.BAD_REQUEST,
"Application service does not have a proxy prefix",
Codes.AS_FEDPROXY_NO_PROXY_PREFIX,
)
pattern = re.compile(
r"^/_matrix/federation/(?:unstable/[^/]+|v[^/]+)/%s(/.*)?$"
% (re.escape(appservice.proxy_prefix),)
)
if pattern.match(path) is None:
raise SynapseError(
HTTPStatus.FORBIDDEN,
f"Path must be under /_matrix/federation/<version>/{appservice.proxy_prefix}",
Codes.AS_FEDPROXY_PATH_NOT_ALLOWED,
)
synapse/http/matrixfederationclient.py
+76
View File
@@ -1468,6 +1468,82 @@ class MatrixFederationHttpClient:
)
return body
async def send_direct_request(
self,
method: str,
destination: str,
path: str,
args: QueryParams | None = None,
json_body: JsonDict | None = None,
timeout: int | None = None,
) -> JsonDict:
"""Sends a single one-off request to the destination, for a caller
that handles backing off itself.
Unlike `get_json`/`put_json`/`post_json`/`delete_json`, this method
always ignores the destination's backoff data, and raises the
destination's real `HttpResponseException` rather than
`RequestSendFailed` for a 5xx or 429 response that survives the
normal short-retry schedule.
Args:
method: The HTTP method to use.
destination: The remote server to send the HTTP request to.
path: The HTTP path.
args: query params
json_body: A dict containing the data that will be used as the
request body. This will be encoded as JSON.
timeout: number of milliseconds to wait for the response.
self._default_timeout (60s) by default.
Returns:
Succeeds when we get a 2xx HTTP response. The
result will be the decoded JSON body.
Raises:
HttpResponseException: If we get an HTTP response code >= 300
(including 429 and 5xx responses that survive retrying).
FederationDeniedError: If this destination is not on our
federation whitelist
RequestSendFailed: If there were problems connecting to the
remote, due to e.g. DNS failures, connection timeouts etc.
"""
request = MatrixFederationRequest(
method=method,
destination=destination,
path=path,
query=args,
json=json_body,
)
start_ms = self.clock.time_msec()
try:
response = await self._send_request(
request,
ignore_backoff=True,
timeout=timeout,
)
except RequestSendFailed as e:
if isinstance(e.inner_exception, HttpResponseException):
raise e.inner_exception from e
raise
if timeout is not None:
_sec_timeout = timeout / 1000
else:
_sec_timeout = self.default_timeout_seconds
return await _handle_response(
self.clock,
self.reactor,
_sec_timeout,
request,
response,
start_ms,
parser=JsonParser(),
)
async def get_file(
self,
destination: str,
synapse/rest/__init__.py
+2
View File
@@ -27,6 +27,7 @@ from synapse.rest.client import (
account,
account_data,
account_validity,
appservice_federation_proxy,
appservice_ping,
appservice_proxy,
auth,
@@ -132,6 +133,7 @@ CLIENT_SERVLET_FUNCTIONS: tuple[RegisterServletsFunc, ...] = (
thread_subscriptions.register_servlets,
room_membership.register_servlets,
appservice_proxy.register_servlets,
appservice_federation_proxy.register_servlets,
)
SERVLET_GROUPS: dict[str, Iterable[RegisterServletsFunc]] = {
synapse/rest/client/appservice_federation_proxy.py
+137
View File
@@ -0,0 +1,137 @@
#
# 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.errors import Codes, SynapseError
from synapse.http.appservice_proxy import send_federation_request_from_appservice
from synapse.http.server import HttpServer
from synapse.http.servlet import RestServlet, parse_json_object_from_request
from synapse.http.site import SynapseRequest
from synapse.types import JsonDict
if TYPE_CHECKING:
from synapse.server import HomeServer
logger = logging.getLogger(__name__)
ALLOWED_METHODS = ("GET", "PUT", "POST", "DELETE")
class AppserviceFederationProxyRestServlet(RestServlet):
PATTERNS = [
re.compile(
r"^/_matrix/client/unstable/io\.element\.msc4512/appservice/fed_proxy$"
)
]
def __init__(self, hs: "HomeServer"):
super().__init__()
self.hs = hs
self.auth = hs.get_auth()
self.store = hs.get_datastores().main
async def on_POST(self, request: SynapseRequest) -> tuple[int, JsonDict]:
requester = await self.auth.get_user_by_req(request)
app_service = (
self.store.get_app_service_by_id(requester.app_service_id)
if requester.app_service_id
else None
)
if not app_service:
raise SynapseError(
HTTPStatus.FORBIDDEN,
"Only application services can use this endpoint",
Codes.FORBIDDEN,
)
content = parse_json_object_from_request(request)
destination = content.get("destination")
if not isinstance(destination, str) or not destination:
raise SynapseError(
HTTPStatus.BAD_REQUEST,
"Missing or invalid destination",
Codes.MISSING_PARAM,
)
method = content.get("method")
if method not in ALLOWED_METHODS:
raise SynapseError(
HTTPStatus.BAD_REQUEST,
f"method must be one of {ALLOWED_METHODS}",
Codes.INVALID_PARAM,
)
path = content.get("path")
if not isinstance(path, str) or not path:
raise SynapseError(
HTTPStatus.BAD_REQUEST,
"Missing or invalid path",
Codes.MISSING_PARAM,
)
body = content.get("body")
if body is not None and not isinstance(body, dict):
raise SynapseError(
HTTPStatus.BAD_REQUEST,
"body must be an object",
Codes.INVALID_PARAM,
)
if body is not None and method in ("GET", "DELETE"):
raise SynapseError(
HTTPStatus.BAD_REQUEST,
f"'body' is not supported for {method}",
Codes.INVALID_PARAM,
)
query = content.get("query")
if query is not None and not isinstance(query, dict):
raise SynapseError(
HTTPStatus.BAD_REQUEST,
"query must be an object",
Codes.INVALID_PARAM,
)
if query is not None and any(
not isinstance(value, str) for value in query.values()
):
raise SynapseError(
HTTPStatus.BAD_REQUEST,
"'query' values must be strings",
Codes.INVALID_PARAM,
)
status, response_content = await send_federation_request_from_appservice(
self.hs,
app_service,
method,
destination,
path,
body,
query,
)
return HTTPStatus.OK, {"status": status, "content": response_content}
def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None:
if not hs.config.experimental.msc4512_enabled:
return
AppserviceFederationProxyRestServlet(hs).register(http_server)
synapse/rest/client/appservice_proxy.py
+3 -3
View File
@@ -55,9 +55,9 @@ def _make_proxy_callback(
def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None:
"""Registers blanket reverse-proxy routes for each application service that has
configured a proxy prefix. This forwards requests under
/_matrix/client/<version>/<prefix>/* (where <version> is either "vN" or "unstable")
to the same path under the application service's proxy URL after verifying request
authentication.
/_matrix/client/<version>/<prefix>/* (where <version> is either "vN" or
"unstable/<namespace>") to the same path under the application service's proxy
URL after verifying request authentication.
"""
if not hs.config.experimental.msc4512_enabled:
return
tests/federation/transport/server/test_appservice_proxy.py
+4 -2
View File
@@ -38,8 +38,10 @@ class ApplicationServiceFederationProxyTestCase(unittest.FederatingHomeserverTes
def default_config(self) -> JsonDict:
config = super().default_config()
_, path = tempfile.mkstemp(prefix="as_fed_proxy_config")
with open(path, "w") as f:
with tempfile.NamedTemporaryFile(
"w", prefix="as_proxy_config", delete=False
) as f:
path = f.name
yaml.dump(
{
"id": "proxy_as",
tests/rest/client/test_appservice_federation_proxy.py
+596
View File
@@ -0,0 +1,596 @@
#
# 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 json
from unittest import mock
from twisted.internet.testing import MemoryReactor
from synapse.api.errors import Codes
from synapse.appservice import ApplicationService
from synapse.rest import admin
from synapse.rest.client import appservice_federation_proxy, login
from synapse.server import HomeServer
from synapse.types import JsonDict, UserID
from synapse.util.clock import Clock
from tests import unittest
from tests.server import FakeChannel
from tests.test_utils import FakeResponse
APPSERVICE_URL = "http://appservice.example.com"
APPSERVICE_PREFIX = "rtc/livekit"
AS_TOKEN = "as_token"
class ApplicationServiceFederationProxyTestCase(unittest.HomeserverTestCase):
"""Tests MSC4512 implementation of ASes sending federation requests"""
servlets = [
admin.register_servlets,
login.register_servlets,
appservice_federation_proxy.register_servlets,
]
def default_config(self) -> JsonDict:
config = super().default_config()
config.setdefault("experimental_features", {}).setdefault(
"msc4512_enabled", True
)
return config
def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
self.appservice = ApplicationService(
AS_TOKEN,
id="proxy_as",
sender=UserID.from_string("@proxy_bot:test"),
namespaces={},
url=APPSERVICE_URL,
proxy_prefix=APPSERVICE_PREFIX,
proxy_url=APPSERVICE_URL,
)
hs.get_datastores().main.services_cache.append(self.appservice)
self.agent_request = mock.AsyncMock()
hs.get_federation_http_client().agent.request = self.agent_request # type: ignore[method-assign]
def _fed_proxy(
self, content: dict, access_token: str | None = AS_TOKEN
) -> FakeChannel:
"""Issues a /fed_proxy POST request with the supplied content and access token
and returns the result."""
return self.make_request(
"POST",
"/_matrix/client/unstable/io.element.msc4512/appservice/fed_proxy",
content,
access_token=access_token,
)
def test_get_is_sent_and_relayed(self) -> None:
"""A GET federation request is relayed to the destination server."""
self.agent_request.return_value = FakeResponse.json(
code=200, payload={"hello": "world"}
)
channel = self._fed_proxy(
{
"destination": "remote.example.com",
"method": "GET",
"path": f"/_matrix/federation/v1/{APPSERVICE_PREFIX}/some/path",
"query": {"foo": "bar"},
}
)
self.assertEqual(channel.code, 200)
self.assertEqual(
channel.json_body, {"status": 200, "content": {"hello": "world"}}
)
((method, uri), kwargs) = self.agent_request.call_args
self.assertEqual(method, b"GET")
self.assertEqual(
uri,
f"matrix-federation://remote.example.com/_matrix/federation/v1/{APPSERVICE_PREFIX}/some/path?foo=bar".encode(),
)
self.assertIsNone(kwargs["bodyProducer"])
headers = kwargs["headers"]
expected_auth_headers = self.hs.get_federation_http_client().build_auth_headers(
b"remote.example.com",
b"GET",
f"/_matrix/federation/v1/{APPSERVICE_PREFIX}/some/path?foo=bar".encode(),
)
self.assertEqual(headers.getRawHeaders(b"Authorization"), expected_auth_headers)
def test_delete_is_sent_and_relayed(self) -> None:
"""A DELETE federation request is relayed to the destination server."""
self.agent_request.return_value = FakeResponse.json(
code=200, payload={"hello": "world"}
)
channel = self._fed_proxy(
{
"destination": "remote.example.com",
"method": "DELETE",
"path": f"/_matrix/federation/v1/{APPSERVICE_PREFIX}/some/path",
"query": {"foo": "bar"},
}
)
self.assertEqual(channel.code, 200)
self.assertEqual(
channel.json_body, {"status": 200, "content": {"hello": "world"}}
)
((method, uri), kwargs) = self.agent_request.call_args
self.assertEqual(method, b"DELETE")
self.assertEqual(
uri,
f"matrix-federation://remote.example.com/_matrix/federation/v1/{APPSERVICE_PREFIX}/some/path?foo=bar".encode(),
)
self.assertIsNone(kwargs["bodyProducer"])
headers = kwargs["headers"]
expected_auth_headers = self.hs.get_federation_http_client().build_auth_headers(
b"remote.example.com",
b"DELETE",
f"/_matrix/federation/v1/{APPSERVICE_PREFIX}/some/path?foo=bar".encode(),
)
self.assertEqual(headers.getRawHeaders(b"Authorization"), expected_auth_headers)
def test_post_with_body_is_sent_and_relayed(self) -> None:
"""A POST federation request is relayed to the destination server."""
self.agent_request.return_value = FakeResponse.json(
code=200, payload={"hello": "world"}
)
channel = self._fed_proxy(
{
"destination": "remote.example.com",
"method": "POST",
"path": f"/_matrix/federation/v1/{APPSERVICE_PREFIX}/some/path",
"body": {"key": "value"},
"query": {"foo": "bar"},
}
)
self.assertEqual(channel.code, 200)
self.assertEqual(
channel.json_body, {"status": 200, "content": {"hello": "world"}}
)
((method, uri), kwargs) = self.agent_request.call_args
self.assertEqual(method, b"POST")
self.assertEqual(
uri,
f"matrix-federation://remote.example.com/_matrix/federation/v1/{APPSERVICE_PREFIX}/some/path?foo=bar".encode(),
)
body_producer = kwargs["bodyProducer"]
self.assertEqual(
json.loads(body_producer._inputFile.getvalue()),
{"key": "value"},
)
headers = kwargs["headers"]
expected_auth_headers = self.hs.get_federation_http_client().build_auth_headers(
b"remote.example.com",
b"POST",
f"/_matrix/federation/v1/{APPSERVICE_PREFIX}/some/path?foo=bar".encode(),
content={"key": "value"},
)
self.assertEqual(headers.getRawHeaders(b"Authorization"), expected_auth_headers)
def test_put_with_body_is_sent_and_relayed(self) -> None:
"""A PUT federation request is relayed to the destination server."""
self.agent_request.return_value = FakeResponse.json(
code=200, payload={"hello": "world"}
)
channel = self._fed_proxy(
{
"destination": "remote.example.com",
"method": "PUT",
"path": f"/_matrix/federation/v1/{APPSERVICE_PREFIX}/some/path",
"body": {"key": "value"},
"query": {"foo": "bar"},
}
)
self.assertEqual(channel.code, 200)
self.assertEqual(
channel.json_body, {"status": 200, "content": {"hello": "world"}}
)
((method, uri), kwargs) = self.agent_request.call_args
self.assertEqual(method, b"PUT")
self.assertEqual(
uri,
f"matrix-federation://remote.example.com/_matrix/federation/v1/{APPSERVICE_PREFIX}/some/path?foo=bar".encode(),
)
body_producer = kwargs["bodyProducer"]
self.assertEqual(
json.loads(body_producer._inputFile.getvalue()),
{"key": "value"},
)
headers = kwargs["headers"]
expected_auth_headers = self.hs.get_federation_http_client().build_auth_headers(
b"remote.example.com",
b"PUT",
f"/_matrix/federation/v1/{APPSERVICE_PREFIX}/some/path?foo=bar".encode(),
content={"key": "value"},
)
self.assertEqual(headers.getRawHeaders(b"Authorization"), expected_auth_headers)
def test_remote_error_response_is_relayed(self) -> None:
"""An error response from the remote destination is relayed back to the caller."""
self.agent_request.return_value = FakeResponse.json(
code=400, payload={"errcode": "M_UNRECOGNIZED"}
)
channel = self._fed_proxy(
{
"destination": "remote.example.com",
"method": "GET",
"path": f"/_matrix/federation/v1/{APPSERVICE_PREFIX}/some/path",
}
)
self.assertEqual(channel.code, 200)
self.assertEqual(
channel.json_body,
{"status": 400, "content": {"errcode": "M_UNRECOGNIZED"}},
)
@unittest.override_config({"federation": {"max_short_retries": 0}})
def test_remote_5xx_response_is_relayed(self) -> None:
"""A 5xx response from the remote destination is relayed back to the caller
with its original error code."""
self.agent_request.return_value = FakeResponse.json(
code=503, payload={"error": "overloaded"}
)
channel = self._fed_proxy(
{
"destination": "remote.example.com",
"method": "GET",
"path": f"/_matrix/federation/v1/{APPSERVICE_PREFIX}/some/path",
}
)
self.assertEqual(channel.code, 200)
self.assertEqual(
channel.json_body,
{"status": 503, "content": {"error": "overloaded"}},
)
def test_destination_backoff_is_ignored(self) -> None:
"""A destination that Synapse is currently backing off from for normal
federation traffic is still reachable via the federation proxy."""
self.get_success(
self.hs.get_datastores().main.set_destination_retry_timings(
"remote.example.com",
None,
self.clock.time_msec(), # We last retried just now...
24 * 60 * 60 * 1000, # ...and we won't retry for another 24h.
)
)
self.agent_request.return_value = FakeResponse.json(
code=200, payload={"ok": True}
)
channel = self._fed_proxy(
{
"destination": "remote.example.com",
"method": "GET",
"path": f"/_matrix/federation/v1/{APPSERVICE_PREFIX}/some/path",
}
)
self.assertEqual(channel.code, 200)
self.assertEqual(channel.json_body, {"status": 200, "content": {"ok": True}})
self.agent_request.assert_called_once()
@unittest.override_config({"federation": {"max_short_retries": 0}})
def test_connection_failure_causes_502(self) -> None:
"""A failure to connect to the remote destination is relayed back to the caller as HTTP 502."""
self.agent_request.side_effect = Exception("boom")
channel = self._fed_proxy(
{
"destination": "remote.example.com",
"method": "GET",
"path": f"/_matrix/federation/v1/{APPSERVICE_PREFIX}/some/path",
}
)
self.assertEqual(channel.code, 502)
self.assertEqual(
channel.json_body["errcode"], Codes.AS_FEDPROXY_CONNECTION_FAILED
)
def test_denied_destination_is_rejected(self) -> None:
"""An attempt to send a federation request to an invalid destination is rejected."""
channel = self._fed_proxy(
{
"destination": "not a valid server name",
"method": "GET",
"path": f"/_matrix/federation/v1/{APPSERVICE_PREFIX}/some/path",
}
)
self.assertEqual(channel.code, 403)
self.assertEqual(
channel.json_body["errcode"], Codes.AS_FEDPROXY_DESTINATION_DENIED
)
self.agent_request.assert_not_called()
def test_self_destination_is_rejected(self) -> None:
"""An attempt to send a federation request to the local server is rejected."""
channel = self._fed_proxy(
{
"destination": self.hs.hostname,
"method": "GET",
"path": f"/_matrix/federation/v1/{APPSERVICE_PREFIX}/some/path",
}
)
self.assertEqual(channel.code, 403)
self.assertEqual(
channel.json_body["errcode"], Codes.AS_FEDPROXY_DESTINATION_DENIED
)
self.agent_request.assert_not_called()
def test_path_traversal_segment_is_rejected(self) -> None:
"""Path traversal components in the path are rejected."""
channel = self._fed_proxy(
{
"destination": "remote.example.com",
"method": "GET",
"path": f"/_matrix/federation/v1/{APPSERVICE_PREFIX}/../../v1/send/txn1",
}
)
self.assertEqual(channel.code, 403)
self.assertEqual(
channel.json_body["errcode"], Codes.AS_FEDPROXY_PATH_NOT_ALLOWED
)
self.agent_request.assert_not_called()
def test_percent_encoded_path_traversal_segment_is_rejected(self) -> None:
"""Percent-encoded path traversal components in the path are rejected."""
channel = self._fed_proxy(
{
"destination": "remote.example.com",
"method": "GET",
"path": f"/_matrix/federation/v1/{APPSERVICE_PREFIX}/%2e%2e/%2e%2e/v1/send/txn1",
}
)
self.assertEqual(channel.code, 403)
self.assertEqual(
channel.json_body["errcode"], Codes.AS_FEDPROXY_PATH_NOT_ALLOWED
)
self.agent_request.assert_not_called()
def test_path_with_query_string_is_rejected(self) -> None:
"""A path containing a query string is rejected and not relayed to the destination."""
channel = self._fed_proxy(
{
"destination": "remote.example.com",
"method": "GET",
"path": f"/_matrix/federation/v1/{APPSERVICE_PREFIX}/some/path?foo=bar",
}
)
self.assertEqual(channel.code, 403)
self.assertEqual(
channel.json_body["errcode"], Codes.AS_FEDPROXY_PATH_NOT_ALLOWED
)
self.agent_request.assert_not_called()
def test_path_with_fragment_is_rejected(self) -> None:
"""A path containing a fragment is rejected and not relayed to the destination."""
channel = self._fed_proxy(
{
"destination": "remote.example.com",
"method": "GET",
"path": f"/_matrix/federation/v1/{APPSERVICE_PREFIX}/some/path#frag",
}
)
self.assertEqual(channel.code, 403)
self.assertEqual(
channel.json_body["errcode"], Codes.AS_FEDPROXY_PATH_NOT_ALLOWED
)
self.agent_request.assert_not_called()
def test_get_with_body_is_rejected(self) -> None:
"""GET requests with a body are rejected and not relayed to the destination."""
channel = self._fed_proxy(
{
"destination": "remote.example.com",
"method": "GET",
"path": f"/_matrix/federation/v1/{APPSERVICE_PREFIX}/some/path",
"body": {"key": "value"},
}
)
self.assertEqual(channel.code, 400)
self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM)
self.agent_request.assert_not_called()
def test_delete_with_body_is_rejected(self) -> None:
"""DELETE requests with a body are rejected and not relayed to the destination."""
channel = self._fed_proxy(
{
"destination": "remote.example.com",
"method": "DELETE",
"path": f"/_matrix/federation/v1/{APPSERVICE_PREFIX}/some/path",
"body": {"key": "value"},
}
)
self.assertEqual(channel.code, 400)
self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM)
self.agent_request.assert_not_called()
def test_non_string_query_value_is_rejected(self) -> None:
"""A request with a non-string query is rejected and not relayed to the destination."""
channel = self._fed_proxy(
{
"destination": "remote.example.com",
"method": "GET",
"path": f"/_matrix/federation/v1/{APPSERVICE_PREFIX}/some/path",
"query": {"active": True},
}
)
self.assertEqual(channel.code, 400)
self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM)
self.agent_request.assert_not_called()
def test_path_outside_prefix_is_rejected(self) -> None:
"""Requests to paths outside the proxy prefix are rejected and not relayed to the destination."""
channel = self._fed_proxy(
{
"destination": "remote.example.com",
"method": "GET",
"path": "/_matrix/federation/v1/send/txnid",
}
)
self.assertEqual(channel.code, 403)
self.assertEqual(
channel.json_body["errcode"], Codes.AS_FEDPROXY_PATH_NOT_ALLOWED
)
self.agent_request.assert_not_called()
def test_path_without_version_segment_is_rejected(self) -> None:
"""Requests to paths that omit the version component are rejected and not relayed to the destination."""
channel = self._fed_proxy(
{
"destination": "remote.example.com",
"method": "GET",
"path": f"/_matrix/federation/{APPSERVICE_PREFIX}/some/path",
}
)
self.assertEqual(channel.code, 403)
self.assertEqual(
channel.json_body["errcode"], Codes.AS_FEDPROXY_PATH_NOT_ALLOWED
)
self.agent_request.assert_not_called()
def test_unstable_version_segment_is_sent_and_relayed(self) -> None:
"""A GET federation request using an unstable version component is relayed to the destination server."""
self.agent_request.return_value = FakeResponse.json(
code=200, payload={"hello": "world"}
)
channel = self._fed_proxy(
{
"destination": "remote.example.com",
"method": "GET",
"path": f"/_matrix/federation/unstable/io.element.msc9999/{APPSERVICE_PREFIX}/some/path",
}
)
self.assertEqual(channel.code, 200)
self.assertEqual(
channel.json_body, {"status": 200, "content": {"hello": "world"}}
)
((method, uri), _) = self.agent_request.call_args
self.assertEqual(method, b"GET")
self.assertEqual(
uri,
f"matrix-federation://remote.example.com/_matrix/federation/unstable/io.element.msc9999/{APPSERVICE_PREFIX}/some/path".encode(),
)
def test_appservice_without_proxy_prefix_is_rejected(self) -> None:
"""Requests from app services without a proxy prefix are rejected and not relayed to the destination."""
other_token = "other_as_token"
other_appservice = ApplicationService(
other_token,
id="other_as",
sender=UserID.from_string("@other_bot:test"),
namespaces={},
url=APPSERVICE_URL,
)
self.hs.get_datastores().main.services_cache.append(other_appservice)
channel = self._fed_proxy(
{
"destination": "remote.example.com",
"method": "GET",
"path": f"/_matrix/federation/v1/{APPSERVICE_PREFIX}/some/path",
},
access_token=other_token,
)
self.assertEqual(channel.code, 400)
self.assertEqual(
channel.json_body["errcode"], Codes.AS_FEDPROXY_NO_PROXY_PREFIX
)
self.agent_request.assert_not_called()
def test_unauthenticated_request_is_rejected(self) -> None:
"""Unauthenticated requests are rejected and not relayed to the destination."""
channel = self._fed_proxy(
{
"destination": "remote.example.com",
"method": "GET",
"path": f"/_matrix/federation/v1/{APPSERVICE_PREFIX}/some/path",
},
access_token=None,
)
self.assertEqual(channel.code, 401)
self.agent_request.assert_not_called()
def test_non_appservice_token_is_rejected(self) -> None:
"""Requests from regular, non-app-service, users are rejected and not relayed to the destination."""
self.register_user("normal_user", "password")
user_token = self.login("normal_user", "password")
channel = self._fed_proxy(
{
"destination": "remote.example.com",
"method": "GET",
"path": f"/_matrix/federation/v1/{APPSERVICE_PREFIX}/some/path",
},
access_token=user_token,
)
self.assertEqual(channel.code, 403)
self.agent_request.assert_not_called()
@unittest.override_config({"experimental_features": {"msc4512_enabled": False}})
def test_endpoint_not_registered_when_msc4512_disabled(self) -> None:
"""The /fed_proxy endpoint 404s when the feature flag is off."""
channel = self._fed_proxy(
{
"destination": "remote.example.com",
"method": "GET",
"path": f"/_matrix/federation/v1/{APPSERVICE_PREFIX}/some/path",
}
)
self.assertEqual(channel.code, 404)
self.agent_request.assert_not_called()