# # 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: # . # 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()