From ad055ea4cc34a415ef82a22e7b38b356ef23dbad Mon Sep 17 00:00:00 2001 From: Matthew Hodgson Date: Wed, 9 Sep 2020 13:25:59 +0100 Subject: [PATCH 1/2] blacklist MSC2753 sytests until it's implemented in synapse (#8285) Dendrite's implementing MSC2753 over at https://github.com/matrix-org/dendrite/pull/1370 to prove the implementation for MSC purposes, and so sytest has sprouted tests for it over at https://github.com/matrix-org/sytest/pull/944. But we don't want them to run on synapse until synapse implements it. --- changelog.d/8285.misc | 1 + sytest-blacklist | 8 ++++++++ 2 files changed, 9 insertions(+) create mode 100644 changelog.d/8285.misc diff --git a/changelog.d/8285.misc b/changelog.d/8285.misc new file mode 100644 index 000000000000..4646664ba1c8 --- /dev/null +++ b/changelog.d/8285.misc @@ -0,0 +1 @@ +Blacklist [MSC2753](https://github.com/matrix-org/matrix-doc/pull/2753) SyTests until it is implemented. \ No newline at end of file diff --git a/sytest-blacklist b/sytest-blacklist index 79b2d4402aaa..b563448016f1 100644 --- a/sytest-blacklist +++ b/sytest-blacklist @@ -36,3 +36,11 @@ Inbound federation of state requires event_id as a mandatory paramater # Blacklisted until https://github.com/matrix-org/synapse/pull/6486 lands Can upload self-signing keys + +# Blacklisted until MSC2753 is implemented +Local users can peek into world_readable rooms by room ID +We can't peek into rooms with shared history_visibility +We can't peek into rooms with invited history_visibility +We can't peek into rooms with joined history_visibility +Local users can peek by room alias +Peeked rooms only turn up in the sync for the device who peeked them From 14b5b48a2285a43f655030ec242b92d217c1994a Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 18 Sep 2020 10:49:29 +0100 Subject: [PATCH 2/2] Fix ratelimiting for federation `/send` requests. (#8342) c.f. #8295 for rationale --- changelog.d/8342.bugfix | 1 + synapse/federation/federation_server.py | 52 +++++++++++++++++++------ synapse/federation/transport/server.py | 13 ++++--- synapse/server.py | 5 +++ 4 files changed, 54 insertions(+), 17 deletions(-) create mode 100644 changelog.d/8342.bugfix diff --git a/changelog.d/8342.bugfix b/changelog.d/8342.bugfix new file mode 100644 index 000000000000..786057facb44 --- /dev/null +++ b/changelog.d/8342.bugfix @@ -0,0 +1 @@ +Fix ratelimitng of federation `/send` requests. diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py index 218df884b02a..ff00f0b3022e 100644 --- a/synapse/federation/federation_server.py +++ b/synapse/federation/federation_server.py @@ -97,10 +97,16 @@ def __init__(self, hs): self.state = hs.get_state_handler() self.device_handler = hs.get_device_handler() + self._federation_ratelimiter = hs.get_federation_ratelimiter() self._server_linearizer = Linearizer("fed_server") self._transaction_linearizer = Linearizer("fed_txn_handler") + # We cache results for transaction with the same ID + self._transaction_resp_cache = ResponseCache( + hs, "fed_txn_handler", timeout_ms=30000 + ) + self.transaction_actions = TransactionActions(self.store) self.registry = hs.get_federation_registry() @@ -135,22 +141,44 @@ async def on_incoming_transaction( request_time = self._clock.time_msec() transaction = Transaction(**transaction_data) + transaction_id = transaction.transaction_id # type: ignore - if not transaction.transaction_id: # type: ignore + if not transaction_id: raise Exception("Transaction missing transaction_id") - logger.debug("[%s] Got transaction", transaction.transaction_id) # type: ignore + logger.debug("[%s] Got transaction", transaction_id) - # use a linearizer to ensure that we don't process the same transaction - # multiple times in parallel. - with ( - await self._transaction_linearizer.queue( - (origin, transaction.transaction_id) # type: ignore - ) - ): - result = await self._handle_incoming_transaction( - origin, transaction, request_time - ) + # We wrap in a ResponseCache so that we de-duplicate retried + # transactions. + return await self._transaction_resp_cache.wrap( + (origin, transaction_id), + self._on_incoming_transaction_inner, + origin, + transaction, + request_time, + ) + + async def _on_incoming_transaction_inner( + self, origin: str, transaction: Transaction, request_time: int + ) -> Tuple[int, Dict[str, Any]]: + # Use a linearizer to ensure that transactions from a remote are + # processed in order. + with await self._transaction_linearizer.queue(origin): + # We rate limit here *after* we've queued up the incoming requests, + # so that we don't fill up the ratelimiter with blocked requests. + # + # This is important as the ratelimiter allows N concurrent requests + # at a time, and only starts ratelimiting if there are more requests + # than that being processed at a time. If we queued up requests in + # the linearizer/response cache *after* the ratelimiting then those + # queued up requests would count as part of the allowed limit of N + # concurrent requests. + with self._federation_ratelimiter.ratelimit(origin) as d: + await d + + result = await self._handle_incoming_transaction( + origin, transaction, request_time + ) return result diff --git a/synapse/federation/transport/server.py b/synapse/federation/transport/server.py index 9325e0f857d1..cc7e9a973ba2 100644 --- a/synapse/federation/transport/server.py +++ b/synapse/federation/transport/server.py @@ -45,7 +45,6 @@ ) from synapse.server import HomeServer from synapse.types import ThirdPartyInstanceID, get_domain_from_id -from synapse.util.ratelimitutils import FederationRateLimiter from synapse.util.versionstring import get_version_string logger = logging.getLogger(__name__) @@ -72,9 +71,7 @@ def __init__(self, hs, servlet_groups=None): super(TransportLayerServer, self).__init__(hs, canonical_json=False) self.authenticator = Authenticator(hs) - self.ratelimiter = FederationRateLimiter( - self.clock, config=hs.config.rc_federation - ) + self.ratelimiter = hs.get_federation_ratelimiter() self.register_servlets() @@ -272,6 +269,8 @@ class BaseFederationServlet: PREFIX = FEDERATION_V1_PREFIX # Allows specifying the API version + RATELIMIT = True # Whether to rate limit requests or not + def __init__(self, handler, authenticator, ratelimiter, server_name): self.handler = handler self.authenticator = authenticator @@ -335,7 +334,7 @@ async def new_func(request, *args, **kwargs): ) with scope: - if origin: + if origin and self.RATELIMIT: with ratelimiter.ratelimit(origin) as d: await d if request._disconnected: @@ -372,6 +371,10 @@ def register(self, server): class FederationSendServlet(BaseFederationServlet): PATH = "/send/(?P[^/]*)/?" + # We ratelimit manually in the handler as we queue up the requests and we + # don't want to fill up the ratelimiter with blocked requests. + RATELIMIT = False + def __init__(self, handler, server_name, **kwargs): super(FederationSendServlet, self).__init__( handler, server_name=server_name, **kwargs diff --git a/synapse/server.py b/synapse/server.py index 9055b97ac317..5e3752c3334f 100644 --- a/synapse/server.py +++ b/synapse/server.py @@ -114,6 +114,7 @@ from synapse.types import DomainSpecificString from synapse.util import Clock from synapse.util.distributor import Distributor +from synapse.util.ratelimitutils import FederationRateLimiter from synapse.util.stringutils import random_string logger = logging.getLogger(__name__) @@ -642,6 +643,10 @@ def get_replication_data_handler(self) -> ReplicationDataHandler: def get_replication_streams(self) -> Dict[str, Stream]: return {stream.NAME: stream(self) for stream in STREAMS_MAP.values()} + @cache_in_self + def get_federation_ratelimiter(self) -> FederationRateLimiter: + return FederationRateLimiter(self.clock, config=self.config.rc_federation) + async def remove_pusher(self, app_id: str, push_key: str, user_id: str): return await self.get_pusherpool().remove_pusher(app_id, push_key, user_id)